byre/internal/store/store.go

package store

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

	_ "modernc.org/sqlite"

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

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

//go:embed migrations/*.sql
var migrationFiles embed.FS

type migration struct {
	name string
	sql  string
}

func loadMigrations() ([]migration, error) {
	entries, err := migrationFiles.ReadDir("migrations")
	if err != nil {
		return nil, err
	}

	migrations := []migration{}
	for _, e := range entries {
		num, name, ok := strings.Cut(strings.TrimSuffix(e.Name(), ".sql"), "-")
		_, err = strconv.Atoi(num)
		if !ok || len(num) != 3 || err != nil {
			return nil, fmt.Errorf("migration file %s: want NNN-name.sql", e.Name())
		}

		sql, err := migrationFiles.ReadFile("migrations/" + e.Name())
		if err != nil {
			return nil, err
		}
		migrations = append(migrations, migration{name: name, sql: string(sql)})
	}
	return migrations, nil
}

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)
	if err != nil {
		db.Close()
		return nil, err
	}

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

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

	_, err = db.Exec(`CREATE TABLE IF NOT EXISTS migrations (name TEXT PRIMARY KEY, applied_at TEXT NOT NULL)`)
	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 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, default_model, review_model, second_review_model FROM users WHERE login = ?`, login).
		Scan(&u.Login, &u.DisplayName, &u.GitName, &u.GitEmail, &prefs, &u.DefaultModel, &u.ReviewModel, &u.SecondReviewModel)
	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
	}
	parentTaskID := ""
	if t.ParentTaskId != nil {
		parentTaskID = *t.ParentTaskId
	}
	return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
		if parentTaskID != "" {
			open := false
			err := tx.QueryRowContext(ctx, `SELECT EXISTS (SELECT 1 FROM tasks WHERE id = ? AND status = ? AND discard_state = '' AND finalizing = '')`,
				parentTaskID, string(api.TaskStatusActive)).Scan(&open)
			if err != nil {
				return err
			}
			if !open {
				return ErrParentNotOpen
			}
		}
		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, parent_task_id)
			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, parentTaskID)
		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 finalizing = '' 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 = ? AND discard_state = '' AND finalizing = '' 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 {
		projectID := ""
		err := tx.QueryRowContext(ctx, `SELECT project_id, merging_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 merging_by = '' WHERE id = ?`, id)
		if err != nil {
			return err
		}
		return emit(api.Event{Kind: "task.merge_ended", ProjectId: &projectID, TaskId: &id, Actor: "byre", Payload: map[string]any{"id": id}})
	})
	return login, err
}

func (s *Store) ClearFinalizing(ctx context.Context, id string) error {
	return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
		projectID := ""
		err := tx.QueryRowContext(ctx, `UPDATE tasks SET finalizing = '', finalizing_commit = '', finalizing_notes = '', finalizing_unchanged = 0 WHERE id = ? AND finalizing != '' RETURNING project_id`, id).Scan(&projectID)
		if errors.Is(err, sql.ErrNoRows) {
			return nil
		}
		if err != nil {
			return err
		}
		return emit(api.Event{Kind: "task.merge_ended", ProjectId: &projectID, TaskId: &id, Actor: "byre", Payload: map[string]any{"id": id}})
	})
}

func (s *Store) SetFinalizingCommit(ctx context.Context, id, commit, notes string) error {
	return s.setFinalizing(ctx, id, commit, notes, false)
}

func (s *Store) SetFinalizingUnchanged(ctx context.Context, id string) error {
	return s.setFinalizing(ctx, id, "", "", true)
}

func (s *Store) setFinalizing(ctx context.Context, id, commit, notes string, unchanged bool) error {
	return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
		res, err := tx.ExecContext(ctx, `UPDATE tasks SET finalizing_commit = ?, finalizing_notes = ?, finalizing_unchanged = ? WHERE id = ? AND status = ? AND finalizing != ''`,
			commit, notes, unchanged, id, string(api.TaskStatusActive))
		if err != nil {
			return err
		}
		n, err := res.RowsAffected()
		if err != nil {
			return err
		}
		if n == 0 {
			return ErrNotFound
		}
		return nil
	})
}

type FinalizingMerge struct {
	Task      api.Task
	By        string
	Commit    string
	Notes     string
	Unchanged bool
}

func (s *Store) FinalizingMerges(ctx context.Context) ([]FinalizingMerge, error) {
	rows, err := s.db.QueryContext(ctx, `SELECT `+taskColumns+`, finalizing, finalizing_commit, finalizing_notes, finalizing_unchanged FROM tasks WHERE status = ? AND finalizing != ''`, string(api.TaskStatusActive))
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	merges := []FinalizingMerge{}
	for rows.Next() {
		f := FinalizingMerge{}
		f.Task, err = scanTask(trailingScanner{rows, []any{&f.By, &f.Commit, &f.Notes, &f.Unchanged}})
		if err != nil {
			return nil, err
		}
		merges = append(merges, f)
	}
	return merges, rows.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) ActiveSubtasks(ctx context.Context, parentID string) ([]api.Task, error) {
	return s.queryTasks(ctx, `SELECT `+taskColumns+` FROM tasks WHERE parent_task_id = ? AND status = ? ORDER BY created_at`, parentID, string(api.TaskStatusActive))
}

func (s *Store) SetParentMerge(ctx context.Context, id, login string) error {
	return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
		projectID, parentID, status, parentMergeBy, discardState, finalizing := "", "", "", "", "", ""
		err := tx.QueryRowContext(ctx, `SELECT project_id, parent_task_id, status, parent_merge_by, discard_state, finalizing FROM tasks WHERE id = ?`, id).
			Scan(&projectID, &parentID, &status, &parentMergeBy, &discardState, &finalizing)
		if errors.Is(err, sql.ErrNoRows) {
			return ErrNotFound
		}
		if err != nil {
			return err
		}
		switch {
		case status != string(api.TaskStatusActive) || parentID == "":
			return ErrNotFound
		case discardState != "":
			return ErrDiscardPending
		case finalizing != "":
			return ErrMerging
		case parentMergeBy != "":
			return ErrParentMergeRequested
		}
		_, err = tx.ExecContext(ctx, `UPDATE tasks SET parent_merge_by = ? WHERE id = ?`, login, id)
		if err != nil {
			return err
		}
		return emit(api.Event{Kind: "task.parent_merge_started", ProjectId: &projectID, TaskId: &id, Actor: login, Payload: map[string]any{"id": id, "parent_task_id": parentID}})
	})
}

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

func (s *Store) ParentMergeBy(ctx context.Context, id string) (string, error) {
	login := ""
	err := s.db.QueryRowContext(ctx, `SELECT parent_merge_by FROM tasks WHERE id = ?`, id).Scan(&login)
	if errors.Is(err, sql.ErrNoRows) {
		return "", ErrNotFound
	}
	return login, 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 = '' AND discard_state = '' 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
}

type MergeAfterReview struct {
	By       string
	ReviewID string
	Target   api.MergeTaskJSONBodyTarget
}

func (s *Store) SetMergeAfterReview(ctx context.Context, id string, mar MergeAfterReview) error {
	return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
		projectID := ""
		snapshotN := 0
		err := tx.QueryRowContext(ctx, `
			UPDATE tasks SET merge_after_review_by = ?, merge_after_review_id = ?, merge_after_review_target = ?
			WHERE id = ? AND status = ? AND discard_state = '' AND merging_by = '' AND finalizing = '' AND merge_queued_by = '' AND parent_merge_by = ''
			AND EXISTS (SELECT 1 FROM reviews WHERE reviews.id = ? AND reviews.task_id = tasks.id AND reviews.state = ?)
			RETURNING project_id, (SELECT snapshot_n FROM reviews WHERE reviews.id = ?)`,
			mar.By, mar.ReviewID, string(mar.Target), id, string(api.TaskStatusActive), mar.ReviewID, string(api.ReviewStateRunning), mar.ReviewID).Scan(&projectID, &snapshotN)
		if errors.Is(err, sql.ErrNoRows) {
			return ErrNotFound
		}
		if err != nil {
			return err
		}
		return emit(api.Event{Kind: "task.merge_after_review", ProjectId: &projectID, TaskId: &id, Actor: mar.By, Payload: map[string]any{"id": id, "review_id": mar.ReviewID, "snapshot_n": snapshotN, "target": string(mar.Target)}})
	})
}

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

func (s *Store) FinishedMergeAfterReviews(ctx context.Context) ([]string, error) {
	rows, err := s.db.QueryContext(ctx, `
		SELECT tasks.merge_after_review_id FROM tasks JOIN reviews ON reviews.id = tasks.merge_after_review_id
		WHERE tasks.merge_after_review_by != '' AND reviews.state != ?`, string(api.ReviewStateRunning))
	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) TakeMergeAfterReview(ctx context.Context, id, reviewID string) (MergeAfterReview, error) {
	mar := MergeAfterReview{}
	err := s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
		projectID, target := "", ""
		err := tx.QueryRowContext(ctx, `SELECT project_id, merge_after_review_by, merge_after_review_id, merge_after_review_target FROM tasks WHERE id = ?`, id).
			Scan(&projectID, &mar.By, &mar.ReviewID, &target)
		if errors.Is(err, sql.ErrNoRows) {
			return ErrNotFound
		}
		if err != nil {
			return err
		}
		mar.Target = api.MergeTaskJSONBodyTarget(target)
		if mar.By == "" || (reviewID != "" && mar.ReviewID != reviewID) {
			mar = MergeAfterReview{}
			return nil
		}
		_, err = tx.ExecContext(ctx, `UPDATE tasks SET merge_after_review_by = '', merge_after_review_id = '', merge_after_review_target = '' WHERE id = ?`, id)
		if err != nil {
			return err
		}
		return emit(api.Event{Kind: "task.merge_after_review_ended", ProjectId: &projectID, TaskId: &id, Actor: "byre", Payload: map[string]any{"id": id}})
	})
	return mar, err
}

func (s *Store) SetTaskSummary(ctx context.Context, id, summary, actor string) error {
	return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
		projectID := ""
		err := tx.QueryRowContext(ctx, `UPDATE tasks SET summary = ? WHERE id = ? AND status = ? RETURNING project_id`,
			summary, 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.summary", ProjectId: &projectID, TaskId: &id, Actor: actor, Payload: map[string]any{"id": id, "summary": summary}})
	})
}

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

func (s *Store) SetReadyPromptN(ctx context.Context, id string, snapshotN int) error {
	_, err := s.db.ExecContext(ctx, `UPDATE tasks SET ready_prompt_n = ? WHERE id = ?`, snapshotN, id)
	return err
}

func (s *Store) MarkReady(ctx context.Context, id, overview, actor string) error {
	return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
		projectID := ""
		err := tx.QueryRowContext(ctx, `SELECT project_id FROM tasks WHERE id = ? AND status = ?`, 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.ready", ProjectId: &projectID, TaskId: &id, Actor: actor, Payload: map[string]any{"id": id, "overview": overview}})
	})
}

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

type ClosedTask struct {
	Task    api.Task
	Project string
	Summary string
}

func (s *Store) SearchClosedTasks(ctx context.Context, projectID, query string, limit int) ([]ClosedTask, error) {
	pattern := "%" + strings.NewReplacer(`\`, `\\`, "%", `\%`, "_", `\_`).Replace(query) + "%"
	rows, err := s.db.QueryContext(ctx, `
		SELECT `+taskColumns+`, (SELECT name FROM projects WHERE projects.id = tasks.project_id), summary FROM tasks
		WHERE status != ? AND (? = '' OR project_id = ?)
			AND (title LIKE ? ESCAPE '\' OR merge_title LIKE ? ESCAPE '\' OR todo_item LIKE ? ESCAPE '\' OR summary LIKE ? ESCAPE '\' OR merge_commit LIKE ? ESCAPE '\')
		ORDER BY closed_at DESC LIMIT ?`,
		string(api.TaskStatusActive), projectID, projectID, pattern, pattern, pattern, pattern, pattern, limit)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	found := []ClosedTask{}
	for rows.Next() {
		c := ClosedTask{}
		c.Task, err = scanTask(trailingScanner{rows, []any{&c.Project, &c.Summary}})
		if err != nil {
			return nil, err
		}
		found = append(found, c)
	}
	return found, rows.Err()
}

type trailingScanner struct {
	row   scanner
	extra []any
}

func (t trailingScanner) Scan(dest ...any) error {
	return t.row.Scan(append(dest, t.extra...)...)
}

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) MergeTaskIntoParent(ctx context.Context, id, parentID, commit, actor string) (api.Task, error) {
	return s.closeTask(ctx, id, api.TaskStatusMerged, commit, actor, "task.merged", map[string]any{"id": id, "commit": commit, "parent_task_id": parentID})
}

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 = ?, finalizing = '', finalizing_commit = '', finalizing_notes = '', finalizing_unchanged = 0, merge_after_review_by = '', merge_after_review_id = '', merge_after_review_target = '' 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 IN ('user.message', 'board.posted') 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) FirstPrompt(ctx context.Context, taskID string) (string, error) {
	text := ""
	err := s.db.QueryRowContext(ctx, `
		SELECT coalesce(json_extract(payload, '$.text'), '') FROM events
		WHERE task_id = ? AND kind = 'user.message' ORDER BY seq LIMIT 1
	`, taskID).Scan(&text)
	return text, err
}

func (s *Store) BoardPosts(ctx context.Context, taskID string, after int64) ([]api.Event, error) {
	rows, err := s.db.QueryContext(ctx, `
		SELECT `+eventColumns+`
		FROM events WHERE task_id = ? AND kind = 'board.posted' AND seq > ? ORDER BY seq
	`, taskID, after)
	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 (s *Store) LeadEventsPage(ctx context.Context, owner string, before int64, turns int) (api.EventPage, error) {
	from := int64(0)
	more := true
	err := s.db.QueryRowContext(ctx, `
		SELECT seq FROM events WHERE owner = ? AND agent_session_id IS NOT NULL AND kind = 'turn.ended' AND seq < ?
		ORDER BY seq DESC LIMIT 1 OFFSET ?
	`, owner, before, turns).Scan(&from)
	if errors.Is(err, sql.ErrNoRows) {
		more = false
	} else if err != nil {
		return api.EventPage{}, err
	}

	rows, err := s.db.QueryContext(ctx, `
		SELECT `+eventColumns+`
		FROM events WHERE owner = ? AND agent_session_id IS NOT NULL AND seq > ? AND seq < ? ORDER BY seq
	`, owner, from, before)
	if err != nil {
		return api.EventPage{}, err
	}
	events, err := scanEvents(rows)
	if err != nil {
		return api.EventPage{}, err
	}
	return api.EventPage{Events: events, More: more}, nil
}

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([]api.Event{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, member, 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.Member, 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 := ""
		role := ""
		discard := ""
		finalizing := ""
		err := tx.QueryRowContext(ctx, `
			SELECT coalesce(t.id, ''), coalesce(t.project_id, ''), s.owner, s.state, s.role, coalesce(t.discard_state, ''), coalesce(t.finalizing, '') FROM agent_sessions s LEFT JOIN tasks t ON t.id = s.task_id WHERE s.id = ?
		`, id).Scan(&taskID, &projectID, &owner, &current, &role, &discard, &finalizing)
		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)
		}
		if state == api.AgentSessionStateRunning && role == string(api.Implementer) && discard != "" {
			return fmt.Errorf("%w: task %s is being discarded", ErrStateChanged, taskID)
		}
		if state == api.AgentSessionStateRunning && role == string(api.Implementer) && finalizing != "" {
			return fmt.Errorf("%w: task %s is finishing its merge", ErrStateChanged, taskID)
		}

		_, 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) SessionsWithSubagents(ctx context.Context) ([]api.AgentSession, error) {
	return s.querySessions(ctx, `SELECT `+sessionColumns+` FROM agent_sessions WHERE subagents > 0`)
}

func (s *Store) SetSubagents(ctx context.Context, id string, n int, 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 subagents = ? WHERE id = ?`, n, id)
		if err != nil {
			return err
		}
		return emit(sessionEvent("session.subagents", projectID, taskID, owner, id, actor, map[string]any{"subagents": n}))
	})
}

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
}

