Monorepo for Tangled
Something went wrong. Try again.
Go
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899190019011902190319041905190619071908190919101911191219131914191519161917191819191920192119221923192419251926192719281929193019311932193319341935193619371938193919401941194219431944194519461947194819491950195119521953195419551956195719581959196019611962196319641965196619671968196919701971197219731974197519761977197819791980198119821983198419851986198719881989199019911992199319941995199619971998199920002001200220032004200520062007200820092010201120122013201420152016201720182019202020212022202320242025202620272028202920302031203220332034203520362037203820392040204120422043204420452046204720482049205020512052205320542055205620572058205920602061206220632064206520662067206820692070207120722073207420752076207720782079208020812082208320842085208620872088208920902091209220932094209520962097209820992100210121022103210421052106210721082109211021112112211321142115211621172118211921202121212221232124212521262127212821292130213121322133213421352136213721382139214021412142214321442145214621472148214921502151215221532154215521562157215821592160216121622163216421652166216721682169217021712172217321742175217621772178217921802181218221832184218521862187218821892190219121922193package appview
import ( "context" "database/sql" "encoding/json" "errors" "fmt" "io" "log/slog" "maps" "net/http" "net/url" "slices" "strings" "sync"
"time"
"github.com/avast/retry-go/v4" "github.com/bluesky-social/indigo/atproto/syntax" jmodels "github.com/bluesky-social/jetstream/pkg/models" "github.com/go-git/go-git/v5/plumbing" "github.com/ipfs/go-cid" "golang.org/x/sync/errgroup" "tangled.org/core/api/tangled" "tangled.org/core/appview/cache" "tangled.org/core/appview/config" "tangled.org/core/appview/db" "tangled.org/core/appview/knotacl" "tangled.org/core/appview/mentions" "tangled.org/core/appview/models" "tangled.org/core/appview/notify" "tangled.org/core/appview/repoverify" "tangled.org/core/appview/serververify" "tangled.org/core/consts" "tangled.org/core/idresolver" "tangled.org/core/orm" "tangled.org/core/rbac")
type RepoPermissionChecker interface { HasRepoPermissionErr(ctx context.Context, repo *models.Repo, userDid, perm string) (bool, error)}
type Ingester struct { Ctx context.Context Db *db.DB Enforcer *rbac.Enforcer Acl RepoPermissionChecker IdResolver *idresolver.Resolver Cache *cache.Cache Config *config.Config Logger *slog.Logger MentionsResolver *mentions.Resolver Notifier notify.Notifier Verifier repoverify.Verifier}
type processFunc func(ctx context.Context, e *jmodels.Event) error
func (i *Ingester) Ingest() processFunc { return func(ctx context.Context, e *jmodels.Event) error { var err error
l := i.Logger.With("kind", e.Kind) switch e.Kind { case jmodels.EventKindAccount: // TODO: sync account state to db if e.Account.Active { break } // TODO: revoke sessions by DID if *e.Account.Status == "deactivated" { err = i.IdResolver.InvalidateIdent(ctx, e.Account.Did) } case jmodels.EventKindIdentity: err = i.IdResolver.InvalidateIdent(ctx, e.Identity.Did) case jmodels.EventKindCommit: l = l.With( "nsid", e.Commit.Collection, "did", e.Did, "rkey", e.Commit.RKey, "op", e.Commit.Operation, ) switch e.Commit.Collection { case tangled.GraphFollowNSID: err = i.ingestFollow(e, l) case tangled.GraphVouchNSID: err = i.ingestVouch(ctx, e, l) case tangled.FeedStarNSID: err = i.ingestStar(ctx, e, l) case tangled.FeedReactionNSID: err = i.ingestReaction(e, l) case tangled.PublicKeyNSID: err = i.ingestPublicKey(e, l) case tangled.RepoArtifactNSID: err = i.ingestArtifact(ctx, e, l) case tangled.ActorProfileNSID: err = i.ingestProfile(ctx, e, l) case tangled.SpindleMemberNSID: err = i.ingestSpindleMember(ctx, e, l) case tangled.SpindleNSID: err = i.ingestSpindle(ctx, e, l) case tangled.KnotMemberNSID: err = i.ingestKnotMember(ctx, e, l) case tangled.KnotNSID: err = i.ingestKnot(ctx, e, l) case tangled.StringNSID: err = i.ingestString(e, l) case tangled.RepoIssueNSID: err = i.ingestIssue(ctx, e, l) case tangled.RepoIssueStateNSID: err = i.ingestState(ctx, e, l, issueStateSpec) case tangled.RepoPullNSID: err = i.ingestPull(ctx, e, l) case tangled.RepoPullStatusNSID: err = i.ingestState(ctx, e, l, pullStatusSpec) case tangled.FeedCommentNSID: err = i.ingestComment(e, l) case tangled.RepoIssueCommentNSID: err = i.ingestIssueComment(e, l) case tangled.RepoPullCommentNSID: err = i.ingestPullComment(e, l) case tangled.LabelDefinitionNSID: err = i.ingestLabelDefinition(e, l) case tangled.LabelOpNSID: err = i.ingestLabelOp(ctx, e, l) case tangled.RepoNSID: err = i.ingestRepo(ctx, e, l) } }
if err != nil { l.Warn("failed to ingest record, skipping", "err", err) }
lastTimeUs := e.TimeUS + 1 if saveErr := i.Db.SaveLastTimeUs(lastTimeUs); saveErr != nil { l.Error("failed to save cursor", "err", saveErr) }
return nil }}
func (i *Ingester) resolveRepoRef(ref string) (*models.Repo, error) { if strings.HasPrefix(ref, "did:") { return db.GetRepoByDid(i.Db, ref) } return db.GetRepoByAtUri(i.Db, ref)}
func (i *Ingester) resolveOldFormatStar(raw json.RawMessage, star *models.Star, l *slog.Logger) (bool, error) { var legacy struct { Subject *string `json:"subject"` SubjectDid *string `json:"subjectDid"` } if err := json.Unmarshal(raw, &legacy); err != nil { return false, err }
switch { case legacy.SubjectDid != nil: repo, err := i.resolveRepoRef(*legacy.SubjectDid) if err != nil { l.Warn("skipping old-format star for unknown repo", "subjectDid", *legacy.SubjectDid) return false, nil } star.SubjectType = models.StarSubjectRepo star.Subject = repo.RepoDid return true, nil
case legacy.Subject != nil: uri, err := syntax.ParseATURI(*legacy.Subject) if err != nil { return false, fmt.Errorf("invalid old-format star subject: %w", err) } switch uri.Collection().String() { case tangled.RepoNSID: repo, err := db.GetRepoByAtUri(i.Db, uri.String()) if err != nil { l.Warn("skipping old-format star for unknown repo", "subject", *legacy.Subject) return false, nil } star.SubjectType = models.StarSubjectRepo star.Subject = repo.RepoDid return true, nil default: star.SubjectType = models.StarSubjectString star.Subject = *legacy.Subject return true, nil }
default: return false, fmt.Errorf("old-format star has neither subject nor subjectDid") }}
func (i *Ingester) ingestStar(ctx context.Context, e *jmodels.Event, l *slog.Logger) error { var err error did := e.Did
l = l.With("handler", "ingestStar")
switch e.Commit.Operation { case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate: raw := json.RawMessage(e.Commit.Record) record := tangled.FeedStar{} unmarshalErr := json.Unmarshal(raw, &record)
createdAt, parseErr := time.Parse(time.RFC3339, record.CreatedAt) if parseErr != nil { createdAt = time.Now() }
star := models.Star{ Did: did, Rkey: e.Commit.RKey, Created: createdAt, }
switch { case unmarshalErr != nil: resolved, resolveErr := i.resolveOldFormatStar(raw, &star, l) if resolveErr != nil { l.Error("invalid record", "newFmtErr", unmarshalErr, "oldFmtErr", resolveErr) return unmarshalErr } if !resolved { return nil }
case record.Subject == nil: return fmt.Errorf("star record has nil subject")
case record.Subject.FeedStar_Repo != nil: repo, repoErr := i.resolveRepoRef(record.Subject.FeedStar_Repo.Did) if repoErr != nil { l.Warn("skipping star for unknown repo", "did", record.Subject.FeedStar_Repo.Did) return nil } star.SubjectType = models.StarSubjectRepo star.Subject = repo.RepoDid
case record.Subject.FeedStar_String != nil: star.SubjectType = models.StarSubjectString star.Subject = record.Subject.FeedStar_String.Uri
default: return fmt.Errorf("star record has empty subject union") }
err = db.UpsertStar(i.Db, star) case jmodels.CommitOperationDelete: err = db.DeleteStarByRkey(i.Db, did, e.Commit.RKey) }
if err != nil { return fmt.Errorf("failed to %s star record: %w", e.Commit.Operation, err) } l.Info("processed star", "operation", e.Commit.Operation, "rkey", e.Commit.RKey)
l.Info("ingested record") return nil}
func (i *Ingester) ingestFollow(e *jmodels.Event, l *slog.Logger) error { var err error did := e.Did
l = l.With("handler", "ingestFollow")
switch e.Commit.Operation { case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate: raw := json.RawMessage(e.Commit.Record) record := tangled.GraphFollow{} err = json.Unmarshal(raw, &record) if err != nil { l.Error("invalid record", "err", err) return err }
err = db.UpsertFollow(i.Db, models.Follow{ UserDid: did, SubjectDid: record.Subject, Rkey: e.Commit.RKey, }) case jmodels.CommitOperationDelete: err = db.DeleteFollowByRkey(i.Db, did, e.Commit.RKey) }
if err != nil { return fmt.Errorf("failed to %s follow record: %w", e.Commit.Operation, err) } l.Info("processed follow", "operation", e.Commit.Operation, "rkey", e.Commit.RKey)
l.Info("ingested record") return nil}
func (i *Ingester) ingestVouch(ctx context.Context, e *jmodels.Event, l *slog.Logger) error { var err error did := e.Did
l = l.With("handler", "ingestVouch")
switch e.Commit.Operation { case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate: raw := json.RawMessage(e.Commit.Record) record := tangled.GraphVouch{} err = json.Unmarshal(raw, &record) if err != nil { l.Error("invalid record", "err", err) return err }
// rkey is the subject_did being vouched for/denounced subjectDID := e.Commit.RKey
_, err = syntax.ParseDID(subjectDID) if err != nil { l.Error("invalid subject_did in rkey", "err", err, "rkey", subjectDID) return fmt.Errorf("invalid subject_did: %w", err) }
if did == subjectDID { l.Warn("attempted self-vouch", "did", did) return fmt.Errorf("cannot vouch for self") }
subjectId, err := i.IdResolver.ResolveIdent(ctx, subjectDID) if err != nil { return err }
if subjectId.Handle.IsInvalidHandle() { return err }
kind, err := models.ParseVouchKind(record.Kind) if err != nil { l.Error("invalid kind", "kind", kind) return fmt.Errorf("invalid kind: %s", kind) }
recordCid, err := cid.Parse(e.Commit.CID) if err != nil { l.Error("invalid cid", "err", err, "cid", e.Commit.CID) return fmt.Errorf("invalid cid: %w", err) }
var evidences []syntax.ATURI for _, raw := range record.Evidences { uri, parseErr := syntax.ParseATURI(raw) if parseErr != nil { l.Warn("invalid evidence AT-URI, skipping", "uri", raw, "err", parseErr) continue } evidences = append(evidences, uri) }
tx, txErr := i.Db.Begin() if txErr != nil { return fmt.Errorf("failed to start transaction: %w", txErr) }
addErr := db.AddVouch(tx, &models.Vouch{ Did: syntax.DID(did), SubjectDid: subjectId.DID, Cid: recordCid, Kind: kind, Reason: record.Reason, Evidences: evidences, }) if addErr != nil { tx.Rollback() err = addErr } else { err = tx.Commit() }
case jmodels.CommitOperationDelete: err = db.DeleteVouchByRkey(i.Db, did, e.Commit.RKey) }
if err != nil { return fmt.Errorf("failed to %s vouch record: %w", e.Commit.Operation, err) }
l.Info("ingested record") return nil}
func (i *Ingester) ingestPublicKey(e *jmodels.Event, l *slog.Logger) error { did := e.Did var err error
l = l.With("handler", "ingestPublicKey")
switch e.Commit.Operation { case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate: l.Debug("processing add of pubkey") raw := json.RawMessage(e.Commit.Record) record := tangled.PublicKey{} err = json.Unmarshal(raw, &record) if err != nil { l.Error("invalid record", "err", err) return err } pubKey, err := models.PublicKeyFromRecord(syntax.DID(did), syntax.RecordKey(e.Commit.RKey), record) if err != nil { l.Error("invalid record", "err", err) return err } if err := pubKey.Validate(); err != nil { l.Error("invalid record", "err", err) return err }
err = db.UpsertPublicKey(i.Db, pubKey) case jmodels.CommitOperationDelete: l.Debug("processing delete of pubkey") err = db.DeletePublicKeyByRkey(i.Db, did, e.Commit.RKey) }
if err != nil { return fmt.Errorf("failed to %s pubkey record: %w", e.Commit.Operation, err) } l.Info("processed pubkey", "operation", e.Commit.Operation, "rkey", e.Commit.RKey)
l.Info("ingested record") return nil}
func (i *Ingester) ingestArtifact(ctx context.Context, e *jmodels.Event, l *slog.Logger) error { did := e.Did var err error
l = l.With("handler", "ingestArtifact")
switch e.Commit.Operation { case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate: raw := json.RawMessage(e.Commit.Record) record := tangled.RepoArtifact{} err = json.Unmarshal(raw, &record) if err != nil { l.Error("invalid record", "err", err) return err }
var repo *models.Repo if record.RepoDid != nil && *record.RepoDid != "" { repo, err = db.GetRepoByDid(i.Db, *record.RepoDid) if err != nil && !errors.Is(err, sql.ErrNoRows) { return fmt.Errorf("failed to look up repo by DID %s: %w", *record.RepoDid, err) } } if repo == nil && record.Repo != nil { repoAt, parseErr := syntax.ParseATURI(*record.Repo) if parseErr != nil { return parseErr } repo, err = db.GetRepoByAtUri(i.Db, repoAt.String()) if err != nil { return err } } if repo == nil { return fmt.Errorf("artifact record has neither valid repoDid nor repo field") }
allowed, permErr := i.Acl.HasRepoPermissionErr(ctx, repo, did, "repo:push") if permErr != nil { l.Warn("ingesting artifact without permission check", "did", did, "repo", repo.RepoIdentifier(), "err", permErr) } else if !allowed { l.Info("skipping unauthorized artifact", "did", did, "repo", repo.RepoIdentifier()) return nil }
repoDid := repo.RepoDid if repoDid == "" && record.RepoDid != nil { repoDid = *record.RepoDid } if repoDid != "" && (record.RepoDid == nil || *record.RepoDid == "") && record.Repo != nil { if enqErr := db.EnqueuePdsRecordMigration(ctx, i.Db, "add-repo-did", syntax.DID(did), syntax.NSID(tangled.RepoArtifactNSID), syntax.RecordKey(e.Commit.RKey)); enqErr != nil { l.Warn("failed to enqueue PDS rewrite for artifact", "err", enqErr, "did", did, "repoDid", repoDid) } }
createdAt, parseErr := time.Parse(time.RFC3339, record.CreatedAt) if parseErr != nil { createdAt = time.Now() }
artifact := models.Artifact{ Did: did, Rkey: e.Commit.RKey, RepoDid: syntax.DID(repo.RepoDid), Tag: plumbing.Hash(record.Tag), CreatedAt: createdAt, BlobCid: cid.Cid(record.Artifact.Ref), Name: record.Name, Size: uint64(record.Artifact.Size), MimeType: record.Artifact.MimeType, }
err = db.AddArtifact(i.Db, artifact) case jmodels.CommitOperationDelete: err = db.DeleteArtifact(i.Db, orm.FilterEq("did", did), orm.FilterEq("rkey", e.Commit.RKey)) }
if err != nil { return fmt.Errorf("failed to %s artifact record: %w", e.Commit.Operation, err) }
l.Info("ingested record") return nil}
func (i *Ingester) ingestProfile(ctx context.Context, e *jmodels.Event, l *slog.Logger) error { did := e.Did var err error
l = l.With("handler", "ingestProfile")
if e.Commit.RKey != "self" { return fmt.Errorf("ingestProfile only ingests `self` record") }
switch e.Commit.Operation { case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate: raw := json.RawMessage(e.Commit.Record) record := tangled.ActorProfile{} err = json.Unmarshal(raw, &record) if err != nil { l.Error("invalid record", "err", err) return err }
avatar := "" if record.Avatar != nil { avatar = record.Avatar.Ref.String() }
description := "" if record.Description != nil { description = *record.Description }
includeBluesky := record.Bluesky
pronouns := "" if record.Pronouns != nil { pronouns = *record.Pronouns }
location := "" if record.Location != nil { location = *record.Location }
var links [5]string for i, l := range record.Links { if i < 5 { links[i] = l } }
var stats [2]models.VanityStat for i, s := range record.Stats { if i < 2 { stats[i].Kind = models.ParseVanityStatKind(s) } }
var pinned [6]string for i, r := range record.PinnedRepositories { if i < 6 { pinned[i] = r } }
var preferredHandle syntax.Handle if record.PreferredHandle != nil { if h, err := syntax.ParseHandle(*record.PreferredHandle); err == nil { ident, identErr := i.IdResolver.ResolveIdent(ctx, did) if identErr == nil && slices.Contains(ident.AlsoKnownAs, "at://"+string(h)) { preferredHandle = h } } }
profile := models.Profile{ Did: did, Avatar: avatar, Description: description, IncludeBluesky: includeBluesky, Location: location, Links: links, Stats: stats, PinnedRepos: pinned, Pronouns: pronouns, PreferredHandle: preferredHandle, }
err = db.ValidateProfile(i.Db, &profile) if err != nil { return fmt.Errorf("invalid profile record") }
err = db.UpsertProfile(i.Db, &profile) if err != nil { return fmt.Errorf("upserting profile: %w", err) }
if i.Cache != nil { pipe := i.Cache.Pipeline() didKey := fmt.Sprintf(cache.PreferredHandleByDid, did) if preferredHandle != "" { pipe.Set(ctx, didKey, string(preferredHandle), cache.PreferredHandleTTL) pipe.Set(ctx, fmt.Sprintf(cache.PreferredHandleByHandle, string(preferredHandle)), did, cache.PreferredHandleTTL) } else { pipe.Del(ctx, didKey) } if _, execErr := pipe.Exec(ctx); execErr != nil { l.Warn("failed to update preferred handle cache", "err", execErr) } } case jmodels.CommitOperationDelete: tx, beginErr := i.Db.Begin() if beginErr != nil { return fmt.Errorf("failed to start transaction: %w", beginErr) }
priorHandle, phErr := db.GetPreferredHandle(tx, did) if phErr != nil && !errors.Is(phErr, sql.ErrNoRows) { l.Warn("failed to read prior preferred handle", "err", phErr) }
err = db.DeleteProfile(tx, did) if err == nil && i.Cache != nil { pipe := i.Cache.Pipeline() pipe.Del(ctx, fmt.Sprintf(cache.PreferredHandleByDid, did)) if priorHandle != "" { pipe.Del(ctx, fmt.Sprintf(cache.PreferredHandleByHandle, string(priorHandle))) } if _, execErr := pipe.Exec(ctx); execErr != nil { l.Warn("failed to evict preferred handle cache", "err", execErr) } } }
if err != nil { return fmt.Errorf("failed to %s profile record: %w", e.Commit.Operation, err) }
l.Info("ingested record") return nil}
func (i *Ingester) ingestSpindleMember(ctx context.Context, e *jmodels.Event, l *slog.Logger) error { did := e.Did var err error
l = l.With("handler", "ingestSpindleMember")
switch e.Commit.Operation { case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate: raw := json.RawMessage(e.Commit.Record) record := tangled.SpindleMember{} err = json.Unmarshal(raw, &record) if err != nil { l.Error("invalid record", "err", err) return err }
// only spindle owner can invite to spindles ok, err := i.Enforcer.IsSpindleInviteAllowed(did, record.Instance) if err != nil { return fmt.Errorf("failed to check invite permission: %w", err) } if !ok { if verifyErr := i.verifySpindle(ctx, record.Instance, did); verifyErr != nil { return fmt.Errorf("invite denied and verify failed: %w", verifyErr) } ok, err = i.Enforcer.IsSpindleInviteAllowed(did, record.Instance) if err != nil { return fmt.Errorf("failed to re-check invite permission: %w", err) } if !ok { return fmt.Errorf("invite denied for did %s on spindle %s", did, record.Instance) } }
memberId, err := i.IdResolver.ResolveIdent(ctx, record.Subject) if err != nil { return err }
if memberId.Handle.IsInvalidHandle() { return fmt.Errorf("invalid handle for member %s", record.Subject) }
existing, err := db.GetSpindleMembers(i.Db, orm.FilterEq("did", did), orm.FilterEq("rkey", e.Commit.RKey), ) if err != nil { return fmt.Errorf("failed to look up existing member: %w", err) } if len(existing) > 1 { return fmt.Errorf("multiple spindle members with rkey %s", e.Commit.RKey) }
tx, err := i.Db.Begin() if err != nil { return fmt.Errorf("failed to start txn: %w", err) } committed := false defer func() { if committed { return } tx.Rollback() i.Enforcer.E.LoadPolicy() }()
if len(existing) == 1 { prev := existing[0] if prev.Instance != record.Instance || prev.Subject != memberId.DID { if err = db.RemoveSpindleMember(tx, orm.FilterEq("did", did), orm.FilterEq("rkey", e.Commit.RKey), ); err != nil { return fmt.Errorf("failed to remove stale row: %w", err) } if err = i.Enforcer.RemoveSpindleMember(prev.Instance, prev.Subject.String()); err != nil { return fmt.Errorf("failed to remove stale ACL: %w", err) } } }
if err = db.AddSpindleMember(tx, models.SpindleMember{ Did: syntax.DID(did), Rkey: e.Commit.RKey, Instance: record.Instance, Subject: memberId.DID, }); err != nil { return fmt.Errorf("failed to add to db: %w", err) }
if err = i.Enforcer.AddSpindleMember(record.Instance, memberId.DID.String()); err != nil { return fmt.Errorf("failed to update ACLs: %w", err) }
if err = tx.Commit(); err != nil { return fmt.Errorf("failed to commit txn: %w", err) }
if err = i.Enforcer.E.SavePolicy(); err != nil { return fmt.Errorf("failed to save ACLs: %w", err) } committed = true
l.Info("upserted spindle member") case jmodels.CommitOperationDelete: rkey := e.Commit.RKey
// get record from db first members, err := db.GetSpindleMembers( i.Db, orm.FilterEq("did", did), orm.FilterEq("rkey", rkey), ) if err != nil || len(members) != 1 { return fmt.Errorf("failed to get member: %w, len(members) = %d", err, len(members)) } member := members[0]
tx, err := i.Db.Begin() if err != nil { return fmt.Errorf("failed to start txn: %w", err) } committed := false defer func() { if committed { return } tx.Rollback() i.Enforcer.E.LoadPolicy() }()
// remove record by rkey && update enforcer if err = db.RemoveSpindleMember( tx, orm.FilterEq("did", did), orm.FilterEq("rkey", rkey), ); err != nil { return fmt.Errorf("failed to remove from db: %w", err) }
// update enforcer err = i.Enforcer.RemoveSpindleMember(member.Instance, member.Subject.String()) if err != nil { return fmt.Errorf("failed to update ACLs: %w", err) }
if err = tx.Commit(); err != nil { return fmt.Errorf("failed to commit txn: %w", err) }
if err = i.Enforcer.E.SavePolicy(); err != nil { return fmt.Errorf("failed to save ACLs: %w", err) } committed = true
l.Info("removed spindle member") }
return nil}
func (i *Ingester) ingestSpindle(ctx context.Context, e *jmodels.Event, l *slog.Logger) error { did := e.Did var err error
l = l.With("handler", "ingestSpindle")
switch e.Commit.Operation { case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate: raw := json.RawMessage(e.Commit.Record) record := tangled.Spindle{} err = json.Unmarshal(raw, &record) if err != nil { l.Error("invalid record", "err", err) return err }
instance := e.Commit.RKey
err := db.AddSpindle(i.Db, models.Spindle{ Owner: syntax.DID(did), Instance: instance, }) if err != nil { l.Error("failed to add spindle to db", "err", err, "instance", instance) return err }
if err := i.verifySpindle(ctx, instance, did); err != nil { l.Warn("failed to verify spindle", "instance", instance, "did", did, "err", err) }
l.Info("ingested record", "instance", instance) return nil
case jmodels.CommitOperationDelete: instance := e.Commit.RKey
// get record from db first spindles, err := db.GetSpindles( ctx, i.Db, orm.FilterEq("owner", did), orm.FilterEq("instance", instance), ) if err != nil || len(spindles) != 1 { return fmt.Errorf("failed to get spindles: %w, len(spindles) = %d", err, len(spindles)) } spindle := spindles[0]
tx, err := i.Db.Begin() if err != nil { return fmt.Errorf("failed to start txn: %w", err) } defer func() { tx.Rollback() i.Enforcer.E.LoadPolicy() }()
// remove spindle members first err = db.RemoveSpindleMember( tx, orm.FilterEq("owner", did), orm.FilterEq("instance", instance), ) if err != nil { return fmt.Errorf("failed to remove spindle members: %w", err) }
err = db.DeleteSpindle( tx, orm.FilterEq("owner", did), orm.FilterEq("instance", instance), ) if err != nil { return fmt.Errorf("failed to delete spindle: %w", err) }
if spindle.Verified != nil { err = i.Enforcer.RemoveSpindle(instance) if err != nil { return fmt.Errorf("failed to remove spindle from enforcer: %w", err) } }
err = tx.Commit() if err != nil { return fmt.Errorf("failed to commit txn: %w", err) }
err = i.Enforcer.E.SavePolicy() if err != nil { return fmt.Errorf("failed to save ACLs: %w", err) }
l.Info("ingested record", "instance", instance) }
return nil}
func (i *Ingester) ingestString(e *jmodels.Event, l *slog.Logger) error { did := e.Did rkey := e.Commit.RKey
var err error
l = l.With("handler", "ingestString")
switch e.Commit.Operation { case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate: raw := json.RawMessage(e.Commit.Record) record := tangled.String{} err = json.Unmarshal(raw, &record) if err != nil { l.Error("invalid record", "err", err) return err }
string := models.StringFromRecord(did, rkey, record)
if err = string.Validate(); err != nil { l.Error("invalid record", "err", err) return err }
if err = db.AddString(i.Db, string); err != nil { l.Error("failed to add string", "err", err) return err }
l.Info("ingested record") return nil
case jmodels.CommitOperationDelete: if err := db.DeleteString( i.Db, orm.FilterEq("did", did), orm.FilterEq("rkey", rkey), ); err != nil { l.Error("failed to delete", "err", err) return fmt.Errorf("failed to delete string record: %w", err) }
l.Info("ingested record") return nil }
return nil}
func (i *Ingester) ingestKnotMember(ctx context.Context, e *jmodels.Event, l *slog.Logger) error { did := e.Did var err error
l = l.With("handler", "ingestKnotMember")
switch e.Commit.Operation { case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate: raw := json.RawMessage(e.Commit.Record) record := tangled.KnotMember{} err = json.Unmarshal(raw, &record) if err != nil { l.Error("invalid record", "err", err) return err }
// only knot owner can invite to knots ok, err := i.Enforcer.IsKnotInviteAllowed(did, record.Domain) if err != nil { return fmt.Errorf("failed to check invite permission: %w", err) } if !ok { if verifyErr := i.verifyKnot(ctx, record.Domain, did); verifyErr != nil { return fmt.Errorf("invite denied and verify failed: %w", verifyErr) } ok, err = i.Enforcer.IsKnotInviteAllowed(did, record.Domain) if err != nil { return fmt.Errorf("failed to re-check invite permission: %w", err) } if !ok { return fmt.Errorf("invite denied for did %s on knot %s", did, record.Domain) } }
memberId, err := i.IdResolver.ResolveIdent(ctx, record.Subject) if err != nil { return err }
if memberId.Handle.IsInvalidHandle() { return fmt.Errorf("invalid handle for member %s", record.Subject) }
existing, err := db.GetKnotMembers(i.Db, orm.FilterEq("did", did), orm.FilterEq("rkey", e.Commit.RKey), ) if err != nil { return fmt.Errorf("failed to look up existing member: %w", err) } if len(existing) > 1 { return fmt.Errorf("multiple knot members with rkey %s", e.Commit.RKey) }
tx, err := i.Db.Begin() if err != nil { return fmt.Errorf("failed to start txn: %w", err) } committed := false defer func() { if committed { return } tx.Rollback() i.Enforcer.E.LoadPolicy() }()
if len(existing) == 1 { prev := existing[0] if prev.Domain != record.Domain || prev.Subject != memberId.DID { if err = db.RemoveKnotMember(tx, orm.FilterEq("did", did), orm.FilterEq("rkey", e.Commit.RKey), ); err != nil { return fmt.Errorf("failed to remove stale row: %w", err) } if err = i.Enforcer.RemoveKnotMember(prev.Domain, prev.Subject.String()); err != nil { return fmt.Errorf("failed to remove stale ACL: %w", err) } } }
if err = db.AddKnotMember(tx, models.KnotMember{ Did: syntax.DID(did), Rkey: e.Commit.RKey, Domain: record.Domain, Subject: memberId.DID, }); err != nil { return fmt.Errorf("failed to add to db: %w", err) }
if err = i.Enforcer.AddKnotMember(record.Domain, memberId.DID.String()); err != nil { return fmt.Errorf("failed to update ACLs: %w", err) }
if err = tx.Commit(); err != nil { return fmt.Errorf("failed to commit txn: %w", err) }
if err = i.Enforcer.E.SavePolicy(); err != nil { return fmt.Errorf("failed to save ACLs: %w", err) } committed = true
l.Info("upserted knot member") case jmodels.CommitOperationDelete: rkey := e.Commit.RKey
members, err := db.GetKnotMembers( i.Db, orm.FilterEq("did", did), orm.FilterEq("rkey", rkey), ) if err != nil { return fmt.Errorf("failed to look up knot member with rkey %s: %w", rkey, err) } if len(members) == 0 { l.Info("knot member already removed", "rkey", rkey) return nil } if len(members) > 1 { return fmt.Errorf("multiple knot members with rkey %s", rkey) } member := members[0]
tx, err := i.Db.Begin() if err != nil { return fmt.Errorf("failed to start txn: %w", err) } committed := false defer func() { if committed { return } tx.Rollback() i.Enforcer.E.LoadPolicy() }()
if err = db.RemoveKnotMember( tx, orm.FilterEq("did", did), orm.FilterEq("rkey", rkey), ); err != nil { return fmt.Errorf("failed to remove from db: %w", err) }
if err = i.Enforcer.RemoveKnotMember(member.Domain, member.Subject.String()); err != nil { return fmt.Errorf("failed to update ACLs: %w", err) }
if err = tx.Commit(); err != nil { return fmt.Errorf("failed to commit txn: %w", err) }
if err = i.Enforcer.E.SavePolicy(); err != nil { return fmt.Errorf("failed to save ACLs: %w", err) } committed = true
l.Info("removed knot member") }
return nil}
func (i *Ingester) ingestKnot(ctx context.Context, e *jmodels.Event, l *slog.Logger) error { did := e.Did var err error
l = l.With("handler", "ingestKnot")
switch e.Commit.Operation { case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate: raw := json.RawMessage(e.Commit.Record) record := tangled.Knot{} err = json.Unmarshal(raw, &record) if err != nil { l.Error("invalid record", "err", err) return err }
domain := e.Commit.RKey
err := db.AddKnot(i.Db, domain, did) if err != nil { l.Error("failed to add knot to db", "err", err, "domain", domain) return err }
if err := i.verifyKnot(ctx, domain, did); err != nil { l.Warn("failed to verify knot", "domain", domain, "did", did, "err", err) }
l.Info("ingested record", "domain", domain) return nil
case jmodels.CommitOperationDelete: domain := e.Commit.RKey
// get record from db first registrations, err := db.GetRegistrations( i.Db, orm.FilterEq("domain", domain), orm.FilterEq("did", did), ) if err != nil { return fmt.Errorf("failed to get registration: %w", err) } if len(registrations) != 1 { return fmt.Errorf("got incorrect number of registrations: %d, expected 1", len(registrations)) } registration := registrations[0]
tx, err := i.Db.Begin() if err != nil { return fmt.Errorf("failed to start txn: %w", err) } defer func() { tx.Rollback() i.Enforcer.E.LoadPolicy() }()
err = db.RemoveKnotMember( tx, orm.FilterEq("did", did), orm.FilterEq("domain", domain), ) if err != nil { return fmt.Errorf("failed to remove knot members: %w", err) }
err = db.DeleteKnot( tx, orm.FilterEq("did", did), orm.FilterEq("domain", domain), ) if err != nil { return fmt.Errorf("failed to delete knot: %w", err) }
err = db.RemoveReposByKnot(tx, domain) if err != nil { return fmt.Errorf("failed to remove repos by knot: %w", err) }
if registration.Registered != nil { err = i.Enforcer.RemoveKnot(domain) if err != nil { return fmt.Errorf("failed to remove knot from enforcer: %w", err) } }
err = tx.Commit() if err != nil { return fmt.Errorf("failed to commit txn: %w", err) }
err = i.Enforcer.E.SavePolicy() if err != nil { return fmt.Errorf("failed to save ACLs: %w", err) }
l.Info("ingested record", "domain", domain) }
return nil}
const ( verifyAttempts = 4 verifyMinDelay = 1 * time.Second verifyMaxDelay = 5 * time.Second)
func (i *Ingester) verifyKnot(ctx context.Context, domain, did string) error { regs, err := db.GetRegistrations(i.Db, orm.FilterEq("domain", domain), orm.FilterEq("did", did), ) if err != nil { return fmt.Errorf("look up registration: %w", err) } if len(regs) != 1 { return fmt.Errorf("no registration for %s by %s", domain, did) } if regs[0].Registered != nil { return nil }
err = retry.Do( func() error { return serververify.RunVerification(ctx, domain, did, i.Config.Core.Dev) }, retry.Context(ctx), retry.Attempts(verifyAttempts), retry.Delay(verifyMinDelay), retry.MaxDelay(verifyMaxDelay), retry.DelayType(retry.BackOffDelay), retry.LastErrorOnly(true), ) if err != nil { return fmt.Errorf("verify: %w", err) } return serververify.MarkKnotVerified(i.Db, i.Enforcer, domain, did)}
func (i *Ingester) verifySpindle(ctx context.Context, instance, did string) error { spindles, err := db.GetSpindles(ctx, i.Db, orm.FilterEq("instance", instance), orm.FilterEq("owner", did), ) if err != nil { return fmt.Errorf("look up spindle: %w", err) } if len(spindles) != 1 { return fmt.Errorf("no spindle for %s by %s", instance, did) } if spindles[0].Verified != nil { return nil }
err = retry.Do( func() error { return serververify.RunVerification(ctx, instance, did, i.Config.Core.Dev) }, retry.Context(ctx), retry.Attempts(verifyAttempts), retry.Delay(verifyMinDelay), retry.MaxDelay(verifyMaxDelay), retry.DelayType(retry.BackOffDelay), retry.LastErrorOnly(true), ) if err != nil { return fmt.Errorf("verify: %w", err) } _, err = serververify.MarkSpindleVerified(i.Db, i.Enforcer, instance, did) return err}
const sweepConcurrency = 4
func (i *Ingester) SweepPendingVerifications() { l := i.Logger.With("handler", "SweepPendingVerifications")
var g errgroup.Group g.SetLimit(sweepConcurrency)
regs, err := db.GetRegistrations(i.Db, orm.FilterIs("registered", nil)) if err != nil { l.Error("failed to list unverified knots", "err", err) } else { for _, reg := range regs { g.Go(func() error { if err := i.verifyKnot(i.Ctx, reg.Domain, reg.ByDid); err != nil { l.Warn("verify knot failed", "domain", reg.Domain, "did", reg.ByDid, "err", err) } return nil }) } }
spindles, err := db.GetSpindles(i.Ctx, i.Db, orm.FilterIs("verified", nil)) if err != nil { l.Error("failed to list unverified spindles", "err", err) g.Wait() return } for _, s := range spindles { g.Go(func() error { if err := i.verifySpindle(i.Ctx, s.Instance, s.Owner.String()); err != nil { l.Warn("verify spindle failed", "instance", s.Instance, "owner", s.Owner, "err", err) } return nil }) } g.Wait()}
func (i *Ingester) ingestIssue(ctx context.Context, e *jmodels.Event, l *slog.Logger) error { did := e.Did rkey := e.Commit.RKey
var err error
l = l.With("handler", "ingestIssue")
switch e.Commit.Operation { case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate: raw := json.RawMessage(e.Commit.Record) record := tangled.RepoIssue{} err = json.Unmarshal(raw, &record) if err != nil { l.Error("invalid record", "err", err) return err }
issue := models.IssueFromRecord(did, rkey, record)
if issue.RepoDid == "" { return fmt.Errorf("issue record has no repo field") } if _, err := syntax.ParseDID(string(issue.RepoDid)); err != nil { return fmt.Errorf("issue record repo field is not a valid DID: %w", err) }
if err := issue.Validate(); err != nil { return fmt.Errorf("failed to validate issue: %w", err) }
if record.Repo != "" && !strings.HasPrefix(record.Repo, "did:") { repo, repoErr := db.GetRepoByAtUri(i.Db, record.Repo) if repoErr == nil && repo.RepoDid != "" { if enqErr := db.EnqueuePdsRecordMigration(ctx, i.Db, "add-repo-did", syntax.DID(did), syntax.NSID(tangled.RepoIssueNSID), syntax.RecordKey(e.Commit.RKey)); enqErr != nil { l.Warn("failed to enqueue PDS rewrite for issue", "err", enqErr, "did", did, "repoDid", repo.RepoDid) } } }
tx, err := i.Db.BeginTx(ctx, nil) if err != nil { l.Error("failed to begin transaction", "err", err) return err } defer tx.Rollback()
err = db.PutIssue(tx, &issue) if err != nil { l.Error("failed to create issue", "err", err) return err }
if err := db.ResolveIssueState(tx, issue.AtUri()); err != nil { l.Error("failed to resolve issue state", "err", err) return err }
err = tx.Commit() if err != nil { l.Error("failed to commit txn", "err", err) return err }
i.drainPendingState(ctx, issue.AtUri(), issueStateSpec, l)
l.Info("ingested record") return nil
case jmodels.CommitOperationDelete: tx, err := i.Db.BeginTx(ctx, nil) if err != nil { l.Error("failed to begin transaction", "err", err) return err } defer tx.Rollback()
if err := db.DeleteIssues( tx, did, rkey, ); err != nil { l.Error("failed to delete", "err", err) return fmt.Errorf("failed to delete issue record: %w", err) } if err := tx.Commit(); err != nil { l.Error("failed to commit txn", "err", err) return err }
l.Info("ingested record") return nil }
return nil}
func (i *Ingester) ingestPull(ctx context.Context, e *jmodels.Event, l *slog.Logger) error { did := e.Did rkey := e.Commit.RKey
var err error
l = l.With("handler", "ingestPull")
switch e.Commit.Operation { case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate: raw := json.RawMessage(e.Commit.Record) record := tangled.RepoPull{} err = json.Unmarshal(raw, &record) if err != nil { l.Error("invalid record", "err", err) return err }
ownerId, err := i.IdResolver.ResolveIdent(ctx, did) if err != nil { l.Error("failed to resolve did", "err", err) return err }
// go through and fetch all blobs in parallel readers := make([]*io.ReadCloser, len(record.Rounds)) var mu sync.Mutex
g, gctx := errgroup.WithContext(ctx)
for idx, b := range record.Rounds { g.Go(func() error { // for some reason, a blob is empty if b.PatchBlob == nil { return fmt.Errorf("missing patchBlob in round %d", idx) }
ownerPds := ownerId.PDSEndpoint() url, _ := url.Parse(fmt.Sprintf("%s/xrpc/com.atproto.sync.getBlob", ownerPds)) q := url.Query() q.Set("cid", b.PatchBlob.Ref.String()) q.Set("did", did) url.RawQuery = q.Encode()
req, err := http.NewRequestWithContext(gctx, http.MethodGet, url.String(), nil) if err != nil { l.Error("failed to create request") return err } req.Header.Set("Content-Type", "application/json")
resp, err := http.DefaultClient.Do(req) if err != nil { l.Error("failed to make request") return err }
mu.Lock() readers[idx] = &resp.Body mu.Unlock()
return nil }) }
if err := g.Wait(); err != nil { for _, r := range readers { if r != nil && *r != nil { (*r).Close() } } return err }
defer func() { for _, r := range readers { if r != nil && *r != nil { (*r).Close() } } }()
pull, err := models.PullFromRecord(did, rkey, record, readers) if err != nil { return fmt.Errorf("failed to parse pull from record: %w", err) } if err := pull.Validate(); err != nil { return fmt.Errorf("failed to validate pull: %w", err) } if pull.DependentOn != nil { if err := func() error { dependentPull, err := db.GetPull( i.Db, orm.FilterEq("dependent_on", pull.DependentOn.String()), ) if errors.Is(err, sql.ErrNoRows) { return nil } if err != nil { return fmt.Errorf("failed to fetch pulls with same dependency: %w", err) } if dependentPull.AtUri() == pull.AtUri() { return nil } return fmt.Errorf("another pull already depends on %s, which would form a DAG, this is presently disallowed", pull.DependentOn.String()) }(); err != nil { return fmt.Errorf("failed to validate pull stack: %w", err) } }
tx, err := i.Db.BeginTx(ctx, nil) if err != nil { l.Error("failed to begin transaction", "err", err) return err } defer tx.Rollback()
err = db.PutPull(tx, pull) if err != nil { l.Error("failed to create pull", "err", err) return err }
if err := db.ResolvePullStatus(tx, pull.AtUri()); err != nil { l.Error("failed to resolve pull status", "err", err) return err }
err = tx.Commit() if err != nil { l.Error("failed to commit txn", "err", err) return err }
i.drainPendingState(ctx, pull.AtUri(), pullStatusSpec, l)
l.Info("ingested record") return nil
case jmodels.CommitOperationDelete: tx, err := i.Db.BeginTx(ctx, nil) if err != nil { l.Error("failed to begin transaction", "err", err) return err } defer tx.Rollback()
if err := db.AbandonPulls( tx, orm.FilterEq("owner_did", did), orm.FilterEq("rkey", rkey), ); err != nil { l.Error("failed to abandon", "err", err) return fmt.Errorf("failed to abandon pull record: %w", err) } if err := tx.Commit(); err != nil { l.Error("failed to commit txn", "err", err) return err }
l.Info("ingested record") return nil }
return nil}
func (i *Ingester) authorizeStateRecord(ctx context.Context, repo *models.Repo, subjectAuthorDid, recordAuthorDid string, l *slog.Logger) (bool, error) { if recordAuthorDid == subjectAuthorDid { return true, nil } if recordAuthorDid == consts.TangledDid { return true, nil }
ok, err := i.Acl.HasRepoPermissionErr(ctx, repo, recordAuthorDid, "repo:push") if err != nil { if errors.Is(err, knotacl.ErrKnotUnreachable) { l.Warn("ingesting state record without permission check", "did", recordAuthorDid, "err", err) return true, nil } return false, err } return ok, nil}
type stateIngestSpec struct { subjectNSID string parse func(did, rkey string, raw json.RawMessage) (models.StateRecord, error) findSubject func(e db.Execer, subject syntax.ATURI) (repo *models.Repo, authorDid string, found bool, err error) put func(tx *sql.Tx, rec models.StateRecord) (syntax.ATURI, error) resolve func(tx *sql.Tx, subject syntax.ATURI) error recompute func(tx *sql.Tx, subject syntax.ATURI) error del func(tx *sql.Tx, did, rkey string) (syntax.ATURI, error)}
var issueStateSpec = stateIngestSpec{ subjectNSID: tangled.RepoIssueNSID, parse: func(did, rkey string, raw json.RawMessage) (models.StateRecord, error) { record := tangled.RepoIssueState{} if err := json.Unmarshal(raw, &record); err != nil { return models.StateRecord{}, fmt.Errorf("invalid record: %w", err) } return models.IssueStateFromRecord(did, rkey, record) }, findSubject: func(e db.Execer, subject syntax.ATURI) (*models.Repo, string, bool, error) { issues, err := db.GetIssues(e, orm.FilterEq("at_uri", subject)) if err != nil { return nil, "", false, err } if len(issues) != 1 || issues[0].Repo == nil { return nil, "", false, nil } return issues[0].Repo, issues[0].Did, true, nil }, put: db.PutIssueState, resolve: db.ResolveIssueState, recompute: db.RecomputeIssueState, del: db.DeleteIssueState,}
var pullStatusSpec = stateIngestSpec{ subjectNSID: tangled.RepoPullNSID, parse: func(did, rkey string, raw json.RawMessage) (models.StateRecord, error) { record := tangled.RepoPullStatus{} if err := json.Unmarshal(raw, &record); err != nil { return models.StateRecord{}, fmt.Errorf("invalid record: %w", err) } return models.PullStatusFromRecord(did, rkey, record) }, findSubject: func(e db.Execer, subject syntax.ATURI) (*models.Repo, string, bool, error) { pulls, err := db.GetPulls(e, orm.FilterEq("at_uri", subject)) if err != nil { return nil, "", false, err } if len(pulls) != 1 || pulls[0].Repo == nil { return nil, "", false, nil } return pulls[0].Repo, pulls[0].OwnerDid, true, nil }, put: db.PutPullStatus, resolve: db.ResolvePullStatus, recompute: db.RecomputePullStatus, del: db.DeletePullStatus,}
func (i *Ingester) ingestState(ctx context.Context, e *jmodels.Event, l *slog.Logger, spec stateIngestSpec) error { did := e.Did rkey := e.Commit.RKey nsid := e.Commit.Collection
l = l.With("handler", "ingestState", "nsid", nsid)
switch e.Commit.Operation { case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate: return i.applyStateRecord(ctx, did, rkey, nsid, e.Commit.Record, spec, l) case jmodels.CommitOperationDelete: return i.deleteStateRecord(ctx, did, rkey, nsid, spec, l) }
return nil}
func (i *Ingester) applyStateRecord(ctx context.Context, did, rkey, nsid string, raw []byte, spec stateIngestSpec, l *slog.Logger) error { rec, err := spec.parse(did, rkey, json.RawMessage(raw)) if err != nil { return err } if string(rec.Subject.Collection()) != spec.subjectNSID { return fmt.Errorf("state subject is not %s: %s", spec.subjectNSID, rec.Subject) }
repo, authorDid, found, err := spec.findSubject(i.Db, rec.Subject) if err != nil { return fmt.Errorf("failed to look up state subject: %w", err) } if !found { return i.parkStateRecord(ctx, did, rkey, nsid, rec.Subject, raw, l) }
authorized, err := i.authorizeStateRecord(ctx, repo, authorDid, did, l) if err != nil { return i.parkStateRecord(ctx, did, rkey, nsid, rec.Subject, raw, l) }
tx, err := i.Db.BeginTx(ctx, nil) if err != nil { return err } defer tx.Rollback()
if err := db.UnparkStateRecord(tx, did, rkey, nsid); err != nil { return fmt.Errorf("failed to unpark state record: %w", err) }
if !authorized { if err := tx.Commit(); err != nil { return err } l.Warn("dropped unauthorized state record", "did", did, "rkey", rkey, "subject", rec.Subject) return nil }
priorSubject, err := spec.put(tx, rec) if err != nil { return fmt.Errorf("failed to put state record: %w", err) } if err := spec.resolve(tx, rec.Subject); err != nil { return fmt.Errorf("failed to resolve state: %w", err) } if priorSubject != "" { if err := spec.recompute(tx, priorSubject); err != nil { return fmt.Errorf("failed to recompute prior subject state: %w", err) } }
if err := tx.Commit(); err != nil { return err }
l.Info("ingested record") return nil}
func (i *Ingester) deleteStateRecord(ctx context.Context, did, rkey, nsid string, spec stateIngestSpec, l *slog.Logger) error { tx, err := i.Db.BeginTx(ctx, nil) if err != nil { return err } defer tx.Rollback()
if err := db.UnparkStateRecord(tx, did, rkey, nsid); err != nil { return fmt.Errorf("failed to unpark state record: %w", err) } subject, err := spec.del(tx, did, rkey) if err != nil { return fmt.Errorf("failed to delete state record: %w", err) } if subject != "" { if err := spec.recompute(tx, subject); err != nil { return fmt.Errorf("failed to recompute state: %w", err) } }
if err := tx.Commit(); err != nil { return err }
l.Info("ingested record") return nil}
func (i *Ingester) parkStateRecord(ctx context.Context, did, rkey, nsid string, subject syntax.ATURI, raw []byte, l *slog.Logger) error { tx, err := i.Db.BeginTx(ctx, nil) if err != nil { return err } defer tx.Rollback()
if err := db.ParkStateRecord(tx, db.PendingStateRecord{ Did: did, Rkey: rkey, Nsid: nsid, Subject: subject, Record: raw, }); err != nil { return fmt.Errorf("failed to park state record: %w", err) }
if err := tx.Commit(); err != nil { return err }
l.Info("parked state record for retry", "subject", subject) return nil}
func (i *Ingester) drainPendingState(ctx context.Context, subject syntax.ATURI, spec stateIngestSpec, l *slog.Logger) { pending, err := db.PendingStateRecordsForSubject(i.Db, subject) if err != nil { l.Error("failed to load pending state records", "err", err, "subject", subject) return } for _, p := range pending { if err := i.applyStateRecord(ctx, p.Did, p.Rkey, p.Nsid, p.Record, spec, l); err != nil { l.Error("failed to drain pending state record", "err", err, "did", p.Did, "rkey", p.Rkey) } }}
const ( pendingStateReconcileInterval = time.Hour pendingStateRecordTTL = 7 * 24 * time.Hour)
func stateSpecForSubject(subject syntax.ATURI) (stateIngestSpec, bool) { switch string(subject.Collection()) { case tangled.RepoIssueNSID: return issueStateSpec, true case tangled.RepoPullNSID: return pullStatusSpec, true default: return stateIngestSpec{}, false }}
func (i *Ingester) StartPendingStateReconciler() { i.ReconcilePendingState()
ticker := time.NewTicker(pendingStateReconcileInterval) defer ticker.Stop() for { select { case <-i.Ctx.Done(): return case <-ticker.C: i.ReconcilePendingState() } }}
func (i *Ingester) ReconcilePendingState() { l := i.Logger.With("handler", "reconcilePendingState")
subjects, err := db.DistinctPendingStateSubjects(i.Db) if err != nil { l.Error("failed to list pending state subjects", "err", err) } for _, subject := range subjects { spec, ok := stateSpecForSubject(subject) if !ok { continue } i.drainPendingState(i.Ctx, subject, spec, l) }
cutoff := time.Now().Add(-pendingStateRecordTTL).UTC().Format(time.RFC3339) evicted, err := db.EvictStalePendingStateRecords(i.Db, cutoff) if err != nil { l.Error("failed to evict stale pending state records", "err", err) return } if evicted > 0 { l.Warn("evicted stale pending state records", "count", evicted, "olderThan", cutoff) }}
// ingestIssueComment ingests legacy sh.tangled.repo.issue.comment deletionsfunc (i *Ingester) ingestIssueComment(e *jmodels.Event, l *slog.Logger) error { l = l.With("handler", "ingestIssueComment")
switch e.Commit.Operation { case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate: // no-op. sh.tangled.repo.issue.comment is deprecated
case jmodels.CommitOperationDelete: if err := db.PurgeComments( i.Db, orm.FilterEq("did", e.Did), orm.FilterEq("collection", e.Commit.Collection), orm.FilterEq("rkey", e.Commit.RKey), ); err != nil { return fmt.Errorf("failed to delete comment record: %w", err) } }
l.Info("ingested record") return nil}
// ingestPullComment ingests legacy sh.tangled.repo.pull.comment deletionsfunc (i *Ingester) ingestPullComment(e *jmodels.Event, l *slog.Logger) error { l = l.With("handler", "ingestPullComment")
switch e.Commit.Operation { case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate: // no-op. sh.tangled.repo.pull.comment is deprecated
case jmodels.CommitOperationDelete: if err := db.PurgeComments( i.Db, orm.FilterEq("did", e.Did), orm.FilterEq("collection", e.Commit.Collection), orm.FilterEq("rkey", e.Commit.RKey), ); err != nil { return fmt.Errorf("failed to delete comment record: %w", err) } }
l.Info("ingested record") return nil}
func (i *Ingester) ingestComment(e *jmodels.Event, l *slog.Logger) error { did := e.Did rkey := e.Commit.RKey cid := e.Commit.CID
var err error
l = l.With("handler", "ingestComment")
ctx := context.Background()
switch e.Commit.Operation { case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate: raw := json.RawMessage(e.Commit.Record) record := tangled.FeedComment{} err = json.Unmarshal(raw, &record) if err != nil { return fmt.Errorf("invalid record: %w", err) }
comment, err := models.CommentFromRecord(syntax.DID(did), syntax.RecordKey(rkey), syntax.CID(cid), record) if err != nil { return fmt.Errorf("failed to parse comment from record: %w", err) }
if err := comment.Validate(); err != nil { return fmt.Errorf("failed to validate comment: %w", err) }
var references []syntax.ATURI if comment.Body.Original != nil { _, references = i.MentionsResolver.Resolve(ctx, *comment.Body.Original) }
tx, err := i.Db.Begin() if err != nil { return fmt.Errorf("failed to start transaction: %w", err) } defer tx.Rollback()
_, err = db.PutComment(tx, comment, references) if err != nil { return fmt.Errorf("failed to create comment: %w", err) }
if err := tx.Commit(); err != nil { return err }
case jmodels.CommitOperationDelete: if err := db.DeleteComments( i.Db, orm.FilterEq("did", did), orm.FilterEq("collection", e.Commit.Collection), orm.FilterEq("rkey", rkey), ); err != nil { return fmt.Errorf("failed to delete comment record: %w", err) } }
l.Info("ingested record") return nil}
func (i *Ingester) ingestReaction(e *jmodels.Event, l *slog.Logger) error { did := e.Did rkey := e.Commit.RKey
l = l.With("handler", "ingestReaction")
switch e.Commit.Operation { case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate: raw := json.RawMessage(e.Commit.Record) record := tangled.FeedReaction{} if err := json.Unmarshal(raw, &record); err != nil { return fmt.Errorf("invalid record: %w", err) }
subjectUri, err := syntax.ParseATURI(record.Subject) if err != nil { return fmt.Errorf("invalid reaction subject %q: %w", record.Subject, err) } subjectUri = models.NormalizeReactionSubject(subjectUri)
kind, ok := models.ParseReactionKind(record.Reaction) if !ok { return fmt.Errorf("invalid reaction kind: %q", record.Reaction) }
created, parseErr := time.Parse(time.RFC3339, record.CreatedAt) if parseErr != nil { created = time.Now() }
reaction := models.Reaction{ ReactedByDid: did, Rkey: rkey, ThreadAt: subjectUri, Kind: kind, Created: created, } if err := db.UpsertReaction(i.Db, reaction); err != nil { return fmt.Errorf("failed to upsert reaction: %w", err) }
case jmodels.CommitOperationDelete: if err := db.DeleteReactionByRkey(i.Db, did, rkey); err != nil { return fmt.Errorf("failed to delete reaction record: %w", err) } }
l.Info("ingested record") return nil}
func (i *Ingester) ingestLabelDefinition(e *jmodels.Event, l *slog.Logger) error { did := e.Did rkey := e.Commit.RKey
var err error
l = l.With("handler", "ingestLabelDefinition")
switch e.Commit.Operation { case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate: raw := json.RawMessage(e.Commit.Record) record := tangled.LabelDefinition{} err = json.Unmarshal(raw, &record) if err != nil { return fmt.Errorf("invalid record: %w", err) }
def, err := models.LabelDefinitionFromRecord(did, rkey, record) if err != nil { return fmt.Errorf("failed to parse labeldef from record: %w", err) }
if err := def.Validate(); err != nil { return fmt.Errorf("failed to validate labeldef: %w", err) }
_, err = db.AddLabelDefinition(i.Db, def) if err != nil { return fmt.Errorf("failed to create labeldef: %w", err) }
l.Info("ingested record") return nil
case jmodels.CommitOperationDelete: if err := db.DeleteLabelDefinition( i.Db, orm.FilterEq("did", did), orm.FilterEq("rkey", rkey), ); err != nil { return fmt.Errorf("failed to delete labeldef record: %w", err) }
l.Info("ingested record") return nil }
return nil}
func (i *Ingester) ingestLabelOp(ctx context.Context, e *jmodels.Event, l *slog.Logger) error { did := e.Did rkey := e.Commit.RKey
var err error
l = l.With("handler", "ingestLabelOp")
switch e.Commit.Operation { case jmodels.CommitOperationCreate: raw := json.RawMessage(e.Commit.Record) record := tangled.LabelOp{} err = json.Unmarshal(raw, &record) if err != nil { return fmt.Errorf("invalid record: %w", err) }
subject := syntax.ATURI(record.Subject) collection := subject.Collection()
var repo *models.Repo switch collection { case tangled.RepoIssueNSID: i, err := db.GetIssues(i.Db, orm.FilterEq("at_uri", subject)) if err != nil || len(i) != 1 { return fmt.Errorf("failed to find subject: %w || subject count %d", err, len(i)) } repo = i[0].Repo case tangled.RepoPullNSID: p, err := db.GetPulls(i.Db, orm.FilterEq("at_uri", subject)) if err != nil || len(p) != 1 { return fmt.Errorf("failed to find subject: %w || subject count %d", err, len(p)) } repo = p[0].Repo default: return fmt.Errorf("unsupported label subject: %s", collection) }
actx, err := db.NewLabelApplicationCtx(i.Db, orm.FilterIn("at_uri", repo.Labels)) if err != nil { return fmt.Errorf("failed to build label application ctx: %w", err) }
ops := models.LabelOpsFromRecord(did, rkey, record)
for _, o := range ops { def, ok := actx.Defs[o.OperandKey] if !ok { return fmt.Errorf("failed to find label def for key: %s, expected: %q", o.OperandKey, slices.Collect(maps.Keys(actx.Defs))) } // validate permissions: only collaborators can apply labels currently // // TODO: introduce a repo:triage permission allowed, permErr := i.Acl.HasRepoPermissionErr(ctx, repo, o.Did, "repo:push") if permErr != nil { if !errors.Is(permErr, knotacl.ErrKnotUnreachable) { return fmt.Errorf("enforcing permission: %w", permErr) } l.Warn("ingesting labelop without permission check", "did", o.Did, "err", permErr) } else if !allowed { return fmt.Errorf("unauthorized label operation") }
if err := def.ValidateOperandValue(&o); err != nil { return fmt.Errorf("failed to validate labelop: %w", err) } }
tx, err := i.Db.Begin() if err != nil { return err } defer tx.Rollback()
for _, o := range ops { _, err = db.AddLabelOp(tx, &o) if err != nil { return fmt.Errorf("failed to add labelop: %w", err) } }
if err = tx.Commit(); err != nil { return err }
l.Info("ingested record") }
return nil}