byre/internal/store/store.go

package store

import (
	"context"
	"crypto/rand"
	"database/sql"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"sync"
	"time"

	_ "modernc.org/sqlite"

	"g.fc.run/misc/byre/internal/api"
)

var ErrNotFound = errors.New("not found")

type migration struct {
	name string
	sql  string
}

var migrations = []migration{{"create-core-tables", `
CREATE TABLE IF NOT EXISTS users (
	login TEXT PRIMARY KEY,
	display_name TEXT NOT NULL,
	git_name TEXT NOT NULL DEFAULT '',
	git_email TEXT NOT NULL DEFAULT '',
	created_at TEXT NOT NULL
);

CREATE TABLE IF NOT EXISTS projects (
	id TEXT PRIMARY KEY,
	name TEXT NOT NULL UNIQUE,
	repo_path TEXT NOT NULL UNIQUE,
	default_branch TEXT NOT NULL,
	remote TEXT NOT NULL,
	created_by TEXT NOT NULL REFERENCES users(login),
	created_at TEXT NOT NULL
);

CREATE TABLE IF NOT EXISTS tasks (
	id TEXT PRIMARY KEY,
	project_id TEXT NOT NULL REFERENCES projects(id),
	title TEXT NOT NULL,
	branch TEXT NOT NULL,
	worktree_path TEXT NOT NULL,
	base_ref TEXT NOT NULL,
	status TEXT NOT NULL,
	created_by TEXT NOT NULL REFERENCES users(login),
	created_at TEXT NOT NULL,
	completed_at TEXT
);

CREATE INDEX IF NOT EXISTS tasks_project ON tasks(project_id, created_at);

CREATE TABLE IF NOT EXISTS events (
	seq INTEGER PRIMARY KEY AUTOINCREMENT,
	kind TEXT NOT NULL,
	project_id TEXT,
	task_id TEXT,
	actor TEXT NOT NULL,
	payload TEXT NOT NULL,
	created_at TEXT NOT NULL
);
`}, {"add-agent-sessions", `
ALTER TABLE events ADD COLUMN agent_session_id TEXT;

CREATE TABLE agent_sessions (
	id TEXT PRIMARY KEY,
	task_id TEXT NOT NULL REFERENCES tasks(id),
	vendor TEXT NOT NULL,
	model TEXT NOT NULL,
	vendor_session_id TEXT NOT NULL,
	state TEXT NOT NULL,
	role TEXT NOT NULL,
	created_by TEXT NOT NULL REFERENCES users(login),
	created_at TEXT NOT NULL
);

CREATE INDEX agent_sessions_task ON agent_sessions(task_id, created_at);
CREATE INDEX agent_sessions_state ON agent_sessions(state);
`}, {"add-inbox", `
CREATE TABLE inbox_items (
	id TEXT PRIMARY KEY,
	project_id TEXT NOT NULL REFERENCES projects(id),
	task_id TEXT NOT NULL REFERENCES tasks(id),
	agent_session_id TEXT NOT NULL REFERENCES agent_sessions(id),
	thread_id TEXT NOT NULL,
	kind TEXT NOT NULL,
	priority INTEGER NOT NULL,
	blocking INTEGER NOT NULL,
	payload TEXT NOT NULL,
	state TEXT NOT NULL,
	snoozed_until TEXT,
	created_at TEXT NOT NULL,
	resolved_by TEXT REFERENCES users(login),
	resolved_at TEXT
);

CREATE INDEX inbox_items_open ON inbox_items(state, priority, created_at);
CREATE INDEX inbox_items_session ON inbox_items(agent_session_id, state);

CREATE TABLE inbox_responses (
	id TEXT PRIMARY KEY,
	item_id TEXT NOT NULL REFERENCES inbox_items(id),
	user_login TEXT NOT NULL REFERENCES users(login),
	type TEXT NOT NULL,
	option TEXT,
	text TEXT,
	attachments TEXT NOT NULL,
	snooze_until TEXT,
	created_at TEXT NOT NULL,
	delivered_at TEXT
);

CREATE INDEX inbox_responses_item ON inbox_responses(item_id, created_at);
`}, {"index-events-by-task", `
CREATE INDEX events_task ON events(task_id, seq);
`}, {"add-turn-queue", `
CREATE TABLE turn_queue (
	session_id TEXT PRIMARY KEY REFERENCES agent_sessions(id),
	actor TEXT NOT NULL,
	text TEXT NOT NULL,
	created_at TEXT NOT NULL
);
`}, {"merge-response-types-into-other", `
UPDATE inbox_responses SET type = 'other' WHERE type IN ('free_form', 'redirect', 'question');
UPDATE inbox_items SET state = 'answered' WHERE state = 'redirected';
`}, {"flatten-report-payloads-to-text", `
UPDATE inbox_items SET payload = json_object('text', coalesce((SELECT group_concat('- ' || value, char(10)) FROM json_each(inbox_items.payload, '$.done')), ''))
WHERE kind = 'report' AND json_type(payload, '$.done') IS NOT NULL;
`}, {"rename-completed-tasks-to-discarded", `
ALTER TABLE tasks RENAME COLUMN completed_at TO closed_at;
ALTER TABLE tasks ADD COLUMN merge_commit TEXT NOT NULL DEFAULT '';
UPDATE tasks SET status = 'discarded' WHERE status = 'completed';
`}, {"dismiss-inbox-of-closed-tasks", `
UPDATE inbox_items SET state = 'dismissed', resolved_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now'), snoozed_until = NULL
WHERE state = 'open' AND task_id IN (SELECT id FROM tasks WHERE status != 'active');
`}, {"add-blobs-and-queued-images", `
CREATE TABLE IF NOT EXISTS blobs (
	id TEXT PRIMARY KEY,
	content_type TEXT NOT NULL,
	path TEXT NOT NULL,
	created_by TEXT NOT NULL REFERENCES users(login),
	created_at TEXT NOT NULL
);

CREATE TABLE turn_queue_next (
	session_id TEXT PRIMARY KEY REFERENCES agent_sessions(id),
	actor TEXT NOT NULL,
	text TEXT NOT NULL,
	images TEXT NOT NULL DEFAULT '[]',
	created_at TEXT NOT NULL
);
INSERT INTO turn_queue_next (session_id, actor, text, created_at) SELECT session_id, actor, text, created_at FROM turn_queue;
DROP TABLE turn_queue;
ALTER TABLE turn_queue_next RENAME TO turn_queue;
`}, {"add-task-merge-title", `
ALTER TABLE tasks ADD COLUMN merge_title TEXT NOT NULL DEFAULT '';
`}, {"add-web-push", `
ALTER TABLE users ADD COLUMN notification_prefs TEXT NOT NULL
	DEFAULT '{"blocking_questions":true,"questions":false,"reports":false,"errors":false,"rate_limits":false}';

CREATE TABLE vapid_keys (
	id INTEGER PRIMARY KEY CHECK (id = 1),
	public_key TEXT NOT NULL,
	private_key TEXT NOT NULL
);

CREATE TABLE push_subscriptions (
	endpoint TEXT PRIMARY KEY,
	user_login TEXT NOT NULL REFERENCES users(login),
	p256dh TEXT NOT NULL,
	auth TEXT NOT NULL,
	subscriber TEXT NOT NULL,
	created_at TEXT NOT NULL
);
`}, {"add-snapshots", `
CREATE TABLE snapshots (
	id TEXT PRIMARY KEY,
	task_id TEXT NOT NULL REFERENCES tasks(id),
	agent_session_id TEXT NOT NULL REFERENCES agent_sessions(id),
	n INTEGER NOT NULL,
	commit_sha TEXT NOT NULL,
	tree_sha TEXT NOT NULL,
	event_seq INTEGER NOT NULL,
	trigger TEXT NOT NULL,
	created_at TEXT NOT NULL,
	UNIQUE (task_id, n)
);
`}, {"inline-question-choices", `
ALTER TABLE inbox_responses ADD COLUMN options TEXT;
UPDATE inbox_responses SET options = json_array(option) WHERE option IS NOT NULL;
ALTER TABLE inbox_responses DROP COLUMN option;

UPDATE inbox_items SET payload = json_set(json_remove(payload, '$.recommended_option'),
	'$.context', json_extract(payload, '$.context') || char(10) || char(10) || (
		SELECT group_concat(CASE WHEN o.value = json_extract(inbox_items.payload, '$.recommended_option') THEN '- (*) ' ELSE '- ( ) ' END || o.value, char(10))
		FROM json_each(inbox_items.payload, '$.options') o),
	'$.multiple', json('false'),
	'$.recommended', json_array(json_extract(payload, '$.recommended_option')))
WHERE kind = 'question' AND json_type(payload, '$.recommended_option') IS NOT NULL;
`}, {"unsnooze-inbox-items", `
UPDATE inbox_items SET snoozed_until = NULL WHERE snoozed_until IS NOT NULL;
`}, {"add-task-todo-item", `
ALTER TABLE tasks ADD COLUMN todo_item TEXT NOT NULL DEFAULT '';
`}, {"add-lead-agents", `
CREATE TABLE agent_sessions_next (
	id TEXT PRIMARY KEY,
	task_id TEXT REFERENCES tasks(id),
	owner TEXT NOT NULL DEFAULT '',
	vendor TEXT NOT NULL,
	model TEXT NOT NULL,
	vendor_session_id TEXT NOT NULL,
	state TEXT NOT NULL,
	role TEXT NOT NULL,
	created_by TEXT NOT NULL REFERENCES users(login),
	created_at TEXT NOT NULL
);
INSERT INTO agent_sessions_next (id, task_id, vendor, model, vendor_session_id, state, role, created_by, created_at)
	SELECT id, task_id, vendor, model, vendor_session_id, state, role, created_by, created_at FROM agent_sessions;
DROP TABLE agent_sessions;
ALTER TABLE agent_sessions_next RENAME TO agent_sessions;
CREATE INDEX agent_sessions_task ON agent_sessions(task_id, created_at);
CREATE INDEX agent_sessions_state ON agent_sessions(state);
CREATE INDEX agent_sessions_owner ON agent_sessions(owner, created_at);

ALTER TABLE events ADD COLUMN owner TEXT;
CREATE INDEX events_owner ON events(owner, seq) WHERE owner IS NOT NULL;

ALTER TABLE users ADD COLUMN focus TEXT NOT NULL DEFAULT '{"tabs":["transcript"],"tab":"transcript"}';

CREATE TABLE lead_inputs (
	id INTEGER PRIMARY KEY AUTOINCREMENT,
	owner TEXT NOT NULL REFERENCES users(login),
	kind TEXT NOT NULL,
	actor TEXT NOT NULL,
	text TEXT NOT NULL,
	images TEXT NOT NULL DEFAULT '[]',
	created_at TEXT NOT NULL
);
CREATE INDEX lead_inputs_owner ON lead_inputs(owner, id);
`}, {"add-focus-project", `
UPDATE users SET focus = json_set(focus, '$.project_id', (SELECT project_id FROM tasks WHERE id = json_extract(users.focus, '$.task_id')))
WHERE json_extract(focus, '$.task_id') IS NOT NULL;
DELETE FROM lead_inputs WHERE text LIKE 'The human switched the right pane %' OR text = 'The human cleared the right pane.';
`}, {"add-task-merging-by", `
ALTER TABLE tasks ADD COLUMN merging_by TEXT NOT NULL DEFAULT '';
`}, {"flag-handoff-prompts", `
UPDATE events SET payload = json_set(payload, '$.handoff', json('true'))
WHERE kind = 'user.message' AND actor = 'byre' AND json_extract(payload, '$.text') LIKE 'Write a handoff for a fresh %';
`}, {"rename-todo-tab-to-project", `
UPDATE users SET focus = json_set(focus,
	'$.tab', CASE json_extract(focus, '$.tab') WHEN 'todo' THEN 'project' ELSE json_extract(focus, '$.tab') END,
	'$.tabs', json((SELECT json_group_array(CASE value WHEN 'todo' THEN 'project' ELSE value END) FROM json_each(users.focus, '$.tabs'))));
`}, {"add-snapshot-forks", `
ALTER TABLE tasks ADD COLUMN forked_from TEXT NOT NULL DEFAULT '';
ALTER TABLE tasks ADD COLUMN forked_at INTEGER NOT NULL DEFAULT 0;
ALTER TABLE agent_sessions ADD COLUMN forked_from TEXT NOT NULL DEFAULT '';
ALTER TABLE agent_sessions ADD COLUMN forked_at TEXT NOT NULL DEFAULT '';
`}, {"add-task-port-base", `
ALTER TABLE tasks ADD COLUMN port_base INTEGER NOT NULL DEFAULT 0;
UPDATE tasks SET port_base = (
	SELECT 41000 + 10 * (count(*) - 1) FROM tasks AS earlier
	WHERE earlier.status = 'active' AND (earlier.created_at < tasks.created_at OR (earlier.created_at = tasks.created_at AND earlier.id <= tasks.id))
) WHERE status = 'active' AND port_base = 0;
`}, {"add-task-merge-queue", `
ALTER TABLE tasks ADD COLUMN merge_queued_by TEXT NOT NULL DEFAULT '';
ALTER TABLE tasks ADD COLUMN merge_queued_at TEXT NOT NULL DEFAULT '';
`}, {"add-task-previews", `
CREATE TABLE previews (
	task_id TEXT NOT NULL REFERENCES tasks(id),
	name TEXT NOT NULL,
	port INTEGER NOT NULL,
	created_at TEXT NOT NULL,
	PRIMARY KEY (task_id, name),
	UNIQUE (task_id, port)
);
`}, {"add-task-color", `
ALTER TABLE tasks ADD COLUMN color INTEGER NOT NULL DEFAULT 0;
UPDATE tasks SET color = (SELECT count(*) FROM tasks AS earlier WHERE earlier.rowid < tasks.rowid) % 11;
UPDATE tasks SET color = (
	SELECT count(*) FROM tasks AS earlier WHERE earlier.status = 'active' AND earlier.rowid < tasks.rowid
) % 11 WHERE status = 'active';
`}, {"drop-turn-queue", `
UPDATE agent_sessions SET state = 'idle' WHERE state = 'queued';
DROP TABLE turn_queue;
`}, {"add-snapshot-reviews", `
CREATE TABLE reviews (
	id TEXT PRIMARY KEY,
	task_id TEXT NOT NULL REFERENCES tasks(id),
	snapshot_n INTEGER NOT NULL,
	agent_session_id TEXT NOT NULL REFERENCES agent_sessions(id),
	vendor TEXT NOT NULL,
	model TEXT NOT NULL,
	state TEXT NOT NULL,
	result TEXT NOT NULL DEFAULT '',
	error TEXT NOT NULL DEFAULT '',
	created_at TEXT NOT NULL,
	finished_at TEXT NOT NULL DEFAULT '',
	UNIQUE (task_id, snapshot_n)
);
CREATE UNIQUE INDEX reviews_session ON reviews(agent_session_id);
`}, {"add-task-inputs", `
CREATE TABLE task_inputs (
	id INTEGER PRIMARY KEY AUTOINCREMENT,
	task_id TEXT NOT NULL REFERENCES tasks(id),
	actor TEXT NOT NULL,
	text TEXT NOT NULL,
	images TEXT NOT NULL DEFAULT '[]',
	created_at TEXT NOT NULL
);
CREATE INDEX task_inputs_task ON task_inputs(task_id, id);
`}, {"drop-web-push", `
DROP TABLE push_subscriptions;
DROP TABLE vapid_keys;
`}, {"delete-report-inbox-items", `
DELETE FROM inbox_responses WHERE item_id IN (SELECT id FROM inbox_items WHERE kind = 'report');
DELETE FROM inbox_items WHERE kind = 'report';
`}, {"allow-sessionless-reviews", `
CREATE TABLE reviews_new (
	id TEXT PRIMARY KEY,
	task_id TEXT NOT NULL REFERENCES tasks(id),
	snapshot_n INTEGER NOT NULL,
	agent_session_id TEXT REFERENCES agent_sessions(id),
	vendor TEXT NOT NULL,
	model TEXT NOT NULL,
	state TEXT NOT NULL,
	result TEXT NOT NULL DEFAULT '',
	error TEXT NOT NULL DEFAULT '',
	created_at TEXT NOT NULL,
	finished_at TEXT NOT NULL DEFAULT '',
	UNIQUE (task_id, snapshot_n)
);
INSERT INTO reviews_new SELECT * FROM reviews;
DROP TABLE reviews;
ALTER TABLE reviews_new RENAME TO reviews;
CREATE UNIQUE INDEX reviews_session ON reviews(agent_session_id);
`}, {"add-workflows", `
CREATE TABLE workflows (
	name TEXT PRIMARY KEY,
	definition TEXT NOT NULL
);
ALTER TABLE tasks ADD COLUMN workflow_name TEXT NOT NULL DEFAULT '';
ALTER TABLE tasks ADD COLUMN workflow_definition TEXT NOT NULL DEFAULT '';
`}, {"run-workflow-stages", `
ALTER TABLE tasks ADD COLUMN workflow_brief TEXT NOT NULL DEFAULT '';
ALTER TABLE agent_sessions ADD COLUMN workflow_stage INTEGER NOT NULL DEFAULT 0;
ALTER TABLE agent_sessions ADD COLUMN stage_output TEXT NOT NULL DEFAULT '';
ALTER TABLE agent_sessions ADD COLUMN stage_finished INTEGER NOT NULL DEFAULT 0;
UPDATE workflows SET definition = json_set(definition, '$.stages',
	(SELECT json_group_array(json_remove(value, '$.vendor')) FROM (SELECT value FROM json_each(definition, '$.stages') ORDER BY key)));
UPDATE tasks SET workflow_definition = json_set(workflow_definition, '$.stages',
	(SELECT json_group_array(json_remove(value, '$.vendor')) FROM (SELECT value FROM json_each(workflow_definition, '$.stages') ORDER BY key)))
	WHERE workflow_definition != '';
`}, {"workflow-stage-tasks", `
ALTER TABLE tasks ADD COLUMN workflow_run TEXT NOT NULL DEFAULT '';
ALTER TABLE tasks ADD COLUMN workflow_stage TEXT NOT NULL DEFAULT '';
ALTER TABLE tasks ADD COLUMN workflow_title TEXT NOT NULL DEFAULT '';
ALTER TABLE tasks ADD COLUMN stage_output TEXT NOT NULL DEFAULT '';
ALTER TABLE tasks ADD COLUMN stage_commit TEXT NOT NULL DEFAULT '';
UPDATE tasks SET workflow_run = id, workflow_title = title,
	workflow_stage = coalesce(
		(SELECT json_extract(tasks.workflow_definition, '$.stages[' || (s.workflow_stage - 1) || '].name') FROM agent_sessions s
			WHERE s.task_id = tasks.id AND s.workflow_stage > 0 ORDER BY s.created_at DESC LIMIT 1),
		json_extract(workflow_definition, '$.stages[0].name'))
	WHERE workflow_name != '' AND workflow_run = '';
`}, {"drop-preview-tab", `
UPDATE users SET focus = json_set(focus,
	'$.tab', CASE json_extract(focus, '$.tab') WHEN 'preview' THEN 'transcript' ELSE json_extract(focus, '$.tab') END,
	'$.tabs', json((SELECT json_group_array(value) FROM json_each(users.focus, '$.tabs') WHERE value != 'preview')));
UPDATE users SET focus = json_set(focus, '$.tabs', json('["transcript"]')) WHERE json_array_length(focus, '$.tabs') = 0;
`}, {"drop-inbox-priority", `
CREATE TABLE inbox_items_next (
	id TEXT PRIMARY KEY,
	project_id TEXT NOT NULL REFERENCES projects(id),
	task_id TEXT NOT NULL REFERENCES tasks(id),
	agent_session_id TEXT NOT NULL REFERENCES agent_sessions(id),
	thread_id TEXT NOT NULL,
	kind TEXT NOT NULL,
	priority INTEGER NOT NULL DEFAULT 0,
	blocking INTEGER NOT NULL DEFAULT 1,
	payload TEXT NOT NULL,
	state TEXT NOT NULL,
	snoozed_until TEXT,
	created_at TEXT NOT NULL,
	resolved_by TEXT REFERENCES users(login),
	resolved_at TEXT
);
INSERT INTO inbox_items_next SELECT id, project_id, task_id, agent_session_id, thread_id, kind, priority, blocking, payload, state, snoozed_until, created_at, resolved_by, resolved_at FROM inbox_items;
DROP TABLE inbox_items;
ALTER TABLE inbox_items_next RENAME TO inbox_items;
CREATE INDEX inbox_items_open ON inbox_items(state, created_at);
CREATE INDEX inbox_items_session ON inbox_items(agent_session_id, state);

CREATE TABLE users_next (
	login TEXT PRIMARY KEY,
	display_name TEXT NOT NULL,
	git_name TEXT NOT NULL DEFAULT '',
	git_email TEXT NOT NULL DEFAULT '',
	created_at TEXT NOT NULL,
	notification_prefs TEXT NOT NULL DEFAULT '{"questions":true,"reports":false,"errors":false,"rate_limits":false}',
	focus TEXT NOT NULL DEFAULT '{"tabs":["transcript"],"tab":"transcript"}'
);
INSERT INTO users_next (login, display_name, git_name, git_email, created_at, notification_prefs, focus)
	SELECT login, display_name, git_name, git_email, created_at,
		json_remove(json_set(notification_prefs, '$.questions', json(CASE WHEN json_extract(notification_prefs, '$.blocking_questions') THEN 'true' ELSE 'false' END)), '$.blocking_questions'),
		focus
	FROM users;
DROP TABLE users;
ALTER TABLE users_next RENAME TO users;
`}}