type Turn struct {
	Actor string
	At    time.Time
}

func (s *Store) RunningTurn(ctx context.Context, taskID string) (Turn, bool, error) {
	t := Turn{}
	at := ""
	err := s.db.QueryRowContext(ctx, `
		SELECT e.actor, e.created_at FROM events e JOIN agent_sessions s ON s.id = e.agent_session_id
		WHERE e.task_id = ? AND e.kind = 'user.message' AND s.role = ? AND s.state = ?
		ORDER BY e.seq DESC LIMIT 1
	`, taskID, string(api.Implementer), string(api.AgentSessionStateRunning)).Scan(&t.Actor, &at)
	if errors.Is(err, sql.ErrNoRows) {
		return t, false, nil
	}
	if err != nil {
		return t, false, err
	}
	t.At, err = parseTime(at)
	return t, err == nil, 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
	}

	touched := []string{}
	seen := map[string]bool{}
	for _, ev := range events {
		if ev.TaskId == nil || seen[*ev.TaskId] {
			continue
		}
		seen[*ev.TaskId] = true
		touched = append(touched, *ev.TaskId)
	}
	for _, taskID := range touched {
		actor := "byre"
		for _, ev := range events {
			if ev.TaskId != nil && *ev.TaskId == taskID {
				actor = ev.Actor
			}
		}
		err = settleTaskState(ctx, tx, emit, taskID, actor)
		if err != nil {
			return err
		}
	}

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

	s.onEvent(events)
	return nil
}

