package 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
}
}