From 912db2aa18b6253352193d66cc7c3cc34632382b Mon Sep 17 00:00:00 2001 From: youfak Date: Wed, 29 Jul 2026 19:55:33 +0800 Subject: [PATCH] feat: add postgres admin state schema --- internal/adapters/postgresadmin/migrations.go | 69 ++++++++++++++ .../migrations/0001_admin_state.sql | 83 ++++++++++++++++ .../adapters/postgresadmin/migrations_test.go | 95 +++++++++++++++++++ 3 files changed, 247 insertions(+) create mode 100644 internal/adapters/postgresadmin/migrations.go create mode 100644 internal/adapters/postgresadmin/migrations/0001_admin_state.sql create mode 100644 internal/adapters/postgresadmin/migrations_test.go diff --git a/internal/adapters/postgresadmin/migrations.go b/internal/adapters/postgresadmin/migrations.go new file mode 100644 index 0000000..0b90b5b --- /dev/null +++ b/internal/adapters/postgresadmin/migrations.go @@ -0,0 +1,69 @@ +package postgresadmin + +import ( + "embed" + "fmt" + "path/filepath" + "regexp" + "sort" + "strconv" + "strings" +) + +type Migration struct { + Version int + Name string + SQL string +} + +//go:embed migrations/*.sql +var migrationFiles embed.FS + +var loadedMigrations = mustLoadMigrations() + +var migrationNamePattern = regexp.MustCompile(`^(\d{4})_([a-z][a-z0-9_]*)\.sql$`) + +func Migrations() []Migration { + return append([]Migration(nil), loadedMigrations...) +} + +func mustLoadMigrations() []Migration { + entries, err := migrationFiles.ReadDir("migrations") + if err != nil { + panic(fmt.Sprintf("postgresadmin: read embedded migrations: %v", err)) + } + result := make([]Migration, 0, len(entries)) + for _, entry := range entries { + if entry.IsDir() { + continue + } + matches := migrationNamePattern.FindStringSubmatch(entry.Name()) + if len(matches) != 3 { + panic("postgresadmin: invalid migration filename " + entry.Name()) + } + version, err := strconv.Atoi(matches[1]) + if err != nil || version <= 0 { + panic("postgresadmin: invalid migration version " + entry.Name()) + } + payload, err := migrationFiles.ReadFile(filepath.ToSlash("migrations/" + entry.Name())) + if err != nil { + panic(fmt.Sprintf("postgresadmin: read migration %s: %v", entry.Name(), err)) + } + if strings.TrimSpace(string(payload)) == "" { + panic("postgresadmin: empty migration " + entry.Name()) + } + result = append(result, Migration{Version: version, Name: matches[2], SQL: string(payload)}) + } + sort.Slice(result, func(left, right int) bool { + return result[left].Version < result[right].Version + }) + for index := 1; index < len(result); index++ { + if result[index-1].Version == result[index].Version { + panic(fmt.Sprintf("postgresadmin: duplicate migration version %d", result[index].Version)) + } + } + if len(result) == 0 { + panic("postgresadmin: no embedded migrations") + } + return result +} diff --git a/internal/adapters/postgresadmin/migrations/0001_admin_state.sql b/internal/adapters/postgresadmin/migrations/0001_admin_state.sql new file mode 100644 index 0000000..25987fb --- /dev/null +++ b/internal/adapters/postgresadmin/migrations/0001_admin_state.sql @@ -0,0 +1,83 @@ +BEGIN; + +CREATE TABLE IF NOT EXISTS control_revisions ( + revision BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, + kind VARCHAR(64) NOT NULL, + created_at TIMESTAMPTZ NOT NULL, + CHECK (kind IN ('config', 'upstream', 'routing')) +); + +CREATE TABLE IF NOT EXISTS config_revisions ( + revision BIGINT PRIMARY KEY, + config_version VARCHAR(128) NOT NULL UNIQUE, + checksum CHAR(64) NOT NULL, + source VARCHAR(512) NOT NULL, + created_at TIMESTAMPTZ NOT NULL, + FOREIGN KEY (revision) REFERENCES control_revisions(revision), + CHECK (checksum ~ '^[0-9A-Fa-f]{64}$') +); + +CREATE TABLE IF NOT EXISTS upstream_admin_state ( + name VARCHAR(128) PRIMARY KEY, + enabled BOOLEAN NOT NULL, + revision BIGINT NOT NULL, + updated_at TIMESTAMPTZ NOT NULL, + FOREIGN KEY (revision) REFERENCES control_revisions(revision) +); + +CREATE TABLE IF NOT EXISTS routing_admin_state ( + name VARCHAR(128) PRIMARY KEY, + enabled BOOLEAN NOT NULL, + upstreams TEXT[] NOT NULL, + current_upstream VARCHAR(128) NOT NULL, + revision BIGINT NOT NULL, + updated_at TIMESTAMPTZ NOT NULL, + FOREIGN KEY (revision) REFERENCES control_revisions(revision), + CHECK (cardinality(upstreams) > 0), + CHECK (current_upstream = ANY(upstreams)) +); + +CREATE TABLE IF NOT EXISTS admin_audit_log ( + id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, + request_id VARCHAR(128) NOT NULL, + actor_id VARCHAR(256) NOT NULL, + source_ip INET, + action VARCHAR(64) NOT NULL, + resource_type VARCHAR(64) NOT NULL, + resource_name VARCHAR(128) NOT NULL, + changed BOOLEAN NOT NULL, + revision BIGINT NOT NULL, + reason VARCHAR(512) NOT NULL DEFAULT '', + occurred_at TIMESTAMPTZ NOT NULL, + FOREIGN KEY (revision) REFERENCES control_revisions(revision), + CHECK (action IN ('commit_config', 'set_upstream_enabled', 'switch_routing')) +); + +CREATE TABLE IF NOT EXISTS admin_outbox ( + id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, + revision BIGINT NOT NULL, + event_type VARCHAR(128) NOT NULL, + aggregate_type VARCHAR(64) NOT NULL, + aggregate_id VARCHAR(128) NOT NULL, + payload JSONB NOT NULL, + occurred_at TIMESTAMPTZ NOT NULL, + claim_owner VARCHAR(128), + claim_until TIMESTAMPTZ, + published_at TIMESTAMPTZ, + FOREIGN KEY (revision) REFERENCES control_revisions(revision), + CHECK (jsonb_typeof(payload) = 'object'), + CHECK ((claim_owner IS NULL) = (claim_until IS NULL)) +); + +CREATE INDEX IF NOT EXISTS admin_audit_log_revision_idx + ON admin_audit_log (revision, id); + +CREATE INDEX IF NOT EXISTS admin_outbox_pending_idx + ON admin_outbox (id) + WHERE published_at IS NULL; + +CREATE INDEX IF NOT EXISTS admin_outbox_claim_idx + ON admin_outbox (claim_until, id) + WHERE published_at IS NULL; + +COMMIT; diff --git a/internal/adapters/postgresadmin/migrations_test.go b/internal/adapters/postgresadmin/migrations_test.go new file mode 100644 index 0000000..44d96f0 --- /dev/null +++ b/internal/adapters/postgresadmin/migrations_test.go @@ -0,0 +1,95 @@ +package postgresadmin + +import ( + "regexp" + "sort" + "strings" + "testing" +) + +func TestMigrationsAreOrderedTransactionalAndImmutable(t *testing.T) { + t.Parallel() + migrations := Migrations() + if len(migrations) != 1 { + t.Fatalf("len(Migrations()) = %d, want 1", len(migrations)) + } + if migrations[0].Version != 1 || migrations[0].Name != "admin_state" { + t.Fatalf("migration metadata = %+v", migrations[0]) + } + normalized := strings.TrimSpace(migrations[0].SQL) + if !strings.HasPrefix(normalized, "BEGIN;") || !strings.HasSuffix(normalized, "COMMIT;") { + t.Fatalf("migration is not transaction wrapped: %q", normalized) + } + for index := 1; index < len(migrations); index++ { + if migrations[index-1].Version >= migrations[index].Version { + t.Fatalf("migration versions are not strictly increasing: %+v", migrations) + } + } + + migrations[0].SQL = "changed" + if Migrations()[0].SQL == "changed" { + t.Fatal("Migrations() returned mutable package storage") + } +} + +func TestAdminSchemaContainsOnlyManagementTables(t *testing.T) { + t.Parallel() + sql := Migrations()[0].SQL + tablePattern := regexp.MustCompile(`(?im)^CREATE TABLE IF NOT EXISTS ([a-z][a-z0-9_]*)\s*\(`) + matches := tablePattern.FindAllStringSubmatch(sql, -1) + tables := make([]string, 0, len(matches)) + for _, match := range matches { + tables = append(tables, match[1]) + } + sort.Strings(tables) + want := []string{ + "admin_audit_log", + "admin_outbox", + "config_revisions", + "control_revisions", + "routing_admin_state", + "upstream_admin_state", + } + if strings.Join(tables, ",") != strings.Join(want, ",") { + t.Fatalf("created tables = %v, want %v", tables, want) + } + + lower := strings.ToLower(sql) + for _, forbidden := range []string{ + "proxy_id", "proxy_host", "proxy_port", "credential", "password", + "extraction_record", "ownership", "idempotency", + } { + if strings.Contains(lower, forbidden) { + t.Errorf("migration contains forbidden detail identifier %q", forbidden) + } + } + for _, destructive := range []string{"drop table", "truncate ", "delete from"} { + if strings.Contains(lower, destructive) { + t.Errorf("forward migration contains destructive statement %q", destructive) + } + } +} + +func TestAdminSchemaContainsTransactionAndOutboxConstraints(t *testing.T) { + t.Parallel() + lower := strings.ToLower(Migrations()[0].SQL) + for _, required := range []string{ + "revision bigint generated by default as identity primary key", + "config_version varchar(128) not null unique", + "checksum char(64) not null", + "upstreams text[] not null", + "current_upstream varchar(128) not null", + "request_id varchar(128) not null", + "source_ip inet", + "payload jsonb not null", + "claim_owner varchar(128)", + "claim_until timestamptz", + "published_at timestamptz", + "foreign key (revision) references control_revisions(revision)", + "where published_at is null", + } { + if !strings.Contains(lower, required) { + t.Errorf("migration missing required schema fragment %q", required) + } + } +}