Something went wrong. Try again.
Monorepo for Tangled
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309package spindle
import ( "context" "database/sql" "encoding/json" "errors" "fmt"
"github.com/bluesky-social/indigo/atproto/syntax" "github.com/bluesky-social/jetstream/pkg/models" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/codes" "tangled.org/core/api/tangled" "tangled.org/core/spindle/db" "tangled.org/core/spindle/observability" "tangled.org/core/tapc")
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.EventKindAccount { return s.ingestAccount(ctx, e) } if e.Kind != models.EventKindCommit { return nil }
ctx, span := observability.Tracer().Start(ctx, "jetstream.ingest") defer span.End()
if span.IsRecording() { var attrs []attribute.KeyValue if e.Did != "" { attrs = append(attrs, attribute.String(observability.UserDIDKey, e.Did)) } if e.Commit != nil { if e.Commit.Collection != "" { attrs = append(attrs, attribute.String(observability.CollectionKey, e.Commit.Collection)) } if e.Commit.RKey != "" { attrs = append(attrs, attribute.String(observability.RKeyKey, e.Commit.RKey)) } } span.SetAttributes(attrs...) } var err error switch e.Commit.Collection { case tangled.SpindleMemberNSID: err = s.ingestMember(ctx, e) case tangled.RepoNSID, tangled.RepoCollaboratorNSID: 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) } case tangled.RepoPullStatusNSID: if evt, ok := jetstreamToTapEvent(e); ok { err = s.processPullStatus(ctx, evt.Record) } }
if err != nil { s.l.WarnContext(ctx, "failed to process message, skipping", "nsid", e.Commit.Collection, "did", e.Did, "rkey", e.Commit.RKey, "err", err) s.metrics.RecordEventIngestion("jetstream", "error") span.SetStatus(codes.Error, "failed to process jetstream event") } else { s.metrics.RecordEventIngestion("jetstream", "success") span.SetStatus(codes.Ok, "success") }
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}
func (s *Spindle) ingestMember(ctx context.Context, e *models.Event) error { did := e.Did rkey := e.Commit.RKey l := s.l.With("component", "ingester", "record", tangled.SpindleMemberNSID, "did", did, "rkey", rkey)
switch e.Commit.Operation { case models.CommitOperationCreate, models.CommitOperationUpdate: raw := e.Commit.Record record := tangled.SpindleMember{} if err := json.Unmarshal(raw, &record); err != nil { return fmt.Errorf("invalid record: %w", err) }
domain := s.cfg.Server.Hostname recordInstance := record.Instance
if recordInstance != domain { return fmt.Errorf("domain mismatch: %s != %s", record.Instance, domain) }
subject, err := syntax.ParseDID(record.Subject) if err != nil { return fmt.Errorf("invalid subject DID %q: %w", record.Subject, err) }
ok, err := s.e.IsSpindleInviteAllowed(did, rbacDomain) if err != nil { return fmt.Errorf("failed to enforce permissions: %w", err) } if !ok { return fmt.Errorf("permission denied for %s", did) }
sqlTx, err := s.db.BeginTx(ctx, nil) if err != nil { return fmt.Errorf("failed to start txn: %w", err) } committed := false defer func() { if !committed { sqlTx.Rollback() } }()
existing, err := db.GetSpindleMember(sqlTx, did, rkey) if err != nil && !errors.Is(err, sql.ErrNoRows) { return fmt.Errorf("failed to look up existing member: %w", err) }
var staleSubject string if existing != nil && existing.Subject != subject { staleSubject = existing.Subject.String() if err := db.RemoveSpindleMember(sqlTx, did, rkey); err != nil { return fmt.Errorf("failed to remove stale member row: %w", err) } }
if err := db.AddSpindleMember(sqlTx, db.SpindleMember{ Did: syntax.DID(did), Rkey: rkey, Instance: recordInstance, Subject: subject, }); err != nil { return fmt.Errorf("failed to add member: %w", err) }
if err := db.AddDid(sqlTx, subject.String()); err != nil { return fmt.Errorf("failed to add did: %w", err) }
dropStaleAcl := false var staleDidDropped bool if staleSubject != "" { remaining, err := db.CountSpindleMembersBySubject(sqlTx, staleSubject) if err != nil { return fmt.Errorf("failed to count stale subject rows: %w", err) } if remaining == 0 { dropStaleAcl = true stillNeeded, err := s.e.WouldHaveAnyPolicyExcludingSpindleMember(staleSubject, rbacDomain) if err != nil { return fmt.Errorf("failed to check residual policies for stale subject: %w", err) } if !stillNeeded { if err := db.RemoveDid(sqlTx, staleSubject); err != nil { return fmt.Errorf("failed to remove stale did: %w", err) } staleDidDropped = true } } l.Info("replaced stale spindle member", "old_subject", staleSubject, "new_subject", subject, "stale_did_dropped", staleDidDropped) }
if err := sqlTx.Commit(); err != nil { return fmt.Errorf("failed to commit txn: %w", err) } committed = true
if dropStaleAcl { if _, err := s.e.TryRemoveSpindleMember(rbacDomain, staleSubject); err != nil { l.Error("post-commit: failed to remove stale ACL", "subject", staleSubject, "err", err) } } if _, err := s.e.TryAddSpindleMember(rbacDomain, subject.String()); err != nil { l.Error("post-commit: failed to add member ACL", "subject", subject, "err", err) }
if staleDidDropped { s.jc.RemoveDid(staleSubject) } s.jc.AddDid(subject.String()) l.Info("added member from firehose", "member", subject) return nil
case models.CommitOperationDelete: sqlTx, err := s.db.BeginTx(ctx, nil) if err != nil { return fmt.Errorf("failed to start txn: %w", err) } committed := false defer func() { if !committed { sqlTx.Rollback() } }()
record, err := db.GetSpindleMember(sqlTx, did, rkey) if errors.Is(err, sql.ErrNoRows) { l.Info("spindle member already removed") return nil } if err != nil { return fmt.Errorf("failed to find member: %w", err) }
staleSubject := record.Subject.String()
if err := db.RemoveSpindleMember(sqlTx, did, rkey); err != nil { return fmt.Errorf("failed to remove member: %w", err) }
remaining, err := db.CountSpindleMembersBySubject(sqlTx, staleSubject) if err != nil { return fmt.Errorf("failed to count remaining member rows: %w", err) }
dropAcl := false var staleDidDropped bool if remaining == 0 { dropAcl = true stillNeeded, err := s.e.WouldHaveAnyPolicyExcludingSpindleMember(staleSubject, rbacDomain) if err != nil { return fmt.Errorf("failed to check residual policies: %w", err) } if !stillNeeded { if err := db.RemoveDid(sqlTx, staleSubject); err != nil { return fmt.Errorf("failed to remove did: %w", err) } staleDidDropped = true } }
if err := sqlTx.Commit(); err != nil { return fmt.Errorf("failed to commit txn: %w", err) } committed = true
if dropAcl { if _, err := s.e.TryRemoveSpindleMember(rbacDomain, staleSubject); err != nil { l.Error("post-commit: failed to remove member ACL", "subject", staleSubject, "err", err) } }
if staleDidDropped { s.jc.RemoveDid(staleSubject) } l.Info("removed member from firehose", "member", record.Subject, "remaining_rows", remaining, "stale_did_dropped", staleDidDropped) } return nil}
func (s *Spindle) ingestAccount(ctx context.Context, e *models.Event) error { if e.Account == nil || e.Account.Active { return nil } status := "unknown" if e.Account.Status != nil { status = *e.Account.Status } did, err := syntax.ParseDID(e.Account.Did) if err != nil { return fmt.Errorf("parsing account did: %w", err) } s.l.Warn("account went inactive, purging its repos", "did", did, "status", status) return s.WipeOwner(ctx, did, "account "+status)}