1 files changed, 135 insertions, 0 deletions
diff --git a/backend/internal/worker/analyzer/analyzer.go b/backend/internal/worker/analyzer/analyzer.go
new file mode 100644
index 0000000..924e571
--- /dev/null
+++ b/backend/internal/worker/analyzer/analyzer.go
@@ -0,0 +1,135 @@
+package analyzer
+
+import (
+	"context"
+	"log/slog"
+	"time"
+
+	"ibd-trader/internal/analyzer"
+	"ibd-trader/internal/database"
+	"ibd-trader/internal/redis/taskqueue"
+
+	"github.com/redis/go-redis/v9"
+)
+
+const (
+	Queue         = "analyzer"
+	QueueEncoding = taskqueue.EncodingJSON
+
+	lockTimeout    = 1 * time.Minute
+	dequeueTimeout = 5 * time.Second
+)
+
+func RunAnalyzer(
+	ctx context.Context,
+	redis *redis.Client,
+	analyzer analyzer.Analyzer,
+	db database.StockStore,
+	name string,
+) error {
+	queue, err := taskqueue.New(
+		ctx,
+		redis,
+		Queue,
+		name,
+		taskqueue.WithEncoding[TaskInfo](QueueEncoding),
+	)
+	if err != nil {
+		return err
+	}
+
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		default:
+			waitForTask(ctx, queue, analyzer, db)
+		}
+	}
+}
+
+func waitForTask(
+	ctx context.Context,
+	queue taskqueue.TaskQueue[TaskInfo],
+	analyzer analyzer.Analyzer,
+	db database.StockStore,
+) {
+	task, err := queue.Dequeue(ctx, lockTimeout, dequeueTimeout)
+	if err != nil {
+		slog.ErrorContext(ctx, "Failed to dequeue task", "error", err)
+		return
+	}
+	if task == nil {
+		// No task available.
+		return
+	}
+
+	ch := make(chan error)
+	defer close(ch)
+	go func() {
+		ch <- analyzeStock(ctx, analyzer, db, task.Data.ID)
+	}()
+
+	ticker := time.NewTicker(lockTimeout / 5)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			// Context was canceled. Return early.
+			return
+		case <-ticker.C:
+			// Extend the lock periodically.
+			func() {
+				ctx, cancel := context.WithTimeout(ctx, lockTimeout/5)
+				defer cancel()
+
+				err := queue.Extend(ctx, task.ID)
+				if err != nil {
+					slog.ErrorContext(ctx, "Failed to extend lock", "error", err)
+				}
+			}()
+		case err = <-ch:
+			// scrapeUrl has completed.
+			if err != nil {
+				slog.ErrorContext(ctx, "Failed to analyze", "error", err)
+				_, err = queue.Return(ctx, task.ID, err)
+				if err != nil {
+					slog.ErrorContext(ctx, "Failed to return task", "error", err)
+					return
+				}
+			} else {
+				slog.DebugContext(ctx, "Analyzed ID", "id", task.Data.ID)
+				err = queue.Complete(ctx, task.ID, nil)
+				if err != nil {
+					slog.ErrorContext(ctx, "Failed to complete task", "error", err)
+					return
+				}
+			}
+			return
+		}
+	}
+}
+
+func analyzeStock(ctx context.Context, a analyzer.Analyzer, db database.StockStore, id string) error {
+	info, err := db.GetStockInfo(ctx, id)
+	if err != nil {
+		return err
+	}
+
+	analysis, err := a.Analyze(
+		ctx,
+		info.Symbol,
+		info.Price,
+		info.ChartAnalysis,
+	)
+	if err != nil {
+		return err
+	}
+
+	return db.AddAnalysis(ctx, id, analysis)
+}
+
+type TaskInfo struct {
+	ID string `json:"id"`
+}