fix: initialize postgres schemas (#1365)

This commit is contained in:
Carlos Florêncio
2026-06-14 17:47:40 +01:00
committed by GitHub
parent a897c67499
commit 66871a6282
19 changed files with 417 additions and 117 deletions
@@ -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)
}
}
@@ -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'`,
@@ -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')" {
@@ -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"
}
@@ -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);
@@ -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
@@ -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.
@@ -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
);
@@ -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
);
+3 -3
View File
@@ -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 {
@@ -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)
}
}
@@ -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)")
@@ -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)
}
}
+18 -15
View File
@@ -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)
}
@@ -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
@@ -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
}
@@ -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)
}
}
@@ -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,
@@ -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
}