diff --git a/AGENTS.md b/AGENTS.md index e49cf75..bb65e1d 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -15,6 +15,12 @@ packages/sdk/src/*.gen.ts ← mise run sdk (source: packages/ap Change the source and re-run the generator. Editing a handler's `@Summary` means `mise run swagger` followed by `mise run sdk`. +## Never edit a merged migration + +Existing databases upgrade in place, and every push to `main` publishes the +`latest` image. Change the schema in a new numbered migration; see +[docs/development.md](docs/development.md#database-migrations). + ## Everything is in English Code, comments, log lines, error messages and documentation. The only place diff --git a/docs/development.md b/docs/development.md index 28d8106..65ee03c 100644 --- a/docs/development.md +++ b/docs/development.md @@ -134,7 +134,40 @@ publishes `ghcr.io/felinics/connect-it`, and finally creates a GitHub Release with generated notes. A failed CI or image publish therefore cannot create a GitHub Release. +## Database migrations + +`packages/service/migrations` is embedded in the binary and applied at +startup, and existing databases upgrade in place. Every push to `main` +publishes the `latest` image, so a migration is frozen once merged: change the +schema in a new numbered migration, never by editing an existing one. + +The binary only migrates up, and an older binary refuses to start against a +newer schema version. To roll back: + +1. Stop every instance. +2. Migrate the service's schema (the first `search_path` entry of + `DATABASE_URL`, or `public` when it sets none) down to the newest version + the older image ships. From a checkout of the newer release, run the + golang-migrate CLI on the database's network: + + ```bash + docker run --rm --network -v "$PWD/packages/service/migrations:/m:ro" \ + migrate/migrate:v4.19.1 -path /m -database '' goto + ``` + +3. Start the older image. + +`0002_connector_settings.down.sql` keeps the table and its settings, so going +from 2 to 1 only resets the version, and this statement can replace the CLI in +step 2. Replace `connect_it` with the service's schema and expect `UPDATE 1`: + +```sql +UPDATE connect_it.schema_migrations SET version = 1 WHERE version = 2 AND NOT dirty; +``` + +v0.1.1 and v0.2.0 ignore `connector_settings`: after rolling back to them, +disabled connectors are available again until the next upgrade. + ## Status -The project is pre-1.0. Migrations only maintain the schema of a fresh -database; in-place upgrades of older development databases are not promised. +The project is pre-1.0. diff --git a/packages/service/migrate_upgrade_test.go b/packages/service/migrate_upgrade_test.go new file mode 100644 index 0000000..722cfdf --- /dev/null +++ b/packages/service/migrate_upgrade_test.go @@ -0,0 +1,223 @@ +package service + +import ( + "context" + "encoding/json" + "net/url" + "os" + "reflect" + "strings" + "testing" + + "github.com/jackc/pgx/v5" + + "github.com/felinics/connect-it/packages/core/connector" + "github.com/felinics/connect-it/packages/core/crypto" + "github.com/felinics/connect-it/packages/core/registry" + "github.com/felinics/connect-it/packages/service/catalogsvc" + "github.com/felinics/connect-it/packages/service/configsvc" + "github.com/felinics/connect-it/packages/service/store" +) + +func TestMigrateUpFromV020(t *testing.T) { + for _, fallback := range []bool{false, true} { + name := "isolated" + if fallback { + name = "with_fallback" + } + t.Run(name, func(t *testing.T) { + dbURL, conn := versionOneDatabase(t, true) + before := storedUpgradeRows(t, conn) + u, err := url.Parse(dbURL) + if err != nil { + t.Fatal(err) + } + target := u.Query().Get("search_path") + var fallbackConn *pgx.Conn + if fallback { + fallbackURL := newTestSchema(t) + fallbackConn, err = pgx.Connect(t.Context(), fallbackURL) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = fallbackConn.Close(context.Background()) }) + _, err = fallbackConn.Exec(t.Context(), `CREATE TABLE connector_settings + (connector_type text PRIMARY KEY, enabled boolean NOT NULL DEFAULT true, updated_at timestamptz NOT NULL); + INSERT INTO connector_settings VALUES ('fixture',false,'2026-01-01Z');`) + if err != nil { + t.Fatal(err) + } + fallbackU, _ := url.Parse(fallbackURL) + q := u.Query() + q.Set("search_path", target+","+fallbackU.Query().Get("search_path")) + u.RawQuery = q.Encode() + dbURL = u.String() + } + for range 2 { + if err := MigrateUp(dbURL); err != nil { + t.Fatal(err) + } + if !tableExists(t, dbURL, target+".connector_settings") { + t.Fatal("upgrade did not create connector_settings in the owning schema") + } + assertUpgradeVersion(t, conn, 2) + if !reflect.DeepEqual(storedUpgradeRows(t, conn), before) { + t.Fatal("upgrade changed existing stored rows") + } + assertUpgradeCatalog(t, conn, true) + } + if fallbackConn != nil { + var enabled bool + if err := fallbackConn.QueryRow(t.Context(), "SELECT enabled FROM connector_settings WHERE connector_type='fixture'").Scan(&enabled); err != nil || enabled { + t.Fatalf("fallback settings changed: enabled=%v err=%v", enabled, err) + } + if tableExists(t, fallbackConn.Config().ConnString(), "schema_migrations") { + t.Fatal("migration metadata leaked into fallback schema") + } + } + }) + } +} + +func TestMigrateUpPreservesVersionOneSettings(t *testing.T) { + dbURL, conn := versionOneDatabase(t, false) + if _, err := conn.Exec(t.Context(), "INSERT INTO connector_settings VALUES ('fixture',false,'2026-01-01Z')"); err != nil { + t.Fatal(err) + } + before := storedUpgradeRows(t, conn) + var setting string + if err := conn.QueryRow(t.Context(), "SELECT row_to_json(s)::text FROM connector_settings s").Scan(&setting); err != nil { + t.Fatal(err) + } + for range 2 { + if err := MigrateUp(dbURL); err != nil { + t.Fatal(err) + } + assertUpgradeVersion(t, conn, 2) + assertUpgradeCatalog(t, conn, false) + } + m, db, err := newMigrator(dbURL) + if err != nil { + t.Fatal(err) + } + if err := m.Steps(-1); err != nil { + t.Fatal(err) + } + _ = db.Close() + assertUpgradeVersion(t, conn, 1) + var after string + if err := conn.QueryRow(t.Context(), "SELECT row_to_json(s)::text FROM connector_settings s").Scan(&after); err != nil || after != setting { + t.Fatalf("rollback changed existing settings: %v", err) + } + if err := MigrateUp(dbURL); err != nil { + t.Fatal(err) + } + assertUpgradeVersion(t, conn, 2) + assertUpgradeCatalog(t, conn, false) + if err := conn.QueryRow(t.Context(), "SELECT row_to_json(s)::text FROM connector_settings s").Scan(&after); err != nil || after != setting { + t.Fatalf("reapplying upgrade changed existing settings: %v", err) + } + if !reflect.DeepEqual(storedUpgradeRows(t, conn), before) { + t.Fatal("upgrade/rollback changed existing stored rows") + } +} + +func TestMigrateUpConcurrentLegacyStart(t *testing.T) { + dbURL, conn := versionOneDatabase(t, true) + before := storedUpgradeRows(t, conn) + errs := make(chan error, 4) + for range 4 { + go func() { errs <- MigrateUp(dbURL) }() + } + for range 4 { + if err := <-errs; err != nil { + t.Fatal(err) + } + } + assertUpgradeVersion(t, conn, 2) + assertUpgradeCatalog(t, conn, true) + if !reflect.DeepEqual(storedUpgradeRows(t, conn), before) { + t.Fatal("concurrent upgrade changed existing stored rows") + } +} + +func versionOneDatabase(t *testing.T, legacy bool) (string, *pgx.Conn) { + t.Helper() + dbURL := newTestSchema(t) + conn, err := pgx.Connect(t.Context(), dbURL) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = conn.Close(context.Background()) }) + var baseline []byte + if legacy { + baseline, err = os.ReadFile("testdata/v0.2.0_init.sql") + } else { + baseline, err = migrationsFS.ReadFile("migrations/0001_init.up.sql") + } + if err != nil { + t.Fatal(err) + } + if _, err := conn.Exec(t.Context(), string(baseline)); err != nil { + t.Fatal(err) + } + if _, err := conn.Exec(t.Context(), `CREATE TABLE schema_migrations (version bigint PRIMARY KEY, dirty boolean NOT NULL); + INSERT INTO schema_migrations VALUES (1,false); + INSERT INTO api_tokens VALUES ('00000000-0000-0000-0000-000000000001','fixture','stored-token-hash','2026-01-01Z',NULL); + INSERT INTO connections (id,connector_type,alias,auth_method,credential,secret_key_version,profile,scopes,status,created_at,updated_at) + VALUES ('00000000-0000-0000-0000-000000000002','fixture','saved','oauth','\x01020304',1,'{"name":"stored"}','{repo}','active','2026-01-01Z','2026-01-01Z');`); err != nil { + t.Fatal(err) + } + kr, err := crypto.ParseKeyring("1:" + strings.Repeat("11", 32)) + if err != nil { + t.Fatal(err) + } + ciphertext, version, err := kr.Encrypt([]byte(`{"client_secret":"fixture-value"}`), []byte("fixture")) + if err != nil { + t.Fatal(err) + } + if _, err := conn.Exec(t.Context(), `INSERT INTO connector_configs + (connector_type,config_schema_version,public_config,secret_config,secret_key_version,created_at,updated_at) + VALUES ('fixture',1,'{"client_id":"stored"}',$1,$2,'2026-01-01Z','2026-01-01Z')`, ciphertext, version); err != nil { + t.Fatal(err) + } + return dbURL, conn +} + +func storedUpgradeRows(t *testing.T, conn *pgx.Conn) map[string]json.RawMessage { + t.Helper() + rows := map[string]json.RawMessage{} + for _, table := range []string{"api_tokens", "connections", "connector_configs"} { + var data []byte + if err := conn.QueryRow(t.Context(), "SELECT coalesce(json_agg(t), '[]') FROM "+table+" t").Scan(&data); err != nil { + t.Fatal(err) + } + rows[table] = data + } + return rows +} + +func assertUpgradeVersion(t *testing.T, conn *pgx.Conn, want int) { + t.Helper() + var version int + var dirty bool + if err := conn.QueryRow(t.Context(), "SELECT version,dirty FROM schema_migrations").Scan(&version, &dirty); err != nil || version != want || dirty { + t.Fatalf("migration version=%d dirty=%v err=%v; want %d clean", version, dirty, err, want) + } +} + +func assertUpgradeCatalog(t *testing.T, conn *pgx.Conn, enabled bool) { + t.Helper() + kr, err := crypto.ParseKeyring("1:" + strings.Repeat("11", 32)) + if err != nil { + t.Fatal(err) + } + reg := registry.New() + reg.MustRegister(connector.Definition{Type: "fixture", Name: "Fixture", ConfigSchemaVersion: 1, + Implementation: connector.RemoteMCP{Endpoint: "https://example.test/mcp"}}) + cat := catalogsvc.New(reg, configsvc.New(store.New(conn), reg, kr)) + items, err := cat.List(t.Context()) + if err != nil || len(items) != 1 || items[0].Type != "fixture" || items[0].Enabled != enabled { + t.Fatalf("catalog unavailable or settings changed: items=%+v err=%v", items, err) + } +} diff --git a/packages/service/migrations/0002_connector_settings.down.sql b/packages/service/migrations/0002_connector_settings.down.sql new file mode 100644 index 0000000..02db8b5 --- /dev/null +++ b/packages/service/migrations/0002_connector_settings.down.sql @@ -0,0 +1,2 @@ +-- The current version-1 baseline already owns this table and its settings. +select 1; diff --git a/packages/service/migrations/0002_connector_settings.up.sql b/packages/service/migrations/0002_connector_settings.up.sql new file mode 100644 index 0000000..b3fd5f9 --- /dev/null +++ b/packages/service/migrations/0002_connector_settings.up.sql @@ -0,0 +1,5 @@ +create table if not exists connector_settings ( + connector_type text primary key, + enabled boolean not null default true, + updated_at timestamptz not null +); diff --git a/packages/service/testdata/v0.2.0_init.sql b/packages/service/testdata/v0.2.0_init.sql new file mode 100644 index 0000000..0fc32c4 --- /dev/null +++ b/packages/service/testdata/v0.2.0_init.sql @@ -0,0 +1,106 @@ +create table admin_account ( + id integer primary key check (id = 1), + username text not null, + password_hash text not null, + updated_at timestamptz not null +); + +create table api_tokens ( + id uuid primary key, + name text not null, + token_hash text not null unique, + created_at timestamptz not null, + revoked_at timestamptz +); + +create table connector_configs ( + connector_type text primary key, + config_schema_version integer not null, + public_config jsonb not null default '{}', + secret_config bytea not null, + secret_key_version integer not null, + created_at timestamptz not null, + updated_at timestamptz not null +); + +create table oauth_clients ( + id uuid primary key, + connector_type text not null, + resource text not null, + issuer text not null, + redirect_uri text not null, + client_id text not null, + client_secret bytea not null, + secret_key_version integer not null, + token_endpoint text not null, + token_endpoint_auth_method text not null, + client_secret_expires_at timestamptz, + created_at timestamptz not null, + updated_at timestamptz not null, + unique (connector_type, resource, issuer, redirect_uri) +); + +create table connections ( + id uuid primary key, + connector_type text not null, + alias text, + auth_method text not null, + credential bytea not null, + secret_key_version integer not null, + profile jsonb not null default '{}', + scopes text[] not null default '{}', + status text not null, + access_token_expires_at timestamptz, + oauth_client_id uuid references oauth_clients(id), + created_at timestamptz not null, + updated_at timestamptz not null +); + +create table oauth_authorizations ( + id uuid primary key, + connector_type text not null, + state_hash text not null unique, + pkce_verifier bytea not null, + secret_key_version integer not null, + auth_method text not null, + connection_id uuid not null references connections(id) on delete cascade, + oauth_client_id uuid references oauth_clients(id), + status text not null, + expires_at timestamptz not null, + created_at timestamptz not null +); + +create table mcp_sessions ( + id uuid primary key, + token_hash text not null unique, + api_token_id uuid not null references api_tokens(id) on delete cascade, + tool_snapshot jsonb not null, + expires_at timestamptz not null, + created_at timestamptz not null +); +create index mcp_sessions_expires_at_idx on mcp_sessions (expires_at); +create index mcp_sessions_api_token_id_idx on mcp_sessions (api_token_id); + +create table mcp_session_connections ( + session_id uuid not null references mcp_sessions(id) on delete cascade, + alias text not null, + connection_id uuid not null references connections(id) on delete cascade, + primary key (session_id, alias), + unique (session_id, connection_id) +); +create index mcp_session_connections_connection_id_idx + on mcp_session_connections (connection_id); + +create table tool_runs ( + id uuid primary key, + connector_type text not null, + connection_id uuid, + tool_id text not null, + session_id uuid, + api_token_id uuid, + status text not null, + error_kind text, + upstream_status integer, + duration_ms integer, + created_at timestamptz not null +);