Shade/packages/shade-inbox-server/src/events.ts

/**
 * Inbox server event emitter.
 *
 * Mirrors `PrekeyServerEvents`. Emits structural facts only — no plaintext,
 * no signatures, no key material. Used by the observer dashboard and
 * operator metrics.
 */

export interface InboxServerEventBase {
  seq: number;
  timestamp: number;
}

export interface InboxServerEventMap {
  'inbox.address_registered': { address: string; signingKeyHash: string };
  'inbox.address_deleted': { address: string };
  'inbox.blob_stored': { address: string; msgId: string; bytes: number; ttlSeconds: number };
  'inbox.blob_idempotent_replay': { address: string; msgId: string };
  'inbox.blob_fetched': { address: string; count: number; bytes: number };
  'inbox.blob_acked': { address: string; msgId: string };
  'inbox.expired_purged': { count: number };
  'inbox.rate_limited': { route: string; key: string };
  'inbox.quota_rejected': { address: string; reason: 'address-quota' | 'sender-quota' | 'body-too-large' };
  // V4.7 — bridge presence transitions. Emitted on the 0↔1 boundary
  // across tracked transports for a given address. Long-poll is
  // intentionally NOT tracked: an LP client toggles in/out of a request
  // every few seconds, and the resulting flapping would dominate the
  // event stream. Push transports (WS, SSE) are also the only ones
  // where the ~50ms revoke window for `BroadcastChannel.removeMember`
  // matters — long-poll users are already on a slow path.
  'inbox.peer_connected': { address: string; bridgeKind: 'ws' | 'sse' };
  'inbox.peer_disconnected': {
    address: string;
    bridgeKind: 'ws' | 'sse';
    reason: 'closed' | 'error';
  };
}

export type InboxServerEventName = keyof InboxServerEventMap;

export type InboxServerEvent = {
  [K in InboxServerEventName]: InboxServerEventBase & { name: K; data: InboxServerEventMap[K] };
}[InboxServerEventName];

export type InboxServerEventListener = (event: InboxServerEvent) => void;

export class InboxServerEvents {
  private listeners = new Set<InboxServerEventListener>();
  private nextSeq = 1;
  private buffer: InboxServerEvent[] = [];
  private readonly maxBuffer: number;

  constructor(options: { bufferSize?: number } = {}) {
    this.maxBuffer = options.bufferSize ?? 1000;
  }

  on(listener: InboxServerEventListener): () => void {
    this.listeners.add(listener);
    return () => this.listeners.delete(listener);
  }

  off(listener: InboxServerEventListener): void {
    this.listeners.delete(listener);
  }

  emit<K extends InboxServerEventName>(name: K, data: InboxServerEventMap[K]): void {
    const event = {
      seq: this.nextSeq++,
      timestamp: Date.now(),
      name,
      data,
    } as InboxServerEvent;

    this.buffer.push(event);
    if (this.buffer.length > this.maxBuffer) this.buffer.shift();

    for (const listener of this.listeners) {
      try {
        listener(event);
      } catch (err) {
        console.error('[Shade] Inbox event listener threw:', err);
      }
    }
  }

  getBufferedSince(since: number): InboxServerEvent[] {
    return this.buffer.filter((e) => e.seq > since);
  }

  getRecent(n: number): InboxServerEvent[] {
    return this.buffer.slice(-n);
  }

  get currentSeq(): number {
    return this.nextSeq - 1;
  }
}

export async function shortHash(key: Uint8Array): Promise<string> {
  const buf = await globalThis.crypto.subtle.digest('SHA-256', key as unknown as ArrayBuffer);
  const arr = new Uint8Array(buf).slice(0, 8);
  return Array.from(arr, (b) => b.toString(16).padStart(2, '0')).join('');
}