diff --git a/go.mod b/go.mod index 64e67ae..9be3ad6 100644 --- a/go.mod +++ b/go.mod @@ -3,11 +3,18 @@ module proxy-pool go 1.26.0 require ( + github.com/jackc/pgx/v5 v5.6.0 github.com/redis/go-redis/v9 v9.19.0 go.yaml.in/yaml/v4 v4.0.0-rc.3 ) require ( github.com/cespare/xxhash/v2 v2.3.0 // indirect + github.com/jackc/pgpassfile v1.0.0 // indirect + github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect + github.com/jackc/puddle/v2 v2.2.1 // indirect go.uber.org/atomic v1.11.0 // indirect + golang.org/x/crypto v0.17.0 // indirect + golang.org/x/sync v0.1.0 // indirect + golang.org/x/text v0.14.0 // indirect ) diff --git a/go.sum b/go.sum index 9b3c79c..10fd021 100644 --- a/go.sum +++ b/go.sum @@ -4,21 +4,43 @@ github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA= github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0= github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= +github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg= +github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a h1:bbPeKD0xmW/Y25WS6cokEszi5g+S0QxI/d45PkRi7Nk= +github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM= +github.com/jackc/pgx/v5 v5.6.0 h1:SWJzexBzPL5jb0GEsrPMLIsi/3jOo7RHlzTjcAeDrPY= +github.com/jackc/pgx/v5 v5.6.0/go.mod h1:DNZ/vlrUnhWCoFGxHAG8U2ljioxukquj7utPDgtQdTw= +github.com/jackc/puddle/v2 v2.2.1 h1:RhxXJtFG022u4ibrCSMSiu5aOq1i77R3OHKNJj77OAk= +github.com/jackc/puddle/v2 v2.2.1/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= github.com/klauspost/cpuid/v2 v2.2.10 h1:tBs3QSyvjDyFTq3uoc/9xFpCuOsJQFNPiAhYdw2skhE= github.com/klauspost/cpuid/v2 v2.2.10/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/redis/go-redis/v9 v9.19.0 h1:XPVaaPSnG6RhYf7p+rmSa9zZfeVAnWsH5h3lxthOm/k= github.com/redis/go-redis/v9 v9.19.0/go.mod h1:v/M13XI1PVCDcm01VtPFOADfZtHf8YW3baQf57KlIkA= -github.com/stretchr/testify v1.3.0 h1:TivCn/peBQ7UY8ooIcPgZFpTNSz0Q2U6UrFlUfqbe0Q= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk= +github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4= github.com/zeebo/xxh3 v1.1.0 h1:s7DLGDK45Dyfg7++yxI0khrfwq9661w9EN78eP/UZVs= github.com/zeebo/xxh3 v1.1.0/go.mod h1:IisAie1LELR4xhVinxWS5+zf1lA4p0MW4T+w+W07F5s= go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE= go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0= go.yaml.in/yaml/v4 v4.0.0-rc.3 h1:3h1fjsh1CTAPjW7q/EMe+C8shx5d8ctzZTrLcs/j8Go= go.yaml.in/yaml/v4 v4.0.0-rc.3/go.mod h1:aZqd9kCMsGL7AuUv/m/PvWLdg5sjJsZ4oHDEnfPPfY0= +golang.org/x/crypto v0.17.0 h1:r8bRNjWL3GshPW3gkd+RpvzWrZAwPS49OmTGZ/uhM4k= +golang.org/x/crypto v0.17.0/go.mod h1:gCAAfMLgwOJRpTjQ2zCCt2OcSfYMTeZVSRtQlPC7Nq4= +golang.org/x/sync v0.1.0 h1:wsuoTGHzEhffawBOhz5CYhcrV4IdKZbEyZjBMuTp12o= +golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sys v0.30.0 h1:QjkSwP/36a20jFYWkSue1YwXzLmsV5Gfq7Eiy72C1uc= golang.org/x/sys v0.30.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +golang.org/x/text v0.14.0 h1:ScX5w1eTa3QqT8oi6+ziP7dTV1S2+ALU0bI+0zXKWiQ= +golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/internal/adapters/postgresadmin/adapter.go b/internal/adapters/postgresadmin/adapter.go new file mode 100644 index 0000000..849e5d0 --- /dev/null +++ b/internal/adapters/postgresadmin/adapter.go @@ -0,0 +1,105 @@ +package postgresadmin + +import ( + "context" + "errors" + "fmt" + "reflect" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" + + "proxy-pool/internal/domain/adminstate" +) + +var _ adminstate.Store = (*Adapter)(nil) + +type transactionBeginner interface { + BeginTx(context.Context, pgx.TxOptions) (pgx.Tx, error) +} + +// Adapter keeps all PostgreSQL transaction and SQL details behind adminstate.Store. +type Adapter struct { + pool transactionBeginner +} + +// New constructs the PostgreSQL management-state store. +func New(pool transactionBeginner) (adminstate.Store, error) { + if isNil(pool) { + return nil, adminstate.ErrInvalidCommand + } + return &Adapter{pool: pool}, nil +} + +func isNil(value any) bool { + if value == nil { + return true + } + kind := reflect.ValueOf(value).Kind() + return (kind == reflect.Chan || kind == reflect.Func || kind == reflect.Interface || + kind == reflect.Map || kind == reflect.Pointer || kind == reflect.Slice) && + reflect.ValueOf(value).IsNil() +} + +func contextError(ctx context.Context) error { + if ctx == nil { + return adminstate.ErrInvalidCommand + } + return ctx.Err() +} + +func (adapter *Adapter) valid() bool { + return adapter != nil && !isNil(adapter.pool) +} + +func (adapter *Adapter) begin(ctx context.Context, options pgx.TxOptions, operation string) (pgx.Tx, error) { + tx, err := adapter.pool.BeginTx(ctx, options) + if err != nil { + return nil, databaseError(ctx, operation, err) + } + if isNil(tx) { + return nil, unavailable(operation) + } + return tx, nil +} + +func rollback(tx pgx.Tx) { + if !isNil(tx) { + _ = tx.Rollback(context.Background()) + } +} + +func commit(ctx context.Context, tx pgx.Tx, operation string) error { + if err := tx.Commit(ctx); err != nil { + return databaseError(ctx, operation, err) + } + return nil +} + +func databaseError(ctx context.Context, operation string, err error) error { + if err == nil { + return nil + } + if ctx != nil && ctx.Err() != nil { + return ctx.Err() + } + if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { + return err + } + + sentinel := adminstate.ErrUnavailable + var postgresError *pgconn.PgError + if errors.As(err, &postgresError) { + switch postgresError.Code { + case "23505": + sentinel = adminstate.ErrConflict + case "22001", "22003", "22P02": + sentinel = adminstate.ErrInvalidCommand + } + } + return fmt.Errorf("postgresadmin: %s: %w", operation, sentinel) +} + +func unavailable(operation string) error { + return fmt.Errorf("postgresadmin: %s: %w", operation, adminstate.ErrUnavailable) +} diff --git a/internal/adapters/postgresadmin/adapter_external_test.go b/internal/adapters/postgresadmin/adapter_external_test.go new file mode 100644 index 0000000..eee450a --- /dev/null +++ b/internal/adapters/postgresadmin/adapter_external_test.go @@ -0,0 +1,232 @@ +package postgresadmin_test + +import ( + "context" + "errors" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" + "github.com/jackc/pgx/v5/pgxpool" + + "proxy-pool/internal/adapters/postgresadmin" + "proxy-pool/internal/domain/adminstate" +) + +func TestNewRejectsNilPool(t *testing.T) { + t.Parallel() + + store, err := postgresadmin.New(nil) + if store != nil { + t.Fatalf("New(nil) store = %T, want nil", store) + } + if !errors.Is(err, adminstate.ErrInvalidCommand) { + t.Fatalf("New(nil) error = %v, want ErrInvalidCommand", err) + } +} + +func TestNewRejectsTypedNilPool(t *testing.T) { + t.Parallel() + + var pool *pgxpool.Pool + store, err := postgresadmin.New(pool) + if store != nil || !errors.Is(err, adminstate.ErrInvalidCommand) { + t.Fatalf("New(typed nil) = %T, %v, want nil, ErrInvalidCommand", store, err) + } +} + +func TestOperationsPrioritizeCanceledContext(t *testing.T) { + t.Parallel() + + pool := &countingPool{} + store, err := postgresadmin.New(pool) + if err != nil { + t.Fatalf("New(): %v", err) + } + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + operations := []struct { + name string + run func() error + }{ + {name: "commit config", run: func() error { + _, operationErr := store.CommitConfig(ctx, adminstate.CommitConfigCommand{}) + return operationErr + }}, + {name: "set upstream", run: func() error { + _, operationErr := store.SetUpstreamEnabled(ctx, adminstate.SetUpstreamCommand{}) + return operationErr + }}, + {name: "switch routing", run: func() error { + _, operationErr := store.SwitchRouting(ctx, adminstate.SwitchRoutingCommand{}) + return operationErr + }}, + {name: "snapshot", run: func() error { + _, operationErr := store.Snapshot(ctx) + return operationErr + }}, + {name: "read audit", run: func() error { + _, operationErr := store.ReadAudit(ctx, adminstate.AuditQuery{}) + return operationErr + }}, + {name: "claim", run: func() error { + _, operationErr := store.Claim(ctx, adminstate.ClaimCommand{}) + return operationErr + }}, + {name: "acknowledge", run: func() error { + return store.Acknowledge(ctx, adminstate.AcknowledgeCommand{}) + }}, + } + + for _, operation := range operations { + operation := operation + t.Run(operation.name, func(t *testing.T) { + if operationErr := operation.run(); !errors.Is(operationErr, context.Canceled) { + t.Fatalf("operation error = %v, want context.Canceled", operationErr) + } + }) + } + if begins := pool.begins.Load(); begins != 0 { + t.Fatalf("BeginTx calls = %d, want 0", begins) + } +} + +func TestInvalidOperationsDoNotBeginTransaction(t *testing.T) { + t.Parallel() + + pool := &countingPool{} + store, err := postgresadmin.New(pool) + if err != nil { + t.Fatalf("New(): %v", err) + } + operations := []struct { + name string + run func() error + }{ + {name: "commit config", run: func() error { + _, operationErr := store.CommitConfig(context.Background(), adminstate.CommitConfigCommand{}) + return operationErr + }}, + {name: "set upstream", run: func() error { + _, operationErr := store.SetUpstreamEnabled(context.Background(), adminstate.SetUpstreamCommand{}) + return operationErr + }}, + {name: "switch routing", run: func() error { + _, operationErr := store.SwitchRouting(context.Background(), adminstate.SwitchRoutingCommand{}) + return operationErr + }}, + {name: "read audit", run: func() error { + _, operationErr := store.ReadAudit(context.Background(), adminstate.AuditQuery{}) + return operationErr + }}, + {name: "claim", run: func() error { + _, operationErr := store.Claim(context.Background(), adminstate.ClaimCommand{}) + return operationErr + }}, + {name: "acknowledge", run: func() error { + return store.Acknowledge(context.Background(), adminstate.AcknowledgeCommand{}) + }}, + } + for _, operation := range operations { + operation := operation + t.Run(operation.name, func(t *testing.T) { + if operationErr := operation.run(); !errors.Is(operationErr, adminstate.ErrInvalidCommand) { + t.Fatalf("operation error = %v, want ErrInvalidCommand", operationErr) + } + }) + } + if begins := pool.begins.Load(); begins != 0 { + t.Fatalf("BeginTx calls = %d, want 0", begins) + } +} + +func TestDatabaseErrorsAreMappedAndRedacted(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + err error + want error + }{ + { + name: "connection", + err: errors.New("postgres://admin:secret@database/private SQL payload"), + want: adminstate.ErrUnavailable, + }, + { + name: "unique constraint", + err: &pgconn.PgError{ + Code: "23505", + Message: "duplicate cfg-sensitive at postgres://admin:secret@database", + }, + want: adminstate.ErrConflict, + }, + { + name: "foreign key constraint", + err: &pgconn.PgError{ + Code: "23503", + Message: "backend revision reference failed with secret details", + }, + want: adminstate.ErrUnavailable, + }, + { + name: "check constraint", + err: &pgconn.PgError{ + Code: "23514", + Message: "backend invariant failed with secret details", + }, + want: adminstate.ErrUnavailable, + }, + } + for _, test := range tests { + test := test + t.Run(test.name, func(t *testing.T) { + t.Parallel() + store, err := postgresadmin.New(&errorPool{err: test.err}) + if err != nil { + t.Fatalf("New(): %v", err) + } + _, operationErr := store.CommitConfig(context.Background(), validConfigCommand()) + if !errors.Is(operationErr, test.want) { + t.Fatalf("CommitConfig() error = %v, want %v", operationErr, test.want) + } + for _, secret := range []string{"secret", "private SQL", "cfg-sensitive", "postgres://"} { + if strings.Contains(operationErr.Error(), secret) { + t.Fatalf("CommitConfig() error leaked %q: %v", secret, operationErr) + } + } + }) + } +} + +type countingPool struct { + begins atomic.Int64 +} + +func (pool *countingPool) BeginTx(context.Context, pgx.TxOptions) (pgx.Tx, error) { + pool.begins.Add(1) + return nil, errors.New("unexpected transaction") +} + +type errorPool struct { + err error +} + +func (pool *errorPool) BeginTx(context.Context, pgx.TxOptions) (pgx.Tx, error) { + return nil, pool.err +} + +func validConfigCommand() adminstate.CommitConfigCommand { + return adminstate.CommitConfigCommand{ + RequestID: "request-a", + Actor: adminstate.Actor{ID: "admin-a", SourceIP: "192.0.2.10"}, + OccurredAt: time.Date(2026, 7, 30, 10, 0, 0, 0, time.UTC), + ConfigVersion: "cfg-sensitive", + Checksum: strings.Repeat("a", adminstate.SHA256HexBytes), + Source: "configs/proxy-pool.yaml", + } +} diff --git a/internal/adapters/postgresadmin/codec.go b/internal/adapters/postgresadmin/codec.go new file mode 100644 index 0000000..dad84fc --- /dev/null +++ b/internal/adapters/postgresadmin/codec.go @@ -0,0 +1,72 @@ +package postgresadmin + +import ( + "encoding/json" + "math" + "net/netip" + "strings" + "time" + + "proxy-pool/internal/domain/adminstate" +) + +func cloneConfigCommand(command adminstate.CommitConfigCommand) adminstate.CommitConfigCommand { + cloned := command + cloned.Checksum = strings.ToLower(command.Checksum) + cloned.Upstreams = append([]adminstate.UpstreamDefinition(nil), command.Upstreams...) + cloned.Routings = make([]adminstate.RoutingDefinition, len(command.Routings)) + for index, routing := range command.Routings { + cloned.Routings[index] = routing + cloned.Routings[index].Upstreams = append([]string(nil), routing.Upstreams...) + } + return cloned +} + +func sourceIPValue(sourceIP string) any { + if sourceIP == "" { + return nil + } + address, err := netip.ParseAddr(sourceIP) + if err != nil { + return nil + } + return address.Unmap() +} + +func encodePayload(value any) (json.RawMessage, error) { + payload, err := json.Marshal(value) + if err != nil { + return nil, err + } + return json.RawMessage(payload), nil +} + +func databaseID(value uint64) (int64, bool) { + if value > math.MaxInt64 { + return 0, false + } + return int64(value), true +} + +func domainID(value int64) (uint64, bool) { + if value < 0 { + return 0, false + } + return uint64(value), true +} + +func eventDatabaseIDs(values []uint64) ([]int64, bool) { + result := make([]int64, len(values)) + for index, value := range values { + converted, ok := databaseID(value) + if !ok { + return nil, false + } + result[index] = converted + } + return result, true +} + +func utc(value time.Time) time.Time { + return value.UTC() +} diff --git a/internal/adapters/postgresadmin/mutate.go b/internal/adapters/postgresadmin/mutate.go new file mode 100644 index 0000000..6bd94dd --- /dev/null +++ b/internal/adapters/postgresadmin/mutate.go @@ -0,0 +1,475 @@ +package postgresadmin + +import ( + "context" + "encoding/json" + "errors" + "math" + "strings" + "time" + + "github.com/jackc/pgx/v5" + + "proxy-pool/internal/domain/adminstate" +) + +const ( + lockRevisionSQL = `LOCK TABLE control_revisions IN SHARE ROW EXCLUSIVE MODE` + currentRevisionSQL = `SELECT COALESCE(MAX(revision), 0)::bigint FROM control_revisions` + insertRevisionSQL = ` + INSERT INTO control_revisions (revision, kind, created_at) + VALUES ($1, $2, $3)` + insertAuditSQL = ` + INSERT INTO admin_audit_log ( + request_id, actor_id, source_ip, action, resource_type, resource_name, + changed, revision, reason, occurred_at + ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)` + insertOutboxSQL = ` + INSERT INTO admin_outbox ( + revision, event_type, aggregate_type, aggregate_id, payload, occurred_at + ) VALUES ($1, $2, $3, $4, $5, $6)` +) + +func (adapter *Adapter) CommitConfig( + ctx context.Context, + command adminstate.CommitConfigCommand, +) (adminstate.MutationResult, error) { + result := adminstate.MutationResult{RequestID: command.RequestID} + if err := contextError(ctx); err != nil { + return result, err + } + if command.Validate() != nil || !adapter.valid() { + return result, adminstate.ErrInvalidCommand + } + command = cloneConfigCommand(command) + + tx, currentRevision, err := adapter.beginMutation(ctx, "commit config") + if err != nil { + return result, err + } + defer rollback(tx) + + var existingChecksum string + var isCurrent bool + err = tx.QueryRow(ctx, ` + SELECT checksum::text, + revision = (SELECT MAX(revision) FROM config_revisions) + FROM config_revisions + WHERE config_version = $1`, command.ConfigVersion).Scan(&existingChecksum, &isCurrent) + switch { + case err == nil: + if !isCurrent || !strings.EqualFold(existingChecksum, command.Checksum) { + return result, adminstate.ErrConflict + } + if err := insertAudit(ctx, tx, command.RequestID, command.Actor, + adminstate.ActionCommitConfig, "config", command.ConfigVersion, false, + currentRevision, "", command.OccurredAt, "audit config no-op"); err != nil { + return result, err + } + if err := commit(ctx, tx, "commit config no-op"); err != nil { + return result, err + } + result.Revision = uint64(currentRevision) + return result, nil + case !errors.Is(err, pgx.ErrNoRows): + return result, databaseError(ctx, "read config revision", err) + } + + nextRevision, err := allocateRevision(ctx, tx, currentRevision, "config", command.OccurredAt) + if err != nil { + return result, err + } + if _, err := tx.Exec(ctx, ` + INSERT INTO config_revisions ( + revision, config_version, checksum, source, created_at + ) VALUES ($1, $2, $3, $4, $5)`, + nextRevision, command.ConfigVersion, command.Checksum, command.Source, utc(command.OccurredAt)); err != nil { + return result, databaseError(ctx, "insert config revision", err) + } + if _, err := tx.Exec(ctx, `DELETE FROM routing_admin_state`); err != nil { + return result, databaseError(ctx, "replace routing state", err) + } + if _, err := tx.Exec(ctx, `DELETE FROM upstream_admin_state`); err != nil { + return result, databaseError(ctx, "replace upstream state", err) + } + if err := copyUpstreams(ctx, tx, command.Upstreams, nextRevision, command.OccurredAt); err != nil { + return result, err + } + if err := copyRoutings(ctx, tx, command.Routings, nextRevision, command.OccurredAt); err != nil { + return result, err + } + if err := insertAudit(ctx, tx, command.RequestID, command.Actor, + adminstate.ActionCommitConfig, "config", command.ConfigVersion, true, + nextRevision, "", command.OccurredAt, "audit config change"); err != nil { + return result, err + } + payload, err := encodePayload(map[string]any{ + "configVersion": command.ConfigVersion, + "checksum": command.Checksum, + "revision": nextRevision, + }) + if err != nil { + return result, unavailable("encode config event") + } + if err := insertOutbox(ctx, tx, nextRevision, "config.committed", "config", + command.ConfigVersion, payload, command.OccurredAt, "insert config event"); err != nil { + return result, err + } + if err := commit(ctx, tx, "commit config change"); err != nil { + return result, err + } + return adminstate.MutationResult{ + RequestID: command.RequestID, + Changed: true, + Revision: uint64(nextRevision), + }, nil +} + +func (adapter *Adapter) SetUpstreamEnabled( + ctx context.Context, + command adminstate.SetUpstreamCommand, +) (adminstate.MutationResult, error) { + result := adminstate.MutationResult{RequestID: command.RequestID} + if err := contextError(ctx); err != nil { + return result, err + } + if command.Validate() != nil || !adapter.valid() { + return result, adminstate.ErrInvalidCommand + } + + tx, currentRevision, err := adapter.beginMutation(ctx, "set upstream") + if err != nil { + return result, err + } + defer rollback(tx) + + var enabled bool + if err := tx.QueryRow(ctx, ` + SELECT enabled + FROM upstream_admin_state + WHERE name = $1 + FOR UPDATE`, command.Name).Scan(&enabled); err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return result, adminstate.ErrNotFound + } + return result, databaseError(ctx, "read upstream state", err) + } + if enabled == command.Enabled { + if err := insertAudit(ctx, tx, command.RequestID, command.Actor, + adminstate.ActionSetUpstream, "upstream", command.Name, false, + currentRevision, "", command.OccurredAt, "audit upstream no-op"); err != nil { + return result, err + } + if err := commit(ctx, tx, "commit upstream no-op"); err != nil { + return result, err + } + result.Revision = uint64(currentRevision) + return result, nil + } + + nextRevision, err := allocateRevision(ctx, tx, currentRevision, "upstream", command.OccurredAt) + if err != nil { + return result, err + } + commandTag, err := tx.Exec(ctx, ` + UPDATE upstream_admin_state + SET enabled = $1, revision = $2, updated_at = $3 + WHERE name = $4`, command.Enabled, nextRevision, utc(command.OccurredAt), command.Name) + if err != nil { + return result, databaseError(ctx, "update upstream state", err) + } + if commandTag.RowsAffected() != 1 { + return result, unavailable("update upstream state") + } + if err := insertAudit(ctx, tx, command.RequestID, command.Actor, + adminstate.ActionSetUpstream, "upstream", command.Name, true, + nextRevision, "", command.OccurredAt, "audit upstream change"); err != nil { + return result, err + } + payload, err := encodePayload(map[string]any{ + "enabled": command.Enabled, + "name": command.Name, + "revision": nextRevision, + }) + if err != nil { + return result, unavailable("encode upstream event") + } + if err := insertOutbox(ctx, tx, nextRevision, "upstream.enabled_changed", "upstream", + command.Name, payload, command.OccurredAt, "insert upstream event"); err != nil { + return result, err + } + if err := commit(ctx, tx, "commit upstream change"); err != nil { + return result, err + } + return adminstate.MutationResult{ + RequestID: command.RequestID, + Changed: true, + Revision: uint64(nextRevision), + }, nil +} + +func (adapter *Adapter) SwitchRouting( + ctx context.Context, + command adminstate.SwitchRoutingCommand, +) (adminstate.MutationResult, error) { + result := adminstate.MutationResult{RequestID: command.RequestID} + if err := contextError(ctx); err != nil { + return result, err + } + if command.Validate() != nil || !adapter.valid() { + return result, adminstate.ErrInvalidCommand + } + + tx, currentRevision, err := adapter.beginMutation(ctx, "switch routing") + if err != nil { + return result, err + } + defer rollback(tx) + + var enabled bool + var candidates []string + var current string + if err := tx.QueryRow(ctx, ` + SELECT enabled, upstreams, current_upstream + FROM routing_admin_state + WHERE name = $1 + FOR UPDATE`, command.Name).Scan(&enabled, &candidates, ¤t); err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return result, adminstate.ErrNotFound + } + return result, databaseError(ctx, "read routing state", err) + } + if !enabled || current != command.ExpectedCurrent { + return result, adminstate.ErrConflict + } + if !contains(candidates, command.Target) { + return result, adminstate.ErrInvalidCommand + } + if current == command.Target { + if err := insertAudit(ctx, tx, command.RequestID, command.Actor, + adminstate.ActionSwitchRoute, "routing", command.Name, false, + currentRevision, command.Reason, command.OccurredAt, "audit routing no-op"); err != nil { + return result, err + } + if err := commit(ctx, tx, "commit routing no-op"); err != nil { + return result, err + } + result.Revision = uint64(currentRevision) + return result, nil + } + + nextRevision, err := allocateRevision(ctx, tx, currentRevision, "routing", command.OccurredAt) + if err != nil { + return result, err + } + commandTag, err := tx.Exec(ctx, ` + UPDATE routing_admin_state + SET current_upstream = $1, revision = $2, updated_at = $3 + WHERE name = $4`, command.Target, nextRevision, utc(command.OccurredAt), command.Name) + if err != nil { + return result, databaseError(ctx, "update routing state", err) + } + if commandTag.RowsAffected() != 1 { + return result, unavailable("update routing state") + } + if err := insertAudit(ctx, tx, command.RequestID, command.Actor, + adminstate.ActionSwitchRoute, "routing", command.Name, true, + nextRevision, command.Reason, command.OccurredAt, "audit routing change"); err != nil { + return result, err + } + payload, err := encodePayload(map[string]any{ + "current": command.Target, + "name": command.Name, + "previous": command.ExpectedCurrent, + "reason": command.Reason, + "revision": nextRevision, + }) + if err != nil { + return result, unavailable("encode routing event") + } + if err := insertOutbox(ctx, tx, nextRevision, "routing.switched", "routing", + command.Name, payload, command.OccurredAt, "insert routing event"); err != nil { + return result, err + } + if err := commit(ctx, tx, "commit routing change"); err != nil { + return result, err + } + return adminstate.MutationResult{ + RequestID: command.RequestID, + Changed: true, + Revision: uint64(nextRevision), + }, nil +} + +func (adapter *Adapter) beginMutation(ctx context.Context, operation string) (pgx.Tx, int64, error) { + tx, err := adapter.begin(ctx, pgx.TxOptions{AccessMode: pgx.ReadWrite}, operation) + if err != nil { + return nil, 0, err + } + if _, err := tx.Exec(ctx, lockRevisionSQL); err != nil { + rollback(tx) + return nil, 0, databaseError(ctx, "lock revision state", err) + } + var currentRevision int64 + if err := tx.QueryRow(ctx, currentRevisionSQL).Scan(¤tRevision); err != nil { + rollback(tx) + return nil, 0, databaseError(ctx, "read current revision", err) + } + if currentRevision < 0 { + rollback(tx) + return nil, 0, unavailable("read current revision") + } + return tx, currentRevision, nil +} + +func allocateRevision( + ctx context.Context, + tx pgx.Tx, + currentRevision int64, + kind string, + occurredAt time.Time, +) (int64, error) { + if currentRevision == math.MaxInt64 { + return 0, adminstate.ErrUnavailable + } + nextRevision := currentRevision + 1 + commandTag, err := tx.Exec(ctx, insertRevisionSQL, nextRevision, kind, utc(occurredAt)) + if err != nil { + return 0, databaseError(ctx, "allocate revision", err) + } + if commandTag.RowsAffected() != 1 { + return 0, unavailable("allocate revision") + } + return nextRevision, nil +} + +func copyUpstreams( + ctx context.Context, + tx pgx.Tx, + definitions []adminstate.UpstreamDefinition, + revision int64, + occurredAt time.Time, +) error { + if len(definitions) == 0 { + return nil + } + updatedAt := utc(occurredAt) + count, err := tx.CopyFrom(ctx, pgx.Identifier{"upstream_admin_state"}, + []string{"name", "enabled", "revision", "updated_at"}, + pgx.CopyFromSlice(len(definitions), func(index int) ([]any, error) { + definition := definitions[index] + return []any{definition.Name, definition.Enabled, revision, updatedAt}, nil + })) + if err != nil { + return databaseError(ctx, "replace upstream state", err) + } + if count != int64(len(definitions)) { + return unavailable("replace upstream state") + } + return nil +} + +func copyRoutings( + ctx context.Context, + tx pgx.Tx, + definitions []adminstate.RoutingDefinition, + revision int64, + occurredAt time.Time, +) error { + if len(definitions) == 0 { + return nil + } + updatedAt := utc(occurredAt) + count, err := tx.CopyFrom(ctx, pgx.Identifier{"routing_admin_state"}, + []string{"name", "enabled", "upstreams", "current_upstream", "revision", "updated_at"}, + pgx.CopyFromSlice(len(definitions), func(index int) ([]any, error) { + definition := definitions[index] + return []any{ + definition.Name, + definition.Enabled, + definition.Upstreams, + definition.CurrentUpstream, + revision, + updatedAt, + }, nil + })) + if err != nil { + return databaseError(ctx, "replace routing state", err) + } + if count != int64(len(definitions)) { + return unavailable("replace routing state") + } + return nil +} + +func insertAudit( + ctx context.Context, + tx pgx.Tx, + requestID string, + actor adminstate.Actor, + action adminstate.Action, + resourceType string, + resourceName string, + changed bool, + revision int64, + reason string, + occurredAt time.Time, + operation string, +) error { + commandTag, err := tx.Exec(ctx, insertAuditSQL, + requestID, + actor.ID, + sourceIPValue(actor.SourceIP), + string(action), + resourceType, + resourceName, + changed, + revision, + reason, + utc(occurredAt), + ) + if err != nil { + return databaseError(ctx, operation, err) + } + if commandTag.RowsAffected() != 1 { + return unavailable(operation) + } + return nil +} + +func insertOutbox( + ctx context.Context, + tx pgx.Tx, + revision int64, + eventType string, + aggregateType string, + aggregateID string, + payload json.RawMessage, + occurredAt time.Time, + operation string, +) error { + commandTag, err := tx.Exec(ctx, insertOutboxSQL, + revision, + eventType, + aggregateType, + aggregateID, + payload, + utc(occurredAt), + ) + if err != nil { + return databaseError(ctx, operation, err) + } + if commandTag.RowsAffected() != 1 { + return unavailable(operation) + } + return nil +} + +func contains(values []string, target string) bool { + for _, value := range values { + if value == target { + return true + } + } + return false +} diff --git a/internal/adapters/postgresadmin/outbox.go b/internal/adapters/postgresadmin/outbox.go new file mode 100644 index 0000000..c50b10d --- /dev/null +++ b/internal/adapters/postgresadmin/outbox.go @@ -0,0 +1,179 @@ +package postgresadmin + +import ( + "context" + "encoding/json" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgtype" + + "proxy-pool/internal/domain/adminstate" +) + +func (adapter *Adapter) Claim( + ctx context.Context, + command adminstate.ClaimCommand, +) ([]adminstate.Event, error) { + if err := contextError(ctx); err != nil { + return nil, err + } + if command.Validate() != nil || !adapter.valid() { + return nil, adminstate.ErrInvalidCommand + } + tx, err := adapter.begin(ctx, pgx.TxOptions{AccessMode: pgx.ReadWrite}, "begin outbox claim") + if err != nil { + return nil, err + } + defer rollback(tx) + + now := utc(command.Now) + claimUntil := now.Add(command.Lease) + rows, err := tx.Query(ctx, ` + WITH candidates AS ( + SELECT id + FROM admin_outbox + WHERE published_at IS NULL + AND (claim_until IS NULL OR claim_until <= $1) + ORDER BY id + FOR UPDATE SKIP LOCKED + LIMIT $2 + ), claimed AS ( + UPDATE admin_outbox AS outbox + SET claim_owner = $3, claim_until = $4 + FROM candidates + WHERE outbox.id = candidates.id + RETURNING outbox.id, outbox.revision, outbox.event_type, + outbox.aggregate_type, outbox.aggregate_id, outbox.payload, + outbox.occurred_at, outbox.claim_owner, outbox.claim_until, + outbox.published_at + ) + SELECT id, revision, event_type, aggregate_type, aggregate_id, payload, + occurred_at, claim_owner, claim_until, published_at + FROM claimed + ORDER BY id`, now, command.Limit, command.ConsumerID, claimUntil) + if err != nil { + return nil, databaseError(ctx, "claim outbox events", err) + } + defer rows.Close() + + events := make([]adminstate.Event, 0, command.Limit) + for rows.Next() { + event := adminstate.Event{} + var id int64 + var revision int64 + var payload []byte + var claimedBy pgtype.Text + var claimedUntil pgtype.Timestamptz + var publishedAt pgtype.Timestamptz + if err := rows.Scan( + &id, + &revision, + &event.Type, + &event.AggregateType, + &event.AggregateID, + &payload, + &event.OccurredAt, + &claimedBy, + &claimedUntil, + &publishedAt, + ); err != nil { + return nil, databaseError(ctx, "decode claimed outbox events", err) + } + convertedID, idOK := domainID(id) + convertedRevision, revisionOK := domainID(revision) + if !idOK || !revisionOK || !claimedBy.Valid || !claimedUntil.Valid { + return nil, unavailable("decode claimed outbox events") + } + event.ID = convertedID + event.Revision = convertedRevision + event.Payload = append(json.RawMessage(nil), payload...) + event.OccurredAt = utc(event.OccurredAt) + event.ClaimedBy = claimedBy.String + value := utc(claimedUntil.Time) + event.ClaimUntil = &value + if publishedAt.Valid { + value := utc(publishedAt.Time) + event.PublishedAt = &value + } + events = append(events, event) + } + if err := rows.Err(); err != nil { + return nil, databaseError(ctx, "claim outbox events", err) + } + if err := commit(ctx, tx, "commit outbox claim"); err != nil { + return nil, err + } + return events, nil +} + +func (adapter *Adapter) Acknowledge( + ctx context.Context, + command adminstate.AcknowledgeCommand, +) error { + if err := contextError(ctx); err != nil { + return err + } + if command.Validate() != nil || !adapter.valid() { + return adminstate.ErrInvalidCommand + } + eventIDs, ok := eventDatabaseIDs(command.EventIDs) + if !ok { + return adminstate.ErrNotFound + } + tx, err := adapter.begin(ctx, pgx.TxOptions{AccessMode: pgx.ReadWrite}, "begin outbox acknowledge") + if err != nil { + return err + } + defer rollback(tx) + + rows, err := tx.Query(ctx, ` + SELECT id, published_at, claim_owner, claim_until + FROM admin_outbox + WHERE id = ANY($1::bigint[]) + ORDER BY id + FOR UPDATE`, eventIDs) + if err != nil { + return databaseError(ctx, "lock acknowledged outbox events", err) + } + count := 0 + now := utc(command.Now) + conflict := false + for rows.Next() { + var id int64 + var publishedAt pgtype.Timestamptz + var claimOwner pgtype.Text + var claimUntil pgtype.Timestamptz + if err := rows.Scan(&id, &publishedAt, &claimOwner, &claimUntil); err != nil { + rows.Close() + return databaseError(ctx, "decode acknowledged outbox events", err) + } + count++ + if publishedAt.Valid || !claimOwner.Valid || claimOwner.String != command.ConsumerID || + !claimUntil.Valid || !now.Before(claimUntil.Time) { + conflict = true + } + } + if err := rows.Err(); err != nil { + rows.Close() + return databaseError(ctx, "lock acknowledged outbox events", err) + } + rows.Close() + if count != len(eventIDs) { + return adminstate.ErrNotFound + } + if conflict { + return adminstate.ErrConflict + } + + commandTag, err := tx.Exec(ctx, ` + UPDATE admin_outbox + SET published_at = $1 + WHERE id = ANY($2::bigint[])`, now, eventIDs) + if err != nil { + return databaseError(ctx, "publish outbox events", err) + } + if commandTag.RowsAffected() != int64(len(eventIDs)) { + return unavailable("publish outbox events") + } + return commit(ctx, tx, "commit outbox acknowledge") +} diff --git a/internal/adapters/postgresadmin/snapshot.go b/internal/adapters/postgresadmin/snapshot.go new file mode 100644 index 0000000..f7daae4 --- /dev/null +++ b/internal/adapters/postgresadmin/snapshot.go @@ -0,0 +1,222 @@ +package postgresadmin + +import ( + "context" + "errors" + + "github.com/jackc/pgx/v5" + + "proxy-pool/internal/domain/adminstate" +) + +func (adapter *Adapter) Snapshot(ctx context.Context) (adminstate.Snapshot, error) { + if err := contextError(ctx); err != nil { + return adminstate.Snapshot{}, err + } + if !adapter.valid() { + return adminstate.Snapshot{}, adminstate.ErrInvalidCommand + } + tx, err := adapter.begin(ctx, pgx.TxOptions{ + IsoLevel: pgx.RepeatableRead, + AccessMode: pgx.ReadOnly, + }, "begin snapshot") + if err != nil { + return adminstate.Snapshot{}, err + } + defer rollback(tx) + + snapshot := adminstate.Snapshot{} + var revision int64 + if err := tx.QueryRow(ctx, currentRevisionSQL).Scan(&revision); err != nil { + return adminstate.Snapshot{}, databaseError(ctx, "read snapshot revision", err) + } + convertedRevision, ok := domainID(revision) + if !ok { + return adminstate.Snapshot{}, unavailable("decode snapshot revision") + } + snapshot.Revision = convertedRevision + + var configRevision int64 + config := adminstate.ConfigRevision{} + err = tx.QueryRow(ctx, ` + SELECT revision, config_version, checksum::text, source, created_at + FROM config_revisions + ORDER BY revision DESC + LIMIT 1`).Scan( + &configRevision, + &config.ConfigVersion, + &config.Checksum, + &config.Source, + &config.CreatedAt, + ) + switch { + case err == nil: + converted, ok := domainID(configRevision) + if !ok { + return adminstate.Snapshot{}, unavailable("decode config revision") + } + config.Revision = converted + config.CreatedAt = utc(config.CreatedAt) + snapshot.Config = &config + case !errors.Is(err, pgx.ErrNoRows): + return adminstate.Snapshot{}, databaseError(ctx, "read config snapshot", err) + } + + upstreams, err := readUpstreamSnapshot(ctx, tx) + if err != nil { + return adminstate.Snapshot{}, err + } + snapshot.Upstreams = upstreams + routings, err := readRoutingSnapshot(ctx, tx) + if err != nil { + return adminstate.Snapshot{}, err + } + snapshot.Routings = routings + if err := commit(ctx, tx, "commit snapshot"); err != nil { + return adminstate.Snapshot{}, err + } + return snapshot, nil +} + +func (adapter *Adapter) ReadAudit( + ctx context.Context, + query adminstate.AuditQuery, +) ([]adminstate.AuditRecord, error) { + if err := contextError(ctx); err != nil { + return nil, err + } + if query.Validate() != nil || !adapter.valid() { + return nil, adminstate.ErrInvalidCommand + } + afterID, ok := databaseID(query.AfterID) + if !ok { + return []adminstate.AuditRecord{}, nil + } + tx, err := adapter.begin(ctx, pgx.TxOptions{AccessMode: pgx.ReadOnly}, "begin audit read") + if err != nil { + return nil, err + } + defer rollback(tx) + + rows, err := tx.Query(ctx, ` + SELECT id, request_id, actor_id, COALESCE(host(source_ip), ''), action, + resource_type, resource_name, changed, revision, reason, occurred_at + FROM admin_audit_log + WHERE id > $1 + ORDER BY id + LIMIT $2`, afterID, query.Limit) + if err != nil { + return nil, databaseError(ctx, "read audit page", err) + } + defer rows.Close() + + records := make([]adminstate.AuditRecord, 0, query.Limit) + for rows.Next() { + record := adminstate.AuditRecord{} + var id int64 + var revision int64 + var action string + if err := rows.Scan( + &id, + &record.RequestID, + &record.Actor.ID, + &record.Actor.SourceIP, + &action, + &record.ResourceType, + &record.ResourceName, + &record.Changed, + &revision, + &record.Reason, + &record.OccurredAt, + ); err != nil { + return nil, databaseError(ctx, "decode audit page", err) + } + convertedID, idOK := domainID(id) + convertedRevision, revisionOK := domainID(revision) + if !idOK || !revisionOK { + return nil, unavailable("decode audit page") + } + record.ID = convertedID + record.Revision = convertedRevision + record.Action = adminstate.Action(action) + record.OccurredAt = utc(record.OccurredAt) + records = append(records, record) + } + if err := rows.Err(); err != nil { + return nil, databaseError(ctx, "read audit page", err) + } + if err := commit(ctx, tx, "commit audit read"); err != nil { + return nil, err + } + return records, nil +} + +func readUpstreamSnapshot(ctx context.Context, tx pgx.Tx) ([]adminstate.UpstreamState, error) { + rows, err := tx.Query(ctx, ` + SELECT name, enabled, revision, updated_at + FROM upstream_admin_state + ORDER BY name`) + if err != nil { + return nil, databaseError(ctx, "read upstream snapshot", err) + } + defer rows.Close() + + states := make([]adminstate.UpstreamState, 0) + for rows.Next() { + state := adminstate.UpstreamState{} + var revision int64 + if err := rows.Scan(&state.Name, &state.Enabled, &revision, &state.UpdatedAt); err != nil { + return nil, databaseError(ctx, "decode upstream snapshot", err) + } + converted, ok := domainID(revision) + if !ok { + return nil, unavailable("decode upstream snapshot") + } + state.Revision = converted + state.UpdatedAt = utc(state.UpdatedAt) + states = append(states, state) + } + if err := rows.Err(); err != nil { + return nil, databaseError(ctx, "read upstream snapshot", err) + } + return states, nil +} + +func readRoutingSnapshot(ctx context.Context, tx pgx.Tx) ([]adminstate.RoutingState, error) { + rows, err := tx.Query(ctx, ` + SELECT name, enabled, upstreams, current_upstream, revision, updated_at + FROM routing_admin_state + ORDER BY name`) + if err != nil { + return nil, databaseError(ctx, "read routing snapshot", err) + } + defer rows.Close() + + states := make([]adminstate.RoutingState, 0) + for rows.Next() { + state := adminstate.RoutingState{} + var revision int64 + if err := rows.Scan( + &state.Name, + &state.Enabled, + &state.Upstreams, + &state.CurrentUpstream, + &revision, + &state.UpdatedAt, + ); err != nil { + return nil, databaseError(ctx, "decode routing snapshot", err) + } + converted, ok := domainID(revision) + if !ok { + return nil, unavailable("decode routing snapshot") + } + state.Revision = converted + state.UpdatedAt = utc(state.UpdatedAt) + state.Upstreams = append([]string(nil), state.Upstreams...) + states = append(states, state) + } + if err := rows.Err(); err != nil { + return nil, databaseError(ctx, "read routing snapshot", err) + } + return states, nil +}