Postgres Storage

Implement durable Crux RecordStore, pgvector VectorStore, and bytea AssetStore adapters on Postgres.

Use a RecordStore when Crux needs durable JSON records. Postgres is a good fit when your app already uses it for product data and you want corpus ledgers, retrieval chunks, workspace metadata, and other Storage Beta records in the same operational database.

This recipe starts with a RecordStore, then shows optional Postgres-backed vector and asset storage. You can use only the pieces you need:

const records = postgresRecordStore(pool);
const vectors = postgresVectorStore(pool); // optional, dense pgvector search
const assets = postgresAssetStore(pool); // optional, modest bytea-backed files

For large production files, object storage such as S3, R2, GCS, or Convex file storage is usually a better AssetStore than Postgres. Use the Postgres asset example when you deliberately want one operational database and the files are modest in size.

RecordStore

Implement the RecordStore interface with a single table:

create table crux_records (
  key text primary key,
  value jsonb not null,
  expires_at timestamptz,
  created_at timestamptz not null default now(),
  updated_at timestamptz not null default now()
);

create index crux_records_key_prefix_idx on crux_records (key text_pattern_ops);
create index crux_records_updated_at_idx on crux_records (updated_at desc);
create index crux_records_expires_at_idx on crux_records (expires_at);

Implementation

This example uses pg, but the same shape works with Prisma, Drizzle, Kysely, Neon, Supabase, or any SQL client.

import type {
  ExactFilter,
  JsonObject,
  RecordListOptions,
  RecordPage,
  RecordStore,
  RecordWriteOptions,
} from "@use-crux/core/storage";

type Row = {
  key: string;
  value: JsonObject;
};

type Queryable = {
  query<T extends Record<string, unknown> = Record<string, unknown>>(
    sql: string,
    values?: readonly unknown[],
  ): Promise<{ rows: T[] }>;
};

export function postgresRecordStore(pool: Queryable): RecordStore {
  return {
    _tag: "RecordStore",

    async get(key) {
      await pool.query(
        `
        delete from crux_records
        where expires_at is not null and expires_at <= now()
        `,
      );

      const row = await pool.query<Row>(
        `
        select key, value
        from crux_records
        where key = $1
          and (expires_at is null or expires_at > now())
        limit 1
        `,
        [key],
      );

      return row.rows[0]?.value ?? null;
    },

    async put(key, value, options?: RecordWriteOptions) {
      const expiresAt =
        options?.ttlMs !== undefined
          ? new Date(Date.now() + options.ttlMs)
          : null;

      await pool.query(
        `
        insert into crux_records (key, value, expires_at, updated_at)
        values ($1, $2::jsonb, $3, now())
        on conflict (key) do update
          set value = excluded.value,
              expires_at = excluded.expires_at,
              updated_at = now()
        `,
        [key, JSON.stringify(value), expiresAt],
      );
    },

    async create(key, value, options?: RecordWriteOptions) {
      const expiresAt =
        options?.ttlMs !== undefined
          ? new Date(Date.now() + options.ttlMs)
          : null;

      const result = await pool.query<{ key: string }>(
        `
        insert into crux_records (key, value, expires_at, updated_at)
        values ($1, $2::jsonb, $3, now())
        on conflict (key) do nothing
        returning key
        `,
        [key, JSON.stringify(value), expiresAt],
      );

      return result.rows.length === 1;
    },

    async delete(key) {
      await pool.query("delete from crux_records where key = $1", [key]);
    },

    async list(prefix, options?: RecordListOptions): Promise<RecordPage> {
      const limit = options?.limit ?? 100;
      if (limit === 0) return { entries: [] };

      const entries: { key: string; value: JsonObject }[] = [];
      let cursor = options?.cursor ?? null;
      let hasMore = false;

      while (entries.length < limit) {
        const rows = await pool.query<Row>(
          `
          select key, value
          from crux_records
          where key like $1
            escape '\\'
            and ($2::text is null or key > $2)
            and (expires_at is null or expires_at > now())
          order by key asc
          limit $3
          `,
          [`${escapeLike(prefix)}%`, cursor, Math.max(limit + 1, 100)],
        );

        const pageHasMore = rows.rows.length === Math.max(limit + 1, 100);
        const scannedRows = pageHasMore ? rows.rows.slice(0, -1) : rows.rows;

        for (let index = 0; index < scannedRows.length; index++) {
          const row = scannedRows[index]!;
          cursor = row.key;
          if (
            !options?.filter ||
            matchesTopLevelFilter(row.value, options.filter)
          ) {
            entries.push({ key: row.key, value: row.value });
          }
          if (entries.length >= limit) {
            hasMore = pageHasMore || index < scannedRows.length - 1;
            break;
          }
        }

        if (hasMore || !pageHasMore || scannedRows.length === 0) break;
      }

      return entries.length >= limit && hasMore
        ? { entries, cursor: cursor ?? undefined }
        : { entries };
    },

    capabilities: () => ({
      ttl: "native",
      filter: "scan",
      watch: false,
      batch: false,
    }),
  };
}

