feat(cdc): native SQLite CDC source (db.cdc.sqlite)

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)

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
+}

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)
+}

api/service/cdc/config.go

@@ -11,6 +11,7 @@ import (
 
 const (
 	Postgres registry.Kind = "db.cdc.postgres"
+	SQLite   registry.Kind = "db.cdc.sqlite"
 )
 
 const (

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)
+}

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))
+}

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"`

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)
 )

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 == "" {

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
 		},
 	})