const (
	DefaultPortFirst = 41000
	PortBlock        = 10
	TaskColors       = 11
)

type Store struct {
	db        *sql.DB
	path      string
	portFirst int
	writeMu   sync.Mutex
	onEvent   func(api.Event)
}

func Open(path string, portFirst int, onEvent func(api.Event)) (*Store, error) {
	dsn := "file:" + path + "?_pragma=journal_mode(WAL)&_pragma=busy_timeout(5000)&_pragma=foreign_keys(1)&_txlock=immediate"
	db, err := sql.Open("sqlite", dsn)
	if err != nil {
		return nil, err
	}

	err = migrate(db, migrations)
	if err != nil {
		db.Close()
		return nil, err
	}

	return &Store{db: db, path: path, portFirst: portFirst, onEvent: onEvent}, nil
}

func migrate(db *sql.DB, migrations []migration) error {
	err := createMigrationsTable(db, migrations)
	if err != nil {
		return err
	}

	applied, err := appliedMigrations(db)
	if err != nil {
		return err
	}

	for _, m := range migrations {
		if applied[m.name] {
			continue
		}

		err = applyMigration(db, m)
		if err != nil {
			return fmt.Errorf("migration %s: %w", m.name, err)
		}
	}

	return nil
}

func createMigrationsTable(db *sql.DB, migrations []migration) error {
	tx, err := db.Begin()
	if err != nil {
		return err
	}
	defer tx.Rollback()

	exists := 0
	err = tx.QueryRow(`SELECT count(*) FROM sqlite_master WHERE type = 'table' AND name = 'migrations'`).Scan(&exists)
	if err != nil {
		return err
	}
	if exists > 0 {
		return nil
	}

	_, err = tx.Exec(`CREATE TABLE migrations (name TEXT PRIMARY KEY, applied_at TEXT NOT NULL)`)
	if err != nil {
		return err
	}

	version := 0
	err = tx.QueryRow(`PRAGMA user_version`).Scan(&version)
	if err != nil {
		return err
	}

	for _, m := range migrations[:min(version, len(migrations))] {
		_, err = tx.Exec(`INSERT INTO migrations (name, applied_at) VALUES (?, ?)`, m.name, now())
		if err != nil {
			return err
		}
	}

	return tx.Commit()
}

