kenn-io--agentsview/internal/duckdb/schema.go

package duckdb

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"sort"
	"strconv"
	"strings"

	"go.kenn.io/agentsview/internal/export"
)

// SchemaVersion is the version of the DuckDB mirror schema created by
// EnsureSchema. Increment it when a non-optional DuckDB column/table is added.
const SchemaVersion = 2

const schemaVersionMetadataKey = "agentsview_schema_version"
const defaultRepairMetadataKey = "agentsview_default_repair_v1"
const usageDedupIndexMetadataKey = "agentsview_usage_dedup_index_v1"
const projectIdentityRemoteScrubMetadataKey = "agentsview_project_identity_remote_scrub_v1"

// DuckDB schema notes:
//
//   - DuckDB stores timestamps as TIMESTAMP for mirror tables; read queries
//     should cast/format them to text when scanning into db.Session/db.Message.
//   - DuckDB BOOLEAN columns scan into Go bools directly, unlike SQLite's
//     integer booleans.
//   - SQLite INTEGER PRIMARY KEY rowids are mirrored as BIGINT values because
//     DuckDB does not provide SQLite rowid/autoincrement semantics.
//   - DuckDB does not support SQLite FTS5 or PostgreSQL GIN indexes here; text
//     search optimization is handled separately from this compatibility schema.
//   - DuckDB Quack currently rejects catalogs with TIMESTAMP DEFAULT
//     current_timestamp columns, so mirror timestamp columns avoid dynamic
//     defaults and writers supply current_timestamp explicitly where needed.

type tableSpec struct {
	name    string
	create  string
	columns []columnSpec
	indexes []string
}

type columnSpec struct {
	name string
	def  string
}

type timestampDefaultSpec struct {
	table  string
	column string
}

var quackIncompatibleTimestampDefaults = []timestampDefaultSpec{
	{"sessions", "created_at"},
	{"secret_findings", "created_at"},
	{"starred_sessions", "created_at"},
	{"pinned_messages", "created_at"},
}

