initial commit

.gitignore

@@ -0,0 +1,24 @@
+# Logs
+logs
+*.log
+npm-debug.log*
+yarn-debug.log*
+yarn-error.log*
+pnpm-debug.log*
+lerna-debug.log*
+
+node_modules
+dist
+*.local
+*.tgz
+
+# Editor directories and files
+.vscode/*
+!.vscode/extensions.json
+.idea
+.DS_Store
+*.suo
+*.ntvs*
+*.njsproj
+*.sln
+*.sw?

LICENSE.md

@@ -0,0 +1,8 @@
+The MIT License (MIT)
+Copyright © 2023 On Set Software Ltd
+
+Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the “Software”), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

README.md

@@ -0,0 +1,67 @@
+# Automerge Repo Network Supabase
+
+A Supabase network adapter for [automerge-repo](https://github.com/automerge/automerge-repo). It comprises a client library for the browser and a Deno package for use in a Supabase edge function.
+
+> **Warning**
+> This package is under active development and is not quite ready for production use. However, it is fully intended to be used in production soon. Please feel free to try it out and report any issues you find.
+
+## Client
+
+### Installation
+
+```bash
+npm install @onsetsoftware/automerge-repo-network-supabase
+```
+
+### Usage
+
+```typescript
+
+import { SupabaseNetworkAdapter } from '@onsetsoftware/automerge-repo-network-supabase';
+import { createClient } from "@supabase/supabase-js";
+export { v4 as uuid } from "uuid";
+
+// create a supabase client with your public URL and anon key
+const supabase = createClient({SUPABASE_PUBLIC_URL}, {SUPABASE_ANON_KEY});
+
+// create a network adapter with the supabase client and the name of supabase edge function you are using
+const supabaseAdapter = new SupabaseNetworkAdapter(supabase, "changes");
+
+// pass the adapter to the repo
+const repo = new Repo({
+  storage: new LocalForageStorageAdapter(),
+  network: [
+    supabaseAdapter,
+  ],
+  // we need to use uuids as the peerId is stored as a uuid type in the database
+  peerId: uuid() as PeerId
+});
+
+// use the repo as normal
+```
+
+## Edge Function
+
+### Usage
+
+```typescript
+
+import { serve } from "https://deno.land/[email protected]/http/server.ts"
+
+import { defaultCorsHeaders, handleRequest } from "https://deno.land/[email protected]/mod.ts";
+
+serve(async (req) => {
+  if (req.method === "OPTIONS") {
+    return new Response("ok", { headers: defaultCorsHeaders });
+  }
+
+  return handleRequest(req);
+});
+
+```
+
+## Database Setup
+
+You will need to create a documents table in your Supabase database and ensure that realtime replication is enabled.
+
+An example migration file can be found in the [migration.example.sql](./migration.example.sql) file.

deno/db/config.ts

@@ -0,0 +1,47 @@
+import { Pool } from "../deps.ts";
+
+export interface PGConfig {
+  initPool(): Pool;
+}
+
+export function getDBConfig(): PGConfig {
+  const supabaseServerConfig = getSupabaseServerConfig();
+
+  return supabaseDBConfig(supabaseServerConfig);
+}
+
+const serverEnvVars = {
+  url: (Deno.env.get("SUPABASE_DB_URL") ?? "").replace("localhost", "host.docker.internal").replace('5432/', '6543/'),
+};
+
+export type SupabaseServerConfig = typeof serverEnvVars;
+
+export function getSupabaseServerConfig() {
+  return validate(serverEnvVars);
+}
+
+function validate<T extends Record<string, string>>(vars: T) {
+  for (const [, v] of Object.entries(vars)) {
+    if (!v) {
+      throw new Error(`Invalid Supabase config: SUPABASE_DB_URL must be set`);
+    }
+  }
+  return vars;
+}
+
+export function supabaseDBConfig(config: SupabaseServerConfig) {
+  const { url } = config;
+  return new PostgresDBConfig(url);
+}
+
+export class PostgresDBConfig implements PGConfig {
+  private _url: string;
+
+  constructor(url: string) {
+    this._url = url;
+  }
+
+  initPool(): Pool {
+    return new Pool(this._url, 20, true);
+  }
+}

deno/db/pg.ts

@@ -0,0 +1,141 @@
+// Low-level config and utilities for Postgres.
+
+import { type Payload, Pool, QueryObjectResult } from "../deps.ts";
+import { getDBConfig } from "./config.ts";
+
+export function getPool(reset = false) {
+  const global = globalThis as unknown as {
+    _pool: Pool;
+  };
+  if (!global._pool || reset) {
+    global._pool = initPool();
+  }
+  return global._pool;
+}
+
+function initPool() {
+  console.log("creating global pool");
+
+  const dbConfig = getDBConfig();
+  return dbConfig.initPool();
+}
+
+export async function withExecutor<R>(f: (executor: Executor) => R) {
+  const p = getPool();
+  return await withExecutorAndPool(f, p);
+}
+
+let brokenPipeFound = false;
+
+async function withExecutorAndPool<R>(
+  f: (executor: Executor) => R,
+  p: Pool
+): Promise<R> {
+  try {
+    const client = await p.connect();
+
+    await client.queryObject(
+      "SET SESSION CHARACTERISTICS AS TRANSACTION ISOLATION LEVEL SERIALIZABLE"
+    );
+
+    // eslint-disable-next-line @typescript-eslint/no-explicit-any
+    const executor: Executor = async (sql: string, params?: any[]) => {
+      try {
+        return await client.queryObject(sql, params);
+      } catch (e) {
+        // console.log(
+        //   // eslint-disable-next-line @typescript-eslint/no-explicit-any
+        //   `Error executing SQL: ${sql}: ${(e as unknown as any).toString()}. Rolling back.`
+        // );
+
+        throw e;
+      }
+    };
+
+    try {
+      return await f(executor);
+    } finally {
+      client.release();
+    }
+  } catch (e) {
+    if (e.toString().includes("Broken pipe")) {
+      console.log("Broken pipe, resetting pool");
+      await p.end();
+      p = getPool(true);
+      if (brokenPipeFound) {
+        brokenPipeFound = false;
+        throw e;
+      } else {
+        brokenPipeFound = true;
+        return withExecutorAndPool(f, p);
+      }
+
+    }
+    throw e;
+  }
+}
+
+export type Executor = <T>(
+  sql: string,
+  params?: any[]
+) => Promise<QueryObjectResult<T>>;
+export type TransactionBodyFn<R> = (executor: Executor) => Promise<R>;
+
+/**
+ * Invokes a supplied function within a transaction.
+ * @param body Function to invoke. If this throws, the transaction will be rolled
+ * back. The thrown error will be re-thrown.
+ * @param auth
+ */
+export async function transact<R>(body: TransactionBodyFn<R>, auth?: Payload) {
+  return await withExecutor(async (executor) => {
+    return await transactWithExecutor(executor, body, auth);
+  });
+}
+
+async function transactWithExecutor<R>(
+  executor: Executor,
+  body: TransactionBodyFn<R>,
+  auth?: Payload
+) {
+  for (let i = 0; i < 10; i++) {
+    try {
+      await executor("begin");
+      try {
+        if (auth) {
+          await executor(`set local role = ${auth.role}`);
+          await executor(
+            `set local request.jwt.claims = '${JSON.stringify(auth)}'`
+          );
+        }
+        const r = await body(executor);
+        await executor("commit");
+        return r;
+      } catch (e) {
+        await executor("rollback");
+        throw e;
+      }
+    } catch (e) {
+      if (shouldRetryTransaction(e)) {
+
+        continue;
+      }
+
+      if (e.toString().includes("violates row-level security policy")) {
+        console.log("row-level security policy violation - rolling back");
+      } else {
+        // this logs all errors caught, whether we retry or not
+        console.log("caught error", e, "rolling back");
+      }
+      throw e;
+    }
+  }
+  throw new Error("Tried to execute transaction too many times. Giving up.");
+}
+
+//stackoverflow.com/questions/60339223/node-js-transaction-coflicts-in-postgresql-optimistic-concurrency-control-and
+function shouldRetryTransaction(err: unknown) {
+  // eslint-disable-next-line @typescript-eslint/no-explicit-any
+  const code = typeof err === "object" ? String((err as any).code) : null;
+  return code === "40001" || code === "40P01" || ((err as any).toString().includes("could not serialize access due to concurrent update")) || (err as any).toString().includes("Broken pipe");
+}

deno/deps.ts

@@ -0,0 +1,11 @@
+export { Pool } from "https://deno.land/x/[email protected]/pool.ts";
+export { QueryObjectResult } from "https://deno.land/x/[email protected]/query/query.ts";
+
+// jwt decoding
+export { decode, type Payload } from "https://deno.land/x/[email protected]/mod.ts";
+
+export * as Automerge from "https://deno.land/x/[email protected]/index.ts";
+export type { Doc } from "https://deno.land/x/[email protected]/types.ts";
+
+export { encode as cborEncode } from "https://deno.land/x/[email protected]/encode.js";
+export { decode as cborDecode } from "https://deno.land/x/[email protected]/decode.js";