diff --git a/apps/backend/internal/agent/settings/store/postgres_schema_test.go b/apps/backend/internal/agent/settings/store/postgres_schema_test.go new file mode 100644 index 0000000000..800dd4c626 --- /dev/null +++ b/apps/backend/internal/agent/settings/store/postgres_schema_test.go @@ -0,0 +1,15 @@ +package store + +import ( + "testing" + + "github.com/kandev/kandev/internal/testutil" +) + +func TestPostgresFreshSchemaInitializes(t *testing.T) { + db := testutil.OpenIsolatedPostgres(t, testutil.PostgresDSNFromEnv(t)) + + if _, err := newSQLiteRepositoryWithDB(db, db, nil); err != nil { + t.Fatalf("init fresh postgres schema: %v", err) + } +} diff --git a/apps/backend/internal/agent/settings/store/sqlite.go b/apps/backend/internal/agent/settings/store/sqlite.go index 6c982f13fa..69eb176389 100644 --- a/apps/backend/internal/agent/settings/store/sqlite.go +++ b/apps/backend/internal/agent/settings/store/sqlite.go @@ -204,6 +204,10 @@ func (r *sqliteRepository) migrateOfficeEnrichmentColumns() { // The migration is idempotent: it inspects sqlite_master for the CHECK keyword // and only proceeds when the constraint is present. func (r *sqliteRepository) migrateDropModelCheckConstraint() error { + if dialect.IsPostgres(r.db.DriverName()) { + return nil + } + var tableDDL string err := r.db.QueryRow( `SELECT sql FROM sqlite_master WHERE type='table' AND name='agent_profiles'`, diff --git a/apps/backend/internal/db/dialect/dialect_test.go b/apps/backend/internal/db/dialect/dialect_test.go index 081bd4a483..f34e2315c5 100644 --- a/apps/backend/internal/db/dialect/dialect_test.go +++ b/apps/backend/internal/db/dialect/dialect_test.go @@ -27,6 +27,15 @@ func TestBoolToInt(t *testing.T) { } } +func TestBlobType(t *testing.T) { + if BlobType(SQLite3) != "BLOB" { + t.Errorf("sqlite: got %q", BlobType(SQLite3)) + } + if BlobType(PGX) != "BYTEA" { + t.Errorf("pgx: got %q", BlobType(PGX)) + } +} + func TestJSONExtract(t *testing.T) { got := JSONExtract(SQLite3, "metadata", "status") if got != "json_extract(metadata, '$.status')" { diff --git a/apps/backend/internal/db/dialect/types.go b/apps/backend/internal/db/dialect/types.go new file mode 100644 index 0000000000..b78c88c79c --- /dev/null +++ b/apps/backend/internal/db/dialect/types.go @@ -0,0 +1,9 @@ +package dialect + +// BlobType returns the portable binary column type for the active driver. +func BlobType(driver string) string { + if IsPostgres(driver) { + return "BYTEA" + } + return "BLOB" +} diff --git a/apps/backend/internal/office/repository/sqlite/base.go b/apps/backend/internal/office/repository/sqlite/base.go index eb49d3ebd5..a62e79c6be 100644 --- a/apps/backend/internal/office/repository/sqlite/base.go +++ b/apps/backend/internal/office/repository/sqlite/base.go @@ -206,8 +206,8 @@ func (r *Repository) createAgentTables() error { is_system INTEGER NOT NULL DEFAULT 0, system_version TEXT NOT NULL DEFAULT '', default_for_roles TEXT NOT NULL DEFAULT '[]', - created_at DATETIME NOT NULL, - updated_at DATETIME NOT NULL, + created_at TIMESTAMP NOT NULL, + updated_at TIMESTAMP NOT NULL, UNIQUE(workspace_id, slug) ); `) @@ -227,8 +227,8 @@ func (r *Repository) createProjectTables() error { budget_cents INTEGER DEFAULT 0, repositories TEXT DEFAULT '[]', executor_config TEXT DEFAULT '{}', - created_at DATETIME NOT NULL, - updated_at DATETIME NOT NULL + created_at TIMESTAMP NOT NULL, + updated_at TIMESTAMP NOT NULL ); `) return err @@ -240,8 +240,8 @@ func (r *Repository) createAgentRuntimeTable() error { agent_id TEXT PRIMARY KEY, status TEXT NOT NULL DEFAULT 'idle', pause_reason TEXT DEFAULT '', - last_run_finished_at DATETIME, - updated_at DATETIME DEFAULT CURRENT_TIMESTAMP + last_run_finished_at TIMESTAMP, + updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); `) return err @@ -266,8 +266,8 @@ func (r *Repository) createCostTables() error { tokens_out INTEGER DEFAULT 0, cost_subcents INTEGER NOT NULL DEFAULT 0, estimated INTEGER NOT NULL DEFAULT 0, - occurred_at DATETIME NOT NULL, - created_at DATETIME NOT NULL + occurred_at TIMESTAMP NOT NULL, + created_at TIMESTAMP NOT NULL ); CREATE INDEX IF NOT EXISTS idx_office_cost_agent ON office_cost_events(agent_profile_id); @@ -283,8 +283,8 @@ func (r *Repository) createCostTables() error { period TEXT NOT NULL, alert_threshold_pct INTEGER DEFAULT 80, action_on_exceed TEXT DEFAULT 'notify_only', - created_at DATETIME NOT NULL, - updated_at DATETIME NOT NULL + created_at TIMESTAMP NOT NULL, + updated_at TIMESTAMP NOT NULL ); `) return err @@ -307,7 +307,7 @@ func (r *Repository) createRunTables() error { failure_reason TEXT NOT NULL DEFAULT '', session_id TEXT NOT NULL DEFAULT '', retry_count INTEGER DEFAULT 0, - scheduled_retry_at DATETIME, + scheduled_retry_at TIMESTAMP, error_message TEXT NOT NULL DEFAULT '', cancel_reason TEXT, -- Provider routing (office-provider-routing). @@ -330,9 +330,9 @@ func (r *Repository) createRunTables() error { result_json TEXT NOT NULL DEFAULT '{}', assembled_prompt TEXT NOT NULL DEFAULT '', summary_injected TEXT NOT NULL DEFAULT '', - requested_at DATETIME NOT NULL, - claimed_at DATETIME, - finished_at DATETIME + requested_at TIMESTAMP NOT NULL, + claimed_at TIMESTAMP, + finished_at TIMESTAMP ); CREATE INDEX IF NOT EXISTS idx_run_status_requested ON runs(status, requested_at); CREATE UNIQUE INDEX IF NOT EXISTS idx_run_idempotency ON runs(idempotency_key) WHERE idempotency_key IS NOT NULL; @@ -363,9 +363,9 @@ func (r *Repository) createRoutineTables() error { catch_up_policy TEXT NOT NULL DEFAULT 'enqueue_missed_with_cap', catch_up_max INTEGER NOT NULL DEFAULT 25, variables TEXT DEFAULT '{}', - last_run_at DATETIME, - created_at DATETIME NOT NULL, - updated_at DATETIME NOT NULL + last_run_at TIMESTAMP, + created_at TIMESTAMP NOT NULL, + updated_at TIMESTAMP NOT NULL ); CREATE TABLE IF NOT EXISTS office_routine_triggers ( @@ -377,11 +377,11 @@ func (r *Repository) createRoutineTables() error { public_id TEXT DEFAULT '', signing_mode TEXT DEFAULT '', secret TEXT DEFAULT '', - next_run_at DATETIME, - last_fired_at DATETIME, + next_run_at TIMESTAMP, + last_fired_at TIMESTAMP, enabled INTEGER DEFAULT 1, - created_at DATETIME NOT NULL, - updated_at DATETIME NOT NULL, + created_at TIMESTAMP NOT NULL, + updated_at TIMESTAMP NOT NULL, FOREIGN KEY (routine_id) REFERENCES office_routines(id) ON DELETE CASCADE ); @@ -395,9 +395,9 @@ func (r *Repository) createRoutineTables() error { linked_task_id TEXT DEFAULT '', coalesced_into_run_id TEXT DEFAULT '', dispatch_fingerprint TEXT DEFAULT '', - started_at DATETIME, - completed_at DATETIME, - created_at DATETIME NOT NULL, + started_at TIMESTAMP, + completed_at TIMESTAMP, + created_at TIMESTAMP NOT NULL, FOREIGN KEY (routine_id) REFERENCES office_routines(id) ON DELETE CASCADE ); `) @@ -415,9 +415,9 @@ func (r *Repository) createApprovalTables() error { payload TEXT DEFAULT '{}', decision_note TEXT DEFAULT '', decided_by TEXT DEFAULT '', - decided_at DATETIME, - created_at DATETIME NOT NULL, - updated_at DATETIME NOT NULL + decided_at TIMESTAMP, + created_at TIMESTAMP NOT NULL, + updated_at TIMESTAMP NOT NULL ); `) return err @@ -436,7 +436,7 @@ func (r *Repository) createActivityTables() error { details TEXT DEFAULT '{}', run_id TEXT NOT NULL DEFAULT '', session_id TEXT NOT NULL DEFAULT '', - created_at DATETIME NOT NULL + created_at TIMESTAMP NOT NULL ); CREATE INDEX IF NOT EXISTS idx_activity_workspace_created ON office_activity_log(workspace_id, created_at DESC); CREATE INDEX IF NOT EXISTS idx_activity_run_id ON office_activity_log(run_id) WHERE run_id != ''; @@ -448,7 +448,7 @@ func (r *Repository) createActivityTables() error { event_type TEXT NOT NULL, level TEXT NOT NULL DEFAULT 'info', payload TEXT NOT NULL DEFAULT '{}', - created_at DATETIME NOT NULL, + created_at TIMESTAMP NOT NULL, PRIMARY KEY (run_id, seq) ); CREATE INDEX IF NOT EXISTS idx_run_events_run_created ON run_events(run_id, created_at); @@ -465,8 +465,8 @@ func (r *Repository) createMemoryTables() error { key TEXT NOT NULL, content TEXT DEFAULT '', metadata TEXT DEFAULT '{}', - created_at DATETIME NOT NULL, - updated_at DATETIME NOT NULL, + created_at TIMESTAMP NOT NULL, + updated_at TIMESTAMP NOT NULL, UNIQUE(agent_profile_id, layer, key) ); `) @@ -484,8 +484,8 @@ func (r *Repository) createChannelTables() error { webhook_secret TEXT NOT NULL DEFAULT '', status TEXT DEFAULT 'active', task_id TEXT DEFAULT '', - created_at DATETIME NOT NULL, - updated_at DATETIME NOT NULL + created_at TIMESTAMP NOT NULL, + updated_at TIMESTAMP NOT NULL ); `) return err @@ -498,8 +498,8 @@ func (r *Repository) createOnboardingTable() error { completed INTEGER NOT NULL DEFAULT 0, ceo_agent_id TEXT DEFAULT '', first_task_id TEXT DEFAULT '', - completed_at DATETIME, - created_at DATETIME DEFAULT CURRENT_TIMESTAMP + completed_at TIMESTAMP, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); `) return err @@ -513,8 +513,8 @@ func (r *Repository) createInstructionTable() error { filename TEXT NOT NULL, content TEXT NOT NULL DEFAULT '', is_entry INTEGER DEFAULT 0, - created_at DATETIME DEFAULT CURRENT_TIMESTAMP, - updated_at DATETIME DEFAULT CURRENT_TIMESTAMP, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, UNIQUE(agent_profile_id, filename) ); `) @@ -528,8 +528,8 @@ func (r *Repository) createLabelTables() error { workspace_id TEXT NOT NULL, name TEXT NOT NULL, color TEXT NOT NULL DEFAULT '#6b7280', - created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, - updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, UNIQUE(workspace_id, name) ); CREATE INDEX IF NOT EXISTS idx_office_labels_workspace ON office_labels(workspace_id); @@ -537,7 +537,7 @@ func (r *Repository) createLabelTables() error { CREATE TABLE IF NOT EXISTS office_task_labels ( task_id TEXT NOT NULL, label_id TEXT NOT NULL, - created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (task_id, label_id), FOREIGN KEY (task_id) REFERENCES tasks(id) ON DELETE CASCADE, FOREIGN KEY (label_id) REFERENCES office_labels(id) ON DELETE CASCADE @@ -552,7 +552,7 @@ func (r *Repository) createTaskExtensionTables() error { CREATE TABLE IF NOT EXISTS task_blockers ( task_id TEXT NOT NULL, blocker_task_id TEXT NOT NULL, - created_at DATETIME NOT NULL, + created_at TIMESTAMP NOT NULL, PRIMARY KEY (task_id, blocker_task_id), CHECK (task_id != blocker_task_id) ); @@ -565,7 +565,7 @@ func (r *Repository) createTaskExtensionTables() error { body TEXT NOT NULL, source TEXT NOT NULL DEFAULT 'user', reply_channel_id TEXT DEFAULT '', - created_at DATETIME NOT NULL + created_at TIMESTAMP NOT NULL ); CREATE INDEX IF NOT EXISTS idx_task_comments_task_created ON task_comments(task_id, created_at); diff --git a/apps/backend/internal/office/repository/sqlite/base_migrations.go b/apps/backend/internal/office/repository/sqlite/base_migrations.go index d1ad4ffa72..157f631847 100644 --- a/apps/backend/internal/office/repository/sqlite/base_migrations.go +++ b/apps/backend/internal/office/repository/sqlite/base_migrations.go @@ -104,7 +104,7 @@ func (r *Repository) migrateFailureColumns() { CREATE TABLE IF NOT EXISTS office_workspace_settings ( workspace_id TEXT PRIMARY KEY, agent_failure_threshold INTEGER NOT NULL DEFAULT 3, - updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP + updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP )`) _, _ = r.db.Exec(` @@ -112,7 +112,7 @@ func (r *Repository) migrateFailureColumns() { user_id TEXT NOT NULL, item_kind TEXT NOT NULL, item_id TEXT NOT NULL, - dismissed_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + dismissed_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (user_id, item_kind, item_id) )`) _, _ = r.db.Exec(`CREATE INDEX IF NOT EXISTS idx_office_inbox_dismissals_kind ON office_inbox_dismissals(item_kind, item_id)`) @@ -126,7 +126,7 @@ func (r *Repository) migrateFailureColumns() { // directly without an interim ALTER step. func (r *Repository) migrateSchedulerColumns() { r.migrate.Apply("tasks.checkout_agent_id", `ALTER TABLE tasks ADD COLUMN checkout_agent_id TEXT`) - r.migrate.Apply("tasks.checkout_at", `ALTER TABLE tasks ADD COLUMN checkout_at DATETIME`) + r.migrate.Apply("tasks.checkout_at", `ALTER TABLE tasks ADD COLUMN checkout_at TIMESTAMP`) } // migrateTaskFTS creates the FTS5 virtual table and triggers for full-text task search. @@ -354,7 +354,7 @@ func taskPriorityMigrationStatements() []string { labels TEXT DEFAULT '[]', identifier TEXT, checkout_agent_id TEXT, - checkout_at DATETIME + checkout_at TIMESTAMP )`, // archived_by_cascade_id is added to the task schema by // task/repository/sqlite/base.go runMigrations() via an diff --git a/apps/backend/internal/office/repository/sqlite/failure.go b/apps/backend/internal/office/repository/sqlite/failure.go index 2f154482b8..d27fd7d8e1 100644 --- a/apps/backend/internal/office/repository/sqlite/failure.go +++ b/apps/backend/internal/office/repository/sqlite/failure.go @@ -262,7 +262,7 @@ func (r *Repository) ListFailedRunsForInbox( } // parseSqliteTime parses the various string formats SQLite stores -// DATETIME values in (RFC3339, with or without timezone, with or +// TIMESTAMP values in (RFC3339, with or without timezone, with or // without nanoseconds). Falls back to zero time on parse failure // rather than erroring, since the inbox doesn't break if a row's // timestamp is unparseable. diff --git a/apps/backend/internal/office/repository/sqlite/tree_holds.go b/apps/backend/internal/office/repository/sqlite/tree_holds.go index 80e60b2cec..7e371ce535 100644 --- a/apps/backend/internal/office/repository/sqlite/tree_holds.go +++ b/apps/backend/internal/office/repository/sqlite/tree_holds.go @@ -20,10 +20,10 @@ func (r *Repository) createTreeHoldTables() error { root_task_id TEXT NOT NULL, mode TEXT NOT NULL, release_policy TEXT NOT NULL DEFAULT '{"strategy":"manual"}', - released_at DATETIME, + released_at TIMESTAMP, released_by TEXT DEFAULT '', released_reason TEXT DEFAULT '', - created_at DATETIME NOT NULL, + created_at TIMESTAMP NOT NULL, FOREIGN KEY (root_task_id) REFERENCES tasks(id) ON DELETE CASCADE ); diff --git a/apps/backend/internal/office/repository/sqlite/workspace_groups.go b/apps/backend/internal/office/repository/sqlite/workspace_groups.go index 6fb2ed1850..2ed4922736 100644 --- a/apps/backend/internal/office/repository/sqlite/workspace_groups.go +++ b/apps/backend/internal/office/repository/sqlite/workspace_groups.go @@ -31,13 +31,13 @@ func (r *Repository) createWorkspaceGroupTables() error { owned_by_kandev INTEGER NOT NULL DEFAULT 0, cleanup_policy TEXT NOT NULL DEFAULT 'never_delete', cleanup_status TEXT NOT NULL DEFAULT 'active', - cleaned_at DATETIME, + cleaned_at TIMESTAMP, cleanup_error TEXT NOT NULL DEFAULT '', restore_status TEXT NOT NULL DEFAULT 'not_needed', restore_error TEXT NOT NULL DEFAULT '', restore_config_json TEXT NOT NULL DEFAULT '', - created_at DATETIME NOT NULL, - updated_at DATETIME NOT NULL + created_at TIMESTAMP NOT NULL, + updated_at TIMESTAMP NOT NULL ); CREATE INDEX IF NOT EXISTS idx_task_workspace_groups_workspace @@ -49,10 +49,10 @@ func (r *Repository) createWorkspaceGroupTables() error { workspace_group_id TEXT NOT NULL, task_id TEXT NOT NULL, role TEXT NOT NULL DEFAULT 'member', - released_at DATETIME, + released_at TIMESTAMP, release_reason TEXT NOT NULL DEFAULT '', released_by_cascade_id TEXT NOT NULL DEFAULT '', - created_at DATETIME NOT NULL, + created_at TIMESTAMP NOT NULL, PRIMARY KEY (workspace_group_id, task_id), FOREIGN KEY (workspace_group_id) REFERENCES task_workspace_groups(id) ON DELETE CASCADE ); diff --git a/apps/backend/internal/persistence/meta.go b/apps/backend/internal/persistence/meta.go index bf482f6686..5089d63611 100644 --- a/apps/backend/internal/persistence/meta.go +++ b/apps/backend/internal/persistence/meta.go @@ -33,7 +33,7 @@ func ensureMetaTable(db *sqlx.DB) error { // readKey returns the value for key, or "" when the key is absent. func readKey(db *sqlx.DB, key string) (string, error) { var value string - err := db.QueryRow(`SELECT value FROM kandev_meta WHERE key = ?`, key).Scan(&value) + err := db.QueryRow(db.Rebind(`SELECT value FROM kandev_meta WHERE key = ?`), key).Scan(&value) if errors.Is(err, sql.ErrNoRows) { return "", nil } @@ -46,8 +46,8 @@ func readKey(db *sqlx.DB, key string) (string, error) { // writeKey upserts key=value into kandev_meta. func writeKey(db *sqlx.DB, key, value string) error { _, err := db.Exec( - `INSERT INTO kandev_meta (key, value) VALUES (?, ?) - ON CONFLICT(key) DO UPDATE SET value = excluded.value`, + db.Rebind(`INSERT INTO kandev_meta (key, value) VALUES (?, ?) + ON CONFLICT(key) DO UPDATE SET value = excluded.value`), key, value, ) if err != nil { diff --git a/apps/backend/internal/persistence/postgres_meta_test.go b/apps/backend/internal/persistence/postgres_meta_test.go new file mode 100644 index 0000000000..926dfe4d17 --- /dev/null +++ b/apps/backend/internal/persistence/postgres_meta_test.go @@ -0,0 +1,28 @@ +package persistence + +import ( + "testing" + "time" + + "github.com/kandev/kandev/internal/testutil" +) + +func TestPostgresLatestVersionMetaRoundTrip(t *testing.T) { + db := testutil.OpenIsolatedPostgres(t, testutil.PostgresDSNFromEnv(t)) + + if err := ensureMetaTable(db); err != nil { + t.Fatalf("ensure meta table: %v", err) + } + + checkedAt := time.Unix(123, 0).UTC() + if err := WriteLatestVersion(db, "v1.2.3", "https://example.test/release", checkedAt); err != nil { + t.Fatalf("write latest version: %v", err) + } + version, url, gotCheckedAt, err := ReadLatestVersion(db) + if err != nil { + t.Fatalf("read latest version: %v", err) + } + if version != "v1.2.3" || url != "https://example.test/release" || !gotCheckedAt.Equal(checkedAt) { + t.Fatalf("meta round-trip = (%q, %q, %s)", version, url, gotCheckedAt) + } +} diff --git a/apps/backend/internal/persistence/provider.go b/apps/backend/internal/persistence/provider.go index 3aa702dd00..1c27a877ff 100644 --- a/apps/backend/internal/persistence/provider.go +++ b/apps/backend/internal/persistence/provider.go @@ -186,6 +186,10 @@ func providePostgres(cfg *config.Config, log *logger.Logger) (*db.Pool, func() e pgDB := sqlx.NewDb(dbConn, "pgx") // For Postgres, writer and reader share the same pool. pool := db.NewPool(pgDB, pgDB) + if err := ensureMetaTable(pgDB); err != nil { + _ = pool.Close() + return nil, nil, fmt.Errorf("ensure meta table: %w", err) + } if log != nil { log.Info("pre-migration backup skipped: postgres driver (use pg_dump)") diff --git a/apps/backend/internal/secrets/postgres_store_test.go b/apps/backend/internal/secrets/postgres_store_test.go new file mode 100644 index 0000000000..cef2e85dda --- /dev/null +++ b/apps/backend/internal/secrets/postgres_store_test.go @@ -0,0 +1,92 @@ +package secrets + +import ( + "context" + "strings" + "testing" + + "github.com/kandev/kandev/internal/testutil" +) + +func TestPostgresStoreRoundTrip(t *testing.T) { + db := testutil.OpenIsolatedPostgres(t, testutil.PostgresDSNFromEnv(t)) + + crypto, err := NewMasterKeyProvider(t.TempDir()) + if err != nil { + t.Fatalf("master key: %v", err) + } + store, cleanup, err := Provide(db, db, crypto) + if err != nil { + t.Fatalf("provide store: %v", err) + } + t.Cleanup(func() { + _ = cleanup() + }) + + ctx := context.Background() + secret := &SecretWithValue{Secret: Secret{Name: "token"}, Value: "s3cr3t"} + if err := store.Create(ctx, secret); err != nil { + t.Fatalf("create secret: %v", err) + } + got, err := store.Reveal(ctx, secret.ID) + if err != nil { + t.Fatalf("reveal secret: %v", err) + } + if got != "s3cr3t" { + t.Fatalf("revealed value = %q, want %q", got, "s3cr3t") + } + + metadata, err := store.Get(ctx, secret.ID) + if err != nil { + t.Fatalf("get secret: %v", err) + } + if metadata.Name != "token" { + t.Fatalf("secret name = %q, want %q", metadata.Name, "token") + } + + nameOnly := "renamed-token" + if err := store.Update(ctx, secret.ID, &UpdateSecretRequest{Name: &nameOnly}); err != nil { + t.Fatalf("update secret name: %v", err) + } + metadata, err = store.Get(ctx, secret.ID) + if err != nil { + t.Fatalf("get renamed secret: %v", err) + } + if metadata.Name != nameOnly { + t.Fatalf("renamed secret name = %q, want %q", metadata.Name, nameOnly) + } + got, err = store.Reveal(ctx, secret.ID) + if err != nil { + t.Fatalf("reveal renamed secret: %v", err) + } + if got != "s3cr3t" { + t.Fatalf("renamed secret value = %q, want %q", got, "s3cr3t") + } + + newValue := "n3w-s3cr3t" + valueName := "renamed-token-with-value" + if err := store.Update(ctx, secret.ID, &UpdateSecretRequest{Name: &valueName, Value: &newValue}); err != nil { + t.Fatalf("update secret value: %v", err) + } + metadata, err = store.Get(ctx, secret.ID) + if err != nil { + t.Fatalf("get updated secret: %v", err) + } + if metadata.Name != valueName { + t.Fatalf("updated secret name = %q, want %q", metadata.Name, valueName) + } + got, err = store.Reveal(ctx, secret.ID) + if err != nil { + t.Fatalf("reveal updated secret: %v", err) + } + if got != newValue { + t.Fatalf("updated secret value = %q, want %q", got, newValue) + } + + if err := store.Delete(ctx, secret.ID); err != nil { + t.Fatalf("delete secret: %v", err) + } + if _, err := store.Get(ctx, secret.ID); err == nil || !strings.Contains(err.Error(), "secret not found") { + t.Fatalf("get deleted secret error = %v, want not found", err) + } +} diff --git a/apps/backend/internal/secrets/sqlite_store.go b/apps/backend/internal/secrets/sqlite_store.go index 631a7fd7cd..82720edb11 100644 --- a/apps/backend/internal/secrets/sqlite_store.go +++ b/apps/backend/internal/secrets/sqlite_store.go @@ -8,6 +8,8 @@ import ( "github.com/google/uuid" "github.com/jmoiron/sqlx" + + "github.com/kandev/kandev/internal/db/dialect" ) type sqliteStore struct { @@ -29,16 +31,17 @@ func Provide(writer, reader *sqlx.DB, crypto *MasterKeyProvider) (*sqliteStore, } func (s *sqliteStore) initSchema() error { - schema := ` + binaryType := dialect.BlobType(s.db.DriverName()) + schema := fmt.Sprintf(` CREATE TABLE IF NOT EXISTS secrets ( id TEXT PRIMARY KEY, name TEXT NOT NULL, - encrypted_value BLOB NOT NULL, - nonce BLOB NOT NULL, + encrypted_value %s NOT NULL, + nonce %s NOT NULL, created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ); - ` + `, binaryType, binaryType) _, err := s.db.Exec(schema) return err } @@ -63,9 +66,9 @@ func (s *sqliteStore) Create(ctx context.Context, secret *SecretWithValue) error return fmt.Errorf("encrypt secret: %w", err) } - _, err = s.db.ExecContext(ctx, ` + _, err = s.db.ExecContext(ctx, s.db.Rebind(` INSERT INTO secrets (id, name, encrypted_value, nonce, created_at, updated_at) - VALUES (?, ?, ?, ?, ?, ?)`, + VALUES (?, ?, ?, ?, ?, ?)`), secret.ID, secret.Name, ciphertext, nonce, now, now, ) if err != nil { @@ -76,9 +79,9 @@ func (s *sqliteStore) Create(ctx context.Context, secret *SecretWithValue) error func (s *sqliteStore) Get(ctx context.Context, id string) (*Secret, error) { var row secretRow - err := s.ro.GetContext(ctx, &row, ` + err := s.ro.GetContext(ctx, &row, s.ro.Rebind(` SELECT id, name, created_at, updated_at - FROM secrets WHERE id = ?`, id) + FROM secrets WHERE id = ?`), id) if err != nil { if err == sql.ErrNoRows { return nil, fmt.Errorf("secret not found: %s", id) @@ -90,8 +93,8 @@ func (s *sqliteStore) Get(ctx context.Context, id string) (*Secret, error) { func (s *sqliteStore) Reveal(ctx context.Context, id string) (string, error) { var ciphertext, nonce []byte - err := s.ro.QueryRowContext(ctx, ` - SELECT encrypted_value, nonce FROM secrets WHERE id = ?`, id). + err := s.ro.QueryRowContext(ctx, s.ro.Rebind(` + SELECT encrypted_value, nonce FROM secrets WHERE id = ?`), id). Scan(&ciphertext, &nonce) if err != nil { if err == sql.ErrNoRows { @@ -124,18 +127,18 @@ func (s *sqliteStore) Update(ctx context.Context, id string, req *UpdateSecretRe if err != nil { return fmt.Errorf("encrypt secret: %w", err) } - _, err = s.db.ExecContext(ctx, ` + _, err = s.db.ExecContext(ctx, s.db.Rebind(` UPDATE secrets SET name = ?, encrypted_value = ?, nonce = ?, updated_at = ? - WHERE id = ?`, + WHERE id = ?`), existing.Name, ciphertext, nonce, now, id, ) if err != nil { return fmt.Errorf("update secret: %w", err) } } else { - _, err = s.db.ExecContext(ctx, ` + _, err = s.db.ExecContext(ctx, s.db.Rebind(` UPDATE secrets SET name = ?, updated_at = ? - WHERE id = ?`, + WHERE id = ?`), existing.Name, now, id, ) if err != nil { @@ -146,7 +149,7 @@ func (s *sqliteStore) Update(ctx context.Context, id string, req *UpdateSecretRe } func (s *sqliteStore) Delete(ctx context.Context, id string) error { - result, err := s.db.ExecContext(ctx, `DELETE FROM secrets WHERE id = ?`, id) + result, err := s.db.ExecContext(ctx, s.db.Rebind(`DELETE FROM secrets WHERE id = ?`), id) if err != nil { return fmt.Errorf("delete secret: %w", err) } diff --git a/apps/backend/internal/task/repository/sqlite/base_migrations.go b/apps/backend/internal/task/repository/sqlite/base_migrations.go index ecc1dc4a65..520ba48d87 100644 --- a/apps/backend/internal/task/repository/sqlite/base_migrations.go +++ b/apps/backend/internal/task/repository/sqlite/base_migrations.go @@ -10,6 +10,8 @@ import ( "github.com/google/uuid" "go.uber.org/zap" + + "github.com/kandev/kandev/internal/db/dialect" ) // migrateExecutorProfiles adds mcp_policy column and drops is_default from executor_profiles. @@ -151,6 +153,10 @@ func (r *Repository) runMigrations() error { // the PRAGMA and the subsequent transaction use the same connection. // Returns true if the migration actually ran (gate fired), false if it was a no-op. func (r *Repository) recreateTable(tableName, triggerPhrase string, statements []string) (bool, error) { + if dialect.IsPostgres(r.db.DriverName()) { + return false, nil + } + var tableSql string err := r.db.QueryRow(`SELECT sql FROM sqlite_master WHERE type='table' AND name=?`, tableName).Scan(&tableSql) if errors.Is(err, sql.ErrNoRows) { @@ -242,6 +248,10 @@ func (r *Repository) migrateTasksRemoveWorkflowFK() error { // // Idempotent: rows that already exist on either side are left untouched. func (r *Repository) backfillExecutorsRunningFromTaskSessions() error { + if dialect.IsPostgres(r.db.DriverName()) { + return nil + } + // Check whether task_sessions still has the column. If migration already ran, // the column is gone and there's nothing to backfill. var tableSql string @@ -491,6 +501,10 @@ type backfillRow struct { // that have sessions but no environment, and links orphaned sessions. // Idempotent: tasks with existing environments are skipped. func (r *Repository) backfillTaskEnvironments() error { + if dialect.IsPostgres(r.db.DriverName()) { + return nil + } + orphaned, err := r.findOrphanedTasks() if err != nil { return err @@ -523,11 +537,11 @@ func (r *Repository) backfillTaskEnvironments() error { func (r *Repository) findOrphanedTasks() ([]backfillRow, error) { rows, err := r.db.Query(` SELECT ts.task_id, - COALESCE(ts.executor_id, ''), - COALESCE(ts.executor_profile_id, ''), - COALESCE(ts.repository_id, ''), - COALESCE(er.container_id, ''), - ts.started_at + MIN(COALESCE(ts.executor_id, '')), + MIN(COALESCE(ts.executor_profile_id, '')), + MIN(COALESCE(ts.repository_id, '')), + MIN(COALESCE(er.container_id, '')), + MIN(ts.started_at) FROM task_sessions ts LEFT JOIN task_environments te ON te.task_id = ts.task_id LEFT JOIN executors_running er ON er.session_id = ts.id diff --git a/apps/backend/internal/task/repository/sqlite/base_schema.go b/apps/backend/internal/task/repository/sqlite/base_schema.go index a8c3d32f4a..f0cc844199 100644 --- a/apps/backend/internal/task/repository/sqlite/base_schema.go +++ b/apps/backend/internal/task/repository/sqlite/base_schema.go @@ -5,6 +5,8 @@ import ( "fmt" "github.com/google/uuid" + + "github.com/kandev/kandev/internal/db/dialect" ) // initSchema creates the database tables if they don't exist and applies @@ -57,10 +59,19 @@ func (r *Repository) ensureWorkspaceIndexes() error { // ensureMessageMetadataIndexes creates indexes on JSON metadata fields for fast lookups. func (r *Repository) ensureMessageMetadataIndexes() error { - if _, err := r.db.Exec(`CREATE INDEX IF NOT EXISTS idx_messages_metadata_tool_call_id ON task_session_messages(task_session_id, json_extract(metadata, '$.tool_call_id'))`); err != nil { + driver := r.db.DriverName() + toolCallIndex := fmt.Sprintf( + `CREATE INDEX IF NOT EXISTS idx_messages_metadata_tool_call_id ON task_session_messages(task_session_id, (%s))`, + dialect.JSONExtract(driver, "metadata", "tool_call_id"), + ) + if _, err := r.db.Exec(toolCallIndex); err != nil { return err } - if _, err := r.db.Exec(`CREATE INDEX IF NOT EXISTS idx_messages_metadata_pending_id ON task_session_messages(task_session_id, json_extract(metadata, '$.pending_id'))`); err != nil { + pendingIndex := fmt.Sprintf( + `CREATE INDEX IF NOT EXISTS idx_messages_metadata_pending_id ON task_session_messages(task_session_id, (%s))`, + dialect.JSONExtract(driver, "metadata", "pending_id"), + ) + if _, err := r.db.Exec(pendingIndex); err != nil { return err } return nil @@ -238,21 +249,6 @@ func (r *Repository) initTaskSchema() error { updated_at TIMESTAMP NOT NULL ); - CREATE TABLE IF NOT EXISTS task_repositories ( - id TEXT PRIMARY KEY, - task_id TEXT NOT NULL, - repository_id TEXT NOT NULL, - base_branch TEXT DEFAULT '', - checkout_branch TEXT DEFAULT '', - position INTEGER DEFAULT 0, - metadata TEXT DEFAULT '{}', - created_at TIMESTAMP NOT NULL, - updated_at TIMESTAMP NOT NULL, - FOREIGN KEY (task_id) REFERENCES tasks(id) ON DELETE CASCADE, - FOREIGN KEY (repository_id) REFERENCES repositories(id) ON DELETE CASCADE, - UNIQUE(task_id, repository_id, base_branch, checkout_branch) - ); - CREATE TABLE IF NOT EXISTS repositories ( id TEXT PRIMARY KEY, workspace_id TEXT NOT NULL, @@ -276,6 +272,21 @@ func (r *Repository) initTaskSchema() error { FOREIGN KEY (workspace_id) REFERENCES workspaces(id) ON DELETE CASCADE ); + CREATE TABLE IF NOT EXISTS task_repositories ( + id TEXT PRIMARY KEY, + task_id TEXT NOT NULL, + repository_id TEXT NOT NULL, + base_branch TEXT DEFAULT '', + checkout_branch TEXT DEFAULT '', + position INTEGER DEFAULT 0, + metadata TEXT DEFAULT '{}', + created_at TIMESTAMP NOT NULL, + updated_at TIMESTAMP NOT NULL, + FOREIGN KEY (task_id) REFERENCES tasks(id) ON DELETE CASCADE, + FOREIGN KEY (repository_id) REFERENCES repositories(id) ON DELETE CASCADE, + UNIQUE(task_id, repository_id, base_branch, checkout_branch) + ); + CREATE TABLE IF NOT EXISTS repository_scripts ( id TEXT PRIMARY KEY, repository_id TEXT NOT NULL, @@ -445,14 +456,30 @@ func (r *Repository) initDocumentsSchema() error { } func (r *Repository) initSessionSchema() error { - if err := r.initMessageTurnSchema(); err != nil { + if err := r.initSessionWorktreeSchema(); err != nil { return err } - return r.initSessionWorktreeSchema() + return r.initMessageTurnSchema() } func (r *Repository) initMessageTurnSchema() error { _, err := r.db.Exec(` + CREATE TABLE IF NOT EXISTS task_session_turns ( + id TEXT PRIMARY KEY, + task_session_id TEXT NOT NULL, + task_id TEXT NOT NULL, + started_at TIMESTAMP NOT NULL, + completed_at TIMESTAMP, + metadata TEXT DEFAULT '{}', + created_at TIMESTAMP NOT NULL, + updated_at TIMESTAMP NOT NULL, + FOREIGN KEY (task_session_id) REFERENCES task_sessions(id) ON DELETE CASCADE + ); + + CREATE INDEX IF NOT EXISTS idx_turns_session_id ON task_session_turns(task_session_id); + CREATE INDEX IF NOT EXISTS idx_turns_session_started ON task_session_turns(task_session_id, started_at); + CREATE INDEX IF NOT EXISTS idx_turns_task_id ON task_session_turns(task_id); + CREATE TABLE IF NOT EXISTS task_session_messages ( id TEXT PRIMARY KEY, task_session_id TEXT NOT NULL, @@ -478,22 +505,6 @@ func (r *Repository) initMessageTurnSchema() error { -- updated_at ADD COLUMN + backfill. Creating it here would fail on existing -- DBs where CREATE TABLE IF NOT EXISTS is a no-op and the column does not -- yet exist (schema init runs before migrations). - - CREATE TABLE IF NOT EXISTS task_session_turns ( - id TEXT PRIMARY KEY, - task_session_id TEXT NOT NULL, - task_id TEXT NOT NULL, - started_at TIMESTAMP NOT NULL, - completed_at TIMESTAMP, - metadata TEXT DEFAULT '{}', - created_at TIMESTAMP NOT NULL, - updated_at TIMESTAMP NOT NULL, - FOREIGN KEY (task_session_id) REFERENCES task_sessions(id) ON DELETE CASCADE - ); - - CREATE INDEX IF NOT EXISTS idx_turns_session_id ON task_session_turns(task_session_id); - CREATE INDEX IF NOT EXISTS idx_turns_session_started ON task_session_turns(task_session_id, started_at); - CREATE INDEX IF NOT EXISTS idx_turns_task_id ON task_session_turns(task_id); `) return err } diff --git a/apps/backend/internal/task/repository/sqlite/postgres_schema_test.go b/apps/backend/internal/task/repository/sqlite/postgres_schema_test.go new file mode 100644 index 0000000000..7bb08a5d3d --- /dev/null +++ b/apps/backend/internal/task/repository/sqlite/postgres_schema_test.go @@ -0,0 +1,52 @@ +package sqlite + +import ( + "testing" + "time" + + "github.com/kandev/kandev/internal/testutil" +) + +func TestPostgresFreshSchemaInitializes(t *testing.T) { + db := testutil.OpenIsolatedPostgres(t, testutil.PostgresDSNFromEnv(t)) + + if _, err := NewWithDB(db, db, nil); err != nil { + t.Fatalf("init fresh postgres schema: %v", err) + } +} + +func TestPostgresSkipsLegacyTaskEnvironmentBackfill(t *testing.T) { + db := testutil.OpenIsolatedPostgres(t, testutil.PostgresDSNFromEnv(t)) + repo, err := NewWithDB(db, db, nil) + if err != nil { + t.Fatalf("init fresh postgres schema: %v", err) + } + + now := time.Now().UTC() + if _, err := db.Exec(db.Rebind(` + INSERT INTO tasks (id, title, created_at, updated_at) + VALUES (?, ?, ?, ?) + `), "task-orphaned", "Orphaned task", now, now); err != nil { + t.Fatalf("insert orphaned task: %v", err) + } + if _, err := db.Exec(db.Rebind(` + INSERT INTO task_sessions (id, task_id, state, started_at, updated_at) + VALUES (?, ?, ?, ?, ?) + `), "session-orphaned", "task-orphaned", "CREATED", now, now); err != nil { + t.Fatalf("insert orphaned session: %v", err) + } + + if err := repo.backfillTaskEnvironments(); err != nil { + t.Fatalf("backfill task environments: %v", err) + } + + var count int + if err := db.Get(&count, db.Rebind(` + SELECT COUNT(*) FROM task_environments WHERE task_id = ? + `), "task-orphaned"); err != nil { + t.Fatalf("count task environments: %v", err) + } + if count != 0 { + t.Fatalf("task environment count = %d, want 0", count) + } +} diff --git a/apps/backend/internal/task/repository/sqlite/task.go b/apps/backend/internal/task/repository/sqlite/task.go index 4641188381..3ad383fdb5 100644 --- a/apps/backend/internal/task/repository/sqlite/task.go +++ b/apps/backend/internal/task/repository/sqlite/task.go @@ -59,6 +59,15 @@ func isFromOfficeProjection(alias string) string { )` } +func excludeConfigModePredicate(driver, col string) string { + if dialect.IsPostgres(driver) { + // Repository writes always marshal metadata as JSON; dirty Postgres rows + // with malformed JSON should fail loudly instead of being silently skipped. + return fmt.Sprintf("COALESCE(%s, '') NOT IN ('true', '1')", dialect.JSONExtract(driver, col, "config_mode")) + } + return fmt.Sprintf("%s IS NOT 1", dialect.JSONExtract(driver, col, "config_mode")) +} + // runnerProjection produces the correlated subquery (without alias) that // resolves the runner for the row of `tasks` with the given alias. Used // inline in SELECT projections. @@ -411,7 +420,7 @@ func (r *Repository) ListTasksByWorkspace(ctx context.Context, workspaceID, work } if excludeConfig { - filter += " AND (metadata IS NULL OR json_extract(metadata, '$.config_mode') IS NOT 1)" + filter += " AND " + excludeConfigModePredicate(r.ro.DriverName(), "metadata") } var rows *sql.Rows @@ -544,7 +553,7 @@ func (r *Repository) searchTasks(ctx context.Context, workspaceID, query, filter tFilter += " AND t.archived_at IS NULL" } if excludeConfig { - tFilter += " AND (t.metadata IS NULL OR json_extract(t.metadata, '$.config_mode') IS NOT 1)" + tFilter += " AND " + excludeConfigModePredicate(r.ro.DriverName(), "t.metadata") } // Collect extra filter args in query-argument order @@ -794,12 +803,12 @@ func (r *Repository) CountOpenWatcherCreatedTasks(ctx context.Context, metadataK if !isSafeMetadataKey(metadataKey) { return 0, fmt.Errorf("invalid watcher metadata key %q", metadataKey) } - query := r.ro.Rebind(` + query := r.ro.Rebind(fmt.Sprintf(` SELECT COUNT(*) FROM tasks WHERE archived_at IS NULL AND state NOT IN (?, ?, ?) - AND json_extract(metadata, '$.` + metadataKey + `') = ? - `) + AND %s = ? + `, dialect.JSONExtract(r.ro.DriverName(), "metadata", metadataKey))) var n int if err := r.ro.QueryRowxContext(ctx, query, v1.TaskStateCompleted, v1.TaskStateFailed, v1.TaskStateCancelled, watchID, diff --git a/apps/backend/internal/testutil/postgres.go b/apps/backend/internal/testutil/postgres.go new file mode 100644 index 0000000000..3290484c15 --- /dev/null +++ b/apps/backend/internal/testutil/postgres.go @@ -0,0 +1,50 @@ +package testutil + +import ( + "os" + "strings" + "testing" + + "github.com/google/uuid" + "github.com/jmoiron/sqlx" + + _ "github.com/jackc/pgx/v5/stdlib" +) + +// OpenIsolatedPostgres opens dsn with a unique schema on a single connection. +// It lets package tests share one Postgres database without racing on +// DROP SCHEMA public when Go runs packages in parallel. +func OpenIsolatedPostgres(t testing.TB, dsn string) *sqlx.DB { + t.Helper() + + schema := "kandev_test_" + strings.ReplaceAll(uuid.NewString(), "-", "") + db, err := sqlx.Open("pgx", dsn) + if err != nil { + t.Fatalf("open postgres: %v", err) + } + db.SetMaxOpenConns(1) + db.SetMaxIdleConns(1) + + if _, err := db.Exec("CREATE SCHEMA " + schema); err != nil { + _ = db.Close() + t.Fatalf("create postgres schema %s: %v", schema, err) + } + t.Cleanup(func() { + _, _ = db.Exec("DROP SCHEMA IF EXISTS " + schema + " CASCADE") + _ = db.Close() + }) + + if _, err := db.Exec("SET search_path TO " + schema); err != nil { + t.Fatalf("set postgres search_path %s: %v", schema, err) + } + return db +} + +func PostgresDSNFromEnv(t testing.TB) string { + t.Helper() + dsn := os.Getenv("KANDEV_TEST_POSTGRES_DSN") + if dsn == "" { + t.Skip("set KANDEV_TEST_POSTGRES_DSN to run Postgres tests") + } + return dsn +}