func settleTaskState(ctx context.Context, tx *sql.Tx, emit emitFunc, taskID, actor string) error {
	projectID := ""
	before := ""
	now := ""
	err := tx.QueryRowContext(ctx, `SELECT project_id, state, `+taskStateExpr+` FROM tasks WHERE id = ?`, taskID).Scan(&projectID, &before, &now)
	if errors.Is(err, sql.ErrNoRows) {
		return nil
	}
	if err != nil {
		return err
	}
	if before == now {
		return nil
	}
	_, err = tx.ExecContext(ctx, `UPDATE tasks SET state = ? WHERE id = ?`, now, taskID)
	if err != nil {
		return err
	}
	return emit(api.Event{
		Kind:      "task.state",
		ProjectId: &projectID,
		TaskId:    &taskID,
		Actor:     actor,
		Payload:   map[string]any{"from": before, "to": now},
	})
}

const taskStateExpr = `CASE
	WHEN tasks.status != 'active' THEN tasks.status
	WHEN tasks.discard_state != '' THEN 'discarding'
	WHEN tasks.merging_by != '' OR tasks.finalizing != '' THEN 'merging'
	ELSE coalesce((
		SELECT CASE
			WHEN s.state != 'idle' THEN s.state
			WHEN s.subagents > 0 THEN 'running'
			WHEN EXISTS (SELECT 1 FROM reviews r WHERE r.task_id = tasks.id AND r.state = 'running') THEN 'reviewing'
			WHEN tasks.review_requests > 0 THEN 'review_pending'
			ELSE 'idle'
		END
		FROM agent_sessions s
		WHERE s.task_id = tasks.id AND s.role != 'reviewer' AND s.state != 'completed'
		ORDER BY s.state = 'running' DESC, s.subagents > 0 DESC, s.created_at DESC LIMIT 1
	), 'no_agent')
END`

