package spindle
import (
"context"
"errors"
"fmt"
"log/slog"
"os"
"slices"
"strings"
"github.com/bluesky-social/indigo/atproto/syntax"
"tangled.org/core/api/org_tangled"
"tangled.org/core/rbac"
"tangled.org/core/spindle/db"
"tangled.org/core/spindle/knotclient"
"tangled.org/core/tapc"
)
// knot and pds reads stay outside the lock, otherwise a hung knot blocks the repo
func (s *Spindle) ensureRepo(ctx context.Context, k knotclient.Knot, repoDid syntax.DID) (repo *db.Repo, err error) {
turn := s.beginTurn(repoDid)
l := s.l.With("repo", repoDid, "knot", k.Host)
// if a later turn already wrote, we return what it left
// only db work goes in fn, cleanup runs after the lock is gone
settle := func(fn func() (cleanup func() error, err error)) (*db.Repo, error) {
var cleanup func() error
applied, err := turn.apply(func() (err error) {
cleanup, err = fn()
return err
})
if err != nil {
return nil, err
}
if !applied {
return s.servedRepo(repoDid)
}
if cleanup != nil {
return nil, cleanup()
}
return nil, nil
}
var manifest org_tangled.RepoManifest
src := k
err = s.knots.GetRecord(ctx, k, repoDid, org_tangled.RepoManifestNSID, knotclient.ManifestRkey, &manifest)
if errors.Is(err, knotclient.ErrNotFound) && s.cfg.Server.FosterKnot != "" {
foster, ok := knotAt(s.cfg.Server.FosterKnot)
if !ok {
return nil, fmt.Errorf("foster knot %q isn't a usable url", s.cfg.Server.FosterKnot)
}
src = foster
err = s.knots.GetRecord(ctx, foster, repoDid, org_tangled.RepoManifestNSID, knotclient.ManifestRkey, &manifest)
}
if errors.Is(err, knotclient.ErrNotFound) {
deleted, err := s.knots.RepoDeleted(ctx, k, repoDid)
if err != nil && !errors.Is(err, knotclient.ErrNotFound) {
return nil, fmt.Errorf("reading status of %s: %w", repoDid, err)
}
return settle(func() (func() error, error) { return s.dropIfStoredKnot(ctx, k, repoDid, deleted) })
}
if err != nil {
return nil, fmt.Errorf("reading manifest of %s: %w", repoDid, err)
}
if manifest.Spindle == nil || *manifest.Spindle != s.cfg.Server.Did().String() {
return settle(func() (func() error, error) { return s.forgetRepo(ctx, repoDid, "manifest names another spindle") })
}
owner, name, rkey, err := manifestIdentity(&manifest)
if err != nil {
l.Warn("ignoring manifest", "err", err)
reason := err.Error()
return settle(func() (func() error, error) { return s.forgetRepo(ctx, repoDid, reason) })
}
// the knot says who owns it, but only the owner's pds can confirm that
declared, err := s.pdsDeclaredRepo(ctx, owner, rkey.String())
switch {
case errors.Is(err, knotclient.ErrNotFound):
// a rename can leave no declaration for a bit, and forgetting here loses the old name
l.Info("owner has no declaration for this repo yet, keeping it as it is", "owner", owner, "rkey", rkey)
return nil, nil
case err != nil:
return nil, fmt.Errorf("reading declaration %s/%s: %w", owner, rkey, err)
case declared != repoDid:
l.Warn("owner's declaration points at another repo", "owner", owner, "rkey", rkey, "declared", declared)
return settle(func() (func() error, error) {
return s.forgetRepo(ctx, repoDid, "owner's declaration points elsewhere")
})
}
if ban, err := s.db.IsBanned(repoDid, owner); err != nil {
return nil, fmt.Errorf("checking bans: %w", err)
} else if ban != nil {
l.Warn("refusing banned repo", "owner", owner)
return settle(func() (func() error, error) { return s.claimWipeIfRows(ctx, repoDid, "banned") })
}
enrolled, err := db.Enrolls(s.db, &s.cfg.Server, owner)
if err != nil {
return nil, fmt.Errorf("checking enrollment: %w", err)
}
if !enrolled {
l.Warn("rejecting repo: owner isn't admitted", "owner", owner)
return nil, nil
}
repo = &db.Repo{Knot: k.Host, Owner: owner, Rkey: rkey, RepoDid: repoDid, Name: name, CreatedAt: manifest.CreatedAt}
var banned bool
var wipe func() error
applied, err := turn.apply(func() error {
err := s.registerRepo(ctx, l, repo)
if errors.Is(err, errBanned) {
banned = true
wipe, err = s.claimWipeIfRows(ctx, repoDid, "banned")
}
return err
})
switch {
case errors.Is(err, errWiping):
return nil, nil
case err != nil:
return nil, err
case banned:
if wipe != nil {
return nil, wipe()
}
return nil, nil
case !applied:
return s.servedRepo(repoDid)
}
desired, err := s.fetchCollaborators(ctx, src, repoDid)
if err != nil {
return repo, err
}
_, err = turn.apply(func() error { return s.applyCollaborators(repoDid, desired) })
return repo, err
}
func manifestIdentity(m *org_tangled.RepoManifest) (owner syntax.DID, name string, rkey syntax.RecordKey, err error) {
if m.Declaration == nil {
return "", "", "", errors.New("manifest has no declaration")
}
if owner, err = syntax.ParseDID(m.Declaration.Owner); err != nil {
return "", "", "", fmt.Errorf("manifest owner %q is malformed: %w", m.Declaration.Owner, err)
}
name = m.Declaration.Name
if rkey, err = syntax.ParseRecordKey(strings.ToLower(name)); err != nil {
return "", "", "", fmt.Errorf("manifest name %q is malformed: %w", name, err)
}
return owner, name, rkey, nil
}
var (
errBanned = errors.New("owner is banned")
errWiping = errors.New("repo is being wiped")
)
// same repo did under a new owner or rkey is a transfer or rename, so the old rows go
func (s *Spindle) registerRepo(ctx context.Context, l *slog.Logger, repo *db.Repo) error {
if wiping, err := s.db.WipeExists(repo.RepoDid); err != nil {
return fmt.Errorf("checking for a wipe: %w", err)
} else if wiping {
return errWiping
}
rows, err := s.db.ReposByDid(repo.RepoDid)
if err != nil {
return fmt.Errorf("looking up repo rows: %w", err)
}
same := func(r db.Repo) bool { return r.Owner == repo.Owner && r.Rkey == repo.Rkey }
stale := slices.DeleteFunc(slices.Clone(rows), same)
var prior *db.Repo
i := slices.IndexFunc(rows, same)
known := i >= 0
if known {
prior = &rows[i]
} else if len(stale) > 0 {
prior = &stale[0]
}
// bans don't take the repo lock, so one can land right after this check
if ban, err := s.db.IsBanned(repo.RepoDid, repo.Owner); err != nil {
return fmt.Errorf("checking bans: %w", err)
} else if ban != nil {
return errBanned
}
if err := s.e.AddRepo(repo.Owner.String(), rbac.ThisServer, repo.RepoDid.String()); err != nil {
return fmt.Errorf("add repo policy: %w", err)
}
if err := s.db.AddRepo(*repo); err != nil {
return fmt.Errorf("add repo: %w", err)
}
// the migrations read the old rows, so the stale ones get deleted after
if !known {
migrateLegacyRepoSecrets(ctx, s.db, s.vault, l, repo.Owner, repo.Name, repo.Rkey, repo.RepoDid)
migrateLegacyRepoCasbin(ctx, s.db, s.e, l, repo.Owner, repo.Name, repo.Rkey, repo.RepoDid)
}
for _, r := range stale {
if err := s.db.DeleteRepoByOwnerRkey(r.Owner, r.Rkey); err != nil {
l.Warn("dropping stale repo row failed", "owner", r.Owner, "rkey", r.Rkey, "err", err)
}
if r.Owner != repo.Owner {
if err := s.e.RemoveRepo(r.Owner.String(), rbac.ThisServer, repo.RepoDid.String()); err != nil {
l.Warn("dropping stale owner policy failed", "owner", r.Owner, "err", err)
}
if err := s.releaseOwnerInterest(ctx, r.Owner); err != nil {
l.Warn("releasing stale owner interest failed", "owner", r.Owner, "err", err)
}
}
}
if s.jc != nil {
s.jc.AddDid(repo.Owner.String())
}
if prior != nil && prior.Name != "" && prior.Name != repo.Name {
l.Info("repo renamed, firing webhook", "old", prior.Name, "new", repo.Name)
s.wh.FireRename(ctx, repo.Owner, repo, prior.Name, repo.Name)
}
if !known && s.scheduler != nil {
// cloning for schedules is slow, so don't block the caller or the repo lock
go func() {
if err := s.scheduler.RefreshRepo(context.WithoutCancel(ctx), *repo); err != nil {
l.Warn("failed to refresh repository schedules", "err", err)
}
}()
}
return nil
}
func (s *Spindle) fetchCollaborators(ctx context.Context, k knotclient.Knot, repoDid syntax.DID) (map[syntax.DID]struct{}, error) {
l := s.l.With("repo", repoDid)
rkeys, err := s.knots.ListRecordKeys(ctx, k, repoDid, org_tangled.GraphCollaboratorNSID)
if err != nil {
return nil, fmt.Errorf("listing collaborators of %s: %w", repoDid, err)
}
desired := make(map[syntax.DID]struct{}, len(rkeys))
for _, rkey := range rkeys {
subject, err := syntax.ParseDID(rkey)
if err != nil {
l.Info("skipping collaborator with a malformed did", "rkey", rkey)
continue
}
desired[subject] = struct{}{}
}
return desired, nil
}
func (s *Spindle) applyCollaborators(repoDid syntax.DID, desired map[syntax.DID]struct{}) error {
l := s.l.With("repo", repoDid)
tracked, err := s.db.ListCollaboratorsByRepoDid(repoDid)
if err != nil {
return err
}
var errs []error
have := make(map[syntax.DID]struct{}, len(tracked))
for _, c := range tracked {
have[c.Subject] = struct{}{}
if _, ok := desired[c.Subject]; ok {
continue
}
if err := s.untrackCollaborator(repoDid, c.Subject); err != nil {
errs = append(errs, fmt.Errorf("remove collaborator %s: %w", c.Subject, err))
continue
}
l.Info("removed collaborator", "subject", c.Subject)
}
for subject := range desired {
if _, ok := have[subject]; ok {
continue
}
if err := s.trackCollaborator(repoDid, subject); err != nil {
errs = append(errs, fmt.Errorf("add collaborator %s: %w", subject, err))
continue
}
l.Info("added collaborator", "subject", subject)
}
return errors.Join(errs...)
}
// access goes first, the clone is removed after the lock is released
func (s *Spindle) forgetRepo(ctx context.Context, repoDid syntax.DID, reason string) (cleanup func() error, err error) {
rows, err := s.db.ReposByDid(repoDid)
if err != nil {
return nil, fmt.Errorf("looking up repo rows: %w", err)
}
if len(rows) == 0 {
return nil, nil
}
l := s.l.With("repo", repoDid, "reason", reason)
l.Warn("no longer serving repo, keeping its secrets and history")
if err := s.revokeRepoAccess(ctx, repoDid, rows); err != nil {
return nil, fmt.Errorf("forgetting %s: %w", repoDid, err)
}
if err := s.db.DeleteReposByDid(repoDid); err != nil {
return nil, err
}
return func() error {
for _, r := range rows {
if err := s.releaseOwnerInterest(ctx, r.Owner); err != nil {
l.Warn("releasing owner interest failed", "owner", r.Owner, "err", err)
}
}
return os.RemoveAll(s.newRepoPath(repoDid))
}, nil
}
// a knot without the repo yet also says not found, so only the stored knot counts.
// secrets only go once that knot says the repo was deleted
func (s *Spindle) dropIfStoredKnot(ctx context.Context, k knotclient.Knot, repoDid syntax.DID, deleted bool) (cleanup func() error, err error) {
rows, err := s.db.ReposByDid(repoDid)
if err != nil {
return nil, fmt.Errorf("looking up repo rows: %w", err)
}
if !slices.ContainsFunc(rows, func(r db.Repo) bool { return r.Knot == k.Host }) {
if len(rows) > 0 {
s.l.Warn("not wiping, the knot that lost the manifest isn't the stored one", "repo", repoDid, "knot", k.Host, "stored", rows[0].Knot)
}
return nil, nil
}
if !deleted {
return s.forgetRepo(ctx, repoDid, "manifest missing on the knot")
}
return s.claimWipeIfRows(ctx, repoDid, "repo deleted on the knot")
}
// a missing declaration comes back as knotclient.ErrNotFound
func (s *Spindle) pdsDeclaredRepo(ctx context.Context, owner syntax.DID, rkey string) (syntax.DID, error) {
pds, err := s.pdsOf(ctx, owner)
if err != nil {
return "", err
}
declared, err := s.declaredRepoAt(ctx, pds, owner, rkey)
if errors.Is(err, knotclient.ErrNotFound) && s.cfg.Server.LiarPds != "" {
liar, ok := knotAt(s.cfg.Server.LiarPds)
if !ok {
return "", fmt.Errorf("liar pds %q isn't a usable url", s.cfg.Server.LiarPds)
}
return s.declaredRepoAt(ctx, liar, owner, rkey)
}
return declared, err
}
func (s *Spindle) declaredRepoAt(ctx context.Context, pds knotclient.Knot, owner syntax.DID, rkey string) (syntax.DID, error) {
var declaration org_tangled.RepoDeclaration
if err := s.knots.GetRecord(ctx, pds, owner, org_tangled.RepoDeclarationNSID, rkey, &declaration); err != nil {
return "", err
}
return syntax.ParseDID(declaration.Repo)
}
// always reads from the knot, for xrpc calls the firehose hasn't caught up to
func (s *Spindle) RefreshRepo(ctx context.Context, repoDid syntax.DID) error {
// a mark landing mid-refresh bumps the epoch, and clearAt only clears when it hasn't moved
epoch, err := s.db.RepoRetryEpoch(repoDid)
if err != nil {
s.l.Error("reading a repo retry", "repo", repoDid, "err", err)
epoch = -1
}
k, err := s.knots.KnotOf(ctx, repoDid)
if err != nil {
return fmt.Errorf("finding the knot of %s: %w", repoDid, err)
}
if _, err := s.ensureRepo(ctx, k, repoDid); err != nil {
return err
}
if err := s.retries.clearAt(s.db, repoDid, epoch); err != nil {
s.l.Error("clearing a repo retry", "repo", repoDid, "err", err)
}
return nil
}
// a live event won't come again, so a failed refresh has to go in the retry set
func (s *Spindle) refreshForEvent(ctx context.Context, rec *tapc.RecordEventData, repoDid syntax.DID) error {
if markOnly(ctx) {
return s.markRepo(repoDid)
}
err := s.RefreshRepo(ctx, repoDid)
// no knot means no retry, waiting won't add one
if err != nil && rec.Live && ctx.Err() == nil && !errors.Is(err, knotclient.ErrNoKnot) {
s.failRepoRefresh(repoDid)
}
return err
}
func (s *Spindle) knowsRepo(repoDid syntax.DID) (bool, error) {
rows, err := s.db.ReposByDid(repoDid)
return len(rows) > 0, err
}