Replicache cvr pull

.github/actions/run-tests/action.yml

@@ -0,0 +1,47 @@
+name: Run tests
+description: Install deps, start the local Supabase stack, and run the vitest suite
+
+runs:
+  using: composite
+  steps:
+    - uses: actions/setup-node@v4
+      with:
+        node-version: 24
+        cache: "npm"
+    - run: npm i
+      shell: bash
+    - uses: supabase/setup-cli@v1
+    - id: cli
+      run: echo "version=$(supabase --version)" >> "$GITHUB_OUTPUT"
+      shell: bash
+    # supabase start spends most of its time pulling images (~2.5GB, dominated
+    # by postgres), so cache them as a tarball keyed on the CLI version, which
+    # determines the image set.
+    - id: image-cache
+      uses: actions/cache@v4
+      with:
+        path: /tmp/supabase-images.tar.zst
+        key: supabase-images-${{ runner.os }}-${{ steps.cli.outputs.version }}
+    - if: steps.image-cache.outputs.cache-hit == 'true'
+      run: zstd -dc /tmp/supabase-images.tar.zst | docker load
+      shell: bash
+    - if: steps.image-cache.outputs.cache-hit != 'true'
+      run: docker images --format '{{.Repository}}:{{.Tag}}' | sort > /tmp/images-before.txt
+      shell: bash
+    # The integration tests need the migrated db + PostgREST behind kong;
+    # everything else is dead weight in CI.
+    - run: supabase start -x gotrue,studio,imgproxy,realtime,storage-api,edge-runtime,logflare,vector,postgres-meta,supavisor
+      shell: bash
+    # Save whatever images supabase start pulled, diffed against the pre-start
+    # list so it tracks registry/tag changes across CLI versions.
+    - if: steps.image-cache.outputs.cache-hit != 'true'
+      run: |
+        new_images=$(comm -13 /tmp/images-before.txt <(docker images --format '{{.Repository}}:{{.Tag}}' | sort))
+        if [ -n "$new_images" ]; then
+          echo "$new_images" | xargs docker save | zstd -T0 > /tmp/supabase-images.tar.zst
+        else
+          echo "No new images pulled; nothing to cache"
+        fi
+      shell: bash
+    - run: npm test
+      shell: bash

.github/workflows/main.yml

@@ -18,6 +18,13 @@ jobs:
       - run: "npm i"
       - run: "npx tsc"
 
+  test:
+    runs-on: ubuntu-latest
+    name: vitest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: ./.github/actions/run-tests
+
   lexicons:
     runs-on: ubuntu-latest
     name: lexicon
@@ -33,7 +40,7 @@ jobs:
       - run: "npm i"
       - run: "npm run publish-lexicons"
   deploy-supabase:
-    needs: [typecheck]
+    needs: [typecheck, test]
     runs-on: ubuntu-latest
     name: Deploy Supabase
     env:

.github/workflows/test.yml

@@ -0,0 +1,28 @@
+name: Test
+
+# On pushes to main the test job in main.yml runs unconditionally and gates
+# the supabase deploy; this workflow only covers PRs.
+on:
+  pull_request:
+    paths:
+      - "**/*.test.ts"
+      - "src/replicache/**"
+      - "supabase/migrations/**"
+      - "supabase/config.toml"
+      - "vitest.config.ts"
+      - "package.json"
+      - "package-lock.json"
+      - ".github/workflows/test.yml"
+      - ".github/actions/run-tests/**"
+
+concurrency:
+  group: test-${{ github.ref }}
+  cancel-in-progress: true
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    name: vitest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: ./.github/actions/run-tests

app/api/rpc/[command]/pull.ts

@@ -1,15 +1,10 @@
 import { z } from "zod";
