Shade/packages/shade-transport-bridge/src/presence-bridge.ts

/**
 * V4.7 — presence subscription client.
 *
 * Consumes the SSE feed at `<base>/v1/bridge/presence?…` and fires
 * `onPresenceChange` whenever a watched address transitions
 * online/offline. Tracking is server-side: the inbox-server emits
 * presence events on the 0↔1 boundary across WS + SSE bridge
 * connections, and this client filters by the watcher's declared
 * address list.
 *
 * Threat model context: the typical consumer (Prism, password
 * managers, anything sender-key-broadcasting) wires this to
 * `BroadcastChannel.removeMember` so a clean WS/SSE close on a
 * paired-peer device revokes its sender-key membership within
 * ~50ms. Long-poll bridges are deliberately NOT tracked on the
 * server (see `inbox-server` `events.ts`); presence here is
 * push-transport only.
 *
 * Watched-list mutations (`addPeer` / `removePeer`) trigger a
 * reconnect with a fresh signed query so the server-side filter
 * reflects the new set. Mutations are expected to be rare (only on
 * pair / unpair, not on every message), so the brief reconnect gap
 * is acceptable.
 */

import type { CryptoProvider } from '@shade/core';
import { signPresenceQuery } from './auth.js';
import { BridgeError } from './errors.js';

export type PresenceVia = 'ws' | 'sse';

export interface PresenceChange {
  address: string;
  status: 'online' | 'offline';
  /** Server's wall-clock time (ms since epoch) when the change happened. */
  at: number;
  /** Which transport carried the connection. Absent on the initial snapshot. */
  via?: PresenceVia;
}

export interface PresenceBridgeOptions {
  /** Bridge base URL — same as `LongPollBridge` / `SseBridge`. */
  baseUrl: string;
  crypto: CryptoProvider;
  /** Watcher's Ed25519 signing key (the address must be a registered inbox). */
  signingPrivateKey: Uint8Array;
  /** Watcher's address (the registered inbox owner). */
  address: string;
  /** Override `fetch` (tests). */
  fetch?: typeof fetch;
  /** Initial reconnect backoff (ms). Default 250. */
  initialBackoffMs?: number;
  /** Max reconnect backoff (ms). Default 10_000. */
  maxBackoffMs?: number;
  /** Disable automatic reconnect. Default false. */
  disableAutoReconnect?: boolean;
}

export interface PresenceSubscribeOptions {
  /** Initial set of addresses to watch. May be empty. */
  watch: readonly string[];
  /** Fired whenever a watched address transitions, plus once per address on initial open. */
  onPresenceChange: (change: PresenceChange) => void | Promise<void>;
  /** Optional reconnect / parse error reporter. */
  onError?: (err: Error) => void;
}

export interface PresenceSubscription {
  /** Add an address to the watched set. Triggers a reconnect. */
  addPeer(address: string): Promise<void>;
  /** Remove an address from the watched set. Triggers a reconnect. */
  removePeer(address: string): Promise<void>;
  /** Snapshot of the currently-watched addresses. */
  watching(): readonly string[];
  /** Tear down. Idempotent. */
  unsubscribe(): Promise<void>;
}

const DEFAULT_INITIAL_BACKOFF = 250;
const DEFAULT_MAX_BACKOFF = 10_000;

export class PresenceBridge {
  private readonly fetchFn: typeof fetch;

  constructor(private readonly options: PresenceBridgeOptions) {
    const f = options.fetch ?? globalThis.fetch;
    this.fetchFn = f.bind(globalThis);
  }

  async subscribe(opts: PresenceSubscribeOptions): Promise<PresenceSubscription> {
    const session = new PresenceSession(this.options, this.fetchFn, opts);
    await session.start();
    return session;
  }
}

class PresenceSession implements PresenceSubscription {
  private watched: string[];
  private abortController: AbortController | null = null;
  private currentReader: ReadableStreamDefaultReader<Uint8Array> | null = null;
  private disposed = false;
  private readLoopPromise: Promise<void> | null = null;
  private readonly onPresenceChange: PresenceSubscribeOptions['onPresenceChange'];
  private readonly onError: NonNullable<PresenceSubscribeOptions['onError']>;
  private firstOpenResolve: (() => void) | null = null;
  private firstOpenReject: ((err: Error) => void) | null = null;
  private firstOpenSettled = false;

  constructor(
    private readonly options: PresenceBridgeOptions,
    private readonly fetchFn: typeof fetch,
    opts: PresenceSubscribeOptions,
  ) {
    this.watched = [...opts.watch];
    this.onPresenceChange = opts.onPresenceChange;
    this.onError =
      opts.onError ?? ((err) => console.warn('[shade-bridge:presence]', err.message));
  }

  watching(): readonly string[] {
    return [...this.watched];
  }

  async start(): Promise<void> {
    return this.openAndPump();
  }

  /**
   * Open one SSE connection and resolve once the first response has
   * been received (so that callers can `await subscribe()` and know
   * the connection is established before the first state change).
   * The read loop continues in the background.
   */
  private openAndPump(): Promise<void> {
    return new Promise<void>((resolve, reject) => {
      this.firstOpenSettled = false;
      this.firstOpenResolve = () => {
        if (this.firstOpenSettled) return;
        this.firstOpenSettled = true;
        resolve();
      };
      this.firstOpenReject = (err: Error) => {
        if (this.firstOpenSettled) return;
        this.firstOpenSettled = true;
        reject(err);
      };
      this.readLoopPromise = this.runLoop();
    });
  }

