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 const now = Math.floor(Date.now() / 1000);128 let manifestHash: string | null = null;129 if (input.path && input.path !== "/") {130 const address = Number.parseInt(input.path.slice(1), 16);131 const site = await env.DB.prepare(132 "SELECT current_manifest FROM sites WHERE address = ?",133 ).bind(address).first<{ current_manifest: string }>();134 if (!site) throw new ApiError(400, "notification target site not found");135 manifestHash = site.current_manifest;136 }137 const devices = await env.DB.prepare(138 `SELECT id, endpoint, p256dh, auth FROM notification_devices139 WHERE status = 'active' AND endpoint IS NOT NULL`,140 ).all<DeviceSubscriptionRow>();141 const inactive = await env.DB.prepare(142 "SELECT COUNT(*) AS count FROM notification_devices WHERE status <> 'active'",143 ).first<{ count: number }>();144 const eventId = crypto.randomUUID();145 const statements = [146 env.DB.prepare(147 `INSERT INTO notification_events148 (id, title, body, path, expected_manifest_hash, tag, created_at)149 VALUES (?, ?, ?, ?, ?, ?, ?)`,150 ).bind(151 eventId,152 input.title ?? "nzip",153 input.body,154 input.path ?? null,155 manifestHash,156 input.tag ?? null,157 now,158 ),159 ...devices.results.map((device) =>160 env.DB.prepare(161 `INSERT INTO notification_deliveries162 (event_id, device_id, status, attempts, next_attempt_at)163 VALUES (?, ?, 'pending', 0, ?)`,164 ).bind(eventId, device.id, now)165 ),166 ];167 await env.DB.batch(statements);168 ctx.waitUntil(drainNotifications(env, now));169 const response: NotifyResponse = {170 eventId,171 queuedDevices: devices.results.length,172 inactiveDevices: inactive?.count ?? 0,173 };174 console.log({175 event: "notify.event_accepted",176 eventId,177 queuedDevices: response.queuedDevices,178 });179 return json(response, 202);180}181182function classify(status: number): "sent" | "disabled" | "retry" | "failed" {183 if (status >= 200 && status < 300) return "sent";184 if (status === 404 || status === 410) return "disabled";185 if (status === 429 || status >= 500) return "retry";186 return "failed";187}188189export async function drainNotifications(190 env: Env,191 now = Math.floor(Date.now() / 1000),192): Promise<void> {193 if (env.NOTIFY_ENABLED !== "true") return;194 const candidates = await env.DB.prepare(195 `SELECT d.event_id, d.device_id AS id, d.attempts, e.title, e.body, e.path, e.tag,196 v.endpoint, v.p256dh, v.auth197 FROM notification_deliveries d198 JOIN notification_events e ON e.id = d.event_id199 JOIN notification_devices v ON v.id = d.device_id AND v.status = 'active'200 WHERE ((d.status IN ('pending','retry') AND COALESCE(d.next_attempt_at,0) <= ?)201 OR (d.status = 'sending' AND d.lease_expires_at < ?))202 ORDER BY COALESCE(d.next_attempt_at,0) LIMIT 25`,203 ).bind(now, now).all<DeliveryRow>();204 const owner = crypto.randomUUID();205 for (const row of candidates.results) {206 const attemptStartedAt = Math.floor(Date.now() / 1000);207 const claim = await env.DB.prepare(208 `UPDATE notification_deliveries SET status='sending', lease_owner=?, lease_expires_at=?209 WHERE event_id=? AND device_id=? AND210 ((status IN ('pending','retry') AND COALESCE(next_attempt_at,0) <= ?)211 OR (status='sending' AND lease_expires_at < ?))`,212 ).bind(213 owner,214 attemptStartedAt + 60,215 row.event_id,216 row.id,217 attemptStartedAt,218 attemptStartedAt,219 ).run();220 if (claim.meta.changes !== 1) continue;221 let outcome: "sent" | "disabled" | "retry" | "failed" = "retry";222 let errorClass = "network";223 let retryAfter: number | null = null;224 try {225 const result = await sendWebPush(env, {226 endpoint: row.endpoint,227 expirationTime: null,228 keys: { p256dh: row.p256dh, auth: row.auth },229 }, {230 eventId: row.event_id,231 title: row.title,232 body: row.body,233 ...(row.path ? { path: row.path } : {}),234 ...(row.tag ? { tag: row.tag } : {}),235 });236 outcome = classify(result.status);237 errorClass = `http_${result.status}`;238 retryAfter = result.retryAfter;239 } catch {240 outcome = "retry";241 }242 const attempts = row.attempts + 1;243 const completedAt = Math.floor(Date.now() / 1000);244 if (outcome === "disabled") {245 const terminal = await env.DB.prepare(246 `UPDATE notification_deliveries SET status='failed', attempts=?, last_error=?, lease_owner=NULL,247 lease_expires_at=NULL WHERE event_id=? AND device_id=? AND status='sending' AND lease_owner=?`,248 ).bind(attempts, errorClass, row.event_id, row.id, owner).run();249 if (terminal.meta.changes === 1) {250 await env.DB.prepare(251 `UPDATE notification_devices SET status='disabled', endpoint=NULL, endpoint_hash=NULL,252 p256dh=NULL, auth=NULL, last_error=? WHERE id=? AND status='active'`,253 ).bind(errorClass, row.id).run();254 }255 } else if (outcome === "sent") {256 const terminal = await env.DB.prepare(257 `UPDATE notification_deliveries SET status='sent', attempts=?, sent_at=?, last_error=NULL,258 lease_owner=NULL, lease_expires_at=NULL WHERE event_id=? AND device_id=? AND status='sending' AND lease_owner=?`,259 ).bind(attempts, completedAt, row.event_id, row.id, owner).run();260 if (terminal.meta.changes === 1) {261 await env.DB.prepare(262 "UPDATE notification_devices SET last_success_at=?, last_error=NULL WHERE id=? AND status='active'",263 ).bind(completedAt, row.id).run();264 }265 } else if (outcome === "retry" && attempts < 5) {266 const delay = retryAfter ??267 RETRY_DELAYS[Math.min(attempts - 1, RETRY_DELAYS.length - 1)];268 await env.DB.prepare(269 `UPDATE notification_deliveries SET status='retry', attempts=?, next_attempt_at=?, last_error=?,270 lease_owner=NULL, lease_expires_at=NULL WHERE event_id=? AND device_id=? AND status='sending' AND lease_owner=?`,271 ).bind(272 attempts,273 completedAt + delay,274 errorClass,275 row.event_id,276 row.id,277 owner,278 )279 .run();280 } else {281 await env.DB.prepare(282 `UPDATE notification_deliveries SET status='failed', attempts=?, last_error=?, lease_owner=NULL,283 lease_expires_at=NULL WHERE event_id=? AND device_id=? AND status='sending' AND lease_owner=?`,284 ).bind(attempts, errorClass, row.event_id, row.id, owner).run();285 }286 console.log({287 event: `notify.delivery_${outcome}`,288 eventId: row.event_id,289 deviceId: row.id,290 attempts,291 });292 }293}294295export async function validateClickTarget(296 env: Env,297 eventId: string,298): Promise<string | null> {299 const event = await env.DB.prepare(300 `SELECT path, expected_manifest_hash FROM notification_events301 WHERE id=? AND created_at >= ?`,302 ).bind(eventId, Math.floor(Date.now() / 1000) - 7 * 86400).first<{303 path: string | null;304 expected_manifest_hash: string | null;305 }>();306 if (!event?.path || event.path === "/") return event?.path ?? null;307 const site = await env.DB.prepare(308 "SELECT current_manifest FROM sites WHERE address=?",309 )310 .bind(Number.parseInt(event.path.slice(1), 16)).first<311 { current_manifest: string }312 >();313 return site?.current_manifest === event.expected_manifest_hash314 ? event.path315 : null;316}317318export async function pruneNotifications(319 env: Env,320 now = Math.floor(Date.now() / 1000),321): Promise<void> {322 await env.DB.batch([323 env.DB.prepare("DELETE FROM notification_events WHERE created_at < ?").bind(324 now - 7 * 86400,325 ),326 env.DB.prepare(327 "UPDATE notification_devices SET status='expired', claim_hash=NULL, pairing_code_hash=NULL, last_seen_at=? WHERE status='pending' AND pairing_expires_at < ?",328 ).bind(now, now),329 env.DB.prepare(330 "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 < ?",331 ).bind(now, now),332 env.DB.prepare(333 "DELETE FROM notification_devices WHERE status IN ('revoked','expired') AND last_seen_at < ?",334 ).bind(now - 7 * 86400),335 ]);336}