package store import ( "context" "crypto/rand" "database/sql" "encoding/hex" "encoding/json" "errors" "fmt" "sync" "time" _ "modernc.org/sqlite" "g.fc.run/misc/byre/internal/api" ) var ErrNotFound = errors.New("not found") const schema = ` CREATE TABLE IF NOT EXISTS users ( login TEXT PRIMARY KEY, display_name TEXT NOT NULL, git_name TEXT NOT NULL DEFAULT '', git_email TEXT NOT NULL DEFAULT '', created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS projects ( id TEXT PRIMARY KEY, name TEXT NOT NULL UNIQUE, repo_path TEXT NOT NULL UNIQUE, default_branch TEXT NOT NULL, remote TEXT NOT NULL, created_by TEXT NOT NULL REFERENCES users(login), created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS tasks ( id TEXT PRIMARY KEY, project_id TEXT NOT NULL REFERENCES projects(id), title TEXT NOT NULL, branch TEXT NOT NULL, worktree_path TEXT NOT NULL, base_ref TEXT NOT NULL, status TEXT NOT NULL, created_by TEXT NOT NULL REFERENCES users(login), created_at TEXT NOT NULL, completed_at TEXT ); CREATE INDEX IF NOT EXISTS tasks_project ON tasks(project_id, created_at); CREATE TABLE IF NOT EXISTS events ( seq INTEGER PRIMARY KEY AUTOINCREMENT, kind TEXT NOT NULL, project_id TEXT, task_id TEXT, actor TEXT NOT NULL, payload TEXT NOT NULL, created_at TEXT NOT NULL ); ` type Store struct { db *sql.DB path string writeMu sync.Mutex onEvent func(api.Event) } func Open(path string, 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 = db.Exec(schema) if err != nil { db.Close() return nil, fmt.Errorf("apply schema: %w", err) } return &Store{db: db, path: path, onEvent: onEvent}, nil } 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{} err := s.db.QueryRowContext(ctx, `SELECT login, display_name, git_name, git_email FROM users WHERE login = ?`, login). Scan(&u.Login, &u.DisplayName, &u.GitName, &u.GitEmail) if errors.Is(err, sql.ErrNoRows) { return u, ErrNotFound } 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) error { return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error { _, err := tx.ExecContext(ctx, ` INSERT INTO tasks (id, project_id, title, branch, worktree_path, base_ref, status, created_by, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) `, t.Id, t.ProjectId, t.Title, t.Branch, t.WorktreePath, t.BaseRef, string(t.Status), t.CreatedBy, formatTime(t.CreatedAt)) if err != nil { return err } return emit(api.Event{Kind: "task.created", ProjectId: &t.ProjectId, TaskId: &t.Id, Actor: t.CreatedBy, Payload: toPayload(t)}) }) } func (s *Store) ListTasks(ctx context.Context, projectID string) ([]api.Task, error) { rows, err := s.db.QueryContext(ctx, `SELECT `+taskColumns+` FROM tasks WHERE project_id = ? ORDER BY created_at DESC`, projectID) 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) 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) CompleteTask(ctx context.Context, id, actor string) (api.Task, error) { err := s.write(ctx, func(tx *sql.Tx, emit emitFunc) error { res, err := tx.ExecContext(ctx, `UPDATE tasks SET status = ?, completed_at = ? WHERE id = ? AND status = ?`, string(api.Completed), now(), id, string(api.Active)) 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 } return emit(api.Event{Kind: "task.completed", ProjectId: &projectID, TaskId: &id, Actor: actor, Payload: map[string]any{"id": id}}) }) if err != nil { return api.Task{}, err } return s.GetTask(ctx, id) } func (s *Store) EventsAfter(ctx context.Context, after int64, limit int) ([]api.Event, error) { rows, err := s.db.QueryContext(ctx, ` SELECT seq, kind, project_id, task_id, actor, payload, created_at FROM events WHERE seq > ? ORDER BY seq LIMIT ? `, after, limit) if err != nil { return nil, err } 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.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() } 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.Active)).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, actor, payload, created_at) VALUES (?, ?, ?, ?, ?, ?) RETURNING seq `, ev.Kind, ev.ProjectId, ev.TaskId, ev.Actor, string(payload), formatTime(ev.CreatedAt)).Scan(&ev.Seq) if err != nil { return err } events = append(events, ev) return nil } err = fn(tx, emit) if err != nil { return err } err = tx.Commit() if err != nil { return err } for _, ev := range events { s.onEvent(ev) } return nil } const projectColumns = `id, name, repo_path, default_branch, remote, created_by, created_at` const taskColumns = `id, project_id, title, branch, worktree_path, base_ref, status, created_by, created_at, completed_at` type scanner interface { Scan(dest ...any) error } func scanProject(row scanner) (api.Project, error) { p := api.Project{} createdAt := "" err := row.Scan(&p.Id, &p.Name, &p.RepoPath, &p.DefaultBranch, &p.Remote, &p.CreatedBy, &createdAt) if err != nil { return p, err } p.CreatedAt, err = parseTime(createdAt) return p, err } func scanTask(row scanner) (api.Task, error) { t := api.Task{} status := "" createdAt := "" completedAt := sql.NullString{} err := row.Scan(&t.Id, &t.ProjectId, &t.Title, &t.Branch, &t.WorktreePath, &t.BaseRef, &status, &t.CreatedBy, &createdAt, &completedAt) if err != nil { return t, err } t.Status = api.TaskStatus(status) t.CreatedAt, err = parseTime(createdAt) if err != nil { return t, err } if completedAt.Valid { ct, err := parseTime(completedAt.String) if err != nil { return t, err } t.CompletedAt = &ct } return t, nil } 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()) } func formatTime(t time.Time) string { return t.UTC().Format(time.RFC3339Nano) } func parseTime(s string) (time.Time, error) { return time.Parse(time.RFC3339Nano, s) }