Mobile Push at 100M Devices: Benchmarking FCM, APNs, and Custom WebSockets

interface WsCapacityInput {
  readonly concurrentConnections: number;
  readonly bytesPerConnection: number;
  readonly nodeRamBytes: number;
  readonly nodeCostPerMonthUsd: number;
  readonly headroomPct: number;
  readonly lbCostPerMillionConnUsd: number;
}

export interface WsCapacityResult {
  readonly connectionsPerNode: number;
  readonly nodesRequired: number;
  readonly monthlyUsd: number;
  readonly perMillionConcurrentUsd: number;
}

export function sizeWebSocketFleet(i: WsCapacityInput): WsCapacityResult {
  if (i.concurrentConnections <= 0) throw new RangeError('concurrentConnections must be > 0');
  if (i.bytesPerConnection <= 0) throw new RangeError('bytesPerConnection must be > 0');
  if (i.headroomPct < 0) throw new RangeError('headroomPct must be >= 0');

  const usableBytes = i.nodeRamBytes * 0.75; // reserve for GC, page cache, TLS sessions
  const connectionsPerNode = Math.floor(usableBytes / i.bytesPerConnection);
  if (connectionsPerNode <= 0) throw new Error('node too small for per-connection footprint');

  const target = i.concurrentConnections * (1 + i.headroomPct / 100);
  const nodesRequired = Math.ceil(target / connectionsPerNode);
  const computeUsd = nodesRequired * i.nodeCostPerMonthUsd;
  const lbUsd = (i.concurrentConnections / 1_000_000) * i.lbCostPerMillionConnUsd;
  const monthlyUsd = computeUsd + lbUsd;

  return {
    connectionsPerNode,
    nodesRequired,
    monthlyUsd,
    perMillionConcurrentUsd: (monthlyUsd / i.concurrentConnections) * 1_000_000,
  };
}

// 100M sockets, 18KB RSS each, 32GB nodes at $212/mo, 30% headroom, $60/M NLB LCU
const plan = sizeWebSocketFleet({
  concurrentConnections: 100_000_000,
  bytesPerConnection: 18 * 1024,
  nodeRamBytes: 32 * 1024 ** 3,
  nodeCostPerMonthUsd: 212,
  headroomPct: 30,
  lbCostPerMillionConnUsd: 60,
});
// => nodesRequired: 470, monthlyUsd: ~$105,640, perMillionConcurrentUsd: ~$1,056

CREATE TABLE notification_outbox (
  id uuid PRIMARY KEY,
  idempotency_key text NOT NULL,
  device_id bigint NOT NULL,
  platform text NOT NULL,
  payload jsonb NOT NULL,
  priority text NOT NULL DEFAULT 'normal',
  status text NOT NULL DEFAULT 'pending',
  attempts int NOT NULL DEFAULT 0,
  next_attempt_at timestamptz NOT NULL DEFAULT now(),
  locked_by text,
  locked_until timestamptz,
  created_at timestamptz NOT NULL DEFAULT now(),
  UNIQUE (device_id, idempotency_key)
);

CREATE INDEX outbox_ready_idx
  ON notification_outbox (next_attempt_at, id)
  WHERE status = 'pending';

-- Claim a batch atomically; concurrent workers never collide.
UPDATE notification_outbox o
SET status = 'inflight',
    locked_by = $1,
    locked_until = now() + interval '30 seconds',
    attempts = o.attempts + 1
FROM (
  SELECT id FROM notification_outbox
  WHERE status = 'pending' AND next_attempt_at <= now()
  ORDER BY next_attempt_at
  LIMIT 500
  FOR UPDATE SKIP LOCKED
) c
WHERE o.id = c.id
RETURNING o.*;

export interface RetryPolicy {
  readonly maxAttempts: number;
  readonly baseDelayMs: number;
  readonly maxDelayMs: number;
}

export async function withRetry<T>(
  operation: (attempt: number, signal: AbortSignal) => Promise<T>,
  policy: RetryPolicy,
  deadline: number,
  sleep: (ms: number, signal: AbortSignal) => Promise<void>,
): Promise<T> {
  let lastError: unknown;
  for (let attempt = 1; attempt <= policy.maxAttempts; attempt++) {
    const remaining = deadline - Date.now();
    if (remaining <= 0) throw new Error('deadline exceeded before attempt');

    const controller = new AbortController();
    const timer = setTimeout(() => controller.abort(), remaining);
    try {
      return await operation(attempt, controller.signal);
    } catch (err) {
      lastError = err;
      if (err instanceof DeliveryError && (err.kind === 'permanent' || err.kind === 'expired')) throw err;
      if (attempt === policy.maxAttempts) break;

      const ceiling = Math.min(policy.maxDelayMs, policy.baseDelayMs * 2 ** (attempt - 1));
      const wait = err instanceof DeliveryError && err.retryAfterMs
        ? err.retryAfterMs
        : Math.random() * ceiling; // full jitter

      if (Date.now() + wait >= deadline) break;
      await sleep(wait, controller.signal);
    } finally {
      clearTimeout(timer);
    }
  }
  throw lastError;
}

export type DeliveryErrorKind = 'permanent' | 'transient' | 'throttled' | 'auth' | 'expired';

export class DeliveryError extends Error {
  constructor(
    readonly kind: DeliveryErrorKind,
    readonly provider: string,
    message: string,
    readonly retryAfterMs?: number,
  ) {
    super(message);
    this.name = 'DeliveryError';
  }
}

export interface PushProvider {
  readonly id: string;
  supports(n: OutboundNotification): boolean;
  send(n: OutboundNotification, signal: AbortSignal): Promise<DeliveryReceipt>;
}

