This is a Deno Postgres Adapter to run Zero Custom Mutators. It was a quick solution, I'm not sure if this is the best way to do it, but I shared with the Zero team to possibly improve it and have an official version :)

DenoPgAdapter.ts

import { Client, type ClientOptions } from "jsr:@db/postgres";
import type { Row } from "@rocicorp/zero/zql/mutate/custom";

/**
 * Adapter for Zero's connectionProvider. Provides .unsafe and .begin methods as required.
 */
export function createZeroPgAdapter(connection: string | ClientOptions = {}) {
  const client = new Client(connection);

  let connected = false;
  async function ensureConnected() {
    if (!connected) {
      await client.connect();
      connected = true;
    }
  }

  return {
    async unsafe(sql: string, params: unknown[]): Promise<Row[]> {
      await ensureConnected();
      const result = await client.queryObject({ text: sql, args: params });
      return result.rows as Row[];
    },
    async begin<T>(
      fn: (tx: { unsafe: (sql: string, params: unknown[]) => Promise<Row[]> }) => Promise<T>,
    ): Promise<T> {
      await ensureConnected();
      await client.queryArray("BEGIN");
      try {
        const tx = {
          async unsafe(sql: string, params: unknown[]): Promise<Row[]> {
            const result = await client.queryObject({ text: sql, args: params });
            return result.rows as Row[];
          },
        };
        const res = await fn(tx);
        await client.queryArray("COMMIT");
        return res;
      } catch (e) {
        await client.queryArray("ROLLBACK");
        throw e;
      }
    },
  };
}

server.ts

import { PushProcessor, connectionProvider } from "@rocicorp/zero/pg";
import { createMutators } from "../shared/public/zero/mutators.ts"; 
import { createZeroPgAdapter } from "./DenoPgAdapter.ts";

const processor = new PushProcessor(
  schema,
  connectionProvider(createZeroPgAdapter(ZERO_UPSTREAM_DB)),
);