Something went wrong. Try again.
Monorepo for Tangled tangled.org
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331package knotfeed
import ( "bytes" "context" "errors" "fmt" "io" "log/slog" "strings" "unicode"
comatproto "github.com/bluesky-social/indigo/api/atproto" atprotorepo "github.com/bluesky-social/indigo/atproto/repo" "github.com/bluesky-social/indigo/atproto/syntax" "github.com/samber/lo" cbg "github.com/whyrusleeping/cbor-gen" "tangled.org/core/tapc")
const ( TypeCommit = "#commit" TypeSync = "#sync" TypeIdentity = "#identity" TypeAccount = "#account" TypeInfo = "#info" TypeError = "#error")
type Message struct { Type string Feed Feed Commit *Commit InfoName string Error string Detail string}
type RecordOp struct { Action tapc.RecordAction Collection syntax.NSID Rkey syntax.RecordKey Bytes []byte Prior PriorSha}
func (op RecordOp) Deleted() bool { return op.Action == tapc.RecordDeleteAction}
const ( sha1HexLen = 40 sha256HexLen = 64)
type ObjectID struct { hex string}
func ParseObjectID(reported string) (ObjectID, bool) { sized := len(reported) == sha1HexLen || len(reported) == sha256HexLen if !sized || strings.ContainsFunc(reported, notHexDigit) { return ObjectID{}, false } if strings.Trim(reported, "0") == "" { return ObjectID{}, true } return ObjectID{hex: reported}, true}
func (o ObjectID) String() string { return o.hex }
func (o ObjectID) IsZero() bool { return o.hex == "" }
func notHexDigit(r rune) bool { return !strings.ContainsRune(hexDigits, unicode.ToLower(r)) }
type PriorSha struct { sha ObjectID known bool}
func ParsePriorSha(reported string) PriorSha { sha, ok := ParseObjectID(reported) if !ok { return PriorSha{} } return PriorSha{sha: sha, known: true}}
func (p PriorSha) Sha() (ObjectID, bool) { return p.sha, p.known }
type Commit struct { Repo syntax.DID Seq int64 Rev string Records []RecordOp}
func Decode(data []byte, log *slog.Logger) (Message, error) { r := bytes.NewReader(data) head, err := readHeader(r) if err != nil { return Message{}, fmt.Errorf("frame header: %w", err) } if head.isError { return decodeError(r, head) } if head.msgType != "" { return decodeTyped(head.msgType, r, log) } return Message{}, fmt.Errorf("frame header names neither a type or an error")}
func decodeError(r io.Reader, head frameHeader) (Message, error) { msg := Message{Type: TypeError, Error: head.errName, Detail: head.errDetail} maj, count, err := cbg.CborReadHeader(r) if errors.Is(err, io.EOF) { return msg, nil } if err != nil { return Message{}, fmt.Errorf("error frame body: %w", err) } if maj != cbg.MajMap { return Message{}, fmt.Errorf("error frame body is major type %d, not a map", maj) } for range count { key, err := cbg.ReadString(r) if err != nil { return Message{}, fmt.Errorf("error frame body: %w", err) } switch key { case "error": if msg.Error, err = cbg.ReadString(r); err != nil { return Message{}, fmt.Errorf("error frame body: %w", err) } case "message": if msg.Detail, err = cbg.ReadString(r); err != nil { return Message{}, fmt.Errorf("error frame body: %w", err) } default: if err := skipValue(r, 0); err != nil { return Message{}, err } } } return msg, nil}
type frameHeader struct { msgType string isError bool errName string errDetail string}
func readHeader(r io.Reader) (frameHeader, error) { maj, count, err := cbg.CborReadHeader(r) if err != nil { return frameHeader{}, err } if maj != cbg.MajMap { return frameHeader{}, fmt.Errorf("expected a cbor map header, got major type %d", maj) } head := frameHeader{} for range count { key, err := cbg.ReadString(r) if err != nil { return frameHeader{}, err } switch key { case "t": if head.msgType, err = cbg.ReadString(r); err != nil { return frameHeader{}, err } case "op": maj, val, err := cbg.CborReadHeader(r) if err != nil { return frameHeader{}, err } head.isError = maj == cbg.MajNegativeInt && val == 0 case "error": if head.errName, err = cbg.ReadString(r); err != nil { return frameHeader{}, err } case "message": if head.errDetail, err = cbg.ReadString(r); err != nil { return frameHeader{}, err } default: if err := skipValue(r, 0); err != nil { return frameHeader{}, err } } } return head, nil}
func decodeTyped(msgType string, r io.Reader, log *slog.Logger) (Message, error) { switch msgType { case TypeCommit: var evt comatproto.SyncSubscribeRepos_Commit if err := evt.UnmarshalCBOR(r); err != nil { return Message{}, fmt.Errorf("commit frame: %w", err) } commit, err := resolveCommit(&evt, log) if err != nil { return Message{}, err } return Message{Type: TypeCommit, Commit: commit}, nil case TypeSync, TypeIdentity, TypeAccount: return Message{Type: msgType}, nil case TypeInfo: var evt comatproto.SyncSubscribeRepos_Info if err := evt.UnmarshalCBOR(r); err != nil { return Message{}, fmt.Errorf("info frame: %w", err) } return Message{Type: TypeInfo, InfoName: evt.Name}, nil default: if err := skipValue(r, 0); err != nil { return Message{}, err } return Message{Type: msgType}, nil }}
func resolveCommit(evt *comatproto.SyncSubscribeRepos_Commit, log *slog.Logger) (*Commit, error) { repoDid, err := syntax.ParseDID(evt.Repo) if err != nil { return nil, fmt.Errorf("commit frame names repo %q, which isn't a did: %w", evt.Repo, err) } commit := &Commit{ Repo: repoDid, Seq: evt.Seq, Rev: evt.Rev, } if len(evt.Ops) == 0 { return commit, nil } ctx := context.Background() _, repo, err := atprotorepo.LoadRepoFromCAR(ctx, bytes.NewReader(evt.Blocks)) if err != nil { log.Error("commit frame blocks didn't decode as a car, dropping the commit", "repo", repoDid, "seq", evt.Seq, "err", err) return commit, nil } commit.Records = lo.FilterMap(evt.Ops, func(op *comatproto.SyncSubscribeRepos_RepoOp, _ int) (RecordOp, bool) { if op == nil { return RecordOp{}, false } record, err := resolveOp(ctx, op, repo) if err != nil { log.Warn("dropping a commit op", "repo", repoDid, "path", op.Path, "err", err) return RecordOp{}, false } return record, true }) return commit, nil}
var recordActions = map[string]tapc.RecordAction{ string(tapc.RecordCreateAction): tapc.RecordCreateAction, string(tapc.RecordUpdateAction): tapc.RecordUpdateAction, string(tapc.RecordDeleteAction): tapc.RecordDeleteAction,}
func resolveOp(ctx context.Context, op *comatproto.SyncSubscribeRepos_RepoOp, repo *atprotorepo.Repo) (RecordOp, error) { action, known := recordActions[op.Action] if !known { return RecordOp{}, fmt.Errorf("op action %q is none of create, update or delete", op.Action) } rawCollection, rawRkey, found := strings.Cut(op.Path, "/") if !found { return RecordOp{}, errors.New("op path has no rkey") } collection, err := syntax.ParseNSID(rawCollection) if err != nil { return RecordOp{}, fmt.Errorf("op collection: %w", err) } rkey, err := syntax.ParseRecordKey(rawRkey) if err != nil { return RecordOp{}, fmt.Errorf("op rkey: %w", err) } record := RecordOp{Action: action, Collection: collection, Rkey: rkey} if record.Deleted() || repo == nil { return record, nil } payload, _, err := repo.GetRecordBytes(ctx, collection, rkey) if err != nil { return RecordOp{}, fmt.Errorf("record bytes missing from the frame car: %w", err) } record.Bytes = payload return record, nil}
const maxCborDepth = 64
func skipValue(r io.Reader, depth int) error { if depth > maxCborDepth { return fmt.Errorf("cbor value nests past %d levels", maxCborDepth) } maj, val, err := cbg.CborReadHeader(r) if err != nil { return err } switch maj { case cbg.MajByteString, cbg.MajTextString: _, err = io.CopyN(io.Discard, r, int64(val)) return err case cbg.MajArray: for range val { if err := skipValue(r, depth+1); err != nil { return err } } return nil case cbg.MajMap: for range val { if err := skipValue(r, depth+1); err != nil { return err } if err := skipValue(r, depth+1); err != nil { return err } } return nil case cbg.MajTag: return skipValue(r, depth+1) default: return nil }}