this repo has no description
backfill: use constants for bufferedOp.kind (#641)
(Part of https://github.com/bluesky-social/indigo/issues/638)
This utilizes constants from the `repomgr` package for buffered
operations.
This was already the case, actually, as they were used
[here](https://github.com/bluesky-social/indigo/blob/main/backfill/backfill.go#L241),
I just added them throughout the code.
They're also used to check the operations emitted by the firehose:
https://github.com/bluesky-social/indigo/blob/main/backfill/backfill.go#L469-L471
The firehose, in turn, uses the events emitted by the repo manager for
the `Action` field in firehose events:
https://github.com/bluesky-social/indigo/blob/main/indexer/indexer.go#L83-L94
This brings us full circle:
Repo manager emits diff operations -> firehose repo commit ops `.Action`
-> backfill buffered op `.kind`
Notably, the constants are also given in the lexicon:
https://github.com/bluesky-social/atproto/blob/main/lexicons/com/atproto/sync/subscribeRepos.json#L178
But `lexgen` does not generate constants for them.
The test in `backfill_test.go` is broken, but I changed the constants
anyway.
+27
-24
+14
-13
backfill/backfill.go
+14
-13
backfill/backfill.go
···
35
35
// Once done it clears the buffer and marks the job as "complete"
36
36
// Allowing the Job interface to abstract away the details of how buffered
37
37
// operations are stored and/or locked
38
-
FlushBufferedOps(ctx context.Context, cb func(kind, rev, path string, rec *[]byte, cid *cid.Cid) error) error
38
+
FlushBufferedOps(ctx context.Context, cb func(kind repomgr.EventKind, rev, path string, rec *[]byte, cid *cid.Cid) error) error
39
39
40
40
ClearBufferedOps(ctx context.Context) error
41
41
}
···
236
236
repo := job.Repo()
237
237
238
238
// Flush buffered operations, clear the buffer, and mark the job as "complete"
239
-
// Clearning and marking are handled by the job interface
240
-
err := job.FlushBufferedOps(ctx, func(kind, rev, path string, rec *[]byte, cid *cid.Cid) error {
241
-
switch repomgr.EventKind(kind) {
239
+
// Clearing and marking are handled by the job interface
240
+
err := job.FlushBufferedOps(ctx, func(kind repomgr.EventKind, rev, path string, rec *[]byte, cid *cid.Cid) error {
241
+
switch kind {
242
242
case repomgr.EvtKindCreateRecord:
243
243
err := b.HandleCreateRecord(ctx, repo, rev, path, rec, cid)
244
244
if err != nil {
···
467
467
468
468
var ops []*bufferedOp
469
469
for _, op := range evt.Ops {
470
-
switch op.Action {
471
-
case "create", "update":
470
+
kind := repomgr.EventKind(op.Action)
471
+
switch kind {
472
+
case repomgr.EvtKindCreateRecord, repomgr.EvtKindUpdateRecord:
472
473
cc, rec, err := bf.getRecord(ctx, r, op)
473
474
if err != nil {
474
475
return fmt.Errorf("getting record failed (%s,%s): %w", op.Action, op.Path, err)
475
476
}
476
477
477
478
ops = append(ops, &bufferedOp{
478
-
kind: op.Action,
479
+
kind: kind,
479
480
path: op.Path,
480
481
rec: rec,
481
482
cid: &cc,
482
483
})
483
-
case "delete":
484
+
case repomgr.EvtKindDeleteRecord:
484
485
ops = append(ops, &bufferedOp{
485
-
kind: op.Action,
486
+
kind: kind,
486
487
path: op.Path,
487
488
})
488
489
default:
···
511
512
512
513
for _, op := range ops {
513
514
switch op.kind {
514
-
case "create":
515
+
case repomgr.EvtKindCreateRecord:
515
516
if err := bf.HandleCreateRecord(ctx, evt.Repo, evt.Rev, op.path, op.rec, op.cid); err != nil {
516
517
return fmt.Errorf("create record failed: %w", err)
517
518
}
518
-
case "update":
519
+
case repomgr.EvtKindUpdateRecord:
519
520
if err := bf.HandleUpdateRecord(ctx, evt.Repo, evt.Rev, op.path, op.rec, op.cid); err != nil {
520
521
return fmt.Errorf("update record failed: %w", err)
521
522
}
522
-
case "delete":
523
+
case repomgr.EvtKindDeleteRecord:
523
524
if err := bf.HandleDeleteRecord(ctx, evt.Repo, evt.Rev, op.path); err != nil {
524
525
return fmt.Errorf("delete record failed: %w", err)
525
526
}
···
533
534
return nil
534
535
}
535
536
536
-
func (bf *Backfiller) BufferOp(ctx context.Context, repo string, since *string, rev, kind, path string, rec *[]byte, cid *cid.Cid) (bool, error) {
537
+
func (bf *Backfiller) BufferOp(ctx context.Context, repo string, since *string, rev string, kind repomgr.EventKind, path string, rec *[]byte, cid *cid.Cid) (bool, error) {
537
538
return bf.BufferOps(ctx, repo, since, rev, []*bufferedOp{{
538
539
path: path,
539
540
kind: kind,
+7
-7
backfill/backfill_test.go
+7
-7
backfill/backfill_test.go
···
63
63
t.Fatal(err)
64
64
}
65
65
if s.State() == backfill.StateInProgress {
66
-
bf.BufferOp(ctx, testRepos[0], "delete", "app.bsky.feed.follow/1", nil, &cid.Undef)
67
-
bf.BufferOp(ctx, testRepos[0], "delete", "app.bsky.feed.follow/2", nil, &cid.Undef)
68
-
bf.BufferOp(ctx, testRepos[0], "delete", "app.bsky.feed.follow/3", nil, &cid.Undef)
69
-
bf.BufferOp(ctx, testRepos[0], "delete", "app.bsky.feed.follow/4", nil, &cid.Undef)
70
-
bf.BufferOp(ctx, testRepos[0], "delete", "app.bsky.feed.follow/5", nil, &cid.Undef)
66
+
bf.BufferOp(ctx, testRepos[0], repomgr.EvtKindDeleteRecord, "app.bsky.feed.follow/1", nil, &cid.Undef)
67
+
bf.BufferOp(ctx, testRepos[0], repomgr.EvtKindDeleteRecord, "app.bsky.feed.follow/2", nil, &cid.Undef)
68
+
bf.BufferOp(ctx, testRepos[0], repomgr.EvtKindDeleteRecord, "app.bsky.feed.follow/3", nil, &cid.Undef)
69
+
bf.BufferOp(ctx, testRepos[0], repomgr.EvtKindDeleteRecord, "app.bsky.feed.follow/4", nil, &cid.Undef)
70
+
bf.BufferOp(ctx, testRepos[0], repomgr.EvtKindDeleteRecord, "app.bsky.feed.follow/5", nil, &cid.Undef)
71
71
72
-
bf.BufferOp(ctx, testRepos[0], "create", "app.bsky.feed.follow/1", nil, &cid.Undef)
72
+
bf.BufferOp(ctx, testRepos[0], repomgr.EvtKindCreateRecord, "app.bsky.feed.follow/1", nil, &cid.Undef)
73
73
74
-
bf.BufferOp(ctx, testRepos[0], "update", "app.bsky.feed.follow/1", nil, &cid.Undef)
74
+
bf.BufferOp(ctx, testRepos[0], repomgr.EvtKindUpdateRecord, "app.bsky.feed.follow/1", nil, &cid.Undef)
75
75
76
76
break
77
77
}
+2
-1
backfill/gormstore.go
+2
-1
backfill/gormstore.go
···
8
8
"sync"
9
9
"time"
10
10
11
+
"github.com/bluesky-social/indigo/repomgr"
11
12
"github.com/ipfs/go-cid"
12
13
"gorm.io/gorm"
13
14
)
···
328
329
return j.db.Save(j.dbj).Error
329
330
}
330
331
331
-
func (j *Gormjob) FlushBufferedOps(ctx context.Context, fn func(kind, rev, path string, rec *[]byte, cid *cid.Cid) error) error {
332
+
func (j *Gormjob) FlushBufferedOps(ctx context.Context, fn func(kind repomgr.EventKind, rev, path string, rec *[]byte, cid *cid.Cid) error) error {
332
333
// TODO: this will block any events for this repo while this flush is ongoing, is that okay?
333
334
j.lk.Lock()
334
335
defer j.lk.Unlock()
+4
-3
backfill/memstore.go
+4
-3
backfill/memstore.go
···
6
6
"sync"
7
7
"time"
8
8
9
+
"github.com/bluesky-social/indigo/repomgr"
9
10
"github.com/ipfs/go-cid"
10
11
)
11
12
12
13
type bufferedOp struct {
13
-
kind string
14
+
kind repomgr.EventKind
14
15
path string
15
16
rec *[]byte
16
17
cid *cid.Cid
···
81
82
return nil
82
83
}
83
84
84
-
func (s *Memstore) BufferOp(ctx context.Context, repo string, since *string, rev, kind, path string, rec *[]byte, cid *cid.Cid) (bool, error) {
85
+
func (s *Memstore) BufferOp(ctx context.Context, repo string, since *string, rev string, kind repomgr.EventKind, path string, rec *[]byte, cid *cid.Cid) (bool, error) {
85
86
s.lk.Lock()
86
87
87
88
// If the job doesn't exist, we can't buffer an op for it
···
199
200
return nil
200
201
}
201
202
202
-
func (j *Memjob) FlushBufferedOps(ctx context.Context, fn func(kind, rev, path string, rec *[]byte, cid *cid.Cid) error) error {
203
+
func (j *Memjob) FlushBufferedOps(ctx context.Context, fn func(kind repomgr.EventKind, rev, path string, rec *[]byte, cid *cid.Cid) error) error {
203
204
panic("TODO: copy what we end up doing from the gormstore")
204
205
/*
205
206
j.lk.Lock()