Something went wrong. Try again.
An ATProto Powered Reputation Network
Something went wrong. Try again.
14 kB · 488 lines
Go
at master
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489package appview
import ( "context" "database/sql" "errors" "log/slog" "slices" "strings" "sync" "time"
"github.com/bluesky-social/indigo/atproto/syntax"
"provenance/internal/atproto" "provenance/internal/domain" "provenance/internal/logging" "provenance/internal/reputation" reputationalg "provenance/internal/reputation/algorithms" "provenance/internal/storage" "provenance/internal/trust" "provenance/internal/trust/algorithms" "provenance/internal/upstream")
const ( softParticipantStream = "jetstream-soft-participants" softAllStream = "jetstream-soft-all" softParticipantListStream = "jetstream-soft-participant-lists" softAllListStream = "jetstream-soft-all-lists" StreamSource = "jetstream" InitialRetryDelay = time.Second MaxRetryDelay = 30 * time.Second ParticipantLimit = 10000 TagLimit = 20 VouchPageLimit = 100 identityTimeout = 5 * time.Second stableStreamInterval = time.Minute)
type Repository interface { Snapshot(context.Context, string) ([]domain.Vouch, error) Resolve(context.Context, string) (string, error)}
type identityInvalidator interface { InvalidateIdentity(context.Context, string) error}
type Stream interface { Consume(context.Context, int64, func(context.Context, string, int64) (bool, error), func(context.Context, domain.Change) error) error}
type Service struct { store storage.Store repo Repository limits trust.Limits algorithm trust.Algorithm reputation reputation.Algorithm cacheTTL time.Duration softMode domain.SoftMode now func() time.Time syncMu sync.Mutex index *indexMonitor}
type Evaluation struct { trust.Result Trusted bool `json:"trusted"`}
func New(store storage.Store, repo Repository, limits trust.Limits) *Service { return NewWithAlgorithm(store, repo, limits, algorithms.Optimistic{})}
// Inject one trust algorithm for direct queries and reputation authority checks.func NewWithAlgorithm(store storage.Store, repo Repository, limits trust.Limits, algorithm trust.Algorithm) *Service { return NewWithOptions(store, repo, limits, Options{Trust: algorithm})}
type Options struct { Trust trust.Algorithm Reputation reputation.Algorithm CacheTTL time.Duration SoftMode domain.SoftMode Indexing domain.IndexPhase}
func NewWithOptions(store storage.Store, repo Repository, limits trust.Limits, options Options) *Service { if options.Trust == nil { options.Trust = algorithms.Optimistic{} } if options.Reputation == nil { options.Reputation = reputationalg.DirectSum{} } if options.CacheTTL == 0 { options.CacheTTL = trust.DefaultCacheTTL } monitor := newIndexMonitor(options.Indexing) monitor.registry.MustRegister(newDatabaseMetrics(store)) return &Service{store: store, repo: repo, limits: limits, algorithm: options.Trust, reputation: options.Reputation, cacheTTL: options.CacheTTL, softMode: options.SoftMode, now: time.Now, index: monitor}}
func (s *Service) Join(ctx context.Context, did string) error { if err := atproto.ValidDID(did); err != nil { return err } s.syncMu.Lock() defer s.syncMu.Unlock() if _, err := s.store.Cursor(ctx, s.streamSource(), time.Now().UnixMicro()); err != nil { return err } if err := s.store.Join(ctx, did); err != nil { return err } p, err := s.store.Participant(ctx, did) if err != nil { return err } if p.BootstrapUS != 0 { if s.softMode == domain.SoftDisabled { return nil } ready, err := s.store.SoftReady(ctx, did) if err != nil { return err } if ready { return nil } } return s.bootstrap(ctx, did)}
func (s *Service) Config(ctx context.Context, did string) (storage.Participant, error) { return s.store.Participant(ctx, did)}
func (s *Service) SetThreshold(ctx context.Context, did string, value float64) error { return s.store.SetThreshold(ctx, did, value)}
func (s *Service) Reputation(ctx context.Context, source, target, tag string, mode domain.Mode, options ...trust.Accounts) (trust.Result, error) { if err := domain.ValidateQuery(tag, mode); err != nil { return trust.Result{}, err } input, err := s.resolveInput(ctx, source, target, options) if err != nil { return trust.Result{}, err } reader := &trustReader{service: s, input: input, tag: tag, mode: mode} return reader.Score(ctx, input.Target)}
// Attach reference text after calculation so that algorithms never select paths by comment content.func pathComments(result trust.Result, records map[[2]string]*domain.Vouch) trust.Result { if len(result.PathScores) == 0 { return result } // Copy the returned nodes because an algorithm can reuse its result buffers. result.PathScores = slices.Clone(result.PathScores) result.PathScores[0].Comment = "" result.PathScores[0].Source = "" result.PathScores[0].ListURI = "" var empty domain.Vouch for i := 1; i < len(result.PathScores); i++ { pair := [2]string{result.PathScores[i-1].DID, result.PathScores[i].DID} record := records[pair] if record == nil { record = &empty } result.PathScores[i].Comment = record.Comment result.PathScores[i].Source = record.Source result.PathScores[i].ListURI = record.ListURI } return result}
func (s *Service) resolveInput(ctx context.Context, source, target string, options []trust.Accounts) (trust.Input, error) { input := trust.Input{Limits: s.limits, Accounts: trust.Accounts{}} if len(options) > 1 { return input, errors.New("only one account options map is accepted") } allowed := map[string]int{} for _, parameter := range s.algorithm.Describe().Accounts { allowed[parameter.Name] = parameter.MaxItems } if len(options) == 1 { for name, values := range options[0] { limit, ok := allowed[name] if !ok { return input, errors.New("unsupported algorithm parameter: " + name) } if len(values) == 0 || len(values) > limit { return input, errors.New("invalid account list: " + name) } } } // One deadline and cache cover every identity in a query, including algorithm options. ctx, cancel := context.WithTimeout(ctx, identityTimeout) defer cancel() cache := map[string]string{} resolve := func(value string) (string, error) { value = strings.TrimSpace(value) if !strings.HasPrefix(value, "did:") { value = strings.ToLower(strings.TrimPrefix(value, "@")) } if did, ok := cache[value]; ok { return did, nil } did, err := s.resolve(ctx, value) if err == nil { cache[value] = did } return did, err } var err error input.Source, err = resolve(source) if err != nil { return input, err } input.Target, err = resolve(target) if err != nil { return input, err } if len(options) == 0 { return input, nil } for name, values := range options[0] { seen := map[string]struct{}{} for _, value := range values { did, err := resolve(value) if err != nil { return input, err } if did == input.Source || did == input.Target { return input, errors.New("source and target cannot appear in algorithm account lists") } if _, ok := seen[did]; ok { continue } seen[did] = struct{}{} input.Accounts[name] = append(input.Accounts[name], did) } } return input, nil}
func (s *Service) resolve(ctx context.Context, value string) (string, error) { // DIDs query the local index without an external identity lookup. if strings.HasPrefix(value, "did:") { return value, atproto.ValidDID(value) } if _, err := syntax.ParseHandle(value); err != nil { return "", err } did, err := s.repo.Resolve(ctx, value) if err != nil { return "", err } return did, atproto.ValidDID(did)}
func (s *Service) Evaluate(ctx context.Context, source, target, tag string, mode domain.Mode, options ...trust.Accounts) (Evaluation, error) { p, err := s.store.Participant(ctx, source) if err != nil { return Evaluation{}, err } result, err := s.Reputation(ctx, source, target, tag, mode, options...) if err != nil { return Evaluation{}, err } trusted := result.Status == trust.SelfStatus || result.Status == trust.CalculatedStatus && result.Score != nil && *result.Score >= p.Threshold return Evaluation{result, trusted}, nil}
func (s *Service) Tags(ctx context.Context, prefix string) ([]string, error) { if err := domain.ValidateTag(prefix); err != nil { return nil, err } return s.store.Tags(ctx, prefix, TagLimit)}
func (s *Service) Vouches(ctx context.Context, did string, offset int) ([]domain.Vouch, error) { if err := atproto.ValidDID(did); err != nil { return nil, err } if offset < 0 { return nil, errors.New("offset must not be negative") } return s.store.OwnVouches(ctx, did, VouchPageLimit, offset)}
func (s *Service) Sync(ctx context.Context, did string) error { s.syncMu.Lock() defer s.syncMu.Unlock() p, err := s.store.Participant(ctx, did) if err != nil { return err } if !p.Active { return errors.New("inactive accounts cannot trigger recovery") } // Explicit recovery forces one fresh identity lookup, not one lookup per collection. if repo, ok := s.repo.(identityInvalidator); ok { if err := repo.InvalidateIdentity(ctx, did); err != nil { return err } } return s.bootstrap(ctx, did)}
// Hold the ingest lock across the snapshot so that replay cannot overwrite newer bootstrap data.func (s *Service) bootstrap(ctx context.Context, did string) error { started := time.Now() vouches, err := s.repo.Snapshot(ctx, did) if err != nil { return err } if domain.SoftLists(s.softMode) { repo, ok := s.repo.(ListRepository) if !ok { return errors.New("repository does not support lists") } lists, err := repo.ListSnapshot(ctx, did) if err != nil { return err } return s.store.ReplaceWithLists(ctx, did, vouches, lists, started) } return s.store.Replace(ctx, did, vouches, started)}
func (s *Service) BootstrapPending(ctx context.Context) error { participants, err := s.store.Participants(ctx, ParticipantLimit+1) if err != nil { return err } if len(participants) > ParticipantLimit { return errors.New("participant bootstrap limit exceeded") } for _, p := range participants { if err := ctx.Err(); err != nil { return err } if !p.Active { continue } if p.BootstrapUS != 0 { if s.softMode == domain.SoftDisabled { continue } ready, err := s.store.SoftReady(ctx, p.DID) if err != nil { return err } if ready { continue } if err := s.Sync(ctx, p.DID); err != nil { slog.Warn("soft bootstrap failed", "did", p.DID, "error", err) } continue } if err := s.Join(ctx, p.DID); err != nil { slog.Warn("participant bootstrap failed", "did", p.DID, "error", err) } } return nil}
func (s *Service) Apply(ctx context.Context, change domain.Change) error { s.syncMu.Lock() defer s.syncMu.Unlock() if err := s.prepareChange(ctx, &change); err != nil { return err } return s.store.Apply(ctx, s.streamSource(), change)}
func (s *Service) prepareChange(ctx context.Context, change *domain.Change) error { if change.RefreshIdentity { if repo, ok := s.repo.(identityInvalidator); ok { if err := repo.InvalidateIdentity(ctx, change.DID); err != nil { return err } } } if domain.SoftLists(s.softMode) && change.Source == domain.ListBlockCollection && change.ListRecord != nil && !(change.Sequence > 0 && domain.SoftAll(s.softMode)) { needed, err := s.store.NeedsList(ctx, *change) if err != nil { return err } if needed { repo, ok := s.repo.(ListRepository) if !ok { return errors.New("repository does not support lists") } data, err := repo.ListMembers(ctx, change.ListRecord.ListURI) if err != nil { return err } change.ListData = &data } } return nil}
// Separate cursors keep disabled periods from advancing an enabled mode's replay position.func (s *Service) streamSource() string { if s.softMode == domain.SoftAllAuthorLists { return softAllListStream } if s.softMode == domain.SoftParticipantLists { return softParticipantListStream } if s.softMode == domain.SoftAllAuthors { return softAllStream } if s.softMode == domain.SoftParticipants { return softParticipantStream } return StreamSource}
func (s *Service) wanted(ctx context.Context, did string, stamp int64) (bool, error) { if domain.SoftLists(s.softMode) { owner, err := s.store.WantsListOwner(ctx, did) if err != nil || owner { return owner, err } } p, err := s.store.Participant(ctx, did) if errors.Is(err, sql.ErrNoRows) { return false, nil } return err == nil && stamp > p.BootstrapUS, err}
// Reconnect to the event source, never to each participant's PDS.func (s *Service) Run(ctx context.Context, stream Stream) { delay := InitialRetryDelay logger := slog.Default().With("component", "legacy_stream", "source", s.streamSource(), "soft_mode", s.softMode) logger.Info("legacy stream started") defer logger.Info("legacy stream worker stopped") prepared := false attempt := 0 for { cursor, err := s.store.Cursor(ctx, s.streamSource(), time.Now().UnixMicro()) if err == nil && !prepared { if err := s.BootstrapPending(ctx); err != nil { logger.Warn("pending bootstrap failed", "error", logging.Error(err)) } prepared = true } processed := 0 started := time.Now() attempt++ logger.Info("legacy stream attempt started", "attempt", attempt, "cursor_time_us", cursor) if err == nil { err = stream.Consume(ctx, cursor, s.wanted, func(ctx context.Context, change domain.Change) error { if err := s.Apply(ctx, change); err != nil { return err } processed++ return nil }) } if ctx.Err() != nil { return } delay = reconnectDelay(delay, processed, time.Since(started)) pause := upstream.Jitter(delay) logger.Warn("legacy stream reconnect scheduled", "attempt", attempt, "events", processed, "duration", time.Since(started), "retry_after", pause, "error", logging.Error(err)) if err := upstream.Wait(ctx, pause); err != nil { return } delay = min(delay*2, MaxRetryDelay) }}
func reconnectDelay(delay time.Duration, processed int, uptime time.Duration) time.Duration { // A few replayed events do not prove recovery from an upstream failure. if processed > 0 && uptime >= stableStreamInterval { return InitialRetryDelay } return delay}