function escapeLike(value: string): string {
  return value.replace(/[\\%_]/g, (match) => `\\${match}`);
}

function matchesTopLevelFilter(
  value: JsonObject,
  filter: ExactFilter,
): boolean {
  return Object.entries(filter).every(([key, expected]) => {
    const actual = value[key];
    return expected === null ? actual === null : actual === expected;
  });
}

VectorStore With pgvector

Postgres can also be a dense VectorStore through the pgvector extension. This is useful when you want a simple deployment and dense semantic search is enough.

It is not the cleanest default for sparse or hybrid retrieval. Crux sparse vectors are { indices, values }, while Postgres full-text search uses tsvector/tsquery. If you want sparse or hybrid search in Postgres, build a custom retriever that combines pgvector and full-text search, or implement a product-specific VectorStore with your own fusion. For first-class sparse/hybrid vector APIs, Upstash Vector is currently the simpler bundled path.

Schema for dense pgvector:

create extension if not exists vector;

create table crux_vectors (
  key text primary key,
  dense vector(1536),
  metadata jsonb not null default '{}'::jsonb,
  created_at timestamptz not null default now(),
  updated_at timestamptz not null default now()
);

create index crux_vectors_dense_idx
  on crux_vectors using hnsw (dense vector_cosine_ops);

create index crux_vectors_metadata_idx
  on crux_vectors using gin (metadata);

Change vector(1536) to match your embedding dimensions.

import type {
  VectorHit,
  VectorRecord,
  VectorSearchQuery,
  VectorStore,
} from "@use-crux/core/storage";

export function postgresVectorStore(pool: Queryable): VectorStore {
  return {
    async upsert(records: readonly VectorRecord[]) {
      for (const record of records) {
        if (!record.dense) {
          throw new Error("postgresVectorStore() supports dense vectors only.");
        }

        await pool.query(
          `
          insert into crux_vectors (key, dense, metadata, updated_at)
          values ($1, $2::vector, $3::jsonb, now())
          on conflict (key) do update
            set dense = excluded.dense,
                metadata = excluded.metadata,
                updated_at = now()
          `,
          [
            record.key,
            toPgVector(record.dense),
            JSON.stringify(record.metadata ?? {}),
          ],
        );
      }
    },

    async delete(keys: readonly string[]) {
      if (keys.length === 0) return;

      await pool.query(
        `
        delete from crux_vectors
        where key = any($1::text[])
        `,
        [keys],
      );
    },

    async search(query: VectorSearchQuery): Promise<readonly VectorHit[]> {
      if (query.mode !== "dense") {
        throw new Error("postgresVectorStore() supports dense search only.");
      }

      const limit = query.limit ?? 10;
      const threshold = query.threshold ?? 0;
      const filter = query.filter ? JSON.stringify(query.filter) : null;

      const rows = await pool.query<{
        key: string;
        score: number;
        metadata: Record<string, unknown>;
      }>(
        `
        select key,
               1 - (dense <=> $1::vector) as score,
               metadata
        from crux_vectors
        where dense is not null
          and ($2::jsonb is null or metadata @> $2::jsonb)
          and 1 - (dense <=> $1::vector) >= $3
        order by dense <=> $1::vector
        limit $4
        `,
        [toPgVector(query.dense), filter, threshold, limit],
      );

      return rows.rows.map((row) => ({
        key: row.key,
        score: row.score,
        metadata: row.metadata,
      }));
    },

    capabilities() {
      return {
        dense: true,
        sparse: false,
        hybrid: false,
        fusion: [],
        filter: "pre",
        consistency: "strong",
      };
    },
  };
}

function toPgVector(vector: readonly number[]): string {
  if (vector.length === 0) {
    throw new Error("Vector must not be empty.");
  }

  for (const value of vector) {
    if (!Number.isFinite(value)) {
      throw new Error("Vector contains a non-finite value.");
    }
  }

  return `[${vector.join(",")}]`;
}

Use it with indexing and retrieval:

const records = postgresRecordStore(pool);
const vectors = postgresVectorStore(pool);

const docsIndexer = indexer({
  id: "docs",
  namespace: "docs",
  records,
  vectors,
  dense,
});

const docs = retriever({
  id: "docs",
  namespace: "docs",
  records,
  vectors,
  dense,
  search: { mode: "dense", limit: 8 },
});

AssetStore With `bytea`

Postgres can store binary content with bytea. This is convenient for tests, internal tools, and small generated files. It is usually not the best choice for large PDFs, images, or user uploads; object storage is cheaper and better at serving bytes.

Schema:

