Skip to content
TanStackAI
Persistence (Advanced)

Build a Generation Adapter (Advanced)

A generation adapter stores the run and the output bytes. A reload can then show a video that is still in progress, and the file remains after the provider URL expires.

Keep both in your own database with three stores. This page builds them against SQLite (the built-in node:sqlite module of Node). Plan about 30 minutes.

Before you start:

Media generation does not use the chat runs store. It uses these stores:

  • Required: generationRuns, a GenerationRunStore keyed by runId (the run id that a generation mints). It is the generation version of runs. Each record also holds the threadId, which is the slot that the run fills.
  • Optional, to keep the generated bytes: artifacts (the metadata) and blobs (the bytes). Provide these two together, or neither.

1. The schema

Three tables. They are separate from the four tables of a chat adapter.

ts
// generation-schema.ts
export const GENERATION_SCHEMA_SQL = `
CREATE TABLE IF NOT EXISTS generation_runs (
  run_id text PRIMARY KEY NOT NULL,
  thread_id text NOT NULL,
  activity text NOT NULL,
  provider text NOT NULL,
  model text NOT NULL,
  status text NOT NULL,
  started_at integer NOT NULL,
  finished_at integer,
  error_json text,
  result_json text,
  artifacts_json text,
  usage_json text
);
CREATE TABLE IF NOT EXISTS artifacts (
  artifact_id text PRIMARY KEY NOT NULL,
  run_id text NOT NULL,
  thread_id text NOT NULL,
  blob_key text NOT NULL,
  name text NOT NULL,
  mime_type text NOT NULL,
  size integer NOT NULL,
  source_url text,
  created_at integer NOT NULL
);
CREATE INDEX IF NOT EXISTS artifacts_run_order
  ON artifacts (run_id, created_at, artifact_id);
CREATE INDEX IF NOT EXISTS artifacts_thread_order
  ON artifacts (thread_id, created_at, artifact_id);
CREATE TABLE IF NOT EXISTS blobs (
  key text PRIMARY KEY NOT NULL,
  bytes blob NOT NULL,
  size integer NOT NULL,
  etag text NOT NULL,
  content_type text,
  custom_metadata_json text,
  created_at integer NOT NULL,
  updated_at integer NOT NULL
);
`

2. Generation runs: idempotent create, patch, latest-for-thread

The store has three rules:

  • createOrResume is idempotent. A second call for a runId returns the stored record unchanged. A resumed run then keeps its startedAt, activity, and status. INSERT ... ON CONFLICT DO NOTHING does this.
  • update on an unknown runId does nothing.
  • findLatestForThread returns the run with the greatest startedAt for a thread. reconstructGeneration calls it to restore the last generation for a thread when a client mounts.
ts
// generation-run-store.ts
import { DatabaseSync } from 'node:sqlite'
import { defineGenerationRunStore } from '@tanstack/ai-persistence'
import type {
  GenerationRunRecord,
  GenerationRunStatus,
} from '@tanstack/ai-persistence'

function toGenerationRunStatus(value: unknown): GenerationRunStatus {
  switch (value) {
    case 'running':
    case 'completed':
    case 'failed':
    case 'interrupted':
    case 'aborted':
      return value
    default:
      throw new TypeError(`Unexpected generation run status: ${String(value)}`)
  }
}

// `node:sqlite` types columns as a SQL-value union, so coerce/narrow each field
// (String / Number / typeof) and JSON-parse the text columns, with no cast.
function mapGenerationRun(row: Record<string, unknown>): GenerationRunRecord {
  return {
    runId: String(row.run_id),
    threadId: String(row.thread_id),
    activity: String(row.activity),
    provider: String(row.provider),
    model: String(row.model),
    status: toGenerationRunStatus(row.status),
    startedAt: Number(row.started_at),
    ...(row.finished_at != null ? { finishedAt: Number(row.finished_at) } : {}),
    ...(typeof row.error_json === 'string'
      ? { error: JSON.parse(row.error_json) }
      : {}),
    ...(typeof row.result_json === 'string'
      ? { result: JSON.parse(row.result_json) }
      : {}),
    ...(typeof row.artifacts_json === 'string'
      ? { artifacts: JSON.parse(row.artifacts_json) }
      : {}),
    ...(typeof row.usage_json === 'string'
      ? { usage: JSON.parse(row.usage_json) }
      : {}),
  }
}

