Something went wrong. Try again.
Monorepo for Tangled
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560package state
import ( "context" "database/sql" "encoding/json" "errors" "fmt" "slices" "strings" "time"
"tangled.org/core/appview/cloudflare" "tangled.org/core/appview/notify"
"tangled.org/core/api/tangled" "tangled.org/core/appview/config" "tangled.org/core/appview/db" "tangled.org/core/appview/knotacl" "tangled.org/core/appview/knotcompat" "tangled.org/core/appview/models" "tangled.org/core/appview/sites" "tangled.org/core/consts" ec "tangled.org/core/eventconsumer" "tangled.org/core/eventstream" knotdb "tangled.org/core/knotserver/db" "tangled.org/core/log" "tangled.org/core/orm" "tangled.org/core/rbac" "tangled.org/core/workflow"
"github.com/bluesky-social/indigo/atproto/syntax" "github.com/go-git/go-git/v5/plumbing" "github.com/posthog/posthog-go")
type aclRoster interface { AddKnotMember(host string, subject syntax.DID, cursor knotacl.Cursor) error RemoveKnotMember(host string, subject syntax.DID, cursor knotacl.Cursor) error AddCollaborator(repoDid, subject syntax.DID, cursor knotacl.Cursor) error RemoveCollaborator(repoDid, subject syntax.DID, cursor knotacl.Cursor) error InvalidateMembers(host string) InvalidateCollaborators(host, repoDid string)}
func Knotstream(ctx context.Context, c *config.Config, d *db.DB, acl *knotacl.Service, enforcer *rbac.Enforcer, posthog posthog.Client, notifier notify.Notifier, cfClient *cloudflare.Client) (*ec.Consumer, error) { knots, err := db.GetRegistrations(d, orm.FilterIsNot("registered", "null")) if err != nil { return nil, err }
hosts := make([]string, len(knots)) for i, k := range knots { hosts[i] = k.Domain }
return bootstrapStream( ctx, "knotstream", ec.KindKnot, hosts, c.Redis.Addr, c.Knotstream, knotIngester(d, acl, enforcer, posthog, notifier, c.Core.Dev, c, cfClient), ), nil}
func resolveRepo(d *db.DB, repoDid *string, ownerDid, repoName string) (*models.Repo, error) { if repoDid != nil && *repoDid != "" { return db.GetRepoByDid(d, *repoDid) } repos, err := db.GetRepos(d, orm.FilterEq("did", ownerDid), orm.FilterEq("rkey", strings.ToLower(repoName))) if err != nil { return nil, err } if len(repos) == 0 { return nil, sql.ErrNoRows } return &repos[0], nil}
func knotIngester(d *db.DB, acl aclRoster, enforcer *rbac.Enforcer, posthog posthog.Client, notifier notify.Notifier, dev bool, c *config.Config, cfClient *cloudflare.Client) ec.ProcessFunc { return func(ctx context.Context, source ec.Source, msg eventstream.Event) error { switch msg.Nsid { case tangled.GitRefUpdateNSID: return ingestRefUpdate(ctx, d, enforcer, posthog, notifier, dev, c, cfClient, source, msg) case tangled.PipelineNSID: return ingestPipeline(d, source, msg) case knotdb.RepoDIDAssignNSID: return ingestDIDAssign(d, enforcer, source, msg, ctx) case knotdb.KnotMemberUpdateNSID: return ingestKnotMemberUpdate(acl, source, msg) case knotdb.RepoCollaboratorUpdateNSID: return ingestCollaboratorUpdate(ctx, d, acl, source, msg) }
return nil }}
const ( aclIngestAttempts = 3 aclIngestBackoff = 50 * time.Millisecond)
func withAclRetry(attempts int, backoff time.Duration, op func() error) error { if err := op(); err == nil || attempts <= 1 { return err } time.Sleep(backoff) return withAclRetry(attempts-1, backoff, op)}
func ingestKnotMemberUpdate(acl aclRoster, source ec.Source, msg eventstream.Event) error { var rec knotdb.KnotMemberUpdate if err := json.Unmarshal(msg.EventJson, &rec); err != nil { return fmt.Errorf("unmarshal memberUpdate: %w", err) }
subject, err := syntax.ParseDID(rec.Subject) if err != nil { return fmt.Errorf("memberUpdate bad subject %q: %w", rec.Subject, err) }
cursor := knotacl.Cursor(msg.Created) switch rec.Op { case knotdb.AclOpAdd: err = withAclRetry(aclIngestAttempts, aclIngestBackoff, func() error { return acl.AddKnotMember(source.Host, subject, cursor) }) case knotdb.AclOpRemove: err = withAclRetry(aclIngestAttempts, aclIngestBackoff, func() error { return acl.RemoveKnotMember(source.Host, subject, cursor) }) default: return fmt.Errorf("memberUpdate unknown op %q", rec.Op) }
if err != nil { acl.InvalidateMembers(source.Host) } return err}
func ingestCollaboratorUpdate(ctx context.Context, d *db.DB, acl aclRoster, source ec.Source, msg eventstream.Event) error { var rec knotdb.RepoCollaboratorUpdate if err := json.Unmarshal(msg.EventJson, &rec); err != nil { return fmt.Errorf("unmarshal collaboratorUpdate: %w", err) }
subject, err := syntax.ParseDID(rec.Subject) if err != nil { return fmt.Errorf("collaboratorUpdate bad subject %q: %w", rec.Subject, err) } repoDid, err := syntax.ParseDID(rec.Repo) if err != nil { return fmt.Errorf("collaboratorUpdate bad repo %q: %w", rec.Repo, err) }
cursor := knotacl.Cursor(msg.Created) switch rec.Op { case knotdb.AclOpAdd: err = withAclRetry(aclIngestAttempts, aclIngestBackoff, func() error { owned, err := repoOwnedBySource(ctx, d, source, repoDid, subject) if err != nil || !owned { return err } return acl.AddCollaborator(repoDid, subject, cursor) }) case knotdb.AclOpRemove: err = withAclRetry(aclIngestAttempts, aclIngestBackoff, func() error { owned, err := repoOwnedBySource(ctx, d, source, repoDid, subject) if err != nil || !owned { return err } return acl.RemoveCollaborator(repoDid, subject, cursor) }) default: return fmt.Errorf("collaboratorUpdate unknown op %q", rec.Op) }
if err != nil { acl.InvalidateCollaborators(source.Host, repoDid.String()) } return err}
func repoOwnedBySource(ctx context.Context, d *db.DB, source ec.Source, repoDid, subject syntax.DID) (bool, error) { repo, err := db.GetRepoByDid(d, repoDid.String()) if errors.Is(err, sql.ErrNoRows) { log.FromContext(ctx).Warn("collaboratorUpdate for unindexed repo, skipping until reconcile", "repo_did", repoDid, "subject", subject) return false, nil } if err != nil { return false, err } if repo.Knot != source.Host { log.FromContext(ctx).Warn("collaboratorUpdate for a repo this knot does not host, dropping", "repo_did", repoDid, "subject", subject, "claimed_by", source.Host, "owner", repo.Knot) return false, nil } return true, nil}
// TODO(boltless): remove this. knotmirror should do all sort of indexingfunc ingestRefUpdate(ctx context.Context, d *db.DB, enforcer *rbac.Enforcer, pc posthog.Client, notifier notify.Notifier, dev bool, c *config.Config, cfClient *cloudflare.Client, source ec.Source, msg eventstream.Event) error { logger := log.FromContext(ctx)
var record tangled.GitRefUpdate err := json.Unmarshal(msg.EventJson, &record) if err != nil { return err }
if !knotcompat.KnotHasCapability(ctx, source.Host, dev, consts.CapKnotACL) { knownKnots, err := enforcer.GetKnotsForUser(record.CommitterDid) switch { case err != nil: logger.Warn("gitRefUpdate membership lookup failed, ingesting without the sanity check", "committer", record.CommitterDid, "knot", source.Host, "err", err) case !slices.Contains(knownKnots, source.Host): logger.Warn("gitRefUpdate committer is not a known member of the knot, ingesting anyway", "committer", record.CommitterDid, "knot", source.Host) } }
if record.Repo == "" { return fmt.Errorf("gitRefUpdate from %s missing repo", source.Host) }
repo, lookupErr := db.GetRepoByDid(d, record.Repo) if lookupErr != nil { return fmt.Errorf("failed to look up repo: %w", lookupErr) }
logger.Info("processing gitRefUpdate event", "repo", repo.RepoIdentifier(), "ref", record.Ref, "old_sha", record.OldSha, "new_sha", record.NewSha)
notifier.Push(ctx, repo, record.Ref, record.OldSha, record.NewSha, record.CommitterDid)
errPunchcard := populatePunchcard(d, record) errLanguages := updateRepoLanguages(d, record)
var errPosthog error if !dev && record.CommitterDid != "" { errPosthog = pc.Enqueue(posthog.Capture{ DistinctId: record.CommitterDid, Event: "git_ref_update", }) }
// Trigger a sites redeploy if this push is to the configured sites branch. if cfClient.Enabled() { go triggerSitesDeployIfNeeded(ctx, d, cfClient, c, record, source) }
return errors.Join(errPunchcard, errLanguages, errPosthog)}
// triggerSitesDeployIfNeeded checks whether the pushed ref matches the sites// branch configured for this repo and, if so, syncs the site to R2func triggerSitesDeployIfNeeded(ctx context.Context, d *db.DB, cfClient *cloudflare.Client, cfg *config.Config, record tangled.GitRefUpdate, source ec.Source) { logger := log.FromContext(ctx)
ref := plumbing.ReferenceName(record.Ref) if !ref.IsBranch() { return } pushedBranch := ref.Short()
repo, err := db.GetRepoByDid(d, record.Repo) if err != nil { return }
siteConfig, err := db.GetRepoSiteConfig(d, repo.RepoDid) if err != nil || siteConfig == nil { return } if siteConfig.Branch != pushedBranch { return }
deploy := &models.SiteDeploy{ RepoDid: syntax.DID(repo.RepoDid), Branch: siteConfig.Branch, Dir: siteConfig.Dir, CommitSHA: record.NewSha, Trigger: models.SiteDeployTriggerPush, }
deployErr := sites.Deploy(ctx, cfClient, cfg, repo, siteConfig.Branch, siteConfig.Dir) if deployErr != nil { logger.Error("sites: R2 sync failed on push", "repo", repo.RepoIdentifier(), "err", deployErr) deploy.Status = models.SiteDeployStatusFailure deploy.Error = deployErr.Error() } else { deploy.Status = models.SiteDeployStatusSuccess }
if err := db.AddSiteDeploy(d, deploy); err != nil { logger.Error("sites: failed to record deploy", "repo", repo.RepoIdentifier(), "err", err) }
if deployErr == nil { logger.Info("site deployed to r2", "repo", repo.RepoIdentifier()) }}
func populatePunchcard(d *db.DB, record tangled.GitRefUpdate) error { if record.CommitterDid == "" { return nil }
knownEmails, err := db.GetAllEmails(d, record.CommitterDid) if err != nil { return err }
count := 0 for _, ke := range knownEmails { if record.Meta == nil { continue } if record.Meta.CommitCount == nil { continue } for _, ce := range record.Meta.CommitCount.ByEmail { if ce == nil { continue } if ce.Email == ke.Address || ce.Email == record.CommitterDid { count += int(ce.Count) } } }
punch := models.Punch{ Did: record.CommitterDid, Date: time.Now(), Count: count, } return db.AddPunch(d, punch)}
func updateRepoLanguages(d *db.DB, record tangled.GitRefUpdate) error { if record.Meta == nil || record.Meta.LangBreakdown == nil || record.Meta.LangBreakdown.Inputs == nil { return fmt.Errorf("empty language data for repo: %s", record.Repo) }
r, lookupErr := db.GetRepoByDid(d, record.Repo) if lookupErr != nil { return fmt.Errorf("failed to look up repo: %w", lookupErr) } repo := *r
ref := plumbing.ReferenceName(record.Ref) if !ref.IsBranch() { return fmt.Errorf("%s is not a valid reference name", ref) }
var langs []models.RepoLanguage for _, l := range record.Meta.LangBreakdown.Inputs { if l == nil { continue }
langs = append(langs, models.RepoLanguage{ RepoDid: syntax.DID(repo.RepoDid), Ref: ref.Short(), IsDefaultRef: record.Meta.IsDefaultRef, Language: l.Lang, Bytes: l.Size, }) }
tx, err := d.Begin() if err != nil { return err } defer tx.Rollback()
// update appview's cache err = db.UpdateRepoLanguages(tx, syntax.DID(repo.RepoDid), ref.Short(), langs) if err != nil { fmt.Printf("failed; %s\n", err) // non-fatal }
return tx.Commit()}
func ingestPipeline(d *db.DB, source ec.Source, msg eventstream.Event) error { var record tangled.Pipeline err := json.Unmarshal(msg.EventJson, &record) if err != nil { return err }
if record.TriggerMetadata == nil { return fmt.Errorf("empty trigger metadata: nsid %s, rkey %s", msg.Nsid, msg.Rkey) }
if record.TriggerMetadata.Repo == nil { return fmt.Errorf("empty repo: nsid %s, rkey %s", msg.Nsid, msg.Rkey) }
repoName := "" if record.TriggerMetadata.Repo.Repo != nil { repoName = *record.TriggerMetadata.Repo.Repo }
repo, lookupErr := resolveRepo(d, record.TriggerMetadata.Repo.RepoDid, record.TriggerMetadata.Repo.Did, repoName) if lookupErr != nil { return fmt.Errorf("failed to look up repo: %w", lookupErr) } if repo.Spindle == "" { return fmt.Errorf("repo does not have a spindle configured yet: nsid %s, rkey %s", msg.Nsid, msg.Rkey) }
// trigger info var trigger models.Trigger var sha string trigger.Kind = workflow.TriggerKind(record.TriggerMetadata.Kind) switch trigger.Kind { case workflow.TriggerKindPush: trigger.PushRef = &record.TriggerMetadata.Push.Ref trigger.PushNewSha = &record.TriggerMetadata.Push.NewSha trigger.PushOldSha = &record.TriggerMetadata.Push.OldSha sha = *trigger.PushNewSha case workflow.TriggerKindPullRequest: trigger.PRSourceBranch = &record.TriggerMetadata.PullRequest.SourceBranch trigger.PRTargetBranch = &record.TriggerMetadata.PullRequest.TargetBranch trigger.PRSourceSha = &record.TriggerMetadata.PullRequest.SourceSha trigger.PRAction = &record.TriggerMetadata.PullRequest.Action sha = *trigger.PRSourceSha }
tx, err := d.Begin() if err != nil { return fmt.Errorf("failed to start txn: %w", err) }
triggerId, err := db.AddTrigger(tx, trigger) if err != nil { return fmt.Errorf("failed to add trigger entry: %w", err) }
pipeline := models.Pipeline{ Rkey: msg.Rkey, Knot: source.Host, RepoOwner: syntax.DID(record.TriggerMetadata.Repo.Did), RepoName: repoName, RepoDid: repo.RepoDid, TriggerId: int(triggerId), Sha: sha, }
err = db.AddPipeline(tx, pipeline) if err != nil { return fmt.Errorf("failed to add pipeline: %w", err) }
err = tx.Commit() if err != nil { return fmt.Errorf("failed to commit txn: %w", err) }
return nil}
func ingestDIDAssign(d *db.DB, enforcer *rbac.Enforcer, source ec.Source, msg eventstream.Event, ctx context.Context) error { logger := log.FromContext(ctx)
var record knotdb.RepoDIDAssign if err := json.Unmarshal(msg.EventJson, &record); err != nil { return fmt.Errorf("unmarshal didAssign: %w", err) }
if record.RepoDid == "" || record.OwnerDid == "" || record.RepoName == "" { return fmt.Errorf("didAssign missing required fields: repoDid=%q ownerDid=%q repoName=%q", record.RepoDid, record.OwnerDid, record.RepoName) }
logger.Info("processing didAssign event", "repo_did", record.RepoDid, "owner_did", record.OwnerDid, "repo_name", record.RepoName)
repos, err := db.GetRepos(d, orm.FilterEq("did", record.OwnerDid), orm.FilterEq("name", record.RepoName), ) if err != nil || len(repos) == 0 { logger.Warn("didAssign for unknown repo, skipping", "owner_did", record.OwnerDid, "repo_name", record.RepoName) return nil } repo := repos[0] knot := source.Host
if repo.Knot != knot { return fmt.Errorf("didAssign from %s for repo hosted on %s, rejecting", knot, repo.Knot) }
repoAtUri := repo.RepoAt().String() legacyResource := record.OwnerDid + "/" + record.RepoName
if repo.RepoDid != record.RepoDid { tx, err := d.Begin() if err != nil { return fmt.Errorf("begin didAssign txn: %w", err) } defer tx.Rollback()
if err := db.CascadeRepoDid(tx, repoAtUri, record.RepoDid); err != nil { return fmt.Errorf("cascade repo_did: %w", err) }
if err := db.EnqueuePdsRewritesForRepo(tx, record.RepoDid, repoAtUri); err != nil { return fmt.Errorf("enqueue pds rewrites: %w", err) }
if err := tx.Commit(); err != nil { return fmt.Errorf("commit didAssign txn: %w", err) } }
if err := enforcer.RemoveRepo(record.OwnerDid, knot, legacyResource); err != nil { return fmt.Errorf("remove legacy RBAC policies for %s: %w", legacyResource, err) } if err := enforcer.AddRepo(record.OwnerDid, knot, record.RepoDid); err != nil { return fmt.Errorf("add RBAC policies for %s: %w", record.RepoDid, err) }
collabs, collabErr := db.GetCollaborators(d, orm.FilterEq("repo_did", record.RepoDid)) if collabErr != nil { return fmt.Errorf("get collaborators for RBAC update: %w", collabErr) } for _, c := range collabs { collabDid := c.SubjectDid.String() if err := enforcer.RemoveCollaborator(collabDid, knot, legacyResource); err != nil { return fmt.Errorf("remove collaborator RBAC for %s: %w", collabDid, err) } if err := enforcer.AddCollaborator(collabDid, knot, record.RepoDid); err != nil { return fmt.Errorf("add collaborator RBAC for %s: %w", collabDid, err) } }
if err := enforcer.E.SavePolicy(); err != nil { return fmt.Errorf("save RBAC policies after didAssign: %w", err) }
logger.Info("didAssign processed successfully", "repo_did", record.RepoDid, "owner_did", record.OwnerDid, "repo_name", record.RepoName)
return nil}