1 files changed, 120 insertions, 0 deletions
diff --git a/internal/storage/events.go b/internal/storage/events.go
new file mode 100644
index 0000000..2051127
--- /dev/null
+++ b/internal/storage/events.go
@@ -0,0 +1,120 @@
+package storage
+
+import (
+	"database/sql"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"time"
+
+	"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")
+	}
+	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()
+	}
+
+	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 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
+// 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
+}