Something went wrong. Try again.
Monorepo for Tangled
Something went wrong. Try again.
Go
at sl/no-verify
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273package spindle
import ( "context"
"tangled.org/core/api/tangled" "tangled.org/core/tapc"
"github.com/bluesky-social/indigo/atproto/syntax" "github.com/bluesky-social/jetstream/pkg/models")
type Ingester func(ctx context.Context, e *models.Event) error
func (s *Spindle) ingest() Ingester { return func(ctx context.Context, e *models.Event) error { if e.Kind != models.EventKindCommit { return nil }
var err error switch e.Commit.Collection { case tangled.RepoNSID: if evt, ok := jetstreamToTapEvent(e); ok { err = s.tap.processEvent(ctx, evt) } case tangled.RepoPullNSID: if evt, ok := jetstreamToTapEvent(e); ok { err = s.processPull(ctx, evt.Record) } }
if err != nil { s.l.Warn("failed to process message, skipping", "nsid", e.Commit.Collection, "did", e.Did, "rkey", e.Commit.RKey, "err", err) }
return nil }}
func jetstreamToTapEvent(e *models.Event) (tapc.Event, bool) { if e.Commit == nil { return tapc.Event{}, false } did, err := syntax.ParseDID(e.Did) if err != nil { return tapc.Event{}, false } var action tapc.RecordAction switch e.Commit.Operation { case models.CommitOperationCreate: action = tapc.RecordCreateAction case models.CommitOperationUpdate: action = tapc.RecordUpdateAction case models.CommitOperationDelete: action = tapc.RecordDeleteAction default: return tapc.Event{}, false } return tapc.Event{ Type: tapc.EvtRecord, Record: &tapc.RecordEventData{ Did: did, Rkey: syntax.RecordKey(e.Commit.RKey), Collection: syntax.NSID(e.Commit.Collection), Action: action, Record: e.Commit.Record, // jetstream is only used for live Live: true, }, }, true}