diff --git a/README.md b/README.md index e1ba2fa..8bf4763 100644 --- a/README.md +++ b/README.md @@ -101,6 +101,36 @@ Keep the service key and encryption key outside the repository. | `PROVENANCE_SLINGSHOT_URL` | `https://slingshot.microcosm.blue` | Record cache for missing event payloads | | `PROVENANCE_HOP_PENALTY` | `0.1` | Penalty for each edge after the first edge | | `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` | + +### Soft vouches + +Soft vouches are optional. A Bluesky follow scores `0.5`. A Bluesky block scores `-0.5`, not the hard-block score `-3`. +These scores affect trust paths, reputation author eligibility, reputation sums, and reputation means. +Explicit vouches take precedence, including zero scores and hard blocks. Otherwise, a Bluesky block overrides a follow. +Duplicate records never multiply a pair's contribution. +Soft vouches apply to default queries and topic fallback queries, but not tag-only queries. +An applicable explicit topic or default assertion overrides a soft assertion. +The public vouch schema and write endpoints retain integer-only scores. + +The participant mode imports follows and blocks from each participant's PDS at bootstrap or explicit sync. +Existing participants receive this import at startup when the feature first becomes active. +The all-author mode also indexes subsequent stream records from nonparticipants. It does not enroll those authors or import their repositories. +Both modes subscribe to the follow and block collections and use Slingshot for absent event payloads. +Disabled mode does not fetch, subscribe to, ingest, or calculate soft assertions. + +SQLite stores soft assertions separately, with their collection, original record key, fixed score, replay timestamps, and account state. +Path responses include the incoming assertion's `source`. The UI shows the Bluesky source for soft path assertions. +Cache keys separate feature modes. Cache dependencies include the author, collection, and original record key. +The existing TTL policy still applies to new edges, precedence changes, and account status changes. +Each feature mode uses a separate stream cursor. Replay remains limited by the stream provider's retention. +After a long outage or a disabled period, use explicit participant sync to recover current records. +The all-author mode has no repository-wide recovery for nonparticipants. + +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. ## API diff --git a/cmd/appview/main.go b/cmd/appview/main.go index c2afdfb..ec60a71 100644 --- a/cmd/appview/main.go +++ b/cmd/appview/main.go @@ -27,7 +27,7 @@ func run() error { if err != nil { return err } - store, err := storage.Open(cfg.Database) + store, err := storage.OpenWithSoft(cfg.Database, cfg.SoftMode) if err != nil { return err } @@ -36,8 +36,8 @@ func run() error { defer cancel() // This selection controls calculation, accepted query parameters, and the API schema. algorithm := algorithms.Optimistic{} - service := appview.NewWithOptions(store, atproto.NewRepository(), cfg.Limits, appview.Options{Trust: algorithm, Reputation: reputationalg.DirectSum{}, CacheTTL: cfg.TrustCacheTTL}) - stream, err := microcosm.New(cfg.JetstreamURL, cfg.SlingshotURL) + service := appview.NewWithOptions(store, atproto.NewRepositoryWithSoft(cfg.SoftMode), cfg.Limits, appview.Options{Trust: algorithm, Reputation: reputationalg.DirectSum{}, CacheTTL: cfg.TrustCacheTTL, SoftMode: cfg.SoftMode}) + stream, err := microcosm.NewWithSoft(cfg.JetstreamURL, cfg.SlingshotURL, cfg.SoftMode) if err != nil { return err } diff --git a/docs/security-review.md b/docs/security-review.md index 7bb0df4..95719b3 100644 --- a/docs/security-review.md +++ b/docs/security-review.md @@ -33,6 +33,10 @@ - Microcosm endpoints come only from operator configuration, not browser input. - Missing payload lookup requires the stream's exact URI and CID. - Vouch comments are public, untrusted text. Clients must not interpret them as HTML. +- Soft records require valid subjects, timestamps, collection types, and repository-bound record URIs. +- 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. ## Correctness @@ -49,6 +53,8 @@ The schema migration gives existing records an empty comment and preserves their Relative reputation sums non-block direct assertions only from authors whose calculated or self trust meets the threshold. Unknown and blocked authors never qualify, even at negative thresholds. Empty sums return zero without a trust-score clamp. The trust cache includes algorithm metadata, limits, topic selection, normalized identities, and account options in its key. +Soft feature modes also isolate cache keys. Explicit assertions override soft assertions, and soft blocks override follows. +Soft blocks remain negative assertions, not hard blocks. Duplicate soft records contribute only once per pair. Record mutations invalidate indexed path dependencies in the same transaction as the mutation. Graph revisions reject cache publication after concurrent mutations. Cache errors and traversal errors never produce partial reputation sums. @@ -74,6 +80,9 @@ Long outages need explicit PDS recovery, not scheduled polling. Bootstrap watermarks depend on synchronized UTC clocks. The health endpoint does not measure stream freshness. 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. 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/comments_test.go b/internal/appview/comments_test.go index f7583f3..920696c 100644 --- a/internal/appview/comments_test.go +++ b/internal/appview/comments_test.go @@ -62,7 +62,7 @@ func TestPathComments(t *testing.T) { func TestCommentResultCopy(t *testing.T) { score := 2.0 - assertion := 2 + assertion := 2.0 original := trust.Result{Status: trust.CalculatedStatus, Score: &score, Path: []string{alice, bob}, PathScores: []trust.PathScore{ {DID: alice, Score: domain.SelfScore}, {DID: bob, Assertion: &assertion, Score: score}, }} diff --git a/internal/appview/http.go b/internal/appview/http.go index d367b53..a91dc44 100644 --- a/internal/appview/http.go +++ b/internal/appview/http.go @@ -63,6 +63,8 @@ func Handler(service *Service, key string) http.Handler { endpoints = append(endpoints, endpoint{method: http.MethodGet, path: "/v1/openapi.json", response: scalarSchema("object")}) 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 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/openapi.go b/internal/appview/openapi.go index f0415af..3645a55 100644 --- a/internal/appview/openapi.go +++ b/internal/appview/openapi.go @@ -155,9 +155,9 @@ 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": "integer", "nullable": true, "minimum": domain.Blocked, "maximum": domain.MaxScore}, "score": score, "comment": comment}, "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, "source": map[string]any{"type": "string", "enum": []string{domain.Collection, domain.FollowCollection, domain.BlockCollection}}}, "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}, "author", "subject", "score", "tag"), + "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 ba2ba52..3be9cc4 100644 --- a/internal/appview/service.go +++ b/internal/appview/service.go @@ -22,13 +22,15 @@ import ( ) const ( - 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" + StreamSource = "jetstream" + InitialRetryDelay = time.Second + MaxRetryDelay = 30 * time.Second + ParticipantLimit = 10000 + TagLimit = 20 + VouchPageLimit = 100 + identityTimeout = 5 * time.Second ) type Repository interface { @@ -47,6 +49,7 @@ type Service struct { algorithm trust.Algorithm reputation reputation.Algorithm cacheTTL time.Duration + softMode domain.SoftMode now func() time.Time syncMu sync.Mutex } @@ -69,6 +72,7 @@ type Options struct { Trust trust.Algorithm Reputation reputation.Algorithm CacheTTL time.Duration + SoftMode domain.SoftMode } func NewWithOptions(store storage.Store, repo Repository, limits trust.Limits, options Options) *Service { @@ -81,7 +85,7 @@ func NewWithOptions(store storage.Store, repo Repository, limits trust.Limits, o if options.CacheTTL == 0 { options.CacheTTL = trust.DefaultCacheTTL } - return &Service{store: store, repo: repo, limits: limits, algorithm: options.Trust, reputation: options.Reputation, cacheTTL: options.CacheTTL, now: time.Now} + return &Service{store: store, repo: repo, limits: limits, algorithm: options.Trust, reputation: options.Reputation, cacheTTL: options.CacheTTL, softMode: options.SoftMode, now: time.Now} } func (s *Service) Join(ctx context.Context, did string) error { @@ -90,7 +94,7 @@ func (s *Service) Join(ctx context.Context, did string) error { } s.syncMu.Lock() defer s.syncMu.Unlock() - if _, err := s.store.Cursor(ctx, StreamSource, time.Now().UnixMicro()); err != nil { + if _, err := s.store.Cursor(ctx, s.streamSource(), time.Now().UnixMicro()); err != nil { return err } if err := s.store.Join(ctx, did); err != nil { @@ -101,7 +105,16 @@ func (s *Service) Join(ctx context.Context, did string) error { return err } if p.BootstrapUS != 0 { - return nil + if s.softMode == domain.SoftDisabled { + return nil + } + ready, err := s.store.SoftReady(ctx, did) + if err != nil { + return err + } + if ready { + return nil + } } return s.bootstrap(ctx, did) } @@ -134,9 +147,11 @@ func pathComments(result trust.Result, records map[[2]string]domain.Vouch) trust // Copy the returned nodes because an algorithm can reuse its result buffers. result.PathScores = slices.Clone(result.PathScores) result.PathScores[0].Comment = "" + result.PathScores[0].Source = "" 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 } return result } @@ -291,7 +306,23 @@ func (s *Service) BootstrapPending(ctx context.Context) error { if err := ctx.Err(); err != nil { return err } - if p.BootstrapUS != 0 || !p.Active { + if !p.Active { + continue + } + if p.BootstrapUS != 0 { + if s.softMode == domain.SoftDisabled { + continue + } + ready, err := s.store.SoftReady(ctx, p.DID) + if err != nil { + return err + } + if ready { + continue + } + if err := s.Sync(ctx, p.DID); err != nil { + slog.Warn("soft bootstrap failed", "did", p.DID, "error", err) + } continue } if err := s.Join(ctx, p.DID); err != nil { @@ -304,7 +335,18 @@ 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() - return s.store.Apply(ctx, StreamSource, change) + 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.SoftAllAuthors { + return softAllStream + } + if s.softMode == domain.SoftParticipants { + return softParticipantStream + } + return StreamSource } func (s *Service) wanted(ctx context.Context, did string, stamp int64) (bool, error) { @@ -320,7 +362,7 @@ func (s *Service) Run(ctx context.Context, stream Stream) { delay := InitialRetryDelay prepared := false for { - cursor, err := s.store.Cursor(ctx, StreamSource, time.Now().UnixMicro()) + cursor, err := s.store.Cursor(ctx, s.streamSource(), time.Now().UnixMicro()) if err == nil && !prepared { if err := s.BootstrapPending(ctx); err != nil { slog.Warn("pending bootstrap failed", "error", err) diff --git a/internal/appview/soft_test.go b/internal/appview/soft_test.go new file mode 100644 index 0000000..decbaa8 --- /dev/null +++ b/internal/appview/soft_test.go @@ -0,0 +1,90 @@ +package appview + +import ( + "context" + "database/sql" + "errors" + "path/filepath" + "testing" + "time" + + "provenance/internal/domain" + "provenance/internal/storage" + "provenance/internal/trust" + "provenance/internal/trust/algorithms" +) + +func TestSoftService(t *testing.T) { + ctx := context.Background() + path := filepath.Join(t.TempDir(), "soft.db") + store, err := storage.OpenWithSoft(path, domain.SoftParticipants) + if err != nil { + t.Fatal(err) + } + defer store.Close() + for _, did := range []string{alice, carol} { + if err := store.Join(ctx, did); err != nil { + t.Fatal(err) + } + } + first := domain.Vouch{Author: alice, Subject: carol, Source: domain.FollowCollection, RecordKey: "follow", Score: domain.FollowScore} + last := domain.Vouch{Author: carol, Subject: bob, Source: domain.BlockCollection, RecordKey: "block", Score: domain.SoftBlockScore} + if err := store.Replace(ctx, alice, []domain.Vouch{first}, time.UnixMicro(100)); err != nil { + t.Fatal(err) + } + if err := store.Replace(ctx, carol, []domain.Vouch{last}, time.UnixMicro(100)); err != nil { + t.Fatal(err) + } + limits := trust.Limits{States: 100, Edges: 100, Penalty: domain.DefaultPenalty} + disabledStore, err := storage.Open(path) + if err != nil { + t.Fatal(err) + } + defer disabledStore.Close() + disabled := New(disabledStore, &repository{}, limits) + got, err := disabled.Reputation(ctx, alice, bob, "", domain.Default) + if err != nil || got.Status != trust.UnknownStatus { + t.Fatalf("disabled score: %+v %v", got, err) + } + for _, algorithm := range []trust.Algorithm{algorithms.Optimistic{}, algorithms.ShortestPath{}} { + s := NewWithOptions(store, &repository{}, limits, Options{Trust: algorithm, SoftMode: domain.SoftParticipants}) + got, err := s.Reputation(ctx, alice, bob, "", domain.Default) + if err != nil || got.Status != trust.CalculatedStatus || got.Score == nil || *got.Score != -.6 { + t.Fatalf("soft path: %+v %v", got, err) + } + if len(got.PathScores) != 3 || got.PathScores[1].Source != domain.FollowCollection || got.PathScores[2].Source != domain.BlockCollection || *got.PathScores[1].Assertion != .5 || *got.PathScores[2].Assertion != -.5 { + t.Fatalf("path provenance: %+v", got) + } + rep, err := s.RelativeReputation(ctx, alice, bob, "", domain.Default, .5) + if err != nil || rep.AbsoluteScore != -.5 || rep.MeanScore != -.5 { + t.Fatalf("soft reputation: %+v %v", rep, err) + } + rep, err = s.RelativeReputation(ctx, alice, bob, "work", domain.Tagged, .5) + if err != nil || rep.Score != 0 { + t.Fatalf("topic-only reputation: %+v %v", rep, err) + } + } + allStore, err := storage.OpenWithSoft(path, domain.SoftAllAuthors) + if err != nil { + t.Fatal(err) + } + defer allStore.Close() + const observer = "did:plc:observer" + for _, v := range []domain.Vouch{{Author: alice, Subject: observer, Source: domain.FollowCollection, RecordKey: "observer", Score: .5}, {Author: observer, Subject: bob, Source: domain.FollowCollection, RecordKey: "follow", Score: .5}} { + if err := allStore.Apply(ctx, "all-stream", domain.Change{Kind: domain.RecordChange, DID: v.Author, Source: v.Source, RecordKey: v.RecordKey, Vouch: &v, TimeUS: 200}); err != nil { + t.Fatal(err) + } + } + all := NewWithOptions(allStore, &repository{}, limits, Options{SoftMode: domain.SoftAllAuthors}) + rep, err := all.RelativeReputation(ctx, alice, bob, "", domain.Default, .5) + if err != nil || rep.Score != 0 || rep.MeanScore != 0 { + t.Fatalf("all-author reputation: %+v %v", rep, err) + } + if _, err := allStore.Participant(ctx, observer); !errors.Is(err, sql.ErrNoRows) { + t.Fatalf("observer enrollment: %v", err) + } + got, err = disabled.Reputation(ctx, alice, bob, "", domain.Default) + if err != nil || got.Status != trust.UnknownStatus { + t.Fatalf("disabled cache isolation: %+v %v", got, err) + } +} diff --git a/internal/appview/trust_cache.go b/internal/appview/trust_cache.go index 7ace3b3..d5e6caa 100644 --- a/internal/appview/trust_cache.go +++ b/internal/appview/trust_cache.go @@ -41,7 +41,7 @@ func (r *trustReader) Score(ctx context.Context, target string) (trust.Result, e } input := r.input input.Target = target - key, err := cacheKey(r.service.algorithm.Describe(), input, r.tag, r.mode) + key, err := cacheKey(r.service.algorithm.Describe(), input, r.tag, r.mode, r.service.softMode) if err != nil { return trust.Result{}, err } @@ -92,7 +92,7 @@ func (r *trustReader) Score(ctx context.Context, target string) (trust.Result, e } // Cache keys include all calculation inputs, but never a private participant threshold. -func cacheKey(descriptor trust.Descriptor, input trust.Input, tag string, mode domain.Mode) (string, error) { +func cacheKey(descriptor trust.Descriptor, input trust.Input, tag string, mode domain.Mode, softMode domain.SoftMode) (string, error) { accounts := trust.Accounts{} for name, values := range input.Accounts { values = slices.Clone(values) @@ -105,7 +105,8 @@ func cacheKey(descriptor trust.Descriptor, input trust.Input, tag string, mode d Mode domain.Mode Limits trust.Limits Accounts trust.Accounts - }{descriptor, input.Source, input.Target, tag, mode, input.Limits, accounts}) + SoftMode domain.SoftMode `json:"soft_mode,omitempty"` + }{descriptor, input.Source, input.Target, tag, mode, input.Limits, accounts, softMode}) if err != nil { return "", err } @@ -129,7 +130,14 @@ func pathDependencies(result trust.Result, records map[[2]string]domain.Vouch) [ if !exists { continue } - dependencies = append(dependencies, trust.Dependency{Author: edge.Author, RecordKey: domain.RecordKey(edge.Subject, edge.Tag)}) + source, key := edge.Source, edge.RecordKey + if source == "" { + source = domain.Collection + } + if !domain.SoftSource(source) { + key = domain.RecordKey(edge.Subject, edge.Tag) + } + dependencies = append(dependencies, trust.Dependency{Author: edge.Author, Source: source, RecordKey: key}) } return dependencies } diff --git a/internal/atproto/oauth.go b/internal/atproto/oauth.go index e033298..c8d0854 100644 --- a/internal/atproto/oauth.go +++ b/internal/atproto/oauth.go @@ -135,9 +135,10 @@ func (o *OAuth) PutVouch(ctx context.Context, session Session, v domain.Vouch) e if err != nil { return err } + score := int(v.Score) body := map[string]any{ "repo": session.DID, "collection": domain.Collection, "rkey": domain.RecordKey(v.Subject, v.Tag), - "record": Record{Type: domain.Collection, Subject: v.Subject, Score: &v.Score, Tag: v.Tag, CreatedAt: time.Now().UTC().Format(time.RFC3339Nano), Comment: v.Comment}, + "record": Record{Type: domain.Collection, Subject: v.Subject, Score: &score, Tag: v.Tag, CreatedAt: time.Now().UTC().Format(time.RFC3339Nano), Comment: v.Comment}, } return sess.APIClient().Post(ctx, "com.atproto.repo.putRecord", body, nil) } diff --git a/internal/atproto/oauth_test.go b/internal/atproto/oauth_test.go index f53bfd0..80a1909 100644 --- a/internal/atproto/oauth_test.go +++ b/internal/atproto/oauth_test.go @@ -183,7 +183,7 @@ func TestPublishComment(t *testing.T) { if err := client.PutVouch(ctx, session, v); err != nil { t.Fatal(err) } - if published.Comment != comment || published.Score == nil || *published.Score != v.Score { + if published.Comment != comment || published.Score == nil || float64(*published.Score) != v.Score { t.Fatalf("published: %+v", published) } } diff --git a/internal/atproto/records.go b/internal/atproto/records.go index 477de83..2552587 100644 --- a/internal/atproto/records.go +++ b/internal/atproto/records.go @@ -47,10 +47,15 @@ type Actor struct { type Repository struct { directory identity.Directory client *http.Client + softMode domain.SoftMode } func NewRepository() *Repository { - return &Repository{identity.DefaultDirectory(), &http.Client{Timeout: RequestTimeout, Transport: ssrf.PublicOnlyTransport()}} + return NewRepositoryWithSoft(domain.SoftDisabled) +} + +func NewRepositoryWithSoft(mode domain.SoftMode) *Repository { + return &Repository{directory: identity.DefaultDirectory(), client: &http.Client{Timeout: RequestTimeout, Transport: ssrf.PublicOnlyTransport()}, softMode: mode} } func ValidDID(did string) error { _, err := syntax.ParseDID(did); return err } @@ -77,7 +82,7 @@ func ValidateRecord(record Record) error { if record.Score == nil { return errors.New("record score is required") } - if err := domain.ValidateScore(*record.Score); err != nil { + if err := domain.ValidateScore(float64(*record.Score)); err != nil { return err } if err := domain.ValidateTag(record.Tag); err != nil { @@ -155,9 +160,16 @@ func (r *Repository) Snapshot(ctx context.Context, did string) ([]domain.Vouch, return nil, errors.New("duplicate repository record") } seenPairs[key] = true - result = append(result, domain.Vouch{Author: did, Subject: v.Subject, Tag: v.Tag, Score: *v.Score, CreatedAt: v.CreatedAt, Comment: v.Comment}) + result = append(result, domain.Vouch{Author: did, Subject: v.Subject, Tag: v.Tag, Score: float64(*v.Score), CreatedAt: v.CreatedAt, Comment: v.Comment}) } if page.Cursor == "" { + if r.softMode != domain.SoftDisabled { + soft, err := r.softSnapshot(ctx, did, host) + if err != nil { + return nil, err + } + result = append(result, soft...) + } return result, nil } if seenCursors[page.Cursor] { diff --git a/internal/atproto/records_test.go b/internal/atproto/records_test.go index b89a010..6366d50 100644 --- a/internal/atproto/records_test.go +++ b/internal/atproto/records_test.go @@ -40,7 +40,7 @@ func TestSnapshotValidation(t *testing.T) { if (err != nil) != test.wantError { t.Fatalf("got %+v, %v", got, err) } - if !test.wantError && (len(got) != 1 || got[0].Author != testDID || got[0].Score != score || got[0].Comment != record.Comment) { + if !test.wantError && (len(got) != 1 || got[0].Author != testDID || got[0].Score != float64(score) || got[0].Comment != record.Comment) { t.Fatalf("got %+v", got) } }) diff --git a/internal/atproto/soft.go b/internal/atproto/soft.go new file mode 100644 index 0000000..3ad1ddc --- /dev/null +++ b/internal/atproto/soft.go @@ -0,0 +1,91 @@ +package atproto + +import ( + "context" + "errors" + "fmt" + "net/url" + "strings" + "time" + + "github.com/bluesky-social/indigo/atproto/syntax" + + "provenance/internal/domain" +) + +type SoftRecord struct { + Type string `json:"$type"` + Subject string `json:"subject"` + CreatedAt string `json:"createdAt"` +} + +func ValidateSoftRecord(record SoftRecord, source string) error { + if !domain.SoftSource(source) || record.Type != source { + return errors.New("invalid soft record type") + } + if err := ValidDID(record.Subject); err != nil { + return err + } + _, err := time.Parse(time.RFC3339Nano, record.CreatedAt) + return err +} + +func (r *Repository) softSnapshot(ctx context.Context, did, host string) ([]domain.Vouch, error) { + result := []domain.Vouch{} + for _, source := range []string{domain.FollowCollection, domain.BlockCollection} { + cursor := "" + cursors := map[string]struct{}{} + keys := 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 SoftRecord `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 soft repository page") + } + examined += len(page.Records) + if examined > MaxRecords || len(cursors) > MaxRecords/PageSize { + return nil, errors.New("soft 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("soft record URI does not match 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 soft repository record") + } + keys[key] = struct{}{} + if ValidateSoftRecord(item.Value, source) != nil { + continue + } + result = append(result, domain.Vouch{Author: did, Subject: item.Value.Subject, Score: domain.SourceScore(source), Source: source, RecordKey: key, CreatedAt: item.Value.CreatedAt}) + } + if page.Cursor == "" { + break + } + if _, exists := cursors[page.Cursor]; exists { + return nil, errors.New("repeated soft repository cursor") + } + cursors[page.Cursor] = struct{}{} + cursor = page.Cursor + } + } + return result, nil +} diff --git a/internal/atproto/soft_test.go b/internal/atproto/soft_test.go new file mode 100644 index 0000000..f29db39 --- /dev/null +++ b/internal/atproto/soft_test.go @@ -0,0 +1,52 @@ +package atproto + +import ( + "context" + "net/http" + "testing" + + "provenance/internal/domain" +) + +func TestSoftSnapshot(t *testing.T) { + for _, mode := range []domain.SoftMode{domain.SoftDisabled, domain.SoftParticipants} { + calls := map[string]int{} + repo := &Repository{directory: directory{}, softMode: mode, client: &http.Client{Transport: transport(func(r *http.Request) (*http.Response, error) { + source := r.URL.Query().Get("collection") + calls[source]++ + if r.URL.Host != "pds.example.com" || r.URL.Query().Get("repo") != testDID { + t.Error("wrong soft snapshot authority") + } + if source == domain.Collection { + return response(map[string]any{"records": []any{}}), nil + } + return response(map[string]any{"records": []any{map[string]any{"uri": "at://" + testDID + "/" + source + "/record", "value": SoftRecord{Type: source, Subject: "did:plc:bob", CreatedAt: "2026-10-09T00:00:00Z"}}}}), nil + })}} + vouches, err := repo.Snapshot(context.Background(), testDID) + if err != nil { + t.Fatal(err) + } + if mode == domain.SoftDisabled { + if len(vouches) != 0 || len(calls) != 1 { + t.Fatalf("disabled soft snapshot: %+v %+v", vouches, calls) + } + continue + } + if len(vouches) != 2 || vouches[0].Source != domain.FollowCollection || vouches[0].Score != .5 || vouches[1].Source != domain.BlockCollection || vouches[1].Score != -.5 || vouches[0].RecordKey != "record" { + t.Fatalf("soft snapshot: %+v", vouches) + } + } +} + +func TestInvalidSoftSnapshot(t *testing.T) { + repo := &Repository{directory: directory{}, softMode: domain.SoftParticipants, client: &http.Client{Transport: transport(func(r *http.Request) (*http.Response, error) { + source := r.URL.Query().Get("collection") + if source == domain.Collection { + return response(map[string]any{"records": []any{}}), nil + } + return response(map[string]any{"records": []any{map[string]any{"uri": "at://did:plc:forged/" + source + "/record", "value": SoftRecord{Type: source, Subject: "did:plc:bob", CreatedAt: "2026-10-09T00:00:00Z"}}}}), nil + })}} + if _, err := repo.Snapshot(context.Background(), testDID); err == nil { + t.Fatal("accepted forged soft record URI") + } +} diff --git a/internal/config/config.go b/internal/config/config.go index ba45391..69d5f53 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -26,6 +26,7 @@ type Appview struct { SlingshotURL string Limits trust.Limits TrustCacheTTL time.Duration + SoftMode domain.SoftMode } type UI struct { @@ -65,10 +66,25 @@ func LoadAppview() (Appview, error) { if err != nil || ttl <= 0 { return Appview{}, errors.New("trust cache TTL must be a positive duration") } + soft, err := strconv.ParseBool(value("ENABLE_SOFT_VOUCH", "false")) + if err != nil { + return Appview{}, errors.New("ENABLE_SOFT_VOUCH must be a boolean") + } + all, err := strconv.ParseBool(value("ENABLE_SOFT_VOUCH_ALL_AUTHORS", "false")) + if err != nil || all && !soft { + return Appview{}, errors.New("ENABLE_SOFT_VOUCH_ALL_AUTHORS requires ENABLE_SOFT_VOUCH=true and a boolean value") + } + softMode := domain.SoftDisabled + if soft { + softMode = domain.SoftParticipants + } + if all { + softMode = domain.SoftAllAuthors + } 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), - Limits: trust.Limits{States: trust.DefaultStateLimit, Edges: trust.DefaultEdgeLimit, Penalty: penalty}, TrustCacheTTL: ttl, + Limits: trust.Limits{States: trust.DefaultStateLimit, Edges: trust.DefaultEdgeLimit, Penalty: penalty}, TrustCacheTTL: ttl, SoftMode: softMode, }, nil } diff --git a/internal/config/config_test.go b/internal/config/config_test.go index c7acd96..b959868 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -5,6 +5,7 @@ import ( "testing" "time" + "provenance/internal/domain" "provenance/internal/trust" ) @@ -13,6 +14,8 @@ func TestServicePorts(t *testing.T) { t.Setenv("PROVENANCE_ENCRYPTION_KEY", "test-key") t.Setenv("PROVENANCE_HOP_PENALTY", "") t.Setenv("PROVENANCE_TRUST_CACHE_TTL", "") + t.Setenv("ENABLE_SOFT_VOUCH", "") + t.Setenv("ENABLE_SOFT_VOUCH_ALL_AUTHORS", "") for _, name := range []string{"PROVENANCE_UI_ADDR", "PROVENANCE_APPVIEW_ADDR", "PROVENANCE_PUBLIC_ORIGIN", "PROVENANCE_APPVIEW_URL"} { t.Setenv(name, "") } @@ -47,6 +50,8 @@ func TestServicePorts(t *testing.T) { } func TestTrustCacheTTL(t *testing.T) { + t.Setenv("ENABLE_SOFT_VOUCH", "") + t.Setenv("ENABLE_SOFT_VOUCH_ALL_AUTHORS", "") t.Setenv("PROVENANCE_SERVICE_KEY", strings.Repeat("x", MinServiceKeyLength)) t.Setenv("PROVENANCE_HOP_PENALTY", "") for _, test := range []struct { @@ -66,3 +71,27 @@ func TestTrustCacheTTL(t *testing.T) { } } } + +func TestSoftFlags(t *testing.T) { + t.Setenv("PROVENANCE_SERVICE_KEY", strings.Repeat("x", MinServiceKeyLength)) + t.Setenv("PROVENANCE_HOP_PENALTY", "") + t.Setenv("PROVENANCE_TRUST_CACHE_TTL", "") + for _, test := range []struct { + enabled, all string + mode domain.SoftMode + }{{"", "", domain.SoftDisabled}, {"false", "false", domain.SoftDisabled}, {"true", "false", domain.SoftParticipants}, {"true", "true", domain.SoftAllAuthors}} { + t.Setenv("ENABLE_SOFT_VOUCH", test.enabled) + t.Setenv("ENABLE_SOFT_VOUCH_ALL_AUTHORS", test.all) + cfg, err := LoadAppview() + if err != nil || cfg.SoftMode != test.mode { + t.Fatalf("flags %s %s: %+v %v", test.enabled, test.all, cfg, err) + } + } + for _, flags := range [][2]string{{"false", "true"}, {"invalid", "false"}, {"true", "invalid"}} { + t.Setenv("ENABLE_SOFT_VOUCH", flags[0]) + t.Setenv("ENABLE_SOFT_VOUCH_ALL_AUTHORS", flags[1]) + if _, err := LoadAppview(); err == nil { + t.Fatalf("accepted flags %v", flags) + } + } +} diff --git a/internal/domain/change.go b/internal/domain/change.go index cdb2dbf..8098532 100644 --- a/internal/domain/change.go +++ b/internal/domain/change.go @@ -13,6 +13,7 @@ type Change struct { DID string TimeUS int64 RecordKey string + Source string Vouch *Vouch Active bool Status string diff --git a/internal/domain/soft.go b/internal/domain/soft.go new file mode 100644 index 0000000..b4182a8 --- /dev/null +++ b/internal/domain/soft.go @@ -0,0 +1,38 @@ +package domain + +import "errors" + +type SoftMode int + +const ( + SoftDisabled SoftMode = iota + SoftParticipants + SoftAllAuthors +) + +const ( + FollowCollection = "app.bsky.graph.follow" + BlockCollection = "app.bsky.graph.block" + FollowScore = 0.5 + SoftBlockScore = -0.5 +) + +func SoftSource(source string) bool { return source == FollowCollection || source == BlockCollection } + +func SourceScore(source string) float64 { + if source == FollowCollection { + return FollowScore + } + return SoftBlockScore +} + +// Virtual assertions have fixed weights. Explicit records retain integer-only validation. +func ValidateVouch(v Vouch) error { + if v.Source == "" || v.Source == Collection { + return ValidateScore(v.Score) + } + if !SoftSource(v.Source) || v.Score != SourceScore(v.Source) || v.Tag != "" || v.Comment != "" || v.RecordKey == "" { + return errors.New("invalid soft vouch") + } + return nil +} diff --git a/internal/domain/soft_test.go b/internal/domain/soft_test.go new file mode 100644 index 0000000..7eb5cae --- /dev/null +++ b/internal/domain/soft_test.go @@ -0,0 +1,24 @@ +package domain + +import ( + "math" + "testing" +) + +func TestSoftValidation(t *testing.T) { + for _, score := range []float64{.5, -.5, math.NaN(), math.Inf(1)} { + if ValidateScore(score) == nil { + t.Fatalf("explicit fractional score accepted: %v", score) + } + } + for _, source := range []string{FollowCollection, BlockCollection} { + v := Vouch{Source: source, RecordKey: "record", Score: SourceScore(source)} + if err := ValidateVouch(v); err != nil { + t.Fatal(err) + } + v.Score = 2 + if ValidateVouch(v) == nil { + t.Fatal("soft source accepted an arbitrary weight") + } + } +} diff --git a/internal/domain/vouch.go b/internal/domain/vouch.go index 8c47fe4..44ed408 100644 --- a/internal/domain/vouch.go +++ b/internal/domain/vouch.go @@ -30,12 +30,14 @@ const ( ) type Vouch struct { - Author string `json:"author"` - Subject string `json:"subject"` - Tag string `json:"tag"` - Score int `json:"score"` - 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"` + CreatedAt string `json:"createdAt"` + Comment string `json:"comment,omitempty"` } func ValidateTag(tag string) error { @@ -45,8 +47,8 @@ func ValidateTag(tag string) error { return nil } -func ValidateScore(score int) error { - if score < Blocked || score > MaxScore { +func ValidateScore(score float64) error { + if math.IsNaN(score) || math.IsInf(score, 0) || math.Trunc(score) != score || score < Blocked || score > MaxScore { return errors.New("assertion score must be an integer from -3 through 2") } return nil diff --git a/internal/domain/vouch_test.go b/internal/domain/vouch_test.go index d52c8d6..62cf5d0 100644 --- a/internal/domain/vouch_test.go +++ b/internal/domain/vouch_test.go @@ -17,7 +17,7 @@ func TestValidation(t *testing.T) { t.Fatal("accepted invalid UTF-8") } for _, score := range []int{-4, 3} { - if ValidateScore(score) == nil { + if ValidateScore(float64(score)) == nil { t.Fatalf("accepted %d", score) } } diff --git a/internal/microcosm/client.go b/internal/microcosm/client.go index 739f3f0..68da2b3 100644 --- a/internal/microcosm/client.go +++ b/internal/microcosm/client.go @@ -28,9 +28,17 @@ type Client struct { streamURL string slingshotURL string http *http.Client + softMode domain.SoftMode } func New(streamURL, slingshotURL string) (*Client, error) { + return NewWithSoft(streamURL, slingshotURL, domain.SoftDisabled) +} + +func NewWithSoft(streamURL, slingshotURL string, mode domain.SoftMode) (*Client, error) { + if mode < domain.SoftDisabled || mode > domain.SoftAllAuthors { + return nil, errors.New("invalid soft vouch mode") + } if err := validateURL(streamURL, "wss", "ws"); err != nil { return nil, err } @@ -44,7 +52,7 @@ func New(streamURL, slingshotURL string) (*Client, error) { // Endpoint configuration is operator-controlled, so explicit private self-hosted instances are valid. transport := http.DefaultTransport.(*http.Transport).Clone() client := &http.Client{Transport: transport, Timeout: atproto.RequestTimeout, CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }} - return &Client{streamURL: streamURL, slingshotURL: strings.TrimRight(slingshotURL, "/"), http: client}, nil + return &Client{streamURL: streamURL, slingshotURL: strings.TrimRight(slingshotURL, "/"), http: client, softMode: mode}, nil } func loopback(host string) bool { return host == "127.0.0.1" || host == "::1" || host == "localhost" } @@ -61,37 +69,43 @@ func validateURL(raw, secure, development string) error { } func (c *Client) record(ctx context.Context, did, key, cid string) (atproto.Record, error) { - params := url.Values{"repo": {did}, "collection": {domain.Collection}, "rkey": {key}, "cid": {cid}} + var record atproto.Record + err := c.rawRecord(ctx, did, domain.Collection, key, cid, &record) + return record, err +} + +func (c *Client) rawRecord(ctx context.Context, did, source, key, cid string, record any) error { + params := url.Values{"repo": {did}, "collection": {source}, "rkey": {key}, "cid": {cid}} req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.slingshotURL+"/xrpc/com.atproto.repo.getRecord?"+params.Encode(), nil) if err != nil { - return atproto.Record{}, err + return err } req.Header.Set("User-Agent", UserAgent) resp, err := c.http.Do(req) if err != nil { - return atproto.Record{}, err + return err } defer resp.Body.Close() if resp.StatusCode != http.StatusOK { - return atproto.Record{}, fmt.Errorf("Slingshot returned HTTP %d for event hydration", resp.StatusCode) + return fmt.Errorf("Slingshot returned HTTP %d for event hydration", resp.StatusCode) } data, err := io.ReadAll(io.LimitReader(resp.Body, atproto.MaxResponseBytes+1)) if err != nil { - return atproto.Record{}, err + return err } if len(data) > atproto.MaxResponseBytes { - return atproto.Record{}, errors.New("Slingshot response exceeds size limit") + return errors.New("Slingshot response exceeds size limit") } var result struct { - URI string `json:"uri"` - CID string `json:"cid"` - Value atproto.Record `json:"value"` + URI string `json:"uri"` + CID string `json:"cid"` + Value json.RawMessage `json:"value"` } if err := json.Unmarshal(data, &result); err != nil { - return atproto.Record{}, err + return err } - if result.URI != "at://"+did+"/"+domain.Collection+"/"+key || result.CID != cid { - return atproto.Record{}, errors.New("Slingshot record does not match the event URI or CID") + if result.URI != "at://"+did+"/"+source+"/"+key || result.CID != cid { + return errors.New("Slingshot record does not match the event URI or CID") } - return result.Value, nil + return json.Unmarshal(result.Value, record) } diff --git a/internal/microcosm/events.go b/internal/microcosm/events.go index 776f589..c94b1b9 100644 --- a/internal/microcosm/events.go +++ b/internal/microcosm/events.go @@ -48,7 +48,8 @@ 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} - if wanted != nil { + allSoft := c.softMode == domain.SoftAllAuthors && (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 { return domain.Change{}, err @@ -73,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.SoftSource(commit.Collection) { + if c.softMode == domain.SoftDisabled { + change.Kind = domain.IdentityChange + return change, nil + } + return c.decodeSoft(ctx, e) + } if commit.Collection != domain.Collection { change.Kind = domain.IdentityChange return change, nil @@ -109,6 +117,6 @@ func (c *Client) decode(ctx context.Context, data []byte, wanted func(context.Co if atproto.ValidateRecord(record) != nil || domain.RecordKey(record.Subject, record.Tag) != change.RecordKey { return change, nil } - change.Vouch = &domain.Vouch{Author: e.DID, Subject: record.Subject, Tag: record.Tag, Score: *record.Score, CreatedAt: record.CreatedAt, Comment: record.Comment} + change.Vouch = &domain.Vouch{Author: e.DID, Subject: record.Subject, Tag: record.Tag, Score: float64(*record.Score), CreatedAt: record.CreatedAt, Comment: record.Comment} return change, nil } diff --git a/internal/microcosm/soft.go b/internal/microcosm/soft.go new file mode 100644 index 0000000..ca65722 --- /dev/null +++ b/internal/microcosm/soft.go @@ -0,0 +1,43 @@ +package microcosm + +import ( + "context" + "encoding/json" + "errors" + + "github.com/bluesky-social/indigo/atproto/syntax" + + "provenance/internal/atproto" + "provenance/internal/domain" +) + +func (c *Client) decodeSoft(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 soft record operation") + } + var record atproto.SoftRecord + if len(commit.Record) == 0 || string(commit.Record) == "null" { + if commit.CID == "" { + return domain.Change{}, errors.New("soft 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 + } + if atproto.ValidateSoftRecord(record, commit.Collection) != nil { + return change, nil + } + change.Vouch = &domain.Vouch{Author: e.DID, Subject: record.Subject, Score: domain.SourceScore(commit.Collection), Source: commit.Collection, RecordKey: commit.RecordKey, CreatedAt: record.CreatedAt} + return change, nil +} diff --git a/internal/microcosm/soft_test.go b/internal/microcosm/soft_test.go new file mode 100644 index 0000000..5cdaad1 --- /dev/null +++ b/internal/microcosm/soft_test.go @@ -0,0 +1,120 @@ +package microcosm + +import ( + "context" + "encoding/json" + "errors" + "github.com/coder/websocket" + "net/http" + "net/http/httptest" + "testing" + + "provenance/internal/atproto" + "provenance/internal/domain" +) + +func TestSoftEvents(t *testing.T) { + for _, source := range []string{domain.FollowCollection, domain.BlockCollection} { + record := atproto.SoftRecord{Type: source, Subject: subject, CreatedAt: "2026-10-09T00:00:00Z"} + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Query().Get("collection") != source { + t.Error("wrong hydration source") + } + json.NewEncoder(w).Encode(map[string]any{"uri": "at://" + author + "/" + source + "/record", "cid": recordCID, "value": record}) + })) + for _, mode := range []domain.SoftMode{domain.SoftDisabled, domain.SoftParticipants, domain.SoftAllAuthors} { + 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, err := json.Marshal(map[string]any{"did": author, "time_us": 100, "kind": domain.RecordChange, "commit": map[string]any{"operation": operation, "collection": source, "rkey": "record", "cid": recordCID, "record": payload}}) + if err != nil { + t.Fatal(err) + } + change, err := client.decode(context.Background(), data, nil) + if err != nil { + t.Fatal(err) + } + if mode == domain.SoftDisabled { + if change.Kind != domain.IdentityChange || change.Vouch != nil { + t.Fatal("disabled soft ingestion") + } + continue + } + if change.Source != source || change.RecordKey != "record" { + t.Fatalf("source lost: %+v", change) + } + if operation == Delete { + if change.Vouch != nil { + t.Fatal("delete kept assertion") + } + continue + } + if change.Vouch == nil || change.Vouch.Score != domain.SourceScore(source) { + t.Fatalf("soft event: %+v", change) + } + } + } + data, _ := json.Marshal(map[string]any{"did": author, "time_us": 200, "kind": domain.RecordChange, "commit": map[string]any{"operation": Update, "collection": source, "rkey": "record", "record": atproto.SoftRecord{Type: source, Subject: "invalid", CreatedAt: record.CreatedAt}}}) + change, err := client.decode(context.Background(), data, nil) + if err != nil || change.Vouch != nil { + t.Fatalf("invalid soft update: %+v %v", change, err) + } + } + server.Close() + } +} + +func TestSoftEventScope(t *testing.T) { + data, _ := json.Marshal(map[string]any{"did": author, "time_us": 100, "kind": domain.RecordChange, "commit": map[string]any{"operation": Create, "collection": domain.FollowCollection, "rkey": "record", "record": atproto.SoftRecord{Type: domain.FollowCollection, Subject: subject, CreatedAt: "2026-10-09T00:00:00Z"}}}) + for _, mode := range []domain.SoftMode{domain.SoftParticipants, domain.SoftAllAuthors} { + client, err := NewWithSoft("ws://127.0.0.1:1/subscribe", "https://records.example.com", mode) + if err != nil { + t.Fatal(err) + } + change, err := client.decode(context.Background(), data, func(context.Context, string, int64) (bool, error) { return false, nil }) + if err != nil || (change.Vouch != nil) != (mode == domain.SoftAllAuthors) { + t.Fatalf("scope: %+v %v", change, err) + } + change, err = client.decode(context.Background(), commit(t, Create, validRecord()), func(context.Context, string, int64) (bool, error) { return false, nil }) + if err != nil || change.Vouch != nil { + t.Fatal("all-author flag enabled explicit records") + } + } +} + +func TestSoftSubscriptions(t *testing.T) { + for _, mode := range []domain.SoftMode{domain.SoftDisabled, domain.SoftParticipants, domain.SoftAllAuthors} { + 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 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) + } + conn, err := websocket.Accept(w, r, nil) + if err != nil { + t.Error(err) + return + } + defer conn.CloseNow() + conn.Write(r.Context(), websocket.MessageText, commit(t, Create, validRecord())) + })) + client, err := NewWithSoft("ws"+server.URL[len("http"):], server.URL, mode) + if err != nil { + t.Fatal(err) + } + stop := errors.New("complete") + if err := client.Consume(context.Background(), 100, nil, func(context.Context, domain.Change) error { return stop }); !errors.Is(err, stop) { + t.Fatal(err) + } + server.Close() + } +} diff --git a/internal/microcosm/stream.go b/internal/microcosm/stream.go index 8012b46..be22b77 100644 --- a/internal/microcosm/stream.go +++ b/internal/microcosm/stream.go @@ -16,6 +16,10 @@ func (c *Client) Consume(ctx context.Context, cursor int64, wanted func(context. u, _ := url.Parse(c.streamURL) params := u.Query() params.Set("wantedCollections", domain.Collection) + if c.softMode != domain.SoftDisabled { + params.Add("wantedCollections", domain.FollowCollection) + params.Add("wantedCollections", domain.BlockCollection) + } 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/reputation/algorithms/direct_sum.go b/internal/reputation/algorithms/direct_sum.go index 18d1e78..e6eb780 100644 --- a/internal/reputation/algorithms/direct_sum.go +++ b/internal/reputation/algorithms/direct_sum.go @@ -42,7 +42,7 @@ func (DirectSum) Calculate(ctx context.Context, input reputation.Input) (reputat return reputation.Result{}, errors.New("reputation requires one assertion per author") } seen[vouch.Author] = struct{}{} - if err := domain.ValidateScore(vouch.Score); err != nil { + if err := domain.ValidateVouch(vouch); err != nil { return reputation.Result{}, err } if vouch.Score == domain.Blocked { diff --git a/internal/reputation/algorithms/scenarios_test.go b/internal/reputation/algorithms/scenarios_test.go index 577cc33..eba620b 100644 --- a/internal/reputation/algorithms/scenarios_test.go +++ b/internal/reputation/algorithms/scenarios_test.go @@ -13,7 +13,7 @@ import ( "provenance/internal/trust/trusttest" ) -func edge(from, to string, score int) trusttest.Edge { +func edge(from, to string, score float64) trusttest.Edge { return trusttest.Edge{From: from, To: to, Score: score} } diff --git a/internal/reputation/algorithms/soft_test.go b/internal/reputation/algorithms/soft_test.go new file mode 100644 index 0000000..7597fab --- /dev/null +++ b/internal/reputation/algorithms/soft_test.go @@ -0,0 +1,26 @@ +package algorithms_test + +import ( + "testing" + + "provenance/internal/domain" + "provenance/internal/reputation/algorithms" + "provenance/internal/reputation/reputationtest" + "provenance/internal/trust" + trustalg "provenance/internal/trust/algorithms" + "provenance/internal/trust/trusttest" +) + +func TestSoftReputation(t *testing.T) { + for _, algorithm := range []trust.Algorithm{trustalg.Optimistic{}, trustalg.ShortestPath{}} { + t.Run(algorithm.Describe().Name, func(t *testing.T) { + reputationtest.Run(t, algorithms.DirectSum{}, algorithm, []reputationtest.Scenario{ + {Name: "follow from source", Source: "alice", Target: "bob", Threshold: 1, Edges: []trusttest.Edge{{From: "alice", To: "bob", Score: domain.FollowScore, Source: domain.FollowCollection}}, Score: .5, MeanScore: .5}, + {Name: "soft block contributes", Source: "alice", Target: "bob", Threshold: 1, Edges: []trusttest.Edge{{From: "alice", To: "bob", Score: domain.SoftBlockScore, Source: domain.BlockCollection}}, Score: -.5, MeanScore: -.5}, + {Name: "soft authority qualifies at lower threshold", Source: "alice", Target: "bob", Threshold: .5, Edges: []trusttest.Edge{{From: "alice", To: "carol", Score: domain.FollowScore, Source: domain.FollowCollection}, {From: "carol", To: "bob", Score: 2}}, Score: 2, MeanScore: 2}, + {Name: "soft authority fails default threshold", Source: "alice", Target: "bob", Threshold: 1, Edges: []trusttest.Edge{{From: "alice", To: "carol", Score: domain.FollowScore, Source: domain.FollowCollection}, {From: "carol", To: "bob", Score: 2}}}, + {Name: "mixed fractional mean", Source: "alice", Target: "bob", Threshold: 1, Edges: []trusttest.Edge{{From: "alice", To: "carol", Score: 2}, {From: "carol", To: "bob", Score: domain.SoftBlockScore, Source: domain.BlockCollection}, {From: "alice", To: "bob", Score: domain.FollowScore, Source: domain.FollowCollection}}}, + }) + }) + } +} diff --git a/internal/storage/comments_test.go b/internal/storage/comments_test.go index dcd72e2..47ac479 100644 --- a/internal/storage/comments_test.go +++ b/internal/storage/comments_test.go @@ -17,7 +17,9 @@ func TestCommentMigration(t *testing.T) { if err != nil { t.Fatal(err) } - legacy := strings.Replace(schema, " comment TEXT NOT NULL DEFAULT '',\n", "", 1) + legacy := strings.SplitN(schema, "CREATE INDEX", 2)[0] + legacy = strings.Replace(legacy, " comment TEXT NOT NULL DEFAULT '',\n", "", 1) + legacy = strings.Replace(legacy, " source TEXT NOT NULL DEFAULT 'computer.ngp.provenance.vouch',\n", "", 1) if _, err := conn.Exec(legacy); err != nil { t.Fatal(err) } diff --git a/internal/storage/db/models.go b/internal/storage/db/models.go index b8b49f7..8d7c356 100644 --- a/internal/storage/db/models.go +++ b/internal/storage/db/models.go @@ -31,6 +31,38 @@ type Secret struct { Expires int64 } +type SoftAccount struct { + Did string + Active int64 + AccountUs int64 + BootstrapUs int64 +} + +type SoftEdge struct { + Author string + Subject string + Source string + RecordKey string + Score float64 + CreatedAt string +} + +type SoftVersion struct { + Author string + Source string + RecordKey string + TimeUs int64 +} + +type SoftVouch struct { + Author string + Subject string + Source string + RecordKey string + Score float64 + CreatedAt string +} + type StreamCursor struct { Source string TimeUs int64 @@ -46,6 +78,7 @@ type TrustDependency struct { CacheKey string Author string RecordKey string + Source string } type Vouch struct { @@ -56,4 +89,5 @@ type Vouch struct { CreatedAt string RecordKey string Comment string + Source string } diff --git a/internal/storage/db/querier.go b/internal/storage/db/querier.go index 8743a98..1ea199c 100644 --- a/internal/storage/db/querier.go +++ b/internal/storage/db/querier.go @@ -13,11 +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) + 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) DeleteRecord(ctx context.Context, arg DeleteRecordParams) error + DeleteSoft(ctx context.Context, arg DeleteSoftParams) error DeleteTrust(ctx context.Context, cacheKey string) error GetSecret(ctx context.Context, arg GetSecretParams) ([]byte, error) Graph(ctx context.Context, arg GraphParams) ([]Vouch, error) @@ -26,6 +29,7 @@ type Querier interface { InitCursor(ctx context.Context, arg InitCursorParams) error InsertVouch(ctx context.Context, arg InsertVouchParams) error Join(ctx context.Context, did string) error + JoinSoft(ctx context.Context, did string) error LegacyKeys(ctx context.Context) ([]LegacyKeysRow, error) OwnVouches(ctx context.Context, arg OwnVouchesParams) ([]Vouch, error) Participant(ctx context.Context, did string) (Participant, error) @@ -34,6 +38,7 @@ type Querier interface { PutSecret(ctx context.Context, arg PutSecretParams) error RecordVersion(ctx context.Context, arg RecordVersionParams) (int64, error) RemoveSecret(ctx context.Context, arg RemoveSecretParams) error + ReplaceSoft(ctx context.Context, arg ReplaceSoftParams) error ReplaceVouch(ctx context.Context, arg ReplaceVouchParams) error SaveDependency(ctx context.Context, arg SaveDependencyParams) error SaveTrust(ctx context.Context, arg SaveTrustParams) error @@ -42,8 +47,17 @@ type Querier interface { SetDeleted(ctx context.Context, arg SetDeletedParams) error SetLegacyKey(ctx context.Context, arg SetLegacyKeyParams) error SetRecordVersion(ctx context.Context, arg SetRecordVersionParams) error + SetSoftVersion(ctx context.Context, arg SetSoftVersionParams) error SetSynced(ctx context.Context, arg SetSyncedParams) error SetThreshold(ctx context.Context, arg SetThresholdParams) (int64, error) + SoftAccount(ctx context.Context, did string) (SoftAccount, error) + SoftAccountState(ctx context.Context, arg SoftAccountStateParams) error + SoftBootstrap(ctx context.Context, arg SoftBootstrapParams) error + SoftDeleted(ctx context.Context, arg SoftDeletedParams) error + SoftGraph(ctx context.Context, arg SoftGraphParams) ([]SoftEdge, error) + SoftIncoming(ctx context.Context, arg SoftIncomingParams) ([]SoftEdge, error) + SoftRecords(ctx context.Context, author string) ([]SoftVouch, error) + SoftVersion(ctx context.Context, arg SoftVersionParams) (int64, error) Tags(ctx context.Context, arg TagsParams) ([]string, error) } diff --git a/internal/storage/db/queries.sql.go b/internal/storage/db/queries.sql.go index 44b2b27..1d49190 100644 --- a/internal/storage/db/queries.sql.go +++ b/internal/storage/db/queries.sql.go @@ -46,7 +46,7 @@ func (q *Queries) AdvanceCursor(ctx context.Context, arg AdvanceCursorParams) er } const authorRecords = `-- name: AuthorRecords :many -SELECT author, subject, tag, score, created_at, record_key, comment FROM vouches WHERE author = ? +SELECT author, subject, tag, score, created_at, record_key, comment, source FROM vouches WHERE author = ? ` func (q *Queries) AuthorRecords(ctx context.Context, author string) ([]Vouch, error) { @@ -66,6 +66,7 @@ func (q *Queries) AuthorRecords(ctx context.Context, author string) ([]Vouch, er &i.CreatedAt, &i.RecordKey, &i.Comment, + &i.Source, ); err != nil { return nil, err } @@ -96,6 +97,24 @@ func (q *Queries) CachedTrust(ctx context.Context, arg CachedTrustParams) (strin return result, err } +const clearSoft = `-- name: ClearSoft :exec +DELETE FROM soft_vouches WHERE author = ? +` + +func (q *Queries) ClearSoft(ctx context.Context, author string) error { + _, err := q.db.ExecContext(ctx, clearSoft, author) + return err +} + +const clearSoftVersions = `-- name: ClearSoftVersions :exec +DELETE FROM soft_versions WHERE author = ? +` + +func (q *Queries) ClearSoftVersions(ctx context.Context, author string) error { + _, err := q.db.ExecContext(ctx, clearSoftVersions, author) + return err +} + const clearVersions = `-- name: ClearVersions :exec DELETE FROM record_versions WHERE author = ? ` @@ -156,6 +175,21 @@ func (q *Queries) DeleteRecord(ctx context.Context, arg DeleteRecordParams) erro return err } +const deleteSoft = `-- name: DeleteSoft :exec +DELETE FROM soft_vouches WHERE author = ? AND source = ? AND record_key = ? +` + +type DeleteSoftParams struct { + Author string + Source string + RecordKey string +} + +func (q *Queries) DeleteSoft(ctx context.Context, arg DeleteSoftParams) error { + _, err := q.db.ExecContext(ctx, deleteSoft, arg.Author, arg.Source, arg.RecordKey) + return err +} + const deleteTrust = `-- name: DeleteTrust :exec DELETE FROM trust_cache WHERE cache_key = ? ` @@ -183,7 +217,7 @@ func (q *Queries) GetSecret(ctx context.Context, arg GetSecretParams) ([]byte, e } const graph = `-- name: Graph :many -SELECT vouches.author, vouches.subject, vouches.tag, vouches.score, vouches.created_at, vouches.record_key, vouches.comment FROM vouches JOIN participants ON participants.did = vouches.author +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 NOT EXISTS (SELECT 1 FROM participants target WHERE target.did = vouches.subject AND target.active = 0) AND (tag = ?1 OR (CAST(?2 AS INTEGER) = 1 AND tag = '')) @@ -213,6 +247,7 @@ func (q *Queries) Graph(ctx context.Context, arg GraphParams) ([]Vouch, error) { &i.CreatedAt, &i.RecordKey, &i.Comment, + &i.Source, ); err != nil { return nil, err } @@ -239,7 +274,7 @@ func (q *Queries) GraphRevision(ctx context.Context) (int64, error) { } const incoming = `-- name: Incoming :many -SELECT vouches.author, vouches.subject, vouches.tag, vouches.score, vouches.created_at, vouches.record_key, vouches.comment FROM vouches JOIN participants ON participants.did = vouches.author +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 subject = ?1 AND participants.active = 1 AND NOT EXISTS (SELECT 1 FROM participants target WHERE target.did = vouches.subject AND target.active = 0) AND (tag = ?2 OR (CAST(?3 AS INTEGER) = 1 AND tag = '')) @@ -275,6 +310,7 @@ func (q *Queries) Incoming(ctx context.Context, arg IncomingParams) ([]Vouch, er &i.CreatedAt, &i.RecordKey, &i.Comment, + &i.Source, ); err != nil { return nil, err } @@ -339,6 +375,15 @@ func (q *Queries) Join(ctx context.Context, did string) error { return err } +const joinSoft = `-- name: JoinSoft :exec +INSERT INTO soft_accounts(did) VALUES (?) ON CONFLICT(did) DO NOTHING +` + +func (q *Queries) JoinSoft(ctx context.Context, did string) error { + _, err := q.db.ExecContext(ctx, joinSoft, did) + return err +} + const legacyKeys = `-- name: LegacyKeys :many SELECT author, subject, tag FROM vouches WHERE record_key = '' ` @@ -373,7 +418,7 @@ func (q *Queries) LegacyKeys(ctx context.Context) ([]LegacyKeysRow, error) { } const ownVouches = `-- name: OwnVouches :many -SELECT vouches.author, vouches.subject, vouches.tag, vouches.score, vouches.created_at, vouches.record_key, vouches.comment FROM vouches JOIN participants ON participants.did = vouches.author +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 ? ` @@ -400,6 +445,7 @@ func (q *Queries) OwnVouches(ctx context.Context, arg OwnVouchesParams) ([]Vouch &i.CreatedAt, &i.RecordKey, &i.Comment, + &i.Source, ); err != nil { return nil, err } @@ -527,6 +573,33 @@ func (q *Queries) RemoveSecret(ctx context.Context, arg RemoveSecretParams) erro return err } +const replaceSoft = `-- name: ReplaceSoft :exec +INSERT INTO soft_vouches(author, subject, source, record_key, score, created_at) VALUES (?, ?, ?, ?, ?, ?) +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 +` + +type ReplaceSoftParams struct { + Author string + Subject string + Source string + RecordKey string + Score float64 + CreatedAt string +} + +func (q *Queries) ReplaceSoft(ctx context.Context, arg ReplaceSoftParams) error { + _, err := q.db.ExecContext(ctx, replaceSoft, + arg.Author, + arg.Subject, + arg.Source, + arg.RecordKey, + arg.Score, + arg.CreatedAt, + ) + return err +} + const replaceVouch = `-- name: ReplaceVouch :exec INSERT INTO vouches(author, subject, tag, score, created_at, record_key, comment) VALUES (?, ?, ?, ?, ?, ?, ?) ON CONFLICT(author, subject, tag) DO UPDATE SET score = excluded.score, created_at = excluded.created_at, record_key = excluded.record_key, comment = excluded.comment @@ -557,17 +630,23 @@ func (q *Queries) ReplaceVouch(ctx context.Context, arg ReplaceVouchParams) erro } const saveDependency = `-- name: SaveDependency :exec -INSERT OR IGNORE INTO trust_dependencies(cache_key, author, record_key) VALUES (?, ?, ?) +INSERT OR IGNORE INTO trust_dependencies(cache_key, author, source, record_key) VALUES (?, ?, ?, ?) ` type SaveDependencyParams struct { CacheKey string Author string + Source string RecordKey string } func (q *Queries) SaveDependency(ctx context.Context, arg SaveDependencyParams) error { - _, err := q.db.ExecContext(ctx, saveDependency, arg.CacheKey, arg.Author, arg.RecordKey) + _, err := q.db.ExecContext(ctx, saveDependency, + arg.CacheKey, + arg.Author, + arg.Source, + arg.RecordKey, + ) return err } @@ -686,6 +765,28 @@ func (q *Queries) SetRecordVersion(ctx context.Context, arg SetRecordVersionPara return err } +const setSoftVersion = `-- name: SetSoftVersion :exec +INSERT INTO soft_versions(author, source, record_key, time_us) VALUES (?, ?, ?, ?) +ON CONFLICT(author, source, record_key) DO UPDATE SET time_us = excluded.time_us +` + +type SetSoftVersionParams struct { + Author string + Source string + RecordKey string + TimeUs int64 +} + +func (q *Queries) SetSoftVersion(ctx context.Context, arg SetSoftVersionParams) error { + _, err := q.db.ExecContext(ctx, setSoftVersion, + arg.Author, + arg.Source, + arg.RecordKey, + arg.TimeUs, + ) + return err +} + const setSynced = `-- name: SetSynced :exec UPDATE participants SET synced_at = ? WHERE did = ? ` @@ -717,6 +818,219 @@ func (q *Queries) SetThreshold(ctx context.Context, arg SetThresholdParams) (int return result.RowsAffected() } +const softAccount = `-- name: SoftAccount :one +SELECT did, active, account_us, bootstrap_us FROM soft_accounts WHERE did = ? +` + +func (q *Queries) SoftAccount(ctx context.Context, did string) (SoftAccount, error) { + row := q.db.QueryRowContext(ctx, softAccount, did) + var i SoftAccount + err := row.Scan( + &i.Did, + &i.Active, + &i.AccountUs, + &i.BootstrapUs, + ) + return i, err +} + +const softAccountState = `-- name: SoftAccountState :exec +UPDATE soft_accounts SET active = ?, account_us = ? WHERE did = ? +` + +type SoftAccountStateParams struct { + Active int64 + AccountUs int64 + Did string +} + +func (q *Queries) SoftAccountState(ctx context.Context, arg SoftAccountStateParams) error { + _, err := q.db.ExecContext(ctx, softAccountState, arg.Active, arg.AccountUs, arg.Did) + return err +} + +const softBootstrap = `-- name: SoftBootstrap :exec +UPDATE soft_accounts SET active = 1, account_us = ?, bootstrap_us = ? WHERE did = ? +` + +type SoftBootstrapParams struct { + AccountUs int64 + BootstrapUs int64 + Did string +} + +func (q *Queries) SoftBootstrap(ctx context.Context, arg SoftBootstrapParams) error { + _, err := q.db.ExecContext(ctx, softBootstrap, arg.AccountUs, arg.BootstrapUs, arg.Did) + return err +} + +const softDeleted = `-- name: SoftDeleted :exec +UPDATE soft_accounts SET active = 0, account_us = ?, bootstrap_us = ? WHERE did = ? +` + +type SoftDeletedParams struct { + AccountUs int64 + BootstrapUs int64 + Did string +} + +func (q *Queries) SoftDeleted(ctx context.Context, arg SoftDeletedParams) error { + _, err := q.db.ExecContext(ctx, softDeleted, arg.AccountUs, arg.BootstrapUs, arg.Did) + return err +} + +const softGraph = `-- name: SoftGraph :many +SELECT s.author, s.subject, s.source, s.record_key, s.score, s.created_at FROM soft_edges s +WHERE (CAST(?1 AS INTEGER) = 1 OR EXISTS (SELECT 1 FROM participants p WHERE p.did = s.author AND p.active = 1)) +AND NOT EXISTS (SELECT 1 FROM vouches v WHERE v.author = s.author AND v.subject = s.subject AND (v.tag = ?2 OR (CAST(?3 AS INTEGER) = 1 AND v.tag = ''))) +ORDER BY s.author, s.subject LIMIT ?4 +` + +type SoftGraphParams struct { + AllAuthors int64 + Tag string + Fallback int64 + MaxRows int64 +} + +func (q *Queries) SoftGraph(ctx context.Context, arg SoftGraphParams) ([]SoftEdge, error) { + rows, err := q.db.QueryContext(ctx, softGraph, + arg.AllAuthors, + arg.Tag, + arg.Fallback, + arg.MaxRows, + ) + if err != nil { + return nil, err + } + defer rows.Close() + items := []SoftEdge{} + for rows.Next() { + var i SoftEdge + if err := rows.Scan( + &i.Author, + &i.Subject, + &i.Source, + &i.RecordKey, + &i.Score, + &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 softIncoming = `-- name: SoftIncoming :many +SELECT s.author, s.subject, s.source, s.record_key, s.score, s.created_at FROM soft_edges s +WHERE s.subject = ?1 +AND (CAST(?2 AS INTEGER) = 1 OR EXISTS (SELECT 1 FROM participants p WHERE p.did = s.author AND p.active = 1)) +AND NOT EXISTS (SELECT 1 FROM vouches v WHERE v.author = s.author AND v.subject = s.subject AND (v.tag = ?3 OR (CAST(?4 AS INTEGER) = 1 AND v.tag = ''))) +ORDER BY s.author LIMIT ?5 +` + +type SoftIncomingParams struct { + Subject string + AllAuthors int64 + Tag string + Fallback int64 + MaxRows int64 +} + +func (q *Queries) SoftIncoming(ctx context.Context, arg SoftIncomingParams) ([]SoftEdge, error) { + rows, err := q.db.QueryContext(ctx, softIncoming, + arg.Subject, + arg.AllAuthors, + arg.Tag, + arg.Fallback, + arg.MaxRows, + ) + if err != nil { + return nil, err + } + defer rows.Close() + items := []SoftEdge{} + for rows.Next() { + var i SoftEdge + if err := rows.Scan( + &i.Author, + &i.Subject, + &i.Source, + &i.RecordKey, + &i.Score, + &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 softRecords = `-- name: SoftRecords :many +SELECT author, subject, source, record_key, score, created_at FROM soft_vouches WHERE author = ? +` + +func (q *Queries) SoftRecords(ctx context.Context, author string) ([]SoftVouch, error) { + rows, err := q.db.QueryContext(ctx, softRecords, author) + if err != nil { + return nil, err + } + defer rows.Close() + items := []SoftVouch{} + for rows.Next() { + var i SoftVouch + if err := rows.Scan( + &i.Author, + &i.Subject, + &i.Source, + &i.RecordKey, + &i.Score, + &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 softVersion = `-- name: SoftVersion :one +SELECT time_us FROM soft_versions WHERE author = ? AND source = ? AND record_key = ? +` + +type SoftVersionParams struct { + Author string + Source string + RecordKey string +} + +func (q *Queries) SoftVersion(ctx context.Context, arg SoftVersionParams) (int64, error) { + row := q.db.QueryRowContext(ctx, softVersion, arg.Author, arg.Source, arg.RecordKey) + var time_us int64 + err := row.Scan(&time_us) + return time_us, err +} + const tags = `-- name: Tags :many SELECT DISTINCT tag FROM vouches JOIN participants ON participants.did = vouches.author WHERE participants.active = 1 AND tag != '' AND substr(tag, 1, length(CAST(?1 AS TEXT))) = CAST(?1 AS TEXT) diff --git a/internal/storage/events.go b/internal/storage/events.go index bc53fa4..30eb6b8 100644 --- a/internal/storage/events.go +++ b/internal/storage/events.go @@ -28,8 +28,15 @@ func (s *SQLite) Apply(ctx context.Context, source string, change domain.Change) } defer tx.Rollback() q := s.queries.WithTx(tx) - if err := applyChange(ctx, q, change); err != nil { - return err + if 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 err := applyChange(ctx, q, change); err != nil { + return err + } } if err := q.AdvanceCursor(ctx, db.AdvanceCursorParams{Source: source, TimeUs: change.TimeUS}); err != nil { return err @@ -38,6 +45,9 @@ func (s *SQLite) Apply(ctx context.Context, source string, change domain.Change) } func applyChange(ctx context.Context, q *db.Queries, change domain.Change) error { + if change.Kind == domain.RecordChange && change.Source != "" && change.Source != domain.Collection { + return errors.New("invalid explicit record source") + } p, err := q.Participant(ctx, change.DID) if errors.Is(err, sql.ErrNoRows) { return nil @@ -85,6 +95,9 @@ func applyChange(ctx context.Context, q *db.Queries, change domain.Change) error return err } if v := change.Vouch; v != nil { + if v.Source != "" && v.Source != domain.Collection { + return errors.New("invalid explicit vouch source") + } if v.Author != change.DID || domain.RecordKey(v.Subject, v.Tag) != change.RecordKey { return errors.New("event record does not match its repository or key") } diff --git a/internal/storage/migration4.sql b/internal/storage/migration4.sql new file mode 100644 index 0000000..f86ce81 --- /dev/null +++ b/internal/storage/migration4.sql @@ -0,0 +1,16 @@ +CREATE TABLE IF NOT EXISTS graph_revision ( + singleton INTEGER PRIMARY KEY CHECK(singleton = 1), + revision INTEGER NOT NULL DEFAULT 0 +); +INSERT OR IGNORE INTO graph_revision(singleton) VALUES (1); +CREATE TABLE IF NOT EXISTS trust_cache ( + cache_key TEXT PRIMARY KEY, + result TEXT NOT NULL, + expires_us INTEGER NOT NULL +); +CREATE TABLE IF NOT EXISTS trust_dependencies ( + cache_key TEXT NOT NULL REFERENCES trust_cache(cache_key) ON DELETE CASCADE, + author TEXT NOT NULL, + record_key TEXT NOT NULL, + PRIMARY KEY(cache_key, author, record_key) +); diff --git a/internal/storage/migration5.sql b/internal/storage/migration5.sql new file mode 100644 index 0000000..667ab61 --- /dev/null +++ b/internal/storage/migration5.sql @@ -0,0 +1,15 @@ +ALTER TABLE vouches ADD COLUMN source TEXT NOT NULL DEFAULT 'computer.ngp.provenance.vouch'; +DROP TRIGGER IF EXISTS trust_vouch_delete; +DROP TRIGGER IF EXISTS trust_vouch_update; +DROP INDEX IF EXISTS trust_dependencies_vouch; +ALTER TABLE trust_dependencies RENAME TO legacy_trust_dependencies; +CREATE TABLE trust_dependencies ( + cache_key TEXT NOT NULL REFERENCES trust_cache(cache_key) ON DELETE CASCADE, + author TEXT NOT NULL, + record_key TEXT NOT NULL, + source TEXT NOT NULL DEFAULT 'computer.ngp.provenance.vouch', + PRIMARY KEY(cache_key, author, source, record_key) +); +INSERT INTO trust_dependencies(cache_key, author, record_key) +SELECT cache_key, author, record_key FROM legacy_trust_dependencies; +DROP TABLE legacy_trust_dependencies; diff --git a/internal/storage/queries.sql b/internal/storage/queries.sql index 2fa21a7..84337df 100644 --- a/internal/storage/queries.sql +++ b/internal/storage/queries.sql @@ -56,7 +56,59 @@ DELETE FROM trust_cache WHERE cache_key = ?; INSERT INTO trust_cache(cache_key, result, expires_us) VALUES (?, ?, ?); -- name: SaveDependency :exec -INSERT OR IGNORE INTO trust_dependencies(cache_key, author, record_key) VALUES (?, ?, ?); +INSERT OR IGNORE INTO trust_dependencies(cache_key, author, source, record_key) VALUES (?, ?, ?, ?); + +-- name: SoftGraph :many +SELECT s.* FROM soft_edges s +WHERE (CAST(@all_authors AS INTEGER) = 1 OR EXISTS (SELECT 1 FROM participants p WHERE p.did = s.author AND p.active = 1)) +AND NOT EXISTS (SELECT 1 FROM vouches v WHERE v.author = s.author AND v.subject = s.subject AND (v.tag = @tag OR (CAST(@fallback AS INTEGER) = 1 AND v.tag = ''))) +ORDER BY s.author, s.subject LIMIT @max_rows; + +-- name: SoftIncoming :many +SELECT s.* FROM soft_edges s +WHERE s.subject = @subject +AND (CAST(@all_authors AS INTEGER) = 1 OR EXISTS (SELECT 1 FROM participants p WHERE p.did = s.author AND p.active = 1)) +AND NOT EXISTS (SELECT 1 FROM vouches v WHERE v.author = s.author AND v.subject = s.subject AND (v.tag = @tag OR (CAST(@fallback AS INTEGER) = 1 AND v.tag = ''))) +ORDER BY s.author LIMIT @max_rows; + +-- name: SoftAccount :one +SELECT * FROM soft_accounts WHERE did = ?; + +-- name: JoinSoft :exec +INSERT INTO soft_accounts(did) VALUES (?) ON CONFLICT(did) DO NOTHING; + +-- name: SoftAccountState :exec +UPDATE soft_accounts SET active = ?, account_us = ? WHERE did = ?; + +-- name: SoftBootstrap :exec +UPDATE soft_accounts SET active = 1, account_us = ?, bootstrap_us = ? WHERE did = ?; + +-- name: SoftDeleted :exec +UPDATE soft_accounts SET active = 0, account_us = ?, bootstrap_us = ? WHERE did = ?; + +-- name: SoftRecords :many +SELECT * FROM soft_vouches WHERE author = ?; + +-- name: SoftVersion :one +SELECT time_us FROM soft_versions WHERE author = ? AND source = ? AND record_key = ?; + +-- name: SetSoftVersion :exec +INSERT INTO soft_versions(author, source, record_key, time_us) VALUES (?, ?, ?, ?) +ON CONFLICT(author, source, record_key) DO UPDATE SET time_us = excluded.time_us; + +-- name: ClearSoftVersions :exec +DELETE FROM soft_versions WHERE author = ?; + +-- name: DeleteSoft :exec +DELETE FROM soft_vouches WHERE author = ? AND source = ? AND record_key = ?; + +-- name: ClearSoft :exec +DELETE FROM soft_vouches WHERE author = ?; + +-- name: ReplaceSoft :exec +INSERT INTO soft_vouches(author, subject, source, record_key, score, created_at) VALUES (?, ?, ?, ?, ?, ?) +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: Graph :many SELECT vouches.* FROM vouches JOIN participants ON participants.did = vouches.author diff --git a/internal/storage/schema.sql b/internal/storage/schema.sql index 588eb15..4b47389 100644 --- a/internal/storage/schema.sql +++ b/internal/storage/schema.sql @@ -15,6 +15,7 @@ CREATE TABLE IF NOT EXISTS vouches ( created_at TEXT NOT NULL, record_key TEXT NOT NULL, comment TEXT NOT NULL DEFAULT '', + source TEXT NOT NULL DEFAULT 'computer.ngp.provenance.vouch', PRIMARY KEY(author, subject, tag) ); @@ -40,19 +41,20 @@ CREATE TABLE IF NOT EXISTS trust_dependencies ( cache_key TEXT NOT NULL REFERENCES trust_cache(cache_key) ON DELETE CASCADE, author TEXT NOT NULL, record_key TEXT NOT NULL, - PRIMARY KEY(cache_key, author, record_key) + source TEXT NOT NULL DEFAULT 'computer.ngp.provenance.vouch', + PRIMARY KEY(cache_key, author, source, record_key) ); -CREATE INDEX IF NOT EXISTS trust_dependencies_vouch ON trust_dependencies(author, record_key, cache_key); +CREATE INDEX IF NOT EXISTS trust_dependencies_vouch ON trust_dependencies(author, source, record_key, cache_key); -- Record mutations invalidate only the paths that use that exact record. CREATE TRIGGER IF NOT EXISTS trust_vouch_delete BEFORE DELETE ON vouches BEGIN DELETE FROM trust_cache WHERE cache_key IN ( - SELECT cache_key FROM trust_dependencies WHERE author = OLD.author AND record_key = OLD.record_key + 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_vouch_update BEFORE UPDATE ON vouches BEGIN DELETE FROM trust_cache WHERE cache_key IN ( - SELECT cache_key FROM trust_dependencies WHERE author = OLD.author AND record_key = OLD.record_key + SELECT cache_key FROM trust_dependencies WHERE author = OLD.author AND source = OLD.source AND record_key = OLD.record_key ); END; diff --git a/internal/storage/soft.go b/internal/storage/soft.go new file mode 100644 index 0000000..d169e4f --- /dev/null +++ b/internal/storage/soft.go @@ -0,0 +1,163 @@ +package storage + +import ( + "context" + "database/sql" + "errors" + "time" + + "provenance/internal/domain" + "provenance/internal/storage/db" +) + +func (s *SQLite) graphRows(ctx context.Context, q *db.Queries, tag string, mode domain.Mode, limit int) ([]domain.Vouch, error) { + fallback, all := int64(0), int64(0) + if mode == domain.Fallback { + fallback = 1 + } + if s.softMode == domain.SoftAllAuthors { + all = 1 + } + rows, err := q.Graph(ctx, db.GraphParams{Tag: tag, Fallback: fallback, MaxRows: int64(limit)}) + if err != nil { + return nil, err + } + result := vouchRows(rows) + if s.softMode == domain.SoftDisabled || mode == domain.Tagged { + 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 +} + +func softRows(rows []db.SoftEdge) []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: row.Score, Source: row.Source, RecordKey: row.RecordKey, CreatedAt: row.CreatedAt}) + } + return result +} + +func (s *SQLite) SoftReady(ctx context.Context, did string) (bool, error) { + a, err := s.queries.SoftAccount(ctx, did) + if errors.Is(err, sql.ErrNoRows) { + return false, nil + } + return a.BootstrapUs != 0, err +} + +// Keep replay watermarks and account state separate from Provenance participation. +func (s *SQLite) applySoft(ctx context.Context, q *db.Queries, change domain.Change) error { + if s.softMode == domain.SoftDisabled { + return nil + } + if s.softMode == domain.SoftParticipants { + if _, err := q.Participant(ctx, change.DID); errors.Is(err, sql.ErrNoRows) { + return nil + } else if err != nil { + return err + } + } + if err := q.JoinSoft(ctx, change.DID); err != nil { + return err + } + a, err := q.SoftAccount(ctx, change.DID) + if err != nil { + return err + } + if change.TimeUS <= a.BootstrapUs { + return nil + } + if change.Kind == domain.AccountChange { + if change.TimeUS <= a.AccountUs { + return nil + } + active := int64(0) + if change.Active { + active = 1 + } + if !change.Active && change.Status == domain.DeletedAccount { + if err := q.ClearSoft(ctx, change.DID); err != nil { + return err + } + if err := q.ClearSoftVersions(ctx, change.DID); err != nil { + return err + } + return q.SoftDeleted(ctx, db.SoftDeletedParams{Did: change.DID, AccountUs: change.TimeUS, BootstrapUs: change.TimeUS}) + } + return q.SoftAccountState(ctx, db.SoftAccountStateParams{Did: change.DID, Active: active, AccountUs: change.TimeUS}) + } + if change.Kind != domain.RecordChange || !domain.SoftSource(change.Source) { + return nil + } + if change.RecordKey == "" { + return errors.New("soft record key is required") + } + version, err := q.SoftVersion(ctx, db.SoftVersionParams{Author: change.DID, Source: change.Source, RecordKey: change.RecordKey}) + if err != nil && !errors.Is(err, sql.ErrNoRows) { + return err + } + if change.TimeUS <= version { + return nil + } + if v := change.Vouch; v != nil { + if v.Author != change.DID || v.Source != change.Source || v.RecordKey != change.RecordKey { + return errors.New("soft record does not match its repository or source") + } + if err := domain.ValidateVouch(*v); err != nil { + return err + } + } + // Every accepted stream update invalidates its old path dependencies, even if its value is unchanged. + if err := q.DeleteSoft(ctx, db.DeleteSoftParams{Author: change.DID, Source: change.Source, RecordKey: change.RecordKey}); err != nil { + return err + } + if v := change.Vouch; v != nil { + if err := q.ReplaceSoft(ctx, db.ReplaceSoftParams{Author: v.Author, Subject: v.Subject, Source: v.Source, RecordKey: v.RecordKey, Score: v.Score, CreatedAt: v.CreatedAt}); err != nil { + return err + } + } + return q.SetSoftVersion(ctx, db.SetSoftVersionParams{Author: change.DID, Source: change.Source, RecordKey: change.RecordKey, TimeUs: change.TimeUS}) +} + +func replaceSoft(ctx context.Context, q *db.Queries, did string, vouches []domain.Vouch, at time.Time) error { + if err := q.JoinSoft(ctx, did); err != nil { + return err + } + previous, err := q.SoftRecords(ctx, did) + if err != nil { + return err + } + keys := make(map[[2]string]struct{}, len(vouches)) + for _, v := range vouches { + if !domain.SoftSource(v.Source) { + continue + } + if v.Author != did { + return errors.New("soft snapshot author does not match repository") + } + if err := domain.ValidateVouch(v); err != nil { + return err + } + key := [2]string{v.Source, v.RecordKey} + if _, exists := keys[key]; exists { + return errors.New("duplicate soft snapshot record") + } + keys[key] = struct{}{} + if err := q.ReplaceSoft(ctx, db.ReplaceSoftParams{Author: did, Subject: v.Subject, Source: v.Source, RecordKey: v.RecordKey, Score: v.Score, CreatedAt: v.CreatedAt}); err != nil { + return err + } + } + for _, v := range previous { + if _, exists := keys[[2]string{v.Source, v.RecordKey}]; exists { + continue + } + if err := q.DeleteSoft(ctx, db.DeleteSoftParams{Author: did, Source: v.Source, RecordKey: v.RecordKey}); err != nil { + return err + } + } + if err := q.ClearSoftVersions(ctx, did); err != nil { + return err + } + return q.SoftBootstrap(ctx, db.SoftBootstrapParams{Did: did, AccountUs: at.UnixMicro(), BootstrapUs: at.UnixMicro()}) +} diff --git a/internal/storage/soft_migration_test.go b/internal/storage/soft_migration_test.go new file mode 100644 index 0000000..31373a7 --- /dev/null +++ b/internal/storage/soft_migration_test.go @@ -0,0 +1,65 @@ +package storage + +import ( + "context" + "database/sql" + "errors" + "path/filepath" + "testing" + "time" + + "provenance/internal/domain" +) + +func TestSoftMigration(t *testing.T) { + path := filepath.Join(t.TempDir(), "version4.db") + conn, err := sql.Open("sqlite", path) + if err != nil { + t.Fatal(err) + } + if _, err := conn.Exec("CREATE TABLE participants(did TEXT PRIMARY KEY, threshold REAL NOT NULL DEFAULT 1, synced_at TEXT NOT NULL DEFAULT '', bootstrap_us INTEGER NOT NULL DEFAULT 0, active INTEGER NOT NULL DEFAULT 1, account_us INTEGER NOT NULL DEFAULT 0); CREATE TABLE vouches(author TEXT NOT NULL, subject TEXT NOT NULL, tag TEXT NOT NULL, score INTEGER NOT NULL, created_at TEXT NOT NULL, record_key TEXT NOT NULL, comment TEXT NOT NULL DEFAULT '', PRIMARY KEY(author,subject,tag));"); err != nil { + t.Fatal(err) + } + if _, err := conn.Exec(migration4); err != nil { + t.Fatal(err) + } + const author = "did:plc:a" + const subject = "did:plc:b" + key := domain.RecordKey(subject, "") + if _, err := conn.Exec("INSERT INTO participants(did) VALUES (?)", author); err != nil { + t.Fatal(err) + } + if _, err := conn.Exec("INSERT INTO vouches(author,subject,tag,score,created_at,record_key,comment) VALUES (?,?,'',2,'',?,'reference')", author, subject, key); err != nil { + t.Fatal(err) + } + now := time.UnixMicro(1000) + if _, err := conn.Exec("INSERT INTO trust_cache(cache_key,result,expires_us) VALUES ('legacy','{\"status\":\"calculated\",\"score\":2}',?)", now.Add(time.Hour).UnixMicro()); err != nil { + t.Fatal(err) + } + if _, err := conn.Exec("INSERT INTO trust_dependencies(cache_key,author,record_key) VALUES ('legacy',?,?)", author, key); err != nil { + t.Fatal(err) + } + if _, err := conn.Exec("PRAGMA user_version=4"); err != nil { + t.Fatal(err) + } + conn.Close() + s, err := OpenWithSoft(path, domain.SoftParticipants) + if err != nil { + t.Fatal(err) + } + defer s.Close() + ctx := context.Background() + graph, err := s.Graph(ctx, "", domain.Default, 10) + if err != nil || len(graph) != 1 || graph[0].Source != domain.Collection || graph[0].Comment != "reference" { + t.Fatalf("migration: %+v %v", graph, err) + } + if _, err := s.CachedTrust(ctx, "legacy", now); err != nil { + t.Fatalf("migration lost cache: %v", err) + } + if err := s.Apply(ctx, "stream", domain.Change{Kind: domain.RecordChange, DID: author, RecordKey: key, TimeUS: 100}); err != nil { + t.Fatal(err) + } + if _, err := s.CachedTrust(ctx, "legacy", now); !errors.Is(err, sql.ErrNoRows) { + t.Fatalf("legacy dependency not invalidated: %v", err) + } +} diff --git a/internal/storage/soft_schema.sql b/internal/storage/soft_schema.sql new file mode 100644 index 0000000..931418e --- /dev/null +++ b/internal/storage/soft_schema.sql @@ -0,0 +1,65 @@ +CREATE TABLE IF NOT EXISTS soft_accounts ( + did TEXT PRIMARY KEY, + active INTEGER NOT NULL DEFAULT 1, + account_us INTEGER NOT NULL DEFAULT 0, + bootstrap_us INTEGER NOT NULL DEFAULT 0 +); + +CREATE TABLE IF NOT EXISTS soft_vouches ( + author TEXT NOT NULL REFERENCES soft_accounts(did) ON DELETE CASCADE, + subject TEXT NOT NULL, + source TEXT NOT NULL CHECK(source IN ('app.bsky.graph.follow', 'app.bsky.graph.block')), + record_key TEXT NOT NULL, + score REAL NOT NULL CHECK( + (source = 'app.bsky.graph.follow' AND score = 0.5) OR + (source = 'app.bsky.graph.block' AND score = -0.5) + ), + created_at TEXT NOT NULL, + PRIMARY KEY(author, source, record_key) +); +CREATE INDEX IF NOT EXISTS soft_vouches_pair ON soft_vouches(author, subject, source, record_key); +CREATE INDEX IF NOT EXISTS soft_vouches_subject ON soft_vouches(subject, author, source, record_key); + +-- One block overrides follows. Duplicate records never multiply a pair's score. +CREATE VIEW IF NOT EXISTS soft_edges AS +SELECT s.* FROM soft_vouches s JOIN soft_accounts a ON a.did = s.author +WHERE a.active = 1 +AND NOT EXISTS (SELECT 1 FROM participants p WHERE p.did = s.subject AND p.active = 0) +AND NOT EXISTS (SELECT 1 FROM soft_accounts p WHERE p.did = s.subject AND p.active = 0) +AND NOT EXISTS ( + SELECT 1 FROM soft_vouches preferred + WHERE preferred.author = s.author AND preferred.subject = s.subject + AND ((preferred.source = 'app.bsky.graph.block' AND s.source = 'app.bsky.graph.follow') + OR (preferred.source = s.source AND preferred.record_key < s.record_key)) +); + +CREATE TABLE IF NOT EXISTS soft_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 TRIGGER IF NOT EXISTS trust_soft_delete BEFORE DELETE ON soft_vouches 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_soft_update BEFORE UPDATE ON soft_vouches 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_soft_insert AFTER INSERT ON soft_vouches BEGIN + UPDATE graph_revision SET revision = revision + 1 WHERE singleton = 1; +END; +CREATE TRIGGER IF NOT EXISTS graph_soft_delete AFTER DELETE ON soft_vouches BEGIN + UPDATE graph_revision SET revision = revision + 1 WHERE singleton = 1; +END; +CREATE TRIGGER IF NOT EXISTS graph_soft_update AFTER UPDATE ON soft_vouches BEGIN + UPDATE graph_revision SET revision = revision + 1 WHERE singleton = 1; +END; +CREATE TRIGGER IF NOT EXISTS graph_soft_account AFTER UPDATE OF active ON soft_accounts WHEN OLD.active != NEW.active BEGIN + UPDATE graph_revision SET revision = revision + 1 WHERE singleton = 1; +END; diff --git a/internal/storage/soft_test.go b/internal/storage/soft_test.go new file mode 100644 index 0000000..4df47a0 --- /dev/null +++ b/internal/storage/soft_test.go @@ -0,0 +1,148 @@ +package storage + +import ( + "context" + "database/sql" + "errors" + "path/filepath" + "testing" + "time" + + "provenance/internal/domain" + "provenance/internal/trust" +) + +func TestSoftStorage(t *testing.T) { + ctx := context.Background() + path := filepath.Join(t.TempDir(), "soft.db") + s, err := OpenWithSoft(path, domain.SoftParticipants) + if err != nil { + t.Fatal(err) + } + defer func() { s.Close() }() + const author = "did:plc:a" + const target = "did:plc:b" + if err := s.Join(ctx, author); err != nil { + t.Fatal(err) + } + v := domain.Vouch{Author: author, Subject: target, Source: domain.FollowCollection, RecordKey: "follow-1", Score: domain.FollowScore} + apply := func(v *domain.Vouch, source, key string, stamp int64) { + t.Helper() + if err := s.Apply(ctx, "soft-stream", domain.Change{Kind: domain.RecordChange, DID: author, Source: source, RecordKey: key, Vouch: v, TimeUS: stamp}); err != nil { + t.Fatal(err) + } + } + apply(&v, v.Source, v.RecordKey, 100) + graph := func(mode domain.Mode, tag string) []domain.Vouch { + t.Helper() + rows, err := s.Graph(ctx, tag, mode, 10) + if err != nil { + t.Fatal(err) + } + return trust.Select(rows, tag, mode) + } + if got := graph(domain.Default, ""); len(got) != 1 || got[0].Score != domain.FollowScore || got[0].Source != domain.FollowCollection { + t.Fatalf("follow: %+v", got) + } + if got := graph(domain.Tagged, "work"); len(got) != 0 { + t.Fatalf("soft topic-only edge: %+v", got) + } + if got := graph(domain.Fallback, "work"); len(got) != 1 { + t.Fatalf("soft fallback: %+v", got) + } + duplicate := v + duplicate.RecordKey = "follow-2" + apply(&duplicate, duplicate.Source, duplicate.RecordKey, 110) + if got := graph(domain.Default, ""); len(got) != 1 || got[0].RecordKey != v.RecordKey { + t.Fatalf("duplicate follows: %+v", got) + } + block := v + block.Source, block.Score = domain.BlockCollection, domain.SoftBlockScore + apply(&block, block.Source, block.RecordKey, 120) + if got := graph(domain.Default, ""); len(got) != 1 || got[0].Score != domain.SoftBlockScore { + t.Fatalf("block precedence: %+v", got) + } + explicit := domain.Vouch{Author: author, Subject: target, Score: 0} + apply(&explicit, domain.Collection, domain.RecordKey(target, ""), 130) + if got := graph(domain.Default, ""); len(got) != 1 || got[0].Score != 0 || got[0].Source != domain.Collection { + t.Fatalf("explicit zero lost: %+v", got) + } + apply(nil, domain.Collection, domain.RecordKey(target, ""), 140) + apply(nil, block.Source, block.RecordKey, 150) + apply(&block, block.Source, block.RecordKey, 120) + if got := graph(domain.Default, ""); len(got) != 1 || got[0].Score != domain.FollowScore { + t.Fatalf("tombstone replay: %+v", got) + } + rows, revision, err := s.TrustGraph(ctx, "", domain.Default, 10) + if err != nil || len(rows) != 1 { + t.Fatal(err) + } + result := trust.Result{Status: trust.CalculatedStatus, Score: &v.Score} + deps := []trust.Dependency{{Author: author, Source: v.Source, RecordKey: v.RecordKey}} + now := time.UnixMicro(1000) + if saved, err := s.SaveTrust(ctx, "follow-cache", result, deps, revision, now.Add(time.Hour)); err != nil || !saved { + t.Fatalf("save: %v %v", saved, err) + } + // The same record key in another collection must not invalidate this path. + apply(nil, domain.BlockCollection, v.RecordKey, 160) + if _, err := s.CachedTrust(ctx, "follow-cache", now); err != nil { + t.Fatalf("source collision: %v", err) + } + apply(nil, v.Source, v.RecordKey, 170) + if _, err := s.CachedTrust(ctx, "follow-cache", now); !errors.Is(err, sql.ErrNoRows) { + t.Fatalf("soft path not invalidated: %v", err) + } + if got := graph(domain.Default, ""); len(got) != 1 || got[0].RecordKey != duplicate.RecordKey { + t.Fatalf("remaining duplicate: %+v", got) + } + if err := s.Replace(ctx, author, []domain.Vouch{v, block}, time.UnixMicro(200)); err != nil { + t.Fatal(err) + } + apply(nil, block.Source, block.RecordKey, 190) + if got := graph(domain.Default, ""); len(got) != 1 || got[0].Score != domain.SoftBlockScore { + t.Fatalf("snapshot watermark: %+v", got) + } + if err := s.Apply(ctx, "soft-stream", domain.Change{Kind: domain.AccountChange, DID: author, TimeUS: 210, Active: false, Status: domain.DeletedAccount}); err != nil { + t.Fatal(err) + } + if err := s.Apply(ctx, "soft-stream", domain.Change{Kind: domain.AccountChange, DID: author, TimeUS: 220, Active: true}); err != nil { + t.Fatal(err) + } + apply(&v, v.Source, v.RecordKey, 205) + if got := graph(domain.Default, ""); len(got) != 0 { + t.Fatalf("deleted account replay: %+v", got) + } +} + +func TestSoftAuthorScope(t *testing.T) { + ctx := context.Background() + for _, mode := range []domain.SoftMode{domain.SoftDisabled, domain.SoftParticipants, domain.SoftAllAuthors} { + s, err := OpenWithSoft(filepath.Join(t.TempDir(), "scope.db"), mode) + if err != nil { + t.Fatal(err) + } + v := domain.Vouch{Author: "did:plc:observer", Subject: "did:plc:target", Source: domain.FollowCollection, RecordKey: "follow", Score: domain.FollowScore} + if err := s.Apply(ctx, "stream", domain.Change{Kind: domain.RecordChange, DID: v.Author, Source: v.Source, RecordKey: v.RecordKey, Vouch: &v, TimeUS: 100}); err != nil { + t.Fatal(err) + } + rows, err := s.Incoming(ctx, v.Subject, "", domain.Default, 10) + if err != nil { + t.Fatal(err) + } + want := 0 + if mode == domain.SoftAllAuthors { + want = 1 + } + if len(rows) != want { + t.Fatalf("scope %v: %+v", mode, rows) + } + if _, err := s.Participant(ctx, v.Author); !errors.Is(err, sql.ErrNoRows) { + t.Fatalf("soft author enrolled: %v", err) + } + var count int + if err := s.conn.QueryRow("SELECT count(*) FROM soft_vouches").Scan(&count); err != nil || count != want { + t.Fatalf("disabled indexing: %d %v", count, err) + } + s.Close() + } +} diff --git a/internal/storage/sqlite.go b/internal/storage/sqlite.go index 2208763..a9aecae 100644 --- a/internal/storage/sqlite.go +++ b/internal/storage/sqlite.go @@ -25,14 +25,31 @@ var migration2 string //go:embed migration3.sql var migration3 string -const SchemaVersion = 4 +//go:embed migration4.sql +var migration4 string + +//go:embed migration5.sql +var migration5 string + +//go:embed soft_schema.sql +var softSchema string + +const SchemaVersion = 5 type SQLite struct { - conn *sql.DB - queries *db.Queries + conn *sql.DB + queries *db.Queries + softMode domain.SoftMode } func Open(path string) (*SQLite, error) { + return OpenWithSoft(path, domain.SoftDisabled) +} + +func OpenWithSoft(path string, mode domain.SoftMode) (*SQLite, error) { + if mode < domain.SoftDisabled || mode > domain.SoftAllAuthors { + return nil, errors.New("invalid soft vouch mode") + } if err := os.MkdirAll(filepath.Dir(path), 0700); err != nil { return nil, err } @@ -62,7 +79,7 @@ func Open(path string) (*SQLite, error) { conn.Close() return nil, err } - return &SQLite{conn, db.New(conn)}, nil + return &SQLite{conn: conn, queries: db.New(conn), softMode: mode}, nil } // Apply schema upgrades atomically and reject databases from a newer release. @@ -103,9 +120,23 @@ func migrate(conn *sql.DB) error { return err } } + // Older schemas first acquire cache tables, then upgrade their dependency identities. + if version > 0 && version < 4 { + if _, err := tx.Exec(migration4); err != nil { + return err + } + } + if version > 0 && version < SchemaVersion { + if _, err := tx.Exec(migration5); err != nil { + return err + } + } if _, err := tx.Exec(schema); err != nil { return err } + if _, err := tx.Exec(softSchema); err != nil { + return err + } if _, err := tx.Exec(fmt.Sprintf("PRAGMA user_version=%d", SchemaVersion)); err != nil { return err } @@ -167,6 +198,15 @@ func (s *SQLite) Replace(ctx context.Context, did string, vouches []domain.Vouch return err } for _, v := range vouches { + if domain.SoftSource(v.Source) { + if s.softMode == domain.SoftDisabled { + return errors.New("soft snapshots are disabled") + } + continue + } + if v.Source != "" && v.Source != domain.Collection { + return errors.New("invalid explicit vouch source") + } if v.Author != did { return errors.New("vouch author does not match repository") } @@ -196,6 +236,11 @@ func (s *SQLite) Replace(ctx context.Context, did string, vouches []domain.Vouch return err } } + if s.softMode != domain.SoftDisabled { + if err := replaceSoft(ctx, q, did, vouches, at); err != nil { + return err + } + } if err := q.SetBootstrap(ctx, db.SetBootstrapParams{Did: did, SyncedAt: at.UTC().Format(time.RFC3339Nano), BootstrapUs: at.UnixMicro(), AccountUs: at.UnixMicro()}); err != nil { return err } @@ -203,12 +248,7 @@ func (s *SQLite) Replace(ctx context.Context, did string, vouches []domain.Vouch } func (s *SQLite) Graph(ctx context.Context, tag string, mode domain.Mode, limit int) ([]domain.Vouch, error) { - fallback := int64(0) - if mode == domain.Fallback { - fallback = 1 - } - rows, err := s.queries.Graph(ctx, db.GraphParams{Tag: tag, Fallback: fallback, MaxRows: int64(limit)}) - return vouchRows(rows), err + return s.graphRows(ctx, s.queries, tag, mode, limit) } func (s *SQLite) OwnVouches(ctx context.Context, did string, limit, offset int) ([]domain.Vouch, error) { @@ -223,7 +263,7 @@ func (s *SQLite) Tags(ctx context.Context, prefix string, limit int) ([]string, func vouchRows(rows []db.Vouch) []domain.Vouch { result := make([]domain.Vouch, 0, len(rows)) for _, v := range rows { - result = append(result, domain.Vouch{Author: v.Author, Subject: v.Subject, Tag: v.Tag, Score: int(v.Score), CreatedAt: v.CreatedAt, Comment: v.Comment}) + result = append(result, domain.Vouch{Author: v.Author, Subject: v.Subject, Tag: v.Tag, Score: float64(v.Score), CreatedAt: v.CreatedAt, Comment: v.Comment, Source: v.Source, RecordKey: v.RecordKey}) } return result } diff --git a/internal/storage/store.go b/internal/storage/store.go index e469761..29464e1 100644 --- a/internal/storage/store.go +++ b/internal/storage/store.go @@ -32,6 +32,7 @@ type Store interface { Incoming(context.Context, string, string, domain.Mode, int) ([]domain.Vouch, error) 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) } type Secrets interface { diff --git a/internal/storage/trust_cache.go b/internal/storage/trust_cache.go index cb01143..8944c38 100644 --- a/internal/storage/trust_cache.go +++ b/internal/storage/trust_cache.go @@ -18,11 +18,7 @@ func (s *SQLite) TrustGraph(ctx context.Context, tag string, mode domain.Mode, l } defer tx.Rollback() q := s.queries.WithTx(tx) - fallback := int64(0) - if mode == domain.Fallback { - fallback = 1 - } - rows, err := q.Graph(ctx, db.GraphParams{Tag: tag, Fallback: fallback, MaxRows: int64(limit)}) + rows, err := s.graphRows(ctx, q, tag, mode, limit) if err != nil { return nil, 0, err } @@ -33,7 +29,7 @@ func (s *SQLite) TrustGraph(ctx context.Context, tag string, mode domain.Mode, l if err := tx.Commit(); err != nil { return nil, 0, err } - return vouchRows(rows), revision, nil + return rows, revision, nil } func (s *SQLite) Incoming(ctx context.Context, target, tag string, mode domain.Mode, limit int) ([]domain.Vouch, error) { @@ -42,7 +38,19 @@ func (s *SQLite) Incoming(ctx context.Context, target, tag string, mode domain.M fallback = 1 } rows, err := s.queries.Incoming(ctx, db.IncomingParams{Subject: target, Tag: tag, Column3: fallback, Limit: int64(limit)}) - return vouchRows(rows), err + if err != nil { + return nil, err + } + result := vouchRows(rows) + if s.softMode == domain.SoftDisabled || mode == domain.Tagged { + return result, nil + } + all := int64(0) + if s.softMode == domain.SoftAllAuthors { + 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 } func (s *SQLite) CachedTrust(ctx context.Context, key string, now time.Time) (trust.Result, error) { @@ -81,7 +89,11 @@ func (s *SQLite) SaveTrust(ctx context.Context, key string, result trust.Result, return false, err } for _, dependency := range dependencies { - if err := q.SaveDependency(ctx, db.SaveDependencyParams{CacheKey: key, Author: dependency.Author, RecordKey: dependency.RecordKey}); err != nil { + source := dependency.Source + if source == "" { + source = domain.Collection + } + if err := q.SaveDependency(ctx, db.SaveDependencyParams{CacheKey: key, Author: dependency.Author, Source: source, RecordKey: dependency.RecordKey}); err != nil { return false, err } } diff --git a/internal/trust/algorithms/soft_test.go b/internal/trust/algorithms/soft_test.go new file mode 100644 index 0000000..d17e1b4 --- /dev/null +++ b/internal/trust/algorithms/soft_test.go @@ -0,0 +1,23 @@ +package algorithms_test + +import ( + "testing" + + "provenance/internal/domain" + "provenance/internal/trust" + "provenance/internal/trust/algorithms" + "provenance/internal/trust/trusttest" +) + +func TestSoftPaths(t *testing.T) { + for _, algorithm := range []trust.Algorithm{algorithms.Optimistic{}, algorithms.ShortestPath{}} { + t.Run(algorithm.Describe().Name, func(t *testing.T) { + trusttest.Run(t, algorithm, []trusttest.Scenario{ + {Name: "follow", Source: "alice", Target: "bob", Edges: []trusttest.Edge{{From: "alice", To: "bob", Score: domain.FollowScore, Source: domain.FollowCollection}}, Want: calculated(.5, "alice", "bob")}, + {Name: "soft block is not a hard block", Source: "alice", Target: "bob", Edges: []trusttest.Edge{{From: "alice", To: "bob", Score: domain.SoftBlockScore, Source: domain.BlockCollection}}, Want: calculated(-.5, "alice", "bob")}, + {Name: "follow chain", Source: "alice", Target: "bob", Edges: []trusttest.Edge{{From: "alice", To: "carol", Score: domain.FollowScore, Source: domain.FollowCollection}, {From: "carol", To: "bob", Score: domain.FollowScore, Source: domain.FollowCollection}}, Want: calculated(.4, "alice", "carol", "bob")}, + {Name: "soft bottleneck", Source: "alice", Target: "bob", Edges: []trusttest.Edge{{From: "alice", To: "carol", Score: 2}, {From: "carol", To: "bob", Score: domain.SoftBlockScore, Source: domain.BlockCollection}}, Want: calculated(-.6, "alice", "carol", "bob")}, + }) + }) + } +} diff --git a/internal/trust/algorithms/traversal.go b/internal/trust/algorithms/traversal.go index a549b95..5ff7f1e 100644 --- a/internal/trust/algorithms/traversal.go +++ b/internal/trust/algorithms/traversal.go @@ -12,14 +12,14 @@ import ( type state struct { node string - capacity int + capacity float64 } type step struct { state state parent int depth int - assertion int + assertion float64 } type pathPolicy int @@ -45,7 +45,7 @@ func calculate(ctx context.Context, edges []domain.Vouch, source, target string, adj := make(map[string][]domain.Vouch) for _, v := range edges { - if err := domain.ValidateScore(v.Score); err != nil { + if err := domain.ValidateVouch(v); err != nil { return trust.Result{}, err } if v.Author == source && v.Subject == target && v.Score == domain.Blocked { diff --git a/internal/trust/cache.go b/internal/trust/cache.go index 982af9c..e3e0783 100644 --- a/internal/trust/cache.go +++ b/internal/trust/cache.go @@ -8,4 +8,5 @@ const DefaultCacheTTL = 24 * time.Hour type Dependency struct { Author string RecordKey string + Source string } diff --git a/internal/trust/graph.go b/internal/trust/graph.go index 33e9e4c..e752cba 100644 --- a/internal/trust/graph.go +++ b/internal/trust/graph.go @@ -31,10 +31,11 @@ type Result struct { } type PathScore struct { - DID string `json:"did"` - Assertion *int `json:"assertion"` - Score float64 `json:"score"` - Comment string `json:"comment,omitempty"` + DID string `json:"did"` + Assertion *float64 `json:"assertion"` + Source string `json:"source,omitempty"` + Score float64 `json:"score"` + Comment string `json:"comment,omitempty"` } type Limits struct { @@ -47,12 +48,15 @@ type Limits struct { func Select(vouches []domain.Vouch, tag string, mode domain.Mode) []domain.Vouch { chosen := make(map[[2]string]domain.Vouch) for _, v := range vouches { + if domain.SoftSource(v.Source) && mode == domain.Tagged { + continue + } if v.Tag != tag && !(mode == domain.Fallback && v.Tag == "") { continue } key := [2]string{v.Author, v.Subject} previous, exists := chosen[key] - if !exists || v.Tag == tag && previous.Tag != tag { + if !exists || preferVouch(v, previous, tag) { chosen[key] = v } } @@ -68,3 +72,31 @@ func Select(vouches []domain.Vouch, tag string, mode domain.Mode) []domain.Vouch }) return result } + +func preferVouch(candidate, previous domain.Vouch, tag string) bool { + if vouchRank(candidate) != vouchRank(previous) { + return vouchRank(candidate) > vouchRank(previous) + } + if candidate.Tag != previous.Tag { + return candidate.Tag == tag + } + return candidate.RecordKey < previous.RecordKey +} + +type selectionRank int + +const ( + followRank selectionRank = iota + softBlockRank + explicitRank +) + +func vouchRank(v domain.Vouch) selectionRank { + if v.Source == domain.FollowCollection { + return followRank + } + if v.Source == domain.BlockCollection { + return softBlockRank + } + return explicitRank +} diff --git a/internal/trust/trusttest/scenario.go b/internal/trust/trusttest/scenario.go index a6bbc52..0d8bd93 100644 --- a/internal/trust/trusttest/scenario.go +++ b/internal/trust/trusttest/scenario.go @@ -20,9 +20,10 @@ const ( ) type Edge struct { - From string - To string - Score int + From string + To string + Score float64 + Source string } type Outcome struct { @@ -61,7 +62,7 @@ func (s Scenario) Input() trust.Input { input.Limits = *s.Limits } for _, edge := range s.Edges { - input.Edges = append(input.Edges, domain.Vouch{Author: Account(edge.From), Subject: Account(edge.To), Score: edge.Score}) + input.Edges = append(input.Edges, domain.Vouch{Author: Account(edge.From), Subject: Account(edge.To), Score: edge.Score, Source: edge.Source, RecordKey: domain.RecordKey(Account(edge.To), "")}) } if s.Accounts != nil { input.Accounts = trust.Accounts{} diff --git a/internal/ui/http.go b/internal/ui/http.go index 989c024..206d400 100644 --- a/internal/ui/http.go +++ b/internal/ui/http.go @@ -234,7 +234,7 @@ func (h *HTTP) put(w http.ResponseWriter, r *http.Request) { h.fail(w, r, invalid("Enter an account and choose a trust level.", err)) return } - result, err := h.service.Put(r.Context(), browser(r), domain.Vouch{Subject: body.Subject, Tag: body.Tag, Score: *body.Score, Comment: body.Comment}) + result, err := h.service.Put(r.Context(), browser(r), domain.Vouch{Subject: body.Subject, Tag: body.Tag, Score: float64(*body.Score), Comment: body.Comment}) if err != nil { h.fail(w, r, err) return diff --git a/internal/ui/web/app.js b/internal/ui/web/app.js index 332be91..18540ca 100644 --- a/internal/ui/web/app.js +++ b/internal/ui/web/app.js @@ -3,6 +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 RATINGS = Object.freeze({ "-3": {label: "Blocked", color: "blocked"}, "-2": {label: "Avoid", color: "red"}, @@ -208,6 +209,13 @@ function rating(score) { if (score === null || score === undefined) return {label: "No connection yet", color: "grey"}; return RATINGS[String(Math.round(score))] || RATINGS["0"]; } + +// Show the source so that soft assertions retain their distinct labels. +function assertionLabel(step) { + 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; +} function dot(score) { const node = document.createElement("span"); node.className = `score-dot ${rating(score).color}`; @@ -270,7 +278,7 @@ function showResult(result, target) { result.path_scores.forEach((step, index) => { const item = text("li", index === 0 ? "You" : ""); if (index > 0) item.append(accountLabel(step.did)); - item.append(text("small", step.assertion === null ? "Your starting point" : `Previous person's vouch: ${rating(step.assertion).label}. Trust along this path: ${step.score.toFixed(2)}.`)); + item.append(text("small", step.assertion === null ? "Your starting point" : `Previous person's vouch: ${assertionLabel(step)}. Trust along this path: ${step.score.toFixed(2)}.`)); path.append(item); }); details.append(path); diff --git a/nix/modules/appview.nix b/nix/modules/appview.nix index e07c4bb..ae1e17a 100644 --- a/nix/modules/appview.nix +++ b/nix/modules/appview.nix @@ -55,10 +55,17 @@ in default = "24h"; description = "Trust cache lifetime as a positive Go duration, such as 24h or 30m."; }; + + enableSoftVouch = lib.mkEnableOption "Bluesky follows and blocks as virtual trust assertions"; + softVouchAllAuthors = lib.mkEnableOption "soft assertions from all observed stream authors"; }; config = lib.mkIf cfg.enable { assertions = [ + { + assertion = !cfg.softVouchAllAuthors || cfg.enableSoftVouch; + message = "softVouchAllAuthors requires enableSoftVouch."; + } { assertion = common.validSecret cfg.serviceKeyFile; message = "services.provenance.appview.serviceKeyFile must name an absolute runtime file outside the Nix store."; @@ -76,6 +83,8 @@ in PROVENANCE_SLINGSHOT_URL = cfg.slingshotURL; PROVENANCE_HOP_PENALTY = toString cfg.hopPenalty; PROVENANCE_TRUST_CACHE_TTL = cfg.trustCacheTTL; + ENABLE_SOFT_VOUCH = lib.boolToString cfg.enableSoftVouch; + ENABLE_SOFT_VOUCH_ALL_AUTHORS = lib.boolToString cfg.softVouchAllAuthors; }; }; }; diff --git a/nix/services/provenance.nix b/nix/services/provenance.nix index ac9c56a..09d54d4 100644 --- a/nix/services/provenance.nix +++ b/nix/services/provenance.nix @@ -71,6 +71,8 @@ let export PROVENANCE_UI_DB="$state/ui.db" 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_ALL_AUTHORS="''${ENABLE_SOFT_VOUCH_ALL_AUTHORS:-${lib.boolToString cfg.softVouchAllAuthors}}" exec ${cfg.package}/bin/${binary} ''; }; @@ -102,6 +104,8 @@ in default = "24h"; description = "Trust cache lifetime as a positive Go duration."; }; + enableSoftVouch = lib.mkEnableOption "Bluesky follows and blocks as virtual assertions"; + softVouchAllAuthors = lib.mkEnableOption "soft assertions from all observed authors"; uiPort = lib.mkOption { type = lib.types.port; default = 8000; diff --git a/nix/tests/modules.nix b/nix/tests/modules.nix index 016266d..5475c14 100644 --- a/nix/tests/modules.nix +++ b/nix/tests/modules.nix @@ -32,6 +32,8 @@ let slingshotURL = "https://records.example.com"; hopPenalty = 0.25; trustCacheTTL = "48h"; + enableSoftVouch = true; + softVouchAllAuthors = true; }; uiConfig.services.provenance.ui = { enable = true; @@ -114,6 +116,8 @@ assert appUnit.environment.PROVENANCE_JETSTREAM_URL == "wss://stream.example.com assert appUnit.environment.PROVENANCE_SLINGSHOT_URL == "https://records.example.com"; 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_ALL_AUTHORS == "true"; assert !(appUnit.environment ? PROVENANCE_SERVICE_KEY); assert !(uiUnit.environment ? PROVENANCE_ENCRYPTION_KEY); assert diff --git a/sqlc.yaml b/sqlc.yaml index adc8ba6..b9ff264 100644 --- a/sqlc.yaml +++ b/sqlc.yaml @@ -1,7 +1,9 @@ version: "2" sql: - engine: sqlite - schema: internal/storage/schema.sql + schema: + - internal/storage/schema.sql + - internal/storage/soft_schema.sql queries: internal/storage/queries.sql gen: go: