engine.go raw
1 package store
2
3 import (
4 "bytes"
5 "errors"
6 "fmt"
7 "math"
8 "os"
9 "path/filepath"
10 "sort"
11 "strconv"
12 "strings"
13 "sync"
14 "time"
15
16 "git.smesh.lol/torly/pkg/nostr"
17 )
18
19 var (
20 // ErrDuplicate means the event id is already stored.
21 ErrDuplicate = errors.New("store: duplicate event")
22 // ErrReplaced means a newer replaceable event already holds the slot.
23 ErrReplaced = errors.New("store: replaced by a newer event")
24 // ErrDeleted means the event id carries a NIP-09 tombstone.
25 ErrDeleted = errors.New("store: event was deleted")
26 // ErrNotFound means no event matches.
27 ErrNotFound = errors.New("store: event not found")
28 // ErrClosed means the engine has been closed.
29 ErrClosed = errors.New("store: closed")
30 )
31
32 // DefaultMaxResults bounds an unbounded query so that a single REQ cannot ask
33 // the relay to decode an entire database.
34 const DefaultMaxResults = 5000
35
36 // Options configures an Engine.
37 type Options struct {
38 // MaxResults is the result cap when a filter carries no limit (or a larger
39 // one). Zero selects DefaultMaxResults.
40 MaxResults int
41
42 // BulkImport defers the social graph indexes until RebuildGraph is called.
43 // A bulk importer writes millions of events; maintaining and backfilling
44 // edges after every one of them is far slower than building the graph in a
45 // single pass once all targets are present.
46 BulkImport bool
47 }
48
49 // Engine is a single-writer storage engine. All public methods serialize on
50 // one mutex: the relay is expected to verify signatures outside it and to let
51 // this engine own the ordering of writes.
52 type Engine struct {
53 dir string
54 mu sync.Mutex
55
56 w *wal
57 ckpt *checkpoint
58
59 nextPubSer uint64
60 lastSer uint64
61 saveN int
62 maxResults int
63 bulk bool
64 bulkRep map[string]uint64
65 bulkIDs map[[32]byte]uint64
66 bulkPub map[string]uint64
67 removedSer map[uint64]struct{}
68 closed bool
69
70 eid *sortedFile
71 sei *sortedFile
72 fpc *sortedFile
73 ca *sortedFile
74 exp *sortedFile
75 kc *sortedFile
76 pc *sortedFile
77 kpc *sortedFile
78 tc *sortedFile
79 tkc *sortedFile
80 tpc *sortedFile
81 tkp *sortedFile
82 wrd *sortedFile
83 rep *sortedFile
84 del *sortedFile
85
86 // Pubkey serials and social graph indexes.
87 pks *sortedFile
88 spk *sortedFile
89 epg *sortedFile
90 peg *sortedFile
91 eeg *sortedFile
92 gee *sortedFile
93 ppg *sortedFile
94 gpp *sortedFile
95 }
96
97 // Open opens or creates a store in dir.
98 func Open(dir string, opts Options) (*Engine, error) {
99 if opts.MaxResults <= 0 {
100 opts.MaxResults = DefaultMaxResults
101 }
102 if err := os.MkdirAll(filepath.Join(dir, "idx"), 0o755); err != nil {
103 return nil, err
104 }
105 w, err := openWAL(filepath.Join(dir, "wal"))
106 if err != nil {
107 return nil, err
108 }
109 w.noSync = opts.BulkImport
110 e := &Engine{dir: dir, w: w, nextPubSer: 1, maxResults: opts.MaxResults, bulk: opts.BulkImport}
111 if e.bulk {
112 e.bulkRep = make(map[string]uint64)
113 e.bulkIDs = make(map[[32]byte]uint64)
114 e.removedSer = make(map[uint64]struct{})
115 }
116 idx := filepath.Join(dir, "idx")
117
118 type spec struct {
119 dst **sortedFile
120 name string
121 recLen int
122 cmpLen int
123 serOff int
124 }
125 last := func(n int) int { return n - SerialLen }
126 specs := []spec{
127 {&e.eid, "eid", eidRecLen, eidCmpLen, 3 + HashLen},
128 {&e.sei, "sei", seiRecLen + seiValLen, seiCmpLen, 3},
129 {&e.fpc, "fpc", fpcRecLen, fpcCmpLen, 3},
130 {&e.ca, "ca", caRecLen, caCmpLen, last(caRecLen)},
131 {&e.exp, "exp", expRecLen, expCmpLen, last(expRecLen)},
132 {&e.kc, "kc", kcRecLen, kcCmpLen, last(kcRecLen)},
133 {&e.pc, "pc", pcRecLen, pcCmpLen, last(pcRecLen)},
134 {&e.kpc, "kpc", kpcRecLen, kpcCmpLen, last(kpcRecLen)},
135 {&e.tc, "tc", tcRecLen, tcCmpLen, last(tcRecLen)},
136 {&e.tkc, "tkc", tkcRecLen, tkcCmpLen, last(tkcRecLen)},
137 {&e.tpc, "tpc", tpcRecLen, tpcCmpLen, last(tpcRecLen)},
138 {&e.tkp, "tkp", tkpRecLen, tkpCmpLen, last(tkpRecLen)},
139 {&e.wrd, "wrd", wrdRecLen, wrdCmpLen, last(wrdRecLen)},
140 {&e.rep, "rep", repRecLen, repCmpLen, last(repRecLen)},
141 {&e.del, "del", delRecLen, delCmpLen, -1},
142 {&e.pks, "pks", pksRecLen, pksCmpLen, -1},
143 {&e.spk, "spk", spkRecLen + spkValLen, spkCmpLen, -1},
144 {&e.epg, "epg", epgRecLen, epgRecLen, -1},
145 {&e.peg, "peg", pegRecLen, pegRecLen, -1},
146 {&e.eeg, "eeg", eegRecLen, eegRecLen, -1},
147 {&e.gee, "gee", geeRecLen, geeRecLen, -1},
148 {&e.ppg, "ppg", ppgRecLen, ppgRecLen, -1},
149 {&e.gpp, "gpp", gppRecLen, gppRecLen, -1},
150 }
151 for _, s := range specs {
152 f, err := openSorted(filepath.Join(idx, s.name+".dat"), s.recLen, s.cmpLen)
153 if err != nil {
154 e.closeFiles()
155 return nil, fmt.Errorf("store: open %s index: %w", s.name, err)
156 }
157 f.bulk = opts.BulkImport
158 f.serOff = s.serOff
159 if opts.BulkImport {
160 f.removed = e.serialRemoved
161 }
162 *s.dst = f
163 }
164
165 c, err := openCheckpoint(filepath.Join(dir, "checkpoint.dat"))
166 if err != nil {
167 e.closeFiles()
168 return nil, err
169 }
170 e.ckpt = c
171 pos, nextPub := c.Get()
172 if nextPub > 0 {
173 e.nextPubSer = nextPub
174 }
175
176 // A checkpoint beyond the log's write position is stale: the log was
177 // truncated or replaced. Rebuild from the beginning.
178 if pos > w.Pos() {
179 pos = 0
180 }
181 // Older stores without graph indexes can be upgraded explicitly with
182 // `torly reindex-graph`. Do not infer that from empty pks/spk here: those
183 // files are also empty while a graph rebuild is in progress, and treating
184 // that as an old store would replay the whole WAL on every resume.
185 if pos == 0 {
186 if err := e.rebuild(); err != nil {
187 e.closeFiles()
188 return nil, err
189 }
190 } else {
191 if err := e.replay(pos); err != nil {
192 e.closeFiles()
193 return nil, err
194 }
195 }
196 // Bulk replay records removals as a serial set that scans consult. The
197 // records are still physically present until a merge, and the set is not
198 // persisted, so materialize the merge before advancing the checkpoint.
199 if e.bulk && len(e.removedSer) > 0 {
200 if err := e.Flush(); err != nil {
201 e.closeFiles()
202 return nil, err
203 }
204 }
205 if err := e.checkpointNow(); err != nil {
206 e.closeFiles()
207 return nil, err
208 }
209 if e.bulk {
210 if err := e.loadBulkIDs(); err != nil {
211 e.closeFiles()
212 return nil, err
213 }
214 if err := e.loadBulkRep(); err != nil {
215 e.closeFiles()
216 return nil, err
217 }
218 }
219 return e, nil
220 }
221
222 // Dir returns the store directory.
223 func (e *Engine) Dir() string { return e.dir }
224
225 // MaxResults returns the configured result cap.
226 func (e *Engine) MaxResults() int { return e.maxResults }
227
228 // MaxSerial returns the WAL serial of the newest stored event.
229 func (e *Engine) MaxSerial() uint64 {
230 e.mu.Lock()
231 defer e.mu.Unlock()
232 return e.lastSer
233 }
234
235 func (e *Engine) indexes() []*sortedFile {
236 return []*sortedFile{
237 e.eid, e.sei, e.fpc, e.ca, e.exp, e.kc, e.pc, e.kpc,
238 e.tc, e.tkc, e.tpc, e.tkp, e.wrd, e.rep, e.del,
239 e.pks, e.spk, e.epg, e.peg, e.eeg, e.gee, e.ppg, e.gpp,
240 }
241 }
242
243 func (e *Engine) closeFiles() {
244 for _, f := range e.indexes() {
245 if f != nil {
246 f.f.Close()
247 }
248 }
249 if e.w != nil {
250 e.w.Close()
251 }
252 if e.ckpt != nil {
253 e.ckpt.Close()
254 }
255 }
256
257 func (e *Engine) rebuild() error {
258 for _, f := range e.indexes() {
259 if err := f.Clear(); err != nil {
260 return err
261 }
262 }
263 e.nextPubSer = 1
264 if e.bulk {
265 e.bulkIDs = make(map[[32]byte]uint64)
266 e.bulkRep = make(map[string]uint64)
267 e.removedSer = make(map[uint64]struct{})
268 }
269 return e.replay(0)
270 }
271
272 // replay re-applies WAL records at or after pos. During a rebuild pos is 0
273 // and the indexes are empty.
274 func (e *Engine) replay(pos uint64) error {
275 var replayErr error
276 err := e.w.ForEachFrom(pos, func(ser uint64, data []byte) bool {
277 ev, derr := decodeEvent(data)
278 if derr != nil {
279 // A record that cannot be decoded is skipped; the log is the
280 // authority for bytes, not the index.
281 return true
282 }
283 e.lastSer = ser
284 if e.isTombstoned(ev.ID[:]) {
285 return true
286 }
287 keep, kerr := e.applyReplacement(ev)
288 if kerr != nil {
289 replayErr = kerr
290 return false
291 }
292 if !keep {
293 return true
294 }
295 if ierr := e.indexEvent(ev, ser); ierr != nil {
296 replayErr = ierr
297 return false
298 }
299 if e.bulk {
300 e.bulkIDs[ev.ID] = ser
301 if ev.IsReplaceable() || ev.IsAddressable() {
302 e.bulkRep[bulkSlotKey(ev)] = ser
303 }
304 }
305 if ev.IsDeletion() {
306 if derr := e.applyDeletion(ev); derr != nil {
307 replayErr = derr
308 return false
309 }
310 }
311 return true
312 })
313 if err != nil {
314 return err
315 }
316 return replayErr
317 }
318
319 // SaveEvent validates nothing about the event beyond its structural identity:
320 // the relay verifies signatures before calling this. It returns ErrDuplicate
321 // when the id is already stored and ErrReplaced when a newer replaceable event
322 // already holds the slot.
323 func (e *Engine) SaveEvent(ev *nostr.Event) (bool, error) {
324 e.mu.Lock()
325 defer e.mu.Unlock()
326 if e.closed {
327 return false, ErrClosed
328 }
329 if e.bulk {
330 if _, ok := e.bulkIDs[ev.ID]; ok {
331 return false, ErrDuplicate
332 }
333 } else {
334 dup, err := e.hasEvent(ev.ID[:])
335 if err != nil {
336 return false, err
337 }
338 if dup {
339 return false, ErrDuplicate
340 }
341 }
342 if e.isTombstoned(ev.ID[:]) {
343 return false, ErrDeleted
344 }
345 if ev.IsReplaceable() || ev.IsAddressable() {
346 if e.bulk {
347 if oldSer, ok := e.bulkRep[bulkSlotKey(ev)]; ok {
348 if old, gerr := e.getBySerial(oldSer); gerr == nil {
349 if !eventWins(ev, old) {
350 return false, ErrReplaced
351 }
352 e.removeEvent(old, oldSer)
353 }
354 }
355 } else {
356 oldSer, ok, err := e.currentReplacement(ev)
357 if err != nil {
358 return false, err
359 }
360 if ok {
361 old, gerr := e.getBySerial(oldSer)
362 if gerr == nil {
363 if !eventWins(ev, old) {
364 return false, ErrReplaced
365 }
366 e.removeEvent(old, oldSer)
367 }
368 }
369 }
370 }
371
372 ser, err := e.w.Append(encodeEvent(ev))
373 if err != nil {
374 return false, err
375 }
376 if err := e.indexEvent(ev, ser); err != nil {
377 return false, err
378 }
379 e.lastSer = ser
380 if e.bulk {
381 e.bulkIDs[ev.ID] = ser
382 if ev.IsReplaceable() || ev.IsAddressable() {
383 e.bulkRep[bulkSlotKey(ev)] = ser
384 }
385 }
386 if ev.IsDeletion() {
387 if err := e.applyDeletion(ev); err != nil {
388 return true, err
389 }
390 }
391 if e.bulk {
392 return true, nil
393 }
394 e.saveN++
395 if e.saveN >= 256 {
396 e.saveN = 0
397 if err := e.checkpointNow(); err != nil {
398 return true, err
399 }
400 }
401 return true, nil
402 }
403
404 // indexEvent writes every index record for one event.
405 func (e *Engine) indexEvent(ev *nostr.Event, ser uint64) error {
406 idHash := Hash8(ev.ID[:])
407 pubHash := Hash8(ev.PubKey[:])
408
409 put := func(f *sortedFile, rec []byte) error { return f.Put(rec) }
410 steps := []struct {
411 f *sortedFile
412 rec []byte
413 }{
414 {e.eid, makeEid(idHash, ser)},
415 {e.sei, makeSeiRec(ser, ev.ID[:])},
416 {e.fpc, makeFpc(ser, ev.ID[:], pubHash[:], ev.CreatedAt)},
417 {e.ca, makeCA(pCA, ev.CreatedAt, ser, caRecLen)},
418 {e.kc, makeKC(ev.Kind, ev.CreatedAt, ser)},
419 {e.pc, makePC(pubHash, ev.CreatedAt, ser)},
420 {e.kpc, makeKPC(ev.Kind, pubHash, ev.CreatedAt, ser)},
421 }
422 for _, s := range steps {
423 if err := put(s.f, s.rec); err != nil {
424 return err
425 }
426 }
427
428 var exp int64
429 for _, tag := range ev.Tags {
430 if len(tag) < 2 || len(tag[0]) != 1 {
431 continue
432 }
433 tagKey := tag[0][0]
434 vh := Hash8([]byte(tag[1]))
435 if err := put(e.tc, makeTC(tagKey, vh, ev.CreatedAt, ser)); err != nil {
436 return err
437 }
438 if err := put(e.tkc, makeTKC(ev.Kind, tagKey, vh, ev.CreatedAt, ser)); err != nil {
439 return err
440 }
441 if err := put(e.tpc, makeTPC(pubHash, tagKey, vh, ev.CreatedAt, ser)); err != nil {
442 return err
443 }
444 if err := put(e.tkp, makeTKP(ev.Kind, pubHash, tagKey, vh, ev.CreatedAt, ser)); err != nil {
445 return err
446 }
447 }
448 if tag, ok := ev.Tags.First("expiration"); ok {
449 if n, err := strconv.ParseInt(tag, 10, 64); err == nil && n > 0 {
450 exp = n
451 if err := put(e.exp, makeCA(pExp, exp, ser, expRecLen)); err != nil {
452 return err
453 }
454 }
455 }
456
457 words := splitWords(ev.Content)
458 if len(words) > 512 {
459 words = words[:512]
460 }
461 for _, word := range words {
462 h := Hash8([]byte(word))
463 if err := put(e.wrd, makeWrd(h, ser)); err != nil {
464 return err
465 }
466 }
467
468 if ev.IsReplaceable() || ev.IsAddressable() {
469 dHash := [HashLen]byte{}
470 if ev.IsAddressable() {
471 dHash = Hash8([]byte(ev.DTag()))
472 }
473 if err := put(e.rep, makeRep(ev.Kind, pubHash, dHash, ev.CreatedAt, ser)); err != nil {
474 return err
475 }
476 }
477 if e.bulk {
478 return nil
479 }
480 if err := e.indexGraph(ev, ser); err != nil {
481 return err
482 }
483 return e.backfillInboundETags(ev, ser)
484 }
485
486 // repKey returns the replacement slot key for an event.
487 func bulkSlotKey(ev *nostr.Event) string {
488 kind, pubHash, dHash := repKey(ev)
489 return bulkSlotKeyParts(kind, pubHash, dHash)
490 }
491
492 func bulkSlotKeyParts(kind int, pubHash, dHash [HashLen]byte) string {
493 var key [2 + HashLen + HashLen]byte
494 PutKind(key[:2], kind)
495 copy(key[2:], pubHash[:])
496 copy(key[2+HashLen:], dHash[:])
497 return string(key[:])
498 }
499
500 // loadBulkRep seeds the in-memory replacement slots from the persisted rep
501 // index. Without it, an import into an existing store would not know which
502 // replaceable event it must displace.
503 // loadBulkIDs seeds the in-memory id map from live events, so a bulk import
504 // into an existing store does not need to sort the growing eid buffer on every
505 // duplicate check.
506 func (e *Engine) loadBulkIDs() error {
507 e.bulkIDs = make(map[[32]byte]uint64)
508 return e.sei.Scan(nil, nil, func(rec []byte) bool {
509 var id [32]byte
510 copy(id[:], rec[seiRecLen:])
511 e.bulkIDs[id] = GetSerial(rec[3:])
512 return true
513 })
514 }
515
516 func (e *Engine) loadBulkRep() error {
517 e.bulkRep = make(map[string]uint64)
518 latestTS := make(map[string]int64)
519 return e.rep.Scan(nil, nil, func(rec []byte) bool {
520 kind := GetKind(rec[3:5])
521 var pubHash, dHash [HashLen]byte
522 copy(pubHash[:], rec[5:5+HashLen])
523 copy(dHash[:], rec[5+HashLen:5+2*HashLen])
524 ts := GetTimestamp(rec[5+2*HashLen:])
525 ser := GetSerial(rec[len(rec)-SerialLen:])
526 key := bulkSlotKeyParts(kind, pubHash, dHash)
527 if oldTS, ok := latestTS[key]; !ok || ts > oldTS {
528 latestTS[key] = ts
529 e.bulkRep[key] = ser
530 }
531 return true
532 })
533 }
534
535 func repKey(ev *nostr.Event) (kind int, pubHash, dHash [HashLen]byte) {
536 pubHash = Hash8(ev.PubKey[:])
537 if ev.IsAddressable() {
538 dHash = Hash8([]byte(ev.DTag()))
539 }
540 return ev.Kind, pubHash, dHash
541 }
542
543 // currentReplacement returns the serial of the newest event in the event's
544 // replacement slot.
545 func (e *Engine) currentReplacement(ev *nostr.Event) (uint64, bool, error) {
546 kind, pubHash, dHash := repKey(ev)
547 prefix := makeRepPrefix(kind, pubHash, dHash)
548 end := prefixEnd(prefix)
549 var last []byte
550 err := e.rep.Scan(prefix, end, func(rec []byte) bool {
551 last = rec
552 return true
553 })
554 if err != nil {
555 return 0, false, err
556 }
557 if last == nil {
558 return 0, false, nil
559 }
560 return GetSerial(last[len(last)-SerialLen:]), true, nil
561 }
562
563 // applyReplacement is the replay-time equivalent of the replacement check in
564 // SaveEvent: it drops an older event when a newer one arrives.
565 func (e *Engine) applyReplacement(ev *nostr.Event) (bool, error) {
566 if !ev.IsReplaceable() && !ev.IsAddressable() {
567 return true, nil
568 }
569 var oldSer uint64
570 var ok bool
571 if e.bulk {
572 oldSer, ok = e.bulkRep[bulkSlotKey(ev)]
573 } else {
574 var err error
575 oldSer, ok, err = e.currentReplacement(ev)
576 if err != nil {
577 return false, err
578 }
579 }
580 if !ok {
581 return true, nil
582 }
583 old, err := e.getBySerial(oldSer)
584 if err != nil {
585 return true, nil
586 }
587 if !eventWins(ev, old) {
588 return false, nil
589 }
590 e.removeEvent(old, oldSer)
591 return true, nil
592 }
593
594 // eventWins reports whether a should replace b: newer created_at wins, and on
595 // a tie the lexicographically lowest id wins (NIP-01).
596 func eventWins(a, b *nostr.Event) bool {
597 if a.CreatedAt != b.CreatedAt {
598 return a.CreatedAt > b.CreatedAt
599 }
600 return bytes.Compare(a.ID[:], b.ID[:]) < 0
601 }
602
603 // removeEvent drops an event from every index. The WAL bytes stay; the event
604 // becomes unreachable and a rebuild replays replacement/deletion semantics.
605 func (e *Engine) removeEvent(ev *nostr.Event, ser uint64) {
606 if e.bulk {
607 e.removedSer[ser] = struct{}{}
608 delete(e.bulkIDs, ev.ID)
609 if ev.IsReplaceable() || ev.IsAddressable() {
610 delete(e.bulkRep, bulkSlotKey(ev))
611 }
612 return
613 }
614 idHash := Hash8(ev.ID[:])
615 pubHash := Hash8(ev.PubKey[:])
616 del := func(f *sortedFile, rec []byte) { f.Delete(rec) }
617
618 del(e.eid, makeEid(idHash, ser))
619 del(e.sei, makeSei(ser))
620 del(e.fpc, makeFpc(ser, ev.ID[:], pubHash[:], ev.CreatedAt))
621 del(e.ca, makeCA(pCA, ev.CreatedAt, ser, caRecLen))
622 del(e.kc, makeKC(ev.Kind, ev.CreatedAt, ser))
623 del(e.pc, makePC(pubHash, ev.CreatedAt, ser))
624 del(e.kpc, makeKPC(ev.Kind, pubHash, ev.CreatedAt, ser))
625
626 for _, tag := range ev.Tags {
627 if len(tag) < 2 || len(tag[0]) != 1 {
628 continue
629 }
630 tagKey := tag[0][0]
631 vh := Hash8([]byte(tag[1]))
632 del(e.tc, makeTC(tagKey, vh, ev.CreatedAt, ser))
633 del(e.tkc, makeTKC(ev.Kind, tagKey, vh, ev.CreatedAt, ser))
634 del(e.tpc, makeTPC(pubHash, tagKey, vh, ev.CreatedAt, ser))
635 del(e.tkp, makeTKP(ev.Kind, pubHash, tagKey, vh, ev.CreatedAt, ser))
636 }
637 if exp := ev.Expiration(); exp > 0 {
638 del(e.exp, makeCA(pExp, exp, ser, expRecLen))
639 }
640 for _, word := range splitWords(ev.Content) {
641 h := Hash8([]byte(word))
642 del(e.wrd, makeWrd(h, ser))
643 }
644 if ev.IsReplaceable() || ev.IsAddressable() {
645 kind, ph, dh := repKey(ev)
646 del(e.rep, makeRep(kind, ph, dh, ev.CreatedAt, ser))
647 if e.bulk {
648 delete(e.bulkRep, bulkSlotKey(ev))
649 }
650 }
651 if e.bulk {
652 delete(e.bulkIDs, ev.ID)
653 }
654 if !e.bulk {
655 e.removeGraph(ev, ser)
656 }
657 }
658
659 // tombstone records a permanent deletion so that a WAL replay cannot bring
660 // the event back.
661 func (e *Engine) serialRemoved(ser uint64) bool {
662 _, ok := e.removedSer[ser]
663 return ok
664 }
665
666 func (e *Engine) tombstone(id []byte) error {
667 return e.del.Put(makeDel(id))
668 }
669
670 func (e *Engine) isTombstoned(id []byte) bool {
671 _, ok, _ := e.del.Get(makeDel(id))
672 return ok
673 }
674
675 // applyDeletion implements NIP-09: the author of the deletion event may remove
676 // their own events, by id (e tag) or by address (a tag).
677 func (e *Engine) applyDeletion(ev *nostr.Event) error {
678 for _, hexID := range ev.Tags.Values("e") {
679 var id [32]byte
680 if nostr.DecodeRaw(hexID, id[:]) != nil {
681 continue
682 }
683 target, ser, err := e.lookup(id[:])
684 if err != nil || target == nil {
685 continue
686 }
687 if !bytes.Equal(target.PubKey[:], ev.PubKey[:]) {
688 continue
689 }
690 if err := e.tombstone(id[:]); err != nil {
691 return err
692 }
693 e.removeEvent(target, ser)
694 }
695 for _, addr := range ev.Tags.Values("a") {
696 kind, pub, d, ok := parseAddress(addr)
697 if !ok || !bytes.Equal(pub[:], ev.PubKey[:]) {
698 continue
699 }
700 probe := &nostr.Event{Kind: kind}
701 copy(probe.PubKey[:], pub[:])
702 probe.Tags = nostr.Tags{{"d", d}}
703 ser, found, err := e.currentReplacement(probe)
704 if err != nil || !found {
705 continue
706 }
707 target, err := e.getBySerial(ser)
708 if err != nil || target == nil {
709 continue
710 }
711 if err := e.tombstone(target.ID[:]); err != nil {
712 return err
713 }
714 e.removeEvent(target, ser)
715 }
716 return nil
717 }
718
719 func parseAddress(s string) (kind int, pub [32]byte, d string, ok bool) {
720 parts := strings.SplitN(s, ":", 3)
721 if len(parts) != 3 {
722 return 0, pub, "", false
723 }
724 k, err := strconv.Atoi(parts[0])
725 if err != nil || k < 0 || k > nostr.MaxKind {
726 return 0, pub, "", false
727 }
728 if nostr.DecodeRaw(parts[1], pub[:]) != nil {
729 return 0, pub, "", false
730 }
731 return k, pub, parts[2], true
732 }
733
734 // hasEvent reports whether id is stored, filtering index hash collisions by
735 // comparing the full id held in the sei record.
736 func (e *Engine) hasEvent(id []byte) (bool, error) {
737 rec, ok, err := e.eid.Get(makeEidPrefix(Hash8(id)))
738 if err != nil || !ok {
739 return false, err
740 }
741 ser := GetSerial(rec[3+HashLen:])
742 seiRec, ok, err := e.sei.Get(makeSei(ser))
743 if err != nil || !ok {
744 return false, err
745 }
746 return bytes.Equal(seiRec[seiRecLen:seiRecLen+32], id), nil
747 }
748
749 // serialByID returns the serial for a live event id without decoding the
750 // event. Graph construction uses this for e-tag targets; decoding every target
751 // from the WAL turned a graph rebuild into a random-read workload.
752 func (e *Engine) serialByID(id []byte) (uint64, bool, error) {
753 if e.bulk {
754 var key [32]byte
755 copy(key[:], id)
756 ser, ok := e.bulkIDs[key]
757 return ser, ok, nil
758 }
759 rec, ok, err := e.eid.Get(makeEidPrefix(Hash8(id)))
760 if err != nil || !ok {
761 return 0, false, err
762 }
763 ser := GetSerial(rec[3+HashLen:])
764 seiRec, ok, err := e.sei.Get(makeSei(ser))
765 if err != nil || !ok || len(seiRec) < seiRecLen+seiValLen {
766 return 0, false, err
767 }
768 if !bytes.Equal(seiRec[seiRecLen:seiRecLen+seiValLen], id) {
769 return 0, false, nil
770 }
771 return ser, true, nil
772 }
773
774 func (e *Engine) lookup(id []byte) (*nostr.Event, uint64, error) {
775 if e.bulk {
776 var key [32]byte
777 copy(key[:], id)
778 ser, ok := e.bulkIDs[key]
779 if !ok {
780 return nil, 0, ErrNotFound
781 }
782 ev, err := e.getBySerial(ser)
783 if err != nil {
784 return nil, 0, err
785 }
786 if !bytes.Equal(ev.ID[:], id) {
787 return nil, 0, ErrNotFound
788 }
789 return ev, ser, nil
790 }
791 rec, ok, err := e.eid.Get(makeEidPrefix(Hash8(id)))
792 if err != nil || !ok {
793 return nil, 0, ErrNotFound
794 }
795 ser := GetSerial(rec[3+HashLen:])
796 ev, err := e.getBySerial(ser)
797 if err != nil {
798 return nil, 0, err
799 }
800 if !bytes.Equal(ev.ID[:], id) {
801 return nil, 0, ErrNotFound
802 }
803 return ev, ser, nil
804 }
805
806 func (e *Engine) getBySerial(ser uint64) (*nostr.Event, error) {
807 data, err := e.w.Read(ser)
808 if err != nil {
809 return nil, err
810 }
811 return decodeEvent(data)
812 }
813
814 // GetByID returns a stored event by id.
815 func (e *Engine) GetByID(id []byte) (*nostr.Event, error) {
816 e.mu.Lock()
817 defer e.mu.Unlock()
818 if e.isTombstoned(id) {
819 return nil, ErrNotFound
820 }
821 ev, _, err := e.lookup(id)
822 return ev, err
823 }
824
825 // GetBySerial returns a stored event by WAL serial.
826 func (e *Engine) GetBySerial(ser uint64) (*nostr.Event, error) {
827 e.mu.Lock()
828 defer e.mu.Unlock()
829 ev, err := e.getBySerial(ser)
830 if err != nil {
831 return nil, err
832 }
833 if e.isTombstoned(ev.ID[:]) {
834 return nil, ErrNotFound
835 }
836 return ev, nil
837 }
838
839 // Delete removes an event if byPub is its author. It returns false when the
840 // event does not exist or the author does not match.
841 func (e *Engine) Delete(id, byPub []byte) (bool, error) {
842 e.mu.Lock()
843 defer e.mu.Unlock()
844 ev, ser, err := e.lookup(id)
845 if err != nil {
846 return false, nil
847 }
848 if !bytes.Equal(ev.PubKey[:], byPub) {
849 return false, nil
850 }
851 if err := e.tombstone(id); err != nil {
852 return false, err
853 }
854 e.removeEvent(ev, ser)
855 return true, nil
856 }
857
858 // PurgeExpired removes NIP-40 events whose expiration is at or before now.
859 func (e *Engine) PurgeExpired(now int64) (int, error) {
860 e.mu.Lock()
861 defer e.mu.Unlock()
862 if err := e.flushPending(); err != nil {
863 return 0, err
864 }
865 start := makeCA(pExp, 0, 0, expRecLen)
866 var end []byte
867 if now >= math.MaxInt64-1 {
868 end = prefixEnd([]byte{'e', 'x', 'p'})
869 } else {
870 end = makeCA(pExp, now+1, 0, expRecLen)
871 }
872 var serials []uint64
873 if err := e.exp.Scan(start, end, func(rec []byte) bool {
874 serials = append(serials, GetSerial(rec[len(rec)-SerialLen:]))
875 return true
876 }); err != nil {
877 return 0, err
878 }
879 n := 0
880 for _, ser := range serials {
881 ev, err := e.getBySerial(ser)
882 if err != nil {
883 continue
884 }
885 exp := ev.Expiration()
886 if exp == 0 || exp > now {
887 continue
888 }
889 if err := e.tombstone(ev.ID[:]); err != nil {
890 return n, err
891 }
892 e.removeEvent(ev, ser)
893 n++
894 }
895 return n, nil
896 }
897
898 // Query returns matching events, newest first, capped by the filter limit and
899 // the engine's MaxResults.
900 func (e *Engine) Query(f *nostr.Filter) ([]*nostr.Event, error) {
901 if f.Graph != nil {
902 return nil, ErrGraph
903 }
904 e.mu.Lock()
905 defer e.mu.Unlock()
906 if e.closed {
907 return nil, ErrClosed
908 }
909 if err := e.flushPending(); err != nil {
910 return nil, fmt.Errorf("store: flush: %w", err)
911 }
912 cands, err := e.candidates(f)
913 if err != nil {
914 return nil, fmt.Errorf("store: candidates: %w", err)
915 }
916 limit := f.Limit
917 if limit <= 0 || limit > e.maxResults {
918 limit = e.maxResults
919 }
920 now := time.Now().Unix()
921
922 // Index scans already carry created_at. The indexes are ordered by
923 // (prefix, created_at), so sorting the candidate list by timestamp and
924 // loading only until the limit is reached avoids reading every matching
925 // event from the WAL. A broad filter such as kinds:[1] used to read and
926 // decode the entire kind before returning one result.
927 timed := true
928 for i := range cands {
929 if !cands[i].hasTS {
930 timed = false
931 break
932 }
933 }
934 if timed {
935 sort.Slice(cands, func(i, j int) bool {
936 if cands[i].ts != cands[j].ts {
937 return cands[i].ts > cands[j].ts
938 }
939 return cands[i].ser > cands[j].ser
940 })
941 out := make([]*nostr.Event, 0, limit)
942 for _, c := range cands {
943 ev, err := e.getBySerial(c.ser)
944 if err != nil {
945 continue
946 }
947 if e.isTombstoned(ev.ID[:]) {
948 continue
949 }
950 if exp := ev.Expiration(); exp != 0 && exp <= now {
951 continue
952 }
953 if !f.Matches(ev) {
954 continue
955 }
956 out = append(out, ev)
957 if len(out) >= limit {
958 break
959 }
960 }
961 return out, nil
962 }
963
964 // ID and full-text searches do not return a timestamp with their
965 // candidates. There are usually few of those, so fall back to loading and
966 // sorting the whole candidate set.
967 out := make([]*nostr.Event, 0, len(cands))
968 for _, c := range cands {
969 ev, err := e.getBySerial(c.ser)
970 if err != nil {
971 continue
972 }
973 if e.isTombstoned(ev.ID[:]) {
974 continue
975 }
976 if exp := ev.Expiration(); exp != 0 && exp <= now {
977 continue
978 }
979 if !f.Matches(ev) {
980 continue
981 }
982 out = append(out, ev)
983 }
984 sortEvents(out)
985 if len(out) > limit {
986 out = out[:limit]
987 }
988 return out, nil
989 }
990
991 // Count returns the number of matching events and whether the value is an
992 // approximation (the engine stopped at its result cap).
993 func (e *Engine) Count(f *nostr.Filter) (int, bool, error) {
994 if f.Graph != nil {
995 return 0, false, ErrGraph
996 }
997 e.mu.Lock()
998 defer e.mu.Unlock()
999 if e.closed {
1000 return 0, false, ErrClosed
1001 }
1002 if err := e.flushPending(); err != nil {
1003 return 0, false, fmt.Errorf("store: flush: %w", err)
1004 }
1005 cands, err := e.candidates(f)
1006 if err != nil {
1007 return 0, false, fmt.Errorf("store: candidates: %w", err)
1008 }
1009 now := time.Now().Unix()
1010 n := 0
1011 for _, c := range cands {
1012 ev, err := e.getBySerial(c.ser)
1013 if err != nil || e.isTombstoned(ev.ID[:]) {
1014 continue
1015 }
1016 if exp := ev.Expiration(); exp != 0 && exp <= now {
1017 continue
1018 }
1019 if !f.Matches(ev) {
1020 continue
1021 }
1022 n++
1023 if n >= e.maxResults {
1024 return n, true, nil
1025 }
1026 }
1027 return n, false, nil
1028 }
1029
1030 func sortEvents(evs []*nostr.Event) {
1031 sort.SliceStable(evs, func(i, j int) bool {
1032 if evs[i].CreatedAt != evs[j].CreatedAt {
1033 return evs[i].CreatedAt > evs[j].CreatedAt
1034 }
1035 return bytes.Compare(evs[i].ID[:], evs[j].ID[:]) < 0
1036 })
1037 }
1038
1039 // flushPending makes buffered index records visible to searches.
1040 func (e *Engine) flushPending() error {
1041 for _, f := range e.indexes() {
1042 if err := f.QuickFlush(false); err != nil {
1043 return err
1044 }
1045 }
1046 return nil
1047 }
1048
1049 // checkpointNow fsyncs the WAL and every index, then advances the checkpoint.
1050 // Only after this returns may recovery start after the checkpoint position:
1051 // before it, the sidecar writes are not guaranteed to have reached disk.
1052 func (e *Engine) checkpointNow() error {
1053 if err := e.w.Sync(); err != nil {
1054 return err
1055 }
1056 for _, f := range e.indexes() {
1057 if err := f.QuickFlush(true); err != nil {
1058 return err
1059 }
1060 }
1061 if err := e.ckpt.Set(e.w.Pos(), e.nextPubSer); err != nil {
1062 return err
1063 }
1064 return e.ckpt.Sync()
1065 }
1066
1067 // Flush writes everything durably and merges each index run into its main
1068 // file. It is safe to call at any time.
1069 func (e *Engine) Flush() error {
1070 e.mu.Lock()
1071 defer e.mu.Unlock()
1072 if e.closed {
1073 return ErrClosed
1074 }
1075 return e.flushLocked()
1076 }
1077
1078 func (e *Engine) flushLocked() error {
1079 if err := e.checkpointNow(); err != nil {
1080 return err
1081 }
1082 for _, f := range e.indexes() {
1083 if err := f.Flush(); err != nil {
1084 return err
1085 }
1086 }
1087 e.removedSer = nil
1088 return e.ckpt.Sync()
1089 }
1090
1091 // Compact merges every index without forcing an additional checkpoint.
1092 func (e *Engine) Compact() error {
1093 e.mu.Lock()
1094 defer e.mu.Unlock()
1095 if e.closed {
1096 return ErrClosed
1097 }
1098 for _, f := range e.indexes() {
1099 if err := f.Flush(); err != nil {
1100 return err
1101 }
1102 }
1103 return nil
1104 }
1105
1106 // Close flushes and closes the engine.
1107 func (e *Engine) Close() error {
1108 e.mu.Lock()
1109 defer e.mu.Unlock()
1110 if e.closed {
1111 return nil
1112 }
1113 e.closed = true
1114 var first error
1115 if err := e.w.Sync(); err != nil && first == nil {
1116 first = err
1117 }
1118 for _, f := range e.indexes() {
1119 if err := f.Flush(); err != nil && first == nil {
1120 first = err
1121 }
1122 }
1123 if err := e.ckpt.Set(e.w.Pos(), e.nextPubSer); err != nil && first == nil {
1124 first = err
1125 }
1126 if err := e.ckpt.Sync(); err != nil && first == nil {
1127 first = err
1128 }
1129 for _, f := range e.indexes() {
1130 if err := f.Close(); err != nil && first == nil {
1131 first = err
1132 }
1133 }
1134 if err := e.w.Close(); err != nil && first == nil {
1135 first = err
1136 }
1137 if err := e.ckpt.Close(); err != nil && first == nil {
1138 first = err
1139 }
1140 return first
1141 }
1142
1143 // --- query planning ---
1144
1145 // candidate is a serial plus the created_at carried by the index that
1146 // produced it. Records from eid/sei/wrd have no timestamp; callers fall back
1147 // to loading those events.
1148 type candidate struct {
1149 ser uint64
1150 ts int64
1151 hasTS bool
1152 }
1153
1154 func (e *Engine) candidates(f *nostr.Filter) ([]candidate, error) {
1155 if len(f.IDs) > 0 {
1156 return e.candidatesByID(f.IDs)
1157 }
1158 if f.Search != "" {
1159 return e.candidatesBySearch(f.Search)
1160 }
1161
1162 since := f.Since
1163 until := f.Until
1164 if until == 0 {
1165 until = math.MaxInt64
1166 }
1167 hasKinds := len(f.Kinds) > 0
1168 hasAuthors := len(f.Authors) > 0
1169 hasTags := len(f.Tags) > 0
1170
1171 var serials []candidate
1172 seen := make(map[uint64]struct{})
1173 add := func(ser uint64, ts int64, hasTS bool) {
1174 if _, ok := seen[ser]; ok {
1175 return
1176 }
1177 seen[ser] = struct{}{}
1178 serials = append(serials, candidate{ser: ser, ts: ts, hasTS: hasTS})
1179 }
1180 collect := func(idx *sortedFile, start, end []byte, tsOff int) error {
1181 return idx.Scan(start, end, func(rec []byte) bool {
1182 ser := GetSerial(rec[len(rec)-SerialLen:])
1183 if tsOff >= 0 && tsOff+8 <= len(rec) {
1184 add(ser, GetTimestamp(rec[tsOff:]), true)
1185 } else {
1186 add(ser, 0, false)
1187 }
1188 return true
1189 })
1190 }
1191
1192 // Range end: exclusive at the first timestamp after until. When until is
1193 // the maximum, the end is the successor of the fixed key prefix.
1194 rngEnd := func(start []byte, prefixLen int, build func(int64) []byte) []byte {
1195 if until >= math.MaxInt64-1 {
1196 return prefixEnd(start[:prefixLen])
1197 }
1198 return build(until + 1)
1199 }
1200
1201 switch {
1202 case hasTags && hasKinds && hasAuthors:
1203 if err := e.forEachTag(f, func(tagKey byte, vh [HashLen]byte) error {
1204 for _, kind := range f.Kinds {
1205 for _, author := range f.Authors {
1206 ph := authorHash(author)
1207 start := makeTKP(kind, ph, tagKey, vh, since, 0)
1208 end := rngEnd(start, 22, func(ts int64) []byte {
1209 return makeTKP(kind, ph, tagKey, vh, ts, 0)
1210 })
1211 if err := collect(e.tkp, start, end, 22); err != nil {
1212 return err
1213 }
1214 }
1215 }
1216 return nil
1217 }); err != nil {
1218 return nil, err
1219 }
1220 case hasTags && hasKinds:
1221 if err := e.forEachTag(f, func(tagKey byte, vh [HashLen]byte) error {
1222 for _, kind := range f.Kinds {
1223 start := makeTKC(kind, tagKey, vh, since, 0)
1224 end := rngEnd(start, 14, func(ts int64) []byte {
1225 return makeTKC(kind, tagKey, vh, ts, 0)
1226 })
1227 if err := collect(e.tkc, start, end, 14); err != nil {
1228 return err
1229 }
1230 }
1231 return nil
1232 }); err != nil {
1233 return nil, err
1234 }
1235 case hasTags && hasAuthors:
1236 if err := e.forEachTag(f, func(tagKey byte, vh [HashLen]byte) error {
1237 for _, author := range f.Authors {
1238 ph := authorHash(author)
1239 start := makeTPC(ph, tagKey, vh, since, 0)
1240 end := rngEnd(start, 20, func(ts int64) []byte {
1241 return makeTPC(ph, tagKey, vh, ts, 0)
1242 })
1243 if err := collect(e.tpc, start, end, 20); err != nil {
1244 return err
1245 }
1246 }
1247 return nil
1248 }); err != nil {
1249 return nil, err
1250 }
1251 case hasTags:
1252 if err := e.forEachTag(f, func(tagKey byte, vh [HashLen]byte) error {
1253 start := makeTC(tagKey, vh, since, 0)
1254 end := rngEnd(start, 12, func(ts int64) []byte {
1255 return makeTC(tagKey, vh, ts, 0)
1256 })
1257 return collect(e.tc, start, end, 12)
1258 }); err != nil {
1259 return nil, err
1260 }
1261 case hasKinds && hasAuthors:
1262 for _, kind := range f.Kinds {
1263 for _, author := range f.Authors {
1264 ph := authorHash(author)
1265 start := makeKPC(kind, ph, since, 0)
1266 end := rngEnd(start, 13, func(ts int64) []byte {
1267 return makeKPC(kind, ph, ts, 0)
1268 })
1269 if err := collect(e.kpc, start, end, 13); err != nil {
1270 return nil, err
1271 }
1272 }
1273 }
1274 case hasKinds:
1275 for _, kind := range f.Kinds {
1276 start := makeKC(kind, since, 0)
1277 end := rngEnd(start, 5, func(ts int64) []byte {
1278 return makeKC(kind, ts, 0)
1279 })
1280 if err := collect(e.kc, start, end, 5); err != nil {
1281 return nil, err
1282 }
1283 }
1284 case hasAuthors:
1285 for _, author := range f.Authors {
1286 ph := authorHash(author)
1287 start := makePC(ph, since, 0)
1288 end := rngEnd(start, 11, func(ts int64) []byte {
1289 return makePC(ph, ts, 0)
1290 })
1291 if err := collect(e.pc, start, end, 11); err != nil {
1292 return nil, err
1293 }
1294 }
1295 default:
1296 start := makeCA(pCA, since, 0, caRecLen)
1297 end := rngEnd(start, 3, func(ts int64) []byte {
1298 return makeCA(pCA, ts, 0, caRecLen)
1299 })
1300 if err := collect(e.ca, start, end, 3); err != nil {
1301 return nil, err
1302 }
1303 }
1304 return serials, nil
1305 }
1306
1307 func (e *Engine) candidatesByID(ids []string) ([]candidate, error) {
1308 var out []candidate
1309 seen := make(map[uint64]struct{})
1310 for _, h := range ids {
1311 var id [32]byte
1312 if nostr.DecodeRaw(h, id[:]) != nil {
1313 continue
1314 }
1315 ev, ser, err := e.lookup(id[:])
1316 if err != nil || ev == nil {
1317 continue
1318 }
1319 if _, ok := seen[ser]; ok {
1320 continue
1321 }
1322 seen[ser] = struct{}{}
1323 out = append(out, candidate{ser: ser})
1324 }
1325 return out, nil
1326 }
1327
1328 func (e *Engine) candidatesBySearch(query string) ([]candidate, error) {
1329 words := splitWords(query)
1330 if len(words) == 0 {
1331 return nil, nil
1332 }
1333 // Deduplicate query words.
1334 uniq := words[:0]
1335 seenWord := make(map[string]struct{}, len(words))
1336 for _, w := range words {
1337 if _, ok := seenWord[w]; ok {
1338 continue
1339 }
1340 seenWord[w] = struct{}{}
1341 uniq = append(uniq, w)
1342 }
1343
1344 var sets [][]uint64
1345 for _, word := range uniq {
1346 h := Hash8([]byte(word))
1347 start := makeWrd(h, 0)
1348 end := prefixEnd(start[:3+HashLen])
1349 var serials []uint64
1350 if err := e.wrd.Scan(start, end, func(rec []byte) bool {
1351 serials = append(serials, GetSerial(rec[len(rec)-SerialLen:]))
1352 return true
1353 }); err != nil {
1354 return nil, err
1355 }
1356 if len(serials) == 0 {
1357 return nil, nil
1358 }
1359 sets = append(sets, serials)
1360 }
1361
1362 result := sets[0]
1363 for _, next := range sets[1:] {
1364 set := make(map[uint64]struct{}, len(next))
1365 for _, s := range next {
1366 set[s] = struct{}{}
1367 }
1368 var keep []uint64
1369 for _, s := range result {
1370 if _, ok := set[s]; ok {
1371 keep = append(keep, s)
1372 }
1373 }
1374 result = keep
1375 if len(result) == 0 {
1376 return nil, nil
1377 }
1378 }
1379 out := make([]candidate, len(result))
1380 for i, ser := range result {
1381 out[i] = candidate{ser: ser}
1382 }
1383 return out, nil
1384 }
1385
1386 // authorHash hashes the raw 32 pubkey bytes of a hex author, matching the way
1387 // the pc/kpc/tpc/tkp indexes hash an event's PubKey field.
1388 func authorHash(hexAuthor string) (h [HashLen]byte) {
1389 var raw [32]byte
1390 if nostr.DecodeRaw(hexAuthor, raw[:]) != nil {
1391 return Hash8([]byte(hexAuthor))
1392 }
1393 return Hash8(raw[:])
1394 }
1395
1396 func (e *Engine) forEachTag(f *nostr.Filter, fn func(tagKey byte, vh [HashLen]byte) error) error {
1397 for name, values := range f.Tags {
1398 if len(name) != 1 {
1399 continue
1400 }
1401 for _, value := range values {
1402 if err := fn(name[0], Hash8([]byte(value))); err != nil {
1403 return err
1404 }
1405 }
1406 }
1407 return nil
1408 }
1409
1410 // splitWords lowercases ASCII runs of letters and digits and keeps runs of at
1411 // least three characters, matching the NIP-50 tokenizer used by morly.
1412 func splitWords(s string) []string {
1413 var out []string
1414 start := -1
1415 buf := make([]byte, 0, 32)
1416 flush := func(end int) {
1417 if start < 0 {
1418 return
1419 }
1420 if len(buf) >= 3 {
1421 out = append(out, string(buf))
1422 }
1423 buf = buf[:0]
1424 start = -1
1425 }
1426 for i := 0; i < len(s); i++ {
1427 c := s[i]
1428 switch {
1429 case c >= 'a' && c <= 'z', c >= '0' && c <= '9':
1430 if start < 0 {
1431 start = i
1432 }
1433 buf = append(buf, c)
1434 case c >= 'A' && c <= 'Z':
1435 if start < 0 {
1436 start = i
1437 }
1438 buf = append(buf, c+32)
1439 default:
1440 flush(i)
1441 }
1442 }
1443 flush(len(s))
1444 return out
1445 }
1446