export class ApnsProvider implements PushProvider {
  readonly id = 'apns';
  private session: http2.ClientHttp2Session | null = null;

  constructor(private readonly cfg: { host: string; topic: string; signingKey: () => Promise<string> }) {}

  supports(n: OutboundNotification): boolean {
    return n.platform === 'ios';
  }

  private getSession(): http2.ClientHttp2Session {
    const s = this.session;
    if (s && !s.closed && !s.destroyed) return s;
    const next = http2.connect(this.cfg.host);
    next.on('error', () => { this.session = null; });
    next.on('goaway', () => { next.close(); this.session = null; });
    this.session = next;
    return next;
  }

  async send(n: OutboundNotification, signal: AbortSignal): Promise<DeliveryReceipt> {
    const jwt = await this.cfg.signingKey();
    const headers: http2.OutgoingHttpHeaders = {
      ':method': 'POST',
      ':path': `/3/device/${n.deviceToken}`,
      authorization: `bearer ${jwt}`,
      'apns-topic': this.cfg.topic,
      'apns-push-type': 'alert',
      'apns-priority': n.priority === 'high' ? '10' : '5',
      'apns-expiration': String(Math.floor(Date.now() / 1000) + n.ttlSeconds),
      'content-type': 'application/json',
    };
    if (n.collapseKey) headers['apns-collapse-id'] = n.collapseKey;

    const body = JSON.stringify({
      aps: { alert: { title: n.data.title, body: n.data.body }, 'thread-id': n.data.threadId },
      ...n.data,
    });

    const req = this.getSession().request(headers, { signal });
    req.end(body);

    const res = await new Promise<{ status: number; payload: string; retryAfter?: string }>((resolve, reject) => {
      let status = 0;
      let payload = '';
      let retryAfter: string | undefined;
      req.setEncoding('utf8');
      req.on('response', (h) => {
        status = Number(h[':status']);
        if (h['retry-after']) retryAfter = String(h['retry-after']);
      });
      req.on('data', (chunk: string) => { payload += chunk; });
      req.on('end', () => resolve({ status, payload, retryAfter }));
      req.on('error', reject);
    });

    if (res.status === 200) return { id: n.id, provider: this.id, acceptedAt: Date.now() };

    const reason = extractApnsReason(res.payload);
    const kind: DeliveryErrorKind =
      res.status === 429 ? 'throttled'
      : res.status === 401 || res.status === 403 ? 'auth'
      : res.status >= 500 ? 'transient'
      : reason === 'BadDeviceToken' || reason === 'Unregistered' || reason === 'DeviceTokenNotForTopic' ? 'permanent'
      : 'transient';

    throw new DeliveryError(
      kind,
      this.id,
      `apns ${res.status} ${reason}`,
      res.retryAfter ? Number(res.retryAfter) * 1000 : undefined,
    );
  }
}

export class PushRouter {
  constructor(
    private readonly providers: readonly PushProvider[],
    private readonly breaker: CircuitBreaker,
  ) {}

  async route(n: OutboundNotification): Promise<DeliveryReceipt> {
    const chain = this.providers.filter((p) => p.supports(n) && this.breaker.isClosed(p.id));
    if (chain.length === 0) throw new Error(`no healthy provider for platform ${n.platform}`);

    let lastError: unknown;
    for (const provider of chain) {
      try {
        const receipt = await withRetry(
          (_attempt, signal) => provider.send(n, signal),
          { maxAttempts: 4, baseDelayMs: 100, maxDelayMs: 8_000 },
          n.deadline,
          sleep,
        );
        this.breaker.recordSuccess(provider.id);
        return receipt;
      } catch (err) {
        this.breaker.recordFailure(provider.id);
        lastError = err;
        if (err instanceof DeliveryError && err.kind === 'permanent') throw err;
      }
    }
    throw lastError;
  }
}

interface Connection {
  readonly socket: WebSocket;
  readonly deviceId: string;
  lastPongAt: number;
}

export class Gateway {
  private readonly byDevice = new Map<string, Connection>();
  private readonly MAX_BUFFERED = 1_000_000;

  constructor(private readonly wss: WebSocketServer) {
    this.wss.on('connection', (socket, req) => this.onConnection(socket, req));
    setInterval(() => this.sweep(), 30_000).unref();
  }

  private onConnection(socket: WebSocket, req: IncomingMessage): void {
    const deviceId = new URL(req.url ?? '/', 'ws://local').searchParams.get('deviceId');
    if (!deviceId) { socket.close(4400, 'missing device id'); return; }

    const conn: Connection = { socket, deviceId, lastPongAt: Date.now() };
    this.byDevice.set(deviceId, conn); // last-writer-wins: newest socket replaces stale one

    socket.on('pong', () => { conn.lastPongAt = Date.now(); });
    socket.on('close', () => {
      if (this.byDevice.get(deviceId) === conn) this.byDevice.delete(deviceId);
    });
    socket.on('error', () => socket.terminate());
  }

  private sweep(): void {
    const cutoff = Date.now() - 90_000;
    for (const conn of this.byDevice.values()) {
      if (conn.lastPongAt < cutoff) { conn.socket.terminate(); continue; }
      if (conn.socket.bufferedAmount > this.MAX_BUFFERED) { conn.socket.close(1013, 'backpressure'); continue; }
      conn.socket.ping();
    }
  }

  deliver(deviceId: string, frame: string): boolean {
    const conn = this.byDevice.get(deviceId);
    if (!conn || conn.socket.readyState !== WebSocket.OPEN) return false;
    if (conn.socket.bufferedAmount > this.MAX_BUFFERED) return false;
    conn.socket.send(frame);
    return true;
  }
}