feat: stream private office screens

server/src/media/ice.ts

@@ -0,0 +1,100 @@
+import { createHmac, randomBytes } from "node:crypto";
+import {
+  ICE_CONFIG_PROTOCOL_VERSION,
+  type IceConfigGrant,
+  type IceConfigRequest,
+  type IceConfigUnavailable,
+  type StunIceServer,
+  type TurnIceServer,
+} from "../../../src/media/iceTypes.ts";
+import type { IceConfig } from "../config.ts";
+
+export type IceCredentialResult =
+  | { ok: true; value: IceConfigGrant }
+  | { ok: false; code: "unavailable" | "rate-limited"; value: IceConfigUnavailable };
+
+export interface IceCredentialProvider {
+  issue(request: IceConfigRequest, rateKey: string): IceCredentialResult;
+}
+
+export interface IceCredentialProviderOptions {
+  now?: () => number;
+  nonce?: () => string;
+}
+
+interface RateEntry { startedAtMs: number; count: number }
+const MAX_RATE_KEYS = 4_096;
+
+export function createIceCredentialProvider(
+  config: IceConfig,
+  options: IceCredentialProviderOptions = {},
+): IceCredentialProvider {
+  const now = options.now ?? Date.now;
+  const nonce = options.nonce ?? (() => randomBytes(18).toString("base64url"));
+  const rates = new Map<string, RateEntry>();
+  const windowMs = config.rateWindowSeconds * 1_000;
+
+  return {
+    issue(request, rateKey) {
+      const at = now();
+      if (!config.configured) return unavailable(request.requestId, windowMs, "unavailable");
+      let entry = rates.get(rateKey);
+      if (!entry && rates.size >= MAX_RATE_KEYS) {
+        for (const [key, value] of rates) {
+          if (at - value.startedAtMs >= windowMs) rates.delete(key);
+        }
+        if (rates.size >= MAX_RATE_KEYS) {
+          return unavailable(request.requestId, windowMs, "rate-limited");
+        }
+        entry = rates.get(rateKey);
+      }
+      if (!entry || at - entry.startedAtMs >= windowMs) {
+        rates.set(rateKey, { startedAtMs: at, count: 1 });
+      } else {
+        entry.count += 1;
+        if (entry.count > config.rateAttempts) {
+          return unavailable(request.requestId, Math.max(1_000, windowMs - (at - entry.startedAtMs)), "rate-limited");
+        }
+      }
+
+      const expiresAtSeconds = Math.floor(at / 1_000) + config.credentialTtlSeconds;
+      // Coturn REST convention: expiration timestamp, colon, opaque username.
+      // No auth subject, email, profile id, IP address, or stable browser id.
+      const username = `${expiresAtSeconds}:${nonce()}`;
+      const credential = createHmac("sha1", config.sharedSecret).update(username).digest("base64");
+      const stunUrls = config.urls.filter((url) => url.startsWith("stun:") || url.startsWith("stuns:"));
+      const turnUrls = config.urls.filter((url) => url.startsWith("turn:") || url.startsWith("turns:"));
+      const iceServers: Array<StunIceServer | TurnIceServer> = [];
+      if (stunUrls.length > 0) iceServers.push({ urls: stunUrls });
+      iceServers.push({ urls: turnUrls, username, credential, credentialType: "password" });
+      return {
+        ok: true,
+        value: {
+          type: "ice-config-grant",
+          protocolVersion: ICE_CONFIG_PROTOCOL_VERSION,
+          requestId: request.requestId,
+          issuedAtMs: at,
+          expiresAtMs: expiresAtSeconds * 1_000,
+          iceServers,
+        },
+      };
+    },
+  };
+}
+
+function unavailable(
+  requestId: string,
+  retryAfterMs: number,
+  code: "unavailable" | "rate-limited",
+): IceCredentialResult {
+  return {
+    ok: false,
+    code,
+    value: {
+      type: "ice-config-unavailable",
+      protocolVersion: ICE_CONFIG_PROTOCOL_VERSION,
+      requestId,
+      retryAfterMs: Math.min(3_600_000, Math.max(1_000, Math.ceil(retryAfterMs))),
+    },
+  };
+}

