+20 -14

knotclient/events.go

··· 7 7 "log/slog" 8 8 "math/rand" 9 9 "net/url" 10 10 + "strconv" 10 11 "sync" 11 12 "time" 12 13 ··· 63 62 } 64 63 65 64 type CursorStore interface { 66 66 - Set(knot, cursor string) 67 67 - Get(knot string) (cursor string) 65 65 + Set(knot string, cursor int64) 66 66 + Get(knot string) (cursor int64) 68 67 } 69 68 70 69 type RedisCursorStore struct { ··· 81 80 cursorKey = "cursor:%s" 82 81 ) 83 82 84 84 - func (r *RedisCursorStore) Set(knot, cursor string) { 83 83 + func (r *RedisCursorStore) Set(knot string, cursor int64) { 85 84 key := fmt.Sprintf(cursorKey, knot) 86 85 r.rdb.Set(context.Background(), key, cursor, 0) 87 86 } 88 87 89 89 - func (r *RedisCursorStore) Get(knot string) (cursor string) { 88 88 + func (r *RedisCursorStore) Get(knot string) (cursor int64) { 90 89 key := fmt.Sprintf(cursorKey, knot) 91 90 val, err := r.rdb.Get(context.Background(), key).Result() 92 91 if err != nil { 93 93 - return "" 92 92 + return 0 94 93 } 95 94 96 96 - return val 95 95 + cursor, err = strconv.ParseInt(val, 10, 64) 96 96 + if err != nil { 97 97 + return 0 // optionally log parsing error 98 98 + } 99 99 + 100 100 + return cursor 97 101 } 98 102 99 103 type MemoryCursorStore struct { 100 104 store sync.Map 101 105 } 102 106 103 103 - func (m *MemoryCursorStore) Set(knot, cursor string) { 107 107 + func (m *MemoryCursorStore) Set(knot string, cursor int64) { 104 108 m.store.Store(knot, cursor) 105 109 } 106 110 107 107 - func (m *MemoryCursorStore) Get(knot string) (cursor string) { 111 111 + func (m *MemoryCursorStore) Get(knot string) (cursor int64) { 108 112 if result, ok := m.store.Load(knot); ok { 109 109 - if val, ok := result.(string); ok { 113 113 + if val, ok := result.(int64); ok { 110 114 return val 111 115 } 112 116 } 113 117 114 114 - return "" 118 118 + return 0 115 119 } 116 120 117 117 - func (e *EventConsumer) buildUrl(s EventSource, cursor string) (*url.URL, error) { 121 121 + func (e *EventConsumer) buildUrl(s EventSource, cursor int64) (*url.URL, error) { 118 122 scheme := "wss" 119 123 if e.cfg.Dev { 120 124 scheme = "ws" ··· 130 124 return nil, err 131 125 } 132 126 133 133 - if cursor != "" { 127 127 + if cursor != 0 { 134 128 query := url.Values{} 135 135 - query.Add("cursor", cursor) 129 129 + query.Add("cursor", fmt.Sprintf("%d", cursor)) 136 130 u.RawQuery = query.Encode() 137 131 } 138 132 return u, nil ··· 228 222 } 229 223 230 224 // update cursor 231 231 - c.cfg.CursorStore.Set(j.source.Knot, msg.Rkey) 225 225 + c.cfg.CursorStore.Set(j.source.Knot, time.Now().Unix()) 232 226 233 227 if err := c.cfg.ProcessFunc(j.source, msg); err != nil { 234 228 c.logger.Error("error processing message", "source", j.source, "err", err)

+10 -6

knotserver/db/events.go

··· 10 10 Rkey string `json:"rkey"` 11 11 Nsid string `json:"nsid"` 12 12 EventJson string `json:"event"` 13 13 + Created int64 `json:"created"` 13 14 } 14 15 15 16 func (d *DB) InsertEvent(event Event, notifier *notifier.Notifier) error { 17 17 + 16 18 _, err := d.db.Exec( 17 19 `insert into events (rkey, nsid, event) values (?, ?, ?)`, 18 20 event.Rkey, ··· 27 25 return err 28 26 } 29 27 30 30 - func (d *DB) GetEvents(cursor string) ([]Event, error) { 28 28 + func (d *DB) GetEvents(cursor int64) ([]Event, error) { 31 29 whereClause := "" 32 30 args := []any{} 33 33 - if cursor != "" { 34 34 - whereClause = "where rkey > ?" 31 31 + if cursor > 0 { 32 32 + whereClause = "where created > ?" 35 33 args = append(args, cursor) 36 34 } 37 35 38 36 query := fmt.Sprintf(` 39 39 - select rkey, nsid, event 37 37 + select rkey, nsid, event, created 40 38 from events 41 39 %s 42 42 - order by rkey asc 40 40 + order by created asc 43 41 limit 100 44 42 `, whereClause) 45 43 ··· 52 50 var evts []Event 53 51 for rows.Next() { 54 52 var ev Event 55 55 - rows.Scan(&ev.Rkey, &ev.Nsid, &ev.EventJson) 53 53 + if err := rows.Scan(&ev.Rkey, &ev.Nsid, &ev.EventJson, &ev.Created); err != nil { 54 54 + return nil, err 55 55 + } 56 56 evts = append(evts, ev) 57 57 } 58 58

+1

knotserver/db/init.go

··· 48 48 rkey text not null, 49 49 nsid text not null, 50 50 event text not null, -- json 51 51 + created integer not null default (strftime('%s', 'now')), 51 52 primary key (rkey, nsid) 52 53 ); 53 54 `)

+8 -3

knotserver/events.go

··· 4 4 "context" 5 5 "encoding/json" 6 6 "net/http" 7 7 + "strconv" 7 8 "time" 8 9 9 10 "github.com/gorilla/websocket" ··· 43 42 } 44 43 }() 45 44 46 46 - cursor := r.URL.Query().Get("cursor") 45 45 + cursorStr := r.URL.Query().Get("cursor") 46 46 + cursor, err := strconv.ParseInt(cursorStr, 10, 64) 47 47 + if err != nil { 48 48 + l.Error("empty or invalid cursor, defaulting to zero", "invalidCursor", cursorStr) 49 49 + } 47 50 48 51 // complete backfill first before going to live data 49 52 l.Debug("going through backfill", "cursor", cursor) ··· 79 74 } 80 75 } 81 76 82 82 - func (h *Handle) streamOps(conn *websocket.Conn, cursor *string) error { 77 77 + func (h *Handle) streamOps(conn *websocket.Conn, cursor *int64) error { 83 78 events, err := h.db.GetEvents(*cursor) 84 79 if err != nil { 85 80 h.l.Error("failed to fetch events from db", "err", err, "cursor", cursor) ··· 110 105 h.l.Debug("err", "err", err) 111 106 return err 112 107 } 113 113 - *cursor = event.Rkey 108 108 + *cursor = event.Created 114 109 } 115 110 116 111 return nil

+1 -1

nix/vm.nix

··· 21 21 g = config.services.tangled-knot.gitUser; 22 22 in [ 23 23 "d /var/lib/knot 0770 ${u} ${g} - -" # Create the directory first 24 24 - "f+ /var/lib/knot/secret 0660 ${u} ${g} - KNOT_SERVER_SECRET=40b4db20544e37a12ba3ed7353d4d4421a30e0593385068d2ef85263495794d8" 24 24 + "f+ /var/lib/knot/secret 0660 ${u} ${g} - KNOT_SERVER_SECRET=16154910ef55fe48121082c0b51fc0e360a8b15eb7bda7991d88dc9f7684427a" 25 25 ]; 26 26 services.tangled-knot = { 27 27 enable = true;