diff --git a/internal/cli/app.go b/internal/cli/app.go index 30a670d..a5b21ba 100644 --- a/internal/cli/app.go +++ b/internal/cli/app.go @@ -12,6 +12,7 @@ import ( "github.com/CrisSTEM/signalscope/internal/config" fetchruntime "github.com/CrisSTEM/signalscope/internal/fetch" + scheduleruntime "github.com/CrisSTEM/signalscope/internal/schedule" "github.com/CrisSTEM/signalscope/internal/storage" ) @@ -81,6 +82,8 @@ func (app *App) Run(ctx context.Context, args []string) int { return app.runSeedDemo(ctx, args[1:]) case "fetch": return app.runFetch(ctx, args[1:]) + case "schedule": + return app.runSchedule(ctx, args[1:]) default: fmt.Fprintf(app.Stderr, "unknown command %q\n\n", args[0]) app.printUsage() @@ -290,7 +293,88 @@ func (app *App) runFetch(ctx context.Context, args []string) int { } fmt.Fprint(app.Stdout, fetchruntime.FormatSummary(summary)) + if summary.HasFailures() { + return 1 + } + + return 0 +} + +func (app *App) runSchedule(ctx context.Context, args []string) int { + flags := flag.NewFlagSet("schedule", flag.ContinueOnError) + flags.SetOutput(app.Stderr) + + configPath := flags.String("config", "", "Path to the config pack directory") + dbPath := flags.String("db", "", "Path to the SQLite database file") + jobID := flags.String("job", "", "Optional schedule job ID to narrow execution to a single configured job") + flags.Usage = func() { + fmt.Fprintln(app.Stderr, "Usage: signalscope schedule --config --db [--job ]") + fmt.Fprintln(app.Stderr) + flags.PrintDefaults() + } + + if err := flags.Parse(args); err != nil { + if errors.Is(err, flag.ErrHelp) { + return 0 + } + return exitUsage + } + if flags.NArg() != 0 { + fmt.Fprintf(app.Stderr, "schedule: unexpected arguments: %s\n\n", strings.Join(flags.Args(), " ")) + flags.Usage() + return exitUsage + } + if strings.TrimSpace(*configPath) == "" { + fmt.Fprintln(app.Stderr, "schedule: --config is required") + fmt.Fprintln(app.Stderr) + flags.Usage() + return exitUsage + } + if strings.TrimSpace(*dbPath) == "" { + fmt.Fprintln(app.Stderr, "schedule: --db is required") + fmt.Fprintln(app.Stderr) + flags.Usage() + return exitUsage + } + pack, err := app.loadPack(*configPath) + if err != nil { + fmt.Fprintf(app.Stderr, "schedule: %v\n", err) + return 1 + } + + db, err := app.openDB(*dbPath) + if err != nil { + fmt.Fprintf(app.Stderr, "schedule: open database: %v\n", err) + return 1 + } + defer db.Close() + + if err := app.bootstrap(ctx, db); err != nil { + fmt.Fprintf(app.Stderr, "schedule: bootstrap database: %v\n", err) + return 1 + } + + if err := app.syncPack(ctx, db, pack); err != nil { + fmt.Fprintf(app.Stderr, "schedule: sync config pack: %v\n", err) + return 1 + } + + runtime := scheduleruntime.Runtime{ + DB: db, + Registry: app.registry(), + Now: app.Now, + } + + summary, err := runtime.Execute(ctx, pack, scheduleruntime.Options{ + JobID: *jobID, + }) + if err != nil { + fmt.Fprintf(app.Stderr, "schedule: %v\n", err) + return 1 + } + + fmt.Fprint(app.Stdout, scheduleruntime.FormatSummary(summary)) if summary.HasFailures() { return 1 } @@ -303,14 +387,16 @@ func (app *App) printUsage() { fmt.Fprintln(app.Stderr, " signalscope check-config --config ") fmt.Fprintln(app.Stderr, " signalscope seed-demo --db [--config ]") fmt.Fprintln(app.Stderr, " signalscope fetch --config --db --source [--binding ]") + fmt.Fprintln(app.Stderr, " signalscope schedule --config --db [--job ]") fmt.Fprintln(app.Stderr) fmt.Fprintln(app.Stderr, "Commands:") fmt.Fprintln(app.Stderr, " check-config Load and validate a config pack") fmt.Fprintln(app.Stderr, " seed-demo Seed a SQLite database from the demo pack") fmt.Fprintln(app.Stderr, " fetch Execute the ingestion runtime for one source kind") + fmt.Fprintln(app.Stderr, " schedule Execute one deterministic scheduler pass") fmt.Fprintln(app.Stderr) fmt.Fprintln(app.Stderr, "Note:") - fmt.Fprintln(app.Stderr, " Only source kinds registered in the runtime can be fetched.") + fmt.Fprintln(app.Stderr, " Only source kinds registered in the runtime can be fetched or scheduled.") } func (app *App) ensureDefaults() { @@ -358,6 +444,7 @@ func (app *App) bootstrap(ctx context.Context, db *sql.DB) error { func (app *App) seedPack(ctx context.Context, db *sql.DB, pack config.Pack) error { return app.SeedPack(ctx, db, pack) } + func (app *App) syncPack(ctx context.Context, db *sql.DB, pack config.Pack) error { return app.SyncPack(ctx, db, pack) } diff --git a/internal/cli/app_schedule_test.go b/internal/cli/app_schedule_test.go new file mode 100644 index 0000000..48ec2f4 --- /dev/null +++ b/internal/cli/app_schedule_test.go @@ -0,0 +1,357 @@ +package cli + +import ( + "context" + "database/sql" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/CrisSTEM/signalscope/internal/config" + fetchruntime "github.com/CrisSTEM/signalscope/internal/fetch" + "github.com/CrisSTEM/signalscope/internal/storage" +) + +type cliScheduleRunRow struct { + Status string + FetchRunsStarted int + FetchRunsSucceeded int + FetchRunsFailed int +} + +func TestRunScheduleRequiresConfigFlag(t *testing.T) { + t.Parallel() + + app, stdout, stderr := newTestApp(cliScheduleFixturePack("cli-schedule-usage")) + + code := app.Run(context.Background(), []string{ + "schedule", + "--db", filepath.Join(t.TempDir(), "signalscope.db"), + }) + if code != exitUsage { + t.Fatalf("Run() exit code = %d, want %d", code, exitUsage) + } + if stdout.Len() != 0 { + t.Fatalf("stdout = %q, want empty", stdout.String()) + } + if !strings.Contains(stderr.String(), "schedule: --config is required") { + t.Fatalf("stderr = %q, want missing config message", stderr.String()) + } +} + +func TestRunScheduleExecutesDueJobAndPersistsScheduleRun(t *testing.T) { + t.Parallel() + + ctx := context.Background() + dbPath := filepath.Join(t.TempDir(), "signalscope.db") + + app, stdout, stderr := newTestApp(cliScheduleFixturePack("cli-schedule-success")) + app.Registry = fetchruntime.NewRegistry() + app.Registry.MustRegister("github", fetchruntime.FetchFunc(func(ctx context.Context, request fetchruntime.Request) (fetchruntime.Result, error) { + return fetchruntime.Result{ + RecordsWritten: 1, + MetricsWritten: 1, + ContentItemsWritten: 0, + }, nil + })) + app.Now = sequenceClock( + time.Date(2026, 4, 27, 11, 0, 0, 0, time.UTC), + time.Date(2026, 4, 27, 11, 0, 1, 0, time.UTC), + time.Date(2026, 4, 27, 11, 0, 2, 0, time.UTC), + time.Date(2026, 4, 27, 11, 0, 3, 0, time.UTC), + ) + + code := app.Run(ctx, []string{ + "schedule", + "--config", "/unused", + "--db", dbPath, + "--job", "job-github", + }) + if code != 0 { + t.Fatalf("Run() exit code = %d, want %d", code, 0) + } + if !strings.Contains(stdout.String(), "schedule summary") { + t.Fatalf("stdout = %q, want schedule summary", stdout.String()) + } + if !strings.Contains(stdout.String(), "due: 1") { + t.Fatalf("stdout = %q, want due count", stdout.String()) + } + if !strings.Contains(stdout.String(), "status=succeeded") { + t.Fatalf("stdout = %q, want succeeded job result", stdout.String()) + } + if stderr.Len() != 0 { + t.Fatalf("stderr = %q, want empty", stderr.String()) + } + + db, err := storage.Open(dbPath) + if err != nil { + t.Fatalf("Open() error = %v", err) + } + defer db.Close() + + count, err := queryCount(ctx, db, `SELECT COUNT(*) FROM schedule_runs WHERE schedule_job_id = ?`, "job-github") + if err != nil { + t.Fatalf("queryCount(schedule_runs) error = %v", err) + } + if count != 1 { + t.Fatalf("schedule run count = %d, want %d", count, 1) + } + + row, err := readLatestCLIScheduleRun(ctx, db, "job-github") + if err != nil { + t.Fatalf("readLatestCLIScheduleRun() error = %v", err) + } + if row.Status != storage.ScheduleRunStatusSucceeded { + t.Fatalf("row.Status = %q, want %q", row.Status, storage.ScheduleRunStatusSucceeded) + } + if row.FetchRunsStarted != 1 { + t.Fatalf("row.FetchRunsStarted = %d, want %d", row.FetchRunsStarted, 1) + } + if row.FetchRunsSucceeded != 1 { + t.Fatalf("row.FetchRunsSucceeded = %d, want %d", row.FetchRunsSucceeded, 1) + } + if row.FetchRunsFailed != 0 { + t.Fatalf("row.FetchRunsFailed = %d, want %d", row.FetchRunsFailed, 0) + } +} + +func TestRunScheduleReturnsZeroWhenJobIsNotDue(t *testing.T) { + t.Parallel() + + ctx := context.Background() + dbPath := filepath.Join(t.TempDir(), "signalscope.db") + pack := cliScheduleFixturePack("cli-schedule-not-due") + + db, err := storage.Open(dbPath) + if err != nil { + t.Fatalf("Open() error = %v", err) + } + if err := storage.Bootstrap(ctx, db); err != nil { + _ = db.Close() + t.Fatalf("Bootstrap() error = %v", err) + } + if err := storage.SeedPack(ctx, db, pack); err != nil { + _ = db.Close() + t.Fatalf("SeedPack() error = %v", err) + } + + nowValue := time.Date(2026, 4, 27, 12, 0, 0, 0, time.UTC) + previousRun, err := storage.StartScheduleRun(ctx, db, storage.StartScheduleRunParams{ + DatasetID: "cli-schedule-not-due", + ScheduleJobID: "job-github", + StartedAt: nowValue.Add(-1 * time.Hour), + }) + if err != nil { + _ = db.Close() + t.Fatalf("StartScheduleRun(previousRun) error = %v", err) + } + if err := storage.FinishScheduleRunSuccess(ctx, db, storage.FinalizeScheduleRunSuccessParams{ + ID: previousRun.ID, + FinishedAt: nowValue.Add(-1*time.Hour + time.Second), + FetchRunsStarted: 1, + FetchRunsSucceeded: 1, + FetchRunsFailed: 0, + }); err != nil { + _ = db.Close() + t.Fatalf("FinishScheduleRunSuccess(previousRun) error = %v", err) + } + if err := db.Close(); err != nil { + t.Fatalf("db.Close() error = %v", err) + } + + var githubCalled bool + + app, stdout, stderr := newTestApp(pack) + app.Registry = fetchruntime.NewRegistry() + app.Registry.MustRegister("github", fetchruntime.FetchFunc(func(ctx context.Context, request fetchruntime.Request) (fetchruntime.Result, error) { + githubCalled = true + return fetchruntime.Result{ + RecordsWritten: 1, + MetricsWritten: 1, + ContentItemsWritten: 0, + }, nil + })) + app.Now = sequenceClock(nowValue) + + code := app.Run(ctx, []string{ + "schedule", + "--config", "/unused", + "--db", dbPath, + "--job", "job-github", + }) + if code != 0 { + t.Fatalf("Run() exit code = %d, want %d", code, 0) + } + if githubCalled { + t.Fatal("github fetcher was called for a not-due job") + } + if !strings.Contains(stdout.String(), "not due: 1") { + t.Fatalf("stdout = %q, want not due count", stdout.String()) + } + if !strings.Contains(stdout.String(), "status=not_due") { + t.Fatalf("stdout = %q, want not_due job result", stdout.String()) + } + if stderr.Len() != 0 { + t.Fatalf("stderr = %q, want empty", stderr.String()) + } + + db, err = storage.Open(dbPath) + if err != nil { + t.Fatalf("Open() error = %v", err) + } + defer db.Close() + + count, err := queryCount(ctx, db, `SELECT COUNT(*) FROM schedule_runs WHERE schedule_job_id = ?`, "job-github") + if err != nil { + t.Fatalf("queryCount(schedule_runs) error = %v", err) + } + if count != 1 { + t.Fatalf("schedule run count = %d, want %d", count, 1) + } +} + +func TestRunScheduleReturnsNonZeroOnPartialJob(t *testing.T) { + t.Parallel() + + ctx := context.Background() + dbPath := filepath.Join(t.TempDir(), "signalscope.db") + + app, stdout, stderr := newTestApp(cliScheduleFixturePackWithTwoGitHubBindings("cli-schedule-partial")) + app.Registry = fetchruntime.NewRegistry() + app.Registry.MustRegister("github", fetchruntime.FetchFunc(func(ctx context.Context, request fetchruntime.Request) (fetchruntime.Result, error) { + switch request.Binding.ID { + case "github-binding-1": + return fetchruntime.Result{ + RecordsWritten: 1, + MetricsWritten: 1, + ContentItemsWritten: 0, + }, nil + case "github-binding-2": + return fetchruntime.Result{}, context.Canceled + default: + return fetchruntime.Result{}, context.DeadlineExceeded + } + })) + app.Now = sequenceClock( + time.Date(2026, 4, 27, 13, 0, 0, 0, time.UTC), + time.Date(2026, 4, 27, 13, 0, 1, 0, time.UTC), + time.Date(2026, 4, 27, 13, 0, 2, 0, time.UTC), + time.Date(2026, 4, 27, 13, 0, 3, 0, time.UTC), + time.Date(2026, 4, 27, 13, 0, 4, 0, time.UTC), + time.Date(2026, 4, 27, 13, 0, 5, 0, time.UTC), + ) + + code := app.Run(ctx, []string{ + "schedule", + "--config", "/unused", + "--db", dbPath, + "--job", "job-github", + }) + if code != 1 { + t.Fatalf("Run() exit code = %d, want %d", code, 1) + } + if !strings.Contains(stdout.String(), "partial: 1") { + t.Fatalf("stdout = %q, want partial count", stdout.String()) + } + if !strings.Contains(stdout.String(), "status=partial") { + t.Fatalf("stdout = %q, want partial job result", stdout.String()) + } + if stderr.Len() != 0 { + t.Fatalf("stderr = %q, want empty", stderr.String()) + } + + db, err := storage.Open(dbPath) + if err != nil { + t.Fatalf("Open() error = %v", err) + } + defer db.Close() + + row, err := readLatestCLIScheduleRun(ctx, db, "job-github") + if err != nil { + t.Fatalf("readLatestCLIScheduleRun() error = %v", err) + } + if row.Status != storage.ScheduleRunStatusPartial { + t.Fatalf("row.Status = %q, want %q", row.Status, storage.ScheduleRunStatusPartial) + } + if row.FetchRunsStarted != 2 { + t.Fatalf("row.FetchRunsStarted = %d, want %d", row.FetchRunsStarted, 2) + } + if row.FetchRunsSucceeded != 1 { + t.Fatalf("row.FetchRunsSucceeded = %d, want %d", row.FetchRunsSucceeded, 1) + } + if row.FetchRunsFailed != 1 { + t.Fatalf("row.FetchRunsFailed = %d, want %d", row.FetchRunsFailed, 1) + } +} + +func cliScheduleFixturePack(datasetID string) config.Pack { + pack := cliFixturePack(datasetID) + pack.Schedules.Jobs = []config.ScheduleJob{ + { + ID: "job-github", + SourceID: "github", + Enabled: true, + Cadence: "24h", + TimeoutSeconds: 30, + JitterSeconds: 5, + Notes: "github schedule job", + }, + } + + pack.Sources.Bindings = []config.Binding{ + pack.Sources.Bindings[0], + } + + return pack +} + +func cliScheduleFixturePackWithTwoGitHubBindings(datasetID string) config.Pack { + pack := cliFixturePack(datasetID) + pack.Schedules.Jobs = []config.ScheduleJob{ + { + ID: "job-github", + SourceID: "github", + Enabled: true, + Cadence: "24h", + TimeoutSeconds: 30, + JitterSeconds: 5, + Notes: "github schedule job", + }, + } + + pack.Sources.Bindings = []config.Binding{ + pack.Sources.Bindings[0], + pack.Sources.Bindings[1], + } + + return pack +} + +func readLatestCLIScheduleRun(ctx context.Context, db *sql.DB, jobID string) (cliScheduleRunRow, error) { + var row cliScheduleRunRow + + err := db.QueryRowContext( + ctx, + `SELECT + status, + fetch_runs_started, + fetch_runs_succeeded, + fetch_runs_failed + FROM schedule_runs + WHERE schedule_job_id = ? + ORDER BY id DESC + LIMIT 1`, + jobID, + ).Scan( + &row.Status, + &row.FetchRunsStarted, + &row.FetchRunsSucceeded, + &row.FetchRunsFailed, + ) + if err != nil { + return cliScheduleRunRow{}, err + } + + return row, nil +} diff --git a/internal/fetch/runtime.go b/internal/fetch/runtime.go index 86a302d..f00965e 100644 --- a/internal/fetch/runtime.go +++ b/internal/fetch/runtime.go @@ -110,6 +110,7 @@ func (result Result) validate() error { type Options struct { SourceKind string + SourceID string BindingID string } @@ -179,7 +180,7 @@ func (runtime Runtime) Execute(ctx context.Context, pack config.Pack, options Op return Summary{}, fmt.Errorf("pack dataset_id is required") } - selected, err := selectBindings(pack, sourceKind, options.BindingID) + selected, err := selectBindings(pack, sourceKind, options.SourceID, options.BindingID) if err != nil { return Summary{}, err } @@ -340,8 +341,9 @@ func invokeFetcher(ctx context.Context, fetcher Fetcher, request Request) (resul return fetcher.Fetch(ctx, request) } -func selectBindings(pack config.Pack, sourceKind, bindingID string) ([]selectedBinding, error) { +func selectBindings(pack config.Pack, sourceKind, sourceID, bindingID string) ([]selectedBinding, error) { sourceKind = strings.TrimSpace(sourceKind) + sourceID = strings.TrimSpace(sourceID) bindingID = strings.TrimSpace(bindingID) if sourceKind == "" { @@ -353,14 +355,31 @@ func selectBindings(pack config.Pack, sourceKind, bindingID string) ([]selectedB for _, source := range pack.Sources.Sources { allSources[source.ID] = source + } - if source.Enabled && strings.TrimSpace(source.Kind) == sourceKind { - matchingSources[source.ID] = source + if sourceID != "" { + source, ok := allSources[sourceID] + if !ok { + return nil, fmt.Errorf("source %q not found", sourceID) + } + if !source.Enabled { + return nil, fmt.Errorf("source %q is disabled", sourceID) + } + if strings.TrimSpace(source.Kind) != sourceKind { + return nil, fmt.Errorf("source %q does not belong to source kind %q", sourceID, sourceKind) } - } - if len(matchingSources) == 0 { - return nil, fmt.Errorf("no enabled sources found for source kind %q", sourceKind) + matchingSources[sourceID] = source + } else { + for _, source := range pack.Sources.Sources { + if source.Enabled && strings.TrimSpace(source.Kind) == sourceKind { + matchingSources[source.ID] = source + } + } + + if len(matchingSources) == 0 { + return nil, fmt.Errorf("no enabled sources found for source kind %q", sourceKind) + } } if bindingID != "" { @@ -383,6 +402,9 @@ func selectBindings(pack config.Pack, sourceKind, bindingID string) ([]selectedB if strings.TrimSpace(source.Kind) != sourceKind { return nil, fmt.Errorf("binding %q does not belong to source kind %q", bindingID, sourceKind) } + if sourceID != "" && binding.SourceID != sourceID { + return nil, fmt.Errorf("binding %q does not belong to source %q", bindingID, sourceID) + } return []selectedBinding{ { @@ -414,6 +436,9 @@ func selectBindings(pack config.Pack, sourceKind, bindingID string) ([]selectedB } if len(selected) == 0 { + if sourceID != "" { + return nil, fmt.Errorf("no enabled bindings found for source %q", sourceID) + } return nil, fmt.Errorf("no enabled bindings found for source kind %q", sourceKind) } diff --git a/internal/fetch/runtime_source_filter_test.go b/internal/fetch/runtime_source_filter_test.go new file mode 100644 index 0000000..b737473 --- /dev/null +++ b/internal/fetch/runtime_source_filter_test.go @@ -0,0 +1,169 @@ +package fetch + +import ( + "context" + "testing" + "time" + + "github.com/CrisSTEM/signalscope/internal/config" + "github.com/CrisSTEM/signalscope/internal/storage" +) + +func TestRuntimeExecuteSourceFilterNarrowsExecutionToSelectedSourceID(t *testing.T) { + t.Parallel() + + ctx := context.Background() + pack := runtimeFixturePackWithMultipleGitHubSources("runtime-fetch-source-filter") + db := openRuntimeTestDB(t, pack) + defer db.Close() + + var seenBindings []string + + registry := NewRegistry() + registry.MustRegister("github", FetchFunc(func(ctx context.Context, request Request) (Result, error) { + seenBindings = append(seenBindings, request.Binding.ID) + return Result{ + RecordsWritten: 1, + MetricsWritten: 1, + ContentItemsWritten: 0, + }, nil + })) + + runtime := Runtime{ + DB: db, + Registry: registry, + Now: sequenceClock( + time.Date(2026, 4, 27, 10, 0, 0, 0, time.UTC), + time.Date(2026, 4, 27, 10, 0, 1, 0, time.UTC), + ), + } + + summary, err := runtime.Execute(ctx, pack, Options{ + SourceKind: "github", + SourceID: "github-secondary", + }) + if err != nil { + t.Fatalf("Execute() error = %v", err) + } + + if summary.Attempted != 1 { + t.Fatalf("summary.Attempted = %d, want %d", summary.Attempted, 1) + } + if summary.Succeeded != 1 { + t.Fatalf("summary.Succeeded = %d, want %d", summary.Succeeded, 1) + } + if summary.Failed != 0 { + t.Fatalf("summary.Failed = %d, want %d", summary.Failed, 0) + } + if len(seenBindings) != 1 { + t.Fatalf("len(seenBindings) = %d, want %d", len(seenBindings), 1) + } + if seenBindings[0] != "github-binding-3" { + t.Fatalf("seenBindings[0] = %q, want %q", seenBindings[0], "github-binding-3") + } + + rows, err := readFetchRuns(ctx, db) + if err != nil { + t.Fatalf("readFetchRuns() error = %v", err) + } + if len(rows) != 1 { + t.Fatalf("len(rows) = %d, want %d", len(rows), 1) + } + if rows[0].BindingID != "github-binding-3" { + t.Fatalf("rows[0].BindingID = %q, want %q", rows[0].BindingID, "github-binding-3") + } + if rows[0].Status != storage.FetchRunStatusSucceeded { + t.Fatalf("rows[0].Status = %q, want %q", rows[0].Status, storage.FetchRunStatusSucceeded) + } +} + +func runtimeFixturePackWithMultipleGitHubSources(datasetID string) config.Pack { + return config.Pack{ + Entities: config.EntitiesFile{ + Version: 1, + DatasetID: datasetID, + DatasetName: "Fetch Runtime Source Filter Fixture Dataset", + Entities: []config.Entity{ + { + ID: "org-1", + Slug: "org-1", + Kind: "organization", + Name: "Org 1", + }, + { + ID: "org-2", + Slug: "org-2", + Kind: "organization", + Name: "Org 2", + }, + { + ID: "org-3", + Slug: "org-3", + Kind: "organization", + Name: "Org 3", + }, + }, + Relationships: []config.Relationship{}, + }, + Sources: config.SourcesFile{ + Version: 1, + DatasetID: datasetID, + Sources: []config.Source{ + { + ID: "github-primary", + Kind: "github", + Enabled: true, + Defaults: config.JSONMap{}, + }, + { + ID: "github-secondary", + Kind: "github", + Enabled: true, + Defaults: config.JSONMap{}, + }, + }, + Bindings: []config.Binding{ + { + ID: "github-binding-1", + EntityID: "org-1", + SourceID: "github-primary", + Enabled: true, + Scope: config.JSONMap{ + "repos": []string{"example/repo-1"}, + }, + Notes: "github primary binding 1", + }, + { + ID: "github-binding-2", + EntityID: "org-2", + SourceID: "github-primary", + Enabled: true, + Scope: config.JSONMap{ + "repos": []string{"example/repo-2"}, + }, + Notes: "github primary binding 2", + }, + { + ID: "github-binding-3", + EntityID: "org-3", + SourceID: "github-secondary", + Enabled: true, + Scope: config.JSONMap{ + "repos": []string{"example/repo-3"}, + }, + Notes: "github secondary binding", + }, + }, + }, + Alerts: config.AlertsFile{ + Version: 1, + DatasetID: datasetID, + Rules: []config.AlertRule{}, + }, + Schedules: config.SchedulesFile{ + Version: 1, + DatasetID: datasetID, + Jobs: []config.ScheduleJob{}, + }, + } +} diff --git a/internal/schedule/runtime.go b/internal/schedule/runtime.go new file mode 100644 index 0000000..cb8ffbe --- /dev/null +++ b/internal/schedule/runtime.go @@ -0,0 +1,428 @@ +package schedule + +import ( + "context" + "database/sql" + "fmt" + "sort" + "strings" + "time" + + "github.com/CrisSTEM/signalscope/internal/config" + fetchruntime "github.com/CrisSTEM/signalscope/internal/fetch" + "github.com/CrisSTEM/signalscope/internal/storage" +) + +const ScheduleJobStatusNotDue = "not_due" + +type Runtime struct { + DB *sql.DB + Registry *fetchruntime.Registry + Now func() time.Time +} + +type Options struct { + JobID string +} + +type JobResult struct { + ScheduleJobID string + SourceID string + SourceKind string + Due bool + Status string + ScheduleRunID int64 + FetchRunsStarted int + FetchRunsSucceeded int + FetchRunsFailed int + ErrorMessage string +} + +type Summary struct { + DatasetID string + Considered int + Due int + NotDue int + Succeeded int + Partial int + Failed int + Results []JobResult +} + +func (summary Summary) HasFailures() bool { + return summary.Partial > 0 || summary.Failed > 0 +} + +type selectedJob struct { + Job config.ScheduleJob + Source config.Source +} + +func (runtime Runtime) Execute(ctx context.Context, pack config.Pack, options Options) (Summary, error) { + ctx = normalizeContext(ctx) + + if runtime.DB == nil { + return Summary{}, fmt.Errorf("database handle is required") + } + + registry := runtime.Registry + if registry == nil { + registry = fetchruntime.DefaultRegistry() + } + + now := runtime.Now + if now == nil { + now = time.Now + } + + datasetID := datasetIDFromPack(pack) + if datasetID == "" { + return Summary{}, fmt.Errorf("pack dataset_id is required") + } + + selected, err := selectJobs(pack, options.JobID) + if err != nil { + return Summary{}, err + } + + fetchRunner := fetchruntime.Runtime{ + DB: runtime.DB, + Registry: registry, + Now: now, + } + + summary := Summary{ + DatasetID: datasetID, + Results: make([]JobResult, 0, len(selected)), + } + + for _, item := range selected { + summary.Considered++ + + evaluatedAt := now().UTC() + due, err := isJobDue(ctx, runtime.DB, datasetID, item.Job, evaluatedAt) + if err != nil { + return summary, fmt.Errorf("determine due state for schedule job %q: %w", item.Job.ID, err) + } + + result := JobResult{ + ScheduleJobID: item.Job.ID, + SourceID: item.Source.ID, + SourceKind: item.Source.Kind, + Due: due, + } + + if !due { + result.Status = ScheduleJobStatusNotDue + summary.NotDue++ + summary.Results = append(summary.Results, result) + continue + } + + summary.Due++ + + scheduleRun, err := storage.StartScheduleRun(ctx, runtime.DB, storage.StartScheduleRunParams{ + DatasetID: datasetID, + ScheduleJobID: item.Job.ID, + StartedAt: evaluatedAt, + }) + if err != nil { + return summary, fmt.Errorf("start schedule run for job %q: %w", item.Job.ID, err) + } + + result.ScheduleRunID = scheduleRun.ID + + // jitter_seconds is intentionally ignored in the deterministic one-shot + // scheduler mode so CI and reviewer demos remain reproducible. + fetchSummary, fetchErr := executeScheduledFetch(ctx, fetchRunner, pack, item) + finishedAt := now().UTC() + + if fetchErr != nil { + result.Status = storage.ScheduleRunStatusFailed + result.ErrorMessage = strings.TrimSpace(fetchErr.Error()) + + if err := storage.FinishScheduleRunFailure(ctx, runtime.DB, storage.FinalizeScheduleRunFailureParams{ + ID: scheduleRun.ID, + FinishedAt: finishedAt, + ErrorMessage: result.ErrorMessage, + FetchRunsStarted: 0, + FetchRunsSucceeded: 0, + FetchRunsFailed: 0, + }); err != nil { + return summary, fmt.Errorf("finalize failed schedule run for job %q: %w", item.Job.ID, err) + } + + summary.Failed++ + summary.Results = append(summary.Results, result) + continue + } + + result.FetchRunsStarted = fetchSummary.Attempted + result.FetchRunsSucceeded = fetchSummary.Succeeded + result.FetchRunsFailed = fetchSummary.Failed + + switch { + case fetchSummary.Failed == 0: + if err := storage.FinishScheduleRunSuccess(ctx, runtime.DB, storage.FinalizeScheduleRunSuccessParams{ + ID: scheduleRun.ID, + FinishedAt: finishedAt, + FetchRunsStarted: fetchSummary.Attempted, + FetchRunsSucceeded: fetchSummary.Succeeded, + FetchRunsFailed: fetchSummary.Failed, + }); err != nil { + return summary, fmt.Errorf("finalize successful schedule run for job %q: %w", item.Job.ID, err) + } + + result.Status = storage.ScheduleRunStatusSucceeded + summary.Succeeded++ + + case fetchSummary.Succeeded > 0: + result.Status = storage.ScheduleRunStatusPartial + result.ErrorMessage = buildFetchFailureSummary(fetchSummary) + + if err := storage.FinishScheduleRunPartial(ctx, runtime.DB, storage.FinalizeScheduleRunPartialParams{ + ID: scheduleRun.ID, + FinishedAt: finishedAt, + ErrorMessage: result.ErrorMessage, + FetchRunsStarted: fetchSummary.Attempted, + FetchRunsSucceeded: fetchSummary.Succeeded, + FetchRunsFailed: fetchSummary.Failed, + }); err != nil { + return summary, fmt.Errorf("finalize partial schedule run for job %q: %w", item.Job.ID, err) + } + + summary.Partial++ + + default: + result.Status = storage.ScheduleRunStatusFailed + result.ErrorMessage = buildFetchFailureSummary(fetchSummary) + + if err := storage.FinishScheduleRunFailure(ctx, runtime.DB, storage.FinalizeScheduleRunFailureParams{ + ID: scheduleRun.ID, + FinishedAt: finishedAt, + ErrorMessage: result.ErrorMessage, + FetchRunsStarted: fetchSummary.Attempted, + FetchRunsSucceeded: fetchSummary.Succeeded, + FetchRunsFailed: fetchSummary.Failed, + }); err != nil { + return summary, fmt.Errorf("finalize failed schedule run for job %q: %w", item.Job.ID, err) + } + + summary.Failed++ + } + + summary.Results = append(summary.Results, result) + } + + return summary, nil +} + +func FormatSummary(summary Summary) string { + var builder strings.Builder + + fmt.Fprintln(&builder, "schedule summary") + fmt.Fprintf(&builder, " dataset: %s\n", summary.DatasetID) + fmt.Fprintf(&builder, " considered: %d\n", summary.Considered) + fmt.Fprintf(&builder, " due: %d\n", summary.Due) + fmt.Fprintf(&builder, " not due: %d\n", summary.NotDue) + fmt.Fprintf(&builder, " succeeded: %d\n", summary.Succeeded) + fmt.Fprintf(&builder, " partial: %d\n", summary.Partial) + fmt.Fprintf(&builder, " failed: %d\n", summary.Failed) + fmt.Fprintln(&builder, "job results:") + + for _, result := range summary.Results { + fmt.Fprintf( + &builder, + " - schedule_job_id=%s source_id=%s due=%t status=%s", + result.ScheduleJobID, + result.SourceID, + result.Due, + result.Status, + ) + + if result.ScheduleRunID > 0 { + fmt.Fprintf(&builder, " schedule_run_id=%d", result.ScheduleRunID) + } + + if result.Due { + fmt.Fprintf( + &builder, + " fetch_runs_started=%d fetch_runs_succeeded=%d fetch_runs_failed=%d", + result.FetchRunsStarted, + result.FetchRunsSucceeded, + result.FetchRunsFailed, + ) + } + + if result.ErrorMessage != "" { + fmt.Fprintf(&builder, " error=%q", result.ErrorMessage) + } + + builder.WriteByte('\n') + } + + return builder.String() +} + +func executeScheduledFetch( + ctx context.Context, + runner fetchruntime.Runtime, + pack config.Pack, + item selectedJob, +) (fetchruntime.Summary, error) { + options := fetchruntime.Options{ + SourceKind: item.Source.Kind, + SourceID: item.Source.ID, + } + + if item.Job.TimeoutSeconds <= 0 { + return runner.Execute(ctx, pack, options) + } + + execCtx, cancel := context.WithTimeout(ctx, time.Duration(item.Job.TimeoutSeconds)*time.Second) + defer cancel() + + return runner.Execute(execCtx, pack, options) +} + +func isJobDue( + ctx context.Context, + db *sql.DB, + datasetID string, + job config.ScheduleJob, + evaluatedAt time.Time, +) (bool, error) { + cadence, err := time.ParseDuration(strings.TrimSpace(job.Cadence)) + if err != nil { + return false, fmt.Errorf("parse cadence for schedule job %q: %w", job.ID, err) + } + + latest, found, err := storage.LookupLatestTerminalScheduleRun(ctx, db, datasetID, job.ID) + if err != nil { + return false, err + } + if !found { + return true, nil + } + + nextDueAt := latest.StartedAt.Add(cadence) + return !evaluatedAt.Before(nextDueAt), nil +} + +func buildFetchFailureSummary(summary fetchruntime.Summary) string { + message := fmt.Sprintf("%d of %d fetch runs failed", summary.Failed, summary.Attempted) + + if firstError := firstFailureMessage(summary.Results); firstError != "" { + message = fmt.Sprintf("%s; first error: %s", message, firstError) + } + + return message +} + +func firstFailureMessage(results []fetchruntime.BindingResult) string { + for _, result := range results { + if strings.TrimSpace(result.ErrorMessage) != "" { + return strings.TrimSpace(result.ErrorMessage) + } + } + + return "" +} + +func selectJobs(pack config.Pack, jobID string) ([]selectedJob, error) { + jobID = strings.TrimSpace(jobID) + + allSources := make(map[string]config.Source, len(pack.Sources.Sources)) + for _, source := range pack.Sources.Sources { + allSources[source.ID] = source + } + + if jobID != "" { + for _, job := range pack.Schedules.Jobs { + if job.ID != jobID { + continue + } + + if !job.Enabled { + return nil, fmt.Errorf("schedule job %q is disabled", jobID) + } + + sourceID := strings.TrimSpace(job.SourceID) + source, ok := allSources[sourceID] + if !ok { + return nil, fmt.Errorf("schedule job %q references unknown source %q", jobID, job.SourceID) + } + if !source.Enabled { + return nil, fmt.Errorf("schedule job %q references disabled source %q", jobID, job.SourceID) + } + + return []selectedJob{ + { + Job: job, + Source: source, + }, + }, nil + } + + return nil, fmt.Errorf("schedule job %q not found", jobID) + } + + selected := make([]selectedJob, 0, len(pack.Schedules.Jobs)) + for _, job := range pack.Schedules.Jobs { + if !job.Enabled { + continue + } + + sourceID := strings.TrimSpace(job.SourceID) + source, ok := allSources[sourceID] + if !ok { + return nil, fmt.Errorf("schedule job %q references unknown source %q", job.ID, job.SourceID) + } + if !source.Enabled { + return nil, fmt.Errorf("schedule job %q references disabled source %q", job.ID, job.SourceID) + } + + selected = append(selected, selectedJob{ + Job: job, + Source: source, + }) + } + + if len(selected) == 0 { + return nil, fmt.Errorf("no enabled schedule jobs found") + } + + sort.Slice(selected, func(i, j int) bool { + return selected[i].Job.ID < selected[j].Job.ID + }) + + return selected, nil +} + +func datasetIDFromPack(pack config.Pack) string { + return firstNonEmpty( + pack.Entities.DatasetID, + pack.Sources.DatasetID, + pack.Alerts.DatasetID, + pack.Schedules.DatasetID, + ) +} + +func firstNonEmpty(values ...string) string { + for _, value := range values { + trimmed := strings.TrimSpace(value) + if trimmed != "" { + return trimmed + } + } + + return "" +} + +func normalizeContext(ctx context.Context) context.Context { + if ctx == nil { + return context.Background() + } + + return ctx +} diff --git a/internal/schedule/runtime_test.go b/internal/schedule/runtime_test.go new file mode 100644 index 0000000..f4fcde6 --- /dev/null +++ b/internal/schedule/runtime_test.go @@ -0,0 +1,646 @@ +package schedule + +import ( + "context" + "database/sql" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/CrisSTEM/signalscope/internal/config" + fetchruntime "github.com/CrisSTEM/signalscope/internal/fetch" + "github.com/CrisSTEM/signalscope/internal/storage" +) + +type scheduleRunRow struct { + Status string + ErrorMessage string + FetchRunsStarted int + FetchRunsSucceeded int + FetchRunsFailed int +} + +func TestRuntimeExecuteRunsDueJobsAndSkipsNotDueDeterministically(t *testing.T) { + t.Parallel() + + ctx := context.Background() + datasetID := "schedule-due-not-due" + pack := scheduleFixturePack(datasetID) + db := openScheduleTestDB(t, pack) + defer db.Close() + + notDueEvaluationTime := time.Date(2026, 4, 26, 10, 0, 6, 0, time.UTC) + previousStartedAt := notDueEvaluationTime.Add(-1 * time.Hour) + + previousRun, err := storage.StartScheduleRun(ctx, db, storage.StartScheduleRunParams{ + DatasetID: datasetID, + ScheduleJobID: "job-news", + StartedAt: previousStartedAt, + }) + if err != nil { + t.Fatalf("StartScheduleRun(previousRun) error = %v", err) + } + if err := storage.FinishScheduleRunSuccess(ctx, db, storage.FinalizeScheduleRunSuccessParams{ + ID: previousRun.ID, + FinishedAt: previousStartedAt.Add(time.Second), + FetchRunsStarted: 1, + FetchRunsSucceeded: 1, + FetchRunsFailed: 0, + }); err != nil { + t.Fatalf("FinishScheduleRunSuccess(previousRun) error = %v", err) + } + + var newsCalled bool + + registry := fetchruntime.NewRegistry() + registry.MustRegister("github", fetchruntime.FetchFunc(func(ctx context.Context, request fetchruntime.Request) (fetchruntime.Result, error) { + return fetchruntime.Result{ + RecordsWritten: 1, + MetricsWritten: 1, + ContentItemsWritten: 0, + }, nil + })) + registry.MustRegister("news_rss", fetchruntime.FetchFunc(func(ctx context.Context, request fetchruntime.Request) (fetchruntime.Result, error) { + newsCalled = true + return fetchruntime.Result{ + RecordsWritten: 1, + MetricsWritten: 1, + ContentItemsWritten: 0, + }, nil + })) + + runtime := Runtime{ + DB: db, + Registry: registry, + Now: scheduleSequenceClock( + time.Date(2026, 4, 26, 10, 0, 0, 0, time.UTC), + time.Date(2026, 4, 26, 10, 0, 1, 0, time.UTC), + time.Date(2026, 4, 26, 10, 0, 2, 0, time.UTC), + time.Date(2026, 4, 26, 10, 0, 3, 0, time.UTC), + time.Date(2026, 4, 26, 10, 0, 4, 0, time.UTC), + time.Date(2026, 4, 26, 10, 0, 5, 0, time.UTC), + notDueEvaluationTime, + ), + } + + summary, err := runtime.Execute(ctx, pack, Options{}) + if err != nil { + t.Fatalf("Execute() error = %v", err) + } + + if summary.Considered != 2 { + t.Fatalf("summary.Considered = %d, want %d", summary.Considered, 2) + } + if summary.Due != 1 { + t.Fatalf("summary.Due = %d, want %d", summary.Due, 1) + } + if summary.NotDue != 1 { + t.Fatalf("summary.NotDue = %d, want %d", summary.NotDue, 1) + } + if summary.Succeeded != 1 { + t.Fatalf("summary.Succeeded = %d, want %d", summary.Succeeded, 1) + } + if summary.Partial != 0 { + t.Fatalf("summary.Partial = %d, want %d", summary.Partial, 0) + } + if summary.Failed != 0 { + t.Fatalf("summary.Failed = %d, want %d", summary.Failed, 0) + } + if len(summary.Results) != 2 { + t.Fatalf("len(summary.Results) = %d, want %d", len(summary.Results), 2) + } + if newsCalled { + t.Fatal("news fetcher was called for a not-due job") + } + + githubResult := mustFindJobResult(t, summary.Results, "job-github") + if !githubResult.Due { + t.Fatal("githubResult.Due = false, want true") + } + if githubResult.Status != storage.ScheduleRunStatusSucceeded { + t.Fatalf("githubResult.Status = %q, want %q", githubResult.Status, storage.ScheduleRunStatusSucceeded) + } + if githubResult.FetchRunsStarted != 2 { + t.Fatalf("githubResult.FetchRunsStarted = %d, want %d", githubResult.FetchRunsStarted, 2) + } + if githubResult.FetchRunsSucceeded != 2 { + t.Fatalf("githubResult.FetchRunsSucceeded = %d, want %d", githubResult.FetchRunsSucceeded, 2) + } + if githubResult.FetchRunsFailed != 0 { + t.Fatalf("githubResult.FetchRunsFailed = %d, want %d", githubResult.FetchRunsFailed, 0) + } + + newsResult := mustFindJobResult(t, summary.Results, "job-news") + if newsResult.Due { + t.Fatal("newsResult.Due = true, want false") + } + if newsResult.Status != ScheduleJobStatusNotDue { + t.Fatalf("newsResult.Status = %q, want %q", newsResult.Status, ScheduleJobStatusNotDue) + } + + githubRun, err := readLatestScheduleRun(ctx, db, "job-github") + if err != nil { + t.Fatalf("readLatestScheduleRun(job-github) error = %v", err) + } + if githubRun.Status != storage.ScheduleRunStatusSucceeded { + t.Fatalf("githubRun.Status = %q, want %q", githubRun.Status, storage.ScheduleRunStatusSucceeded) + } + if githubRun.FetchRunsStarted != 2 { + t.Fatalf("githubRun.FetchRunsStarted = %d, want %d", githubRun.FetchRunsStarted, 2) + } + if githubRun.FetchRunsSucceeded != 2 { + t.Fatalf("githubRun.FetchRunsSucceeded = %d, want %d", githubRun.FetchRunsSucceeded, 2) + } + if githubRun.FetchRunsFailed != 0 { + t.Fatalf("githubRun.FetchRunsFailed = %d, want %d", githubRun.FetchRunsFailed, 0) + } + + newsRunCount, err := scheduleQueryCount( + ctx, + db, + `SELECT COUNT(*) FROM schedule_runs WHERE dataset_id = ? AND schedule_job_id = ?`, + datasetID, + "job-news", + ) + if err != nil { + t.Fatalf("scheduleQueryCount(job-news) error = %v", err) + } + if newsRunCount != 1 { + t.Fatalf("newsRunCount = %d, want %d", newsRunCount, 1) + } +} + +func TestRuntimeExecutePersistsPartialWhenAnyBindingFails(t *testing.T) { + t.Parallel() + + ctx := context.Background() + datasetID := "schedule-partial" + pack := scheduleFixturePack(datasetID) + db := openScheduleTestDB(t, pack) + defer db.Close() + + registry := fetchruntime.NewRegistry() + registry.MustRegister("github", fetchruntime.FetchFunc(func(ctx context.Context, request fetchruntime.Request) (fetchruntime.Result, error) { + switch request.Binding.ID { + case "github-binding-1": + return fetchruntime.Result{ + RecordsWritten: 1, + MetricsWritten: 1, + ContentItemsWritten: 0, + }, nil + case "github-binding-2": + return fetchruntime.Result{}, context.DeadlineExceeded + default: + return fetchruntime.Result{}, context.Canceled + } + })) + + runtime := Runtime{ + DB: db, + Registry: registry, + Now: scheduleSequenceClock( + time.Date(2026, 4, 26, 11, 0, 0, 0, time.UTC), + time.Date(2026, 4, 26, 11, 0, 1, 0, time.UTC), + time.Date(2026, 4, 26, 11, 0, 2, 0, time.UTC), + time.Date(2026, 4, 26, 11, 0, 3, 0, time.UTC), + time.Date(2026, 4, 26, 11, 0, 4, 0, time.UTC), + time.Date(2026, 4, 26, 11, 0, 5, 0, time.UTC), + ), + } + + summary, err := runtime.Execute(ctx, pack, Options{ + JobID: "job-github", + }) + if err != nil { + t.Fatalf("Execute() error = %v", err) + } + + if summary.Considered != 1 { + t.Fatalf("summary.Considered = %d, want %d", summary.Considered, 1) + } + if summary.Due != 1 { + t.Fatalf("summary.Due = %d, want %d", summary.Due, 1) + } + if summary.Partial != 1 { + t.Fatalf("summary.Partial = %d, want %d", summary.Partial, 1) + } + if summary.Succeeded != 0 { + t.Fatalf("summary.Succeeded = %d, want %d", summary.Succeeded, 0) + } + if summary.Failed != 0 { + t.Fatalf("summary.Failed = %d, want %d", summary.Failed, 0) + } + if len(summary.Results) != 1 { + t.Fatalf("len(summary.Results) = %d, want %d", len(summary.Results), 1) + } + + result := summary.Results[0] + if result.Status != storage.ScheduleRunStatusPartial { + t.Fatalf("result.Status = %q, want %q", result.Status, storage.ScheduleRunStatusPartial) + } + if result.FetchRunsStarted != 2 { + t.Fatalf("result.FetchRunsStarted = %d, want %d", result.FetchRunsStarted, 2) + } + if result.FetchRunsSucceeded != 1 { + t.Fatalf("result.FetchRunsSucceeded = %d, want %d", result.FetchRunsSucceeded, 1) + } + if result.FetchRunsFailed != 1 { + t.Fatalf("result.FetchRunsFailed = %d, want %d", result.FetchRunsFailed, 1) + } + if !strings.Contains(result.ErrorMessage, "1 of 2 fetch runs failed") { + t.Fatalf("result.ErrorMessage = %q, want failure summary", result.ErrorMessage) + } + if !strings.Contains(result.ErrorMessage, "context deadline exceeded") { + t.Fatalf("result.ErrorMessage = %q, want first failure detail", result.ErrorMessage) + } + + storedRun, err := readLatestScheduleRun(ctx, db, "job-github") + if err != nil { + t.Fatalf("readLatestScheduleRun() error = %v", err) + } + if storedRun.Status != storage.ScheduleRunStatusPartial { + t.Fatalf("storedRun.Status = %q, want %q", storedRun.Status, storage.ScheduleRunStatusPartial) + } + if storedRun.FetchRunsStarted != 2 { + t.Fatalf("storedRun.FetchRunsStarted = %d, want %d", storedRun.FetchRunsStarted, 2) + } + if storedRun.FetchRunsSucceeded != 1 { + t.Fatalf("storedRun.FetchRunsSucceeded = %d, want %d", storedRun.FetchRunsSucceeded, 1) + } + if storedRun.FetchRunsFailed != 1 { + t.Fatalf("storedRun.FetchRunsFailed = %d, want %d", storedRun.FetchRunsFailed, 1) + } +} + +func TestRuntimeExecutePersistsFailedWhenAllBindingsFail(t *testing.T) { + t.Parallel() + + ctx := context.Background() + datasetID := "schedule-failed" + pack := scheduleFixturePack(datasetID) + db := openScheduleTestDB(t, pack) + defer db.Close() + + registry := fetchruntime.NewRegistry() + registry.MustRegister("github", fetchruntime.FetchFunc(func(ctx context.Context, request fetchruntime.Request) (fetchruntime.Result, error) { + return fetchruntime.Result{}, context.Canceled + })) + + runtime := Runtime{ + DB: db, + Registry: registry, + Now: scheduleSequenceClock( + time.Date(2026, 4, 26, 12, 0, 0, 0, time.UTC), + time.Date(2026, 4, 26, 12, 0, 1, 0, time.UTC), + time.Date(2026, 4, 26, 12, 0, 2, 0, time.UTC), + time.Date(2026, 4, 26, 12, 0, 3, 0, time.UTC), + time.Date(2026, 4, 26, 12, 0, 4, 0, time.UTC), + time.Date(2026, 4, 26, 12, 0, 5, 0, time.UTC), + ), + } + + summary, err := runtime.Execute(ctx, pack, Options{ + JobID: "job-github", + }) + if err != nil { + t.Fatalf("Execute() error = %v", err) + } + + if summary.Failed != 1 { + t.Fatalf("summary.Failed = %d, want %d", summary.Failed, 1) + } + if summary.Partial != 0 { + t.Fatalf("summary.Partial = %d, want %d", summary.Partial, 0) + } + if summary.Succeeded != 0 { + t.Fatalf("summary.Succeeded = %d, want %d", summary.Succeeded, 0) + } + if len(summary.Results) != 1 { + t.Fatalf("len(summary.Results) = %d, want %d", len(summary.Results), 1) + } + + result := summary.Results[0] + if result.Status != storage.ScheduleRunStatusFailed { + t.Fatalf("result.Status = %q, want %q", result.Status, storage.ScheduleRunStatusFailed) + } + if result.FetchRunsStarted != 2 { + t.Fatalf("result.FetchRunsStarted = %d, want %d", result.FetchRunsStarted, 2) + } + if result.FetchRunsSucceeded != 0 { + t.Fatalf("result.FetchRunsSucceeded = %d, want %d", result.FetchRunsSucceeded, 0) + } + if result.FetchRunsFailed != 2 { + t.Fatalf("result.FetchRunsFailed = %d, want %d", result.FetchRunsFailed, 2) + } + if !strings.Contains(result.ErrorMessage, "2 of 2 fetch runs failed") { + t.Fatalf("result.ErrorMessage = %q, want all-failed summary", result.ErrorMessage) + } + + storedRun, err := readLatestScheduleRun(ctx, db, "job-github") + if err != nil { + t.Fatalf("readLatestScheduleRun() error = %v", err) + } + if storedRun.Status != storage.ScheduleRunStatusFailed { + t.Fatalf("storedRun.Status = %q, want %q", storedRun.Status, storage.ScheduleRunStatusFailed) + } + if storedRun.FetchRunsStarted != 2 { + t.Fatalf("storedRun.FetchRunsStarted = %d, want %d", storedRun.FetchRunsStarted, 2) + } + if storedRun.FetchRunsSucceeded != 0 { + t.Fatalf("storedRun.FetchRunsSucceeded = %d, want %d", storedRun.FetchRunsSucceeded, 0) + } + if storedRun.FetchRunsFailed != 2 { + t.Fatalf("storedRun.FetchRunsFailed = %d, want %d", storedRun.FetchRunsFailed, 2) + } +} + +func TestRuntimeExecuteJobFilterNarrowsSelection(t *testing.T) { + t.Parallel() + + ctx := context.Background() + datasetID := "schedule-job-filter" + pack := scheduleFixturePack(datasetID) + db := openScheduleTestDB(t, pack) + defer db.Close() + + var githubCalled bool + + registry := fetchruntime.NewRegistry() + registry.MustRegister("github", fetchruntime.FetchFunc(func(ctx context.Context, request fetchruntime.Request) (fetchruntime.Result, error) { + githubCalled = true + return fetchruntime.Result{ + RecordsWritten: 1, + MetricsWritten: 1, + ContentItemsWritten: 0, + }, nil + })) + registry.MustRegister("news_rss", fetchruntime.FetchFunc(func(ctx context.Context, request fetchruntime.Request) (fetchruntime.Result, error) { + return fetchruntime.Result{ + RecordsWritten: 1, + MetricsWritten: 1, + ContentItemsWritten: 0, + }, nil + })) + + runtime := Runtime{ + DB: db, + Registry: registry, + Now: scheduleSequenceClock( + time.Date(2026, 4, 26, 13, 0, 0, 0, time.UTC), + time.Date(2026, 4, 26, 13, 0, 1, 0, time.UTC), + time.Date(2026, 4, 26, 13, 0, 2, 0, time.UTC), + time.Date(2026, 4, 26, 13, 0, 3, 0, time.UTC), + ), + } + + summary, err := runtime.Execute(ctx, pack, Options{ + JobID: "job-news", + }) + if err != nil { + t.Fatalf("Execute() error = %v", err) + } + + if githubCalled { + t.Fatal("github fetcher was called despite --job filtering to job-news") + } + if summary.Considered != 1 { + t.Fatalf("summary.Considered = %d, want %d", summary.Considered, 1) + } + if summary.Due != 1 { + t.Fatalf("summary.Due = %d, want %d", summary.Due, 1) + } + if summary.Succeeded != 1 { + t.Fatalf("summary.Succeeded = %d, want %d", summary.Succeeded, 1) + } + if len(summary.Results) != 1 { + t.Fatalf("len(summary.Results) = %d, want %d", len(summary.Results), 1) + } + if summary.Results[0].ScheduleJobID != "job-news" { + t.Fatalf("summary.Results[0].ScheduleJobID = %q, want %q", summary.Results[0].ScheduleJobID, "job-news") + } + + newsRunCount, err := scheduleQueryCount( + ctx, + db, + `SELECT COUNT(*) FROM schedule_runs WHERE dataset_id = ? AND schedule_job_id = ?`, + datasetID, + "job-news", + ) + if err != nil { + t.Fatalf("scheduleQueryCount(job-news) error = %v", err) + } + if newsRunCount != 1 { + t.Fatalf("newsRunCount = %d, want %d", newsRunCount, 1) + } + + githubRunCount, err := scheduleQueryCount( + ctx, + db, + `SELECT COUNT(*) FROM schedule_runs WHERE dataset_id = ? AND schedule_job_id = ?`, + datasetID, + "job-github", + ) + if err != nil { + t.Fatalf("scheduleQueryCount(job-github) error = %v", err) + } + if githubRunCount != 0 { + t.Fatalf("githubRunCount = %d, want %d", githubRunCount, 0) + } +} + +func openScheduleTestDB(t *testing.T, pack config.Pack) *sql.DB { + t.Helper() + + ctx := context.Background() + + db, err := storage.Open(filepath.Join(t.TempDir(), "signalscope.db")) + if err != nil { + t.Fatalf("Open() error = %v", err) + } + + if err := storage.Bootstrap(ctx, db); err != nil { + _ = db.Close() + t.Fatalf("Bootstrap() error = %v", err) + } + if err := storage.SeedPack(ctx, db, pack); err != nil { + _ = db.Close() + t.Fatalf("SeedPack() error = %v", err) + } + + return db +} + +func scheduleFixturePack(datasetID string) config.Pack { + return config.Pack{ + Entities: config.EntitiesFile{ + Version: 1, + DatasetID: datasetID, + DatasetName: "Schedule Fixture Dataset", + Entities: []config.Entity{ + { + ID: "org-1", + Slug: "org-1", + Kind: "organization", + Name: "Org 1", + }, + { + ID: "org-2", + Slug: "org-2", + Kind: "organization", + Name: "Org 2", + }, + }, + Relationships: []config.Relationship{}, + }, + Sources: config.SourcesFile{ + Version: 1, + DatasetID: datasetID, + Sources: []config.Source{ + { + ID: "github", + Kind: "github", + Enabled: true, + Defaults: config.JSONMap{}, + }, + { + ID: "news-rss", + Kind: "news_rss", + Enabled: true, + Defaults: config.JSONMap{}, + }, + }, + Bindings: []config.Binding{ + { + ID: "github-binding-1", + EntityID: "org-1", + SourceID: "github", + Enabled: true, + Scope: config.JSONMap{ + "repos": []string{"example/repo-1"}, + }, + Notes: "github binding 1", + }, + { + ID: "github-binding-2", + EntityID: "org-2", + SourceID: "github", + Enabled: true, + Scope: config.JSONMap{ + "repos": []string{"example/repo-2"}, + }, + Notes: "github binding 2", + }, + { + ID: "news-binding-1", + EntityID: "org-1", + SourceID: "news-rss", + Enabled: true, + Scope: config.JSONMap{ + "query": "Example Org", + }, + Notes: "news binding 1", + }, + }, + }, + Alerts: config.AlertsFile{ + Version: 1, + DatasetID: datasetID, + Rules: []config.AlertRule{}, + }, + Schedules: config.SchedulesFile{ + Version: 1, + DatasetID: datasetID, + Jobs: []config.ScheduleJob{ + { + ID: "job-github", + SourceID: "github", + Enabled: true, + Cadence: "24h", + TimeoutSeconds: 30, + JitterSeconds: 5, + Notes: "github schedule job", + }, + { + ID: "job-news", + SourceID: "news-rss", + Enabled: true, + Cadence: "24h", + TimeoutSeconds: 30, + JitterSeconds: 5, + Notes: "news schedule job", + }, + }, + }, + } +} + +func readLatestScheduleRun(ctx context.Context, db *sql.DB, jobID string) (scheduleRunRow, error) { + var row scheduleRunRow + + err := db.QueryRowContext( + ctx, + `SELECT + status, + error_message, + fetch_runs_started, + fetch_runs_succeeded, + fetch_runs_failed + FROM schedule_runs + WHERE schedule_job_id = ? + ORDER BY id DESC + LIMIT 1`, + jobID, + ).Scan( + &row.Status, + &row.ErrorMessage, + &row.FetchRunsStarted, + &row.FetchRunsSucceeded, + &row.FetchRunsFailed, + ) + if err != nil { + return scheduleRunRow{}, err + } + + return row, nil +} + +func scheduleQueryCount(ctx context.Context, db *sql.DB, query string, args ...any) (int, error) { + var count int + if err := db.QueryRowContext(ctx, query, args...).Scan(&count); err != nil { + return 0, err + } + + return count, nil +} + +func mustFindJobResult(t *testing.T, results []JobResult, jobID string) JobResult { + t.Helper() + + for _, result := range results { + if result.ScheduleJobID == jobID { + return result + } + } + + t.Fatalf("job result %q not found", jobID) + return JobResult{} +} + +func scheduleSequenceClock(values ...time.Time) func() time.Time { + index := 0 + + return func() time.Time { + if len(values) == 0 { + return time.Date(2026, 4, 26, 0, 0, 0, 0, time.UTC) + } + + if index >= len(values) { + return values[len(values)-1] + } + + value := values[index] + index++ + return value + } +}