summaryrefslogtreecommitdiff
path: root/internal/storage
diff options
context:
space:
mode:
Diffstat (limited to 'internal/storage')
-rw-r--r--internal/storage/db.go60
-rw-r--r--internal/storage/events.go35
-rw-r--r--internal/storage/events_test.go106
3 files changed, 185 insertions, 16 deletions
diff --git a/internal/storage/db.go b/internal/storage/db.go
index 3a4f059..42d7800 100644
--- a/internal/storage/db.go
+++ b/internal/storage/db.go
@@ -7,6 +7,7 @@ import (
"strings"
"time"
+ "github.com/thepeterstone/claudomator/internal/event"
"github.com/thepeterstone/claudomator/internal/task"
)
@@ -261,8 +262,17 @@ func (s *DB) ListSubtasks(parentID string) ([]*task.Task, error) {
return tasks, rows.Err()
}
-// UpdateTaskState atomically updates a task's state, enforcing valid transitions.
+// UpdateTaskState atomically updates a task's state, enforcing valid
+// transitions. The transition is attributed to the system actor; callers
+// acting on behalf of a user should use UpdateTaskStateBy.
func (s *DB) UpdateTaskState(id string, newState task.State) error {
+ return s.UpdateTaskStateBy(id, newState, event.ActorSystem)
+}
+
+// UpdateTaskStateBy is UpdateTaskState with explicit actor attribution. It
+// writes a state_change event in the same transaction as the row update so the
+// event stream and task state never diverge.
+func (s *DB) UpdateTaskStateBy(id string, newState task.State, actor event.Actor) error {
tx, err := s.db.Begin()
if err != nil {
return err
@@ -285,6 +295,14 @@ func (s *DB) UpdateTaskState(id string, newState task.State) error {
if _, err := tx.Exec(`UPDATE tasks SET state = ?, updated_at = ? WHERE id = ?`, string(newState), now, id); err != nil {
return err
}
+ if err := createEventTx(tx, &event.Event{
+ TaskID: id,
+ Kind: event.KindStateChange,
+ Actor: actor,
+ Payload: stateChangePayload(task.State(currentState), newState),
+ }); err != nil {
+ return err
+ }
return tx.Commit()
}
@@ -319,6 +337,15 @@ func (s *DB) ResetTaskForRetry(id string) (*task.Task, error) {
return nil, err
}
+ if err := createEventTx(tx, &event.Event{
+ TaskID: id,
+ Kind: event.KindStateChange,
+ Actor: event.ActorUser,
+ Payload: stateChangePayload(t.State, task.StateQueued),
+ }); err != nil {
+ return nil, err
+ }
+
if err := tx.Commit(); err != nil {
return nil, err
}
@@ -340,20 +367,35 @@ func (s *DB) UpdateTaskAgent(id string, agent task.AgentConfig) error {
// RejectTask sets a task's state to PENDING and stores the rejection comment.
func (s *DB) RejectTask(id, comment string) error {
- now := time.Now().UTC()
- result, err := s.db.Exec(`UPDATE tasks SET state = ?, rejection_comment = ?, updated_at = ? WHERE id = ?`,
- string(task.StatePending), comment, now, id)
+ tx, err := s.db.Begin()
if err != nil {
return err
}
- n, err := result.RowsAffected()
- if err != nil {
+ defer tx.Rollback() //nolint:errcheck
+
+ var currentState string
+ if err := tx.QueryRow(`SELECT state FROM tasks WHERE id = ?`, id).Scan(&currentState); err != nil {
+ if err == sql.ErrNoRows {
+ return fmt.Errorf("task %q not found", id)
+ }
return err
}
- if n == 0 {
- return fmt.Errorf("task %q not found", id)
+
+ now := time.Now().UTC()
+ if _, err := tx.Exec(`UPDATE tasks SET state = ?, rejection_comment = ?, updated_at = ? WHERE id = ?`,
+ string(task.StatePending), comment, now, id); err != nil {
+ return err
}
- return nil
+
+ if err := createEventTx(tx, &event.Event{
+ TaskID: id,
+ Kind: event.KindStateChange,
+ Actor: event.ActorUser,
+ Payload: stateChangePayload(task.State(currentState), task.StatePending),
+ }); err != nil {
+ return err
+ }
+ return tx.Commit()
}
// TaskUpdate holds the fields that UpdateTask may change.
diff --git a/internal/storage/events.go b/internal/storage/events.go
index 82016ab..2051127 100644
--- a/internal/storage/events.go
+++ b/internal/storage/events.go
@@ -1,6 +1,7 @@
package storage
import (
+ "database/sql"
"encoding/json"
"errors"
"fmt"
@@ -8,12 +9,29 @@ import (
"github.com/google/uuid"
"github.com/thepeterstone/claudomator/internal/event"
+ "github.com/thepeterstone/claudomator/internal/task"
)
// CreateEvent appends an event to a task's stream. Seq is assigned by the
// database as MAX(seq)+1 within a transaction. On return, e.ID, e.Seq, and
// e.Timestamp are populated (timestamps default to time.Now().UTC()).
func (s *DB) CreateEvent(e *event.Event) error {
+ tx, err := s.db.Begin()
+ if err != nil {
+ return err
+ }
+ defer tx.Rollback()
+ if err := createEventTx(tx, e); err != nil {
+ return err
+ }
+ return tx.Commit()
+}
+
+// createEventTx validates, defaults, assigns a per-task monotonic seq, and
+// inserts an event using the supplied transaction. The caller commits. This
+// lets state-changing methods write their state_change event atomically with
+// the row update.
+func createEventTx(tx *sql.Tx, e *event.Event) error {
if e == nil {
return errors.New("event is nil")
}
@@ -38,12 +56,6 @@ func (s *DB) CreateEvent(e *event.Event) error {
e.Timestamp = e.Timestamp.UTC()
}
- tx, err := s.db.Begin()
- if err != nil {
- return err
- }
- defer tx.Rollback()
-
var nextSeq int64
if err := tx.QueryRow(`SELECT COALESCE(MAX(seq), 0) + 1 FROM events WHERE task_id = ?`, e.TaskID).Scan(&nextSeq); err != nil {
return fmt.Errorf("computing next seq: %w", err)
@@ -56,7 +68,16 @@ func (s *DB) CreateEvent(e *event.Event) error {
); err != nil {
return fmt.Errorf("inserting event: %w", err)
}
- return tx.Commit()
+ return nil
+}
+
+// stateChangePayload builds the JSON payload for a state_change event.
+func stateChangePayload(from, to task.State) json.RawMessage {
+ b, _ := json.Marshal(struct {
+ From task.State `json:"from"`
+ To task.State `json:"to"`
+ }{From: from, To: to})
+ return b
}
// ListEvents returns events for a task in seq order. Pass sinceSeq=0 for the
diff --git a/internal/storage/events_test.go b/internal/storage/events_test.go
index 3b342c7..1f03aeb 100644
--- a/internal/storage/events_test.go
+++ b/internal/storage/events_test.go
@@ -160,6 +160,112 @@ func TestListEvents_PreservesPayloadAndKind(t *testing.T) {
}
}
+func TestUpdateTaskState_WritesStateChangeEvent(t *testing.T) {
+ db := testDB(t)
+ makeTaskForEvent(t, db, "task-1") // starts PENDING
+
+ if err := db.UpdateTaskState("task-1", task.StateQueued); err != nil {
+ t.Fatalf("UpdateTaskState: %v", err)
+ }
+
+ events, err := db.ListEvents("task-1", 0)
+ if err != nil {
+ t.Fatalf("ListEvents: %v", err)
+ }
+ if len(events) != 1 {
+ t.Fatalf("expected 1 event, got %d", len(events))
+ }
+ e := events[0]
+ if e.Kind != event.KindStateChange {
+ t.Errorf("kind: got %v want state_change", e.Kind)
+ }
+ if e.Actor != event.ActorSystem {
+ t.Errorf("actor: got %v want system", e.Actor)
+ }
+ var payload struct {
+ From task.State `json:"from"`
+ To task.State `json:"to"`
+ }
+ if err := json.Unmarshal(e.Payload, &payload); err != nil {
+ t.Fatalf("unmarshal payload: %v", err)
+ }
+ if payload.From != task.StatePending || payload.To != task.StateQueued {
+ t.Errorf("payload: got %s→%s want PENDING→QUEUED", payload.From, payload.To)
+ }
+}
+
+func TestUpdateTaskStateBy_AttributesActor(t *testing.T) {
+ db := testDB(t)
+ makeTaskForEvent(t, db, "task-1")
+
+ if err := db.UpdateTaskStateBy("task-1", task.StateCancelled, event.ActorUser); err != nil {
+ t.Fatalf("UpdateTaskStateBy: %v", err)
+ }
+
+ events, err := db.ListEvents("task-1", 0)
+ if err != nil {
+ t.Fatalf("ListEvents: %v", err)
+ }
+ if len(events) != 1 {
+ t.Fatalf("expected 1 event, got %d", len(events))
+ }
+ if events[0].Actor != event.ActorUser {
+ t.Errorf("actor: got %v want user", events[0].Actor)
+ }
+}
+
+func TestUpdateTaskState_InvalidTransition_WritesNoEvent(t *testing.T) {
+ db := testDB(t)
+ makeTaskForEvent(t, db, "task-1") // PENDING
+
+ // PENDING → COMPLETED is not a valid transition.
+ if err := db.UpdateTaskState("task-1", task.StateCompleted); err == nil {
+ t.Fatal("expected invalid transition error, got nil")
+ }
+
+ events, err := db.ListEvents("task-1", 0)
+ if err != nil {
+ t.Fatalf("ListEvents: %v", err)
+ }
+ if len(events) != 0 {
+ t.Fatalf("expected no events after rejected transition, got %d", len(events))
+ }
+}
+
+func TestRejectTask_WritesUserStateChangeEvent(t *testing.T) {
+ db := testDB(t)
+ makeTaskForEvent(t, db, "task-1")
+ // Drive PENDING → QUEUED → RUNNING → READY so reject is valid.
+ for _, st := range []task.State{task.StateQueued, task.StateRunning, task.StateReady} {
+ if err := db.UpdateTaskState("task-1", st); err != nil {
+ t.Fatalf("transition to %s: %v", st, err)
+ }
+ }
+
+ if err := db.RejectTask("task-1", "needs work"); err != nil {
+ t.Fatalf("RejectTask: %v", err)
+ }
+
+ events, err := db.ListEvents("task-1", 0)
+ if err != nil {
+ t.Fatalf("ListEvents: %v", err)
+ }
+ last := events[len(events)-1]
+ if last.Kind != event.KindStateChange || last.Actor != event.ActorUser {
+ t.Errorf("reject event: got kind=%v actor=%v want state_change/user", last.Kind, last.Actor)
+ }
+ var payload struct {
+ From task.State `json:"from"`
+ To task.State `json:"to"`
+ }
+ if err := json.Unmarshal(last.Payload, &payload); err != nil {
+ t.Fatalf("unmarshal: %v", err)
+ }
+ if payload.From != task.StateReady || payload.To != task.StatePending {
+ t.Errorf("payload: got %s→%s want READY→PENDING", payload.From, payload.To)
+ }
+}
+
func TestCreateEvent_RequiresFields(t *testing.T) {
db := testDB(t)
generated by cgit v1.2.3 (git 2.39.1) at 2026-07-16 12:56:17 +0000