Something went wrong. Try again.
Monorepo for Tangled
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436package knotacl
import ( "context" "database/sql" "errors" "fmt" "log/slog" "slices" "sync" "sync/atomic" "time"
"github.com/bluesky-social/indigo/atproto/syntax" "golang.org/x/sync/singleflight"
"tangled.org/core/appview/db" "tangled.org/core/appview/models" "tangled.org/core/orm")
const ( reconcileTTL = 5 * time.Minute reconcileBackoff = 15 * time.Second cursorRetention = 1 * time.Hour)
var errReconcileBackoff = errors.New("reconcile suppressed during backoff")
type Cursor int64
type scopeState struct { mu sync.Mutex gen atomic.Uint64 failedAt time.Time refs int}
type roster struct { store *db.DB src lister ttl time.Duration now func() time.Time log *slog.Logger group singleflight.Group
mu sync.Mutex scopes map[string]*scopeState}
func newRoster(store *db.DB, src lister, ttl time.Duration, now func() time.Time, logger *slog.Logger) *roster { if now == nil { now = time.Now } if logger == nil { logger = slog.Default() } return &roster{ store: store, src: src, ttl: ttl, now: now, log: logger, scopes: map[string]*scopeState{}, }}
func (r *roster) GetKnotMembers(ctx context.Context, host string) ([]string, error) { return r.serve(ctx, memberScope(host), func() error { return r.reconcileMembers(ctx, host) }, func() ([]string, error) { rows, err := db.GetKnotMembers(r.store, orm.FilterEq("domain", host)) if err != nil { return nil, err } return mapSlice(rows, func(km models.KnotMember) string { return km.Subject.String() }), nil }, )}
func (r *roster) GetRepoCollaborators(ctx context.Context, host, repoDid string) ([]string, error) { return r.serve(ctx, collabScope(repoDid), func() error { return r.reconcileCollaborators(ctx, host, repoDid) }, func() ([]string, error) { rows, err := db.GetCollaborators(r.store, orm.FilterEq("repo_did", repoDid)) if err != nil { return nil, err } return mapSlice(rows, func(c models.Collaborator) string { return c.SubjectDid.String() }), nil }, )}
func (r *roster) AddKnotMember(host string, subject syntax.DID, cursor Cursor) error { return r.applyDelta(memberScope(host), subject, cursor, func(tx *sql.Tx) error { if err := db.RemoveKnotMember(tx, orm.FilterEq("domain", host), orm.FilterEq("subject", subject.String()), ); err != nil { return err } return db.AddKnotMember(tx, models.KnotMember{Domain: host, Subject: subject}) })}
func (r *roster) RemoveKnotMember(host string, subject syntax.DID, cursor Cursor) error { return r.applyDelta(memberScope(host), subject, cursor, func(tx *sql.Tx) error { return db.RemoveKnotMember(tx, orm.FilterEq("domain", host), orm.FilterEq("subject", subject.String()), ) })}
// NOTE: maybe TODO or not, but no Did of adder means no ability to suggest a vouch for the person they just added as collaborator.func (r *roster) AddCollaborator(repoDid, subject syntax.DID, cursor Cursor) error { return r.applyDelta(collabScope(repoDid.String()), subject, cursor, func(tx *sql.Tx) error { return db.AddCollaborator(tx, models.Collaborator{SubjectDid: subject, RepoDid: repoDid}) })}
func (r *roster) RemoveCollaborator(repoDid, subject syntax.DID, cursor Cursor) error { return r.applyDelta(collabScope(repoDid.String()), subject, cursor, func(tx *sql.Tx) error { return db.DeleteCollaborator(tx, orm.FilterEq("repo_did", repoDid.String()), orm.FilterEq("subject_did", subject.String()), ) })}
func (r *roster) applyDelta(scope string, subject syntax.DID, cursor Cursor, mutate func(*sql.Tx) error) error { st := r.acquire(scope) defer r.release(scope, st)
st.mu.Lock() defer st.mu.Unlock()
tx, err := r.store.Begin() if err != nil { return err } defer tx.Rollback()
seen, ok, err := seenCursor(tx, scope, subject) if err != nil { return err } if ok && cursor <= seen { return nil }
if err := mutate(tx); err != nil { return err } if err := recordCursor(tx, scope, subject, cursor); err != nil { return err } if err := tx.Commit(); err != nil { return err }
st.gen.Add(1) return nil}
func (r *roster) InvalidateMembers(host string) { _ = clearSyncedAt(r.store, memberScope(host))}
func (r *roster) InvalidateCollaborators(host, repoDid string) { _ = clearSyncedAt(r.store, collabScope(repoDid))}
func (r *roster) serve(ctx context.Context, key string, reconcile func() error, read func() ([]string, error)) ([]string, error) { if memo := memoFrom(ctx); memo != nil { if v, ok := memo.get(key); ok { return slices.Clone(v), nil } }
recErr := r.maybeReconcile(key, reconcile)
subjects, err := read() if err != nil { return nil, err }
if len(subjects) == 0 && recErr != nil && !r.everSynced(key) { return nil, fmt.Errorf("%w: %v", ErrKnotUnreachable, recErr) }
subjects = dedup(subjects) if memo := memoFrom(ctx); memo != nil { memo.put(key, subjects) } return slices.Clone(subjects), nil}
func (r *roster) maybeReconcile(key string, reconcile func() error) error { if r.fresh(key) { return nil } if r.backingOff(key) { return errReconcileBackoff } _, err, _ := r.group.Do(key, func() (any, error) { if r.fresh(key) { return nil, nil } if r.backingOff(key) { return nil, errReconcileBackoff } return nil, reconcile() }) return err}
func (r *roster) reconcileMembers(ctx context.Context, host string) error { scope := memberScope(host) st := r.acquire(scope) defer r.release(scope, st)
genBefore := st.gen.Load() subjects, err := r.src.GetKnotMembers(ctx, host) if err != nil { r.markFailed(st) return err } return r.commitReconcile(st, scope, genBefore, func(tx *sql.Tx) error { if err := db.RemoveKnotMember(tx, orm.FilterEq("domain", host)); err != nil { return err } for _, s := range subjects { did, perr := syntax.ParseDID(s) if perr != nil { r.log.Warn("dropping malformed member DID from reconcile", "host", host, "subject", s, "error", perr) continue } if err := db.AddKnotMember(tx, models.KnotMember{Domain: host, Subject: did}); err != nil { return err } } return nil })}
func (r *roster) reconcileCollaborators(ctx context.Context, host, repoDid string) error { repo, perr := syntax.ParseDID(repoDid) if perr != nil { return perr } scope := collabScope(repoDid) st := r.acquire(scope) defer r.release(scope, st)
genBefore := st.gen.Load() subjects, err := r.src.GetRepoCollaborators(ctx, host, repoDid) if err != nil { r.markFailed(st) return err } return r.commitReconcile(st, scope, genBefore, func(tx *sql.Tx) error { if err := db.DeleteCollaborator(tx, orm.FilterEq("repo_did", repoDid)); err != nil { return err } for _, s := range subjects { did, perr := syntax.ParseDID(s) if perr != nil { r.log.Warn("dropping malformed collaborator DID from reconcile", "repo_did", repoDid, "subject", s, "error", perr) continue } if err := db.AddCollaborator(tx, models.Collaborator{SubjectDid: did, RepoDid: repo}); err != nil { return err } } return nil })}
func (r *roster) commitReconcile(st *scopeState, scope string, genBefore uint64, replace func(*sql.Tx) error) error { st.mu.Lock() defer st.mu.Unlock()
if st.gen.Load() != genBefore { if err := setSyncedAt(r.store, scope, r.now()); err != nil { return err } r.clearFailed(st) return nil }
tx, err := r.store.Begin() if err != nil { return err } defer tx.Rollback() if err := replace(tx); err != nil { return err } if err := setSyncedAt(tx, scope, r.now()); err != nil { return err } if err := pruneCursors(tx, scope, int64(cursorRetention)); err != nil { return err } if err := tx.Commit(); err != nil { return err } r.clearFailed(st) return nil}
func (r *roster) acquire(scope string) *scopeState { r.mu.Lock() defer r.mu.Unlock() st := r.scopes[scope] if st == nil { st = &scopeState{} r.scopes[scope] = st } st.refs++ return st}
func (r *roster) release(scope string, st *scopeState) { r.mu.Lock() defer r.mu.Unlock() st.refs-- if st.refs == 0 && st.failedAt.IsZero() { delete(r.scopes, scope) }}
func (r *roster) markFailed(st *scopeState) { r.mu.Lock() defer r.mu.Unlock() st.failedAt = r.now()}
func (r *roster) clearFailed(st *scopeState) { r.mu.Lock() defer r.mu.Unlock() st.failedAt = time.Time{}}
func (r *roster) backingOff(scope string) bool { r.mu.Lock() defer r.mu.Unlock() st, ok := r.scopes[scope] if !ok { return false } return !st.failedAt.IsZero() && r.now().Sub(st.failedAt) < reconcileBackoff}
func (r *roster) fresh(key string) bool { at, ok, err := getSyncedAt(r.store, key) if err != nil || !ok { return false } return r.now().Sub(at) < r.ttl}
func (r *roster) everSynced(key string) bool { _, ok, _ := getSyncedAt(r.store, key) return ok}
func memberScope(host string) string { return "m\x00" + host }
func collabScope(repoDid string) string { return "c\x00" + repoDid }
func getSyncedAt(e db.Execer, key string) (time.Time, bool, error) { var raw string err := e.QueryRow(`select synced_at from knotacl_sync where scope_key = ?`, key).Scan(&raw) if errors.Is(err, sql.ErrNoRows) { return time.Time{}, false, nil } if err != nil { return time.Time{}, false, err } at, err := time.Parse(time.RFC3339, raw) if err != nil { return time.Time{}, false, err } return at, true, nil}
func setSyncedAt(e db.Execer, key string, at time.Time) error { _, err := e.Exec( `insert into knotacl_sync (scope_key, synced_at) values (?, ?) on conflict(scope_key) do update set synced_at = excluded.synced_at`, key, at.UTC().Format(time.RFC3339), ) return err}
func clearSyncedAt(e db.Execer, key string) error { _, err := e.Exec(`delete from knotacl_sync where scope_key = ?`, key) return err}
func seenCursor(e db.Execer, scope string, subject syntax.DID) (Cursor, bool, error) { var raw int64 err := e.QueryRow( `select cursor from knotacl_delta_cursor where scope_key = ? and subject = ?`, scope, subject.String(), ).Scan(&raw) if errors.Is(err, sql.ErrNoRows) { return 0, false, nil } if err != nil { return 0, false, err } return Cursor(raw), true, nil}
func recordCursor(e db.Execer, scope string, subject syntax.DID, cursor Cursor) error { _, err := e.Exec( `insert into knotacl_delta_cursor (scope_key, subject, cursor) values (?, ?, ?) on conflict(scope_key, subject) do update set cursor = excluded.cursor`, scope, subject.String(), int64(cursor), ) return err}
func pruneCursors(e db.Execer, scope string, retentionNanos int64) error { _, err := e.Exec( `delete from knotacl_delta_cursor where scope_key = ? and cursor < (select max(cursor) from knotacl_delta_cursor where scope_key = ?) - ?`, scope, scope, retentionNanos, ) return err}