Something went wrong. Try again.
Read-only ActivityPub → atproto bridge for the threadiverse using Coves lexicons
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133// Package sync serves the com.atproto.sync.* XRPC surface — the endpoints a// relay or Jetstream needs to treat Tidepool as a valid subscribeRepos// upstream. The repo layer (internal/repo) owns all reads against repo// tables; this package turns those reads into wire frames and HTTP// responses, and fans out commit notifications to WebSocket subscribers.package sync
import ( "bytes" "fmt"
comatproto "github.com/bluesky-social/indigo/api/atproto" "github.com/bluesky-social/indigo/events" lexutil "github.com/bluesky-social/indigo/lex/util"
"github.com/ipfs/go-cid"
"tidepool/internal/repo")
// frameTimeFormat is the atproto datetime layout used for the #commit `time`// field: RFC 3339, UTC, millisecond precision (the spec's preferred form).const frameTimeFormat = "2006-01-02T15:04:05.000Z"
// commitFrame converts a stored firehose event into the subscribeRepos// #commit message, using indigo's lexicon types so the CBOR encoding is// exactly what relays and Jetstream expect.func commitFrame(ev *repo.Event) (*events.XRPCStreamEvent, error) { commitCID, err := cid.Parse(ev.CommitCID) if err != nil { return nil, fmt.Errorf("sync: parse commit cid %q for seq %d: %w", ev.CommitCID, ev.Seq, err) } var prevData *lexutil.LexLink if ev.PrevDataCID != nil { c, err := cid.Parse(*ev.PrevDataCID) if err != nil { return nil, fmt.Errorf("sync: parse prevData cid %q for seq %d: %w", *ev.PrevDataCID, ev.Seq, err) } prevData = (*lexutil.LexLink)(&c) }
ops := make([]*comatproto.SyncSubscribeRepos_RepoOp, 0, len(ev.Ops)) for _, op := range ev.Ops { frameOp := &comatproto.SyncSubscribeRepos_RepoOp{ Action: string(op.Action), Path: op.Path, } if op.CID != "" { c, err := cid.Parse(op.CID) if err != nil { return nil, fmt.Errorf("sync: parse op cid %q for seq %d: %w", op.CID, ev.Seq, err) } frameOp.Cid = (*lexutil.LexLink)(&c) } if op.Prev != "" { c, err := cid.Parse(op.Prev) if err != nil { return nil, fmt.Errorf("sync: parse op prev cid %q for seq %d: %w", op.Prev, ev.Seq, err) } frameOp.Prev = (*lexutil.LexLink)(&c) } ops = append(ops, frameOp) }
return &events.XRPCStreamEvent{ RepoCommit: &comatproto.SyncSubscribeRepos_Commit{ Seq: ev.Seq, Repo: ev.DID, Commit: lexutil.LexLink(commitCID), Rev: ev.Rev, Since: ev.SinceRev, PrevData: prevData, Blocks: ev.CAR, Ops: ops, Blobs: []lexutil.LexLink{}, Time: ev.CreatedAt.UTC().Format(frameTimeFormat), }, }, nil}
// accountFrame converts a stored account event (repo.EventKindAccount) into// the subscribeRepos #account message: the signal relays use to update// account state (bigsky tombstones the repo and purges its data on// status "deleted") instead of inferring it from scrub delete-commits.func accountFrame(ev *repo.Event) *events.XRPCStreamEvent { account := &comatproto.SyncSubscribeRepos_Account{ Seq: ev.Seq, Did: ev.DID, Active: ev.AccountActive, Time: ev.CreatedAt.UTC().Format(frameTimeFormat), } if !ev.AccountActive && ev.AccountStatus != "" { status := ev.AccountStatus account.Status = &status } return &events.XRPCStreamEvent{RepoAccount: account}}
// infoFrame builds a #info message (e.g. OutdatedCursor).func infoFrame(name, message string) *events.XRPCStreamEvent { info := &comatproto.SyncSubscribeRepos_Info{Name: name} if message != "" { info.Message = &message } return &events.XRPCStreamEvent{RepoInfo: info}}
// errorFrame builds a terminal error frame (e.g. FutureCursor); the spec// requires closing the stream after sending one.func errorFrame(name, message string) *events.XRPCStreamEvent { return &events.XRPCStreamEvent{Error: &events.ErrorFrame{Error: name, Message: message}}}
// serializeFrame renders header + body as the binary WebSocket payload.func serializeFrame(evt *events.XRPCStreamEvent) ([]byte, error) { var buf bytes.Buffer if err := evt.Serialize(&buf); err != nil { return nil, fmt.Errorf("sync: serialize stream frame: %w", err) } return buf.Bytes(), nil}
// identityFrame converts a stored identity event into a #identity frame.func identityFrame(ev *repo.Event) *events.XRPCStreamEvent { handle := ev.Handle return &events.XRPCStreamEvent{ RepoIdentity: &comatproto.SyncSubscribeRepos_Identity{ Did: ev.DID, Seq: ev.Seq, Time: ev.CreatedAt.UTC().Format(frameTimeFormat), Handle: &handle, }, }}