refactor(core): extract transcript projection

packages/core/src/session/projection/transcript.ts

@@ -0,0 +1,201 @@
+export * as TranscriptProjection from "./transcript.js"
+
+import { and, desc, eq, sql } from "drizzle-orm"
+import { DateTime, Effect, Schema } from "effect"
+import { Database } from "../../database/database.js"
+import { Agent } from "@opencode/schema/agent"
+import { Model } from "@opencode/schema/model"
+import { Workspace } from "@opencode/schema/workspace"
+import { AbsolutePath, RelativePath } from "../../schema.js"
+import { SessionEvent } from "../event.js"
+import { SessionMessage } from "../message.js"
+import { SessionMessageUpdater } from "../message-updater.js"
+import { SessionMessageTable, SessionTable } from "../sql.js"
+
+type DatabaseService = Database.Interface["db"]
+type MessageEvent = Exclude<
+  SessionEvent.DurableEvent,
+  typeof SessionEvent.Forked.Type | typeof SessionEvent.Deleted.Type
+>
+
+const decodeMessage = Schema.decodeUnknownSync(SessionMessage.Info)
+const encodeMessage = Schema.encodeSync(SessionMessage.Info)
+
+export function project(db: DatabaseService, event: MessageEvent) {
+  return Effect.gen(function* () {
+    const decodeRow = (row: typeof SessionMessageTable.$inferSelect) =>
+      decodeMessage({ ...row.data, id: row.id, type: row.type })
+    const updateMessage = (message: SessionMessage.Info) => {
+      const encoded = encodeMessage(message)
+      const { id, type, ...data } = encoded
+      return db
+        .update(SessionMessageTable)
+        .set({ type, time_created: DateTime.toEpochMillis(message.time.created), data })
+        .where(
+          and(
+            eq(SessionMessageTable.id, SessionMessage.ID.make(id)),
+            eq(SessionMessageTable.session_id, event.data.sessionID),
+          ),
+        )
+        .run()
+        .pipe(Effect.orDie)
+    }
+    const appendMessage = (message: SessionMessage.Info) => appendAtEventSequence(db, event, message)
+    const adapter: SessionMessageUpdater.Adapter = {
+      getAgent() {
+        return db
+          .select({ agent: SessionTable.agent })
+          .from(SessionTable)
+          .where(eq(SessionTable.id, event.data.sessionID))
+          .get()
+          .pipe(
+            Effect.orDie,
+            Effect.map((row) => (row?.agent ? Agent.ID.make(row.agent) : undefined)),
+          )
+      },
+      getModel() {
+        return db
+          .select({ model: SessionTable.model })
+          .from(SessionTable)
+          .where(eq(SessionTable.id, event.data.sessionID))
+          .get()
+          .pipe(
+            Effect.orDie,
+            Effect.map((row) => (row?.model ? Schema.decodeUnknownSync(Model.Ref)(row.model) : undefined)),
+          )
+      },
+      getLocation() {
+        return db
+          .select({
+            directory: SessionTable.directory,
+            workspaceID: SessionTable.workspace_id,
+            projectID: SessionTable.project_id,
+            subpath: SessionTable.path,
+          })
+          .from(SessionTable)
+          .where(eq(SessionTable.id, event.data.sessionID))
+          .get()
+          .pipe(
+            Effect.orDie,
+            Effect.map((row) =>
+              row
+                ? {
+                    location: {
+                      directory: AbsolutePath.make(row.directory),
+                      workspaceID: row.workspaceID ? Workspace.ID.make(row.workspaceID) : undefined,
+                    },
+                    projectID: row.projectID,
+                    subpath: row.subpath === null ? undefined : RelativePath.make(row.subpath),
+                  }
+                : undefined,
+            ),
+          )
+      },
+      getCurrentAssistant() {
+        return Effect.gen(function* () {
+          // A newer step supersedes stale incomplete rows; never resume an older assistant projection.
+          const row = yield* db
+            .select()
+            .from(SessionMessageTable)
+            .where(
+              and(eq(SessionMessageTable.session_id, event.data.sessionID), eq(SessionMessageTable.type, "assistant")),
+            )
+            .orderBy(desc(SessionMessageTable.seq))
+            .limit(1)
+            .get()
+            .pipe(Effect.orDie)
+          if (!row) return
+          const message = decodeRow(row)
+          return message.type === "assistant" && !message.time.completed ? message : undefined
+        })
+      },
+      getAssistant(messageID) {
+        return Effect.gen(function* () {
+          const row = yield* db
+            .select()
+            .from(SessionMessageTable)
+            .where(
+              and(
+                eq(SessionMessageTable.id, messageID),
+                eq(SessionMessageTable.session_id, event.data.sessionID),
+                eq(SessionMessageTable.type, "assistant"),
+              ),
+            )
+            .get()
+            .pipe(Effect.orDie)
+          if (!row) return
+          const message = decodeRow(row)
+          return message.type === "assistant" ? message : undefined
+        })
+      },
+      getShell(shellID) {
+        return Effect.gen(function* () {
+          const row = yield* db
+            .select()
+            .from(SessionMessageTable)
+            .where(
+              and(
+                eq(SessionMessageTable.session_id, event.data.sessionID),
+                eq(SessionMessageTable.type, "shell"),
+                sql`json_extract(${SessionMessageTable.data}, '$.shellID') = ${shellID}`,
+              ),
+            )
+            .orderBy(desc(SessionMessageTable.seq))
+            .limit(1)
+            .get()
+            .pipe(Effect.orDie)
+          if (!row) return
+          const message = decodeRow(row)
+          return message.type === "shell" ? message : undefined
+        })
+      },
+      getCompaction() {
+        return Effect.gen(function* () {
+          const row = yield* db
+            .select()
+            .from(SessionMessageTable)
+            .where(
+              and(
+                eq(SessionMessageTable.session_id, event.data.sessionID),
+                eq(SessionMessageTable.type, "compaction"),
+                sql`json_extract(${SessionMessageTable.data}, '$.status') = 'running'`,
+              ),
+            )
+            .orderBy(desc(SessionMessageTable.seq))
+            .limit(1)
+            .get()
+            .pipe(Effect.orDie)
+          if (!row) return
+          const message = decodeRow(row)
+          return message.type === "compaction" ? message : undefined
+        })
+      },
+      updateAssistant: updateMessage,
+      updateShell: updateMessage,
+      updateCompaction: updateMessage,
+      appendMessage,
+    }
+    yield* SessionMessageUpdater.update(adapter, event)
+  })
+}
+
+export function appendAtEventSequence(
+  db: DatabaseService,
+  event: SessionEvent.DurableEvent,
+  message: SessionMessage.Info,
+) {
+  const encoded = encodeMessage(message)
+  const { id, type, ...data } = encoded
+  return db
+    .insert(SessionMessageTable)
+    .values({
+      id: SessionMessage.ID.make(id),
+      session_id: event.data.sessionID,
+      type,
+      seq: event.durable.seq,
+      time_created: DateTime.toEpochMillis(message.time.created),
+      data,
+    })
+    .run()
+    .pipe(Effect.orDie)
+}