var mirrorTables = []tableSpec{
	{
		name: "sync_metadata",
		create: `CREATE TABLE IF NOT EXISTS sync_metadata (
			key TEXT PRIMARY KEY,
			value TEXT NOT NULL
		)`,
		columns: []columnSpec{
			{"key", "key TEXT"},
			{"value", "value TEXT NOT NULL DEFAULT ''"},
		},
	},
	{
		name: "source_archives",
		create: `CREATE TABLE IF NOT EXISTS source_archives (
			source_archive_id TEXT PRIMARY KEY,
			source_archive_salt TEXT NOT NULL
		)`,
		columns: []columnSpec{
			{"source_archive_id", "source_archive_id TEXT"},
			{"source_archive_salt", "source_archive_salt TEXT NOT NULL DEFAULT ''"},
		},
	},
	{
		name: "sessions",
		create: `CREATE TABLE IF NOT EXISTS sessions (
			id TEXT PRIMARY KEY,
			project TEXT NOT NULL,
			machine TEXT NOT NULL DEFAULT 'local',
			agent TEXT NOT NULL DEFAULT 'claude',
			first_message TEXT,
			display_name TEXT,
			session_name TEXT,
			started_at TIMESTAMP,
			ended_at TIMESTAMP,
			message_count INTEGER NOT NULL DEFAULT 0,
			user_message_count INTEGER NOT NULL DEFAULT 0,
			file_path TEXT,
			file_size BIGINT,
			file_mtime BIGINT,
			file_inode BIGINT,
			file_device BIGINT,
			file_hash TEXT,
			local_modified_at TIMESTAMP,
			transcript_revision TEXT NOT NULL DEFAULT '0',
			parent_session_id TEXT,
			relationship_type TEXT NOT NULL DEFAULT '',
			total_output_tokens INTEGER NOT NULL DEFAULT 0,
			peak_context_tokens INTEGER NOT NULL DEFAULT 0,
			has_total_output_tokens BOOLEAN NOT NULL DEFAULT FALSE,
			has_peak_context_tokens BOOLEAN NOT NULL DEFAULT FALSE,
			is_automated BOOLEAN NOT NULL DEFAULT FALSE,
			tool_failure_signal_count INTEGER NOT NULL DEFAULT 0,
			tool_retry_count INTEGER NOT NULL DEFAULT 0,
			edit_churn_count INTEGER NOT NULL DEFAULT 0,
			consecutive_failure_max INTEGER NOT NULL DEFAULT 0,
			outcome TEXT NOT NULL DEFAULT 'unknown',
			outcome_confidence TEXT NOT NULL DEFAULT 'low',
			ended_with_role TEXT NOT NULL DEFAULT '',
			final_failure_streak INTEGER NOT NULL DEFAULT 0,
			signals_pending_since TEXT,
			compaction_count INTEGER NOT NULL DEFAULT 0,
			mid_task_compaction_count INTEGER NOT NULL DEFAULT 0,
			context_pressure_max DOUBLE,
			health_score INTEGER,
			health_grade TEXT,
			has_tool_calls BOOLEAN NOT NULL DEFAULT FALSE,
			has_context_data BOOLEAN NOT NULL DEFAULT FALSE,
			quality_signal_version INTEGER NOT NULL DEFAULT 0,
			short_prompt_count INTEGER NOT NULL DEFAULT 0,
			unstructured_start BOOLEAN NOT NULL DEFAULT FALSE,
			missing_success_criteria_count INTEGER NOT NULL DEFAULT 0,
			missing_verification_count INTEGER NOT NULL DEFAULT 0,
			duplicate_prompt_count INTEGER NOT NULL DEFAULT 0,
			no_code_context_count INTEGER NOT NULL DEFAULT 0,
			runaway_tool_loop_count INTEGER NOT NULL DEFAULT 0,
			data_version INTEGER NOT NULL DEFAULT 0,
			cwd TEXT NOT NULL DEFAULT '',
			git_branch TEXT NOT NULL DEFAULT '',
			source_session_id TEXT NOT NULL DEFAULT '',
			source_version TEXT NOT NULL DEFAULT '',
			transcript_fidelity TEXT NOT NULL DEFAULT '',
			parser_malformed_lines INTEGER NOT NULL DEFAULT 0,
			is_truncated BOOLEAN NOT NULL DEFAULT FALSE,
			deleted_at TIMESTAMP,
			created_at TIMESTAMP,
			termination_status TEXT,
			secret_leak_count INTEGER NOT NULL DEFAULT 0,
			secrets_rules_version TEXT NOT NULL DEFAULT ''
		)`,
		columns: []columnSpec{
			{"id", "id TEXT"},
			{"project", "project TEXT NOT NULL DEFAULT ''"},
			{"machine", "machine TEXT NOT NULL DEFAULT 'local'"},
			{"agent", "agent TEXT NOT NULL DEFAULT 'claude'"},
			{"first_message", "first_message TEXT"},
			{"display_name", "display_name TEXT"},
			{"session_name", "session_name TEXT"},
			{"started_at", "started_at TIMESTAMP"},
			{"ended_at", "ended_at TIMESTAMP"},
			{"message_count", "message_count INTEGER NOT NULL DEFAULT 0"},
			{"user_message_count", "user_message_count INTEGER NOT NULL DEFAULT 0"},
			{"file_path", "file_path TEXT"},
			{"file_size", "file_size BIGINT"},
			{"file_mtime", "file_mtime BIGINT"},
			{"file_inode", "file_inode BIGINT"},
			{"file_device", "file_device BIGINT"},
			{"file_hash", "file_hash TEXT"},
			{"local_modified_at", "local_modified_at TIMESTAMP"},
			{"transcript_revision", "transcript_revision TEXT NOT NULL DEFAULT '0'"},
			{"parent_session_id", "parent_session_id TEXT"},
			{"relationship_type", "relationship_type TEXT NOT NULL DEFAULT ''"},
			{"total_output_tokens", "total_output_tokens INTEGER NOT NULL DEFAULT 0"},
			{"peak_context_tokens", "peak_context_tokens INTEGER NOT NULL DEFAULT 0"},
			{"has_total_output_tokens", "has_total_output_tokens BOOLEAN NOT NULL DEFAULT FALSE"},
			{"has_peak_context_tokens", "has_peak_context_tokens BOOLEAN NOT NULL DEFAULT FALSE"},
			{"is_automated", "is_automated BOOLEAN NOT NULL DEFAULT FALSE"},
			{"tool_failure_signal_count", "tool_failure_signal_count INTEGER NOT NULL DEFAULT 0"},
			{"tool_retry_count", "tool_retry_count INTEGER NOT NULL DEFAULT 0"},
			{"edit_churn_count", "edit_churn_count INTEGER NOT NULL DEFAULT 0"},
			{"consecutive_failure_max", "consecutive_failure_max INTEGER NOT NULL DEFAULT 0"},
			{"outcome", "outcome TEXT NOT NULL DEFAULT 'unknown'"},
			{"outcome_confidence", "outcome_confidence TEXT NOT NULL DEFAULT 'low'"},
			{"ended_with_role", "ended_with_role TEXT NOT NULL DEFAULT ''"},
			{"final_failure_streak", "final_failure_streak INTEGER NOT NULL DEFAULT 0"},
			{"signals_pending_since", "signals_pending_since TEXT"},
			{"compaction_count", "compaction_count INTEGER NOT NULL DEFAULT 0"},
			{"mid_task_compaction_count", "mid_task_compaction_count INTEGER NOT NULL DEFAULT 0"},
			{"context_pressure_max", "context_pressure_max DOUBLE"},
			{"health_score", "health_score INTEGER"},
			{"health_grade", "health_grade TEXT"},
			{"has_tool_calls", "has_tool_calls BOOLEAN NOT NULL DEFAULT FALSE"},
			{"has_context_data", "has_context_data BOOLEAN NOT NULL DEFAULT FALSE"},
			{"quality_signal_version", "quality_signal_version INTEGER NOT NULL DEFAULT 0"},
			{"short_prompt_count", "short_prompt_count INTEGER NOT NULL DEFAULT 0"},
			{"unstructured_start", "unstructured_start BOOLEAN NOT NULL DEFAULT FALSE"},
			{"missing_success_criteria_count", "missing_success_criteria_count INTEGER NOT NULL DEFAULT 0"},
			{"missing_verification_count", "missing_verification_count INTEGER NOT NULL DEFAULT 0"},
			{"duplicate_prompt_count", "duplicate_prompt_count INTEGER NOT NULL DEFAULT 0"},
			{"no_code_context_count", "no_code_context_count INTEGER NOT NULL DEFAULT 0"},
			{"runaway_tool_loop_count", "runaway_tool_loop_count INTEGER NOT NULL DEFAULT 0"},
			{"data_version", "data_version INTEGER NOT NULL DEFAULT 0"},
			{"cwd", "cwd TEXT NOT NULL DEFAULT ''"},
			{"git_branch", "git_branch TEXT NOT NULL DEFAULT ''"},
			{"source_session_id", "source_session_id TEXT NOT NULL DEFAULT ''"},
			{"source_version", "source_version TEXT NOT NULL DEFAULT ''"},
			{"transcript_fidelity", "transcript_fidelity TEXT NOT NULL DEFAULT ''"},
			{"parser_malformed_lines", "parser_malformed_lines INTEGER NOT NULL DEFAULT 0"},
			{"is_truncated", "is_truncated BOOLEAN NOT NULL DEFAULT FALSE"},
			{"deleted_at", "deleted_at TIMESTAMP"},
			{"created_at", "created_at TIMESTAMP"},
			{"termination_status", "termination_status TEXT"},
			{"secret_leak_count", "secret_leak_count INTEGER NOT NULL DEFAULT 0"},
			{"secrets_rules_version", "secrets_rules_version TEXT NOT NULL DEFAULT ''"},
		},
		indexes: []string{
			"CREATE INDEX IF NOT EXISTS idx_sessions_ended ON sessions(ended_at, id)",
			"CREATE INDEX IF NOT EXISTS idx_sessions_project ON sessions(project)",
			"CREATE INDEX IF NOT EXISTS idx_sessions_machine ON sessions(machine)",
			"CREATE INDEX IF NOT EXISTS idx_sessions_parent ON sessions(parent_session_id)",
			"CREATE INDEX IF NOT EXISTS idx_sessions_started ON sessions(started_at)",
			"CREATE INDEX IF NOT EXISTS idx_sessions_agent ON sessions(agent)",
			"CREATE INDEX IF NOT EXISTS idx_sessions_termination_status ON sessions(termination_status)",
		},
	},
	{
		name: "messages",
		create: `CREATE TABLE IF NOT EXISTS messages (
			id BIGINT,
			session_id TEXT NOT NULL,
			ordinal INTEGER NOT NULL,
			role TEXT NOT NULL,
			content TEXT NOT NULL,
			thinking_text TEXT NOT NULL DEFAULT '',
			timestamp TIMESTAMP,
			has_thinking BOOLEAN NOT NULL DEFAULT FALSE,
			has_tool_use BOOLEAN NOT NULL DEFAULT FALSE,
			content_length INTEGER NOT NULL DEFAULT 0,
			is_system BOOLEAN NOT NULL DEFAULT FALSE,
			model TEXT NOT NULL DEFAULT '',
			token_usage TEXT NOT NULL DEFAULT '',
			context_tokens INTEGER NOT NULL DEFAULT 0,
			output_tokens INTEGER NOT NULL DEFAULT 0,
			has_context_tokens BOOLEAN NOT NULL DEFAULT FALSE,
			has_output_tokens BOOLEAN NOT NULL DEFAULT FALSE,
			claude_message_id TEXT NOT NULL DEFAULT '',
			claude_request_id TEXT NOT NULL DEFAULT '',
			source_type TEXT NOT NULL DEFAULT '',
			source_subtype TEXT NOT NULL DEFAULT '',
			source_uuid TEXT NOT NULL DEFAULT '',
			source_parent_uuid TEXT NOT NULL DEFAULT '',
			is_sidechain BOOLEAN NOT NULL DEFAULT FALSE,
			is_compact_boundary BOOLEAN NOT NULL DEFAULT FALSE,
			UNIQUE(session_id, ordinal)
		)`,
		columns: []columnSpec{
			{"id", "id BIGINT"},
			{"session_id", "session_id TEXT NOT NULL DEFAULT ''"},
			{"ordinal", "ordinal INTEGER NOT NULL DEFAULT 0"},
			{"role", "role TEXT NOT NULL DEFAULT ''"},
			{"content", "content TEXT NOT NULL DEFAULT ''"},
			{"thinking_text", "thinking_text TEXT NOT NULL DEFAULT ''"},
			{"timestamp", "timestamp TIMESTAMP"},
			{"has_thinking", "has_thinking BOOLEAN NOT NULL DEFAULT FALSE"},
			{"has_tool_use", "has_tool_use BOOLEAN NOT NULL DEFAULT FALSE"},
			{"content_length", "content_length INTEGER NOT NULL DEFAULT 0"},
			{"is_system", "is_system BOOLEAN NOT NULL DEFAULT FALSE"},
			{"model", "model TEXT NOT NULL DEFAULT ''"},
			{"token_usage", "token_usage TEXT NOT NULL DEFAULT ''"},
			{"context_tokens", "context_tokens INTEGER NOT NULL DEFAULT 0"},
			{"output_tokens", "output_tokens INTEGER NOT NULL DEFAULT 0"},
			{"has_context_tokens", "has_context_tokens BOOLEAN NOT NULL DEFAULT FALSE"},
			{"has_output_tokens", "has_output_tokens BOOLEAN NOT NULL DEFAULT FALSE"},
			{"claude_message_id", "claude_message_id TEXT NOT NULL DEFAULT ''"},
			{"claude_request_id", "claude_request_id TEXT NOT NULL DEFAULT ''"},
			{"source_type", "source_type TEXT NOT NULL DEFAULT ''"},
			{"source_subtype", "source_subtype TEXT NOT NULL DEFAULT ''"},
			{"source_uuid", "source_uuid TEXT NOT NULL DEFAULT ''"},
			{"source_parent_uuid", "source_parent_uuid TEXT NOT NULL DEFAULT ''"},
			{"is_sidechain", "is_sidechain BOOLEAN NOT NULL DEFAULT FALSE"},
			{"is_compact_boundary", "is_compact_boundary BOOLEAN NOT NULL DEFAULT FALSE"},
		},
		indexes: []string{
			"CREATE INDEX IF NOT EXISTS idx_messages_session_ordinal ON messages(session_id, ordinal)",
			"CREATE INDEX IF NOT EXISTS idx_messages_session_role ON messages(session_id, role)",
			"CREATE INDEX IF NOT EXISTS idx_messages_timestamp ON messages(timestamp)",
		},
	},
	{
		name: "usage_events",
		create: `CREATE TABLE IF NOT EXISTS usage_events (
			id BIGINT,
			session_id TEXT NOT NULL,
			message_ordinal INTEGER,
			source TEXT NOT NULL,
			model TEXT NOT NULL,
			input_tokens INTEGER NOT NULL DEFAULT 0,
			output_tokens INTEGER NOT NULL DEFAULT 0,
			cache_creation_input_tokens INTEGER NOT NULL DEFAULT 0,
			cache_read_input_tokens INTEGER NOT NULL DEFAULT 0,
			reasoning_tokens INTEGER NOT NULL DEFAULT 0,
			cost_usd DOUBLE,
			cost_status TEXT NOT NULL DEFAULT '',
			cost_source TEXT NOT NULL DEFAULT '',
			occurred_at TIMESTAMP,
			dedup_key TEXT NOT NULL DEFAULT ''
		)`,
		columns: []columnSpec{
			{"id", "id BIGINT"},
			{"session_id", "session_id TEXT NOT NULL DEFAULT ''"},
			{"message_ordinal", "message_ordinal INTEGER"},
			{"source", "source TEXT NOT NULL DEFAULT ''"},
			{"model", "model TEXT NOT NULL DEFAULT ''"},
			{"input_tokens", "input_tokens INTEGER NOT NULL DEFAULT 0"},
			{"output_tokens", "output_tokens INTEGER NOT NULL DEFAULT 0"},
			{"cache_creation_input_tokens", "cache_creation_input_tokens INTEGER NOT NULL DEFAULT 0"},
			{"cache_read_input_tokens", "cache_read_input_tokens INTEGER NOT NULL DEFAULT 0"},
			{"reasoning_tokens", "reasoning_tokens INTEGER NOT NULL DEFAULT 0"},
			{"cost_usd", "cost_usd DOUBLE"},
			{"cost_status", "cost_status TEXT NOT NULL DEFAULT ''"},
			{"cost_source", "cost_source TEXT NOT NULL DEFAULT ''"},
			{"occurred_at", "occurred_at TIMESTAMP"},
			{"dedup_key", "dedup_key TEXT NOT NULL DEFAULT ''"},
		},
		indexes: []string{
			"CREATE INDEX IF NOT EXISTS idx_usage_events_session ON usage_events(session_id)",
			"CREATE INDEX IF NOT EXISTS idx_usage_events_occurred ON usage_events(occurred_at)",
			"CREATE INDEX IF NOT EXISTS idx_usage_events_dedup ON usage_events(session_id, source, dedup_key)",
		},
	},
	{
		name: "cursor_usage_events",
		create: `CREATE TABLE IF NOT EXISTS cursor_usage_events (
			id BIGINT,
			occurred_at TIMESTAMP NOT NULL,
			model TEXT NOT NULL,
			kind TEXT NOT NULL DEFAULT '',
			input_tokens INTEGER NOT NULL DEFAULT 0,
			output_tokens INTEGER NOT NULL DEFAULT 0,
			cache_write_tokens INTEGER NOT NULL DEFAULT 0,
			cache_read_tokens INTEGER NOT NULL DEFAULT 0,
			charged_cents DOUBLE NOT NULL DEFAULT 0,
			cursor_token_fee DOUBLE NOT NULL DEFAULT 0,
			user_id TEXT NOT NULL DEFAULT '',
			user_email TEXT NOT NULL DEFAULT '',
			is_headless BOOLEAN NOT NULL DEFAULT FALSE,
			dedup_key TEXT NOT NULL DEFAULT ''
		)`,
		columns: []columnSpec{
			{"id", "id BIGINT"},
			{"occurred_at", "occurred_at TIMESTAMP NOT NULL"},
			{"model", "model TEXT NOT NULL DEFAULT ''"},
			{"kind", "kind TEXT NOT NULL DEFAULT ''"},
			{"input_tokens", "input_tokens INTEGER NOT NULL DEFAULT 0"},
			{"output_tokens", "output_tokens INTEGER NOT NULL DEFAULT 0"},
			{"cache_write_tokens", "cache_write_tokens INTEGER NOT NULL DEFAULT 0"},
			{"cache_read_tokens", "cache_read_tokens INTEGER NOT NULL DEFAULT 0"},
			{"charged_cents", "charged_cents DOUBLE NOT NULL DEFAULT 0"},
			{"cursor_token_fee", "cursor_token_fee DOUBLE NOT NULL DEFAULT 0"},
			{"user_id", "user_id TEXT NOT NULL DEFAULT ''"},
			{"user_email", "user_email TEXT NOT NULL DEFAULT ''"},
			{"is_headless", "is_headless BOOLEAN NOT NULL DEFAULT FALSE"},
			{"dedup_key", "dedup_key TEXT NOT NULL DEFAULT ''"},
		},
		indexes: []string{
			"CREATE UNIQUE INDEX IF NOT EXISTS idx_cursor_usage_events_dedup ON cursor_usage_events(dedup_key)",
			"CREATE INDEX IF NOT EXISTS idx_cursor_usage_events_occurred ON cursor_usage_events(occurred_at)",
			"CREATE INDEX IF NOT EXISTS idx_cursor_usage_events_model ON cursor_usage_events(model)",
		},
	},
	{
		name: "model_pricing",
		create: `CREATE TABLE IF NOT EXISTS model_pricing (
			model_pattern TEXT PRIMARY KEY,
			input_per_mtok DOUBLE NOT NULL DEFAULT 0,
			output_per_mtok DOUBLE NOT NULL DEFAULT 0,
			cache_creation_per_mtok DOUBLE NOT NULL DEFAULT 0,
			cache_read_per_mtok DOUBLE NOT NULL DEFAULT 0,
			updated_at TEXT NOT NULL DEFAULT ''
		)`,
		columns: []columnSpec{
			{"model_pattern", "model_pattern TEXT"},
			{"input_per_mtok", "input_per_mtok DOUBLE NOT NULL DEFAULT 0"},
			{"output_per_mtok", "output_per_mtok DOUBLE NOT NULL DEFAULT 0"},
			{"cache_creation_per_mtok", "cache_creation_per_mtok DOUBLE NOT NULL DEFAULT 0"},
			{"cache_read_per_mtok", "cache_read_per_mtok DOUBLE NOT NULL DEFAULT 0"},
			{"updated_at", "updated_at TEXT NOT NULL DEFAULT ''"},
		},
	},
	{
		name: "source_project_identity_observations",
		create: `CREATE TABLE IF NOT EXISTS source_project_identity_observations (
			source_archive_id TEXT NOT NULL DEFAULT '',
			source_archive_salt TEXT NOT NULL DEFAULT '',
			project TEXT NOT NULL,
			machine TEXT NOT NULL,
			root_path TEXT NOT NULL DEFAULT '',
			git_remote TEXT NOT NULL DEFAULT '',
			git_remote_name TEXT NOT NULL DEFAULT '',
			repository_path TEXT NOT NULL DEFAULT '',
			worktree_name TEXT NOT NULL DEFAULT '',
			worktree_root_path TEXT NOT NULL DEFAULT '',
			worktree_relationship TEXT NOT NULL DEFAULT 'unknown',
			checkout_state TEXT NOT NULL DEFAULT 'unknown',
			git_branch TEXT NOT NULL DEFAULT '',
			remote_resolution TEXT NOT NULL DEFAULT 'unknown',
			remote_candidate_count INTEGER NOT NULL DEFAULT 0,
			observed_at TIMESTAMP NOT NULL,
			normalized_remote TEXT NOT NULL DEFAULT '',
			key_source TEXT NOT NULL DEFAULT '',
			key TEXT NOT NULL DEFAULT '',
			PRIMARY KEY (source_archive_id, project, machine, root_path, git_remote)
		)`,
		columns: []columnSpec{
			{"source_archive_id", "source_archive_id TEXT NOT NULL DEFAULT ''"},
			{"source_archive_salt", "source_archive_salt TEXT NOT NULL DEFAULT ''"},
			{"project", "project TEXT NOT NULL DEFAULT ''"},
			{"machine", "machine TEXT NOT NULL DEFAULT ''"},
			{"root_path", "root_path TEXT NOT NULL DEFAULT ''"},
			{"git_remote", "git_remote TEXT NOT NULL DEFAULT ''"},
			{"git_remote_name", "git_remote_name TEXT NOT NULL DEFAULT ''"},
			{"repository_path", "repository_path TEXT NOT NULL DEFAULT ''"},
			{"worktree_name", "worktree_name TEXT NOT NULL DEFAULT ''"},
			{"worktree_root_path", "worktree_root_path TEXT NOT NULL DEFAULT ''"},
			{"worktree_relationship", "worktree_relationship TEXT NOT NULL DEFAULT 'unknown'"},
			{"checkout_state", "checkout_state TEXT NOT NULL DEFAULT 'unknown'"},
			{"git_branch", "git_branch TEXT NOT NULL DEFAULT ''"},
			{"remote_resolution", "remote_resolution TEXT NOT NULL DEFAULT 'unknown'"},
			{"remote_candidate_count", "remote_candidate_count INTEGER NOT NULL DEFAULT 0"},
			{"observed_at", "observed_at TIMESTAMP"},
			{"normalized_remote", "normalized_remote TEXT NOT NULL DEFAULT ''"},
			{"key_source", "key_source TEXT NOT NULL DEFAULT ''"},
			{"key", "key TEXT NOT NULL DEFAULT ''"},
		},
		indexes: []string{
			"CREATE INDEX IF NOT EXISTS idx_source_project_identity_observations_project ON source_project_identity_observations(project)",
		},
	},
	{
		name: "source_session_project_identity_snapshots",
		create: `CREATE TABLE IF NOT EXISTS source_session_project_identity_snapshots (
			source_archive_id TEXT NOT NULL,
			source_database_generation TEXT NOT NULL,
			source_session_id TEXT NOT NULL,
			project TEXT NOT NULL,
			machine TEXT NOT NULL,
			root_path TEXT NOT NULL DEFAULT '',
			git_remote TEXT NOT NULL DEFAULT '',
			git_remote_name TEXT NOT NULL DEFAULT '',
			repository_path TEXT NOT NULL DEFAULT '',
			worktree_name TEXT NOT NULL DEFAULT '',
			worktree_root_path TEXT NOT NULL DEFAULT '',
			worktree_relationship TEXT NOT NULL DEFAULT 'unknown',
			checkout_state TEXT NOT NULL DEFAULT 'unknown',
			git_branch TEXT NOT NULL DEFAULT '',
			remote_resolution TEXT NOT NULL DEFAULT 'unknown',
			remote_candidate_count INTEGER NOT NULL DEFAULT 0,
			observed_at TIMESTAMP NOT NULL,
			normalized_remote TEXT NOT NULL DEFAULT '',
			key_source TEXT NOT NULL DEFAULT '',
			key TEXT NOT NULL DEFAULT '',
			PRIMARY KEY (
				source_archive_id, source_database_generation, source_session_id
			)
		)`,
		columns: []columnSpec{
			{"source_archive_id", "source_archive_id TEXT NOT NULL DEFAULT ''"},
			{"source_database_generation", "source_database_generation TEXT NOT NULL DEFAULT ''"},
			{"source_session_id", "source_session_id TEXT NOT NULL DEFAULT ''"},
			{"project", "project TEXT NOT NULL DEFAULT ''"},
			{"machine", "machine TEXT NOT NULL DEFAULT ''"},
			{"root_path", "root_path TEXT NOT NULL DEFAULT ''"},
			{"git_remote", "git_remote TEXT NOT NULL DEFAULT ''"},
			{"git_remote_name", "git_remote_name TEXT NOT NULL DEFAULT ''"},
			{"repository_path", "repository_path TEXT NOT NULL DEFAULT ''"},
			{"worktree_name", "worktree_name TEXT NOT NULL DEFAULT ''"},
			{"worktree_root_path", "worktree_root_path TEXT NOT NULL DEFAULT ''"},
			{"worktree_relationship", "worktree_relationship TEXT NOT NULL DEFAULT 'unknown'"},
			{"checkout_state", "checkout_state TEXT NOT NULL DEFAULT 'unknown'"},
			{"git_branch", "git_branch TEXT NOT NULL DEFAULT ''"},
			{"remote_resolution", "remote_resolution TEXT NOT NULL DEFAULT 'unknown'"},
			{"remote_candidate_count", "remote_candidate_count INTEGER NOT NULL DEFAULT 0"},
			{"observed_at", "observed_at TIMESTAMP"},
			{"normalized_remote", "normalized_remote TEXT NOT NULL DEFAULT ''"},
			{"key_source", "key_source TEXT NOT NULL DEFAULT ''"},
			{"key", "key TEXT NOT NULL DEFAULT ''"},
		},
		indexes: []string{
			"CREATE INDEX IF NOT EXISTS idx_source_session_project_identity_snapshots_project ON source_session_project_identity_snapshots(source_archive_id, project)",
		},
	},
	{
		name: "tool_calls",
		create: `CREATE TABLE IF NOT EXISTS tool_calls (
			id BIGINT,
			message_id BIGINT,
			session_id TEXT NOT NULL,
			tool_name TEXT NOT NULL,
			category TEXT NOT NULL,
			call_index INTEGER NOT NULL DEFAULT 0,
			tool_use_id TEXT NOT NULL DEFAULT '',
			input_json TEXT,
			skill_name TEXT,
			result_content_length INTEGER,
			result_content TEXT,
			subagent_session_id TEXT,
			file_path TEXT
		)`,
		columns: []columnSpec{
			{"id", "id BIGINT"},
			{"message_id", "message_id BIGINT"},
			{"session_id", "session_id TEXT NOT NULL DEFAULT ''"},
			{"tool_name", "tool_name TEXT NOT NULL DEFAULT ''"},
			{"category", "category TEXT NOT NULL DEFAULT ''"},
			{"call_index", "call_index INTEGER NOT NULL DEFAULT 0"},
			{"tool_use_id", "tool_use_id TEXT NOT NULL DEFAULT ''"},
			{"input_json", "input_json TEXT"},
			{"skill_name", "skill_name TEXT"},
			{"result_content_length", "result_content_length INTEGER"},
			{"result_content", "result_content TEXT"},
			{"subagent_session_id", "subagent_session_id TEXT"},
			{"file_path", "file_path TEXT"},
		},
		indexes: []string{
			"CREATE UNIQUE INDEX IF NOT EXISTS idx_tool_calls_dedup ON tool_calls(session_id, message_id, call_index)",
			"CREATE INDEX IF NOT EXISTS idx_tool_calls_session ON tool_calls(session_id)",
			"CREATE INDEX IF NOT EXISTS idx_tool_calls_message ON tool_calls(message_id)",
			"CREATE INDEX IF NOT EXISTS idx_tool_calls_category ON tool_calls(category)",
			// DuckDB has no partial indexes, so this mirrors SQLite's
			// idx_tool_calls_file_path without the WHERE file_path IS NOT NULL
			// clause; it backs the cross-session Recent Edits feed.
			"CREATE INDEX IF NOT EXISTS idx_tool_calls_file_path ON tool_calls(file_path)",
		},
	},
	{
		name: "tool_result_events",
		create: `CREATE TABLE IF NOT EXISTS tool_result_events (
			id BIGINT,
			session_id TEXT NOT NULL,
			tool_call_message_ordinal INTEGER NOT NULL,
			call_index INTEGER NOT NULL DEFAULT 0,
			tool_use_id TEXT,
			agent_id TEXT,
			subagent_session_id TEXT,
			source TEXT NOT NULL,
			status TEXT NOT NULL,
			content TEXT NOT NULL,
			content_length INTEGER NOT NULL DEFAULT 0,
			timestamp TIMESTAMP,
			event_index INTEGER NOT NULL DEFAULT 0
		)`,
		columns: []columnSpec{
			{"id", "id BIGINT"},
			{"session_id", "session_id TEXT NOT NULL DEFAULT ''"},
			{"tool_call_message_ordinal", "tool_call_message_ordinal INTEGER NOT NULL DEFAULT 0"},
			{"call_index", "call_index INTEGER NOT NULL DEFAULT 0"},
			{"tool_use_id", "tool_use_id TEXT"},
			{"agent_id", "agent_id TEXT"},
			{"subagent_session_id", "subagent_session_id TEXT"},
			{"source", "source TEXT NOT NULL DEFAULT ''"},
			{"status", "status TEXT NOT NULL DEFAULT ''"},
			{"content", "content TEXT NOT NULL DEFAULT ''"},
			{"content_length", "content_length INTEGER NOT NULL DEFAULT 0"},
			{"timestamp", "timestamp TIMESTAMP"},
			{"event_index", "event_index INTEGER NOT NULL DEFAULT 0"},
		},
		indexes: []string{
			"CREATE INDEX IF NOT EXISTS idx_tool_result_events_session ON tool_result_events(session_id)",
			"CREATE UNIQUE INDEX IF NOT EXISTS idx_tool_result_events_dedup ON tool_result_events(session_id, tool_call_message_ordinal, call_index, event_index)",
		},
	},
	{
		name: "secret_findings",
		create: `CREATE TABLE IF NOT EXISTS secret_findings (
			id BIGINT,
			session_id TEXT NOT NULL,
			rule_name TEXT NOT NULL,
			confidence TEXT NOT NULL,
			location_kind TEXT NOT NULL,
			message_ordinal INTEGER NOT NULL,
			call_index INTEGER,
			event_index INTEGER,
			match_start INTEGER NOT NULL,
			match_end INTEGER NOT NULL,
			match_index INTEGER NOT NULL,
			redacted_match TEXT NOT NULL,
			rules_version TEXT NOT NULL,
			created_at TIMESTAMP
		)`,
		columns: []columnSpec{
			{"id", "id BIGINT"},
			{"session_id", "session_id TEXT NOT NULL DEFAULT ''"},
			{"rule_name", "rule_name TEXT NOT NULL DEFAULT ''"},
			{"confidence", "confidence TEXT NOT NULL DEFAULT ''"},
			{"location_kind", "location_kind TEXT NOT NULL DEFAULT ''"},
			{"message_ordinal", "message_ordinal INTEGER NOT NULL DEFAULT 0"},
			{"call_index", "call_index INTEGER"},
			{"event_index", "event_index INTEGER"},
			{"match_start", "match_start INTEGER NOT NULL DEFAULT 0"},
			{"match_end", "match_end INTEGER NOT NULL DEFAULT 0"},
			{"match_index", "match_index INTEGER NOT NULL DEFAULT 0"},
			{"redacted_match", "redacted_match TEXT NOT NULL DEFAULT ''"},
			{"rules_version", "rules_version TEXT NOT NULL DEFAULT ''"},
			{"created_at", "created_at TIMESTAMP"},
		},
		indexes: []string{
			"CREATE INDEX IF NOT EXISTS idx_secret_findings_session ON secret_findings(session_id)",
			"CREATE INDEX IF NOT EXISTS idx_secret_findings_rule ON secret_findings(rule_name)",
		},
	},
	{
		name: "starred_sessions",
		create: `CREATE TABLE IF NOT EXISTS starred_sessions (
			session_id TEXT PRIMARY KEY,
			created_at TIMESTAMP
		)`,
		columns: []columnSpec{
			{"session_id", "session_id TEXT"},
			{"created_at", "created_at TIMESTAMP"},
		},
	},
	{
		name: "pinned_messages",
		create: `CREATE TABLE IF NOT EXISTS pinned_messages (
			id BIGINT,
			session_id TEXT NOT NULL,
			message_id BIGINT NOT NULL,
			ordinal INTEGER NOT NULL,
			source_uuid TEXT NOT NULL DEFAULT '',
			note TEXT,
			created_at TIMESTAMP,
			UNIQUE(session_id, message_id)
		)`,
		columns: []columnSpec{
			{"id", "id BIGINT"},
			{"session_id", "session_id TEXT NOT NULL DEFAULT ''"},
			{"message_id", "message_id BIGINT NOT NULL DEFAULT 0"},
			{"ordinal", "ordinal INTEGER NOT NULL DEFAULT 0"},
			{"source_uuid", "source_uuid TEXT NOT NULL DEFAULT ''"},
			{"note", "note TEXT"},
			{"created_at", "created_at TIMESTAMP"},
		},
		indexes: []string{
			"CREATE INDEX IF NOT EXISTS idx_pinned_session ON pinned_messages(session_id)",
			"CREATE INDEX IF NOT EXISTS idx_pinned_message ON pinned_messages(message_id)",
			"CREATE INDEX IF NOT EXISTS idx_pinned_created ON pinned_messages(created_at)",
		},
	},
}

