ai-persistence-build-prisma-adapter

# Prisma Chat Persistence The deliverable is **one file in the app** — `src/lib/chat-persistence.ts` — exporting a `ChatPersistence` built from the app's existing `PrismaClient`. Plus four models added to the app's existing `schema.prisma` and a migration created with the app's own `prisma migrate`. Do not create a package, a second client, a datasource block, a generator, or a hand-written SQL migration. The app has those. Read the **Store Reference** (`docs/persistence/store-reference.md`) for the store contracts and invariants, and **ai-persistence/stores** for the shape rules. Every store below mirrors the reference in-memory backend in `@tanstack/ai-persistence` (`memory.ts`); the shared conformance testkit is the proof. ## 1. Read the app before writing anything | Find | Where to look | What it decides | | -------------------- | ------------------------------------------------------------------------------ | --------------------------------------------------------------- | | Schema location | `prisma/schema.prisma`, or a multi-file `prisma/schema/` dir | Append to the existing file, or add one new `.prisma` file | | Provider | the `datasource` block | Whether `Json` is available; nothing else changes | | Client singleton | `src/lib/prisma.ts`, `src/db.ts`, `globalThis` dev cache | What `chat-persistence.ts` imports — never `new PrismaClient()` | | Generated client | the `generator client` block (`output`, `prisma-client-js` vs `prisma-client`) | Where `ChatRun`/`ChatInterrupt` row types come from | | Existing model names | the schema | Whether `Message`/`Run` are taken — prefix if so | | Migration flow | `prisma/migrations/`, or `db push` in scripts | `prisma migrate dev` vs `prisma db push` | Prisma 6 and 7 both work: the delegate query API (`findUnique`, `upsert`, `update`, `findMany`, `delete`) is unchanged, so it does not matter which client the app generated. **Never invent a migration path.** Add the models, then have the user run their own `npx prisma migrate dev --name chat-persistence` (or `db push`) and `prisma generate`. ## 2. Add the models to their schema IDs are `String`, timestamps are `BigInt` (portable epoch ms — `Int` overflows in 2038, `DateTime` forces a conversion at every boundary), JSON payloads are `String`. Use `@map`/`@@map` to match the app's database naming. ```prisma model ChatThread { threadId String @id @map("thread_id") messagesJson String @map("messages_json") updatedAt BigInt @map("updated_at") @@map("chat_threads") } model ChatRun { runId String @id @map("run_id") threadId String @map("thread_id") status String startedAt BigInt @map("started_at") finishedAt BigInt? @map("finished_at") error String? errorCode String? @map("error_code") usageJson String? @map("usage_json") sandboxKey String? @map("sandbox_key") detachedSince BigInt? @map("detached_since") cancelRequested Boolean? @map("cancel_requested") driverEpoch Int? @map("driver_epoch") parentRunId String? @map("parent_run_id") subagentRunId String? @map("subagent_run_id") name String? @@index([threadId, status]) @@index([threadId, startedAt]) // Powers listByParentRun: children of one parent, oldest startedAt first. @@index([parentRunId, startedAt]) // Powers listReclaimable: status = 'running' AND detachedSince <= cutoff. @@index([status, detachedSince]) @@map("chat_runs") } model ChatInterrupt { interruptId String @id @map("interrupt_id") runId String @map("run_id") threadId String @map("thread_id") status String requestedAt BigInt @map("requested_at") resolvedAt BigInt? @map("resolved_at") payloadJson String @map("payload_json") responseJson String? @map("response_json") @@index([threadId, requestedAt]) @@map("chat_interrupts") } model ChatMetadata { namespace String key String valueJson String @map("value_json") @@id([namespace, key]) @@map("chat_metadata") } ``` Rename models freely to fit the app — the store code below is the only thing that references them. Extra app-owned fields (a `userId`, audit columns) are fine as long as they are optional or defaulted, so the stores' creates still succeed. `namespace` is the `MetadataStore` first argument; the stock SQL in the guide calls the same column `scope`. `RunRecord.error` is a structured `RunError` (`{ message: string, code?: string }`), so it gets two columns rather than one JSON blob: `error` for the provider's prose and `errorCode` for the stable classification an operator filters and groups by. `error` and `errorCode` always move together in `update`, so a later code-less failure can never leave a stale `code` from an earlier one behind. On **Postgres or MySQL** you can switch the `*Json` fields to Prisma's `Json` type and drop the `JSON.stringify`/`parse` in the mappers below. Keep `String` if the app targets SQLite or if it is multi-provider. ## 3. Write `src/lib/chat-persistence.ts` Two conversions the SQL backends do not need: `BigInt` timestamps in and out, and JSON as strings. Everything else is the shared invariant set. ```ts ignore import { defineAIPersistence } from '@tanstack/ai-persistence' import type { ChatInterrupt, ChatRun, Prisma, PrismaClient, } from '@prisma/client' import type { ModelMessage, TokenUsage } from '@tanstack/ai' import type { ChatPersistence, InterruptRecord, InterruptStatus, InterruptStore, MessageStore, MetadataStore, RunRecord, RunStatus, RunStore, } from '@tanstack/ai-persistence' import { prisma } from '@/lib/prisma' // Trusts the shape the stores themselves wrote — nothing else writes these // columns. function parseJson<T>(raw: string): T { return JSON.parse(raw) } const RUN_STATUSES: ReadonlyArray<RunStatus> = [ 'running', 'interrupted', 'completed', 'failed', 'aborted', ] const INTERRUPT_STATUSES: ReadonlyArray<InterruptStatus> = [ 'pending', 'resolved', 'cancelled', ] // The column is a plain String, so narrow instead of trusting it. function toRunStatus(value: string): RunStatus { const status = RUN_STATUSES.find((candidate) => candidate === value) if (!status) throw new Error(`Unknown run status: ${value}`) return status } function toInterruptStatus(value: string): InterruptStatus { const status = INTERRUPT_STATUSES.find((candidate) => candidate === value) if (!status) throw new Error(`Unknown interrupt status: ${value}`) return status } // Records omit absent optionals so they compare cleanly against the reference // in-memory backend. function mapRun(row: ChatRun): RunRecord { return { runId: row.runId, threadId: row.threadId, status: toRunStatus(row.status), startedAt: Number(row.startedAt), ...(row.finishedAt != null ? { finishedAt: Number(row.finishedAt) } : {}), ...(row.error != null ? { error: { message: row.error, ...(row.errorCode != null ? { code: row.errorCode } : {}), }, } : {}), ...(row.usageJson != null ? { usage: parseJson<TokenUsage>(row.usageJson) } : {}), ...(row.sandboxKey != null ? { sandboxKey: row.sandboxKey } : {}), ...(row.detachedSince != null ? { detachedSince: Number(row.detachedSince) } : {}), ...(row.cancelRequested != null ? { cancelRequested: row.cancelRequested } : {}), ...(row.driverEpoch != null ? { driverEpoch: row.driverEpoch } : {}), ...(row.parentRunId != null ? { parentRunId: row.parentRunId } : {}), ...(row.subagentRunId != null ? { subagentRunId: row.subagentRunId } : {}), ...(row.name != null ? { name: row.name } : {}), } } function mapInterrupt(row: ChatInterrupt): InterruptRecord { return { interruptId: row.interruptId, runId: row.runId, threadId: row.threadId, status: toInterruptStatus(row.status), requestedAt: Number(row.requestedAt), payload: parseJson<Record<string, unknown>>(row.payloadJson), ...(row.resolvedAt != null ? { resolvedAt: Number(row.resolvedAt) } : {}), ...(row.responseJson != null ? { response: parseJson<unknown>(row.responseJson) } : {}), } } function createMessageStore(db: PrismaClient): MessageStore { return { async loadThread(threadId) { const row = await db.chatThread.findUnique({ where: { threadId } }) // Unknown thread is [], never null. return row ? parseJson<Array<ModelMessage>>(row.messagesJson) : [] }, // Full overwrite — `messages` is the complete authoritative transcript. async saveThread(threadId, messages) { const messagesJson = JSON.stringify(messages) const updatedAt = BigInt(Date.now()) await db.chatThread.upsert({ where: { threadId }, create: { threadId, messagesJson, updatedAt }, update: { messagesJson, updatedAt }, }) }, } } function createRunStore(db: PrismaClient): RunStore { return { async get(runId) { const row = await db.chatRun.findUnique({ where: { runId } }) return row ? mapRun(row) : null }, // An empty `update` is Prisma's ON CONFLICT DO NOTHING: an existing runId // comes back untouched, so resume and double-submit are safe. async createOrResume(input) { const { runId, threadId, startedAt, status } = input const row = await db.chatRun.upsert({ where: { runId }, create: { runId, threadId, status: status ?? 'running', startedAt: BigInt(startedAt), ...(input.parentRunId !== undefined ? { parentRunId: input.parentRunId } : {}), ...(input.subagentRunId !== undefined ? { subagentRunId: input.subagentRunId } : {}), ...(input.name !== undefined ? { name: input.name } : {}), }, update: {}, }) return mapRun(row) }, // Patching an unknown runId is a no-op: never throws, never inserts. async update(runId, patch) { const data: Prisma.ChatRunUpdateManyMutationInput = {} if (patch.status !== undefined) data.status = patch.status if (patch.finishedAt !== undefined) { data.finishedAt = BigInt(patch.finishedAt) } // Both columns move together, so a later code-less failure cannot // leave a stale errorCode from an earlier one behind. if (patch.error !== undefined) { data.error = patch.error.message data.errorCode = patch.error.code ?? null } if (patch.usage !== undefined) data.usageJson = JSON.stringify(patch.usage) // The four durable-run fields use `'field' in patch`, NOT // `!== undefined`: a reattach clears `detachedSince` by passing it // explicitly as `undefined`, and that must still write NULL, not be // silently dropped from the update. Checking `!== undefined` cannot // tell "clear this" from "didn't mention this", and would leave every // reattached run looking permanently detached to the reaper. Same // reasoning for `cancelRequested` (`false` is a meaningful value). if ('sandboxKey' in patch) data.sandboxKey = patch.sandboxKey ?? null if ('detachedSince' in patch) { data.detachedSince = patch.detachedSince === undefined ? null : BigInt(patch.detachedSince) } if ('cancelRequested' in patch) data.cancelRequested = patch.cancelRequested ?? null if ('driverEpoch' in patch) data.driverEpoch = patch.driverEpoch ?? null if (Object.keys(data).length === 0) return await db.chatRun.updateMany({ where: { runId }, data }) }, // Optional in the contract; enables reconnect without a client-held run id. async findActiveRun(threadId) { const row = await db.chatRun.findFirst({ where: { threadId, status: 'running' }, orderBy: { startedAt: 'desc' }, }) return row ? mapRun(row) : null }, // Optional; every run for the thread, ascending by startedAt. Uses the // (threadId, startedAt) index. async listByThread(threadId) {