func (s *Store) BeginReviewRequest(ctx context.Context, taskID string) error {
	return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
		_, err := tx.ExecContext(ctx, `UPDATE tasks SET review_requests = review_requests + 1 WHERE id = ?`, taskID)
		if err != nil {
			return err
		}
		return settleTaskState(ctx, tx, emit, taskID, "byre")
	})
}

func (s *Store) EndReviewRequest(ctx context.Context, taskID string) error {
	return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
		_, err := tx.ExecContext(ctx, `UPDATE tasks SET review_requests = max(review_requests - 1, 0) WHERE id = ?`, taskID)
		if err != nil {
			return err
		}
		return settleTaskState(ctx, tx, emit, taskID, "byre")
	})
}

func (s *Store) ClearReviewRequests(ctx context.Context) error {
	return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
		rows, err := tx.QueryContext(ctx, `UPDATE tasks SET review_requests = 0 WHERE review_requests > 0 RETURNING id`)
		if err != nil {
			return err
		}
		ids := []string{}
		for rows.Next() {
			id := ""
			err = rows.Scan(&id)
			if err != nil {
				rows.Close()
				return err
			}
			ids = append(ids, id)
		}
		rows.Close()
		err = rows.Err()
		if err != nil {
			return err
		}
		for _, id := range ids {
			err = settleTaskState(ctx, tx, emit, id, "byre")
			if err != nil {
				return err
			}
		}
		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, parent_task_id, parent_merge_by, merge_after_review_by != '', state, merging_by != '' OR finalizing != '', discard_state != '' AND status = 'active', (SELECT coalesce(max(seq), 0) FROM events WHERE events.task_id = tasks.id AND kind = 'turn.ended' AND json_extract(payload, '$.summary') = 1)`

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

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 := ""
	parentTaskID := ""
	parentMergeBy := ""
	taskState := ""
	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, &parentTaskID, &parentMergeBy, &t.MergeAfterReview, &taskState, &t.Merging, &t.Discarding, &t.SummarySeq)
	if err != nil {
		return t, err
	}
	t.State = api.TaskState(taskState)
	if parentTaskID != "" {
		t.ParentTaskId = &parentTaskID
	}
	t.MergingIntoParent = parentMergeBy != ""
	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.Member, &sess.CreatedBy, &createdAt, &forkedFrom, &forkedAt, &sess.Subagents)
	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)
}