ai-persistence-build-cloudflare-artifact-store

name

ai-persistence/build-cloudflare-artifact-store

description

Use when a Cloudflare Worker needs durable byte storage for TanStack AI generated media (images, audio, video, transcripts) — writes a BlobStore backed by R2 and an ArtifactStore backed by D1, composes them onto the generation persistence so withGenerationPersistence persists artifact bytes, and serves them back from a Worker GET route. Includes one-line sketches for S3, GCS, Vercel Blob, Supabase, and a dev filesystem BlobStore.

Cloudflare Artifact + Blob Store

withGenerationPersistence(persistence) needs only stores.generationRuns to track a generation's lifecycle. Add stores.artifacts (metadata) and stores.blobs (the bytes) — both, or neither — and the middleware also persists the generated media: image/audio/TTS/video/transcription bytes land at blob key artifacts/<runId>/<artifactId>, with an ArtifactRecord row describing each.

The deliverable is one file in the Worker — e.g. src/lib/generation-persistence.ts — exporting a factory that builds an AIPersistence from the request's R2 + D1 bindings, plus a GET route that serves artifact bytes with retrieveArtifact / retrieveBlob.

Read the sibling ai-persistence/build-cloudflare-adapter skill for the per-request-binding rule, wrangler config shape, D1 migration workflow, and the chat (generation-run/message) side. This skill covers only the two byte-storage stores and how to compose them.

The two contracts

Both come from @tanstack/ai-persistence. defineBlobStore / defineArtifactStore type an object literal inline (autocomplete + contract checking, no separate annotation).

// BlobStore — the byte layer. R2 backs it.
interface BlobStore {
  put: (
    key: string,
    body: BlobBody,
    options?: BlobPutOptions,
  ) => Promise<BlobRecord>
  // metadata + byte accessors; `options.range` reads one slice (for `206`s)
  get: (key: string, options?: BlobGetOptions) => Promise<BlobObject | null>
  head: (key: string) => Promise<BlobRecord | null> // metadata only
  delete: (key: string) => Promise<void> // no-op if absent
  list: (options?: BlobListOptions) => Promise<BlobListPage>
}

// ArtifactStore — the metadata layer. D1 backs it.
interface ArtifactStore {
  save: (record: ArtifactRecord) => Promise<void> // insert or overwrite
  get: (artifactId: string) => Promise<ArtifactRecord | null>
  list: (runId: string) => Promise<Array<ArtifactRecord>> // [] when none
  listForThread: (threadId: string) => Promise<Array<ArtifactRecord>>
  delete: (artifactId: string) => Promise<void>
  deleteForRun: (runId: string) => Promise<void>
}

list and listForThread return records ordered by createdAt, then by the ordinal bytewise order of artifactId. Compare UTF-8 bytes from left to right. Do not use locale collation.

BlobBody is ReadableStream<Uint8Array> | ArrayBuffer | ArrayBufferView | string | Blob. The non-stream shapes flow straight into R2Bucket.put unchanged — but a ReadableStream body does not, in the general case: workerd's put requires a stream with a known length (a Response body or the readable half of a FixedLengthStream), and the artifact middleware hands you a TransformStream-wrapped body whenever it had to cap a fetched body as it drains. Passing that stream to bucket.put throws TypeError: Provided readable stream must have a known length.

When does that actually happen? The wrapper only exists to enforce maxArtifactBytes during the drain, so the middleware applies it only when nothing else bounds the transfer:

Provider response	Body handed to put	R2 path
content-length, no content-encoding	untouched, declared length intact	bucket.put direct
chunked (no declared length)	wrapped, length-less	multipart
content-encoding: gzip	wrapped, length-less	multipart

A provider CDN normally sends content-length, so the first row is the common case and bucket.put(key, body) just works. The recipe below is what makes the other two rows work: it re-declares the length from BlobPutOptions.expectedLength when the middleware could vouch for one, and otherwise streams through a multipart upload (one 8 MiB part at a time — flat memory at any artifact size). Write it once and every response shape is covered.

