1591 lines
48 KiB
Go
1591 lines
48 KiB
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) 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, ¤t, &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)
|
|
}
|