1import { env, runInDurableObject } from "cloudflare:test";2import { describe, expect, it } from "vitest";3import { generateSecretKey } from "nostr-tools/pure";4import type { Relay } from "../../src/relay.ts";5import { KIND_PUSH_REGISTRATION } from "../../src/kinds.ts";6import { queuePush } from "../../src/push.ts";7import { ev, now, pk, rpc } from "../helpers/relay.ts";8import { WS } from "../helpers/ws.ts";910const callback = "https://push.example.com/hooks/device-1";1112async function enable(host: string, owner: Uint8Array) {13 await rpc(host, owner, "claim");14 await rpc(host, owner, "setpolicy", { features: { push: true }, pushCallbacks: ["https://push.example.com"] });15}1617async function trustAndCapture(name: string, capture: (req: Request) => Response | Promise<Response>) {18 await runInDurableObject(env.RELAY.getByName(name), async (r: Relay) => {19 Object.defineProperty(r, "pushCallbackOrigins", { value: ["https://push.example.com"], configurable: true });20 r.fetcher = async (url, init) => capture(new Request(String(url), init));21 });22}2324describe("NIP-9a push engine", () => {25 it.each([{ order: "older", offset: -1 }, { order: "newer", offset: 1 }])("delivers both payloads when the id-only registration is $order", async ({ order, offset }) => {26 const name = "push-engine-" + order;27 const host = name + ".bind.ws";28 const registeredAt = now() - 10;29 const owner = generateSecretKey();30 await enable(host, owner);31 const socket = await WS.connect(host);32 await socket.auth(owner, host);33 const seen: { url: string; body: string }[] = [];34 await trustAndCapture(name, async (req) => { seen.push({ url: req.url, body: await req.text() }); return new Response(null, { status: 204 }); });35 const registration = ev(owner, KIND_PUSH_REGISTRATION, "", [36 ["d", "one"], ["relay", "wss://" + host], ["filter", JSON.stringify({ kinds: [1] })], ["callback", callback], ["include_event"],37 ], registeredAt);38 expect((await socket.ok(registration)).ok).toBe(true);39 const note = ev(owner, 1, "hello");40 expect((await socket.ok(note)).ok).toBe(true);41 await runInDurableObject(env.RELAY.getByName(name), async (r: Relay) => { await r.alarm(); });42 expect(seen).toHaveLength(1);43 expect(seen[0].url).toBe(callback);44 expect(JSON.parse(seen[0].body)).toEqual({ id: note.id, relay: "wss://" + host + "/", event: JSON.parse(JSON.stringify(note)) });45 const idsOnly = ev(owner, KIND_PUSH_REGISTRATION, "", [46 ["d", "two"], ["relay", "wss://" + host], ["filter", JSON.stringify({ kinds: [1] })], ["callback", callback],47 ], registeredAt + offset);48 expect((await socket.ok(idsOnly)).ok).toBe(true);49 const second = ev(owner, 1, "id only");50 expect((await socket.ok(second)).ok).toBe(true);51 await runInDurableObject(env.RELAY.getByName(name), async (r: Relay) => { await r.alarm(); });52 expect(seen).toHaveLength(3);53 const deliveries = seen.map(({ url, body }) => ({ url, payload: JSON.parse(body) })).filter(({ payload }) => payload.id === second.id);54 expect(deliveries).toHaveLength(2);55 // Registration timestamps exercise both selection orders; notification56 // identity and payload content establish correctness, not arrival position.57 expect(deliveries).toEqual(expect.arrayContaining([58 { url: callback, payload: { id: second.id, relay: "wss://" + host + "/", event: JSON.parse(JSON.stringify(second)) } },59 { url: callback, payload: { id: second.id, relay: "wss://" + host + "/" } },60 ]));61 });6263 it("does not deliver ignored events, duplicate broadcasts, or events after member removal", async () => {64 const host = "push-auth.bind.ws";65 const owner = generateSecretKey();66 const member = generateSecretKey();67 await enable(host, owner);68 await rpc(host, owner, "setmember", pk(member));69 const socket = await WS.connect(host);70 const seen: Request[] = [];71 await trustAndCapture("push-auth", (req) => { seen.push(req); return new Response(null, { status: 204 }); });72 const registration = ev(member, KIND_PUSH_REGISTRATION, "", [73 ["d", "member"], ["relay", "wss://" + host], ["filter", JSON.stringify({ kinds: [1] })], ["ignore", JSON.stringify({ "#t": ["skip"] })], ["callback", callback],74 ]);75 await socket.auth(member, host);76 expect((await socket.ok(registration)).ok).toBe(true);77 const ignored = ev(owner, 1, "skip", [["t", "skip"]]);78 await socket.ok(ignored);79 await runInDurableObject(env.RELAY.getByName("push-auth"), async (r: Relay) => { r.broadcast(ignored); await r.alarm(); });80 expect(seen).toHaveLength(0);81 const queued = ev(owner, 1, "queued before removal");82 await socket.ok(queued);83 await rpc(host, owner, "removemember", pk(member));84 const after = ev(owner, 1, "after");85 await socket.ok(after);86 await rpc(host, owner, "setmember", pk(member));87 await runInDurableObject(env.RELAY.getByName("push-auth"), async (r: Relay) => { await r.alarm(); });88 expect(seen).toHaveLength(0);89 });9091 it("retries transient failures, removes exhausted work, and deletes registrations on 404", async () => {92 const host = "push-retry.bind.ws";93 const owner = generateSecretKey();94 await enable(host, owner);95 const socket = await WS.connect(host);96 await socket.auth(owner, host);97 let status = 500;98 const seen: Request[] = [];99 await trustAndCapture("push-retry", (req) => { seen.push(req); return new Response(null, { status }); });100 const registration = ev(owner, KIND_PUSH_REGISTRATION, "", [101 ["d", "retry"], ["relay", "wss://" + host], ["filter", JSON.stringify({ kinds: [1] })], ["callback", callback],102 ]);103 expect((await socket.ok(registration)).ok).toBe(true);104 const note = ev(owner, 1, "retry me");105 await socket.ok(note);106 for (let i = 0; i < 4; i++) await runInDurableObject(env.RELAY.getByName("push-retry"), async (r: Relay) => { r.sql.exec("UPDATE push_queue SET due=0"); await r.alarm(); });107 expect(seen).toHaveLength(4);108 await runInDurableObject(env.RELAY.getByName("push-retry"), async (r: Relay) => { expect(r.sql.exec("SELECT 1 FROM push_queue").toArray()).toHaveLength(0); });109 status = 404;110 const second = ev(owner, KIND_PUSH_REGISTRATION, "", [111 ["d", "gone"], ["relay", "wss://" + host], ["filter", JSON.stringify({ kinds: [1] })], ["callback", callback],112 ]);113 expect((await socket.ok(second)).ok).toBe(true);114 const last = ev(owner, 1, "gone");115 await socket.ok(last);116 await runInDurableObject(env.RELAY.getByName("push-retry"), async (r: Relay) => { r.sql.exec("UPDATE push_queue SET due=0"); await r.alarm(); });117 await runInDurableObject(env.RELAY.getByName("push-retry"), async (r: Relay) => { expect(r.sql.exec("SELECT 1 FROM events WHERE id=?", second.id).toArray()).toHaveLength(0); });118 });119120 it("bounds queued work and drops queued references when policy, visibility, expiry, or registrations change", async () => {121 const host = "push-bounds.bind.ws";122 const owner = generateSecretKey();123 await enable(host, owner);124 const seen: string[] = [];125 await trustAndCapture("push-bounds", async (req) => { seen.push(await req.text()); return new Response(null, { status: 204 }); });126 await runInDurableObject(env.RELAY.getByName("push-bounds"), async (r: Relay) => {127 const registration = ev(owner, KIND_PUSH_REGISTRATION, "", [["d", "bounded"], ["relay", "wss://" + host], ["filter", JSON.stringify({ kinds: [1] })], ["callback", callback]]);128 expect(r.store.save(registration, now())).toBe("");129 for (let i = 0; i < 300; i++) {130 const note = ev(owner, 1, "queued " + i);131 expect(r.store.save(note, now())).toBe("");132 queuePush(r, note);133 }134 expect(r.sql.exec("SELECT count(*) AS n FROM push_queue").one().n).toBe(256);135 queuePush(r, ev(owner, 1, "overflow"));136 expect(r.sql.exec("SELECT count(*) AS n FROM push_queue").one().n).toBe(256);137 r.sql.exec("DELETE FROM push_queue");138 const disabled = ev(owner, 1, "disabled");139 r.store.save(disabled, now());140 queuePush(r, disabled);141 r.settings.update({ features: { ...r.settings.policy.features, push: false } });142 await r.alarm();143 r.settings.update({ features: { ...r.settings.policy.features, push: true } });144 await r.alarm();145 expect(seen.some((x) => x.includes(disabled.id))).toBe(false);146 // A hidden event is removed from the delivery set at tick time.147 r.sql.exec("DELETE FROM push_queue");148 const hidden = ev(owner, 1, "hidden");149 r.store.save(hidden, now());150 queuePush(r, hidden);151 r.settings.setEvent(hidden.id, "hide", "test", now());152 await r.alarm();153 expect(seen.some((x) => x.includes(hidden.id))).toBe(false);154 });155 });156157 it("does not fetch after an exhausted attempt and handles a network failure", async () => {158 const host = "push-attempts.bind.ws";159 const owner = generateSecretKey();160 await enable(host, owner);161 let calls = 0;162 await trustAndCapture("push-attempts", () => { calls++; throw new Error("network down"); });163 await runInDurableObject(env.RELAY.getByName("push-attempts"), async (r: Relay) => {164 const registration = ev(owner, KIND_PUSH_REGISTRATION, "", [["d", "attempts"], ["relay", "wss://" + host], ["filter", JSON.stringify({ kinds: [1] })], ["callback", callback]]);165 const note = ev(owner, 1, "network");166 r.store.save(registration, now());167 r.store.save(note, now());168 queuePush(r, note);169 r.sql.exec("UPDATE push_queue SET attempts=3,due=0");170 await r.alarm();171 expect(calls).toBe(1);172 r.sql.exec("INSERT OR IGNORE INTO push_queue(registration_id,event_id,due,attempts,expires) VALUES(?,?,?,?,?)", registration.id, note.id, 0, 4, now() + 100);173 await r.alarm();174 expect(calls).toBe(1);175 });176 });177178 it("delivers private events only when the registration author is a party", async () => {179 const host = "push-private-events.bind.ws";180 const owner = generateSecretKey();181 const recipient = generateSecretKey();182 const stranger = generateSecretKey();183 await enable(host, owner);184 const socket = await WS.connect(host);185 await socket.auth(owner, host);186 const seen: string[] = [];187 await trustAndCapture("push-private-events", async (req) => { seen.push(await req.text()); return new Response(null, { status: 204 }); });188 // Sign the event first, then register an ID filter for the ID-only path.189 const dm = ev(owner, 4, "secret", [["p", pk(recipient)]]);190 const actual = ev(owner, KIND_PUSH_REGISTRATION, "", [["d", "private"], ["relay", "wss://" + host], ["filter", JSON.stringify({ ids: [dm.id] })], ["callback", callback]]);191 expect((await socket.ok(actual)).ok).toBe(true);192 await socket.ok(dm);193 await runInDurableObject(env.RELAY.getByName("push-private-events"), async (r: Relay) => { await r.alarm(); });194 expect(seen).toHaveLength(1);195 const unrelated = ev(owner, 4, "other", [["p", pk(stranger)]]);196 await socket.ok(unrelated);197 await runInDurableObject(env.RELAY.getByName("push-private-events"), async (r: Relay) => { await r.alarm(); });198 expect(seen).toHaveLength(1);199 });200});201202describe("push admission and callback isolation", () => {203 it("rejects unapproved and malformed registrations and permits replacement at the author cap", async () => {204 const host = "push-admission.bind.ws";205 const owner = generateSecretKey();206 const outsider = generateSecretKey();207 await enable(host, owner);208 const socket = await WS.connect(host);209 const registration = (d: string, url = callback, filter = "{}", relay = "wss://" + host + "/", time = Math.floor(Date.now() / 1000)) => ev(owner, KIND_PUSH_REGISTRATION, "", [["d", d], ["relay", relay], ["filter", filter], ["callback", url]], time);210 expect((await socket.ok(registration("no-auth"))).msg).toContain("auth-required:");211 await socket.auth(owner, host);212 expect((await socket.ok(registration("no-host-approval"))).ok).toBe(false);213 await trustAndCapture("push-admission", () => new Response(null, { status: 204 }));214 for (const bad of [registration("path", callback, "{}", "wss://" + host + "/wrong"), registration("filter", callback, '{"unknown":true}'), registration("search", callback, '{"search":"topic"}'), registration("port", "https://push.example.com:444/path")]) expect((await socket.ok(bad)).ok).toBe(false);215 const stranger = await WS.connect(host);216 await stranger.auth(outsider, host);217 expect((await stranger.ok(ev(outsider, KIND_PUSH_REGISTRATION, "", registration("stranger").tags))).msg).toContain("members");218 for (let i = 0; i < 4; i++) expect((await socket.ok(registration(String(i)))).ok).toBe(true);219 expect((await socket.ok(registration("fifth"))).ok).toBe(false);220 expect((await socket.ok(registration("0", callback + "-updated", "{}", "wss://" + host + "/", Math.floor(Date.now() / 1000) + 1))).ok).toBe(true);221 socket.ws.close(); stranger.ws.close();222 });223224 it("accepts another event while a callback is still waiting", async () => {225 const host = "push-slow.bind.ws";226 const owner = generateSecretKey();227 await enable(host, owner);228 const socket = await WS.connect(host);229 await socket.auth(owner, host);230 await trustAndCapture("push-slow", () => new Response(null, { status: 204 }));231 expect((await socket.ok(ev(owner, KIND_PUSH_REGISTRATION, "", [["d", "slow"], ["relay", "wss://" + host], ["filter", '{"kinds":[1]}'], ["callback", callback]]))).ok).toBe(true);232 await runInDurableObject(env.RELAY.getByName("push-slow"), async (r: Relay) => {233 let release!: () => void;234 let started!: () => void;235 const pending = new Promise<void>((resolve) => { release = resolve; });236 const fetching = new Promise<void>((resolve) => { started = resolve; });237 r.fetcher = async () => { started(); await pending; return new Response(null, { status: 204 }); };238 const conn = r.virtualConn(host, pk(owner));239 const first = ev(owner, 1, "first");240 expect((await r.acceptAny(first, conn)).ok).toBe(true);241 r.broadcast(first);242 const tick = r.alarm();243 await fetching;244 try { expect((await r.acceptAny(ev(owner, 1, "during callback"), conn)).ok).toBe(true); }245 finally { release(); await tick; }246 });247 });248});249250describe("push alarm durability", () => {251 it("aborts a stalled callback and persists a retry instead of waiting indefinitely", async () => {252 const host = "push-timeout.bind.ws";253 const owner = generateSecretKey();254 await enable(host, owner);255 await trustAndCapture("push-timeout", () => new Response(null, { status: 204 }));256 const socket = await WS.connect(host);257 await socket.auth(owner, host);258 expect((await socket.ok(ev(owner, KIND_PUSH_REGISTRATION, "", [["d", "timeout"], ["relay", "wss://" + host], ["filter", '{"kinds":[1]}'], ["callback", callback]]))).ok).toBe(true);259 await runInDurableObject(env.RELAY.getByName("push-timeout"), async (r: Relay) => {260 let aborted = false;261 r.fetcher = async (_url, init) => new Promise<Response>((_resolve, reject) => {262 expect(init?.redirect).toBe("manual");263 init?.signal?.addEventListener("abort", () => { aborted = true; reject(new Error("aborted")); }, { once: true });264 });265 const note = ev(owner, 1, "stalled");266 r.accept(note, null); r.broadcast(note);267 await r.alarm();268 expect(aborted).toBe(true);269 const job = r.sql.exec<{ attempts: number; due: number }>("SELECT attempts,due FROM push_queue WHERE event_id=?", note.id).one();270 expect(job.attempts).toBe(1);271 expect(job.due).toBeGreaterThan(Math.floor(Date.now() / 1000));272 });273 });274275 it("schedules work created later in an alarm and recreates callback tables after teardown", async () => {276 const host = "push-late-alarm.bind.ws";277 const owner = generateSecretKey();278 await enable(host, owner);279 await trustAndCapture("push-late-alarm", () => new Response(null, { status: 204 }));280 const socket = await WS.connect(host);281 await socket.auth(owner, host);282 expect((await socket.ok(ev(owner, KIND_PUSH_REGISTRATION, "", [["d", "late"], ["relay", "wss://" + host], ["filter", '{"kinds":[1]}'], ["callback", callback]]))).ok).toBe(true);283 await runInDurableObject(env.RELAY.getByName("push-late-alarm"), async (r: Relay) => {284 const publish = r.publishDiscovery;285 r.publishDiscovery = async () => {286 const note = ev(owner, 1, "late alarm event");287 if (r.accept(note, null).stored) r.broadcast(note);288 };289 try { await r.alarm(); }290 finally { r.publishDiscovery = publish; }291 expect(r.sql.exec("SELECT 1 FROM push_queue").toArray()).toHaveLength(1);292 expect(await r.storage.getAlarm()).toBeLessThan(Date.now() + 3000);293 await r.teardown();294 await r.alarm();295 expect(r.sql.exec("SELECT 1 FROM push_queue").toArray()).toHaveLength(0);296 });297 });298});