func appliedMigrations(db *sql.DB) (map[string]bool, error) {
	rows, err := db.Query(`SELECT name FROM migrations`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	applied := map[string]bool{}
	for rows.Next() {
		name := ""
		err = rows.Scan(&name)
		if err != nil {
			return nil, err
		}
		applied[name] = true
	}
	return applied, rows.Err()
}

func applyMigration(db *sql.DB, m migration) error {
	ctx := context.Background()
	conn, err := db.Conn(ctx)
	if err != nil {
		return err
	}
	defer conn.Close()

	_, err = conn.ExecContext(ctx, `PRAGMA foreign_keys = OFF`)
	if err != nil {
		return err
	}
	defer conn.ExecContext(ctx, `PRAGMA foreign_keys = ON`)

	tx, err := conn.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer tx.Rollback()

	_, err = tx.Exec(m.sql)
	if err != nil {
		return err
	}

	err = checkForeignKeys(tx)
	if err != nil {
		return err
	}

	_, err = tx.Exec(`INSERT INTO migrations (name, applied_at) VALUES (?, ?)`, m.name, now())
	if err != nil {
		return err
	}

	return tx.Commit()
}

func checkForeignKeys(tx *sql.Tx) error {
	rows, err := tx.Query(`PRAGMA foreign_key_check`)
	if err != nil {
		return err
	}
	defer rows.Close()

	if rows.Next() {
		table := ""
		rowid := sql.NullInt64{}
		parent := ""
		fkid := 0
		err = rows.Scan(&table, &rowid, &parent, &fkid)
		if err != nil {
			return err
		}
		return fmt.Errorf("foreign key violation: %s row %d references a missing %s", table, rowid.Int64, parent)
	}
	return rows.Err()
}

func (s *Store) Close() error {
	return s.db.Close()
}

func (s *Store) Path() string {
	return s.path
}

func NewID() string {
	b := make([]byte, 6)
	rand.Read(b)
	return hex.EncodeToString(b)
}

func (s *Store) EnsureUser(ctx context.Context, login, displayName string) (api.User, error) {
	_, err := s.db.ExecContext(ctx, `
		INSERT INTO users (login, display_name, created_at) VALUES (?, ?, ?)
		ON CONFLICT (login) DO UPDATE SET display_name = excluded.display_name
	`, login, displayName, now())
	if err != nil {
		return api.User{}, err
	}
	return s.GetUser(ctx, login)
}

func (s *Store) GetUser(ctx context.Context, login string) (api.User, error) {
	u := api.User{}
	prefs := ""
	err := s.db.QueryRowContext(ctx, `SELECT login, display_name, git_name, git_email, notification_prefs FROM users WHERE login = ?`, login).
		Scan(&u.Login, &u.DisplayName, &u.GitName, &u.GitEmail, &prefs)
	if errors.Is(err, sql.ErrNoRows) {
		return u, ErrNotFound
	}
	if err != nil {
		return u, err
	}
	err = json.Unmarshal([]byte(prefs), &u.NotificationPrefs)
	return u, err
}

func (s *Store) UpdateUserGit(ctx context.Context, login, gitName, gitEmail string) (api.User, error) {
	err := s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
		_, err := tx.ExecContext(ctx, `UPDATE users SET git_name = ?, git_email = ? WHERE login = ?`, gitName, gitEmail, login)
		if err != nil {
			return err
		}
		return emit(api.Event{Kind: "user.updated", Actor: login, Payload: map[string]any{"login": login}})
	})
	if err != nil {
		return api.User{}, err
	}
	return s.GetUser(ctx, login)
}

