1 // Package db is the database domain. It is the only owner of the store
2 // engine: no other domain ever sees the Engine value, a WAL handle or an
3 // index file. Requests and responses travel over fixed duplex channels.
4 package db
5
6 import (
7 "errors"
8 "fmt"
9 "sort"
10 "time"
11
12 "git.smesh.lol/torly/pkg/logx"
13 "git.smesh.lol/torly/pkg/nostr"
14 "git.smesh.lol/torly/pkg/store"
15 tinyspawn "git.smesh.lol/torly/pkg/tinyspawn"
16 "git.smesh.lol/torly/pkg/wire"
17 )
18
19 // Data is the domain's startup configuration.
20 type Data struct {
21 Dir string
22 MaxResults int
23 Expiry time.Duration
24 }
25
26 // Chans are the channels bound at spawn.
27 type Chans struct {
28 Req tinyspawn.Duplex[wire.DBRequest]
29 Resp tinyspawn.Duplex[wire.DBResponse]
30 Ready chan<- wire.Ready
31 Stop tinyspawn.Breaker
32 }
33
34 // Run is the domain entry point. It opens the store itself, so the Engine never
35 // crosses a domain boundary.
36 func Run(data Data, ch Chans) {
37 eng, err := store.Open(data.Dir, store.Options{MaxResults: data.MaxResults})
38 if err != nil {
39 ch.Ready <- wire.Ready{Err: err.Error()}
40 return
41 }
42 ch.Ready <- wire.Ready{OK: true}
43
44 interval := data.Expiry
45 if interval <= 0 {
46 interval = time.Minute
47 }
48 ticker := time.NewTicker(interval)
49 defer ticker.Stop()
50 defer eng.Close()
51
52 for {
53 select {
54 case req := <-ch.Req.Recv():
55 start := time.Now()
56 resp := handle(eng, req)
57 if took := time.Since(start); took > time.Second {
58 logx.Errorf("slow db request kind=%d id=%d took=%s events=%d", req.Kind, req.ID, took, len(resp.Events))
59 }
60 select {
61 case ch.Resp.Tx <- resp:
62 case <-ch.Stop.Done():
63 return
64 }
65 case <-ticker.C:
66 eng.PurgeExpired(time.Now().Unix())
67 case <-ch.Stop.Done():
68 return
69 }
70 }
71 }
72
73 func handle(eng *store.Engine, req wire.DBRequest) wire.DBResponse {
74 resp := wire.DBResponse{ID: req.ID, Kind: req.Kind}
75 switch req.Kind {
76 case wire.DBSave:
77 ev := req.Event
78 stored, err := eng.SaveEvent(&ev)
79 resp.Code = resultCode(err)
80 if err != nil {
81 resp.Err = err.Error()
82 }
83 resp.Stored = stored
84 case wire.DBQuery:
85 events, err := query(eng, req.Filters, req.Limit)
86 if err != nil {
87 resp.Code = wire.ResError
88 resp.Err = err.Error()
89 break
90 }
91 resp.Events = events
92 case wire.DBCount:
93 n, approx, err := count(eng, req.Filters)
94 if err != nil {
95 resp.Code = wire.ResError
96 resp.Err = err.Error()
97 break
98 }
99 resp.Count = n
100 resp.Approximate = approx
101 case wire.DBGraph:
102 res, err := eng.TraverseGraph(&req.Graph)
103 if err != nil {
104 resp.Code = wire.ResError
105 resp.Err = err.Error()
106 break
107 }
108 resp.Graph = res
109 default:
110 resp.Code = wire.ResError
111 resp.Err = "unknown database request"
112 }
113 return resp
114 }
115
116 func resultCode(err error) uint8 {
117 switch {
118 case err == nil:
119 return wire.ResOK
120 case errors.Is(err, store.ErrDuplicate):
121 return wire.ResDuplicate
122 case errors.Is(err, store.ErrReplaced):
123 return wire.ResReplaced
124 case errors.Is(err, store.ErrDeleted):
125 return wire.ResDeleted
126 default:
127 return wire.ResError
128 }
129 }
130
131 // query runs every filter and returns a de-duplicated, newest-first slice.
132 func query(eng *store.Engine, filters []nostr.Filter, limit int) ([]nostr.Event, error) {
133 if len(filters) == 0 {
134 return nil, nil
135 }
136 seen := make(map[[32]byte]bool)
137 var events []nostr.Event
138 for i := range filters {
139 got, err := eng.Query(&filters[i])
140 if err != nil {
141 return nil, fmt.Errorf("store query: %w", err)
142 }
143 for _, ev := range got {
144 if seen[ev.ID] {
145 continue
146 }
147 seen[ev.ID] = true
148 events = append(events, ev.Clone())
149 }
150 }
151 sort.SliceStable(events, func(i, j int) bool {
152 if events[i].CreatedAt != events[j].CreatedAt {
153 return events[i].CreatedAt > events[j].CreatedAt
154 }
155 return string(events[i].ID[:]) < string(events[j].ID[:])
156 })
157 if limit > 0 && len(events) > limit {
158 events = events[:limit]
159 }
160 return events, nil
161 }
162
163 func count(eng *store.Engine, filters []nostr.Filter) (int, bool, error) {
164 total := 0
165 approx := false
166 for i := range filters {
167 n, a, err := eng.Count(&filters[i])
168 if err != nil {
169 return 0, false, fmt.Errorf("store count: %w", err)
170 }
171 total += n
172 approx = approx || a
173 }
174 if len(filters) > 1 {
175 approx = true
176 }
177 return total, approx, nil
178 }
179