Add capacity snapshot

Author: sarahsturgeon

web/src/index.ts

@@ -6,6 +6,7 @@ import {
   serializeContext,
   parseContext,
   OBS_CONTEXT_HEADER,
+  type CapacitySnapshot,
   type CloseReason,
   type RequestContext,
 } from "./observability";
@@ -53,6 +54,18 @@ export class BaseSession extends Container<Env> {
   // embed, which is cosmetic — never a correctness concern.
   protected obsContext?: RequestContext;
 
+  private async fetchCapacitySnapshot(): Promise<CapacitySnapshot | undefined> {
+    try {
+      const queueDO = this.env.QUEUE_DO.get(this.env.QUEUE_DO.idFromName("global-queue"));
+      const res = await queueDO.fetch(`http://do/internal/capacity?branch=${encodeURIComponent(this.branch)}`);
+      if (!res.ok) return undefined;
+      return await res.json<CapacitySnapshot>();
+    } catch (e) {
+      console.error("[obs] fetchCapacitySnapshot failed:", e);
+      return undefined;
+    }
+  }
+
   constructor(ctx: DurableObjectState, env: Env) {
     // Pass the container config to the super constructor.
     super(ctx, env, {
@@ -92,11 +105,15 @@ export class BaseSession extends Container<Env> {
       this.sessionState = "PROVISIONING";
 
       this.obsContext = parseContext(request.headers.get(OBS_CONTEXT_HEADER));
-      void notify.sessionStarted(this.env, {
-        sessionId,
-        branch: this.branch,
-        context: this.obsContext ?? { ip: "unknown" },
-      });
+      void (async () => {
+        const capacity = await this.fetchCapacitySnapshot();
+        await notify.sessionStarted(this.env, {
+          sessionId,
+          branch: this.branch,
+          context: this.obsContext ?? { ip: "unknown" },
+          capacity,
+        });
+      })();
 
       // Start the container, but don't wait for it to finish.
       void this.startContainer(sessionId);
@@ -384,17 +401,21 @@ export class BaseSession extends Container<Env> {
 
     await this.notifyQueueManagerOfClosure();
 
-    void notify.sessionEnded(this.env, {
-      sessionId: this.ctx.id.name ?? "unknown",
-      branch: this.branch,
-      reason,
-      startedAt: this.sessionMetadata?.startedAt,
-      endedAt,
-      scriptCount: this.scriptCount,
-      logLineCount: this.logLineCount,
-      extensionGranted: this.extensionGranted,
-      context: this.obsContext,
-    });
+    void (async () => {
+      const capacity = await this.fetchCapacitySnapshot();
+      await notify.sessionEnded(this.env, {
+        sessionId: this.ctx.id.name ?? "unknown",
+        branch: this.branch,
+        reason,
+        startedAt: this.sessionMetadata?.startedAt,
+        endedAt,
+        scriptCount: this.scriptCount,
+        logLineCount: this.logLineCount,
+        extensionGranted: this.extensionGranted,
+        context: this.obsContext,
+        capacity,
+      });
+    })();
   }
 
   async notifyQueueManagerOfClosure() {
@@ -576,9 +597,31 @@ export class QueueDO extends DurableObject<Env> {
     return this.activeCountForType(type) < max && this.activeSessions.size < this.maxTotalSessions;
   }
 
+  private snapshotFor(branch: string): CapacitySnapshot {
+    return {
+      branch,
+      branchUsed: this.activeCountForType(branch),
+      branchMax: this.maxPerType[branch] ?? 2,
+      totalUsed: this.activeSessions.size,
+      totalMax: this.maxTotalSessions,
+      queueDepth: this.waitingQueue.length,
+    };
+  }
+
   async fetch(request: Request): Promise<Response> {
     const url = new URL(request.url);
 
+    // Internal-only: the worker entrypoint routes /api/* and /ws/* to DOs,
+    // so /internal/* paths are unreachable externally (404 at the worker)
+    // but work over DO stub-to-stub calls, which bypass the entrypoint.
+    if (url.pathname === "/internal/capacity") {
+      const branch = url.searchParams.get("branch");
+      if (!branch) return new Response("Missing branch", { status: 400 });
+      return new Response(JSON.stringify(this.snapshotFor(branch)), {
+        headers: { "Content-Type": "application/json" },
+      });
+    }
+
     if (url.pathname === "/api/request-session") {
       const sessionType = url.searchParams.get("type") || "public";
       const rawIP = request.headers.get("CF-Connecting-IP") || "unknown";
@@ -613,6 +656,16 @@ export class QueueDO extends DurableObject<Env> {
         }});
         const position = this.waitingQueue.filter(w => w.sessionType === sessionType).length;
 
+        const obsContext = parseContext(request.headers.get(OBS_CONTEXT_HEADER));
+        if (obsContext) {
+          void notify.queueEntered(this.env, {
+            sessionType,
+            position,
+            capacity: this.snapshotFor(sessionType),
+            context: obsContext,
+          });
+        }
+
         return new Response(JSON.stringify({
           status: "QUEUED",
           ticketId,

web/src/observability/embeds.ts

@@ -1,6 +1,8 @@
 import type {
+  CapacitySnapshot,
   CloseReason,
   ErrorEvent,
+  QueueEnteredEvent,
   RequestContext,
   SessionEndedEvent,
   SessionStartedEvent,
@@ -107,6 +109,9 @@ const networkFields = (ctx: RequestContext): DiscordEmbedField[] => [
   { name: "Network", value: ispLink(ctx),   inline: true },
 ];
 
+const capacityLine = (c: CapacitySnapshot, opts?: { skipQueue: boolean }): string =>
+  `${c.branch} ${c.branchUsed}/${c.branchMax} · total ${c.totalUsed}/${c.totalMax}${!opts?.skipQueue && c.queueDepth > 0 ? ` · queue ${c.queueDepth}` : ""}`;
+
 const contextSubtext = (ctx: RequestContext | undefined): string | undefined => {
   if (!ctx) return undefined;
 
@@ -139,6 +144,7 @@ export function buildSessionStartedEmbed(e: SessionStartedEvent): DiscordEmbed {
     `### ${locationHeader(ctx)}`,
     sub(`started ${relativeTimestamp()}${ctx.colo ? ` · edge ${code(ctx.colo)}` : ""}`),
   ];
+  if (e.capacity) lines.push(sub(capacityLine(e.capacity)));
 
   const fields: DiscordEmbedField[] = [
     { name: "Branch", value: `${emoji} ${code(label)}`, inline: true },
@@ -175,6 +181,7 @@ export function buildSessionEndedEmbed(e: SessionEndedEvent): DiscordEmbed {
     `### ${ctx ? locationHeader(ctx) : "🌐 Unknown location"}`,
     sub(`ended ${relativeTimestamp()}${durationMs !== undefined ? ` · ran for **${duration(durationMs)}**` : ""}`),
   ];
+  if (e.capacity) lines.push(sub(capacityLine(e.capacity)));
 
   const fields: DiscordEmbedField[] = [
     { name: "Branch", value: `${emoji} ${code(label)}`, inline: true },
@@ -236,6 +243,30 @@ export function buildErrorEmbed(e: ErrorEvent): DiscordEmbed {
   };
 }
 
+export function buildQueueEnteredEmbed(e: QueueEnteredEvent): DiscordEmbed {
+  const { emoji, label } = branchMeta(e.sessionType);
+  const ctx = e.context;
+
+  const lines: string[] = [
+    `### ${locationHeader(ctx)}`,
+    sub(`position ${code("#" + e.position)} · ${capacityLine(e.capacity, { skipQueue: true })}`),
+  ];
+
+  const fields: DiscordEmbedField[] = [
+    { name: "Branch", value: `${emoji} ${code(label)}`, inline: true },
+    ...networkFields(ctx),
+  ];
+
+  return {
+    title: "⏳ Queue entered",
+    description: truncate(lines.join("\n"), LIMIT_DESCRIPTION),
+    color: COLORS.info,
+    timestamp: new Date().toISOString(),
+    fields,
+    footer: FOOTER,
+  };
+}
+
 export function buildWarningEmbed(e: WarningEvent): DiscordEmbed {
   const lines: string[] = [
     `## ⚠️ ${truncate(e.title, 120)}`,

web/src/observability/index.ts

@@ -1,18 +1,20 @@
 import { post } from "./webhook";
 import {
   buildErrorEmbed,
+  buildQueueEnteredEmbed,
   buildSessionEndedEmbed,
   buildSessionStartedEmbed,
   buildWarningEmbed,
 } from "./embeds";
 import type {
   ErrorEvent,
+  QueueEnteredEvent,
   SessionEndedEvent,
   SessionStartedEvent,
   WarningEvent,
 } from "./types";
 
-export type { CloseReason, RequestContext } from "./types";
+export type { CapacitySnapshot, CloseReason, RequestContext } from "./types";
 export { extractRequestContext, serializeContext, parseContext, OBS_CONTEXT_HEADER } from "./context";
 
 type Env = { DISCORD_WEBHOOK_URL?: string };
@@ -22,6 +24,8 @@ export const notify = {
     post(env, { embeds: [buildSessionStartedEmbed(event)] }),
   sessionEnded: (env: Env, event: SessionEndedEvent) =>
     post(env, { embeds: [buildSessionEndedEmbed(event)] }),
+  queueEntered: (env: Env, event: QueueEnteredEvent) =>
+    post(env, { embeds: [buildQueueEnteredEmbed(event)] }),
   error: (env: Env, event: ErrorEvent) =>
     post(env, { embeds: [buildErrorEmbed(event)] }),
   warning: (env: Env, event: WarningEvent) =>

web/src/observability/types.ts

@@ -19,10 +19,20 @@ export interface RequestContext {
   userAgent?: string;
 }
 
+export interface CapacitySnapshot {
+  branch: string;
+  branchUsed: number;
+  branchMax: number;
+  totalUsed: number;
+  totalMax: number;
+  queueDepth: number;
+}
+
 export interface SessionStartedEvent {
   sessionId: string;
   branch: string;
   context: RequestContext;
+  capacity?: CapacitySnapshot;
 }
 
 export interface SessionEndedEvent {
@@ -35,6 +45,14 @@ export interface SessionEndedEvent {
   logLineCount: number;
   extensionGranted: boolean;
   context?: RequestContext;
+  capacity?: CapacitySnapshot;
+}
+
+export interface QueueEnteredEvent {
+  sessionType: string;
+  position: number;
+  capacity: CapacitySnapshot;
+  context: RequestContext;
 }
 
 export interface ErrorEvent {