func (s *Store) CreateProject(ctx context.Context, p api.Project) error {
	return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
		_, err := tx.ExecContext(ctx, `
			INSERT INTO projects (id, name, repo_path, default_branch, remote, created_by, created_at)
			VALUES (?, ?, ?, ?, ?, ?, ?)
		`, p.Id, p.Name, p.RepoPath, p.DefaultBranch, p.Remote, p.CreatedBy, formatTime(p.CreatedAt))
		if err != nil {
			return err
		}
		return emit(api.Event{Kind: "project.created", ProjectId: &p.Id, Actor: p.CreatedBy, Payload: ToPayload(p)})
	})
}

func (s *Store) ListProjects(ctx context.Context) ([]api.Project, error) {
	rows, err := s.db.QueryContext(ctx, `SELECT `+projectColumns+` FROM projects ORDER BY name`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	projects := []api.Project{}
	for rows.Next() {
		p, err := scanProject(rows)
		if err != nil {
			return nil, err
		}
		projects = append(projects, p)
	}
	return projects, rows.Err()
}

func (s *Store) GetProject(ctx context.Context, id string) (api.Project, error) {
	p, err := scanProject(s.db.QueryRowContext(ctx, `SELECT `+projectColumns+` FROM projects WHERE id = ?`, id))
	if errors.Is(err, sql.ErrNoRows) {
		return p, ErrNotFound
	}
	return p, err
}

func (s *Store) CreateTask(ctx context.Context, t api.Task, frozen *FrozenWorkflow, actor string) error {
	workflowName := ""
	workflowDefinition := ""
	workflowBrief := ""
	workflowTitle := ""
	workflowRun := ""
	workflowStage := ""
	if frozen != nil {
		b, err := json.Marshal(frozen.Definition)
		if err != nil {
			return err
		}
		workflowName = frozen.Name
		workflowDefinition = string(b)
		workflowBrief = frozen.Brief
		workflowTitle = frozen.Title
		workflowRun = *t.WorkflowRun
		workflowStage = *t.WorkflowStage
		t.Workflow = &workflowName
	}
	todoItem := ""
	if t.TodoItem != nil {
		todoItem = *t.TodoItem
	}
	mergeTitle := ""
	if t.MergeTitle != t.Title {
		mergeTitle = t.MergeTitle
	}
	forkedFrom := ""
	if t.ForkedFrom != nil {
		forkedFrom = *t.ForkedFrom
	}
	forkedAt := 0
	if t.ForkedAt != nil {
		forkedAt = *t.ForkedAt
	}
	return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
		err := tx.QueryRowContext(ctx, `
			SELECT min(candidate) FROM (
				SELECT ? AS candidate
				UNION ALL SELECT port_base + ? FROM tasks WHERE status = ?
			) WHERE candidate NOT IN (SELECT port_base FROM tasks WHERE status = ?)
		`, s.portFirst, PortBlock, string(api.TaskStatusActive), string(api.TaskStatusActive)).Scan(&t.PortBase)
		if err != nil {
			return err
		}
		err = tx.QueryRowContext(ctx, `
			WITH RECURSIVE palette(c) AS (SELECT 0 UNION ALL SELECT c + 1 FROM palette WHERE c + 1 < ?)
			SELECT c FROM palette
			ORDER BY c IN (SELECT color FROM tasks WHERE status = ?), coalesce((SELECT max(rowid) FROM tasks WHERE color = c), -1), c
			LIMIT 1
		`, TaskColors, string(api.TaskStatusActive)).Scan(&t.Color)
		if err != nil {
			return err
		}
		_, err = tx.ExecContext(ctx, `
			INSERT INTO tasks (id, project_id, title, merge_title, branch, worktree_path, base_ref, port_base, color, status, created_by, created_at, todo_item, forked_from, forked_at, workflow_name, workflow_definition, workflow_brief, workflow_title, workflow_run, workflow_stage)
			VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
		`, t.Id, t.ProjectId, t.Title, mergeTitle, t.Branch, t.WorktreePath, t.BaseRef, t.PortBase, t.Color, string(t.Status), t.CreatedBy, formatTime(t.CreatedAt), todoItem, forkedFrom, forkedAt, workflowName, workflowDefinition, workflowBrief, workflowTitle, workflowRun, workflowStage)
		if err != nil {
			return err
		}
		return emit(api.Event{Kind: "task.created", ProjectId: &t.ProjectId, TaskId: &t.Id, Actor: actor, Payload: ToPayload(t)})
	})
}

