summaryrefslogtreecommitdiff
path: root/internal/storage/events.go
diff options
context:
space:
mode:
authorClaude <noreply@anthropic.com>2026-05-23 08:07:50 +0000
committerClaude <noreply@anthropic.com>2026-05-23 08:07:50 +0000
commitdc97d7ae00bd25668ba495fba249ceade0c6b100 (patch)
tree0c4f48b4cea2b20ab3f814e7655c4ce5c1fc044c /internal/storage/events.go
parent39cc108b5f0bf3d0238a1707d657a77b7a3df184 (diff)
feat(storage): emit state_change events on task transitions
UpdateTaskState now writes a state_change event in the same transaction as the row update, so the event stream and task state never diverge. Adds UpdateTaskStateBy(id, newState, actor) for explicit actor attribution; UpdateTaskState delegates with ActorSystem (correct for the executor, which drives the state machine). RejectTask and ResetTaskForRetry also emit state_change events attributed to the user. API accept and cancel handlers now attribute their transitions to the user via UpdateTaskStateBy. The executor's Store interface is unchanged, so no test doubles needed updating. https://claude.ai/code/session_01SESwn7kQ7oP62trWw6pc39
Diffstat (limited to 'internal/storage/events.go')
-rw-r--r--internal/storage/events.go35
1 files changed, 28 insertions, 7 deletions
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