graph.go raw
1 package store
2
3 import (
4 "bytes"
5 "errors"
6 "fmt"
7 "os"
8 "sort"
9 "time"
10
11 "git.smesh.lol/torly/pkg/nostr"
12 )
13
14 // This file owns the social graph indexes and the _graph traversal.
15 //
16 // Every edge is stored twice, once in each direction:
17 //
18 // epg event -> pubkey peg pubkey -> event
19 // eeg event -> event gee event -> event
20 // ppg pubkey -> pubkey gpp pubkey -> pubkey
21 //
22 // The forward direction is what a traversal scans; the reverse direction is
23 // what makes deletion and inbound traversal one prefix scan instead of a full
24 // table scan.
25 //
26 // Pubkeys are interned in pks/spk first, so an edge carries two five-byte
27 // serials rather than two 32-byte keys.
28
29 // ErrGraph is returned for a malformed graph query.
30 var ErrGraph = errors.New("store: invalid graph query")
31
32 // lookupPubkeySerial resolves a pubkey without allocating a serial.
33 func (e *Engine) lookupPubkeySerial(pub []byte) (uint64, bool, error) {
34 if len(pub) < 32 {
35 return 0, false, errors.New("store: short pubkey")
36 }
37 h := Hash8(pub)
38 prefix := makePksPrefix(h)
39 // The common case is a direct point lookup. Hash collisions are possible
40 // but vanishingly rare, so only the fallback scans every row for the hash.
41 if rec, ok, err := e.pks.Get(prefix); err != nil {
42 return 0, false, err
43 } else if ok {
44 ser := GetSerial(rec[3+HashLen:])
45 if row, exists, err := e.spk.Get(makeSpk(ser)); err != nil {
46 return 0, false, err
47 } else if exists && len(row) >= spkRecLen+spkValLen &&
48 bytes.Equal(row[spkRecLen:spkRecLen+spkValLen], pub[:32]) {
49 return ser, true, nil
50 }
51 }
52
53 var found uint64
54 var ok bool
55 var scanErr error
56 err := e.pks.Scan(prefix, prefixEnd(prefix), func(rec []byte) bool {
57 ser := GetSerial(rec[3+HashLen:])
58 row, exists, err := e.spk.Get(makeSpk(ser))
59 if err != nil {
60 scanErr = err
61 return false
62 }
63 if exists && len(row) >= spkRecLen+spkValLen &&
64 bytes.Equal(row[spkRecLen:spkRecLen+spkValLen], pub[:32]) {
65 found = ser
66 ok = true
67 return false
68 }
69 return true
70 })
71 if err != nil {
72 return 0, false, err
73 }
74 if scanErr != nil {
75 return 0, false, scanErr
76 }
77 return found, ok, nil
78 }
79
80 // pubkeySerial interns a pubkey, allocating a serial the first time it is seen.
81 func (e *Engine) pubkeySerial(pub []byte) (uint64, error) {
82 if e.bulkPub != nil {
83 if ser, ok := e.bulkPub[string(pub[:32])]; ok {
84 return ser, nil
85 }
86 }
87 if ser, ok, err := e.lookupPubkeySerial(pub); err != nil {
88 return 0, err
89 } else if ok {
90 return ser, nil
91 }
92 ser := e.nextPubSer
93 if ser == 0 || ser > MaxSerial {
94 return 0, errors.New("store: pubkey serial space exhausted")
95 }
96 e.nextPubSer++
97 // The value is written before the lookup row: a crash between the two
98 // leaves an unreachable pubkey value, never a lookup row that resolves to
99 // a missing key.
100 if err := e.spk.Put(makeSpkRec(ser, pub)); err != nil {
101 return 0, err
102 }
103 if err := e.pks.Put(makePks(Hash8(pub), ser)); err != nil {
104 return 0, err
105 }
106 if e.bulkPub != nil {
107 e.bulkPub[string(pub[:32])] = ser
108 }
109 return ser, nil
110 }
111
112 func (e *Engine) pubkeyBySerial(ser uint64) ([32]byte, bool) {
113 var out [32]byte
114 row, ok, err := e.spk.Get(makeSpk(ser))
115 if err != nil || !ok || len(row) < spkRecLen+spkValLen {
116 return out, false
117 }
118 copy(out[:], row[spkRecLen:spkRecLen+spkValLen])
119 return out, true
120 }
121
122 func (e *Engine) eventIDBySerial(ser uint64) ([32]byte, bool) {
123 var out [32]byte
124 row, ok, err := e.sei.Get(makeSei(ser))
125 if err != nil || !ok || len(row) < seiRecLen+seiValLen {
126 return out, false
127 }
128 copy(out[:], row[seiRecLen:seiRecLen+seiValLen])
129 return out, true
130 }
131
132 // A single event can carry an unbounded tag list. Such events are not graph
133 // indexed at all: partial truncation would silently change the meaning of a
134 // follow list or thread, while indexing them makes one event dominate a graph
135 // rebuild. The event itself is still stored and available to NIP-01 queries.
136 const maxGraphTags = 256
137
138 func graphTagCount(ev *nostr.Event) int {
139 for i := range ev.Tags {
140 if i >= maxGraphTags {
141 return i + 1
142 }
143 }
144 return len(ev.Tags)
145 }
146
147 // indexGraph writes the author, p-tag and e-tag edges for one event. An event
148 // with more than maxGraphTags total tags contributes no graph edges at all.
149 func (e *Engine) indexGraph(ev *nostr.Event, ser uint64) error {
150 if graphTagCount(ev) > maxGraphTags {
151 return nil
152 }
153 authorSer, err := e.pubkeySerial(ev.PubKey[:])
154 if err != nil {
155 return err
156 }
157 if err := e.epg.Put(makeEpg(ser, authorSer, ev.Kind, DirAuthor)); err != nil {
158 return err
159 }
160 if err := e.peg.Put(makePeg(authorSer, ev.Kind, DirAuthor, ser)); err != nil {
161 return err
162 }
163
164 seenPK := map[uint64]struct{}{authorSer: {}}
165 for _, tag := range ev.Tags {
166 if tag.Name() != "p" || len(tag) < 2 {
167 continue
168 }
169 var pub [32]byte
170 if nostr.Decode32(tag[1], pub[:]) != nil {
171 continue
172 }
173 pkSer, err := e.pubkeySerial(pub[:])
174 if err != nil {
175 return err
176 }
177 if _, ok := seenPK[pkSer]; ok {
178 continue
179 }
180 seenPK[pkSer] = struct{}{}
181 if err := e.epg.Put(makeEpg(ser, pkSer, ev.Kind, DirPTagOut)); err != nil {
182 return err
183 }
184 if err := e.peg.Put(makePeg(pkSer, ev.Kind, DirPTagIn, ser)); err != nil {
185 return err
186 }
187 if err := e.ppg.Put(makePpg(authorSer, pkSer, ev.Kind, DirPKOut, ser)); err != nil {
188 return err
189 }
190 if err := e.gpp.Put(makeGpp(pkSer, ev.Kind, DirPKIn, authorSer, ser)); err != nil {
191 return err
192 }
193 }
194
195 // A single event can carry an unbounded tag list. The graph is an
196 // acceleration structure, so cap the work one event can cause while the
197 // event itself remains fully stored and queryable.
198 seenEV := make(map[uint64]struct{})
199 for _, tag := range ev.Tags {
200 if tag.Name() != "e" || len(tag) < 2 {
201 continue
202 }
203 var id [32]byte
204 if nostr.Decode32(tag[1], id[:]) != nil {
205 continue
206 }
207 tgtSer, ok, err := e.serialByID(id[:])
208 if err != nil || !ok {
209 // The target is not stored yet. backfillInboundETags creates this
210 // edge when it does arrive.
211 continue
212 }
213 if _, ok := seenEV[tgtSer]; ok {
214 continue
215 }
216 seenEV[tgtSer] = struct{}{}
217 if err := e.eeg.Put(makeEeg(ser, tgtSer, ev.Kind, DirETagOut)); err != nil {
218 return err
219 }
220 if err := e.gee.Put(makeGee(tgtSer, ev.Kind, DirETagIn, ser)); err != nil {
221 return err
222 }
223 }
224 return nil
225 }
226
227 // backfillInboundETags creates the inbound edges for events that were stored
228 // before the event they reference. Without it, publishing a reply before its
229 // root would leave the thread graph incomplete forever. The tag index already
230 // holds the reference, so the backfill costs one prefix scan per new event.
231 func (e *Engine) backfillInboundETags(ev *nostr.Event, ser uint64) error {
232 targetHex := nostr.Hex(ev.ID[:])
233 vh := Hash8([]byte(targetHex))
234 start := makeTC('e', vh, 0, 0)
235 end := prefixEnd(start[:3+1+HashLen])
236 var scanErr error
237 err := e.tc.Scan(start, end, func(rec []byte) bool {
238 srcSer := GetSerial(rec[len(rec)-SerialLen:])
239 if srcSer == ser {
240 return true
241 }
242 src, err := e.getBySerial(srcSer)
243 if err != nil || !src.Tags.Has("e", targetHex) {
244 return true
245 }
246 if graphTagCount(src) > maxGraphTags {
247 return true
248 }
249 row, ok, err := e.eeg.Get(makeEeg(srcSer, ser, src.Kind, DirETagOut))
250 if err != nil {
251 scanErr = err
252 return false
253 }
254 if ok && len(row) > 0 {
255 return true
256 }
257 if err := e.eeg.Put(makeEeg(srcSer, ser, src.Kind, DirETagOut)); err != nil {
258 scanErr = err
259 return false
260 }
261 if err := e.gee.Put(makeGee(ser, src.Kind, DirETagIn, srcSer)); err != nil {
262 scanErr = err
263 return false
264 }
265 return true
266 })
267 if err != nil {
268 return err
269 }
270 return scanErr
271 }
272
273 // removeGraph deletes every graph edge that names the event. Scans are used
274 // instead of key reconstruction because a referenced event may already have
275 // been removed and its serial can no longer be looked up.
276 func (e *Engine) removeGraph(ev *nostr.Event, ser uint64) {
277 // event -> pubkey, plus the matching reverse pubkey -> event rows.
278 _ = e.scanSerial(e.epg, pEpg, ser, func(rec []byte) bool {
279 pkSer := GetSerial(rec[8:13])
280 kind := GetKind(rec[13:15])
281 e.epg.Delete(rec)
282 switch rec[15] {
283 case DirAuthor:
284 e.peg.Delete(makePeg(pkSer, kind, DirAuthor, ser))
285 case DirPTagOut:
286 e.peg.Delete(makePeg(pkSer, kind, DirPTagIn, ser))
287 }
288 return true
289 })
290
291 // event -> event, outbound e-tags.
292 _ = e.scanSerial(e.eeg, pEeg, ser, func(rec []byte) bool {
293 tgtSer := GetSerial(rec[8:13])
294 kind := GetKind(rec[13:15])
295 e.eeg.Delete(rec)
296 e.gee.Delete(makeGee(tgtSer, kind, DirETagIn, ser))
297 return true
298 })
299
300 // event -> event, inbound e-tags where this event is the target.
301 _ = e.scanSerial(e.gee, pGee, ser, func(rec []byte) bool {
302 kind := GetKind(rec[8:10])
303 srcSer := GetSerial(rec[11:16])
304 e.gee.Delete(rec)
305 e.eeg.Delete(makeEeg(srcSer, ser, kind, DirETagOut))
306 return true
307 })
308
309 authorSer, ok, err := e.lookupPubkeySerial(ev.PubKey[:])
310 if err != nil || !ok {
311 return
312 }
313
314 // pubkey -> pubkey, edges sourced by this event's author.
315 _ = e.scanSerial(e.ppg, pPpg, authorSer, func(rec []byte) bool {
316 if GetSerial(rec[16:21]) != ser {
317 return true
318 }
319 tgtPK := GetSerial(rec[8:13])
320 kind := GetKind(rec[13:15])
321 e.ppg.Delete(rec)
322 e.gpp.Delete(makeGpp(tgtPK, kind, DirPKIn, authorSer, ser))
323 return true
324 })
325 }
326
327 func (e *Engine) scanSerial(f *sortedFile, p prefix, ser uint64, fn func(rec []byte) bool) error {
328 start := graphPrefix(p, ser)
329 return f.Scan(start, prefixEnd(start), fn)
330 }
331
332 // TraverseGraph runs a _graph query against the materialized indexes.
333 func (e *Engine) TraverseGraph(q *nostr.GraphQuery) (*nostr.GraphResult, error) {
334 e.mu.Lock()
335 defer e.mu.Unlock()
336 if e.closed {
337 return nil, ErrClosed
338 }
339 if err := q.Validate(); err != nil {
340 return nil, err
341 }
342 if err := e.flushPending(); err != nil {
343 return nil, err
344 }
345 start := time.Now()
346 var (
347 res *nostr.GraphResult
348 err error
349 )
350 switch q.Edge {
351 case nostr.GraphEdgePubkeyPubkey:
352 res, err = e.traversePubkeyPubkey(q)
353 case nostr.GraphEdgePubkeyEvent:
354 res, err = e.traversePubkeyEvent(q)
355 case nostr.GraphEdgeEventEvent:
356 res, err = e.traverseEventEvent(q)
357 default:
358 return nil, ErrGraph
359 }
360 if err != nil {
361 return nil, err
362 }
363 res.Sort()
364 res.Elapsed = time.Since(start).String()
365 return res, nil
366 }
367
368 func (e *Engine) seedPubkey(q *nostr.GraphQuery) (uint64, bool, error) {
369 var raw [32]byte
370 if err := nostr.Decode32(q.Seed, raw[:]); err != nil {
371 return 0, false, errGraphSeed()
372 }
373 return e.lookupPubkeySerial(raw[:])
374 }
375
376 func errGraphSeed() error { return errors.New("store: graph seed is not a pubkey") }
377
378 // traversePubkeyPubkey is a breadth-first walk of kind-3 p-tag edges. Kind 3
379 // is the contact list: a p tag there is a follow.
380 func (e *Engine) traversePubkeyPubkey(q *nostr.GraphQuery) (*nostr.GraphResult, error) {
381 res := nostr.NewGraphResult()
382 seed, ok, err := e.seedPubkey(q)
383 if err != nil || !ok {
384 return res, err
385 }
386 visited := map[uint64]struct{}{seed: {}}
387 frontier := []uint64{seed}
388 var scanErr error
389
390 for depth := 1; depth <= q.Depth; depth++ {
391 var next []uint64
392 for _, ser := range frontier {
393 if res.TotalPubkeys >= e.maxResults {
394 break
395 }
396 visit := func(neighbor uint64) bool {
397 if res.TotalPubkeys >= e.maxResults {
398 return false
399 }
400 if _, seen := visited[neighbor]; seen {
401 return true
402 }
403 visited[neighbor] = struct{}{}
404 next = append(next, neighbor)
405 pub, ok := e.pubkeyBySerial(neighbor)
406 if ok {
407 res.AddPubkeyAtDepth(nostr.Hex(pub[:]), depth)
408 }
409 return res.TotalPubkeys < e.maxResults
410 }
411 if q.IsOutbound() {
412 scanErr = e.scanSerial(e.ppg, pPpg, ser, func(rec []byte) bool {
413 if GetKind(rec[13:15]) != 3 {
414 return true
415 }
416 return visit(GetSerial(rec[8:13]))
417 })
418 }
419 if scanErr == nil && q.IsInbound() {
420 scanErr = e.scanSerial(e.gpp, pGpp, ser, func(rec []byte) bool {
421 if GetKind(rec[8:10]) != 3 {
422 return true
423 }
424 return visit(GetSerial(rec[11:16]))
425 })
426 }
427 if scanErr != nil {
428 return nil, scanErr
429 }
430 }
431 if len(next) == 0 {
432 break
433 }
434 frontier = next
435 }
436 return res, nil
437 }
438
439 // traversePubkeyEvent returns the events directly connected to the seed
440 // pubkey: authored events for "out", p-tag mentions for "in". The pubkey↔event
441 // graph is bipartite, so a second hop would change the result type; orly's
442 // extension also defines pe as a one-hop query.
443 func (e *Engine) traversePubkeyEvent(q *nostr.GraphQuery) (*nostr.GraphResult, error) {
444 res := nostr.NewGraphResult()
445 seed, ok, err := e.seedPubkey(q)
446 if err != nil || !ok {
447 return res, err
448 }
449 visit := func(ser uint64) bool {
450 if res.TotalEvents >= e.maxResults {
451 return false
452 }
453 id, ok := e.eventIDBySerial(ser)
454 if ok {
455 res.AddEventAtDepth(nostr.Hex(id[:]), 1)
456 }
457 return res.TotalEvents < e.maxResults
458 }
459 if q.IsOutbound() {
460 if err := e.scanSerial(e.peg, pPeg, seed, func(rec []byte) bool {
461 if rec[10] != DirAuthor {
462 return true
463 }
464 return visit(GetSerial(rec[11:16]))
465 }); err != nil {
466 return nil, err
467 }
468 }
469 if q.IsInbound() {
470 if err := e.scanSerial(e.peg, pPeg, seed, func(rec []byte) bool {
471 if rec[10] != DirPTagIn {
472 return true
473 }
474 return visit(GetSerial(rec[11:16]))
475 }); err != nil {
476 return nil, err
477 }
478 }
479 return res, nil
480 }
481
482 // traverseEventEvent starts from every event authored by the seed pubkey and
483 // follows e-tag edges. The authored events are roots, not results.
484 func (e *Engine) traverseEventEvent(q *nostr.GraphQuery) (*nostr.GraphResult, error) {
485 res := nostr.NewGraphResult()
486 seed, ok, err := e.seedPubkey(q)
487 if err != nil || !ok {
488 return res, err
489 }
490 roots, err := e.authoredEventSerials(seed)
491 if err != nil {
492 return nil, err
493 }
494 sort.Slice(roots, func(i, j int) bool { return roots[i] < roots[j] })
495
496 seen := make(map[uint64]struct{}, len(roots))
497 for _, root := range roots {
498 seen[root] = struct{}{}
499 }
500 for _, root := range roots {
501 local := map[uint64]struct{}{root: {}}
502 frontier := []uint64{root}
503 for depth := 1; depth <= q.Depth; depth++ {
504 var next []uint64
505 for _, ser := range frontier {
506 if res.TotalEvents >= e.maxResults {
507 break
508 }
509 visit := func(neighbor uint64) bool {
510 if _, ok := local[neighbor]; ok {
511 return true
512 }
513 local[neighbor] = struct{}{}
514 next = append(next, neighbor)
515 if _, ok := seen[neighbor]; !ok {
516 seen[neighbor] = struct{}{}
517 if id, ok := e.eventIDBySerial(neighbor); ok {
518 res.AddEventAtDepth(nostr.Hex(id[:]), depth)
519 }
520 }
521 return res.TotalEvents < e.maxResults
522 }
523 if q.IsOutbound() {
524 err = e.scanSerial(e.eeg, pEeg, ser, func(rec []byte) bool {
525 return visit(GetSerial(rec[8:13]))
526 })
527 }
528 if err == nil && q.IsInbound() {
529 err = e.scanSerial(e.gee, pGee, ser, func(rec []byte) bool {
530 return visit(GetSerial(rec[11:16]))
531 })
532 }
533 if err != nil {
534 return nil, err
535 }
536 }
537 if len(next) == 0 {
538 break
539 }
540 frontier = next
541 }
542 }
543 return res, nil
544 }
545
546 func (e *Engine) authoredEventSerials(pubSer uint64) ([]uint64, error) {
547 var out []uint64
548 seen := make(map[uint64]struct{})
549 err := e.scanSerial(e.peg, pPeg, pubSer, func(rec []byte) bool {
550 if rec[10] != DirAuthor {
551 return true
552 }
553 ser := GetSerial(rec[11:16])
554 if _, ok := seen[ser]; ok {
555 return true
556 }
557 seen[ser] = struct{}{}
558 out = append(out, ser)
559 return true
560 })
561 return out, err
562 }
563
564 // RebuildGraph clears the pubkey and graph indexes and materializes them from
565 // the events that survived import. Bulk imports skip graph maintenance while
566 // writing; by the time this runs every e-tag target that exists is already
567 // stored, so no per-event backfill scan is needed.
568 func (e *Engine) RebuildGraph() error {
569 e.mu.Lock()
570 defer e.mu.Unlock()
571 if e.closed {
572 return ErrClosed
573 }
574 for _, f := range []*sortedFile{e.pks, e.spk, e.epg, e.peg, e.eeg, e.gee, e.ppg, e.gpp} {
575 if err := f.Clear(); err != nil {
576 return err
577 }
578 }
579 e.nextPubSer = 1
580 e.bulkPub = make(map[string]uint64)
581 if err := e.spk.Scan(nil, nil, func(rec []byte) bool {
582 var pub [32]byte
583 copy(pub[:], rec[spkRecLen:spkRecLen+spkValLen])
584 e.bulkPub[string(pub[:])] = GetSerial(rec[3:])
585 return true
586 }); err != nil {
587 return err
588 }
589
590 live := e.bulkIDs
591 if live == nil {
592 live = make(map[[32]byte]uint64)
593 if err := e.sei.Scan(nil, nil, func(rec []byte) bool {
594 var id [32]byte
595 copy(id[:], rec[seiRecLen:])
596 live[id] = GetSerial(rec[3:])
597 return true
598 }); err != nil {
599 return err
600 }
601 }
602 total := len(live)
603 done := 0
604 var graphErr error
605 err := e.w.ForEachFrom(0, func(ser uint64, data []byte) bool {
606 ev, derr := decodeEvent(data)
607 if derr != nil {
608 return true
609 }
610 if _, ok := live[ev.ID]; !ok {
611 return true
612 }
613 if ierr := e.indexGraph(ev, ser); ierr != nil {
614 graphErr = ierr
615 return false
616 }
617 done++
618 if done%100000 == 0 {
619 fmt.Fprintf(os.Stderr, "graph: %d/%d live events\n", done, total)
620 }
621 return true
622 })
623 if err != nil {
624 return err
625 }
626 if graphErr != nil {
627 return graphErr
628 }
629
630 e.bulk = false
631 e.bulkRep = nil
632 e.bulkIDs = nil
633 e.bulkPub = nil
634 return e.flushPending()
635 }
636