-import {
-  PullRequest,
-  PullResponseV1,
-  VersionNotSupportedResponse,
-} from "replicache";
-import type { Fact } from "src/replicache";
-import { FactWithIndexes } from "src/replicache/utils";
-import type { Attribute } from "src/replicache/attributes";
+import { VersionNotSupportedResponse } from "replicache";
+import Client from "ioredis";
+import { computePull } from "src/replicache/serverPullData";
+import { makeCVRStore } from "src/replicache/cvrStore";
 import { makeRoute } from "../lib";
 import type { Env } from "./route";
-import type { Json } from "supabase/database.types";
 
 // First define the sub-types for V0 and V1 requests
 const pullRequestV0 = z.object({
@@ -21,7 +16,9 @@ const pullRequestV0 = z.object({
   lastMutationID: z.number(),
 });
 
-// For the Cookie type used in V1
+// For the Cookie type used in V1. New clients hold a CVR cookie (an object
+// with an order property); pre-CVR clients hold a Date.now() number, which
+// computePull answers with a full snapshot.
 const cookieType = z.union([
   z.null(),
   z.string(),
@@ -44,98 +41,21 @@ const pullRequestV1 = z.object({
 // Combined PullRequest type
 const PullRequestSchema = z.union([pullRequestV0, pullRequestV1]);
 
+// Without Redis (dev) every CVR lookup misses and pulls degrade to full
+// snapshots — the pre-CVR behavior.
+const cvrStore = makeCVRStore(
+  process.env.NODE_ENV === "production" && process.env.REDIS_URL
+    ? new Client(process.env.REDIS_URL)
+    : null,
+);
+
 export const pull = makeRoute({
   route: "pull",
   input: z.object({ pullRequest: PullRequestSchema, token_id: z.string() }),
   handler: async ({ pullRequest, token_id }, { supabase }: Env) => {
     let body = pullRequest;
     if (body.pullVersion === 0) return versionNotSupported;
-    let { data, error } = await supabase.rpc("pull_data", {
-      token_id,
-      client_group_id: body.clientGroupID,
-    });
-    if (!data) {
-      console.log(error);
-
-      return {
-        error: "ClientStateNotFound",
-      } as const;
-    }
-
-    let facts = data.facts as {
-      attribute: string;
-      author_did: string | null;
-      created_at: string;
-      data: any;
-      entity: string;
-      id: string;
-      updated_at: string | null;
-      version: number;
-    }[];
-    let publication_data = data.publications as {
-      description: string;
-      title: string;
-      tags: string[];
-      cover_image: string | null;
-      preferences: Json | null;
-    }[];
-    let draft_contributors = (data.draft_contributors as string[] | null) ?? [];
-    let pub_patch = publication_data?.[0]
-      ? [
-          {
-            op: "put",
-            key: "publication_description",
-            value: publication_data[0].description,
-          },
-          {
-            op: "put",
-            key: "publication_title",
-            value: publication_data[0].title,
-          },
-          {
-            op: "put",
-            key: "publication_tags",
-            value: publication_data[0].tags || [],
-          },
-          {
-            op: "put",
-            key: "post_preferences",
-            value: publication_data[0].preferences || null,
-          },
-        ]
-      : [];
-
-    let clientGroup = (
-      (data.client_groups as {
-        client_id: string;
-        client_group: string;
-        last_mutation: number;
-      }[]) || []
-    ).reduce(
-      (acc, clientRecord) => {
-        acc[clientRecord.client_id] = clientRecord.last_mutation;
-        return acc;
-      },
-      {} as { [clientID: string]: number },
-    );
-
-    return {
-      cookie: Date.now(),
-      lastMutationIDChanges: clientGroup,
-      patch: [
-        { op: "clear" },
-        { op: "put", key: "initialized", value: true },
-        ...(facts || []).map((f) => {
-          return {
-            op: "put",
-            key: f.id,
-            value: FactWithIndexes(f as unknown as Fact<Attribute>),
-          } as const;
-        }),
-        ...pub_patch,
-        { op: "put", key: "draft_contributors", value: draft_contributors },
-      ],
-    } as PullResponseV1;
+    return await computePull(supabase, cvrStore, body, token_id, Date.now());
   },
 });