Something went wrong. Try again.
An ATProto Powered Reputation Network
Something went wrong. Try again.
9.7 kB · 304 lines
Go
at master
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305package storage
import ( "context" "database/sql" "errors" "slices" "time"
"provenance/internal/domain" "provenance/internal/storage/db")
func appendListRows(result []domain.Vouch, rows []db.ListEdge) []domain.Vouch { result = slices.Grow(result, len(rows)) for _, row := range rows { result = append(result, domain.Vouch{Author: row.Author, Subject: row.Subject, Score: domain.SoftBlockScore, Source: row.Source, RecordKey: row.RecordKey, CreatedAt: row.CreatedAt, ListURI: row.ListUri, ListAuthor: row.ListAuthor, ListKey: row.ListKey, ItemKey: row.ItemKey}) } return result}
func (s *SQLite) WantsListOwner(ctx context.Context, did string) (bool, error) { if !domain.SoftLists(s.softMode) { return false, nil } wanted, err := s.queries.WantsListOwner(ctx, did) return wanted, err}
func (s *SQLite) listSubscriber(ctx context.Context, q *db.Queries, did string) (bool, error) { if domain.SoftAll(s.softMode) { return true, nil } _, err := q.Participant(ctx, did) if errors.Is(err, sql.ErrNoRows) { return false, nil } return err == nil, err}
func listFresh(ctx context.Context, q *db.Queries, change domain.Change) (bool, error) { version, err := q.ListVersion(ctx, db.ListVersionParams{Author: change.DID, Source: change.Source, RecordKey: change.RecordKey}) if err != nil && !errors.Is(err, sql.ErrNoRows) { return false, err } if change.Source == domain.ListBlockCollection { bootstrap, err := q.ListBootstrap(ctx, change.DID) if err != nil && !errors.Is(err, sql.ErrNoRows) { return false, err } version = max(version, bootstrap) } account, err := q.SoftAccount(ctx, change.DID) if err != nil && !errors.Is(err, sql.ErrNoRows) { return false, err } version = max(version, account.BootstrapUs) if account.Active == 0 { version = max(version, account.AccountUs) } if change.Sequence > 0 && change.TimeUS == version { // Snapshot watermarks still override archive records with the same timestamp. if change.TimeUS <= account.BootstrapUs { return false, nil } if change.Source == domain.ListBlockCollection { bootstrap, err := q.ListBootstrap(ctx, change.DID) if err != nil && !errors.Is(err, sql.ErrNoRows) { return false, err } if change.TimeUS <= bootstrap { return false, nil } } return true, nil } return change.TimeUS > version, nil}
func (s *SQLite) NeedsList(ctx context.Context, change domain.Change) (bool, error) { if !domain.SoftLists(s.softMode) { return false, nil } wanted, err := s.listSubscriber(ctx, s.queries, change.DID) if err != nil || !wanted { return false, err } return listFresh(ctx, s.queries, change)}
func putList(ctx context.Context, q *db.Queries, r domain.ListRecord) error { if err := domain.ValidateList(r); err != nil { return err } if err := joinSoft(ctx, q, r.Author); err != nil { return err } if r.Subject != "" { if err := q.PutIdentity(ctx, r.Subject); err != nil { return err } } return q.PutListRecord(ctx, db.PutListRecordParams{Author: r.Author, Source: r.Source, RecordKey: r.RecordKey, ListUri: r.ListURI, Subject: r.Subject, Purpose: r.Purpose, CreatedAt: r.CreatedAt})}
// Shared list snapshots retain unchanged records so that unrelated paths keep their caches.func replaceList(ctx context.Context, q *db.Queries, data domain.ListData) error { owner, _, err := domain.ListOwner(data.URI) if err != nil { return err } if data.TimeUS <= 0 { return errors.New("list snapshot timestamp is required") } previous, err := q.ListURIRecords(ctx, data.URI) if err != nil { return err } keys := make(map[[2]string]struct{}, len(data.Records)) for _, r := range data.Records { if r.Author != owner || r.ListURI != data.URI || r.Source == domain.ListBlockCollection { return errors.New("list snapshot record does not match its owner or URI") } key := [2]string{r.Source, r.RecordKey} if _, exists := keys[key]; exists { return errors.New("duplicate list snapshot record") } keys[key] = struct{}{} if err := putList(ctx, q, r); err != nil { return err } } for _, r := range previous { if _, exists := keys[[2]string{r.Source, r.RecordKey}]; exists { continue } if err := q.DeleteListRecord(ctx, db.DeleteListRecordParams{Author: r.Author, Source: r.Source, RecordKey: r.RecordKey}); err != nil { return err } } if err := joinSoft(ctx, q, owner); err != nil { return err } return q.PutListWatermark(ctx, db.PutListWatermarkParams{ListUri: data.URI, Owner: owner, BootstrapUs: data.TimeUS})}
func replaceLists(ctx context.Context, q *db.Queries, did string, snapshot domain.ListSnapshot, at time.Time) error { if err := joinSoft(ctx, q, did); err != nil { return err } previous, err := q.ListAuthorRecords(ctx, db.ListAuthorRecordsParams{Author: did, Source: domain.ListBlockCollection}) if err != nil { return err } keys := make(map[string]struct{}, len(snapshot.Blocks)) needed := make(map[string]struct{}, len(snapshot.Blocks)) for _, r := range snapshot.Blocks { if r.Author != did || r.Source != domain.ListBlockCollection { return errors.New("listblock snapshot author or source mismatch") } if _, exists := keys[r.RecordKey]; exists { return errors.New("duplicate listblock snapshot record") } keys[r.RecordKey] = struct{}{} needed[r.ListURI] = struct{}{} if err := putList(ctx, q, r); err != nil { return err } } for _, r := range previous { if _, exists := keys[r.RecordKey]; exists { continue } if err := q.DeleteListRecord(ctx, db.DeleteListRecordParams{Author: did, Source: r.Source, RecordKey: r.RecordKey}); err != nil { return err } } for _, data := range snapshot.Lists { if _, exists := needed[data.URI]; !exists { return errors.New("unexpected or duplicate list snapshot") } if err := replaceList(ctx, q, data); err != nil { return err } delete(needed, data.URI) } if len(needed) != 0 { return errors.New("incomplete blocked list snapshot") } return q.PutListBootstrap(ctx, db.PutListBootstrapParams{Author: did, BootstrapUs: at.UnixMicro()})}
func (s *SQLite) applyList(ctx context.Context, q *db.Queries, change domain.Change) error { if change.Kind == domain.AccountChange { return s.listAccount(ctx, q, change) } if change.Source == domain.ListBlockCollection { wanted, err := s.listSubscriber(ctx, q, change.DID) if err != nil || !wanted { return err } } else { wanted, err := q.WantsListOwner(ctx, change.DID) if err != nil || !wanted && !(change.Sequence > 0 && domain.SoftAll(s.softMode)) { return err } } fresh, err := listFresh(ctx, q, change) if err != nil || !fresh { return err } old, err := q.ListRecord(ctx, db.ListRecordParams{Author: change.DID, Source: change.Source, RecordKey: change.RecordKey}) if err != nil && !errors.Is(err, sql.ErrNoRows) { return err } uri := old.ListUri applyNew := true if change.ListRecord != nil { r := *change.ListRecord if r.Author != change.DID || r.Source != change.Source || r.RecordKey != change.RecordKey { return errors.New("list event record mismatch") } if err := domain.ValidateList(r); err != nil { return err } uri = r.ListURI } if change.Source != domain.ListBlockCollection { if old.ListUri != "" { stamp, err := q.ListWatermark(ctx, old.ListUri) if err != nil && !errors.Is(err, sql.ErrNoRows) { return err } if change.TimeUS <= stamp { return nil } } stamp, err := q.ListWatermark(ctx, uri) if err != nil && !errors.Is(err, sql.ErrNoRows) { return err } applyNew = change.TimeUS > stamp wanted, err := q.WantsListURI(ctx, uri) if err != nil { return err } // Retain unknown deletions as tombstones, but do not store unrelated list contents. applyNew = applyNew && (wanted || change.Sequence > 0 && domain.SoftAll(s.softMode)) } if change.RecordKey == "" { return errors.New("list event key is required") } if err := joinSoft(ctx, q, change.DID); err != nil { return err } if err := q.DeleteListRecord(ctx, db.DeleteListRecordParams{Author: change.DID, Source: change.Source, RecordKey: change.RecordKey}); err != nil { return err } if change.ListRecord != nil && applyNew { if err := putList(ctx, q, *change.ListRecord); err != nil { return err } } if change.ListData != nil { if change.ListRecord == nil || change.ListData.URI != change.ListRecord.ListURI { return errors.New("list hydration URI mismatch") } if err := replaceList(ctx, q, *change.ListData); err != nil { return err } } return q.PutListVersion(ctx, db.PutListVersionParams{Author: change.DID, Source: change.Source, RecordKey: change.RecordKey, TimeUs: change.TimeUS})}
func (s *SQLite) listAccount(ctx context.Context, q *db.Queries, change domain.Change) error { a, err := q.SoftAccount(ctx, change.DID) if errors.Is(err, sql.ErrNoRows) { return nil } if err != nil { return err } if oldEvent(change, a.AccountUs) || change.TimeUS <= a.BootstrapUs { return nil } if !change.Active && change.Status == domain.DeletedAccount { if err := q.ClearListAuthor(ctx, change.DID); err != nil { return err } if err := q.ClearSoft(ctx, change.DID); err != nil { return err } if err := q.ClearSoftVersions(ctx, change.DID); err != nil { return err } // Retain the deletion watermark so that old membership events cannot restore records. return q.SoftDeleted(ctx, db.SoftDeletedParams{Did: change.DID, AccountUs: change.TimeUS, BootstrapUs: change.TimeUS}) } active := int64(0) if change.Active { active = 1 } return q.SoftAccountState(ctx, db.SoftAccountStateParams{Did: change.DID, Active: active, AccountUs: change.TimeUS})}