Crux
CookbookBasics

Postgres Storage

Implement durable Crux RecordStore, SearchStore, 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 search and asset storage. You can use only the pieces you need:

const records = postgresRecordStore(pool);
const search = postgresSearchStore({ pool, dimensions: 1536, lexical: true });
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;
  });
}

Use the first-party PostgreSQL SearchStore when retrieval search should live in the same database as records:

import { postgresSearchStore } from "@use-crux/postgres";

const search = postgresSearchStore({
  pool,
  dimensions: 1536,
  sparseDimensions: 30_000,
  lexical: { configuration: "simple" },
});

const setup = await search.setup.check();
if (!setup.ok) await search.setup.apply();

At least one of dimensions, sparseDimensions, or lexical is required. Dense and sparse legs use pgvector. Lexical search uses native PostgreSQL full-text search over the indexed chunk content.

The Crux-owned search table includes only the configured payload columns. With dense, sparse, and lexical enabled, the relevant shape is:

create extension if not exists vector;

create table crux_search (
  key text primary key,
  dense vector(1536),
  sparse sparsevec(30000),
  content text,
  search_document tsvector GENERATED ALWAYS AS (
    to_tsvector('simple'::regconfig, coalesce(content, ''))
  ) STORED,
  metadata jsonb not null default '{}'::jsonb,
  created_at timestamptz not null default now(),
  updated_at timestamptz not null default now()
);

create index crux_search_dense_idx
  on crux_search using hnsw (dense vector_cosine_ops);

create index crux_search_sparse_idx
  on crux_search using hnsw (sparse sparsevec_ip_ops);

create index crux_search_document_idx
  on crux_search using gin (search_document);

create index crux_search_metadata_idx
  on crux_search using gin (metadata);

Crux resolves the lexical configuration with a parameterized regconfig lookup before DDL or queries. Lexical queries use websearch_to_tsquery(configuration, query) and rank candidates with PostgreSQL full-text ranking. Empty parsed lexical queries return no lexical candidates.

Use it with indexing and retrieval:

const records = postgresRecordStore(pool);
const search = postgresSearchStore({
  pool,
  dimensions: 1536,
  lexical: true,
});

await search.setup.apply();

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

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

const hits = await docs.retrieve({
  query: "billing credits",
  search: {
    dense: { candidates: 120 },
    lexical: { candidates: 80 },
    fusion: { strategy: "rrf", k: 60 },
  },
  limit: 8,
});

setup.check() is non-mutating. It reports missing columns, generated expression/configuration mismatches, missing GIN indexes, pgvector requirements, dimensions, and presence constraints with stable finding codes. setup.apply() creates missing schemas, tables, nullable columns, the vector extension when required, constraints, and indexes. It does not rewrite content from the records table or silently change a text-search configuration.

After enabling lexical search on an existing knowledge base, run knowledgeBase.reindex() so active chunks write content into their SearchRecords. Until then, setup.check() reports POSTGRES_SEARCH_LEXICAL_CONTENT_MISSING for active indexed chunks with null content.

For multi-leg queries, PostgreSQL ranks each leg in filtered candidate CTEs, full-joins candidates by key, and calculates deterministic normalized RRF in one server-side statement. Final ordering is fused score descending, then key ascending.

PostgreSQL lexical search is not BM25 in this release. True BM25 is deferred to RFC #352, and learned sparse providers are deferred to RFC #353. Sparse search remains a composable leg when sparseDimensions is configured.

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 { postgresSearchStore } from "@use-crux/postgres";
import { Pool } from "pg";

const pool = new Pool({ connectionString: process.env.DATABASE_URL });
const records = postgresRecordStore(pool);
const search = postgresSearchStore({ pool, dimensions: 1536, lexical: true });

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

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

Postgres owns JSON record hydration. SearchStore owns retrieval search. AssetStore owns bytes for workspaces. Keeping those capabilities separate makes each adapter simpler and avoids pretending Postgres is also object storage 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.

RecordStore guide

The full RecordStore contract and adapter expectations.

SearchStore guide

Pair Postgres records with retrieval search.

AssetStore guide

Store workspace bytes in object storage.

Editable workspace sources

Let a custom source-backed mount write changes back to its provider.

Node request-scoped defer

Use primitive-local work directly, or add a Node HTTP boundary for response-scoped work.

On this page

RecordStoreImplementationSearchStore With pgvector And Lexical SearchAssetStore With byteaUse ItProduction NotesRelated