  private async runLoop(): Promise<void> {
    let backoff = this.options.initialBackoffMs ?? DEFAULT_INITIAL_BACKOFF;
    const maxBackoff = this.options.maxBackoffMs ?? DEFAULT_MAX_BACKOFF;
    let firstAttempt = true;
    while (!this.disposed) {
      try {
        await this.openOnce();
        if (firstAttempt) {
          firstAttempt = false;
          this.firstOpenResolve?.();
        }
        // Reset backoff on a successful open.
        backoff = this.options.initialBackoffMs ?? DEFAULT_INITIAL_BACKOFF;
        await this.consume();
      } catch (err) {
        if (this.disposed) return;
        if (firstAttempt) {
          // Failed before we ever got a 200 — surface to the caller of subscribe().
          this.firstOpenReject?.(err as Error);
          return;
        }
        this.onError(err as Error);
      }
      this.currentReader = null;
      if (this.disposed || this.options.disableAutoReconnect) return;
      await sleep(backoff);
      backoff = Math.min(backoff * 2, maxBackoff);
    }
  }

  private async openOnce(): Promise<void> {
    const qs = await signPresenceQuery({
      crypto: this.options.crypto,
      signingPrivateKey: this.options.signingPrivateKey,
      address: this.options.address,
      watched: this.watched,
    });
    const url = `${stripTrailingSlash(this.options.baseUrl)}/v1/bridge/presence?${qs.toString()}`;
    this.abortController = new AbortController();
    let res: Response;
    try {
      res = await this.fetchFn(url, {
        method: 'GET',
        headers: { accept: 'text/event-stream', 'cache-control': 'no-cache' },
        signal: this.abortController.signal,
      });
    } catch (err) {
      throw new BridgeError(`presence connect failed: ${(err as Error).message}`);
    }
    if (!res.ok) {
      throw new BridgeError(`presence connect failed: HTTP ${res.status}`, res.status);
    }
    if (!res.body) {
      throw new BridgeError('presence response has no body');
    }
    this.currentReader = res.body.getReader() as ReadableStreamDefaultReader<Uint8Array>;
  }

  private async consume(): Promise<void> {
    const reader = this.currentReader;
    if (!reader) return;
    const decoder = new TextDecoder();
    let buf = '';
    let dataLines: string[] = [];
    let eventName: string | null = null;
    while (true) {
      let chunk: Awaited<ReturnType<typeof reader.read>>;
      try {
        chunk = await reader.read();
      } catch (err) {
        // Reader cancelled (mutation / unsubscribe) — exit cleanly.
        if (this.disposed || (err as Error).name === 'AbortError') return;
        throw err;
      }
      if (chunk.done) return;
      buf += decoder.decode(chunk.value, { stream: true });
      let idx;
      while ((idx = buf.indexOf('\n')) !== -1) {
        const rawLine = buf.slice(0, idx);
        buf = buf.slice(idx + 1);
        const line = rawLine.endsWith('\r') ? rawLine.slice(0, -1) : rawLine;
        if (line === '') {
          if (dataLines.length > 0) {
            await this.dispatch(eventName, dataLines.join('\n'));
          }
          dataLines = [];
          eventName = null;
          continue;
        }
        if (line.startsWith(':')) continue;
        const colon = line.indexOf(':');
        const field = colon === -1 ? line : line.slice(0, colon);
        let val = colon === -1 ? '' : line.slice(colon + 1);
        if (val.startsWith(' ')) val = val.slice(1);
        if (field === 'data') dataLines.push(val);
        else if (field === 'event') eventName = val;
      }
    }
  }

  private async dispatch(name: string | null, data: string): Promise<void> {
    if (name !== null && name !== '' && name !== 'presence') return;
    let parsed: unknown;
    try {
      parsed = JSON.parse(data);
    } catch (err) {
      this.onError(new BridgeError(`malformed presence data: ${(err as Error).message}`));
      return;
    }
    const change = parsed as PresenceChange;
    if (
      typeof change.address !== 'string' ||
      (change.status !== 'online' && change.status !== 'offline') ||
      typeof change.at !== 'number'
    ) {
      this.onError(new BridgeError('presence frame missing required fields'));
      return;
    }
    try {
      await this.onPresenceChange(change);
    } catch (err) {
      this.onError(err as Error);
    }
  }

  async addPeer(address: string): Promise<void> {
    if (this.disposed) throw new BridgeError('PresenceBridge subscription disposed');
    if (this.watched.includes(address)) return;
    this.watched = [...this.watched, address];
    await this.reconnect();
  }

  async removePeer(address: string): Promise<void> {
    if (this.disposed) throw new BridgeError('PresenceBridge subscription disposed');
    if (!this.watched.includes(address)) return;
    this.watched = this.watched.filter((a) => a !== address);
    await this.reconnect();
  }

  /**
   * Tear down the current SSE connection so the run loop reopens with
   * the new watched list. Cancels via abort + reader.cancel — both are
   * tolerated by the consume() catch path.
   */
  private async reconnect(): Promise<void> {
    const reader = this.currentReader;
    this.currentReader = null;
    this.abortController?.abort();
    if (reader) {
      try {
        await reader.cancel();
      } catch {
        /* ignore */
      }
    }
  }

  async unsubscribe(): Promise<void> {
    if (this.disposed) return;
    this.disposed = true;
    const reader = this.currentReader;
    this.currentReader = null;
    this.abortController?.abort();
    if (reader) {
      try {
        await reader.cancel();
      } catch {
        /* ignore */
      }
    }
    if (this.readLoopPromise) {
      try {
        await this.readLoopPromise;
      } catch {
        /* ignore */
      }
    }
  }
}

function stripTrailingSlash(s: string): string {
  return s.endsWith('/') ? s.slice(0, -1) : s;
}

function sleep(ms: number): Promise<void> {
  return new Promise((r) => setTimeout(r, ms));
}