feat: use Sync Streams

Author: marcus-pousette

packages/e2ee-chat/README.md

@@ -51,7 +51,7 @@ Run the frontend at the printed URL once the dev server starts.
    -- You can scope this to specific tables if needed, but the publication name must stay "powersync".
    CREATE PUBLICATION powersync FOR ALL TABLES;
    ```
-4. Copy `infra/powersync/sync_rules.yaml` into your PowerSync dashboard so the client can sync the encrypted tables.  
+4. Copy `infra/powersync/sync_streams.yaml` into your PowerSync dashboard so the client can sync the encrypted tables using sync streams.  
 5. Populate `frontend/.env.local` with the Supabase URL and anon key from the dashboard (see Quickstart above).
 
 If you ever need a fresh database during development, run `pnpm --filter @app/chat-e2ee migrate:reset` to drop and reapply the schema locally. To repair discrepancies between your migrations and the remote database, use the Supabase CLI’s `migration repair` command as documented by Supabase.
@@ -91,14 +91,14 @@ If you ever need a fresh database during development, run `pnpm --filter @app/ch
 - **Anonymous sessions** – Enable the Supabase Anonymous provider and the launch screen shows a "Continue as guest" button. Guest users still unlock a local vault, but their messages display the Supabase user UUID unless you add a dedicated `sender_id` column to your schema.
 - **Mirrors** – `startChatMirrors` decrypts encrypted rows per-room, writing plaintext representations into `chat_rooms_plain` and `chat_messages_plain` so the UI can query unencrypted data locally.
 
-## Schema & sync rules
+## Schema & sync streams
 
 - Supabase schema lives at `packages/e2ee-chat/infra/schema.sql`.
-- PowerSync sync rules live at `packages/e2ee-chat/infra/powersync/sync_rules.yaml` and replicate:
-  - The personal vault tables (`chat_e2ee_keys`, `chat_identity_*`).
-  - Per-room buckets: encrypted rooms, messages, membership, and the wrapped room keys for the current user.
+- PowerSync sync streams live at `packages/e2ee-chat/infra/powersync/sync_streams.yaml` and configure: 
+  - Auto-subscribed streams for personal vault tables (`chat_e2ee_keys`, `chat_identity_*`), room metadata, membership records, and wrapped keys.
+  - Parameterized streams (`room_members`, `room_messages`) that the client subscribes to on demand when a room becomes active.
 
-Run the provided Supabase scripts (see `package.json` scripts) to push the schema to your project. Use the PowerSync dashboard to paste the sync rules.
+Run the provided Supabase scripts (see `package.json` scripts) to push the schema to your project. Use the PowerSync dashboard to paste the sync streams configuration.
 
 After editing `infra/schema.sql`, generate and push a fresh migration:
 

packages/e2ee-chat/frontend/package.json

@@ -29,9 +29,9 @@
     "@crypto/webauthn": "workspace:*",
     "@heroicons/react": "^2.2.0",
     "@journeyapps/wa-sqlite": "^1.3.1",
-    "@powersync/common": "^1.38.1",
-    "@powersync/react": "^1.7.4",
-    "@powersync/web": "^1.26.2",
+    "@powersync/common": "^1.39.0",
+    "@powersync/react": "^1.8.0",
+    "@powersync/web": "^1.27.0",
     "@supabase/supabase-js": "^2.45.0",
     "libsodium-wrappers-sumo": "^0.7.13",
     "react": "19.1.1",

packages/e2ee-chat/frontend/src/App.tsx

@@ -1,5 +1,5 @@
 import { useEffect, useMemo, useState } from "react";
