1// Jobs: owner backfill from the relays in a kind 10002, rebroadcast to2// other relays with a cursor, standing jobs on an interval, caps and3// removal. Relays on this host are dialled object to object.4import { env, runInDurableObject } from "cloudflare:test";5import { describe, it, expect } from "vitest";6import { generateSecretKey, type Event } from "nostr-tools/pure";7import type { Relay } from "../../src/relay.ts";8import type { Job } from "../../src/jobs.ts";9import { now, ev, pk, rpc, drive } from "../helpers/relay.ts";10import { WS } from "../helpers/ws.ts";1112// relay claims a name and posts the given events to it.13async function relay(host: string, owner: Uint8Array, events: Event[] = []) {14 await rpc(host, owner, "claim");15 const c = await WS.connect(host);16 for (const e of events) expect((await c.ok(e)).ok, e.id).toBe(true);17 return c;18}1920describe("owner backfill", () => {21 it("fetches the owner's own history from the relays in their kind 10002, and nothing else", async () => {22 const me = generateSecretKey();23 const stranger = generateSecretKey();24 const a = "hist-a.bind.ws";25 const b = "hist-b.bind.ws";26 const mine = [ev(me, 1, "on a", [], now() - 30), ev(me, 1, "also on a", [], now() - 20), ev(me, 30023, "an article", [["d", "x"]], now() - 10)];27 await relay(a, generateSecretKey(), [mine[0], mine[1], ev(stranger, 1, "not mine")]);28 await relay(b, generateSecretKey(), [mine[2], ev(stranger, 1, "also not mine")]);2930 const host = "hist.bind.ws";31 const c = await relay(host, me);32 expect((await rpc(host, me, "backfill")).status).toBe(400);33 expect((await c.ok(ev(me, 10002, "", [["r", "wss://" + a, "read"], ["r", "wss://" + b, "write"], ["r", "wss://" + host]]))).ok).toBe(true);3435 const started = await rpc(host, me, "backfill");36 expect(started.status).toBe(200);37 expect(started.result.label).toBe("backfill");38 expect(started.result.relays).toEqual(["wss://" + a, "wss://" + b]);39 expect(started.result.filter).toEqual({ authors: [pk(me)] });40 const jobs = await drive(host, me);41 const job = jobs.find((j) => j.id === started.result.id)!;42 expect(job.last?.error).toBe("");43 expect(job.last?.stored).toBe(3);44 expect(job.nextRun).toBe(0);45 const got = await c.req({ kinds: [1, 30023] });46 expect(got.map((e) => e.id).sort()).toEqual(mine.map((e) => e.id).sort());47 expect(got.every((e) => e.pubkey === pk(me))).toBe(true);48 // A member without the jobs action cannot start one; a bad list is refused.49 expect((await rpc(host, stranger, "backfill")).status).toBe(403);50 expect((await rpc(host, me, "backfill", ["https://" + a])).status).toBe(400);51 });52});5354describe("rebroadcast", () => {55 it("forwards matching events to every target, counts duplicates as sent, and moves the cursor", async () => {56 const owner = generateSecretKey();57 const src = "blast.bind.ws";58 const e1 = ev(owner, 1, "one", [], now() - 30);59 const e2 = ev(owner, 1, "two", [], now() - 20);60 const e7 = ev(owner, 7, "+", [["e", e1.id]], now() - 10);61 const c = await relay(src, owner, [e1, e2, e7]);62 const t1 = "target-a.bind.ws";63 const t2 = "target-b.bind.ws";64 const c1 = await relay(t1, generateSecretKey(), [e1]); // already there: a duplicate counts as sent65 const c2 = await relay(t2, generateSecretKey());6667 const bad = await rpc(src, owner, "addjob", { kind: "push", relays: ["wss://" + src], filter: { kinds: [1] } });68 expect(bad.error).toMatch(/itself/);69 const started = await rpc(src, owner, "addjob", { kind: "push", relays: ["wss://" + t1, "wss://" + t2], filter: { kinds: [1] } });70 expect(started.status, JSON.stringify(started)).toBe(200);71 let jobs = await drive(src, owner);72 let job = jobs.find((j) => j.id === started.result.id)!;73 expect(job.last?.error).toBe("");74 expect(job.last?.sent).toBe(4);75 expect(job.last?.refused).toBe(0);76 expect(job.cursor).toBeGreaterThan(0);77 expect((await c1.req({ kinds: [1, 7] })).map((e) => e.id).sort()).toEqual([e1.id, e2.id].sort());78 expect((await c2.req({ kinds: [1, 7] })).map((e) => e.id).sort()).toEqual([e1.id, e2.id].sort());7980 // Nothing on the source is stored by a push; a second run sends only what is new.81 const e3 = ev(owner, 1, "three");82 expect((await c.ok(e3)).ok).toBe(true);83 expect((await rpc(src, owner, "runjob", started.result.id)).result).toBe(true);84 jobs = await drive(src, owner);85 job = jobs.find((j) => j.id === started.result.id)!;86 expect(job.last?.sent).toBe(2);87 expect((await c2.req({ ids: [e3.id] })).length).toBe(1);8889 // A target that refuses counts refusals and does not stop the others.90 const closed = "target-c.bind.ws";91 const cOwner = generateSecretKey();92 await relay(closed, cOwner);93 await rpc(closed, cOwner, "setpolicy", { writes: "owner" });94 const again = await rpc(src, owner, "addjob", { kind: "push", relays: ["wss://" + closed], filter: { kinds: [1] } });95 jobs = await drive(src, owner);96 job = jobs.find((j) => j.id === again.result.id)!;97 expect(job.last?.refused).toBe(3);98 expect(job.last?.sent).toBe(0);99 });100101 it("never rebroadcasts private kinds from a members-only relay", async () => {102 const owner = generateSecretKey();103 const src = "quiet.bind.ws";104 const friend = generateSecretKey();105 const dm = ev(owner, 4, "secret", [["p", pk(friend)]]);106 const note = ev(owner, 1, "public");107 await relay(src, owner);108 await rpc(src, owner, "setpolicy", { reads: "members" });109 const target = "loud.bind.ws";110 const ct = await relay(target, generateSecretKey());111 const stub = env.RELAY.getByName("quiet");112 await runInDurableObject(stub, async (r: Relay) => {113 expect(r.accept(dm, null).stored).toBe(true);114 expect(r.accept(note, null).stored).toBe(true);115 });116 const refused = await rpc(src, owner, "addjob", { kind: "push", relays: ["wss://" + target], filter: { kinds: [4] } });117 expect(refused.error).toMatch(/private kinds/);118 const started = await rpc(src, owner, "addjob", { kind: "push", relays: ["wss://" + target] });119 const jobs = await drive(src, owner);120 const job = jobs.find((j) => j.id === started.result.id)!;121 // The note and the relay's own signed records go; the message does not.122 expect(job.last?.sent).toBeGreaterThanOrEqual(1);123 expect(job.last?.skipped).toBeGreaterThanOrEqual(1);124 expect((await ct.req({ kinds: [1, 4] })).map((e) => e.kind)).toEqual([1]);125 });126});127128describe("rebroadcast cursor safety", () => {129 it("preserves a failed target independently and resumes it after a successful target", async () => {130 const owner = generateSecretKey(), targetOwner = generateSecretKey();131 const src = "mixed-source.bind.ws", good = "mixed-good.bind.ws", bad = "mixed-bad.bind.ws";132 const events = Array.from({ length: 6 }, (_, i) => ev(owner, 1, "mixed " + i));133 await relay(src, owner, events);134 await relay(good, targetOwner);135 const blocked = await relay(bad, targetOwner);136 await rpc(bad, targetOwner, "setpolicy", { writes: "owner" });137 const added = await rpc(src, owner, "addjob", { kind: "push", relays: ["wss://" + good, "wss://" + bad], filter: { kinds: [1] } });138 let job = (await drive(src, owner)).find((j) => j.id === added.result.id)!;139 expect(job.targetCursors?.["wss://" + good]).toBeGreaterThan(0);140 expect(job.targetCursors?.["wss://" + bad]).toBe(0);141 expect(job.last?.error).toBeTruthy();142 await rpc(bad, targetOwner, "setpolicy", { writes: "open" });143 await rpc(src, owner, "runjob", job.id);144 job = (await drive(src, owner)).find((j) => j.id === added.result.id)!;145 expect(job.last?.sent).toBe(6);146 expect((await blocked.req({ kinds: [1] })).map((e) => e.id).sort()).toEqual(events.map((e) => e.id).sort());147 });148149 it("does not advance a target past an unprocessed refusal tail", async () => {150 const owner = generateSecretKey();151 const src = "cursor-tail.bind.ws", target = "cursor-tail-target.bind.ws";152 const events = Array.from({ length: 6 }, (_, i) => ev(owner, 7, "tail-" + i));153 await relay(src, owner, events);154 const targetOwner = generateSecretKey();155 await relay(target, targetOwner);156 await rpc(target, targetOwner, "setpolicy", { writes: "owner" });157 const started = await rpc(src, owner, "addjob", { kind: "push", relays: ["wss://" + target], filter: { kinds: [7] } });158 const jobs = await drive(src, owner);159 const job = jobs.find((j) => j.id === started.result.id)!;160 expect(job.targetCursors?.["wss://" + target]).toBe(0);161 expect((await (await WS.connect(target)).req({ kinds: [7] })).length).toBe(0);162 });163});164165describe("standing jobs", () => {166 it("runs a recurring pull on its interval, caps standing jobs, and can be removed", async () => {167 const owner = generateSecretKey();168 const src = "upstream.bind.ws";169 const e1 = ev(owner, 1, "first", [], now() - 10);170 const cs = await relay(src, owner, [e1]);171 const host = "mirror.bind.ws";172 const c = await relay(host, owner);173174 expect((await rpc(host, owner, "addjob", { kind: "pull", relays: ["wss://" + src], every: 2 })).status).toBe(400);175 const started = await rpc(host, owner, "addjob", { kind: "pull", relays: ["wss://" + src], every: 1 });176 expect(started.status).toBe(200);177 let jobs = await drive(host, owner);178 let job = jobs.find((j) => j.id === started.result.id)!;179 // The note plus the upstream relay's own signed records.180 expect(job.last?.stored).toBeGreaterThanOrEqual(1);181 expect(job.nextRun).toBeGreaterThan(now() + 3500);182 expect(job.nextRun).toBeLessThanOrEqual(now() + 3600);183 expect((await c.req({ ids: [e1.id] })).length).toBe(1);184 const firstFinish = job.last!.finishedAt;185 // The daily alarm is not scheduled past the next run.186 const stub = env.RELAY.getByName("mirror");187 await runInDurableObject(stub, async (_r: Relay, state) => {188 const at = await state.storage.getAlarm();189 expect(at).not.toBeNull();190 expect(at!).toBeLessThanOrEqual(job.nextRun * 1000 + 500);191 });192193 // Time passes: the interval elapses and the mirror picks up what is new.194 const e2 = ev(owner, 1, "second");195 expect((await cs.ok(e2)).ok).toBe(true);196 await runInDurableObject(stub, async (_r: Relay, state) => {197 const list = (await state.storage.get<Job[]>("jobs"))!;198 list.find((j) => j.id === started.result.id)!.nextRun = now() - 1;199 await state.storage.put("jobs", list);200 });201 await new Promise((r) => setTimeout(r, 1100));202 jobs = await drive(host, owner);203 job = jobs.find((j) => j.id === started.result.id)!;204 expect(job.last!.finishedAt).toBeGreaterThan(firstFinish);205 expect(job.last?.stored).toBeGreaterThanOrEqual(1);206 expect(job.last?.stored).toBeLessThan(5);207 expect((await c.req({ ids: [e2.id] })).length).toBe(1);208209 // At most five standing jobs; a once job still fits; removal frees the slot.210 for (let i = 0; i < 4; i++) expect((await rpc(host, owner, "addjob", { kind: "pull", relays: [`wss://other-${i}.bind.ws`], every: 24 })).status).toBe(200);211 const sixth = await rpc(host, owner, "addjob", { kind: "pull", relays: ["wss://other-9.bind.ws"], every: 24 });212 expect(sixth.status).toBe(409);213 expect(sixth.error).toMatch(/standing/);214 expect((await rpc(host, owner, "removejob", started.result.id)).result).toBe(true);215 expect((await rpc(host, owner, "removejob", started.result.id)).result).toBe(false);216 expect(((await rpc(host, owner, "listjobs")).result as Job[]).some((j) => j.id === started.result.id)).toBe(false);217 expect((await rpc(host, owner, "addjob", { kind: "pull", relays: ["wss://other-9.bind.ws"], every: 24 })).status).toBe(200);218 // The old one-pull view still answers.219 const st = (await rpc(host, owner, "pullstatus")).result;220 expect(st.last === null || typeof st.last.stored === "number").toBe(true);221 });222});