fix: serialize extension arrays after provisioning

CHANGELOG.md

@@ -7,6 +7,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+
+- `upsert` and `upsertMany` now work on the same `postgres.js` pool connection
+  that `createIndex` used to install `ltree` and `pgvector`. Batch arrays are
+  bound through the built-in `text[]` serializer and cast to their qualified
+  destination types, avoiding stale per-connection custom-array type maps while
+  preserving parameterization, nulls, and array escaping.
+
 ## [0.1.0] - 2026-09-03
 
 ### Added

packages/core/src/write.ts

@@ -14,6 +14,8 @@ import { runSql } from "./sql/exec.ts";
 import { normalizeTemporalTuple, temporalTupleSchema } from "./temporal.ts";
 
 const MAX_UPSERT_BATCH_SIZE = 1000;
+/** Built-in PostgreSQL `text` OID, used as sql.array's element type. */
+const POSTGRES_TEXT_OID = 25;
 
 const recordSchema = z
   .object({
@@ -141,17 +143,27 @@ export async function upsertMany(
   if (ids.length === 0) return [];
 
   const { sql } = index;
+  // postgres.js discovers array serializers when each connection starts. If
+  // createIndex installs extension types on that same connection, its cached
+  // map does not yet contain ltree[] or vector[]. Bind every parallel array as
+  // the always-known text[] type, then let PostgreSQL cast each element to the
+  // routine's qualified destination type. sql.array owns array escaping, nulls,
+  // and empty-array encoding; values remain parameters.
+  const embeddingArray =
+    index.vectorType === "halfvec"
+      ? sql`${sql.array(embeddings, POSTGRES_TEXT_OID)}::text[]::public.halfvec[]`
+      : sql`${sql.array(embeddings, POSTGRES_TEXT_OID)}::text[]::public.vector[]`;
   const rows = await runBatchUpsert(
     sql<BatchUpsertRow[]>`
       select ord, id, status
       from ${sql(index.schema)}.batch_upsert
-      ( ${ids}
-      , ${contents}
+      ( ${sql.array(ids, POSTGRES_TEXT_OID)}::text[]::uuid[]
+      , ${sql.array(contents, POSTGRES_TEXT_OID)}::text[]
       , ${sql.json(metas)}
-      , ${trees}
-      , ${temporals}
-      , ${names}
-      , ${embeddings}
+      , ${sql.array(trees, POSTGRES_TEXT_OID)}::text[]::public.ltree[]
+      , ${sql.array(temporals, POSTGRES_TEXT_OID)}::text[]::tstzrange[]
+      , ${sql.array(names, POSTGRES_TEXT_OID)}::text[]
+      , ${embeddingArray}
       , ${parsedOptions.onConflict}
       )
     `,

packages/core/test/fresh-extension-arrays.integration.test.ts

@@ -0,0 +1,135 @@
+import assert from "node:assert/strict";
+import { test } from "node:test";
+import type { Sql } from "postgres";
+import { createIndex } from "../src/create-index.ts";
+import { openIndex } from "../src/open-index.ts";
+import {
+  connect,
+  connectToDatabase,
+  createTestDatabase,
+  dropTestDatabase,
+  randomTestDatabase,
+} from "./support/db.ts";
+
+/**
+ * Regression: postgres.js caches custom array serializers when a connection
+ * starts. createIndex may install ltree/vector after that discovery has run, so
+ * a bare JS array sent on the same connection degrades to `a,b` instead of a
+ * PostgreSQL array literal. A one-connection pool makes that sequence
+ * deterministic and also proves a prepared statement can be reused.
+ */
+test("batch writes work when createIndex installs extension array types on the same connection", async () => {
+  const admin = connect();
+  const database = randomTestDatabase();
+  let indexSql: Sql | undefined;
+
+  try {
+    await createTestDatabase(admin, database);
+    indexSql = connectToDatabase(database, 1);
+
+    // Open the only pool connection before the extension types exist. This
+    // freezes postgres.js's initial array-type map without ltree[]/halfvec[].
+    const installed = await indexSql<{ readonly extname: string }[]>`
+      select extname
+      from pg_catalog.pg_extension
+      where extname in ('vector', 'pg_textsearch', 'ltree')
+    `;
+    assert.deepEqual(Array.from(installed), []);
+
+    await createIndex(indexSql, "docs", { dimensions: 4 });
+    const index = await openIndex(indexSql, "docs", {
+      embedding: "mock-embedding",
+    });
+
+    const first = await index.upsertMany([
+      {
+        content: "Auth tokens rotate daily.",
+        tree: "docs.auth",
+        temporal: ["2026-01-01T00:00:00Z"],
+        embedding: [1, 0, 0, 0],
+      },
+      {
+        content: "Rate limits apply per API key.",
+        tree: "docs.api",
+        name: "rate-limit",
+        temporal: ["2026-02-01T00:00:00Z", "2026-03-01T00:00:00Z"],
+      },
+      {
+        content: "Backups are retained for thirty days.",
+        tree: "docs.ops",
+      },
+    ]);
+    assert.deepEqual(
+      first.map((result) => result.status),
+      ["inserted", "inserted", "inserted"],
+    );
+
+    // Reuse the prepared batch statement and exercise text-array escaping.
+    const second = await index.upsertMany([
+      {
+        content: 'Cache keys may contain commas, quotes " and backslashes \\.',
+        tree: "docs.cache",
+        name: 'key,"primary\\name',
+        embedding: [0, 1, 0, 0],
+      },
+    ]);
+    assert.equal(second[0]?.status, "inserted");
+
+    const rows = await indexSql<
+      {
+        readonly content: string;
+        readonly tree: string;
+        readonly name: string | null;
+        readonly temporal: string | null;
+        readonly embedding: string | null;
+      }[]
+    >`
+      select content, tree::text as tree, name, temporal, embedding
+      from docs.record
+      order by tree
+    `;
+    assert.deepEqual(
+      rows.map((row) => row.tree),
+      ["docs.api", "docs.auth", "docs.cache", "docs.ops"],
+    );
+    assert.equal(rows[0]?.name, "rate-limit");
+    assert.ok(rows[0]?.temporal);
+    assert.equal(rows[0]?.embedding, null);
+    assert.equal(rows[1]?.name, null);
+    assert.ok(rows[1]?.temporal);
+    assert.equal(rows[1]?.embedding, "[1,0,0,0]");
+    assert.equal(rows[2]?.name, 'key,"primary\\name');
+    assert.equal(
+      rows[2]?.content,
+      'Cache keys may contain commas, quotes " and backslashes \\.',
+    );
+    assert.equal(rows[2]?.embedding, "[0,1,0,0]");
+    assert.equal(rows[3]?.name, null);
+    assert.equal(rows[3]?.temporal, null);
+    assert.equal(rows[3]?.embedding, null);
+
+    // The alternate pgvector storage type uses a distinct array cast path.
+    await createIndex(indexSql, "docs_vector", {
+      dimensions: 4,
+      vectorType: "vector",
+    });
+    const vectorIndex = await openIndex(indexSql, "docs_vector", {
+      embedding: "mock-embedding",
+    });
+    const vectorResult = await vectorIndex.upsert({
+      content: "Full precision vector.",
+      tree: "docs.vectors",
+      embedding: [0, 0, 1, 0],
+    });
+    const [vectorRow] = await indexSql<{ readonly embedding: string | null }[]>`
+      select embedding
+      from docs_vector.record
+      where id = ${vectorResult.id}
+    `;
+    assert.equal(vectorRow?.embedding, "[0,0,1,0]");
+  } finally {
+    await indexSql?.end();
+    await dropTestDatabase(admin, database);
+    await admin.end();
+  }
+});

packages/core/test/support/db.ts

@@ -8,15 +8,16 @@ export function connect(): Sql {
   return connectToUrl(testDatabaseUrl());
 }
 
-export function connectToDatabase(database: string): Sql {
+export function connectToDatabase(database: string, max?: number): Sql {
   const url = new URL(testDatabaseUrl());
   url.pathname = `/${database}`;
-  return connectToUrl(url.toString());
+  return connectToUrl(url.toString(), max);
 }
 
-function connectToUrl(url: string): Sql {
+function connectToUrl(url: string, max?: number): Sql {
   return postgres(url, {
     onnotice: () => {},
+    ...(max === undefined ? {} : { max }),
   });
 }