From e1ceafeec5e3e9743fc4fe9fd9e41c8a2f00ca18 Mon Sep 17 00:00:00 2001 From: Bretton <36870434+BrettM86@users.noreply.github.com> Date: Thu, 9 Jul 2026 23:33:21 -0700 Subject: [PATCH] =?UTF-8?q?Task=2008=20(2/2):=20e2e=20scenarios,=20helpers?= =?UTF-8?q?,=20FOLLOWUPS=20=E2=80=94=20and=20a=20real=20vote-retraction=20?= =?UTF-8?q?bug=20the=20suite=20caught?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Eight scenarios against the real Lemmy/PLC/Jetstream stack: subscribe, post ordering, comment strongRefs (wire-captured uri+cid), edit/delete, votes via side-channel XRPC, backfill (author-profile-first + seeded counts), restart idempotency (backfill-completion poll, gap-post cursor -resume proof, op-agnostic duplicate keys), concurrent burst. Review fixes (8-reviewer panel): drain() now fails on a dead listener instead of passing negative assertions vacuously; every consumed create/update is lexicon-validated and unknown collections fail the suite (votes-never-records enforced suite-wide); reader goroutine joined on cleanup; subscribe retries only pending states; embed .external crosses the wire validated. The new retract-to-zero assertion caught a production bug: Lemmy federates vote-clears as Undo with a RECONSTRUCTED inner vote (fresh activity id, typed Like even when the live vote is a dislike; flips are bare opposite votes with no Undo) — the id-targeted RetractVote never matched, so every Lemmy vote-clear was a silent no-op. Fixed: id-targeted first, known-id probe stops replays, unknown-id falls back to retracting the voter's live vote. +3 unit tests. make e2e: ok tidepool/tests/e2e 100.036s (8/8). Full unit suite green. All 8 PLAN.md tasks complete. Co-Authored-By: Claude Fable 5 --- FOLLOWUPS.md | 176 ++++++ LOOP_STATE.md | 4 +- README.md | 81 ++- internal/votes/aggregator.go | 119 ++-- internal/votes/aggregator_test.go | 58 ++ tests/e2e/bridge_test.go | 612 +++++++++++++++++++ tests/e2e/helpers.go | 971 ++++++++++++++++++++++++++++++ 7 files changed, 1983 insertions(+), 38 deletions(-) create mode 100644 FOLLOWUPS.md create mode 100644 tests/e2e/bridge_test.go create mode 100644 tests/e2e/helpers.go diff --git a/FOLLOWUPS.md b/FOLLOWUPS.md new file mode 100644 index 0000000..703a10b --- /dev/null +++ b/FOLLOWUPS.md @@ -0,0 +1,176 @@ +# Follow-ups + +Everything known-deferred at the end of the v1 build loop (tasks 01–08), +collected from `LOOP_STATE.md`'s cross-task notes plus discoveries made +while building the e2e harness. Organized by area; items marked **(e2e)** +were discovered or confirmed by the task-08 harness. + +## Federation & interop + +- **Lemmy first-contact Accept race (e2e).** Lemmy's federation queue + initializes a newly-seen instance's cursor at the *current* max activity + id ("skip all past activities", `crates/federate/src/worker.rs`), so the + Accept answering the very first Follow to a given Lemmy instance is + usually skipped: the instance row is created by that same Follow, and the + per-instance worker spawns after the Accept is queued. Re-sending the + Follow (fresh activity id → fresh Accept) recovers. The harness retries in + `subscribeCommunity`; production operators hit this at most once (usually) + per peer instance. Consider an automatic Follow re-send in the follow + lifecycle when a subscription stays `pending` past a threshold. +- **Author auto-upvotes do not federate (e2e).** Lemmy casts a local Like + by the post author but never announces it, so live-bridged posts read one + upvote lower than Lemmy's UI until a backfill re-seed. Accepted drift; + document for the AppView. +- **PieFed / Mbin untested** (PLAN.md deferred: verify against Lemmy only). + Known gap: `communityRef` uses a Lemmy `/c/` heuristic — Mbin uses `/m/`. +- **Lemmy 1.0 / API v4.** Everything (harness helpers, seeder's + `/api/v3/post`, webfinger expectations) targets Lemmy 0.19.x. 1.0 moves to + `/api/v4` and changes some protocol structs; revisit `e2e/lemmy/Dockerfile`'s + pinned tag and `tests/e2e/helpers.go` together. +- **Service actor type is `Service`** (changed in task 08): Lemmy's Person + protocol enum accepts only `Person|Service|Organization` — an + `Application` actor fails deserialization and the Follow is dropped. If + another platform chokes on `Service`, this needs per-platform content + negotiation (unlikely; Mastodon et al. accept it). +- **WebFinger https→http fallback** exists only when the SSRF guard is + relaxed (`ALLOW_PRIVATE_FETCH`, dev/e2e). Production is https-only by + construction — fine until someone runs a bridge against an http-only + internal instance on purpose. + +## Write side & production rollout (PLAN.md §9, explicitly out of v1) + +- Outbound Coves→Lemmy direction; AP actors for Coves users. +- Key claiming/migration for bridged users (keys are escrowed under the + rotation key precisely to enable this later). On handle-collision retry, + callers should eventually REUSE an orphaned minted DID via a PLC + updateHandle op instead of re-minting (task 03 note). +- Moderation federation, DMs. +- Relay `requestCrawl` is wired but has never been exercised against a real + relay (dev logs instead of sending). +- `ENVIRONMENT=production` has never been end-to-end tested (the harness + runs development mode for migrations-on-start, strict validation, http + scheme, private fetch). +- Strict lexicon validation is dev/test-only; production logs-and-writes. + Wire a metric on validation failures and consider a strict-first rollout + (task 05 note). + +## Sync surface (task 04 notes) + +- **No `#account{active:false}` frame on consent revocation** — subscribers + currently rely on the scrub delete-commits; tombstoned actors' historical + events stay replayable until retention expires. +- No connection cap / per-IP rate limit on the public sync surface + (pre-internet-facing hardening). +- `getRepo` buffers full CARs in memory; `ExportCAR` includes unreachable + historical blocks (consider reachable-set-only). +- `PruneEvents` is one unbatched DELETE per hourly sweep. +- MST loads are full-tree (one SELECT per node) → `PutRecord` is O(repo + size); needs a per-DID tree cache before big-community scale. +- `SigningKeys` could become a `SignCommit` capability (keeps key plaintext + inside identity; enables KMS later) — revisit before the interface + calcifies. + +## Ingestion (task 06 notes) + +- **No per-signer/per-IP rate limit on `/inbox`** — queue-flood DoS via + many self-signed identities remains the top hardening item. +- `ap_tombstones` grows unbounded (no pruner — mirror `FIREHOSE_RETENTION` + treatment). +- A `Delete` arriving before its object was ever materialized leaves + nothing to tombstone → a later `Create` still materializes (needs dedup / + tombstone-of-unseen-ids). +- `ClaimNext` does an O(N) row scan when one community's queue backs up + behind a failing event (per-key serialization cost; revisit at scale). +- A shutdown-interrupted attempt still consumes its ClaimNext attempt + increment (cosmetic). +- `MAX_BLOB_BYTES` above 5 MiB is a silent no-op (clamped by the AP + client's fixed `maxResponseBytes`). +- Mint-gate ("retry via queue backoff") is verified at unit level only — + the harness never drives minting into the rate limiter (a low + `MINT_RATE_PER_MINUTE` stack variant would need its own compose profile). + +## Votes (task 07 notes) + +- **Lemmy vote-clear Undo carries a RECONSTRUCTED inner vote (e2e, + fixed).** Measured on the wire against Lemmy 0.19: a flip federates as a + bare opposite vote (no Undo at all), and a clear as Undo{Like} with a + freshly generated activity id, typed Like even when the live vote is a + Dislike — task 07's "Lemmy inlines the original Like" was false. The + original id-targeted-only retraction therefore made EVERY Lemmy + vote-clear a silent no-op; caught by the e2e retract-to-zero assertion + and fixed in `internal/votes/aggregator.go` (id-targeted update first, + known-id replay guard, then direction-agnostic live-vote fallback). +- `vote_events` grows unbounded (no pruning of superseded/undone rows). +- Actor-delete / consent revocation does **not** scrub that actor's + `vote_events` rows (inconsistent with the scrub posture elsewhere; counts + are anonymous on the wire, so exposure is low). +- No true concurrency stress on the aggregate-row locking claim — the e2e + burst scenario exercises concurrent queue workers across communities but + contains **no votes at all**, so the vote-concurrency path is untested + beyond unit level; a many-voters-one-post hammer is still missing. +- Subject resolution happens outside the mutation tx (narrow TOCTOU with a + racing Delete, documented in task 07). +- No upper sanity cap on seeded counts; comment count seeding skipped + (per-comment API calls would triple backfill egress). +- Baseline-voter drift: a voter counted only in the seeded baseline who + later flips federates a bare Dislike (no Undo), leaving the baseline + upvote next to the new live downvote; a clear sends an Undo the live-vote + fallback cannot act on (no live row). Counts stale until re-seed. + +## Materializer (task 05 notes) + +- Transient media-fetch failure on profile refresh drops existing blobs + (no carry-forward); a stale actor behind a 403-ing instance drops content + instead of serving stale. +- `commitRecord`'s PutRecord→PutMapping is not one tx (self-heals on retry; + a Delete landing in the crash window logs Warn). +- `DeleteActor` scrubs records but not blobs stored under community DIDs. +- Test gap: `embed.images` arm + nsfw label shapes are never + lexicon-validated by unit tests (only the external-embed arm is). The e2e + suite lexicon-validates every create/update its listeners consume — the + external-embed arm crosses the wire via scenario 2's link post — but no + scenario posts an image, so `embed.images` never appears on the wire + (needs pictrs-backed image upload in the harness). +- Residual TOCTOU (task 03): a consent flip racing an in-flight commit can + let that one commit land (consent read is outside the commit tx; fine for + single-writer v1). + +## Storage / housekeeping + +- `service_keys.private_key_pem` column name lies for the "plc-rotation" + row (it holds sealed ciphertext) — rename candidate (task 03 note). +- `blocks` is append-only with no GC (load-bearing for GetRecord read + consistency; revisit together with the getRepo memory item). + +## E2E harness itself (task 08) + +- **PLC directory image is pinned** to a did-method-plc commit + (`PLC_COMMIT` in `e2e/plc/Dockerfile`); bump it deliberately via + `git ls-remote` when upstream fixes/features are needed. +- Jetstream **exits** when its upstream drops; `restart: unless-stopped` + papers over it. If Jetstream grows reconnect logic upstream, drop the + policy. +- Unexpected-collection enforcement runs on **every** commit event any + listener consumes (await and drain both fail fast on a collection outside + the four emitted ones), and the vote scenario watches the firehose + unfiltered while votes flow. Remaining gap: events emitted while no + unfiltered listener is subscribed go unchecked — a stack-wide + "nothing else ever appeared" sweep at suite end would close it. +- Scenario ideas not yet covered: image post (pictrs → blob → embed.images + lexicon validation), consent (`#nobridge` in a Lemmy bio → suppression), + `Delete(Actor)` tombstone flow, unsubscribe (Undo{Follow}) stopping + announces, community profile *update* propagation. + +## CI + +- **The e2e job has never run on GitHub Actions**, and it cold-builds + **Lemmy from source every run.** BuildKit cache + mounts (the cargo target-dir cache in `e2e/lemmy/Dockerfile`) do not + persist on GitHub-hosted runners, so each CI run pays the full debug + Rust compile — plausibly 30–60+ min on a 2-core runner, with disk + pressure to match (the workflow prunes preinstalled images up front as a + stopgap). Mitigations, in preference order: build the lemmy-debug image + once, push it to GHCR, and have the compose file pull it (rebuild only on + version bumps); or wire buildx's `gha` cache backend via + docker/build-push-action. Neither is verifiable locally, so this stays a + follow-up until the job has run on Actions at least once. diff --git a/LOOP_STATE.md b/LOOP_STATE.md index 906ccf5..786c44b 100644 --- a/LOOP_STATE.md +++ b/LOOP_STATE.md @@ -14,7 +14,9 @@ update this file → schedule next. Stop the loop when every task is `done`. | 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 | 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 (infra: Dockerfile, compose, Lemmy federation, Makefile, CI, lexicon-sync) | done | (see git log) | 7/7 reviewers (4 Claude + codex/gemini/glm, first full external panel since 03); fixes: PRODUCTION https→http redirect-downgrade guard (codex unique catch), webfinger fallback narrowed to transport failures + both-legs errors + 4 tests, minter PDS-endpoint scheme threading, PLC image commit-pin, --wait-timeout + CI logs if:always() + Makefile up-failure cleanup/teardown-status, loopback-only host binds, check-lexicons fail-open holes, sync-lexicons bridge-nesting guard, 2 false compose-header claims rewritten (invented env var, wrong --wait semantics) | -| 9 | 08-e2e-harness (tests: tests/e2e helpers + 7 scenarios, FOLLOWUPS.md, README) | in-progress | | impl done in loop 8's agent run; both pre-fix and post-fix `make e2e` passed (8/8 scenarios) | +| 9 | 08-e2e-harness (tests: tests/e2e helpers + 8 scenarios, FOLLOWUPS.md, README) | done | (see git log) | 8/8 reviewers (5 Claude + codex/gemini/glm; gemini zero-issue "excellent", codex sharpest); fixes: drain() dead-listener vacuous-pass (5/8 flagged), centralized vetEvent (unknown-collection Fatalf + lexicon-validate every consumed create/update, suite-wide locked-decision-7 enforcement), scenario-7 backfill-completion poll + gap-post cursor-resume proof + op-agnostic dup keys, readLoop goroutine join, subscribe fail-fast on explicit reject, embed.external e2e coverage, seeder e2e assertion. NEW ASSERTION CAUGHT REAL BUG: Lemmy vote-clear federates Undo with RECONSTRUCTED inner vote (fresh id, type Like even for live dislike; flips are bare opposite votes, no Undo) → id-targeted RetractVote no-oped every production vote-clear; fixed with known-id replay probe + live-vote fallback + 3 unit tests | + +ALL TASKS DONE — loop complete. `make e2e` green (8/8 scenarios, ~100s), full unit suite green. Statuses: pending → in-progress → review → done (or blocked: ). diff --git a/README.md b/README.md index fd77ff3..00550e9 100644 --- a/README.md +++ b/README.md @@ -32,6 +32,80 @@ directory — the test suite can never create DIDs on the public `plc.directory`. Point `TIDEPOOL_TEST_PLC_URL` elsewhere-on-localhost if your directory is on a different port. +## Running the stack (e2e harness) + +The end-to-end harness runs the whole read path against **real +infrastructure** — a real Lemmy federating with the bridge, a real did:plc +directory backing DID minting, and a real Jetstream decoding the firehose — +in one compose network: + +``` + docker-compose.e2e.yml + ┌─────────────────────────────────────────────────────────────────────┐ + │ http://lemmy (debug build) http://tidepool (this repo) │ + │ ┌───────────────────────┐ Follow ◀─ ┌───────────────────────────┐ │ + │ │ lemmy + lemmy-postgres│ Accept ─▶ │ tidepool + its postgres │ │ + │ │ + pictrs │ Announce ─▶│ inbox → queue → material │ │ + │ └───────────────────────┘ (plain │ izer → virtual PDS │ │ + │ HTTP) └─────┬──────────────┬──────┘ │ + │ mint DIDs │ CBOR frames│ │ + │ ┌─────────────▼──┐ ┌────────▼──────┐ │ + │ │ plc (did:plc │ │ jetstream │ │ + │ │ + plc-postgres)│ │ (JSON events) │ │ + │ └────────────────┘ └────────┬──────┘ │ + └───────────────────────────────────────────────────────────┼─────────┘ + host (127.0.0.1 only): tidepool :8092, lemmy :8541, jetstream :6028 ◀─┘ + tests/e2e (go test -tags e2e) +``` + +The host ports bind **loopback-only** (`127.0.0.1:8092/8541/6028`): the +stack carries an admin token and runs with `ALLOW_PRIVATE_FETCH=1`, so it +must not be reachable from the local network. + +```sh +make e2e # build + start the stack, run tests/e2e, tear down +make e2e-up # start and leave running (iterate with make e2e-test) +make e2e-test # run the suite against the running stack +make e2e-logs # tail everything +make e2e-down # tear down (removes volumes) +``` + +The first `make e2e` builds Lemmy **from source in debug mode** (a full Rust +compile; cached afterwards). That is deliberate: whether Lemmy accepts +`http://` federation URLs is a compile-time property (`cfg!(debug_assertions)` +→ the AP crate's `allow_http_urls`), and every published Lemmy image is a +release build that refuses plain HTTP, explicit ports, and private addresses. +A debug build — the same thing LemmyNet's own `docker/federation` compose +uses — federates happily over `http://lemmy` ↔ `http://tidepool` inside the +network, with `BRIDGE_SCHEME=http` (dev-only) making the bridge emit +plain-HTTP AP ids to match. See the header comments in +`docker-compose.e2e.yml` and `e2e/lemmy/Dockerfile` for the full story. + +The suite (`tests/e2e/bridge_test.go`, build tag `e2e`) covers: subscribe → +`community.profile` on the firehose; a link post → `actor.profile` then +`community.post` in order, with the shared url crossing the wire as +`embed.external`; comment threads in the authors' repos with resolving +strongRefs; edits/deletes as update/delete ops; post **and** comment votes +reaching `getVoteAggregates` — upvote, flip, retract — while **never** +appearing as records; pre-subscribe backfill with each author's profile +emitted before their posts and Lemmy's pre-existing vote counts seeded into +the aggregates; a mid-test container restart proving deterministic-rkey +replay idempotency (the forced backfill redo is confirmed complete via the +admin API's `last_backfill_at` before asserting it emitted nothing) plus +exactly-once delivery of a post created in the recovery window and Jetstream +cursor resume; and a concurrent-ingestion burst accounted per +(did, collection/rkey). Every create/update the tests consume from Jetstream +is validated against the vendored Coves lexicons on the consumer side of the +wire, and any collection outside the four the bridge emits fails the suite +immediately (votes must never become records). Two scripts keep the vendored +lexicons honest: `scripts/sync-lexicons.sh` copies them from a Coves +checkout; `scripts/check-lexicons.sh` verifies the committed manifest (the +layer that runs in CI) and additionally byte-compares against +`~/Code/coves` when that checkout exists (local only). + +Everything is local-only: the harness never contacts `plc.directory`, public +relays, or public Lemmy instances. + ## Configuration Environment variables with logged dev defaults (see @@ -43,6 +117,7 @@ production**: | `DATABASE_URL` | local dev postgres | bridge state | | `LISTEN_ADDR` | `:8091` | HTTP bind address | | `BRIDGE_HOSTNAME` | `localhost` | public domain of the bridge; anchors handles and the PDS endpoint in minted DID docs | +| `BRIDGE_SCHEME` | `https` | scheme of the bridge's own AP URLs (actor id, inbox, activity ids). `http` is dev-only — the e2e harness federates with a debug-mode Lemmy over plain HTTP | | `PLC_DIRECTORY_URL` | `http://localhost:3002` (local, `make plc-up`) | did:plc directory; production uses `https://plc.directory` | | `BRIDGE_KEK` | fixed public dev key | 32-byte key-encryption key (64 hex chars or base64) sealing per-actor signing keys and the escrow rotation key at rest (AES-256-GCM) | | `BRIDGE_SERVICE_DID` | *(optional)* | pre-provisioned service DID for the bridge's own actor | @@ -169,9 +244,9 @@ 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 -watch it re-emit our records as JSON. Not automated in CI (it needs Docker -networking to the host); the manual runbook is: +**Automated:** the e2e harness (`make e2e`, above) runs a real Jetstream +against the bridge and asserts on the decoded events — this manual runbook +survives for ad-hoc poking at a dev bridge: Until the materializer (task 05) generates organic writes, the repo test suite is the write driver — so point the bridge at the **test** database and diff --git a/internal/votes/aggregator.go b/internal/votes/aggregator.go index 7fbd955..5add9d4 100644 --- a/internal/votes/aggregator.go +++ b/internal/votes/aggregator.go @@ -179,16 +179,33 @@ func (a *Aggregator) ApplyVote(ctx context.Context, vote *ap.Object, communityIR } // 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. +// voter's live vote on the subject is marked undone and the aggregate +// recomputed. +// +// What Lemmy actually federates (measured by the e2e suite against a real +// Lemmy 0.19, and NOT what task 07 originally assumed): +// +// - flip (up → down): a bare Dislike, no Undo at all — ApplyVote's +// supersede handles it; +// - clear (score 0): Announce{Undo{Like}} whose inner vote is +// RECONSTRUCTED — a freshly generated activity id, and typed "Like" +// even when the voter's live vote is a Dislike. +// +// So the inner vote's id and type are hints, not truth. The retraction +// runs in two steps: first an id-targeted update (correct for +// implementations that inline the original activity, and what keeps a +// REPLAYED Undo{Like id=A} after a re-like id=B from retracting B); if that +// matches nothing and the id was never seen at all (Lemmy's regenerated +// id, or no id), fall back to retracting the voter's current live vote on +// the subject regardless of direction — Undo means "remove my vote". The +// (voter, subject) scoping plus the announced-undo community binding keep +// forged undos harmless. +// +// 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 voter with no live +// vote (out-of-order delivery, or history that only exists as a seeded +// baseline), a replayed undo naming an already-superseded activity, 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 @@ -241,31 +258,63 @@ func (a *Aggregator) RetractVote(ctx context.Context, vote *ap.Object, community 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} + // Step 1: target the undone activity by its own id (correct for + // implementations that inline the original vote). The voter/subject + // predicates stay as defense — a forged undo naming someone else's + // activity id retracts nothing. No direction predicate: the id is + // the authoritative target and the inner type is unreliable. + var retracted int64 if vote.ID != "" { - query += ` AND activity_id = $4` - args = append(args, vote.ID) + result, 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) + if err != nil { + return fmt.Errorf("retract vote %q by %q on %q: %w", vote.ID, voter, subject, err) + } + if retracted, err = result.RowsAffected(); err != nil { + return fmt.Errorf("retract vote %q by %q on %q: rows affected: %w", vote.ID, voter, subject, err) + } + if retracted == 0 { + // Zero rows means either a REPLAY (the id is known but its + // vote was already undone/superseded — must retract nothing, + // above all not a newer re-vote by the same voter) or a + // Lemmy-style RECONSTRUCTED inner vote (fresh id the bridge + // has never seen). Distinguish by whether the id exists. + var known bool + if err := tx.QueryRowContext(ctx, ` + SELECT EXISTS (SELECT 1 FROM vote_events WHERE activity_id = $1)`, + vote.ID).Scan(&known); err != nil { + return fmt.Errorf("probe undone activity %q: %w", vote.ID, err) + } + if known { + a.logger.Debug("vote retraction dropped: replayed undo of a superseded vote", + "subject", subject, "voter", voter, "activity", vote.ID) + return nil + } + } } - 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) + + // Step 2: unknown or absent id — "remove my vote". Retract the + // voter's current live vote on the subject regardless of direction + // (the reconstructed inner vote is typed Like even when the live + // vote is a Dislike). + if retracted == 0 { + result, err := tx.ExecContext(ctx, ` + UPDATE vote_events + SET undone = TRUE + WHERE subject_ap_id = $1 AND voter_ap_id = $2 AND NOT undone`, + subject, voter) + if err != nil { + return fmt.Errorf("retract vote by %q on %q: %w", voter, subject, err) + } + if retracted, err = result.RowsAffected(); 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", + a.logger.Debug("vote retraction dropped: no live vote to retract", "subject", subject, "voter", voter, "direction", direction) return nil } @@ -278,9 +327,11 @@ func (a *Aggregator) RetractVote(ctx context.Context, vote *ap.Object, community // 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. +// counted only in the baseline who later flips federates a bare Dislike +// (Lemmy sends no Undo on flips), so the retired upvote stays in the +// baseline next to the new live downvote; a clear sends an Undo the +// fallback cannot act on (no live vote row). 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 { diff --git a/internal/votes/aggregator_test.go b/internal/votes/aggregator_test.go index 5117c64..097c3dd 100644 --- a/internal/votes/aggregator_test.go +++ b/internal/votes/aggregator_test.go @@ -140,6 +140,64 @@ func TestReplayedUndoDoesNotRetractReLike(t *testing.T) { assert.Equal(t, 0, down) } +// TestLemmyClearAfterFlipRetractsLiveVote pins the vote-clear wire behavior +// the e2e suite measured against a real Lemmy 0.19: a flip federates as a +// bare opposite vote (no Undo), and a clear federates as an Undo whose inner +// vote is RECONSTRUCTED — a freshly generated activity id, typed Like even +// though the voter's live vote is the dislike. The retraction must fall back +// to removing the voter's live vote regardless of the inner id/type. +// (Before this pin, every Lemmy vote-clear was a silent no-op.) +func TestLemmyClearAfterFlipRetractsLiveVote(t *testing.T) { + database := testDB(t) + agg, objects := testAggregator(t, database) + bridgeSubject(t, objects, subjectPost, "3jzfcijpj2z2a") + ctx := context.Background() + + // Like → flip (bare Dislike) → clear (Undo{Like} with a fresh id). + require.NoError(t, agg.ApplyVote(ctx, like(activityID(t, 1), voterAlice, subjectPost), "")) + require.NoError(t, agg.ApplyVote(ctx, dislike(activityID(t, 2), voterAlice, subjectPost), "")) + require.NoError(t, agg.RetractVote(ctx, like(activityID(t, 3), voterAlice, subjectPost), "")) + + up, down, found := counts(t, database, subjectPost) + require.True(t, found) + assert.Equal(t, 0, up) + assert.Equal(t, 0, down, "the regenerated-id Undo{Like} must retract the live dislike") +} + +// TestIDLessUndoRetractsLiveVote: an inline undo whose inner vote carries no +// id at all still removes the voter's live vote. +func TestIDLessUndoRetractsLiveVote(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("", voterAlice, subjectPost), "")) + + up, down, found := counts(t, database, subjectPost) + require.True(t, found) + assert.Equal(t, 0, up) + assert.Equal(t, 0, down) +} + +// TestUndoByOtherVoterRetractsNothing: the unknown-id fallback is scoped to +// (voter, subject) — an undo by someone who never voted must not touch +// another voter's live vote. +func TestUndoByOtherVoterRetractsNothing(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, 2), voterBob, subjectPost), "")) + + up, down, _ := counts(t, database, subjectPost) + assert.Equal(t, 1, up, "bob's undo must not retract alice's live vote") + assert.Equal(t, 0, down) +} + func TestDuplicateActivityIDOnAnotherSubjectMintsNoAggregate(t *testing.T) { database := testDB(t) agg, objects := testAggregator(t, database) diff --git a/tests/e2e/bridge_test.go b/tests/e2e/bridge_test.go new file mode 100644 index 0000000..61455cd --- /dev/null +++ b/tests/e2e/bridge_test.go @@ -0,0 +1,612 @@ +//go:build e2e + +package e2e + +import ( + "fmt" + "testing" + "time" +) + +// setupSubscribedCommunity creates a fresh Lemmy community and subscribes +// the bridge to it via the admin API (WebFinger handle form, exercising +// webfinger against a real Lemmy — a task-06 deferred gap). It returns both +// sides of the mapping: the Lemmy community (for further API writes) and +// the bridge's view (DID, group IRI). +func setupSubscribedCommunity(t *testing.T, h *harness, prefix string) (lemmyCommunity, adminCommunity) { + t.Helper() + name := h.uniqueName(t, prefix) + community := h.admin.createCommunity(t, name, "E2E "+name) + sub := h.subscribeCommunity(t, "!"+name+"@lemmy") + if sub.DID == "" { + t.Fatalf("subscribed community %s has no DID", name) + } + return community, sub +} + +// Scenario 1: Subscribe !testing@lemmy → community.profile appears on the +// firehose and validates against the Coves lexicon (as every create/update +// consumed by the suite does, via the listener's centralized vetting). +func TestSubscribe_CommunityProfileOnFirehose(t *testing.T) { + h := newHarness(t) + + cursor := cursorNow() + l := h.newListener(t, cursor, colCommunityProfile) + + community, sub := setupSubscribedCommunity(t, h, "sub") + + ev := l.await("community.profile create", func(e *jsEvent) bool { + return e.Did == sub.DID && e.Commit.Collection == colCommunityProfile && + e.Commit.Operation == opCreate + }) + if ev.Commit.RKey != rkeySelf { + t.Errorf("community.profile rkey = %q, want %q", ev.Commit.RKey, rkeySelf) + } + if got := recordField(t, ev.Commit.Record, "name"); got != community.Name { + t.Errorf("community.profile name = %q, want %q", got, community.Name) + } +} + +// Scenario 2: a Lemmy user shares a link → actor.profile first, then +// community.post in the COMMUNITY's repo with author = the user's DID +// (PLAN.md locked decision 3), in that order — and the shared url crosses +// the wire as an embed.external whose uri survives byte-identical (the +// classic CBOR→JSON breakage point). +func TestPost_ActorProfileThenPost(t *testing.T) { + h := newHarness(t) + community, sub := setupSubscribedCommunity(t, h, "post") + + username := h.uniqueName(t, "bob") + user := h.registerUser(t, username) + + cursor := cursorNow() + l := h.newListener(t, cursor, colActorProfile, colPost) + + title := "Hello from " + username + // A link post. The url stays on the compose network (LOCAL-ONLY): Lemmy + // fetches it for opengraph metadata; the bridge itself never fetches the + // link — it copies the AP fields into the embed. + linkURL := "http://lemmy/?e2e=" + h.suffix + post := user.createLinkPost(t, community.ID, title, "first bridged post", linkURL) + t.Logf("created lemmy post %d (%s)", post.ID, post.APID) + + // Both events funnel through one listener; the FIRST match for this + // scenario must be the author's profile — the AppView rejects posts + // whose author isn't indexed yet, so emission order is load-bearing. + first := l.await("actor.profile or community.post for "+username, func(e *jsEvent) bool { + switch e.Commit.Collection { + case colActorProfile: + name, _ := fieldOf(e.Commit.Record, "displayName") + return name == username + case colPost: + got, _ := fieldOf(e.Commit.Record, "title") + return e.Did == sub.DID && got == title + } + return false + }) + if first.Commit.Collection != colActorProfile { + t.Fatalf("first event was %s — actor.profile must be emitted before the post", first) + } + profileEv := first + if profileEv.Commit.RKey != rkeySelf { + t.Errorf("actor.profile rkey = %q, want %q", profileEv.Commit.RKey, rkeySelf) + } + + postEv := l.await("community.post create", func(e *jsEvent) bool { + got, _ := fieldOf(e.Commit.Record, "title") + return e.Commit.Collection == colPost && e.Did == sub.DID && + e.Commit.Operation == opCreate && got == title + }) + + if got := recordField(t, postEv.Commit.Record, "author"); got != profileEv.Did { + t.Errorf("post author = %q, want the author's DID %q", got, profileEv.Did) + } + if got := recordField(t, postEv.Commit.Record, "community"); got != sub.DID { + t.Errorf("post community = %q, want repo DID %q (posts live in the community's repo)", got, sub.DID) + } + if got := recordField(t, postEv.Commit.Record, "embed", "external", "uri"); got != linkURL { + t.Errorf("post embed.external.uri = %q, want the shared link %q", got, linkURL) + } +} + +// Scenario 3: comment + nested reply → comment records in the AUTHOR's repo +// (PLAN.md locked decision 3: comments live in the bridged user's repo, +// pinned against the DID of the author's actor.profile) whose +// reply.root/reply.parent strongRefs resolve to the exact uri+cid of the +// earlier firehose events. +func TestComments_StrongRefsResolve(t *testing.T) { + h := newHarness(t) + community, sub := setupSubscribedCommunity(t, h, "cmt") + + username := h.uniqueName(t, "carol") + user := h.registerUser(t, username) + + cursor := cursorNow() + l := h.newListener(t, cursor, colActorProfile, colPost, colComment) + + title := "Comment thread " + h.suffix + post := user.createPost(t, community.ID, title, "root post") + + // The author's profile precedes their first content (scenario 2 asserts + // the ordering; here it pins the author's DID). + profileEv := l.await("author actor.profile", func(e *jsEvent) bool { + name, _ := fieldOf(e.Commit.Record, "displayName") + return e.Commit.Collection == colActorProfile && name == username + }) + authorDID := profileEv.Did + + postEv := l.await("post create", func(e *jsEvent) bool { + got, _ := fieldOf(e.Commit.Record, "title") + return e.Commit.Collection == colPost && e.Did == sub.DID && got == title + }) + postURI, postCID := postEv.atURI(), postEv.Commit.CID + + comment := user.createComment(t, post.ID, 0, "top-level comment") + commentEv := l.await("top-level comment create", func(e *jsEvent) bool { + got, _ := fieldOf(e.Commit.Record, "reply", "parent", "uri") + return e.Commit.Collection == colComment && e.Commit.Operation == opCreate && + got == postURI + }) + if commentEv.Did != authorDID { + t.Errorf("comment emitted from repo %s, want the author's repo %s (comments live in the bridged user's repo)", + commentEv.Did, authorDID) + } + + // Top-level comment: parent and root are both the post, cid included. + for _, ref := range []string{"parent", "root"} { + if got := recordField(t, commentEv.Commit.Record, "reply", ref, "uri"); got != postURI { + t.Errorf("comment reply.%s.uri = %q, want %q", ref, got, postURI) + } + if got := recordField(t, commentEv.Commit.Record, "reply", ref, "cid"); got != postCID { + t.Errorf("comment reply.%s.cid = %q, want the post event's cid %q", ref, got, postCID) + } + } + + commentURI, commentCID := commentEv.atURI(), commentEv.Commit.CID + + user.createComment(t, post.ID, comment.ID, "nested reply") + replyEv := l.await("nested reply create", func(e *jsEvent) bool { + got, _ := fieldOf(e.Commit.Record, "reply", "parent", "uri") + return e.Commit.Collection == colComment && e.Commit.Operation == opCreate && + got == commentURI + }) + if replyEv.Did != authorDID { + t.Errorf("nested reply emitted from repo %s, want the author's repo %s", replyEv.Did, authorDID) + } + + // Nested reply: parent = the comment, root = the post — resolved + // against the uri+cid observed on the wire, i.e. exactly what a + // strongRef-following AppView would look up. + if got := recordField(t, replyEv.Commit.Record, "reply", "parent", "cid"); got != commentCID { + t.Errorf("reply parent.cid = %q, want %q", got, commentCID) + } + if got := recordField(t, replyEv.Commit.Record, "reply", "root", "uri"); got != postURI { + t.Errorf("reply root.uri = %q, want %q", got, postURI) + } + if got := recordField(t, replyEv.Commit.Record, "reply", "root", "cid"); got != postCID { + t.Errorf("reply root.cid = %q, want %q", got, postCID) + } +} + +// Scenario 4: edit post → update event on the same rkey; delete comment → +// delete op on the firehose. +func TestUpdateAndDelete(t *testing.T) { + h := newHarness(t) + community, sub := setupSubscribedCommunity(t, h, "upd") + + username := h.uniqueName(t, "dave") + user := h.registerUser(t, username) + + cursor := cursorNow() + l := h.newListener(t, cursor, colPost, colComment) + + title := "Editable " + h.suffix + post := user.createPost(t, community.ID, title, "original body") + postEv := l.await("post create", func(e *jsEvent) bool { + got, _ := fieldOf(e.Commit.Record, "title") + return e.Commit.Collection == colPost && e.Commit.Operation == opCreate && + e.Did == sub.DID && got == title + }) + + comment := user.createComment(t, post.ID, 0, "doomed comment") + commentEv := l.await("comment create", func(e *jsEvent) bool { + got, _ := fieldOf(e.Commit.Record, "reply", "parent", "uri") + return e.Commit.Collection == colComment && e.Commit.Operation == opCreate && + got == postEv.atURI() + }) + + user.editPost(t, post.ID, "edited body") + updateEv := l.await("post update", func(e *jsEvent) bool { + return e.Commit.Collection == colPost && e.Commit.Operation == opUpdate && + e.Did == sub.DID && e.Commit.RKey == postEv.Commit.RKey + }) + if got := recordField(t, updateEv.Commit.Record, "content"); got != "edited body" { + t.Errorf("updated post content = %q, want %q", got, "edited body") + } + if updateEv.Commit.CID == postEv.Commit.CID { + t.Error("update event carries the same cid as the create — no new commit?") + } + + user.deleteComment(t, comment.ID) + deleteEv := l.await("comment delete", func(e *jsEvent) bool { + return e.Commit.Collection == colComment && e.Commit.Operation == opDelete && + e.Did == commentEv.Did && e.Commit.RKey == commentEv.Commit.RKey + }) + if len(deleteEv.Commit.Record) != 0 && string(deleteEv.Commit.Record) != "null" { + t.Errorf("delete op unexpectedly carries a record: %s", truncate(deleteEv.Commit.Record, 200)) + } +} + +// Scenario 5: votes — on posts AND comments — update the getVoteAggregates +// side channel and NEVER appear as records on the firehose (PLAN.md locked +// decision 7). Covers the full lifecycle: upvote, flip to downvote +// (Undo{Like} + Dislike), retract (bare Undo), and a comment vote (task +// 07's reply.root community-binding path). +func TestVotes_SideChannelOnly(t *testing.T) { + h := newHarness(t) + community, sub := setupSubscribedCommunity(t, h, "vote") + + author := h.registerUser(t, h.uniqueName(t, "erin")) + voter := h.registerUser(t, h.uniqueName(t, "frank")) + + cursor := cursorNow() + // No collection filter: watch EVERYTHING that hits the firehose while + // votes flow. + l := h.newListener(t, cursor) + + title := "Votable " + h.suffix + post := author.createPost(t, community.ID, title, "vote on me") + postEv := l.await("post create", func(e *jsEvent) bool { + got, _ := fieldOf(e.Commit.Record, "title") + return e.Commit.Collection == colPost && e.Did == sub.DID && got == title + }) + postURI := postEv.atURI() + + // Upvote → aggregates show it. + voter.likePost(t, post.ID, 1) + awaitAggregates(t, h, postURI, 1, 0) + + // Flip to downvote → Lemmy sends Undo{Like} + Dislike; the aggregator + // must retract the upvote and apply the downvote. + voter.likePost(t, post.ID, -1) + awaitAggregates(t, h, postURI, 0, 1) + + // Retract → a bare Undo{Dislike}; back to zero on both sides. + voter.likePost(t, post.ID, 0) + awaitAggregates(t, h, postURI, 0, 0) + + // Comment votes ride the same side channel (the aggregator binds a + // comment to its community via reply.root — pinned here end-to-end). + comment := author.createComment(t, post.ID, 0, "votable comment") + commentEv := l.await("comment create", func(e *jsEvent) bool { + got, _ := fieldOf(e.Commit.Record, "reply", "parent", "uri") + return e.Commit.Collection == colComment && e.Commit.Operation == opCreate && + got == postURI + }) + voter.likeComment(t, comment.ID, 1) + awaitAggregates(t, h, commentEv.atURI(), 1, 0) + + // The whole time, nothing vote-shaped may have hit the firehose. The + // listener's centralized vetting already fails fast on any unexpected + // collection; this explicit unfiltered sweep is the belt to that + // suspenders. + for _, ev := range l.drain(3 * time.Second) { + if ev.Kind == kindCommit && ev.Commit != nil && !expectedCollections[ev.Commit.Collection] { + t.Errorf("unexpected collection on firehose during voting: %s", ev) + } + } +} + +// awaitAggregates polls the side-channel XRPC until the subject shows the +// expected live counts (votes flow through Lemmy's federation queue and the +// bridge's inbox queue, so counts converge, not snap). +func awaitAggregates(t *testing.T, h *harness, uri string, up, down int64) { + t.Helper() + deadline := time.Now().Add(eventTimeout) + var last voteAggregate + for time.Now().Before(deadline) { + aggs := h.getVoteAggregates(t, uri) + last = aggs[uri] + if last.Upvotes == up && last.Downvotes == down { + return + } + time.Sleep(time.Second) + } + t.Fatalf("aggregates for %s = %d up / %d down, want %d/%d", uri, last.Upvotes, last.Downvotes, up, down) +} + +// Scenario 6: backfill — posts that existed BEFORE the bridge subscribed +// appear on the firehose after the Follow is accepted (outbox walk), with +// the community profile first and every post's author profile before the +// post itself; pre-existing vote counts are seeded from Lemmy's API +// (SEED_COUNTS_FROM_API, on in the e2e compose). +func TestBackfill_PreexistingPosts(t *testing.T) { + h := newHarness(t) + + name := h.uniqueName(t, "bf") + community := h.admin.createCommunity(t, name, "Backfill "+name) + + author := h.registerUser(t, h.uniqueName(t, "gina")) + titles := make(map[string]bool, 3) + var votedPost lemmyPost + var votedTitle string + for i := range 3 { + title := fmt.Sprintf("Pre-existing %d %s", i, h.suffix) + p := author.createPost(t, community.ID, title, fmt.Sprintf("history %d", i)) + titles[title] = true + if i == 0 { + votedPost, votedTitle = p, title + } + } + // A second user upvotes one pre-existing post BEFORE the bridge knows + // this community exists. Lemmy's API then reports upvotes=2 for it (the + // author auto-like + this vote) — the number the backfill seeder must + // carry over, since neither vote will ever federate as an activity. + voter := h.registerUser(t, h.uniqueName(t, "hal")) + voter.likePost(t, votedPost.ID, 1) + + cursor := cursorNow() + l := h.newListener(t, cursor, colCommunityProfile, colActorProfile, colPost) + + sub := h.subscribeCommunity(t, "!"+name+"@lemmy") + + l.await("community.profile create", func(e *jsEvent) bool { + return e.Did == sub.DID && e.Commit.Collection == colCommunityProfile + }) + + // All three historical posts must materialize (accept triggers the + // outbox backfill). Order among the posts is newest-first outbox order — + // only presence is asserted — but each post's AUTHOR must have hit the + // firehose as an actor.profile before the post itself (the Coves AppView + // rejects posts by unindexed authors, so backfill emission order is as + // load-bearing as the live path's). + profileDIDs := map[string]bool{} + votedURI := "" + remaining := len(titles) + for remaining > 0 { + ev := l.await(fmt.Sprintf("backfilled post or author profile (%d posts to go)", remaining), func(e *jsEvent) bool { + switch e.Commit.Collection { + case colActorProfile: + return true // consume every profile to track author-first ordering + case colPost: + title, _ := fieldOf(e.Commit.Record, "title") + return e.Did == sub.DID && e.Commit.Operation == opCreate && titles[title] + } + return false + }) + if ev.Commit.Collection == colActorProfile { + profileDIDs[ev.Did] = true + continue + } + title := recordField(t, ev.Commit.Record, "title") + if got := recordField(t, ev.Commit.Record, "author"); !profileDIDs[got] { + t.Errorf("backfilled post %q emitted before its author's actor.profile (author %s)", title, got) + } + if title == votedTitle { + votedURI = ev.atURI() + } + delete(titles, title) + remaining-- + } + + // The seeded baseline shows through the side channel: 2 up (author + // auto-like + the pre-subscribe vote), 0 down. + if votedURI == "" { + t.Fatal("the voted pre-existing post never appeared on the firehose") + } + awaitAggregates(t, h, votedURI, 2, 0) +} + +// Scenario 7: recovery across restart — bounce the Tidepool container and +// prove three things: +// +// 1. A forced backfill redo — confirmed to have RUN via the admin API's +// last_backfill_at advancing — re-materializes everything with NO +// duplicate commits (deterministic rkeys + identical-re-put-is-a-noop). +// 2. A "gap" post created right after /healthz, while Jetstream may still +// be reconnecting, is delivered exactly once (cursor resume, not luck). +// 3. Replaying the ORIGINAL cursor after the bounce yields the entire +// pre-restart history exactly once — keyed by (did, collection/rkey) +// across ALL operations, so a broken replay emitting create+update for +// the same record is caught too. +func TestRestart_ReplayIsIdempotent(t *testing.T) { + h := newHarness(t) + + name := h.uniqueName(t, "rst") + community := h.admin.createCommunity(t, name, "Restart "+name) + author := h.registerUser(t, h.uniqueName(t, "hank")) + + title := "Survivor " + h.suffix + author.createPost(t, community.ID, title, "pre-restart post") + + cursor := cursorNow() + l := h.newListener(t, cursor, colCommunityProfile, colActorProfile, colPost) + + handle := "!" + name + "@lemmy" + sub := h.subscribeCommunity(t, handle) + + firstEv := l.await("pre-restart post create", func(e *jsEvent) bool { + got, _ := fieldOf(e.Commit.Record, "title") + return e.Did == sub.DID && e.Commit.Collection == colPost && got == title + }) + if firstEv.Commit.Operation != opCreate { + t.Fatalf("pre-restart post arrived as %q, want %q", firstEv.Commit.Operation, opCreate) + } + + // Bounce the bridge. Jetstream exits when its upstream drops and docker + // revives it (see docker-compose.e2e.yml), so the pre-restart listener's + // connection is gone — close it deliberately first. + l.close() + h.restartTidepool(t) + + // Immediately after /healthz — deliberately NOT waiting for Jetstream's + // own recovery — create a post into the reconnect gap. Wherever in the + // recovery dance it lands, it must come out exactly once. + gapTitle := "Gap " + h.suffix + author.createPost(t, community.ID, gapTitle, "created in the recovery window") + + // Force a full backfill redo and wait for the admin API to report a NEW + // last_backfill_at (the run is async): only after the redo has actually + // finished is "it emitted no duplicates" an assertion rather than a race + // against a slow redo. + before, ok := h.findCommunity(t, handle, sub.Community) + if !ok { + t.Fatalf("community %s missing from the admin list after restart", handle) + } + h.triggerBackfill(t, handle) + backfillDeadline := time.Now().Add(eventTimeout) + for { + state, found := h.findCommunity(t, handle, sub.Community) + if found && state.LastBackfillAt != "" && state.LastBackfillAt != before.LastBackfillAt { + t.Logf("backfill redo completed at %s (previous run: %q)", state.LastBackfillAt, before.LastBackfillAt) + break + } + if time.Now().After(backfillDeadline) { + t.Fatalf("backfill redo never completed: last_backfill_at stuck at %q", before.LastBackfillAt) + } + time.Sleep(time.Second) + } + + // Replay everything from the ORIGINAL cursor. Key the accounting by + // (did, collection/rkey) — NOT operation — so a create+update pair for + // one record counts as the duplicate it is. + l2 := h.newListener(t, cursor, colCommunityProfile, colActorProfile, colPost) + + title2 := "Post-restart " + h.suffix + author.createPost(t, community.ID, title2, "after the bounce") + + seen := map[string]int{} + keyOf := func(ev *jsEvent) string { + return fmt.Sprintf("%s %s/%s", ev.Did, ev.Commit.Collection, ev.Commit.RKey) + } + gapKey := "" + sawNew, sawGap := false, false + deadline := time.Now().Add(eventTimeout) + for !(sawNew && sawGap) && time.Now().Before(deadline) { + for _, ev := range l2.drain(time.Second) { + if ev.Kind != kindCommit || ev.Commit == nil { + continue + } + seen[keyOf(ev)]++ + if ev.Did == sub.DID && ev.Commit.Collection == colPost { + switch got, _ := fieldOf(ev.Commit.Record, "title"); got { + case gapTitle: + sawGap, gapKey = true, keyOf(ev) + case title2: + sawNew = true + } + } + } + } + if !sawGap { + t.Fatal("gap post (created during the recovery window) never reached jetstream — cursor resume dropped it") + } + if !sawNew { + t.Fatal("post-restart post never reached jetstream — pipeline did not survive the restart") + } + // The backfill redo finished before l2 dialed; a short trailing drain + // still catches any late duplicate emission in flight. + for _, ev := range l2.drain(8 * time.Second) { + if ev.Kind == kindCommit && ev.Commit != nil { + seen[keyOf(ev)]++ + } + } + + // The replayed pre-restart post must appear EXACTLY once: zero would + // mean Jetstream lost history, twice would mean the backfill redo + // re-committed it (deterministic-rkey idempotency broken). + firstKey := fmt.Sprintf("%s %s/%s", sub.DID, colPost, firstEv.Commit.RKey) + if n := seen[firstKey]; n != 1 { + t.Errorf("pre-restart post replayed %d times (want exactly 1): %s", n, firstKey) + } + if n := seen[gapKey]; n != 1 { + t.Errorf("gap post appeared %d times (want exactly 1): %s", n, gapKey) + } + for key, n := range seen { + if n > 1 { + t.Errorf("record %s committed %d times after restart — duplicate emission", key, n) + } + } +} + +// Scenario 8 (bonus, closes a task-06 deferred gap end-to-end): a burst of +// activity across TWO communities and several authors — concurrent queue +// workers (INGEST_WORKERS=4) with distinct ordering keys — lands every +// record exactly once, counting every commit per (did, collection/rkey) +// regardless of operation. +func TestBurst_ConcurrentIngestionExactlyOnce(t *testing.T) { + h := newHarness(t) + + commA, subA := setupSubscribedCommunity(t, h, "ba") + commB, subB := setupSubscribedCommunity(t, h, "bb") + + users := []*lemmyClient{ + h.registerUser(t, h.uniqueName(t, "ivy")), + h.registerUser(t, h.uniqueName(t, "jack")), + h.registerUser(t, h.uniqueName(t, "kim")), + } + + cursor := cursorNow() + l := h.newListener(t, cursor, colPost) + + const perCommunity = 6 + titles := make(map[string]bool, 2*perCommunity) + for i := range perCommunity { + for j, id := range []int{commA.ID, commB.ID} { + title := fmt.Sprintf("Burst %d.%d %s", j, i, h.suffix) + users[(i+j)%len(users)].createPost(t, id, title, "burst body") + titles[title] = true + } + } + + // Account for EVERY commit on the two communities' repos by + // (did, collection/rkey) — an update sneaking in after a create is a + // duplicate commit on that record, not a separate event. + seen := map[string]int{} + keyTitle := map[string]string{} + count := func(e *jsEvent) { + if e.Kind != kindCommit || e.Commit == nil { + return + } + if e.Did != subA.DID && e.Did != subB.DID { + return + } + key := fmt.Sprintf("%s %s/%s", e.Did, e.Commit.Collection, e.Commit.RKey) + seen[key]++ + if title, ok := fieldOf(e.Commit.Record, "title"); ok && titles[title] { + keyTitle[key] = title + } + } + + // Every post arrives… + matched := map[string]bool{} + for len(matched) < len(titles) { + ev := l.await(fmt.Sprintf("burst post (%d to go)", len(titles)-len(matched)), func(e *jsEvent) bool { + count(e) + title, _ := fieldOf(e.Commit.Record, "title") + return e.Commit.Operation == opCreate && + (e.Did == subA.DID || e.Did == subB.DID) && + titles[title] && !matched[title] + }) + matched[recordField(t, ev.Commit.Record, "title")] = true + } + // …and exactly once: nothing trailing, no second commit on any rkey. + for _, ev := range l.drain(4 * time.Second) { + count(ev) + } + for key, n := range seen { + if n != 1 { + t.Errorf("record %s (%q) committed %d times, want exactly 1", key, keyTitle[key], n) + } + } + // …and no burst title may have landed under two different rkeys. + perTitle := map[string]int{} + for _, title := range keyTitle { + perTitle[title]++ + } + for title := range titles { + if perTitle[title] != 1 { + t.Errorf("burst post %q landed on %d distinct records, want exactly 1", title, perTitle[title]) + } + } +} diff --git a/tests/e2e/helpers.go b/tests/e2e/helpers.go new file mode 100644 index 0000000..5530013 --- /dev/null +++ b/tests/e2e/helpers.go @@ -0,0 +1,971 @@ +//go:build e2e + +// Package e2e drives the docker-compose.e2e.yml stack end to end: a real +// Lemmy (debug build, plain-HTTP federation) federating with Tidepool, a +// real did:plc directory backing DID minting, and a real Jetstream decoding +// the bridge's subscribeRepos firehose. +// +// Coves-style: E2E tests test REAL infrastructure, not mocks. Run them with +// `make e2e` (compose up --build → wait for health → go test -tags e2e → +// compose down -v), or against an already-running stack with +// `go test -tags e2e ./tests/e2e/...`. +// +// Everything here is LOCAL-ONLY: the stack never touches plc.directory, +// public relays, or public Lemmy instances. +package e2e + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "net/url" + "os" + "os/exec" + "path/filepath" + "runtime" + "strings" + "sync" + "testing" + "time" + + "github.com/bluesky-social/indigo/atproto/lexicon" + "github.com/gorilla/websocket" + + "tidepool/lexicons" +) + +// ── Configuration ────────────────────────────────────────────────────────── + +// envOr reads an env var with a default (the docker-compose.e2e.yml host +// ports). +func envOr(name, fallback string) string { + if v := os.Getenv(name); v != "" { + return v + } + return fallback +} + +func tidepoolURL() string { return envOr("TIDEPOOL_E2E_URL", "http://localhost:8092") } +func lemmyURL() string { return envOr("LEMMY_E2E_URL", "http://localhost:8541") } +func jetstreamURL() string { + return envOr("JETSTREAM_E2E_URL", "ws://localhost:6028") +} +func adminToken() string { return envOr("TIDEPOOL_E2E_ADMIN_TOKEN", "e2e-admin-token") } + +// stackTimeout bounds the initial wait-for-healthy loop. Container startup +// (Lemmy migrations, PLC boot) can be slow on a cold machine; `make e2e` +// already waited for compose health, so this is usually instant. +const stackTimeout = 5 * time.Minute + +// eventTimeout bounds one wait for a federation → firehose → Jetstream +// round trip. LEMMY_TEST_FAST_FEDERATION makes deliveries near-instant, but +// leave generous slack for slow CI. +const eventTimeout = 90 * time.Second + +// ── Firehose vocabulary ──────────────────────────────────────────────────── + +// Collections the bridge emits (task 05's materializer). +const ( + colCommunityProfile = "social.coves.community.profile" + colActorProfile = "social.coves.actor.profile" + colPost = "social.coves.community.post" + colComment = "social.coves.community.comment" +) + +// Event kinds, operations, and rkeys as they appear on the Jetstream wire. +// Consts, not inline strings: a typo'd operation in an await predicate would +// not fail compilation — it would burn a 90s timeout instead. +const ( + kindCommit = "commit" + + opCreate = "create" + opUpdate = "update" + opDelete = "delete" + + rkeySelf = "self" +) + +// expectedCollections is the complete set of record collections that may +// legally appear on the firehose. Anything else — vote records above all +// (PLAN.md locked decision 7: votes NEVER become records) — is a bug, and +// the listener enforces this on every consumed commit event (vetEvent). +var expectedCollections = map[string]bool{ + colCommunityProfile: true, + colActorProfile: true, + colPost: true, + colComment: true, +} + +// ── Harness ──────────────────────────────────────────────────────────────── + +// harness bundles the three service endpoints plus a logged-in Lemmy admin. +type harness struct { + http *http.Client + admin *lemmyClient // Lemmy admin (setup credentials from lemmy.hjson) + // suffix makes names unique per run so the suite can run repeatedly + // against a long-lived stack (deterministic rkeys make true re-runs + // idempotent, but distinct communities keep scenarios independent). + suffix string +} + +var ( + stackOnce sync.Once + stackErr error + setupOnce sync.Once + setupErr error + adminJWT string + nameSerial int + nameMu sync.Mutex +) + +// newHarness waits for the stack once per process, logs in the Lemmy admin, +// and (once) applies the site settings the suite needs: open registration, +// no captcha, federation on, and the tidepool host allowlisted. +func newHarness(t *testing.T) *harness { + t.Helper() + + stackOnce.Do(func() { stackErr = waitForStack() }) + if stackErr != nil { + t.Fatalf("e2e stack not ready: %v (start it with `make e2e-up`)", stackErr) + } + + h := &harness{ + http: &http.Client{Timeout: 30 * time.Second}, + suffix: strings.ToLower(fmt.Sprintf("%x", time.Now().UnixNano()%0xffffff)), + } + + setupOnce.Do(func() { setupErr = h.setupLemmySite() }) + if setupErr != nil { + t.Fatalf("lemmy site setup: %v", setupErr) + } + h.admin = &lemmyClient{h: h, jwt: adminJWT} + return h +} + +// uniqueName mints a lemmy-legal name ([a-z0-9_], ≤20 chars) unique across +// the process and across runs. Truncating would cut off the unique suffix +// and silently collide across runs, so an over-long compose is fatal. +func (h *harness) uniqueName(t *testing.T, prefix string) string { + t.Helper() + nameMu.Lock() + defer nameMu.Unlock() + nameSerial++ + name := fmt.Sprintf("%s_%s%d", prefix, h.suffix, nameSerial) + if len(name) > 20 { + t.Fatalf("uniqueName(%q) = %q exceeds lemmy's 20-char limit — shorten the prefix", prefix, name) + } + return name +} + +// waitForStack polls every service the tests talk to until healthy. +// Wait-for-healthy loops, not fixed sleeps: container startup time varies +// wildly between a warm laptop and cold CI. +func waitForStack() error { + deadline := time.Now().Add(stackTimeout) + client := &http.Client{Timeout: 5 * time.Second} + + probes := []struct { + name string + check func() error + }{ + {"tidepool /healthz", func() error { + return probeHTTP(client, tidepoolURL()+"/healthz") + }}, + {"lemmy /api/v3/site", func() error { + return probeHTTP(client, lemmyURL()+"/api/v3/site") + }}, + {"jetstream /subscribe", func() error { + u := jetstreamURL() + "/subscribe?cursor=" + fmt.Sprint(time.Now().UnixMicro()) + conn, _, err := websocket.DefaultDialer.Dial(u, nil) + if err != nil { + return err + } + return conn.Close() + }}, + } + + for _, probe := range probes { + for { + err := probe.check() + if err == nil { + break + } + if time.Now().After(deadline) { + return fmt.Errorf("%s: %w", probe.name, err) + } + time.Sleep(2 * time.Second) + } + } + return nil +} + +func probeHTTP(client *http.Client, url string) error { + resp, err := client.Get(url) + if err != nil { + return err + } + defer func() { _ = resp.Body.Close() }() + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("GET %s: status %d", url, resp.StatusCode) + } + return nil +} + +// setupLemmySite logs in the setup admin and applies idempotent site +// settings: open registration (the default is require-application, which +// blocks user creation), captcha off, federation on, and — per the task +// spec — the tidepool host allowlisted (Lemmy then federates ONLY with the +// bridge, which doubles as a guard against accidental external federation). +func (h *harness) setupLemmySite() error { + admin := &lemmyClient{h: h} + if err := admin.login("admin", "lemmylemmy"); err != nil { + return fmt.Errorf("admin login: %w", err) + } + adminJWT = admin.jwt + + body := map[string]any{ + "registration_mode": "Open", + "captcha_enabled": false, + "federation_enabled": true, + "allowed_instances": []string{"tidepool"}, + } + var out json.RawMessage + if err := admin.do(http.MethodPut, "/api/v3/site", body, &out); err != nil { + return fmt.Errorf("edit site: %w", err) + } + return nil +} + +// ── Lemmy HTTP API client (v3, lemmy 0.19) ───────────────────────────────── + +type lemmyClient struct { + h *harness + jwt string +} + +// do issues one JSON API call with the client's bearer token. +func (c *lemmyClient) do(method, path string, body, out any) error { + var reqBody io.Reader + if body != nil { + raw, err := json.Marshal(body) + if err != nil { + return err + } + reqBody = bytes.NewReader(raw) + } + req, err := http.NewRequest(method, lemmyURL()+path, reqBody) + if err != nil { + return err + } + if body != nil { + req.Header.Set("Content-Type", "application/json") + } + if c.jwt != "" { + req.Header.Set("Authorization", "Bearer "+c.jwt) + } + resp, err := c.h.http.Do(req) + if err != nil { + return err + } + defer func() { _ = resp.Body.Close() }() + raw, err := io.ReadAll(resp.Body) + if err != nil { + return err + } + if resp.StatusCode/100 != 2 { + return fmt.Errorf("lemmy %s %s: status %d: %s", method, path, resp.StatusCode, truncate(raw, 300)) + } + if out != nil { + if err := json.Unmarshal(raw, out); err != nil { + return fmt.Errorf("lemmy %s %s: decode: %w (%s)", method, path, err, truncate(raw, 200)) + } + } + return nil +} + +func truncate(b []byte, n int) string { + if len(b) > n { + return string(b[:n]) + "…" + } + return string(b) +} + +func (c *lemmyClient) login(user, password string) error { + var out struct { + JWT string `json:"jwt"` + } + err := c.do(http.MethodPost, "/api/v3/user/login", + map[string]any{"username_or_email": user, "password": password}, &out) + if err != nil { + return err + } + if out.JWT == "" { + return fmt.Errorf("login %s: empty jwt", user) + } + c.jwt = out.JWT + return nil +} + +// registerUser creates a fresh Lemmy user and returns a logged-in client. +func (h *harness) registerUser(t *testing.T, username string) *lemmyClient { + t.Helper() + c := &lemmyClient{h: h} + password := "password_" + username + var out struct { + JWT string `json:"jwt"` + } + err := c.do(http.MethodPost, "/api/v3/user/register", map[string]any{ + "username": username, + "password": password, + "password_verify": password, + "show_nsfw": true, + }, &out) + if err != nil { + t.Fatalf("register lemmy user %s: %v", username, err) + } + if out.JWT == "" { + t.Fatalf("register lemmy user %s: no jwt in response (site requires verification?)", username) + } + c.jwt = out.JWT + return c +} + +type lemmyCommunity struct { + ID int + Name string + APID string +} + +func (c *lemmyClient) createCommunity(t *testing.T, name, title string) lemmyCommunity { + t.Helper() + var out struct { + CommunityView struct { + Community struct { + ID int `json:"id"` + Name string `json:"name"` + ActorID string `json:"actor_id"` + } `json:"community"` + } `json:"community_view"` + } + err := c.do(http.MethodPost, "/api/v3/community", + map[string]any{"name": name, "title": title}, &out) + if err != nil { + t.Fatalf("create community %s: %v", name, err) + } + comm := out.CommunityView.Community + return lemmyCommunity{ID: comm.ID, Name: comm.Name, APID: comm.ActorID} +} + +type lemmyPost struct { + ID int + APID string +} + +func (c *lemmyClient) createPost(t *testing.T, communityID int, title, body string) lemmyPost { + t.Helper() + return c.createLinkPost(t, communityID, title, body, "") +} + +// createLinkPost creates a post with a shared link (url ""), which the +// bridge materializes as an embed.external. The url must stay on the +// compose network (LOCAL-ONLY) — Lemmy fetches it for opengraph metadata. +func (c *lemmyClient) createLinkPost(t *testing.T, communityID int, title, body, url string) lemmyPost { + t.Helper() + req := map[string]any{"name": title, "community_id": communityID, "body": body} + if url != "" { + req["url"] = url + } + var out struct { + PostView struct { + Post struct { + ID int `json:"id"` + APID string `json:"ap_id"` + } `json:"post"` + } `json:"post_view"` + } + if err := c.do(http.MethodPost, "/api/v3/post", req, &out); err != nil { + t.Fatalf("create post %q: %v", title, err) + } + return lemmyPost{ID: out.PostView.Post.ID, APID: out.PostView.Post.APID} +} + +func (c *lemmyClient) editPost(t *testing.T, postID int, newBody string) { + t.Helper() + if err := c.do(http.MethodPut, "/api/v3/post", + map[string]any{"post_id": postID, "body": newBody}, nil); err != nil { + t.Fatalf("edit post %d: %v", postID, err) + } +} + +type lemmyComment struct { + ID int + APID string +} + +// createComment creates a comment; parentID 0 means a top-level comment. +func (c *lemmyClient) createComment(t *testing.T, postID, parentID int, content string) lemmyComment { + t.Helper() + body := map[string]any{"post_id": postID, "content": content} + if parentID != 0 { + body["parent_id"] = parentID + } + var out struct { + CommentView struct { + Comment struct { + ID int `json:"id"` + APID string `json:"ap_id"` + } `json:"comment"` + } `json:"comment_view"` + } + if err := c.do(http.MethodPost, "/api/v3/comment", body, &out); err != nil { + t.Fatalf("create comment on post %d: %v", postID, err) + } + return lemmyComment{ID: out.CommentView.Comment.ID, APID: out.CommentView.Comment.APID} +} + +func (c *lemmyClient) deleteComment(t *testing.T, commentID int) { + t.Helper() + if err := c.do(http.MethodPost, "/api/v3/comment/delete", + map[string]any{"comment_id": commentID, "deleted": true}, nil); err != nil { + t.Fatalf("delete comment %d: %v", commentID, err) + } +} + +// likePost casts a vote: score 1 (up), -1 (down), 0 (retract). +func (c *lemmyClient) likePost(t *testing.T, postID, score int) { + t.Helper() + if err := c.do(http.MethodPost, "/api/v3/post/like", + map[string]any{"post_id": postID, "score": score}, nil); err != nil { + t.Fatalf("vote %+d on post %d: %v", score, postID, err) + } +} + +// likeComment casts a comment vote: score 1 (up), -1 (down), 0 (retract). +func (c *lemmyClient) likeComment(t *testing.T, commentID, score int) { + t.Helper() + if err := c.do(http.MethodPost, "/api/v3/comment/like", + map[string]any{"comment_id": commentID, "score": score}, nil); err != nil { + t.Fatalf("vote %+d on comment %d: %v", score, commentID, err) + } +} + +// ── Tidepool admin client ────────────────────────────────────────────────── + +type adminCommunity struct { + Community string `json:"community"` + DID string `json:"did"` + FollowState string `json:"follow_state"` + LastBackfillAt string `json:"last_backfill_at"` // RFC3339, empty until first backfill completes +} + +func (h *harness) adminDo(method, path string, body, out any) error { + var reqBody io.Reader + if body != nil { + raw, err := json.Marshal(body) + if err != nil { + return err + } + reqBody = bytes.NewReader(raw) + } + req, err := http.NewRequest(method, tidepoolURL()+path, reqBody) + if err != nil { + return err + } + req.Header.Set("Authorization", "Bearer "+adminToken()) + if body != nil { + req.Header.Set("Content-Type", "application/json") + } + resp, err := h.http.Do(req) + if err != nil { + return err + } + defer func() { _ = resp.Body.Close() }() + raw, err := io.ReadAll(resp.Body) + if err != nil { + return err + } + if resp.StatusCode/100 != 2 { + return fmt.Errorf("tidepool %s %s: status %d: %s", method, path, resp.StatusCode, truncate(raw, 300)) + } + if out != nil { + if err := json.Unmarshal(raw, out); err != nil { + return fmt.Errorf("tidepool %s %s: decode: %w", method, path, err) + } + } + return nil +} + +// subscribeCommunity drives POST /admin/communities until the Follow is +// accepted and returns the community's DID. +// +// The retry loop is deliberate, not paranoia: Lemmy's federation queue +// skips all activities queued before an instance's per-instance worker +// first starts ("skip all past activities", crates/federate/src/worker.rs). +// On FIRST contact the instance row is created by our Follow itself, so the +// Accept usually races the worker spawn and is dropped. Re-sending the +// Follow (each send has a fresh activity id) makes Lemmy emit a new Accept, +// which the by-now-running worker delivers. Real deployments hit the same +// race at most once (usually) per peer instance. +// +// Only pending/empty states are retried: an explicit rejection or failure +// state is a real answer and fails immediately instead of masquerading as a +// timeout. +func (h *harness) subscribeCommunity(t *testing.T, community string) adminCommunity { + t.Helper() + deadline := time.Now().Add(eventTimeout) + attempt := 0 + requirePending := func(state adminCommunity) { + switch state.FollowState { + case "", "pending", "accepted": + default: + t.Fatalf("subscribe %s: follow state %q (community %s) — explicit non-pending answer, not retrying", + community, state.FollowState, state.Community) + } + } + var last adminCommunity + for { + attempt++ + var resp adminCommunity + if err := h.adminDo(http.MethodPost, "/admin/communities", + map[string]any{"community": community}, &resp); err != nil { + t.Fatalf("subscribe %s (attempt %d): %v", community, attempt, err) + } + last = resp + if resp.FollowState == "accepted" { + return resp + } + requirePending(resp) + // Poll for the Accept before re-sending. + pollUntil := time.Now().Add(10 * time.Second) + for time.Now().Before(pollUntil) { + time.Sleep(time.Second) + if state, ok := h.findCommunity(t, community, resp.Community); ok { + last = state + if state.FollowState == "accepted" { + return state + } + requirePending(state) + } + } + if time.Now().After(deadline) { + t.Fatalf("subscribe %s: follow not accepted after %d attempts (last state %q, community %s)", + community, attempt, last.FollowState, last.Community) + } + t.Logf("subscribe %s: still pending after attempt %d, re-sending follow (lemmy first-contact race)", community, attempt) + } +} + +// findCommunity looks a community up in GET /admin/communities by either +// the requested handle or the resolved group IRI. +func (h *harness) findCommunity(t *testing.T, handle, groupIRI string) (adminCommunity, bool) { + t.Helper() + var list struct { + Communities []adminCommunity `json:"communities"` + } + if err := h.adminDo(http.MethodGet, "/admin/communities", nil, &list); err != nil { + t.Fatalf("list communities: %v", err) + } + for _, c := range list.Communities { + if c.Community == groupIRI || c.Community == handle { + return c, true + } + } + return adminCommunity{}, false +} + +// triggerBackfill forces a backfill run for a subscribed community. +func (h *harness) triggerBackfill(t *testing.T, community string) { + t.Helper() + if err := h.adminDo(http.MethodPost, "/admin/communities/backfill", + map[string]any{"community": community}, nil); err != nil { + t.Fatalf("trigger backfill %s: %v", community, err) + } +} + +// getVoteAggregates reads the side-channel XRPC. +type voteAggregate struct { + URI string `json:"uri"` + Upvotes int64 `json:"upvotes"` + Downvotes int64 `json:"downvotes"` +} + +func (h *harness) getVoteAggregates(t *testing.T, uris ...string) map[string]voteAggregate { + t.Helper() + q := url.Values{} + for _, u := range uris { + q.Add("uris", u) + } + resp, err := h.http.Get(tidepoolURL() + "/xrpc/social.coves.bridge.getVoteAggregates?" + q.Encode()) + if err != nil { + t.Fatalf("getVoteAggregates: %v", err) + } + defer func() { _ = resp.Body.Close() }() + raw, err := io.ReadAll(resp.Body) + if err != nil { + t.Fatalf("getVoteAggregates: read body: %v", err) + } + if resp.StatusCode != http.StatusOK { + t.Fatalf("getVoteAggregates: status %d: %s", resp.StatusCode, truncate(raw, 200)) + } + var out struct { + Aggregates []voteAggregate `json:"aggregates"` + } + if err := json.Unmarshal(raw, &out); err != nil { + t.Fatalf("getVoteAggregates: decode: %v", err) + } + byURI := make(map[string]voteAggregate, len(out.Aggregates)) + for _, a := range out.Aggregates { + byURI[a.URI] = a + } + return byURI +} + +// ── Jetstream WebSocket listener ─────────────────────────────────────────── + +// jsEvent is Jetstream's JSON event shape (kind "commit" only — the bridge +// emits no identity/account frames yet). +type jsEvent struct { + Did string `json:"did"` + TimeUs int64 `json:"time_us"` + Kind string `json:"kind"` + Commit *jsCommit `json:"commit"` +} + +type jsCommit struct { + Rev string `json:"rev"` + Operation string `json:"operation"` + Collection string `json:"collection"` + RKey string `json:"rkey"` + Record json.RawMessage `json:"record"` + CID string `json:"cid"` +} + +func (e *jsEvent) String() string { + if e.Commit == nil { + return fmt.Sprintf("%s %s", e.Kind, e.Did) + } + return fmt.Sprintf("%s %s %s/%s (%s)", e.Commit.Operation, e.Did, e.Commit.Collection, e.Commit.RKey, e.Kind) +} + +// atURI is the at-uri of the committed record. +func (e *jsEvent) atURI() string { + return fmt.Sprintf("at://%s/%s/%s", e.Did, e.Commit.Collection, e.Commit.RKey) +} + +// jsListener consumes /subscribe and buffers events for matching. +type jsListener struct { + t *testing.T + conn *websocket.Conn + events chan *jsEvent + closed chan struct{} // closed by close(): deliberate shutdown + done chan struct{} // closed by readLoop's defer: goroutine exited + once sync.Once + + mu sync.Mutex + readErr error // readLoop's terminal error, nil on deliberate close +} + +// setReadErr records why readLoop died — unless the listener was closed +// deliberately, in which case the read error is just the connection teardown. +func (l *jsListener) setReadErr(err error) { + select { + case <-l.closed: + return + default: + } + l.mu.Lock() + l.readErr = err + l.mu.Unlock() +} + +// readError returns readLoop's terminal error (nil if the listener was +// closed deliberately or is still running). +func (l *jsListener) readError() error { + l.mu.Lock() + defer l.mu.Unlock() + return l.readErr +} + +// newListener subscribes from cursorMicros (0 = now). Collections filter +// server-side via wantedCollections; none means all collections (used for +// negative assertions like "votes never hit the firehose"). +// +// Anti-flake convention (from the Coves harness): capture the cursor BEFORE +// triggering the action under test, so a subscription opened after the +// write still replays it. +func (h *harness) newListener(t *testing.T, cursorMicros int64, collections ...string) *jsListener { + t.Helper() + q := url.Values{} + if cursorMicros > 0 { + q.Set("cursor", fmt.Sprint(cursorMicros)) + } + for _, c := range collections { + q.Add("wantedCollections", c) + } + u := jetstreamURL() + "/subscribe" + if enc := q.Encode(); enc != "" { + u += "?" + enc + } + // Dial with retries: Jetstream exits when its upstream drops (the + // restart scenario provokes exactly that) and docker revives it, so a + // fresh listener may race the reboot. + var conn *websocket.Conn + dialDeadline := time.Now().Add(eventTimeout) + for { + var err error + conn, _, err = websocket.DefaultDialer.Dial(u, nil) + if err == nil { + break + } + if time.Now().After(dialDeadline) { + t.Fatalf("dial jetstream %s: %v", u, err) + } + t.Logf("dial jetstream: %v (retrying)", err) + time.Sleep(2 * time.Second) + } + l := &jsListener{ + t: t, + conn: conn, + events: make(chan *jsEvent, 1024), + closed: make(chan struct{}), + done: make(chan struct{}), + } + go l.readLoop() + t.Cleanup(l.close) + return l +} + +// cursorNow returns a Jetstream cursor a couple of seconds in the past — +// capture it immediately before the action under test. +func cursorNow() int64 { + return time.Now().Add(-2 * time.Second).UnixMicro() +} + +func (l *jsListener) readLoop() { + defer close(l.done) + defer close(l.events) + for { + select { + case <-l.closed: + return + default: + } + _ = l.conn.SetReadDeadline(time.Now().Add(5 * time.Minute)) + _, raw, err := l.conn.ReadMessage() + if err != nil { + l.setReadErr(err) + return + } + var ev jsEvent + if err := json.Unmarshal(raw, &ev); err != nil { + // Guarded log: never Logf after a deliberate close (close() + // joins this goroutine, but belt-and-braces). + select { + case <-l.closed: + return + default: + l.t.Logf("jetstream: undecodable event: %v (%s)", err, truncate(raw, 200)) + } + continue + } + select { + case l.events <- &ev: + case <-l.closed: + return + } + } +} + +// close shuts the listener down and JOINS the read goroutine, so readLoop +// can never touch t after the test (and its cleanup) completed. +func (l *jsListener) close() { + l.once.Do(func() { + close(l.closed) + _ = l.conn.Close() + <-l.done + }) +} + +// vetEvent enforces the suite-wide wire contract on EVERY consumed commit +// event, matched or skipped: +// +// - only the known emitted collections may appear — anything else, votes +// above all (PLAN.md locked decision 7: votes NEVER become records), is +// an immediate failure, no matter which scenario's await/drain window it +// lands in; +// - every create/update record must validate against the vendored Coves +// lexicons (deletes carry no record). +func (l *jsListener) vetEvent(ev *jsEvent) { + l.t.Helper() + if ev.Kind != kindCommit || ev.Commit == nil { + return + } + if !expectedCollections[ev.Commit.Collection] { + l.t.Fatalf("unexpected collection on firehose: %s — only community/actor profiles, posts, and comments may ever appear (votes never become records)", ev) + } + if op := ev.Commit.Operation; op == opCreate || op == opUpdate { + validateLexicon(l.t, ev.Commit) + } +} + +// await returns the first commit event matching pred, failing the test +// after eventTimeout. Non-matching events are vetted (vetEvent), logged, +// and kept out of the way (each scenario matches on its own +// community/author to stay independent of concurrent traffic). +func (l *jsListener) await(desc string, pred func(*jsEvent) bool) *jsEvent { + l.t.Helper() + timer := time.NewTimer(eventTimeout) + defer timer.Stop() + for { + select { + case ev, ok := <-l.events: + if !ok { + l.t.Fatalf("await %s: jetstream events channel closed unexpectedly (read error: %v)", desc, l.readError()) + return nil + } + l.vetEvent(ev) + if ev.Kind == kindCommit && ev.Commit != nil && pred(ev) { + l.t.Logf("await %s: matched %s", desc, ev) + return ev + } + l.t.Logf("await %s: skipping %s", desc, ev) + case <-timer.C: + l.t.Fatalf("await %s: no matching jetstream event within %s", desc, eventTimeout) + return nil + } + } +} + +// drain collects everything that arrives within d (for negative +// assertions: "nothing else showed up"). A dead reader would make every +// negative assertion pass vacuously, so an unexpectedly closed channel is +// fatal — silence must mean "connected and nothing arrived". +func (l *jsListener) drain(d time.Duration) []*jsEvent { + l.t.Helper() + timer := time.NewTimer(d) + defer timer.Stop() + var out []*jsEvent + for { + select { + case ev, ok := <-l.events: + if !ok { + l.t.Fatalf("drain: jetstream events channel closed unexpectedly after %d events (read error: %v)", len(out), l.readError()) + return out + } + l.vetEvent(ev) + out = append(out, ev) + case <-timer.C: + return out + } + } +} + +// ── Lexicon conformance ──────────────────────────────────────────────────── + +// validateLexicon checks a firehose record against the vendored Coves +// lexicons with the same indigo validator (and flags) the materializer uses +// — but on the CONSUMER side of the wire: what Jetstream decoded is what +// the AppView would index. +func validateLexicon(t *testing.T, commit *jsCommit) { + t.Helper() + catalog, err := lexicons.Catalog() + if err != nil { + t.Fatalf("load lexicon catalog: %v", err) + } + var data any + if err := json.Unmarshal(commit.Record, &data); err != nil { + t.Fatalf("decode %s record: %v", commit.Collection, err) + } + rec, ok := data.(map[string]any) + if !ok { + t.Fatalf("%s record is not an object", commit.Collection) + } + recType, _ := rec["$type"].(string) + if recType != commit.Collection { + t.Fatalf("record $type %q != collection %q", recType, commit.Collection) + } + if err := lexicon.ValidateRecord(catalog, data, recType, lexicon.ValidateFlags(0)); err != nil { + t.Errorf("%s record fails lexicon validation: %v\nrecord: %s", + recType, err, truncate(commit.Record, 600)) + } +} + +// fieldOf digs a string field out of a raw record, reporting ok=false for +// anything that doesn't match: empty/null records (delete events!), +// undecodable bytes, non-object intermediates, missing keys, non-string +// leaves. Total over every event shape, so it is the ONLY record accessor +// allowed inside await predicates, which see foreign and delete events. +func fieldOf(record json.RawMessage, path ...string) (string, bool) { + if len(record) == 0 { + return "", false + } + var cur any + if err := json.Unmarshal(record, &cur); err != nil { + return "", false + } + for _, key := range path { + m, ok := cur.(map[string]any) + if !ok { + return "", false + } + if cur, ok = m[key]; !ok { + return "", false + } + } + s, ok := cur.(string) + return s, ok +} + +// recordField is the fatal variant of fieldOf for post-match assertions on +// an event that is already known to carry the record. +func recordField(t *testing.T, record json.RawMessage, path ...string) string { + t.Helper() + s, ok := fieldOf(record, path...) + if !ok { + t.Fatalf("record field %q missing or not a string in: %s", + strings.Join(path, "."), truncate(record, 300)) + } + return s +} + +// ── Container control (scenario 7) ───────────────────────────────────────── + +// composeFile locates docker-compose.e2e.yml relative to this source file. +func composeFile(t *testing.T) string { + t.Helper() + if v := os.Getenv("TIDEPOOL_E2E_COMPOSE"); v != "" { + return v + } + _, self, _, ok := runtime.Caller(0) + if !ok { + t.Fatal("cannot locate test source for compose file discovery") + } + return filepath.Join(filepath.Dir(self), "..", "..", "docker-compose.e2e.yml") +} + +// restartTidepool restarts the bridge container and waits for /healthz — +// the scenario-7 crash/redeploy simulation. +func (h *harness) restartTidepool(t *testing.T) { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute) + defer cancel() + cmd := exec.CommandContext(ctx, "docker", "compose", "-f", composeFile(t), "restart", "tidepool") + out, err := cmd.CombinedOutput() + if err != nil { + t.Fatalf("restart tidepool: %v\n%s", err, out) + } + client := &http.Client{Timeout: 5 * time.Second} + deadline := time.Now().Add(2 * time.Minute) + for { + if err := probeHTTP(client, tidepoolURL()+"/healthz"); err == nil { + return + } + if time.Now().After(deadline) { + t.Fatal("tidepool did not come back healthy after restart") + } + time.Sleep(time.Second) + } +} -- 2.51.2