fix: process insight jobs during backfill

cmd/worker/main.go

@@ -47,13 +47,7 @@ func main() {
 	defer poll.Stop()
 	cleanup := time.NewTicker(24 * time.Hour)
 	defer cleanup.Stop()
-	nightly := time.NewTicker(time.Hour)
-	defer nightly.Stop()
-	if created, err := insightService.EnqueueAll(ctx, "NIGHTLY"); err != nil {
-		logger.Error("initial insight scheduling failed", "error", err)
-	} else {
-		logger.Info("initial insight scheduling complete", "created_jobs", created)
-	}
+	go runInsightScheduler(ctx, logger, insightService)
 	for {
 		select {
 		case <-ctx.Done():
@@ -80,15 +74,32 @@ func main() {
 			} else if deleted > 0 {
 				logger.Info("growth cleanup complete", "deleted_reports", deleted)
 			}
-		case <-nightly.C:
-			scheduleCtx, scheduleCancel := context.WithTimeout(ctx, 5*time.Minute)
-			created, err := insightService.EnqueueAll(scheduleCtx, "NIGHTLY")
-			scheduleCancel()
+		}
+	}
+}
+
+func runInsightScheduler(ctx context.Context, logger *slog.Logger, service *insight.Service) {
+	run := func() {
+		scheduleCtx, cancel := context.WithTimeout(ctx, 30*time.Minute)
+		defer cancel()
+		created, err := service.EnqueueAll(scheduleCtx, "NIGHTLY")
 		if err != nil {
-				logger.Error("nightly insight scheduling failed", "error", err)
-			} else if created > 0 {
-				logger.Info("nightly insight scheduling complete", "created_jobs", created)
-			}
+			if !errors.Is(err, context.Canceled) {
+				logger.Error("insight scheduling failed", "error", err)
+			}
+			return
+		}
+		logger.Info("insight scheduling complete", "created_jobs", created)
+	}
+	run()
+	ticker := time.NewTicker(time.Hour)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			run()
 		}
 	}
 }

internal/insight/repository.go