server/src/media/index.ts

@@ -6,3 +6,9 @@ export {
   type MediaSignalServiceOptions,
 } from "./service.ts";
 export { bundledMediaOffice, officeHasMediaBinding } from "./bindings.ts";
+export {
+  createIceCredentialProvider,
+  type IceCredentialProvider,
+  type IceCredentialProviderOptions,
+  type IceCredentialResult,
+} from "./ice.ts";

server/src/media/service.ts

@@ -63,6 +63,7 @@ interface Participant {
   rateStartedAt: number;
   rateCount: number;
   queue: ScreenSharePeerMessage[];
+  queueOverflowed: boolean;
   listeners: Set<(message: ScreenSharePeerMessage) => void>;
 }
 
@@ -121,6 +122,7 @@ export function createMediaSignalService(options: MediaSignalServiceOptions = {}
       // snapshot, so dropping the oldest relay is safer than letting a sender
       // evict another participant by filling its queue.
       participant.queue.shift();
+      participant.queueOverflowed = true;
     }
     participant.queue.push(message);
   }
@@ -207,10 +209,9 @@ export function createMediaSignalService(options: MediaSignalServiceOptions = {}
     sequence: number,
   ): MediaSignalResult<null> {
     if (!sameBinding(session.binding, binding)) return { ok: false, code: "unauthorized", message: "Media binding does not match the grant." };
-    if (!Number.isSafeInteger(sequence) || sequence <= participant.lastClientSequence) {
+    if (!clientSequenceHasSuccessor(sequence) || sequence <= participant.lastClientSequence) {
       return { ok: false, code: "conflict", message: "Media client sequence is not monotonic." };
     }
-    participant.lastClientSequence = sequence;
     return { ok: true, value: null };
   }
 
@@ -220,10 +221,14 @@ export function createMediaSignalService(options: MediaSignalServiceOptions = {}
     return { token, participant: {
       id: opaque(18), subject, role, tokenHash: hash(token), issuedAt, expiresAt: issuedAt + ttl,
       lastClientSequence: clientSequence, rateStartedAt: issuedAt, rateCount: 0, queue: [], listeners: new Set(),
+      queueOverflowed: false,
     } };
   }
 
   function join(request: ScreenShareCreateRequest | ScreenShareJoinRequest, subject: string): MediaSignalResult<ScreenShareCreateGrant | ScreenShareJoinGrant> {
+    if (!clientSequenceHasSuccessor(request.sequence)) {
+      return { ok: false, code: "invalid", message: "Media client sequence cannot advance." };
+    }
     cleanup();
     if (request.type === "screen-share-create-request") {
       if (sessions.size >= maximumSessions) return { ok: false, code: "capacity", message: "Media signaling is at capacity." };
@@ -263,22 +268,35 @@ export function createMediaSignalService(options: MediaSignalServiceOptions = {}
     const found = authenticate(request.credential, subject);
     if (!found.ok) return found;
     const { session, participant } = found.value;
+    if (request.lastReceivedSequence > session.sequence) {
+      return { ok: false, code: "conflict", message: "Media server sequence is ahead of this session." };
+    }
     const valid = validateRequest(session, participant, request.binding, request.sequence);
     if (!valid.ok) return valid;
+    participant.lastClientSequence = request.sequence;
     const token = opaque(32);
     participant.tokenHash.fill(0);
     participant.tokenHash = hash(token);
     participant.issuedAt = now();
     participant.expiresAt = now() + ttl;
-    // A resume grant carries a complete participant snapshot. Dropping queued
-    // relays avoids replaying a message with a sequence older than that grant;
-    // clients renegotiate when `continuous` is false.
-    participant.queue.length = 0;
+    const continuous = !participant.queueOverflowed;
+    const queued = participant.queue.filter((message) => message.sequence > request.lastReceivedSequence);
+    const grantSequence = nextSequence(session);
+    const resumedAt = now();
+    // The resume grant is written before subscribe drains this queue. Rebase
+    // retained messages onto fresh server sequences so a strict client cursor
+    // can accept them after the grant instead of treating them as stale.
+    participant.queue = queued.map((message) => ({
+      ...message,
+      sequence: nextSequence(session),
+      timestampMs: resumedAt,
+    }));
+    participant.queueOverflowed = false;
     return { ok: true, value: {
       type: "screen-share-resume-grant", protocolVersion: SCREEN_SHARE_SIGNALING_PROTOCOL_VERSION,
-      requestId: request.requestId, sequence: nextSequence(session), timestampMs: now(), binding: session.binding,
+      requestId: request.requestId, sequence: grantSequence, timestampMs: resumedAt, binding: session.binding,
       grant: grant(session, participant, token), nextClientSequence: request.sequence + 1,
-      participants: peerList(session, participant.id), continuous: false,
+      participants: peerList(session, participant.id), continuous,
     } };
   }
 
@@ -296,6 +314,7 @@ export function createMediaSignalService(options: MediaSignalServiceOptions = {}
       const permitted = participant.role === "presenter" ? request.signal.descriptionType === "offer" : request.signal.descriptionType === "answer";
       if (!permitted) return { ok: false, code: "unauthorized", message: "Media SDP role is not permitted." };
     }
+    participant.lastClientSequence = request.sequence;
     const relay: ScreenShareSignalRelay = {
       type: "screen-share-signal-relay", protocolVersion: SCREEN_SHARE_SIGNALING_PROTOCOL_VERSION,
       sequence: nextSequence(session), timestampMs: now(), sessionId: session.id, binding: session.binding,
@@ -313,12 +332,14 @@ export function createMediaSignalService(options: MediaSignalServiceOptions = {}
     if (!valid.ok) return valid;
     if (request.type === "screen-share-stop-request") {
       if (participant.role !== "presenter") return { ok: false, code: "unauthorized", message: "Only a presenter may stop a share." };
+      participant.lastClientSequence = request.sequence;
       destroySession(session, stopNotice(session, request.reason));
       return { ok: true, value: null };
     }
     if (request.reason === "moderator-action" && !admin) return { ok: false, code: "unauthorized", message: "Moderator action requires an administrator." };
     if (request.scope === "session") {
       if (participant.role !== "presenter" && !admin) return { ok: false, code: "unauthorized", message: "Only a presenter or administrator may revoke a session." };
+      participant.lastClientSequence = request.sequence;
       destroySession(session, {
         type: "screen-share-revoked", protocolVersion: SCREEN_SHARE_SIGNALING_PROTOCOL_VERSION,
         sequence: nextSequence(session), timestampMs: now(), sessionId: session.id, binding: session.binding,
@@ -333,6 +354,7 @@ export function createMediaSignalService(options: MediaSignalServiceOptions = {}
       return { ok: false, code: "unauthorized", message: "Viewer may only leave its own session." };
     }
     if (target.role === "presenter") return { ok: false, code: "unauthorized", message: "Presenter must stop the session." };
+    participant.lastClientSequence = request.sequence;
     removeParticipant(session, target, request.reason);
     return { ok: true, value: null };
   }
@@ -400,6 +422,9 @@ function positive(value: number | undefined, fallback: number): number {
   return typeof value === "number" && Number.isFinite(value) && value > 0 ? value : fallback;
 }
 function count(value: number | undefined, fallback: number): number { return Math.max(1, Math.floor(positive(value, fallback))); }
+function clientSequenceHasSuccessor(value: number): boolean {
+  return Number.isSafeInteger(value) && value >= 0 && value < Number.MAX_SAFE_INTEGER;
+}
 function copyBinding(binding: ScreenShareBinding): ScreenShareBinding { return { ...binding }; }
 function sameBinding(a: ScreenShareBinding, b: ScreenShareBinding): boolean {
   return a.officeId === b.officeId && a.levelId === b.levelId && a.roomId === b.roomId && a.screenId === b.screenId;