diff --git a/README.md b/README.md index 2129057..367a38a 100644 --- a/README.md +++ b/README.md @@ -12,7 +12,7 @@ Go backend for merchant authentication, one-time payment-data ingestion, determi | `go run ./cmd/api` | Start the HTTP API on `HTTP_ADDR` | Long-running | | `go run ./cmd/worker` | Process queued growth reports, page insights, and growth-coach jobs | Long-running alongside the API | | `go run ./cmd/scheduler` | Enqueue due growth-coach work when `AI_COACH_ENABLED=true` | Optional long-running process | -| `go run ./cmd/calendar --help` | Import, approve, or activate reviewed calendar catalogs | Administrative/optional | +| `go run ./cmd/calendar --help` | Bootstrap or rotate independently reviewed calendar catalogs | Administrative/optional | | `go run ./cmd/admin-reset-ai` | Reset AI-related administrative state | Administrative/recovery only | The importer is checksum-idempotent: importing the same archive again is a safe no-op. It is not a file watcher or scheduled ETL process. The worker does not replace the API; both processes should run at the same time. @@ -122,6 +122,17 @@ Run these commands from the repository root and in this order: go mod download go run ./cmd/migrate go run ./cmd/importer +``` + +When calendar-dependent growth features are enabled, bootstrap an independently reviewed external catalog after the importer and before `cmd/admin`/application startup: + +```bash +go run ./cmd/calendar -action bootstrap -file /path/to/reviewed-calendar.json -current-jalali-year 1405 -reviewer reviewer-id -actor operator-id +``` + +The command rejects bundled review candidates, resumes the same reviewed version safely, and refuses to replace a different active version. Catalog rotation continues to use the explicit import, approve, and activate actions. Finish setup by creating the demo user: + +```bash go run ./cmd/admin ``` @@ -243,6 +254,7 @@ docker compose build backend frontend docker compose up -d postgres redis docker compose --profile tools run --rm migrate docker compose --profile import run --rm importer +docker compose --profile tools run --rm calendar docker compose --profile tools run --rm admin docker compose up -d backend worker frontend docker compose ps diff --git a/cmd/calendar/main.go b/cmd/calendar/main.go index 573c1a0..cae025f 100644 --- a/cmd/calendar/main.go +++ b/cmd/calendar/main.go @@ -3,90 +3,185 @@ package main import ( "context" "encoding/json" + "errors" "flag" "fmt" + "io" "log/slog" "os" "os/signal" + "strings" "git.amiirkhl.ir/starcoach-hackathon/zarinpal-analytics-backend/internal/config" "git.amiirkhl.ir/starcoach-hackathon/zarinpal-analytics-backend/internal/database" "git.amiirkhl.ir/starcoach-hackathon/zarinpal-analytics-backend/internal/eventcatalog" "git.amiirkhl.ir/starcoach-hackathon/zarinpal-analytics-backend/internal/insight" + "github.com/jackc/pgx/v5/pgxpool" ) -func main() { - action := flag.String("action", "", "import, approve, or activate") - filePath := flag.String("file", "", "reviewed JSON catalog for import") - version := flag.String("version", "", "catalog version for approve or activate") - actor := flag.String("actor", "", "audited reviewer or operator identity") - currentJalaliYear := flag.Int("current-jalali-year", 0, "current Jalali year used for coverage validation") - flag.Parse() +type options struct { + action string + filePath string + version string + actor string + reviewer string + currentJalaliYear int +} +func main() { + os.Exit(realMain(os.Args[1:])) +} + +func realMain(arguments []string) int { logger := slog.New(slog.NewJSONHandler(os.Stdout, nil)) ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt) defer cancel() + if err := run(ctx, arguments, logger); err != nil { + if errors.Is(err, flag.ErrHelp) { + fmt.Fprint(os.Stdout, calendarUsage) + return 0 + } + logger.Error("calendar command failed", "error", err) + return 1 + } + return 0 +} + +const calendarUsage = `Usage: calendar -action [options] + + -file PATH reviewed JSON catalog for import or bootstrap + -current-jalali-year YEAR current Jalali year for coverage validation + -version VERSION catalog version for approve or activate + -reviewer ID independent reviewer identity for bootstrap + -actor ID approval or activation operator identity +` + +func run(ctx context.Context, arguments []string, logger *slog.Logger) error { + command, err := parseOptions(arguments) + if err != nil { + return err + } + var input eventcatalog.Import + if command.action == "import" || command.action == "bootstrap" { + input, err = readImport(command.filePath) + if err != nil { + return err + } + if err := eventcatalog.ValidateImport(input, command.currentJalaliYear); err != nil { + return fmt.Errorf("validating calendar catalog: %w", err) + } + } + cfg, err := config.Load() if err != nil { - fail(logger, "configuration failed", err) + return fmt.Errorf("loading configuration: %w", err) } pool, err := database.Open(ctx, cfg.DatabaseURL) if err != nil { - fail(logger, "database connection failed", err) + return fmt.Errorf("connecting to database: %w", err) } defer pool.Close() repository := eventcatalog.NewRepository(pool) - switch *action { + switch command.action { case "import": - if *filePath == "" || *currentJalaliYear == 0 { - fail(logger, "invalid import arguments", fmt.Errorf("file and current-jalali-year are required")) - } - file, err := os.Open(*filePath) - if err != nil { - fail(logger, "opening calendar catalog failed", err) - } - defer file.Close() - var input eventcatalog.Import - decoder := json.NewDecoder(file) - decoder.DisallowUnknownFields() - if err := decoder.Decode(&input); err != nil { - fail(logger, "decoding calendar catalog failed", err) - } - if err := eventcatalog.ValidateImport(input, *currentJalaliYear); err != nil { - fail(logger, "calendar validation failed", err) - } if err := repository.ImportDraft(ctx, input); err != nil { - fail(logger, "calendar import failed", err) + return fmt.Errorf("importing calendar catalog: %w", err) } logger.Info("calendar draft imported", "version", input.Version, "events", len(input.Events)) case "approve": - if *version == "" || *actor == "" { - fail(logger, "invalid approval arguments", fmt.Errorf("version and actor are required")) + if err := repository.Approve(ctx, command.version, command.actor); err != nil { + return fmt.Errorf("approving calendar catalog: %w", err) } - if err := repository.Approve(ctx, *version, *actor); err != nil { - fail(logger, "calendar approval failed", err) - } - logger.Info("calendar approved", "version", *version, "reviewer", *actor) + logger.Info("calendar approved", "version", command.version, "reviewer", command.actor) case "activate": - if *version == "" || *actor == "" { - fail(logger, "invalid activation arguments", fmt.Errorf("version and actor are required")) + if err := repository.Activate(ctx, command.version, command.actor); err != nil { + return fmt.Errorf("activating calendar catalog: %w", err) } - if err := repository.Activate(ctx, *version, *actor); err != nil { - fail(logger, "calendar activation failed", err) - } - service := insight.NewService(insight.NewRepository(pool), insight.NewPostgreSQLBuilder(pool)) - created, err := service.EnqueueAll(ctx, "CALENDAR") + created, err := scheduleInsights(ctx, pool) if err != nil { - fail(logger, "calendar refresh scheduling failed", err) + return err } - logger.Info("calendar activated", "version", *version, "actor", *actor, "created_jobs", created) - default: - fail(logger, "invalid action", fmt.Errorf("action must be import, approve, or activate")) + logger.Info("calendar activated", "version", command.version, "actor", command.actor, "created_jobs", created) + case "bootstrap": + result, err := eventcatalog.BootstrapCatalog(ctx, repository, input, command.currentJalaliYear, command.reviewer, command.actor, func(ctx context.Context) (int, error) { + return scheduleInsights(ctx, pool) + }) + if err != nil { + return fmt.Errorf("bootstrapping calendar catalog: %w", err) + } + logger.Info("calendar bootstrap complete", "version", result.Version, "status", result.Status, "transitions", result.Transitions, "no_op", result.NoOp, "created_jobs", result.CreatedJobs) } + return nil } -func fail(logger *slog.Logger, message string, err error) { - logger.Error(message, "error", err) - os.Exit(1) +func parseOptions(arguments []string) (options, error) { + var result options + flags := flag.NewFlagSet("calendar", flag.ContinueOnError) + flags.SetOutput(io.Discard) + flags.StringVar(&result.action, "action", "", "import, approve, activate, or bootstrap") + flags.StringVar(&result.filePath, "file", "", "reviewed JSON catalog for import or bootstrap") + flags.StringVar(&result.version, "version", "", "catalog version for approve or activate") + flags.StringVar(&result.actor, "actor", "", "audited reviewer or activation operator identity") + flags.StringVar(&result.reviewer, "reviewer", "", "independent reviewer identity for bootstrap") + flags.IntVar(&result.currentJalaliYear, "current-jalali-year", 0, "current Jalali year used for coverage validation") + if err := flags.Parse(arguments); err != nil { + return options{}, fmt.Errorf("parsing calendar arguments: %w", err) + } + if flags.NArg() != 0 { + return options{}, errors.New("calendar: positional arguments are not supported") + } + result.action = strings.TrimSpace(result.action) + result.filePath = strings.TrimSpace(result.filePath) + result.version = strings.TrimSpace(result.version) + result.actor = strings.TrimSpace(result.actor) + result.reviewer = strings.TrimSpace(result.reviewer) + switch result.action { + case "import": + if result.filePath == "" || result.currentJalaliYear == 0 { + return options{}, errors.New("calendar: file and current-jalali-year are required for import") + } + case "approve", "activate": + if result.version == "" || result.actor == "" { + return options{}, fmt.Errorf("calendar: version and actor are required for %s", result.action) + } + case "bootstrap": + if result.filePath == "" || result.currentJalaliYear == 0 || result.reviewer == "" || result.actor == "" { + return options{}, errors.New("calendar: file, current-jalali-year, reviewer, and actor are required for bootstrap") + } + default: + return options{}, errors.New("calendar: action must be import, approve, activate, or bootstrap") + } + return result, nil +} + +func readImport(path string) (eventcatalog.Import, error) { + file, err := os.Open(path) + if err != nil { + return eventcatalog.Import{}, fmt.Errorf("opening calendar catalog: %w", err) + } + defer file.Close() + var input eventcatalog.Import + decoder := json.NewDecoder(file) + decoder.DisallowUnknownFields() + if err := decoder.Decode(&input); err != nil { + return eventcatalog.Import{}, fmt.Errorf("decoding calendar catalog: %w", err) + } + var extra any + if err := decoder.Decode(&extra); !errors.Is(err, io.EOF) { + if err == nil { + return eventcatalog.Import{}, errors.New("decoding calendar catalog: trailing JSON document") + } + return eventcatalog.Import{}, fmt.Errorf("decoding calendar catalog trailing data: %w", err) + } + return input, nil +} + +func scheduleInsights(ctx context.Context, pool *pgxpool.Pool) (int, error) { + service := insight.NewService(insight.NewRepository(pool), insight.NewPostgreSQLBuilder(pool)) + created, err := service.EnqueueAll(ctx, "CALENDAR") + if err != nil { + return 0, fmt.Errorf("enqueueing calendar insights: %w", err) + } + return created, nil } diff --git a/cmd/calendar/main_test.go b/cmd/calendar/main_test.go new file mode 100644 index 0000000..07ae708 --- /dev/null +++ b/cmd/calendar/main_test.go @@ -0,0 +1,72 @@ +package main + +import ( + "errors" + "flag" + "os" + "path/filepath" + "testing" +) + +func TestParseOptionsSupportsHelp(t *testing.T) { + t.Parallel() + + if _, err := parseOptions([]string{"--help"}); !errors.Is(err, flag.ErrHelp) { + t.Fatalf("error=%v, want flag.ErrHelp", err) + } +} + +func TestParseOptionsRequiresBootstrapInputs(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + args []string + }{ + {name: "file", args: []string{"-action", "bootstrap", "-current-jalali-year", "1405", "-reviewer", "reviewer", "-actor", "operator"}}, + {name: "current year", args: []string{"-action", "bootstrap", "-file", "/calendar/reviewed.json", "-reviewer", "reviewer", "-actor", "operator"}}, + {name: "reviewer", args: []string{"-action", "bootstrap", "-file", "/calendar/reviewed.json", "-current-jalali-year", "1405", "-actor", "operator"}}, + {name: "operator", args: []string{"-action", "bootstrap", "-file", "/calendar/reviewed.json", "-current-jalali-year", "1405", "-reviewer", "reviewer"}}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + if _, err := parseOptions(test.args); err == nil { + t.Fatal("expected missing bootstrap input error") + } + }) + } + + options, err := parseOptions([]string{"-action", "bootstrap", "-file", "/calendar/reviewed.json", "-current-jalali-year", "1405", "-reviewer", "reviewer", "-actor", "operator"}) + if err != nil { + t.Fatal(err) + } + if options.action != "bootstrap" || options.currentJalaliYear != 1405 || options.reviewer != "reviewer" || options.actor != "operator" { + t.Fatalf("options=%+v", options) + } +} + +func TestReadImportRejectsUnknownFieldsAndTrailingDocuments(t *testing.T) { + t.Parallel() + + directory := t.TempDir() + tests := []struct { + name string + json string + }{ + {name: "unknown field", json: `{"version":"test","unexpected":true}`}, + {name: "trailing document", json: `{"version":"test"} {}`}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + path := filepath.Join(directory, test.name+".json") + if err := os.WriteFile(path, []byte(test.json), 0o600); err != nil { + t.Fatal(err) + } + if _, err := readImport(path); err == nil { + t.Fatal("expected strict JSON error") + } + }) + } +} diff --git a/internal/eventcatalog/bootstrap.go b/internal/eventcatalog/bootstrap.go new file mode 100644 index 0000000..a04e15e --- /dev/null +++ b/internal/eventcatalog/bootstrap.go @@ -0,0 +1,140 @@ +package eventcatalog + +import ( + "context" + "errors" + "fmt" + "strings" +) + +var ( + ErrCatalogNotFound = errors.New("event catalog: version not found") + ErrCatalogConflict = errors.New("event catalog: bootstrap conflict") +) + +type CatalogStatus string + +const ( + CatalogStatusDraft CatalogStatus = "DRAFT" + CatalogStatusApproved CatalogStatus = "APPROVED" + CatalogStatusActive CatalogStatus = "ACTIVE" + CatalogStatusRetired CatalogStatus = "RETIRED" +) + +type CatalogState struct { + Version string + Status CatalogStatus + CoverageStartJalaliYear int + CoverageEndJalaliYear int + SourceURI string + SourceVersion string + SourceChecksum string + ImportedBy string +} + +type BootstrapResult struct { + Version string + Status CatalogStatus + Transitions []string + NoOp bool + CreatedJobs int +} + +type bootstrapStore interface { + CatalogState(context.Context, string) (CatalogState, error) + ActiveCatalogState(context.Context) (CatalogState, error) + ImportDraft(context.Context, Import) error + Approve(context.Context, string, string) error + ActivateInitial(context.Context, string, string) error +} + +type BootstrapScheduler func(context.Context) (int, error) + +func BootstrapCatalog(ctx context.Context, store bootstrapStore, input Import, currentJalaliYear int, reviewer, operator string, schedule BootstrapScheduler) (BootstrapResult, error) { + reviewer = strings.TrimSpace(reviewer) + operator = strings.TrimSpace(operator) + if err := ValidateImport(input, currentJalaliYear); err != nil { + return BootstrapResult{}, err + } + if err := ValidateReviewableChecksum(input.SourceChecksum); err != nil { + return BootstrapResult{}, err + } + if err := ValidateApproval(input.ImportedBy, reviewer); err != nil { + return BootstrapResult{}, err + } + if operator == "" { + return BootstrapResult{}, errors.New("event catalog: activation operator is required") + } + if schedule == nil { + return BootstrapResult{}, errors.New("event catalog: bootstrap scheduler is required") + } + + active, err := store.ActiveCatalogState(ctx) + if err != nil && !errors.Is(err, ErrCatalogNotFound) { + return BootstrapResult{}, fmt.Errorf("loading active calendar catalog: %w", err) + } + if active.Version != "" && active.Version != input.Version { + return BootstrapResult{}, fmt.Errorf("%w: active version %q differs from requested version %q", ErrCatalogConflict, active.Version, input.Version) + } + + imported := false + state, err := store.CatalogState(ctx, input.Version) + if errors.Is(err, ErrCatalogNotFound) { + if err := store.ImportDraft(ctx, input); err != nil { + return BootstrapResult{}, fmt.Errorf("importing calendar draft: %w", err) + } + state = stateFromImport(input, CatalogStatusDraft) + imported = true + } else if err != nil { + return BootstrapResult{}, fmt.Errorf("loading calendar catalog version: %w", err) + } else if !state.matches(input) { + return BootstrapResult{}, fmt.Errorf("%w: version %q metadata does not match the reviewed file", ErrCatalogConflict, input.Version) + } + + result := BootstrapResult{Version: input.Version} + if imported { + result.Transitions = append(result.Transitions, "IMPORTED") + } + if state.Status == CatalogStatusDraft && active.Version == "" { + if err := store.Approve(ctx, input.Version, reviewer); err != nil { + return BootstrapResult{}, fmt.Errorf("approving calendar catalog: %w", err) + } + result.Transitions = append(result.Transitions, "APPROVED") + state.Status = CatalogStatusApproved + } + if state.Status == CatalogStatusApproved && active.Version == "" { + if err := store.ActivateInitial(ctx, input.Version, operator); err != nil { + return BootstrapResult{}, fmt.Errorf("activating calendar catalog: %w", err) + } + result.Transitions = append(result.Transitions, "ACTIVATED") + state.Status = CatalogStatusActive + } + if state.Status != CatalogStatusActive { + return BootstrapResult{}, fmt.Errorf("%w: version %q has status %q", ErrCatalogConflict, input.Version, state.Status) + } + created, err := schedule(ctx) + if err != nil { + return BootstrapResult{}, fmt.Errorf("scheduling calendar refresh: %w", err) + } + result.Status = CatalogStatusActive + result.CreatedJobs = created + result.NoOp = len(result.Transitions) == 0 && created == 0 + return result, nil +} + +func stateFromImport(input Import, status CatalogStatus) CatalogState { + return CatalogState{ + Version: input.Version, + Status: status, + CoverageStartJalaliYear: input.CoverageStartJalaliYear, + CoverageEndJalaliYear: input.CoverageEndJalaliYear, + SourceURI: input.SourceURI, + SourceVersion: input.SourceVersion, + SourceChecksum: input.SourceChecksum, + ImportedBy: input.ImportedBy, + } +} + +func (state CatalogState) matches(input Import) bool { + return state == stateFromImport(input, state.Status) +} diff --git a/internal/eventcatalog/bootstrap_integration_test.go b/internal/eventcatalog/bootstrap_integration_test.go new file mode 100644 index 0000000..d020998 --- /dev/null +++ b/internal/eventcatalog/bootstrap_integration_test.go @@ -0,0 +1,87 @@ +package eventcatalog + +import ( + "context" + "errors" + "fmt" + "os" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgxpool" +) + +func TestRepositoryBootstrapStateAndInitialActivation(t *testing.T) { + pool := eventCatalogTestPool(t) + repository := NewRepository(pool) + ctx := t.Context() + prefix := fmt.Sprintf("bootstrap-%d", time.Now().UnixNano()) + first := catalogImportForIntegration(prefix + "-first") + second := catalogImportForIntegration(prefix + "-second") + t.Cleanup(func() { + _, _ = pool.Exec(context.Background(), `DELETE FROM calendar_catalog_versions WHERE version IN ($1,$2)`, first.Version, second.Version) + }) + + if err := repository.ImportDraft(ctx, first); err != nil { + t.Fatal(err) + } + state, err := repository.CatalogState(ctx, first.Version) + if err != nil || state.Status != CatalogStatusDraft || !state.matches(first) { + t.Fatalf("draft state=%+v error=%v", state, err) + } + if err := repository.Approve(ctx, first.Version, "reviewer"); err != nil { + t.Fatal(err) + } + if err := repository.ActivateInitial(ctx, first.Version, "operator"); err != nil { + t.Fatal(err) + } + active, err := repository.ActiveCatalogState(ctx) + if err != nil || active.Version != first.Version || active.Status != CatalogStatusActive { + t.Fatalf("active state=%+v error=%v", active, err) + } + var auditCount int + if err := pool.QueryRow(ctx, `SELECT count(*) FROM calendar_catalog_audit a JOIN calendar_catalog_versions v ON v.id=a.catalog_version_id WHERE v.version=$1`, first.Version).Scan(&auditCount); err != nil || auditCount != 3 { + t.Fatalf("first audit count=%d error=%v", auditCount, err) + } + + if err := repository.ImportDraft(ctx, second); err != nil { + t.Fatal(err) + } + if err := repository.Approve(ctx, second.Version, "reviewer"); err != nil { + t.Fatal(err) + } + if err := repository.ActivateInitial(ctx, second.Version, "operator"); !errors.Is(err, ErrCatalogConflict) { + t.Fatalf("activation error=%v, want ErrCatalogConflict", err) + } + state, err = repository.CatalogState(ctx, second.Version) + if err != nil || state.Status != CatalogStatusApproved { + t.Fatalf("conflicting state=%+v error=%v", state, err) + } + if err := pool.QueryRow(ctx, `SELECT count(*) FROM calendar_catalog_audit a JOIN calendar_catalog_versions v ON v.id=a.catalog_version_id WHERE v.version=$1`, second.Version).Scan(&auditCount); err != nil || auditCount != 2 { + t.Fatalf("second audit count=%d error=%v", auditCount, err) + } +} + +func eventCatalogTestPool(t *testing.T) *pgxpool.Pool { + t.Helper() + databaseURL := os.Getenv("TEST_DATABASE_URL") + if databaseURL == "" { + t.Skip("TEST_DATABASE_URL is not configured") + } + pool, err := pgxpool.New(context.Background(), databaseURL) + if err != nil { + t.Fatal(err) + } + t.Cleanup(pool.Close) + return pool +} + +func catalogImportForIntegration(version string) Import { + input := validCoveredImport() + input.Version = version + input.SourceChecksum = "sha256:" + fmt.Sprintf("%064x", len(version)) + for index := range input.Events { + input.Events[index].ID = fmt.Sprintf("%s-event-%d", version, index) + } + return input +} diff --git a/internal/eventcatalog/bootstrap_test.go b/internal/eventcatalog/bootstrap_test.go new file mode 100644 index 0000000..0ffd49b --- /dev/null +++ b/internal/eventcatalog/bootstrap_test.go @@ -0,0 +1,156 @@ +package eventcatalog + +import ( + "context" + "errors" + "reflect" + "testing" +) + +func TestBootstrapCatalogTransitionsToActive(t *testing.T) { + t.Parallel() + + store := &bootstrapStoreFake{} + createdJobs := 7 + result, err := BootstrapCatalog(t.Context(), store, validCoveredImport(), 1405, "reviewer", "operator", func(context.Context) (int, error) { + return createdJobs, nil + }) + if err != nil { + t.Fatal(err) + } + if want := []string{"IMPORTED", "APPROVED", "ACTIVATED"}; !reflect.DeepEqual(result.Transitions, want) { + t.Fatalf("transitions=%v, want %v", result.Transitions, want) + } + if result.Status != CatalogStatusActive || result.NoOp || result.CreatedJobs != createdJobs { + t.Fatalf("result=%+v", result) + } +} + +func TestBootstrapCatalogResumesAndNoOps(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + status CatalogStatus + transitions []string + }{ + {name: "draft", status: CatalogStatusDraft, transitions: []string{"APPROVED", "ACTIVATED"}}, + {name: "approved", status: CatalogStatusApproved, transitions: []string{"ACTIVATED"}}, + {name: "active", status: CatalogStatusActive}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + input := validCoveredImport() + store := &bootstrapStoreFake{state: stateFromImport(input, test.status), active: stateFromImport(input, CatalogStatusActive)} + if test.status != CatalogStatusActive { + store.active = CatalogState{} + } + result, err := BootstrapCatalog(t.Context(), store, input, 1405, "reviewer", "operator", func(context.Context) (int, error) { return 0, nil }) + if err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(result.Transitions, test.transitions) { + t.Fatalf("transitions=%v, want %v", result.Transitions, test.transitions) + } + if result.NoOp != (test.status == CatalogStatusActive) { + t.Fatalf("no_op=%t", result.NoOp) + } + }) + } +} + +func TestBootstrapCatalogFailsClosed(t *testing.T) { + t.Parallel() + + input := validCoveredImport() + tests := []struct { + name string + state CatalogState + active CatalogState + mutate func(*Import) + }{ + {name: "retired version", state: stateFromImport(input, CatalogStatusRetired)}, + {name: "different active version", active: CatalogState{Version: "newer", Status: CatalogStatusActive}}, + {name: "checksum mismatch", state: stateFromImport(input, CatalogStatusDraft), mutate: func(input *Import) { input.SourceChecksum = "different" }}, + {name: "metadata mismatch", state: stateFromImport(input, CatalogStatusDraft), mutate: func(input *Import) { input.SourceVersion = "different" }}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + candidate := input + if test.mutate != nil { + test.mutate(&candidate) + } + store := &bootstrapStoreFake{state: test.state, active: test.active} + _, err := BootstrapCatalog(t.Context(), store, candidate, 1405, "reviewer", "operator", func(context.Context) (int, error) { return 0, nil }) + if !errors.Is(err, ErrCatalogConflict) { + t.Fatalf("error=%v, want ErrCatalogConflict", err) + } + if len(store.transitions) != 0 { + t.Fatalf("unexpected transitions=%v", store.transitions) + } + }) + } +} + +func TestBootstrapCatalogRejectsInvalidReviewAndRetriesScheduling(t *testing.T) { + t.Parallel() + + input := validCoveredImport() + input.SourceChecksum = "review-required-before-activation" + store := &bootstrapStoreFake{} + if _, err := BootstrapCatalog(t.Context(), store, input, 1405, "reviewer", "operator", func(context.Context) (int, error) { return 0, nil }); err == nil { + t.Fatal("expected placeholder checksum error") + } + + input = validCoveredImport() + store = &bootstrapStoreFake{state: stateFromImport(input, CatalogStatusActive), active: stateFromImport(input, CatalogStatusActive)} + wantErr := errors.New("scheduling failed") + if _, err := BootstrapCatalog(t.Context(), store, input, 1405, "reviewer", "operator", func(context.Context) (int, error) { return 0, wantErr }); !errors.Is(err, wantErr) { + t.Fatalf("error=%v, want scheduling error", err) + } + result, err := BootstrapCatalog(t.Context(), store, input, 1405, "reviewer", "operator", func(context.Context) (int, error) { return 2, nil }) + if err != nil || result.CreatedJobs != 2 || result.NoOp { + t.Fatalf("result=%+v error=%v", result, err) + } +} + +type bootstrapStoreFake struct { + state CatalogState + active CatalogState + transitions []string +} + +func (f *bootstrapStoreFake) CatalogState(context.Context, string) (CatalogState, error) { + if f.state.Version == "" { + return CatalogState{}, ErrCatalogNotFound + } + return f.state, nil +} + +func (f *bootstrapStoreFake) ActiveCatalogState(context.Context) (CatalogState, error) { + if f.active.Version == "" { + return CatalogState{}, ErrCatalogNotFound + } + return f.active, nil +} + +func (f *bootstrapStoreFake) ImportDraft(_ context.Context, input Import) error { + f.state = stateFromImport(input, CatalogStatusDraft) + f.transitions = append(f.transitions, "IMPORTED") + return nil +} + +func (f *bootstrapStoreFake) Approve(_ context.Context, _ string, _ string) error { + f.state.Status = CatalogStatusApproved + f.transitions = append(f.transitions, "APPROVED") + return nil +} + +func (f *bootstrapStoreFake) ActivateInitial(_ context.Context, _ string, _ string) error { + f.state.Status = CatalogStatusActive + f.active = f.state + f.transitions = append(f.transitions, "ACTIVATED") + return nil +} diff --git a/internal/eventcatalog/import.go b/internal/eventcatalog/import.go index 6d8c13a..68c655a 100644 --- a/internal/eventcatalog/import.go +++ b/internal/eventcatalog/import.go @@ -60,6 +60,26 @@ type Repository struct{ pool *pgxpool.Pool } func NewRepository(pool *pgxpool.Pool) *Repository { return &Repository{pool: pool} } +func (r *Repository) CatalogState(ctx context.Context, version string) (CatalogState, error) { + return r.catalogState(ctx, `SELECT version,status,coverage_start_jalali_year,coverage_end_jalali_year,source_uri,source_version,source_checksum,imported_by FROM calendar_catalog_versions WHERE version=$1`, version) +} + +func (r *Repository) ActiveCatalogState(ctx context.Context) (CatalogState, error) { + return r.catalogState(ctx, `SELECT version,status,coverage_start_jalali_year,coverage_end_jalali_year,source_uri,source_version,source_checksum,imported_by FROM calendar_catalog_versions WHERE status='ACTIVE'`) +} + +func (r *Repository) catalogState(ctx context.Context, query string, arguments ...any) (CatalogState, error) { + var state CatalogState + err := r.pool.QueryRow(ctx, query, arguments...).Scan(&state.Version, &state.Status, &state.CoverageStartJalaliYear, &state.CoverageEndJalaliYear, &state.SourceURI, &state.SourceVersion, &state.SourceChecksum, &state.ImportedBy) + if errors.Is(err, pgx.ErrNoRows) { + return CatalogState{}, ErrCatalogNotFound + } + if err != nil { + return CatalogState{}, fmt.Errorf("querying calendar catalog state: %w", err) + } + return state, nil +} + func (r *Repository) ImportDraft(ctx context.Context, input Import) error { if err := ValidateEvents(input.Events); err != nil { return err @@ -169,3 +189,43 @@ func (r *Repository) Activate(ctx context.Context, version, actor string) error } return tx.Commit(ctx) } + +func (r *Repository) ActivateInitial(ctx context.Context, version, actor string) error { + if strings.TrimSpace(actor) == "" { + return errors.New("event catalog: activation operator is required") + } + tx, err := r.pool.BeginTx(ctx, pgx.TxOptions{}) + if err != nil { + return fmt.Errorf("starting initial calendar activation: %w", err) + } + defer func() { _ = tx.Rollback(context.WithoutCancel(ctx)) }() + if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended('calendar-catalog-activation',0))`); err != nil { + return fmt.Errorf("locking initial calendar activation: %w", err) + } + var activeVersion string + err = tx.QueryRow(ctx, `SELECT version FROM calendar_catalog_versions WHERE status='ACTIVE'`).Scan(&activeVersion) + if err == nil { + if activeVersion == version { + return tx.Commit(ctx) + } + return fmt.Errorf("%w: active version %q differs from requested version %q", ErrCatalogConflict, activeVersion, version) + } + if !errors.Is(err, pgx.ErrNoRows) { + return fmt.Errorf("loading active calendar catalog: %w", err) + } + var id int64 + var sourceChecksum string + if err := tx.QueryRow(ctx, `SELECT id,source_checksum FROM calendar_catalog_versions WHERE version=$1 AND status='APPROVED' FOR UPDATE`, version).Scan(&id, &sourceChecksum); err != nil { + return fmt.Errorf("loading approved calendar catalog: %w", err) + } + if err := ValidateReviewableChecksum(sourceChecksum); err != nil { + return err + } + if _, err := tx.Exec(ctx, `UPDATE calendar_catalog_versions SET status='ACTIVE',activated_at=now() WHERE id=$1`, id); err != nil { + return fmt.Errorf("activating calendar catalog: %w", err) + } + if _, err := tx.Exec(ctx, `INSERT INTO calendar_catalog_audit(catalog_version_id,action,actor) VALUES($1,'ACTIVATED',$2)`, id, actor); err != nil { + return fmt.Errorf("auditing calendar activation: %w", err) + } + return tx.Commit(ctx) +}