eventlog.go
1// Package eventlog is a Tangled-compatible event stream: an SQLite log of
2// {rkey, nsid, event, created} served on a websocket as upstream's
3// eventstream does (knot and spindle /events). created is a monotonic local
4// clock (UnixNano) and is the cursor: GET /events?cursor=N replays events
5// with created > N as JSON text messages, then tails.
6//
7// The bridge serves sh.tangled.git.refUpdate on it; the runner serves
8// sh.tangled.pipeline, sh.tangled.pipeline.status and run.rtw.ci.* records.
9package eventlog
10
11import (
12 "context"
13 "crypto/subtle"
14 "database/sql"
15 "encoding/json"
16 "fmt"
17 "net/http"
18 "strconv"
19 "sync"
20 "time"
21
22 "github.com/gorilla/websocket"
23 _ "modernc.org/sqlite"
24
25 "git.kilimanjaro.io/ci/internal/lex"
26 "git.kilimanjaro.io/ci/internal/tid"
27)
28
29const (
30 batchSize = 500
31 keepAliveInterval = 30 * time.Second
32 writeDeadline = 10 * time.Second
33)
34
35// Log is an append-only event log.
36type Log struct {
37 db *sql.DB
38
39 mu sync.Mutex
40 last int64
41 subs map[chan struct{}]struct{}
42}
43
44// Open opens (or creates) the log at path.
45func Open(path string) (*Log, error) {
46 db, err := sql.Open("sqlite", "file:"+path+"?_pragma=journal_mode(WAL)&_pragma=busy_timeout(5000)&_pragma=synchronous(NORMAL)")
47 if err != nil {
48 return nil, err
49 }
50 db.SetMaxOpenConns(1) // one writer; reads are fast and serialised with it
51 if _, err := db.Exec(`create table if not exists events (
52 created integer primary key,
53 rkey text not null,
54 nsid text not null,
55 event text not null
56 )`); err != nil {
57 db.Close()
58 return nil, err
59 }
60 l := &Log{db: db, subs: map[chan struct{}]struct{}{}}
61 if err := db.QueryRow(`select coalesce(max(created), 0) from events`).Scan(&l.last); err != nil {
62 db.Close()
63 return nil, err
64 }
65 return l, nil
66}
67
68// DB exposes the database for packages that keep their own tables next to
69// the log (the runner's cursor and pipelines).
70func (l *Log) DB() *sql.DB { return l.db }
71
72// Close closes the database.
73func (l *Log) Close() error { return l.db.Close() }
74
75// Append adds a record (marshalled to JSON) with a new TID rkey unless rkey
76// is given, and wakes subscribers.
77func (l *Log) Append(nsid, rkey string, record any) (lex.Event, error) {
78 b, err := json.Marshal(record)
79 if err != nil {
80 return lex.Event{}, err
81 }
82 if rkey == "" {
83 rkey = tid.New()
84 }
85 l.mu.Lock()
86 created := time.Now().UnixNano()
87 if created <= l.last {
88 created = l.last + 1
89 }
90 _, err = l.db.Exec(`insert into events (created, rkey, nsid, event) values (?, ?, ?, ?)`, created, rkey, nsid, string(b))
91 if err == nil {
92 l.last = created
93 }
94 subs := make([]chan struct{}, 0, len(l.subs))
95 for c := range l.subs {
96 subs = append(subs, c)
97 }
98 l.mu.Unlock()
99 if err != nil {
100 return lex.Event{}, err
101 }
102 for _, c := range subs {
103 select {
104 case c <- struct{}{}:
105 default:
106 }
107 }
108 return lex.Event{Rkey: rkey, Nsid: nsid, Event: b, Created: created}, nil
109}
110
111// List returns up to limit events after cursor, oldest first.
112func (l *Log) List(cursor int64, limit int) ([]lex.Event, error) {
113 rows, err := l.db.Query(`select created, rkey, nsid, event from events where created > ? order by created limit ?`, cursor, limit)
114 if err != nil {
115 return nil, err
116 }
117 defer rows.Close()
118 var out []lex.Event
119 for rows.Next() {
120 var ev lex.Event
121 var body string
122 if err := rows.Scan(&ev.Created, &ev.Rkey, &ev.Nsid, &body); err != nil {
123 return nil, err
124 }
125 ev.Event = json.RawMessage(body)
126 out = append(out, ev)
127 }
128 return out, rows.Err()
129}
130
131// subscribe returns a channel poked after every append.
132func (l *Log) subscribe() (chan struct{}, func()) {
133 c := make(chan struct{}, 1)
134 l.mu.Lock()
135 l.subs[c] = struct{}{}
136 l.mu.Unlock()
137 return c, func() {
138 l.mu.Lock()
139 delete(l.subs, c)
140 l.mu.Unlock()
141 }
142}
143
144var upgrader = websocket.Upgrader{ReadBufferSize: 1024, WriteBufferSize: 1024}
145
146// ServeHTTP serves the stream (GET /events?cursor=N).
147func (l *Log) ServeHTTP(w http.ResponseWriter, r *http.Request) {
148 var cursor int64
149 if raw := r.URL.Query().Get("cursor"); raw != "" {
150 c, err := strconv.ParseInt(raw, 10, 64)
151 if err != nil {
152 http.Error(w, "invalid cursor", http.StatusBadRequest)
153 return
154 }
155 cursor = c
156 }
157 conn, err := upgrader.Upgrade(w, r, nil)
158 if err != nil {
159 return
160 }
161 defer conn.Close()
162 poke, unsubscribe := l.subscribe()
163 defer unsubscribe()
164
165 ctx, cancel := context.WithCancel(r.Context())
166 defer cancel()
167 go func() {
168 for {
169 if _, _, err := conn.NextReader(); err != nil {
170 cancel()
171 return
172 }
173 }
174 }()
175 drain := func() error {
176 for {
177 evs, err := l.List(cursor, batchSize)
178 if err != nil {
179 return err
180 }
181 for _, ev := range evs {
182 b, _ := json.Marshal(ev)
183 _ = conn.SetWriteDeadline(time.Now().Add(writeDeadline))
184 if err := conn.WriteMessage(websocket.TextMessage, b); err != nil {
185 return err
186 }
187 cursor = ev.Created
188 }
189 if len(evs) < batchSize {
190 return nil
191 }
192 }
193 }
194 if drain() != nil {
195 return
196 }
197 for {
198 select {
199 case <-ctx.Done():
200 return
201 case <-poke:
202 if drain() != nil {
203 return
204 }
205 case <-time.After(keepAliveInterval):
206 if conn.WriteControl(websocket.PingMessage, nil, time.Now().Add(writeDeadline)) != nil {
207 return
208 }
209 }
210 }
211}
212
213// Subscribe follows a remote stream from cursor, calling f for each event in
214// order, reconnecting with backoff until ctx ends. f's error stops the
215// subscription (return it to retry later from the last good cursor). token,
216// if set, is sent as a bearer token.
217func Subscribe(ctx context.Context, url string, cursor int64, token string, f func(lex.Event) error) error {
218 backoff := time.Second
219 for {
220 err := follow(ctx, url, &cursor, token, f)
221 if ctx.Err() != nil {
222 return ctx.Err()
223 }
224 if fe, ok := err.(handlerError); ok {
225 return fe.err
226 }
227 select {
228 case <-ctx.Done():
229 return ctx.Err()
230 case <-time.After(backoff):
231 }
232 backoff = min(backoff*2, time.Minute)
233 }
234}
235
236type handlerError struct{ err error }
237
238func (h handlerError) Error() string { return h.err.Error() }
239
240func follow(ctx context.Context, url string, cursor *int64, token string, f func(lex.Event) error) error {
241 h := http.Header{}
242 if token != "" {
243 h.Set("Authorization", "Bearer "+token)
244 }
245 conn, resp, err := websocket.DefaultDialer.DialContext(ctx, fmt.Sprintf("%s?cursor=%d", url, *cursor), h)
246 if err != nil {
247 if resp != nil {
248 return fmt.Errorf("%s: %s", url, resp.Status)
249 }
250 return err
251 }
252 defer conn.Close()
253 go func() {
254 <-ctx.Done()
255 conn.Close()
256 }()
257 for {
258 _, msg, err := conn.ReadMessage()
259 if err != nil {
260 return err
261 }
262 var ev lex.Event
263 if err := json.Unmarshal(msg, &ev); err != nil {
264 return err
265 }
266 if err := f(ev); err != nil {
267 return handlerError{err}
268 }
269 *cursor = ev.Created
270 }
271}
272
273// RequireToken wraps h so that requests must carry "Authorization: Bearer
274// <token>". An empty token disables the check.
275func RequireToken(token string, h http.Handler) http.Handler {
276 if token == "" {
277 return h
278 }
279 want := []byte("Bearer " + token)
280 return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
281 if subtle.ConstantTimeCompare([]byte(r.Header.Get("Authorization")), want) != 1 {
282 http.Error(w, "unauthorized", http.StatusUnauthorized)
283 return
284 }
285 h.ServeHTTP(w, r)
286 })
287}
