alice.pds.demo.boltless.dev
This repository has no description
1package spindle
2
3import (
4 "context"
5
6 "tangled.org/core/api/tangled"
7 "tangled.org/core/tapc"
8
9 "github.com/bluesky-social/indigo/atproto/syntax"
10 "github.com/bluesky-social/jetstream/pkg/models"
11)
12
13type Ingester func(ctx context.Context, e *models.Event) error
14
15func (s *Spindle) ingest() Ingester {
16 return func(ctx context.Context, e *models.Event) error {
17 if e.Kind != models.EventKindCommit {
18 return nil
19 }
20
21 var err error
22 switch e.Commit.Collection {
23 case tangled.RepoNSID:
24 if evt, ok := jetstreamToTapEvent(e); ok {
25 err = s.tap.processEvent(ctx, evt)
26 }
27 case tangled.RepoPullNSID:
28 if evt, ok := jetstreamToTapEvent(e); ok {
29 err = s.processPull(ctx, evt.Record)
30 }
31 }
32
33 if err != nil {
34 s.l.Warn("failed to process message, skipping", "nsid", e.Commit.Collection, "did", e.Did, "rkey", e.Commit.RKey, "err", err)
35 }
36
37 return nil
38 }
39}
40
41func jetstreamToTapEvent(e *models.Event) (tapc.Event, bool) {
42 if e.Commit == nil {
43 return tapc.Event{}, false
44 }
45 did, err := syntax.ParseDID(e.Did)
46 if err != nil {
47 return tapc.Event{}, false
48 }
49 var action tapc.RecordAction
50 switch e.Commit.Operation {
51 case models.CommitOperationCreate:
52 action = tapc.RecordCreateAction
53 case models.CommitOperationUpdate:
54 action = tapc.RecordUpdateAction
55 case models.CommitOperationDelete:
56 action = tapc.RecordDeleteAction
57 default:
58 return tapc.Event{}, false
59 }
60 return tapc.Event{
61 Type: tapc.EvtRecord,
62 Record: &tapc.RecordEventData{
63 Did: did,
64 Rkey: syntax.RecordKey(e.Commit.RKey),
65 Collection: syntax.NSID(e.Commit.Collection),
66 Action: action,
67 Record: e.Commit.Record,
68 // jetstream is only used for live
69 Live: true,
70 },
71 }, true
72}