type schemaOptions struct {
	createIndexes bool
}

// EnsureSchema creates and additively migrates the DuckDB mirror schema.
func EnsureSchema(ctx context.Context, db *sql.DB) error {
	return ensureSchema(ctx, db, schemaOptions{createIndexes: true})
}

func ensureSchema(ctx context.Context, db *sql.DB, opts schemaOptions) error {
	for _, table := range mirrorTables {
		if _, err := db.ExecContext(ctx, table.create); err != nil {
			return fmt.Errorf("creating duckdb table %s: %w", table.name, err)
		}
	}
	if !opts.createIndexes {
		if err := checkSchemaShapeCompat(ctx, db); err != nil {
			return err
		}
		return checkSchemaRepairsViaQuack(ctx, db)
	}

	existing, err := loadColumns(ctx, db)
	if err != nil {
		return err
	}
	defaultRepairDone, err := metadataKeyExists(ctx, db, defaultRepairMetadataKey)
	if err != nil {
		return err
	}
	for _, table := range mirrorTables {
		have := existing[table.name]
		if have == nil {
			have = make(map[string]bool)
			existing[table.name] = have
		}
		for _, column := range table.columns {
			added := false
			if !have[column.name] {
				stmt := fmt.Sprintf(
					"ALTER TABLE %s ADD COLUMN %s",
					table.name, relaxedColumnDef(column.def),
				)
				if _, err := db.ExecContext(ctx, stmt); err != nil {
					return fmt.Errorf(
						"adding duckdb column %s.%s: %w",
						table.name, column.name, err,
					)
				}
				have[column.name] = true
				added = true
			}
			if added || !defaultRepairDone {
				if err := backfillAddedColumnDefault(ctx, db, table.name, column); err != nil {
					return err
				}
			}
		}
	}

	if err := migrateMessagesIDPrimaryKey(ctx, db); err != nil {
		return err
	}

	if err := dropQuackIncompatibleTimestampDefaults(ctx, db); err != nil {
		return err
	}

	projectIdentityRemoteScrubDone, err := metadataKeyExists(
		ctx, db, projectIdentityRemoteScrubMetadataKey,
	)
	if err != nil {
		return err
	}
	if !projectIdentityRemoteScrubDone {
		if err := scrubProjectIdentityGitRemoteCredentials(ctx, db); err != nil {
			return err
		}
	}

	recordUsageDedupIndexMigration, err := migrateUsageEventsDedupIndex(ctx, db)
	if err != nil {
		return err
	}

	for _, table := range mirrorTables {
		for _, stmt := range table.indexes {
			if _, err := db.ExecContext(ctx, stmt); err != nil {
				return fmt.Errorf("creating duckdb index for %s: %w", table.name, err)
			}
		}
	}

	if !defaultRepairDone {
		if err := recordMetadataKey(ctx, db, defaultRepairMetadataKey, "1"); err != nil {
			return err
		}
	}
	if recordUsageDedupIndexMigration {
		if err := recordMetadataKey(ctx, db, usageDedupIndexMetadataKey, "1"); err != nil {
			return err
		}
	}
	if !projectIdentityRemoteScrubDone {
		if err := recordMetadataKey(
			ctx, db, projectIdentityRemoteScrubMetadataKey, "1",
		); err != nil {
			return err
		}
	}
	if err := recordMetadataKey(
		ctx, db, schemaVersionMetadataKey, strconv.Itoa(SchemaVersion),
	); err != nil {
		return fmt.Errorf("recording duckdb schema version: %w", err)
	}
	return nil
}

