1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
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
}
|
