1import type { NotifyRequest, NotifyResponse } from "../../shared/mod.ts";2import { ApiError } from "./api/errors.ts";3import type { Env } from "./env.ts";4import { json } from "./env.ts";5import { sendWebPush } from "./web_push.ts";67interface DeviceSubscriptionRow {8 id: string;9 endpoint: string;10 p256dh: string;11 auth: string;12}1314interface DeliveryRow extends DeviceSubscriptionRow {15 event_id: string;16 title: string;17 body: string;18 path: string | null;19 tag: string | null;20 attempts: number;21}2223const RETRY_DELAYS = [300, 900, 3600, 21600] as const;2425function scalarLength(value: string): number {26 return [...value].length;27}2829function boundedString(30 value: unknown,31 name: string,32 min: number,33 max: number,34): string {35 if (typeof value !== "string") {36 throw new ApiError(400, `${name} must be a string`);37 }38 const length = scalarLength(value);39 if (length < min || length > max) {40 throw new ApiError(400, `${name} must be ${min}-${max} characters`);41 }42 return value;43}4445function normalizePath(value: unknown): string | undefined {46 if (value === undefined) return undefined;47 const path = boundedString(value, "path", 1, 128);48 if (49 !path.startsWith("/") || path.startsWith("//") || path.includes("\\") ||50 /[\u0000-\u001f\u007f]/.test(path)51 ) throw new ApiError(400, "path must be a same-origin absolute path");52 let decoded: string;53 try {54 decoded = decodeURIComponent(path);55 } catch {56 throw new ApiError(400, "path has invalid encoding");57 }58 if (decoded !== path || !/^\/(?:[0-9a-f]{4})?$/.test(path)) {59 throw new ApiError(400, "path must be / or an existing nzip site path");60 }61 return path;62}6364export function validateNotifyRequest(value: unknown): NotifyRequest {65 if (!value || typeof value !== "object" || Array.isArray(value)) {66 throw new ApiError(400, "request body must be an object");67 }68 const record = value as Record<string, unknown>;69 const allowed = new Set(["title", "body", "path", "tag"]);70 if (Object.keys(record).some((key) => !allowed.has(key))) {71 throw new ApiError(400, "request contains unknown fields");72 }73 const body = boundedString(record.body, "body", 1, 240);74 const title = record.title === undefined75 ? undefined76 : boundedString(record.title, "title", 1, 80);77 const path = normalizePath(record.path);78 let tag: string | undefined;79 if (record.tag !== undefined) {80 tag = boundedString(record.tag, "tag", 1, 64);81 if (!/^[A-Za-z0-9._:-]+$/.test(tag)) {82 throw new ApiError(400, "tag is invalid");83 }84 }85 const request = {86 body,87 ...(title ? { title } : {}),88 ...(path ? { path } : {}),89 ...(tag ? { tag } : {}),90 };91 if (new TextEncoder().encode(JSON.stringify(request)).byteLength > 2048) {92 throw new ApiError(400, "notification payload is too large");93 }94 return request;95}9697async function readJson(request: Request): Promise<unknown> {98 if (99 request.headers.get("content-type")?.split(";", 1)[0] !== "application/json"100 ) {101 throw new ApiError(415, "content-type must be application/json");102 }103 const length = Number(request.headers.get("content-length") ?? "0");104 if (Number.isFinite(length) && length > 4096) {105 throw new ApiError(413, "request too large");106 }107 const text = await request.text();108 if (new TextEncoder().encode(text).byteLength > 4096) {109 throw new ApiError(413, "request too large");110 }111 try {112 return JSON.parse(text);113 } catch {114 throw new ApiError(400, "invalid JSON");115 }116}117118export async function handleNotifySend(119 request: Request,120 env: Env,121 ctx: ExecutionContext,122): Promise<Response> {123 if (env.NOTIFY_ENABLED !== "true") {124 throw new ApiError(503, "notifications are disabled");125 }126 const input = validateNotifyRequest(await readJson(request));127 return json(await queueNotification(env, ctx, input), 202);128}129130/**131 * Persist a validated event and its deliveries in the durable outbox, then start draining.132 * Callers must check `NOTIFY_ENABLED` and validate the request first.133 */134export async function queueNotification(135 env: Env,136 ctx: ExecutionContext,137 input: NotifyRequest,138): Promise<NotifyResponse> {139 const now = Math.floor(Date.now() / 1000);140 let manifestHash: string | null = null;141 if (input.path && input.path !== "/") {142 const address = Number.parseInt(input.path.slice(1), 16);143 const site = await env.DB.prepare(144 "SELECT current_manifest FROM sites WHERE address = ?",145 ).bind(address).first<{ current_manifest: string }>();146 if (!site) throw new ApiError(400, "notification target site not found");147 manifestHash = site.current_manifest;148 }149 const devices = await env.DB.prepare(150 `SELECT id, endpoint, p256dh, auth FROM notification_devices151 WHERE status = 'active' AND endpoint IS NOT NULL`,152 ).all<DeviceSubscriptionRow>();153 const inactive = await env.DB.prepare(154 "SELECT COUNT(*) AS count FROM notification_devices WHERE status <> 'active'",155 ).first<{ count: number }>();156 const eventId = crypto.randomUUID();157 const statements = [158 env.DB.prepare(159 `INSERT INTO notification_events160 (id, title, body, path, expected_manifest_hash, tag, created_at)161 VALUES (?, ?, ?, ?, ?, ?, ?)`,162 ).bind(163 eventId,164 input.title ?? "nzip",165 input.body,166 input.path ?? null,167 manifestHash,168 input.tag ?? null,169 now,170 ),171 ...devices.results.map((device) =>172 env.DB.prepare(173 `INSERT INTO notification_deliveries174 (event_id, device_id, status, attempts, next_attempt_at)175 VALUES (?, ?, 'pending', 0, ?)`,176 ).bind(eventId, device.id, now)177 ),178 ];179 await env.DB.batch(statements);180 ctx.waitUntil(drainNotifications(env, now));181 const response: NotifyResponse = {182 eventId,183 queuedDevices: devices.results.length,184 inactiveDevices: inactive?.count ?? 0,185 };186 console.log({187 event: "notify.event_accepted",188 eventId,189 queuedDevices: response.queuedDevices,190 });191 return response;192}193194function classify(status: number): "sent" | "disabled" | "retry" | "failed" {195 if (status >= 200 && status < 300) return "sent";196 if (status === 404 || status === 410) return "disabled";197 if (status === 429 || status >= 500) return "retry";198 return "failed";199}200201export async function drainNotifications(202 env: Env,203 now = Math.floor(Date.now() / 1000),204): Promise<void> {205 if (env.NOTIFY_ENABLED !== "true") return;206 const candidates = await env.DB.prepare(207 `SELECT d.event_id, d.device_id AS id, d.attempts, e.title, e.body, e.path, e.tag,208 v.endpoint, v.p256dh, v.auth209 FROM notification_deliveries d210 JOIN notification_events e ON e.id = d.event_id211 JOIN notification_devices v ON v.id = d.device_id AND v.status = 'active'212 WHERE ((d.status IN ('pending','retry') AND COALESCE(d.next_attempt_at,0) <= ?)213 OR (d.status = 'sending' AND d.lease_expires_at < ?))214 ORDER BY COALESCE(d.next_attempt_at,0) LIMIT 25`,215 ).bind(now, now).all<DeliveryRow>();216 const owner = crypto.randomUUID();217 for (const row of candidates.results) {218 const attemptStartedAt = Math.floor(Date.now() / 1000);219 const claim = await env.DB.prepare(220 `UPDATE notification_deliveries SET status='sending', lease_owner=?, lease_expires_at=?221 WHERE event_id=? AND device_id=? AND222 ((status IN ('pending','retry') AND COALESCE(next_attempt_at,0) <= ?)223 OR (status='sending' AND lease_expires_at < ?))`,224 ).bind(225 owner,226 attemptStartedAt + 60,227 row.event_id,228 row.id,229 attemptStartedAt,230 attemptStartedAt,231 ).run();232 if (claim.meta.changes !== 1) continue;233 let outcome: "sent" | "disabled" | "retry" | "failed" = "retry";234 let errorClass = "network";235 let retryAfter: number | null = null;236 try {237 const result = await sendWebPush(env, {238 endpoint: row.endpoint,239 expirationTime: null,240 keys: { p256dh: row.p256dh, auth: row.auth },241 }, {242 eventId: row.event_id,243 title: row.title,244 body: row.body,245 ...(row.path ? { path: row.path } : {}),246 ...(row.tag ? { tag: row.tag } : {}),247 });248 outcome = classify(result.status);249 errorClass = `http_${result.status}`;250 retryAfter = result.retryAfter;251 } catch {252 outcome = "retry";253 }254 const attempts = row.attempts + 1;255 const completedAt = Math.floor(Date.now() / 1000);256 if (outcome === "disabled") {257 const terminal = await env.DB.prepare(258 `UPDATE notification_deliveries SET status='failed', attempts=?, last_error=?, lease_owner=NULL,259 lease_expires_at=NULL WHERE event_id=? AND device_id=? AND status='sending' AND lease_owner=?`,260 ).bind(attempts, errorClass, row.event_id, row.id, owner).run();261 if (terminal.meta.changes === 1) {262 await env.DB.prepare(263 `UPDATE notification_devices SET status='disabled', endpoint=NULL, endpoint_hash=NULL,264 p256dh=NULL, auth=NULL, last_error=? WHERE id=? AND status='active'`,265 ).bind(errorClass, row.id).run();266 }267 } else if (outcome === "sent") {268 const terminal = await env.DB.prepare(269 `UPDATE notification_deliveries SET status='sent', attempts=?, sent_at=?, last_error=NULL,270 lease_owner=NULL, lease_expires_at=NULL WHERE event_id=? AND device_id=? AND status='sending' AND lease_owner=?`,271 ).bind(attempts, completedAt, row.event_id, row.id, owner).run();272 if (terminal.meta.changes === 1) {273 await env.DB.prepare(274 "UPDATE notification_devices SET last_success_at=?, last_error=NULL WHERE id=? AND status='active'",275 ).bind(completedAt, row.id).run();276 }277 } else if (outcome === "retry" && attempts < 5) {278 const delay = retryAfter ??279 RETRY_DELAYS[Math.min(attempts - 1, RETRY_DELAYS.length - 1)];280 await env.DB.prepare(281 `UPDATE notification_deliveries SET status='retry', attempts=?, next_attempt_at=?, last_error=?,282 lease_owner=NULL, lease_expires_at=NULL WHERE event_id=? AND device_id=? AND status='sending' AND lease_owner=?`,283 ).bind(284 attempts,285 completedAt + delay,286 errorClass,287 row.event_id,288 row.id,289 owner,290 )291 .run();292 } else {293 await env.DB.prepare(294 `UPDATE notification_deliveries SET status='failed', attempts=?, last_error=?, lease_owner=NULL,295 lease_expires_at=NULL WHERE event_id=? AND device_id=? AND status='sending' AND lease_owner=?`,296 ).bind(attempts, errorClass, row.event_id, row.id, owner).run();297 }298 console.log({299 event: `notify.delivery_${outcome}`,300 eventId: row.event_id,301 deviceId: row.id,302 attempts,303 });304 }305}306307export async function validateClickTarget(308 env: Env,309 eventId: string,310): Promise<string | null> {311 const event = await env.DB.prepare(312 `SELECT path, expected_manifest_hash FROM notification_events313 WHERE id=? AND created_at >= ?`,314 ).bind(eventId, Math.floor(Date.now() / 1000) - 7 * 86400).first<{315 path: string | null;316 expected_manifest_hash: string | null;317 }>();318 if (!event?.path || event.path === "/") return event?.path ?? null;319 const site = await env.DB.prepare(320 "SELECT current_manifest FROM sites WHERE address=?",321 )322 .bind(Number.parseInt(event.path.slice(1), 16)).first<323 { current_manifest: string }324 >();325 return site?.current_manifest === event.expected_manifest_hash326 ? event.path327 : null;328}329330export async function pruneNotifications(331 env: Env,332 now = Math.floor(Date.now() / 1000),333): Promise<void> {334 await env.DB.batch([335 env.DB.prepare("DELETE FROM notification_events WHERE created_at < ?").bind(336 now - 7 * 86400,337 ),338 env.DB.prepare(339 "UPDATE notification_devices SET status='expired', claim_hash=NULL, pairing_code_hash=NULL, last_seen_at=? WHERE status='pending' AND pairing_expires_at < ?",340 ).bind(now, now),341 env.DB.prepare(342 "UPDATE notification_devices SET status='expired', claim_hash=NULL, pairing_code_hash=NULL, last_seen_at=? WHERE status='approved' AND active_at IS NULL AND claim_expires_at < ?",343 ).bind(now, now),344 env.DB.prepare(345 "DELETE FROM notification_devices WHERE status IN ('revoked','expired') AND last_seen_at < ?",346 ).bind(now - 7 * 86400),347 ]);348}