func migrateUsageEventsDedupIndex(ctx context.Context, db *sql.DB) (bool, error) {
	done, err := metadataKeyExists(ctx, db, usageDedupIndexMetadataKey)
	if err != nil {
		return false, err
	}
	if done {
		return false, nil
	}
	if _, err := db.ExecContext(ctx, `DROP INDEX IF EXISTS idx_usage_events_dedup`); err != nil {
		return false, fmt.Errorf("dropping duckdb usage_events dedup index: %w", err)
	}
	return true, nil
}

func scrubProjectIdentityGitRemoteCredentials(
	ctx context.Context, db *sql.DB,
) error {
	tx, err := db.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf(
			"beginning duckdb project identity remote scrub: %w", err,
		)
	}
	defer func() {
		_ = tx.Rollback()
	}()

	rows, err := tx.QueryContext(ctx, `
		SELECT source_archive_id, source_archive_salt,
			project, machine, root_path, git_remote, git_remote_name,
			repository_path, worktree_name, worktree_root_path,
			worktree_relationship, checkout_state, git_branch,
			remote_resolution, remote_candidate_count, observed_at,
			normalized_remote, key_source, key
		FROM source_project_identity_observations
		WHERE git_remote != ''`)
	if err != nil {
		return fmt.Errorf(
			"listing duckdb project identity remotes for scrub: %w", err,
		)
	}
	rowsClosed := false
	defer func() {
		if !rowsClosed {
			_ = rows.Close()
		}
	}()

	type pendingScrub struct {
		obs       export.ProjectIdentityObservation
		rawRemote string
	}
	var pending []pendingScrub
	for rows.Next() {
		var obs export.ProjectIdentityObservation
		if err := rows.Scan(
			&obs.SourceArchiveID,
			&obs.SourceArchiveSalt,
			&obs.Project,
			&obs.Machine,
			&obs.RootPath,
			&obs.GitRemote,
			&obs.GitRemoteName,
			&obs.RepositoryPath,
			&obs.WorktreeName,
			&obs.WorktreeRootPath,
			&obs.WorktreeRelationship,
			&obs.CheckoutState,
			&obs.GitBranch,
			&obs.RemoteResolution,
			&obs.RemoteCandidateCount,
			&obs.ObservedAt,
			&obs.NormalizedRemote,
			&obs.KeySource,
			&obs.Key,
		); err != nil {
			return fmt.Errorf(
				"scanning duckdb project identity remote for scrub: %w", err,
			)
		}
		sanitized := export.SanitizeGitRemoteForStorage(obs.GitRemote)
		if sanitized == obs.GitRemote {
			continue
		}
		pending = append(pending, pendingScrub{
			obs:       obs,
			rawRemote: obs.GitRemote,
		})
	}
	if err := rows.Err(); err != nil {
		return fmt.Errorf(
			"iterating duckdb project identity remotes for scrub: %w", err,
		)
	}
	if err := rows.Close(); err != nil {
		return fmt.Errorf(
			"closing duckdb project identity remotes scrub rows: %w", err,
		)
	}
	rowsClosed = true

	for _, scrub := range pending {
		obs := export.SanitizeStoredProjectIdentityObservation(scrub.obs)
		if err := upsertProjectIdentityObservation(
			func(stmt string, args ...any) error {
				_, err := tx.ExecContext(ctx, stmt, args...)
				return err
			},
			func(stmt string, args ...any) *sql.Row {
				return tx.QueryRowContext(ctx, stmt, args...)
			},
			obs, scrub.rawRemote,
		); err != nil {
			return fmt.Errorf(
				"upserting scrubbed duckdb project identity remote: %w", err,
			)
		}
		if _, err := tx.ExecContext(ctx, `
			DELETE FROM source_project_identity_observations
			WHERE source_archive_id = ? AND project = ? AND machine = ?
			  AND root_path = ? AND git_remote = ?`,
			scrub.obs.SourceArchiveID, scrub.obs.Project, scrub.obs.Machine,
			scrub.obs.RootPath, scrub.rawRemote,
		); err != nil {
			return fmt.Errorf(
				"removing raw duckdb project identity remote: %w", err,
			)
		}
	}
	if err := tx.Commit(); err != nil {
		return fmt.Errorf(
			"committing duckdb project identity remote scrub: %w", err,
		)
	}
	return nil
}

