diff --git a/cmd/worker/main.go b/cmd/worker/main.go index 66d867c..ceab2b3 100644 --- a/cmd/worker/main.go +++ b/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() - if err != nil { - logger.Error("nightly insight scheduling failed", "error", err) - } else if created > 0 { - logger.Info("nightly insight scheduling complete", "created_jobs", created) - } + } + } +} + +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 { + 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() } } } diff --git a/internal/insight/repository.go b/internal/insight/repository.go index b107cca..5fe753f 100644 --- a/internal/insight/repository.go +++ b/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 } diff --git a/internal/insight/service.go b/internal/insight/service.go index 2bef805..1f3b524 100644 --- a/internal/insight/service.go +++ b/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(`{}`)}) diff --git a/internal/insight/service_test.go b/internal/insight/service_test.go new file mode 100644 index 0000000..966bfc0 --- /dev/null +++ b/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) + } + } +}