Something went wrong. Try again.
APNs relay for Prefect for iOS: pushes a flow run's pause to your phone
Something went wrong. Try again.
4.9 kB · 119 lines
TypeScript
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120import { createHash } from "node:crypto";import type { Environment, Store } from "./db.ts";import type { Prefect } from "./prefect.ts";import type { Relay } from "./relay.ts";
type Deps = { store: Store; prefect: Prefect; relay: Relay; basePath: string; status: () => { subscribed: boolean; apns: boolean }; now?: () => number;};
const ID = /^[A-Za-z0-9._:-]{1,80}$/;const TOKEN = /^[0-9a-f]{32,400}$/i;const ACCEPTED_FOR_MS = 5 * 60_000;
function json(status: number, body: unknown): Response { return Response.json(body, { status });}
function field(body: Record<string, unknown>, name: string, shape: RegExp): string | null { const value = body[name]; return typeof value === "string" && shape.test(value) ? value : null;}
async function bodyOf(req: Request): Promise<Record<string, unknown> | null> { try { const parsed: unknown = await req.json(); return typeof parsed === "object" && parsed !== null && !Array.isArray(parsed) ? (parsed as Record<string, unknown>) : null; } catch { return null; }}
function isEnvironment(value: unknown): value is Environment { return value === "production" || value === "development";}
/** * The device registry. Whoever can sign in to the Prefect server can register * a device for it: the relay holds no credential list of its own, it asks the * server whether it accepts the Authorization header it was shown. */export function handler(deps: Deps): (req: Request) => Promise<Response> { const now = deps.now ?? Date.now; const accepted = new Map<string, number>();
const authorized = async (req: Request): Promise<boolean> => { const header = req.headers.get("authorization"); const key = createHash("sha256").update(header ?? "").digest("hex"); const until = accepted.get(key); if (until && until > now()) return true; if (!(await deps.prefect.accepts(header))) return false; accepted.set(key, now() + ACCEPTED_FOR_MS); return true; };
return async (req) => { const url = new URL(req.url); if (!url.pathname.startsWith(deps.basePath)) return json(404, { detail: "not found" }); const path = url.pathname.slice(deps.basePath.length).replace(/\/+$/, "") || "/"; const parts = path.split("/").filter(Boolean);
if (req.method === "GET" && path === "/health") { return json(200, { ok: true, ...deps.status(), devices: deps.store.devices().length, waiting: deps.store.waits().length }); } if (parts[0] !== "devices") return json(404, { detail: "not found" });
try { if (!(await authorized(req))) return json(401, { detail: "the Prefect server refused these credentials" }); } catch { return json(502, { detail: "couldn't reach the Prefect server to check these credentials" }); }
if (req.method === "POST" && parts.length === 1) { const body = await bodyOf(req); const id = body && field(body, "id", ID); const token = body && field(body, "token", TOKEN); const serverId = body && field(body, "server_id", ID); if (!body || !id || !token || !serverId || !isEnvironment(body.environment)) { return json(422, { detail: "expected id, token, server_id and environment (production or development)" }); } const startToken = body.start_token == null ? null : field(body, "start_token", TOKEN); if (body.start_token != null && !startToken) return json(422, { detail: "start_token is not a token" }); deps.store.saveDevice({ id, token: token.toLowerCase(), startToken: startToken?.toLowerCase() ?? null, environment: body.environment, serverId }); return json(200, { id, live_activities: !!startToken }); }
const id = parts[1]; if (!id || !ID.test(id)) return json(404, { detail: "not found" });
if (req.method === "DELETE" && parts.length === 2) { deps.store.removeDevice(id); return new Response(null, { status: 204 }); }
const device = deps.store.device(id); if (!device) return json(404, { detail: "this device is not registered" });
if (req.method === "POST" && parts.length === 3 && parts[2] === "test") { const result = await deps.relay.test(device); return result.status === "sent" ? json(200, { sent: true }) : json(result.status === "gone" ? 410 : 502, { sent: false, detail: result.reason }); }
if (req.method === "PUT" && parts.length === 4 && parts[2] === "activities") { const runId = parts[3]; const body = await bodyOf(req); const token = body && field(body, "token", TOKEN); if (!runId || !ID.test(runId) || !token) return json(422, { detail: "expected a run id and a token" }); deps.store.saveActivity(id, runId, token.toLowerCase()); void deps.relay.activityRegistered(runId).catch(() => {}); return new Response(null, { status: 204 }); }
return json(404, { detail: "not found" }); };}