store_js.go raw
1 //go:build js && wasm
2
3 package idbstore
4
5 import (
6 "context"
7 "errors"
8 "fmt"
9 "sync"
10 "syscall/js"
11 "time"
12
13 "git.smesh.lol/nostrino/pkg/btcwire"
14 "git.smesh.lol/nostrino/pkg/crypto/chainhash"
15 "git.smesh.lol/nostrino/pkg/neutrino"
16 )
17
18 const (
19 storeHeaders = "headers"
20 storeFHeaders = "fheaders"
21 storeFilters = "filters"
22 storeMeta = "meta"
23
24 anchorKey = "anchor"
25 queueDepth = 512
26 batchLimit = 200
27 opTimeout = 30 * time.Second
28 )
29
30 // Config opens one chain's database.
31 type Config struct {
32 // DBName is the IndexedDB database. Empty means "nostrino-<chain>".
33 DBName string
34 // Chain is used for the default database name.
35 Chain string
36 // FiltersFrom skips loading filters below this height. Filters are the
37 // large records; a swap wallet that scans recent history sets this to its
38 // birth height.
39 FiltersFrom uint32
40 }
41
42 // Store is a neutrino.Store backed by IndexedDB, with an in-memory working
43 // copy. Writes go to memory first and are persisted in batches.
44 type Store struct {
45 *neutrino.Memory
46
47 db js.Value
48 name string
49
50 queue chan op
51 stop chan struct{}
52 wg sync.WaitGroup
53
54 errMu sync.Mutex
55 err error
56
57 closeOnce sync.Once
58 closed chan struct{}
59 }
60
61 type op struct {
62 store string
63 value js.Value
64 barrier chan error
65 isDelete bool
66 deleteFrom uint32
67 }
68
69 // Open opens or creates the database and loads its records into memory.
70 func Open(ctx context.Context, cfg Config) (*Store, error) {
71 name := cfg.DBName
72 if name == "" {
73 name = "nostrino-" + cfg.Chain
74 }
75 db, err := openDB(ctx, name)
76 if err != nil {
77 return nil, err
78 }
79 s := &Store{
80 Memory: neutrino.NewMemory(),
81 db: db,
82 name: name,
83 queue: make(chan op, queueDepth),
84 stop: make(chan struct{}),
85 closed: make(chan struct{}),
86 }
87 s.wg.Add(1)
88 go s.flusher()
89 if err := s.load(ctx, cfg.FiltersFrom); err != nil {
90 s.Close(context.Background())
91 return nil, err
92 }
93 return s, nil
94 }
95
96 // Name is the database name.
97 func (s *Store) Name() string { return s.name }
98
99 const (
100 headersSchema = "headers"
101 fheadersSchema = "fheaders"
102 filtersSchema = "filters"
103 metaSchema = "meta"
104 )
105
106 func openDB(ctx context.Context, name string) (js.Value, error) {
107 idb := js.Global().Get("indexedDB")
108 if idb.Type() == js.TypeUndefined {
109 return js.Value{}, errors.New("idbstore: indexedDB is not available")
110 }
111 req := idb.Call("open", name, 1)
112 var upgrade, success, failure js.Func
113 done := make(chan error, 1)
114 upgrade = js.FuncOf(func(this js.Value, args []js.Value) any {
115 db := req.Get("result")
116 names := db.Get("objectStoreNames")
117 for _, spec := range []struct{ name, key string }{
118 {storeHeaders, "h"},
119 {storeFHeaders, "h"},
120 {storeFilters, "h"},
121 {storeMeta, "k"},
122 } {
123 if names.Call("contains", spec.name).Bool() {
124 continue
125 }
126 opts := js.Global().Get("Object").New()
127 opts.Set("keyPath", spec.key)
128 db.Call("createObjectStore", spec.name, opts)
129 }
130 return nil
131 })
132 success = js.FuncOf(func(this js.Value, args []js.Value) any {
133 select {
134 case done <- nil:
135 default:
136 }
137 return nil
138 })
139 failure = js.FuncOf(func(this js.Value, args []js.Value) any {
140 select {
141 case done <- errors.New("idbstore: open failed"):
142 default:
143 }
144 return nil
145 })
146 req.Set("onupgradeneeded", upgrade)
147 req.Set("onsuccess", success)
148 req.Set("onerror", failure)
149 defer func() {
150 upgrade.Release()
151 success.Release()
152 failure.Release()
153 }()
154 select {
155 case err := <-done:
156 if err != nil {
157 return js.Value{}, err
158 }
159 return req.Get("result"), nil
160 case <-ctx.Done():
161 return js.Value{}, ctx.Err()
162 }
163 }
164
165 func (s *Store) load(ctx context.Context, filtersFrom uint32) error {
166 anchor, ok, err := s.loadAnchor(ctx)
167 if err != nil {
168 return err
169 }
170 if ok {
171 if err := s.Memory.SetAnchor(anchor); err != nil {
172 return fmt.Errorf("idbstore: anchor: %w", err)
173 }
174 }
175 if err := s.loadHeaders(ctx); err != nil {
176 return err
177 }
178 if err := s.loadFilterHeaders(ctx); err != nil {
179 return err
180 }
181 return s.loadFilters(ctx, filtersFrom)
182 }
183
184 func (s *Store) loadAnchor(ctx context.Context) (neutrino.Anchor, bool, error) {
185 arr, err := s.getAll(ctx, storeMeta, nil)
186 if err != nil {
187 return neutrino.Anchor{}, false, err
188 }
189 for i := 0; i < arr.Length(); i++ {
190 rec := arr.Index(i)
191 if rec.Get("k").String() != anchorKey {
192 continue
193 }
194 var a neutrino.Anchor
195 a.Height = uint32(rec.Get("h").Int())
196 if err := copyBytes(a.BlockHash[:], rec.Get("bh")); err != nil {
197 return a, false, err
198 }
199 a.FilterHeight = int32(rec.Get("fh").Int())
200 if a.FilterHeight >= 0 {
201 if err := copyBytes(a.FilterHeader[:], rec.Get("fhd")); err != nil {
202 return a, false, err
203 }
204 }
205 return a, true, nil
206 }
207 return neutrino.Anchor{}, false, nil
208 }
209
210 func (s *Store) loadHeaders(ctx context.Context) error {
211 arr, err := s.getAll(ctx, storeHeaders, nil)
212 if err != nil {
213 return err
214 }
215 var raw [btcwire.BlockHeaderSize]byte
216 for i := 0; i < arr.Length(); i++ {
217 rec := arr.Index(i)
218 height := uint32(rec.Get("h").Int())
219 if err := copyBytes(raw[:], rec.Get("d")); err != nil {
220 return fmt.Errorf("idbstore: header %d: %w", height, err)
221 }
222 var h btcwire.BlockHeader
223 if err := h.Deserialize(raw[:]); err != nil {
224 return fmt.Errorf("idbstore: header %d: %w", height, err)
225 }
226 if err := s.Memory.PutHeaders(height, []btcwire.BlockHeader{h}); err != nil {
227 return fmt.Errorf("idbstore: load headers: %w", err)
228 }
229 }
230 return nil
231 }
232
233 func (s *Store) loadFilterHeaders(ctx context.Context) error {
234 arr, err := s.getAll(ctx, storeFHeaders, nil)
235 if err != nil {
236 return err
237 }
238 var fh, hd chainhash.Hash
239 for i := 0; i < arr.Length(); i++ {
240 rec := arr.Index(i)
241 height := uint32(rec.Get("h").Int())
242 if err := copyBytes(fh[:], rec.Get("fh")); err != nil {
243 return fmt.Errorf("idbstore: filter header %d: %w", height, err)
244 }
245 if err := copyBytes(hd[:], rec.Get("hd")); err != nil {
246 return fmt.Errorf("idbstore: filter header %d: %w", height, err)
247 }
248 if err := s.Memory.PutFilterHeaders(height, []chainhash.Hash{fh}, []chainhash.Hash{hd}); err != nil {
249 return fmt.Errorf("idbstore: load filter headers: %w", err)
250 }
251 }
252 return nil
253 }
254
255 func (s *Store) loadFilters(ctx context.Context, from uint32) error {
256 var rangeVal js.Value
257 if from > 0 {
258 kr := js.Global().Get("IDBKeyRange")
259 if kr.Type() != js.TypeUndefined {
260 rangeVal = kr.Call("lowerBound", int(from))
261 }
262 }
263 arr, err := s.getAll(ctx, storeFilters, &rangeVal)
264 if err != nil {
265 return err
266 }
267 for i := 0; i < arr.Length(); i++ {
268 rec := arr.Index(i)
269 height := uint32(rec.Get("h").Int())
270 var bh, fh chainhash.Hash
271 if err := copyBytes(bh[:], rec.Get("bh")); err != nil {
272 return fmt.Errorf("idbstore: filter %d: %w", height, err)
273 }
274 if err := copyBytes(fh[:], rec.Get("fh")); err != nil {
275 return fmt.Errorf("idbstore: filter %d: %w", height, err)
276 }
277 data := rec.Get("d")
278 buf := make([]byte, data.Length())
279 if err := copyBytes(buf, data); err != nil {
280 return fmt.Errorf("idbstore: filter %d: %w", height, err)
281 }
282 if err := s.Memory.PutFilter(height, bh, fh, buf); err != nil {
283 return fmt.Errorf("idbstore: load filters: %w", err)
284 }
285 }
286 return nil
287 }
288
289 func (s *Store) getAll(ctx context.Context, store string, rangeVal *js.Value) (js.Value, error) {
290 tx := s.db.Call("transaction", store, "readonly")
291 os := tx.Call("objectStore", store)
292 var req js.Value
293 if rangeVal != nil && rangeVal.Type() != js.TypeUndefined {
294 req = os.Call("getAll", *rangeVal)
295 } else {
296 req = os.Call("getAll")
297 }
298 var success, failure js.Func
299 done := make(chan error, 1)
300 success = js.FuncOf(func(this js.Value, args []js.Value) any {
301 select {
302 case done <- nil:
303 default:
304 }
305 return nil
306 })
307 failure = js.FuncOf(func(this js.Value, args []js.Value) any {
308 select {
309 case done <- errors.New("idbstore: read " + store + " failed"):
310 default:
311 }
312 return nil
313 })
314 req.Set("onsuccess", success)
315 req.Set("onerror", failure)
316 select {
317 case err := <-done:
318 success.Release()
319 failure.Release()
320 if err != nil {
321 return js.Value{}, err
322 }
323 return req.Get("result"), nil
324 case <-ctx.Done():
325 return js.Value{}, ctx.Err()
326 }
327 }
328
329 func (s *Store) flusher() {
330 defer s.wg.Done()
331 for {
332 select {
333 case <-s.stop:
334 return
335 case first := <-s.queue:
336 batch := []op{first}
337 for len(batch) < batchLimit {
338 select {
339 case next := <-s.queue:
340 batch = append(batch, next)
341 default:
342 goto write
343 }
344 }
345 write:
346 err := s.writeBatch(batch)
347 if err != nil {
348 s.setErr(err)
349 }
350 for _, o := range batch {
351 if o.barrier != nil {
352 o.barrier <- err
353 }
354 }
355 }
356 }
357 }
358
359 func (s *Store) writeBatch(batch []op) error {
360 var names []string
361 seen := make(map[string]bool, 4)
362 for _, o := range batch {
363 if o.store == "" || seen[o.store] {
364 continue
365 }
366 seen[o.store] = true
367 names = append(names, o.store)
368 }
369 if len(names) == 0 {
370 return nil
371 }
372 namesArray := js.Global().Get("Array").New()
373 for _, name := range names {
374 namesArray.Call("push", name)
375 }
376 tx := s.db.Call("transaction", namesArray, "readwrite")
377 for _, o := range batch {
378 if o.store == "" {
379 continue
380 }
381 os := tx.Call("objectStore", o.store)
382 if o.isDelete {
383 kr := js.Global().Get("IDBKeyRange")
384 if kr.Type() == js.TypeUndefined {
385 return errors.New("idbstore: IDBKeyRange is not available")
386 }
387 os.Call("delete", kr.Call("lowerBound", int(o.deleteFrom)))
388 continue
389 }
390 os.Call("put", o.value)
391 }
392 var success, failure js.Func
393 done := make(chan error, 1)
394 success = js.FuncOf(func(this js.Value, args []js.Value) any {
395 select {
396 case done <- nil:
397 default:
398 }
399 return nil
400 })
401 failure = js.FuncOf(func(this js.Value, args []js.Value) any {
402 select {
403 case done <- errors.New("idbstore: write failed"):
404 default:
405 }
406 return nil
407 })
408 tx.Set("oncomplete", success)
409 tx.Set("onerror", failure)
410 select {
411 case err := <-done:
412 success.Release()
413 failure.Release()
414 return err
415 case <-time.After(opTimeout):
416 success.Release()
417 failure.Release()
418 return errors.New("idbstore: write timed out")
419 }
420 }
421
422 func (s *Store) setErr(err error) {
423 s.errMu.Lock()
424 if s.err == nil {
425 s.err = err
426 }
427 s.errMu.Unlock()
428 }
429
430 func (s *Store) firstErr() error {
431 s.errMu.Lock()
432 defer s.errMu.Unlock()
433 return s.err
434 }
435
436 func (s *Store) enqueue(o op) error {
437 select {
438 case <-s.closed:
439 return errors.New("idbstore: closed")
440 default:
441 }
442 select {
443 case s.queue <- o:
444 return nil
445 case <-s.closed:
446 return errors.New("idbstore: closed")
447 }
448 }
449
450 // Flush waits until every queued write has committed.
451 func (s *Store) Flush(ctx context.Context) error {
452 barrier := make(chan error, 1)
453 if err := s.enqueue(op{barrier: barrier}); err != nil {
454 return err
455 }
456 select {
457 case err := <-barrier:
458 if err != nil {
459 return err
460 }
461 return s.firstErr()
462 case <-ctx.Done():
463 return ctx.Err()
464 }
465 }
466
467 // Close flushes, stops the writer and closes the database.
468 func (s *Store) Close(ctx context.Context) error {
469 var err error
470 s.closeOnce.Do(func() {
471 err = s.Flush(ctx)
472 close(s.stop)
473 s.wg.Wait()
474 close(s.closed)
475 s.db.Call("close")
476 })
477 return err
478 }
479
480 // SetAnchor stores the anchor through the same write-behind queue.
481 func (s *Store) SetAnchor(a neutrino.Anchor) error {
482 if err := s.Memory.SetAnchor(a); err != nil {
483 return err
484 }
485 rec := js.Global().Get("Object").New()
486 rec.Set("k", anchorKey)
487 rec.Set("h", int(a.Height))
488 rec.Set("bh", bytesValue(a.BlockHash[:]))
489 rec.Set("fh", int(a.FilterHeight))
490 rec.Set("fhd", bytesValue(a.FilterHeader[:]))
491 barrier := make(chan error, 1)
492 if err := s.enqueue(op{store: storeMeta, value: rec, barrier: barrier}); err != nil {
493 return err
494 }
495 select {
496 case err := <-barrier:
497 return err
498 case <-time.After(opTimeout):
499 return errors.New("idbstore: anchor write timed out")
500 }
501 }
502
503 // Truncate removes the in-memory state and deletes the persisted key ranges.
504 func (s *Store) Truncate(from uint32) error {
505 if err := s.Memory.Truncate(from); err != nil {
506 return err
507 }
508 for _, name := range []string{storeHeaders, storeFHeaders, storeFilters} {
509 if err := s.enqueue(op{store: name, isDelete: true, deleteFrom: from}); err != nil {
510 return err
511 }
512 }
513 barrier := make(chan error, 1)
514 if err := s.enqueue(op{barrier: barrier}); err != nil {
515 return err
516 }
517 select {
518 case err := <-barrier:
519 return err
520 case <-time.After(opTimeout):
521 return errors.New("idbstore: truncate timed out")
522 }
523 }
524
525 // PutHeaders stores headers in memory and queues them for persistence.
526 func (s *Store) PutHeaders(start uint32, headers []btcwire.BlockHeader) error {
527 if err := s.Memory.PutHeaders(start, headers); err != nil {
528 return err
529 }
530 for i, h := range headers {
531 raw := h.Serialize()
532 rec := js.Global().Get("Object").New()
533 rec.Set("h", int(start)+i)
534 rec.Set("d", bytesValue(raw[:]))
535 if err := s.enqueue(op{store: storeHeaders, value: rec}); err != nil {
536 return err
537 }
538 }
539 return nil
540 }
541
542 // PutFilterHeaders stores filter hashes and headers, then queues them.
543 func (s *Store) PutFilterHeaders(start uint32, filterHashes, headers []chainhash.Hash) error {
544 if err := s.Memory.PutFilterHeaders(start, filterHashes, headers); err != nil {
545 return err
546 }
547 for i, h := range headers {
548 rec := js.Global().Get("Object").New()
549 rec.Set("h", int(start)+i)
550 rec.Set("fh", bytesValue(filterHashes[i][:]))
551 rec.Set("hd", bytesValue(h[:]))
552 if err := s.enqueue(op{store: storeFHeaders, value: rec}); err != nil {
553 return err
554 }
555 }
556 return nil
557 }
558
559 // PutFilter stores a verified filter, then queues it.
560 func (s *Store) PutFilter(height uint32, blockHash, filterHash chainhash.Hash, filter []byte) error {
561 if err := s.Memory.PutFilter(height, blockHash, filterHash, filter); err != nil {
562 return err
563 }
564 rec := js.Global().Get("Object").New()
565 rec.Set("h", int(height))
566 rec.Set("bh", bytesValue(blockHash[:]))
567 rec.Set("fh", bytesValue(filterHash[:]))
568 rec.Set("d", bytesValue(filter))
569 return s.enqueue(op{store: storeFilters, value: rec})
570 }
571
572 func bytesValue(b []byte) js.Value {
573 arr := js.Global().Get("Uint8Array").New(len(b))
574 js.CopyBytesToJS(arr, b)
575 return arr
576 }
577
578 func copyBytes(dst []byte, v js.Value) error {
579 if v.Type() != js.TypeObject {
580 return errors.New("idbstore: field is not a typed array")
581 }
582 if v.Length() != len(dst) {
583 return fmt.Errorf("idbstore: field has %d bytes, want %d", v.Length(), len(dst))
584 }
585 js.CopyBytesToGo(dst, v)
586 return nil
587 }
588
589 var _ neutrino.Store = (*Store)(nil)
590