-import { usePowerSync, useQuery } from "@powersync/react";
+import { usePowerSync, useQuery, useStatus } from "@powersync/react";
 import { createPasswordCrypto } from "@crypto/password";
 import {
   createDEKCrypto,
@@ -74,21 +74,30 @@ function useSupabaseUser(): {
 }
 
 function useVaultProviders(userId: string | null) {
-  const { data } = useQuery(
+  const { data, isLoading } = useQuery(
     "SELECT provider FROM chat_e2ee_keys WHERE user_id = ?",
     [userId ?? ""],
-    { throttleMs: 150 },
+    {
+      throttleMs: 150,
+      streams: userId
+        ? [{ ...USER_VAULT_KEYS_STREAM, waitForStream: true }]
+        : undefined,
+    },
   );
   return useMemo(() => {
+    if (isLoading) {
+      return { haveAny: false, havePassword: false, loading: true };
+    }
     const rows = Array.isArray(data)
       ? (data as Array<{ provider: string }>)
       : [];
     const providers = new Set(rows.map((r) => r.provider));
     return {
       haveAny: providers.size > 0,
       havePassword: providers.has("password"),
+      loading: false,
     };
-  }, [data]);
+  }, [data, isLoading]);
 }
 
 type RoomPlain = {
@@ -127,10 +136,18 @@ type RoomKeyRow = {
   kdf_salt_b64: string;
 };
 
+const USER_VAULT_KEYS_STREAM = {
+  name: "user_vault_keys",
+  parameters: undefined,
+} as const;
+const ROOM_KEYS_STREAM = { name: "room_keys", parameters: undefined } as const;
+const ROOM_DETAILS_STREAM = { name: "room_details", parameters: undefined } as const;
+
 export default function App() {
   const db = usePowerSync();
   const [dbReady, setDbReady] = useState(false);
   const { userId, authEvent } = useSupabaseUser();
+  const status = useStatus();
   const providers = useVaultProviders(userId);
 
   if (typeof window !== "undefined" && import.meta.env.MODE !== "production") {
@@ -147,7 +164,6 @@ export default function App() {
   const [activeRoomId, setActiveRoomId] = useState<string | null>(null);
   const [pendingRoomSelection, setPendingRoomSelection] =
     useState<string | null>(null);
-  const [mirrorsStarted, setMirrorsStarted] = useState(false);
   const [passwordResetPending, setPasswordResetPending] = useState(false);
   const [guestSupported, setGuestSupported] = useState(() =>
     isAnonymousSupported(),
@@ -212,10 +228,8 @@ export default function App() {
         },
       },
     );
-    setMirrorsStarted(true);
     return () => {
       stop?.();
-      setMirrorsStarted(false);
     };
   }, [db, userId, dataCrypto, roomProviders]);
 
@@ -238,7 +252,12 @@ export default function App() {
   const { data: roomKeyRows } = useQuery(
     "SELECT * FROM chat_room_keys WHERE user_id = ?",
     [userId ?? ""],
-    { throttleMs: 250 },
+    {
+      throttleMs: 250,
+      streams: userId
+        ? [{ ...ROOM_KEYS_STREAM, waitForStream: true }]
+        : undefined,
+    },
   );
 
   useEffect(() => {
@@ -305,7 +324,12 @@ export default function App() {
     ? `SELECT * FROM ${ROOMS_MIRROR_TABLE} ORDER BY updated_at DESC`
     : "SELECT NULL as id WHERE 1 = 0";
 
-  const roomsQuery = useQuery(roomsSql, [], { throttleMs: 200 });
+  const roomsQuery = useQuery(roomsSql, [], {
+    throttleMs: 200,
+    streams: dbReady
+      ? [{ ...ROOM_DETAILS_STREAM, waitForStream: true }]
+      : undefined,
+  });
 
   const roomsData = useMemo(() => {
     if (roomsQuery?.error) {
@@ -368,10 +392,31 @@ export default function App() {
     }
   }, [rooms, activeRoomId, roomKeys, pendingRoomSelection]);
 
+  const roomMessagesStreamDescriptor = useMemo(
+    () =>
+      activeRoomId
+        ? { name: "room_messages", parameters: { room_id: activeRoomId } }
+        : null,
+    [activeRoomId],
+  );
+
+  const roomMembersStreamDescriptor = useMemo(
+    () =>
+      activeRoomId
+        ? { name: "room_members", parameters: { room_id: activeRoomId } }
+        : null,
+    [activeRoomId],
+  );
+
   const { data: messagesData } = useQuery(
     `SELECT * FROM ${MESSAGES_MIRROR_TABLE} WHERE room_id = ? ORDER BY sent_at ASC`,
     [activeRoomId ?? ""],
-    { throttleMs: 120 },
+    {
+      throttleMs: 120,
+      streams: roomMessagesStreamDescriptor
+        ? [{ ...roomMessagesStreamDescriptor, waitForStream: true }]
+        : undefined,
+    },
   );
   const remoteMessages: MessagePlain[] = useMemo(() => {
     if (!Array.isArray(messagesData)) return [];
@@ -407,7 +452,12 @@ export default function App() {
   const { data: membershipData } = useQuery(
     "SELECT * FROM chat_room_members WHERE room_id = ? ORDER BY joined_at ASC",
     [activeRoomId ?? ""],
-    { throttleMs: 300 },
+    {
+      throttleMs: 300,
+      streams: roomMembersStreamDescriptor
+        ? [{ ...roomMembersStreamDescriptor, waitForStream: true }]
+        : undefined,
+    },
   );
   const members: MemberPlain[] = useMemo(() => {
     if (!Array.isArray(membershipData)) return [];
@@ -420,6 +470,16 @@ export default function App() {
     }));
   }, [membershipData]);
 
+  const roomsStreamStatus = status.forStream(ROOM_DETAILS_STREAM);
+  const activeRoomStreamStatus = roomMessagesStreamDescriptor
+    ? status.forStream(roomMessagesStreamDescriptor)
+    : null;
+  const mirrorsStarted = dataCrypto
+    ? (roomsStreamStatus?.subscription?.hasSynced ?? false) &&
+      (!roomMessagesStreamDescriptor ||
+        (activeRoomStreamStatus?.subscription?.hasSynced ?? false))
+    : false;
+
   const canSendToActiveRoom = activeRoomId ? roomKeys.has(activeRoomId) : false;
 
   const handleSignIn = async (email: string, password: string) => {
@@ -715,6 +775,13 @@ export default function App() {
   }
 
   if (!dataCrypto) {
+    if (providers.loading) {
+      return (
+        <div className="min-h-screen bg-slate-950/5 flex items-center justify-center">
+          <div className="text-sm text-slate-500">Preparing encrypted vault…</div>
+        </div>
+      );
+    }
     return (
       <VaultScreen
         hasVault={providers.haveAny}

packages/e2ee-chat/infra/powersync/sync_rules.yaml

@@ -0,0 +1,77 @@
+# PowerSync Sync Streams for the E2EE Chat demo.
+#
+# The configuration combines auto-subscribed streams for user-specific metadata
+# with parameterized streams that are activated on demand by the client when a
+# room is opened. This keeps sensitive encrypted payloads scoped to rooms the
+# user is actively viewing while still syncing key material needed for cryptography.
+
+streams:
+  user_vault_keys:
+    query: |
+      SELECT *
+      FROM public.chat_e2ee_keys
+      WHERE user_id = auth.user_id()
+    auto_subscribe: true
+
+  identity_private_keys:
+    query: |
+      SELECT *
+      FROM public.chat_identity_private_keys
+      WHERE user_id = auth.user_id()
+    auto_subscribe: true
+
+  identity_public_keys:
+    query: |
+      SELECT *
+      FROM public.chat_identity_public_keys
+    auto_subscribe: true
+
+  membership_records:
+    query: |
+      SELECT *
+      FROM public.chat_room_members
+      WHERE user_id = auth.user_id()
+    auto_subscribe: true
+
+  room_details:
+    query: |
+      SELECT *
+      FROM public.chat_rooms
+      WHERE id IN (
+        SELECT room_id
+        FROM public.chat_room_members
+        WHERE user_id = auth.user_id()
+      )
+    auto_subscribe: true
+
+  room_keys:
+    query: |
+      SELECT *
+      FROM public.chat_room_keys
+      WHERE user_id = auth.user_id()
+    auto_subscribe: true
+
+  room_members:
+    query: |
+      SELECT *
+      FROM public.chat_room_members
+      WHERE room_id = subscription.parameter('room_id')
+        AND room_id IN (
+          SELECT room_id
+          FROM public.chat_room_members
+          WHERE user_id = auth.user_id()
+        )
+
+  room_messages:
+    query: |
+      SELECT *
+      FROM public.chat_messages
+      WHERE bucket_id = subscription.parameter('room_id')
+        AND bucket_id IN (
+          SELECT room_id
+          FROM public.chat_room_members
+          WHERE user_id = auth.user_id()
+        )
+
+config:
+  edition: 2