create table crux_assets (
  uri text primary key,
  content bytea not null,
  mime_type text not null,
  size integer not null,
  metadata jsonb not null default '{}'::jsonb,
  created_at timestamptz not null default now(),
  updated_at timestamptz not null default now()
);

Implementation:

import type {
  Asset,
  StoredAsset,
  AssetRef,
  AssetStore,
} from "@use-crux/core/storage";

export function postgresAssetStore(pool: Queryable): AssetStore {
  return {
    async put(asset: Asset, options): Promise<StoredAsset> {
      if (asset.type !== "data") {
        throw new Error("This example only persists data assets");
      }
      const key = options?.key ?? crypto.randomUUID();
      const uri = `postgres://crux_assets/${encodeURIComponent(key)}`;
      const bytes = await dataAssetBytes(asset.data);

      await pool.query(
        `
        insert into crux_assets (uri, content, mime_type, size, metadata, updated_at)
        values ($1, $2, $3, $4, $5::jsonb, now())
        on conflict (uri) do update
          set content = excluded.content,
              mime_type = excluded.mime_type,
              size = excluded.size,
              metadata = excluded.metadata,
              updated_at = now()
        `,
        [
          uri,
          Buffer.from(bytes),
          asset.mediaType,
          asset.size ?? bytes.byteLength,
          JSON.stringify(options?.metadata ?? {}),
        ],
      );

      return {
        type: "data",
        data: new Uint8Array(bytes),
        mediaType: asset.mediaType,
        size: asset.size ?? bytes.byteLength,
        ref: { uri },
      };
    },

    async get(ref: AssetRef): Promise<StoredAsset> {
      const row = await pool.query<{
        content: Buffer;
        mime_type: string;
        size: number;
      }>(
        `
        select content, mime_type, size
        from crux_assets
        where uri = $1
        limit 1
        `,
        [ref.uri],
      );

      const asset = row.rows[0];
      if (!asset) {
        throw new Error(`Asset not found: ${ref.uri}`);
      }

      return {
        type: "data",
        data: new Uint8Array(asset.content),
        mediaType: asset.mime_type,
        size: asset.size,
        ref,
      };
    },

    async delete(ref: AssetRef) {
      await pool.query("delete from crux_assets where uri = $1", [ref.uri]);
    },
  };
}

async function dataAssetBytes(data: Uint8Array | Blob): Promise<Uint8Array> {
  if (data instanceof Uint8Array) return new Uint8Array(data);
  return new Uint8Array(await data.arrayBuffer());
}

Wire it into a workspace:

import { storage } from "@use-crux/core/storage";
import { workspace } from "@use-crux/core/workspace";

const files = workspace({
  id: "thread-files",
  namespace: threadId,
  storage: storage({
    records: postgresRecordStore(pool),
    assets: postgresAssetStore(pool),
  }),
});

await files.write("/outputs/report.pdf", pdfBytes, {
  mimeType: "application/pdf",
});

Use It

import { indexer } from "@use-crux/core/indexing";
import { retriever } from "@use-crux/core/retrieval";
import { Pool } from "pg";

const pool = new Pool({ connectionString: process.env.DATABASE_URL });
const records = postgresRecordStore(pool);
const vectors = postgresVectorStore(pool);

const docsIndexer = indexer({
  id: "docs",
  namespace: "docs",
  records,
  vectors,
  dense,
});

const docs = retriever({
  id: "docs",
  namespace: "docs",
  records,
  vectors,
  dense,
});

Postgres owns JSON record hydration. VectorStore owns similarity search. AssetStore owns bytes for workspaces. Keeping those capabilities separate makes each adapter simpler and avoids pretending Postgres is also object storage or vector infrastructure unless you explicitly add those capabilities.

Production Notes

Use a tenant-aware key prefix or separate databases when tenants need hard isolation. Crux keys are already namespaced by feature, but tenant isolation is an application security boundary.

Run a periodic cleanup job if you use TTL heavily:

delete from crux_records
where expires_at is not null and expires_at <= now();

For high-write workloads, add indexes that match your app’s actual prefixes and keep list() limits bounded.

For pgvector, benchmark hnsw vs ivfflat indexes with your corpus size and update pattern. For small corpora, exact scan can be acceptable; for larger corpora, use an ANN index and monitor recall.

For asset storage, keep a hard size limit if you use Postgres bytea. Store large user uploads and generated artifacts in object storage and keep only metadata/URIs in Postgres.

Postgres Storage

RecordStore

Implementation

VectorStore With pgvector

AssetStore With `bytea`

Use It

Production Notes

RecordStore guide

VectorStore guide

AssetStore guide

On this page

Postgres Storage

RecordStore

Implementation

VectorStore With pgvector

AssetStore With bytea

Use It

Production Notes

Related

RecordStore guide

VectorStore guide

AssetStore guide

On this page

AssetStore With `bytea`