func migrateMessagesIDPrimaryKey(ctx context.Context, db *sql.DB) error {
	hasPrimary, err := tableHasPrimaryKey(ctx, db, "messages")
	if err != nil {
		return err
	}
	if !hasPrimary {
		return nil
	}
	if _, err := db.ExecContext(ctx, `ALTER TABLE messages RENAME TO messages_with_id_pk`); err != nil {
		return fmt.Errorf("renaming duckdb messages table for rekey: %w", err)
	}
	create := mirrorTableCreate("messages")
	cols := mirrorTableColumns("messages")
	if create == "" || len(cols) == 0 {
		return fmt.Errorf("missing duckdb messages table spec")
	}
	if _, err := db.ExecContext(ctx, create); err != nil {
		return fmt.Errorf("creating rekeyed duckdb messages table: %w", err)
	}
	colList := strings.Join(cols, ", ")
	if _, err := db.ExecContext(ctx, fmt.Sprintf(
		`INSERT INTO messages (%[1]s) SELECT %[1]s FROM messages_with_id_pk`,
		colList,
	)); err != nil {
		return fmt.Errorf("copying rekeyed duckdb messages: %w", err)
	}
	if _, err := db.ExecContext(ctx, `DROP TABLE messages_with_id_pk`); err != nil {
		return fmt.Errorf("dropping old duckdb messages table: %w", err)
	}
	return nil
}

