diff --git a/Makefile b/Makefile index ad3b72592..89b926828 100644 --- a/Makefile +++ b/Makefile @@ -13,9 +13,9 @@ test: go test ./system/... -v -race -short go test ./service/... -v -race -short go test ./cluster/... -v -race -short - go test --tags "fts5 sqlite_vec treesitter" ./runtime/... -v -race -short + go test --tags "fts5 sqlite_vec treesitter sqlite_preupdate_hook" ./runtime/... -v -race -short go test ./boot/... -v -race -short - go test --tags "fts5 sqlite_vec treesitter" ./cmd/... -v -race -short + go test --tags "fts5 sqlite_vec treesitter sqlite_preupdate_hook" ./cmd/... -v -race -short test-system: go test ./internal/... -v -race @@ -25,7 +25,7 @@ test-system: test-runtime: go test ./internal/... -v -race go test ./api/... -v -race - go test --tags "fts5 sqlite_vec treesitter" ./runtime/... -v -race + go test --tags "fts5 sqlite_vec treesitter sqlite_preupdate_hook" ./runtime/... -v -race test-service: go test ./internal/... -v -race @@ -45,7 +45,7 @@ test-network: .PHONY: lint lint: - go run github.com/golangci/golangci-lint/v2/cmd/golangci-lint@v2.8.0 run --timeout=10m --build-tags=race ./... + go run github.com/golangci/golangci-lint/v2/cmd/golangci-lint@v2.8.0 run --timeout=10m --build-tags=race,sqlite_preupdate_hook ./... # Mutation testing with gremlins. Coverage is scoped to the directory gremlins # runs from, so target a package subtree via MUTATE_DIR. workers=1 keeps per- @@ -87,7 +87,7 @@ build-wippy: build-wippy-local .PHONY: build-wippy-local build-wippy-local: mkdir -p ./dist - CGO_ENABLED=1 go build --tags "fts5 sqlite_vec treesitter" \ + CGO_ENABLED=1 go build --tags "fts5 sqlite_vec treesitter sqlite_preupdate_hook" \ -ldflags="$(WIPPY_LDFLAGS)" \ -trimpath \ -o ./dist/wippy-$(shell go env GOOS)-$(shell go env GOARCH) \ @@ -99,7 +99,7 @@ build-wippy-all: build-wippy-linux-amd64 build-wippy-linux-arm64 build-wippy-dar .PHONY: build-wippy-linux-amd64 build-wippy-linux-amd64: mkdir -p ./dist - CGO_ENABLED=1 GOOS=linux GOARCH=amd64 go build --tags "fts5 sqlite_vec treesitter" \ + CGO_ENABLED=1 GOOS=linux GOARCH=amd64 go build --tags "fts5 sqlite_vec treesitter sqlite_preupdate_hook" \ -ldflags="$(WIPPY_LDFLAGS)" \ -trimpath \ -o ./dist/wippy-linux-amd64 \ @@ -109,7 +109,7 @@ build-wippy-linux-amd64: build-wippy-linux-arm64: mkdir -p ./dist CGO_LDFLAGS="" CGO_CFLAGS="" CC=aarch64-linux-gnu-gcc \ - CGO_ENABLED=1 GOOS=linux GOARCH=arm64 go build --tags "fts5 sqlite_vec treesitter" \ + CGO_ENABLED=1 GOOS=linux GOARCH=arm64 go build --tags "fts5 sqlite_vec treesitter sqlite_preupdate_hook" \ -ldflags="$(WIPPY_LDFLAGS)" \ -trimpath \ -o ./dist/wippy-linux-arm64 \ @@ -118,7 +118,7 @@ build-wippy-linux-arm64: .PHONY: build-wippy-darwin-amd64 build-wippy-darwin-amd64: mkdir -p ./dist - CGO_ENABLED=1 GOOS=darwin GOARCH=amd64 go build --tags "fts5 sqlite_vec treesitter" \ + CGO_ENABLED=1 GOOS=darwin GOARCH=amd64 go build --tags "fts5 sqlite_vec treesitter sqlite_preupdate_hook" \ -ldflags="$(WIPPY_LDFLAGS)" \ -trimpath \ -o ./dist/wippy-darwin-amd64 \ @@ -127,7 +127,7 @@ build-wippy-darwin-amd64: .PHONY: build-wippy-darwin-arm64 build-wippy-darwin-arm64: mkdir -p ./dist - CGO_ENABLED=1 GOOS=darwin GOARCH=arm64 go build --tags "fts5 sqlite_vec treesitter" \ + CGO_ENABLED=1 GOOS=darwin GOARCH=arm64 go build --tags "fts5 sqlite_vec treesitter sqlite_preupdate_hook" \ -ldflags="$(WIPPY_LDFLAGS)" \ -trimpath \ -o ./dist/wippy-darwin-arm64 \ @@ -137,7 +137,7 @@ build-wippy-darwin-arm64: build-wippy-windows-amd64: mkdir -p ./dist CGO_LDFLAGS="" CGO_CFLAGS="" CC=x86_64-w64-mingw32-gcc \ - CGO_ENABLED=1 GOOS=windows GOARCH=amd64 go build --tags "fts5 sqlite_vec treesitter" \ + CGO_ENABLED=1 GOOS=windows GOARCH=amd64 go build --tags "fts5 sqlite_vec treesitter sqlite_preupdate_hook" \ -ldflags="$(WIPPY_LDFLAGS)" \ -trimpath \ -o ./dist/wippy-windows-amd64.exe \ @@ -170,4 +170,4 @@ build-sign-wippy-windows: build-wippy-windows-amd64 sign-wippy-windows .PHONY: run-wippy run-wippy: - go run --tags "fts5 sqlite_vec treesitter" -ldflags="$(WIPPY_LDFLAGS)" ./cmd/wippy/ $(ARGS) + go run --tags "fts5 sqlite_vec treesitter sqlite_preupdate_hook" -ldflags="$(WIPPY_LDFLAGS)" ./cmd/wippy/ $(ARGS) diff --git a/api/service/cdc/composite.go b/api/service/cdc/composite.go new file mode 100644 index 000000000..a5f6adb6c --- /dev/null +++ b/api/service/cdc/composite.go @@ -0,0 +1,44 @@ +// SPDX-License-Identifier: MPL-2.0 + +package cdc + +import "context" + +type Engine interface { + SourceInspector + SourceStreamer +} + +type composite struct { + engines []Engine +} + +func NewComposite(engines ...Engine) *composite { + return &composite{engines: engines} +} + +func (c *composite) List() []SourceInfo { + out := make([]SourceInfo, 0) + for _, e := range c.engines { + out = append(out, e.List()...) + } + return out +} + +func (c *composite) Get(name string) (SourceInfo, bool) { + for _, e := range c.engines { + if info, ok := e.Get(name); ok { + return info, true + } + } + return SourceInfo{}, false +} + +func (c *composite) Stream(ctx context.Context, name string, opts StreamOptions) (ChangeStream, SourceInfo, error) { + for _, e := range c.engines { + if _, ok := e.Get(name); ok { + return e.Stream(ctx, name, opts) + } + } + return nil, SourceInfo{}, ErrSourceNotFound +} diff --git a/api/service/cdc/composite_test.go b/api/service/cdc/composite_test.go new file mode 100644 index 000000000..b06a4997e --- /dev/null +++ b/api/service/cdc/composite_test.go @@ -0,0 +1,69 @@ +// SPDX-License-Identifier: MPL-2.0 + +package cdc + +import ( + "context" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +type fakeStream struct{ ch chan Change } + +func (f *fakeStream) Changes() <-chan Change { return f.ch } +func (f *fakeStream) Close() {} + +type fakeEngine struct { + infos map[string]SourceInfo + opened []string +} + +func (e *fakeEngine) List() []SourceInfo { + out := make([]SourceInfo, 0, len(e.infos)) + for _, i := range e.infos { + out = append(out, i) + } + return out +} + +func (e *fakeEngine) Get(name string) (SourceInfo, bool) { + i, ok := e.infos[name] + return i, ok +} + +func (e *fakeEngine) Stream(_ context.Context, name string, _ StreamOptions) (ChangeStream, SourceInfo, error) { + e.opened = append(e.opened, name) + return &fakeStream{ch: make(chan Change)}, e.infos[name], nil +} + +func TestCompositeListAggregates(t *testing.T) { + a := &fakeEngine{infos: map[string]SourceInfo{"pg": {Name: "pg", Engine: "postgres"}}} + b := &fakeEngine{infos: map[string]SourceInfo{"lite": {Name: "lite", Engine: "sqlite"}}} + c := NewComposite(a, b) + + infos := c.List() + assert.Len(t, infos, 2) +} + +func TestCompositeGetAndStreamRouting(t *testing.T) { + a := &fakeEngine{infos: map[string]SourceInfo{"pg": {Name: "pg", Engine: "postgres"}}} + b := &fakeEngine{infos: map[string]SourceInfo{"lite": {Name: "lite", Engine: "sqlite"}}} + c := NewComposite(a, b) + + info, ok := c.Get("lite") + require.True(t, ok) + assert.Equal(t, "sqlite", info.Engine) + + _, _, err := c.Stream(context.Background(), "lite", StreamOptions{}) + require.NoError(t, err) + assert.Equal(t, []string{"lite"}, b.opened) + assert.Empty(t, a.opened) +} + +func TestCompositeStreamNotFound(t *testing.T) { + c := NewComposite(&fakeEngine{infos: map[string]SourceInfo{}}) + _, _, err := c.Stream(context.Background(), "missing", StreamOptions{}) + assert.ErrorIs(t, err, ErrSourceNotFound) +} diff --git a/api/service/cdc/config.go b/api/service/cdc/config.go index 51b59e33a..201d00721 100644 --- a/api/service/cdc/config.go +++ b/api/service/cdc/config.go @@ -11,6 +11,7 @@ import ( const ( Postgres registry.Kind = "db.cdc.postgres" + SQLite registry.Kind = "db.cdc.sqlite" ) const ( diff --git a/api/service/cdc/config_sqlite.go b/api/service/cdc/config_sqlite.go new file mode 100644 index 000000000..928192aab --- /dev/null +++ b/api/service/cdc/config_sqlite.go @@ -0,0 +1,36 @@ +// SPDX-License-Identifier: MPL-2.0 + +package cdc + +import ( + "time" + + "github.com/wippyai/runtime/api/supervisor" +) + +type SQLiteConfig struct { + DBResource string `json:"db_resource"` + Name string `json:"name,omitempty"` + StatusInterval string `json:"status_interval,omitempty"` + Tables []string `json:"tables,omitempty"` + Lifecycle supervisor.LifecycleConfig `json:"lifecycle"` + Snapshot bool `json:"snapshot,omitempty"` +} + +func (c *SQLiteConfig) InitDefaults() { + c.Lifecycle.InitDefaults() +} + +func (c *SQLiteConfig) Validate() error { + if c.DBResource == "" { + return ErrDBResourceRequired + } + if _, err := c.StatusDuration(); err != nil { + return err + } + return nil +} + +func (c *SQLiteConfig) StatusDuration() (time.Duration, error) { + return parseInterval(c.StatusInterval) +} diff --git a/api/service/cdc/config_sqlite_test.go b/api/service/cdc/config_sqlite_test.go new file mode 100644 index 000000000..b49f6a292 --- /dev/null +++ b/api/service/cdc/config_sqlite_test.go @@ -0,0 +1,35 @@ +// SPDX-License-Identifier: MPL-2.0 + +package cdc + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestSQLiteConfigValidate(t *testing.T) { + missing := &SQLiteConfig{} + assert.ErrorIs(t, missing.Validate(), ErrDBResourceRequired) + + badInterval := &SQLiteConfig{DBResource: "app:db", StatusInterval: "nope"} + assert.ErrorIs(t, badInterval.Validate(), ErrInvalidInterval) + + negative := &SQLiteConfig{DBResource: "app:db", StatusInterval: "-5s"} + assert.ErrorIs(t, negative.Validate(), ErrInvalidInterval) + + ok := &SQLiteConfig{DBResource: "app:db", StatusInterval: "5s", Tables: []string{"users"}, Snapshot: true} + require.NoError(t, ok.Validate()) + + d, err := ok.StatusDuration() + require.NoError(t, err) + assert.Equal(t, "5s", d.String()) +} + +func TestSQLiteConfigZeroInterval(t *testing.T) { + cfg := &SQLiteConfig{DBResource: "app:db"} + d, err := cfg.StatusDuration() + require.NoError(t, err) + assert.Equal(t, int64(0), int64(d)) +} diff --git a/api/service/cdc/context.go b/api/service/cdc/context.go index cc0f98df0..2ea298b19 100644 --- a/api/service/cdc/context.go +++ b/api/service/cdc/context.go @@ -8,6 +8,9 @@ type SourceInfo struct { Name string `json:"name"` Slot string `json:"slot"` Publication string `json:"publication,omitempty"` + Engine string `json:"engine,omitempty"` + File string `json:"file,omitempty"` + DBResource string `json:"db_resource,omitempty"` Tables []string `json:"tables,omitempty"` Streaming bool `json:"streaming,omitempty"` Failover bool `json:"failover,omitempty"` diff --git a/api/service/cdc/errors.go b/api/service/cdc/errors.go index ffdeae95b..f26752037 100644 --- a/api/service/cdc/errors.go +++ b/api/service/cdc/errors.go @@ -15,4 +15,6 @@ var ( ErrInvalidInterval = apierror.New(apierror.Invalid, "interval must be a non-negative duration (e.g. 10s)").WithRetryable(apierror.False) ErrFailoverTemporary = apierror.New(apierror.Invalid, "failover cannot be set on a temporary slot").WithRetryable(apierror.False) ErrInvalidSnapshotFetchSize = apierror.New(apierror.Invalid, "snapshot_fetch_size must be non-negative").WithRetryable(apierror.False) + ErrDBResourceRequired = apierror.New(apierror.Invalid, "db_resource is required").WithRetryable(apierror.False) + ErrSourceNotFound = apierror.New(apierror.NotFound, "cdc source not found").WithRetryable(apierror.False) ) diff --git a/api/service/sql/config.go b/api/service/sql/config.go index 40ed40511..230001078 100644 --- a/api/service/sql/config.go +++ b/api/service/sql/config.go @@ -36,6 +36,14 @@ const ( DefaultMaxLifetime = 1 * time.Hour ) +// EngineConfig is the contract every engine configuration satisfies, letting the +// generic pool lifecycle validate and read lifecycle settings without knowing the +// concrete engine type. +type EngineConfig interface { + Validate() error + LifecycleConfig() supervisor.LifecycleConfig +} + type ( // PoolConfig defines settings for a database connection pool PoolConfig struct { @@ -106,6 +114,11 @@ func (c *SQLiteConfig) InitDefaults() { c.Lifecycle.InitDefaults() } +// LifecycleConfig returns the supervisor lifecycle settings for the database. +func (c *DBConfig) LifecycleConfig() supervisor.LifecycleConfig { + return c.Lifecycle +} + // Validate checks if the DBConfig has all required fields set to valid values func (c *DBConfig) Validate() error { if c.Host == "" { @@ -143,6 +156,11 @@ func (c *DBConfig) Validate() error { return nil } +// LifecycleConfig returns the supervisor lifecycle settings for the database. +func (c *SQLiteConfig) LifecycleConfig() supervisor.LifecycleConfig { + return c.Lifecycle +} + // Validate checks if the SQLiteConfig has all required fields set to valid values func (c *SQLiteConfig) Validate() error { if c.File == "" { diff --git a/boot/components/service/storage/cdc.go b/boot/components/service/storage/cdc.go index 77c206962..196020b21 100644 --- a/boot/components/service/storage/cdc.go +++ b/boot/components/service/storage/cdc.go @@ -9,30 +9,41 @@ import ( "github.com/wippyai/runtime/api/event" logapi "github.com/wippyai/runtime/api/logs" "github.com/wippyai/runtime/api/payload" + resourceapi "github.com/wippyai/runtime/api/resource" cdcapi "github.com/wippyai/runtime/api/service/cdc" bootpkg "github.com/wippyai/runtime/boot" bootsystem "github.com/wippyai/runtime/boot/components/system" - cdc "github.com/wippyai/runtime/service/cdc/postgres" + pgcdc "github.com/wippyai/runtime/service/cdc/postgres" + sqlitecdc "github.com/wippyai/runtime/service/cdc/sqlite" ) func CDC() boot.Component { return boot.New(boot.P{ Name: CDCName, - DependsOn: []boot.Name{bootsystem.EnvironmentName}, + DependsOn: []boot.Name{bootsystem.EnvironmentName, bootsystem.ResourcesName}, Load: func(ctx context.Context) (context.Context, error) { logger := logapi.GetLogger(ctx) dtt := payload.GetTranscoder(ctx) bus := event.GetBus(ctx) + resReg := resourceapi.GetRegistry(ctx) handlers := bootpkg.GetHandlerRegistry(ctx) - manager, err := cdc.NewManager(dtt, bus, logger.Named("cdc")) + pgManager, err := pgcdc.NewManager(dtt, bus, logger.Named("cdc.postgres")) if err != nil { return ctx, NewCDCManagerError(err) } - handlers.RegisterListener("db.cdc.*", manager) - ctx = cdcapi.WithSourceInspector(ctx, manager) - ctx = cdcapi.WithSourceStreamer(ctx, manager) + sqliteManager, err := sqlitecdc.NewManager(dtt, bus, logger.Named("cdc.sqlite"), resReg) + if err != nil { + return ctx, NewCDCManagerError(err) + } + + handlers.RegisterListener("db.cdc.postgres", pgManager) + handlers.RegisterListener("db.cdc.sqlite", sqliteManager) + + composite := cdcapi.NewComposite(pgManager, sqliteManager) + ctx = cdcapi.WithSourceInspector(ctx, composite) + ctx = cdcapi.WithSourceStreamer(ctx, composite) return ctx, nil }, }) diff --git a/boot/components/service/storage/sql.go b/boot/components/service/storage/sql.go index 70e078c5c..b59786e91 100644 --- a/boot/components/service/storage/sql.go +++ b/boot/components/service/storage/sql.go @@ -6,12 +6,16 @@ import ( "context" "github.com/wippyai/runtime/api/boot" + envapi "github.com/wippyai/runtime/api/env" "github.com/wippyai/runtime/api/event" logapi "github.com/wippyai/runtime/api/logs" "github.com/wippyai/runtime/api/payload" bootpkg "github.com/wippyai/runtime/boot" bootsystem "github.com/wippyai/runtime/boot/components/system" "github.com/wippyai/runtime/service/sql" + + // Register the built-in SQL engines (postgres, mysql, sqlite) with the manager. + _ "github.com/wippyai/runtime/service/sql/engine/all" ) func SQL() boot.Component { @@ -22,12 +26,14 @@ func SQL() boot.Component { logger := logapi.GetLogger(ctx) dtt := payload.GetTranscoder(ctx) bus := event.GetBus(ctx) + envRegistry := envapi.GetRegistry(ctx) handlers := bootpkg.GetHandlerRegistry(ctx) manager, err := sql.NewManager( dtt, bus, logger.Named("sql"), + envRegistry, ) if err != nil { return ctx, NewSQLManagerError(err) diff --git a/service/cdc/sqlite/bench_test.go b/service/cdc/sqlite/bench_test.go new file mode 100644 index 000000000..3de73c668 --- /dev/null +++ b/service/cdc/sqlite/bench_test.go @@ -0,0 +1,20 @@ +// SPDX-License-Identifier: MPL-2.0 + +package sqlite + +import "testing" + +func BenchmarkMapRow(b *testing.B) { + cols := []columnInfo{ + {name: "id"}, + {name: "email", text: true}, + {name: "balance"}, + {name: "blob"}, + } + vals := []any{int64(1), []byte("user@example.com"), 42.5, []byte{0x00, 0x01, 0x02}} + + b.ReportAllocs() + for i := 0; i < b.N; i++ { + _ = mapRow(cols, vals) + } +} diff --git a/service/cdc/sqlite/checkpoint.go b/service/cdc/sqlite/checkpoint.go new file mode 100644 index 000000000..bc1a7c3aa --- /dev/null +++ b/service/cdc/sqlite/checkpoint.go @@ -0,0 +1,60 @@ +// SPDX-License-Identifier: MPL-2.0 + +//go:build sqlite_preupdate_hook + +package sqlite + +import ( + "context" + "database/sql" +) + +const createOffsetsSQL = `CREATE TABLE IF NOT EXISTS ` + offsetsTable + ` ( + source TEXT PRIMARY KEY, + last_seq INTEGER NOT NULL DEFAULT 0, + snapshot_done INTEGER NOT NULL DEFAULT 0, + updated_at TEXT NOT NULL DEFAULT (datetime('now')) +)` + +func ensureOffsets(ctx context.Context, db *sql.DB) error { + _, err := db.ExecContext(ctx, createOffsetsSQL) + return err +} + +func loadOffset(ctx context.Context, db *sql.DB, source string) (snapshotDone bool, lastSeq uint64, err error) { + row := db.QueryRowContext(ctx, "SELECT snapshot_done, last_seq FROM "+offsetsTable+" WHERE source = ?", source) + var done int + var seq int64 + switch scanErr := row.Scan(&done, &seq); scanErr { + case nil: + return done != 0, uint64(seq), nil + case sql.ErrNoRows: + return false, 0, nil + default: + return false, 0, scanErr + } +} + +func saveSnapshotDone(ctx context.Context, db *sql.DB, source string) error { + _, err := db.ExecContext(ctx, + "INSERT INTO "+offsetsTable+" (source, snapshot_done, updated_at) VALUES (?, 1, datetime('now')) "+ + "ON CONFLICT(source) DO UPDATE SET snapshot_done = 1, updated_at = datetime('now')", + source) + return err +} + +func saveOffset(ctx context.Context, db *sql.DB, source string, seq uint64) error { + if db == nil { + return nil + } + _, err := db.ExecContext(ctx, + "INSERT INTO "+offsetsTable+" (source, last_seq, updated_at) VALUES (?, ?, datetime('now')) "+ + "ON CONFLICT(source) DO UPDATE SET last_seq = MAX(last_seq, excluded.last_seq), updated_at = datetime('now')", + source, int64(seq)) + return err +} + +func deleteOffset(ctx context.Context, db *sql.DB, source string) error { + _, err := db.ExecContext(ctx, "DELETE FROM "+offsetsTable+" WHERE source = ?", source) + return err +} diff --git a/service/cdc/sqlite/decode.go b/service/cdc/sqlite/decode.go new file mode 100644 index 000000000..9decc12d0 --- /dev/null +++ b/service/cdc/sqlite/decode.go @@ -0,0 +1,52 @@ +// SPDX-License-Identifier: MPL-2.0 + +package sqlite + +import ( + "strconv" + "strings" + "unicode/utf8" +) + +type columnInfo struct { + name string + text bool +} + +func textAffinity(declType string) bool { + t := strings.ToUpper(declType) + if strings.Contains(t, "BLOB") || t == "" { + return false + } + return strings.Contains(t, "CHAR") || strings.Contains(t, "CLOB") || strings.Contains(t, "TEXT") +} + +func mapRow(cols []columnInfo, vals []any) map[string]any { + if vals == nil { + return nil + } + out := make(map[string]any, len(vals)) + for i, v := range vals { + name, text := columnAt(cols, i) + out[name] = normalizeValue(v, text) + } + return out +} + +func columnAt(cols []columnInfo, i int) (string, bool) { + if i < len(cols) { + return cols[i].name, cols[i].text + } + return "column" + strconv.Itoa(i), false +} + +func normalizeValue(v any, text bool) any { + b, ok := v.([]byte) + if !ok { + return v + } + if text && utf8.Valid(b) { + return string(b) + } + return b +} diff --git a/service/cdc/sqlite/decode_test.go b/service/cdc/sqlite/decode_test.go new file mode 100644 index 000000000..f1307111f --- /dev/null +++ b/service/cdc/sqlite/decode_test.go @@ -0,0 +1,64 @@ +// SPDX-License-Identifier: MPL-2.0 + +package sqlite + +import ( + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestTextAffinity(t *testing.T) { + cases := map[string]bool{ + "TEXT": true, + "VARCHAR(255)": true, + "CLOB": true, + "nchar": true, + "INTEGER": false, + "REAL": false, + "BLOB": false, + "": false, + "NUMERIC": false, + } + for decl, want := range cases { + assert.Equalf(t, want, textAffinity(decl), "decl=%q", decl) + } +} + +func TestMapRowNilForMissingSide(t *testing.T) { + assert.Nil(t, mapRow([]columnInfo{{name: "id"}}, nil)) +} + +func TestMapRowDecodesTextBytesByAffinity(t *testing.T) { + cols := []columnInfo{ + {name: "id", text: false}, + {name: "email", text: true}, + {name: "payload", text: false}, + } + vals := []any{int64(7), []byte("a@b.com"), []byte{0x00, 0x01, 0x02}} + + out := mapRow(cols, vals) + + assert.Equal(t, int64(7), out["id"]) + assert.Equal(t, "a@b.com", out["email"]) + assert.Equal(t, []byte{0x00, 0x01, 0x02}, out["payload"]) +} + +func TestMapRowKeepsInvalidUTF8AsBytes(t *testing.T) { + cols := []columnInfo{{name: "data", text: true}} + invalid := []byte{0xff, 0xfe, 0xfd} + out := mapRow(cols, []any{invalid}) + assert.Equal(t, invalid, out["data"]) +} + +func TestMapRowFallbackColumnNames(t *testing.T) { + out := mapRow(nil, []any{int64(1), "x"}) + assert.Equal(t, int64(1), out["column0"]) + assert.Equal(t, "x", out["column1"]) +} + +func TestNormalizeValuePassThrough(t *testing.T) { + assert.Equal(t, int64(3), normalizeValue(int64(3), true)) + assert.Equal(t, 1.5, normalizeValue(1.5, true)) + assert.Nil(t, normalizeValue(nil, true)) +} diff --git a/service/cdc/sqlite/errors.go b/service/cdc/sqlite/errors.go new file mode 100644 index 000000000..714a00c1f --- /dev/null +++ b/service/cdc/sqlite/errors.go @@ -0,0 +1,55 @@ +// SPDX-License-Identifier: MPL-2.0 + +package sqlite + +import ( + "errors" + + "github.com/wippyai/runtime/api/attrs" + apierror "github.com/wippyai/runtime/api/error" + "github.com/wippyai/runtime/api/registry" +) + +var ( + ErrSourceClosed = errors.New("cdc: source is closed") + ErrTranscoderRequired = apierror.New(apierror.Invalid, "transcoder is required").WithRetryable(apierror.False) + ErrEventBusRequired = apierror.New(apierror.Invalid, "event bus is required").WithRetryable(apierror.False) + ErrResourceRegRequired = apierror.New(apierror.Invalid, "resource registry is required").WithRetryable(apierror.False) + ErrPreupdateTagRequired = apierror.New(apierror.Internal, "sqlite cdc requires the sqlite_preupdate_hook build tag").WithRetryable(apierror.False) + ErrNoSourceStreamer = apierror.New(apierror.Internal, "cdc source streamer not available").WithRetryable(apierror.False) + ErrChangeBacklogOverflow = apierror.New(apierror.Unavailable, "sqlite cdc change backlog overflow").WithRetryable(apierror.True) +) + +func NewUnsupportedEntryKindError(kind registry.Kind) apierror.Error { + return apierror.New(apierror.Invalid, "unsupported entry kind"). + WithRetryable(apierror.False). + WithDetails(attrs.NewBagFrom(map[string]any{"kind": kind})) +} + +func NewServiceExistsError(id registry.ID) apierror.Error { + return apierror.New(apierror.Conflict, "cdc service already exists"). + WithRetryable(apierror.False). + WithDetails(attrs.NewBagFrom(map[string]any{"id": id.String()})) +} + +func NewServiceNotFoundError(id registry.ID) apierror.Error { + return apierror.New(apierror.NotFound, "cdc service not found"). + WithRetryable(apierror.False). + WithDetails(attrs.NewBagFrom(map[string]any{"id": id.String()})) +} + +func NewInvalidConfigError(err error) apierror.Error { + apiErr := apierror.New(apierror.Invalid, "invalid cdc configuration").WithRetryable(apierror.False) + if err != nil { + apiErr = apiErr.WithDetails(attrs.NewBagFrom(map[string]any{"cause": err.Error()})).WithCause(err) + } + return apiErr +} + +func NewSourceCreationError(err error) apierror.Error { + apiErr := apierror.New(apierror.Internal, "failed to create cdc source").WithRetryable(apierror.False) + if err != nil { + apiErr = apiErr.WithDetails(attrs.NewBagFrom(map[string]any{"cause": err.Error()})).WithCause(err) + } + return apiErr +} diff --git a/service/cdc/sqlite/hook.go b/service/cdc/sqlite/hook.go new file mode 100644 index 000000000..a8a2a5f27 --- /dev/null +++ b/service/cdc/sqlite/hook.go @@ -0,0 +1,152 @@ +// SPDX-License-Identifier: MPL-2.0 + +//go:build sqlite_preupdate_hook + +package sqlite + +import ( + "path/filepath" + "sync" + + "database/sql" + + "github.com/mattn/go-sqlite3" + + apierror "github.com/wippyai/runtime/api/error" + sqlconfig "github.com/wippyai/runtime/api/service/sql" + sqlservice "github.com/wippyai/runtime/service/sql" +) + +// sqliteCDCDriver is a SQLite driver variant whose ConnectHook rebinds preupdate +// hooks on every connection the pool opens to a file with a registered sink. +const sqliteCDCDriver = "sqlite3_wippy" + +const ( + cdcInsert = sqlite3.SQLITE_INSERT + cdcUpdate = sqlite3.SQLITE_UPDATE + cdcDelete = sqlite3.SQLITE_DELETE +) + +var ( + errNotSQLiteConn = apierror.New(apierror.Invalid, "underlying connection is not a SQLite connection").WithRetryable(apierror.False) + errCDCMemoryUnsupported = apierror.New(apierror.Invalid, "sqlite cdc requires a file-backed database").WithRetryable(apierror.False) +) + +// cdcSink receives row-level changes observed on the writer connection. +type cdcSink interface { + PreUpdate(op int, table string, rowid int64, old, new []any) + Commit() + Rollback() +} + +var ( + cdcMu sync.RWMutex + cdcSinks = make(map[string]cdcSink) +) + +func init() { + sql.Register(sqliteCDCDriver, &sqlite3.SQLiteDriver{ConnectHook: cdcConnectHook}) + sqlservice.RegisterDriver(sqlconfig.SQLite, sqliteCDCDriver) +} + +func cdcConnectHook(conn *sqlite3.SQLiteConn) error { + file := normalizeCDCPath(conn.GetFilename("main")) + if file == "" { + return nil + } + cdcMu.RLock() + sink, ok := cdcSinks[file] + cdcMu.RUnlock() + if ok { + bindCDCHooks(conn, sink) + } + return nil +} + +func registerSink(file string, sink cdcSink) { + cdcMu.Lock() + cdcSinks[file] = sink + cdcMu.Unlock() +} + +func unregisterSink(file string) { + cdcMu.Lock() + delete(cdcSinks, file) + cdcMu.Unlock() +} + +func installHooksOnRaw(raw any, sink cdcSink) (string, error) { + conn, ok := raw.(*sqlite3.SQLiteConn) + if !ok { + return "", errNotSQLiteConn + } + file := normalizeCDCPath(conn.GetFilename("main")) + if file == "" { + return "", errCDCMemoryUnsupported + } + bindCDCHooks(conn, sink) + return file, nil +} + +func clearHooksOnRaw(raw any) error { + conn, ok := raw.(*sqlite3.SQLiteConn) + if !ok { + return errNotSQLiteConn + } + conn.RegisterPreUpdateHook(nil) + conn.RegisterCommitHook(nil) + conn.RegisterRollbackHook(nil) + return nil +} + +func normalizeCDCPath(path string) string { + if path == "" { + return "" + } + abs, err := filepath.Abs(path) + if err != nil { + return filepath.Clean(path) + } + return abs +} + +func bindCDCHooks(conn *sqlite3.SQLiteConn, sink cdcSink) { + conn.RegisterPreUpdateHook(func(d sqlite3.SQLitePreUpdateData) { + count := d.Count() + var oldRow, newRow []any + var rowid int64 + switch d.Op { + case sqlite3.SQLITE_INSERT: + newRow = scanPreUpdateRow(&d, count, true) + rowid = d.NewRowID + case sqlite3.SQLITE_DELETE: + oldRow = scanPreUpdateRow(&d, count, false) + rowid = d.OldRowID + case sqlite3.SQLITE_UPDATE: + oldRow = scanPreUpdateRow(&d, count, false) + newRow = scanPreUpdateRow(&d, count, true) + rowid = d.NewRowID + } + sink.PreUpdate(d.Op, d.TableName, rowid, oldRow, newRow) + }) + conn.RegisterCommitHook(func() int { + sink.Commit() + return 0 + }) + conn.RegisterRollbackHook(func() { + sink.Rollback() + }) +} + +func scanPreUpdateRow(d *sqlite3.SQLitePreUpdateData, count int, isNew bool) []any { + if count <= 0 { + return nil + } + vals := make([]any, count) + if isNew { + _ = d.New(vals...) + } else { + _ = d.Old(vals...) + } + return vals +} diff --git a/service/cdc/sqlite/integration_test.go b/service/cdc/sqlite/integration_test.go new file mode 100644 index 000000000..0c1a7718a --- /dev/null +++ b/service/cdc/sqlite/integration_test.go @@ -0,0 +1,286 @@ +// SPDX-License-Identifier: MPL-2.0 + +//go:build integration && sqlite_preupdate_hook + +package sqlite + +import ( + "context" + "database/sql" + "path/filepath" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/wippyai/runtime/api/registry" + "github.com/wippyai/runtime/api/resource" + config "github.com/wippyai/runtime/api/service/cdc" + sqlconfig "github.com/wippyai/runtime/api/service/sql" + sqlservice "github.com/wippyai/runtime/service/sql" +) + +type fakeResource struct{ res sqlservice.DBResource } + +func (f *fakeResource) Get() (any, error) { return f.res, nil } +func (f *fakeResource) Release() {} + +type fakeRegistry struct{ db *sql.DB } + +func (r *fakeRegistry) Acquire(context.Context, registry.ID, resource.AccessMode) (resource.Resource[any], error) { + return &fakeResource{res: sqlservice.DBResource{DB: r.db, Type: sqlconfig.SQLite}}, nil +} +func (r *fakeRegistry) List() ([]registry.ID, error) { return nil, nil } +func (r *fakeRegistry) Exists(registry.ID) bool { return true } + +func openPool(t *testing.T) (*sql.DB, string) { + t.Helper() + file := filepath.Join(t.TempDir(), "app.db") + db, err := sql.Open("sqlite3_wippy", "file:"+file+"?mode=rwc") + require.NoError(t, err) + db.SetMaxOpenConns(1) + db.SetMaxIdleConns(1) + _, err = db.Exec("PRAGMA journal_mode=WAL") + require.NoError(t, err) + return db, file +} + +func newSource(t *testing.T, db *sql.DB, opts sourceOptions) *Source { + t.Helper() + opts.res = &fakeRegistry{db: db} + opts.dbResource = registry.NewID("app", "db") + if opts.name == "" { + opts.name = "test-src" + } + if opts.statusInterval == "" { + opts.statusInterval = "1s" + } + h, err := buildSource(opts) + require.NoError(t, err) + return h.(*Source) +} + +func waitChange(t *testing.T, ch <-chan config.Change) config.Change { + t.Helper() + select { + case c := <-ch: + return c + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for change") + return config.Change{} + } +} + +func TestIntegrationInsertUpdateDelete(t *testing.T) { + db, _ := openPool(t) + _, err := db.Exec(`CREATE TABLE users (id INTEGER PRIMARY KEY, email TEXT, balance REAL)`) + require.NoError(t, err) + + src := newSource(t, db, sourceOptions{}) + _, err = src.Start(context.Background()) + require.NoError(t, err) + defer func() { _ = src.Stop(context.Background()) }() + + stream := src.Subscribe(config.StreamOptions{}) + defer stream.Close() + + _, err = db.Exec(`INSERT INTO users (id, email, balance) VALUES (1, 'a@b.com', 42.5)`) + require.NoError(t, err) + ins := waitChange(t, stream.Changes()) + assert.Equal(t, "insert", ins.Op) + assert.Equal(t, "users", ins.Table) + assert.Equal(t, "a@b.com", ins.After["email"]) + assert.Equal(t, 42.5, ins.After["balance"]) + assert.Equal(t, int64(1), ins.After["id"]) + assert.Nil(t, ins.Before) + + _, err = db.Exec(`UPDATE users SET balance = 99.0 WHERE id = 1`) + require.NoError(t, err) + upd := waitChange(t, stream.Changes()) + assert.Equal(t, "update", upd.Op) + assert.Equal(t, 42.5, upd.Before["balance"]) + assert.Equal(t, 99.0, upd.After["balance"]) + + _, err = db.Exec(`DELETE FROM users WHERE id = 1`) + require.NoError(t, err) + del := waitChange(t, stream.Changes()) + assert.Equal(t, "delete", del.Op) + assert.Equal(t, "a@b.com", del.Before["email"]) + assert.Nil(t, del.After) +} + +func TestIntegrationValueFidelity(t *testing.T) { + db, _ := openPool(t) + _, err := db.Exec(`CREATE TABLE items (id INTEGER PRIMARY KEY, name TEXT, qty INTEGER, price REAL, blob BLOB, note TEXT)`) + require.NoError(t, err) + + src := newSource(t, db, sourceOptions{}) + _, err = src.Start(context.Background()) + require.NoError(t, err) + defer func() { _ = src.Stop(context.Background()) }() + + stream := src.Subscribe(config.StreamOptions{}) + defer stream.Close() + + _, err = db.Exec(`INSERT INTO items (id, name, qty, price, blob, note) VALUES (1, 'widget', 7, 1.25, X'00ff10', NULL)`) + require.NoError(t, err) + + c := waitChange(t, stream.Changes()) + assert.Equal(t, "widget", c.After["name"]) + assert.Equal(t, int64(7), c.After["qty"]) + assert.Equal(t, 1.25, c.After["price"]) + assert.Equal(t, []byte{0x00, 0xff, 0x10}, c.After["blob"]) + assert.Nil(t, c.After["note"]) +} + +func TestIntegrationRollbackDiscarded(t *testing.T) { + db, _ := openPool(t) + _, err := db.Exec(`CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT)`) + require.NoError(t, err) + + src := newSource(t, db, sourceOptions{}) + _, err = src.Start(context.Background()) + require.NoError(t, err) + defer func() { _ = src.Stop(context.Background()) }() + + stream := src.Subscribe(config.StreamOptions{}) + defer stream.Close() + + tx, err := db.Begin() + require.NoError(t, err) + _, err = tx.Exec(`INSERT INTO t (id, v) VALUES (1, 'rolled-back')`) + require.NoError(t, err) + require.NoError(t, tx.Rollback()) + + _, err = db.Exec(`INSERT INTO t (id, v) VALUES (2, 'committed')`) + require.NoError(t, err) + + c := waitChange(t, stream.Changes()) + assert.Equal(t, "insert", c.Op) + assert.Equal(t, "committed", c.After["v"]) + assert.Equal(t, int64(2), c.After["id"]) +} + +func TestIntegrationSnapshotBootstrap(t *testing.T) { + db, _ := openPool(t) + _, err := db.Exec(`CREATE TABLE users (id INTEGER PRIMARY KEY, email TEXT)`) + require.NoError(t, err) + _, err = db.Exec(`INSERT INTO users (id, email) VALUES (1, 'existing@b.com')`) + require.NoError(t, err) + + src := newSource(t, db, sourceOptions{snapshot: true}) + stream := src.Subscribe(config.StreamOptions{}) + defer stream.Close() + + _, err = src.Start(context.Background()) + require.NoError(t, err) + defer func() { _ = src.Stop(context.Background()) }() + + snap := waitChange(t, stream.Changes()) + assert.Equal(t, "snapshot", snap.Op) + assert.Equal(t, "existing@b.com", snap.After["email"]) + + _, err = db.Exec(`INSERT INTO users (id, email) VALUES (2, 'new@b.com')`) + require.NoError(t, err) + live := waitChange(t, stream.Changes()) + assert.Equal(t, "insert", live.Op) + assert.Equal(t, "new@b.com", live.After["email"]) +} + +func TestIntegrationRestartKeepsCheckpoint(t *testing.T) { + db, _ := openPool(t) + _, err := db.Exec(`CREATE TABLE users (id INTEGER PRIMARY KEY, email TEXT)`) + require.NoError(t, err) + _, err = db.Exec(`INSERT INTO users (id, email) VALUES (1, 'existing@b.com')`) + require.NoError(t, err) + + first := newSource(t, db, sourceOptions{snapshot: true, name: "src"}) + s1 := first.Subscribe(config.StreamOptions{}) + _, err = first.Start(context.Background()) + require.NoError(t, err) + snap := waitChange(t, s1.Changes()) + require.Equal(t, "snapshot", snap.Op) + s1.Close() + require.NoError(t, first.Stop(context.Background())) + + second := newSource(t, db, sourceOptions{snapshot: true, name: "src"}) + s2 := second.Subscribe(config.StreamOptions{}) + defer s2.Close() + _, err = second.Start(context.Background()) + require.NoError(t, err) + defer func() { _ = second.Stop(context.Background()) }() + + _, err = db.Exec(`INSERT INTO users (id, email) VALUES (2, 'new@b.com')`) + require.NoError(t, err) + + got := waitChange(t, s2.Changes()) + assert.Equal(t, "insert", got.Op, "restart must not re-snapshot; first event should be the live insert") + assert.Equal(t, "new@b.com", got.After["email"]) +} + +func TestIntegrationLaggardDoesNotStallWrites(t *testing.T) { + db, _ := openPool(t) + _, err := db.Exec(`CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT)`) + require.NoError(t, err) + + src := newSource(t, db, sourceOptions{}) + _, err = src.Start(context.Background()) + require.NoError(t, err) + defer func() { _ = src.Stop(context.Background()) }() + + laggard := src.Subscribe(config.StreamOptions{Buffer: 1}) + defer laggard.Close() + + done := make(chan error, 1) + go func() { + for i := 0; i < 500; i++ { + if _, e := db.Exec(`INSERT INTO t (v) VALUES ('x')`); e != nil { + done <- e + return + } + } + done <- nil + }() + + select { + case e := <-done: + require.NoError(t, e) + case <-time.After(10 * time.Second): + t.Fatal("writes stalled: a non-reading subscriber blocked the writer") + } + + reader := src.Subscribe(config.StreamOptions{}) + defer reader.Close() + _, err = db.Exec(`INSERT INTO t (v) VALUES ('final')`) + require.NoError(t, err) + + got := waitChange(t, reader.Changes()) + assert.Equal(t, "insert", got.Op) + assert.Equal(t, "final", got.After["v"]) +} + +func TestIntegrationTableAllowlist(t *testing.T) { + db, _ := openPool(t) + _, err := db.Exec(`CREATE TABLE users (id INTEGER PRIMARY KEY, v TEXT)`) + require.NoError(t, err) + _, err = db.Exec(`CREATE TABLE orders (id INTEGER PRIMARY KEY, v TEXT)`) + require.NoError(t, err) + + src := newSource(t, db, sourceOptions{tables: []string{"users"}}) + _, err = src.Start(context.Background()) + require.NoError(t, err) + defer func() { _ = src.Stop(context.Background()) }() + + stream := src.Subscribe(config.StreamOptions{}) + defer stream.Close() + + _, err = db.Exec(`INSERT INTO orders (id, v) VALUES (1, 'ignored')`) + require.NoError(t, err) + _, err = db.Exec(`INSERT INTO users (id, v) VALUES (1, 'captured')`) + require.NoError(t, err) + + c := waitChange(t, stream.Changes()) + assert.Equal(t, "users", c.Table) + assert.Equal(t, "captured", c.After["v"]) +} diff --git a/service/cdc/sqlite/manager.go b/service/cdc/sqlite/manager.go new file mode 100644 index 000000000..58d9f8828 --- /dev/null +++ b/service/cdc/sqlite/manager.go @@ -0,0 +1,248 @@ +// SPDX-License-Identifier: MPL-2.0 + +package sqlite + +import ( + "context" + "sync" + + "github.com/wippyai/runtime/api/event" + "github.com/wippyai/runtime/api/payload" + "github.com/wippyai/runtime/api/registry" + "github.com/wippyai/runtime/api/resource" + config "github.com/wippyai/runtime/api/service/cdc" + "github.com/wippyai/runtime/api/supervisor" + entryutil "github.com/wippyai/runtime/system/entry" + "go.uber.org/zap" +) + +type sourceHandle interface { + supervisor.Service + Subscribe(opts config.StreamOptions) config.ChangeStream + closeSubscriptions() + markDrop() +} + +type sourceOptions struct { + res resource.Registry + log *zap.Logger + dbResource registry.ID + name string + statusInterval string + tables []string + snapshot bool +} + +type Manager struct { + dtt payload.Transcoder + bus event.Bus + res resource.Registry + log *zap.Logger + sources map[registry.ID]sourceHandle + infos map[registry.ID]config.SourceInfo + infosByName map[string]registry.ID + mu sync.Mutex +} + +func NewManager(dtt payload.Transcoder, bus event.Bus, log *zap.Logger, res resource.Registry) (*Manager, error) { + if dtt == nil { + return nil, ErrTranscoderRequired + } + if bus == nil { + return nil, ErrEventBusRequired + } + if res == nil { + return nil, ErrResourceRegRequired + } + if log == nil { + log = zap.NewNop() + } + return &Manager{ + dtt: dtt, + bus: bus, + res: res, + log: log, + sources: make(map[registry.ID]sourceHandle), + infos: make(map[registry.ID]config.SourceInfo), + infosByName: make(map[string]registry.ID), + }, nil +} + +func (m *Manager) Add(ctx context.Context, entry registry.Entry) error { + m.mu.Lock() + defer m.mu.Unlock() + + if entry.Kind != config.SQLite { + return NewUnsupportedEntryKindError(entry.Kind) + } + if _, exists := m.sources[entry.ID]; exists { + return NewServiceExistsError(entry.ID) + } + + cfg, err := entryutil.DecodeEntryConfig[config.SQLiteConfig](ctx, m.dtt, entry) + if err != nil { + return NewInvalidConfigError(err) + } + if err := cfg.Validate(); err != nil { + return NewInvalidConfigError(err) + } + + src, err := buildSource(m.sourceOptions(entry, cfg)) + if err != nil { + return NewSourceCreationError(err) + } + + m.sources[entry.ID] = src + m.storeInfo(entry, cfg) + m.register(ctx, entry, src, cfg.Lifecycle) + return nil +} + +func (m *Manager) Update(ctx context.Context, entry registry.Entry) error { + m.mu.Lock() + defer m.mu.Unlock() + + if entry.Kind != config.SQLite { + return NewUnsupportedEntryKindError(entry.Kind) + } + if _, exists := m.sources[entry.ID]; !exists { + return NewServiceNotFoundError(entry.ID) + } + + cfg, err := entryutil.DecodeEntryConfig[config.SQLiteConfig](ctx, m.dtt, entry) + if err != nil { + return NewInvalidConfigError(err) + } + if err := cfg.Validate(); err != nil { + return NewInvalidConfigError(err) + } + + src, err := buildSource(m.sourceOptions(entry, cfg)) + if err != nil { + return NewSourceCreationError(err) + } + + if old := m.sources[entry.ID]; old != nil { + old.closeSubscriptions() + } + m.removeInfo(entry.ID) + m.unregister(ctx, entry) + + m.sources[entry.ID] = src + m.storeInfo(entry, cfg) + m.register(ctx, entry, src, cfg.Lifecycle) + return nil +} + +func (m *Manager) Delete(ctx context.Context, entry registry.Entry) error { + m.mu.Lock() + defer m.mu.Unlock() + + src, exists := m.sources[entry.ID] + if !exists { + return NewServiceNotFoundError(entry.ID) + } + src.markDrop() + src.closeSubscriptions() + m.removeInfo(entry.ID) + m.unregister(ctx, entry) + delete(m.sources, entry.ID) + return nil +} + +func (m *Manager) sourceOptions(entry registry.Entry, cfg *config.SQLiteConfig) sourceOptions { + return sourceOptions{ + res: m.res, + log: m.log.With(zap.String("id", entry.ID.String())), + name: entry.ID.String(), + dbResource: registry.ParseID(cfg.DBResource), + tables: cfg.Tables, + statusInterval: cfg.StatusInterval, + snapshot: cfg.Snapshot, + } +} + +func (m *Manager) storeInfo(entry registry.Entry, cfg *config.SQLiteConfig) { + info := config.SourceInfo{ + Name: entry.ID.String(), + Engine: "sqlite", + DBResource: cfg.DBResource, + Tables: append([]string(nil), cfg.Tables...), + Snapshot: cfg.Snapshot, + } + m.infos[entry.ID] = info + m.infosByName[info.Name] = entry.ID +} + +func (m *Manager) removeInfo(id registry.ID) { + if info, ok := m.infos[id]; ok { + if current, present := m.infosByName[info.Name]; present && current == id { + delete(m.infosByName, info.Name) + } + delete(m.infos, id) + } +} + +func (m *Manager) List() []config.SourceInfo { + m.mu.Lock() + defer m.mu.Unlock() + + out := make([]config.SourceInfo, 0, len(m.infos)) + for _, info := range m.infos { + out = append(out, info) + } + return out +} + +func (m *Manager) Get(name string) (config.SourceInfo, bool) { + m.mu.Lock() + defer m.mu.Unlock() + + if id, ok := m.infosByName[name]; ok { + if info, present := m.infos[id]; present { + return info, true + } + } + return config.SourceInfo{}, false +} + +func (m *Manager) Stream(_ context.Context, name string, opts config.StreamOptions) (config.ChangeStream, config.SourceInfo, error) { + m.mu.Lock() + src, info, ok := m.lookupSourceLocked(name) + m.mu.Unlock() + if !ok { + return nil, config.SourceInfo{}, NewServiceNotFoundError(registry.ParseID(name)) + } + return src.Subscribe(opts), info, nil +} + +func (m *Manager) lookupSourceLocked(name string) (sourceHandle, config.SourceInfo, bool) { + if id, ok := m.infosByName[name]; ok { + if src := m.sources[id]; src != nil { + return src, m.infos[id], true + } + } + return nil, config.SourceInfo{}, false +} + +func (m *Manager) register(ctx context.Context, entry registry.Entry, src sourceHandle, lifecycle supervisor.LifecycleConfig) { + m.bus.Send(ctx, event.Event{ + System: supervisor.System, + Kind: supervisor.ServiceRegister, + Path: entry.ID.String(), + Data: &supervisor.Entry{ + Service: src, + Config: lifecycle, + }, + }) + m.log.Info("added sqlite cdc source", zap.String("id", entry.ID.String()), zap.String("kind", entry.Kind)) +} + +func (m *Manager) unregister(ctx context.Context, entry registry.Entry) { + m.bus.Send(ctx, event.Event{ + System: supervisor.System, + Kind: supervisor.ServiceRemove, + Path: entry.ID.String(), + }) + m.log.Info("removed sqlite cdc source", zap.String("id", entry.ID.String())) +} diff --git a/service/cdc/sqlite/manager_test.go b/service/cdc/sqlite/manager_test.go new file mode 100644 index 000000000..c4dca30dc --- /dev/null +++ b/service/cdc/sqlite/manager_test.go @@ -0,0 +1,74 @@ +// SPDX-License-Identifier: MPL-2.0 + +package sqlite + +import ( + "sort" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/wippyai/runtime/api/registry" + config "github.com/wippyai/runtime/api/service/cdc" +) + +func newInspectorManager() *Manager { + return &Manager{ + sources: map[registry.ID]sourceHandle{}, + infos: map[registry.ID]config.SourceInfo{}, + infosByName: map[string]registry.ID{}, + } +} + +func TestNewManagerValidation(t *testing.T) { + _, err := NewManager(nil, nil, nil, nil) + assert.ErrorIs(t, err, ErrTranscoderRequired) +} + +func TestManagerStoreAndList(t *testing.T) { + m := newInspectorManager() + idA := registry.NewID("test", "id-a") + idB := registry.NewID("test", "id-b") + + m.storeInfo(registry.Entry{ID: idA, Kind: config.SQLite}, &config.SQLiteConfig{ + DBResource: "app:db", + Tables: []string{"users"}, + Snapshot: true, + }) + m.storeInfo(registry.Entry{ID: idB, Kind: config.SQLite}, &config.SQLiteConfig{ + DBResource: "app:db2", + }) + + infos := m.List() + require.Len(t, infos, 2) + names := []string{infos[0].Name, infos[1].Name} + sort.Strings(names) + assert.Equal(t, []string{idA.String(), idB.String()}, names) + + got, ok := m.Get(idA.String()) + require.True(t, ok) + assert.Equal(t, "sqlite", got.Engine) + assert.Equal(t, "app:db", got.DBResource) + assert.Equal(t, []string{"users"}, got.Tables) + assert.True(t, got.Snapshot) +} + +func TestManagerGetMiss(t *testing.T) { + m := newInspectorManager() + _, ok := m.Get("missing") + assert.False(t, ok) +} + +func TestManagerRemoveInfo(t *testing.T) { + m := newInspectorManager() + id := registry.NewID("test", "id-a") + m.storeInfo(registry.Entry{ID: id, Kind: config.SQLite}, &config.SQLiteConfig{DBResource: "app:db"}) + + m.removeInfo(id) + + _, ok := m.Get(id.String()) + assert.False(t, ok) + assert.Empty(t, m.List()) + assert.NotContains(t, m.infosByName, id.String()) +} diff --git a/service/cdc/sqlite/snapshot.go b/service/cdc/sqlite/snapshot.go new file mode 100644 index 000000000..132b49705 --- /dev/null +++ b/service/cdc/sqlite/snapshot.go @@ -0,0 +1,118 @@ +// SPDX-License-Identifier: MPL-2.0 + +//go:build sqlite_preupdate_hook + +package sqlite + +import ( + "context" + "database/sql" + "strconv" + "strings" + + config "github.com/wippyai/runtime/api/service/cdc" +) + +func (s *Source) runSnapshot(ctx context.Context, conn *sql.Conn) error { + tables, err := s.snapshotTables(ctx, conn) + if err != nil { + return err + } + for _, table := range tables { + if err := s.snapshotTable(ctx, conn, table); err != nil { + return err + } + } + return nil +} + +func (s *Source) snapshotTables(ctx context.Context, conn *sql.Conn) ([]string, error) { + rows, err := conn.QueryContext(ctx, "SELECT name FROM sqlite_master WHERE type = 'table' AND name NOT LIKE 'sqlite_%'") + if err != nil { + return nil, err + } + defer func() { _ = rows.Close() }() + + var tables []string + for rows.Next() { + var name string + if err := rows.Scan(&name); err != nil { + return nil, err + } + if s.tableAllowed(name) { + tables = append(tables, name) + } + } + return tables, rows.Err() +} + +func (s *Source) snapshotTable(ctx context.Context, conn *sql.Conn, table string) error { + cols := s.columnsFor(ctx, table) + + rows, err := conn.QueryContext(ctx, "SELECT * FROM "+quoteIdent(table)) //nolint:gosec // quoted identifier from sqlite_master; SQLite cannot bind table names as parameters + if err != nil { + return err + } + defer func() { _ = rows.Close() }() + + names, err := rows.Columns() + if err != nil { + return err + } + if len(cols) == 0 { + cols = columnsFromNames(names) + } + + for rows.Next() { + vals := make([]any, len(names)) + ptrs := make([]any, len(names)) + for i := range vals { + ptrs[i] = &vals[i] + } + if err := rows.Scan(ptrs...); err != nil { + return err + } + seq := s.seq.Add(1) + s.subs.publish(ctx, config.Change{ + Source: s.name, + Op: "snapshot", + Table: table, + Relation: table, + After: mapRow(cols, vals), + LSN: strconv.FormatUint(seq, 10), + }) + } + return rows.Err() +} + +func resolveColumns(ctx context.Context, db *sql.DB, table string) ([]columnInfo, error) { + rows, err := db.QueryContext(ctx, "PRAGMA table_info("+quoteIdent(table)+")") + if err != nil { + return nil, err + } + defer func() { _ = rows.Close() }() + + var cols []columnInfo + for rows.Next() { + var cid, notnull, pk int + var name, declType string + var dflt any + if err := rows.Scan(&cid, &name, &declType, ¬null, &dflt, &pk); err != nil { + return nil, err + } + cols = append(cols, columnInfo{name: name, text: textAffinity(declType)}) + } + return cols, rows.Err() +} + +func columnsFromNames(names []string) []columnInfo { + cols := make([]columnInfo, len(names)) + for i, n := range names { + cols[i] = columnInfo{name: n} + } + return cols +} + +func quoteIdent(name string) string { + return `"` + strings.ReplaceAll(name, `"`, `""`) + `"` +} diff --git a/service/cdc/sqlite/source.go b/service/cdc/sqlite/source.go new file mode 100644 index 000000000..aaa1aa27e --- /dev/null +++ b/service/cdc/sqlite/source.go @@ -0,0 +1,466 @@ +// SPDX-License-Identifier: MPL-2.0 + +//go:build sqlite_preupdate_hook + +package sqlite + +import ( + "context" + "database/sql" + "fmt" + "os" + "strconv" + "strings" + "sync" + "sync/atomic" + "time" + + "go.uber.org/zap" + + "github.com/wippyai/runtime/api/metrics" + "github.com/wippyai/runtime/api/registry" + "github.com/wippyai/runtime/api/resource" + config "github.com/wippyai/runtime/api/service/cdc" + sqlconfig "github.com/wippyai/runtime/api/service/sql" + sqlservice "github.com/wippyai/runtime/service/sql" +) + +const ( + offsetsTable = "wippy_cdc_offsets" + changesCounter = "wippy_cdc_changes_total" + walGauge = "wippy_cdc_wal_size_bytes" + defaultStatusInterval = 30 * time.Second + commitQueueSize = 256 + auxBusyTimeoutMillisec = 5000 +) + +type capturedChange struct { + table string + old []any + new []any + op int + rowid int64 +} + +type Source struct { + poolRes resource.Resource[any] + res resource.Registry + readDB *sql.DB + checkpointDB *sql.DB + runDone chan struct{} + commits chan []capturedChange + subs *subscribers + writerDB *sql.DB + cols map[string][]columnInfo + cancel context.CancelFunc + tables map[string]struct{} + log *zap.Logger + dbResID registry.ID + file string + name string + pending []capturedChange + statusInterval time.Duration + seq atomic.Uint64 + colMu sync.RWMutex + mu sync.Mutex + pendMu sync.Mutex + stopped atomic.Bool + dropCP atomic.Bool + snap bool +} + +func buildSource(opts sourceOptions) (sourceHandle, error) { + log := opts.log + if log == nil { + log = zap.NewNop() + } + interval := defaultStatusInterval + if opts.statusInterval != "" { + d, err := time.ParseDuration(opts.statusInterval) + if err != nil || d < 0 { + return nil, fmt.Errorf("invalid status_interval %q", opts.statusInterval) + } + if d > 0 { + interval = d + } + } + return &Source{ + log: log, + res: opts.res, + subs: newSubscribers(), + name: opts.name, + statusInterval: interval, + dbResID: opts.dbResource, + tables: filterSet(opts.tables), + cols: make(map[string][]columnInfo), + snap: opts.snapshot, + }, nil +} + +func (s *Source) Subscribe(opts config.StreamOptions) config.ChangeStream { + return s.subs.subscribe(opts) +} + +func (s *Source) closeSubscriptions() { + s.subs.closeAll() +} + +func (s *Source) markDrop() { + s.dropCP.Store(true) +} + +func (s *Source) PreUpdate(op int, table string, rowid int64, old, new []any) { + if !s.tableAllowed(table) { + return + } + s.pendMu.Lock() + s.pending = append(s.pending, capturedChange{op: op, table: table, rowid: rowid, old: old, new: new}) + s.pendMu.Unlock() +} + +func (s *Source) Commit() { + s.pendMu.Lock() + batch := s.pending + s.pending = nil + s.pendMu.Unlock() + if len(batch) == 0 { + return + } + select { + case s.commits <- batch: + case <-s.runDone: + } +} + +func (s *Source) Rollback() { + s.pendMu.Lock() + s.pending = nil + s.pendMu.Unlock() +} + +func (s *Source) tableAllowed(table string) bool { + lower := strings.ToLower(table) + if lower == offsetsTable { + return false + } + if len(s.tables) == 0 { + return true + } + _, ok := s.tables[lower] + return ok +} + +func (s *Source) Start(ctx context.Context) (<-chan any, error) { + if s.stopped.Load() { + return nil, ErrSourceClosed + } + + dbRes, res, err := s.acquirePool(ctx) + if err != nil { + return nil, err + } + writerDB := dbRes.DB + + conn, err := writerDB.Conn(ctx) + if err != nil { + res.Release() + return nil, fmt.Errorf("acquire writer conn: %w", err) + } + + var file string + if rawErr := conn.Raw(func(dc any) error { + f, e := installHooksOnRaw(dc, s) + file = f + return e + }); rawErr != nil { + _ = conn.Close() + res.Release() + return nil, rawErr + } + registerSink(file, s) + + s.mu.Lock() + s.poolRes = res + s.writerDB = writerDB + s.file = file + s.mu.Unlock() + + readDB, checkpointDB, err := openAuxConns(file) + if err != nil { + s.abortStart(ctx, conn, writerDB, file) + return nil, err + } + + s.mu.Lock() + s.readDB = readDB + s.checkpointDB = checkpointDB + s.mu.Unlock() + + if err := ensureOffsets(ctx, checkpointDB); err != nil { + s.abortStart(ctx, conn, writerDB, file) + return nil, err + } + snapDone, lastSeq, loadErr := loadOffset(ctx, checkpointDB, s.name) + if loadErr != nil { + s.log.Warn("load cdc offset failed; treating as fresh", zap.Error(loadErr)) + } + if lastSeq > s.seq.Load() { + s.seq.Store(lastSeq) + } + + runCtx, cancel := context.WithCancel(ctx) + status := make(chan any, 8) + runDone := make(chan struct{}) + commits := make(chan []capturedChange, commitQueueSize) + + s.mu.Lock() + if s.stopped.Load() { + s.mu.Unlock() + cancel() + s.abortStart(ctx, conn, writerDB, file) + return nil, ErrSourceClosed + } + s.cancel = cancel + s.runDone = runDone + s.commits = commits + s.mu.Unlock() + + doSnapshot := s.snap && !snapDone + if doSnapshot { + if err := s.runSnapshot(runCtx, conn); err != nil { + cancel() + s.abortStart(ctx, conn, writerDB, file) + return nil, fmt.Errorf("snapshot: %w", err) + } + if serr := saveSnapshotDone(runCtx, checkpointDB, s.name); serr != nil { + s.log.Warn("persist snapshot completion failed; restart may re-snapshot", zap.Error(serr)) + } + } + + go s.run(runCtx, status, runDone, metrics.GetCollector(ctx)) + + _ = conn.Close() + + select { + case status <- "sqlite cdc started": + default: + } + s.log.Info("sqlite cdc source started", + zap.String("file", s.file), + zap.Bool("snapshot", doSnapshot)) + return status, nil +} + +func (s *Source) abortStart(ctx context.Context, conn *sql.Conn, writerDB *sql.DB, file string) { + _ = conn.Close() + s.detachHooks(ctx, writerDB, file) + s.releaseResources(ctx) +} + +func (s *Source) acquirePool(ctx context.Context) (sqlservice.DBResource, resource.Resource[any], error) { + res, err := s.res.Acquire(ctx, s.dbResID, resource.ModeNormal) + if err != nil { + return sqlservice.DBResource{}, nil, fmt.Errorf("acquire db resource: %w", err) + } + dbAny, err := res.Get() + if err != nil { + res.Release() + return sqlservice.DBResource{}, nil, fmt.Errorf("get db resource: %w", err) + } + dbRes, ok := dbAny.(sqlservice.DBResource) + if !ok { + res.Release() + return sqlservice.DBResource{}, nil, fmt.Errorf("resource %s is not a database", s.name) + } + if dbRes.Type != sqlconfig.SQLite { + res.Release() + return sqlservice.DBResource{}, nil, fmt.Errorf("resource %s is not a sqlite database (kind %s)", s.name, dbRes.Type) + } + return dbRes, res, nil +} + +func (s *Source) Stop(ctx context.Context) error { + if !s.stopped.CompareAndSwap(false, true) { + return nil + } + defer s.closeSubscriptions() + + s.mu.Lock() + cancel := s.cancel + runDone := s.runDone + writerDB := s.writerDB + file := s.file + s.mu.Unlock() + + if cancel != nil { + cancel() + } + if runDone != nil { + select { + case <-runDone: + case <-ctx.Done(): + return ctx.Err() + } + } + + if writerDB != nil { + s.detachHooks(ctx, writerDB, file) + } + if s.dropCP.Load() { + s.mu.Lock() + cpDB := s.checkpointDB + s.mu.Unlock() + if cpDB != nil { + _ = deleteOffset(ctx, cpDB, s.name) + } + } + s.releaseResources(ctx) + return nil +} + +func (s *Source) detachHooks(ctx context.Context, writerDB *sql.DB, file string) { + unregisterSink(file) + conn, err := writerDB.Conn(ctx) + if err != nil { + return + } + _ = conn.Raw(clearHooksOnRaw) + _ = conn.Close() +} + +func (s *Source) releaseResources(_ context.Context) { + s.mu.Lock() + readDB := s.readDB + cpDB := s.checkpointDB + res := s.poolRes + s.readDB = nil + s.checkpointDB = nil + s.poolRes = nil + s.mu.Unlock() + + if readDB != nil { + _ = readDB.Close() + } + if cpDB != nil { + _ = cpDB.Close() + } + if res != nil { + res.Release() + } +} + +func (s *Source) run(ctx context.Context, status chan any, runDone chan struct{}, mc metrics.Collector) { + defer close(runDone) + defer close(status) + + ticker := time.NewTicker(s.statusInterval) + defer ticker.Stop() + + for { + select { + case batch := <-s.commits: + s.process(ctx, batch, mc) + case <-ticker.C: + s.onTick(ctx, mc) + case <-ctx.Done(): + s.drainRemaining(ctx, mc) + return + } + } +} + +func (s *Source) drainRemaining(ctx context.Context, mc metrics.Collector) { + for { + select { + case batch := <-s.commits: + s.process(ctx, batch, mc) + default: + return + } + } +} + +func (s *Source) process(ctx context.Context, batch []capturedChange, mc metrics.Collector) { + for _, ch := range batch { + cols := s.columnsFor(ctx, ch.table) + op := opString(ch.op) + seq := s.seq.Add(1) + change := config.Change{ + Source: s.name, + Op: op, + Table: ch.table, + Relation: ch.table, + Before: mapRow(cols, ch.old), + After: mapRow(cols, ch.new), + LSN: strconv.FormatUint(seq, 10), + } + s.subs.publish(ctx, change) + if mc != nil { + mc.CounterInc(changesCounter, metrics.Labels{"source": s.name, "op": op}) + } + } +} + +func (s *Source) onTick(ctx context.Context, mc metrics.Collector) { + if mc != nil { + if info, err := os.Stat(s.file + "-wal"); err == nil { + mc.GaugeSet(walGauge, float64(info.Size()), metrics.Labels{"source": s.name}) + } + } + if seq := s.seq.Load(); seq > 0 { + if err := saveOffset(ctx, s.checkpointDB, s.name, seq); err != nil { + s.log.Warn("persist cdc offset failed", zap.Error(err)) + } + } +} + +func (s *Source) columnsFor(ctx context.Context, table string) []columnInfo { + s.colMu.RLock() + cols, ok := s.cols[table] + s.colMu.RUnlock() + if ok { + return cols + } + + cols, err := resolveColumns(ctx, s.readDB, table) + if err != nil { + s.log.Warn("resolve columns failed; emitting positional column names", + zap.String("table", table), zap.Error(err)) + return nil + } + s.colMu.Lock() + s.cols[table] = cols + s.colMu.Unlock() + return cols +} + +func opString(op int) string { + switch op { + case cdcInsert: + return "insert" + case cdcUpdate: + return "update" + case cdcDelete: + return "delete" + default: + return "unknown" + } +} + +func openAuxConns(file string) (read, checkpoint *sql.DB, err error) { + read, err = sql.Open("sqlite3", "file:"+file+"?mode=rwc&_busy_timeout="+strconv.Itoa(auxBusyTimeoutMillisec)+"&_query_only=ON") + if err != nil { + return nil, nil, fmt.Errorf("open read connection: %w", err) + } + read.SetMaxOpenConns(1) + read.SetMaxIdleConns(1) + + checkpoint, err = sql.Open("sqlite3", "file:"+file+"?mode=rwc&_busy_timeout="+strconv.Itoa(auxBusyTimeoutMillisec)) + if err != nil { + _ = read.Close() + return nil, nil, fmt.Errorf("open checkpoint connection: %w", err) + } + checkpoint.SetMaxOpenConns(1) + checkpoint.SetMaxIdleConns(1) + return read, checkpoint, nil +} diff --git a/service/cdc/sqlite/source_stub.go b/service/cdc/sqlite/source_stub.go new file mode 100644 index 000000000..8feb9cd2a --- /dev/null +++ b/service/cdc/sqlite/source_stub.go @@ -0,0 +1,9 @@ +// SPDX-License-Identifier: MPL-2.0 + +//go:build !sqlite_preupdate_hook + +package sqlite + +func buildSource(_ sourceOptions) (sourceHandle, error) { + return nil, ErrPreupdateTagRequired +} diff --git a/service/cdc/sqlite/source_stub_test.go b/service/cdc/sqlite/source_stub_test.go new file mode 100644 index 000000000..086d7859c --- /dev/null +++ b/service/cdc/sqlite/source_stub_test.go @@ -0,0 +1,17 @@ +// SPDX-License-Identifier: MPL-2.0 + +//go:build !sqlite_preupdate_hook + +package sqlite + +import ( + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestBuildSourceRequiresTag(t *testing.T) { + src, err := buildSource(sourceOptions{name: "x"}) + assert.Nil(t, src) + assert.ErrorIs(t, err, ErrPreupdateTagRequired) +} diff --git a/service/cdc/sqlite/source_tagged_test.go b/service/cdc/sqlite/source_tagged_test.go new file mode 100644 index 000000000..c98672d36 --- /dev/null +++ b/service/cdc/sqlite/source_tagged_test.go @@ -0,0 +1,27 @@ +// SPDX-License-Identifier: MPL-2.0 + +//go:build sqlite_preupdate_hook + +package sqlite + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/wippyai/runtime/api/registry" +) + +func TestBuildSourceWithTag(t *testing.T) { + h, err := buildSource(sourceOptions{name: "x", dbResource: registry.NewID("app", "db")}) + require.NoError(t, err) + require.NotNil(t, h) + _, ok := h.(*Source) + assert.True(t, ok) +} + +func TestBuildSourceRejectsBadInterval(t *testing.T) { + _, err := buildSource(sourceOptions{name: "x", statusInterval: "nope"}) + assert.Error(t, err) +} diff --git a/service/cdc/sqlite/subscribers.go b/service/cdc/sqlite/subscribers.go new file mode 100644 index 000000000..05c451505 --- /dev/null +++ b/service/cdc/sqlite/subscribers.go @@ -0,0 +1,182 @@ +// SPDX-License-Identifier: MPL-2.0 + +package sqlite + +import ( + "context" + "strings" + "sync" + "sync/atomic" + + config "github.com/wippyai/runtime/api/service/cdc" +) + +const ( + defaultStreamBuffer = 128 + maxStreamBuffer = 65536 +) + +type subscribers struct { + m map[uint64]*subscription + mu sync.RWMutex + next uint64 +} + +func newSubscribers() *subscribers { + return &subscribers{m: make(map[uint64]*subscription)} +} + +func (s *subscribers) subscribe(opts config.StreamOptions) config.ChangeStream { + buffer := opts.Buffer + if buffer <= 0 { + buffer = defaultStreamBuffer + } + if buffer > maxStreamBuffer { + buffer = maxStreamBuffer + } + + s.mu.Lock() + s.next++ + sub := &subscription{ + parent: s, + id: s.next, + in: make(chan config.Change, buffer), + out: make(chan config.Change, buffer), + done: make(chan struct{}), + tables: filterSet(opts.Tables), + ops: filterSet(opts.Ops), + } + s.m[sub.id] = sub + s.mu.Unlock() + + go sub.run() + return sub +} + +func (s *subscribers) publish(ctx context.Context, change config.Change) { + s.mu.RLock() + matched := make([]*subscription, 0, len(s.m)) + for _, sub := range s.m { + if sub.matches(change) { + matched = append(matched, sub) + } + } + s.mu.RUnlock() + + for _, sub := range matched { + sub.send(ctx, change) + } +} + +func (s *subscribers) remove(id uint64) { + s.mu.Lock() + delete(s.m, id) + s.mu.Unlock() +} + +func (s *subscribers) closeAll() { + s.mu.Lock() + subs := make([]*subscription, 0, len(s.m)) + for id, sub := range s.m { + subs = append(subs, sub) + delete(s.m, id) + } + s.mu.Unlock() + + for _, sub := range subs { + sub.Close() + } +} + +type subscription struct { + parent *subscribers + in chan config.Change + out chan config.Change + done chan struct{} + tables map[string]struct{} + ops map[string]struct{} + id uint64 + once sync.Once + closed atomic.Bool +} + +func (s *subscription) Changes() <-chan config.Change { + return s.out +} + +func (s *subscription) Close() { + s.once.Do(func() { + s.closed.Store(true) + if s.parent != nil { + s.parent.remove(s.id) + } + close(s.done) + }) +} + +func (s *subscription) run() { + defer close(s.out) + for { + select { + case <-s.done: + return + default: + } + select { + case change := <-s.in: + select { + case <-s.done: + return + case s.out <- change: + } + case <-s.done: + return + } + } +} + +func (s *subscription) send(_ context.Context, change config.Change) { + if s.closed.Load() { + return + } + select { + case s.in <- change: + default: + s.Close() + } +} + +func (s *subscription) matches(change config.Change) bool { + if len(s.ops) > 0 { + if _, ok := s.ops[strings.ToLower(change.Op)]; !ok { + return false + } + } + if len(s.tables) > 0 { + if _, ok := s.tables[strings.ToLower(change.Relation)]; ok { + return true + } + if _, ok := s.tables[strings.ToLower(change.Table)]; ok { + return true + } + return false + } + return true +} + +func filterSet(values []string) map[string]struct{} { + if len(values) == 0 { + return nil + } + out := make(map[string]struct{}, len(values)) + for _, v := range values { + v = strings.ToLower(strings.TrimSpace(v)) + if v != "" { + out[v] = struct{}{} + } + } + if len(out) == 0 { + return nil + } + return out +} diff --git a/service/cdc/sqlite/subscribers_test.go b/service/cdc/sqlite/subscribers_test.go new file mode 100644 index 000000000..20e192a91 --- /dev/null +++ b/service/cdc/sqlite/subscribers_test.go @@ -0,0 +1,131 @@ +// SPDX-License-Identifier: MPL-2.0 + +package sqlite + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + config "github.com/wippyai/runtime/api/service/cdc" +) + +func TestFilterSet(t *testing.T) { + assert.Nil(t, filterSet(nil)) + assert.Nil(t, filterSet([]string{" ", ""})) + + got := filterSet([]string{"Users", " orders ", "users"}) + assert.Contains(t, got, "users") + assert.Contains(t, got, "orders") + assert.Len(t, got, 2) +} + +func TestSubscriptionMatches(t *testing.T) { + all := &subscription{} + assert.True(t, all.matches(config.Change{Op: "insert", Table: "users"})) + + byOp := &subscription{ops: map[string]struct{}{"insert": {}}} + assert.True(t, byOp.matches(config.Change{Op: "insert"})) + assert.False(t, byOp.matches(config.Change{Op: "delete"})) + + byTable := &subscription{tables: map[string]struct{}{"users": {}}} + assert.True(t, byTable.matches(config.Change{Op: "insert", Table: "users"})) + assert.True(t, byTable.matches(config.Change{Op: "insert", Relation: "users"})) + assert.False(t, byTable.matches(config.Change{Op: "insert", Table: "orders"})) +} + +func TestSubscribersPublishAndClose(t *testing.T) { + subs := newSubscribers() + stream := subs.subscribe(config.StreamOptions{}) + + subs.publish(context.Background(), config.Change{Op: "insert", Table: "users", Source: "s"}) + + select { + case change := <-stream.Changes(): + assert.Equal(t, "insert", change.Op) + assert.Equal(t, "users", change.Table) + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for change") + } + + subs.closeAll() + select { + case _, ok := <-stream.Changes(): + assert.False(t, ok, "channel should be closed after closeAll") + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for close") + } +} + +func TestSubscribeBufferClamp(t *testing.T) { + subs := newSubscribers() + + def := subs.subscribe(config.StreamOptions{Buffer: 0}).(*subscription) + assert.Equal(t, defaultStreamBuffer, cap(def.in)) + + neg := subs.subscribe(config.StreamOptions{Buffer: -5}).(*subscription) + assert.Equal(t, defaultStreamBuffer, cap(neg.in)) + + exact := subs.subscribe(config.StreamOptions{Buffer: 7}).(*subscription) + assert.Equal(t, 7, cap(exact.in)) + + huge := subs.subscribe(config.StreamOptions{Buffer: maxStreamBuffer + 100}).(*subscription) + assert.Equal(t, maxStreamBuffer, cap(huge.in)) +} + +func TestSubscribeAssignsUniqueIncreasingIDs(t *testing.T) { + subs := newSubscribers() + a := subs.subscribe(config.StreamOptions{}).(*subscription) + b := subs.subscribe(config.StreamOptions{}).(*subscription) + assert.Equal(t, uint64(1), a.id) + assert.Equal(t, uint64(2), b.id) +} + +func TestPublishNeverBlocksAndClosesLaggard(t *testing.T) { + subs := newSubscribers() + stream := subs.subscribe(config.StreamOptions{Buffer: 1}) + + done := make(chan struct{}) + go func() { + for i := 0; i < 1000; i++ { + subs.publish(context.Background(), config.Change{Op: "insert", Table: "t"}) + } + close(done) + }() + + select { + case <-done: + case <-time.After(3 * time.Second): + t.Fatal("publish blocked on a non-reading subscriber") + } + + for { + select { + case _, ok := <-stream.Changes(): + if !ok { + return + } + case <-time.After(2 * time.Second): + t.Fatal("laggard subscription was not closed on overflow") + } + } +} + +func TestSubscribersFilterByOp(t *testing.T) { + subs := newSubscribers() + stream := subs.subscribe(config.StreamOptions{Ops: []string{"delete"}}) + defer stream.Close() + + subs.publish(context.Background(), config.Change{Op: "insert", Table: "users"}) + subs.publish(context.Background(), config.Change{Op: "delete", Table: "users"}) + + select { + case change := <-stream.Changes(): + require.Equal(t, "delete", change.Op) + case <-time.After(2 * time.Second): + t.Fatal("timed out") + } +} diff --git a/service/sql/conn.go b/service/sql/conn.go index 54039c1fe..372862b3d 100644 --- a/service/sql/conn.go +++ b/service/sql/conn.go @@ -5,12 +5,6 @@ package sql import ( "context" "database/sql" - "errors" - "fmt" - "net/url" - "sort" - "strconv" - "strings" "sync" "sync/atomic" @@ -172,75 +166,84 @@ func (p *ConnPool) Stop(ctx context.Context) error { p.current = nil p.db = nil p.mu.Unlock() + if gen == nil { + return nil + } gen.closeWhenIdle() return gen.waitClosed(ctx) } } -// UpdateConfig updates the pool configuration +// UpdateConfig updates the pool configuration. It delegates engine-specific +// validation and tuning to the engine registered for the pool's kind. func (p *ConnPool) UpdateConfig(cfg any) error { if p.closed.Load() { return ErrPoolClosed } - switch c := cfg.(type) { - case *config.DBConfig: - if p.kind == config.SQLite { - return NewInvalidConfigTypeError("DBConfig", config.SQLite) - } - - if err := c.Validate(); err != nil { - return NewInvalidConfigError(err) - } - - newDB, err := openStandardDB(p.kind, c) - if err != nil { - return err - } + ec, ok := cfg.(config.EngineConfig) + if !ok { + return NewUnsupportedConfigTypeError(p.kind) + } - newGen := newDBGeneration(newDB) - p.mu.Lock() - if p.closed.Load() { - p.mu.Unlock() - _ = newDB.Close() - return ErrPoolClosed - } - oldGen := p.current - if oldGen == nil && p.db != nil { - oldGen = newDBGeneration(p.db) - } - p.current = newGen - p.db = newDB - p.mu.Unlock() + eng, ok := engineFor(p.kind) + if !ok { + return NewUnsupportedConfigTypeError(p.kind) + } - if oldGen != nil { - oldGen.closeWhenIdle() - } + return p.updateConfig(context.Background(), eng, ec) +} - var cfg any = c - p.config.Store(&cfg) +func (p *ConnPool) updateConfig(ctx context.Context, eng Engine, ec config.EngineConfig) error { + if p.closed.Load() { + return ErrPoolClosed + } - case *config.SQLiteConfig: - if p.kind != config.SQLite { - return NewInvalidConfigTypeError("SQLiteConfig", p.kind) - } + if err := eng.ValidateConfigType(ec); err != nil { + return err + } - if err := c.Validate(); err != nil { - return NewInvalidConfigError(err) - } + if err := ec.Validate(); err != nil { + return NewInvalidConfigError(err) + } + if p.kind == config.SQLite { gen := p.currentGeneration() if gen == nil { return ErrPoolClosed } - gen.db.SetConnMaxLifetime(c.Pool.MaxLifetime) + eng.Tune(gen.db, ec) + var stored any = ec + p.config.Store(&stored) + return nil + } - var cfg any = c - p.config.Store(&cfg) + newDB, err := openEngineDB(ctx, eng, ec) + if err != nil { + return err + } + newGen := newDBGeneration(newDB) - default: - return NewUnsupportedConfigTypeError(p.kind) + p.mu.Lock() + if p.closed.Load() { + p.mu.Unlock() + _ = newDB.Close() + return ErrPoolClosed + } + oldGen := p.current + if oldGen == nil && p.db != nil { + oldGen = newDBGeneration(p.db) } + p.current = newGen + p.db = newDB + p.mu.Unlock() + + if oldGen != nil { + oldGen.closeWhenIdle() + } + + var stored any = ec + p.config.Store(&stored) return nil } @@ -280,160 +283,6 @@ func (p *ConnPool) Acquire( } } -func openStandardDB(kind registry.Kind, cfg *config.DBConfig) (*sql.DB, error) { - dsn, err := buildDSN(kind, cfg) - if err != nil { - return nil, NewInvalidDSNError(err) - } - - db, err := sql.Open(getDriver(kind), dsn) - if err != nil { - return nil, NewConnectionPoolCreationError(err) - } - - db.SetMaxOpenConns(cfg.Pool.MaxOpen) - db.SetMaxIdleConns(cfg.Pool.MaxIdle) - db.SetConnMaxLifetime(cfg.Pool.MaxLifetime) - return db, nil -} - -// Helper to build DSN string for different database types -func buildDSN(kind registry.Kind, cfg *config.DBConfig) (string, error) { - switch kind { - case config.Postgres: - if err := validateDSNFields(cfg); err != nil { - return "", err - } - opts := buildPostgresOptionsString(cfg.Options) - var b strings.Builder - b.Grow(128) - b.WriteString("host=") - b.WriteString(quotePostgresValue(cfg.Host)) - b.WriteString(" port=") - b.WriteString(strconv.Itoa(cfg.Port)) - b.WriteString(" user=") - b.WriteString(quotePostgresValue(cfg.Username)) - b.WriteString(" password=") - b.WriteString(quotePostgresValue(cfg.Password)) - b.WriteString(" dbname=") - b.WriteString(quotePostgresValue(cfg.Database)) - if opts != "" { - b.WriteString(" ") - b.WriteString(opts) - } - return b.String(), nil - - case config.MySQL: - if err := validateDSNFields(cfg); err != nil { - return "", err - } - opts := buildMySQLOptionsString(cfg.Options) - var b strings.Builder - b.Grow(128) - b.WriteString(cfg.Username) - b.WriteString(":") - b.WriteString(cfg.Password) - b.WriteString("@tcp(") - b.WriteString(cfg.Host) - b.WriteString(":") - b.WriteString(strconv.Itoa(cfg.Port)) - b.WriteString(")/") - b.WriteString(cfg.Database) - if opts != "" { - b.WriteString("?") - b.WriteString(opts) - } - return b.String(), nil - - default: - return "", NewUnsupportedDatabaseTypeError(kind) - } -} - -func validateDSNFields(cfg *config.DBConfig) error { - switch { - case cfg.Host == "": - return NewInvalidDSNError(errors.New("host is empty")) - case cfg.Port <= 0: - return NewInvalidDSNError(fmt.Errorf("port is invalid: %d", cfg.Port)) - case cfg.Username == "": - return NewInvalidDSNError(errors.New("username is empty")) - case cfg.Database == "": - return NewInvalidDSNError(errors.New("database is empty")) - } - return nil -} - -func quotePostgresValue(value string) string { - var b strings.Builder - b.Grow(len(value) + 2) - b.WriteByte('\'') - for i := 0; i < len(value); i++ { - c := value[i] - if c == '\\' || c == '\'' { - b.WriteByte('\\') - } - b.WriteByte(c) - } - b.WriteByte('\'') - return b.String() -} - -func getDriver(kind registry.Kind) string { - switch kind { - case config.Postgres: - return "postgres" - case config.MySQL: - return "mysql" - default: - return kind - } -} - -// buildPostgresOptionsString renders lib/pq keyword/value options. -func buildPostgresOptionsString(options map[string]string) string { - if len(options) == 0 { - return "" - } - - keys := make([]string, 0, len(options)) - for k := range options { - keys = append(keys, k) - } - sort.Strings(keys) - - var b strings.Builder - b.Grow(len(options) * 20) - for i, k := range keys { - if i > 0 { - b.WriteString(" ") - } - b.WriteString(k) - b.WriteString("=") - b.WriteString(quotePostgresValue(options[k])) - } - - return b.String() -} - -func buildMySQLOptionsString(options map[string]string) string { - if len(options) == 0 { - return "" - } - - values := url.Values{} - for k, v := range options { - values.Set(k, v) - } - return values.Encode() -} - -// Helper kept for older internal tests and benchmarks. PostgreSQL was the only -// historical caller that used this space-separated keyword/value form. -func buildOptionsString(options map[string]string) string { - return buildPostgresOptionsString(options) -} - // DBConn represents a database connection resource type DBConn struct { pool *ConnPool diff --git a/service/sql/conn_test.go b/service/sql/conn_test.go index 0da2e344f..d41059e05 100644 --- a/service/sql/conn_test.go +++ b/service/sql/conn_test.go @@ -363,133 +363,6 @@ func TestConnPool_UpdateConfigSwapsStandardDBAndRetiresOldAfterRelease(t *testin require.NoError(t, pool.Stop(ctx)) } -func TestBuildDSN(t *testing.T) { - tests := []struct { - cfg *apiconfig.DBConfig - name string - kind string - wantErr bool - }{ - { - name: "postgres", - kind: apiconfig.Postgres, - cfg: &apiconfig.DBConfig{ - Host: "localhost", Port: 5432, Database: "db", - Username: "user", Password: "pass", - }, - wantErr: false, - }, - { - name: "mysql", - kind: apiconfig.MySQL, - cfg: &apiconfig.DBConfig{ - Host: "localhost", Port: 3306, Database: "db", - Username: "user", Password: "pass", - }, - wantErr: false, - }, - { - name: "unsupported", - kind: "db.unknown", - cfg: &apiconfig.DBConfig{}, - wantErr: true, - }, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - _, err := buildDSN(tt.kind, tt.cfg) - if tt.wantErr { - assert.Error(t, err) - } else { - assert.NoError(t, err) - } - }) - } -} - -func TestBuildOptionsString(t *testing.T) { - t.Run("empty options", func(t *testing.T) { - result := buildOptionsString(nil) - assert.Empty(t, result) - }) - - t.Run("single option", func(t *testing.T) { - result := buildOptionsString(map[string]string{"sslmode": "disable"}) - assert.Equal(t, "sslmode='disable'", result) - }) - - t.Run("postgres options are stable and space separated", func(t *testing.T) { - result := buildPostgresOptionsString(map[string]string{ - "sslmode": "disable", - "connect_timeout": "10", - "application_name": "test", - }) - assert.Equal(t, "application_name='test' connect_timeout='10' sslmode='disable'", result) - }) - - t.Run("mysql options are stable query parameters", func(t *testing.T) { - result := buildMySQLOptionsString(map[string]string{ - "charset": "utf8mb4", - "parseTime": "true", - "timeout": "2s", - }) - assert.Equal(t, "charset=utf8mb4&parseTime=true&timeout=2s", result) - }) -} - -func TestQuotePostgresValue(t *testing.T) { - assert.Equal(t, "'alice'", quotePostgresValue("alice")) - assert.Equal(t, "''", quotePostgresValue("")) - assert.Equal(t, "'se cret'", quotePostgresValue("se cret")) - assert.Equal(t, `'O\'Brien'`, quotePostgresValue("O'Brien")) - assert.Equal(t, `'a\\b'`, quotePostgresValue(`a\b`)) -} - -func TestValidateDSNFields(t *testing.T) { - base := func() *apiconfig.DBConfig { - return &apiconfig.DBConfig{Host: "h", Port: 5432, Username: "u", Database: "d"} - } - - require.NoError(t, validateDSNFields(base())) - - c := base() - c.Host = "" - assert.ErrorContains(t, validateDSNFields(c), "host is empty") - - c = base() - c.Port = 0 - assert.ErrorContains(t, validateDSNFields(c), "port is invalid") - - c = base() - c.Username = "" - assert.ErrorContains(t, validateDSNFields(c), "username is empty") - - c = base() - c.Database = "" - assert.ErrorContains(t, validateDSNFields(c), "database is empty") -} - -func TestBuildDSN_EmptyUsernameDoesNotAbsorbNextToken(t *testing.T) { - cfg := &apiconfig.DBConfig{ - Host: "h", - Port: 5432, - Database: "d", - Username: "", - Password: "secret", - } - - _, err := buildDSN(apiconfig.Postgres, cfg) - require.Error(t, err) - assert.Contains(t, err.Error(), "username is empty") -} - -func TestGetDriver(t *testing.T) { - assert.Equal(t, "postgres", getDriver(apiconfig.Postgres)) - assert.Equal(t, "mysql", getDriver(apiconfig.MySQL)) - assert.Equal(t, "unknown", getDriver("unknown")) -} - // Benchmarks func newBenchPool(b *testing.B) *ConnPool { @@ -559,31 +432,3 @@ func BenchmarkConnPool_ConcurrentAcquire(b *testing.B) { } }) } - -func BenchmarkBuildDSN_Postgres(b *testing.B) { - cfg := &apiconfig.DBConfig{ - Host: "localhost", Port: 5432, Database: "db", - Username: "user", Password: "pass", - Options: map[string]string{"sslmode": "disable"}, - } - - b.ResetTimer() - b.ReportAllocs() - for i := 0; i < b.N; i++ { - _, _ = buildDSN(apiconfig.Postgres, cfg) - } -} - -func BenchmarkBuildOptionsString(b *testing.B) { - opts := map[string]string{ - "sslmode": "disable", - "connect_timeout": "10", - "application_name": "test", - } - - b.ResetTimer() - b.ReportAllocs() - for i := 0; i < b.N; i++ { - _ = buildOptionsString(opts) - } -} diff --git a/service/sql/driver.go b/service/sql/driver.go new file mode 100644 index 000000000..7cafa44f9 --- /dev/null +++ b/service/sql/driver.go @@ -0,0 +1,38 @@ +// SPDX-License-Identifier: MPL-2.0 + +package sql + +import ( + "sync" + + "github.com/wippyai/runtime/api/registry" +) + +// driverOverrides lets an out-of-tree extension swap the database/sql driver an +// engine opens, keyed by registry kind. It is the only seam through which engine +// connection behavior (for example, a SQLite build that installs preupdate hooks) +// is altered without modifying the core package. +var ( + driverMu sync.RWMutex + driverOverrides = make(map[registry.Kind]string) +) + +// RegisterDriver overrides the database/sql driver name used for the given kind. +// Extensions call this from an init function, before any pool is created. +func RegisterDriver(kind registry.Kind, name string) { + driverMu.Lock() + driverOverrides[kind] = name + driverMu.Unlock() +} + +// driverName returns the registered override for kind, falling back to def. +func driverName(kind registry.Kind, def string) string { + driverMu.RLock() + name, ok := driverOverrides[kind] + driverMu.RUnlock() + if ok { + return name + } + + return def +} diff --git a/service/sql/driver_test.go b/service/sql/driver_test.go new file mode 100644 index 000000000..9ae3cab82 --- /dev/null +++ b/service/sql/driver_test.go @@ -0,0 +1,95 @@ +// SPDX-License-Identifier: MPL-2.0 + +package sql + +import ( + "context" + "database/sql" + "errors" + "testing" + "time" + + _ "github.com/mattn/go-sqlite3" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/wippyai/runtime/api/payload" + "github.com/wippyai/runtime/api/registry" + config "github.com/wippyai/runtime/api/service/sql" + "go.uber.org/zap" +) + +// fixedConfigEngine returns a preset SQLite config regardless of the entry, so driver +// tests can drive createPool to sql.Open without a transcoder. Prepare can be forced +// to fail to exercise the create lifecycle's close-on-error branch. +type fixedConfigEngine struct { + prepareErr error + kind registry.Kind + driver string +} + +func (e fixedConfigEngine) Kind() registry.Kind { return e.kind } + +func (e fixedConfigEngine) DriverName() string { return e.driver } + +func (fixedConfigEngine) DecodeConfig(context.Context, payload.Transcoder, registry.Entry) (config.EngineConfig, error) { + return &config.SQLiteConfig{File: ":memory:", Pool: config.PoolConfig{MaxLifetime: time.Hour}}, nil +} + +func (fixedConfigEngine) ResolveEnv(context.Context, EngineDeps, config.EngineConfig) error { + return nil +} + +func (fixedConfigEngine) BuildDSN(config.EngineConfig) (string, error) { + return ":memory:", nil +} + +func (e fixedConfigEngine) Prepare(context.Context, *sql.DB, config.EngineConfig) error { + return e.prepareErr +} + +func (fixedConfigEngine) Tune(*sql.DB, config.EngineConfig) {} + +func (fixedConfigEngine) ValidateConfigType(config.EngineConfig) error { + return nil +} + +func TestDriverNameFallbackAndOverride(t *testing.T) { + const kind = registry.Kind("db.sql.drivernametest") + + assert.Equal(t, "default-drv", driverName(kind, "default-drv")) + + RegisterDriver(kind, "override-drv") + assert.Equal(t, "override-drv", driverName(kind, "default-drv")) +} + +func TestCreatePoolAppliesDriverOverride(t *testing.T) { + const kind = registry.Kind("db.sql.overridetest") + RegisterEngine(fixedConfigEngine{kind: kind, driver: "sqlite3"}) + + factory := &DefaultPoolFactory{} + deps := EngineDeps{Log: zap.NewNop()} + entry := registry.Entry{ID: registry.NewID("test", "ov"), Kind: kind, Data: payload.New("x")} + + pool, _, err := factory.CreatePool(context.Background(), deps, entry) + require.NoError(t, err) + require.NotNil(t, pool) + require.NoError(t, pool.Stop(context.Background())) + + RegisterDriver(kind, "sentinel-missing-driver") + overridden, _, err := factory.CreatePool(context.Background(), deps, entry) + require.Error(t, err) + assert.Nil(t, overridden) +} + +func TestCreatePoolClosesOnPrepareError(t *testing.T) { + const kind = registry.Kind("db.sql.preparefailtest") + prepErr := errors.New("prepare boom") + RegisterEngine(fixedConfigEngine{kind: kind, driver: "sqlite3", prepareErr: prepErr}) + + entry := registry.Entry{ID: registry.NewID("test", "pf"), Kind: kind, Data: payload.New("x")} + pool, _, err := (&DefaultPoolFactory{}).CreatePool(context.Background(), EngineDeps{Log: zap.NewNop()}, entry) + + require.Error(t, err) + assert.Nil(t, pool) + assert.ErrorIs(t, err, prepErr) +} diff --git a/service/sql/engine.go b/service/sql/engine.go new file mode 100644 index 000000000..93aac432c --- /dev/null +++ b/service/sql/engine.go @@ -0,0 +1,122 @@ +// SPDX-License-Identifier: MPL-2.0 + +package sql + +import ( + "context" + "database/sql" + + envapi "github.com/wippyai/runtime/api/env" + "github.com/wippyai/runtime/api/payload" + "github.com/wippyai/runtime/api/registry" + config "github.com/wippyai/runtime/api/service/sql" + "go.uber.org/zap" +) + +// EngineDeps carries the shared collaborators an engine needs to turn a registry +// entry into a configured pool. +type EngineDeps struct { + Transcoder payload.Transcoder + Env envapi.Registry + Log *zap.Logger +} + +// Engine is a self-contained SQL dialect. Each engine knows how to decode its +// configuration, resolve environment overrides, open and tune a pool, and run any +// post-open preparation. Engines register themselves with RegisterEngine, so adding +// a database never touches the factory or manager dispatch surface. +type Engine interface { + Kind() registry.Kind + DriverName() string + DecodeConfig(ctx context.Context, dtt payload.Transcoder, entry registry.Entry) (config.EngineConfig, error) + ResolveEnv(ctx context.Context, deps EngineDeps, cfg config.EngineConfig) error + BuildDSN(cfg config.EngineConfig) (string, error) + Prepare(ctx context.Context, db *sql.DB, cfg config.EngineConfig) error + Tune(db *sql.DB, cfg config.EngineConfig) + ValidateConfigType(cfg config.EngineConfig) error +} + +var engines = make(map[registry.Kind]Engine) + +// RegisterEngine adds an engine to the registry under its kind. Intended to be +// called from engine package init functions. +func RegisterEngine(e Engine) { + engines[e.Kind()] = e +} + +// engineFor looks up the engine registered for a kind. +func engineFor(kind registry.Kind) (Engine, bool) { + e, ok := engines[kind] + return e, ok +} + +// createPool runs the generic create lifecycle for a known engine. +func createPool(ctx context.Context, deps EngineDeps, eng Engine, entry registry.Entry) (*ConnPool, config.EngineConfig, error) { + cfg, err := eng.DecodeConfig(ctx, deps.Transcoder, entry) + if err != nil { + return nil, nil, NewInvalidConfigError(err) + } + + if err := eng.ResolveEnv(ctx, deps, cfg); err != nil { + return nil, nil, err + } + + if err := cfg.Validate(); err != nil { + return nil, nil, NewInvalidConfigError(err) + } + + db, err := openEngineDB(ctx, eng, cfg) + if err != nil { + return nil, nil, err + } + + pool := &ConnPool{ + kind: eng.Kind(), + db: db, + current: newDBGeneration(db), + status: make(chan any, 1), + } + + var cfgAny any = cfg + pool.config.Store(&cfgAny) + + return pool, cfg, nil +} + +// updatePool runs the generic update lifecycle for a known engine. +func updatePool(ctx context.Context, deps EngineDeps, eng Engine, pool *ConnPool, entry registry.Entry) (config.EngineConfig, error) { + cfg, err := eng.DecodeConfig(ctx, deps.Transcoder, entry) + if err != nil { + return nil, NewInvalidConfigError(err) + } + + if err := eng.ResolveEnv(ctx, deps, cfg); err != nil { + return nil, err + } + + if err := pool.updateConfig(ctx, eng, cfg); err != nil { + return nil, NewPoolUpdateError(err) + } + + return cfg, nil +} + +func openEngineDB(ctx context.Context, eng Engine, cfg config.EngineConfig) (*sql.DB, error) { + dsn, err := eng.BuildDSN(cfg) + if err != nil { + return nil, NewInvalidDSNError(err) + } + + db, err := sql.Open(driverName(eng.Kind(), eng.DriverName()), dsn) + if err != nil { + return nil, NewConnectionPoolCreationError(err) + } + + if err := eng.Prepare(ctx, db, cfg); err != nil { + _ = db.Close() + return nil, err + } + + eng.Tune(db, cfg) + return db, nil +} diff --git a/service/sql/engine/all/all.go b/service/sql/engine/all/all.go new file mode 100644 index 000000000..d63543976 --- /dev/null +++ b/service/sql/engine/all/all.go @@ -0,0 +1,12 @@ +// SPDX-License-Identifier: MPL-2.0 + +// Package all registers every built-in SQL engine via blank import. A composition +// root (for example the storage boot component) imports this package so the standard +// dialects are available; out-of-tree engines register themselves the same way. +package all + +import ( + // Blank imports register the built-in engines with service/sql via init. + _ "github.com/wippyai/runtime/service/sql/engine/sqlite" + _ "github.com/wippyai/runtime/service/sql/engine/standard" +) diff --git a/service/sql/engine/sqlite/sqlite.go b/service/sql/engine/sqlite/sqlite.go new file mode 100644 index 000000000..20da5a918 --- /dev/null +++ b/service/sql/engine/sqlite/sqlite.go @@ -0,0 +1,89 @@ +// SPDX-License-Identifier: MPL-2.0 + +// Package sqlite implements the file-backed SQLite engine. It registers itself with +// service/sql through the public engine seam; a CDC-enabled build overrides the +// underlying driver via service/sql.RegisterDriver, which core applies transparently. +package sqlite + +import ( + "context" + "database/sql" + "fmt" + + "github.com/wippyai/runtime/api/payload" + "github.com/wippyai/runtime/api/registry" + config "github.com/wippyai/runtime/api/service/sql" + sqlservice "github.com/wippyai/runtime/service/sql" + entryutil "github.com/wippyai/runtime/system/entry" +) + +// defaultDriver is the stock SQLite driver. A build with the preupdate hook overrides +// it via service/sql.RegisterDriver; the engine itself stays override-agnostic. +const defaultDriver = "sqlite3" + +type engine struct{} + +func init() { + sqlservice.RegisterEngine(engine{}) +} + +func (engine) Kind() registry.Kind { + return config.SQLite +} + +func (engine) DriverName() string { + return defaultDriver +} + +func (engine) DecodeConfig(ctx context.Context, dtt payload.Transcoder, entry registry.Entry) (config.EngineConfig, error) { + cfg, err := entryutil.DecodeEntryConfig[config.SQLiteConfig](ctx, dtt, entry) + if err != nil { + return nil, err + } + + return cfg, nil +} + +func (engine) ResolveEnv(context.Context, sqlservice.EngineDeps, config.EngineConfig) error { + return nil +} + +func (engine) BuildDSN(ec config.EngineConfig) (string, error) { + cfg, ok := ec.(*config.SQLiteConfig) + if !ok { + return "", sqlservice.NewInvalidConfigTypeError(fmt.Sprintf("%T", ec), config.SQLite) + } + + if cfg.File == ":memory:" { + return ":memory:", nil + } + + return "file:" + cfg.File + "?mode=rwc", nil +} + +func (engine) Prepare(ctx context.Context, db *sql.DB, _ config.EngineConfig) error { + if _, err := db.ExecContext(ctx, "PRAGMA journal_mode=WAL;"); err != nil { + return sqlservice.NewWALModeError(err) + } + + return nil +} + +func (engine) Tune(db *sql.DB, ec config.EngineConfig) { + cfg, ok := ec.(*config.SQLiteConfig) + if !ok { + return + } + + db.SetMaxOpenConns(1) + db.SetMaxIdleConns(1) + db.SetConnMaxLifetime(cfg.Pool.MaxLifetime) +} + +func (engine) ValidateConfigType(ec config.EngineConfig) error { + if _, ok := ec.(*config.SQLiteConfig); !ok { + return sqlservice.NewInvalidConfigTypeError(fmt.Sprintf("%T", ec), config.SQLite) + } + + return nil +} diff --git a/service/sql/engine/sqlite/sqlite_test.go b/service/sql/engine/sqlite/sqlite_test.go new file mode 100644 index 000000000..a5223cf8d --- /dev/null +++ b/service/sql/engine/sqlite/sqlite_test.go @@ -0,0 +1,78 @@ +// SPDX-License-Identifier: MPL-2.0 + +package sqlite + +import ( + "context" + "database/sql" + "path/filepath" + "testing" + "time" + + _ "github.com/mattn/go-sqlite3" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/wippyai/runtime/api/registry" + config "github.com/wippyai/runtime/api/service/sql" + sqlservice "github.com/wippyai/runtime/service/sql" + "go.uber.org/zap" +) + +func TestKindDriverRegistered(t *testing.T) { + e := engine{} + assert.Equal(t, config.SQLite, e.Kind()) + assert.Equal(t, "sqlite3", e.DriverName()) + + _, _, err := (&sqlservice.DefaultPoolFactory{}).CreatePool( + context.Background(), + sqlservice.EngineDeps{Log: zap.NewNop()}, + registry.Entry{ID: registry.NewID("t", "x"), Kind: config.SQLite, Data: nil}, + ) + require.Error(t, err) + assert.NotContains(t, err.Error(), "unsupported entry kind") +} + +func TestBuildDSN(t *testing.T) { + e := engine{} + + mem, err := e.BuildDSN(&config.SQLiteConfig{File: ":memory:"}) + require.NoError(t, err) + assert.Equal(t, ":memory:", mem) + + file, err := e.BuildDSN(&config.SQLiteConfig{File: "/tmp/app.db"}) + require.NoError(t, err) + assert.Equal(t, "file:/tmp/app.db?mode=rwc", file) + + _, err = e.BuildDSN(&config.DBConfig{}) + require.Error(t, err) + assert.Contains(t, err.Error(), "invalid config type") +} + +func TestPrepareEnablesWAL(t *testing.T) { + file := filepath.Join(t.TempDir(), "app.db") + db, err := sql.Open("sqlite3", "file:"+file+"?mode=rwc") + require.NoError(t, err) + defer func() { _ = db.Close() }() + + require.NoError(t, engine{}.Prepare(context.Background(), db, &config.SQLiteConfig{File: file})) + + var mode string + require.NoError(t, db.QueryRowContext(context.Background(), "PRAGMA journal_mode;").Scan(&mode)) + assert.Equal(t, "wal", mode) +} + +func TestTuneSingleWriter(t *testing.T) { + db, err := sql.Open("sqlite3", ":memory:") + require.NoError(t, err) + defer func() { _ = db.Close() }() + + engine{}.Tune(db, &config.SQLiteConfig{Pool: config.PoolConfig{MaxLifetime: time.Hour}}) + assert.Equal(t, 1, db.Stats().MaxOpenConnections) +} + +func TestValidateConfigType(t *testing.T) { + require.NoError(t, engine{}.ValidateConfigType(&config.SQLiteConfig{File: ":memory:"})) + err := engine{}.ValidateConfigType(&config.DBConfig{}) + require.Error(t, err) + assert.Contains(t, err.Error(), "invalid config type") +} diff --git a/service/sql/engine/standard/standard.go b/service/sql/engine/standard/standard.go new file mode 100644 index 000000000..936b13976 --- /dev/null +++ b/service/sql/engine/standard/standard.go @@ -0,0 +1,207 @@ +// SPDX-License-Identifier: MPL-2.0 + +// Package standard implements the network SQL engines (PostgreSQL and MySQL) that +// share DBConfig. It registers itself with service/sql through the public engine +// seam, so the core package carries no knowledge of these dialects. +package standard + +import ( + "context" + "database/sql" + "errors" + "fmt" + "net/url" + "sort" + "strconv" + "strings" + + "github.com/wippyai/runtime/api/payload" + "github.com/wippyai/runtime/api/registry" + config "github.com/wippyai/runtime/api/service/sql" + sqlservice "github.com/wippyai/runtime/service/sql" + entryutil "github.com/wippyai/runtime/system/entry" +) + +// engine serves a network SQL dialect. Postgres and MySQL share the same DBConfig, +// env resolution, and pool tuning, differing only in driver name and DSN format, so +// each is a separate instance carrying its own DSN builder. +type engine struct { + dsn func(*config.DBConfig) (string, error) + kind registry.Kind + driver string +} + +func init() { + sqlservice.RegisterEngine(engine{kind: config.Postgres, driver: "postgres", dsn: buildPostgresDSN}) + sqlservice.RegisterEngine(engine{kind: config.MySQL, driver: "mysql", dsn: buildMySQLDSN}) +} + +func (e engine) Kind() registry.Kind { + return e.kind +} + +func (e engine) DriverName() string { + return e.driver +} + +func (engine) DecodeConfig(ctx context.Context, dtt payload.Transcoder, entry registry.Entry) (config.EngineConfig, error) { + cfg, err := entryutil.DecodeEntryConfig[config.DBConfig](ctx, dtt, entry) + if err != nil { + return nil, err + } + + return cfg, nil +} + +func (engine) ResolveEnv(context.Context, sqlservice.EngineDeps, config.EngineConfig) error { + return nil +} + +func (e engine) BuildDSN(ec config.EngineConfig) (string, error) { + cfg, ok := ec.(*config.DBConfig) + if !ok { + return "", sqlservice.NewInvalidConfigTypeError(fmt.Sprintf("%T", ec), e.kind) + } + + return e.dsn(cfg) +} + +func (engine) Prepare(context.Context, *sql.DB, config.EngineConfig) error { + return nil +} + +func (engine) Tune(db *sql.DB, ec config.EngineConfig) { + cfg, ok := ec.(*config.DBConfig) + if !ok { + return + } + + db.SetMaxOpenConns(cfg.Pool.MaxOpen) + db.SetMaxIdleConns(cfg.Pool.MaxIdle) + db.SetConnMaxLifetime(cfg.Pool.MaxLifetime) +} + +func (e engine) ValidateConfigType(ec config.EngineConfig) error { + if _, ok := ec.(*config.DBConfig); !ok { + return sqlservice.NewInvalidConfigTypeError(fmt.Sprintf("%T", ec), e.kind) + } + + return nil +} + +func buildPostgresDSN(cfg *config.DBConfig) (string, error) { + if err := validateDSNFields(cfg); err != nil { + return "", err + } + opts := buildPostgresOptionsString(cfg.Options) + var b strings.Builder + b.Grow(128) + b.WriteString("host=") + b.WriteString(quotePostgresValue(cfg.Host)) + b.WriteString(" port=") + b.WriteString(strconv.Itoa(cfg.Port)) + b.WriteString(" user=") + b.WriteString(quotePostgresValue(cfg.Username)) + b.WriteString(" password=") + b.WriteString(quotePostgresValue(cfg.Password)) + b.WriteString(" dbname=") + b.WriteString(quotePostgresValue(cfg.Database)) + if opts != "" { + b.WriteString(" ") + b.WriteString(opts) + } + + return b.String(), nil +} + +func buildMySQLDSN(cfg *config.DBConfig) (string, error) { + if err := validateDSNFields(cfg); err != nil { + return "", err + } + opts := buildMySQLOptionsString(cfg.Options) + var b strings.Builder + b.Grow(128) + b.WriteString(cfg.Username) + b.WriteString(":") + b.WriteString(cfg.Password) + b.WriteString("@tcp(") + b.WriteString(cfg.Host) + b.WriteString(":") + b.WriteString(strconv.Itoa(cfg.Port)) + b.WriteString(")/") + b.WriteString(cfg.Database) + if opts != "" { + b.WriteString("?") + b.WriteString(opts) + } + + return b.String(), nil +} + +// buildPostgresOptionsString renders lib/pq keyword/value options. +func buildPostgresOptionsString(options map[string]string) string { + if len(options) == 0 { + return "" + } + + keys := make([]string, 0, len(options)) + for k := range options { + keys = append(keys, k) + } + sort.Strings(keys) + + var b strings.Builder + b.Grow(len(options) * 20) + for i, k := range keys { + if i > 0 { + b.WriteString(" ") + } + b.WriteString(k) + b.WriteString("=") + b.WriteString(quotePostgresValue(options[k])) + } + + return b.String() +} + +func buildMySQLOptionsString(options map[string]string) string { + if len(options) == 0 { + return "" + } + + values := url.Values{} + for k, v := range options { + values.Set(k, v) + } + + return values.Encode() +} + +func validateDSNFields(cfg *config.DBConfig) error { + switch { + case cfg.Host == "": + return sqlservice.NewInvalidDSNError(errors.New("host is empty")) + case cfg.Port <= 0: + return sqlservice.NewInvalidDSNError(fmt.Errorf("port is invalid: %d", cfg.Port)) + case cfg.Username == "": + return sqlservice.NewInvalidDSNError(errors.New("username is empty")) + case cfg.Database == "": + return sqlservice.NewInvalidDSNError(errors.New("database is empty")) + } + return nil +} + +func quotePostgresValue(value string) string { + var b strings.Builder + b.Grow(len(value) + 2) + b.WriteByte('\'') + for i := 0; i < len(value); i++ { + c := value[i] + if c == '\\' || c == '\'' { + b.WriteByte('\\') + } + b.WriteByte(c) + } + b.WriteByte('\'') + return b.String() +} diff --git a/service/sql/engine/standard/standard_test.go b/service/sql/engine/standard/standard_test.go new file mode 100644 index 000000000..8ce3946b7 --- /dev/null +++ b/service/sql/engine/standard/standard_test.go @@ -0,0 +1,148 @@ +// SPDX-License-Identifier: MPL-2.0 + +package standard + +import ( + "context" + "database/sql" + "testing" + "time" + + _ "github.com/mattn/go-sqlite3" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/wippyai/runtime/api/registry" + config "github.com/wippyai/runtime/api/service/sql" + sqlservice "github.com/wippyai/runtime/service/sql" + "go.uber.org/zap" +) + +func sqlOpenMemory(*testing.T) (*sql.DB, error) { + return sql.Open("sqlite3", ":memory:") +} + +func TestRegistered(t *testing.T) { + for _, k := range []registry.Kind{config.Postgres, config.MySQL} { + _, _, err := (&sqlservice.DefaultPoolFactory{}).CreatePool( + context.Background(), + sqlservice.EngineDeps{Log: zap.NewNop()}, + registry.Entry{ID: registry.NewID("t", "x"), Kind: k, Data: nil}, + ) + require.Error(t, err) + assert.NotContains(t, err.Error(), "unsupported entry kind", "engine %s must be registered", k) + } +} + +func TestBuildDSN(t *testing.T) { + tests := []struct { + name string + kind registry.Kind + cfg *config.DBConfig + expected string + }{ + { + name: "postgres", + kind: config.Postgres, + cfg: &config.DBConfig{Host: "localhost", Port: 5432, Database: "testdb", Username: "user", Password: "pass", Options: map[string]string{"sslmode": "disable"}}, + expected: "host='localhost' port=5432 user='user' password='pass' dbname='testdb' sslmode='disable'", + }, + { + name: "postgres with connect timeout", + kind: config.Postgres, + cfg: &config.DBConfig{Host: "localhost", Port: 5432, Database: "testdb", Username: "user", Password: "pass", Options: map[string]string{"connect_timeout": "2", "sslmode": "disable"}}, + expected: "host='localhost' port=5432 user='user' password='pass' dbname='testdb' connect_timeout='2' sslmode='disable'", + }, + { + name: "mysql", + kind: config.MySQL, + cfg: &config.DBConfig{Host: "localhost", Port: 3306, Database: "testdb", Username: "user", Password: "pass", Options: map[string]string{"charset": "utf8mb4"}}, + expected: "user:pass@tcp(localhost:3306)/testdb?charset=utf8mb4", + }, + { + name: "mysql with query options", + kind: config.MySQL, + cfg: &config.DBConfig{Host: "localhost", Port: 3306, Database: "testdb", Username: "user", Password: "pass", Options: map[string]string{"charset": "utf8mb4", "parseTime": "true", "timeout": "2s"}}, + expected: "user:pass@tcp(localhost:3306)/testdb?charset=utf8mb4&parseTime=true&timeout=2s", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + e := engine{kind: tt.kind} + if tt.kind == config.Postgres { + e.dsn = buildPostgresDSN + } else { + e.dsn = buildMySQLDSN + } + dsn, err := e.BuildDSN(tt.cfg) + require.NoError(t, err) + assert.Equal(t, tt.expected, dsn) + }) + } +} + +func TestBuildDSN_WrongType(t *testing.T) { + e := engine{kind: config.Postgres, dsn: buildPostgresDSN} + _, err := e.BuildDSN(&config.SQLiteConfig{File: ":memory:"}) + require.Error(t, err) + assert.Contains(t, err.Error(), "invalid config type") +} + +func TestDriverNameAndKind(t *testing.T) { + e := engine{kind: config.Postgres, driver: "postgres"} + assert.Equal(t, config.Postgres, e.Kind()) + assert.Equal(t, "postgres", e.DriverName()) +} + +func TestOptionsStrings(t *testing.T) { + assert.Empty(t, buildPostgresOptionsString(nil)) + assert.Equal(t, "application_name='test' connect_timeout='10' sslmode='disable'", + buildPostgresOptionsString(map[string]string{"sslmode": "disable", "connect_timeout": "10", "application_name": "test"})) + assert.Empty(t, buildMySQLOptionsString(nil)) + assert.Equal(t, "charset=utf8mb4&parseTime=true&timeout=2s", + buildMySQLOptionsString(map[string]string{"charset": "utf8mb4", "parseTime": "true", "timeout": "2s"})) +} + +func TestTuneAndValidateConfigType(t *testing.T) { + e := engine{kind: config.Postgres} + require.NoError(t, e.ValidateConfigType(&config.DBConfig{})) + require.Error(t, e.ValidateConfigType(&config.SQLiteConfig{File: ":memory:"})) + + db, err := sqlOpenMemory(t) + require.NoError(t, err) + defer func() { _ = db.Close() }() + e.Tune(db, &config.DBConfig{Pool: config.PoolConfig{MaxOpen: 7, MaxIdle: 3, MaxLifetime: time.Hour}}) + assert.Equal(t, 7, db.Stats().MaxOpenConnections) +} + +func TestQuotePostgresValue(t *testing.T) { + assert.Equal(t, "'alice'", quotePostgresValue("alice")) + assert.Equal(t, "''", quotePostgresValue("")) + assert.Equal(t, "'se cret'", quotePostgresValue("se cret")) + assert.Equal(t, `'O\'Brien'`, quotePostgresValue("O'Brien")) + assert.Equal(t, `'a\\b'`, quotePostgresValue(`a\b`)) +} + +func TestValidateDSNFields(t *testing.T) { + base := func() *config.DBConfig { + return &config.DBConfig{Host: "h", Port: 5432, Username: "u", Database: "d"} + } + + require.NoError(t, validateDSNFields(base())) + + c := base() + c.Host = "" + assert.ErrorContains(t, validateDSNFields(c), "host is empty") + + c = base() + c.Port = 0 + assert.ErrorContains(t, validateDSNFields(c), "port is invalid") + + c = base() + c.Username = "" + assert.ErrorContains(t, validateDSNFields(c), "username is empty") + + c = base() + c.Database = "" + assert.ErrorContains(t, validateDSNFields(c), "database is empty") +} diff --git a/service/sql/engines_stub_test.go b/service/sql/engines_stub_test.go new file mode 100644 index 000000000..73bb12451 --- /dev/null +++ b/service/sql/engines_stub_test.go @@ -0,0 +1,101 @@ +// SPDX-License-Identifier: MPL-2.0 + +package sql + +import ( + "context" + "database/sql" + "fmt" + + "github.com/wippyai/runtime/api/payload" + "github.com/wippyai/runtime/api/registry" + config "github.com/wippyai/runtime/api/service/sql" + entryutil "github.com/wippyai/runtime/system/entry" +) + +// stubEngine is a faithful-but-minimal engine registered only for the core dispatch +// tests. The real engines live in service/sql/engine/* and cannot be imported here +// (that would be an import cycle), so these stubs exercise the registry, factory, +// manager, and ConnPool plumbing. Real DSN/env/WAL behavior is tested in the engine +// sub-packages. +type stubEngine struct { + kind registry.Kind + driver string + isSQLite bool +} + +func init() { + RegisterEngine(stubEngine{kind: config.Postgres, driver: "postgres"}) + RegisterEngine(stubEngine{kind: config.MySQL, driver: "mysql"}) + RegisterEngine(stubEngine{kind: config.SQLite, driver: "sqlite3", isSQLite: true}) +} + +func (e stubEngine) Kind() registry.Kind { + return e.kind +} + +func (e stubEngine) DriverName() string { + return e.driver +} + +func (e stubEngine) DecodeConfig(ctx context.Context, dtt payload.Transcoder, entry registry.Entry) (config.EngineConfig, error) { + if e.isSQLite { + cfg, err := entryutil.DecodeEntryConfig[config.SQLiteConfig](ctx, dtt, entry) + if err != nil { + return nil, err + } + return cfg, nil + } + + cfg, err := entryutil.DecodeEntryConfig[config.DBConfig](ctx, dtt, entry) + if err != nil { + return nil, err + } + return cfg, nil +} + +func (stubEngine) ResolveEnv(context.Context, EngineDeps, config.EngineConfig) error { + return nil +} + +func (e stubEngine) BuildDSN(config.EngineConfig) (string, error) { + if e.isSQLite { + return ":memory:", nil + } + return "host=stub", nil +} + +func (stubEngine) Prepare(context.Context, *sql.DB, config.EngineConfig) error { + return nil +} + +func (e stubEngine) Tune(db *sql.DB, ec config.EngineConfig) { + if e.isSQLite { + db.SetMaxOpenConns(1) + db.SetMaxIdleConns(1) + if cfg, ok := ec.(*config.SQLiteConfig); ok { + db.SetConnMaxLifetime(cfg.Pool.MaxLifetime) + } + return + } + + if cfg, ok := ec.(*config.DBConfig); ok { + db.SetMaxOpenConns(cfg.Pool.MaxOpen) + db.SetMaxIdleConns(cfg.Pool.MaxIdle) + db.SetConnMaxLifetime(cfg.Pool.MaxLifetime) + } +} + +func (e stubEngine) ValidateConfigType(ec config.EngineConfig) error { + if e.isSQLite { + if _, ok := ec.(*config.SQLiteConfig); !ok { + return NewInvalidConfigTypeError(fmt.Sprintf("%T", ec), e.kind) + } + return nil + } + + if _, ok := ec.(*config.DBConfig); !ok { + return NewInvalidConfigTypeError(fmt.Sprintf("%T", ec), e.kind) + } + return nil +} diff --git a/service/sql/errors.go b/service/sql/errors.go index 42b00daa8..2b3e4a05a 100644 --- a/service/sql/errors.go +++ b/service/sql/errors.go @@ -52,12 +52,6 @@ func NewUnsupportedAccessModeError(mode string) apierror.Error { WithDetails(attrs.NewBagFrom(map[string]any{"mode": mode})) } -func NewUnsupportedDatabaseTypeError(kind registry.Kind) apierror.Error { - return apierror.New(apierror.Invalid, "unsupported database type"). - WithRetryable(apierror.False). - WithDetails(attrs.NewBagFrom(map[string]any{"database_type": kind})) -} - func NewConnectionPoolCreationError(err error) apierror.Error { apiErr := apierror.New(apierror.Internal, "failed to create connection pool").WithRetryable(apierror.False) if err != nil { @@ -66,14 +60,6 @@ func NewConnectionPoolCreationError(err error) apierror.Error { return apiErr } -func NewSQLiteConnectionCreationError(err error) apierror.Error { - apiErr := apierror.New(apierror.Internal, "failed to create SQLite connection").WithRetryable(apierror.False) - if err != nil { - apiErr = apiErr.WithDetails(attrs.NewBagFrom(map[string]any{"cause": err.Error()})).WithCause(err) - } - return apiErr -} - func NewWALModeError(err error) apierror.Error { apiErr := apierror.New(apierror.Internal, "failed to enable WAL mode").WithRetryable(apierror.False) if err != nil { @@ -115,11 +101,3 @@ func NewPoolUpdateError(err error) apierror.Error { } return apiErr } - -func NewSQLiteUpdateError(err error) apierror.Error { - apiErr := apierror.New(apierror.Internal, "failed to update SQLite config").WithRetryable(apierror.False) - if err != nil { - apiErr = apiErr.WithDetails(attrs.NewBagFrom(map[string]any{"cause": err.Error()})).WithCause(err) - } - return apiErr -} diff --git a/service/sql/factory.go b/service/sql/factory.go index 38e370ae8..3789a26c6 100644 --- a/service/sql/factory.go +++ b/service/sql/factory.go @@ -4,94 +4,42 @@ package sql import ( "context" - "database/sql" "github.com/wippyai/runtime/api/registry" config "github.com/wippyai/runtime/api/service/sql" ) -// PoolFactoryAPI defines the interface for creating database connection pools -type PoolFactoryAPI interface { - // CreateStandardPool creates a connection pool for standard SQL databases (Postgres, MySQL) - CreateStandardPool(ctx context.Context, kind registry.Kind, cfg *config.DBConfig) (*ConnPool, error) - - // CreateSQLitePool creates a connection pool for SQLite databases - CreateSQLitePool(ctx context.Context, cfg *config.SQLiteConfig) (*ConnPool, error) +// Factory creates and updates connection pools. It dispatches to the engine +// registered for an entry's kind, so it never needs per-engine branches. +type Factory interface { + CreatePool(ctx context.Context, deps EngineDeps, entry registry.Entry) (*ConnPool, config.EngineConfig, error) + UpdatePool(ctx context.Context, deps EngineDeps, pool *ConnPool, entry registry.Entry) (config.EngineConfig, error) } -// DefaultPoolFactory is the default implementation of PoolFactoryAPI +// DefaultPoolFactory is the registry-backed Factory used in production. type DefaultPoolFactory struct{} -// NewDefaultPoolFactory creates a new default pool factory -func NewDefaultPoolFactory() PoolFactoryAPI { +// NewDefaultPoolFactory creates a new default pool factory. +func NewDefaultPoolFactory() Factory { return &DefaultPoolFactory{} } -// CreateStandardPool implements PoolFactoryAPI.CreateStandardPool -func (f *DefaultPoolFactory) CreateStandardPool(_ context.Context, kind registry.Kind, cfg *config.DBConfig) (*ConnPool, error) { - if err := cfg.Validate(); err != nil { - return nil, NewInvalidConfigError(err) - } - - db, err := openStandardDB(kind, cfg) - if err != nil { - return nil, err - } - - pool := &ConnPool{ - kind: kind, - db: db, - current: newDBGeneration(db), - status: make(chan any, 1), +// CreatePool implements Factory.CreatePool. +func (f *DefaultPoolFactory) CreatePool(ctx context.Context, deps EngineDeps, entry registry.Entry) (*ConnPool, config.EngineConfig, error) { + eng, ok := engineFor(entry.Kind) + if !ok { + return nil, nil, NewUnsupportedEntryKindError(entry.Kind) } - var cfgAny any = cfg - pool.config.Store(&cfgAny) - - return pool, nil + return createPool(ctx, deps, eng, entry) } -// CreateSQLitePool implements PoolFactoryAPI.CreateSQLitePool -func (f *DefaultPoolFactory) CreateSQLitePool(ctx context.Context, cfg *config.SQLiteConfig) (*ConnPool, error) { - if err := cfg.Validate(); err != nil { - return nil, NewInvalidConfigError(err) - } - - var dsn string - - // Handle in-memory database - if cfg.File == ":memory:" { - dsn = ":memory:" - } else { - // Use the file path directly - dsn = "file:" + cfg.File + "?mode=rwc" +// UpdatePool implements Factory.UpdatePool. +func (f *DefaultPoolFactory) UpdatePool(ctx context.Context, deps EngineDeps, pool *ConnPool, entry registry.Entry) (config.EngineConfig, error) { + eng, ok := engineFor(entry.Kind) + if !ok { + return nil, NewUnsupportedEntryKindError(entry.Kind) } - db, err := sql.Open("sqlite3", dsn) - if err != nil { - return nil, NewSQLiteConnectionCreationError(err) - } - - // Enable WAL mode for better concurrency - if _, err := db.ExecContext(ctx, "PRAGMA journal_mode=WAL;"); err != nil { - _ = db.Close() - return nil, NewWALModeError(err) - } - - // SQLite specific settings - db.SetMaxOpenConns(1) // SQLite supports only one writer - db.SetMaxIdleConns(1) - db.SetConnMaxLifetime(cfg.Pool.MaxLifetime) - - pool := &ConnPool{ - kind: config.SQLite, - db: db, - current: newDBGeneration(db), - status: make(chan any, 1), - } - - var cfgAny any = cfg - pool.config.Store(&cfgAny) - - return pool, nil + return updatePool(ctx, deps, eng, pool, entry) } diff --git a/service/sql/factory_test.go b/service/sql/factory_test.go index 7b21637cc..d8cc20775 100644 --- a/service/sql/factory_test.go +++ b/service/sql/factory_test.go @@ -4,14 +4,18 @@ package sql import ( "context" + "fmt" "testing" "time" _ "github.com/mattn/go-sqlite3" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "github.com/wippyai/runtime/api/payload" "github.com/wippyai/runtime/api/registry" config "github.com/wippyai/runtime/api/service/sql" + "github.com/wippyai/runtime/api/supervisor" + "go.uber.org/zap" ) func createTestDBConfig() *config.DBConfig { @@ -32,128 +36,45 @@ func createTestDBConfig() *config.DBConfig { } } -// TestDefaultPoolFactory_BuildDSN tests DSN string building without connecting to actual databases -func TestDefaultPoolFactory_BuildDSN(t *testing.T) { - tests := []struct { - name string - kind registry.Kind - cfg *config.DBConfig - expected string - isError bool - }{ - { - name: "PostgreSQL DSN", - kind: config.Postgres, - cfg: &config.DBConfig{ - Host: "localhost", - Port: 5432, - Database: "testdb", - Username: "user", - Password: "pass", - Options: map[string]string{ - "sslmode": "disable", - }, - }, - expected: "host='localhost' port=5432 user='user' password='pass' dbname='testdb' sslmode='disable'", - isError: false, - }, - { - name: "PostgreSQL DSN with connect timeout", - kind: config.Postgres, - cfg: &config.DBConfig{ - Host: "localhost", - Port: 5432, - Database: "testdb", - Username: "user", - Password: "pass", - Options: map[string]string{ - "connect_timeout": "2", - "sslmode": "disable", - }, - }, - expected: "host='localhost' port=5432 user='user' password='pass' dbname='testdb' connect_timeout='2' sslmode='disable'", - isError: false, - }, - { - name: "MySQL DSN", - kind: config.MySQL, - cfg: &config.DBConfig{ - Host: "localhost", - Port: 3306, - Database: "testdb", - Username: "user", - Password: "pass", - Options: map[string]string{ - "charset": "utf8mb4", - }, - }, - expected: "user:pass@tcp(localhost:3306)/testdb?charset=utf8mb4", - isError: false, - }, - { - name: "MySQL DSN with query options", - kind: config.MySQL, - cfg: &config.DBConfig{ - Host: "localhost", - Port: 3306, - Database: "testdb", - Username: "user", - Password: "pass", - Options: map[string]string{ - "charset": "utf8mb4", - "parseTime": "true", - "timeout": "2s", - }, - }, - expected: "user:pass@tcp(localhost:3306)/testdb?charset=utf8mb4&parseTime=true&timeout=2s", - isError: false, - }, - { - name: "Unsupported database type", - kind: "db.unsupported", - cfg: createTestDBConfig(), - expected: "", - isError: true, - }, - { - name: "PostgreSQL DSN with empty username is rejected", - kind: config.Postgres, - cfg: &config.DBConfig{ - Host: "localhost", - Port: 5432, - Database: "testdb", - Username: "", - Password: "secret", - Options: map[string]string{ - "sslmode": "disable", - }, - }, - expected: "", - isError: true, - }, - } +// fixedTranscoder decodes registry entries into a preset configuration, letting +// factory tests exercise the engine lifecycle without a real registry payload. +type fixedTranscoder struct{ cfg any } - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - dsn, err := buildDSN(tt.kind, tt.cfg) +func (f fixedTranscoder) Marshal(v any) (payload.Payload, error) { + return payload.New(v), nil +} - if tt.isError { - assert.Error(t, err) - } else { - assert.NoError(t, err) - assert.Equal(t, tt.expected, dsn) - } - }) +func (f fixedTranscoder) Unmarshal(_ payload.Payload, v any) error { + switch target := v.(type) { + case *config.DBConfig: + if c, ok := f.cfg.(*config.DBConfig); ok { + *target = *c + return nil + } + case *config.SQLiteConfig: + if c, ok := f.cfg.(*config.SQLiteConfig); ok { + *target = *c + return nil + } } + return fmt.Errorf("unexpected decode target %T", v) } -// TestDefaultPoolFactory_CreateStandardPool tests standard pool factory methods validation -func TestDefaultPoolFactory_CreateStandardPool(t *testing.T) { - // We'll test the validation logic without actually connecting +func (f fixedTranscoder) Transcode(p payload.Payload, format payload.Format) (payload.Payload, error) { + return payload.NewPayload(p.Data(), format), nil +} + +func depsFor(cfg any) EngineDeps { + return EngineDeps{Transcoder: fixedTranscoder{cfg: cfg}, Env: NewMockEnvRegistry(), Log: zap.NewNop()} +} + +// TestDefaultPoolFactory_CreatePoolValidation tests pool creation validation through +// the registry-backed factory. +func TestDefaultPoolFactory_CreatePoolValidation(t *testing.T) { factory := &DefaultPoolFactory{} tests := []struct { - cfg *config.DBConfig + cfg any name string kind registry.Kind errMsg string @@ -162,50 +83,58 @@ func TestDefaultPoolFactory_CreateStandardPool(t *testing.T) { { name: "Invalid configuration - empty host", kind: config.Postgres, - cfg: &config.DBConfig{Host: "", Port: 5432, Database: "db", Username: "user", Password: "pass"}, + cfg: &config.DBConfig{Host: "", Port: 5432, Database: "db", Username: "user", Password: "pass", Pool: config.PoolConfig{MaxLifetime: time.Hour}}, isError: true, errMsg: "invalid configuration", }, { name: "Invalid configuration - zero port", kind: config.Postgres, - cfg: &config.DBConfig{Host: "localhost", Port: 0, Database: "db", Username: "user", Password: "pass"}, + cfg: &config.DBConfig{Host: "localhost", Port: 0, Database: "db", Username: "user", Password: "pass", Pool: config.PoolConfig{MaxLifetime: time.Hour}}, isError: true, errMsg: "invalid configuration", }, { name: "Invalid configuration - empty database", kind: config.Postgres, - cfg: &config.DBConfig{Host: "localhost", Port: 5432, Database: "", Username: "user", Password: "pass"}, + cfg: &config.DBConfig{Host: "localhost", Port: 5432, Database: "", Username: "user", Password: "pass", Pool: config.PoolConfig{MaxLifetime: time.Hour}}, isError: true, errMsg: "invalid configuration", }, { name: "Invalid configuration - empty username", kind: config.Postgres, - cfg: &config.DBConfig{Host: "localhost", Port: 5432, Database: "db", Username: "", Password: "pass"}, + cfg: &config.DBConfig{Host: "localhost", Port: 5432, Database: "db", Username: "", Password: "pass", Pool: config.PoolConfig{MaxLifetime: time.Hour}}, isError: true, errMsg: "invalid configuration", }, { name: "Invalid configuration - empty password", kind: config.Postgres, - cfg: &config.DBConfig{Host: "localhost", Port: 5432, Database: "db", Username: "user", Password: ""}, + cfg: &config.DBConfig{Host: "localhost", Port: 5432, Database: "db", Username: "user", Password: "", Pool: config.PoolConfig{MaxLifetime: time.Hour}}, isError: true, errMsg: "invalid configuration", }, { - name: "Unsupported database type", + name: "Invalid configuration - SQLite empty file", + kind: config.SQLite, + cfg: &config.SQLiteConfig{File: "", Pool: config.PoolConfig{MaxLifetime: time.Hour}}, + isError: true, + errMsg: "invalid configuration", + }, + { + name: "Unsupported entry kind", kind: "db.unsupported", cfg: createTestDBConfig(), isError: true, - errMsg: "invalid connection config", + errMsg: "unsupported entry kind", }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - pool, err := factory.CreateStandardPool(context.Background(), tt.kind, tt.cfg) + entry := registry.Entry{ID: registry.NewID("test", "x"), Kind: tt.kind, Data: payload.New("x")} + pool, _, err := factory.CreatePool(context.Background(), depsFor(tt.cfg), entry) if tt.isError { require.Error(t, err) @@ -216,39 +145,22 @@ func TestDefaultPoolFactory_CreateStandardPool(t *testing.T) { } } -// TestDefaultPoolFactory_CreateSQLitePoolValidation tests SQLite pool validation -func TestDefaultPoolFactory_CreateSQLitePoolValidation(t *testing.T) { - factory := &DefaultPoolFactory{} - - tests := []struct { - cfg *config.SQLiteConfig - name string - errMsg string - isError bool - }{ - { - name: "Invalid configuration - empty file", - cfg: &config.SQLiteConfig{File: "", Pool: config.PoolConfig{MaxLifetime: time.Hour}}, - isError: true, - errMsg: "invalid configuration", - }, - { - name: "Invalid configuration - zero max lifetime", - cfg: &config.SQLiteConfig{File: ":memory:", Pool: config.PoolConfig{MaxLifetime: 0}}, - isError: true, - errMsg: "invalid configuration", - }, +// TestDefaultPoolFactory_CreatePoolSQLiteSuccess exercises the full create lifecycle +// (open, WAL prepare, tune, store) for a SQLite pool. +func TestDefaultPoolFactory_CreatePoolSQLiteSuccess(t *testing.T) { + cfg := &config.SQLiteConfig{ + File: ":memory:", + Lifecycle: supervisor.LifecycleConfig{StartTimeout: time.Minute}, + Pool: config.PoolConfig{MaxLifetime: time.Hour}, } + entry := registry.Entry{ID: registry.NewID("test", "lite"), Kind: config.SQLite, Data: payload.New("x")} - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - pool, err := factory.CreateSQLitePool(context.Background(), tt.cfg) + pool, ec, err := (&DefaultPoolFactory{}).CreatePool(context.Background(), depsFor(cfg), entry) + require.NoError(t, err) + require.NotNil(t, pool) + assert.Equal(t, config.SQLite, pool.kind) + require.NotNil(t, ec) + assert.Equal(t, time.Minute, ec.LifecycleConfig().StartTimeout) - if tt.isError { - require.Error(t, err) - assert.Contains(t, err.Error(), tt.errMsg) - assert.Nil(t, pool) - } - }) - } + require.NoError(t, pool.Stop(context.Background())) } diff --git a/service/sql/manager.go b/service/sql/manager.go index 3d11f0598..2836e8221 100644 --- a/service/sql/manager.go +++ b/service/sql/manager.go @@ -6,13 +6,12 @@ import ( "context" "sync" + envapi "github.com/wippyai/runtime/api/env" "github.com/wippyai/runtime/api/event" "github.com/wippyai/runtime/api/payload" "github.com/wippyai/runtime/api/registry" "github.com/wippyai/runtime/api/resource" - config "github.com/wippyai/runtime/api/service/sql" "github.com/wippyai/runtime/api/supervisor" - entryutil "github.com/wippyai/runtime/system/entry" "go.uber.org/zap" ) @@ -20,7 +19,8 @@ import ( type Manager struct { dtt payload.Transcoder bus event.Bus - factory PoolFactoryAPI + factory Factory + env envapi.Registry log *zap.Logger services map[registry.ID]*ConnPool mu sync.RWMutex @@ -31,8 +31,9 @@ func NewManager( dtt payload.Transcoder, bus event.Bus, log *zap.Logger, + envRegistry envapi.Registry, ) (*Manager, error) { - return NewManagerWithFactory(dtt, bus, log, NewDefaultPoolFactory()) + return NewManagerWithFactory(dtt, bus, log, envRegistry, NewDefaultPoolFactory()) } // NewManagerWithFactory creates a new SQL service manager with the specified pool factory @@ -40,7 +41,8 @@ func NewManagerWithFactory( dtt payload.Transcoder, bus event.Bus, log *zap.Logger, - factory PoolFactoryAPI, + envRegistry envapi.Registry, + factory Factory, ) (*Manager, error) { if dtt == nil { return nil, ErrTranscoderRequired @@ -60,123 +62,65 @@ func NewManagerWithFactory( dtt: dtt, bus: bus, factory: factory, + env: envRegistry, services: make(map[registry.ID]*ConnPool), }, nil } -// Add implements registry.EntryListener -func (m *Manager) Add(ctx context.Context, entry registry.Entry) error { - m.mu.Lock() - defer m.mu.Unlock() - - switch entry.Kind { - case config.Postgres, config.MySQL: - return m.handleStandardDBAdd(ctx, entry) - case config.SQLite: - return m.handleSQLiteAdd(ctx, entry) - default: - return NewUnsupportedEntryKindError(entry.Kind) - } +// deps bundles the manager's collaborators for the engine lifecycle. +func (m *Manager) deps() EngineDeps { + return EngineDeps{Transcoder: m.dtt, Env: m.env, Log: m.log} } -// Update implements registry.EntryListener -func (m *Manager) Update(ctx context.Context, entry registry.Entry) error { +// Add implements registry.EntryListener +func (m *Manager) Add(ctx context.Context, entry registry.Entry) error { m.mu.Lock() defer m.mu.Unlock() - switch entry.Kind { - case config.Postgres, config.MySQL: - return m.handleStandardDBUpdate(ctx, entry) - case config.SQLite: - return m.handleSQLiteUpdate(ctx, entry) - default: + if _, ok := engineFor(entry.Kind); !ok { return NewUnsupportedEntryKindError(entry.Kind) } -} -// Delete implements registry.EntryListener -func (m *Manager) Delete(ctx context.Context, entry registry.Entry) error { - m.mu.Lock() - defer m.mu.Unlock() - - return m.handleDBDelete(ctx, entry) -} - -func (m *Manager) handleStandardDBAdd(ctx context.Context, entry registry.Entry) error { if _, exists := m.services[entry.ID]; exists { return NewServiceExistsError(entry.ID) } - cfg, err := entryutil.DecodeEntryConfig[config.DBConfig](ctx, m.dtt, entry) + pool, cfg, err := m.factory.CreatePool(ctx, m.deps(), entry) if err != nil { - return NewInvalidConfigError(err) + return err } - pool, err := m.factory.CreateStandardPool(ctx, entry.Kind, cfg) - if err != nil { - return NewConnectionPoolCreationError(err) - } - - return m.registerService(ctx, entry, pool, cfg.Lifecycle) + return m.registerService(ctx, entry, pool, cfg.LifecycleConfig()) } -func (m *Manager) handleSQLiteAdd(ctx context.Context, entry registry.Entry) error { - if _, exists := m.services[entry.ID]; exists { - return NewServiceExistsError(entry.ID) - } - - cfg, err := entryutil.DecodeEntryConfig[config.SQLiteConfig](ctx, m.dtt, entry) - if err != nil { - return NewInvalidConfigError(err) - } +// Update implements registry.EntryListener +func (m *Manager) Update(ctx context.Context, entry registry.Entry) error { + m.mu.Lock() + defer m.mu.Unlock() - pool, err := m.factory.CreateSQLitePool(ctx, cfg) - if err != nil { - return NewSQLiteConnectionCreationError(err) + if _, ok := engineFor(entry.Kind); !ok { + return NewUnsupportedEntryKindError(entry.Kind) } - return m.registerService(ctx, entry, pool, cfg.Lifecycle) -} - -func (m *Manager) handleStandardDBUpdate(ctx context.Context, entry registry.Entry) error { pool, exists := m.services[entry.ID] if !exists { return NewServiceNotFoundError(entry.ID) } - cfg, err := entryutil.DecodeEntryConfig[config.DBConfig](ctx, m.dtt, entry) + cfg, err := m.factory.UpdatePool(ctx, m.deps(), pool, entry) if err != nil { - return NewInvalidConfigError(err) - } - - if err := pool.UpdateConfig(cfg); err != nil { - return NewPoolUpdateError(err) + return err } - m.updateService(ctx, entry, cfg.Lifecycle) + m.updateService(ctx, entry, cfg.LifecycleConfig()) return nil } -func (m *Manager) handleSQLiteUpdate(ctx context.Context, entry registry.Entry) error { - pool, exists := m.services[entry.ID] - if !exists { - return NewServiceNotFoundError(entry.ID) - } - - cfg, err := entryutil.DecodeEntryConfig[config.SQLiteConfig](ctx, m.dtt, entry) - if err != nil { - return NewInvalidConfigError(err) - } - - if err := pool.UpdateConfig(cfg); err != nil { - return NewSQLiteUpdateError(err) - } - - m.updateService(ctx, entry, cfg.Lifecycle) - return nil -} +// Delete implements registry.EntryListener +func (m *Manager) Delete(ctx context.Context, entry registry.Entry) error { + m.mu.Lock() + defer m.mu.Unlock() -func (m *Manager) handleDBDelete(ctx context.Context, entry registry.Entry) error { _, exists := m.services[entry.ID] if !exists { return NewServiceNotFoundError(entry.ID) diff --git a/service/sql/manager_test.go b/service/sql/manager_test.go index 5f7880df9..ac60292ac 100644 --- a/service/sql/manager_test.go +++ b/service/sql/manager_test.go @@ -143,16 +143,21 @@ func NewMockConnPool(kind registry.Kind) *ConnPool { return pool } +type standardPoolCall struct { + Cfg *apiconfig.DBConfig + Kind registry.Kind +} + +type sqlitePoolCall struct { + Cfg *apiconfig.SQLiteConfig + Kind registry.Kind +} + // Mock factory implementation type TestPoolFactory struct { - standardPoolCalls []struct { - Cfg *apiconfig.DBConfig - Kind registry.Kind - } - sqlitePoolCalls []struct { - Cfg *apiconfig.SQLiteConfig - } - shouldFailNext bool + standardPoolCalls []standardPoolCall + sqlitePoolCalls []sqlitePoolCall + shouldFailNext bool } func NewTestPoolFactory() *TestPoolFactory { @@ -162,32 +167,58 @@ func NewTestPoolFactory() *TestPoolFactory { } } -func (f *TestPoolFactory) CreateStandardPool(_ context.Context, kind registry.Kind, cfg *apiconfig.DBConfig) (*ConnPool, error) { - f.standardPoolCalls = append(f.standardPoolCalls, struct { - Cfg *apiconfig.DBConfig - Kind registry.Kind - }{ - Kind: kind, - Cfg: cfg, - }) - +func (f *TestPoolFactory) CreatePool(ctx context.Context, deps EngineDeps, entry registry.Entry) (*ConnPool, apiconfig.EngineConfig, error) { if f.shouldFailNext { - return nil, assert.AnError + return nil, nil, assert.AnError } - return NewMockConnPool(kind), nil -} -func (f *TestPoolFactory) CreateSQLitePool(_ context.Context, cfg *apiconfig.SQLiteConfig) (*ConnPool, error) { - f.sqlitePoolCalls = append(f.sqlitePoolCalls, struct { - Cfg *apiconfig.SQLiteConfig - }{ - Cfg: cfg, - }) + eng, ok := engineFor(entry.Kind) + if !ok { + return nil, nil, NewUnsupportedEntryKindError(entry.Kind) + } + cfg, err := eng.DecodeConfig(ctx, deps.Transcoder, entry) + if err != nil { + return nil, nil, err + } + if err := eng.ResolveEnv(ctx, deps, cfg); err != nil { + return nil, nil, err + } + if err := cfg.Validate(); err != nil { + return nil, nil, NewInvalidConfigError(err) + } + + if c, ok := cfg.(*apiconfig.SQLiteConfig); ok { + f.sqlitePoolCalls = append(f.sqlitePoolCalls, sqlitePoolCall{Kind: entry.Kind, Cfg: c}) + } else if c, ok := cfg.(*apiconfig.DBConfig); ok { + f.standardPoolCalls = append(f.standardPoolCalls, standardPoolCall{Kind: entry.Kind, Cfg: c}) + } + pool := NewMockConnPool(entry.Kind) + var cfgAny any = cfg + pool.config.Store(&cfgAny) + return pool, cfg, nil +} + +func (f *TestPoolFactory) UpdatePool(ctx context.Context, deps EngineDeps, pool *ConnPool, entry registry.Entry) (apiconfig.EngineConfig, error) { if f.shouldFailNext { return nil, assert.AnError } - return NewMockConnPool(apiconfig.SQLite), nil + + eng, ok := engineFor(entry.Kind) + if !ok { + return nil, NewUnsupportedEntryKindError(entry.Kind) + } + cfg, err := eng.DecodeConfig(ctx, deps.Transcoder, entry) + if err != nil { + return nil, err + } + if err := eng.ResolveEnv(ctx, deps, cfg); err != nil { + return nil, err + } + if err := pool.updateConfig(ctx, eng, cfg); err != nil { + return nil, err + } + return cfg, nil } // MockEnvRegistry implements envapi.Registry for testing @@ -255,7 +286,7 @@ func newTestManager(t *testing.T) (*Manager, event.Bus, *TestPoolFactory) { transcoder := &TestTranscoder{} factory := NewTestPoolFactory() - manager, err := NewManagerWithFactory(transcoder, bus, logger, factory) + manager, err := NewManagerWithFactory(transcoder, bus, logger, NewMockEnvRegistry(), factory) require.NoError(t, err) return manager, bus, factory } @@ -267,7 +298,7 @@ func TestNewManagerWithFactory(t *testing.T) { factory := NewTestPoolFactory() t.Run("Valid initialization", func(t *testing.T) { - manager, err := NewManagerWithFactory(transcoder, bus, logger, factory) + manager, err := NewManagerWithFactory(transcoder, bus, logger, NewMockEnvRegistry(), factory) assert.NoError(t, err) assert.NotNil(t, manager) assert.Equal(t, logger, manager.log) @@ -278,21 +309,21 @@ func TestNewManagerWithFactory(t *testing.T) { }) t.Run("Nil transcoder", func(t *testing.T) { - manager, err := NewManagerWithFactory(nil, bus, logger, factory) + manager, err := NewManagerWithFactory(nil, bus, logger, NewMockEnvRegistry(), factory) require.Error(t, err) assert.Nil(t, manager) assert.Contains(t, err.Error(), "transcoder is required") }) t.Run("Nil event bus", func(t *testing.T) { - manager, err := NewManagerWithFactory(transcoder, nil, logger, factory) + manager, err := NewManagerWithFactory(transcoder, nil, logger, NewMockEnvRegistry(), factory) require.Error(t, err) assert.Nil(t, manager) assert.Contains(t, err.Error(), "event bus is required") }) t.Run("Nil factory", func(t *testing.T) { - manager, err := NewManagerWithFactory(transcoder, bus, logger, nil) + manager, err := NewManagerWithFactory(transcoder, bus, logger, NewMockEnvRegistry(), nil) require.Error(t, err) assert.Nil(t, manager) assert.Contains(t, err.Error(), "pool factory is required") @@ -350,7 +381,12 @@ func TestManager_Add(t *testing.T) { id registry.ID shouldFail bool expectSuccess bool - }{} + }{ + {name: "add postgres", kind: apiconfig.Postgres, id: registry.NewID("test", "add-pg"), shouldFail: false, expectSuccess: true}, + {name: "add sqlite", kind: apiconfig.SQLite, id: registry.NewID("test", "add-lite"), shouldFail: false, expectSuccess: true}, + {name: "add failure", kind: apiconfig.Postgres, id: registry.NewID("test", "add-fail"), shouldFail: true, expectSuccess: false}, + {name: "unsupported kind", kind: "db.unsupported", id: registry.NewID("test", "add-bad"), shouldFail: false, expectSuccess: false}, + } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { diff --git a/test.sh b/test.sh index 9c5ec9fb3..0a2c7e0cc 100755 --- a/test.sh +++ b/test.sh @@ -32,6 +32,7 @@ go test \ ./system/env/... \ ./service/env/... \ ./service/sql/... \ + ./service/cdc/sqlite \ ./service/cdc/postgres \ ./runtime/lua/modules/cdc \ ./runtime/lua/modules/hub \ @@ -39,6 +40,9 @@ go test \ ./api/fs \ ./boot/components/dispatchers +echo "running sqlite cdc integration tests (local temp file, no docker)" +CGO_ENABLED=1 go test -tags "integration sqlite_preupdate_hook" ./service/cdc/sqlite + if [[ -n "${WIPPY_CDC_IT_REPL_DSN:-}" && -n "${WIPPY_CDC_IT_ADMIN_DSN:-}" ]]; then go test -tags integration ./service/cdc/postgres exit 0