411 lines
18 KiB
Plaintext
411 lines
18 KiB
Plaintext
---
|
|
title: "Database persistence for chat"
|
|
sidebarTitle: "Database persistence"
|
|
description: "Split conversation state and live session metadata across hooks — preload, turn start, turn complete — without tying the pattern to a specific ORM or schema."
|
|
---
|
|
|
|
Durable chat runs can span **hours** and **many turns**. You usually want:
|
|
|
|
1. **Conversation state** — full **`UIMessage[]`** (or equivalent) keyed by **`chatId`**, so reloads and history views work.
|
|
2. **Live session state** — a **scoped access token** for the session and optionally **`lastEventId`** for stream resume.
|
|
|
|
This page describes a **hook mapping** that works with any database. Adapt table and column names to your stack.
|
|
|
|
## Conceptual data model
|
|
|
|
You can use one table or two; the important split is **semantic**:
|
|
|
|
| Concept | Purpose | Typical fields |
|
|
| ------- | ------- | -------------- |
|
|
| **Conversation** | Durable transcript + display metadata | Stable id (same as **`chatId`**), serialized **`uiMessages`**, title, model choice, owner/user id, timestamps |
|
|
| **Active session** | Hydrate the transport on page reload | Same **`chatId`** as key (or FK), **`publicAccessToken`**, optional **`lastEventId`** |
|
|
|
|
The **conversation** row is what your UI lists as "chats." The **session** row is what the **transport** needs after a refresh: a session-scoped PAT (so the transport doesn't have to re-mint on first paint) and the SSE resume cursor.
|
|
|
|
Storing the current **`runId`** is optional — useful for telemetry / dashboard linking ("View this run") but not required for resume. The Session row owns its current run server-side; the transport reads from `session.out` keyed on `chatId`, so a run swap (continuation, upgrade) is invisible to your DB schema.
|
|
|
|
<Note>
|
|
Store **`UIMessage[]`** in a JSON-compatible column, or normalize to a messages table — the pattern is *when* you read/write, not *how* you encode rows.
|
|
</Note>
|
|
|
|
## Where each hook writes
|
|
|
|
This pattern covers **durable DB rows** (the conversation and the active session). Per-process in-memory state ([`chat.local`](/ai-chat/chat-local), [DB connection pools](/database-connections), sandboxes, etc.) belongs in [`onBoot`](/ai-chat/lifecycle-hooks#onboot) — it fires on every fresh worker including continuation runs, where `onPreload` and `onChatStart` do not.
|
|
|
|
### `onPreload` (optional)
|
|
|
|
When the user triggers [preload](/ai-chat/fast-starts#preload), the run starts **before** the first user message.
|
|
|
|
- Ensure the **conversation** row exists (create or no-op).
|
|
- **Upsert session**: **`chatAccessToken`** from the event (a session-scoped PAT covering both `read:sessions:{chatId}` and `write:sessions:{chatId}`).
|
|
- Load any **user / tenant context** you need for prompts (`clientData`).
|
|
|
|
If you skip preload, do the equivalent in **`onChatStart`** when **`preloaded`** is false.
|
|
|
|
### `onChatStart` (chat's first message, non-preloaded path)
|
|
|
|
- Fires **once per chat**, on the very first user message. Does NOT fire on continuation runs (post-`endRun`, post-waitpoint-timeout, post-`chat.requestUpgrade`) or on OOM-retry attempts.
|
|
- If **`preloaded`** is true, return early — **`onPreload`** already ran.
|
|
- Otherwise mirror preload: user/context, conversation create, session upsert.
|
|
- No need to gate the conversation create on `continuation` — it's always a brand-new chat at this point.
|
|
- For continuation runs that need to refresh per-run state (new PAT, new `lastEventId`), do it in **`onTurnStart`** / **`onTurnComplete`** — both fire on every turn including the first turn of a continuation run.
|
|
|
|
### `onTurnStart`
|
|
|
|
- **`await`** persist **`uiMessages`** (full accumulated history including the new user turn) **before** the hook returns — `chat.agent` does not begin streaming until `onTurnStart` resolves, so this is what bounds "user message is durable before the stream".
|
|
|
|
<Warning>
|
|
**Don't use [`chat.defer()`](/ai-chat/background-injection#chat-defer-standalone) for the message write here.** `chat.defer` is fire-and-forget — the hook resolves before the write lands and the stream starts immediately. If the user refreshes mid-stream, the next page load reads `[]` from your DB, the resumed SSE stream pushes the assistant into an empty array, and the user's message disappears from the rendered conversation forever.
|
|
|
|
```ts
|
|
// ❌ Bad — non-blocking write, mid-stream refresh drops the user message.
|
|
onTurnStart: async ({ chatId, uiMessages }) => {
|
|
chat.defer(db.chat.update({ where: { id: chatId }, data: { messages: uiMessages } }));
|
|
},
|
|
|
|
// ✅ Good — awaited, durable before the model starts.
|
|
onTurnStart: async ({ chatId, uiMessages }) => {
|
|
await db.chat.update({ where: { id: chatId }, data: { messages: uiMessages } });
|
|
},
|
|
```
|
|
|
|
`chat.defer` is for writes whose timing doesn't matter for resume — analytics, audit logs, search-index updates, etc. Anything the next page load reads needs to land before the stream begins.
|
|
</Warning>
|
|
|
|
### `onTurnComplete`
|
|
|
|
- Persist **`uiMessages`** again with the **assistant** reply finalized.
|
|
- **Upsert session** with the fresh **`chatAccessToken`** and **`lastEventId`** from the event.
|
|
|
|
**`lastEventId`** lets the frontend [resume](/ai-chat/frontend) without replaying SSE events it already applied. Treat it as part of session state, not optional polish, if you care about duplicate chunks after refresh.
|
|
|
|
<Warning>
|
|
**Write the messages and `lastEventId` in a single transaction.** Both values are read in parallel on the next page load (one fetches the conversation, the other fetches the session). If a refresh races between the two writes, the page can see the assistant message persisted (full history) but a stale `lastEventId` from the previous turn. The transport then resumes from that stale cursor and replays this turn's chunks on top of the already-persisted assistant message, producing a duplicated render.
|
|
|
|
```ts
|
|
// ✅ Atomic — refresh on the next page load reads both writes consistently.
|
|
await db.$transaction([
|
|
db.chat.update({ where: { id: chatId }, data: { messages: uiMessages } }),
|
|
db.chatSession.upsert({
|
|
where: { id: chatId },
|
|
create: { id: chatId, publicAccessToken: chatAccessToken, lastEventId },
|
|
update: { publicAccessToken: chatAccessToken, lastEventId },
|
|
}),
|
|
]);
|
|
|
|
// ❌ Two awaits — narrow race window where messages are post-write but
|
|
// lastEventId is still pre-write. A page refresh that lands here will
|
|
// duplicate the assistant message on resume.
|
|
await db.chat.update({ where: { id: chatId }, data: { messages: uiMessages } });
|
|
await db.chatSession.upsert({ /* ... */ });
|
|
```
|
|
</Warning>
|
|
|
|
## Token renewal (app server)
|
|
|
|
The persisted PAT has a TTL (see **`chatAccessTokenTTL`** on **`chat.agent`**, default 1h). When the transport gets a **401** on a session-PAT-authed request, it calls your **`accessToken`** callback to mint a fresh PAT — no DB lookup required, since the session is keyed on `chatId` (which the transport already has).
|
|
|
|
Your `accessToken` callback typically just wraps `auth.createPublicToken`:
|
|
|
|
```ts
|
|
"use server";
|
|
import { auth } from "@trigger.dev/sdk";
|
|
|
|
export async function mintChatAccessToken(chatId: string) {
|
|
return auth.createPublicToken({
|
|
scopes: { read: { sessions: chatId }, write: { sessions: chatId } },
|
|
expirationTime: "1h",
|
|
});
|
|
}
|
|
```
|
|
|
|
If you want to keep your DB session row in sync, the transport's **`onSessionChange`** callback fires every time the cached PAT changes — persist the new value there.
|
|
|
|
No Trigger task code needs to run for renewal.
|
|
|
|
## Minimal pseudocode
|
|
|
|
```typescript
|
|
// Pseudocode — replace saveConversation / saveSession with your DB layer.
|
|
|
|
chat.agent({
|
|
id: "my-chat",
|
|
clientDataSchema: z.object({ userId: z.string() }),
|
|
|
|
onPreload: async ({ chatId, chatAccessToken, clientData }) => {
|
|
if (!clientData) return;
|
|
await ensureUser(clientData.userId);
|
|
await upsertConversation({ id: chatId, userId: clientData.userId /* ... */ });
|
|
await upsertSession({ chatId, publicAccessToken: chatAccessToken });
|
|
},
|
|
|
|
onChatStart: async ({ chatId, chatAccessToken, clientData, preloaded }) => {
|
|
if (preloaded) return;
|
|
// Fires once per chat — no continuation gate needed.
|
|
await ensureUser(clientData.userId);
|
|
await upsertConversation({ id: chatId, userId: clientData.userId /* ... */ });
|
|
await upsertSession({ chatId, publicAccessToken: chatAccessToken });
|
|
},
|
|
|
|
onTurnStart: async ({ chatId, uiMessages }) => {
|
|
// Awaited, not chat.defer — see the warning in `onTurnStart` above.
|
|
await saveConversationMessages(chatId, uiMessages);
|
|
},
|
|
|
|
onTurnComplete: async ({ chatId, uiMessages, chatAccessToken, lastEventId }) => {
|
|
// Atomic: messages + lastEventId must be readable consistently on resume.
|
|
// See the warning above for why a non-atomic write causes duplicate renders.
|
|
await db.$transaction([
|
|
saveConversationMessagesQuery(chatId, uiMessages),
|
|
upsertSessionQuery({ chatId, publicAccessToken: chatAccessToken, lastEventId }),
|
|
]);
|
|
},
|
|
|
|
run: async ({ messages, signal }) => {
|
|
/* streamText, etc. */
|
|
},
|
|
});
|
|
```
|
|
|
|
## Alternative: `hydrateMessages`
|
|
|
|
For apps that need the backend to be the single source of truth for message history — abuse prevention, branching conversations, or rollback support — use [`hydrateMessages`](/ai-chat/lifecycle-hooks#hydratemessages) instead of relying on the frontend's accumulated state.
|
|
|
|
With hydration, the hook loads messages from your database on every turn. The frontend's messages are ignored (except for the new user message, which arrives in `incomingMessages`):
|
|
|
|
```ts
|
|
import { chat, upsertIncomingMessage } from "@trigger.dev/sdk/ai";
|
|
|
|
export const myChat = chat.agent({
|
|
id: "my-chat",
|
|
hydrateMessages: async ({ chatId, trigger, incomingMessages }) => {
|
|
const record = await db.chat.findUnique({ where: { id: chatId } });
|
|
const stored = record?.messages ?? [];
|
|
|
|
// `upsertIncomingMessage` pushes a fresh user message and no-ops
|
|
// on HITL continuations (the runtime overlays the new tool-state
|
|
// advance onto the existing entry). See lifecycle hooks for the
|
|
// full pattern: /ai-chat/lifecycle-hooks#hydratemessages
|
|
if (upsertIncomingMessage(stored, { trigger, incomingMessages })) {
|
|
// Upsert, not update: on a head-start first turn no preload ran,
|
|
// so the row may not exist yet when this hook fires.
|
|
await db.chat.upsert({
|
|
where: { id: chatId },
|
|
create: { id: chatId, messages: stored },
|
|
update: { messages: stored },
|
|
});
|
|
}
|
|
|
|
return stored;
|
|
},
|
|
onTurnComplete: async ({ chatId, uiMessages, chatAccessToken, lastEventId }) => {
|
|
// Persist the response and refresh session state atomically — see the
|
|
// warning in the previous section for why these two writes have to be
|
|
// in the same transaction.
|
|
await db.$transaction([
|
|
db.chat.update({ where: { id: chatId }, data: { messages: uiMessages } }),
|
|
db.chatSession.upsert({
|
|
where: { id: chatId },
|
|
create: { id: chatId, publicAccessToken: chatAccessToken, lastEventId },
|
|
update: { publicAccessToken: chatAccessToken, lastEventId },
|
|
}),
|
|
]);
|
|
},
|
|
run: async ({ messages, signal }) => {
|
|
return streamText({ model: anthropic("claude-sonnet-4-5"), messages, abortSignal: signal });
|
|
},
|
|
});
|
|
```
|
|
|
|
This replaces the `onTurnStart` persistence pattern — the hook handles both loading and persisting the new message in one place.
|
|
|
|
Hydration composes with [Head Start](/ai-chat/fast-starts#with-hydratemessages): on a head-start first turn the route handler's history arrives as `incomingMessages`, and the write path must be an upsert because no preload ran to create the row.
|
|
|
|
## Design notes
|
|
|
|
- **`chatId`** is stable for the life of a thread and is the only identifier the transport persists. Runs come and go (idle continuation, upgrade, cancel/restart) but the chat keeps its identity.
|
|
- **`continuation: true`** means "same logical chat, new run" — refresh the persisted PAT, don't assume an empty conversation.
|
|
- The current `runId` is available on every hook event for telemetry / dashboard linking ("View this run"), but you don't need to persist it for resume to work — the transport addresses by `chatId`.
|
|
- Keep **task modules** that perform writes **out of** browser bundles; the pattern assumes persistence runs **in the worker** (or your BFF that the task calls).
|
|
|
|
## Complete example
|
|
|
|
End-to-end implementation across the three files involved: agent task, server actions, and React component.
|
|
|
|
<Warning>
|
|
The example below trusts raw `chatId` and returns rows without filtering by user. In a real multi-user app, **scope every query by the authenticated user** — read the user from your auth/session in each server action and add `where: { userId }` to all `db.chat.*` and `db.chatSession.*` queries. Without that, one client could read or delete another user's chat state, and `getAllSessions()` would leak other users' `publicAccessToken`s. The snippet keeps auth out of the way to focus on the persistence shape.
|
|
</Warning>
|
|
|
|
<CodeGroup>
|
|
```ts trigger/chat.ts
|
|
import { chat } from "@trigger.dev/sdk/ai";
|
|
import { streamText, stepCountIs } from "ai";
|
|
import { anthropic } from "@ai-sdk/anthropic";
|
|
import { z } from "zod";
|
|
import { db } from "@/lib/db";
|
|
|
|
export const myChat = chat.agent({
|
|
id: "my-chat",
|
|
clientDataSchema: z.object({
|
|
userId: z.string(),
|
|
}),
|
|
onChatStart: async ({ chatId, clientData }) => {
|
|
await db.chat.create({
|
|
data: { id: chatId, userId: clientData.userId, title: "New chat", messages: [] },
|
|
});
|
|
},
|
|
onTurnStart: async ({ chatId, uiMessages, runId, chatAccessToken }) => {
|
|
// Persist messages + session before streaming
|
|
await db.chat.update({
|
|
where: { id: chatId },
|
|
data: { messages: uiMessages },
|
|
});
|
|
await db.chatSession.upsert({
|
|
where: { id: chatId },
|
|
create: { id: chatId, runId, publicAccessToken: chatAccessToken },
|
|
update: { runId, publicAccessToken: chatAccessToken },
|
|
});
|
|
},
|
|
onTurnComplete: async ({ chatId, uiMessages, runId, chatAccessToken, lastEventId }) => {
|
|
// Persist assistant response + stream position atomically — see the
|
|
// race-condition warning earlier on this page.
|
|
await db.$transaction([
|
|
db.chat.update({
|
|
where: { id: chatId },
|
|
data: { messages: uiMessages },
|
|
}),
|
|
db.chatSession.upsert({
|
|
where: { id: chatId },
|
|
create: { id: chatId, runId, publicAccessToken: chatAccessToken, lastEventId },
|
|
update: { runId, publicAccessToken: chatAccessToken, lastEventId },
|
|
}),
|
|
]);
|
|
},
|
|
run: async ({ messages, signal }) => {
|
|
return streamText({
|
|
model: anthropic("claude-sonnet-4-5"),
|
|
messages,
|
|
abortSignal: signal,
|
|
stopWhen: stepCountIs(15),
|
|
});
|
|
},
|
|
});
|
|
```
|
|
|
|
```ts app/actions.ts
|
|
"use server";
|
|
|
|
import { auth } from "@trigger.dev/sdk";
|
|
import { chat } from "@trigger.dev/sdk/ai";
|
|
import { db } from "@/lib/db";
|
|
|
|
export const startChatSession = chat.createStartSessionAction("my-chat");
|
|
|
|
export async function mintChatAccessToken(chatId: string) {
|
|
return auth.createPublicToken({
|
|
scopes: { read: { sessions: chatId }, write: { sessions: chatId } },
|
|
expirationTime: "1h",
|
|
});
|
|
}
|
|
|
|
export async function getChatMessages(chatId: string) {
|
|
const found = await db.chat.findUnique({ where: { id: chatId } });
|
|
return found?.messages ?? [];
|
|
}
|
|
|
|
export async function getAllSessions() {
|
|
const sessions = await db.chatSession.findMany();
|
|
const result: Record<
|
|
string,
|
|
{
|
|
publicAccessToken: string;
|
|
lastEventId?: string;
|
|
}
|
|
> = {};
|
|
for (const s of sessions) {
|
|
result[s.id] = {
|
|
publicAccessToken: s.publicAccessToken,
|
|
lastEventId: s.lastEventId ?? undefined,
|
|
};
|
|
}
|
|
return result;
|
|
}
|
|
|
|
export async function deleteSession(chatId: string) {
|
|
await db.chatSession.delete({ where: { id: chatId } }).catch(() => {});
|
|
}
|
|
```
|
|
|
|
```tsx app/components/chat.tsx
|
|
"use client";
|
|
|
|
import { useChat } from "@ai-sdk/react";
|
|
import { useTriggerChatTransport } from "@trigger.dev/sdk/chat/react";
|
|
import type { myChat } from "@/trigger/chat";
|
|
import { mintChatAccessToken, startChatSession, deleteSession } from "@/app/actions";
|
|
|
|
export function Chat({ chatId, initialMessages, initialSessions }) {
|
|
const transport = useTriggerChatTransport<typeof myChat>({
|
|
task: "my-chat",
|
|
accessToken: ({ chatId }) => mintChatAccessToken(chatId),
|
|
startSession: ({ chatId, clientData }) =>
|
|
startChatSession({ chatId, clientData }),
|
|
clientData: { userId: currentUser.id }, // Type-checked against clientDataSchema
|
|
sessions: initialSessions,
|
|
onSessionChange: (id, session) => {
|
|
if (!session) deleteSession(id);
|
|
},
|
|
});
|
|
|
|
const { messages, sendMessage, stop, status } = useChat({
|
|
id: chatId,
|
|
messages: initialMessages,
|
|
transport,
|
|
resume: initialMessages.length > 0,
|
|
});
|
|
|
|
return (
|
|
<div>
|
|
{messages.map((m) => (
|
|
<div key={m.id}>
|
|
<strong>{m.role}:</strong>
|
|
{m.parts.map((part, i) =>
|
|
part.type === "text" ? <span key={i}>{part.text}</span> : null
|
|
)}
|
|
</div>
|
|
))}
|
|
|
|
<form
|
|
onSubmit={(e) => {
|
|
e.preventDefault();
|
|
const input = e.currentTarget.querySelector("input");
|
|
if (input?.value) {
|
|
sendMessage({ text: input.value });
|
|
input.value = "";
|
|
}
|
|
}}
|
|
>
|
|
<input placeholder="Type a message..." />
|
|
<button type="submit" disabled={status === "streaming"}>
|
|
Send
|
|
</button>
|
|
{status === "streaming" && (
|
|
<button type="button" onClick={stop}>
|
|
Stop
|
|
</button>
|
|
)}
|
|
</form>
|
|
</div>
|
|
);
|
|
}
|
|
```
|
|
|
|
</CodeGroup>
|
|
|
|
## See also
|
|
|
|
- [Lifecycle hooks](/ai-chat/lifecycle-hooks)
|
|
- [Session management](/ai-chat/frontend#session-management) — `resume`, `lastEventId`, transport
|
|
- [`chat.defer()`](/ai-chat/background-injection#chat-defer-standalone) — non-blocking writes during a turn
|
|
- [Code execution sandbox](/ai-chat/patterns/code-sandbox) — combines **`onWait`** / **`onComplete`** with this persistence model
|