API Gateway WebSocket Fan-Out: Coalesce Pushes After Reconnect Storms

// ws/coalesce.ts
type Outbound = { topic: string; payload: unknown; ts: number };

const buffers = new Map<string, Map<string, Outbound>>(); // connId → topic → last

export function enqueue(connId: string, msg: Outbound, windowMs = 50) {
 let topics = buffers.get(connId);
 if (!topics) {
 topics = new Map();
 buffers.set(connId, topics);
 setTimeout(() => flush(connId), windowMs); // ✅ one timer per conn
 }
 topics.set(msg.topic, msg); // collapse duplicates by topic
}

async function flush(connId: string) {
 const topics = buffers.get(connId);
 buffers.delete(connId);
 if (!topics) return;
 const batch = [...topics.values()];
 await apiGw.postToConnection({
 ConnectionId: connId,
 Data: JSON.stringify({ type: "batch", items: batch }),
 });
}

// ws/catchup.ts
export async function onConnect(connId: string, userId: string) {
 const delay = 200 + Math.random() * 1800; // ✅ spread load
 await sleep(delay);
 const missed = await loadMissed(userId, since: lastAck(userId));
 // Cap catch-up size — rest goes to "refresh from snapshot"
 const slice = missed.slice(0, 100);
 for (const m of slice) enqueue(connId, m, 50);
 if (missed.length > 100) enqueue(connId, snapshotHint(userId), 0);
}

// ws/fanout.ts
import { Semaphore } from "./sem";

const postSem = new Semaphore(200); // global postToConnection concurrency

export async function fanout(connIds: string[], payload: Outbound) {
 const critical = payload.topic === "billing" || payload.topic === "security";
 if (!critical && postSem.pressure() > 0.8) {
 metrics.hit("FanoutShed");
 return; // ✅ shed presence/typing indicators
 }
 await Promise.all(
 connIds.map((id) =>
 postSem.run(() =>
 apiGw.postToConnection({ ConnectionId: id, Data: JSON.stringify(payload) })
 .catch((e) => {
 if (e.statusCode === 410) return registry.drop(id); // gone
 throw e;
 })
 )
 )
 );
}