withGenerationPersistence(persistence, { maxArtifactBytes: false }) drops the ceiling and the wrapper altogether, so even a chunked reply arrives untouched. It buys nothing extra for R2 (a chunked body has no length to preserve), so choose it on its own merits: no application limit on what an origin can stream into your bucket. R2's own limits still apply — 5 GiB per single-shot put, and 10,000 multipart parts (~80 GiB at the 8 MiB part size below). Keep the cap when allowInputUrl lets callers name the URL.

BlobPutOptions is { contentType?, customMetadata?, expectedLength? }; BlobGetOptions is { range?: { offset: number, length?: number } } and maps onto R2's own range; BlobListOptions is { prefix?, cursor?, limit? }; BlobListPage is { objects: BlobRecord[], cursor?, truncated? }.

1. BlobStore backed by R2

R2Object carries size, etag, httpMetadata.contentType, customMetadata, and uploaded (a Date). BlobRecord wants createdAt / updatedAt as epoch ms — R2 tracks only the single uploaded instant, so map it to both. get / head are the byte-body vs metadata-only split; R2ObjectBody already exposes body, arrayBuffer(), and text(), so a BlobObject is essentially the R2 object plus the mapped metadata.

import { defineBlobStore, resolveBlobRange } from '@tanstack/ai-persistence'
import type { BlobObject, BlobRecord } from '@tanstack/ai-persistence'

// R2 multipart parts must be ≥ 5 MiB and — except for the last — all exactly
// the SAME size, so a part reader has to cut on an exact boundary and carry the
// remainder. 8 MiB × the 10,000-part ceiling puts the multipart path's limit at
// ~80 GiB; raise this for larger objects, and check R2's current object-size
// limits before promising more.
const MULTIPART_PART_SIZE = 8 * 1024 * 1024

/**
 * Cut exactly `limit` bytes off the stream (fewer only at EOF), carrying any
 * overshoot into the next part.
 *
 * `carry` is the leftover from the previous call. Chunks are collected by
 * reference and copied once per part: growing a `Uint8Array` chunk-by-chunk
 * instead would re-copy the whole part on every read — quadratic work for a
 * part built from hundreds of small chunks.
 */
async function readPart(
  reader: ReadableStreamDefaultReader<Uint8Array>,
  limit: number,
  carry: Uint8Array,
): Promise<{ bytes: Uint8Array; carry: Uint8Array; eof: boolean }> {
  const chunks: Array<Uint8Array> = carry. >  ? [carry] : []
   total = carry.
   eof = 
   (total < limit) {
     { value, done } =  reader.()
     (done) {
      eof = 
      
    }
    chunks.(value)
    total += value.
  }
   joined =  (total)
   offset = 
   ( chunk  chunks) {
    joined.(chunk, offset)
    offset += chunk.
  }
  
  
   (!eof && total > limit) {
     {
      : joined.(, limit),
      : joined.(limit),
      eof,
    }
  }
   { : joined, :  (), eof }
}



  =  *  *  * 


  (): <R2Object | > {
   (expectedLength !==  && expectedLength <= ) {
     bucket.(
      key,
      body.( (expectedLength)),
      options,
    )
  }
  
   reader = body.()
   first =  (reader, ,  ())
   (first.) {
    
     bucket.(key, first., options)
  }
   upload =  bucket.(key, options)
   {
     : <R2UploadedPart> = [
       upload.(, first.),
    ]
     carry = first.
     partNumber = 
     (;;) {
       part =  (reader, , carry)
      carry = part.
       (part.. > ) {
        
        
         (partNumber > ) {
            (
            ,
          )
        }
        parts.( upload.(partNumber, part.))
        partNumber += 
      }
       (part.) 
    }
      upload.(parts)
  }  (error) {
     upload.().( )
     error
  }
}

 ():  {
   uploaded = obj..()
   {
    : obj.,
    : obj.,
    : obj.,
    ...(obj.?.
      ? { : obj.. }
      : {}),
    ...(obj. ? { : obj. } : {}),
    : uploaded,
    : uploaded,
  }
}

  () {
   ({
     () {
       r2Options = {
        ...(options?.
          ? { : { : options. } }
          : {}),
        ...(options?.
          ? { : options. }
          : {}),
      }
       obj =
        body  
          ?  (
              bucket,
              key,
              body,
              r2Options,
              options?.,
            )
          :  bucket.(key, body, r2Options)
      
       (!obj)   ()
       (obj)
    },

     (key, options): < | > {
       (!options?.) {
         whole =  bucket.(key)
         (!whole)  
         {
          ...(whole),
          : whole.,
          :  whole.(),
          :  whole.(),
        }
      }
      
      
      
      
      
      
      
      
      
      
       ( attempt = ; attempt < ; attempt++) {
         head =  bucket.(key)
        
        
        
         (!head)  
         served = (head., options.)
         obj =  bucket.(key, {
          : { : served., : served. },
          
          
          
          : { : head. },
        })
        
         (!obj)  
         (!(  obj)) 
         {
          
          
          
          ...(obj),
          : served,
          : obj.,
          :  obj.(),
          :  obj.(),
        }
      }
        ()
    },

     () {
       obj =  bucket.(key)
       obj ? (obj) : 
    },

     () {
       bucket.(key)
    },

     () {
      
       (options?. === ) {
         { : [] }
      }
       page =  bucket.({
        ...(options?. !==  ? { : options. } : {}),
        ...(options?. !==  ? { : options. } : {}),
        ...(options?. !==  ? { : options. } : {}),
        
        : [, ],
      })
       {
        : page..(toRecord),
        ...(page. ? { : page., :  } : {}),
      }
    },
  })
}