export function createGenerationRunStore(db: DatabaseSync) {
  const select = db.prepare('SELECT * FROM generation_runs WHERE run_id = ?')
  const insert = db.prepare(
    `INSERT INTO generation_runs
       (run_id, thread_id, activity, provider, model, status, started_at)
     VALUES (?, ?, ?, ?, ?, ?, ?)
     ON CONFLICT(run_id) DO NOTHING`,
  )
  const latest = db.prepare(
    `SELECT * FROM generation_runs WHERE thread_id = ?
     ORDER BY started_at DESC LIMIT 1`,
  )
  return defineGenerationRunStore({
    async createOrResume(input) {
      const existing = select.get(input.runId)
      if (existing) return mapGenerationRun(existing)
      const status: GenerationRunStatus = input.status ?? 'running'
      insert.run(
        input.runId,
        input.threadId,
        input.activity,
        input.provider,
        input.model,
        status,
        input.startedAt,
      )
      return {
        runId: input.runId,
        threadId: input.threadId,
        activity: input.activity,
        provider: input.provider,
        model: input.model,
        status,
        startedAt: input.startedAt,
      }
    },
    async update(runId, patch) {
      const sets: Array<string> = []
      const params: Array<string | number> = []
      if (patch.status !== undefined) {
        sets.push('status = ?')
        params.push(patch.status)
      }
      if (patch.finishedAt !== undefined) {
        sets.push('finished_at = ?')
        params.push(patch.finishedAt)
      }
      if (patch.error !== undefined) {
        sets.push('error_json = ?')
        params.push(JSON.stringify(patch.error))
      }
      if (patch.result !== undefined) {
        sets.push('result_json = ?')
        params.push(JSON.stringify(patch.result))
      }
      if (patch.artifacts !== undefined) {
        sets.push('artifacts_json = ?')
        params.push(JSON.stringify(patch.artifacts))
      }
      if (patch.usage !== undefined) {
        sets.push('usage_json = ?')
        params.push(JSON.stringify(patch.usage))
      }
      // Empty patch, or an unknown run id, touches nothing (UPDATE no-ops).
      if (sets.length === 0) return
      params.push(runId)
      db.prepare(
        `UPDATE generation_runs SET ${sets.join(', ')} WHERE run_id = ?`,
      ).run(...params)
    },
    async get(runId) {
      const row = select.get(runId)
      return row ? mapGenerationRun(row) : null
    },
    // The most recent run linked to a thread. `reconstructGeneration` calls this
    // so a server-driven client (`persistence: true`) hydrates the last
    // generation for its thread by the stable thread id, without a run id.
    async findLatestForThread(threadId) {
      const row = latest.get(threadId)
      return row ? mapGenerationRun(row) : null
    },
  })
}

3. Artifacts: media metadata

ArtifactStore holds one metadata row for each generated file: its runId, mimeType, size, and createdAt. The bytes go in the blob store (step 4).

  • save is an upsert.
  • list(runId) returns every artifact for a run, or [] when there are none.
  • listForThread(threadId) returns every artifact for the thread, in exact (createdAt, artifactId) ascending order. It must return the complete thread history, not a page and not only the latest run. Snapshot capture uses this list.
  • delete and deleteForRun are required. You store media durably so that you can control retention and erasure. They mirror BlobStore.delete.

Persist blobKey exactly as you get it. It records where the bytes went. A storageKey mapper can put them anywhere, so a reader cannot calculate the path again. A record without blobKey does not load.

ts
// artifact-store.ts
import { DatabaseSync } from 'node:sqlite'
import { defineArtifactStore } from '@tanstack/ai-persistence'
import type { ArtifactRecord } from '@tanstack/ai-persistence'

function mapArtifact(row: Record<string, unknown>): ArtifactRecord {
  return {
    artifactId: String(row.artifact_id),
    runId: String(row.run_id),
    threadId: String(row.thread_id),
    blobKey: String(row.blob_key),
    name: String(row.name),
    mimeType: String(row.mime_type),
    size: Number(row.size),
    ...(typeof row.source_url === 'string'
      ? { sourceUrl: row.source_url }
      : {}),
    createdAt: Number(row.created_at),
  }
}

