flo-bit.dev / atproto-notify
[READ-ONLY] Mirror of https://github.com/flo-bit/atproto-notify.
0
Fork 0

Configure Feed

Select the types of activity you want to include in your feed.

web pushes

author
Florian date (May 23, 2026, 11:53 AM +0200) commit 67f3353c parent 3f28f797
+818 -41
+13
apps/relay/migrations/0003_push.sql
··· 1 + -- Web push subscriptions (Phase 2). One row per browser PushSubscription; a 2 + -- user can have several (multiple devices/browsers). Telegram channels stay in 3 + -- the `channels` table; web push lives here because keys + endpoint don't fit 4 + -- the (did, platform) shape and there are many per user. 5 + 6 + CREATE TABLE push_subscriptions ( 7 + endpoint TEXT PRIMARY KEY, -- unique per subscription; the push service URL 8 + did TEXT NOT NULL, 9 + p256dh TEXT NOT NULL, -- client public key (base64url, uncompressed point) 10 + auth TEXT NOT NULL, -- client auth secret (base64url, 16 bytes) 11 + created_at INTEGER NOT NULL 12 + ); 13 + CREATE INDEX push_subs_by_did ON push_subscriptions (did);
+1
apps/relay/package.json
··· 10 10 "test:watch": "vitest", 11 11 "typecheck": "tsc -b", 12 12 "deploy": "wrangler deploy", 13 + "vapid:keygen": "node scripts/generate-vapid.js", 13 14 "db:migrate": "wrangler d1 migrations apply notifs-relay --remote", 14 15 "db:migrate:local": "wrangler d1 migrations apply notifs-relay --local" 15 16 },
+22
apps/relay/scripts/generate-vapid.js
··· 1 + // Generate a VAPID keypair for Web Push. Run: `pnpm vapid:keygen`. 2 + // 3 + // Prints three values: 4 + // VAPID_PUBLIC_KEY — base64url uncompressed point. Goes in the relay's 5 + // wrangler.toml [vars] AND the web app's VAPID_PUBLIC_KEY 6 + // config (it's the browser's applicationServerKey; public). 7 + // VAPID_PRIVATE_JWK — the private signing key. Set as a relay SECRET: 8 + // `wrangler secret put VAPID_PRIVATE_JWK` (never commit). 9 + // VAPID_SUBJECT — a mailto: or https: contact URL (edit before use). 10 + import { webcrypto as crypto } from 'node:crypto'; 11 + 12 + const kp = await crypto.subtle.generateKey({ name: 'ECDSA', namedCurve: 'P-256' }, true, [ 13 + 'sign', 14 + 'verify', 15 + ]); 16 + const privateJwk = await crypto.subtle.exportKey('jwk', kp.privateKey); 17 + const rawPublic = new Uint8Array(await crypto.subtle.exportKey('raw', kp.publicKey)); 18 + const b64url = (bytes) => Buffer.from(bytes).toString('base64url'); 19 + 20 + console.log('VAPID_PUBLIC_KEY=' + b64url(rawPublic)); 21 + console.log('VAPID_PRIVATE_JWK=' + JSON.stringify(privateJwk)); 22 + console.log('VAPID_SUBJECT=mailto:you@example.com');
+73
apps/relay/src/db/queries.ts
··· 436 436 const result = await db.prepare('DELETE FROM delivery_log WHERE created_at < ?').bind(beforeMs).run(); 437 437 return result.meta.changes ?? 0; 438 438 } 439 + 440 + // --------------------------------------------------------------------------- 441 + // push_subscriptions (web push) 442 + // --------------------------------------------------------------------------- 443 + 444 + export interface PushSubscriptionRow { 445 + endpoint: string; 446 + did: Did; 447 + p256dh: string; 448 + auth: string; 449 + created_at: number; 450 + } 451 + 452 + export interface UpsertPushSubscriptionInput { 453 + endpoint: string; 454 + did: Did; 455 + p256dh: string; 456 + auth: string; 457 + createdAt: number; 458 + } 459 + 460 + /** Insert/replace a subscription, keyed by its endpoint (re-subscribing refreshes the keys + owner). */ 461 + export async function upsertPushSubscription( 462 + db: D1Database, 463 + input: UpsertPushSubscriptionInput, 464 + ): Promise<void> { 465 + await db 466 + .prepare( 467 + 'INSERT OR REPLACE INTO push_subscriptions (endpoint, did, p256dh, auth, created_at) VALUES (?, ?, ?, ?, ?)', 468 + ) 469 + .bind(input.endpoint, input.did, input.p256dh, input.auth, input.createdAt) 470 + .run(); 471 + } 472 + 473 + export async function listPushSubscriptionsForDid( 474 + db: D1Database, 475 + did: Did, 476 + ): Promise<PushSubscriptionRow[]> { 477 + const { results } = await db 478 + .prepare('SELECT * FROM push_subscriptions WHERE did = ? ORDER BY created_at DESC') 479 + .bind(did) 480 + .all<PushSubscriptionRow>(); 481 + return results; 482 + } 483 + 484 + export function countPushSubscriptionsForDid(db: D1Database, did: Did): Promise<{ c: number } | null> { 485 + return db 486 + .prepare('SELECT COUNT(*) AS c FROM push_subscriptions WHERE did = ?') 487 + .bind(did) 488 + .first<{ c: number }>(); 489 + } 490 + 491 + /** Unregister a subscription owned by `did` (scoped so a user can't drop another's). */ 492 + export async function deletePushSubscriptionForDid( 493 + db: D1Database, 494 + did: Did, 495 + endpoint: string, 496 + ): Promise<boolean> { 497 + const result = await db 498 + .prepare('DELETE FROM push_subscriptions WHERE did = ? AND endpoint = ?') 499 + .bind(did, endpoint) 500 + .run(); 501 + return changed(result); 502 + } 503 + 504 + /** Reap a dead subscription by endpoint (push service returned 404/410). */ 505 + export async function deletePushSubscription(db: D1Database, endpoint: string): Promise<boolean> { 506 + const result = await db 507 + .prepare('DELETE FROM push_subscriptions WHERE endpoint = ?') 508 + .bind(endpoint) 509 + .run(); 510 + return changed(result); 511 + }
+37 -9
apps/relay/src/delivery/dispatcher.ts
··· 1 - import { deleteChannelByPlatformUser } from '../db/queries'; 1 + import { deleteChannelByPlatformUser, deletePushSubscription } from '../db/queries'; 2 2 import type { DispatchJob, Env } from '../env'; 3 3 4 4 import { ··· 7 7 sendMessage, 8 8 TelegramApiError, 9 9 } from './telegram'; 10 + import { sendWebPush, WebPushError } from './webpush'; 10 11 11 12 /** 12 13 * Queue consumer. Each message is an independent delivery; we ack on success and 13 - * on permanent failure (dead channel), and retry on transient failure so Queues' 14 - * built-in retry/backoff handles it. 14 + * on permanent failure (dead channel/subscription), and retry on transient 15 + * failure so Queues' built-in retry/backoff handles it. 15 16 */ 16 17 export async function handleQueue(batch: MessageBatch<DispatchJob>, env: Env): Promise<void> { 17 18 for (const message of batch.messages) { ··· 19 20 await dispatch(env, message.body); 20 21 message.ack(); 21 22 } catch (err) { 22 - if (err instanceof TelegramApiError && isDeadChannel(err)) { 23 - // The user blocked the bot or the chat is gone: reap the channel and 24 - // stop retrying this message. 25 - const { channel } = message.body; 26 - await deleteChannelByPlatformUser(env.DB, channel.platform, channel.platformUserId); 27 - console.error(`dispatch: dropping dead channel ${channel.platformUserId}: ${err.description}`); 23 + if (await reapIfDead(env, message.body, err)) { 28 24 message.ack(); 29 25 } else { 30 26 console.error('dispatch: transient failure, retrying', err); ··· 34 30 } 35 31 } 36 32 33 + /** If `err` means the target is permanently undeliverable, reap it and return true. */ 34 + async function reapIfDead(env: Env, job: DispatchJob, err: unknown): Promise<boolean> { 35 + const { channel } = job; 36 + if (channel.platform === 'telegram' && err instanceof TelegramApiError && isDeadChannel(err)) { 37 + // The user blocked the bot or the chat is gone. 38 + await deleteChannelByPlatformUser(env.DB, channel.platform, channel.platformUserId); 39 + console.error(`dispatch: dropping dead telegram channel ${channel.platformUserId}: ${err.description}`); 40 + return true; 41 + } 42 + if ( 43 + channel.platform === 'webpush' && 44 + err instanceof WebPushError && 45 + (err.statusCode === 404 || err.statusCode === 410) 46 + ) { 47 + // The push subscription expired or was unsubscribed. 48 + await deletePushSubscription(env.DB, channel.endpoint); 49 + console.error(`dispatch: dropping dead push subscription (${err.statusCode})`); 50 + return true; 51 + } 52 + return false; 53 + } 54 + 37 55 /** Telegram errors that mean the channel is permanently undeliverable. */ 38 56 function isDeadChannel(err: TelegramApiError): boolean { 39 57 if (err.errorCode === 403) { ··· 48 66 49 67 async function dispatch(env: Env, job: DispatchJob): Promise<void> { 50 68 if (job.kind === 'notification') { 69 + if (job.channel.platform === 'webpush') { 70 + await sendWebPush(env, job.channel, { 71 + title: job.title, 72 + body: job.body, 73 + uri: job.uri, 74 + senderDid: job.senderDid, 75 + }); 76 + return; 77 + } 78 + 51 79 const text = `*${escapeMd(job.title)}*\n${escapeMd(job.body)}`; 52 80 const replyMarkup: InlineKeyboardMarkup | undefined = 53 81 job.uri !== undefined ? { inline_keyboard: [[{ text: 'Open', url: job.uri }]] } : undefined;
+108
apps/relay/src/delivery/push-crypto.ts
··· 1 + // Low-level WebCrypto helpers for Web Push (RFC 8291 / 8188 / 8292). No 2 + // node:crypto, so this runs on Cloudflare Workers. Binary I/O is Uint8Array; 3 + // base64url is the unpadded URL-safe variant used throughout the Web Push specs. 4 + 5 + export function b64urlEncode(bytes: Uint8Array): string { 6 + let bin = ''; 7 + for (const b of bytes) bin += String.fromCharCode(b); 8 + return btoa(bin).replace(/\+/g, '-').replace(/\//g, '_').replace(/=+$/, ''); 9 + } 10 + 11 + export function b64urlDecode(s: string): Uint8Array { 12 + const padded = s.replace(/-/g, '+').replace(/_/g, '/'); 13 + const bin = atob(padded + '='.repeat((4 - (padded.length % 4)) % 4)); 14 + const out = new Uint8Array(bin.length); 15 + for (let i = 0; i < bin.length; i++) out[i] = bin.charCodeAt(i); 16 + return out; 17 + } 18 + 19 + const encoder = new TextEncoder(); 20 + export function utf8(s: string): Uint8Array { 21 + return encoder.encode(s); 22 + } 23 + 24 + export function concatBytes(...parts: Uint8Array[]): Uint8Array { 25 + const total = parts.reduce((n, p) => n + p.length, 0); 26 + const out = new Uint8Array(total); 27 + let off = 0; 28 + for (const p of parts) { 29 + out.set(p, off); 30 + off += p.length; 31 + } 32 + return out; 33 + } 34 + 35 + export function randomBytes(n: number): Uint8Array { 36 + return crypto.getRandomValues(new Uint8Array(n)); 37 + } 38 + 39 + /** HKDF (extract + expand) via WebCrypto; returns `length` bytes. */ 40 + export async function hkdf( 41 + salt: Uint8Array, 42 + ikm: Uint8Array, 43 + info: Uint8Array, 44 + length: number, 45 + ): Promise<Uint8Array> { 46 + const key = await crypto.subtle.importKey('raw', ikm, 'HKDF', false, ['deriveBits']); 47 + const bits = await crypto.subtle.deriveBits( 48 + { name: 'HKDF', hash: 'SHA-256', salt, info }, 49 + key, 50 + length * 8, 51 + ); 52 + return new Uint8Array(bits); 53 + } 54 + 55 + /** Generate an ephemeral P-256 ECDH keypair; returns the raw uncompressed public point (65 bytes). */ 56 + export async function generateEcdhKeypair(): Promise<{ publicKey: Uint8Array; privateKey: CryptoKey }> { 57 + const kp = (await crypto.subtle.generateKey({ name: 'ECDH', namedCurve: 'P-256' }, true, [ 58 + 'deriveBits', 59 + ])) as CryptoKeyPair; 60 + const raw = (await crypto.subtle.exportKey('raw', kp.publicKey)) as ArrayBuffer; 61 + return { publicKey: new Uint8Array(raw), privateKey: kp.privateKey }; 62 + } 63 + 64 + /** ECDH shared secret (the X coordinate, 32 bytes) between our private key and a raw peer public point. */ 65 + export async function ecdhSharedSecret( 66 + privateKey: CryptoKey, 67 + peerPublicRaw: Uint8Array, 68 + ): Promise<Uint8Array> { 69 + const peer = await crypto.subtle.importKey( 70 + 'raw', 71 + peerPublicRaw, 72 + { name: 'ECDH', namedCurve: 'P-256' }, 73 + false, 74 + [], 75 + ); 76 + // workerd uses the standard `public` field; @cloudflare/workers-types names it 77 + // `$public`, so cast to satisfy the compiler while passing the runtime-correct shape. 78 + const bits = await crypto.subtle.deriveBits( 79 + { name: 'ECDH', public: peer } as unknown as Parameters<typeof crypto.subtle.deriveBits>[0], 80 + privateKey, 81 + 256, 82 + ); 83 + return new Uint8Array(bits); 84 + } 85 + 86 + /** AES-128-GCM encrypt; WebCrypto appends the 16-byte auth tag to the ciphertext. */ 87 + export async function aesGcmEncrypt( 88 + key: Uint8Array, 89 + iv: Uint8Array, 90 + data: Uint8Array, 91 + ): Promise<Uint8Array> { 92 + const k = await crypto.subtle.importKey('raw', key, { name: 'AES-GCM' }, false, ['encrypt']); 93 + const ct = await crypto.subtle.encrypt({ name: 'AES-GCM', iv, tagLength: 128 }, k, data); 94 + return new Uint8Array(ct); 95 + } 96 + 97 + /** ECDSA P-256 / SHA-256 signature as raw r||s (64 bytes) — the JOSE ES256 form for VAPID JWTs. */ 98 + export async function ecdsaSign(privateJwk: JsonWebKey, data: Uint8Array): Promise<Uint8Array> { 99 + const key = await crypto.subtle.importKey( 100 + 'jwk', 101 + privateJwk, 102 + { name: 'ECDSA', namedCurve: 'P-256' }, 103 + false, 104 + ['sign'], 105 + ); 106 + const sig = await crypto.subtle.sign({ name: 'ECDSA', hash: 'SHA-256' }, key, data); 107 + return new Uint8Array(sig); 108 + }
+124
apps/relay/src/delivery/webpush.ts
··· 1 + // Web Push delivery: RFC 8291 (aes128gcm message encryption) + RFC 8292 (VAPID). 2 + // Pure WebCrypto via ./push-crypto, so it runs on Cloudflare Workers. 3 + import type { Env } from '../env'; 4 + 5 + import { 6 + aesGcmEncrypt, 7 + b64urlDecode, 8 + b64urlEncode, 9 + concatBytes, 10 + ecdhSharedSecret, 11 + ecdsaSign, 12 + generateEcdhKeypair, 13 + hkdf, 14 + randomBytes, 15 + utf8, 16 + } from './push-crypto'; 17 + 18 + /** A browser PushSubscription, flattened (keys base64url-encoded). */ 19 + export interface PushSubscriptionData { 20 + endpoint: string; 21 + p256dh: string; 22 + auth: string; 23 + } 24 + 25 + /** The JSON payload the service worker receives in its `push` handler. */ 26 + export interface PushPayload { 27 + title: string; 28 + body: string; 29 + uri?: string; 30 + senderDid: string; 31 + } 32 + 33 + /** Thrown on a non-2xx push response; `statusCode` 404/410 means the subscription is dead. */ 34 + export class WebPushError extends Error { 35 + readonly statusCode: number; 36 + constructor(statusCode: number, detail: string) { 37 + super(`web push failed: ${statusCode} ${detail}`); 38 + this.name = 'WebPushError'; 39 + this.statusCode = statusCode; 40 + } 41 + } 42 + 43 + const RECORD_SIZE = 4096; 44 + 45 + /** 46 + * Encrypt `plaintext` for a subscription's keys using aes128gcm (RFC 8291). 47 + * Returns the full message body: `salt(16) | rs(4) | idlen(1) | as_public(65) | ciphertext`. 48 + */ 49 + export async function encryptPayload( 50 + uaPublicRaw: Uint8Array, 51 + authSecret: Uint8Array, 52 + plaintext: Uint8Array, 53 + ): Promise<Uint8Array> { 54 + const as = await generateEcdhKeypair(); 55 + const ecdhSecret = await ecdhSharedSecret(as.privateKey, uaPublicRaw); 56 + 57 + // IKM = HKDF(salt = auth_secret, ikm = ecdh_secret, info = "WebPush: info\0"|ua|as, 32) 58 + const keyInfo = concatBytes(utf8('WebPush: info\0'), uaPublicRaw, as.publicKey); 59 + const ikm = await hkdf(authSecret, ecdhSecret, keyInfo, 32); 60 + 61 + const salt = randomBytes(16); 62 + const cek = await hkdf(salt, ikm, utf8('Content-Encoding: aes128gcm\0'), 16); 63 + const nonce = await hkdf(salt, ikm, utf8('Content-Encoding: nonce\0'), 12); 64 + 65 + // Single record: plaintext followed by the 0x02 last-record padding delimiter. 66 + const padded = concatBytes(plaintext, Uint8Array.of(0x02)); 67 + const ciphertext = await aesGcmEncrypt(cek, nonce, padded); 68 + 69 + const rs = new Uint8Array(4); 70 + new DataView(rs.buffer).setUint32(0, RECORD_SIZE, false); 71 + const header = concatBytes(salt, rs, Uint8Array.of(as.publicKey.length), as.publicKey); 72 + return concatBytes(header, ciphertext); 73 + } 74 + 75 + /** Build the `Authorization: vapid ...` header value for `endpoint` (RFC 8292). */ 76 + export async function vapidAuthHeader( 77 + endpoint: string, 78 + privateJwk: JsonWebKey, 79 + publicKey: string, 80 + subject: string, 81 + ): Promise<string> { 82 + const aud = new URL(endpoint).origin; 83 + const header = b64urlEncode(utf8(JSON.stringify({ typ: 'JWT', alg: 'ES256' }))); 84 + const exp = Math.floor(Date.now() / 1000) + 12 * 60 * 60; // ≤ 24h 85 + const payload = b64urlEncode(utf8(JSON.stringify({ aud, exp, sub: subject }))); 86 + const signingInput = `${header}.${payload}`; 87 + const sig = await ecdsaSign(privateJwk, utf8(signingInput)); 88 + const jwt = `${signingInput}.${b64urlEncode(sig)}`; 89 + return `vapid t=${jwt}, k=${publicKey}`; 90 + } 91 + 92 + /** Deliver one web push message. Throws {@link WebPushError} on a non-2xx response. */ 93 + export async function sendWebPush( 94 + env: Env, 95 + sub: PushSubscriptionData, 96 + payload: PushPayload, 97 + ): Promise<void> { 98 + const body = await encryptPayload( 99 + b64urlDecode(sub.p256dh), 100 + b64urlDecode(sub.auth), 101 + utf8(JSON.stringify(payload)), 102 + ); 103 + const privateJwk = JSON.parse(env.VAPID_PRIVATE_JWK) as JsonWebKey; 104 + const authorization = await vapidAuthHeader( 105 + sub.endpoint, 106 + privateJwk, 107 + env.VAPID_PUBLIC_KEY, 108 + env.VAPID_SUBJECT, 109 + ); 110 + 111 + const res = await fetch(sub.endpoint, { 112 + method: 'POST', 113 + headers: { 114 + Authorization: authorization, 115 + 'Content-Encoding': 'aes128gcm', 116 + 'Content-Type': 'application/octet-stream', 117 + TTL: '86400', 118 + }, 119 + body, 120 + }); 121 + if (!res.ok) { 122 + throw new WebPushError(res.status, await res.text().catch(() => '')); 123 + } 124 + }
+22 -3
apps/relay/src/env.ts
··· 1 1 import type { ServiceJwtVerifier } from '@atcute/xrpc-server/auth'; 2 2 3 - /** A delivery channel on a third-party platform. v1 supports Telegram only. */ 3 + /** A Telegram delivery channel. */ 4 4 export interface TelegramChannel { 5 5 platform: 'telegram'; 6 6 /** Telegram chat id (stored as `channels.platform_user_id`). */ 7 7 platformUserId: string; 8 8 } 9 + 10 + /** A web push delivery channel (a browser PushSubscription). */ 11 + export interface WebPushChannel { 12 + platform: 'webpush'; 13 + endpoint: string; 14 + p256dh: string; 15 + auth: string; 16 + } 17 + 18 + /** Where a notification can be delivered. */ 19 + export type DeliveryChannel = TelegramChannel | WebPushChannel; 9 20 10 21 /** 11 22 * Work item placed on `DISPATCH_QUEUE` and handled by the `queue` consumer. 12 - * Discriminated on `kind`. 23 + * Discriminated on `kind`. Notifications fan out to any channel; pending-request 24 + * prompts are Telegram-only (they use inline approve/deny buttons). 13 25 */ 14 26 export type DispatchJob = 15 27 | { 16 28 kind: 'notification'; 17 - channel: TelegramChannel; 29 + channel: DeliveryChannel; 18 30 title: string; 19 31 body: string; 20 32 uri?: string; ··· 48 60 TELEGRAM_BOT_TOKEN: string; 49 61 /** Shared secret embedded in the Telegram webhook path (secret). */ 50 62 TELEGRAM_WEBHOOK_SECRET: string; 63 + 64 + /** VAPID public key — base64url uncompressed point; also the browser's applicationServerKey (var). */ 65 + VAPID_PUBLIC_KEY: string; 66 + /** VAPID contact subject, e.g. "mailto:you@example.com" (var). */ 67 + VAPID_SUBJECT: string; 68 + /** VAPID private signing key as JWK JSON (secret). */ 69 + VAPID_PRIVATE_JWK: string; 51 70 } 52 71 53 72 /**
+7
apps/relay/src/rpc/entrypoint.ts
··· 3 3 import type { Did } from '@atcute/lexicons'; 4 4 import type { 5 5 NotifsRpc, 6 + PushSubscriptionInput, 6 7 ToolsAtmoNotifsDenyPending, 7 8 ToolsAtmoNotifsGetSettings, 8 9 ToolsAtmoNotifsGrant, ··· 62 63 } 63 64 getSettings(did: Did): Promise<ToolsAtmoNotifsGetSettings.$output> { 64 65 return ops.getSettings(this.env, did); 66 + } 67 + registerWebPush(did: Did, sub: PushSubscriptionInput) { 68 + return ops.registerWebPush(this.env, did, sub); 69 + } 70 + unregisterWebPush(did: Did, endpoint: string) { 71 + return ops.unregisterWebPush(this.env, did, endpoint); 65 72 } 66 73 }
+26
apps/relay/src/rpc/ops.ts
··· 6 6 import type { Did } from '@atcute/lexicons'; 7 7 8 8 import type { 9 + PushSubscriptionInput, 9 10 ToolsAtmoNotifsDenyPending, 10 11 ToolsAtmoNotifsGetSettings, 11 12 ToolsAtmoNotifsGrant, ··· 207 208 notifyPendingViaTelegram: (user?.notify_pending_via_telegram ?? 0) === 1, 208 209 }; 209 210 } 211 + 212 + export async function registerWebPush( 213 + env: Env, 214 + did: Did, 215 + sub: PushSubscriptionInput, 216 + ): Promise<{ registered: boolean }> { 217 + await q.ensureUser(env.DB, did, now()); 218 + await q.upsertPushSubscription(env.DB, { 219 + endpoint: sub.endpoint, 220 + did, 221 + p256dh: sub.p256dh, 222 + auth: sub.auth, 223 + createdAt: now(), 224 + }); 225 + return { registered: true }; 226 + } 227 + 228 + export async function unregisterWebPush( 229 + env: Env, 230 + did: Did, 231 + endpoint: string, 232 + ): Promise<{ unregistered: boolean }> { 233 + const unregistered = await q.deletePushSubscriptionForDid(env.DB, did, endpoint); 234 + return { unregistered }; 235 + }
+39 -22
apps/relay/src/xrpc/send.ts
··· 55 55 throw rateLimited(perDay.resetIn, 'Daily notification limit reached for this recipient'); 56 56 } 57 57 58 - // 4. No linked channels → accept but deliver to nobody. 59 - const channels = await q.listChannelsForDid(app.env.DB, recipient); 60 - if (channels.length === 0) { 58 + // 4. Collect delivery targets: Telegram channels + web push subscriptions. 59 + const telegramChannels = (await q.listChannelsForDid(app.env.DB, recipient)).filter( 60 + (channel) => channel.platform === 'telegram', 61 + ); 62 + const pushSubs = await q.listPushSubscriptionsForDid(app.env.DB, recipient); 63 + const deliveredCount = telegramChannels.length + pushSubs.length; 64 + 65 + // No targets → accept but deliver to nobody. 66 + if (deliveredCount === 0) { 61 67 await logDelivery(app, id, recipient, senderDid, input.title, 0); 62 68 return json({ id, delivered: 0 }); 63 69 } 64 70 65 - // 5. Enqueue one dispatch job per channel. 66 - const jobs = channels 67 - .filter((channel) => channel.platform === 'telegram') 68 - .map( 69 - (channel): { body: DispatchJob } => ({ 70 - body: { 71 - kind: 'notification', 72 - channel: { platform: 'telegram', platformUserId: channel.platform_user_id }, 73 - title: input.title, 74 - body: input.body, 75 - uri: input.uri, 76 - senderDid, 71 + // 5. Enqueue one dispatch job per target. 72 + const jobs: { body: DispatchJob }[] = [ 73 + ...telegramChannels.map((channel) => ({ 74 + body: { 75 + kind: 'notification' as const, 76 + channel: { platform: 'telegram' as const, platformUserId: channel.platform_user_id }, 77 + title: input.title, 78 + body: input.body, 79 + uri: input.uri, 80 + senderDid, 81 + }, 82 + })), 83 + ...pushSubs.map((sub) => ({ 84 + body: { 85 + kind: 'notification' as const, 86 + channel: { 87 + platform: 'webpush' as const, 88 + endpoint: sub.endpoint, 89 + p256dh: sub.p256dh, 90 + auth: sub.auth, 77 91 }, 78 - }), 79 - ); 80 - if (jobs.length > 0) { 81 - await app.env.DISPATCH_QUEUE.sendBatch(jobs); 82 - } 92 + title: input.title, 93 + body: input.body, 94 + uri: input.uri, 95 + senderDid, 96 + }, 97 + })), 98 + ]; 99 + await app.env.DISPATCH_QUEUE.sendBatch(jobs); 83 100 84 101 // 6. Record the delivery. 85 - await logDelivery(app, id, recipient, senderDid, input.title, channels.length); 86 - return json({ id, delivered: channels.length }); 102 + await logDelivery(app, id, recipient, senderDid, input.title, deliveredCount); 103 + return json({ id, delivered: deliveredCount }); 87 104 }, 88 105 }; 89 106 }
+117
apps/relay/test/webpush.test.ts
··· 1 + import type { Did } from '@atcute/lexicons'; 2 + import { env } from 'cloudflare:test'; 3 + import { expect, it } from 'vitest'; 4 + 5 + import * as q from '../src/db/queries'; 6 + import { 7 + b64urlDecode, 8 + b64urlEncode, 9 + concatBytes, 10 + ecdhSharedSecret, 11 + hkdf, 12 + utf8, 13 + } from '../src/delivery/push-crypto'; 14 + import { encryptPayload, vapidAuthHeader } from '../src/delivery/webpush'; 15 + import * as ops from '../src/rpc/ops'; 16 + 17 + it('encryptPayload produces a body the subscription keypair can decrypt (RFC 8291 round-trip)', async () => { 18 + // The "user agent" ECDH keypair stands in for the browser's subscription keys. 19 + const ua = await crypto.subtle.generateKey({ name: 'ECDH', namedCurve: 'P-256' }, true, [ 20 + 'deriveBits', 21 + ]); 22 + const uaPublic = new Uint8Array(await crypto.subtle.exportKey('raw', ua.publicKey)); 23 + const authSecret = crypto.getRandomValues(new Uint8Array(16)); 24 + 25 + const original = JSON.stringify({ title: 'Hi', body: 'there', uri: 'https://x.example' }); 26 + const body = await encryptPayload(uaPublic, authSecret, utf8(original)); 27 + 28 + // Parse the aes128gcm header: salt(16) | rs(4) | idlen(1) | as_public(idlen) | ciphertext. 29 + const salt = body.slice(0, 16); 30 + const idlen = body[20]!; 31 + const asPublic = body.slice(21, 21 + idlen); 32 + const ciphertext = body.slice(21 + idlen); 33 + expect(idlen).toBe(65); 34 + 35 + // Re-derive the key schedule from the UA side and decrypt. 36 + const ecdh = await ecdhSharedSecret(ua.privateKey, asPublic); 37 + const ikm = await hkdf( 38 + authSecret, 39 + ecdh, 40 + concatBytes(utf8('WebPush: info\0'), uaPublic, asPublic), 41 + 32, 42 + ); 43 + const cek = await hkdf(salt, ikm, utf8('Content-Encoding: aes128gcm\0'), 16); 44 + const nonce = await hkdf(salt, ikm, utf8('Content-Encoding: nonce\0'), 12); 45 + 46 + const key = await crypto.subtle.importKey('raw', cek, { name: 'AES-GCM' }, false, ['decrypt']); 47 + const padded = new Uint8Array( 48 + await crypto.subtle.decrypt({ name: 'AES-GCM', iv: nonce, tagLength: 128 }, key, ciphertext), 49 + ); 50 + 51 + expect(padded[padded.length - 1]).toBe(0x02); // last-record delimiter 52 + expect(new TextDecoder().decode(padded.slice(0, -1))).toBe(original); 53 + }); 54 + 55 + it('vapidAuthHeader signs a verifiable ES256 JWT with the right aud/sub', async () => { 56 + const kp = await crypto.subtle.generateKey({ name: 'ECDSA', namedCurve: 'P-256' }, true, [ 57 + 'sign', 58 + 'verify', 59 + ]); 60 + const privateJwk = await crypto.subtle.exportKey('jwk', kp.privateKey); 61 + const rawPublic = new Uint8Array(await crypto.subtle.exportKey('raw', kp.publicKey)); 62 + const publicKey = b64urlEncode(rawPublic); 63 + 64 + const header = await vapidAuthHeader( 65 + 'https://fcm.googleapis.com/fcm/send/abc', 66 + privateJwk, 67 + publicKey, 68 + 'mailto:t@example.com', 69 + ); 70 + 71 + const match = /^vapid t=([^,]+), k=(.+)$/.exec(header); 72 + expect(match).not.toBeNull(); 73 + const [, jwt, k] = match!; 74 + expect(k).toBe(publicKey); 75 + 76 + const [h64, p64, s64] = jwt!.split('.'); 77 + const payload = JSON.parse(new TextDecoder().decode(b64urlDecode(p64!))) as Record<string, unknown>; 78 + expect(payload.aud).toBe('https://fcm.googleapis.com'); 79 + expect(payload.sub).toBe('mailto:t@example.com'); 80 + expect(typeof payload.exp).toBe('number'); 81 + 82 + const verifyKey = await crypto.subtle.importKey( 83 + 'raw', 84 + rawPublic, 85 + { name: 'ECDSA', namedCurve: 'P-256' }, 86 + false, 87 + ['verify'], 88 + ); 89 + const ok = await crypto.subtle.verify( 90 + { name: 'ECDSA', hash: 'SHA-256' }, 91 + verifyKey, 92 + b64urlDecode(s64!), 93 + utf8(`${h64}.${p64}`), 94 + ); 95 + expect(ok).toBe(true); 96 + }); 97 + 98 + it('registerWebPush stores a subscription; unregisterWebPush removes it (scoped to the user)', async () => { 99 + const did = 'did:plc:pushuser' as Did; 100 + const sub = { endpoint: 'https://push.example/abc', p256dh: 'fakePub', auth: 'fakeAuth' }; 101 + 102 + expect(await ops.registerWebPush(env, did, sub)).toEqual({ registered: true }); 103 + let rows = await q.listPushSubscriptionsForDid(env.DB, did); 104 + expect(rows).toHaveLength(1); 105 + expect(rows[0]?.endpoint).toBe(sub.endpoint); 106 + 107 + // Another user can't drop it. 108 + expect(await ops.unregisterWebPush(env, 'did:plc:other' as Did, sub.endpoint)).toEqual({ 109 + unregistered: false, 110 + }); 111 + expect(await q.listPushSubscriptionsForDid(env.DB, did)).toHaveLength(1); 112 + 113 + // The owner can. 114 + expect(await ops.unregisterWebPush(env, did, sub.endpoint)).toEqual({ unregistered: true }); 115 + rows = await q.listPushSubscriptionsForDid(env.DB, did); 116 + expect(rows).toHaveLength(0); 117 + });
+5
apps/relay/wrangler.toml
··· 38 38 [vars] 39 39 RELAY_DID = "did:web:notifs.atmo.tools" 40 40 BOT_USERNAME = "atmo_notify_bot" # e.g. "atmonotifsbot" 41 + # Web push (VAPID). Generate with `pnpm vapid:keygen`. The public key is also the 42 + # browser applicationServerKey — paste the SAME value into apps/web VAPID_PUBLIC_KEY. 43 + VAPID_PUBLIC_KEY = "" 44 + VAPID_SUBJECT = "mailto:you@example.com" 41 45 42 46 # Secrets set via `wrangler secret put`: 43 47 # - TELEGRAM_BOT_TOKEN 44 48 # - TELEGRAM_WEBHOOK_SECRET 49 + # - VAPID_PRIVATE_JWK (the private JWK from `pnpm vapid:keygen`)
+9
apps/web/src/lib/config.ts
··· 13 13 * `atproto` (identity) is sufficient. 14 14 */ 15 15 export const OAUTH_SCOPE = 'atproto'; 16 + 17 + /** 18 + * VAPID public key (base64url, uncompressed point) — the browser's 19 + * applicationServerKey for web push. MUST equal the relay's `VAPID_PUBLIC_KEY`. 20 + * Generate once with `pnpm --filter @atmo/notifs-relay vapid:keygen`, then paste 21 + * the same value here and into the relay's `wrangler.toml`. Empty hides the push 22 + * controls (so the app still works before keys are configured). 23 + */ 24 + export const VAPID_PUBLIC_KEY = '';
+66
apps/web/src/lib/push.ts
··· 1 + // Browser-only web push helpers. Functions assume they run in the browser 2 + // (Settings click handlers / onMount); `pushSupported()` gates them for SSR and 3 + // unsupported browsers, and when no VAPID key is configured. 4 + import { VAPID_PUBLIC_KEY } from '$lib/config'; 5 + 6 + export interface FlatPushSubscription { 7 + endpoint: string; 8 + p256dh: string; 9 + auth: string; 10 + } 11 + 12 + export function pushSupported(): boolean { 13 + return ( 14 + typeof window !== 'undefined' && 15 + 'serviceWorker' in navigator && 16 + 'PushManager' in window && 17 + 'Notification' in window && 18 + VAPID_PUBLIC_KEY !== '' 19 + ); 20 + } 21 + 22 + function urlBase64ToUint8Array(base64: string): Uint8Array<ArrayBuffer> { 23 + const padded = (base64 + '='.repeat((4 - (base64.length % 4)) % 4)) 24 + .replace(/-/g, '+') 25 + .replace(/_/g, '/'); 26 + const raw = atob(padded); 27 + const out = new Uint8Array(raw.length); 28 + for (let i = 0; i < raw.length; i++) out[i] = raw.charCodeAt(i); 29 + return out; 30 + } 31 + 32 + function flatten(sub: PushSubscription): FlatPushSubscription { 33 + const keys = sub.toJSON().keys; 34 + return { endpoint: sub.endpoint, p256dh: keys?.p256dh ?? '', auth: keys?.auth ?? '' }; 35 + } 36 + 37 + export async function currentSubscription(): Promise<FlatPushSubscription | null> { 38 + if (!pushSupported()) return null; 39 + const reg = await navigator.serviceWorker.ready; 40 + const sub = await reg.pushManager.getSubscription(); 41 + return sub ? flatten(sub) : null; 42 + } 43 + 44 + /** Request permission + subscribe this browser; returns the flattened subscription. */ 45 + export async function subscribe(): Promise<FlatPushSubscription> { 46 + const permission = await Notification.requestPermission(); 47 + if (permission !== 'granted') { 48 + throw new Error('Notification permission was not granted'); 49 + } 50 + const reg = await navigator.serviceWorker.ready; 51 + const sub = await reg.pushManager.subscribe({ 52 + userVisibleOnly: true, 53 + applicationServerKey: urlBase64ToUint8Array(VAPID_PUBLIC_KEY) 54 + }); 55 + return flatten(sub); 56 + } 57 + 58 + /** Unsubscribe this browser; returns the endpoint that was removed (or null). */ 59 + export async function unsubscribe(): Promise<string | null> { 60 + const reg = await navigator.serviceWorker.ready; 61 + const sub = await reg.pushManager.getSubscription(); 62 + if (!sub) return null; 63 + const { endpoint } = sub; 64 + await sub.unsubscribe(); 65 + return endpoint; 66 + }
+11
apps/web/src/lib/remote/notifs.remote.ts
··· 54 54 export const unlinkTelegram = command(async () => { 55 55 await requireRelay().unlinkChannel({ platform: 'telegram' }); 56 56 }); 57 + 58 + export const registerPush = command( 59 + v.object({ endpoint: v.string(), p256dh: v.string(), auth: v.string() }), 60 + async (sub) => { 61 + await requireRelay().registerWebPush(sub); 62 + } 63 + ); 64 + 65 + export const unregisterPush = command(v.object({ endpoint: v.string() }), async ({ endpoint }) => { 66 + await requireRelay().unregisterWebPush(endpoint); 67 + });
+4 -1
apps/web/src/lib/server/relay.ts
··· 10 10 // `relayFor` throws a clear error rather than failing cryptically. 11 11 import type { Did } from '@atcute/lexicons'; 12 12 import type { 13 + PushSubscriptionInput, 13 14 ToolsAtmoNotifsDenyPending, 14 15 ToolsAtmoNotifsGrant, 15 16 ToolsAtmoNotifsLinkChannel, ··· 47 48 linkChannel: (input: ToolsAtmoNotifsLinkChannel.$input) => svc.linkChannel(did, input), 48 49 unlinkChannel: (input: ToolsAtmoNotifsUnlinkChannel.$input) => svc.unlinkChannel(did, input), 49 50 updateSettings: (input: ToolsAtmoNotifsUpdateSettings.$input) => 50 - svc.updateSettings(did, input) 51 + svc.updateSettings(did, input), 52 + registerWebPush: (sub: PushSubscriptionInput) => svc.registerWebPush(did, sub), 53 + unregisterWebPush: (endpoint: string) => svc.unregisterWebPush(did, endpoint) 51 54 }; 52 55 }
+87 -6
apps/web/src/routes/(app)/settings/+page.svelte
··· 1 1 <script lang="ts"> 2 + import { onMount } from 'svelte'; 2 3 import { invalidateAll } from '$app/navigation'; 3 4 import Icon from '$lib/components/Icon.svelte'; 4 5 import IOSToggle from '$lib/components/IOSToggle.svelte'; 5 6 import RelativeTime from '$lib/components/RelativeTime.svelte'; 6 - import { linkTelegram, setNotifyPending, unlinkTelegram } from '$lib/remote/notifs.remote'; 7 + import { currentSubscription, pushSupported, subscribe, unsubscribe } from '$lib/push'; 8 + import { 9 + linkTelegram, 10 + registerPush, 11 + setNotifyPending, 12 + unlinkTelegram, 13 + unregisterPush 14 + } from '$lib/remote/notifs.remote'; 7 15 import type { PageData } from './$types'; 8 16 9 17 let { data }: { data: PageData } = $props(); ··· 38 46 busy['link'] = false; 39 47 } 40 48 } 49 + 50 + // Web push is per-browser: check this device's subscription on mount. 51 + type PushState = 'loading' | 'unsupported' | 'on' | 'off'; 52 + let pushState = $state<PushState>('loading'); 53 + 54 + onMount(async () => { 55 + if (!pushSupported()) { 56 + pushState = 'unsupported'; 57 + return; 58 + } 59 + try { 60 + pushState = (await currentSubscription()) ? 'on' : 'off'; 61 + } catch { 62 + pushState = 'off'; 63 + } 64 + }); 65 + 66 + async function enablePush() { 67 + busy['push'] = true; 68 + errorMsg = ''; 69 + try { 70 + await registerPush(await subscribe()); 71 + pushState = 'on'; 72 + } catch (err) { 73 + errorMsg = err instanceof Error ? err.message : 'Could not enable push'; 74 + } finally { 75 + busy['push'] = false; 76 + } 77 + } 78 + 79 + async function disablePush() { 80 + busy['push'] = true; 81 + errorMsg = ''; 82 + try { 83 + const endpoint = await unsubscribe(); 84 + if (endpoint) await unregisterPush({ endpoint }); 85 + pushState = 'off'; 86 + } catch (err) { 87 + errorMsg = err instanceof Error ? err.message : 'Could not disable push'; 88 + } finally { 89 + busy['push'] = false; 90 + } 91 + } 41 92 </script> 42 93 43 94 <svelte:head><title>Settings · atmo.pub</title></svelte:head> ··· 99 150 {/if} 100 151 </div> 101 152 102 - <!-- Push (coming soon) --> 103 - <div class="flex items-center gap-3 p-4 opacity-60"> 153 + <!-- Push --> 154 + <div class="flex items-center gap-3 p-4" class:opacity-60={pushState === 'unsupported'}> 104 155 <div 105 - class="grid size-10 shrink-0 place-items-center rounded-full bg-surface-2 text-muted" 156 + class="grid size-10 shrink-0 place-items-center rounded-full {pushState === 'on' 157 + ? 'bg-accent-soft text-accent' 158 + : 'bg-surface-2 text-muted'}" 106 159 aria-hidden="true" 107 160 > 108 161 <Icon name="push" size={20} /> 109 162 </div> 110 163 <div class="min-w-0 flex-1"> 111 164 <div class="text-sm font-semibold text-fg">Push</div> 112 - <div class="text-xs text-muted">Native push on iOS, Android & web.</div> 165 + <div class="text-xs text-muted"> 166 + {#if pushState === 'unsupported'} 167 + Not available in this browser. 168 + {:else if pushState === 'on'} 169 + Enabled on this device. 170 + {:else} 171 + Get notifications in this browser, even when it's closed. 172 + {/if} 173 + </div> 113 174 </div> 114 - <span class="font-mono text-xs text-muted-2">Coming soon</span> 175 + {#if pushState === 'on'} 176 + <button 177 + class="rounded-md border border-line px-3 py-1.5 text-sm font-medium text-danger transition-colors hover:bg-danger/10 disabled:opacity-50" 178 + disabled={busy['push']} 179 + onclick={disablePush} 180 + > 181 + {busy['push'] ? '…' : 'Disable'} 182 + </button> 183 + {:else if pushState === 'off'} 184 + <button 185 + class="rounded-md bg-accent px-3 py-1.5 text-sm font-medium text-accent-fg transition-opacity hover:opacity-90 disabled:opacity-50" 186 + disabled={busy['push']} 187 + onclick={enablePush} 188 + > 189 + {busy['push'] ? 'Enabling…' : 'Enable'} 190 + </button> 191 + {:else if pushState === 'loading'} 192 + <span class="font-mono text-xs text-muted-2">…</span> 193 + {:else} 194 + <span class="font-mono text-xs text-muted-2">Unavailable</span> 195 + {/if} 115 196 </div> 116 197 </div> 117 198 </section>
+36
apps/web/src/service-worker.ts
··· 49 49 ); 50 50 } 51 51 }); 52 + 53 + // Web push: show the notification the relay delivered. 54 + sw.addEventListener('push', (event) => { 55 + let data: { title?: string; body?: string; uri?: string } = {}; 56 + try { 57 + data = event.data?.json() ?? {}; 58 + } catch { 59 + /* non-JSON payload — fall back to defaults */ 60 + } 61 + event.waitUntil( 62 + sw.registration.showNotification(data.title ?? 'atmo.pub', { 63 + body: data.body ?? '', 64 + icon: '/icon.svg', 65 + badge: '/icon.svg', 66 + data: { uri: data.uri } 67 + }) 68 + ); 69 + }); 70 + 71 + // Focus an existing tab (navigating it to the notification's link) or open one. 72 + sw.addEventListener('notificationclick', (event) => { 73 + event.notification.close(); 74 + const uri = (event.notification.data as { uri?: string } | null)?.uri; 75 + const target = uri && /^https?:/.test(uri) ? uri : new URL('/inbox', sw.location.origin).href; 76 + event.waitUntil( 77 + (async () => { 78 + const clients = await sw.clients.matchAll({ type: 'window', includeUncontrolled: true }); 79 + for (const client of clients) { 80 + await client.focus(); 81 + if (uri) await client.navigate(target); 82 + return; 83 + } 84 + await sw.clients.openWindow(target); 85 + })() 86 + ); 87 + });
+11
packages/lexicons/src/rpc.ts
··· 27 27 ToolsAtmoNotifsUpdateSettings, 28 28 } from './lexicons/index.js'; 29 29 30 + /** A browser PushSubscription, flattened (binding-only; no public lexicon). */ 31 + export interface PushSubscriptionInput { 32 + endpoint: string; 33 + p256dh: string; 34 + auth: string; 35 + } 36 + 30 37 export interface NotifsRpc { 31 38 grant(did: Did, input: ToolsAtmoNotifsGrant.$input): Promise<ToolsAtmoNotifsGrant.$output>; 32 39 revoke(did: Did, input: ToolsAtmoNotifsRevoke.$input): Promise<ToolsAtmoNotifsRevoke.$output>; ··· 54 61 listPending(did: Did): Promise<ToolsAtmoNotifsListPending.$output>; 55 62 listChannels(did: Did): Promise<ToolsAtmoNotifsListChannels.$output>; 56 63 getSettings(did: Did): Promise<ToolsAtmoNotifsGetSettings.$output>; 64 + 65 + // Web push (binding-only; no public lexicon). 66 + registerWebPush(did: Did, sub: PushSubscriptionInput): Promise<{ registered: boolean }>; 67 + unregisterWebPush(did: Did, endpoint: string): Promise<{ unregistered: boolean }>; 57 68 }