Keep active task transcripts in memory for instant task switching

web/src/state.ts

@@ -17,7 +17,9 @@ export type Topic =
   | "focus"
   | "lead.sessions"
   | "lead.transcript"
-  | "lead.stream";
+  | "lead.stream"
+  | "ready"
+  | "lead.ready";
 
 export const tabs: TaskTab[] = ["transcript", "changes", "files", "terminal", "inbox", "reviews", "project"];
 
@@ -130,6 +132,12 @@ export async function loadProjects(): Promise<void> {
 export async function loadTasks(projectId: string): Promise<void> {
   const tasks = await get("/api/projects/{projectId}/tasks", { path: { projectId } });
   state.tasks.set(projectId, tasks);
+  for (const t of tasks) {
+    if (t.status === "active") {
+      taskFeed(t.id);
+    }
+  }
+  dropClosedFeeds();
   await Promise.all(tasks.filter((t) => t.status === "active").map((t) => loadSessions(t.id)));
   notify("tasks");
 }
@@ -193,7 +201,8 @@ export function focusTab(tab: TaskTab): void {
 }
 
 class Feed {
-  events: ByreEvent[] = [];
+  readonly events: ByreEvent[] = [];
+  ready = false;
   private last = 0;
   private buffer: ByreEvent[] = [];
   private timer = 0;
@@ -201,26 +210,34 @@ class Feed {
   constructor(
     private readonly transcriptTopic: Topic,
     private readonly streamTopic: Topic,
-    private readonly store: (events: ByreEvent[]) => void,
+    private readonly readyTopic: Topic,
+    private readonly shown: () => boolean,
   ) {}
 
-  reset(): void {
-    clearTimeout(this.timer);
-    this.timer = 0;
-    this.buffer = [];
-    this.events = [];
-    this.last = 0;
-    this.store(this.events);
-  }
-
   readonly push = (ev: ByreEvent): void => {
+    if (ev.seq === 0 && (!this.ready || !this.shown())) {
+      return;
+    }
     this.buffer.push(ev);
-    if (!this.timer) {
+    if (this.ready && !this.timer) {
       this.timer = setTimeout(() => this.flush(), 0);
     }
   };
 
+  readonly load = (history: ByreEvent[]): void => {
+    if (this.ready) {
+      return;
+    }
+    this.ready = true;
+    this.buffer = [...history, ...this.buffer];
+    this.flush();
+    if (this.shown()) {
+      notify(this.readyTopic);
+    }
+  };
+
   private flush(): void {
+    clearTimeout(this.timer);
     const events = this.buffer;
     this.buffer = [];
     this.timer = 0;
@@ -232,30 +249,84 @@ class Feed {
       }
       this.append(stored);
       stored = [];
-      notify(this.streamTopic, ev);
+      if (this.shown()) {
+        notify(this.streamTopic, ev);
+      }
     }
     this.append(stored);
   }
 
   private append(events: ByreEvent[]): void {
-    const fresh = events.filter((ev) => ev.seq > this.last);
+    const fresh: ByreEvent[] = [];
+    for (const ev of events) {
+      if (ev.seq > this.last) {
+        fresh.push(ev);
+        this.last = ev.seq;
+      }
+    }
     if (fresh.length === 0) {
       return;
     }
     this.events.push(...fresh);
-    this.last = fresh.at(-1)!.seq;
-    notify(this.transcriptTopic, fresh);
+    if (this.shown()) {
+      notify(this.transcriptTopic, fresh);
+    }
   }
 }
 
-const taskFeed = new Feed("transcript", "stream", (events) => {
-  state.transcript = events;
-});
-const leadFeed = new Feed("lead.transcript", "lead.stream", (events) => {
-  state.leadTranscript = events;
-});
+const historyPage = 1000;
 
-let closeTaskStream: (() => void) | undefined;
+const taskFeeds = new Map<string, Feed>();
+const leadFeed = new Feed("lead.transcript", "lead.stream", "lead.ready", () => true);
+state.leadTranscript = leadFeed.events;
+
+async function loadHistory(taskId: string, feed: Feed): Promise<void> {
+  const events: ByreEvent[] = [];
+  for (;;) {
+    const page = await get("/api/tasks/{taskId}/events", { path: { taskId }, query: { after: events.at(-1)?.seq ?? 0, limit: historyPage } });
+    events.push(...page);
+    if (page.length < historyPage) {
+      break;
+    }
+  }
+  feed.load(events);
+}
+
+const historyLoads = new Set<Feed>();
+
+function taskFeed(taskId: string): Feed {
+  let feed = taskFeeds.get(taskId);
+  if (!feed) {
+    feed = new Feed("transcript", "stream", "ready", () => state.selectedTaskId === taskId);
+    taskFeeds.set(taskId, feed);
+  }
+  if (feed.ready || historyLoads.has(feed)) {
+    return feed;
+  }
+  const loading = feed;
+  historyLoads.add(loading);
+  loadHistory(taskId, loading)
+    .catch((err: unknown) => console.error("load task history", taskId, err))
+    .finally(() => historyLoads.delete(loading));
+  return feed;
+}
+
+function dropClosedFeeds(): void {
+  for (const taskId of taskFeeds.keys()) {
+    const task = findTask(taskId);
+    if (taskId !== state.selectedTaskId && task && task.status !== "active") {
+      taskFeeds.delete(taskId);
+    }
+  }
+}
+
+export function transcriptLoading(): boolean {
+  return state.selectedTaskId !== undefined && taskFeeds.get(state.selectedTaskId)?.ready !== true;
+}
+
+export function leadLoading(): boolean {
+  return !leadFeed.ready;
+}
 
 const visitsKey = "byre.visits";
 
@@ -273,27 +344,22 @@ function markVisited(taskId: string | undefined): void {
 export function selectTask(taskId: string | undefined): void {
   markVisited(state.selectedTaskId);
   markVisited(taskId);
-  closeTaskStream?.();
-  closeTaskStream = undefined;
-  taskFeed.reset();
   state.selectedTaskId = taskId;
+  state.transcript = taskId ? taskFeed(taskId).events : [];
   state.pinned = undefined;
   state.reviewed = undefined;
+  dropClosedFeeds();
   notify("selection");
-
-  if (!taskId) {
-    return;
-  }
-
-  closeTaskStream = subscribe(0, { task_id: taskId }, taskFeed.push, () => undefined);
 }
 
 export function startLeadStream(): void {
-  leadFeed.reset();
-  subscribe(0, { lead: true }, leadFeed.push, () => undefined);
+  subscribe(0, { lead: true }, leadFeed.push, () => undefined, () => leadFeed.load([]));
 }
 
 export function handleEvent(ev: ByreEvent): void {
+  if (ev.task_id) {
+    taskFeeds.get(ev.task_id)?.push(ev);
+  }
   if (ev.seq === 0) {
     notify("event", ev);
     return;