func tableHasPrimaryKey(ctx context.Context, db *sql.DB, table string) (bool, error) {
	var count int
	if err := db.QueryRowContext(ctx, `
		SELECT COUNT(*)
		FROM information_schema.table_constraints
		WHERE table_schema = current_schema()
		  AND table_name = ?
		  AND constraint_type = 'PRIMARY KEY'`,
		strings.ToLower(table),
	).Scan(&count); err != nil {
		return false, fmt.Errorf("checking duckdb primary key for %s: %w", table, err)
	}
	return count > 0, nil
}

func mirrorTableCreate(name string) string {
	for _, table := range mirrorTables {
		if table.name == name {
			return table.create
		}
	}
	return ""
}

func mirrorTableColumns(name string) []string {
	for _, table := range mirrorTables {
		if table.name != name {
			continue
		}
		cols := make([]string, len(table.columns))
		for i, column := range table.columns {
			cols[i] = column.name
		}
		return cols
	}
	return nil
}

func metadataKeyExists(ctx context.Context, db *sql.DB, key string) (bool, error) {
	var exists bool
	if err := db.QueryRowContext(ctx,
		`SELECT COUNT(*) > 0 FROM sync_metadata WHERE key = ?`,
		key,
	).Scan(&exists); err != nil {
		return false, fmt.Errorf("checking duckdb metadata key %s: %w", key, err)
	}
	return exists, nil
}

