diff --git a/knotclient/events.go b/knotclient/events.go --- a/knotclient/events.go +++ b/knotclient/events.go @@ -7,6 +7,7 @@ "fmt" "log/slog" "math/rand" "net/url" + "strconv" "sync" "time" @@ -62,8 +63,8 @@ mu sync.RWMutex } type CursorStore interface { - Set(knot, cursor string) - Get(knot string) (cursor string) + Set(knot string, cursor int64) + Get(knot string) (cursor int64) } type RedisCursorStore struct { @@ -80,40 +81,45 @@ const ( cursorKey = "cursor:%s" ) -func (r *RedisCursorStore) Set(knot, cursor string) { +func (r *RedisCursorStore) Set(knot string, cursor int64) { key := fmt.Sprintf(cursorKey, knot) r.rdb.Set(context.Background(), key, cursor, 0) } -func (r *RedisCursorStore) Get(knot string) (cursor string) { +func (r *RedisCursorStore) Get(knot string) (cursor int64) { key := fmt.Sprintf(cursorKey, knot) val, err := r.rdb.Get(context.Background(), key).Result() if err != nil { - return "" + return 0 } - return val + cursor, err = strconv.ParseInt(val, 10, 64) + if err != nil { + return 0 // optionally log parsing error + } + + return cursor } type MemoryCursorStore struct { store sync.Map } -func (m *MemoryCursorStore) Set(knot, cursor string) { +func (m *MemoryCursorStore) Set(knot string, cursor int64) { m.store.Store(knot, cursor) } -func (m *MemoryCursorStore) Get(knot string) (cursor string) { +func (m *MemoryCursorStore) Get(knot string) (cursor int64) { if result, ok := m.store.Load(knot); ok { - if val, ok := result.(string); ok { + if val, ok := result.(int64); ok { return val } } - return "" + return 0 } -func (e *EventConsumer) buildUrl(s EventSource, cursor string) (*url.URL, error) { +func (e *EventConsumer) buildUrl(s EventSource, cursor int64) (*url.URL, error) { scheme := "wss" if e.cfg.Dev { scheme = "ws" @@ -124,9 +130,9 @@ if err != nil { return nil, err } - if cursor != "" { + if cursor != 0 { query := url.Values{} - query.Add("cursor", cursor) + query.Add("cursor", fmt.Sprintf("%d", cursor)) u.RawQuery = query.Encode() } return u, nil @@ -222,7 +228,7 @@ return } // update cursor - c.cfg.CursorStore.Set(j.source.Knot, msg.Rkey) + c.cfg.CursorStore.Set(j.source.Knot, time.Now().Unix()) if err := c.cfg.ProcessFunc(j.source, msg); err != nil { c.logger.Error("error processing message", "source", j.source, "err", err) diff --git a/knotserver/db/events.go b/knotserver/db/events.go --- a/knotserver/db/events.go +++ b/knotserver/db/events.go @@ -10,9 +10,11 @@ type Event struct { Rkey string `json:"rkey"` Nsid string `json:"nsid"` EventJson string `json:"event"` + Created int64 `json:"created"` } func (d *DB) InsertEvent(event Event, notifier *notifier.Notifier) error { + _, err := d.db.Exec( `insert into events (rkey, nsid, event) values (?, ?, ?)`, event.Rkey, @@ -25,19 +27,19 @@ return err } -func (d *DB) GetEvents(cursor string) ([]Event, error) { +func (d *DB) GetEvents(cursor int64) ([]Event, error) { whereClause := "" args := []any{} - if cursor != "" { - whereClause = "where rkey > ?" + if cursor > 0 { + whereClause = "where created > ?" args = append(args, cursor) } query := fmt.Sprintf(` - select rkey, nsid, event + select rkey, nsid, event, created from events %s - order by rkey asc + order by created asc limit 100 `, whereClause) @@ -50,7 +52,9 @@ var evts []Event for rows.Next() { var ev Event - rows.Scan(&ev.Rkey, &ev.Nsid, &ev.EventJson) + if err := rows.Scan(&ev.Rkey, &ev.Nsid, &ev.EventJson, &ev.Created); err != nil { + return nil, err + } evts = append(evts, ev) } diff --git a/knotserver/db/init.go b/knotserver/db/init.go --- a/knotserver/db/init.go +++ b/knotserver/db/init.go @@ -48,6 +48,7 @@ create table if not exists events ( rkey text not null, nsid text not null, event text not null, -- json + created integer not null default (strftime('%s', 'now')), primary key (rkey, nsid) ); `) diff --git a/knotserver/events.go b/knotserver/events.go --- a/knotserver/events.go +++ b/knotserver/events.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "net/http" + "strconv" "time" "github.com/gorilla/websocket" @@ -42,7 +43,11 @@ } } }() - cursor := r.URL.Query().Get("cursor") + cursorStr := r.URL.Query().Get("cursor") + cursor, err := strconv.ParseInt(cursorStr, 10, 64) + if err != nil { + l.Error("empty or invalid cursor, defaulting to zero", "invalidCursor", cursorStr) + } // complete backfill first before going to live data l.Debug("going through backfill", "cursor", cursor) @@ -74,7 +79,7 @@ } } } -func (h *Handle) streamOps(conn *websocket.Conn, cursor *string) error { +func (h *Handle) streamOps(conn *websocket.Conn, cursor *int64) error { events, err := h.db.GetEvents(*cursor) if err != nil { h.l.Error("failed to fetch events from db", "err", err, "cursor", cursor) @@ -105,7 +110,7 @@ if err := conn.WriteMessage(websocket.TextMessage, jsonMsg); err != nil { h.l.Debug("err", "err", err) return err } - *cursor = event.Rkey + *cursor = event.Created } return nil diff --git a/nix/vm.nix b/nix/vm.nix --- a/nix/vm.nix +++ b/nix/vm.nix @@ -21,7 +21,7 @@ u = config.services.tangled-knot.gitUser; g = config.services.tangled-knot.gitUser; in [ "d /var/lib/knot 0770 ${u} ${g} - -" # Create the directory first - "f+ /var/lib/knot/secret 0660 ${u} ${g} - KNOT_SERVER_SECRET=40b4db20544e37a12ba3ed7353d4d4421a30e0593385068d2ef85263495794d8" + "f+ /var/lib/knot/secret 0660 ${u} ${g} - KNOT_SERVER_SECRET=16154910ef55fe48121082c0b51fc0e360a8b15eb7bda7991d88dc9f7684427a" ]; services.tangled-knot = { enable = true;