Invariants that matter (asserted by the conformance testkit):

• get / head return null for a missing key; delete is a silent no-op.
• put overwrites an existing key.
• put accepts a ReadableStream body with no declared length — the middleware streams URL-fetched artifacts as exactly that. This is where the naive "pass the body straight to bucket.put" recipe fails at runtime (workerd requires a known length), which is what putStream above handles.
• get honours options.range: it returns only that slice, reports it as range, and keeps size on the whole object. That is the 206 a video player's seeking depends on, and R2 slices in the bucket so the bytes never cross the Worker.
• list filters by prefix literally (R2 prefix is a literal byte prefix — no glob), returns keys in ascending order, and pages via the opaque cursor when truncated. R2's own cursor is opaque and satisfies this directly. limit: 0 must yield an empty, untruncated page — R2 treats limit: 0 as "use the default", so special-case it: if (options?.limit === 0) return { objects: [] }.

2. ArtifactStore backed by D1

ArtifactRecord is { artifactId, runId, threadId, blobKey?, name, mimeType, size, sourceUrl?, createdAt } (createdAt epoch ms). One flat table is keyed by artifact_id. It has run and thread ordered indexes for list and listForThread.

CREATE TABLE IF NOT EXISTS generation_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 INDEX IF NOT EXISTS generation_artifacts_run_order
  ON generation_artifacts (run_id, created_at, artifact_id);
CREATE INDEX IF NOT EXISTS generation_artifacts_thread_order
  ON generation_artifacts (thread_id, created_at, artifact_id);

import { defineArtifactStore } from '@tanstack/ai-persistence'
import type { ArtifactRecord } from '@tanstack/ai-persistence'

interface ArtifactRow {
  artifact_id: string
  run_id: string
  thread_id: string
  blob_key: string | null
  name: string
  mime_type: string
  size: number
  source_url: string | null
  created_at: number
}

function fromRow(row: ArtifactRow): ArtifactRecord {
  return {
    artifactId: row.artifact_id,
    runId: row.run_id,
    threadId: row.thread_id,
    ...(row.blob_key != null ? { blobKey: row.blob_key } : {}),
    name: row.name,
    mimeType: row.mime_type,
    size: row.size,
    ...(row. !=  ? { : row. } : {}),
    : row.,
  }
}

  () {
   ({
     () {
      
       db
        .(
          ,
        )
        .(
          record.,
          record.,
          record.,
          record. ?? ,
          record.,
          record.,
          record.,
          record. ?? ,
          record.,
        )
        .()
    },

     () {
       row =  db
        .()
        .(artifactId)
        .<>()
       row ? (row) : 
    },

     () {
       { results } =  db
        .(
          ,
        )
        .(runId)
        .<>()
       results.(fromRow)
    },

     () {
       { results } =  db
        .(
          ,
        )
        .(threadId)
        .<>()
       results.(fromRow)
    },

     () {
       db
        .()
        .(artifactId)
        .()
    },

     () {
       db
        .()
        .(runId)
        .()
    },
  })
}