func recordMetadataKey(
	ctx context.Context, db *sql.DB, key string, value string,
) error {
	var existing string
	err := db.QueryRowContext(ctx,
		`SELECT value FROM sync_metadata WHERE key = ?`,
		key,
	).Scan(&existing)
	if err == nil && existing == value {
		return nil
	}
	if err != nil && !errors.Is(err, sql.ErrNoRows) {
		return fmt.Errorf("checking duckdb metadata key %s: %w", key, err)
	}
	if _, err := db.ExecContext(ctx, `
		INSERT INTO sync_metadata (key, value)
		VALUES (?, ?)
		ON CONFLICT(key) DO UPDATE SET value = excluded.value`,
		key, value,
	); err != nil {
		return fmt.Errorf("recording duckdb metadata key %s: %w", key, err)
	}
	return nil
}

func dropQuackIncompatibleTimestampDefaults(ctx context.Context, db *sql.DB) error {
	for _, spec := range quackIncompatibleTimestampDefaults {
		defaultValue, err := columnDefault(ctx, db, spec.table, spec.column)
		if err != nil {
			return err
		}
		if !strings.Contains(strings.ToLower(defaultValue), "current_timestamp") {
			continue
		}
		stmt := fmt.Sprintf(
			"ALTER TABLE %s ALTER COLUMN %s DROP DEFAULT",
			spec.table,
			spec.column,
		)
		if _, err := db.ExecContext(ctx, stmt); err != nil {
			return fmt.Errorf(
				"dropping quack-incompatible duckdb default %s.%s: %w",
				spec.table,
				spec.column,
				err,
			)
		}
	}
	return nil
}

func columnDefault(
	ctx context.Context, db *sql.DB, table, column string,
) (string, error) {
	var defaultValue sql.NullString
	err := db.QueryRowContext(ctx, `
		SELECT column_default
		FROM information_schema.columns
		WHERE table_schema = current_schema()
		  AND lower(table_name) = ?
		  AND lower(column_name) = ?`,
		strings.ToLower(table),
		strings.ToLower(column),
	).Scan(&defaultValue)
	if errors.Is(err, sql.ErrNoRows) {
		return "", nil
	}
	if err != nil {
		return "", fmt.Errorf(
			"checking duckdb column default %s.%s: %w",
			table,
			column,
			err,
		)
	}
	if !defaultValue.Valid {
		return "", nil
	}
	return defaultValue.String, nil
}

func relaxedColumnDef(def string) string {
	def = strings.ReplaceAll(def, " NOT NULL", "")
	if i := strings.Index(def, " DEFAULT "); i >= 0 {
		def = def[:i]
	}
	return def
}

