summaryrefslogtreecommitdiff
path: root/internal/storage/events.go
diff options
context:
space:
mode:
authorClaude <noreply@anthropic.com>2026-05-22 09:08:52 +0000
committerClaude <noreply@anthropic.com>2026-05-22 09:08:52 +0000
commit39cc108b5f0bf3d0238a1707d657a77b7a3df184 (patch)
tree84ee8bcc039b058a308a542f1e1f4a4d11f9e144 /internal/storage/events.go
parentc24fdaa9b6bce49f5c0a1c589fd064d46b62b88d (diff)
feat(event,storage): add events table substrate for OSS redesign
New internal/event package defines the Event type with canonical Kind and Actor constants. New events table is appended via additive migration (id, task_id, seq, ts, kind, actor, payload_json) with seq as per-task monotonic, assigned by the storage layer inside a transaction. CreateEvent + ListEvents support the future migration of question_json, interactions_json, summary, and elaboration_input off the tasks table into a unified, replayable event stream. No existing code paths are modified yet; this is pure substrate for Phase 1. https://claude.ai/code/session_01SESwn7kQ7oP62trWw6pc39
Diffstat (limited to 'internal/storage/events.go')
-rw-r--r--internal/storage/events.go99
1 files changed, 99 insertions, 0 deletions
diff --git a/internal/storage/events.go b/internal/storage/events.go
new file mode 100644
index 0000000..82016ab
--- /dev/null
+++ b/internal/storage/events.go
@@ -0,0 +1,99 @@
+package storage
+
+import (
+ "encoding/json"
+ "errors"
+ "fmt"
+ "time"
+
+ "github.com/google/uuid"
+ "github.com/thepeterstone/claudomator/internal/event"
+)
+
+// 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 {
+ if e == nil {
+ return errors.New("event is nil")
+ }
+ if e.TaskID == "" {
+ return errors.New("event.TaskID required")
+ }
+ if e.Kind == "" {
+ return errors.New("event.Kind required")
+ }
+ if e.Actor == "" {
+ return errors.New("event.Actor required")
+ }
+ if len(e.Payload) == 0 {
+ e.Payload = json.RawMessage("{}")
+ }
+ if e.ID == "" {
+ e.ID = uuid.NewString()
+ }
+ if e.Timestamp.IsZero() {
+ e.Timestamp = time.Now().UTC()
+ } else {
+ 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)
+ }
+ e.Seq = nextSeq
+
+ if _, err := tx.Exec(
+ `INSERT INTO events (id, task_id, seq, ts, kind, actor, payload_json) VALUES (?, ?, ?, ?, ?, ?, ?)`,
+ e.ID, e.TaskID, e.Seq, e.Timestamp, string(e.Kind), string(e.Actor), string(e.Payload),
+ ); err != nil {
+ return fmt.Errorf("inserting event: %w", err)
+ }
+ return tx.Commit()
+}
+
+// ListEvents returns events for a task in seq order. Pass sinceSeq=0 for the
+// full history, or a higher value to fetch only events strictly after that seq
+// (for incremental polling / SSE catch-up).
+func (s *DB) ListEvents(taskID string, sinceSeq int64) ([]*event.Event, error) {
+ rows, err := s.db.Query(
+ `SELECT id, task_id, seq, ts, kind, actor, payload_json FROM events WHERE task_id = ? AND seq > ? ORDER BY seq ASC`,
+ taskID, sinceSeq,
+ )
+ if err != nil {
+ return nil, err
+ }
+ defer rows.Close()
+ var out []*event.Event
+ for rows.Next() {
+ e, err := scanEvent(rows)
+ if err != nil {
+ return nil, err
+ }
+ out = append(out, e)
+ }
+ return out, rows.Err()
+}
+
+func scanEvent(row scanner) (*event.Event, error) {
+ var (
+ e event.Event
+ kind string
+ actor string
+ payload string
+ )
+ if err := row.Scan(&e.ID, &e.TaskID, &e.Seq, &e.Timestamp, &kind, &actor, &payload); err != nil {
+ return nil, err
+ }
+ e.Kind = event.Kind(kind)
+ e.Actor = event.Actor(actor)
+ e.Payload = json.RawMessage(payload)
+ return &e, nil
+}