export function createArtifactStore(db: DatabaseSync) {
  const upsert = db.prepare(
    `INSERT INTO artifacts
       (artifact_id, run_id, thread_id, blob_key, name, mime_type, size, source_url, created_at)
     VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
     ON CONFLICT(artifact_id) DO UPDATE SET
       run_id = excluded.run_id, thread_id = excluded.thread_id,
       blob_key = excluded.blob_key, name = excluded.name,
       mime_type = excluded.mime_type, size = excluded.size,
       source_url = excluded.source_url, created_at = excluded.created_at`,
  )
  const selectOne = db.prepare('SELECT * FROM artifacts WHERE artifact_id = ?')
  const byRun = db.prepare(
    'SELECT * FROM artifacts WHERE run_id = ? ORDER BY created_at ASC, artifact_id ASC',
  )
  const byThread = db.prepare(
    'SELECT * FROM artifacts WHERE thread_id = ? ORDER BY created_at ASC, artifact_id ASC',
  )
  return defineArtifactStore({
    async save(record) {
      upsert.run(
        record.artifactId,
        record.runId,
        record.threadId,
        record.blobKey,
        record.name,
        record.mimeType,
        record.size,
        record.sourceUrl ?? null,
        record.createdAt,
      )
    },
    async get(artifactId) {
      const row = selectOne.get(artifactId)
      return row ? mapArtifact(row) : null
    },
    async list(runId) {
      return byRun.all(runId).map(mapArtifact)
    },
    async listForThread(threadId) {
      return byThread.all(threadId).map(mapArtifact)
    },
    async delete(artifactId) {
      db.prepare('DELETE FROM artifacts WHERE artifact_id = ?').run(artifactId)
    },
    async deleteForRun(runId) {
      db.prepare('DELETE FROM artifacts WHERE run_id = ?').run(runId)
    },
  })
}

4. Blobs: the bytes

BlobStore is a small object store. withGenerationPersistence writes each generated file under the key artifacts/<runId>/<artifactId>. A prefix list, list({ prefix: 'artifacts/<runId>/' }), then finds all the media of a run.

  • put accepts any BlobBody: a stream, a buffer, a string, or a Blob. The toBytes helper below turns it into bytes.
  • get honors options.range and returns only that slice. A media player needs this to seek. See the BlobStore contracts.
  • list matches prefix literally and pages with a keyset cursor.
ts
// blob-store.ts
import { DatabaseSync } from 'node:sqlite'
import { defineBlobStore, resolveBlobRange } from '@tanstack/ai-persistence'
import type {
  BlobBody,
  BlobObject,
  BlobRecord,
} from '@tanstack/ai-persistence'

async function toBytes(body: BlobBody): Promise<Uint8Array> {
  if (typeof body === 'string') return new TextEncoder().encode(body)
  if (body instanceof ArrayBuffer) return new Uint8Array(body.slice(0))
  if (ArrayBuffer.isView(body)) {
    return new Uint8Array(body.buffer, body.byteOffset, body.byteLength).slice()
  }
  if (body instanceof Blob) {
    return new Uint8Array(await body.arrayBuffer())
  }
  // ReadableStream<Uint8Array>: drain it into one buffer.
  const reader = body.getReader()
  const chunks: Array<Uint8Array> = []
  let total = 0
  for (;;) {
    const { done, value } = await reader.read()
    if (done) break
    chunks.push(value)
    total += value.byteLength
  }
  const bytes = new Uint8Array(total)
  let offset = 0
  for (const chunk of chunks) {
    bytes.set(chunk, offset)
    offset += chunk.byteLength
  }
  return bytes
}

function mapBlobRecord(row: Record<string, unknown>): BlobRecord {
  return {
    key: String(row.key),
    ...(row.size != null ? { size: Number(row.size) } : {}),
    ...(typeof row.etag === 'string' ? { etag: row.etag } : {}),
    ...(typeof row.content_type === 'string'
      ? { contentType: row.content_type }
      : {}),
    ...(typeof row.custom_metadata_json === 'string'
      ? { customMetadata: JSON.parse(row.custom_metadata_json) }
      : {}),
    ...(row.created_at != null ? { createdAt: Number(row.created_at) } : {}),
    ...(row.updated_at != null ? { updatedAt: Number(row.updated_at) } : {}),
  }
}

