Replicache and Remix boilerplate

.env.example

@@ -0,0 +1 @@
+DATABASE_URL="postgresql://docker:docker@localhost:5432/postgres"

README.md

@@ -2,6 +2,15 @@
 
 📖 See the [Remix docs](https://remix.run/docs) and the [Remix Vite docs](https://remix.run/docs/en/main/guides/vite) for details on supported features.
 
+## Postgres db
+
+- You'll need docker installed on your system
+
+`docker pull postgres`
+`docker run --name replimix-db -e POSTGRES_USER=docker -e POSTGRES_PASSWORD=docker -e POSTGRES_DB=postgres -p 5432:5432 -d postgres`
+
+- Update your .env DATABASE_URL to point to your running db, see the example.
+
 ## Development
 
 Run the Vite dev server:

app/routes/_index.tsx

@@ -1,41 +1,134 @@
 import type { MetaFunction } from "@remix-run/node";
 
+import { createId } from "@paralleldrive/cuid2";
+import { useEffect, useRef, useState } from "react";
+import {
+  Replicache,
+  TEST_LICENSE_KEY,
+  type WriteTransaction,
+} from "replicache";
+import { useSubscribe } from "replicache-react";
+
+import { useEventSource } from "remix-utils/sse/react";
+import { Message, MessageWithID } from "~/utils/replicache";
+
 export const meta: MetaFunction = () => {
   return [
-    { title: "New Remix App" },
-    { name: "description", content: "Welcome to Remix!" },
+    { title: "New Replimix App" },
+    { name: "description", content: "Welcome to Replimix!" },
   ];
 };
 
-export default function Index() {
+const licenseKey =
+  import.meta.env.VITE_REPLICACHE_LICENSE_KEY || TEST_LICENSE_KEY;
+
+if (!licenseKey) {
+  throw new Error("Missing VITE_REPLICACHE_LICENSE_KEY");
+}
+
+/**
+ * Replicache instance for the document with mutators defined.
+ */
+type DocumentReplicache = Replicache<{
+  createMessage: (
+    tx: WriteTransaction,
+    { id, from, content, order }: MessageWithID,
+  ) => Promise<void>;
+}>;
+
+function usePullOnPoke(r: DocumentReplicache | null) {
+  const lastEventId = useEventSource("/api/replicache/poke", {
+    event: "poke",
+  });
+
+  useEffect(() => {
+    r?.pull();
+  }, [lastEventId]);
+}
+
+export default function ReplicachePlayground() {
+  const [replicache, setReplicache] = useState<DocumentReplicache | null>(null);
+
+  usePullOnPoke(replicache);
+
+  useEffect(() => {
+    const r = new Replicache({
+      name: "chat-user-id",
+      licenseKey,
+      schemaVersion: "1",
+      mutators: {
+        async createMessage(
+          tx: WriteTransaction,
+          { id, from, content, order }: MessageWithID,
+        ) {
+          await tx.set(`messages/${id}`, {
+            from,
+            content,
+            order,
+          });
+        },
+      },
+
+      pushURL: "/api/replicache/push",
+      pullURL: "/api/replicache/pull",
+    });
+    setReplicache(r);
+    return () => {
+      void r.close();
+    };
+  }, []);
+
+  const messages = useSubscribe(
+    replicache,
+    async (tx) => {
+      const list = await tx
+        .scan<Message>({ prefix: "messages/" })
+        .entries()
+        .toArray();
+      list.sort(([, { order: a }], [, { order: b }]) => a - b);
+      return list;
+    },
+    { default: [] },
+  );
+
+  const usernameRef = useRef<HTMLInputElement>(null);
+  const contentRef = useRef<HTMLInputElement>(null);
+
+  const onSubmit = async (e: React.FormEvent<HTMLFormElement>) => {
+    e.preventDefault();
+    let last: Message | null = null;
+    if (messages.length) {
+      const lastMessageTuple = messages[messages.length - 1];
+      last = lastMessageTuple[1];
+    }
+    const order = (last?.order ?? 0) + 1;
+    const username = usernameRef.current?.value ?? "";
+    const content = contentRef.current?.value ?? "";
+
+    await replicache?.mutate.createMessage({
+      id: createId(),
+      from: username,
+      content,
+      order,
+    });
+
+    if (contentRef.current) {
+      contentRef.current.value = "";
+    }
+  };
+
   return (
-    <div style={{ fontFamily: "system-ui, sans-serif", lineHeight: "1.8" }}>
-      <h1>Welcome to Remix</h1>
-      <ul>
-        <li>
-          <a
-            target="_blank"
-            href="https://remix.run/tutorials/blog"
-            rel="noreferrer"
-          >
-            15m Quickstart Blog Tutorial
-          </a>
-        </li>
-        <li>
-          <a
-            target="_blank"
-            href="https://remix.run/tutorials/jokes"
-            rel="noreferrer"
-          >
-            Deep Dive Jokes App Tutorial
-          </a>
-        </li>
-        <li>
-          <a target="_blank" href="https://remix.run/docs" rel="noreferrer">
-            Remix Docs
-          </a>
-        </li>
-      </ul>
+    <div>
+      <form onSubmit={onSubmit}>
+        <input ref={usernameRef} required /> says:
+        <input ref={contentRef} required /> <input type="submit" />
+      </form>
+      {messages.map(([k, v]) => (
+        <div key={k}>
+          <b>{v.from}: </b>
+          {v.content}
+        </div>
+      ))}
     </div>
   );
 }

app/routes/api.replicache.poke.tsx

@@ -0,0 +1,26 @@
+import { EventEmitter } from "node:events";
+import { createId } from "@paralleldrive/cuid2";
+import type { LoaderFunctionArgs } from "@remix-run/node";
+import { eventStream } from "remix-utils/sse/server";
+import { replog } from "~/utils/replicache";
+
+const emitter = new EventEmitter();
+
+export const sendPoke = () => emitter.emit("poke");
+
+// TODO: This should be grouped into channels/documents somehow.
+export async function loader({ request }: LoaderFunctionArgs) {
+  return eventStream(request.signal, function setup(send) {
+    function handle() {
+      const id = createId();
+      send({ event: "poke", data: id });
+      replog("poked", id);
+    }
+
+    emitter.on("poke", handle);
+
+    return function clear() {
+      emitter.off("poke", handle);
+    };
+  });
+}

app/routes/api.replicache.pull.tsx

@@ -0,0 +1,104 @@
+import { type ActionFunctionArgs, json } from "@remix-run/node";
+import { replicacheServerId } from "db";
+import { sql } from "drizzle-orm";
+import type { PatchOperation, PullResponse } from "replicache";
+import { type TransactionExecutor, tx } from "./api.replicache.push";
+import { ereplog, replog } from "~/utils/replicache";
+
+export async function action({ request }: ActionFunctionArgs) {
+  const resp = await pull(request);
+  return json(resp ?? "{}");
+}
+
+async function pull(req: Request) {
+  const pull = await req.json();
+  replog("Processing pull", JSON.stringify(pull));
+  const { clientGroupID } = pull;
+  const fromVersion = pull.cookie ?? 0;
+  const t0 = Date.now();
+
+  try {
+    // Read all data in a single transaction so it's consistent.
+    return await tx(async (t) => {
+      // Get current version.
+      const [{ version: currentVersion }] = await t.execute<{
+        version: number;
+      }>(sql`select version from replicache_server where id = ${replicacheServerId}`);
+
+      if (fromVersion > currentVersion) {
+        throw new Error(
+          `fromVersion ${fromVersion} is from the future - aborting. This can happen in development if the server restarts. In that case, clear appliation data in browser and refresh.`,
+        );
+      }
+
+      // Get lmids for requesting client groups.
+      const lastMutationIDChanges = await getLastMutationIDChanges(
+        t,
+        clientGroupID,
+        fromVersion,
+      );
+
+      // Get changed domain objects since requested version.
+      const changed = await t.execute<{
+        id: string;
+        sender: string;
+        content: string;
+        ord: number;
+        version: number;
+        deleted: boolean;
+      }>(
+        sql`select id, sender, content, ord, version, deleted from messages where version > ${fromVersion}`,
+      );
+
+      // Build and return response.
+      const patch: PatchOperation[] = [];
+      for (const row of changed) {
+        const { id, sender, content, ord, version: rowVersion, deleted } = row;
+        if (deleted) {
+          if (rowVersion > fromVersion) {
+            patch.push({
+              op: "del",
+              key: `messages/${id}`,
+            });
+          }
+        } else {
+          patch.push({
+            op: "put",
+            key: `messages/${id}`,
+            value: {
+              from: sender,
+              content,
+              order: ord,
+            },
+          });
+        }
+      }
+
+      const body: PullResponse = {
+        lastMutationIDChanges: lastMutationIDChanges ?? {},
+        cookie: currentVersion,
+        patch,
+      };
+
+      return body;
+    });
+  } catch (e) {
+    ereplog(e);
+    return json({ error: e }, { status: 500 });
+  } finally {
+    replog("Processed pull in", Date.now() - t0);
+  }
+}
+
+async function getLastMutationIDChanges(
+  t: TransactionExecutor,
+  clientGroupID: string,
+  fromVersion: number,
+) {
+  const rows = await t.execute<{ id: string; last_mutation_id: number }>(
+    sql`select id, last_mutation_id
+      from replicache_client
+      where client_group_id = ${clientGroupID} and version > ${fromVersion}`,
+  );
+  return Object.fromEntries(rows.map((r) => [r.id, r.last_mutation_id]));
+}