From 4237e418f7332f9edfad30964d2d6a51aa6c1db2 Mon Sep 17 00:00:00 2001 From: Adam Soltys Date: Fri, 6 Mar 2026 23:22:12 -0800 Subject: add group registry for MLS push notifications Server now tracks MLS groups (synced from clients) and subscribes to kind 445 events on group relays, sending push to all members except the sender. Registry persisted to Redis across restarts. Co-Authored-By: Claude Opus 4.6 --- lib/dmNotifications.ts | 143 ++++++++++++++++++++++++++++++++++++++++++------- 1 file changed, 125 insertions(+), 18 deletions(-) (limited to 'lib') diff --git a/lib/dmNotifications.ts b/lib/dmNotifications.ts index 41591450..ca6b032a 100644 --- a/lib/dmNotifications.ts +++ b/lib/dmNotifications.ts @@ -9,56 +9,163 @@ if (config.vapid) { webpush.setVapidDetails(`mailto:${config.support}`, config.vapid.pk, config.vapid.sk); } -const handleDmEvent = async (event: any) => { - const pTag = event.tags.find((t: any) => t[0] === "p"); - l(`dm event ${event.id.slice(0, 8)} p=${pTag?.[1]?.slice(0, 12) || "none"}`); - if (!pTag) return; - - const recipientPubkey = pTag[1]; +// --------------------------------------------------------------------------- +// Push notification delivery +// --------------------------------------------------------------------------- +const sendPush = async (pubkey: string, url = "/messages") => { try { - const user = await getUser(recipientPubkey); + const user = await getUser(pubkey); if (!user?.id) return; const subscriptions = await db.sMembers(`${user.id}:subscriptions`); if (!subscriptions?.length) return; - l(`dm push to ${user.username} (${subscriptions.length} subs)`); + l(`push to ${user.username} (${subscriptions.length} subs)`); const payload = JSON.stringify({ title: "New message", body: "You have a new encrypted message", - url: "/messages", + url, }); for (const s of subscriptions as any) { webpush .sendNotification(JSON.parse(s as string), payload) .catch((e) => { - warn("dm push failed", e.message); + warn("push failed", e.message); db.sRem(`${user.id}:subscriptions`, s); }); } - } catch (e) { - // User not found for this pubkey - not a coinos user + } catch { + // User not found — not a coinos user + } +}; + +// --------------------------------------------------------------------------- +// Kind 1059 (gift wrap / invite) — notifies recipient via p tag +// --------------------------------------------------------------------------- + +const handleInviteEvent = async (event: any) => { + const pTag = event.tags.find((t: any) => t[0] === "p"); + if (!pTag) return; + await sendPush(pTag[1]); +}; + +// --------------------------------------------------------------------------- +// Kind 445 (MLS group message) — notifies all group members except sender +// --------------------------------------------------------------------------- + +// In-memory group registry: nostrGroupId → { members, relays } +interface GroupEntry { + members: string[]; + relays: string[]; +} + +const groupRegistry = new Map(); + +// Track active relay subscriptions for kind 445 +const activeGroupRelays = new Set(); +const groupRelaySubs: (() => void)[] = []; + +const handleGroupMessage = async (event: any) => { + const hTag = event.tags.find((t: any) => t[0] === "h"); + if (!hTag) return; + + const nostrGroupId = hTag[1]; + const group = groupRegistry.get(nostrGroupId); + if (!group) return; + + const sender = event.pubkey; + for (const member of group.members) { + if (member !== sender) sendPush(member); + } +}; + +const subscribeGroupRelay = (url: string) => { + if (activeGroupRelays.has(url)) return; + activeGroupRelays.add(url); + + Relay.connect(url) + .then((relay) => { + const sub = relay.subscribe( + [{ kinds: [445], since: Math.floor(Date.now() / 1000) }], + { onevent: handleGroupMessage }, + ); + groupRelaySubs.push(() => sub.close()); + l(`listening for group messages on ${url}`); + }) + .catch((e) => { + warn(`group listener failed ${url}`, e.message); + activeGroupRelays.delete(url); + setTimeout(() => subscribeGroupRelay(url), 5000); + }); +}; + +/** Called when a client syncs its groups. Updates registry and subscribes to new relays. */ +export const syncGroups = async (groups: { nostrGroupId: string; members: string[]; relays: string[] }[]) => { + for (const g of groups) { + groupRegistry.set(g.nostrGroupId, { members: g.members, relays: g.relays }); + for (const url of g.relays) subscribeGroupRelay(url); } + + l(`group registry: ${groupRegistry.size} groups, ${activeGroupRelays.size} relays`); }; -const subscribeRelay = (url: string) => { +// --------------------------------------------------------------------------- +// Kind 1059 relay subscriptions (invites / NIP-17 DMs) +// --------------------------------------------------------------------------- + +const subscribeInviteRelay = (url: string) => { Relay.connect(url) .then((relay) => { relay.subscribe([{ kinds: [1059], since: Math.floor(Date.now() / 1000) }], { - onevent: handleDmEvent, + onevent: handleInviteEvent, }); - l(`listening for DM notifications on ${url}`); + l(`listening for invites on ${url}`); }) .catch((e) => { - warn(`DM listener failed ${url}`, e.message); - setTimeout(() => subscribeRelay(url), 5000); + warn(`invite listener failed ${url}`, e.message); + setTimeout(() => subscribeInviteRelay(url), 5000); }); }; +// --------------------------------------------------------------------------- +// Boot: restore group registry from Redis, start listeners +// --------------------------------------------------------------------------- + +const REGISTRY_KEY = "mls:groupRegistry"; + +const saveRegistry = async () => { + const obj: Record = {}; + for (const [id, entry] of groupRegistry) obj[id] = entry; + await db.set(REGISTRY_KEY, JSON.stringify(obj)); +}; + +const loadRegistry = async () => { + try { + const raw = await db.get(REGISTRY_KEY); + if (!raw) return; + const obj = JSON.parse(raw as string); + for (const [id, entry] of Object.entries(obj) as [string, GroupEntry][]) { + groupRegistry.set(id, entry); + for (const url of entry.relays) subscribeGroupRelay(url); + } + l(`restored ${groupRegistry.size} groups from registry`); + } catch {} +}; + +/** Called from syncGroups to persist after updates */ +export const syncGroupsAndSave = async (groups: { nostrGroupId: string; members: string[]; relays: string[] }[]) => { + await syncGroups(groups); + await saveRegistry(); +}; + export const listenForDMs = () => { + // Subscribe to kind 1059 on configured relays const relays = new Set([config.nostr, ...config.relays.filter((r: string) => r.startsWith("wss://"))]); - for (const url of relays) subscribeRelay(url); + for (const url of relays) subscribeInviteRelay(url); + + // Restore persisted group registry and subscribe to kind 445 + loadRegistry(); }; -- cgit v1.2.3