function blobObject(
  record: BlobRecord,
  bytes: Uint8Array,
  range?: { offset: number; length: number },
): BlobObject {
  return {
    ...record,
    // `size` keeps describing the whole object; `range` describes these bytes.
    ...(range ? { range } : {}),
    body: new ReadableStream<Uint8Array>({
      start(controller) {
        controller.enqueue(bytes.slice())
        controller.close()
      },
    }),
    arrayBuffer() {
      const copy = new ArrayBuffer(bytes.byteLength)
      new Uint8Array(copy).set(bytes)
      return Promise.resolve(copy)
    },
    text: () => Promise.resolve(new TextDecoder().decode(bytes)),
  }
}

export function createBlobStore(db: DatabaseSync) {
  const upsert = db.prepare(
    `INSERT INTO blobs
       (key, bytes, size, etag, content_type, custom_metadata_json, created_at, updated_at)
     VALUES (?, ?, ?, ?, ?, ?, ?, ?)
     ON CONFLICT(key) DO UPDATE SET
       bytes = excluded.bytes, size = excluded.size, etag = excluded.etag,
       content_type = excluded.content_type,
       custom_metadata_json = excluded.custom_metadata_json,
       updated_at = excluded.updated_at`,
  )
  const selectCreated = db.prepare('SELECT created_at FROM blobs WHERE key = ?')
  const selectOne = db.prepare('SELECT * FROM blobs WHERE key = ?')
  // Metadata without the bytes, and the bounded slice: a ranged read must not
  // load the whole object to hand back a piece of it.
  const selectMeta = db.prepare(
    `SELECT key, size, etag, content_type, custom_metadata_json,
            created_at, updated_at
       FROM blobs WHERE key = ?`,
  )
  const selectSlice = db.prepare(
    'SELECT substr(bytes, ?, ?) AS bytes FROM blobs WHERE key = ?',
  )
  return defineBlobStore({
    async put(key, body, options) {
      const bytes = await toBytes(body)
      const now = Date.now()
      const prior = selectCreated.get(key)
      const createdAt =
        prior && prior.created_at != null ? Number(prior.created_at) : now
      const etag = String(now)
      upsert.run(
        key,
        bytes,
        bytes.byteLength,
        etag,
        options?.contentType ?? null,
        options?.customMetadata ? JSON.stringify(options.customMetadata) : null,
        createdAt,
        now,
      )
      return {
        key,
        size: bytes.byteLength,
        etag,
        createdAt,
        updatedAt: now,
        ...(options?.contentType !== undefined
          ? { contentType: options.contentType }
          : {}),
        ...(options?.customMetadata !== undefined
          ? { customMetadata: options.customMetadata }
          : {}),
      }
    },
    async get(key, options) {
      if (!options?.range) {
        const row = selectOne.get(key)
        if (!row) return null
        const bytes =
          row.bytes instanceof Uint8Array ? row.bytes : new Uint8Array()
        return blobObject(mapBlobRecord(row), bytes)
      }
      // Metadata first, WITHOUT the bytes, so the clamp costs no I/O...
      const meta = selectMeta.get(key)
      if (!meta) return null
      const served = resolveBlobRange(Number(meta.size), options.range)
      // ...then let SQLite cut the slice (`substr` is 1-based and byte-wise
      // over a BLOB). Reading the row whole and slicing in JS would load the
      // entire object on every video seek, the cost ranges exist to avoid.
      const slice = selectSlice.get(served.offset + 1, served.length, key)
      if (!slice) return null
      const bytes =
        slice.bytes instanceof Uint8Array ? slice.bytes : new Uint8Array()
      return blobObject(mapBlobRecord(meta), bytes, served)
    },
    async head(key) {
      // Metadata only: never pull the bytes to answer a question about them.
      const row = selectMeta.get(key)
      return row ? mapBlobRecord(row) : null
    },
    async delete(key) {
      db.prepare('DELETE FROM blobs WHERE key = ?').run(key)
    },
    async list(options) {
      if (options?.limit === 0) return { objects: [], truncated: false }
      // Match the prefix with `substr(...) = ?` rather than LIKE: SQLite's LIKE
      // is case-INsensitive for ASCII and treats `%`/`_` as wildcards, while the
      // contract says a prefix matches literally and case-sensitively. Then page
      // with a keyset cursor (keys strictly greater than the last one returned).
      const prefix = options?.prefix ?? ''
      const params: Array<string | number> = [prefix, prefix]
      let where = 'substr(key, 1, length(?)) = ?'
      if (options?.cursor !== undefined) {
        where += ' AND key > ?'
        params.push(options.cursor)
      }
      let sql = `SELECT * FROM blobs WHERE ${where} ORDER BY key ASC`
      const limit = options?.limit
      if (limit !== undefined) {
        sql += ' LIMIT ?' // fetch one extra row to detect truncation
        params.push(limit + 1)
      }
      const rows = db
        .prepare(sql)
        .all(...params)
        .map(mapBlobRecord)
      if (limit !== undefined && rows.length > limit) {
        const page = rows.slice(0, limit)
        const cursor = page.at(-1)?.key
        return {
          objects: page,
          truncated: true,
          ...(cursor !== undefined ? { cursor } : {}),
        }
      }
      return { objects: rows, truncated: false }
    },
  })
}

