diff --git a/README.md b/README.md index 8bf4763..cb33e24 100644 --- a/README.md +++ b/README.md @@ -103,6 +103,7 @@ Keep the service key and encryption key outside the repository. | `PROVENANCE_TRUST_CACHE_TTL` | `24h` | Positive Go duration for cached trust results and paths | | `ENABLE_SOFT_VOUCH` | `false` | Use participant Bluesky follows and blocks in trust and reputation | | `ENABLE_SOFT_VOUCH_ALL_AUTHORS` | `false` | Index soft assertions from all observed stream authors. Requires `ENABLE_SOFT_VOUCH=true` | +| `ENABLE_SOFT_VOUCH_LISTS` | `false` | Use public blocked lists as soft blocks. Has no effect without `ENABLE_SOFT_VOUCH=true` | ### Soft vouches @@ -132,6 +133,30 @@ CAUTION: All-author mode can greatly increase database size and stream traffic. Graph limits still apply. Large graphs return resource-limit errors rather than partial scores. NixOS and native development services expose `enableSoftVouch` and `softVouchAllAuthors`, both false by default. +#### Blocked lists + +With `ENABLE_SOFT_VOUCH_LISTS=true`, each member of a public blocked moderation list contributes `-0.5` from the subscriber. +The flag has no effect when `ENABLE_SOFT_VOUCH=false`. Positive lists and private list mutes remain outside this feature. +Individual follow records already cover accounts that a user follows through a list's bulk-follow action. +Explicit vouches still take precedence. Overlapping lists and individual soft blocks never multiply a pair's contribution. +Blocked-list assertions apply only to default queries and topic fallback queries. +Participant mode imports public listblock records and their referenced lists at bootstrap or explicit sync. +All-author mode also accepts observed listblock events from nonparticipants. +List owners do not need to participate. Their list definitions and memberships come from their current PDS. +Snapshots fetch each owner's list collections once and reject incomplete pages or forged repository URIs. +Each collection permits at most 10,000 records. A participant snapshot permits at most 100 distinct blocked lists. +If an owner exceeds these limits or its PDS fails, the snapshot fails without replacing the index. +Stream updates cover listblock records, list definitions, and memberships, including missing-payload hydration. +The index retains referenced lists and replay tombstones after subscribers unblock them. No automatic cleanup exists. +List-enabled modes have separate cache keys and stream cursors. +Selected path dependencies include the subscription, list definition, and membership record. +Path responses include `source: "app.bsky.graph.listblock"` and `list_uri`. The UI shows the blocked-list source. +The existing TTL policy still applies to new members, new subscriptions, and account status changes. +NixOS and native development services expose `softVouchLists`, false by default. + +CAUTION: List owners can change members and therefore change subscribers' trust graphs. +List subscriptions add stream traffic, database records, and PDS requests. Replay gaps still require explicit participant sync. + ## API ### Trust algorithms diff --git a/docs/security-review.md b/docs/security-review.md index 95719b3..1df00cd 100644 --- a/docs/security-review.md +++ b/docs/security-review.md @@ -37,6 +37,9 @@ - Soft sources have fixed fractional scores. Explicit writes reject fractional scores. - Soft account state and records remain separate from participant enrollment and private thresholds. - Cache dependencies include collections so that identical record keys cannot invalidate unrelated sources. +- Blocked-list imports require DID-based list URIs and repository-bound definitions and membership records. +- List members depend on their public subscription, moderation-list definition, and membership records, not private mute state. +- Participant snapshots commit list imports and vouches together. Failed imports preserve the index. ## Correctness @@ -83,6 +86,9 @@ Monitor cursor lag, disconnect logs, and database growth. All-author soft ingestion has unbounded total record volume. Per-query graph limits do not bound database size. Nonparticipant soft records come only from observed events. The index does not reconstruct their complete repositories after replay gaps. Participant sync recovers soft records after long outages or disabled periods. +Blocked-list owners control membership and can change subscribers' calculated trust. A subscription delegates this part of the graph to its owner. +Blocked-list imports have bounded record counts, but total indexed list data and retained replay tombstones have no automatic cleanup. +An unavailable list owner's PDS can delay a new subscription event and its stream cursor until recovery. Run only one appview ingestor per database. Repository operations serialize within one UI process. Multiple UI replicas need shared refresh coordination. Live PDS OAuth interoperability still needs an account-based smoke test. diff --git a/internal/appview/http.go b/internal/appview/http.go index a91dc44..9fdc163 100644 --- a/internal/appview/http.go +++ b/internal/appview/http.go @@ -64,7 +64,8 @@ func Handler(service *Service, key string) http.Handler { document := apiDocument(endpoints, descriptor) document["x-provenance-reputation-algorithm"] = service.reputation.Describe().Name document["x-provenance-soft-vouch-enabled"] = service.softMode != domain.SoftDisabled - document["x-provenance-soft-vouch-all-authors"] = service.softMode == domain.SoftAllAuthors + document["x-provenance-soft-vouch-all-authors"] = domain.SoftAll(service.softMode) + document["x-provenance-soft-vouch-lists"] = domain.SoftLists(service.softMode) mux.HandleFunc("GET /v1/openapi.json", func(w http.ResponseWriter, r *http.Request) { httpx.JSON(w, http.StatusOK, document) }) return mux } diff --git a/internal/appview/lists.go b/internal/appview/lists.go new file mode 100644 index 0000000..497a4e0 --- /dev/null +++ b/internal/appview/lists.go @@ -0,0 +1,11 @@ +package appview + +import ( + "context" + "provenance/internal/domain" +) + +type ListRepository interface { + ListSnapshot(context.Context, string) (domain.ListSnapshot, error) + ListMembers(context.Context, string) (domain.ListData, error) +} diff --git a/internal/appview/lists_test.go b/internal/appview/lists_test.go new file mode 100644 index 0000000..2f8a828 --- /dev/null +++ b/internal/appview/lists_test.go @@ -0,0 +1,90 @@ +package appview + +import ( + "context" + "errors" + "path/filepath" + "provenance/internal/domain" + "provenance/internal/storage" + "provenance/internal/trust" + "provenance/internal/trust/algorithms" + "testing" + "time" +) + +type listRepository struct { + repository + snapshot domain.ListSnapshot + err error + reads int +} + +func (r *listRepository) ListSnapshot(context.Context, string) (domain.ListSnapshot, error) { + return r.snapshot, r.err +} +func (r *listRepository) ListMembers(context.Context, string) (domain.ListData, error) { + r.reads++ + return r.snapshot.Lists[0], r.err +} + +func TestListService(t *testing.T) { + ctx := context.Background() + const owner = "did:plc:listowner" + const uri = "at://" + owner + "/app.bsky.graph.list/mod" + const date = "2026-10-09T00:00:00Z" + for _, algorithm := range []trust.Algorithm{algorithms.Optimistic{}, algorithms.ShortestPath{}} { + t.Run(algorithm.Describe().Name, func(t *testing.T) { + store, err := storage.OpenWithSoft(filepath.Join(t.TempDir(), "lists.db"), domain.SoftParticipantLists) + if err != nil { + t.Fatal(err) + } + defer store.Close() + repo := &listRepository{snapshot: domain.ListSnapshot{Blocks: []domain.ListRecord{{Author: alice, Source: domain.ListBlockCollection, RecordKey: "block", ListURI: uri, CreatedAt: date}}, Lists: []domain.ListData{{URI: uri, TimeUS: 100, Records: []domain.ListRecord{{Author: owner, Source: domain.ListCollection, RecordKey: "mod", ListURI: uri, Purpose: domain.ModerationList, CreatedAt: date}, {Author: owner, Source: domain.ListItemCollection, RecordKey: "member", ListURI: uri, Subject: bob, CreatedAt: date}}}}}} + duplicate := repo.snapshot.Blocks[0] + duplicate.RecordKey = "block-2" + repo.snapshot.Blocks = append(repo.snapshot.Blocks, duplicate) + s := NewWithOptions(store, repo, trust.Limits{States: 100, Edges: 100, Penalty: .1}, Options{Trust: algorithm, SoftMode: domain.SoftParticipantLists}) + if err := s.Join(ctx, alice); err != nil { + t.Fatal(err) + } + got, err := s.Reputation(ctx, alice, bob, "", domain.Default) + if err != nil || got.Status != trust.CalculatedStatus || got.Score == nil || *got.Score != -.5 || got.PathScores[1].ListURI != uri || got.PathScores[1].Source != domain.ListBlockCollection { + t.Fatalf("blocked list trust: %+v %v", got, err) + } + rep, err := s.RelativeReputation(ctx, alice, bob, "", domain.Default, 1) + if err != nil || rep.Score != -.5 || rep.MeanScore != -.5 { + t.Fatalf("blocked list reputation: %+v %v", rep, err) + } + wanted, err := s.wanted(ctx, owner, time.Now().UnixMicro()) + if err != nil || !wanted { + t.Fatalf("nonparticipant list owner: %v %v", wanted, err) + } + stamp := time.Now().Add(time.Second).UnixMicro() + if err := s.Apply(ctx, domain.Change{Kind: domain.RecordChange, DID: owner, Source: domain.ListItemCollection, RecordKey: "member", TimeUS: stamp}); err != nil { + t.Fatal(err) + } + got, err = s.Reputation(ctx, alice, bob, "", domain.Default) + if err != nil || got.Status != trust.UnknownStatus { + t.Fatalf("stale list path: %+v %v", got, err) + } + repo.err = errors.New("owner unavailable") + block := repo.snapshot.Blocks[0] + err = s.Apply(ctx, domain.Change{Kind: domain.RecordChange, DID: alice, Source: block.Source, RecordKey: block.RecordKey, ListRecord: &block, TimeUS: stamp + 1}) + if !errors.Is(err, repo.err) { + t.Fatalf("hydration error ignored: %v", err) + } + repo.err = nil + repo.snapshot.Lists[0].TimeUS = stamp + 2 + if err := s.Apply(ctx, domain.Change{Kind: domain.RecordChange, DID: alice, Source: block.Source, RecordKey: block.RecordKey, ListRecord: &block, TimeUS: stamp + 3}); err != nil { + t.Fatal(err) + } + reads := repo.reads + if err := s.Apply(ctx, domain.Change{Kind: domain.RecordChange, DID: alice, Source: block.Source, RecordKey: block.RecordKey, ListRecord: &block, TimeUS: stamp + 3}); err != nil { + t.Fatal(err) + } + if repo.reads != reads { + t.Fatal("replay fetched list again") + } + }) + } +} diff --git a/internal/appview/openapi.go b/internal/appview/openapi.go index 3645a55..5d69f55 100644 --- a/internal/appview/openapi.go +++ b/internal/appview/openapi.go @@ -155,7 +155,7 @@ func apiSchemas() map[string]any { "mean_score": map[string]any{"type": "number", "description": "Arithmetic mean of qualifying non-block vouch scores, including zeros. An empty set returns 0."}, }, "score", "absolute_score", "mean_score"), "Evaluation": objectSchema(evaluation, "status", "score", "trusted"), - "PathScore": objectSchema(map[string]any{"did": stringSchema, "assertion": map[string]any{"type": "number", "nullable": true, "minimum": domain.Blocked, "maximum": domain.MaxScore}, "score": score, "comment": comment, "source": map[string]any{"type": "string", "enum": []string{domain.Collection, domain.FollowCollection, domain.BlockCollection}}}, "did", "assertion", "score"), + "PathScore": objectSchema(map[string]any{"did": stringSchema, "assertion": map[string]any{"type": "number", "nullable": true, "minimum": domain.Blocked, "maximum": domain.MaxScore}, "score": score, "comment": comment, "list_uri": stringSchema, "source": map[string]any{"type": "string", "enum": []string{domain.Collection, domain.FollowCollection, domain.BlockCollection, domain.ListBlockCollection}}}, "did", "assertion", "score"), "Error": objectSchema(map[string]any{"error": stringSchema}, "error"), "Vouch": objectSchema(map[string]any{"author": stringSchema, "subject": stringSchema, "score": map[string]any{"type": "integer", "minimum": domain.Blocked, "maximum": domain.MaxScore}, "tag": stringSchema, "createdAt": stringSchema, "comment": comment, "source": stringSchema, "record_key": stringSchema}, "author", "subject", "score", "tag"), "Participant": objectSchema(map[string]any{"did": stringSchema, "threshold": scalarSchema("number"), "active": scalarSchema("boolean"), "syncedAt": stringSchema, "bootstrapUs": scalarSchema("integer")}, "did", "threshold", "active", "syncedAt", "bootstrapUs"), diff --git a/internal/appview/service.go b/internal/appview/service.go index 3be9cc4..e3b2072 100644 --- a/internal/appview/service.go +++ b/internal/appview/service.go @@ -22,15 +22,17 @@ import ( ) const ( - softParticipantStream = "jetstream-soft-participants" - softAllStream = "jetstream-soft-all" - StreamSource = "jetstream" - InitialRetryDelay = time.Second - MaxRetryDelay = 30 * time.Second - ParticipantLimit = 10000 - TagLimit = 20 - VouchPageLimit = 100 - identityTimeout = 5 * time.Second + 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 ) type Repository interface { @@ -148,10 +150,12 @@ func pathComments(result trust.Result, records map[[2]string]domain.Vouch) trust result.PathScores = slices.Clone(result.PathScores) result.PathScores[0].Comment = "" result.PathScores[0].Source = "" + result.PathScores[0].ListURI = "" for i := 1; i < len(result.PathScores); i++ { pair := [2]string{result.PathScores[i-1].DID, result.PathScores[i].DID} result.PathScores[i].Comment = records[pair].Comment result.PathScores[i].Source = records[pair].Source + result.PathScores[i].ListURI = records[pair].ListURI } return result } @@ -291,6 +295,17 @@ func (s *Service) bootstrap(ctx context.Context, did string) error { 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) } @@ -335,11 +350,34 @@ func (s *Service) BootstrapPending(ctx context.Context) error { func (s *Service) Apply(ctx context.Context, change domain.Change) error { s.syncMu.Lock() defer s.syncMu.Unlock() + if domain.SoftLists(s.softMode) && change.Source == domain.ListBlockCollection && change.ListRecord != nil { + 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 s.store.Apply(ctx, s.streamSource(), change) } // 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 } @@ -350,6 +388,12 @@ func (s *Service) streamSource() string { } 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 diff --git a/internal/appview/trust_cache.go b/internal/appview/trust_cache.go index d5e6caa..886cd09 100644 --- a/internal/appview/trust_cache.go +++ b/internal/appview/trust_cache.go @@ -138,6 +138,11 @@ func pathDependencies(result trust.Result, records map[[2]string]domain.Vouch) [ key = domain.RecordKey(edge.Subject, edge.Tag) } dependencies = append(dependencies, trust.Dependency{Author: edge.Author, Source: source, RecordKey: key}) + if source == domain.ListBlockCollection { + dependencies = append(dependencies, + trust.Dependency{Author: edge.ListAuthor, Source: domain.ListCollection, RecordKey: edge.ListKey}, + trust.Dependency{Author: edge.ListAuthor, Source: domain.ListItemCollection, RecordKey: edge.ItemKey}) + } } return dependencies } diff --git a/internal/atproto/lists.go b/internal/atproto/lists.go new file mode 100644 index 0000000..0e644b5 --- /dev/null +++ b/internal/atproto/lists.go @@ -0,0 +1,182 @@ +package atproto + +import ( + "context" + "errors" + "fmt" + "github.com/bluesky-social/indigo/atproto/syntax" + "net/url" + "provenance/internal/domain" + "strings" + "time" +) + +type ListRecord struct { + Type string `json:"$type"` + Subject string `json:"subject"` + List string `json:"list"` + Purpose string `json:"purpose"` + CreatedAt string `json:"createdAt"` +} + +const maxBlockedLists = 100 + +type listRead struct { + records map[string]domain.ListData + timeUS int64 +} + +func DecodeList(record ListRecord, author, source, key string) (domain.ListRecord, error) { + if record.Type != source { + return domain.ListRecord{}, errors.New("list record type mismatch") + } + r := domain.ListRecord{Author: author, Source: source, RecordKey: key, ListURI: record.List, Subject: record.Subject, Purpose: record.Purpose, CreatedAt: record.CreatedAt} + if source == domain.ListBlockCollection { + r.ListURI, r.Subject = record.Subject, "" + } + if source == domain.ListCollection { + r.ListURI = "at://" + author + "/" + source + "/" + key + } + return r, domain.ValidateList(r) +} + +func (r *Repository) listRecords(ctx context.Context, did, source string) ([]domain.ListRecord, error) { + host, err := r.pdsHost(ctx, did) + if err != nil { + return nil, err + } + result := []domain.ListRecord{} + cursor := "" + cursors, keys := map[string]struct{}{}, map[string]struct{}{} + examined := 0 + for { + params := url.Values{"repo": {did}, "collection": {source}, "limit": {fmt.Sprint(PageSize)}} + if cursor != "" { + params.Set("cursor", cursor) + } + var page struct { + Cursor string `json:"cursor"` + Records []struct { + URI string `json:"uri"` + Value ListRecord `json:"value"` + } `json:"records"` + } + if err := getJSON(ctx, r.client, strings.TrimRight(host, "/")+"/xrpc/com.atproto.repo.listRecords?"+params.Encode(), &page); err != nil { + return nil, err + } + if page.Records == nil || len(page.Records) > PageSize { + return nil, errors.New("invalid list repository page") + } + examined += len(page.Records) + if examined > MaxRecords || len(cursors) > MaxRecords/PageSize { + return nil, errors.New("list record limit exceeded") + } + for _, item := range page.Records { + uri, err := syntax.ParseATURI(item.URI) + if err != nil || uri.Authority().String() != did || uri.Collection().String() != source { + return nil, errors.New("list record URI does not match its repository") + } + key := uri.RecordKey().String() + if _, err := syntax.ParseRecordKey(key); err != nil { + return nil, err + } + if _, exists := keys[key]; exists { + return nil, errors.New("duplicate list repository record") + } + keys[key] = struct{}{} + r, err := DecodeList(item.Value, did, source, key) + if err != nil { + continue + } + result = append(result, r) + } + if page.Cursor == "" { + return result, nil + } + if _, exists := cursors[page.Cursor]; exists { + return nil, errors.New("repeated list repository cursor") + } + cursors[page.Cursor] = struct{}{} + cursor = page.Cursor + } +} + +func (r *Repository) ListSnapshot(ctx context.Context, did string) (domain.ListSnapshot, error) { + if !domain.SoftLists(r.softMode) { + return domain.ListSnapshot{}, errors.New("soft lists are disabled") + } + blocks, err := r.listRecords(ctx, did, domain.ListBlockCollection) + if err != nil { + return domain.ListSnapshot{}, err + } + snapshot := domain.ListSnapshot{Blocks: blocks} + seen := map[string]struct{}{} + owners := map[string]listRead{} + for _, block := range blocks { + if _, exists := seen[block.ListURI]; exists { + continue + } + seen[block.ListURI] = struct{}{} + if len(seen) > maxBlockedLists { + return domain.ListSnapshot{}, errors.New("blocked list limit exceeded") + } + owner, _, err := domain.ListOwner(block.ListURI) + if err != nil { + return domain.ListSnapshot{}, err + } + data, exists := owners[owner] + if !exists { + data, err = r.ownerLists(ctx, owner) + if err != nil { + return domain.ListSnapshot{}, err + } + owners[owner] = data + } + list := data.records[block.ListURI] + list.URI = block.ListURI + if list.TimeUS == 0 { + list.TimeUS = data.timeUS + } + snapshot.Lists = append(snapshot.Lists, list) + } + return snapshot, nil +} + +func (r *Repository) ListMembers(ctx context.Context, uri string) (domain.ListData, error) { + if !domain.SoftLists(r.softMode) { + return domain.ListData{}, errors.New("soft lists are disabled") + } + owner, _, err := domain.ListOwner(uri) + if err != nil { + return domain.ListData{}, err + } + lists, err := r.ownerLists(ctx, owner) + if err != nil { + return domain.ListData{}, err + } + data := lists.records[uri] + data.URI = uri + if data.TimeUS == 0 { + data.TimeUS = lists.timeUS + } + return data, nil +} + +// Fetch each owner's bounded collections once, even when several blocked lists share an owner. +func (r *Repository) ownerLists(ctx context.Context, owner string) (listRead, error) { + stamp := time.Now().UnixMicro() + lists := map[string]domain.ListData{} + for _, source := range []string{domain.ListCollection, domain.ListItemCollection} { + records, err := r.listRecords(ctx, owner, source) + if err != nil { + return listRead{}, err + } + for _, record := range records { + data := lists[record.ListURI] + data.URI, data.TimeUS = record.ListURI, stamp + data.Records = append(data.Records, record) + lists[record.ListURI] = data + } + } + return listRead{records: lists, timeUS: stamp}, nil +} diff --git a/internal/atproto/lists_test.go b/internal/atproto/lists_test.go new file mode 100644 index 0000000..ba23d01 --- /dev/null +++ b/internal/atproto/lists_test.go @@ -0,0 +1,56 @@ +package atproto + +import ( + "context" + "net/http" + "provenance/internal/domain" + "testing" +) + +func TestListRepository(t *testing.T) { + const uri = "at://" + testDID + "/app.bsky.graph.list/mod" + calls := map[string]int{} + repo := &Repository{directory: directory{}, softMode: domain.SoftParticipantLists, client: &http.Client{Transport: transport(func(r *http.Request) (*http.Response, error) { + source := r.URL.Query().Get("collection") + calls[source]++ + key := "record" + record := ListRecord{Type: source, CreatedAt: "2026-10-09T00:00:00Z"} + switch source { + case domain.ListBlockCollection: + record.Subject = uri + case domain.ListCollection: + key, record.Purpose = "mod", domain.ModerationList + case domain.ListItemCollection: + record.List, record.Subject = uri, "did:plc:bob" + } + return response(map[string]any{"records": []any{map[string]any{"uri": "at://" + testDID + "/" + source + "/" + key, "value": record}}}), nil + })}} + got, err := repo.ListSnapshot(context.Background(), testDID) + if err != nil || len(got.Blocks) != 1 || len(got.Lists) != 1 || len(got.Lists[0].Records) != 2 { + t.Fatalf("list snapshot: %+v %v", got, err) + } + if calls[domain.ListBlockCollection] != 1 || calls[domain.ListCollection] != 1 || calls[domain.ListItemCollection] != 1 { + t.Fatalf("fetches: %+v", calls) + } + repo.softMode = domain.SoftParticipants + if _, err := repo.ListSnapshot(context.Background(), testDID); err == nil { + t.Fatal("disabled list fetch") + } + if calls[domain.ListBlockCollection] != 1 { + t.Fatal("disabled list request") + } +} + +func TestListRecordValidation(t *testing.T) { + const uri = "at://" + testDID + "/app.bsky.graph.list/mod" + for _, record := range []ListRecord{ + {Type: domain.ListBlockCollection, Subject: "https://example.com/list", CreatedAt: "2026-10-09T00:00:00Z"}, + {Type: domain.ListItemCollection, List: "at://did:plc:forged/app.bsky.graph.list/mod", Subject: "did:plc:bob", CreatedAt: "2026-10-09T00:00:00Z"}, + {Type: domain.ListItemCollection, List: uri, Subject: "invalid", CreatedAt: "2026-10-09T00:00:00Z"}, + {Type: domain.ListBlockCollection, Subject: uri, CreatedAt: "invalid"}, + } { + if _, err := DecodeList(record, testDID, record.Type, "record"); err == nil { + t.Fatalf("accepted invalid record: %+v", record) + } + } +} diff --git a/internal/atproto/records.go b/internal/atproto/records.go index 2552587..dd99238 100644 --- a/internal/atproto/records.go +++ b/internal/atproto/records.go @@ -97,25 +97,37 @@ func ValidateRecord(record Record) error { // Snapshot reads only the DID's current PDS and rejects incomplete repository pages. func (r *Repository) Snapshot(ctx context.Context, did string) ([]domain.Vouch, error) { - parsed, err := syntax.ParseDID(did) + host, err := r.pdsHost(ctx, did) if err != nil { return nil, err } + return r.snapshot(ctx, did, host) +} + +func (r *Repository) pdsHost(ctx context.Context, did string) (string, error) { + parsed, err := syntax.ParseDID(did) + if err != nil { + return "", err + } identifier, err := syntax.ParseAtIdentifier(did) if err != nil { - return nil, err + return "", err } if err := r.directory.Purge(ctx, identifier); err != nil { - return nil, err + return "", err } identity, err := r.directory.LookupDID(ctx, parsed) if err != nil { - return nil, err + return "", err } host := identity.PDSEndpoint() if !strings.HasPrefix(host, "https://") { - return nil, errors.New("PDS must use HTTPS") + return "", errors.New("PDS must use HTTPS") } + return host, nil +} + +func (r *Repository) snapshot(ctx context.Context, did, host string) ([]domain.Vouch, error) { var result []domain.Vouch cursor := "" seenCursors := map[string]bool{} diff --git a/internal/config/config.go b/internal/config/config.go index 69d5f53..75aa682 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -81,6 +81,16 @@ func LoadAppview() (Appview, error) { if all { softMode = domain.SoftAllAuthors } + lists, err := strconv.ParseBool(value("ENABLE_SOFT_VOUCH_LISTS", "false")) + if err != nil { + return Appview{}, errors.New("ENABLE_SOFT_VOUCH_LISTS must be a boolean") + } + if lists && soft { + softMode = domain.SoftParticipantLists + if all { + softMode = domain.SoftAllAuthorLists + } + } return Appview{ Address: value("PROVENANCE_APPVIEW_ADDR", defaultAppviewAddress), Database: value("PROVENANCE_APPVIEW_DB", "data/appview.db"), ServiceKey: serviceKey, JetstreamURL: value("PROVENANCE_JETSTREAM_URL", microcosm.DefaultJetstreamURL), SlingshotURL: value("PROVENANCE_SLINGSHOT_URL", microcosm.DefaultSlingshotURL), diff --git a/internal/config/config_test.go b/internal/config/config_test.go index b959868..9808c85 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -73,6 +73,7 @@ func TestTrustCacheTTL(t *testing.T) { } func TestSoftFlags(t *testing.T) { + t.Setenv("ENABLE_SOFT_VOUCH_LISTS", "") t.Setenv("PROVENANCE_SERVICE_KEY", strings.Repeat("x", MinServiceKeyLength)) t.Setenv("PROVENANCE_HOP_PENALTY", "") t.Setenv("PROVENANCE_TRUST_CACHE_TTL", "") @@ -95,3 +96,25 @@ func TestSoftFlags(t *testing.T) { } } } + +func TestSoftListFlags(t *testing.T) { + t.Setenv("PROVENANCE_SERVICE_KEY", strings.Repeat("x", MinServiceKeyLength)) + t.Setenv("PROVENANCE_HOP_PENALTY", "") + t.Setenv("PROVENANCE_TRUST_CACHE_TTL", "") + t.Setenv("ENABLE_SOFT_VOUCH_LISTS", "true") + for _, test := range []struct { + soft, all string + mode domain.SoftMode + }{{"false", "false", domain.SoftDisabled}, {"true", "false", domain.SoftParticipantLists}, {"true", "true", domain.SoftAllAuthorLists}} { + t.Setenv("ENABLE_SOFT_VOUCH", test.soft) + t.Setenv("ENABLE_SOFT_VOUCH_ALL_AUTHORS", test.all) + cfg, err := LoadAppview() + if err != nil || cfg.SoftMode != test.mode { + t.Fatalf("list flags: %+v %v", cfg, err) + } + } + t.Setenv("ENABLE_SOFT_VOUCH_LISTS", "invalid") + if _, err := LoadAppview(); err == nil { + t.Fatal("invalid list flag accepted") + } +} diff --git a/internal/domain/change.go b/internal/domain/change.go index 8098532..0c8be94 100644 --- a/internal/domain/change.go +++ b/internal/domain/change.go @@ -9,14 +9,16 @@ const ( ) type Change struct { - Kind ChangeKind - DID string - TimeUS int64 - RecordKey string - Source string - Vouch *Vouch - Active bool - Status string + Kind ChangeKind + DID string + TimeUS int64 + RecordKey string + Source string + ListRecord *ListRecord + ListData *ListData + Vouch *Vouch + Active bool + Status string } const DeletedAccount = "deleted" diff --git a/internal/domain/lists.go b/internal/domain/lists.go new file mode 100644 index 0000000..54ef7ad --- /dev/null +++ b/internal/domain/lists.go @@ -0,0 +1,78 @@ +package domain + +import ( + "errors" + "github.com/bluesky-social/indigo/atproto/syntax" + "time" +) + +type ListRecord struct { + Author string + Source string + RecordKey string + ListURI string + Subject string + Purpose string + CreatedAt string +} + +type ListData struct { + URI string + Records []ListRecord + TimeUS int64 +} + +type ListSnapshot struct { + Blocks []ListRecord + Lists []ListData +} + +func ListOwner(uri string) (string, string, error) { + parsed, err := syntax.ParseATURI(uri) + if err != nil { + return "", "", err + } + owner, key := parsed.Authority().String(), parsed.RecordKey().String() + if _, err := syntax.ParseDID(owner); err != nil { + return "", "", err + } + if parsed.Collection().String() != ListCollection { + return "", "", errors.New("invalid list collection") + } + if _, err := syntax.ParseRecordKey(key); err != nil { + return "", "", err + } + if uri != "at://"+owner+"/"+ListCollection+"/"+key { + return "", "", errors.New("invalid list URI") + } + return owner, key, nil +} + +func ValidateList(r ListRecord) error { + if _, err := syntax.ParseDID(r.Author); err != nil { + return err + } + if _, err := syntax.ParseRecordKey(r.RecordKey); err != nil { + return err + } + owner, key, err := ListOwner(r.ListURI) + if err != nil { + return err + } + if !ListSource(r.Source) { + return errors.New("invalid list source") + } + if r.Source != ListBlockCollection && r.Author != owner { + return errors.New("list record author is not its owner") + } + if r.Source == ListCollection && r.RecordKey != key { + return errors.New("list key does not match its URI") + } + if r.Source == ListItemCollection { + if _, err := syntax.ParseDID(r.Subject); err != nil { + return err + } + } + _, err = time.Parse(time.RFC3339Nano, r.CreatedAt) + return err +} diff --git a/internal/domain/soft.go b/internal/domain/soft.go index b4182a8..7590856 100644 --- a/internal/domain/soft.go +++ b/internal/domain/soft.go @@ -8,16 +8,30 @@ const ( SoftDisabled SoftMode = iota SoftParticipants SoftAllAuthors + SoftParticipantLists + SoftAllAuthorLists ) const ( - FollowCollection = "app.bsky.graph.follow" - BlockCollection = "app.bsky.graph.block" - FollowScore = 0.5 - SoftBlockScore = -0.5 + FollowCollection = "app.bsky.graph.follow" + BlockCollection = "app.bsky.graph.block" + FollowScore = 0.5 + SoftBlockScore = -0.5 + ListBlockCollection = "app.bsky.graph.listblock" + ListItemCollection = "app.bsky.graph.listitem" + ListCollection = "app.bsky.graph.list" + ModerationList = "app.bsky.graph.defs#modlist" ) -func SoftSource(source string) bool { return source == FollowCollection || source == BlockCollection } +func SoftSource(source string) bool { + return source == FollowCollection || source == BlockCollection || source == ListBlockCollection +} + +func ListSource(source string) bool { + return source == ListCollection || source == ListItemCollection || source == ListBlockCollection +} +func SoftLists(mode SoftMode) bool { return mode == SoftParticipantLists || mode == SoftAllAuthorLists } +func SoftAll(mode SoftMode) bool { return mode == SoftAllAuthors || mode == SoftAllAuthorLists } func SourceScore(source string) float64 { if source == FollowCollection { diff --git a/internal/domain/vouch.go b/internal/domain/vouch.go index 44ed408..26ff2f0 100644 --- a/internal/domain/vouch.go +++ b/internal/domain/vouch.go @@ -30,14 +30,18 @@ const ( ) type Vouch struct { - Author string `json:"author"` - Subject string `json:"subject"` - Tag string `json:"tag"` - Score float64 `json:"score"` - Source string `json:"source,omitempty"` - RecordKey string `json:"record_key,omitempty"` - CreatedAt string `json:"createdAt"` - Comment string `json:"comment,omitempty"` + Author string `json:"author"` + Subject string `json:"subject"` + Tag string `json:"tag"` + Score float64 `json:"score"` + Source string `json:"source,omitempty"` + RecordKey string `json:"record_key,omitempty"` + ListURI string `json:"list_uri,omitempty"` + ListAuthor string `json:"-"` + ListKey string `json:"-"` + ItemKey string `json:"-"` + CreatedAt string `json:"createdAt"` + Comment string `json:"comment,omitempty"` } func ValidateTag(tag string) error { diff --git a/internal/microcosm/client.go b/internal/microcosm/client.go index 68da2b3..4c5c95a 100644 --- a/internal/microcosm/client.go +++ b/internal/microcosm/client.go @@ -36,7 +36,7 @@ func New(streamURL, slingshotURL string) (*Client, error) { } func NewWithSoft(streamURL, slingshotURL string, mode domain.SoftMode) (*Client, error) { - if mode < domain.SoftDisabled || mode > domain.SoftAllAuthors { + if mode < domain.SoftDisabled || mode > domain.SoftAllAuthorLists { return nil, errors.New("invalid soft vouch mode") } if err := validateURL(streamURL, "wss", "ws"); err != nil { diff --git a/internal/microcosm/events.go b/internal/microcosm/events.go index c94b1b9..a732f28 100644 --- a/internal/microcosm/events.go +++ b/internal/microcosm/events.go @@ -48,7 +48,7 @@ func (c *Client) decode(ctx context.Context, data []byte, wanted func(context.Co return domain.Change{}, err } change := domain.Change{Kind: e.Kind, DID: e.DID, TimeUS: e.TimeUS} - allSoft := c.softMode == domain.SoftAllAuthors && (e.Kind == domain.AccountChange || e.Commit != nil && domain.SoftSource(e.Commit.Collection)) + allSoft := domain.SoftAll(c.softMode) && (e.Kind == domain.AccountChange || e.Commit != nil && domain.SoftSource(e.Commit.Collection)) if wanted != nil && !allSoft { participates, err := wanted(ctx, e.DID, e.TimeUS) if err != nil { @@ -74,6 +74,13 @@ func (c *Client) decode(ctx context.Context, data []byte, wanted func(context.Co return domain.Change{}, errors.New("Jetstream commit is missing") } commit := e.Commit + if domain.ListSource(commit.Collection) { + if !domain.SoftLists(c.softMode) { + change.Kind = domain.IdentityChange + return change, nil + } + return c.decodeList(ctx, e) + } if domain.SoftSource(commit.Collection) { if c.softMode == domain.SoftDisabled { change.Kind = domain.IdentityChange diff --git a/internal/microcosm/lists.go b/internal/microcosm/lists.go new file mode 100644 index 0000000..0d37c01 --- /dev/null +++ b/internal/microcosm/lists.go @@ -0,0 +1,42 @@ +package microcosm + +import ( + "context" + "encoding/json" + "errors" + "github.com/bluesky-social/indigo/atproto/syntax" + "provenance/internal/atproto" + "provenance/internal/domain" +) + +func (c *Client) decodeList(ctx context.Context, e event) (domain.Change, error) { + commit := e.Commit + change := domain.Change{Kind: domain.RecordChange, DID: e.DID, TimeUS: e.TimeUS, Source: commit.Collection, RecordKey: commit.RecordKey} + if _, err := syntax.ParseRecordKey(commit.RecordKey); err != nil { + change.Kind = domain.IdentityChange + return change, nil + } + if commit.Operation == Delete { + return change, nil + } + if commit.Operation != Create && commit.Operation != Update { + return domain.Change{}, errors.New("unsupported list operation") + } + var record atproto.ListRecord + if len(commit.Record) == 0 || string(commit.Record) == "null" { + if commit.CID == "" { + return domain.Change{}, errors.New("list event lacks its record and CID") + } + if err := c.rawRecord(ctx, e.DID, commit.Collection, commit.RecordKey, commit.CID, &record); err != nil { + return domain.Change{}, err + } + } else if err := json.Unmarshal(commit.Record, &record); err != nil { + return change, nil + } + r, err := atproto.DecodeList(record, e.DID, commit.Collection, commit.RecordKey) + if err != nil { + return change, nil + } + change.ListRecord = &r + return change, nil +} diff --git a/internal/microcosm/lists_test.go b/internal/microcosm/lists_test.go new file mode 100644 index 0000000..5299908 --- /dev/null +++ b/internal/microcosm/lists_test.go @@ -0,0 +1,70 @@ +package microcosm + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "provenance/internal/atproto" + "provenance/internal/domain" + "testing" +) + +func TestListEvents(t *testing.T) { + const uri = "at://" + author + "/app.bsky.graph.list/mod" + for _, source := range []string{domain.ListBlockCollection, domain.ListCollection, domain.ListItemCollection} { + record := atproto.ListRecord{Type: source, CreatedAt: "2026-10-09T00:00:00Z"} + key := "record" + switch source { + case domain.ListBlockCollection: + record.Subject = uri + case domain.ListCollection: + key, record.Purpose = "mod", domain.ModerationList + case domain.ListItemCollection: + record.List, record.Subject = uri, subject + } + calls := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls++ + if r.URL.Query().Get("collection") != source { + t.Error("wrong list hydration collection") + } + json.NewEncoder(w).Encode(map[string]any{"uri": "at://" + author + "/" + source + "/" + key, "cid": recordCID, "value": record}) + })) + for _, mode := range []domain.SoftMode{domain.SoftDisabled, domain.SoftParticipants, domain.SoftParticipantLists, domain.SoftAllAuthorLists} { + client, err := NewWithSoft("ws://127.0.0.1:1/subscribe", server.URL, mode) + if err != nil { + t.Fatal(err) + } + for _, operation := range []Operation{Create, Update, Delete} { + for _, payload := range []any{record, nil} { + data, _ := json.Marshal(map[string]any{"did": author, "time_us": 100, "kind": domain.RecordChange, "commit": map[string]any{"operation": operation, "collection": source, "rkey": key, "record": payload, "cid": recordCID}}) + before := calls + change, err := client.decode(context.Background(), data, nil) + if err != nil { + t.Fatal(err) + } + if !domain.SoftLists(mode) { + if change.ListRecord != nil || change.Kind != domain.IdentityChange || calls != before { + t.Fatal("disabled list ingestion or hydration") + } + continue + } + if change.Source != source || change.RecordKey != key { + t.Fatalf("list event provenance: %+v", change) + } + if operation == Delete { + if change.ListRecord != nil { + t.Fatal("list deletion retained record") + } + continue + } + if change.ListRecord == nil || change.ListRecord.ListURI != uri { + t.Fatalf("list payload: %+v", change) + } + } + } + } + server.Close() + } +} diff --git a/internal/microcosm/soft_test.go b/internal/microcosm/soft_test.go index 5cdaad1..2c179cf 100644 --- a/internal/microcosm/soft_test.go +++ b/internal/microcosm/soft_test.go @@ -86,19 +86,25 @@ func TestSoftEventScope(t *testing.T) { } func TestSoftSubscriptions(t *testing.T) { - for _, mode := range []domain.SoftMode{domain.SoftDisabled, domain.SoftParticipants, domain.SoftAllAuthors} { + for _, mode := range []domain.SoftMode{domain.SoftDisabled, domain.SoftParticipants, domain.SoftAllAuthors, domain.SoftParticipantLists, domain.SoftAllAuthorLists} { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { collections := r.URL.Query()["wantedCollections"] want := 1 if mode != domain.SoftDisabled { want = 3 } + if domain.SoftLists(mode) { + want = 6 + } if len(collections) != want || collections[0] != domain.Collection { t.Errorf("subscriptions: %v", collections) } if want == 3 && (collections[1] != domain.FollowCollection || collections[2] != domain.BlockCollection) { t.Errorf("soft subscriptions: %v", collections) } + if want == 6 && (collections[3] != domain.ListBlockCollection || collections[4] != domain.ListCollection || collections[5] != domain.ListItemCollection) { + t.Errorf("list subscriptions: %v", collections) + } conn, err := websocket.Accept(w, r, nil) if err != nil { t.Error(err) diff --git a/internal/microcosm/stream.go b/internal/microcosm/stream.go index be22b77..3ae1b7b 100644 --- a/internal/microcosm/stream.go +++ b/internal/microcosm/stream.go @@ -20,6 +20,11 @@ func (c *Client) Consume(ctx context.Context, cursor int64, wanted func(context. params.Add("wantedCollections", domain.FollowCollection) params.Add("wantedCollections", domain.BlockCollection) } + if domain.SoftLists(c.softMode) { + params.Add("wantedCollections", domain.ListBlockCollection) + params.Add("wantedCollections", domain.ListCollection) + params.Add("wantedCollections", domain.ListItemCollection) + } params.Set("cursor", strconv.FormatInt(max(1, cursor-ReplayOverlap.Microseconds()), 10)) u.RawQuery = params.Encode() conn, response, err := websocket.Dial(ctx, u.String(), &websocket.DialOptions{HTTPClient: c.http, HTTPHeader: http.Header{"User-Agent": {UserAgent}}}) diff --git a/internal/storage/db/models.go b/internal/storage/db/models.go index 8d7c356..a12b585 100644 --- a/internal/storage/db/models.go +++ b/internal/storage/db/models.go @@ -9,6 +9,46 @@ type GraphRevision struct { Revision int64 } +type ListBootstrap struct { + Author string + BootstrapUs int64 +} + +type ListEdge struct { + Author string + Subject string + Source string + RecordKey string + CreatedAt string + ListUri string + ListAuthor string + ListKey string + ItemKey string +} + +type ListRecord struct { + Author string + Source string + RecordKey string + ListUri string + Subject string + Purpose string + CreatedAt string +} + +type ListSnapshot struct { + ListUri string + Owner string + BootstrapUs int64 +} + +type ListVersion struct { + Author string + Source string + RecordKey string + TimeUs int64 +} + type Participant struct { Did string Threshold float64 diff --git a/internal/storage/db/querier.go b/internal/storage/db/querier.go index 1ea199c..05fcf0a 100644 --- a/internal/storage/db/querier.go +++ b/internal/storage/db/querier.go @@ -13,12 +13,14 @@ type Querier interface { AdvanceCursor(ctx context.Context, arg AdvanceCursorParams) error AuthorRecords(ctx context.Context, author string) ([]Vouch, error) CachedTrust(ctx context.Context, arg CachedTrustParams) (string, error) + ClearListAuthor(ctx context.Context, author string) error ClearSoft(ctx context.Context, author string) error ClearSoftVersions(ctx context.Context, author string) error ClearVersions(ctx context.Context, author string) error ClearVouches(ctx context.Context, author string) error ConsumeSecret(ctx context.Context, arg ConsumeSecretParams) ([]byte, error) Cursor(ctx context.Context, source string) (int64, error) + DeleteListRecord(ctx context.Context, arg DeleteListRecordParams) error DeleteRecord(ctx context.Context, arg DeleteRecordParams) error DeleteSoft(ctx context.Context, arg DeleteSoftParams) error DeleteTrust(ctx context.Context, cacheKey string) error @@ -31,10 +33,22 @@ type Querier interface { Join(ctx context.Context, did string) error JoinSoft(ctx context.Context, did string) error LegacyKeys(ctx context.Context) ([]LegacyKeysRow, error) + ListAuthorRecords(ctx context.Context, arg ListAuthorRecordsParams) ([]ListRecord, error) + ListBootstrap(ctx context.Context, author string) (int64, error) + ListGraph(ctx context.Context, arg ListGraphParams) ([]ListEdge, error) + ListIncoming(ctx context.Context, arg ListIncomingParams) ([]ListEdge, error) + ListRecord(ctx context.Context, arg ListRecordParams) (ListRecord, error) + ListURIRecords(ctx context.Context, listUri string) ([]ListRecord, error) + ListVersion(ctx context.Context, arg ListVersionParams) (int64, error) + ListWatermark(ctx context.Context, listUri string) (int64, error) OwnVouches(ctx context.Context, arg OwnVouchesParams) ([]Vouch, error) Participant(ctx context.Context, did string) (Participant, error) Participants(ctx context.Context, limit int64) ([]Participant, error) PruneSecrets(ctx context.Context, expires int64) error + PutListBootstrap(ctx context.Context, arg PutListBootstrapParams) error + PutListRecord(ctx context.Context, arg PutListRecordParams) error + PutListVersion(ctx context.Context, arg PutListVersionParams) error + PutListWatermark(ctx context.Context, arg PutListWatermarkParams) error PutSecret(ctx context.Context, arg PutSecretParams) error RecordVersion(ctx context.Context, arg RecordVersionParams) (int64, error) RemoveSecret(ctx context.Context, arg RemoveSecretParams) error @@ -59,6 +73,8 @@ type Querier interface { SoftRecords(ctx context.Context, author string) ([]SoftVouch, error) SoftVersion(ctx context.Context, arg SoftVersionParams) (int64, error) Tags(ctx context.Context, arg TagsParams) ([]string, error) + WantsListOwner(ctx context.Context, owner string) (bool, error) + WantsListURI(ctx context.Context, listUri string) (bool, error) } var _ Querier = (*Queries)(nil) diff --git a/internal/storage/db/queries.sql.go b/internal/storage/db/queries.sql.go index 1d49190..e0eaedb 100644 --- a/internal/storage/db/queries.sql.go +++ b/internal/storage/db/queries.sql.go @@ -97,6 +97,15 @@ func (q *Queries) CachedTrust(ctx context.Context, arg CachedTrustParams) (strin return result, err } +const clearListAuthor = `-- name: ClearListAuthor :exec +DELETE FROM list_records WHERE author = ? +` + +func (q *Queries) ClearListAuthor(ctx context.Context, author string) error { + _, err := q.db.ExecContext(ctx, clearListAuthor, author) + return err +} + const clearSoft = `-- name: ClearSoft :exec DELETE FROM soft_vouches WHERE author = ? ` @@ -161,6 +170,21 @@ func (q *Queries) Cursor(ctx context.Context, source string) (int64, error) { return time_us, err } +const deleteListRecord = `-- name: DeleteListRecord :exec +DELETE FROM list_records WHERE author = ? AND source = ? AND record_key = ? +` + +type DeleteListRecordParams struct { + Author string + Source string + RecordKey string +} + +func (q *Queries) DeleteListRecord(ctx context.Context, arg DeleteListRecordParams) error { + _, err := q.db.ExecContext(ctx, deleteListRecord, arg.Author, arg.Source, arg.RecordKey) + return err +} + const deleteRecord = `-- name: DeleteRecord :exec DELETE FROM vouches WHERE author = ? AND record_key = ? ` @@ -417,6 +441,252 @@ func (q *Queries) LegacyKeys(ctx context.Context) ([]LegacyKeysRow, error) { return items, nil } +const listAuthorRecords = `-- name: ListAuthorRecords :many +SELECT author, source, record_key, list_uri, subject, purpose, created_at FROM list_records WHERE author = ? AND source = ? +` + +type ListAuthorRecordsParams struct { + Author string + Source string +} + +func (q *Queries) ListAuthorRecords(ctx context.Context, arg ListAuthorRecordsParams) ([]ListRecord, error) { + rows, err := q.db.QueryContext(ctx, listAuthorRecords, arg.Author, arg.Source) + if err != nil { + return nil, err + } + defer rows.Close() + items := []ListRecord{} + for rows.Next() { + var i ListRecord + if err := rows.Scan( + &i.Author, + &i.Source, + &i.RecordKey, + &i.ListUri, + &i.Subject, + &i.Purpose, + &i.CreatedAt, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const listBootstrap = `-- name: ListBootstrap :one +SELECT bootstrap_us FROM list_bootstraps WHERE author = ? +` + +func (q *Queries) ListBootstrap(ctx context.Context, author string) (int64, error) { + row := q.db.QueryRowContext(ctx, listBootstrap, author) + var bootstrap_us int64 + err := row.Scan(&bootstrap_us) + return bootstrap_us, err +} + +const listGraph = `-- name: ListGraph :many +SELECT e.author, e.subject, e.source, e.record_key, e.created_at, e.list_uri, e.list_author, e.list_key, e.item_key FROM list_edges e +WHERE (CAST(?1 AS INTEGER) = 1 OR EXISTS (SELECT 1 FROM participants p WHERE p.did = e.author AND p.active = 1)) +AND NOT EXISTS (SELECT 1 FROM vouches v WHERE v.author = e.author AND v.subject = e.subject AND (v.tag = ?2 OR (CAST(?3 AS INTEGER) = 1 AND v.tag = ''))) +ORDER BY e.author, e.subject, e.list_uri LIMIT ?4 +` + +type ListGraphParams struct { + AllAuthors int64 + Tag string + Fallback int64 + MaxRows int64 +} + +func (q *Queries) ListGraph(ctx context.Context, arg ListGraphParams) ([]ListEdge, error) { + rows, err := q.db.QueryContext(ctx, listGraph, + arg.AllAuthors, + arg.Tag, + arg.Fallback, + arg.MaxRows, + ) + if err != nil { + return nil, err + } + defer rows.Close() + items := []ListEdge{} + for rows.Next() { + var i ListEdge + if err := rows.Scan( + &i.Author, + &i.Subject, + &i.Source, + &i.RecordKey, + &i.CreatedAt, + &i.ListUri, + &i.ListAuthor, + &i.ListKey, + &i.ItemKey, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const listIncoming = `-- name: ListIncoming :many +SELECT e.author, e.subject, e.source, e.record_key, e.created_at, e.list_uri, e.list_author, e.list_key, e.item_key FROM list_edges e +WHERE e.subject = ?1 +AND (CAST(?2 AS INTEGER) = 1 OR EXISTS (SELECT 1 FROM participants p WHERE p.did = e.author AND p.active = 1)) +AND NOT EXISTS (SELECT 1 FROM vouches v WHERE v.author = e.author AND v.subject = e.subject AND (v.tag = ?3 OR (CAST(?4 AS INTEGER) = 1 AND v.tag = ''))) +ORDER BY e.author, e.list_uri LIMIT ?5 +` + +type ListIncomingParams struct { + Subject string + AllAuthors int64 + Tag string + Fallback int64 + MaxRows int64 +} + +func (q *Queries) ListIncoming(ctx context.Context, arg ListIncomingParams) ([]ListEdge, error) { + rows, err := q.db.QueryContext(ctx, listIncoming, + arg.Subject, + arg.AllAuthors, + arg.Tag, + arg.Fallback, + arg.MaxRows, + ) + if err != nil { + return nil, err + } + defer rows.Close() + items := []ListEdge{} + for rows.Next() { + var i ListEdge + if err := rows.Scan( + &i.Author, + &i.Subject, + &i.Source, + &i.RecordKey, + &i.CreatedAt, + &i.ListUri, + &i.ListAuthor, + &i.ListKey, + &i.ItemKey, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const listRecord = `-- name: ListRecord :one +SELECT author, source, record_key, list_uri, subject, purpose, created_at FROM list_records WHERE author = ? AND source = ? AND record_key = ? +` + +type ListRecordParams struct { + Author string + Source string + RecordKey string +} + +func (q *Queries) ListRecord(ctx context.Context, arg ListRecordParams) (ListRecord, error) { + row := q.db.QueryRowContext(ctx, listRecord, arg.Author, arg.Source, arg.RecordKey) + var i ListRecord + err := row.Scan( + &i.Author, + &i.Source, + &i.RecordKey, + &i.ListUri, + &i.Subject, + &i.Purpose, + &i.CreatedAt, + ) + return i, err +} + +const listURIRecords = `-- name: ListURIRecords :many +SELECT author, source, record_key, list_uri, subject, purpose, created_at FROM list_records WHERE list_uri = ? AND source != 'app.bsky.graph.listblock' +` + +func (q *Queries) ListURIRecords(ctx context.Context, listUri string) ([]ListRecord, error) { + rows, err := q.db.QueryContext(ctx, listURIRecords, listUri) + if err != nil { + return nil, err + } + defer rows.Close() + items := []ListRecord{} + for rows.Next() { + var i ListRecord + if err := rows.Scan( + &i.Author, + &i.Source, + &i.RecordKey, + &i.ListUri, + &i.Subject, + &i.Purpose, + &i.CreatedAt, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const listVersion = `-- name: ListVersion :one +SELECT time_us FROM list_versions WHERE author = ? AND source = ? AND record_key = ? +` + +type ListVersionParams struct { + Author string + Source string + RecordKey string +} + +func (q *Queries) ListVersion(ctx context.Context, arg ListVersionParams) (int64, error) { + row := q.db.QueryRowContext(ctx, listVersion, arg.Author, arg.Source, arg.RecordKey) + var time_us int64 + err := row.Scan(&time_us) + return time_us, err +} + +const listWatermark = `-- name: ListWatermark :one +SELECT bootstrap_us FROM list_snapshots WHERE list_uri = ? +` + +func (q *Queries) ListWatermark(ctx context.Context, listUri string) (int64, error) { + row := q.db.QueryRowContext(ctx, listWatermark, listUri) + var bootstrap_us int64 + err := row.Scan(&bootstrap_us) + return bootstrap_us, err +} + const ownVouches = `-- name: OwnVouches :many SELECT vouches.author, vouches.subject, vouches.tag, vouches.score, vouches.created_at, vouches.record_key, vouches.comment, vouches.source FROM vouches JOIN participants ON participants.did = vouches.author WHERE participants.active = 1 AND author = ? ORDER BY subject, tag LIMIT ? OFFSET ? @@ -521,6 +791,88 @@ func (q *Queries) PruneSecrets(ctx context.Context, expires int64) error { return err } +const putListBootstrap = `-- name: PutListBootstrap :exec +INSERT INTO list_bootstraps(author, bootstrap_us) VALUES (?, ?) +ON CONFLICT(author) DO UPDATE SET bootstrap_us = excluded.bootstrap_us +` + +type PutListBootstrapParams struct { + Author string + BootstrapUs int64 +} + +func (q *Queries) PutListBootstrap(ctx context.Context, arg PutListBootstrapParams) error { + _, err := q.db.ExecContext(ctx, putListBootstrap, arg.Author, arg.BootstrapUs) + return err +} + +const putListRecord = `-- name: PutListRecord :exec +INSERT INTO list_records(author, source, record_key, list_uri, subject, purpose, created_at) VALUES (?, ?, ?, ?, ?, ?, ?) +ON CONFLICT(author, source, record_key) DO UPDATE SET list_uri = excluded.list_uri, subject = excluded.subject, purpose = excluded.purpose, created_at = excluded.created_at +WHERE list_uri != excluded.list_uri OR subject != excluded.subject OR purpose != excluded.purpose OR created_at != excluded.created_at +` + +type PutListRecordParams struct { + Author string + Source string + RecordKey string + ListUri string + Subject string + Purpose string + CreatedAt string +} + +func (q *Queries) PutListRecord(ctx context.Context, arg PutListRecordParams) error { + _, err := q.db.ExecContext(ctx, putListRecord, + arg.Author, + arg.Source, + arg.RecordKey, + arg.ListUri, + arg.Subject, + arg.Purpose, + arg.CreatedAt, + ) + return err +} + +const putListVersion = `-- name: PutListVersion :exec +INSERT INTO list_versions(author, source, record_key, time_us) VALUES (?, ?, ?, ?) +ON CONFLICT(author, source, record_key) DO UPDATE SET time_us = excluded.time_us +` + +type PutListVersionParams struct { + Author string + Source string + RecordKey string + TimeUs int64 +} + +func (q *Queries) PutListVersion(ctx context.Context, arg PutListVersionParams) error { + _, err := q.db.ExecContext(ctx, putListVersion, + arg.Author, + arg.Source, + arg.RecordKey, + arg.TimeUs, + ) + return err +} + +const putListWatermark = `-- name: PutListWatermark :exec +INSERT INTO list_snapshots(list_uri, owner, bootstrap_us) VALUES (?, ?, ?) +ON CONFLICT(list_uri) DO UPDATE SET bootstrap_us = excluded.bootstrap_us +` + +type PutListWatermarkParams struct { + ListUri string + Owner string + BootstrapUs int64 +} + +func (q *Queries) PutListWatermark(ctx context.Context, arg PutListWatermarkParams) error { + _, err := q.db.ExecContext(ctx, putListWatermark, arg.ListUri, arg.Owner, arg.BootstrapUs) + return err +} + const putSecret = `-- name: PutSecret :exec INSERT INTO secrets(kind, id, value, expires) VALUES (?, ?, ?, ?) ON CONFLICT(kind, id) DO UPDATE SET value = excluded.value, expires = excluded.expires @@ -1064,3 +1416,25 @@ func (q *Queries) Tags(ctx context.Context, arg TagsParams) ([]string, error) { } return items, nil } + +const wantsListOwner = `-- name: WantsListOwner :one +SELECT EXISTS(SELECT 1 FROM list_snapshots WHERE owner = ?) AS wanted +` + +func (q *Queries) WantsListOwner(ctx context.Context, owner string) (bool, error) { + row := q.db.QueryRowContext(ctx, wantsListOwner, owner) + var wanted bool + err := row.Scan(&wanted) + return wanted, err +} + +const wantsListURI = `-- name: WantsListURI :one +SELECT EXISTS(SELECT 1 FROM list_snapshots WHERE list_uri = ?) AS wanted +` + +func (q *Queries) WantsListURI(ctx context.Context, listUri string) (bool, error) { + row := q.db.QueryRowContext(ctx, wantsListURI, listUri) + var wanted bool + err := row.Scan(&wanted) + return wanted, err +} diff --git a/internal/storage/events.go b/internal/storage/events.go index 30eb6b8..746fb98 100644 --- a/internal/storage/events.go +++ b/internal/storage/events.go @@ -28,12 +28,17 @@ func (s *SQLite) Apply(ctx context.Context, source string, change domain.Change) } defer tx.Rollback() q := s.queries.WithTx(tx) - if domain.SoftSource(change.Source) || change.Kind == domain.AccountChange { + if domain.SoftLists(s.softMode) && (domain.ListSource(change.Source) || change.Kind == domain.AccountChange) { + if err := s.applyList(ctx, q, change); err != nil { + return err + } + } + if !domain.ListSource(change.Source) && (domain.SoftSource(change.Source) || change.Kind == domain.AccountChange) { if err := s.applySoft(ctx, q, change); err != nil { return err } } - if !domain.SoftSource(change.Source) { + if !domain.SoftSource(change.Source) && !domain.ListSource(change.Source) { if err := applyChange(ctx, q, change); err != nil { return err } diff --git a/internal/storage/lists.go b/internal/storage/lists.go new file mode 100644 index 0000000..0a5ad0e --- /dev/null +++ b/internal/storage/lists.go @@ -0,0 +1,282 @@ +package storage + +import ( + "context" + "database/sql" + "errors" + "time" + + "provenance/internal/domain" + "provenance/internal/storage/db" +) + +func listRows(rows []db.ListEdge) []domain.Vouch { + result := make([]domain.Vouch, 0, 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) + } + 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 := q.JoinSoft(ctx, r.Author); 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 := q.JoinSoft(ctx, 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 := q.JoinSoft(ctx, 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 { + 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 + } + if change.RecordKey == "" { + return errors.New("list event key is required") + } + if err := q.JoinSoft(ctx, 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 change.TimeUS <= max(a.AccountUs, 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}) +} diff --git a/internal/storage/lists_schema.sql b/internal/storage/lists_schema.sql new file mode 100644 index 0000000..b0b0be0 --- /dev/null +++ b/internal/storage/lists_schema.sql @@ -0,0 +1,65 @@ +CREATE TABLE IF NOT EXISTS list_records ( + author TEXT NOT NULL REFERENCES soft_accounts(did) ON DELETE CASCADE, + source TEXT NOT NULL CHECK(source IN ('app.bsky.graph.list', 'app.bsky.graph.listitem', 'app.bsky.graph.listblock')), + record_key TEXT NOT NULL, + list_uri TEXT NOT NULL, + subject TEXT NOT NULL DEFAULT '', + purpose TEXT NOT NULL DEFAULT '', + created_at TEXT NOT NULL, + PRIMARY KEY(author, source, record_key) +); +CREATE INDEX IF NOT EXISTS list_records_uri ON list_records(list_uri, source, subject, author); +CREATE INDEX IF NOT EXISTS list_records_subject ON list_records(subject, source, author); + +CREATE TABLE IF NOT EXISTS list_versions ( + author TEXT NOT NULL REFERENCES soft_accounts(did) ON DELETE CASCADE, + source TEXT NOT NULL, + record_key TEXT NOT NULL, + time_us INTEGER NOT NULL, + PRIMARY KEY(author, source, record_key) +); +CREATE TABLE IF NOT EXISTS list_snapshots ( + list_uri TEXT PRIMARY KEY, + owner TEXT NOT NULL REFERENCES soft_accounts(did) ON DELETE CASCADE, + bootstrap_us INTEGER NOT NULL +); +CREATE INDEX IF NOT EXISTS list_snapshots_owner ON list_snapshots(owner); +CREATE TABLE IF NOT EXISTS list_bootstraps ( + author TEXT PRIMARY KEY REFERENCES soft_accounts(did) ON DELETE CASCADE, + bootstrap_us INTEGER NOT NULL +); + +CREATE VIEW IF NOT EXISTS list_edges AS +SELECT b.author, i.subject, b.source, b.record_key, b.created_at, + b.list_uri, l.author AS list_author, l.record_key AS list_key, i.record_key AS item_key +FROM list_records b +JOIN list_records l ON l.list_uri = b.list_uri AND l.source = 'app.bsky.graph.list' AND l.purpose = 'app.bsky.graph.defs#modlist' +JOIN list_records i ON i.list_uri = l.list_uri AND i.author = l.author AND i.source = 'app.bsky.graph.listitem' +JOIN soft_accounts subscriber ON subscriber.did = b.author AND subscriber.active = 1 +JOIN soft_accounts owner ON owner.did = l.author AND owner.active = 1 +WHERE b.source = 'app.bsky.graph.listblock' +AND NOT EXISTS (SELECT 1 FROM participants p WHERE p.did = i.subject AND p.active = 0) +AND NOT EXISTS (SELECT 1 FROM soft_accounts p WHERE p.did = i.subject AND p.active = 0) +AND NOT EXISTS (SELECT 1 FROM list_records x WHERE x.author = b.author AND x.source = b.source AND x.list_uri = b.list_uri AND x.record_key < b.record_key) +AND NOT EXISTS (SELECT 1 FROM list_records x WHERE x.author = i.author AND x.source = i.source AND x.list_uri = i.list_uri AND x.subject = i.subject AND x.record_key < i.record_key); + +-- A selected list edge depends on its subscription, list definition, and membership. +CREATE TRIGGER IF NOT EXISTS trust_list_delete BEFORE DELETE ON list_records BEGIN + DELETE FROM trust_cache WHERE cache_key IN ( + SELECT cache_key FROM trust_dependencies WHERE author = OLD.author AND source = OLD.source AND record_key = OLD.record_key + ); +END; +CREATE TRIGGER IF NOT EXISTS trust_list_update BEFORE UPDATE ON list_records BEGIN + DELETE FROM trust_cache WHERE cache_key IN ( + SELECT cache_key FROM trust_dependencies WHERE author = OLD.author AND source = OLD.source AND record_key = OLD.record_key + ); +END; +CREATE TRIGGER IF NOT EXISTS graph_list_insert AFTER INSERT ON list_records BEGIN + UPDATE graph_revision SET revision = revision + 1 WHERE singleton = 1; +END; +CREATE TRIGGER IF NOT EXISTS graph_list_delete AFTER DELETE ON list_records BEGIN + UPDATE graph_revision SET revision = revision + 1 WHERE singleton = 1; +END; +CREATE TRIGGER IF NOT EXISTS graph_list_update AFTER UPDATE ON list_records BEGIN + UPDATE graph_revision SET revision = revision + 1 WHERE singleton = 1; +END; diff --git a/internal/storage/lists_test.go b/internal/storage/lists_test.go new file mode 100644 index 0000000..de6912d --- /dev/null +++ b/internal/storage/lists_test.go @@ -0,0 +1,330 @@ +package storage + +import ( + "context" + "database/sql" + "errors" + "path/filepath" + "provenance/internal/domain" + "provenance/internal/trust" + "testing" + "time" +) + +const listDate = "2026-10-09T00:00:00Z" + +func listFixture() domain.ListSnapshot { + const uri = "at://did:plc:owner/app.bsky.graph.list/mod" + return domain.ListSnapshot{ + Blocks: []domain.ListRecord{{Author: "did:plc:a", Source: domain.ListBlockCollection, RecordKey: "subscription", ListURI: uri, CreatedAt: listDate}}, + Lists: []domain.ListData{{URI: uri, TimeUS: 100, Records: []domain.ListRecord{ + {Author: "did:plc:owner", Source: domain.ListCollection, RecordKey: "mod", ListURI: uri, Purpose: domain.ModerationList, CreatedAt: listDate}, + {Author: "did:plc:owner", Source: domain.ListItemCollection, RecordKey: "member", ListURI: uri, Subject: "did:plc:b", CreatedAt: listDate}, + }}}, + } +} + +func TestBlockedListStorage(t *testing.T) { + ctx := context.Background() + s, err := OpenWithSoft(filepath.Join(t.TempDir(), "lists.db"), domain.SoftParticipantLists) + if err != nil { + t.Fatal(err) + } + defer s.Close() + snapshot := listFixture() + author := snapshot.Blocks[0].Author + if err := s.Join(ctx, author); err != nil { + t.Fatal(err) + } + follow := domain.Vouch{Author: author, Subject: "did:plc:b", Source: domain.FollowCollection, RecordKey: "follow", Score: .5} + if err := s.ReplaceWithLists(ctx, author, []domain.Vouch{follow}, snapshot, time.UnixMicro(100)); err != nil { + t.Fatal(err) + } + graph := func(tag string, mode domain.Mode) []domain.Vouch { + t.Helper() + rows, err := s.Graph(ctx, tag, mode, 100) + if err != nil { + t.Fatal(err) + } + return trust.Select(rows, tag, mode) + } + if got := graph("", domain.Default); len(got) != 1 || got[0].Score != -.5 || got[0].Source != domain.ListBlockCollection || got[0].ListURI != snapshot.Lists[0].URI || got[0].ItemKey != "member" { + t.Fatalf("list edge: %+v", got) + } + if got := graph("work", domain.Tagged); len(got) != 0 { + t.Fatalf("tag-only: %+v", got) + } + if got := graph("work", domain.Fallback); len(got) != 1 || got[0].Score != -.5 { + t.Fatalf("fallback: %+v", got) + } + if _, err := s.Participant(ctx, "did:plc:owner"); !errors.Is(err, sql.ErrNoRows) { + t.Fatalf("owner enrolled: %v", err) + } + // Explicit zero scores override list blocks in both default and topic fallback queries. + for index, tag := range []string{"", "work"} { + v := domain.Vouch{Author: author, Subject: follow.Subject, Tag: tag, Score: 0} + key := domain.RecordKey(v.Subject, tag) + stamp := int64(101 + 2*index) + if err := s.Apply(ctx, "explicit", domain.Change{Kind: domain.RecordChange, DID: author, RecordKey: key, Vouch: &v, TimeUS: stamp}); err != nil { + t.Fatal(err) + } + mode := domain.Default + if tag != "" { + mode = domain.Fallback + } + if got := graph(tag, mode); len(got) != 1 || got[0].Score != 0 || got[0].Source != domain.Collection { + t.Fatalf("explicit precedence: %+v", got) + } + if err := s.Apply(ctx, "explicit", domain.Change{Kind: domain.RecordChange, DID: author, RecordKey: key, TimeUS: stamp + 1}); err != nil { + t.Fatal(err) + } + } + apply := func(r *domain.ListRecord, author, source, key string, stamp int64) { + t.Helper() + if err := s.Apply(ctx, "lists", domain.Change{Kind: domain.RecordChange, DID: author, Source: source, RecordKey: key, ListRecord: r, TimeUS: stamp}); err != nil { + t.Fatal(err) + } + } + // A duplicate membership cannot multiply the negative assertion. + duplicate := snapshot.Lists[0].Records[1] + duplicate.RecordKey = "member-2" + apply(&duplicate, duplicate.Author, duplicate.Source, duplicate.RecordKey, 110) + if got := graph("", domain.Default); len(got) != 1 { + t.Fatalf("duplicate members: %+v", got) + } + rows, revision, err := s.TrustGraph(ctx, "", domain.Default, 100) + if err != nil || len(rows) < 1 { + t.Fatal(err) + } + score := -.5 + deps := []trust.Dependency{{Author: author, Source: domain.ListBlockCollection, RecordKey: "subscription"}, {Author: duplicate.Author, Source: domain.ListItemCollection, RecordKey: "member"}, {Author: duplicate.Author, Source: domain.ListCollection, RecordKey: "mod"}} + now := time.UnixMicro(1000) + if saved, err := s.SaveTrust(ctx, "list-cache", trust.Result{Status: trust.CalculatedStatus, Score: &score}, deps, revision, now.Add(time.Hour)); !saved || err != nil { + t.Fatalf("save: %v %v", saved, err) + } + apply(nil, duplicate.Author, duplicate.Source, "member-2", 120) + if _, err := s.CachedTrust(ctx, "list-cache", now); err != nil { + t.Fatalf("unselected membership invalidation: %v", err) + } + apply(nil, duplicate.Author, duplicate.Source, "member", 130) + if _, err := s.CachedTrust(ctx, "list-cache", now); !errors.Is(err, sql.ErrNoRows) { + t.Fatalf("selected membership retained cache: %v", err) + } + apply(&snapshot.Lists[0].Records[1], duplicate.Author, duplicate.Source, "member", 125) + if got := graph("", domain.Default); len(got) != 1 || got[0].Score != .5 { + t.Fatalf("membership replay: %+v", got) + } + apply(&snapshot.Lists[0].Records[1], duplicate.Author, duplicate.Source, "member", 140) + apply(nil, author, domain.ListBlockCollection, "subscription", 150) + if got := graph("", domain.Default); len(got) != 1 || got[0].Score != .5 { + t.Fatalf("unblocked list: %+v", got) + } + apply(&snapshot.Blocks[0], author, domain.ListBlockCollection, "subscription", 160) + apply(nil, duplicate.Author, domain.ListCollection, "mod", 170) + if got := graph("", domain.Default); len(got) != 1 || got[0].Score != .5 { + t.Fatalf("deleted list: %+v", got) + } + apply(&snapshot.Lists[0].Records[0], duplicate.Author, domain.ListCollection, "mod", 165) + if got := graph("", domain.Default); len(got) != 1 || got[0].Score != .5 { + t.Fatalf("list deletion replay: %+v", got) + } +} + +func TestListScopeAndSnapshots(t *testing.T) { + ctx := context.Background() + for _, mode := range []domain.SoftMode{domain.SoftDisabled, domain.SoftParticipants, domain.SoftAllAuthors, domain.SoftParticipantLists, domain.SoftAllAuthorLists} { + s, err := OpenWithSoft(filepath.Join(t.TempDir(), "scope.db"), mode) + if err != nil { + t.Fatal(err) + } + snapshot := listFixture() + block := snapshot.Blocks[0] + data := snapshot.Lists[0] + if err := s.Apply(ctx, "stream", domain.Change{Kind: domain.RecordChange, DID: block.Author, Source: block.Source, RecordKey: block.RecordKey, ListRecord: &block, ListData: &data, TimeUS: 200}); err != nil { + t.Fatal(err) + } + rows, err := s.Incoming(ctx, "did:plc:b", "", domain.Default, 100) + if err != nil { + t.Fatal(err) + } + want := 0 + if mode == domain.SoftAllAuthorLists { + want = 1 + } + if len(rows) != want { + t.Fatalf("scope %v: %+v", mode, rows) + } + s.Close() + } +} + +func TestListSnapshotRollback(t *testing.T) { + ctx := context.Background() + s, err := OpenWithSoft(filepath.Join(t.TempDir(), "rollback.db"), domain.SoftParticipantLists) + if err != nil { + t.Fatal(err) + } + defer s.Close() + snapshot := listFixture() + if err := s.Join(ctx, "did:plc:a"); err != nil { + t.Fatal(err) + } + if err := s.ReplaceWithLists(ctx, "did:plc:a", nil, snapshot, time.UnixMicro(100)); err != nil { + t.Fatal(err) + } + snapshot.Lists[0].Records[1].Author = "did:plc:forged" + if err := s.ReplaceWithLists(ctx, "did:plc:a", nil, snapshot, time.UnixMicro(200)); err == nil { + t.Fatal("forged list member accepted") + } + rows, err := s.Incoming(ctx, "did:plc:b", "", domain.Default, 100) + if err != nil || len(rows) != 1 { + t.Fatalf("rollback lost member: %+v %v", rows, err) + } + snapshot.Lists = nil + if err := s.ReplaceWithLists(ctx, "did:plc:a", nil, snapshot, time.UnixMicro(200)); err == nil { + t.Fatal("incomplete list snapshot accepted") + } +} + +func TestMovedListMember(t *testing.T) { + ctx := context.Background() + s, err := OpenWithSoft(filepath.Join(t.TempDir(), "move.db"), domain.SoftParticipantLists) + if err != nil { + t.Fatal(err) + } + defer s.Close() + first := listFixture() + if err := s.Join(ctx, "did:plc:a"); err != nil { + t.Fatal(err) + } + if err := s.ReplaceWithLists(ctx, "did:plc:a", nil, first, time.UnixMicro(100)); err != nil { + t.Fatal(err) + } + const secondURI = "at://did:plc:owner/app.bsky.graph.list/second" + second := domain.ListSnapshot{Blocks: []domain.ListRecord{{Author: "did:plc:c", Source: domain.ListBlockCollection, RecordKey: "block", ListURI: secondURI, CreatedAt: listDate}}, Lists: []domain.ListData{{URI: secondURI, TimeUS: 300, Records: []domain.ListRecord{{Author: "did:plc:owner", Source: domain.ListCollection, RecordKey: "second", ListURI: secondURI, Purpose: domain.ModerationList, CreatedAt: listDate}}}}} + if err := s.Join(ctx, "did:plc:c"); err != nil { + t.Fatal(err) + } + if err := s.ReplaceWithLists(ctx, "did:plc:c", nil, second, time.UnixMicro(300)); err != nil { + t.Fatal(err) + } + moved := first.Lists[0].Records[1] + moved.ListURI = secondURI + if err := s.Apply(ctx, "stream", domain.Change{Kind: domain.RecordChange, DID: moved.Author, Source: moved.Source, RecordKey: moved.RecordKey, ListRecord: &moved, TimeUS: 200}); err != nil { + t.Fatal(err) + } + rows, err := s.Incoming(ctx, moved.Subject, "", domain.Default, 100) + if err != nil || len(rows) != 0 { + t.Fatalf("stale membership after move: %+v %v", rows, err) + } +} + +func TestUnknownListDeletion(t *testing.T) { + ctx := context.Background() + s, err := OpenWithSoft(filepath.Join(t.TempDir(), "replay.db"), domain.SoftParticipantLists) + if err != nil { + t.Fatal(err) + } + defer s.Close() + snapshot := listFixture() + if err := s.Join(ctx, "did:plc:a"); err != nil { + t.Fatal(err) + } + if err := s.ReplaceWithLists(ctx, "did:plc:a", nil, snapshot, time.UnixMicro(100)); err != nil { + t.Fatal(err) + } + item := snapshot.Lists[0].Records[1] + item.RecordKey, item.Subject = "unseen", "did:plc:unseen" + if err := s.Apply(ctx, "stream", domain.Change{Kind: domain.RecordChange, DID: item.Author, Source: item.Source, RecordKey: item.RecordKey, TimeUS: 300}); err != nil { + t.Fatal(err) + } + if err := s.Apply(ctx, "stream", domain.Change{Kind: domain.RecordChange, DID: item.Author, Source: item.Source, RecordKey: item.RecordKey, ListRecord: &item, TimeUS: 200}); err != nil { + t.Fatal(err) + } + rows, err := s.Incoming(ctx, item.Subject, "", domain.Default, 100) + if err != nil || len(rows) != 0 { + t.Fatalf("unknown deletion lost its tombstone: %+v %v", rows, err) + } +} + +func TestListDependencies(t *testing.T) { + ctx := context.Background() + for _, source := range []string{domain.ListCollection, domain.ListBlockCollection} { + t.Run(source, func(t *testing.T) { + s, err := OpenWithSoft(filepath.Join(t.TempDir(), "deps.db"), domain.SoftParticipantLists) + if err != nil { + t.Fatal(err) + } + defer s.Close() + snapshot := listFixture() + if err := s.Join(ctx, "did:plc:a"); err != nil { + t.Fatal(err) + } + if err := s.ReplaceWithLists(ctx, "did:plc:a", nil, snapshot, time.UnixMicro(100)); err != nil { + t.Fatal(err) + } + _, revision, err := s.TrustGraph(ctx, "", domain.Default, 100) + if err != nil { + t.Fatal(err) + } + now := time.UnixMicro(1000) + record := snapshot.Blocks[0] + if source == domain.ListCollection { + record = snapshot.Lists[0].Records[0] + } + score := -.5 + if saved, err := s.SaveTrust(ctx, "path", trust.Result{Status: trust.CalculatedStatus, Score: &score}, []trust.Dependency{{Author: record.Author, Source: record.Source, RecordKey: record.RecordKey}}, revision, now.Add(time.Hour)); !saved || err != nil { + t.Fatalf("cache: %v %v", saved, err) + } + snapshot.Lists[0].TimeUS = 200 + if err := s.ReplaceWithLists(ctx, "did:plc:a", nil, snapshot, time.UnixMicro(200)); err != nil { + t.Fatal(err) + } + if _, err := s.CachedTrust(ctx, "path", now); err != nil { + t.Fatalf("unchanged snapshot invalidated path: %v", err) + } + if err := s.Apply(ctx, "stream", domain.Change{Kind: domain.RecordChange, DID: record.Author, Source: record.Source, RecordKey: record.RecordKey, TimeUS: 300}); err != nil { + t.Fatal(err) + } + if _, err := s.CachedTrust(ctx, "path", now); !errors.Is(err, sql.ErrNoRows) { + t.Fatalf("list dependency retained cache: %v", err) + } + }) + } +} + +func TestListOwnerLifecycle(t *testing.T) { + ctx := context.Background() + s, err := OpenWithSoft(filepath.Join(t.TempDir(), "lifecycle.db"), domain.SoftParticipantLists) + if err != nil { + t.Fatal(err) + } + defer s.Close() + snapshot := listFixture() + if err := s.Join(ctx, "did:plc:a"); err != nil { + t.Fatal(err) + } + if err := s.ReplaceWithLists(ctx, "did:plc:a", nil, snapshot, time.UnixMicro(100)); err != nil { + t.Fatal(err) + } + for _, change := range []domain.Change{ + {Kind: domain.AccountChange, DID: "did:plc:owner", Active: false, TimeUS: 200}, + {Kind: domain.AccountChange, DID: "did:plc:owner", Active: true, TimeUS: 300}, + {Kind: domain.AccountChange, DID: "did:plc:owner", Active: false, Status: domain.DeletedAccount, TimeUS: 400}, + {Kind: domain.AccountChange, DID: "did:plc:owner", Active: true, TimeUS: 500}, + {Kind: domain.RecordChange, DID: "did:plc:owner", Source: domain.ListCollection, RecordKey: "mod", ListRecord: &snapshot.Lists[0].Records[0], TimeUS: 350}, + {Kind: domain.RecordChange, DID: "did:plc:owner", Source: domain.ListItemCollection, RecordKey: "member", ListRecord: &snapshot.Lists[0].Records[1], TimeUS: 360}, + } { + if err := s.Apply(ctx, "stream", change); err != nil { + t.Fatal(err) + } + rows, err := s.Incoming(ctx, "did:plc:b", "", domain.Default, 100) + want := 0 + if change.TimeUS == 300 { + want = 1 + } + if err != nil || len(rows) != want { + t.Fatalf("lifecycle %v: %+v %v", change.TimeUS, rows, err) + } + } +} diff --git a/internal/storage/queries.sql b/internal/storage/queries.sql index 84337df..a548879 100644 --- a/internal/storage/queries.sql +++ b/internal/storage/queries.sql @@ -110,6 +110,66 @@ INSERT INTO soft_vouches(author, subject, source, record_key, score, created_at) ON CONFLICT(author, source, record_key) DO UPDATE SET subject = excluded.subject, score = excluded.score, created_at = excluded.created_at WHERE subject != excluded.subject OR score != excluded.score OR created_at != excluded.created_at; +-- name: ListGraph :many +SELECT e.* FROM list_edges e +WHERE (CAST(@all_authors AS INTEGER) = 1 OR EXISTS (SELECT 1 FROM participants p WHERE p.did = e.author AND p.active = 1)) +AND NOT EXISTS (SELECT 1 FROM vouches v WHERE v.author = e.author AND v.subject = e.subject AND (v.tag = @tag OR (CAST(@fallback AS INTEGER) = 1 AND v.tag = ''))) +ORDER BY e.author, e.subject, e.list_uri LIMIT @max_rows; + +-- name: ListIncoming :many +SELECT e.* FROM list_edges e +WHERE e.subject = @subject +AND (CAST(@all_authors AS INTEGER) = 1 OR EXISTS (SELECT 1 FROM participants p WHERE p.did = e.author AND p.active = 1)) +AND NOT EXISTS (SELECT 1 FROM vouches v WHERE v.author = e.author AND v.subject = e.subject AND (v.tag = @tag OR (CAST(@fallback AS INTEGER) = 1 AND v.tag = ''))) +ORDER BY e.author, e.list_uri LIMIT @max_rows; + +-- name: ListRecord :one +SELECT * FROM list_records WHERE author = ? AND source = ? AND record_key = ?; + +-- name: ListAuthorRecords :many +SELECT * FROM list_records WHERE author = ? AND source = ?; + +-- name: ListURIRecords :many +SELECT * FROM list_records WHERE list_uri = ? AND source != 'app.bsky.graph.listblock'; + +-- name: PutListRecord :exec +INSERT INTO list_records(author, source, record_key, list_uri, subject, purpose, created_at) VALUES (?, ?, ?, ?, ?, ?, ?) +ON CONFLICT(author, source, record_key) DO UPDATE SET list_uri = excluded.list_uri, subject = excluded.subject, purpose = excluded.purpose, created_at = excluded.created_at +WHERE list_uri != excluded.list_uri OR subject != excluded.subject OR purpose != excluded.purpose OR created_at != excluded.created_at; + +-- name: DeleteListRecord :exec +DELETE FROM list_records WHERE author = ? AND source = ? AND record_key = ?; + +-- name: ClearListAuthor :exec +DELETE FROM list_records WHERE author = ?; + +-- name: ListVersion :one +SELECT time_us FROM list_versions WHERE author = ? AND source = ? AND record_key = ?; + +-- name: PutListVersion :exec +INSERT INTO list_versions(author, source, record_key, time_us) VALUES (?, ?, ?, ?) +ON CONFLICT(author, source, record_key) DO UPDATE SET time_us = excluded.time_us; + +-- name: ListWatermark :one +SELECT bootstrap_us FROM list_snapshots WHERE list_uri = ?; + +-- name: PutListWatermark :exec +INSERT INTO list_snapshots(list_uri, owner, bootstrap_us) VALUES (?, ?, ?) +ON CONFLICT(list_uri) DO UPDATE SET bootstrap_us = excluded.bootstrap_us; + +-- name: ListBootstrap :one +SELECT bootstrap_us FROM list_bootstraps WHERE author = ?; + +-- name: PutListBootstrap :exec +INSERT INTO list_bootstraps(author, bootstrap_us) VALUES (?, ?) +ON CONFLICT(author) DO UPDATE SET bootstrap_us = excluded.bootstrap_us; + +-- name: WantsListOwner :one +SELECT EXISTS(SELECT 1 FROM list_snapshots WHERE owner = ?) AS wanted; + +-- name: WantsListURI :one +SELECT EXISTS(SELECT 1 FROM list_snapshots WHERE list_uri = ?) AS wanted; + -- name: Graph :many SELECT vouches.* FROM vouches JOIN participants ON participants.did = vouches.author WHERE participants.active = 1 diff --git a/internal/storage/soft.go b/internal/storage/soft.go index d169e4f..43b7980 100644 --- a/internal/storage/soft.go +++ b/internal/storage/soft.go @@ -15,7 +15,7 @@ func (s *SQLite) graphRows(ctx context.Context, q *db.Queries, tag string, mode if mode == domain.Fallback { fallback = 1 } - if s.softMode == domain.SoftAllAuthors { + if domain.SoftAll(s.softMode) { all = 1 } rows, err := q.Graph(ctx, db.GraphParams{Tag: tag, Fallback: fallback, MaxRows: int64(limit)}) @@ -27,7 +27,15 @@ func (s *SQLite) graphRows(ctx context.Context, q *db.Queries, tag string, mode return result, nil } soft, err := q.SoftGraph(ctx, db.SoftGraphParams{Tag: tag, Fallback: fallback, AllAuthors: all, MaxRows: int64(limit)}) - return append(result, softRows(soft)...), err + if err != nil { + return nil, err + } + result = append(result, softRows(soft)...) + if !domain.SoftLists(s.softMode) { + return result, nil + } + lists, err := q.ListGraph(ctx, db.ListGraphParams{Tag: tag, Fallback: fallback, AllAuthors: all, MaxRows: int64(limit)}) + return append(result, listRows(lists)...), err } func softRows(rows []db.SoftEdge) []domain.Vouch { @@ -43,7 +51,14 @@ func (s *SQLite) SoftReady(ctx context.Context, did string) (bool, error) { if errors.Is(err, sql.ErrNoRows) { return false, nil } - return a.BootstrapUs != 0, err + if err != nil || !domain.SoftLists(s.softMode) || a.BootstrapUs == 0 { + return a.BootstrapUs != 0, err + } + _, err = s.queries.ListBootstrap(ctx, did) + if errors.Is(err, sql.ErrNoRows) { + return false, nil + } + return err == nil, err } // Keep replay watermarks and account state separate from Provenance participation. @@ -51,7 +66,7 @@ func (s *SQLite) applySoft(ctx context.Context, q *db.Queries, change domain.Cha if s.softMode == domain.SoftDisabled { return nil } - if s.softMode == domain.SoftParticipants { + if !domain.SoftAll(s.softMode) { if _, err := q.Participant(ctx, change.DID); errors.Is(err, sql.ErrNoRows) { return nil } else if err != nil { diff --git a/internal/storage/sqlite.go b/internal/storage/sqlite.go index a9aecae..fa3c6f5 100644 --- a/internal/storage/sqlite.go +++ b/internal/storage/sqlite.go @@ -34,7 +34,10 @@ var migration5 string //go:embed soft_schema.sql var softSchema string -const SchemaVersion = 5 +//go:embed lists_schema.sql +var listsSchema string + +const SchemaVersion = 6 type SQLite struct { conn *sql.DB @@ -47,7 +50,7 @@ func Open(path string) (*SQLite, error) { } func OpenWithSoft(path string, mode domain.SoftMode) (*SQLite, error) { - if mode < domain.SoftDisabled || mode > domain.SoftAllAuthors { + if mode < domain.SoftDisabled || mode > domain.SoftAllAuthorLists { return nil, errors.New("invalid soft vouch mode") } if err := os.MkdirAll(filepath.Dir(path), 0700); err != nil { @@ -126,7 +129,7 @@ func migrate(conn *sql.DB) error { return err } } - if version > 0 && version < SchemaVersion { + if version > 0 && version < 5 { if _, err := tx.Exec(migration5); err != nil { return err } @@ -137,6 +140,9 @@ func migrate(conn *sql.DB) error { if _, err := tx.Exec(softSchema); err != nil { return err } + if _, err := tx.Exec(listsSchema); err != nil { + return err + } if _, err := tx.Exec(fmt.Sprintf("PRAGMA user_version=%d", SchemaVersion)); err != nil { return err } @@ -183,12 +189,28 @@ func (s *SQLite) SetThreshold(ctx context.Context, did string, value float64) er } func (s *SQLite) Replace(ctx context.Context, did string, vouches []domain.Vouch, at time.Time) error { + return s.replace(ctx, did, vouches, nil, at) +} + +func (s *SQLite) ReplaceWithLists(ctx context.Context, did string, vouches []domain.Vouch, lists domain.ListSnapshot, at time.Time) error { + if !domain.SoftLists(s.softMode) { + return errors.New("soft lists are disabled") + } + return s.replace(ctx, did, vouches, &lists, at) +} + +func (s *SQLite) replace(ctx context.Context, did string, vouches []domain.Vouch, lists *domain.ListSnapshot, at time.Time) error { tx, err := s.conn.BeginTx(ctx, nil) if err != nil { return err } defer tx.Rollback() q := s.queries.WithTx(tx) + if lists != nil { + if err := replaceLists(ctx, q, did, *lists, at); err != nil { + return err + } + } previous, err := q.AuthorRecords(ctx, did) if err != nil { return err diff --git a/internal/storage/store.go b/internal/storage/store.go index 29464e1..9ff3c2d 100644 --- a/internal/storage/store.go +++ b/internal/storage/store.go @@ -33,6 +33,9 @@ type Store interface { CachedTrust(context.Context, string, time.Time) (trust.Result, error) SaveTrust(context.Context, string, trust.Result, []trust.Dependency, int64, time.Time) (bool, error) SoftReady(context.Context, string) (bool, error) + ReplaceWithLists(context.Context, string, []domain.Vouch, domain.ListSnapshot, time.Time) error + NeedsList(context.Context, domain.Change) (bool, error) + WantsListOwner(context.Context, string) (bool, error) } type Secrets interface { diff --git a/internal/storage/trust_cache.go b/internal/storage/trust_cache.go index 8944c38..e99d90c 100644 --- a/internal/storage/trust_cache.go +++ b/internal/storage/trust_cache.go @@ -46,11 +46,19 @@ func (s *SQLite) Incoming(ctx context.Context, target, tag string, mode domain.M return result, nil } all := int64(0) - if s.softMode == domain.SoftAllAuthors { + if domain.SoftAll(s.softMode) { all = 1 } soft, err := s.queries.SoftIncoming(ctx, db.SoftIncomingParams{Subject: target, Tag: tag, Fallback: fallback, AllAuthors: all, MaxRows: int64(limit)}) - return append(result, softRows(soft)...), err + if err != nil { + return nil, err + } + result = append(result, softRows(soft)...) + if !domain.SoftLists(s.softMode) { + return result, nil + } + lists, err := s.queries.ListIncoming(ctx, db.ListIncomingParams{Subject: target, Tag: tag, Fallback: fallback, AllAuthors: all, MaxRows: int64(limit)}) + return append(result, listRows(lists)...), err } func (s *SQLite) CachedTrust(ctx context.Context, key string, now time.Time) (trust.Result, error) { diff --git a/internal/trust/graph.go b/internal/trust/graph.go index e752cba..85f2f5a 100644 --- a/internal/trust/graph.go +++ b/internal/trust/graph.go @@ -34,6 +34,7 @@ type PathScore struct { DID string `json:"did"` Assertion *float64 `json:"assertion"` Source string `json:"source,omitempty"` + ListURI string `json:"list_uri,omitempty"` Score float64 `json:"score"` Comment string `json:"comment,omitempty"` } @@ -95,7 +96,7 @@ func vouchRank(v domain.Vouch) selectionRank { if v.Source == domain.FollowCollection { return followRank } - if v.Source == domain.BlockCollection { + if v.Source == domain.BlockCollection || v.Source == domain.ListBlockCollection { return softBlockRank } return explicitRank diff --git a/internal/ui/web/app.js b/internal/ui/web/app.js index 18540ca..67679fb 100644 --- a/internal/ui/web/app.js +++ b/internal/ui/web/app.js @@ -3,7 +3,7 @@ const AUTOCOMPLETE_DELAY_MS = 250; const VOUCH_PAGE_SIZE = 100; const Severity = Object.freeze({INFO: "info", ERROR: "error"}); -const VouchSource = Object.freeze({FOLLOW: "app.bsky.graph.follow", BLOCK: "app.bsky.graph.block"}); +const VouchSource = Object.freeze({FOLLOW: "app.bsky.graph.follow", BLOCK: "app.bsky.graph.block", LIST_BLOCK: "app.bsky.graph.listblock"}); const RATINGS = Object.freeze({ "-3": {label: "Blocked", color: "blocked"}, "-2": {label: "Avoid", color: "red"}, @@ -212,6 +212,7 @@ function rating(score) { // Show the source so that soft assertions retain their distinct labels. function assertionLabel(step) { + if (step.source === VouchSource.LIST_BLOCK) return `Bluesky blocked list (${step.assertion}): ${step.list_uri}`; if (step.source === VouchSource.FOLLOW) return `Bluesky follow (${step.assertion})`; if (step.source === VouchSource.BLOCK) return `Bluesky block (${step.assertion})`; return rating(step.assertion).label; diff --git a/nix/modules/appview.nix b/nix/modules/appview.nix index ae1e17a..a60708e 100644 --- a/nix/modules/appview.nix +++ b/nix/modules/appview.nix @@ -57,6 +57,7 @@ in }; enableSoftVouch = lib.mkEnableOption "Bluesky follows and blocks as virtual trust assertions"; + softVouchLists = lib.mkEnableOption "blocked-list soft assertions when enableSoftVouch is active"; softVouchAllAuthors = lib.mkEnableOption "soft assertions from all observed stream authors"; }; @@ -84,6 +85,7 @@ in PROVENANCE_HOP_PENALTY = toString cfg.hopPenalty; PROVENANCE_TRUST_CACHE_TTL = cfg.trustCacheTTL; ENABLE_SOFT_VOUCH = lib.boolToString cfg.enableSoftVouch; + ENABLE_SOFT_VOUCH_LISTS = lib.boolToString cfg.softVouchLists; ENABLE_SOFT_VOUCH_ALL_AUTHORS = lib.boolToString cfg.softVouchAllAuthors; }; }; diff --git a/nix/services/provenance.nix b/nix/services/provenance.nix index 09d54d4..594465a 100644 --- a/nix/services/provenance.nix +++ b/nix/services/provenance.nix @@ -72,6 +72,7 @@ let trust_cache_ttl=${lib.escapeShellArg cfg.trustCacheTTL} export PROVENANCE_TRUST_CACHE_TTL="''${PROVENANCE_TRUST_CACHE_TTL:-$trust_cache_ttl}" export ENABLE_SOFT_VOUCH="''${ENABLE_SOFT_VOUCH:-${lib.boolToString cfg.enableSoftVouch}}" + export ENABLE_SOFT_VOUCH_LISTS="''${ENABLE_SOFT_VOUCH_LISTS:-${lib.boolToString cfg.softVouchLists}}" export ENABLE_SOFT_VOUCH_ALL_AUTHORS="''${ENABLE_SOFT_VOUCH_ALL_AUTHORS:-${lib.boolToString cfg.softVouchAllAuthors}}" exec ${cfg.package}/bin/${binary} ''; @@ -105,6 +106,7 @@ in description = "Trust cache lifetime as a positive Go duration."; }; enableSoftVouch = lib.mkEnableOption "Bluesky follows and blocks as virtual assertions"; + softVouchLists = lib.mkEnableOption "blocked-list soft assertions when enableSoftVouch is active"; softVouchAllAuthors = lib.mkEnableOption "soft assertions from all observed authors"; uiPort = lib.mkOption { type = lib.types.port; diff --git a/nix/tests/modules.nix b/nix/tests/modules.nix index 5475c14..f28af40 100644 --- a/nix/tests/modules.nix +++ b/nix/tests/modules.nix @@ -33,6 +33,7 @@ let hopPenalty = 0.25; trustCacheTTL = "48h"; enableSoftVouch = true; + softVouchLists = true; softVouchAllAuthors = true; }; uiConfig.services.provenance.ui = { @@ -117,6 +118,7 @@ assert appUnit.environment.PROVENANCE_SLINGSHOT_URL == "https://records.example. assert appUnit.environment.PROVENANCE_HOP_PENALTY == toString 0.25; assert appUnit.environment.PROVENANCE_TRUST_CACHE_TTL == "48h"; assert appUnit.environment.ENABLE_SOFT_VOUCH == "true"; +assert appUnit.environment.ENABLE_SOFT_VOUCH_LISTS == "true"; assert appUnit.environment.ENABLE_SOFT_VOUCH_ALL_AUTHORS == "true"; assert !(appUnit.environment ? PROVENANCE_SERVICE_KEY); assert !(uiUnit.environment ? PROVENANCE_ENCRYPTION_KEY); diff --git a/sqlc.yaml b/sqlc.yaml index b9ff264..d6e86b2 100644 --- a/sqlc.yaml +++ b/sqlc.yaml @@ -4,6 +4,7 @@ sql: schema: - internal/storage/schema.sql - internal/storage/soft_schema.sql + - internal/storage/lists_schema.sql queries: internal/storage/queries.sql gen: go: