From 20d98a396d448d150125e94243937afdda0806cb Mon Sep 17 00:00:00 2001 From: Bretton <36870434+BrettM86@users.noreply.github.com> Date: Wed, 8 Jul 2026 18:33:44 -0700 Subject: [PATCH] =?UTF-8?q?Task=2007:=20vote=20aggregation=20side=20channe?= =?UTF-8?q?l=20=E2=80=94=20aggregator,=20XRPC,=20seeding?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Lemmy Like/Dislike/Undo activities become bridge-side aggregate counts served over social.coves.bridge.getVoteAggregates (lexicon published under lexicons/social/coves/bridge/). Votes never become PDS records (PLAN decision 7). - Migration 010: vote_aggregates (+ seeded_* baseline columns so re-seeding is idempotent and never clobbers live votes) and append-only vote_events (activity_id dedupe, ≤1 live row per voter+subject, recompute-in-tx under the aggregate row lock). - internal/votes.Aggregator fills the task-06 ingest.VoteAggregator seam: latest-state counting (flips, undos, replays), id-targeted retraction, malformed/unknown/unbridged votes drop as processed no-ops (queue ordering keys can never wedge on vote noise). - Authority binding from review: announced votes require subject ∈ announcing community (post mapping DID / comment reply.root DID — not SameAuthority, since Lemmy hosts post objects on the author's instance); bare Undo{vote} binds inner actor to signer like bare Delete. - XRPC: ≤100 uris, at-uri validation, unknown uris omitted, request-order response, Cache-Control, per-IP limiter (fail-closed 50k cap, throttled sweep). - Optional Lemmy API count seeding during backfill (SEED_COUNTS_FROM_API, SSRF-guarded, presence-checked decode so a wrong-shape 200 cannot zero-clobber a baseline). Reviewed by 6 reviewers (4 Claude specialty + codex gpt-5.5 + security; Gemini perm-denied, GLM watchdog-killed); 13 fixes + ~20 regression tests. Co-Authored-By: Claude Fable 5 --- LOOP_STATE.md | 51 +- README.md | 35 ++ cmd/tidepool/main.go | 30 +- internal/config/config.go | 29 ++ .../db/migrations/010_vote_aggregates.sql | 58 +++ internal/ingest/backfill.go | 33 ++ internal/ingest/backfill_test.go | 61 +++ internal/ingest/consent.go | 25 + internal/ingest/handler.go | 8 + internal/ingest/handler_test.go | 78 +++ internal/store/migrations_test.go | 3 +- internal/votes/aggregator.go | 473 ++++++++++++++++++ internal/votes/aggregator_test.go | 425 ++++++++++++++++ internal/votes/e2e_test.go | 203 ++++++++ internal/votes/ratelimit.go | 103 ++++ internal/votes/ratelimit_test.go | 98 ++++ internal/votes/seed.go | 147 ++++++ internal/votes/seed_test.go | 148 ++++++ internal/votes/votes_test.go | 161 ++++++ internal/votes/xrpc.go | 221 ++++++++ internal/votes/xrpc_test.go | 264 ++++++++++ .../coves/bridge/getVoteAggregates.json | 66 +++ 22 files changed, 2716 insertions(+), 4 deletions(-) create mode 100644 internal/db/migrations/010_vote_aggregates.sql create mode 100644 internal/votes/aggregator.go create mode 100644 internal/votes/aggregator_test.go create mode 100644 internal/votes/e2e_test.go create mode 100644 internal/votes/ratelimit.go create mode 100644 internal/votes/ratelimit_test.go create mode 100644 internal/votes/seed.go create mode 100644 internal/votes/seed_test.go create mode 100644 internal/votes/votes_test.go create mode 100644 internal/votes/xrpc.go create mode 100644 internal/votes/xrpc_test.go create mode 100644 lexicons/social/coves/bridge/getVoteAggregates.json diff --git a/LOOP_STATE.md b/LOOP_STATE.md index 944ad06..4ad09cb 100644 --- a/LOOP_STATE.md +++ b/LOOP_STATE.md @@ -12,7 +12,7 @@ update this file → schedule next. Stop the loop when every task is `done`. | 4 | 04-sync-firehose | done | (see git log) | 7/8 reviewers (glm watchdog-killed); fixes: ping starvation, prune-mid-replay OutdatedCursor, broadcaster closed-channel, SeqBounds dirty-read, pruner fail-closed; consent-on-firehose deferred to 06 | | 5 | 05-materializer | done | (see git log) | 8 reviewers; fixes: id-authority binding, Note-root panic, create-after-delete, nobridge scrub, embedded-actor trust, byte caps, uri scheme, Group-type check + 9 regression tests | | 6 | 06-ingestion | done | (see git log) | 8 reviewers (5 Claude + codex/gemini; glm wandered, no JSON); fixes: announced-Delete/Undo scoped to announcer authority (+actor-delete only self), bare Update{Person/Group} no-mint gate, announce content community-authority check, Undo{Delete} restore compensation, handleAccept pending-only, queue lease fencing token + shutdown-cancel handling + processed/poisoned exclusivity, backfillReplies tombstone check, truncation leaves resumable, activityID rand-fail propagates + 14 regression tests | -| 7 | 07-vote-aggregates | pending | | | +| 7 | 07-vote-aggregates | done | (see git log) | 6/8 reviewers (Gemini perm-denied, glm watchdog-killed); fixes: announced-vote subject↔community binding (post mapping-DID / comment reply.root), bare Undo{Like} signer binding, RetractVote id-targeted undo, dup-id 0/0 aggregate-row leak, seeder zero-clobber presence check, limiter sweep-throttle + 50k fail-closed cap, at-uri validation + ~20 regression tests | | 8 | 08-e2e-harness | pending | | | Statuses: pending → in-progress → review → done (or blocked: ). @@ -342,3 +342,52 @@ and deferred TODOs here) vs what real Lemmy actually queries (needs live Lemmy). activityID rand- fail path is guarded but unit-untestable (Go 1.24+ crypto/rand failure is a fatal crash, not a returnable error). + +### From task 07 (vote aggregates — task 08 consumes this) +- internal/votes: Aggregator implements ingest.VoteAggregator. + NewAggregator(db, apObjects, communities, records, logger) — `records` is + a narrow votes.RecordReader (satisfied by repo.Manager, wired in main.go). + XRPC: GET /xrpc/social.coves.bridge.getVoteAggregates (lexicon at + lexicons/social/coves/bridge/, README documents it as the AppView + integration point). ≤100 uris counted PRE-dedupe; malformed at-uri → + InvalidRequest 400 (validated via syntax.ParseATURI); unknown uris + omitted; response preserves request order; Cache-Control public/max-age=30; + per-IP token bucket on RemoteAddr only (XFF deliberately ignored), + FAIL-CLOSED at 50k buckets (unknown IPs get 429 during rotation floods), + sweep throttled to 1/min, injectable clock for tests. +- Vote state machine: append-only vote_events, invariant ≤1 non-undone row + per (voter,subject) is APP-enforced only — every mutation MUST take the + vote_aggregates row lock FIRST (lockAggregate upsert / SELECT FOR UPDATE), + then recompute-in-tx. Dedupe by activity_id (unique, first-writer-wins — + forged-id squatting narrowed by authority binding but ids remain + unauthenticated). RetractVote targets the undone activity's own id when + vote.ID != "" (replay-proof), direction-match fallback for id-less undos. + Dup activity id is probed BEFORE creating an aggregate row (no 0/0 rows). +- AUTHORITY MODEL for votes: announced votes require the announcer to + resolve to a followed community AND subject∈that community — posts via + mapping.DID == community DID (NOT SameAuthority: Lemmy hosts post objects + on the AUTHOR's instance), comments via one GetRecord read of reply.root's + DID. Bare Like/Dislike: inbox already binds actor↔signer (403). + Bare Undo{vote}: Handler.authorizeBareVote (consent.go, host-granularity + SameAuthority like bare Delete). All mismatches drop at debug as nil — + queue outcome contract (nil=processed) preserved everywhere; only real DB + errors are retryable. +- Seeding: seeded_upvotes/seeded_downvotes baseline columns (beyond spec — + deliberate: served = seeded + live, re-seed idempotent, never clobbers + live votes). LemmySeeder.SeedPostCounts posts-only via SSRF-guarded + client, presence-checked decode (missing post_view/counts → error, old + baseline survives; a wrong-shape 200 can NOT zero-clobber). Wired as + optional ingest.CountSeeder in backfill (best-effort, Warn on failure, + never affects run outcome). SEED_COUNTS_FROM_API (default on; strict bool + parse rejects typos). Known accepted drift: a baseline voter who flips + sends Undo for a like we never saw (no-op) — stale until re-seed. +- DEFERRED (task 08+): vote_events grows unbounded (no pruning of + superseded/undone rows — mirror FIREHOSE_RETENTION treatment alongside + ap_tombstones); actor-delete/consent revocation does NOT scrub that + actor's vote_events rows (inconsistent with scrub posture elsewhere; + counts are anonymous on the wire so exposure is low); no concurrency + stress test of the aggregate-row locking claim (sequential tests only); + subject resolution happens outside the mutation tx (narrow TOCTOU with a + racing Delete, documented); no upper sanity cap on seeded counts; + comment count seeding skipped (per-comment API calls would triple + backfill egress). diff --git a/README.md b/README.md index a597b35..fd77ff3 100644 --- a/README.md +++ b/README.md @@ -54,6 +54,7 @@ production**: | `BACKFILL_MAX_POSTS` | `100` | posts materialized per community backfill run | | `MINT_RATE_PER_MINUTE` / `MINT_BURST` | `60` / `120` | rate gate on inbound DID minting (PLC registrations are forever; unseen authors in delivered content trigger mints) | | `INGEST_WORKERS` | `4` | inbox queue worker-pool size | +| `SEED_COUNTS_FROM_API` | on | seed backfilled posts' vote aggregates from the origin instance's public API (`/api/v3/post` `counts`); set `0` to disable | ## Subscribing to communities (admin API) @@ -132,6 +133,40 @@ it as a `subscribeRepos` upstream: Repos whose actor revoked consent (tombstoned) report `RepoDeactivated` / `active: false` and their content endpoints stop serving. +## Vote aggregates (the AppView integration point) + +Votes never become records (nothing may strongRef a vote): Lemmy +`Like`/`Dislike`/`Undo` activities maintain bridge-side aggregate counts, +served over **one sanctioned side-channel XRPC** the Coves AppView polls: + +``` +GET /xrpc/social.coves.bridge.getVoteAggregates?uris=at://…&uris=at://… +``` + +- `uris`: at-uris of bridged posts/comments — repeated `uris` params (the + atproto convention; comma-separated values are also accepted), at most + **100** per call (`InvalidRequest` beyond that). +- Response: `{"aggregates":[{"uri","upvotes","downvotes","updatedAt"}]}` in + request order. Unknown or never-voted uris are **omitted**, not an error. +- Public read, `Cache-Control: public, max-age=30`, rate limited per client + IP (token bucket; deployments behind a proxy should rate-limit the real + client at the edge — the bridge deliberately ignores `X-Forwarded-For`). + +The contract is the lexicon at +[`lexicons/social/coves/bridge/getVoteAggregates.json`](lexicons/social/coves/bridge/getVoteAggregates.json) +and is versioned by nsid: breaking changes ship under a new name. + +Counts reflect each distinct voter's **latest** state — flips +(`Like` → `Dislike`) and `Undo`s are folded in, re-delivered activities are +deduplicated by activity id. Votes on content the bridge never materialized +are dropped (logged at debug). Known limitation: AP delivers votes only +going forward, and Lemmy outboxes announce historical Likes sparsely — so +backfilled posts would start near zero. `SEED_COUNTS_FROM_API` (default on) +compensates by seeding a baseline from the origin's public API during +backfill; live votes stack on top, and an undo of a vote that only exists in +the baseline is a no-op (accepted drift, refreshed on re-seed). Comment +scores are not seeded in v1 — comments accumulate live votes only. + ## Verifying with Jetstream The task-04 integration proof: run a real Jetstream against the bridge and diff --git a/cmd/tidepool/main.go b/cmd/tidepool/main.go index 20b5720..6d70239 100644 --- a/cmd/tidepool/main.go +++ b/cmd/tidepool/main.go @@ -26,6 +26,7 @@ import ( "tidepool/internal/repo" "tidepool/internal/store" tidepoolsync "tidepool/internal/sync" + "tidepool/internal/votes" ) const ( @@ -205,11 +206,31 @@ func run(logger *slog.Logger) error { return err } + // The vote aggregation side channel (task 07): Like/Dislike activities + // maintain bridge-side counts (never records), served over + // social.coves.bridge.getVoteAggregates. + voteAggregator, err := votes.NewAggregator(database, objects, communities, repoManager, logger) + if err != nil { + return err + } + // Seeding imports historical scores for backfilled posts from the origin + // instance's public API (AP alone cannot provide them). + var seeder ingest.CountSeeder + if cfg.SeedCountsFromAPI { + lemmySeeder, err := votes.NewLemmySeeder(voteAggregator, + ap.NewGuardedHTTPClient(cfg.AllowPrivateAddresses, 30*time.Second), cfg.UserAgent, logger) + if err != nil { + return err + } + seeder = lemmySeeder + } + backfill, err := ingest.NewBackfill(ingest.BackfillOptions{ Fetcher: apClient, Materializer: materializer, Communities: communities, Tombstones: tombstones, + Seeder: seeder, MaxPosts: cfg.BackfillMaxPosts, // Async runs derive from the run context so a mid-run backfill stops // pulling remote pages once shutdown starts; the drain below waits for @@ -226,7 +247,7 @@ func run(logger *slog.Logger) error { Objects: objects, Communities: communities, Tombstones: tombstones, - Votes: ingest.NewNoopVotes(logger), // task 07 replaces this + Votes: voteAggregator, Backfill: backfill, ServiceActorID: serviceActor.ID, Logger: logger, @@ -271,7 +292,12 @@ func run(logger *slog.Logger) error { } admin.Routes(router) - // Task 07 registers here: vote aggregates XRPC. + // The vote-aggregate XRPC (the AppView's side-channel read). + votesXRPC, err := votes.NewXRPC(votes.XRPCOptions{DB: database, Logger: logger}) + if err != nil { + return err + } + votesXRPC.Routes(router) server := &http.Server{ Addr: cfg.ListenAddr, diff --git a/internal/config/config.go b/internal/config/config.go index 73846aa..850f4fb 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -95,6 +95,11 @@ type Config struct { // IngestWorkers is the inbox queue worker-pool size (INGEST_WORKERS, // default 4). IngestWorkers int + // SeedCountsFromAPI enables seeding backfilled posts' vote aggregates + // from the origin instance's public API (Lemmy's `counts` field) — + // history whose individual Like activities AP never delivers + // (SEED_COUNTS_FROM_API, default on; set to 0/false to disable). + SeedCountsFromAPI bool } // Load reads configuration from the environment. logger must not be nil; @@ -234,6 +239,10 @@ func Load(logger *slog.Logger) (*Config, error) { if err != nil { return nil, err } + cfg.SeedCountsFromAPI, err = boolVarDefault(logger, "SEED_COUNTS_FROM_API", true) + if err != nil { + return nil, err + } defaultUserAgent := fmt.Sprintf("tidepool/0.1 (+https://%s)", cfg.BridgeHostname) cfg.UserAgent = os.Getenv("USER_AGENT") @@ -297,6 +306,26 @@ func boolVar(name string) bool { return false } +// boolVarDefault parses a boolean environment variable with an explicit +// default when unset (tuning-knob semantics: same default in every +// environment, logged when applied). Unlike boolVar it rejects +// unrecognized values instead of silently reading them as false — a +// default-on flag "disabled" by a typo would be invisible. +func boolVarDefault(logger *slog.Logger, name string, fallback bool) (bool, error) { + raw := strings.ToLower(strings.TrimSpace(os.Getenv(name))) + if raw == "" { + logger.Info(name+" not set, using default", "value", fallback) + return fallback, nil + } + switch raw { + case "1", "true", "yes", "on": + return true, nil + case "0", "false", "no", "off": + return false, nil + } + return false, fmt.Errorf("config: %s must be a boolean (1/0, true/false, yes/no, on/off), got %q", name, raw) +} + // stringVar returns the value of an environment variable. When unset it // falls back to the logged dev default in development and errors in // production. diff --git a/internal/db/migrations/010_vote_aggregates.sql b/internal/db/migrations/010_vote_aggregates.sql new file mode 100644 index 0000000..f65a3fa --- /dev/null +++ b/internal/db/migrations/010_vote_aggregates.sql @@ -0,0 +1,58 @@ +-- +goose Up +-- Task 07: the vote side channel (PLAN.md locked decision 7 — votes never +-- become records). Lemmy Like/Dislike/Undo activities maintain bridge-side +-- aggregate counts served over social.coves.bridge.getVoteAggregates. +-- +-- Two tables because AP vote delivery is a stream of per-voter state +-- changes, not increments: voters flip votes (Like → Undo{Like} → Dislike) +-- and instances re-deliver activities. Counts must reflect each distinct +-- voter's LATEST state, so vote_events tracks per-(voter, subject) state and +-- vote_aggregates holds the served totals, recomputed transactionally. + +-- vote_aggregates: one row per voted-on bridged subject (post/comment). +-- upvotes/downvotes are the SERVED totals: seeded_* (a baseline imported +-- from the origin's public API during backfill — history whose individual +-- Like activities the bridge never saw) plus the live count of non-undone +-- vote_events. The aggregate row also serializes vote writes per subject: +-- every ApplyVote/RetractVote/Seed transaction locks it first, so the +-- recompute always sees a consistent event set. +CREATE TABLE vote_aggregates ( + subject_ap_id TEXT PRIMARY KEY CHECK (subject_ap_id <> ''), -- canonical AP id of the voted-on object + subject_at_uri TEXT NOT NULL CHECK (subject_at_uri <> ''), -- the materialized record (ap_objects.at_uri) + upvotes INTEGER NOT NULL DEFAULT 0 CHECK (upvotes >= 0), + downvotes INTEGER NOT NULL DEFAULT 0 CHECK (downvotes >= 0), + seeded_upvotes INTEGER NOT NULL DEFAULT 0 CHECK (seeded_upvotes >= 0), + seeded_downvotes INTEGER NOT NULL DEFAULT 0 CHECK (seeded_downvotes >= 0), + updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP +); + +-- The XRPC queries by at-uri (that is what the AppView knows). +CREATE INDEX idx_vote_aggregates_at_uri ON vote_aggregates (subject_at_uri); + +-- vote_events: one row per vote ACTIVITY (append-only). activity_id is the +-- dedupe key — a re-delivered Like is a no-op even if the queue's fencing +-- ever lets one through. Invariant maintained by the aggregator: at most one +-- non-undone row per (voter, subject) — that row IS the voter's current +-- vote; superseded and undone votes keep their rows with undone = TRUE, so +-- a stale activity id can never be re-applied. +CREATE TABLE vote_events ( + id BIGSERIAL PRIMARY KEY, + activity_id TEXT NOT NULL CHECK (activity_id <> ''), -- AP Like/Dislike activity id + voter_ap_id TEXT NOT NULL CHECK (voter_ap_id <> ''), -- AP actor id of the voter + subject_ap_id TEXT NOT NULL CHECK (subject_ap_id <> ''), -- AP id of the voted-on object + direction TEXT NOT NULL CHECK (direction IN ('up', 'down')), + undone BOOLEAN NOT NULL DEFAULT FALSE, -- superseded by a newer vote, or Undo{Like|Dislike} + created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, + CONSTRAINT vote_events_activity_id_key UNIQUE (activity_id) +); + +-- Current-state lookup (supersede on flip, Undo targeting) and the +-- per-subject recompute both scan live events only. +CREATE INDEX idx_vote_events_live ON vote_events (subject_ap_id, voter_ap_id) + WHERE NOT undone; + +-- +goose Down +DROP INDEX IF EXISTS idx_vote_events_live; +DROP TABLE IF EXISTS vote_events; +DROP INDEX IF EXISTS idx_vote_aggregates_at_uri; +DROP TABLE IF EXISTS vote_aggregates; diff --git a/internal/ingest/backfill.go b/internal/ingest/backfill.go index 2a202df..67793ac 100644 --- a/internal/ingest/backfill.go +++ b/internal/ingest/backfill.go @@ -33,6 +33,15 @@ type BackfillFetcher interface { FetchCollection(ctx context.Context, iri string, visit func(*ap.Object) error) error } +// CountSeeder imports a backfilled post's historical vote counts from its +// origin's public API (task 07's votes.LemmySeeder; AP alone cannot provide +// them — outboxes announce historical Likes only sparsely). Optional and +// best-effort: a nil seeder or a seeding failure never affects the backfill +// outcome. +type CountSeeder interface { + SeedPostCounts(ctx context.Context, postAPID string) error +} + // BackfillOptions configures NewBackfill. Fetcher, Materializer, // Communities, and Tombstones are required. type BackfillOptions struct { @@ -40,6 +49,9 @@ type BackfillOptions struct { Materializer Materializer Communities store.Communities Tombstones store.Tombstones + // Seeder, when set, seeds each backfilled post's vote aggregates from + // the origin's public API (config SEED_COUNTS_FROM_API). + Seeder CountSeeder // MaxPosts caps posts per run (default 100, config.BackfillMaxPosts). MaxPosts int // MinInterval is the freshness window for un-forced triggers @@ -63,6 +75,7 @@ type Backfill struct { mat Materializer communities store.Communities tombstones store.Tombstones + seeder CountSeeder maxPosts int minInterval time.Duration baseCtx context.Context @@ -101,6 +114,7 @@ func NewBackfill(opts BackfillOptions) (*Backfill, error) { mat: opts.Materializer, communities: opts.Communities, tombstones: opts.Tombstones, + seeder: opts.Seeder, maxPosts: opts.MaxPosts, minInterval: opts.MinInterval, baseCtx: baseCtx, @@ -271,6 +285,7 @@ func (b *Backfill) materializeOutboxItem(ctx context.Context, item *ap.Object, c if _, err := b.mat.MaterializePost(ctx, obj); err != nil { return false, err } + b.seedCounts(ctx, obj.ID) b.backfillReplies(ctx, obj) return true, nil case ap.TypeNote: @@ -283,6 +298,24 @@ func (b *Backfill) materializeOutboxItem(ctx context.Context, item *ap.Object, c } } +// seedCounts imports a backfilled post's historical vote counts (task 07). +// Best-effort: failures are logged and never affect the run — a post with a +// zero score is strictly better than no post. Warn (matching the +// backfillReplies convention) so a systemic seeding outage is visible at +// default log levels; a cancellation during shutdown is not an outage. +func (b *Backfill) seedCounts(ctx context.Context, postAPID string) { + if b.seeder == nil { + return + } + if err := b.seeder.SeedPostCounts(ctx, postAPID); err != nil { + if ctx.Err() != nil { + b.logger.Debug("backfill vote-count seeding canceled", "post", postAPID, "error", err) + return + } + b.logger.Warn("backfill vote-count seeding failed", "post", postAPID, "error", err) + } +} + // backfillReplies pages a post's advertised replies collection. Failures // are logged, never fatal — replies are best-effort garnish on backfill. func (b *Backfill) backfillReplies(ctx context.Context, post *ap.Object) { diff --git a/internal/ingest/backfill_test.go b/internal/ingest/backfill_test.go index 082d827..e2615d2 100644 --- a/internal/ingest/backfill_test.go +++ b/internal/ingest/backfill_test.go @@ -3,9 +3,11 @@ package ingest import ( "context" "encoding/json" + "fmt" "net/http" "os" "path/filepath" + "sync" "testing" "time" @@ -99,6 +101,65 @@ func TestBackfillProducesMappedHistory(t *testing.T) { "an un-forced re-trigger inside the freshness window must not re-run") } +// recordingSeeder captures CountSeeder invocations; a non-nil err makes +// every call fail. +type recordingSeeder struct { + mu sync.Mutex + seeded []string + err error +} + +func (s *recordingSeeder) SeedPostCounts(_ context.Context, postAPID string) error { + s.mu.Lock() + defer s.mu.Unlock() + s.seeded = append(s.seeded, postAPID) + return s.err +} + +// TestBackfillSeedsVoteCounts: every materialized post gets exactly one +// seeding call (replies do not — comments are not seeded in v1), and the +// seeder is best-effort: a failing one is still invoked per post but never +// affects the run's outcome. Nil-seeder safety is exercised by every other +// backfill test (newBackfill leaves Seeder unset). +func TestBackfillSeedsVoteCounts(t *testing.T) { + h := newHarness(t) + h.subscribeTechnology() + serveOutboxFixtures(t, h) + ctx := context.Background() + + seeded := func(seeder *recordingSeeder) *Backfill { + b, err := NewBackfill(BackfillOptions{ + Fetcher: h.client, + Materializer: h.mat, + Communities: h.communities, + Tombstones: h.tombstones, + Seeder: seeder, + MaxPosts: 10, + }) + require.NoError(t, err) + return b + } + + seeder := &recordingSeeder{} + community, err := h.communities.GetByAPGroupID(ctx, groupID) + require.NoError(t, err) + require.NoError(t, seeded(seeder).Run(ctx, community, true)) + assert.Equal(t, []string{pageID, secondPageID}, seeder.seeded, + "each materialized post is seeded once (newest first); replies are not") + + // A failing seeder is invisible to the run: no error, and the clean + // completion still stamps last_backfill_at. + failing := &recordingSeeder{err: fmt.Errorf("origin API is down")} + require.NoError(t, seeded(failing).Run(ctx, community, true), + "seeding failures must never fail the backfill") + assert.Equal(t, []string{pageID, secondPageID}, failing.seeded, + "the failing seeder is still invoked per post") + community, err = h.communities.GetByAPGroupID(ctx, groupID) + require.NoError(t, err) + assert.NotNil(t, community.LastBackfillAt, + "a run with seeding failures is still a clean completion") +} + // TestBackfillHonorsMaxPosts: the post cap stops the walk early. func TestBackfillHonorsMaxPosts(t *testing.T) { h := newHarness(t) diff --git a/internal/ingest/consent.go b/internal/ingest/consent.go index b5e1103..2121965 100644 --- a/internal/ingest/consent.go +++ b/internal/ingest/consent.go @@ -112,6 +112,17 @@ func (h *Handler) handleUndo(ctx context.Context, undo *ap.Object, signer, annou } switch inner.Type { case ap.TypeLike, ap.TypeDislike: + // The inbox binds only the OUTER Undo's actor to the signature; the + // inner vote's actor is unverified. A bare undo may therefore only + // retract votes attributed to the signer's own instance — otherwise + // any signer could retract other instances' users' votes. Announced + // undos ride the announcing community's vouching, exactly like + // announced votes (FEP-1b12 group fan-out). + if announcer == "" { + if err := h.authorizeBareVote(undo.ID, inner, signer); err != nil { + return err + } + } return h.votes.RetractVote(ctx, inner, announcer) case ap.TypeDelete: return h.handleUndoDelete(ctx, undo, inner, signer, announcer) @@ -177,6 +188,20 @@ func (h *Handler) handleUndoDelete(ctx context.Context, undo, del *ap.Object, si return nil } +// authorizeBareVote enforces who may cast (or retract) a BARE, un-announced +// vote: the vote's actor must live on the verified signer's authority — host +// granularity, the same instance-is-the-trust-unit rule as bare Delete (an +// instance may speak for its own users, never for another instance's). +// Mismatches (including an actorless vote) drop as processed skips, never +// retryable errors — a retry would wedge the ordering key over a vote. +func (h *Handler) authorizeBareVote(activityID string, vote *ap.Object, signer string) error { + if actor := refID(vote.Actor); !ap.SameAuthority(actor, signer) { + return skip(activityID, fmt.Sprintf( + "bare vote attributed to cross-authority actor %q signed by %s", actor, signer)) + } + return nil +} + // authorizeDelete enforces who may Delete (or Undo{Delete}) a target id. // // - Bare (unannounced): only the target id's OWN authority may delete it — diff --git a/internal/ingest/handler.go b/internal/ingest/handler.go index 98356bd..54674f6 100644 --- a/internal/ingest/handler.go +++ b/internal/ingest/handler.go @@ -142,6 +142,14 @@ func (h *Handler) Process(ctx context.Context, event *store.InboxEvent) error { return h.handleReject(ctx, activity, signer) case ap.TypeLike, ap.TypeDislike: // Bare votes (rare; Lemmy normally announces them via the group). + // The inbox already bound this top-level activity's actor to the + // signer's authority, so for a well-formed vote the check below is + // redundant belt-and-braces; it keeps the dispatch layer's bare-vote + // rule self-contained (and handleUndo, where the inner vote's actor + // is NOT inbox-bound, shares it). + if err := h.authorizeBareVote(activity.ID, activity, signer); err != nil { + return err + } return h.votes.ApplyVote(ctx, activity, "") default: return skip(activity.ID, "unsupported activity type "+activity.Type) diff --git a/internal/ingest/handler_test.go b/internal/ingest/handler_test.go index 959d381..969e7a9 100644 --- a/internal/ingest/handler_test.go +++ b/internal/ingest/handler_test.go @@ -190,6 +190,84 @@ func TestAnnounceLikeRoutedToVotes(t *testing.T) { assert.Equal(t, "Like "+pageID, h.votes.applied[0]) } +// TestBareVoteCrossAuthorityDropped: the inbox binds only the TOP-LEVEL +// activity's actor to the HTTP signature, so vote dispatch must not let a +// signer submit or retract votes attributed to other instances' users. A +// bare Like claiming a foreign actor dies at the inbox binding; a bare +// Undo{Like} carries the unverified actor INSIDE the undo, so the dispatch +// layer drops it — as a processed skip, never a retry that would wedge the +// ordering key. +func TestBareVoteCrossAuthorityDropped(t *testing.T) { + h := newHarness(t) + mallory := h.newRemoteActor("https://evil.example/u/mallory", + person("https://evil.example/u/mallory", "mallory", nil)) + ctx := context.Background() + + // A bare Like attributed to a victim on another instance never clears + // the inbox's actor/signer authority binding. + status := h.deliver(mallory, map[string]any{ + "id": "https://evil.example/activities/like/1", + "type": "Like", + "actor": "https://lemmy.world/u/victim", + "object": pageID, + }) + assert.Equal(t, http.StatusForbidden, status, + "a bare vote with a cross-authority actor must be rejected at the inbox") + + // The Undo passes the inbox (the OUTER actor is mallory's own) but + // attributes the undone vote to the victim: dropped in dispatch. + require.Equal(t, http.StatusAccepted, h.deliver(mallory, map[string]any{ + "id": "https://evil.example/activities/undo/1", + "type": "Undo", + "actor": mallory.id, + "object": map[string]any{ + "id": "https://lemmy.world/activities/like/1", + "type": "Like", + "actor": "https://lemmy.world/u/victim", + "object": pageID, + }, + })) + h.drain() + + event, err := h.events.GetEvent(ctx, "https://evil.example/activities/undo/1") + require.NoError(t, err) + assert.NotNil(t, event.ProcessedAt, "the drop is a processed skip, never a retry") + assert.Empty(t, h.votes.applied) + assert.Empty(t, h.votes.retracted, "a cross-authority undo must never reach the aggregator") +} + +// TestBareVoteSameAuthorityRouted: bare votes and undos attributed to a user +// on the signer's own instance reach the aggregator (host granularity — the +// instance is the trust unit, as with bare Delete). +func TestBareVoteSameAuthorityRouted(t *testing.T) { + h := newHarness(t) + voter := h.newRemoteActor(personID, person(personID, "LeftLeaningFreedomFighters", nil)) + + require.Equal(t, http.StatusAccepted, h.deliver(voter, map[string]any{ + "id": "https://lemmy.world/activities/like/bare-1", + "type": "Like", + "actor": personID, + "object": pageID, + })) + require.Equal(t, http.StatusAccepted, h.deliver(voter, map[string]any{ + "id": "https://lemmy.world/activities/undo/bare-1", + "type": "Undo", + "actor": personID, + "object": map[string]any{ + "id": "https://lemmy.world/activities/like/bare-1", + "type": "Like", + "actor": personID, + "object": pageID, + }, + })) + h.drain() + + require.Len(t, h.votes.applied, 1) + assert.Equal(t, "Like "+pageID, h.votes.applied[0]) + require.Len(t, h.votes.retracted, 1) + assert.Equal(t, "Like "+pageID, h.votes.retracted[0]) +} + // TestEchoSuppression: an inbound activity whose object maps to a record // the bridge itself created (origin=bridge) is dropped. func TestEchoSuppression(t *testing.T) { diff --git a/internal/store/migrations_test.go b/internal/store/migrations_test.go index a92e086..5dd0cc8 100644 --- a/internal/store/migrations_test.go +++ b/internal/store/migrations_test.go @@ -25,7 +25,7 @@ func TestMigrations_UpDownUp(t *testing.T) { SELECT COUNT(*) FROM information_schema.tables WHERE table_schema = 'public' AND table_name IN ('ap_objects', 'bridged_actors', 'communities', 'inbox_events', 'service_keys', - 'blocks', 'repo_state', 'firehose_events') + 'blocks', 'repo_state', 'firehose_events', 'vote_aggregates', 'vote_events') `).Scan(&remaining) require.NoError(t, err) assert.Zero(t, remaining, "down migrations must drop every Tidepool table") @@ -56,6 +56,7 @@ func TestMigrations_UniqueConstraintNames(t *testing.T) { "communities_did_key", "inbox_events_activity_id_key", "service_keys_name_key", + "vote_events_activity_id_key", // the vote dedupe key (task 07) } for _, name := range expected { var exists bool diff --git a/internal/votes/aggregator.go b/internal/votes/aggregator.go new file mode 100644 index 0000000..7fbd955 --- /dev/null +++ b/internal/votes/aggregator.go @@ -0,0 +1,473 @@ +// Package votes is the bridge-side vote aggregation store (task 07, PLAN.md +// locked decision 7): Lemmy Like/Dislike activities never become records — +// they maintain per-subject aggregate counts served over one sanctioned +// side-channel XRPC, social.coves.bridge.getVoteAggregates. +// +// The counting model: AP vote delivery is a stream of per-voter state +// changes, not increments. Voters flip votes (Like → Undo{Like} → Dislike), +// instances re-deliver activities, and Undos arrive for votes the bridge +// never saw. vote_events therefore keeps at most one live (non-undone) row +// per (voter, subject) — the voter's current vote — with every activity id +// recorded for dedupe, and vote_aggregates holds the served totals, +// recomputed from the live events inside the same transaction. The aggregate +// row doubles as the per-subject write lock: every mutation locks it first, +// so concurrent workers can never interleave a stale recompute. +package votes + +import ( + "context" + "database/sql" + stderrors "errors" + "fmt" + "log/slog" + "strings" + + "tidepool/internal/ap" + "tidepool/internal/errors" + "tidepool/internal/materialize" + "tidepool/internal/store" +) + +// Vote directions (vote_events.direction). +const ( + directionUp = "up" + directionDown = "down" +) + +// RecordReader is the slice of *repo.Manager the aggregator uses to read a +// bridged comment's stored record: the record's reply.root strongRef names +// the thread's root post in the community repo, which is how an announced +// comment vote is bound to its announcing community. +type RecordReader interface { + GetRecord(ctx context.Context, did, collection, rkey string) (record map[string]any, recordCID string, err error) +} + +// Aggregator implements ingest.VoteAggregator over the vote_aggregates / +// vote_events tables. It owns its SQL (multi-statement transactions across +// both tables, like internal/repo) and uses store.APObjects only to resolve +// the voted-on subject to its materialized record. Communities and records +// bind announced votes to the announcing community (a followed community may +// only vouch for votes on its OWN content, never inflate or deflate another +// community's scores). +type Aggregator struct { + db *sql.DB + objects store.APObjects + communities store.Communities + records RecordReader + logger *slog.Logger +} + +// NewAggregator validates dependencies and builds an Aggregator. +func NewAggregator(db *sql.DB, objects store.APObjects, communities store.Communities, records RecordReader, logger *slog.Logger) (*Aggregator, error) { + if db == nil { + return nil, errors.NewValidationError("db", "must not be nil") + } + if objects == nil { + return nil, errors.NewValidationError("objects", "must not be nil") + } + if communities == nil { + return nil, errors.NewValidationError("communities", "must not be nil") + } + if records == nil { + return nil, errors.NewValidationError("records", "must not be nil") + } + if logger == nil { + logger = slog.Default() + } + return &Aggregator{db: db, objects: objects, communities: communities, records: records, logger: logger}, nil +} + +// ApplyVote records one Like or Dislike: insert the activity (duplicate +// activity id → no-op), supersede the voter's previous vote on the subject, +// and recompute the aggregate — all in one transaction. Votes on subjects +// the bridge never materialized (or whose mapping is soft-deleted) are +// dropped and logged at debug: the v1 decision is no pending bucket, and +// vote volume makes anything louder than debug unusable. Malformed votes +// (no activity id, voter, or subject) are dropped the same way — there is +// nothing to retry and poisoning the ordering key over a vote helps nobody. +func (a *Aggregator) ApplyVote(ctx context.Context, vote *ap.Object, communityIRI string) error { + if vote == nil { + return nil + } + direction, ok := directionFor(vote.Type) + if !ok { + a.logger.Debug("vote dropped: not a Like/Dislike", "type", vote.Type, "activity", vote.ID) + return nil + } + voter, subject := refID(vote.Actor), refID(vote.Object) + if vote.ID == "" || voter == "" || subject == "" { + a.logger.Debug("vote dropped: missing activity id, voter, or subject", + "activity", vote.ID, "voter", voter, "subject", subject, "community", communityIRI) + return nil + } + + mapping, err := a.subjectMapping(ctx, subject) + if err != nil { + return err + } + if mapping == nil { + a.logger.Debug("vote dropped: subject not bridged", + "subject", subject, "activity", vote.ID, "community", communityIRI) + return nil + } + if communityIRI != "" { + ok, err := a.subjectBelongsToCommunity(ctx, mapping, communityIRI) + if err != nil { + return err + } + if !ok { + a.logger.Debug("vote dropped: subject not in announcing community", + "subject", subject, "activity", vote.ID, "community", communityIRI) + return nil + } + } + + return a.inTx(ctx, func(tx *sql.Tx) error { + // Dedupe BEFORE creating the aggregate row: a duplicate activity id + // aimed at a DIFFERENT subject must not mint a spurious 0/0 aggregate + // for a never-voted subject (the XRPC contract omits never-voted + // uris). A plain SELECT before the lock has no lock-ordering hazard; + // the race with a concurrent first delivery of the same id is closed + // by the unique constraint on the insert below. + var seen bool + if err := tx.QueryRowContext(ctx, ` + SELECT EXISTS (SELECT 1 FROM vote_events WHERE activity_id = $1)`, + vote.ID).Scan(&seen); err != nil { + return fmt.Errorf("dedupe vote event %q: %w", vote.ID, err) + } + if seen { + a.logger.Debug("vote dropped: duplicate activity id", "activity", vote.ID) + return nil + } + + if err := lockAggregate(ctx, tx, subject, mapping.ATURI); err != nil { + return err + } + + // Dedupe backstop: the activity id is unique forever. A concurrent + // first delivery that slipped past the probe above changes nothing. + result, err := tx.ExecContext(ctx, ` + INSERT INTO vote_events (activity_id, voter_ap_id, subject_ap_id, direction) + VALUES ($1, $2, $3, $4) + ON CONFLICT (activity_id) DO NOTHING`, + vote.ID, voter, subject, direction) + if err != nil { + return fmt.Errorf("insert vote event %q: %w", vote.ID, err) + } + inserted, err := result.RowsAffected() + if err != nil { + return fmt.Errorf("insert vote event %q: rows affected: %w", vote.ID, err) + } + if inserted == 0 { + a.logger.Debug("vote dropped: duplicate activity id", "activity", vote.ID) + return nil + } + + // Supersede: a voter has one current vote per subject. The previous + // live vote (same or opposite direction — re-like and flip both land + // here) is marked undone; only the new event stays live. + if _, err := tx.ExecContext(ctx, ` + UPDATE vote_events + SET undone = TRUE + WHERE subject_ap_id = $1 AND voter_ap_id = $2 AND NOT undone AND activity_id <> $3`, + subject, voter, vote.ID); err != nil { + return fmt.Errorf("supersede votes by %q on %q: %w", voter, subject, err) + } + + return recomputeAggregate(ctx, tx, subject) + }) +} + +// RetractVote undoes a previously applied vote (Undo{Like|Dislike}): the +// voter's live vote in the undone direction is marked undone and the +// aggregate recomputed. When the undone activity's own id is known it is +// targeted directly, so a replayed undo can never retract a NEWER re-vote +// that merely shares (voter, subject, direction). Everything that cannot be +// acted on is a logged no-op, never an error: a nil or bare-IRI +// vote.Object/vote.Actor, an undo for a vote the bridge never saw +// (out-of-order delivery, or history that only exists as a seeded baseline), +// an undo whose direction no longer matches the voter's current vote (the +// like it undoes was already superseded by a flip), and an announced undo +// whose subject does not belong to the announcing community. +func (a *Aggregator) RetractVote(ctx context.Context, vote *ap.Object, communityIRI string) error { + if vote == nil { + return nil + } + direction, ok := directionFor(vote.Type) + if !ok { + a.logger.Debug("vote retraction dropped: not a Like/Dislike", "type", vote.Type, "activity", vote.ID) + return nil + } + voter, subject := refID(vote.Actor), refID(vote.Object) + if voter == "" || subject == "" { + a.logger.Debug("vote retraction dropped: missing voter or subject", + "activity", vote.ID, "voter", voter, "subject", subject, "community", communityIRI) + return nil + } + if communityIRI != "" { + mapping, err := a.subjectMapping(ctx, subject) + if err != nil { + return err + } + if mapping == nil { + a.logger.Debug("vote retraction dropped: subject not bridged", + "subject", subject, "activity", vote.ID, "community", communityIRI) + return nil + } + ok, err := a.subjectBelongsToCommunity(ctx, mapping, communityIRI) + if err != nil { + return err + } + if !ok { + a.logger.Debug("vote retraction dropped: subject not in announcing community", + "subject", subject, "activity", vote.ID, "community", communityIRI) + return nil + } + } + + return a.inTx(ctx, func(tx *sql.Tx) error { + // The aggregate row is the per-subject lock; no row means no vote was + // ever counted for this subject, so there is nothing to retract. + var locked int + err := tx.QueryRowContext(ctx, ` + SELECT 1 FROM vote_aggregates WHERE subject_ap_id = $1 FOR UPDATE`, + subject).Scan(&locked) + if stderrors.Is(err, sql.ErrNoRows) { + a.logger.Debug("vote retraction dropped: no aggregate for subject", + "subject", subject, "voter", voter) + return nil + } + if err != nil { + return fmt.Errorf("lock vote aggregate for %q: %w", subject, err) + } + + // Target the undone activity itself when its id is known (Lemmy + // inlines the original Like, so it usually is): a replayed + // Undo{Like id=A} after a re-like (id=B) must retract nothing, not B. + // The voter/subject/direction predicates stay as defense — a forged + // undo naming someone else's activity id retracts nothing. Id-less + // inline undos fall back to the voter's current live vote. + query := ` + UPDATE vote_events + SET undone = TRUE + WHERE subject_ap_id = $1 AND voter_ap_id = $2 AND direction = $3 AND NOT undone` + args := []any{subject, voter, direction} + if vote.ID != "" { + query += ` AND activity_id = $4` + args = append(args, vote.ID) + } + result, err := tx.ExecContext(ctx, query, args...) + if err != nil { + return fmt.Errorf("retract vote by %q on %q: %w", voter, subject, err) + } + retracted, err := result.RowsAffected() + if err != nil { + return fmt.Errorf("retract vote by %q on %q: rows affected: %w", voter, subject, err) + } + if retracted == 0 { + a.logger.Debug("vote retraction dropped: no matching live vote", + "subject", subject, "voter", voter, "direction", direction) + return nil + } + return recomputeAggregate(ctx, tx, subject) + }) +} + +// SeedAggregates imports a baseline (upvotes, downvotes) for a bridged +// subject from its origin's public API — history whose individual Like +// activities the bridge never saw (Lemmy outboxes announce historical votes +// only sparsely). Live vote_events stack on top of the baseline; re-seeding +// (backfill redo) overwrites the baseline idempotently. Known drift: a voter +// counted in the baseline who later flips sends Undo{Like} for a like the +// bridge never saw (no-op) plus a fresh Dislike — the retired upvote stays +// in the baseline. Accepted for v1; the baseline refreshes on re-seed. +// Subjects not present in ap_objects are dropped and logged at debug, like +// ApplyVote. +func (a *Aggregator) SeedAggregates(ctx context.Context, subjectAPID string, upvotes, downvotes int) error { + if subjectAPID == "" { + return errors.NewValidationError("subject_ap_id", "must not be empty") + } + if upvotes < 0 || downvotes < 0 { + return errors.NewValidationError("counts", "must not be negative") + } + + mapping, err := a.subjectMapping(ctx, subjectAPID) + if err != nil { + return err + } + if mapping == nil { + a.logger.Debug("vote seed dropped: subject not bridged", "subject", subjectAPID) + return nil + } + atURI := mapping.ATURI + + return a.inTx(ctx, func(tx *sql.Tx) error { + if _, err := tx.ExecContext(ctx, ` + INSERT INTO vote_aggregates ( + subject_ap_id, subject_at_uri, + seeded_upvotes, seeded_downvotes, upvotes, downvotes + ) VALUES ($1, $2, $3, $4, $3, $4) + ON CONFLICT (subject_ap_id) DO UPDATE SET + subject_at_uri = EXCLUDED.subject_at_uri, + seeded_upvotes = EXCLUDED.seeded_upvotes, + seeded_downvotes = EXCLUDED.seeded_downvotes`, + subjectAPID, atURI, upvotes, downvotes); err != nil { + return fmt.Errorf("seed vote aggregate for %q: %w", subjectAPID, err) + } + // Fold any live events that landed before the seed into the totals. + return recomputeAggregate(ctx, tx, subjectAPID) + }) +} + +// subjectMapping resolves a voted-on AP id to its materialized mapping. It +// returns nil (drop the vote) when the subject was never materialized or its +// mapping is soft-deleted — voting on deleted content stays a no-op. +func (a *Aggregator) subjectMapping(ctx context.Context, subject string) (*store.APObjectMapping, error) { + mapping, err := a.objects.GetByAPID(ctx, subject) + if errors.IsNotFound(err) { + return nil, nil + } + if err != nil { + return nil, fmt.Errorf("votes: resolve subject %s: %w", subject, err) + } + if mapping.IsDeleted() { + return nil, nil + } + return mapping, nil +} + +// subjectBelongsToCommunity reports whether a bridged subject's content +// belongs to the announcing community — the vote counterpart of ingest's +// announced-content and announced-delete authority checks. Without it, ONE +// malicious followed community could Announce Like/Dislike/Undo against ANY +// bridged subject and skew other communities' scores (fabricated voter +// strings need no mint). +// +// The binding is by community DID, not IRI authority: Lemmy hosts a post's +// AP object on the AUTHOR's instance, so a legitimate cross-instance-authored +// post would fail any SameAuthority(subject, announcer) check. Posts are +// written into the community's own repo (PLAN.md decision 3), so the +// mapping's DID IS the community DID. Comments live in the author's repo; +// their stored record's reply.root strongRef names the thread's root post in +// the community repo, so one record read recovers the community DID. +func (a *Aggregator) subjectBelongsToCommunity(ctx context.Context, mapping *store.APObjectMapping, communityIRI string) (bool, error) { + community, err := a.communities.GetByAPGroupID(ctx, communityIRI) + if errors.IsNotFound(err) { + return false, nil + } + if err != nil { + return false, fmt.Errorf("votes: resolve announcing community %s: %w", communityIRI, err) + } + switch mapping.Collection { + case materialize.CollectionPost: + return mapping.DID == community.DID, nil + case materialize.CollectionComment: + record, _, err := a.records.GetRecord(ctx, mapping.DID, mapping.Collection, mapping.RKey) + if errors.IsNotFound(err) { + return false, nil + } + if err != nil { + return false, fmt.Errorf("votes: read comment record %s: %w", mapping.ATURI, err) + } + rootDID := replyRootDID(record) + return rootDID != "" && rootDID == community.DID, nil + default: + // Votes bind to posts and comments only. + return false, nil + } +} + +// replyRootDID extracts the repo DID from a comment record's reply.root +// strongRef uri (at://did/collection/rkey). Malformed records yield "". +func replyRootDID(record map[string]any) string { + reply, ok := record["reply"].(map[string]any) + if !ok { + return "" + } + root, ok := reply["root"].(map[string]any) + if !ok { + return "" + } + uri, _ := root["uri"].(string) + rest, ok := strings.CutPrefix(uri, "at://") + if !ok { + return "" + } + did, _, _ := strings.Cut(rest, "/") + return did +} + +// inTx runs fn inside a transaction, committing on nil and rolling back on +// error. +func (a *Aggregator) inTx(ctx context.Context, fn func(tx *sql.Tx) error) error { + tx, err := a.db.BeginTx(ctx, nil) + if err != nil { + return fmt.Errorf("votes: begin tx: %w", err) + } + if err := fn(tx); err != nil { + _ = tx.Rollback() + return err + } + if err := tx.Commit(); err != nil { + return fmt.Errorf("votes: commit tx: %w", err) + } + return nil +} + +// lockAggregate upserts the subject's aggregate row, taking its row lock — +// the per-subject serialization point every vote mutation goes through +// (DO UPDATE always fires, so the lock is taken on the existing-row path +// too). Counts are recomputed later in the same transaction. +func lockAggregate(ctx context.Context, tx *sql.Tx, subject, atURI string) error { + if _, err := tx.ExecContext(ctx, ` + INSERT INTO vote_aggregates (subject_ap_id, subject_at_uri) + VALUES ($1, $2) + ON CONFLICT (subject_ap_id) DO UPDATE SET subject_at_uri = EXCLUDED.subject_at_uri`, + subject, atURI); err != nil { + return fmt.Errorf("lock vote aggregate for %q: %w", subject, err) + } + return nil +} + +// recomputeAggregate rewrites the subject's served totals from the seeded +// baseline plus the live (non-undone) events. Recompute-per-subject over +// incremental arithmetic: a Lemmy post sees at most a few thousand votes, +// and recomputing inside the locking transaction cannot drift. +func recomputeAggregate(ctx context.Context, tx *sql.Tx, subject string) error { + if _, err := tx.ExecContext(ctx, ` + UPDATE vote_aggregates a + SET upvotes = a.seeded_upvotes + live.up, + downvotes = a.seeded_downvotes + live.down, + updated_at = CURRENT_TIMESTAMP + FROM ( + SELECT + COUNT(*) FILTER (WHERE direction = 'up') AS up, + COUNT(*) FILTER (WHERE direction = 'down') AS down + FROM vote_events + WHERE subject_ap_id = $1 AND NOT undone + ) live + WHERE a.subject_ap_id = $1`, + subject); err != nil { + return fmt.Errorf("recompute vote aggregate for %q: %w", subject, err) + } + return nil +} + +// directionFor maps an AP activity type to a vote direction. +func directionFor(activityType string) (string, bool) { + switch activityType { + case ap.TypeLike: + return directionUp, true + case ap.TypeDislike: + return directionDown, true + } + return "", false +} + +// refID returns the id of a possibly-nil object reference. +func refID(obj *ap.Object) string { + if obj == nil { + return "" + } + return obj.ID +} diff --git a/internal/votes/aggregator_test.go b/internal/votes/aggregator_test.go new file mode 100644 index 0000000..5117c64 --- /dev/null +++ b/internal/votes/aggregator_test.go @@ -0,0 +1,425 @@ +package votes + +import ( + "context" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "tidepool/internal/ap" + "tidepool/internal/materialize" +) + +// Fixtures for announced-vote community binding. +const ( + testGroupIRI = "https://lemmy.world/c/golang" + + // A second community's repo DID and a comment author's user DID. + otherCommunityDID = "did:plc:yk4dd2qkboz2yv6tpubpc6co" + commentAuthorDID = "did:plc:44ybard66vv44zksje25o7dz" +) + +func TestApplyVoteCountsDistinctVoters(t *testing.T) { + database := testDB(t) + agg, objects := testAggregator(t, database) + bridgeSubject(t, objects, subjectPost, "3jzfcijpj2z2a") + ctx := context.Background() + + require.NoError(t, agg.ApplyVote(ctx, like(activityID(t, 1), voterAlice, subjectPost), "")) + require.NoError(t, agg.ApplyVote(ctx, like(activityID(t, 2), voterBob, subjectPost), "")) + require.NoError(t, agg.ApplyVote(ctx, dislike(activityID(t, 3), voterCarol, subjectPost), "")) + + up, down, found := counts(t, database, subjectPost) + require.True(t, found) + assert.Equal(t, 2, up) + assert.Equal(t, 1, down) +} + +func TestApplyVoteFlipCountsLatestState(t *testing.T) { + database := testDB(t) + agg, objects := testAggregator(t, database) + bridgeSubject(t, objects, subjectPost, "3jzfcijpj2z2a") + ctx := context.Background() + + // Alice upvotes, then flips to a downvote WITHOUT an Undo in between + // (implementations differ; the count must reflect her latest state). + require.NoError(t, agg.ApplyVote(ctx, like(activityID(t, 1), voterAlice, subjectPost), "")) + require.NoError(t, agg.ApplyVote(ctx, dislike(activityID(t, 2), voterAlice, subjectPost), "")) + + up, down, _ := counts(t, database, subjectPost) + assert.Equal(t, 0, up, "the flipped-away upvote must not linger") + assert.Equal(t, 1, down) + + // The Lemmy-style flip (Undo{Dislike} then Like) nets out too. + require.NoError(t, agg.RetractVote(ctx, dislike(activityID(t, 2), voterAlice, subjectPost), "")) + require.NoError(t, agg.ApplyVote(ctx, like(activityID(t, 3), voterAlice, subjectPost), "")) + + up, down, _ = counts(t, database, subjectPost) + assert.Equal(t, 1, up) + assert.Equal(t, 0, down) +} + +func TestRetractVoteDecrements(t *testing.T) { + database := testDB(t) + agg, objects := testAggregator(t, database) + bridgeSubject(t, objects, subjectPost, "3jzfcijpj2z2a") + ctx := context.Background() + + require.NoError(t, agg.ApplyVote(ctx, like(activityID(t, 1), voterAlice, subjectPost), "")) + require.NoError(t, agg.RetractVote(ctx, like(activityID(t, 1), voterAlice, subjectPost), "")) + + up, down, found := counts(t, database, subjectPost) + require.True(t, found) + assert.Equal(t, 0, up) + assert.Equal(t, 0, down) + + // Redelivered undo: still zero, still no error. + require.NoError(t, agg.RetractVote(ctx, like(activityID(t, 1), voterAlice, subjectPost), "")) + up, down, _ = counts(t, database, subjectPost) + assert.Equal(t, 0, up) + assert.Equal(t, 0, down) +} + +func TestDuplicateActivityIDIsNoOp(t *testing.T) { + database := testDB(t) + agg, objects := testAggregator(t, database) + bridgeSubject(t, objects, subjectPost, "3jzfcijpj2z2a") + ctx := context.Background() + + dup := like(activityID(t, 1), voterAlice, subjectPost) + require.NoError(t, agg.ApplyVote(ctx, dup, "")) + require.NoError(t, agg.ApplyVote(ctx, dup, "")) + + up, down, _ := counts(t, database, subjectPost) + assert.Equal(t, 1, up, "a re-delivered vote must not double-count") + assert.Equal(t, 0, down) + + // Even a forged reuse of the id (different voter) changes nothing. + require.NoError(t, agg.ApplyVote(ctx, like(activityID(t, 1), voterBob, subjectPost), "")) + up, _, _ = counts(t, database, subjectPost) + assert.Equal(t, 1, up) +} + +func TestUndoBeforeLikeIsNoOp(t *testing.T) { + database := testDB(t) + agg, objects := testAggregator(t, database) + bridgeSubject(t, objects, subjectPost, "3jzfcijpj2z2a") + ctx := context.Background() + + // The undo of a vote the bridge never saw must not error or go negative. + require.NoError(t, agg.RetractVote(ctx, like(activityID(t, 1), voterAlice, subjectPost), "")) + _, _, found := counts(t, database, subjectPost) + assert.False(t, found, "an unactionable undo must not create an aggregate") + + // The (out-of-order) like still lands afterwards. + require.NoError(t, agg.ApplyVote(ctx, like(activityID(t, 1), voterAlice, subjectPost), "")) + up, down, _ := counts(t, database, subjectPost) + assert.Equal(t, 1, up) + assert.Equal(t, 0, down) +} + +func TestReplayedUndoDoesNotRetractReLike(t *testing.T) { + database := testDB(t) + agg, objects := testAggregator(t, database) + bridgeSubject(t, objects, subjectPost, "3jzfcijpj2z2a") + ctx := context.Background() + + // Like(A), Undo(A), re-Like(B), then Undo(A) REPLAYED (a fresh Announce + // wrapper gets it past inbox dedupe): the undo must target activity A — + // long since undone — and never B, which merely shares + // (voter, subject, direction). + likeA := like(activityID(t, 1), voterAlice, subjectPost) + require.NoError(t, agg.ApplyVote(ctx, likeA, "")) + require.NoError(t, agg.RetractVote(ctx, likeA, "")) + require.NoError(t, agg.ApplyVote(ctx, like(activityID(t, 2), voterAlice, subjectPost), "")) + require.NoError(t, agg.RetractVote(ctx, likeA, "")) + + up, down, _ := counts(t, database, subjectPost) + assert.Equal(t, 1, up, "the replayed undo of A must not retract the newer like B") + assert.Equal(t, 0, down) +} + +func TestDuplicateActivityIDOnAnotherSubjectMintsNoAggregate(t *testing.T) { + database := testDB(t) + agg, objects := testAggregator(t, database) + bridgeSubject(t, objects, subjectPost, "3jzfcijpj2z2a") + bridgeSubject(t, objects, subjectComment, "3jzfcijpj2z3a") + ctx := context.Background() + + require.NoError(t, agg.ApplyVote(ctx, like(activityID(t, 1), voterAlice, subjectPost), "")) + // The same activity id aimed at a DIFFERENT bridged subject is a dedupe + // no-op that must not mint a spurious 0/0 aggregate row — the XRPC + // contract omits never-voted uris. + require.NoError(t, agg.ApplyVote(ctx, like(activityID(t, 1), voterAlice, subjectComment), "")) + + _, _, found := counts(t, database, subjectComment) + assert.False(t, found, "a deduped vote must not create an aggregate for its claimed subject") + up, down, _ := counts(t, database, subjectPost) + assert.Equal(t, 1, up) + assert.Equal(t, 0, down) +} + +func TestReLikeAfterUndoCounts(t *testing.T) { + database := testDB(t) + agg, objects := testAggregator(t, database) + bridgeSubject(t, objects, subjectPost, "3jzfcijpj2z2a") + ctx := context.Background() + + require.NoError(t, agg.ApplyVote(ctx, like(activityID(t, 1), voterAlice, subjectPost), "")) + require.NoError(t, agg.RetractVote(ctx, like(activityID(t, 1), voterAlice, subjectPost), "")) + require.NoError(t, agg.ApplyVote(ctx, like(activityID(t, 2), voterAlice, subjectPost), "")) + + up, down, _ := counts(t, database, subjectPost) + assert.Equal(t, 1, up) + assert.Equal(t, 0, down) +} + +func TestVoteOnUnbridgedSubjectDropped(t *testing.T) { + database := testDB(t) + agg, _ := testAggregator(t, database) + ctx := context.Background() + + // No ap_objects mapping exists: the vote is dropped, not retried. + require.NoError(t, agg.ApplyVote(ctx, + like(activityID(t, 1), voterAlice, "https://lemmy.world/post/999"), "")) + + _, _, found := counts(t, database, "https://lemmy.world/post/999") + assert.False(t, found) + var events int + require.NoError(t, database.QueryRow(`SELECT COUNT(*) FROM vote_events`).Scan(&events)) + assert.Zero(t, events) +} + +func TestVoteOnDeletedSubjectDropped(t *testing.T) { + database := testDB(t) + agg, objects := testAggregator(t, database) + bridgeSubject(t, objects, subjectPost, "3jzfcijpj2z2a") + ctx := context.Background() + require.NoError(t, objects.SoftDelete(ctx, subjectPost)) + + require.NoError(t, agg.ApplyVote(ctx, like(activityID(t, 1), voterAlice, subjectPost), "")) + _, _, found := counts(t, database, subjectPost) + assert.False(t, found, "votes on tombstoned content must not resurrect state") +} + +func TestVotesAreKeyedPerSubject(t *testing.T) { + database := testDB(t) + agg, objects := testAggregator(t, database) + bridgeSubject(t, objects, subjectPost, "3jzfcijpj2z2a") + bridgeSubject(t, objects, subjectComment, "3jzfcijpj2z3a") + ctx := context.Background() + + // The same voter holds independent votes on different subjects. + require.NoError(t, agg.ApplyVote(ctx, like(activityID(t, 1), voterAlice, subjectPost), "")) + require.NoError(t, agg.ApplyVote(ctx, dislike(activityID(t, 2), voterAlice, subjectComment), "")) + + up, down, _ := counts(t, database, subjectPost) + assert.Equal(t, 1, up) + assert.Equal(t, 0, down) + up, down, _ = counts(t, database, subjectComment) + assert.Equal(t, 0, up) + assert.Equal(t, 1, down) +} + +func TestMalformedVotesAreDroppedNotRetried(t *testing.T) { + database := testDB(t) + agg, objects := testAggregator(t, database) + bridgeSubject(t, objects, subjectPost, "3jzfcijpj2z2a") + ctx := context.Background() + + // The LOOP contract: nil/bare refs must be no-ops, never errors (an + // error would wedge the community's ordering key on retries). + require.NoError(t, agg.ApplyVote(ctx, nil, "")) + require.NoError(t, agg.ApplyVote(ctx, &ap.Object{Type: ap.TypeLike}, "")) + require.NoError(t, agg.ApplyVote(ctx, + &ap.Object{ID: activityID(t, 1), Type: ap.TypeLike, Object: &ap.Object{ID: subjectPost}}, "")) + require.NoError(t, agg.ApplyVote(ctx, // no activity id + &ap.Object{Type: ap.TypeLike, Actor: &ap.Object{ID: voterAlice}, Object: &ap.Object{ID: subjectPost}}, "")) + require.NoError(t, agg.ApplyVote(ctx, // not a vote type + vote(ap.TypeAnnounce, activityID(t, 2), voterAlice, subjectPost), "")) + + require.NoError(t, agg.RetractVote(ctx, nil, "")) + require.NoError(t, agg.RetractVote(ctx, &ap.Object{Type: ap.TypeLike}, "")) + require.NoError(t, agg.RetractVote(ctx, + &ap.Object{ID: activityID(t, 3), Type: ap.TypeLike, Actor: &ap.Object{ID: voterAlice}}, "")) + + var events int + require.NoError(t, database.QueryRow(`SELECT COUNT(*) FROM vote_events`).Scan(&events)) + assert.Zero(t, events) +} + +func TestSeedAggregates(t *testing.T) { + database := testDB(t) + agg, objects := testAggregator(t, database) + atURI := bridgeSubject(t, objects, subjectPost, "3jzfcijpj2z2a") + ctx := context.Background() + + require.NoError(t, agg.SeedAggregates(ctx, subjectPost, 40, 3)) + up, down, found := counts(t, database, subjectPost) + require.True(t, found) + assert.Equal(t, 40, up) + assert.Equal(t, 3, down) + + // Live votes stack on top of the baseline. + require.NoError(t, agg.ApplyVote(ctx, like(activityID(t, 1), voterAlice, subjectPost), "")) + up, down, _ = counts(t, database, subjectPost) + assert.Equal(t, 41, up) + assert.Equal(t, 3, down) + + // Re-seeding (backfill redo) replaces the baseline and keeps live votes. + require.NoError(t, agg.SeedAggregates(ctx, subjectPost, 50, 5)) + up, down, _ = counts(t, database, subjectPost) + assert.Equal(t, 51, up) + assert.Equal(t, 5, down) + + // The served at-uri is the mapping's. + var storedURI string + require.NoError(t, database.QueryRow(` + SELECT subject_at_uri FROM vote_aggregates WHERE subject_ap_id = $1`, + subjectPost).Scan(&storedURI)) + assert.Equal(t, atURI, storedURI) +} + +func TestSeedAggregatesBeforeLiveVotesFoldsThem(t *testing.T) { + database := testDB(t) + agg, objects := testAggregator(t, database) + bridgeSubject(t, objects, subjectPost, "3jzfcijpj2z2a") + ctx := context.Background() + + // A live vote lands BEFORE the seed (announce raced the backfill): the + // seed must fold it in, not clobber it. + require.NoError(t, agg.ApplyVote(ctx, like(activityID(t, 1), voterAlice, subjectPost), "")) + require.NoError(t, agg.SeedAggregates(ctx, subjectPost, 10, 1)) + + up, down, _ := counts(t, database, subjectPost) + assert.Equal(t, 11, up) + assert.Equal(t, 1, down) +} + +func TestSeedAggregatesUnbridgedSubjectDropped(t *testing.T) { + database := testDB(t) + agg, _ := testAggregator(t, database) + + require.NoError(t, agg.SeedAggregates(context.Background(), + "https://lemmy.world/post/999", 7, 2)) + _, _, found := counts(t, database, "https://lemmy.world/post/999") + assert.False(t, found) +} + +// --- Announced-vote community binding --- +// +// An announced vote (communityIRI != "") counts only when its subject +// belongs to the announcing community; otherwise one malicious followed +// community could skew any bridged subject's score bridge-wide. + +func TestAnnouncedVoteCrossCommunitySubjectDropped(t *testing.T) { + database := testDB(t) + agg, objects := testAggregator(t, database) + // The announcer's repo DID differs from the DID of the community repo + // the post lives in — a vote injected against another community's post. + followCommunity(t, database, testGroupIRI, otherCommunityDID) + bridgeSubject(t, objects, subjectPost, "3jzfcijpj2z2a") // repo DID: testDID + ctx := context.Background() + + require.NoError(t, agg.ApplyVote(ctx, like(activityID(t, 1), voterAlice, subjectPost), testGroupIRI)) + + _, _, found := counts(t, database, subjectPost) + assert.False(t, found, "a cross-community announced vote must not create an aggregate") + var events int + require.NoError(t, database.QueryRow(`SELECT COUNT(*) FROM vote_events`).Scan(&events)) + assert.Zero(t, events, "a cross-community announced vote must not record an event") +} + +func TestAnnouncedVoteUnknownCommunityDropped(t *testing.T) { + database := testDB(t) + agg, objects := testAggregator(t, database) + bridgeSubject(t, objects, subjectPost, "3jzfcijpj2z2a") + ctx := context.Background() + + // The announcer was never registered as a community: no binding is + // possible, so the vote is dropped (nil, never a retryable error). + require.NoError(t, agg.ApplyVote(ctx, + like(activityID(t, 1), voterAlice, subjectPost), "https://lemmy.world/c/unknown")) + _, _, found := counts(t, database, subjectPost) + assert.False(t, found) +} + +func TestAnnouncedVoteCrossInstanceAuthoredPostCounts(t *testing.T) { + database := testDB(t) + agg, objects := testAggregator(t, database) + followCommunity(t, database, testGroupIRI, testDID) + // Lemmy hosts a post's AP object on the AUTHOR's instance: a lemmy.zip + // user's post in a lemmy.world community carries a lemmy.zip AP id. The + // binding is by community repo DID, not IRI authority, so it must count. + remotePost := "https://lemmy.zip/post/4242" + bridgeSubjectAs(t, objects, remotePost, "3jzfcijpj2z4a", testDID, testCollection) + ctx := context.Background() + + require.NoError(t, agg.ApplyVote(ctx, like(activityID(t, 1), voterBob, remotePost), testGroupIRI)) + + up, down, found := counts(t, database, remotePost) + require.True(t, found, "a cross-instance-authored post in the announcing community must count") + assert.Equal(t, 1, up) + assert.Equal(t, 0, down) +} + +func TestAnnouncedCommentVoteBindsThreadRootCommunity(t *testing.T) { + database := testDB(t) + agg, objects, records := testAggregatorWithRecords(t, database) + followCommunity(t, database, testGroupIRI, testDID) + ctx := context.Background() + + // A comment lives in its AUTHOR's repo; its stored record's reply.root + // names the thread's root post in the community repo. Rooted in the + // announcing community: counts. + bridgeSubjectAs(t, objects, subjectComment, "3jzfcijpj2z3a", + commentAuthorDID, materialize.CollectionComment) + records.put(commentAuthorDID, materialize.CollectionComment, "3jzfcijpj2z3a", map[string]any{ + "reply": map[string]any{ + "root": map[string]any{ + "uri": "at://" + testDID + "/" + testCollection + "/3jzfcijpj2z2a", + "cid": testCID, + }, + }, + }) + require.NoError(t, agg.ApplyVote(ctx, like(activityID(t, 1), voterAlice, subjectComment), testGroupIRI)) + up, down, found := counts(t, database, subjectComment) + require.True(t, found) + assert.Equal(t, 1, up) + assert.Equal(t, 0, down) + + // Rooted in ANOTHER community's repo: dropped. + otherComment := "https://lemmy.world/comment/300" + bridgeSubjectAs(t, objects, otherComment, "3jzfcijpj2z5a", + commentAuthorDID, materialize.CollectionComment) + records.put(commentAuthorDID, materialize.CollectionComment, "3jzfcijpj2z5a", map[string]any{ + "reply": map[string]any{ + "root": map[string]any{ + "uri": "at://" + otherCommunityDID + "/" + testCollection + "/3jzfcijpj2z6a", + "cid": testCID, + }, + }, + }) + require.NoError(t, agg.ApplyVote(ctx, like(activityID(t, 2), voterAlice, otherComment), testGroupIRI)) + _, _, found = counts(t, database, otherComment) + assert.False(t, found, "a comment rooted in another community must not be countable via this announcer") +} + +func TestAnnouncedRetractCrossCommunityDropped(t *testing.T) { + database := testDB(t) + agg, objects := testAggregator(t, database) + followCommunity(t, database, testGroupIRI, otherCommunityDID) + bridgeSubject(t, objects, subjectPost, "3jzfcijpj2z2a") // repo DID: testDID + ctx := context.Background() + + // The vote lands bare (no announcer binding), then a DIFFERENT community + // announces the undo: it must not deflate the score. + likeA := like(activityID(t, 1), voterAlice, subjectPost) + require.NoError(t, agg.ApplyVote(ctx, likeA, "")) + require.NoError(t, agg.RetractVote(ctx, likeA, testGroupIRI)) + + up, down, _ := counts(t, database, subjectPost) + assert.Equal(t, 1, up, "a cross-community announced undo must not retract the vote") + assert.Equal(t, 0, down) +} diff --git a/internal/votes/e2e_test.go b/internal/votes/e2e_test.go new file mode 100644 index 0000000..7af5c47 --- /dev/null +++ b/internal/votes/e2e_test.go @@ -0,0 +1,203 @@ +package votes + +import ( + "context" + "encoding/json" + "net/http" + "net/url" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "tidepool/internal/ap" + "tidepool/internal/errors" + "tidepool/internal/ingest" + "tidepool/internal/materialize" + "tidepool/internal/store" +) + +// The aggregator is the real implementation behind task 06's seam. +var _ ingest.VoteAggregator = (*Aggregator)(nil) + +// The seeder is what backfill's optional hook expects. +var _ ingest.CountSeeder = (*LemmySeeder)(nil) + +const ( + e2eGroupID = "https://lemmy.world/c/technology" + e2eServiceID = "https://bridge.test/actor" +) + +// stubMaterializer fails loudly if the dispatcher ever routes a vote to the +// materializer — votes must stay bridge-side. +type stubMaterializer struct{ t *testing.T } + +func (s *stubMaterializer) MaterializePost(context.Context, *ap.Object) (*materialize.Result, error) { + s.t.Fatal("votes must never reach MaterializePost") + return nil, nil +} + +func (s *stubMaterializer) MaterializeComment(context.Context, *ap.Object) (*materialize.Result, error) { + s.t.Fatal("votes must never reach MaterializeComment") + return nil, nil +} + +func (s *stubMaterializer) HandleUpdate(context.Context, *ap.Object) (*materialize.Result, error) { + s.t.Fatal("votes must never reach HandleUpdate") + return nil, nil +} + +func (s *stubMaterializer) HandleDelete(context.Context, string) error { + s.t.Fatal("votes must never reach HandleDelete") + return nil +} + +func (s *stubMaterializer) RefreshActor(context.Context, *ap.Object) (*store.BridgedActor, error) { + s.t.Fatal("votes must never reach RefreshActor") + return nil, nil +} + +func (s *stubMaterializer) RefreshCommunity(context.Context, *ap.Object) (*store.Community, error) { + s.t.Fatal("votes must never reach RefreshCommunity") + return nil, nil +} + +func (s *stubMaterializer) EnsureCommunity(context.Context, *ap.Object) (*store.Community, error) { + s.t.Fatal("votes must never reach EnsureCommunity") + return nil, nil +} + +// stubFetcher fails loudly on any fetch: inline vote activities must be +// dispatched without touching the network. +type stubFetcher struct{ t *testing.T } + +func (s *stubFetcher) FetchObject(_ context.Context, iri string) (*ap.Object, error) { + s.t.Errorf("unexpected fetch of %s while dispatching votes", iri) + return nil, errors.NewNotFoundError("object", iri) +} + +// deliverVote runs one activity through Handler.Process the way the queue +// worker does (signature verification and dedupe happened at the inbox; the +// handler receives the verified payload + bound actor). +func deliverVote(t *testing.T, handler *ingest.Handler, signer string, activity map[string]any) { + t.Helper() + payload, err := json.Marshal(activity) + require.NoError(t, err) + require.NoError(t, handler.Process(context.Background(), &store.InboxEvent{ + ActivityID: activity["id"].(string), + Payload: payload, + ActorID: signer, + })) +} + +// announceVote wraps a Like/Dislike in FEP-1b12 group fan-out exactly as +// Lemmy emits it (inline inner activity). +func announceVote(announceID string, inner map[string]any) map[string]any { + return map[string]any{ + "id": announceID, + "type": "Announce", + "actor": e2eGroupID, + "to": []any{ap.PublicAudience}, + "audience": e2eGroupID, + "object": inner, + } +} + +func inlineVote(voteType, id, voter, subject string) map[string]any { + return map[string]any{ + "id": id, + "type": voteType, + "actor": voter, + "object": subject, + "audience": e2eGroupID, + } +} + +// TestFakeLemmyVoteE2E is the definition-of-done flow: Announce{Like} from +// three distinct voters + one Announce{Dislike} + one Announce{Undo{Like}}, +// dispatched through the real ingest handler, must serve {up: 2, down: 1} +// over the XRPC side channel. +func TestFakeLemmyVoteE2E(t *testing.T) { + database := testDB(t) + agg, objects := testAggregator(t, database) + postURI := bridgeSubject(t, objects, subjectPost, "3jzfcijpj2z2a") + ctx := context.Background() + + communities := store.NewCommunities(database) + _, err := communities.UpsertCommunity(ctx, store.Community{ + APGroupID: e2eGroupID, + DID: testDID, + PreferredUsername: "technology", + Instance: testInstance, + }) + require.NoError(t, err) + require.NoError(t, communities.SetFollowState(ctx, e2eGroupID, store.FollowStateAccepted)) + + handler, err := ingest.NewHandler(ingest.HandlerOptions{ + Materializer: &stubMaterializer{t: t}, + Fetcher: &stubFetcher{t: t}, + Objects: objects, + Communities: communities, + Tombstones: store.NewTombstones(database), + Votes: agg, + ServiceActorID: e2eServiceID, + }) + require.NoError(t, err) + + base := "https://lemmy.world/activities" + like1 := inlineVote("Like", base+"/like/1", voterAlice, subjectPost) + deliverVote(t, handler, e2eGroupID, announceVote(base+"/announce/1", like1)) + deliverVote(t, handler, e2eGroupID, announceVote(base+"/announce/2", + inlineVote("Like", base+"/like/2", voterBob, subjectPost))) + deliverVote(t, handler, e2eGroupID, announceVote(base+"/announce/3", + inlineVote("Like", base+"/like/3", voterCarol, subjectPost))) + deliverVote(t, handler, e2eGroupID, announceVote(base+"/announce/4", + inlineVote("Dislike", base+"/dislike/4", "https://lemmy.world/u/dan", subjectPost))) + // Alice takes her upvote back: Announce{Undo{Like}} with the original + // Like inlined, the shape Lemmy sends. + deliverVote(t, handler, e2eGroupID, announceVote(base+"/announce/5", map[string]any{ + "id": base + "/undo/5", + "type": "Undo", + "actor": voterAlice, + "object": like1, + })) + + // ... and the AppView reads the net result over the side channel. + router := newXRPCRouter(t, database, 1000, 1000) + rec := get(t, router, "uris="+url.QueryEscape(postURI)) + require.Equal(t, http.StatusOK, rec.Code, rec.Body.String()) + + var out xrpcResponse + require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &out)) + require.Len(t, out.Aggregates, 1) + assert.Equal(t, postURI, out.Aggregates[0].URI) + assert.Equal(t, 2, out.Aggregates[0].Upvotes) + assert.Equal(t, 1, out.Aggregates[0].Downvotes) +} + +// TestBareVoteDispatch: a Like delivered directly (not group-announced) +// rides Handler.Process's bare-vote arm with communityIRI "". +func TestBareVoteDispatch(t *testing.T) { + database := testDB(t) + agg, objects := testAggregator(t, database) + bridgeSubject(t, objects, subjectPost, "3jzfcijpj2z2a") + + handler, err := ingest.NewHandler(ingest.HandlerOptions{ + Materializer: &stubMaterializer{t: t}, + Fetcher: &stubFetcher{t: t}, + Objects: objects, + Communities: store.NewCommunities(database), + Tombstones: store.NewTombstones(database), + Votes: agg, + ServiceActorID: e2eServiceID, + }) + require.NoError(t, err) + + deliverVote(t, handler, voterAlice, + inlineVote("Like", "https://lemmy.world/activities/like/bare-1", voterAlice, subjectPost)) + + up, down, found := counts(t, database, subjectPost) + require.True(t, found) + assert.Equal(t, 1, up) + assert.Equal(t, 0, down) +} diff --git a/internal/votes/ratelimit.go b/internal/votes/ratelimit.go new file mode 100644 index 0000000..93b4868 --- /dev/null +++ b/internal/votes/ratelimit.go @@ -0,0 +1,103 @@ +package votes + +import ( + "sync" + "time" + + "golang.org/x/time/rate" +) + +const ( + // limiterIdleTTL is how long an idle per-IP bucket survives before a + // sweep may reclaim it (a reclaimed bucket restarts full, which only + // ever favors the client). + limiterIdleTTL = 10 * time.Minute + // limiterSweepInterval throttles sweeps to at most one full-map scan + // per interval, however hot the endpoint runs — without it a flood of + // distinct IPs would provoke an O(n) scan under the mutex on every + // request, turning the limiter itself into the DoS vector. + limiterSweepInterval = time.Minute + // limiterSweepThreshold is the map size below which sweeps don't + // bother running: a few thousand lingering buckets cost less than the + // scans that would reclaim them. + limiterSweepThreshold = 10_000 + // limiterMaxBuckets hard-caps the bucket map. At the cap, requests + // from IPs without a bucket are refused (fail closed): the cap is only + // reachable during an address-rotation flood (one IPv6 /64 is + // effectively unlimited addresses, none idle long enough for the TTL + // sweep), refusing unknown IPs leaves established clients — the + // AppView's long-lived poller — untouched, and admitting them instead + // would hand every rotated address a fresh full burst, which is + // exactly the abuse the limiter exists to stop. Idle buckets age out + // via the TTL sweep, so a genuinely new client is locked out only + // while the flood lasts. + limiterMaxBuckets = 50_000 +) + +// ipLimiter is a per-client-IP token bucket set (the same golang.org/x/time +// machinery as ingest's mint gate, keyed by IP). Stale buckets are swept +// inline — at most once per sweepInterval — and the map is hard-capped, so +// an address-rotating scraper can balloon neither memory nor scan time. +type ipLimiter struct { + perSecond rate.Limit + burst int + + // Tunables default to the package constants; they are fields (like + // now) so tests can shrink them. + idleTTL time.Duration + sweepInterval time.Duration + sweepThreshold int + maxBuckets int + now func() time.Time + + mu sync.Mutex + lastSweep time.Time + buckets map[string]*ipBucket +} + +type ipBucket struct { + limiter *rate.Limiter + lastSeen time.Time +} + +func newIPLimiter(perSecond float64, burst int) *ipLimiter { + return &ipLimiter{ + perSecond: rate.Limit(perSecond), + burst: burst, + idleTTL: limiterIdleTTL, + sweepInterval: limiterSweepInterval, + sweepThreshold: limiterSweepThreshold, + maxBuckets: limiterMaxBuckets, + now: time.Now, + buckets: map[string]*ipBucket{}, + } +} + +// allow reports whether one request from ip fits in its bucket. Unknown IPs +// are refused outright while the bucket map sits at maxBuckets (see the +// constant for why fail-closed is the right shape there). +func (l *ipLimiter) allow(ip string) bool { + l.mu.Lock() + defer l.mu.Unlock() + now := l.now() + + if len(l.buckets) >= l.sweepThreshold && now.Sub(l.lastSweep) >= l.sweepInterval { + l.lastSweep = now + for key, bucket := range l.buckets { + if now.Sub(bucket.lastSeen) > l.idleTTL { + delete(l.buckets, key) + } + } + } + + bucket, ok := l.buckets[ip] + if !ok { + if len(l.buckets) >= l.maxBuckets { + return false + } + bucket = &ipBucket{limiter: rate.NewLimiter(l.perSecond, l.burst)} + l.buckets[ip] = bucket + } + bucket.lastSeen = now + return bucket.limiter.AllowN(now, 1) +} diff --git a/internal/votes/ratelimit_test.go b/internal/votes/ratelimit_test.go new file mode 100644 index 0000000..d9f3e47 --- /dev/null +++ b/internal/votes/ratelimit_test.go @@ -0,0 +1,98 @@ +package votes + +import ( + "fmt" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// testClock is a hand-cranked clock for driving the limiter's TTL, sweep, +// and refill paths deterministically. +type testClock struct{ t time.Time } + +func newTestClock() *testClock { + return &testClock{t: time.Date(2026, 7, 8, 12, 0, 0, 0, time.UTC)} +} + +func (c *testClock) now() time.Time { return c.t } +func (c *testClock) advance(d time.Duration) { c.t = c.t.Add(d) } + +func TestIPLimiterIndependentBuckets(t *testing.T) { + limiter := newIPLimiter(0.001, 1) + + require.True(t, limiter.allow("192.0.2.1")) + require.False(t, limiter.allow("192.0.2.1"), "the first IP's burst is spent") + assert.True(t, limiter.allow("192.0.2.2"), "a second IP must have its own bucket") +} + +func TestIPLimiterHardCap(t *testing.T) { + limiter := newIPLimiter(0.001, 1) + limiter.maxBuckets = 3 + + // Many distinct IPs: the map must never grow past the cap. + for i := 0; i < 50; i++ { + limiter.allow(fmt.Sprintf("2001:db8::%x", i)) + require.LessOrEqual(t, len(limiter.buckets), 3, "after IP %d", i) + } + assert.Len(t, limiter.buckets, 3) + + // At the cap: unknown IPs are refused, established ones still tracked. + assert.False(t, limiter.allow("198.51.100.9"), "unknown IP at cap is refused") + assert.False(t, limiter.allow("2001:db8::0"), + "a known IP still hits its own (spent) bucket, not the cap") +} + +func TestIPLimiterCapRecoversAfterSweep(t *testing.T) { + clock := newTestClock() + limiter := newIPLimiter(0.001, 1) + limiter.now = clock.now + limiter.maxBuckets = 2 + limiter.sweepThreshold = 1 + + require.True(t, limiter.allow("192.0.2.1")) + require.True(t, limiter.allow("192.0.2.2")) + require.False(t, limiter.allow("192.0.2.3"), "the map is at its cap") + + // Once the residents idle past the TTL and the sweep throttle window + // passes, the next request sweeps them out and the new IP fits. + clock.advance(limiterIdleTTL + limiterSweepInterval) + assert.True(t, limiter.allow("192.0.2.3")) + assert.Len(t, limiter.buckets, 1) +} + +func TestIPLimiterSweepThrottled(t *testing.T) { + clock := newTestClock() + limiter := newIPLimiter(0.001, 1) + limiter.now = clock.now + limiter.sweepThreshold = 1 + limiter.idleTTL = time.Second + + limiter.allow("192.0.2.1") + limiter.allow("192.0.2.2") // crosses the threshold: sweeps, stamps lastSweep + + // Both buckets idle past the (tiny) TTL, but the throttle window has + // not elapsed — the next allow must not sweep them. + clock.advance(2 * time.Second) + limiter.allow("192.0.2.3") + assert.Len(t, limiter.buckets, 3, "no sweep inside the throttle window") + + // Past the window, the sweep runs and reclaims every idle bucket. + clock.advance(limiterSweepInterval) + limiter.allow("192.0.2.4") + assert.Len(t, limiter.buckets, 1, "idle buckets reclaimed once the window passes") +} + +func TestIPLimiterRefillFollowsClock(t *testing.T) { + clock := newTestClock() + limiter := newIPLimiter(1, 1) + limiter.now = clock.now + + require.True(t, limiter.allow("192.0.2.1")) + require.False(t, limiter.allow("192.0.2.1"), "burst spent, no time has passed") + + clock.advance(1500 * time.Millisecond) + assert.True(t, limiter.allow("192.0.2.1"), "one token refills after a second") +} diff --git a/internal/votes/seed.go b/internal/votes/seed.go new file mode 100644 index 0000000..e2567ea --- /dev/null +++ b/internal/votes/seed.go @@ -0,0 +1,147 @@ +package votes + +import ( + "context" + "encoding/json" + "fmt" + "io" + "log/slog" + "net/http" + "net/url" + "strings" + + "tidepool/internal/errors" +) + +// maxSeedResponseBytes caps a Lemmy API counts response — a full post_view +// is a few KB; anything near the cap is not the endpoint we think it is. +const maxSeedResponseBytes = 1 << 20 // 1 MiB + +// SeedStore is the slice of *Aggregator the seeder needs (tests inject +// recorders). +type SeedStore interface { + SeedAggregates(ctx context.Context, subjectAPID string, upvotes, downvotes int) error +} + +// LemmySeeder imports historical vote counts for backfilled posts from +// Lemmy's public HTTP API (GET /api/v3/post?id=N, the `counts` field) — +// the seeding hook task 06's backfill calls, gated behind +// SEED_COUNTS_FROM_API. AP alone cannot provide this: Lemmy outboxes +// announce historical Likes only sparsely, so without seeding backfilled +// posts start near zero. Comment counts are deliberately not seeded in v1 +// (one API call per comment would triple backfill egress for garnish); +// comments accumulate live votes only. +type LemmySeeder struct { + store SeedStore + client *http.Client + userAgent string + logger *slog.Logger +} + +// NewLemmySeeder builds a seeder. client must be an SSRF-guarded HTTP +// client (ap.NewGuardedHTTPClient) — the API URL is derived from a remote +// object's self-asserted AP id. +func NewLemmySeeder(store SeedStore, client *http.Client, userAgent string, logger *slog.Logger) (*LemmySeeder, error) { + if store == nil { + return nil, errors.NewValidationError("store", "must not be nil") + } + if client == nil { + return nil, errors.NewValidationError("client", "must not be nil") + } + if logger == nil { + logger = slog.Default() + } + return &LemmySeeder{store: store, client: client, userAgent: userAgent, logger: logger}, nil +} + +// lemmyCountsResponse is the slice of Lemmy's GET /api/v3/post response the +// seeder reads. Every level is a pointer so ABSENCE is detectable: a 200 +// whose JSON lacks the expected nesting (a proxy error page served as JSON, +// renamed fields in a future Lemmy API, the wrong endpoint) must fail +// loudly, not decode cleanly to zeros — re-seeding REPLACES the baseline, so +// a silent 0/0 would overwrite a previously good one. +type lemmyCountsResponse struct { + PostView *struct { + Counts *struct { + Upvotes *int `json:"upvotes"` + Downvotes *int `json:"downvotes"` + } `json:"counts"` + } `json:"post_view"` +} + +// SeedPostCounts fetches the post's current score from its origin instance +// and stores it as the subject's seeded baseline. Non-Lemmy-shaped AP ids +// are a silent no-op (only Lemmy's URL scheme is recognized in v1); fetch +// and decode failures return an error the caller logs — seeding is +// best-effort garnish on backfill, never fatal. +func (s *LemmySeeder) SeedPostCounts(ctx context.Context, postAPID string) error { + apiURL, ok := lemmyPostAPIURL(postAPID) + if !ok { + return nil + } + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, apiURL, nil) + if err != nil { + return fmt.Errorf("votes: build counts request for %s: %w", postAPID, err) + } + req.Header.Set("Accept", "application/json") + if s.userAgent != "" { + req.Header.Set("User-Agent", s.userAgent) + } + + resp, err := s.client.Do(req) + if err != nil { + return fmt.Errorf("votes: fetch counts for %s: %w", postAPID, err) + } + defer func() { _ = resp.Body.Close() }() + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("votes: fetch counts for %s: unexpected status %d", postAPID, resp.StatusCode) + } + + var payload lemmyCountsResponse + if err := json.NewDecoder(io.LimitReader(resp.Body, maxSeedResponseBytes)).Decode(&payload); err != nil { + return fmt.Errorf("votes: decode counts for %s: %w", postAPID, err) + } + if payload.PostView == nil || payload.PostView.Counts == nil || + payload.PostView.Counts.Upvotes == nil || payload.PostView.Counts.Downvotes == nil { + return fmt.Errorf("votes: counts response for %s lacks post_view.counts.{upvotes,downvotes}", postAPID) + } + upvotes, downvotes := *payload.PostView.Counts.Upvotes, *payload.PostView.Counts.Downvotes + if upvotes < 0 || downvotes < 0 { + return fmt.Errorf("votes: counts for %s are negative (up=%d down=%d)", + postAPID, upvotes, downvotes) + } + + if err := s.store.SeedAggregates(ctx, postAPID, upvotes, downvotes); err != nil { + return err + } + s.logger.Debug("vote counts seeded from origin API", + "post", postAPID, "upvotes", upvotes, "downvotes", downvotes) + return nil +} + +// lemmyPostAPIURL derives the public API counts endpoint from a Lemmy post +// AP id: https://host/post/123 → https://host/api/v3/post?id=123. Anything +// else (comments, non-Lemmy URL shapes) reports ok=false. +func lemmyPostAPIURL(postAPID string) (string, bool) { + parsed, err := url.Parse(postAPID) + if err != nil || parsed.Host == "" || (parsed.Scheme != "http" && parsed.Scheme != "https") { + return "", false + } + postID, ok := strings.CutPrefix(parsed.Path, "/post/") + if !ok || postID == "" || strings.ContainsRune(postID, '/') { + return "", false + } + for _, r := range postID { + if r < '0' || r > '9' { + return "", false + } + } + api := url.URL{ + Scheme: parsed.Scheme, + Host: parsed.Host, + Path: "/api/v3/post", + RawQuery: "id=" + postID, + } + return api.String(), true +} diff --git a/internal/votes/seed_test.go b/internal/votes/seed_test.go new file mode 100644 index 0000000..4595411 --- /dev/null +++ b/internal/votes/seed_test.go @@ -0,0 +1,148 @@ +package votes + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "tidepool/internal/ap" +) + +func TestLemmyPostAPIURL(t *testing.T) { + tests := []struct { + name string + apID string + want string + wantOK bool + }{ + {"lemmy post", "https://lemmy.world/post/49131386", + "https://lemmy.world/api/v3/post?id=49131386", true}, + {"http with port (tests)", "http://127.0.0.1:8080/post/7", + "http://127.0.0.1:8080/api/v3/post?id=7", true}, + {"comment", "https://lemmy.world/comment/123", "", false}, + {"non-numeric id", "https://lemmy.world/post/abc", "", false}, + {"trailing path", "https://lemmy.world/post/1/extra", "", false}, + {"empty id", "https://lemmy.world/post/", "", false}, + {"not a url", "::::", "", false}, + {"wrong scheme", "ftp://lemmy.world/post/1", "", false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, ok := lemmyPostAPIURL(tt.apID) + assert.Equal(t, tt.wantOK, ok) + assert.Equal(t, tt.want, got) + }) + } +} + +func TestSeedPostCountsFromFakeLemmyAPI(t *testing.T) { + database := testDB(t) + agg, objects := testAggregator(t, database) + + // The fake Lemmy public API. Tests always talk to loopback httptest + // servers, never real instances; the guarded client needs + // allowPrivate=true for that (same rule as every other AP test). + mux := http.NewServeMux() + mux.HandleFunc("GET /api/v3/post", func(w http.ResponseWriter, r *http.Request) { + require.Equal(t, "321", r.URL.Query().Get("id")) + w.Header().Set("Content-Type", "application/json") + _, _ = fmt.Fprint(w, `{"post_view":{"counts":{"upvotes":128,"downvotes":9,"score":119}}}`) + }) + server := httptest.NewServer(mux) + t.Cleanup(server.Close) + + postID := server.URL + "/post/321" + bridgeSubject(t, objects, postID, "3jzfcijpj2z2a") + + seeder, err := NewLemmySeeder(agg, + ap.NewGuardedHTTPClient(true, 5*time.Second), "tidepool-test/0", nil) + require.NoError(t, err) + require.NoError(t, seeder.SeedPostCounts(context.Background(), postID)) + + up, down, found := counts(t, database, postID) + require.True(t, found) + assert.Equal(t, 128, up) + assert.Equal(t, 9, down) +} + +func TestSeedPostCountsNonLemmyShapeIsNoOp(t *testing.T) { + database := testDB(t) + agg, _ := testAggregator(t, database) + seeder, err := NewLemmySeeder(agg, ap.NewGuardedHTTPClient(true, time.Second), "", nil) + require.NoError(t, err) + + // Comments (and anything else that is not /post/N) are not seeded in v1; + // no fetch happens and no error is returned. + require.NoError(t, seeder.SeedPostCounts(context.Background(), + "https://lemmy.world/comment/555")) + _, _, found := counts(t, database, "https://lemmy.world/comment/555") + assert.False(t, found) +} + +// TestSeedPostCountsMalformedResponseKeepsBaseline: a 200 whose body lacks +// the expected post_view.counts nesting (a proxy error page served as JSON, +// renamed fields, the wrong endpoint) or is truncated must return an error +// and must NOT seed — re-seeding replaces the baseline, so decoding absent +// fields to zero would silently overwrite a previously good baseline with +// 0/0 on a backfill redo. +func TestSeedPostCountsMalformedResponseKeepsBaseline(t *testing.T) { + database := testDB(t) + agg, objects := testAggregator(t, database) + + tests := []struct { + name string + body string + }{ + {"empty object", `{}`}, + {"missing counts", `{"post_view":{"creator_banned":false}}`}, + {"missing count fields", `{"post_view":{"counts":{"score":119}}}`}, + {"error page as json", `{"error":"couldnt_find_post"}`}, + {"truncated json", `{"post_view":{"counts":{"upvotes":128,`}, + {"not json at all", `502 Bad Gateway`}, + } + for i, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + _, _ = fmt.Fprint(w, tt.body) + })) + t.Cleanup(server.Close) + + postID := fmt.Sprintf("%s/post/%d", server.URL, i+1) + bridgeSubject(t, objects, postID, fmt.Sprintf("3jzfcijpj2z%da", i)) + // A good baseline from an earlier run: the broken response must + // leave it untouched. + require.NoError(t, agg.SeedAggregates(context.Background(), postID, 42, 7)) + + seeder, err := NewLemmySeeder(agg, + ap.NewGuardedHTTPClient(true, 5*time.Second), "tidepool-test/0", nil) + require.NoError(t, err) + assert.Error(t, seeder.SeedPostCounts(context.Background(), postID)) + + up, down, found := counts(t, database, postID) + require.True(t, found) + assert.Equal(t, 42, up, "a malformed response must not overwrite the baseline") + assert.Equal(t, 7, down, "a malformed response must not overwrite the baseline") + }) + } +} + +func TestSeedPostCountsAPIFailureReturnsError(t *testing.T) { + database := testDB(t) + agg, _ := testAggregator(t, database) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "boom", http.StatusInternalServerError) + })) + t.Cleanup(server.Close) + + seeder, err := NewLemmySeeder(agg, ap.NewGuardedHTTPClient(true, time.Second), "", nil) + require.NoError(t, err) + // The caller (backfill) logs and moves on; the seeder just reports it. + assert.Error(t, seeder.SeedPostCounts(context.Background(), server.URL+"/post/1")) +} diff --git a/internal/votes/votes_test.go b/internal/votes/votes_test.go new file mode 100644 index 0000000..6af017c --- /dev/null +++ b/internal/votes/votes_test.go @@ -0,0 +1,161 @@ +package votes + +import ( + "context" + "database/sql" + stderrors "errors" + "fmt" + "log/slog" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "tidepool/internal/ap" + "tidepool/internal/errors" + "tidepool/internal/store" + "tidepool/internal/testutil" +) + +// The votes tests run against a real postgres database (Coves convention: +// real infrastructure, no mocks). They skip cleanly when +// TIDEPOOL_TEST_DATABASE_URL is unset; `make test` starts the postgres-test +// container and sets it. + +// Shared fixtures. +const ( + testDID = "did:plc:ewvi7nxzyoun6zhxrhs64oiz" + testCID = "bafyreib2rxk3rybk3aobmv5cjuql3bm2twh4jo5uxgf5kpqrsqxi3jgxte" + testCollection = "social.coves.community.post" + testInstance = "lemmy.world" + + subjectPost = "https://lemmy.world/post/100" + subjectComment = "https://lemmy.world/comment/200" + + voterAlice = "https://lemmy.world/u/alice" + voterBob = "https://lemmy.zip/u/bob" + voterCarol = "https://sh.itjust.works/u/carol" +) + +// testDB returns a migrated connection with the vote tables (and the +// ap_objects spine the aggregator resolves subjects through) truncated. +func testDB(t *testing.T) *sql.DB { + t.Helper() + database := testutil.DB(t) + testutil.Truncate(t, database, + "vote_events", "vote_aggregates", "ap_objects", "communities") + return database +} + +// testAggregator builds an Aggregator over the test database. +func testAggregator(t *testing.T, database *sql.DB) (*Aggregator, store.APObjects) { + t.Helper() + agg, objects, _ := testAggregatorWithRecords(t, database) + return agg, objects +} + +// testAggregatorWithRecords additionally exposes the fake record reader the +// aggregator resolves comment thread roots through (the real repo layer is +// exercised in internal/repo and internal/materialize; here the record read +// is one narrow seam). +func testAggregatorWithRecords(t *testing.T, database *sql.DB) (*Aggregator, store.APObjects, *fakeRecords) { + t.Helper() + objects := store.NewAPObjects(database) + records := &fakeRecords{records: map[string]map[string]any{}} + agg, err := NewAggregator(database, objects, store.NewCommunities(database), records, slog.Default()) + require.NoError(t, err) + return agg, objects, records +} + +// fakeRecords is a map-backed RecordReader. +type fakeRecords struct{ records map[string]map[string]any } + +func (f *fakeRecords) put(did, collection, rkey string, record map[string]any) { + f.records[did+"/"+collection+"/"+rkey] = record +} + +func (f *fakeRecords) GetRecord(_ context.Context, did, collection, rkey string) (map[string]any, string, error) { + record, ok := f.records[did+"/"+collection+"/"+rkey] + if !ok { + return nil, "", errors.NewNotFoundError("record", did+"/"+collection+"/"+rkey) + } + return record, testCID, nil +} + +// followCommunity registers an announcing community (group IRI → repo DID) +// for announced-vote binding tests. +func followCommunity(t *testing.T, database *sql.DB, groupIRI, did string) { + t.Helper() + communities := store.NewCommunities(database) + _, err := communities.UpsertCommunity(context.Background(), store.Community{ + APGroupID: groupIRI, + DID: did, + PreferredUsername: "votes-test", + Instance: testInstance, + }) + require.NoError(t, err) +} + +// bridgeSubject materializes a fake mapping for an AP id (the subject must +// exist in ap_objects for votes to count) and returns its at-uri. +func bridgeSubject(t *testing.T, objects store.APObjects, apID, rkey string) string { + t.Helper() + return bridgeSubjectAs(t, objects, apID, rkey, testDID, testCollection) +} + +// bridgeSubjectAs is bridgeSubject with an explicit repo DID and collection +// (announced-vote binding tests place subjects in specific community repos). +func bridgeSubjectAs(t *testing.T, objects store.APObjects, apID, rkey, did, collection string) string { + t.Helper() + publishedAt := time.Date(2026, 7, 1, 12, 0, 0, 0, time.UTC) + mapping, err := objects.PutMapping(context.Background(), store.APObjectMapping{ + APID: apID, + APType: "Page", + OriginInstance: testInstance, + DID: did, + Collection: collection, + RKey: rkey, + CID: testCID, + PublishedAt: &publishedAt, + }) + require.NoError(t, err) + return mapping.ATURI +} + +// like/dislike build vote activities the way the dispatcher hands them over +// (Announce-unwrapped: Type, ID, Actor, Object). +func like(activityID, voter, subject string) *ap.Object { + return vote(ap.TypeLike, activityID, voter, subject) +} + +func dislike(activityID, voter, subject string) *ap.Object { + return vote(ap.TypeDislike, activityID, voter, subject) +} + +func vote(voteType, activityID, voter, subject string) *ap.Object { + return &ap.Object{ + ID: activityID, + Type: voteType, + Actor: &ap.Object{ID: voter}, + Object: &ap.Object{ID: subject}, + } +} + +// activityID mints per-test activity ids. +func activityID(t *testing.T, n int) string { + return fmt.Sprintf("https://lemmy.world/activities/like/%s-%d", t.Name(), n) +} + +// counts reads the served aggregate for a subject straight from the table. +// found=false means no aggregate row exists at all. +func counts(t *testing.T, database *sql.DB, subject string) (up, down int, found bool) { + t.Helper() + err := database.QueryRow(` + SELECT upvotes, downvotes FROM vote_aggregates WHERE subject_ap_id = $1`, + subject).Scan(&up, &down) + if stderrors.Is(err, sql.ErrNoRows) { + return 0, 0, false + } + require.NoError(t, err) + return up, down, true +} diff --git a/internal/votes/xrpc.go b/internal/votes/xrpc.go new file mode 100644 index 0000000..004f76e --- /dev/null +++ b/internal/votes/xrpc.go @@ -0,0 +1,221 @@ +package votes + +import ( + "database/sql" + "encoding/json" + "fmt" + "log/slog" + "net" + "net/http" + "strings" + "time" + + "github.com/bluesky-social/indigo/atproto/syntax" + "github.com/go-chi/chi/v5" + "github.com/lib/pq" + + "tidepool/internal/errors" +) + +// NSIDGetVoteAggregates is the sanctioned side-channel query +// (lexicons/social/coves/bridge/getVoteAggregates.json). The nsid is the +// version: breaking changes mean a new name, never a changed shape. +const NSIDGetVoteAggregates = "social.coves.bridge.getVoteAggregates" + +const ( + // maxURIsPerQuery caps one getVoteAggregates call (mirrors the lexicon's + // maxLength). + maxURIsPerQuery = 100 + // cacheMaxAge is the Cache-Control freshness window. Vote counts are + // eventually consistent by nature; 30s keeps AppView polling cheap + // without making scores feel stale. + cacheMaxAge = 30 * time.Second + + // Rate-limit defaults (per client IP): the endpoint is a public read the + // AppView polls in batches, so sustained per-IP throughput can stay low + // while the burst absorbs a page-load fan-out. + defaultRatePerSecond = 10 + defaultRateBurst = 30 +) + +// XRPCOptions configures NewXRPC. DB is required. +type XRPCOptions struct { + DB *sql.DB + Logger *slog.Logger + // RatePerSecond / RateBurst tune the per-IP token bucket; zero values + // take the defaults. Tests shrink them. + RatePerSecond float64 + RateBurst int +} + +// XRPC serves the vote-aggregate side channel: the one sanctioned read the +// AppView uses to display scores for bridged content (PLAN.md decision 7). +// Public, cacheable, rate limited by client IP. +type XRPC struct { + db *sql.DB + limiter *ipLimiter + logger *slog.Logger +} + +// NewXRPC validates options and builds the handler. +func NewXRPC(opts XRPCOptions) (*XRPC, error) { + if opts.DB == nil { + return nil, errors.NewValidationError("db", "must not be nil") + } + logger := opts.Logger + if logger == nil { + logger = slog.Default() + } + perSecond := opts.RatePerSecond + if perSecond <= 0 { + perSecond = defaultRatePerSecond + } + burst := opts.RateBurst + if burst <= 0 { + burst = defaultRateBurst + } + return &XRPC{ + db: opts.DB, + limiter: newIPLimiter(perSecond, burst), + logger: logger, + }, nil +} + +// Routes mounts the side channel on a chi router. +func (x *XRPC) Routes(r chi.Router) { + r.Get("/xrpc/"+NSIDGetVoteAggregates, x.handleGetVoteAggregates) +} + +// aggregateView is one output entry ({uri, upvotes, downvotes, updatedAt}). +type aggregateView struct { + URI string `json:"uri"` + Upvotes int64 `json:"upvotes"` + Downvotes int64 `json:"downvotes"` + UpdatedAt string `json:"updatedAt"` +} + +// handleGetVoteAggregates serves GET /xrpc/social.coves.bridge. +// getVoteAggregates?uris=at://…&uris=at://… (repeated params, the atproto +// convention; comma-separated values inside one param are accepted too). +// Well-formed but unknown uris are omitted from the response, never an +// error — the AppView batches optimistically over content that may predate +// vote ingestion. Malformed at-uris are InvalidRequest. +func (x *XRPC) handleGetVoteAggregates(w http.ResponseWriter, r *http.Request) { + if !x.limiter.allow(clientIP(r)) { + x.writeXRPCError(w, http.StatusTooManyRequests, "RateLimitExceeded", "rate limit exceeded") + return + } + + uris, err := parseURIs(r.URL.Query()["uris"]) + if err != nil { + x.writeXRPCError(w, http.StatusBadRequest, "InvalidRequest", err.Error()) + return + } + if len(uris) == 0 { + x.writeXRPCError(w, http.StatusBadRequest, "InvalidRequest", "missing required parameter: uris") + return + } + + rows, err := x.db.QueryContext(r.Context(), ` + SELECT subject_at_uri, upvotes, downvotes, updated_at + FROM vote_aggregates + WHERE subject_at_uri = ANY($1)`, + pq.Array(uris)) + if err != nil { + x.logger.Error("votes: query aggregates", "error", err) + x.writeXRPCError(w, http.StatusInternalServerError, "InternalServerError", "internal error") + return + } + defer func() { _ = rows.Close() }() + + byURI := make(map[string]aggregateView, len(uris)) + for rows.Next() { + var view aggregateView + var updatedAt time.Time + if err := rows.Scan(&view.URI, &view.Upvotes, &view.Downvotes, &updatedAt); err != nil { + x.logger.Error("votes: scan aggregate", "error", err) + x.writeXRPCError(w, http.StatusInternalServerError, "InternalServerError", "internal error") + return + } + // Millisecond-precision UTC, the datetime shape Coves records use. + view.UpdatedAt = updatedAt.UTC().Format("2006-01-02T15:04:05.000Z") + byURI[view.URI] = view + } + if err := rows.Err(); err != nil { + x.logger.Error("votes: iterate aggregates", "error", err) + x.writeXRPCError(w, http.StatusInternalServerError, "InternalServerError", "internal error") + return + } + + // Respond in request order; unknown uris are simply absent. + aggregates := make([]aggregateView, 0, len(byURI)) + for _, uri := range uris { + if view, ok := byURI[uri]; ok { + aggregates = append(aggregates, view) + } + } + + w.Header().Set("Cache-Control", fmt.Sprintf("public, max-age=%d", int(cacheMaxAge.Seconds()))) + x.writeJSON(w, http.StatusOK, map[string]any{"aggregates": aggregates}) +} + +// parseURIs flattens repeated uris params and comma-separated lists into a +// deduplicated, order-preserving slice of syntactically valid at-uris. A +// malformed at-uri is an error (the lexicon declares format at-uri; only +// well-formed-but-unknown uris are silently omitted downstream). The +// 100-uri cap counts raw values, before de-duplication — a request carrying +// more than the lexicon's maxLength has already broken the contract even if +// it contains duplicates — which also lets the check short-circuit before +// allocating anything proportional to an arbitrarily long parameter list. +func parseURIs(params []string) ([]string, error) { + var uris []string + raw := 0 + seen := map[string]bool{} + for _, param := range params { + for value := range strings.SplitSeq(param, ",") { + value = strings.TrimSpace(value) + if value == "" { + continue + } + raw++ + if raw > maxURIsPerQuery { + return nil, fmt.Errorf("too many uris: more than the maximum of %d", maxURIsPerQuery) + } + if seen[value] { + continue + } + if _, err := syntax.ParseATURI(value); err != nil { + return nil, fmt.Errorf("malformed at-uri: %q", value) + } + seen[value] = true + uris = append(uris, value) + } + } + return uris, nil +} + +// clientIP extracts the connection's remote IP. Deliberately not +// X-Forwarded-For: the bridge cannot know which proxies to trust, and a +// spoofable header would let one client exhaust every bucket. Deployments +// behind a load balancer rate-limit the real client at the edge. +func clientIP(r *http.Request) string { + host, _, err := net.SplitHostPort(r.RemoteAddr) + if err != nil { + return r.RemoteAddr + } + return host +} + +func (x *XRPC) writeJSON(w http.ResponseWriter, status int, body any) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + if err := json.NewEncoder(w).Encode(body); err != nil { + // The status line and headers are already out, so nothing is + // recoverable — but a truncated body should leave a trace. + x.logger.Debug("votes: write response body", "error", err) + } +} + +func (x *XRPC) writeXRPCError(w http.ResponseWriter, status int, code, message string) { + x.writeJSON(w, status, map[string]string{"error": code, "message": message}) +} diff --git a/internal/votes/xrpc_test.go b/internal/votes/xrpc_test.go new file mode 100644 index 0000000..f525dc2 --- /dev/null +++ b/internal/votes/xrpc_test.go @@ -0,0 +1,264 @@ +package votes + +import ( + "context" + "database/sql" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "net/url" + "strings" + "testing" + + "github.com/go-chi/chi/v5" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// xrpcResponse mirrors the getVoteAggregates output schema. +type xrpcResponse struct { + Aggregates []struct { + URI string `json:"uri"` + Upvotes int `json:"upvotes"` + Downvotes int `json:"downvotes"` + UpdatedAt string `json:"updatedAt"` + } `json:"aggregates"` +} + +// newXRPCRouter mounts the side channel with test-friendly rate limits. +func newXRPCRouter(t *testing.T, database *sql.DB, perSecond float64, burst int) chi.Router { + t.Helper() + xrpc, err := NewXRPC(XRPCOptions{DB: database, RatePerSecond: perSecond, RateBurst: burst}) + require.NoError(t, err) + router := chi.NewRouter() + xrpc.Routes(router) + return router +} + +// get performs one getVoteAggregates request with the given raw query. +func get(t *testing.T, router chi.Router, rawQuery string) *httptest.ResponseRecorder { + t.Helper() + return getFrom(t, router, rawQuery, "") +} + +// getFrom is get with an explicit client RemoteAddr (empty keeps the +// httptest default). +func getFrom(t *testing.T, router chi.Router, rawQuery, remoteAddr string) *httptest.ResponseRecorder { + t.Helper() + req := httptest.NewRequest(http.MethodGet, + "https://bridge.test/xrpc/"+NSIDGetVoteAggregates+"?"+rawQuery, nil) + if remoteAddr != "" { + req.RemoteAddr = remoteAddr + } + rec := httptest.NewRecorder() + router.ServeHTTP(rec, req) + return rec +} + +// testATURI builds a valid at-uri under the test repo for the given rkey. +func testATURI(rkey string) string { + return "at://" + testDID + "/" + testCollection + "/" + rkey +} + +func TestGetVoteAggregates(t *testing.T) { + database := testDB(t) + agg, objects := testAggregator(t, database) + postURI := bridgeSubject(t, objects, subjectPost, "3jzfcijpj2z2a") + commentURI := bridgeSubject(t, objects, subjectComment, "3jzfcijpj2z3a") + ctx := context.Background() + + require.NoError(t, agg.ApplyVote(ctx, like(activityID(t, 1), voterAlice, subjectPost), "")) + require.NoError(t, agg.ApplyVote(ctx, like(activityID(t, 2), voterBob, subjectPost), "")) + require.NoError(t, agg.ApplyVote(ctx, dislike(activityID(t, 3), voterCarol, subjectComment), "")) + + router := newXRPCRouter(t, database, 1000, 1000) + unknown := "at://" + testDID + "/" + testCollection + "/3jzzzzzzzzzza" + rec := get(t, router, "uris="+url.QueryEscape(commentURI)+ + "&uris="+url.QueryEscape(unknown)+"&uris="+url.QueryEscape(postURI)) + require.Equal(t, http.StatusOK, rec.Code, rec.Body.String()) + assert.Equal(t, "public, max-age=30", rec.Header().Get("Cache-Control")) + assert.Contains(t, rec.Header().Get("Content-Type"), "application/json") + + var out xrpcResponse + require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &out)) + require.Len(t, out.Aggregates, 2, "the unknown uri must be omitted, not an error") + + // Request order is preserved. + assert.Equal(t, commentURI, out.Aggregates[0].URI) + assert.Equal(t, 0, out.Aggregates[0].Upvotes) + assert.Equal(t, 1, out.Aggregates[0].Downvotes) + assert.Equal(t, postURI, out.Aggregates[1].URI) + assert.Equal(t, 2, out.Aggregates[1].Upvotes) + assert.Equal(t, 0, out.Aggregates[1].Downvotes) + assert.NotEmpty(t, out.Aggregates[0].UpdatedAt) +} + +func TestGetVoteAggregatesCommaSeparated(t *testing.T) { + database := testDB(t) + agg, objects := testAggregator(t, database) + postURI := bridgeSubject(t, objects, subjectPost, "3jzfcijpj2z2a") + commentURI := bridgeSubject(t, objects, subjectComment, "3jzfcijpj2z3a") + ctx := context.Background() + require.NoError(t, agg.ApplyVote(ctx, like(activityID(t, 1), voterAlice, subjectPost), "")) + require.NoError(t, agg.ApplyVote(ctx, like(activityID(t, 2), voterBob, subjectComment), "")) + + router := newXRPCRouter(t, database, 1000, 1000) + rec := get(t, router, "uris="+url.QueryEscape(postURI+","+commentURI)) + require.Equal(t, http.StatusOK, rec.Code, rec.Body.String()) + var out xrpcResponse + require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &out)) + assert.Len(t, out.Aggregates, 2) +} + +func TestGetVoteAggregatesHundredURIBatch(t *testing.T) { + database := testDB(t) + agg, objects := testAggregator(t, database) + postURI := bridgeSubject(t, objects, subjectPost, "3jzfcijpj2z2a") + require.NoError(t, agg.ApplyVote(context.Background(), + like(activityID(t, 1), voterAlice, subjectPost), "")) + router := newXRPCRouter(t, database, 1000, 1000) + + // Exactly 100 uris (99 unknown + 1 real) is accepted. + uris := make([]string, 0, 100) + for i := 0; i < 99; i++ { + uris = append(uris, fmt.Sprintf("at://%s/%s/3jz%09dza", testDID, testCollection, i)) + } + uris = append(uris, postURI) + params := make([]string, len(uris)) + for i, uri := range uris { + params[i] = "uris=" + url.QueryEscape(uri) + } + rec := get(t, router, strings.Join(params, "&")) + require.Equal(t, http.StatusOK, rec.Code, rec.Body.String()) + var out xrpcResponse + require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &out)) + require.Len(t, out.Aggregates, 1) + assert.Equal(t, postURI, out.Aggregates[0].URI) + + // The 101st tips it over: InvalidRequest. + params = append(params, "uris="+url.QueryEscape( + "at://"+testDID+"/"+testCollection+"/3jzoverflowza")) + rec = get(t, router, strings.Join(params, "&")) + require.Equal(t, http.StatusBadRequest, rec.Code) + var xrpcErr map[string]string + require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &xrpcErr)) + assert.Equal(t, "InvalidRequest", xrpcErr["error"]) +} + +func TestGetVoteAggregatesMissingURIs(t *testing.T) { + database := testDB(t) + router := newXRPCRouter(t, database, 1000, 1000) + + for _, query := range []string{"", "uris=", "uris=%2C%2C"} { + rec := get(t, router, query) + require.Equal(t, http.StatusBadRequest, rec.Code, "query %q", query) + var xrpcErr map[string]string + require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &xrpcErr)) + assert.Equal(t, "InvalidRequest", xrpcErr["error"]) + } +} + +func TestGetVoteAggregatesMalformedURI(t *testing.T) { + database := testDB(t) + router := newXRPCRouter(t, database, 1000, 1000) + + for _, uri := range []string{ + "not-a-uri", + "https://lemmy.world/post/100", + "at://", + "at://" + testDID + "/not_an_nsid/3jzfcijpj2z2a", + } { + rec := get(t, router, "uris="+url.QueryEscape(uri)) + require.Equal(t, http.StatusBadRequest, rec.Code, "uri %q", uri) + var xrpcErr map[string]string + require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &xrpcErr)) + assert.Equal(t, "InvalidRequest", xrpcErr["error"]) + assert.Contains(t, xrpcErr["message"], "malformed at-uri") + } + + // One malformed uri poisons the whole batch, even alongside valid ones. + rec := get(t, router, "uris="+url.QueryEscape(testATURI("3jzfcijpj2z2a"))+ + "&uris="+url.QueryEscape("not-a-uri")) + require.Equal(t, http.StatusBadRequest, rec.Code) +} + +func TestParseURIs(t *testing.T) { + a, b, c := testATURI("3jzaaaaaaaaaa"), testATURI("3jzbbbbbbbbbb"), testATURI("3jzcccccccccc") + + t.Run("trims whitespace", func(t *testing.T) { + uris, err := parseURIs([]string{" " + a + " , " + b + "\t"}) + require.NoError(t, err) + assert.Equal(t, []string{a, b}, uris) + }) + + t.Run("dedupes preserving order", func(t *testing.T) { + uris, err := parseURIs([]string{a, b, a, b, a}) + require.NoError(t, err) + assert.Equal(t, []string{a, b}, uris) + }) + + t.Run("mixes commas and repeated params", func(t *testing.T) { + uris, err := parseURIs([]string{a + "," + b, c, b}) + require.NoError(t, err) + assert.Equal(t, []string{a, b, c}, uris) + }) + + t.Run("skips empty values", func(t *testing.T) { + uris, err := parseURIs([]string{",,", "", " , "}) + require.NoError(t, err) + assert.Empty(t, uris) + }) + + t.Run("rejects malformed at-uris", func(t *testing.T) { + _, err := parseURIs([]string{a, "not-a-uri"}) + require.Error(t, err) + assert.Contains(t, err.Error(), "malformed at-uri") + }) + + t.Run("caps the raw count before dedupe", func(t *testing.T) { + // Exactly 100 raw values pass, even though they collapse to one. + params := make([]string, maxURIsPerQuery) + for i := range params { + params[i] = a + } + uris, err := parseURIs(params) + require.NoError(t, err) + assert.Equal(t, []string{a}, uris) + + // The 101st raw value is over the cap despite being a duplicate. + _, err = parseURIs(append(params, a)) + require.Error(t, err) + assert.Contains(t, err.Error(), "too many uris") + }) +} + +func TestGetVoteAggregatesRateLimitPerIP(t *testing.T) { + database := testDB(t) + // One-token buckets with negligible refill: exhausting one client's + // bucket must not touch another client's. + router := newXRPCRouter(t, database, 0.001, 1) + query := "uris=" + url.QueryEscape(testATURI("3jzfcijpj2z2a")) + + require.Equal(t, http.StatusOK, getFrom(t, router, query, "192.0.2.10:4000").Code) + require.Equal(t, http.StatusTooManyRequests, getFrom(t, router, query, "192.0.2.10:4001").Code, + "same IP, different port: one bucket") + assert.Equal(t, http.StatusOK, getFrom(t, router, query, "192.0.2.11:4000").Code, + "a different IP gets its own bucket") +} + +func TestGetVoteAggregatesRateLimited(t *testing.T) { + database := testDB(t) + // A bucket of 2 with negligible refill: the third request must be + // refused with RateLimitExceeded. + router := newXRPCRouter(t, database, 0.001, 2) + query := "uris=" + url.QueryEscape("at://"+testDID+"/"+testCollection+"/3jzfcijpj2z2a") + + require.Equal(t, http.StatusOK, get(t, router, query).Code) + require.Equal(t, http.StatusOK, get(t, router, query).Code) + rec := get(t, router, query) + require.Equal(t, http.StatusTooManyRequests, rec.Code) + var xrpcErr map[string]string + require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &xrpcErr)) + assert.Equal(t, "RateLimitExceeded", xrpcErr["error"]) +} diff --git a/lexicons/social/coves/bridge/getVoteAggregates.json b/lexicons/social/coves/bridge/getVoteAggregates.json new file mode 100644 index 0000000..7247285 --- /dev/null +++ b/lexicons/social/coves/bridge/getVoteAggregates.json @@ -0,0 +1,66 @@ +{ + "lexicon": 1, + "id": "social.coves.bridge.getVoteAggregates", + "defs": { + "main": { + "type": "query", + "description": "Get bridge-side vote aggregates for bridged posts and comments. Votes on bridged (fediverse) content never become atproto records; Tidepool aggregates them and serves the counts over this query — the one sanctioned side channel an AppView polls to display scores. Unknown or unvoted uris are omitted from the response, not an error. Public, cacheable, rate limited by IP. Versioned by nsid: breaking changes ship under a new name.", + "parameters": { + "type": "params", + "required": ["uris"], + "properties": { + "uris": { + "type": "array", + "items": { + "type": "string", + "format": "at-uri" + }, + "maxLength": 100, + "description": "at-uris of the subject records (posts or comments) to fetch aggregates for. At most 100 per call." + } + } + }, + "output": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": ["aggregates"], + "properties": { + "aggregates": { + "type": "array", + "items": { + "type": "ref", + "ref": "#voteAggregate" + } + } + } + } + } + }, + "voteAggregate": { + "type": "object", + "description": "Current vote totals for one bridged subject. Counts reflect each distinct voter's latest state (flips and undos are folded in), plus a baseline seeded from the origin instance's API for pre-bridge history.", + "required": ["uri", "upvotes", "downvotes", "updatedAt"], + "properties": { + "uri": { + "type": "string", + "format": "at-uri", + "description": "The subject record's at-uri" + }, + "upvotes": { + "type": "integer", + "minimum": 0 + }, + "downvotes": { + "type": "integer", + "minimum": 0 + }, + "updatedAt": { + "type": "string", + "format": "datetime", + "description": "When the aggregate last changed" + } + } + } + } +} -- 2.51.2