func (s *Store) ListTasks(ctx context.Context, projectID string) ([]api.Task, error) {
	return s.queryTasks(ctx, `SELECT `+taskColumns+` FROM tasks WHERE project_id = ? ORDER BY created_at DESC`, projectID)
}

func (s *Store) GetTask(ctx context.Context, id string) (api.Task, error) {
	t, err := scanTask(s.db.QueryRowContext(ctx, `SELECT `+taskColumns+` FROM tasks WHERE id = ?`, id))
	if errors.Is(err, sql.ErrNoRows) {
		return t, ErrNotFound
	}
	return t, err
}

func (s *Store) SetMergeTitle(ctx context.Context, id, title, actor string) error {
	return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
		projectID := ""
		err := tx.QueryRowContext(ctx, `UPDATE tasks SET merge_title = ? WHERE id = ? AND status = ? AND merging_by = '' AND merge_queued_by = '' RETURNING project_id`,
			title, id, string(api.TaskStatusActive)).Scan(&projectID)
		if errors.Is(err, sql.ErrNoRows) {
			return ErrNotFound
		}
		if err != nil {
			return err
		}
		return emit(api.Event{Kind: "task.merge_title", ProjectId: &projectID, TaskId: &id, Actor: actor, Payload: map[string]any{"id": id, "merge_title": title}})
	})
}

func (s *Store) SetMergingBy(ctx context.Context, id, login string) error {
	return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
		projectID := ""
		err := tx.QueryRowContext(ctx, `UPDATE tasks SET merging_by = ? WHERE id = ? RETURNING project_id`, login, id).Scan(&projectID)
		if errors.Is(err, sql.ErrNoRows) {
			return ErrNotFound
		}
		if err != nil {
			return err
		}
		return emit(api.Event{Kind: "task.merge_started", ProjectId: &projectID, TaskId: &id, Actor: login, Payload: map[string]any{"id": id}})
	})
}

func (s *Store) MergingBy(ctx context.Context, id string) (string, error) {
	login := ""
	err := s.db.QueryRowContext(ctx, `SELECT merging_by FROM tasks WHERE id = ?`, id).Scan(&login)
	if errors.Is(err, sql.ErrNoRows) {
		return "", ErrNotFound
	}
	return login, err
}

func (s *Store) TakeMergingBy(ctx context.Context, id string) (string, error) {
	login := ""
	err := s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
		err := tx.QueryRowContext(ctx, `SELECT merging_by FROM tasks WHERE id = ?`, id).Scan(&login)
		if errors.Is(err, sql.ErrNoRows) {
			return ErrNotFound
		}
		if err != nil {
			return err
		}
		_, err = tx.ExecContext(ctx, `UPDATE tasks SET merging_by = '' WHERE id = ?`, id)
		return err
	})
	return login, err
}

func (s *Store) MergingTasks(ctx context.Context) ([]api.Task, error) {
	return s.queryTasks(ctx, `SELECT `+taskColumns+` FROM tasks WHERE status = ? AND merging_by != ''`, string(api.TaskStatusActive))
}

func (s *Store) QueuedMerges(ctx context.Context, projectID string) ([]api.Task, error) {
	return s.queryTasks(ctx, `SELECT `+taskColumns+` FROM tasks WHERE project_id = ? AND status = ? AND merge_queued_by != '' ORDER BY merge_queued_at`,
		projectID, string(api.TaskStatusActive))
}

func (s *Store) ProjectsWithQueuedMerges(ctx context.Context) ([]string, error) {
	rows, err := s.db.QueryContext(ctx, `SELECT DISTINCT project_id FROM tasks WHERE status = ? AND merge_queued_by != ''`, string(api.TaskStatusActive))
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	ids := []string{}
	for rows.Next() {
		id := ""
		err = rows.Scan(&id)
		if err != nil {
			return nil, err
		}
		ids = append(ids, id)
	}
	return ids, rows.Err()
}

func (s *Store) queryTasks(ctx context.Context, query string, args ...any) ([]api.Task, error) {
	rows, err := s.db.QueryContext(ctx, query, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	tasks := []api.Task{}
	for rows.Next() {
		t, err := scanTask(rows)
		if err != nil {
			return nil, err
		}
		tasks = append(tasks, t)
	}
	return tasks, rows.Err()
}

func (s *Store) QueueMerge(ctx context.Context, id, login string) error {
	return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
		projectID := ""
		err := tx.QueryRowContext(ctx, `UPDATE tasks SET merge_queued_by = ?, merge_queued_at = ? WHERE id = ? AND status = ? AND merge_queued_by = '' RETURNING project_id`,
			login, now(), id, string(api.TaskStatusActive)).Scan(&projectID)
		if errors.Is(err, sql.ErrNoRows) {
			return ErrNotFound
		}
		if err != nil {
			return err
		}
		return emit(api.Event{Kind: "task.merge_queued", ProjectId: &projectID, TaskId: &id, Actor: login, Payload: map[string]any{"id": id}})
	})
}

func (s *Store) TakeQueuedMerge(ctx context.Context, id string) (string, error) {
	login := ""
	err := s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
		projectID := ""
		err := tx.QueryRowContext(ctx, `SELECT project_id, merge_queued_by FROM tasks WHERE id = ?`, id).Scan(&projectID, &login)
		if errors.Is(err, sql.ErrNoRows) {
			return ErrNotFound
		}
		if err != nil {
			return err
		}
		if login == "" {
			return nil
		}
		_, err = tx.ExecContext(ctx, `UPDATE tasks SET merge_queued_by = '', merge_queued_at = '' WHERE id = ?`, id)
		if err != nil {
			return err
		}
		return emit(api.Event{Kind: "task.merge_unqueued", ProjectId: &projectID, TaskId: &id, Actor: "byre", Payload: map[string]any{"id": id}})
	})
	return login, err
}

func (s *Store) MergeTask(ctx context.Context, id, commit, pushed, actor string) (api.Task, error) {
	return s.closeTask(ctx, id, api.TaskStatusMerged, commit, actor, "task.merged", map[string]any{"id": id, "commit": commit, "pushed": pushed})
}

func (s *Store) DiscardTask(ctx context.Context, id, actor string) (api.Task, error) {
	return s.closeTask(ctx, id, api.TaskStatusDiscarded, "", actor, "task.discarded", map[string]any{"id": id})
}

func (s *Store) closeTask(ctx context.Context, id string, status api.TaskStatus, commit, actor, kind string, payload map[string]any) (api.Task, error) {
	err := s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
		res, err := tx.ExecContext(ctx, `UPDATE tasks SET status = ?, closed_at = ?, merge_commit = ? WHERE id = ? AND status = ?`,
			string(status), now(), commit, id, string(api.TaskStatusActive))
		if err != nil {
			return err
		}
		n, err := res.RowsAffected()
		if err != nil {
			return err
		}
		if n == 0 {
			return ErrNotFound
		}

		projectID := ""
		err = tx.QueryRowContext(ctx, `SELECT project_id FROM tasks WHERE id = ?`, id).Scan(&projectID)
		if err != nil {
			return err
		}
		err = emit(api.Event{Kind: kind, ProjectId: &projectID, TaskId: &id, Actor: actor, Payload: payload})
		if err != nil {
			return err
		}
		_, err = tx.ExecContext(ctx, `DELETE FROM task_inputs WHERE task_id = ?`, id)
		if err != nil {
			return err
		}
		return dismissTaskInbox(ctx, tx, emit, id, actor)
	})
	if err != nil {
		return api.Task{}, err
	}
	return s.GetTask(ctx, id)
}

func (s *Store) Participants(ctx context.Context, taskID string) ([]api.Participant, error) {
	rows, err := s.db.QueryContext(ctx, `
		SELECT u.login, u.display_name, u.git_name, u.git_email
		FROM events e JOIN users u ON u.login = e.actor
		WHERE e.task_id = ? AND (e.kind = 'user.message' OR (e.kind = 'task.created' AND json_extract(e.payload, '$.forked_from') IS NOT NULL))
		GROUP BY u.login ORDER BY min(e.seq)
	`, taskID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	participants := []api.Participant{}
	for rows.Next() {
		p := api.Participant{}
		err = rows.Scan(&p.Login, &p.DisplayName, &p.GitName, &p.GitEmail)
		if err != nil {
			return nil, err
		}
		participants = append(participants, p)
	}
	return participants, rows.Err()
}

const eventColumns = `seq, kind, project_id, task_id, agent_session_id, owner, actor, payload, created_at`

func (s *Store) EventsAfter(ctx context.Context, viewer string, after int64, limit int) ([]api.Event, error) {
	rows, err := s.db.QueryContext(ctx, `
		SELECT `+eventColumns+`
		FROM events WHERE seq > ? AND (owner IS NULL OR owner = ?) ORDER BY seq LIMIT ?
	`, after, viewer, limit)
	if err != nil {
		return nil, err
	}
	return scanEvents(rows)
}

func (s *Store) TaskEvents(ctx context.Context, taskID string, after int64, limit int) ([]api.Event, error) {
	rows, err := s.db.QueryContext(ctx, `
		SELECT `+eventColumns+`
		FROM events WHERE task_id = ? AND seq > ? ORDER BY seq LIMIT ?
	`, taskID, after, limit)
	if err != nil {
		return nil, err
	}
	return scanEvents(rows)
}

func (s *Store) TaskEventsTail(ctx context.Context, taskID string, limit int) ([]api.Event, error) {
	rows, err := s.db.QueryContext(ctx, `
		SELECT * FROM (SELECT `+eventColumns+` FROM events WHERE task_id = ? ORDER BY seq DESC LIMIT ?) ORDER BY seq
	`, taskID, limit)
	if err != nil {
		return nil, err
	}
	return scanEvents(rows)
}

type Handoff struct {
	SessionID string
	At        time.Time
	Text      string
}

func (s *Store) TaskHandoffs(ctx context.Context, taskID string) ([]Handoff, error) {
	rows, err := s.db.QueryContext(ctx, `
		SELECT agent_session_id, created_at, coalesce(json_extract(payload, '$.result'), '') FROM events
		WHERE task_id = ? AND kind = 'turn.ended' AND json_extract(payload, '$.handoff') = 1
		ORDER BY seq
	`, taskID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	handoffs := []Handoff{}
	for rows.Next() {
		h := Handoff{}
		at := ""
		err = rows.Scan(&h.SessionID, &at, &h.Text)
		if err != nil {
			return nil, err
		}
		h.At, err = parseTime(at)
		if err != nil {
			return nil, err
		}
		handoffs = append(handoffs, h)
	}
	return handoffs, rows.Err()
}

func (s *Store) LeadEvents(ctx context.Context, owner string, after int64, limit int) ([]api.Event, error) {
	rows, err := s.db.QueryContext(ctx, `
		SELECT `+eventColumns+`
		FROM events WHERE owner = ? AND agent_session_id IS NOT NULL AND seq > ? ORDER BY seq LIMIT ?
	`, owner, after, limit)
	if err != nil {
		return nil, err
	}
	return scanEvents(rows)
}

func scanEvents(rows *sql.Rows) ([]api.Event, error) {
	defer rows.Close()

	events := []api.Event{}
	for rows.Next() {
		ev := api.Event{}
		payload := ""
		createdAt := ""
		err := rows.Scan(&ev.Seq, &ev.Kind, &ev.ProjectId, &ev.TaskId, &ev.AgentSessionId, &ev.Owner, &ev.Actor, &payload, &createdAt)
		if err != nil {
			return nil, err
		}
		err = json.Unmarshal([]byte(payload), &ev.Payload)
		if err != nil {
			return nil, err
		}
		ev.CreatedAt, err = parseTime(createdAt)
		if err != nil {
			return nil, err
		}
		events = append(events, ev)
	}
	return events, rows.Err()
}

func (s *Store) AppendEvent(ctx context.Context, ev api.Event) error {
	return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
		return emit(ev)
	})
}