@@ -174,7 +174,8 @@ func (r *Repository) Claim(ctx context.Context) (claimedJob, error) {
 	var terminal *string
 	err = tx.QueryRow(ctx, `SELECT j.id,j.snapshot_id,j.merchant_id,s.terminal_source_key,j.module,j.created_at,j.updated_at
 		FROM ai_insight_jobs j JOIN ai_insight_snapshots s ON s.id=j.snapshot_id
-		WHERE j.id=(SELECT id FROM ai_insight_jobs WHERE status='QUEUED' AND scheduled_at<=now() ORDER BY scheduled_at,created_at FOR UPDATE SKIP LOCKED LIMIT 1)`).Scan(&job.JobID, &job.SnapshotID, &job.MerchantID, &terminal, &job.Module, &job.CreatedAt, &job.UpdatedAt)
+		WHERE j.id=(SELECT id FROM ai_insight_jobs WHERE status='QUEUED' AND scheduled_at<=now()
+		ORDER BY CASE WHEN trigger_reason='MANUAL' THEN 0 ELSE 1 END,scheduled_at,created_at FOR UPDATE SKIP LOCKED LIMIT 1)`).Scan(&job.JobID, &job.SnapshotID, &job.MerchantID, &terminal, &job.Module, &job.CreatedAt, &job.UpdatedAt)
 	if errors.Is(err, pgx.ErrNoRows) {
 		return claimedJob{}, ErrNotFound
 	}

internal/insight/service.go

@@ -10,12 +10,24 @@ import (
 )
 
 type Service struct {
-	repository *Repository
+	repository repositoryStore
 	builder    Builder
 	now        func() time.Time
 }
 
-func NewService(repository *Repository, builder Builder) *Service {
+type repositoryStore interface {
+	DatasetMetadata(context.Context) (string, time.Time, error)
+	ActiveCatalogVersion(context.Context) (string, error)
+	MerchantScopes(context.Context) ([]MerchantScope, error)
+	Create(context.Context, CreateRequest) (Job, bool, error)
+	Latest(context.Context, int64, string, Module) (Snapshot, error)
+	GetJob(context.Context, int64, string) (Job, error)
+	EventsAfter(context.Context, int64, string, int64) ([]Event, error)
+	Claim(context.Context) (claimedJob, error)
+	Complete(context.Context, claimedJob, Analysis, string, string) error
+}
+
+func NewService(repository repositoryStore, builder Builder) *Service {
 	return &Service{repository: repository, builder: builder, now: time.Now}
 }
 
@@ -42,11 +54,23 @@ func (s *Service) EnqueueAll(ctx context.Context, reason string) (int, error) {
 	if err != nil {
 		return 0, err
 	}
+	watermark, dataThrough, err := s.repository.DatasetMetadata(ctx)
+	if err != nil {
+		return 0, err
+	}
+	catalog, err := s.repository.ActiveCatalogVersion(ctx)
+	if err != nil {
+		return 0, err
+	}
+	timeContext := BuildTimeContext(s.now(), dataThrough)
 	modules := []Module{ModuleOverview, ModulePayments, ModuleFees, ModulePeers, ModuleTransactions, ModuleGrowth}
 	created := 0
 	for _, scope := range scopes {
 		for _, module := range modules {
-			if _, cached, err := s.Enqueue(ctx, scope.MerchantID, scope.TerminalID, module, reason); err != nil {
+			if err := ctx.Err(); err != nil {
+				return created, err
+			}
+			if _, cached, err := s.enqueuePrepared(ctx, scope.MerchantID, scope.TerminalID, module, reason, false, "", watermark, catalog, timeContext); err != nil {
 				return created, err
 			} else if !cached {
 				created++
@@ -66,6 +90,10 @@ func (s *Service) enqueue(ctx context.Context, merchantID int64, terminalID stri
 		return Job{}, false, err
 	}
 	tc := BuildTimeContext(s.now(), dataThrough)
+	return s.enqueuePrepared(ctx, merchantID, terminalID, module, reason, manual, key, watermark, catalog, tc)
+}
+
+func (s *Service) enqueuePrepared(ctx context.Context, merchantID int64, terminalID string, module Module, reason string, manual bool, key, watermark, catalog string, tc TimeContext) (Job, bool, error) {
 	fingerprint := hash(fmt.Sprintf("%d|%s|%s|%s|%s|%s|%s|%s", merchantID, terminalID, module, tc.AsOfDate, watermark, catalog, AnalyticsVersion, "1.0.0"))
 	requestHash := hash(fmt.Sprintf("%d|%s|%s|%t", merchantID, terminalID, module, manual))
 	return s.repository.Create(ctx, CreateRequest{MerchantID: merchantID, TerminalID: terminalID, Module: module, TriggerReason: reason, ManualRefresh: manual, IdempotencyKey: key, RequestHash: requestHash, Fingerprint: fingerprint, Time: tc, Confidence: confidenceFor(tc.Freshness), CatalogVersion: catalog, Evidence: []byte(`{}`)})

internal/insight/service_test.go

@@ -0,0 +1,73 @@
+package insight
+
+import (
+	"context"
+	"encoding/json"
+	"testing"
+	"time"
+)
+
+type schedulingRepository struct {
+	datasetCalls int
+	catalogCalls int
+	created      []CreateRequest
+}
+
+func (r *schedulingRepository) DatasetMetadata(context.Context) (string, time.Time, error) {
+	r.datasetCalls++
+	return "watermark", time.Date(2026, 6, 30, 0, 0, 0, 0, time.UTC), nil
+}
+func (r *schedulingRepository) ActiveCatalogVersion(context.Context) (string, error) {
+	r.catalogCalls++
+	return "catalog-v1", nil
+}
+func (r *schedulingRepository) MerchantScopes(context.Context) ([]MerchantScope, error) {
+	return []MerchantScope{{MerchantID: 1}, {MerchantID: 1, TerminalID: "terminal-1"}}, nil
+}
+func (r *schedulingRepository) Create(_ context.Context, request CreateRequest) (Job, bool, error) {
+	r.created = append(r.created, request)
+	return Job{}, false, nil
+}
+func (*schedulingRepository) Latest(context.Context, int64, string, Module) (Snapshot, error) {
+	panic("unexpected Latest call")
+}
+func (*schedulingRepository) GetJob(context.Context, int64, string) (Job, error) {
+	panic("unexpected GetJob call")
+}
+func (*schedulingRepository) EventsAfter(context.Context, int64, string, int64) ([]Event, error) {
+	panic("unexpected EventsAfter call")
+}
+func (*schedulingRepository) Claim(context.Context) (claimedJob, error) {
+	panic("unexpected Claim call")
+}
+func (*schedulingRepository) Complete(context.Context, claimedJob, Analysis, string, string) error {
+	panic("unexpected Complete call")
+}
+
+func TestService_EnqueueAllLoadsBatchMetadataOnce(t *testing.T) {
+	repository := &schedulingRepository{}
+	service := NewService(repository, nil)
+	service.now = func() time.Time { return time.Date(2026, 8, 21, 12, 0, 0, 0, time.UTC) }
+
+	created, err := service.EnqueueAll(t.Context(), "NIGHTLY")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if created != 12 {
+		t.Fatalf("created %d jobs, want 12", created)
+	}
+	if repository.datasetCalls != 1 {
+		t.Fatalf("dataset metadata loaded %d times, want once", repository.datasetCalls)
+	}
+	if repository.catalogCalls != 1 {
+		t.Fatalf("catalog metadata loaded %d times, want once", repository.catalogCalls)
+	}
+	if len(repository.created) != 12 {
+		t.Fatalf("repository received %d create requests, want 12", len(repository.created))
+	}
+	for _, request := range repository.created {
+		if request.CatalogVersion != "catalog-v1" || request.Fingerprint == "" || !json.Valid(request.Evidence) {
+			t.Fatalf("invalid batched request: %#v", request)
+		}
+	}
+}