From 089e196dbcc8c7ba315ddf045b26d660a1a9c98c Mon Sep 17 00:00:00 2001 From: Hermes Date: Fri, 14 Aug 2026 23:38:57 +0500 Subject: [PATCH] =?UTF-8?q?internal/storage:=20SQLite-=D0=91=D0=94=20?= =?UTF-8?q?=D0=B4=D0=BB=D1=8F=20=D0=B7=D0=B0=D0=B4=D0=B0=D1=87=20=D0=B8=20?= =?UTF-8?q?=D1=82=D1=80=D0=B0=D1=81=D1=81=D0=B8=D1=80=D0=BE=D0=B2=D0=BE?= =?UTF-8?q?=D0=BA=20(=D1=82=D0=B0=D1=81=D0=BA=D0=B8=20+=20=D1=81=D1=82?= =?UTF-8?q?=D0=B0=D1=82=D1=83=D1=81-=D0=BC=D0=B0=D1=88=D0=B8=D0=BD=D0=B0,?= =?UTF-8?q?=20=D1=82=D1=80=D0=B0=D1=81=D1=81=D1=8B=20=D1=81=20=D0=B3=D1=80?= =?UTF-8?q?=D1=83=D0=BF=D0=BF=D0=B8=D1=80=D0=BE=D0=B2=D0=BA=D0=BE=D0=B9=20?= =?UTF-8?q?=D0=BF=D0=BE=20task=5Fid)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/storage/errors.go | 21 +++ internal/storage/models.go | 104 ++++++++++++++ internal/storage/models_test.go | 49 +++++++ internal/storage/storage.go | 158 ++++++++++++++++++++ internal/storage/storage_test.go | 239 +++++++++++++++++++++++++++++++ internal/storage/tasks.go | 138 ++++++++++++++++++ internal/storage/traces.go | 112 +++++++++++++++ 7 files changed, 821 insertions(+) create mode 100644 internal/storage/errors.go create mode 100644 internal/storage/models.go create mode 100644 internal/storage/models_test.go create mode 100644 internal/storage/storage.go create mode 100644 internal/storage/storage_test.go create mode 100644 internal/storage/tasks.go create mode 100644 internal/storage/traces.go diff --git a/internal/storage/errors.go b/internal/storage/errors.go new file mode 100644 index 0000000..3420945 --- /dev/null +++ b/internal/storage/errors.go @@ -0,0 +1,21 @@ +package storage + +import "errors" + +// Классы ошибок Ratatoskr-стиля: S1–S5 +var ( + // S1 — сущность не найдена + ErrNotFound = errors.New("S1: task not found") + + // S2 — дубликат (task_tag, title+chat_id) + ErrDuplicate = errors.New("S2: duplicate entity") + + // S3 — невалидный переход статуса + ErrInvalidStatus = errors.New("S3: invalid status transition") + + // S4 — внутренняя ошибка БД + ErrDB = errors.New("S4: database error") + + // S5 — трасса не найдена + ErrTraceNotFound = errors.New("S5: trace not found") +) \ No newline at end of file diff --git a/internal/storage/models.go b/internal/storage/models.go new file mode 100644 index 0000000..2db18e2 --- /dev/null +++ b/internal/storage/models.go @@ -0,0 +1,104 @@ +package storage + +// Status — статус задачи (state machine). +type Status string + +const ( + StatusDraft Status = "draft" // только что создана + StatusCollecting Status = "collecting" // аналитик собирает детали + StatusReady Status = "ready" // черновик готов, ждёт запуска + StatusRunning Status = "running" // opencode работает + StatusSuccess Status = "success" // задача выполнена + StatusFailed Status = "failed" // ошибка выполнения + StatusTimeout Status = "timeout" // таймаут opencode + StatusCancelled Status = "cancelled" // отменена пользователем + StatusAborted Status = "aborted" // сбой сбора, черновик выброшен + StatusClosed Status = "closed" // закрыта вручную +) + +// AllStatuses — все возможные статусы для валидации. +var AllStatuses = []Status{ + StatusDraft, StatusCollecting, StatusReady, + StatusRunning, StatusSuccess, StatusFailed, StatusTimeout, + StatusCancelled, StatusAborted, StatusClosed, +} + +// validTransitions задаёт разрешённые переходы статусов. +var validTransitions = map[Status][]Status{ + StatusDraft: {StatusCollecting, StatusCancelled, StatusAborted}, + StatusCollecting: {StatusReady, StatusDraft, StatusCancelled, StatusAborted}, + StatusReady: {StatusRunning, StatusCancelled, StatusAborted, StatusClosed}, + StatusRunning: {StatusSuccess, StatusFailed, StatusTimeout, StatusCancelled}, + StatusSuccess: {StatusClosed}, + StatusFailed: {StatusReady, StatusClosed, StatusCancelled}, // retry + StatusTimeout: {StatusReady, StatusClosed, StatusCancelled}, // retry + StatusCancelled: {StatusClosed}, + StatusAborted: {StatusClosed}, + StatusClosed: {}, // терминальный +} + +// IsValidTransition проверяет, допустим ли переход from → to. +func IsValidTransition(from, to Status) bool { + allowed, ok := validTransitions[from] + if !ok { + return false + } + for _, s := range allowed { + if s == to { + return true + } + } + return false +} + +// IsTerminal возвращает true, если статус терминальный. +func IsTerminal(s Status) bool { + return s == StatusSuccess || s == StatusCancelled || + s == StatusAborted || s == StatusClosed +} + +// Task — запись задачи в БД. +type Task struct { + ID int64 `json:"id"` + ChatID string `json:"chat_id"` // tg:// + Title string `json:"title"` + Goal string `json:"goal"` + Repo string `json:"repo"` + Why string `json:"why"` + AC string `json:"ac"` // acceptance criteria + TaskTag string `json:"task_tag"` // UUID, стабильный на всю жизнь + Status Status `json:"status"` + CreatedAt SQLiteTime `json:"created_at"` + UpdatedAt SQLiteTime `json:"updated_at"` +} + +// Trace — запись трассировки выполнения. +type TraceStatus string + +const ( + TraceRunning TraceStatus = "running" + TraceSuccess TraceStatus = "success" + TraceFailed TraceStatus = "failed" + TraceTimeout TraceStatus = "timeout" +) + +// Trace — лог одного субагента. +type Trace struct { + ID int64 `json:"id"` + TaskID int64 `json:"task_id"` + Agent string `json:"agent"` // analyst | researcher | dev | reviewer + SessionID string `json:"session_id"` // opencode session_id + Prompt string `json:"prompt"` + Output string `json:"output"` // полный NDJSON или summary + Status TraceStatus `json:"status"` + StartedAt SQLiteTime `json:"started_at"` + FinishedAt NullSQLiteTime `json:"finished_at,omitempty"` +} + +// TaskFilter — параметры фильтрации списка задач. +type TaskFilter struct { + ChatID string + Status Status + Limit int + Offset int +} \ No newline at end of file diff --git a/internal/storage/models_test.go b/internal/storage/models_test.go new file mode 100644 index 0000000..9b9581b --- /dev/null +++ b/internal/storage/models_test.go @@ -0,0 +1,49 @@ +package storage + +import ( + "testing" +) + +func TestIsValidTransition(t *testing.T) { + tests := []struct { + from, to Status + want bool + }{ + {StatusDraft, StatusCollecting, true}, + {StatusDraft, StatusCancelled, true}, + {StatusDraft, StatusRunning, false}, + {StatusDraft, StatusClosed, false}, + {StatusRunning, StatusSuccess, true}, + {StatusRunning, StatusFailed, true}, + {StatusRunning, StatusTimeout, true}, + {StatusSuccess, StatusClosed, true}, + {StatusSuccess, StatusDraft, false}, + {StatusFailed, StatusReady, true}, // retry + {StatusFailed, StatusClosed, true}, + {StatusFailed, StatusRunning, false}, + {StatusTimeout, StatusReady, true}, // retry + {StatusClosed, StatusDraft, false}, + {StatusClosed, StatusRunning, false}, + } + for _, tc := range tests { + got := IsValidTransition(tc.from, tc.to) + if got != tc.want { + t.Errorf("IsValidTransition(%q → %q) = %v, want %v", tc.from, tc.to, got, tc.want) + } + } +} + +func TestIsTerminal(t *testing.T) { + if !IsTerminal(StatusSuccess) { + t.Error("StatusSuccess should be terminal") + } + if !IsTerminal(StatusCancelled) { + t.Error("StatusCancelled should be terminal") + } + if IsTerminal(StatusDraft) { + t.Error("StatusDraft should NOT be terminal") + } + if IsTerminal(StatusRunning) { + t.Error("StatusRunning should NOT be terminal") + } +} \ No newline at end of file diff --git a/internal/storage/storage.go b/internal/storage/storage.go new file mode 100644 index 0000000..6272c10 --- /dev/null +++ b/internal/storage/storage.go @@ -0,0 +1,158 @@ +package storage + +import ( + "context" + "database/sql" + "database/sql/driver" + "fmt" + "time" + + _ "modernc.org/sqlite" +) + +// SQLiteTime — time.Time с sql.Scanner и driver.Valuer для modernc.org/sqlite. +// Хранит как INTEGER (Unix-секунды). +type SQLiteTime time.Time + +func (st *SQLiteTime) Scan(src any) error { + if src == nil { + *st = SQLiteTime(time.Time{}) + return nil + } + switch v := src.(type) { + case int64: + *st = SQLiteTime(time.Unix(v, 0).UTC()) + return nil + case float64: + *st = SQLiteTime(time.Unix(int64(v), 0).UTC()) + return nil + case string: + // fallback: RFC3339 + t, err := time.Parse(time.RFC3339, v) + if err != nil { + return fmt.Errorf("parse time %q: %w", v, err) + } + *st = SQLiteTime(t) + return nil + default: + return fmt.Errorf("cannot scan %T as SQLiteTime", src) + } +} + +func (st SQLiteTime) Value() (driver.Value, error) { + return time.Time(st).Unix(), nil +} + +func (st SQLiteTime) Time() time.Time { + return time.Time(st) +} + +// NullSQLiteTime — nullable версия SQLiteTime. +type NullSQLiteTime struct { + Time SQLiteTime + Valid bool +} + +func (n *NullSQLiteTime) Scan(src any) error { + if src == nil { + n.Valid = false + return nil + } + n.Valid = true + return n.Time.Scan(src) +} + +func (n NullSQLiteTime) Value() (driver.Value, error) { + if !n.Valid { + return nil, nil + } + return n.Time.Value() +} + +// Storage — компонент БД для задач и трассировок. +type Storage struct { + db *sql.DB +} + +// Open открывает SQLite-БД и выполняет миграции. +// path — путь к файлу БД; ":memory:" для тестов. +func Open(ctx context.Context, path string) (*Storage, error) { + db, err := sql.Open("sqlite", path) + if err != nil { + return nil, fmt.Errorf("%w: open: %w", ErrDB, err) + } + // Прагмы: WAL + синхронность + pragmas := []string{ + "PRAGMA journal_mode=WAL", + "PRAGMA synchronous=NORMAL", + "PRAGMA busy_timeout=5000", + "PRAGMA foreign_keys=ON", + } + for _, p := range pragmas { + if _, err := db.ExecContext(ctx, p); err != nil { + db.Close() + return nil, fmt.Errorf("%w: pragma %q: %w", ErrDB, p, err) + } + } + s := &Storage{db: db} + if err := s.migrate(ctx); err != nil { + db.Close() + return nil, err + } + return s, nil +} + +// Close закрывает БД. +func (s *Storage) Close() error { + return s.db.Close() +} + +// DB возвращает сырой *sql.DB для использования в транзакциях. +func (s *Storage) DB() *sql.DB { + return s.db +} + +// migrate создаёт таблицы при первом запуске. +func (s *Storage) migrate(ctx context.Context) error { + schema := ` + CREATE TABLE IF NOT EXISTS tasks ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + chat_id TEXT NOT NULL DEFAULT '', + title TEXT NOT NULL DEFAULT '', + goal TEXT NOT NULL DEFAULT '', + repo TEXT NOT NULL DEFAULT '', + why TEXT NOT NULL DEFAULT '', + ac TEXT NOT NULL DEFAULT '', + task_tag TEXT NOT NULL DEFAULT '', + status TEXT NOT NULL DEFAULT 'draft', + created_at INTEGER NOT NULL DEFAULT (unixepoch()), + updated_at INTEGER NOT NULL DEFAULT (unixepoch()) + ); + + CREATE INDEX IF NOT EXISTS idx_tasks_chat ON tasks(chat_id); + CREATE INDEX IF NOT EXISTS idx_tasks_status ON tasks(status); + + CREATE TABLE IF NOT EXISTS traces ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + task_id INTEGER NOT NULL REFERENCES tasks(id) ON DELETE CASCADE, + agent TEXT NOT NULL DEFAULT '', + session_id TEXT NOT NULL DEFAULT '', + prompt TEXT NOT NULL DEFAULT '', + output TEXT NOT NULL DEFAULT '', + status TEXT NOT NULL DEFAULT 'running', + started_at INTEGER NOT NULL DEFAULT (unixepoch()), + finished_at INTEGER + ); + + CREATE INDEX IF NOT EXISTS idx_traces_task ON traces(task_id); + ` + if _, err := s.db.ExecContext(ctx, schema); err != nil { + return fmt.Errorf("%w: migrate: %w", ErrDB, err) + } + return nil +} + +// Now возвращает текущее время UTC как SQLiteTime. +func Now() SQLiteTime { + return SQLiteTime(time.Now().UTC()) +} \ No newline at end of file diff --git a/internal/storage/storage_test.go b/internal/storage/storage_test.go new file mode 100644 index 0000000..64a391b --- /dev/null +++ b/internal/storage/storage_test.go @@ -0,0 +1,239 @@ +package storage + +import ( + "context" + "errors" + "testing" +) + +func setupTestDB(t *testing.T) (*Storage, context.Context) { + t.Helper() + ctx := context.Background() + s, err := Open(ctx, ":memory:") + if err != nil { + t.Fatalf("Open(:memory:): %v", err) + } + t.Cleanup(func() { s.Close() }) + return s, ctx +} + +func TestCreateAndGetTask(t *testing.T) { + s, ctx := setupTestDB(t) + task := &Task{ + ChatID: "tg://123", + Title: "Test task", + Goal: "Test goal", + Repo: "test/repo", + TaskTag: "abc-123", + } + id, err := s.CreateTask(ctx, task) + if err != nil { + t.Fatalf("CreateTask: %v", err) + } + if id == 0 { + t.Fatal("expected non-zero ID") + } + if task.Status != StatusDraft { + t.Fatalf("expected draft status, got %s", task.Status) + } + + got, err := s.GetTask(ctx, id) + if err != nil { + t.Fatalf("GetTask: %v", err) + } + if got.Title != "Test task" { + t.Fatalf("title = %q, want %q", got.Title, "Test task") + } + if got.Status != StatusDraft { + t.Fatalf("status = %s, want draft", got.Status) + } +} + +func TestGetTaskNotFound(t *testing.T) { + s, ctx := setupTestDB(t) + _, err := s.GetTask(ctx, 999) + if !IsNotFound(err) { + t.Fatalf("expected ErrNotFound, got %v", err) + } +} + +func TestUpdateTaskStatus(t *testing.T) { + s, ctx := setupTestDB(t) + task := &Task{ + ChatID: "tg://123", + Title: "Update test", + TaskTag: "def-456", + } + _, _ = s.CreateTask(ctx, task) + + // draft → collecting — валидно + task.Status = StatusCollecting + if err := s.UpdateTask(ctx, task); err != nil { + t.Fatalf("UpdateTask draft→collecting: %v", err) + } + got, _ := s.GetTask(ctx, task.ID) + if got.Status != StatusCollecting { + t.Fatalf("status = %s, want collecting", got.Status) + } + + // collecting → ready + task.Status = StatusReady + if err := s.UpdateTask(ctx, task); err != nil { + t.Fatalf("UpdateTask collecting→ready: %v", err) + } + + // ready → running + task.Status = StatusRunning + if err := s.UpdateTask(ctx, task); err != nil { + t.Fatalf("UpdateTask ready→running: %v", err) + } + + // running → success + task.Status = StatusSuccess + if err := s.UpdateTask(ctx, task); err != nil { + t.Fatalf("UpdateTask running→success: %v", err) + } +} + +func TestUpdateTaskInvalidTransition(t *testing.T) { + s, ctx := setupTestDB(t) + task := &Task{ChatID: "tg://1", TaskTag: "invalid"} + _, _ = s.CreateTask(ctx, task) + + task.Status = StatusRunning // draft → running невалидно + err := s.UpdateTask(ctx, task) + if !IsInvalidStatus(err) { + t.Fatalf("expected ErrInvalidStatus, got %v", err) + } +} + +func TestUpdateTaskNotFound(t *testing.T) { + s, ctx := setupTestDB(t) + err := s.UpdateTask(ctx, &Task{ID: 999, Status: StatusReady}) + if !IsNotFound(err) { + t.Fatalf("expected ErrNotFound, got %v", err) + } +} + +func TestListTasks(t *testing.T) { + s, ctx := setupTestDB(t) + for i := 0; i < 5; i++ { + _, _ = s.CreateTask(ctx, &Task{ + ChatID: "tg://user", + Title: "Task", + TaskTag: "tag", + }) + } + tasks, err := s.ListTasks(ctx, TaskFilter{ChatID: "tg://user", Limit: 10}) + if err != nil { + t.Fatalf("ListTasks: %v", err) + } + if len(tasks) != 5 { + t.Fatalf("expected 5 tasks, got %d", len(tasks)) + } +} + +func TestDeleteTask(t *testing.T) { + s, ctx := setupTestDB(t) + task := &Task{ChatID: "tg://del", TaskTag: "delete-me"} + id, _ := s.CreateTask(ctx, task) + + if err := s.DeleteTask(ctx, id); err != nil { + t.Fatalf("DeleteTask: %v", err) + } + _, err := s.GetTask(ctx, id) + if !IsNotFound(err) { + t.Fatal("expected ErrNotFound after delete") + } +} + +func TestAppendAndGetTraces(t *testing.T) { + s, ctx := setupTestDB(t) + task := &Task{ChatID: "tg://trace", TaskTag: "trace-test"} + id, _ := s.CreateTask(ctx, task) + + tr := &Trace{ + TaskID: id, + Agent: "analyst", + SessionID: "sess-001", + Prompt: "analyze this", + Output: "", + } + traceID, err := s.AppendTrace(ctx, tr) + if err != nil { + t.Fatalf("AppendTrace: %v", err) + } + if traceID == 0 { + t.Fatal("expected non-zero trace ID") + } + if tr.Status != TraceRunning { + t.Fatalf("trace status = %s, want running", tr.Status) + } + + // апдейт статуса + if err := s.UpdateTraceStatus(ctx, traceID, TraceSuccess); err != nil { + t.Fatalf("UpdateTraceStatus: %v", err) + } + + // добавим вторую трассу + _, _ = s.AppendTrace(ctx, &Trace{ + TaskID: id, + Agent: "dev", + SessionID: "sess-002", + }) + + traces, err := s.GetTraces(ctx, id) + if err != nil { + t.Fatalf("GetTraces: %v", err) + } + if len(traces) != 2 { + t.Fatalf("expected 2 traces, got %d", len(traces)) + } + if traces[0].Status != TraceSuccess { + t.Fatalf("first trace status = %s, want success", traces[0].Status) + } + if traces[1].Agent != "dev" { + t.Fatalf("second trace agent = %s, want dev", traces[1].Agent) + } +} + +func TestGetLatestTrace(t *testing.T) { + s, ctx := setupTestDB(t) + task := &Task{ChatID: "tg://latest", TaskTag: "latest-trace"} + id, _ := s.CreateTask(ctx, task) + + _, _ = s.AppendTrace(ctx, &Trace{TaskID: id, Agent: "analyst", SessionID: "s1"}) + _, _ = s.AppendTrace(ctx, &Trace{TaskID: id, Agent: "analyst", SessionID: "s2"}) + + latest, err := s.GetLatestTrace(ctx, id, "analyst") + if err != nil { + t.Fatalf("GetLatestTrace: %v", err) + } + if latest.SessionID != "s2" { + t.Fatalf("latest session = %s, want s2", latest.SessionID) + } +} + +func TestUpdateTraceOutput(t *testing.T) { + s, ctx := setupTestDB(t) + task := &Task{ChatID: "tg://out", TaskTag: "output"} + id, _ := s.CreateTask(ctx, task) + traceID, _ := s.AppendTrace(ctx, &Trace{TaskID: id, Agent: "dev"}) + + if err := s.UpdateTraceOutput(ctx, traceID, "hello world"); err != nil { + t.Fatalf("UpdateTraceOutput: %v", err) + } + traces, _ := s.GetTraces(ctx, id) + if traces[0].Output != "hello world" { + t.Fatalf("output = %q, want %q", traces[0].Output, "hello world") + } +} + +// хелперы для проверки классов ошибок +func IsNotFound(err error) bool { + return errors.Is(err, ErrNotFound) +} + +func IsInvalidStatus(err error) bool { + return errors.Is(err, ErrInvalidStatus) +} \ No newline at end of file diff --git a/internal/storage/tasks.go b/internal/storage/tasks.go new file mode 100644 index 0000000..db48fb5 --- /dev/null +++ b/internal/storage/tasks.go @@ -0,0 +1,138 @@ +package storage + +import ( + "context" + "database/sql" + "fmt" +) + +// CreateTask создаёт задачу со статусом draft. Возвращает её ID. +// chat_id и task_tag передаются извне (в диалоге — chat.Address + uuid). +func (s *Storage) CreateTask(ctx context.Context, t *Task) (int64, error) { + now := Now() + res, err := s.db.ExecContext(ctx, ` + INSERT INTO tasks (chat_id, title, goal, repo, why, ac, task_tag, status, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + t.ChatID, t.Title, t.Goal, t.Repo, t.Why, t.AC, + t.TaskTag, StatusDraft, now, now, + ) + if err != nil { + return 0, fmt.Errorf("%w: create task: %w", ErrDB, err) + } + id, err := res.LastInsertId() + if err != nil { + return 0, fmt.Errorf("%w: last insert id: %w", ErrDB, err) + } + t.ID = id + t.Status = StatusDraft + t.CreatedAt = now + t.UpdatedAt = now + return id, nil +} + +// GetTask возвращает задачу по ID. +func (s *Storage) GetTask(ctx context.Context, id int64) (*Task, error) { + t := &Task{} + err := s.db.QueryRowContext(ctx, ` + SELECT id, chat_id, title, goal, repo, why, ac, task_tag, status, created_at, updated_at + FROM tasks WHERE id = ?`, id).Scan( + &t.ID, &t.ChatID, &t.Title, &t.Goal, &t.Repo, + &t.Why, &t.AC, &t.TaskTag, &t.Status, &t.CreatedAt, &t.UpdatedAt, + ) + if err == sql.ErrNoRows { + return nil, fmt.Errorf("%w: task %d", ErrNotFound, id) + } + if err != nil { + return nil, fmt.Errorf("%w: get task %d: %w", ErrDB, id, err) + } + return t, nil +} + +// UpdateTask обновляет поля задачи. Если newStatus указан и отличен от текущего — +// проверяет валидность перехода. Возвращает ошибку ErrInvalidStatus при недопустимом переходе. +func (s *Storage) UpdateTask(ctx context.Context, t *Task) error { + // получаем текущий статус для валидации перехода + var oldStatus Status + err := s.db.QueryRowContext(ctx, `SELECT status FROM tasks WHERE id = ?`, t.ID).Scan(&oldStatus) + if err == sql.ErrNoRows { + return fmt.Errorf("%w: task %d", ErrNotFound, t.ID) + } + if err != nil { + return fmt.Errorf("%w: get status %d: %w", ErrDB, t.ID, err) + } + + if t.Status != "" && t.Status != oldStatus { + if !IsValidTransition(oldStatus, t.Status) { + return fmt.Errorf("%w: %s → %s for task %d", ErrInvalidStatus, oldStatus, t.Status, t.ID) + } + } else if t.Status == "" { + t.Status = oldStatus + } + + now := Now() + res, err := s.db.ExecContext(ctx, ` + UPDATE tasks + SET title=?, goal=?, repo=?, why=?, ac=?, status=?, updated_at=? + WHERE id=?`, + t.Title, t.Goal, t.Repo, t.Why, t.AC, t.Status, now, t.ID, + ) + if err != nil { + return fmt.Errorf("%w: update task %d: %w", ErrDB, t.ID, err) + } + n, _ := res.RowsAffected() + if n == 0 { + return fmt.Errorf("%w: task %d", ErrNotFound, t.ID) + } + t.UpdatedAt = now + return nil +} + +// ListTasks возвращает список задач с фильтрацией. +func (s *Storage) ListTasks(ctx context.Context, filter TaskFilter) ([]*Task, error) { + if filter.Limit <= 0 { + filter.Limit = 50 + } + where := "1=1" + args := []any{} + if filter.ChatID != "" { + where += " AND chat_id = ?" + args = append(args, filter.ChatID) + } + if filter.Status != "" { + where += " AND status = ?" + args = append(args, string(filter.Status)) + } + args = append(args, filter.Limit, filter.Offset) + + rows, err := s.db.QueryContext(ctx, ` + SELECT id, chat_id, title, goal, repo, why, ac, task_tag, status, created_at, updated_at + FROM tasks WHERE `+where+` ORDER BY updated_at DESC LIMIT ? OFFSET ?`, args...) + if err != nil { + return nil, fmt.Errorf("%w: list tasks: %w", ErrDB, err) + } + defer rows.Close() + + var tasks []*Task + for rows.Next() { + t := &Task{} + if err := rows.Scan(&t.ID, &t.ChatID, &t.Title, &t.Goal, &t.Repo, + &t.Why, &t.AC, &t.TaskTag, &t.Status, &t.CreatedAt, &t.UpdatedAt); err != nil { + return nil, fmt.Errorf("%w: scan task: %w", ErrDB, err) + } + tasks = append(tasks, t) + } + return tasks, rows.Err() +} + +// DeleteTask удаляет задачу (каскадно — трассы). Только для тестов/админки. +func (s *Storage) DeleteTask(ctx context.Context, id int64) error { + res, err := s.db.ExecContext(ctx, `DELETE FROM tasks WHERE id = ?`, id) + if err != nil { + return fmt.Errorf("%w: delete task %d: %w", ErrDB, id, err) + } + n, _ := res.RowsAffected() + if n == 0 { + return fmt.Errorf("%w: task %d", ErrNotFound, id) + } + return nil +} \ No newline at end of file diff --git a/internal/storage/traces.go b/internal/storage/traces.go new file mode 100644 index 0000000..7dd2abd --- /dev/null +++ b/internal/storage/traces.go @@ -0,0 +1,112 @@ +package storage + +import ( + "context" + "database/sql" + "fmt" +) + +// AppendTrace добавляет трассу выполнения к задаче. Возвращает её ID. +func (s *Storage) AppendTrace(ctx context.Context, tr *Trace) (int64, error) { + now := Now() + res, err := s.db.ExecContext(ctx, ` + INSERT INTO traces (task_id, agent, session_id, prompt, output, status, started_at) + VALUES (?, ?, ?, ?, ?, ?, ?)`, + tr.TaskID, tr.Agent, tr.SessionID, tr.Prompt, tr.Output, TraceRunning, now, + ) + if err != nil { + return 0, fmt.Errorf("%w: append trace: %w", ErrDB, err) + } + id, err := res.LastInsertId() + if err != nil { + return 0, fmt.Errorf("%w: last insert id: %w", ErrDB, err) + } + tr.ID = id + tr.Status = TraceRunning + tr.StartedAt = now + return id, nil +} + +// UpdateTraceStatus обновляет статус и finished_at трассы. +func (s *Storage) UpdateTraceStatus(ctx context.Context, traceID int64, status TraceStatus) error { + now := Now() + res, err := s.db.ExecContext(ctx, ` + UPDATE traces SET status=?, finished_at=? WHERE id=?`, + status, now, traceID, + ) + if err != nil { + return fmt.Errorf("%w: update trace %d: %w", ErrDB, traceID, err) + } + n, _ := res.RowsAffected() + if n == 0 { + return fmt.Errorf("%w: trace %d", ErrTraceNotFound, traceID) + } + return nil +} + +// UpdateTraceOutput обновляет output трассы (например, по мере поступления данных). +func (s *Storage) UpdateTraceOutput(ctx context.Context, traceID int64, output string) error { + _, err := s.db.ExecContext(ctx, `UPDATE traces SET output=? WHERE id=?`, output, traceID) + if err != nil { + return fmt.Errorf("%w: update trace output %d: %w", ErrDB, traceID, err) + } + return nil +} + +// GetTraces возвращает трассы для задачи, отсортированные по started_at. +func (s *Storage) GetTraces(ctx context.Context, taskID int64) ([]*Trace, error) { + rows, err := s.db.QueryContext(ctx, ` + SELECT id, task_id, agent, session_id, prompt, output, status, started_at, finished_at + FROM traces WHERE task_id=? ORDER BY started_at ASC`, taskID) + if err != nil { + return nil, fmt.Errorf("%w: get traces for task %d: %w", ErrDB, taskID, err) + } + defer rows.Close() + + var traces []*Trace + for rows.Next() { + tr := &Trace{} + var finishedAt NullSQLiteTime + if err := rows.Scan(&tr.ID, &tr.TaskID, &tr.Agent, &tr.SessionID, + &tr.Prompt, &tr.Output, &tr.Status, &tr.StartedAt, &finishedAt); err != nil { + return nil, fmt.Errorf("%w: scan trace: %w", ErrDB, err) + } + tr.FinishedAt = finishedAt + traces = append(traces, tr) + } + return traces, rows.Err() +} + +// GetLatestTrace возвращает последнюю трассу задачи для заданного агента. +func (s *Storage) GetLatestTrace(ctx context.Context, taskID int64, agent string) (*Trace, error) { + tr := &Trace{} + var finishedAt NullSQLiteTime + err := s.db.QueryRowContext(ctx, ` + SELECT id, task_id, agent, session_id, prompt, output, status, started_at, finished_at + FROM traces WHERE task_id=? AND agent=? ORDER BY id DESC LIMIT 1`, + taskID, agent).Scan( + &tr.ID, &tr.TaskID, &tr.Agent, &tr.SessionID, + &tr.Prompt, &tr.Output, &tr.Status, &tr.StartedAt, &finishedAt, + ) + if err == sql.ErrNoRows { + return nil, fmt.Errorf("%w: trace for task %d agent %s", ErrTraceNotFound, taskID, agent) + } + if err != nil { + return nil, fmt.Errorf("%w: get latest trace: %w", ErrDB, err) + } + tr.FinishedAt = finishedAt + return tr, nil +} + +// DeleteTrace удаляет трассу. Только для тестов/админки. +func (s *Storage) DeleteTrace(ctx context.Context, id int64) error { + res, err := s.db.ExecContext(ctx, `DELETE FROM traces WHERE id = ?`, id) + if err != nil { + return fmt.Errorf("%w: delete trace %d: %w", ErrDB, id, err) + } + n, _ := res.RowsAffected() + if n == 0 { + return fmt.Errorf("%w: trace %d", ErrTraceNotFound, id) + } + return nil +} \ No newline at end of file