ai-persistence-build-drizzle-adapter

name	ai-persistence/build-drizzle-adapter
description	Use when an app already runs Drizzle ORM and needs TanStack AI chat persistence — writes a chat-persistence.ts into the app against its existing db handle, schema file, and drizzle-kit journal. Covers the four tables (SQLite/Postgres/MySQL), the onConflict idempotency rules, JSON columns, and per-request bindings like D1.

Drizzle Chat Persistence

The deliverable is one file in the app — src/lib/chat-persistence.ts — exporting a ChatPersistence built from the app's existing Drizzle db. Plus four tables added to the app's existing schema file and a migration generated through the app's existing drizzle-kit setup.

Do not create a package, a second db instance, a migration runner, or a drizzle.config.ts. 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
Dialect	drizzle.config.ts dialect:, or the drizzle-orm/*-core import	sqlite-core vs pg-core vs mysql-core column builders
Schema file(s)	drizzle.config.ts schema: glob	Where the four tables go — append, never start a new file
The db handle	src/db/index.ts, src/db.ts, src/server/db.ts	Module singleton (export const db) vs factory (getDb())
Migration flow	drizzle.config.ts out:, the migrations/ or drizzle/ journal	Which generate/apply commands to tell the user to run
Naming conventions	Existing tables in the schema file	Table prefix, var casing, snake_case column names
Import alias	tsconfig.json paths	@/db, ~/db, #/db/index, or a relative path

Match what is already there. If their tables are chat_*-prefixed and their vars are camelCase, so are yours. If they already have a messages table for something else, prefix — the store code reads database names off the table objects, so any name works.

Never invent a migration path. Add the tables to their schema file, then have them run their own commands (npx drizzle-kit generate then migrate/push, or wrangler d1 migrations apply for D1). A parallel migration table behind their back is how schemas drift.

2. Add the tables to their schema file

SQLite. JSON payloads use text({ mode: 'json' }) so Drizzle round-trips objects for you; timestamps are integer epoch ms.

import {
  index,
  integer,
  primaryKey,
  sqliteTable,
  text,
} from 'drizzle-orm/sqlite-core'
import type { ModelMessage, TokenUsage } from '@tanstack/ai'
import type { InterruptRecord, RunStatus } from '@tanstack/ai-persistence'

export const chatThreads = sqliteTable('chat_threads', {
  threadId: text('thread_id').primaryKey(),
  messagesJson: text('messages_json', { mode: 'json' })
    .$type<Array<ModelMessage>>()
    .notNull(),
  updatedAt: integer('updated_at').notNull(),
})

export const chatRuns = sqliteTable(
  'chat_runs',
  {
    runId: text('run_id').primaryKey(),
    threadId: text('thread_id').notNull(),
    status: text().<>().(),
    : ().(),
    : (),
    : (),
    : (),
    : (, { :  }).<>(),
    : (),
    : (),
    : (, { :  }),
    : (),
  },
   [
    
    ().(table., table.),
    
    ().(table., table.),
  ],
)

  chatInterrupts = (, {
  : ().(),
  : ().(),
  : ().(),
  : ().<[]>().(),
  : ().(),
  : (),
  : (, { :  })
    .<<, >>()
    .(),
  : (, { :  }).<>(),
})

  chatMetadata = (
  ,
  {
    : ().(),
    : ().(),
    : (, { :  }).<>().(),
  },
   [({ : [table., table.] })],
)

updatedAt on threads is an app-owned extra, not part of any contract — the stores never read columns they do not know about, so add userId, tenant ids, or audit columns the same way (nullable or defaulted so inserts still succeed). The namespace column 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.

Postgres (drizzle-orm/pg-core): jsonb() for the JSON payloads, bigint({ mode: 'number' }) for epoch-ms timestamps (including detachedSince), integer() for driverEpoch, boolean() for cancelRequested, text() elsewhere, composite primaryKey on (namespace, key) unchanged. MySQL (drizzle-orm/mysql-core): json(), bigint({ mode: 'number' }), boolean() for cancelRequested, and varchar(..., { length: 255 }) for the primary-key columns. The store bodies below are identical across all three, only onConflictDoUpdate becomes onDuplicateKeyUpdate on MySQL, and the (status, detachedSince) / (threadId, startedAt) indexes carry over as is.

3. Write src/lib/chat-persistence.ts

The whole file. Idempotency is the entire game — the comments below mark the rules the conformance suite checks.

import { and, asc, desc, eq, isNotNull, lte } from 'drizzle-orm'
import { defineAIPersistence } from '@tanstack/ai-persistence'
import type { SQL } from 'drizzle-orm'
import type {
  ChatPersistence,
  InterruptRecord,
  InterruptStore,
  MessageStore,
  MetadataStore,
  RunRecord,
  RunStore,
} from '@tanstack/ai-persistence'

import { db } from '@/db'
import {
  chatInterrupts,
  chatMetadata,
  chatRuns,
  chatThreads,
} from '@/db/schema'

type Db = typeof db

// Records omit absent optionals so they compare cleanly against the reference
// in-memory backend.
function mapRun(row: typeof chatRuns.$inferSelect): RunRecord {
  return {
    runId: row.runId,
    threadId: row.threadId,
    status: row.status,
    startedAt: row.startedAt,
    ...(row.finishedAt !=  ? { : row. } : {}),
    ...(row. != 
      ? {
          : {
            : row.,
            ...(row. !=  ? { : row. } : {}),
          },
        }
      : {}),
    ...(row. !=  ? { : row. } : {}),
    ...(row. !=  ? { : row. } : {}),
    ...(row. !=  ? { : row. } : {}),
    ...(row. != 
      ? { : row. }
      : {}),
    ...(row. !=  ? { : row. } : {}),
  }
}

 ():  {
   {
    : row.,
    : row.,
    : row.,
    : row.,
    : row.,
    : row.,
    ...(row. !=  ? { : row. } : {}),
    ...(row. !=  ? { : row. } : {}),
  }
}

 ():  {
   {
     () {
       rows =  db
        .({ : chatThreads. })
        .(chatThreads)
        .((chatThreads., threadId))
        .()
      
       rows[]?. ?? []
    },
    
     () {
       updatedAt = .()
       db
        .(chatThreads)
        .({ threadId, : messages, updatedAt })
        .({
          : chatThreads.,
          : { : messages, updatedAt },
        })
    },
  }
}

 ():  {
    () {
     rows =  db
      .()
      .(chatRuns)
      .((chatRuns., runId))
      .()
     rows[] ? (rows[]) : 
  }

   {
    get,
    
    
     () {
       existing =  (runId)
       (existing)  existing

       db
        .(chatRuns)
        .({ runId, threadId, : status ?? , startedAt })
        .({ : chatRuns. })

      
      
       stored =  (runId)
       (
        stored ?? { runId, threadId, : status ?? , startedAt }
      )
    },
    
     () {
       : < chatRuns.> = {}
       (patch. !== ) set. = patch.
       (patch. !== ) set. = patch.
      
      
       (patch. !== ) {
        set. = patch..
        set. = patch.. ?? 
      }
       (patch. !== ) set. = patch.
      
      
      
      
      
      
      
       (  patch) set. = patch. ?? 
       (  patch)
        set. = patch. ?? 
       (  patch)
        set. = patch. ?? 
       (  patch) set. = patch. ?? 
       (.(set). === ) 

       db.(chatRuns).(set).((chatRuns., runId))
    },
    
     () {
       rows =  db
        .()
        .(chatRuns)
        .(
          ((chatRuns., threadId), (chatRuns., )),
        )
        .((chatRuns.))
        .()
       rows[] ? (rows[]) : 
    },
    
    
     () {
       rows =  db
        .()
        .(chatRuns)
        .((chatRuns., threadId))
        .((chatRuns.))
       rows.(mapRun)
    },
    
    
     () {
       cutoff = now - ttlMs
       rows =  db
        .()
        .(chatRuns)
        .(
          (
            (chatRuns., ),
            (chatRuns.),
            (chatRuns., cutoff),
          ),
        )
       rows.(mapRun)
    },
  }
}

 ():  {
  
    =  () => {
     rows =  db
      .()
      .(chatInterrupts)
      .(where)
      .((chatInterrupts.))
     rows.(mapInterrupt)
  }

   {
    
    
     () {
       db
        .(chatInterrupts)
        .({
          : record.,
          : record.,
          : record.,
          : ,
          : record.,
          : record.,
          ...(record. !== 
            ? { : record. }
            : {}),
        })
        .({ : chatInterrupts. })
    },
     () {
       db
        .(chatInterrupts)
        .({
          : ,
          : .(),
          ...(response !==  ? { : response } : {}),
        })
        .((chatInterrupts., interruptId))
    },
     () {
       db
        .(chatInterrupts)
        .({ : , : .() })
        .((chatInterrupts., interruptId))
    },
     () {
       rows =  db
        .()
        .(chatInterrupts)
        .((chatInterrupts., interruptId))
        .()
       rows[] ? (rows[]) : 
    },
    :  ((chatInterrupts., threadId)),
    : 
      (
        (
          (chatInterrupts., threadId),
          (chatInterrupts., ),
        ),
      ),
    :  ((chatInterrupts., runId)),
    : 
      (
        (
          (chatInterrupts., runId),
          (chatInterrupts., ),
        ),
      ),
  }
}

 ():  {
   {
     () {
       rows =  db
        .({ : chatMetadata. })
        .(chatMetadata)
        .(
          ((chatMetadata., namespace), (chatMetadata., key)),
        )
        .()
       rows[]?. ?? 
    },
     () {
      
      
       (value == ) {
          (
          ,
        )
      }
       db
        .(chatMetadata)
        .({ namespace, key, : value })
        .({
          : [chatMetadata., chatMetadata.],
          : { : value },
        })
    },
     () {
       db
        .(chatMetadata)
        .(
          ((chatMetadata., namespace), (chatMetadata., key)),
        )
    },
  }
}


  :  = ({
  : {
    : (db),
    : (db),
    : (db),
    : (db),
  },
})

Annotate ChatPersistence — bare AIPersistence is the all-optional bag and withPersistence rejects it. There is no locks store: stores accepts only those four keys, and coordination is wired separately with withLocks (see ai-core/locks).

If db is per-request

Workers/D1 and any request-scoped client cannot read a binding at module scope. Export a factory instead, and call it inside the handler:

type Db = ReturnType<typeof getDb>

export function chatPersistence(): ChatPersistence {
  const db = getDb()
  return defineAIPersistence({
    stores: {
      messages: createMessageStore(db),
      runs: createRunStore(db),
      interrupts: createInterruptStore(db),
      metadata: createMetadataStore(db),
    },
  })
}

The store factories are unchanged — only the export flips from a const to a function. For D1 specifically, see ai-persistence/build-cloudflare-adapter.

4. Wire it into the chat route

import {
  chat,
  chatParamsFromRequest,
  toServerSentEventsResponse,
} from '@tanstack/ai'
import { openaiText } from '@tanstack/ai-openai'
import { withPersistence } from '@tanstack/ai-persistence'
import { chatPersistence } from '@/lib/chat-persistence'

export async function POST(request: Request) {
  const params = await chatParamsFromRequest(request)
  const stream = chat({
    adapter: openaiText('gpt-5.5'),
    messages: params.messages,
    threadId: params.threadId,
    runId: params.runId,
    ...(params.resume ? { resume: params.resume } : {}),
    middleware: [withPersistence(chatPersistence)],
  })
  return toServerSentEventsResponse(stream)
}

threadId is a bare string to the stores. Authorize thread access at the route — derive the user from the session, never trust a client-supplied id.

5. Verify

import { runPersistenceConformance } from '@tanstack/ai-persistence/testkit'
import { chatPersistence } from '../src/lib/chat-persistence'

runPersistenceConformance('app-drizzle', () => chatPersistence, {
  skip: ['generationRuns', 'artifacts', 'blobs'],
})

Point it at a throwaway database (:memory: SQLite, a scratch schema, PGlite) that has the migration applied, and reset between runs. The suite covers all seven stores, so a chat adapter declares the generation half it omits; drop the skip once you add those tables. skip never accepts 'locks', which is not a store.

If your recipe leaves an optional runs method (listByThread/listReclaimable) unimplemented, declare it with skipMethods, e.g. { skipMethods: ['runs.listByThread'] }. An omitted method that is not declared fails the suite instead of silently passing.

Only if you are publishing this as a package

Everything above assumes the file lives in the app. If instead you are shipping a reusable drizzle adapter to npm, the same store bodies apply, plus:

• Peer deps @tanstack/ai, @tanstack/ai-persistence, drizzle-orm >=0.44.0; dev dep drizzle-kit. Keep the module root free of Node built-ins so it is edge-safe, and put any node:sqlite convenience factory behind a /sqlite subpath.
• Type db structurally so a consumer's client is assignable: Pick<BaseSQLiteDatabase<'sync' | 'async', unknown>, 'select' | 'insert' | 'update' | 'delete'>.
• Multi-dialect: take a provider: 'sqlite' | 'pg' option, declare overloads so db and schema must agree, and add a runtime dialect check so a mismatched pair fails at construction rather than on first query.
• BYO schema: accept drizzlePersistence(db, { schema }), validate the tables/columns exist at construction, and pin the required column shapes with a compile-time contract type.
• Never bundle SQL migrations or a runner. Either re-export the stock tables from a /sqlite-schema subpath so the consumer's drizzle-kit picks them up, or emit an owned starter schema file with a small CLI. An opt-in ensureTables(db) issuing CREATE TABLE IF NOT EXISTS is fine for local dev, kept clearly separate from their journal. Pick one DDL owner per database.
• Run runPersistenceConformance once per dialect.