func (s *Store) Broadcast(ev api.Event) {
	s.writeMu.Lock()
	defer s.writeMu.Unlock()
	ev.CreatedAt = time.Now().UTC()
	s.onEvent(ev)
}

func (s *Store) CreateSession(ctx context.Context, sess api.AgentSession, projectID string) error {
	forkedFrom := ""
	forkedAt := ""
	if sess.ForkedFrom != nil && sess.ForkedAt != nil {
		forkedFrom = *sess.ForkedFrom
		forkedAt = *sess.ForkedAt
	}
	return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
		_, err := tx.ExecContext(ctx, `
			INSERT INTO agent_sessions (id, task_id, owner, vendor, model, vendor_session_id, state, role, created_by, created_at, forked_from, forked_at)
			VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
		`, sess.Id, nullable(sess.TaskId), sess.Owner, string(sess.Vendor), sess.Model, sess.VendorSessionId, string(sess.State), string(sess.Role), sess.CreatedBy, formatTime(sess.CreatedAt), forkedFrom, forkedAt)
		if err != nil {
			return err
		}
		return emit(sessionEvent("session.created", projectID, sess.TaskId, sess.Owner, sess.Id, sess.CreatedBy, ToPayload(sess)))
	})
}

func sessionEvent(kind, projectID, taskID, owner, sessionID, actor string, payload map[string]any) api.Event {
	return api.Event{
		Kind:           kind,
		ProjectId:      nullable(projectID),
		TaskId:         nullable(taskID),
		AgentSessionId: &sessionID,
		Owner:          nullable(owner),
		Actor:          actor,
		Payload:        payload,
	}
}

func nullable(s string) *string {
	if s == "" {
		return nil
	}
	return &s
}

func (s *Store) LeadSessions(ctx context.Context, owner string) ([]api.AgentSession, error) {
	return s.querySessions(ctx, `SELECT `+sessionColumns+` FROM agent_sessions WHERE owner = ? ORDER BY created_at`, owner)
}

func (s *Store) LeadOwners(ctx context.Context) ([]string, error) {
	rows, err := s.db.QueryContext(ctx, `SELECT DISTINCT owner FROM agent_sessions WHERE owner != '' ORDER BY owner`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	owners := []string{}
	for rows.Next() {
		owner := ""
		err = rows.Scan(&owner)
		if err != nil {
			return nil, err
		}
		owners = append(owners, owner)
	}
	return owners, rows.Err()
}

func (s *Store) GetSession(ctx context.Context, id string) (api.AgentSession, error) {
	sess, err := scanSession(s.db.QueryRowContext(ctx, `SELECT `+sessionColumns+` FROM agent_sessions WHERE id = ?`, id))
	if errors.Is(err, sql.ErrNoRows) {
		return sess, ErrNotFound
	}
	return sess, err
}

func (s *Store) ListSessions(ctx context.Context, taskID string) ([]api.AgentSession, error) {
	return s.querySessions(ctx, `SELECT `+sessionColumns+` FROM agent_sessions WHERE task_id = ? AND role != ? ORDER BY created_at`, taskID, string(api.Reviewer))
}

func (s *Store) SessionsInState(ctx context.Context, state api.AgentSessionState) ([]api.AgentSession, error) {
	return s.querySessions(ctx, `SELECT `+sessionColumns+` FROM agent_sessions WHERE state = ? ORDER BY created_at`, string(state))
}

func (s *Store) querySessions(ctx context.Context, query string, args ...any) ([]api.AgentSession, error) {
	rows, err := s.db.QueryContext(ctx, query, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	sessions := []api.AgentSession{}
	for rows.Next() {
		sess, err := scanSession(rows)
		if err != nil {
			return nil, err
		}
		sessions = append(sessions, sess)
	}
	return sessions, rows.Err()
}

var ErrStateChanged = errors.New("session state changed")

func (s *Store) SetSessionState(ctx context.Context, id string, state api.AgentSessionState, actor string, payload map[string]any) error {
	return s.transitionSession(ctx, id, nil, state, actor, payload)
}

func (s *Store) TransitionSession(ctx context.Context, id string, from, to api.AgentSessionState, actor string) error {
	return s.transitionSession(ctx, id, &from, to, actor, nil)
}

func (s *Store) transitionSession(ctx context.Context, id string, from *api.AgentSessionState, state api.AgentSessionState, actor string, payload map[string]any) error {
	return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
		taskID := ""
		projectID := ""
		owner := ""
		current := ""
		err := tx.QueryRowContext(ctx, `
			SELECT coalesce(t.id, ''), coalesce(t.project_id, ''), s.owner, s.state FROM agent_sessions s LEFT JOIN tasks t ON t.id = s.task_id WHERE s.id = ?
		`, id).Scan(&taskID, &projectID, &owner, &current)
		if errors.Is(err, sql.ErrNoRows) {
			return ErrNotFound
		}
		if err != nil {
			return err
		}

		if from != nil && current != string(*from) {
			return fmt.Errorf("%w: session %s is %s, not %s", ErrStateChanged, id, current, *from)
		}

		_, err = tx.ExecContext(ctx, `UPDATE agent_sessions SET state = ? WHERE id = ?`, string(state), id)
		if err != nil {
			return err
		}

		if payload == nil {
			payload = map[string]any{}
		}
		payload["state"] = string(state)
		return emit(sessionEvent("session.state", projectID, taskID, owner, id, actor, payload))
	})
}

func (s *Store) SessionStarted(ctx context.Context, sessionID string) (bool, error) {
	started := false
	err := s.db.QueryRowContext(ctx, `SELECT EXISTS (SELECT 1 FROM events WHERE agent_session_id = ? AND kind = 'user.message')`, sessionID).Scan(&started)
	return started, err
}

func (s *Store) SetVendorSessionID(ctx context.Context, id, vendorSessionID string) error {
	return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
		res, err := tx.ExecContext(ctx, `UPDATE agent_sessions SET vendor_session_id = ? WHERE id = ?`, vendorSessionID, id)
		if err != nil {
			return err
		}
		n, err := res.RowsAffected()
		if err != nil {
			return err
		}
		if n == 0 {
			return ErrNotFound
		}
		return nil
	})
}