Omitting source_url / blob_key from the record when the column is NULL keeps records comparing cleanly against the reference in-memory store. Persist blob_key verbatim: a storageKey mapper can put the bytes anywhere, so a reader cannot recompute the path — resolveArtifactBlobKey(record) falls back to the default convention only for rows written before the column existed. Cloudflare KV is not an equivalent ArtifactStore backend. The required ordering and indexed reads need a transactional indexed database. Use D1 or another transactional indexed database for artifact metadata. Store the bytes in R2.

3. Compose and wire

Bindings are per-request on Workers, so export a factory. Combine the byte stores with a generation-run store (and, if this Worker also does chat, the chat stores). Either build the whole AIPersistence with defineAIPersistence, or layer the artifact stores onto an existing chat persistence with composePersistence:

import {
  defineAIPersistence,
  composePersistence,
  withGenerationPersistence,
} from '@tanstack/ai-persistence'
import { r2BlobStore } from './r2-blob-store'
import { d1ArtifactStore } from './d1-artifact-store'
import { d1GenerationRunStore } from './d1-job-store' // your GenerationRunStore

/** Call inside a request handler — bindings are not available at module scope. */
export function generationPersistence(env: Env) {
  return defineAIPersistence({
    stores: {
      generationRuns: d1GenerationRunStore(env.DB),
      artifacts: d1ArtifactStore(env.DB),
      blobs: r2BlobStore(env.ARTIFACTS_BUCKET),
    },
  })
}

// …or add bytes to a persistence that already has chat + generation runs:
// composePersistence(chatAndJobsPersistence(env), {
//   overrides: {
//     artifacts: d1ArtifactStore(env.DB),
//     blobs: r2BlobStore(env.ARTIFACTS_BUCKET),
//   },
// })

withGenerationPersistence throws if exactly one of artifacts / blobs is present — provide both or neither. Wire it as generation middleware:

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

export default {
  async fetch(request: Request, env: Env) {
    const { prompt, threadId } = await request.json()
    const stream = generateImage({
      adapter: openaiImage('gpt-image-1'),
      prompt,
      threadId, // the slot recorded on the job + artifacts
      stream: true,
      middleware: [
        // Nothing is buffered: the artifact streams from the provider CDN
        // into R2. Add `maxArtifactBytes: false` if you want no ceiling at
        // all on what an origin can stream into your bucket (the default is
        // 1 GiB); it is not needed for the streaming itself.
        withGenerationPersistence(generationPersistence(env), { threadId }),
      ],
    })
    return toServerSentEventsResponse(stream)
  },
}

4. Serve the bytes back

A GET route resolves an artifactId to its record and its stored bytes. retrieveArtifact returns the ArtifactRecord (or null → 404); retrieveBlob returns the BlobObject (metadata + a streamable body). Both resolve the blob key from the record internally, so you never build the key yourself.

Honour Range requests: <video> seeking is built on 206 / Content-Range, and Safari refuses to play a source that ignores Range entirely. Images never notice; a few-hundred-MB clip is unwatchable without it. Pass range to retrieveBlob — the store slices in R2 — rather than reaching into the bucket binding from the route, which would tie the route to R2 and bypass the store's own key resolution.

import {
  parseRangeHeader,
  retrieveArtifact,
  retrieveBlob,
} from '@tanstack/ai-persistence'
import { generationPersistence } from './lib/generation-persistence'

export async function GET(request: Request, env: Env) {
  const artifactId = new URL(request.url).searchParams.get('id') ?? ''
  const persistence = generationPersistence(env)

  // Authorize before serving — derive the owner from the session, never trust
  // a client-supplied id. (The record carries runId/threadId to check against.)
  const record = await retrieveArtifact(persistence, artifactId)
  if (!record) return new Response('Not found', { status: 404 })

  // `parseRangeHeader` resolves the header against the size on the record —
  // suffix ranges included — so an unsatisfiable range is a 416 here and the
  // store only ever sees a range it can serve.
  const range = parseRangeHeader(request.headers.get('range'), record.size)
   (range === ) {
      (, {
      : ,
      : { :  },
    })
  }

  
   blob =  (
    persistence,
    record,
    range ? { range } : ,
  )
   (!blob?.)   (, { :  })

  
  
   headers = {
    : record.,
    : ,
  }
   (!blob.) {
      (blob., {
      : { ...headers, : (record.) },
    })
  }
   { offset, length } = blob.
    (blob., {
    : ,
    : {
      ...headers,
      : (length),
      : ,
    },
  })
}

