feat: add postgres admin state schema
This commit is contained in:
youfak
parent
b53b9f1adc
commit
912db2aa18
69
internal/adapters/postgresadmin/migrations.go
Normal file
69
internal/adapters/postgresadmin/migrations.go
Normal file
@ -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
|
||||
}
|
||||
@ -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;
|
||||
95
internal/adapters/postgresadmin/migrations_test.go
Normal file
95
internal/adapters/postgresadmin/migrations_test.go
Normal file
@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user