1import { env, runInDurableObject } from "cloudflare:test";2import { describe, expect, it } from "vitest";3import { generateSecretKey, getPublicKey, type Event } from "nostr-tools/pure";4import type { Relay } from "../../src/relay.ts";5import { Socket, dial, runPullRound, type PullJob, type PullSocket } from "../../src/pull.ts";6import { runPullSourceRound, type Job } from "../../src/jobs.ts";7import { ev, now, rpc } from "../helpers/relay.ts";8import { WS } from "../helpers/ws.ts";910async function claimed(host: string, owner: Uint8Array) {11 await rpc(host, owner, "claim");12 return WS.connect(host);13}1415async function run(name: string, job: PullJob) {16 return runInDurableObject(env.RELAY.getByName(name), async (relay: Relay) => runPullRound(relay, job));17}1819class RefusingSocket implements PullSocket {20 constructor(private query: boolean) {}21 send() { /* scripted refusal */ }22 async recv() { return (this.query ? ["CLOSED", "history", "auth-required: source refused history"] : ["NEG-ERR", "pull", "unsupported: sync is switched off"]) as unknown[]; }23 close() { /* scripted socket */ }24}2526class ScriptedSocket implements PullSocket {27 constructor(private messages: unknown[][]) {}28 send() { /* scripted socket */ }29 async recv() { return this.messages.shift() ?? ["EOSE", "history"]; }30 close() { /* scripted socket */ }31}3233describe("history pull fallback", () => {34 it("falls back from a refused NIP-77 sync and resumes ordinary queries", async () => {35 const owner = generateSecretKey();36 const source = "pull-query-source.bind.ws";37 const destination = "pull-query-destination.bind.ws";38 const sourceClient = await claimed(source, owner);39 await rpc(source, owner, "setpolicy", { features: { sync: false } });40 const first = ev(owner, 1, "first", [], now() - 20);41 const second = ev(owner, 1, "second", [], now() - 10);42 expect((await sourceClient.ok(first)).ok).toBe(true);43 expect((await sourceClient.ok(second)).ok).toBe(true);44 await claimed(destination, owner);4546 const job: PullJob = {47 url: "wss://" + source,48 startedAt: now(),49 rounds: 0,50 stored: 0,51 skipped: 0,52 blobs: 0,53 failures: 0,54 filter: { authors: [getPublicKey(owner)] },55 };56 const firstRound = await run("pull-query-destination", job);57 expect(firstRound).toEqual({ more: true, error: "" });58 expect(job.progress).toMatchObject({ mode: "query", status: "running", pages: 0 });59 expect(job.progress?.windows).toEqual([{ since: 0, until: job.startedAt }]);6061 const secondRound = await run("pull-query-destination", job);62 expect(secondRound).toEqual({ more: false, error: "" });63 expect(job.progress).toMatchObject({ mode: "query", status: "best-effort", pages: 1, partial: false });64 expect(job.stored).toBe(2);65 });6667 it("marks a saturated one-second query window partial", async () => {68 const owner = generateSecretKey();69 const source = "pull-query-saturated-source.bind.ws";70 const destination = "pull-query-saturated-destination.bind.ws";71 await claimed(source, owner);72 await rpc(source, owner, "setpolicy", { features: { sync: false } });73 const timestamp = now() - 30;74 const events: Event[] = [];75 await runInDurableObject(env.RELAY.getByName("pull-query-saturated-source"), async (relay: Relay) => {76 for (let i = 0; i < 501; i++) {77 const event = ev(owner, 1, "same timestamp " + i, [], timestamp);78 events.push(event);79 expect(relay.store.save(event, now())).toBe("");80 }81 });82 await claimed(destination, owner);83 const job: PullJob = {84 url: "wss://" + source,85 startedAt: timestamp,86 rounds: 0,87 stored: 0,88 skipped: 0,89 blobs: 0,90 failures: 0,91 filter: { authors: [getPublicKey(owner)], since: timestamp },92 };93 expect((await run("pull-query-saturated-destination", job)).more).toBe(true);94 const result = await run("pull-query-saturated-destination", job);95 expect(result).toEqual({ more: false, error: "" });96 expect(job.progress).toMatchObject({ mode: "query", status: "partial", pages: 1, partial: true });97 expect(job.progress?.warning).toMatch(/one-second window reached/);98 expect(job.stored).toBe(500);99 expect(events).toHaveLength(501);100 });101102 it("advances past a refused source and resumes the next source after JSON persistence", async () => {103 const owner = generateSecretKey();104 const source = "pull-refused-source.bind.ws";105 const destination = "pull-refused-destination.bind.ws";106 await claimed(source, owner);107 await rpc(source, owner, "setpolicy", { features: { sync: false } });108 const sourceClient = await WS.connect(source);109 const note = ev(owner, 1, "from the second source", [], now() - 10);110 expect((await sourceClient.ok(note)).ok).toBe(true);111 await claimed(destination, owner);112 const job = {113 id: "pull-test",114 kind: "pull",115 label: "pull",116 relays: ["wss://first.example", "wss://" + source],117 filter: { authors: [getPublicKey(owner)] },118 every: 0,119 createdAt: now(),120 nextRun: 0,121 running: true,122 startedAt: now(),123 rounds: 0,124 failures: 0,125 relayIndex: 0,126 cursor: 0,127 stored: 0,128 skipped: 0,129 blobs: 0,130 sent: 0,131 refused: 0,132 last: null,133 } as Job;134 let refusalRounds = 0;135 const connect = async (relay: Relay, url: string): Promise<PullSocket> => {136 if (url === "wss://first.example") return new RefusingSocket(refusalRounds++ > 0);137 return new Socket(await dial(relay, url));138 };139 const first = await runInDurableObject(env.RELAY.getByName("pull-refused-destination"), async (relay: Relay) => runPullSourceRound(relay, job, connect));140 expect(first).toEqual({ more: true, error: "" });141 expect(job.relayIndex).toBe(0);142 expect(job.pullSources?.[0]).toMatchObject({ mode: "query", status: "running" });143 const persisted = JSON.parse(JSON.stringify(job)) as Job;144 const refused = await runInDurableObject(env.RELAY.getByName("pull-refused-destination"), async (relay: Relay) => runPullSourceRound(relay, persisted, connect));145 expect(refused).toEqual({ more: true, error: "" });146 expect(persisted.relayIndex).toBe(1);147 expect(persisted.pullSources?.[0]).toMatchObject({ status: "refused", error: expect.stringContaining("auth-required") });148 const third = await runInDurableObject(env.RELAY.getByName("pull-refused-destination"), async (relay: Relay) => runPullSourceRound(relay, persisted));149 expect(third).toEqual({ more: true, error: "" });150 expect(persisted.relayIndex).toBe(1);151 expect(persisted.pullSources?.[1]).toMatchObject({ mode: "query", status: "running" });152 const fourth = await runInDurableObject(env.RELAY.getByName("pull-refused-destination"), async (relay: Relay) => runPullSourceRound(relay, persisted));153 expect(fourth).toEqual({ more: false, error: "" });154 expect(persisted.relayIndex).toBe(2);155 expect(persisted.pullSources?.[1]).toMatchObject({ status: "best-effort", pages: 1, stored: 1 });156 });157158 it("retries transient connection failures three times before advancing", async () => {159 const owner = generateSecretKey();160 const destination = "pull-retry-destination.bind.ws";161 await claimed(destination, owner);162 const job = {163 id: "pull-retry-test", kind: "pull", label: "pull", relays: ["wss://flaky.example", "wss://next.example"], filter: {}, every: 0,164 createdAt: now(), nextRun: 0, running: true, startedAt: now(), rounds: 0, failures: 0, relayIndex: 0, cursor: 0,165 stored: 0, skipped: 0, blobs: 0, sent: 0, refused: 0, last: null,166 } as Job;167 let attempts = 0;168 const connect = async (_relay: Relay, url: string): Promise<PullSocket> => {169 if (url === "wss://flaky.example") { attempts++; throw new Error("temporary network failure"); }170 return new ScriptedSocket([["NEG-ERR", "pull", "unsupported"]]);171 };172 for (let i = 0; i < 2; i++) {173 const r = await runInDurableObject(env.RELAY.getByName("pull-retry-destination"), async (relay: Relay) => runPullSourceRound(relay, job, connect));174 expect(r.error).toMatch(/temporary network failure/);175 expect(job.relayIndex).toBe(0);176 }177 const third = await runInDurableObject(env.RELAY.getByName("pull-retry-destination"), async (relay: Relay) => runPullSourceRound(relay, job, connect));178 expect(third).toEqual({ more: true, error: "" });179 expect(attempts).toBe(3);180 expect(job.pullSources?.[0]).toMatchObject({ status: "failed", failures: 3 });181 expect(job.relayIndex).toBe(1);182 });183184 it("skips invalid, wrong-author, and out-of-window source events", async () => {185 const owner = generateSecretKey();186 const stranger = generateSecretKey();187 const destination = "pull-validation-destination.bind.ws";188 await claimed(destination, owner);189 const good = ev(owner, 1, "in window", [], 100);190 const wrongAuthor = ev(stranger, 1, "wrong author", [], 100);191 const outOfWindow = ev(owner, 1, "too old", [], 50);192 const invalid = { ...good, content: "tampered" };193 let round = 0;194 const connect = async (): Promise<PullSocket> => round++ === 0195 ? new ScriptedSocket([["NEG-ERR", "pull", "unsupported"]])196 : new ScriptedSocket([["EVENT", "history", invalid], ["EVENT", "history", wrongAuthor], ["EVENT", "history", outOfWindow], ["EVENT", "history", good], ["EOSE", "history"]]);197 const job: PullJob = { url: "wss://scripted.example", startedAt: 100, rounds: 0, stored: 0, skipped: 0, blobs: 0, failures: 0, filter: { authors: [getPublicKey(owner)], since: 80 } };198 await runInDurableObject(env.RELAY.getByName("pull-validation-destination"), async (relay: Relay) => runPullRound(relay, job, connect));199 const result = await runInDurableObject(env.RELAY.getByName("pull-validation-destination"), async (relay: Relay) => runPullRound(relay, job, connect));200 expect(result).toEqual({ more: false, error: "" });201 expect(job.stored).toBe(1);202 expect(job.skipped).toBe(3);203 expect(job.progress).toMatchObject({ status: "partial", partial: true, pages: 1 });204 });205206 it("splits a saturated multi-second window and preserves the split after JSON cloning", async () => {207 const owner = generateSecretKey();208 const destination = "pull-split-destination.bind.ws";209 await claimed(destination, owner);210 const repeated = ev(owner, 1, "repeated", [], 50);211 let round = 0;212 const connect = async (): Promise<PullSocket> => {213 if (round++ === 0) return new ScriptedSocket([["NEG-ERR", "pull", "unsupported"]]);214 return new ScriptedSocket([...Array.from({ length: 500 }, () => ["EVENT", "history", repeated] as unknown[]), ["EOSE", "history"]]);215 };216 const job: PullJob = { url: "wss://scripted.example", startedAt: 100, rounds: 0, stored: 0, skipped: 0, blobs: 0, failures: 0, filter: { authors: [getPublicKey(owner)], since: 0 } };217 expect((await runInDurableObject(env.RELAY.getByName("pull-split-destination"), async (relay: Relay) => runPullRound(relay, job, connect))).more).toBe(true);218 expect((await runInDurableObject(env.RELAY.getByName("pull-split-destination"), async (relay: Relay) => runPullRound(relay, job, connect))).more).toBe(true);219 expect(job.progress?.windows).toEqual([{ since: 0, until: 50 }, { since: 51, until: 100 }]);220 const persisted = JSON.parse(JSON.stringify(job)) as PullJob;221 await runInDurableObject(env.RELAY.getByName("pull-split-destination"), async (relay: Relay) => runPullRound(relay, persisted, connect));222 expect(persisted.progress?.windows).toBeDefined();223 expect(persisted.progress?.pages).toBe(2);224 });225});