beaver-gateway/ui/src/lib/gateway.svelte.ts

import { LiveStream } from "./api/live.svelte";
import type {
  BusEvent,
  ConversationSummary,
  LimitsResponse,
  LimitWindow,
} from "./api/types";
import { session } from "./session.svelte";

const TAPE_SIZE = 120;
const RELOAD_DEBOUNCE_MS = 1500;
const QUIET = new Set(["stream", "hello"]);

// Gateway-wide live state: the conversation index kept fresh by
// ``/api/events``, the last non-chatty events as a tape, and the quota
// windows updated the moment the SDK reports them.
class Gateway {
  conversations = $state<ConversationSummary[]>([]);
  tape = $state<BusEvent[]>([]);
  limits = $state<LimitsResponse | null>(null);
  loaded = $state(false);
  readonly live = new LiveStream(() => session.client, "/api/events", {
    onEvent: (event) => this.apply(event),
    prepare: () => this.load(),
  });

  get running(): ConversationSummary[] {
    return this.conversations.filter((row) => row.running_turn);
  }

  get open(): ConversationSummary[] {
    return this.conversations.filter((row) => row.status === "open");
  }

  byId(id: string): ConversationSummary | undefined {
    return this.conversations.find((row) => row.id === id);
  }

  async load(): Promise<void> {
    const { client } = session;
    if (!client) {
      return;
    }
    const [list, limits] = await Promise.all([
      client.conversations({ limit: 500 }),
      client.limits().catch(() => null),
    ]);
    this.conversations = list.conversations;
    if (limits) {
      this.limits = limits;
    }
    this.loaded = true;
  }

  async refreshLimits(): Promise<void> {
    const { client } = session;
    if (client) {
      this.limits = await client.limits();
    }
  }

  private upsert(row: ConversationSummary): void {
    const index = this.conversations.findIndex((c) => c.id === row.id);
    if (index >= 0) {
      this.conversations[index] = { ...this.conversations[index], ...row };
    } else {
      this.conversations.unshift(row);
    }
  }

  apply(event: BusEvent): void {
    if (!QUIET.has(event.type)) {
      this.tape.unshift(event);
      if (this.tape.length > TAPE_SIZE) {
        this.tape.length = TAPE_SIZE;
      }
    }
    switch (event.type) {
      case "conversation.created":
      case "conversation.updated": {
        if (typeof event.id === "string") {
          this.upsert(event as unknown as ConversationSummary);
        }
        return;
      }
      case "turn.start":
      case "turn.end": {
        const row = event.conversation_id
          ? this.byId(event.conversation_id)
          : undefined;
        if (row) {
          row.running_turn =
            event.type === "turn.start" ? (event.turn_id ?? null) : null;
          row.last_activity_at = event.ts;
        }
        return;
      }
      case "rate_limit": {
        this.applyLimit(event);
        return;
      }
      case "message.queued":
      case "inject.queued":
      case "reply": {
        this.reloadSoon();
        return;
      }
      default:
    }
  }

  private reloadTimer: ReturnType<typeof setTimeout> | null = null;

  private reloadSoon(): void {
    if (this.reloadTimer) {
      return;
    }
    this.reloadTimer = setTimeout(() => {
      this.reloadTimer = null;
      this.load().catch(() => undefined);
    }, RELOAD_DEBOUNCE_MS);
  }

  private applyLimit(event: BusEvent): void {
    if (!this.limits || typeof event.window !== "string") {
      this.refreshLimits().catch(() => undefined);
      return;
    }
    const current = this.limits.windows.find((w) => w.window === event.window);
    const patch = {
      overage_status: event.overage_status ?? null,
      resets_at: event.resets_at,
      status: event.status,
      ts: event.ts,
      utilization: event.utilization,
    } as Partial<LimitWindow>;
    if (current) {
      Object.assign(current, patch);
    } else {
      this.refreshLimits().catch(() => undefined);
    }
  }

  start(): void {
    this.live.start();
  }

  stop(): void {
    this.live.stop();
  }
}

export const gateway = new Gateway();