Something went wrong. Try again.
Monorepo for Tangled forked from tangled.org/core
Something went wrong. Try again.
1.4 kB · 79 lines
Go
at master
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980package eventstream
import ( "database/sql" "encoding/json" "sync" "time"
"tangled.org/core/notifier")
type Store interface { Exec(query string, args ...any) (sql.Result, error) Query(query string, args ...any) (*sql.Rows, error)}
var ( clockMu sync.Mutex lastNanos int64)
func Insert(s Store, ev Event, n *notifier.Notifier) error { clockMu.Lock() defer clockMu.Unlock()
if ev.Created == 0 { now := time.Now().UnixNano() if now <= lastNanos { now = lastNanos + 1 } ev.Created = now } if ev.Created > lastNanos { lastNanos = ev.Created }
if _, err := s.Exec( `insert into events (rkey, nsid, event, created) values (?, ?, ?, ?)`, ev.Rkey, ev.Nsid, []byte(ev.EventJson), ev.Created, ); err != nil { return err } n.NotifyAll() return nil}
func List(s Store, cursor int64, limit int) ([]Event, error) { rows, err := s.Query(` select rkey, nsid, event, created from events where created > ? order by created asc limit ? `, cursor, limit) if err != nil { return nil, err } defer rows.Close()
var out []Event for rows.Next() { var ev Event var eventJsonStr string if err := rows.Scan(&ev.Rkey, &ev.Nsid, &eventJsonStr, &ev.Created); err != nil { return nil, err } ev.EventJson = json.RawMessage(eventJsonStr) out = append(out, ev) }
if err := rows.Err(); err != nil { return nil, err }
return out, nil}