5. Assemble a generation adapter

Pass the three stores to defineAIPersistence.

  • generationRuns alone is a valid generation adapter. It keeps run records and no bytes.
  • Add artifacts and blobs together to keep the media.
ts
// generation-persistence.ts
import { DatabaseSync } from 'node:sqlite'
import { defineAIPersistence } from '@tanstack/ai-persistence'
// The three generation store factories and the schema string, from your modules.
import { createArtifactStore } from './artifact-store'
import { createBlobStore } from './blob-store'
import { createGenerationRunStore } from './generation-run-store'
import { GENERATION_SCHEMA_SQL } from './generation-schema'

export function generationPersistence(options: {
  url: string
  migrate?: boolean
}) {
  const db = new DatabaseSync(options.url)
  if (options.migrate) db.exec(GENERATION_SCHEMA_SQL)
  return defineAIPersistence({
    stores: {
      generationRuns: createGenerationRunStore(db),
      artifacts: createArtifactStore(db),
      blobs: createBlobStore(db),
    },
  })
}

6. Use it in your generation route

Create the .data folder. Then pass the adapter to withGenerationPersistence:

ts
import {
  generateImage,
  generationParamsFromRequest,
  toServerSentEventsResponse,
} from '@tanstack/ai'
import { openaiImage } from '@tanstack/ai-openai'
import { withGenerationPersistence } from '@tanstack/ai-persistence'
import { generationPersistence } from './generation-persistence'

// Creating tables on open is for local development. See Migrations.
const persistence = generationPersistence({
  url: 'file:.data/generation.sqlite',
  migrate: true,
})

export async function POST(request: Request) {
  const { input, threadId } = await generationParamsFromRequest(
    'image',
    request,
  )
  if (typeof input.prompt !== 'string') {
    throw new Error('This endpoint accepts text image prompts only.')
  }

  const stream = generateImage({
    adapter: openaiImage('gpt-image-2.5-flare'),
    prompt: input.prompt,
    threadId,
    stream: true,
    middleware: [withGenerationPersistence(persistence)],
  })
  return toServerSentEventsResponse(stream)
}

Generate an image with a threadId. The generation_runs table now has the run, and the blobs table has the image bytes. To restore the run in the browser and serve the bytes back, wire the client and the GET routes from Generation persistence and Keep generated files.

How it works

  • withGenerationPersistence calls createOrResume when a generation starts, then update with the status, the result metadata, and the artifact references.
  • With artifacts and blobs, it writes each generated file to blobs and records an ArtifactRecord for it.
  • reconstructGeneration calls findLatestForThread to find the run to restore for a slot.
  • retrieveArtifact and retrieveBlob read the record and the bytes back for your serve route.

Advanced

One backend for chat and generation

Fold these stores into an existing chat adapter with composePersistence. One backend then serves both withPersistence and withGenerationPersistence. See Controls.

Prove the stores

Run the conformance suite against the adapter. A generation-only adapter skips the chat stores (messages, activities, runs, interrupts, metadata). See Verify with the conformance suite.

Next

Edit on GitHub
Build a Chat Adapter
Build a Sandbox Adapter