func (s *Store) SetSessionModel(ctx context.Context, id, model, actor string) error {
	return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
		taskID := ""
		projectID := ""
		owner := ""
		err := tx.QueryRowContext(ctx, `
			SELECT coalesce(t.id, ''), coalesce(t.project_id, ''), s.owner FROM agent_sessions s LEFT JOIN tasks t ON t.id = s.task_id WHERE s.id = ?
		`, id).Scan(&taskID, &projectID, &owner)
		if errors.Is(err, sql.ErrNoRows) {
			return ErrNotFound
		}
		if err != nil {
			return err
		}

		_, err = tx.ExecContext(ctx, `UPDATE agent_sessions SET model = ? WHERE id = ?`, model, id)
		if err != nil {
			return err
		}

		return emit(sessionEvent("session.model", projectID, taskID, owner, id, actor, map[string]any{"model": model}))
	})
}

type Counts struct {
	LastEventSeq int64
	Projects     int
	ActiveTasks  int
}

func (s *Store) Counts(ctx context.Context) (Counts, error) {
	c := Counts{}
	err := s.db.QueryRowContext(ctx, `
		SELECT
			(SELECT COALESCE(MAX(seq), 0) FROM events),
			(SELECT COUNT(*) FROM projects),
			(SELECT COUNT(*) FROM tasks WHERE status = ?)
	`, string(api.TaskStatusActive)).Scan(&c.LastEventSeq, &c.Projects, &c.ActiveTasks)
	return c, err
}

type emitFunc func(api.Event) error

func (s *Store) write(ctx context.Context, fn func(tx *sql.Tx, emit emitFunc) error) error {
	s.writeMu.Lock()
	defer s.writeMu.Unlock()

	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer tx.Rollback()

	events := []api.Event{}
	emit := func(ev api.Event) error {
		ev.CreatedAt = time.Now().UTC()
		payload, err := json.Marshal(ev.Payload)
		if err != nil {
			return err
		}
		err = tx.QueryRowContext(ctx, `
			INSERT INTO events (kind, project_id, task_id, agent_session_id, owner, actor, payload, created_at)
			VALUES (?, ?, ?, ?, ?, ?, ?, ?) RETURNING seq
		`, ev.Kind, ev.ProjectId, ev.TaskId, ev.AgentSessionId, ev.Owner, ev.Actor, string(payload), formatTime(ev.CreatedAt)).Scan(&ev.Seq)
		if err != nil {
			return err
		}
		events = append(events, ev)
		return nil
	}

	err = fn(tx, emit)
	if err != nil {
		return err
	}

	err = tx.Commit()
	if err != nil {
		return err
	}

	for _, ev := range events {
		s.onEvent(ev)
	}
	return nil
}

const projectColumns = `id, name, repo_path, default_branch, remote, created_by, created_at`

const taskColumns = `id, project_id, title, merge_title, branch, worktree_path, base_ref, port_base, color, status, created_by, created_at, closed_at, merge_commit, todo_item, forked_from, forked_at, merge_queued_at, workflow_name, workflow_run, workflow_stage, EXISTS (SELECT 1 FROM reviews WHERE reviews.task_id = tasks.id AND reviews.state = 'running')`

const sessionColumns = `id, coalesce(task_id, ''), owner, vendor, model, vendor_session_id, state, role, created_by, created_at, forked_from, forked_at`

type scanner interface {
	Scan(dest ...any) error
}

func scanProject(row scanner) (api.Project, error) {
	p := api.Project{}
	createdAt := ""
	err := row.Scan(&p.Id, &p.Name, &p.RepoPath, &p.DefaultBranch, &p.Remote, &p.CreatedBy, &createdAt)
	if err != nil {
		return p, err
	}
	p.CreatedAt, err = parseTime(createdAt)
	return p, err
}

func scanTask(row scanner) (api.Task, error) {
	t := api.Task{}
	status := ""
	createdAt := ""
	closedAt := sql.NullString{}
	mergeCommit := ""
	todoItem := ""
	forkedFrom := ""
	forkedAt := 0
	mergeQueuedAt := ""
	workflowName := ""
	workflowRun := ""
	workflowStage := ""
	err := row.Scan(&t.Id, &t.ProjectId, &t.Title, &t.MergeTitle, &t.Branch, &t.WorktreePath, &t.BaseRef, &t.PortBase, &t.Color, &status, &t.CreatedBy, &createdAt, &closedAt, &mergeCommit, &todoItem, &forkedFrom, &forkedAt, &mergeQueuedAt, &workflowName, &workflowRun, &workflowStage, &t.Reviewing)
	if err != nil {
		return t, err
	}
	if workflowName != "" {
		t.Workflow = &workflowName
		t.WorkflowRun = &workflowRun
		t.WorkflowStage = &workflowStage
	}
	if t.MergeTitle == "" {
		t.MergeTitle = t.Title
	}
	t.Status = api.TaskStatus(status)
	t.CreatedAt, err = parseTime(createdAt)
	if err != nil {
		return t, err
	}
	if closedAt.Valid {
		ct, err := parseTime(closedAt.String)
		if err != nil {
			return t, err
		}
		t.ClosedAt = &ct
	}
	if mergeCommit != "" {
		t.MergeCommit = &mergeCommit
	}
	if todoItem != "" {
		t.TodoItem = &todoItem
	}
	if forkedFrom != "" {
		t.ForkedFrom = &forkedFrom
		t.ForkedAt = &forkedAt
	}
	if mergeQueuedAt != "" {
		qt, err := parseTime(mergeQueuedAt)
		if err != nil {
			return t, err
		}
		t.MergeQueuedAt = &qt
	}
	return t, nil
}

func scanSession(row scanner) (api.AgentSession, error) {
	sess := api.AgentSession{}
	vendor := ""
	state := ""
	role := ""
	createdAt := ""
	forkedFrom := ""
	forkedAt := ""
	err := row.Scan(&sess.Id, &sess.TaskId, &sess.Owner, &vendor, &sess.Model, &sess.VendorSessionId, &state, &role, &sess.CreatedBy, &createdAt, &forkedFrom, &forkedAt)
	if err != nil {
		return sess, err
	}
	if forkedFrom != "" {
		sess.ForkedFrom = &forkedFrom
		sess.ForkedAt = &forkedAt
	}
	sess.Vendor = api.AgentSessionVendor(vendor)
	sess.State = api.AgentSessionState(state)
	sess.Role = api.AgentSessionRole(role)
	sess.CreatedAt, err = parseTime(createdAt)
	return sess, err
}

func ToPayload(v any) map[string]any {
	b, err := json.Marshal(v)
	if err != nil {
		panic(err)
	}
	m := map[string]any{}
	err = json.Unmarshal(b, &m)
	if err != nil {
		panic(err)
	}
	return m
}

func now() string {
	return formatTime(time.Now())
}

const timeLayout = "2006-01-02T15:04:05.000000000Z07:00"

func formatTime(t time.Time) string {
	return t.UTC().Format(timeLayout)
}

func parseTime(s string) (time.Time, error) {
	return time.Parse(time.RFC3339Nano, s)
}