To hydrate a server-driven generation client (persistence: true + a stable threadId) on mount, also expose reconstructGeneration(persistence, request) on a GET that reads ?threadId= / ?runId= — see ai-core/client-persistence.

Other backends — the contract is tiny, here's how each maps

BlobStore is five methods over an object store. Any of these backs it; swap the factory, keep everything else. put maps to the SDK's upload, get to a download that exposes body/arrayBuffer/text, head to a metadata fetch, delete to a delete, list to a prefixed, cursor-paged list.

Backend	npm	One-line sketch
AWS S3	@aws-sdk/client-s3	put→PutObjectCommand; get→GetObjectCommand (Body is a stream → body, .transformToByteArray()/.transformToString()); head→HeadObjectCommand; delete→DeleteObjectCommand; list→ListObjectsV2Command (Prefix, ContinuationToken↔cursor, MaxKeys↔limit, IsTruncated↔truncated).
Google Cloud Storage	@google-cloud/storage	bucket.file(key): put→.save(body, { contentType, metadata }); get→.createReadStream() for body + .download() for bytes; head→.getMetadata(); delete→.delete({ ignoreNotFound: true }); list→bucket.getFiles({ prefix, maxResults, pageToken }).
Vercel Blob	@vercel/blob	put→put(key, body, { access: 'public', contentType }); get→fetch(head(key).url) (stream res.body); head→head(key) (returns null→catch as absent); delete→del(key); list→list({ prefix, cursor, limit }) (hasMore↔truncated).
Supabase Storage	@supabase/supabase-js

For each: contentType and customMetadata ride the SDK's own metadata fields; BlobRecord.createdAt/updatedAt come from the object's stored timestamps (epoch ms); return null from get/head on a not-found rather than throwing.

Verify

The shared runPersistenceConformance testkit covers all seven stores, including generationRuns, artifacts, and blobs — point it at your factory rather than hand-writing these assertions:

import { runPersistenceConformance } from '@tanstack/ai-persistence/testkit'
import { env } from 'cloudflare:test'
import { generationPersistence } from '../src/lib/generation-persistence'

// A generation-only Worker declares the chat state stores it does not provide.
runPersistenceConformance('app-r2', () => generationPersistence(env), {
  skip: ['messages', 'runs', 'interrupts', 'metadata'],
})

Run it against a Miniflare R2 + D1 binding with the migration applied, reset between runs (see ai-persistence/build-cloudflare-adapter for the cloudflare:test harness pattern). It exercises, among the rest:

• put then get round-trips bytes and metadata; get/head return null for a missing key; delete is a silent no-op on an absent key.
• put accepts a ReadableStream body with no declared length (a TransformStream-wrapped stream) and records the real drained size — the shape every URL-fetched artifact arrives in.
• get with a range returns just that slice, reports it as range, and still reports the whole object's size — what a 206 / Content-Range response is built from.
• put overwrites an existing key (and its contentType/customMetadata).
• list filters by prefix literally and case-sensitively, returns ascending keys, pages through the cursor when truncated without gaps or repeats, and returns an empty untruncated page for limit: 0.
• The ArtifactStore: save is insert-or-overwrite, get returns null when absent, list(runId) returns [] for an unknown run, listForThread(threadId) returns the complete ordered history, and delete / deleteForRun remove exactly the expected rows.
• The GenerationRunStore: createOrResume idempotency, no-op update on an unknown id, and findLatestForThread returning the most recently started linked run (terminal ones included).

An end-to-end check is the strongest signal: run generateImage through withGenerationPersistence(generationPersistence(env), { threadId }), then confirm the blob exists at artifacts/<runId>/<artifactId> and retrieveBlob streams it back.