# Build a Generation Adapter

# Build a Generation Adapter

Image, audio, and video runs need their own persistence: a run record so a reload
can find the last generation for a slot, and byte storage so the media itself
comes back. This page builds those three stores against SQLite (Node's built-in
`node:sqlite`).

Read [Build your own adapter](./build-your-own-adapter) first for the shape of an
adapter and which stores your app needs. Every method signature and invariant is
in the [store reference](./store-reference).

[Media generation](./generation-persistence) persists differently from chat: it
does not use the chat `runs` store at all.

- **Required:** a `generationRuns` store, a `GenerationRunStore` keyed by
  `runId` (the run/request id a generation mints). It is the counterpart to
  `runs`.
- **Optional, to keep the generated bytes:** an `artifacts` store (metadata) and
  a `blobs` store (the bytes). These two must be provided **together**.

`threadId` is the slot the run belongs to, recorded on each run record.

Three tables, independent of the four a
[chat adapter](./build-your-own-chat-adapter) uses:

```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,
  name text NOT NULL,
  mime_type text NOT NULL,
  size integer NOT NULL,
  source_url text,
  created_at integer NOT NULL
);
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
);
```

## Generation runs: idempotent create, patch, latest-for-thread

`GenerationRunStore` is the generation analogue of `RunStore`. Three contracts
to hold:

- `createOrResume` is idempotent. A second call for a `runId` returns the stored
  record unchanged, so resuming a run never resets its `startedAt`, `activity`,
  or status. `INSERT ... ON CONFLICT DO NOTHING` gives you that.
- `update` on an unknown `runId` is a no-op.
- `findLatestForThread` returns the run with the greatest `startedAt` linked to a
  thread. `reconstructGeneration` calls it to hydrate the last generation for a
  thread on a server-driven client's mount.

```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':
      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) }
      : {}),
  }
}

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
    },
  })
}
```

## Artifacts: media metadata

`ArtifactStore` holds one metadata row per generated file: its `runId`,
`mimeType`, `size`, and a `createdAt`. The bytes live in the blob store below.

- `save` is an upsert.
- `list(runId)` returns every artifact for a run, `[]` when there are none.
- `delete` / `deleteForRun` are required. Retention and erasure are the point of
  storing media durably, and they mirror `BlobStore.delete`.

Persist `blobKey` verbatim. It records where these bytes actually went, and a
`storageKey` mapper can put them anywhere, so a reader cannot recompute the
path. `resolveArtifactBlobKey(record)` falls back to the default convention
only for rows written before the column existed. Drop it and every artifact
stored under a custom key becomes unreadable.

```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),
    ...(typeof row.blob_key === 'string' ? { blobKey: 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),
  }
}

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',
  )
  return defineArtifactStore({
    async save(record) {
      upsert.run(
        record.artifactId,
        record.runId,
        record.threadId,
        record.blobKey ?? null,
        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 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)
    },
  })
}
```

## Blobs: the bytes

`BlobStore` is a small object store. `withGenerationPersistence` writes each
generated file under the key `artifacts/<runId>/<artifactId>`, so a
prefix-filtered `list({ prefix: 'artifacts/<runId>/' })` enumerates a run's
media.

- `put` accepts any `BlobBody`: a stream, buffer, string, or `Blob`. The helper
  below normalizes it to bytes.
- `get` honours `options.range` by returning only that slice, which is what a
  media player's seeking needs. See the
  [`BlobStore` contracts](./store-reference#blobstore).
- `list` matches `prefix` literally and pages with a keyset cursor.

```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)),
  }
}

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 }
    },
  })
}
```

## Assemble a generation adapter

Hand the three stores to `defineAIPersistence` the same way. `generationRuns` alone is a
valid generation adapter (run records, no byte storage); add `artifacts` +
`blobs` (together) to keep the media:

```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),
    },
  })
}
```

Pass the result to `withGenerationPersistence` on a `generateImage` /
`generateVideo` / … call; see [Generation persistence](./generation-persistence).
You can also fold these stores into an existing chat adapter with
`composePersistence`, so one backend serves both `withPersistence` and
`withGenerationPersistence`.

## Where to go next

- [Generation persistence](./generation-persistence): wire this adapter into a generation route.
- [Keep generated files](./keep-generated-files): serve the stored bytes back.
- [Build your own adapter](./build-your-own-adapter#verify-with-the-conformance-suite): run the conformance suite against what you just built.