fix(migrations): add connector settings to existing databases

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

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 <network> -v "$PWD/packages/service/migrations:/m:ro" \
+     migrate/migrate:v4.19.1 -path /m -database '<DATABASE_URL with scheme pgx5://>' goto <version>
+   ```
+
+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.

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

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;

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

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