func backfillAddedColumnDefault(
	ctx context.Context, db *sql.DB, table string, column columnSpec,
) error {
	defaultValue, ok := columnDefaultLiteral(column.def)
	if !ok {
		return nil
	}
	stmt := fmt.Sprintf(
		"UPDATE %s SET %s = %s WHERE %s IS NULL",
		table, column.name, defaultValue, column.name,
	)
	if _, err := db.ExecContext(ctx, stmt); err != nil {
		return fmt.Errorf(
			"backfilling duckdb column %s.%s default: %w",
			table, column.name, err,
		)
	}
	return nil
}

func columnDefaultLiteral(def string) (string, bool) {
	idx := strings.Index(strings.ToUpper(def), " DEFAULT ")
	if idx < 0 {
		return "", false
	}
	value := strings.TrimSpace(def[idx+len(" DEFAULT "):])
	if value == "" || strings.Contains(strings.ToLower(value), "current_timestamp") {
		return "", false
	}
	return value, true
}

// CheckSchemaCompat verifies that the DuckDB mirror has the required
// read/push tables and columns. It does not mutate the database.
func CheckSchemaCompat(ctx context.Context, db *sql.DB) error {
	if err := checkSchemaShapeCompat(ctx, db); err != nil {
		return err
	}
	pendingRepairs, err := pendingSchemaRepairs(ctx, db)
	if err != nil {
		return err
	}
	return schemaRepairError(pendingRepairs)
}

func CheckSchemaCompatViaQuack(ctx context.Context, db *sql.DB) error {
	if err := checkSchemaShapeCompat(ctx, db); err != nil {
		return err
	}
	return checkSchemaRepairsViaQuack(ctx, db)
}

func checkSchemaShapeCompat(ctx context.Context, db *sql.DB) error {
	existing, err := loadColumns(ctx, db)
	if err != nil {
		return err
	}
	var missing []string
	for _, table := range mirrorTables {
		have, ok := existing[table.name]
		if !ok || len(have) == 0 {
			missing = append(missing, "missing table "+table.name)
			continue
		}
		for _, column := range table.columns {
			if !have[column.name] {
				missing = append(missing, table.name+"."+column.name)
			}
		}
	}
	if len(missing) > 0 {
		sort.Strings(missing)
		return fmt.Errorf(
			"duckdb schema incompatible; run agentsview duckdb push to migrate; missing: %s",
			strings.Join(missing, ", "),
		)
	}

	var version string
	err = db.QueryRowContext(ctx,
		`SELECT value FROM sync_metadata WHERE key = ?`,
		schemaVersionMetadataKey,
	).Scan(&version)
	if errors.Is(err, sql.ErrNoRows) {
		return fmt.Errorf(
			"duckdb schema incompatible; missing %s in sync_metadata",
			schemaVersionMetadataKey,
		)
	}
	if err != nil {
		return fmt.Errorf("checking duckdb schema version: %w", err)
	}
	got, err := strconv.Atoi(version)
	if err != nil {
		return fmt.Errorf(
			"duckdb schema incompatible; invalid schema version %q",
			version,
		)
	}
	if got < SchemaVersion {
		return fmt.Errorf(
			"duckdb schema incompatible; version %d is older than required %d",
			got, SchemaVersion,
		)
	}

	return nil
}

func checkSchemaRepairsViaQuack(ctx context.Context, db *sql.DB) error {
	rows, err := db.QueryContext(ctx,
		"SELECT issue FROM "+quackAttachmentName+".query(?)",
		pendingSchemaRepairsQuery(),
	)
	if err != nil {
		return fmt.Errorf("checking duckdb schema repairs through quack: %w", err)
	}
	pendingRepairs, err := collectSchemaRepairIssues(rows)
	if err != nil {
		return err
	}
	return schemaRepairError(pendingRepairs)
}

func pendingSchemaRepairs(ctx context.Context, db *sql.DB) ([]string, error) {
	rows, err := db.QueryContext(ctx, pendingSchemaRepairsQuery())
	if err != nil {
		return nil, fmt.Errorf("checking duckdb schema repairs: %w", err)
	}
	return collectSchemaRepairIssues(rows)
}

func collectSchemaRepairIssues(rows *sql.Rows) ([]string, error) {
	defer func() {
		_ = rows.Close()
	}()
	var pendingRepairs []string
	for rows.Next() {
		var issue string
		if err := rows.Scan(&issue); err != nil {
			return nil, fmt.Errorf("scanning duckdb schema repair check: %w", err)
		}
		pendingRepairs = append(pendingRepairs, issue)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterating duckdb schema repair check: %w", err)
	}
	return pendingRepairs, nil
}

func schemaRepairError(pendingRepairs []string) error {
	if len(pendingRepairs) == 0 {
		return nil
	}
	sort.Strings(pendingRepairs)
	return fmt.Errorf(
		"duckdb schema incompatible; run full DuckDB schema migration on the base database; pending repairs: %s",
		strings.Join(pendingRepairs, ", "),
	)
}

func pendingSchemaRepairsQuery() string {
	metadataValues := []string{
		"(" + duckLiteral(defaultRepairMetadataKey) + ")",
		"(" + duckLiteral(usageDedupIndexMetadataKey) + ")",
		"(" + duckLiteral(projectIdentityRemoteScrubMetadataKey) + ")",
	}
	defaultValues := make([]string, len(quackIncompatibleTimestampDefaults))
	for i, spec := range quackIncompatibleTimestampDefaults {
		defaultValues[i] = "(" + duckLiteral(spec.table) + ", " +
			duckLiteral(spec.column) + ")"
	}
	return `
		WITH required_metadata(key) AS (
			VALUES ` + strings.Join(metadataValues, ", ") + `
		),
		checked_defaults(table_name, column_name) AS (
			VALUES ` + strings.Join(defaultValues, ", ") + `
		)
		SELECT 'missing ' || key AS issue
		FROM required_metadata m
		WHERE NOT EXISTS (
			SELECT 1 FROM sync_metadata sm WHERE sm.key = m.key
		)
		UNION ALL
		SELECT 'messages.id primary key' AS issue
		WHERE EXISTS (
			SELECT 1
			FROM information_schema.table_constraints
			WHERE table_schema = current_schema()
			  AND lower(table_name) = 'messages'
			  AND constraint_type = 'PRIMARY KEY'
		)
		UNION ALL
		SELECT lower(c.table_name) || '.' || lower(c.column_name) ||
			' current_timestamp default' AS issue
		FROM information_schema.columns c
		JOIN checked_defaults d
		  ON lower(c.table_name) = d.table_name
		 AND lower(c.column_name) = d.column_name
		WHERE c.table_schema = current_schema()
		  AND lower(coalesce(c.column_default, '')) LIKE '%current_timestamp%'`
}

func loadColumns(ctx context.Context, db *sql.DB) (map[string]map[string]bool, error) {
	rows, err := db.QueryContext(ctx, `
		SELECT lower(table_name), lower(column_name)
		FROM information_schema.columns
		WHERE table_schema = current_schema()`)
	if err != nil {
		return nil, fmt.Errorf("loading duckdb schema columns: %w", err)
	}
	defer rows.Close()

	columns := make(map[string]map[string]bool)
	for rows.Next() {
		var table, column string
		if err := rows.Scan(&table, &column); err != nil {
			return nil, fmt.Errorf("scanning duckdb schema columns: %w", err)
		}
		if columns[table] == nil {
			columns[table] = make(map[string]bool)
		}
		columns[table][column] = true
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterating duckdb schema columns: %w", err)
	}
	return columns, nil
}