feat: add idempotent calendar bootstrap
Bootstrap independently reviewed catalogs through audited import, approval, and initial activation transitions, then resume insight scheduling safely.
This commit is contained in:
140
internal/eventcatalog/bootstrap.go
Normal file
140
internal/eventcatalog/bootstrap.go
Normal file
@@ -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)
|
||||
}
|
||||
87
internal/eventcatalog/bootstrap_integration_test.go
Normal file
87
internal/eventcatalog/bootstrap_integration_test.go
Normal file
@@ -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
|
||||
}
|
||||
156
internal/eventcatalog/bootstrap_test.go
Normal file
156
internal/eventcatalog/bootstrap_test.go
Normal file
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user