Release

when having a co.encoded(Schema.NullOr(Schema.String)), construction
with null works well, but mutating a value into null throws.
This commit fixes it and adds a test that verifies it actually works.

.changeset/config.json

@@ -12,7 +12,7 @@
             "jazz-react",
             "jazz-nodejs",
             "jazz-run",
-            "cojson-transport-nodejs-ws",
+            "cojson-transport-ws",
             "cojson-storage-indexeddb",
             "cojson-storage-sqlite"
         ]

.github/workflows/build-and-deploy.yaml

@@ -11,7 +11,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        example: ["chat", "pets", "todo"]
+        example: ["chat", "pets", "todo", "inspector"]
         # example: ["twit", "chat", "counter-js-auth0", "pets", "twit", "file-drop", "inspector"]
 
     steps:
@@ -105,7 +105,7 @@ jobs:
     needs: build-examples
     strategy:
       matrix:
-        example: ["chat", "pets", "todo"]
+        example: ["chat", "pets", "todo", "inspector"]
         # example: ["twit", "chat", "counter-js-auth0", "pets", "twit", "file-drop", "inspector"]
 
     steps:

examples/chat/CHANGELOG.md

@@ -1,5 +1,63 @@
 # jazz-example-chat
 
+## 0.0.64
+
+### Patch Changes
+
+- Updated dependencies
+  - cojson@0.7.17
+  - jazz-react@0.7.17
+  - jazz-tools@0.7.17
+
+## 0.0.63
+
+### Patch Changes
+
+- Updated dependencies
+  - jazz-tools@0.7.16
+  - jazz-react@0.7.16
+
+## 0.0.62
+
+### Patch Changes
+
+- Updated dependencies
+  - jazz-react@0.7.15
+
+## 0.0.61
+
+### Patch Changes
+
+- Updated dependencies
+  - cojson@0.7.14
+  - jazz-tools@0.7.14
+  - jazz-react@0.7.14
+
+## 0.0.60
+
+### Patch Changes
+
+- Updated dependencies
+  - jazz-tools@0.7.13
+  - jazz-react@0.7.13
+
+## 0.0.59
+
+### Patch Changes
+
+- Updated dependencies
+  - jazz-tools@0.7.12
+  - jazz-react@0.7.12
+
+## 0.0.58
+
+### Patch Changes
+
+- Updated dependencies
+  - cojson@0.7.11
+  - jazz-react@0.7.11
+  - jazz-tools@0.7.11
+
 ## 0.0.57
 
 ### Patch Changes

examples/chat/package.json

@@ -1,7 +1,7 @@
 {
   "name": "jazz-example-chat",
   "private": true,
-  "version": "0.0.57",
+  "version": "0.0.64",
   "type": "module",
   "scripts": {
     "dev": "vite",

examples/inspector/CHANGELOG.md

@@ -1,5 +1,13 @@
 # jazz-example-chat
 
+## 0.0.48
+
+### Patch Changes
+
+- Updated dependencies
+  - cojson@0.7.17
+  - cojson-transport-ws@0.7.17
+
 ## 0.0.47
 
 ### Patch Changes

examples/inspector/package.json

@@ -1,7 +1,7 @@
 {
-  "name": "inspector",
+  "name": "jazz-inspector",
   "private": true,
-  "version": "0.0.47",
+  "version": "0.0.48",
   "type": "module",
   "scripts": {
     "dev": "vite",
@@ -17,9 +17,9 @@
     "class-variance-authority": "^0.7.0",
     "clsx": "^2.0.0",
     "hash-slash": "workspace:*",
-    "jazz-react": "workspace:*",
-    "jazz-react-auth-local": "workspace:*",
     "cojson": "workspace:*",
+    "cojson-transport-ws": "workspace:*",
+    "effect": "^3.1.5",
     "lucide-react": "^0.274.0",
     "qrcode": "^1.5.3",
     "react": "^18.2.0",

examples/inspector/src/app.tsx

@@ -0,0 +1,309 @@
+import ReactDOM from "react-dom/client";
+import {
+    RawAccount,
+    CoID,
+    RawCoValue,
+    SessionID,
+    LocalNode,
+    AgentSecret,
+    AccountID,
+    cojsonInternals,
+    WasmCrypto,
+} from "cojson";
+import { clsx } from "clsx";
+import { AccountInfo, CoJsonTree, Tag } from "./cojson-tree";
+import { useEffect, useState } from "react";
+import { createWebSocketPeer } from "cojson-transport-ws";
+import { Effect } from "effect";
+
+ReactDOM.createRoot(document.getElementById("root")!).render(<App />);
+
+function App() {
+    const [accountID, setAccountID] = useState<CoID<RawAccount>>(
+        localStorage["inspectorAccountID"]
+    );
+    const [accountSecret, setAccountSecret] = useState<AgentSecret>(
+        localStorage["inspectorAccountSecret"]
+    );
+
+    const [coValueId, setCoValueId] = useState<CoID<RawCoValue>>(
+        window.location.hash.slice(2) as CoID<RawCoValue>
+    );
+
+    useEffect(() => {
+        window.addEventListener("hashchange", () => {
+            setCoValueId(window.location.hash.slice(2) as CoID<RawCoValue>);
+        });
+    });
+
+    const [localNode, setLocalNode] = useState<LocalNode>();
+
+    useEffect(() => {
+        if (!accountID || !accountSecret) return;
+        WasmCrypto.create().then(async (crypto) => {
+            const wsPeer = await Effect.runPromise(
+                createWebSocketPeer({
+                    id: "mesh",
+                    websocket: new WebSocket("wss://mesh.jazz.tools"),
+                    role: "server",
+                })
+            );
+            const node = await LocalNode.withLoadedAccount({
+                accountID: accountID,
+                accountSecret: accountSecret,
+                sessionID: cojsonInternals.newRandomSessionID(accountID),
+                peersToLoadFrom: [wsPeer],
+                crypto,
+                migration: async () => {
+                    console.log("Not running any migration in inspector");
+                },
+            });
+            setLocalNode(node);
+        });
+    }, [accountID, accountSecret]);
+
+    return (
+        <div className="flex flex-col items-center w-screen h-screen p-2 gap-2">
+            <div className="flex gap-2 items-center">
+                Account
+                <input
+                    className="border p-2 rounded"
+                    placeholder="Account ID"
+                    value={accountID}
+                    onChange={(e) => {
+                        setAccountID(e.target.value as AccountID);
+                        localStorage["inspectorAccountID"] = e.target.value;
+                    }}
+                />
+                <input
+                    type="password"
+                    className="border p-2 rounded"
+                    placeholder="Account Secret"
+                    value={accountSecret}
+                    onChange={(e) => {
+                        setAccountSecret(e.target.value as AgentSecret);
+                        localStorage["inspectorAccountSecret"] = e.target.value;
+                    }}
+                />
+                {localNode ? (
+                    <AccountInfo accountID={accountID} node={localNode} />
+                ) : (
+                    ""
+                )}
+            </div>
+            <div className="flex gap-2 items-center">
+                CoValue ID
+                <input
+                    className="border p-2 rounded min-w-[20rem]"
+                    placeholder="CoValue ID"
+                    value={coValueId}
+                    onChange={(e) =>
+                        setCoValueId(e.target.value as CoID<RawCoValue>)
+                    }
+                />
+            </div>
+            {coValueId && localNode ? (
+                <Inspect coValueId={coValueId} node={localNode} />
+            ) : null}
+        </div>
+    );
+}
+
+// function ImageCoValue({ value }: { value: ImageDefinition["_shape"] }) {
+//     const keys = Object.keys(value);
+//     const keyIncludingRes = keys.find((key) => key.includes("x"));
+//     const idToResolve = keyIncludingRes
+//         ? value[keyIncludingRes as `${number}x${number}`]
+//         : null;
+
+//     if (!idToResolve) return <div>Can't find image</div>;
+
+//     const [blobURL, setBlobURL] = useState<string>();
+
+//     useEffect(() => {
+
+//     })
+
+//     return (
+//         <img
+//             src={image?.blobURL || value.placeholderDataURL}
+//             alt="placeholder"
+//         />
+//     );
+// }
+
+function Inspect({
+    coValueId,
+    node,
+}: {
+    coValueId: CoID<RawCoValue>;
+    node: LocalNode;
+}) {
+    const [coValue, setCoValue] = useState<RawCoValue | "unavailable">();
+
+    useEffect(() => {
+        return node.subscribe(coValueId, (coValue) => {
+            setCoValue(coValue);
+        });
+    }, [node, coValueId]);
+
+    if (coValue === "unavailable") {
+        return <div>Unavailable</div>;
+    }
+
+    const values = coValue?.toJSON() || {};
+    const isImage =
+        typeof values === "object" && "placeholderDataURL" in values;
+    const isGroup = coValue?.core.header.ruleset?.type === "group";
+
+    const entires = Object.entries(values as any) as [string, string][];
+    const onlyCoValues = entires.filter(([key]) => key.startsWith("co_"));
+
+    let title = "";
+    if (isImage) {
+        title = "Image";
+    } else if (isGroup) {
+        title = "Group";
+    }
+
+    return (
+        <div className="mb-auto">
+            <h1 className="text-xl font-bold mb-2">
+                Inspecting {title}{" "}
+                <span className="text-gray-500 text-sm">{coValueId}</span>
+            </h1>
+
+            {isGroup ? (
+                <p>
+                    {onlyCoValues.length > 0 ? <h3>Permissions</h3> : ""}
+                    <div className="flex gap-2 flex-col">
+                        {onlyCoValues?.map(([key, value]) => (
+                            <div className="flex gap-1 items-center">
+                                <span className="bg-gray-200 text-xs px-2 py-0.5 rounded">
+                                    {value}
+                                </span>
+                                <AccountInfo
+                                    accountID={key as CoID<RawAccount>}
+                                    node={node}
+                                />
+                            </div>
+                        ))}
+                    </div>
+                </p>
+            ) : (
+                <span className="">
+                    Group{" "}
+                    <Tag href={`#/${coValue?.group.id}`}>
+                        {coValue?.group.id}
+                    </Tag>
+                </span>
+            )}
+            {/* {isImage ? (
+                <div className="my-2">
+                    <ImageCoValue value={values as any} />
+                </div>
+            ) : null} */}
+            <pre className="max-w-[80vw] overflow-scroll text-sm mt-4">
+                <CoJsonTree coValueId={coValueId} node={node} />
+            </pre>
+            <h2 className="text-lg font-semibold mt-10 mb-4">Sessions</h2>
+            {coValue && <Sessions coValue={coValue} node={node} />}
+        </div>
+    );
+}
+
+function Sessions({ coValue, node }: { coValue: RawCoValue; node: LocalNode }) {
+    const validTx = coValue.core.getValidSortedTransactions();
+    return (
+        <div className="max-w-[80vw] border rounded">
+            {[...coValue.core.sessionLogs.entries()].map(
+                ([sessionID, session]) => (
+                    <div
+                        key={sessionID}
+                        className="mv-10 flex gap-2 border-b p-5 flex-wrap  flex-col"
+                    >
+                        <div className="flex gap-2 flex-row">
+                            <SessionInfo
+                                sessionID={sessionID}
+                                transactionCount={session.transactions.length}
+                                node={node}
+                            />
+                        </div>
+                        <div className="flex gap-1 flex-wrap max-h-64 overflow-y-auto p-1 bg-gray-50 rounded">
+                            {session.transactions.map((tx, txIdx) => {
+                                const correspondingValidTx = validTx.find(
+                                    (validTx) =>
+                                        validTx.txID.sessionID === sessionID &&
+                                        validTx.txID.txIndex == txIdx
+                                );
+                                return (
+                                    <div
+                                        key={txIdx}
+                                        className={clsx(
+                                            "text-xs flex-1 p-2 border rounded min-w-36 max-w-40 overflow-scroll bg-white",
+                                            !correspondingValidTx &&
+                                                "bg-red-50 border-red-100"
+                                        )}
+                                    >
+                                        <div>
+                                            {new Date(
+                                                tx.madeAt
+                                            ).toLocaleString()}
+                                        </div>
+                                        <div>{tx.privacy}</div>
+                                        <pre>
+                                            {correspondingValidTx
+                                                ? JSON.stringify(
+                                                      correspondingValidTx.changes,
+                                                      undefined,
+                                                      2
+                                                  )
+                                                : "invalid/undecryptable"}
+                                        </pre>
+                                    </div>
+                                );
+                            })}
+                        </div>
+                        <div className="text-xs">
+                            {session.lastHash} / {session.lastSignature}{" "}
+                        </div>
+                    </div>
+                )
+            )}
+        </div>
+    );
+}
+
+function SessionInfo({
+    sessionID,
+    transactionCount,
+    node,
+}: {
+    sessionID: SessionID;
+    transactionCount: number;
+    node: LocalNode;
+}) {
+    let Prefix = sessionID.startsWith("co_") ? (
+        <AccountInfo
+            accountID={sessionID.split("_session_")[0] as CoID<RawAccount>}
+            node={node}
+        />
+    ) : (
+        <pre className="text-xs">{sessionID.split("_session_")[0]}</pre>
+    );
+
+    return (
+        <div>
+            {Prefix}
+            <div>
+                <span className="text-xs">
+                    Session {sessionID.split("_session_")[1]}
+                </span>
+                <span className="text-xs text-gray-600 font-medium">
+                    {" "}
+                    - {transactionCount} txs
+                </span>
+            </div>
+        </div>
+    );
+}

examples/inspector/src/cojson-tree.tsx

@@ -0,0 +1,249 @@
+import clsx from "clsx";
+import { AccountID, CoID, LocalNode, RawAccount, RawCoMap, RawCoValue } from "cojson";
+import { useEffect, useState } from "react";
+import { LinkIcon } from "./link-icon";
+
+export function CoJsonTree({
+    coValueId,
+    node,
+}: {
+    coValueId: CoID<RawCoValue>;
+    node: LocalNode;
+}) {
+    const [coValue, setCoValue] = useState<RawCoValue | "unavailable">();
+
+    useEffect(() => {
+        return node.subscribe(coValueId, (value) => {
+            setCoValue(value);
+        });
+    });
+
+    if (coValue === "unavailable") {
+        return <div className="text-red-500">Unavailable</div>;
+    }
+
+    const values = coValue?.toJSON() || {};
+
+    return <RenderCoValueJSON json={values} node={node} />;
+}
+
+function RenderObject({
+    json,
+    node,
+}: {
+    json: Record<string, any>;
+    node: LocalNode;
+}) {
+    const [limit, setLimit] = useState(10);
+    const hasMore = Object.keys(json).length > limit;
+
+    const entries = Object.entries(json).slice(0, limit);
+    return (
+        <div className="flex gap-x-1 flex-col font-mono text-xs overflow-auto">
+            {"{"}
+            {entries.map(([key, value]) => {
+                return (
+                    <RenderObjectValue
+                        property={key}
+                        value={value}
+                        node={node}
+                    />
+                );
+            })}
+            {hasMore ? (
+                <div
+                    className="text-gray-500 cursor-pointer"
+                    onClick={() => setLimit((l) => l + 10)}
+                >
+                    ... {Object.keys(json).length - limit} more
+                </div>
+            ) : null}
+            {"}"}
+        </div>
+    );
+}
+
+function RenderObjectValue({
+    property,
+    value,
+    node,
+}: {
+    property: string;
+    value: any;
+    node: LocalNode;
+}) {
+    const [shouldLoad, setShouldLoad] = useState(false);
+
+    const isCoValue =
+        typeof value === "string" ? value?.startsWith("co_") : false;
+
+    return (
+        <div className={clsx(`flex group`)}>
+            <div className="text-gray-500 flex items-start">
+                <div className="flex items-center">
+                    <RenderCoValueJSON json={property} node={node} />:{" "}
+                </div>
+            </div>
+
+            {isCoValue ? (
+                <div className={clsx(shouldLoad && "pb-2")}>
+                    <div className="flex items-center ">
+                        <div onClick={() => setShouldLoad((s) => !s)}>
+                            <div className="w-8 text-center text-gray-700 font-mono px-1 text-xs rounded hover:bg-gray-300 cursor-pointer">
+                                {shouldLoad ? `-` : `...`}
+                            </div>
+                        </div>
+                        <a
+                            href={`#/${value}`}
+                            className="ml-2 group-hover:block hidden"
+                        >
+                            <LinkIcon />
+                        </a>
+                    </div>
+                    <span>
+                        {shouldLoad ? (
+                            <CoJsonTree coValueId={value} node={node} />
+                        ) : null}
+                    </span>
+                </div>
+            ) : (
+                <div className="">
+                    <RenderCoValueJSON json={value} node={node} />
+                </div>
+            )}
+        </div>
+    );
+}
+
+function RenderCoValueArray({ json, node }: { json: any[]; node: LocalNode }) {
+    const [limit, setLimit] = useState(10);
+    const hasMore = json.length > limit;
+
+    const entries = json.slice(0, limit);
+    return (
+        <div className="flex gap-x-1 flex-col font-mono text-xs overflow-auto">
+            {entries.map((value, idx) => {
+                return (
+                    <div key={idx} className="flex gap-x-1">
+                        <RenderCoValueJSON json={value} node={node} />
+                    </div>
+                );
+            })}
+            {hasMore ? (
+                <div
+                    className="text-gray-500 cursor-pointer"
+                    onClick={() => setLimit((l) => l + 10)}
+                >
+                    ... {json.length - limit} more
+                </div>
+            ) : null}
+        </div>
+    );
+}
+
+function RenderCoValueJSON({
+    json,
+    node,
+}: {
+    json:
+        | Record<string, any>
+        | any[]
+        | string
+        | null
+        | number
+        | boolean
+        | undefined;
+    node: LocalNode;
+}) {
+    if (typeof json === "undefined") {
+        return <>"undefined"</>;
+    } else if (Array.isArray(json)) {
+        return (
+            <div className="">
+                <span className="text-gray-500">[</span>
+                <div className="ml-2">
+                    <RenderCoValueArray json={json} node={node} />
+                </div>
+                <span className="text-gray-500">]</span>
+            </div>
+        );
+    } else if (
+        typeof json === "object" &&
+        json &&
+        Object.getPrototypeOf(json) === Object.prototype
+    ) {
+        return <RenderObject json={json} node={node} />;
+    } else if (typeof json === "string") {
+        if (json?.startsWith("co_")) {
+            if (json.includes("_session_")) {
+                return (
+                    <>
+                        <AccountInfo accountID={json.split("_session_")[0] as AccountID} node={node}/>{" "}
+                        (sess {json.split("_session_")[1]})
+                    </>
+                );
+            } else {
+                return (
+                    <>
+                        <a className="underline" href={`#/${json}`}>
+                            {'"'}
+                            {json}
+                            {'"'}
+                        </a>
+                    </>
+                );
+            }
+        } else {
+            return <div className="truncate max-w-64 ml-1">{json}</div>;
+        }
+    } else {
+        return <div className="truncate max-w-64">{JSON.stringify(json)}</div>;
+    }
+}
+
+export function AccountInfo({ accountID, node }: { accountID: CoID<RawAccount>, node: LocalNode }) {
+    const [name, setName] = useState<string | null>(null);
+
+    useEffect(() => {
+        (async () => {
+            const account = await node.load(accountID);
+            if (account === "unavailable") return;
+            const profileID = account?.get("profile");
+            if (profileID === undefined) return;
+            const profile = await node.load(profileID as CoID<RawCoMap>);
+            if (profile === "unavailable") return;
+            setName(profile?.get("name") as string);
+        })()
+    }, [accountID, node]);
+
+    return name ? (
+            <Tag href={`#/${accountID}`} title={accountID}><h1>{name}</h1></Tag>
+    ) : (
+        <Tag href={`#/${accountID}`}>{accountID}</Tag>
+    );
+}
+
+export function Tag({
+    children,
+    href,
+    title
+}: {
+    children: React.ReactNode;
+    href?: string;
+    title?: string;
+}) {
+    if (href) {
+        return (
+            <a
+                href={href}
+                className="border text-xs px-2 py-0.5 rounded hover:underline"
+                title={title}
+            >
+                {children}
+            </a>
+        );
+    }
+    return (
+        <span className="border text-xs px-2 py-0.5 rounded">{children}</span>
+    );
+}

examples/pets/CHANGELOG.md

@@ -1,5 +1,64 @@
 # jazz-example-pets
 
+## 0.0.82
+
+### Patch Changes
+
+-   jazz-react@0.7.17
+-   jazz-tools@0.7.17
+-   jazz-browser-media-images@0.7.17
+
+## 0.0.81
+
+### Patch Changes
+
+-   Updated dependencies
+    -   jazz-tools@0.7.16
+    -   jazz-browser-media-images@0.7.16
+    -   jazz-react@0.7.16
+
+## 0.0.80
+
+### Patch Changes
+
+-   Updated dependencies
+    -   jazz-react@0.7.15
+
+## 0.0.79
+
+### Patch Changes
+
+-   Updated dependencies
+    -   jazz-tools@0.7.14
+    -   jazz-react@0.7.14
+    -   jazz-browser-media-images@0.7.14
+
+## 0.0.78
+
+### Patch Changes
+
+-   Updated dependencies
+    -   jazz-tools@0.7.13
+    -   jazz-browser-media-images@0.7.13
+    -   jazz-react@0.7.13
+
+## 0.0.77
+
+### Patch Changes
+
+-   Updated dependencies
+    -   jazz-tools@0.7.12
+    -   jazz-browser-media-images@0.7.12
+    -   jazz-react@0.7.12
+
+## 0.0.76
+
+### Patch Changes
+
+-   jazz-react@0.7.11
+-   jazz-tools@0.7.11
+-   jazz-browser-media-images@0.7.11
+
 ## 0.0.75
 
 ### Patch Changes

examples/pets/package.json

@@ -1,7 +1,7 @@
 {
   "name": "jazz-example-pets",
   "private": true,
-  "version": "0.0.75",
+  "version": "0.0.82",
   "type": "module",
   "scripts": {
     "dev": "vite",

examples/todo/CHANGELOG.md

@@ -1,5 +1,58 @@
 # jazz-example-todo
 
+## 0.0.81
+
+### Patch Changes
+
+-   jazz-react@0.7.17
+-   jazz-tools@0.7.17
+
+## 0.0.80
+
+### Patch Changes
+
+-   Updated dependencies
+    -   jazz-tools@0.7.16
+    -   jazz-react@0.7.16
+
+## 0.0.79
+
+### Patch Changes
+
+-   Updated dependencies
+    -   jazz-react@0.7.15
+
+## 0.0.78
+
+### Patch Changes
+
+-   Updated dependencies
+    -   jazz-tools@0.7.14
+    -   jazz-react@0.7.14
+
+## 0.0.77
+
+### Patch Changes
+
+-   Updated dependencies
+    -   jazz-tools@0.7.13
+    -   jazz-react@0.7.13
+
+## 0.0.76
+
+### Patch Changes
+
+-   Updated dependencies
+    -   jazz-tools@0.7.12
+    -   jazz-react@0.7.12
+
+## 0.0.75
+
+### Patch Changes
+
+-   jazz-react@0.7.11
+-   jazz-tools@0.7.11
+
 ## 0.0.74
 
 ### Patch Changes

examples/todo/package.json

@@ -1,7 +1,7 @@
 {
   "name": "jazz-example-todo",
   "private": true,
-  "version": "0.0.74",
+  "version": "0.0.81",
   "type": "module",
   "scripts": {
     "dev": "vite",

homepage/homepage/app/docs/guide.mdx

@@ -178,7 +178,7 @@ function App() {// old
     if (issue) {// old
         return <IssueComponent issue={issue} />; // old
     } else { // old
-        return <button onClick={createIssue}>Create Issue</button>; // old
+        return <button onClick={createIssue}>Create Issue</button>;
     } // old
 } // old
 // old

homepage/homepage/app/layout.tsx

@@ -9,9 +9,11 @@ import localFont from "next/font/local";
 import { GcmpLogo, JazzLogo } from "@/components/logos";
 import { SiGithub, SiDiscord, SiTwitter } from "@icons-pack/react-simple-icons";
 import { Nav, NavLink, Newsletter, NewsletterButton } from "@/components/nav";
-import { MailIcon } from "lucide-react";
 import { DocNav } from "@/components/docs/nav";
 
+import { SpeedInsights } from "@vercel/speed-insights/next"
+import { Analytics } from "@vercel/analytics/react"
+
 // If loading a variable font, you don't need to specify the font weight
 const manrope = Manrope({
     subsets: ["latin"],
@@ -48,6 +50,8 @@ export default function RootLayout({
                     "flex flex-col items-center bg-stone-50 dark:bg-stone-950 overflow-x-hidden",
                 ].join(" ")}
             >
+                <SpeedInsights/>
+                <Analytics/>
                 <ThemeProvider
                     attribute="class"
                     defaultTheme="system"
@@ -192,12 +196,6 @@ export default function RootLayout({
                         </div>
                     </footer>
                 </ThemeProvider>
-                <script
-                    defer
-                    data-api="/api/event"
-                    data-domain="jazz.tools"
-                    src="/js/script.js"
-                ></script>
             </body>
         </html>
     );

homepage/homepage/app/page.mdx

@@ -28,12 +28,6 @@ import Link from "next/link";
 
 <Prose>
 
-<a href="https://app.localfirstconf.com/schedule/conference/every-app-secretly-wants-to-be-local-first" className="-mt-8 md:-mt-20 float-right top-[5rem] right-4 border border-stone-700 dark:border-stone-300 rounded flex gap-3 items-center px-4 py-2 mb-4 rotate-2 md:rotate-6 no-underline hover:scale-105 transition-transform">
-<div className="text-sm font-bold uppercase">See you in Berlin<br/>May 30-31!</div>
-<LocalFirstConfLogo className="w-24"/>
-</a>
-
-
 # Instant sync.
 
 <Slogan>A new way to build apps with distributed state.</Slogan>

homepage/homepage/package.json

@@ -14,6 +14,7 @@
         "*.{ts,tsx}": "eslint --fix",
         "*.{js,jsx,mdx,json}": "prettier --write"
     },
+    "packageManager": "pnpm@9.1.4",
     "dependencies": {
         "@evilmartians/harmony": "^1.0.0",
         "@icons-pack/react-simple-icons": "^9.1.0",
@@ -21,6 +22,8 @@
         "@mdx-js/react": "^2.3.0",
         "@next/mdx": "^13.5.4",
         "@types/mdx": "^2.0.8",
+        "@vercel/analytics": "^1.3.1",
+        "@vercel/speed-insights": "^1.0.12",
         "class-variance-authority": "^0.7.0",
         "clsx": "^2.0.0",
         "lucide-react": "^0.284.0",

oldExamples/inspector/src/app.tsx

@@ -1,243 +0,0 @@
-import {
-  WithJazz,
-  useJazz,
-  DemoAuth,
-  useAutoSub,
-  useBinaryStream,
-} from "jazz-react";
-import ReactDOM from "react-dom/client";
-import { HashRoute } from "hash-slash";
-import { Account, CoID, CoValue, SessionID } from "cojson";
-import { clsx } from "clsx";
-import { ImageDefinition } from "cojson/src/media";
-import { CoJsonTree } from "./cojson-tree";
-
-ReactDOM.createRoot(document.getElementById("root")!).render(
-  <WithJazz
-    auth={DemoAuth({ appName: "Jazz Chat Example" })}
-    apiKey="api_z9d034j3t34ht034ir"
-  >
-    <App />
-  </WithJazz>
-);
-
-function App() {
-  return (
-    <div className="flex flex-col items-center justify-between w-screen h-screen p-2 ">
-      <button
-        onClick={useJazz().logOut}
-        className="rounded mb-5 px-2 py-1 bg-stone-200 dark:bg-stone-800 dark:text-white self-end"
-      >
-        Log Out
-      </button>
-      {HashRoute(
-        {
-          "/": <Home />,
-          "/:id": (id) => <Inspect coValueId={id as CoID<CoValue>} />,
-        },
-        { reportToParentFrame: true }
-      )}
-    </div>
-  );
-}
-
-function Home() {
-  return (
-    <form
-      className="mb-auto"
-      onSubmit={(event) => {
-        const coValueId = (event.target as any).coValueId
-          .value as CoID<CoValue>;
-        location.hash = "/" + coValueId;
-        event.preventDefault();
-      }}
-    >
-      <input name="coValueId" className="border" />
-      <button>Inspect</button>
-    </form>
-  );
-}
-
-function Tag({ children, href }: { children: React.ReactNode; href?: string }) {
-  if (href) {
-    return (
-      <a
-        href={href}
-        className="border text-xs px-2 py-0.5 rounded hover:underline"
-      >
-        {children}
-      </a>
-    );
-  }
-  return <span className="border text-xs px-2 py-0.5 rounded">{children}</span>;
-}
-
-function ImageCoValue({ value }: { value: ImageDefinition["_shape"] }) {
-  const keys = Object.keys(value);
-  const keyIncludingRes = keys.find((key) => key.includes("x"));
-  const idToResolve = keyIncludingRes
-    ? value[keyIncludingRes as `${number}x${number}`]
-    : null;
-
-  if (!idToResolve) return <div>Can't find image</div>;
-
-  const image = useBinaryStream(idToResolve);
-
-  return (
-    <img src={image?.blobURL || value.placeholderDataURL} alt="placeholder" />
-  );
-}
-
-function Inspect({ coValueId }: { coValueId: CoID<CoValue> }) {
-  const coValue = useAutoSub(coValueId);
-
-  const values = coValue?.meta.coValue.toJSON() || {};
-  const isImage = "placeholderDataURL" in values;
-  const isGroup = coValue?.meta.group.id === coValueId;
-
-  const entires = Object.entries(values as any) as [string, string][];
-  const onlyCoValues = entires.filter(([key]) => key.startsWith("co_"));
-
-  let title = "";
-  if (isImage) {
-    title = "Image";
-  } else if (isGroup) {
-    title = "Group";
-  }
-
-  return (
-    <div className="mb-auto">
-      <h1 className="text-xl font-bold mb-2">
-        Inspecting {title}{" "}
-        <span className="text-gray-500 text-sm">{coValueId}</span>
-      </h1>
-
-      {isGroup ? (
-        <p>
-          {onlyCoValues.length > 0 ? <h3>Permissions</h3> : ""}
-          <div className="flex gap-2 flex-col">
-            {onlyCoValues?.map(([key, value]) => (
-              <div className="flex gap-1 items-center">
-                <span className="bg-gray-200 text-xs px-2 py-0.5 rounded">
-                  {value}
-                </span>
-                <AccountInfo accountID={key as CoID<Account>} />
-              </div>
-            ))}
-          </div>
-        </p>
-      ) : (
-        <span className="">
-          Group{" "}
-          <Tag href={`#/${coValue?.meta.group.id}`}>
-            {coValue?.meta.group.id}
-          </Tag>
-        </span>
-      )}
-      {isImage ? (
-        <div className="my-2">
-          <ImageCoValue value={values as any} />
-        </div>
-      ) : null}
-      <pre className="max-w-[80vw] overflow-scroll text-sm mt-4">
-        <CoJsonTree coValueId={coValueId} />
-      </pre>
-      <h2 className="text-lg font-semibold mt-10 mb-4">Sessions</h2>
-      {coValue && <Sessions coValue={coValue.meta.coValue} />}
-    </div>
-  );
-}
-
-function Sessions({ coValue }: { coValue: CoValue }) {
-  const validTx = coValue.core.getValidSortedTransactions();
-  return (
-    <div className="max-w-[80vw] border rounded">
-      {[...coValue.core.sessionLogs.entries()].map(([sessionID, session]) => (
-        <div
-          key={sessionID}
-          className="mv-10 flex gap-2 border-b p-5 flex-wrap  flex-col"
-        >
-          <div className="flex gap-2 flex-row">
-            <SessionInfo
-              sessionID={sessionID}
-              transactionCount={session.transactions.length}
-            />
-          </div>
-          <div className="flex gap-1 flex-wrap max-h-64 overflow-y-auto p-1 bg-gray-50 rounded">
-            {session.transactions.map((tx, txIdx) => {
-              const correspondingValidTx = validTx.find(
-                (validTx) =>
-                  validTx.txID.sessionID === sessionID &&
-                  validTx.txID.txIndex == txIdx
-              );
-              return (
-                <div
-                  key={txIdx}
-                  className={clsx(
-                    "text-xs flex-1 p-2 border rounded min-w-36 max-w-40 overflow-scroll bg-white",
-                    !correspondingValidTx && "bg-red-50 border-red-100"
-                  )}
-                >
-                  <div>{new Date(tx.madeAt).toLocaleString()}</div>
-                  <div>{tx.privacy}</div>
-                  <pre>
-                    {correspondingValidTx
-                      ? JSON.stringify(
-                          correspondingValidTx.changes,
-                          undefined,
-                          2
-                        )
-                      : "invalid/undecryptable"}
-                  </pre>
-                </div>
-              );
-            })}
-          </div>
-          <div className="text-xs">
-            {session.lastHash} / {session.lastSignature}{" "}
-          </div>
-        </div>
-      ))}
-    </div>
-  );
-}
-
-function SessionInfo({
-  sessionID,
-  transactionCount,
-}: {
-  sessionID: SessionID;
-  transactionCount: number;
-}) {
-  let Prefix = sessionID.startsWith("co_") ? (
-    <AccountInfo accountID={sessionID.split("_session_")[0] as CoID<Account>} />
-  ) : (
-    <pre className="text-xs">{sessionID.split("_session_")[0]}</pre>
-  );
-
-  return (
-    <div>
-      {Prefix}
-      <div>
-        <span className="text-xs">
-          Session {sessionID.split("_session_")[1]}
-        </span>
-        <span className="text-xs text-gray-600 font-medium">
-          {" "}
-          - {transactionCount} txs
-        </span>
-      </div>
-    </div>
-  );
-}
-
-function AccountInfo({ accountID }: { accountID: CoID<Account> }) {
-  const account = useAutoSub(accountID);
-  return (
-    <div className="flex items-center gap-2">
-      <h1>{account?.profile?.name}</h1>
-
-      <Tag href={`#/${accountID}`}>{account?.id}</Tag>
-    </div>
-  );
-}

oldExamples/inspector/src/cojson-tree.tsx

@@ -1,151 +0,0 @@
-import clsx from "clsx";
-import { CoID, CoValue } from "cojson";
-import { useAutoSub } from "jazz-react";
-import { useState } from "react";
-import { LinkIcon } from "./link-icon";
-
-export function CoJsonTree({ coValueId }: { coValueId: CoID<CoValue> }) {
-  const coValue = useAutoSub(coValueId);
-
-  const values = coValue?.meta.coValue.toJSON() || {};
-
-  return <RenderCoValueJSON json={values} />;
-}
-
-function RenderObject({ json }: { json: Record<string, any> }) {
-  const [limit, setLimit] = useState(10);
-  const hasMore = Object.keys(json).length > limit;
-
-  const entries = Object.entries(json).slice(0, limit);
-  return (
-    <div className="flex gap-x-1 flex-col font-mono text-xs overflow-auto">
-      {entries.map(([key, value]) => {
-        return <RenderObjectValue property={key} value={value} />;
-      })}
-      {hasMore ? (
-        <div
-          className="text-gray-500 cursor-pointer"
-          onClick={() => setLimit((l) => l + 10)}
-        >
-          ... {Object.keys(json).length - limit} more
-        </div>
-      ) : null}
-    </div>
-  );
-}
-
-function RenderObjectValue({
-  property,
-  value,
-}: {
-  property: string;
-  value: any;
-}) {
-  const [shouldLoad, setShouldLoad] = useState(false);
-
-  const isCoValue =
-    typeof value === "string" ? value?.startsWith("co_") : false;
-
-  return (
-    <div className={clsx(`flex group`)}>
-      <span className="text-gray-500 flex">
-        <RenderCoValueJSON json={property} />:{" "}
-      </span>
-
-      {isCoValue ? (
-        <div className={clsx(shouldLoad && "pb-2")}>
-          <div className="flex items-center ">
-            <div onClick={() => setShouldLoad((s) => !s)}>
-              <div className="w-8 text-center text-gray-700 font-mono px-1 text-xs rounded hover:bg-gray-300 cursor-pointer">
-                {shouldLoad ? `-` : `...`}
-              </div>
-            </div>
-            <a href={`#/${value}`} className="ml-2 group-hover:block hidden">
-              <LinkIcon />
-            </a>
-          </div>
-          <span>{shouldLoad ? <CoJsonTree coValueId={value} /> : null}</span>
-        </div>
-      ) : (
-        <div className="">
-          <RenderCoValueJSON json={value} />
-        </div>
-      )}
-    </div>
-  );
-}
-
-function RenderCoValueArray({ json }: { json: any[] }) {
-  const [limit, setLimit] = useState(10);
-  const hasMore = json.length > limit;
-
-  const entries = json.slice(0, limit);
-  return (
-    <div className="flex gap-x-1 flex-col font-mono text-xs overflow-auto">
-      {entries.map((value, idx) => {
-        return (
-          <div key={idx} className="flex gap-x-1">
-            <RenderCoValueJSON json={value} />
-          </div>
-        );
-      })}
-      {hasMore ? (
-        <div
-          className="text-gray-500 cursor-pointer"
-          onClick={() => setLimit((l) => l + 10)}
-        >
-          ... {json.length - limit} more
-        </div>
-      ) : null}
-    </div>
-  );
-}
-
-function RenderCoValueJSON({
-  json,
-}: {
-  json:
-    | Record<string, any>
-    | any[]
-    | string
-    | null
-    | number
-    | boolean
-    | undefined;
-}) {
-  if (typeof json === "undefined") {
-    return <>"undefined"</>;
-  } else if (Array.isArray(json)) {
-    return (
-      <div className="">
-        <span className="text-gray-500">[</span>
-        <div className="ml-2">
-          <RenderCoValueArray json={json} />
-        </div>
-        <span className="text-gray-500">]</span>
-      </div>
-    );
-  } else if (
-    typeof json === "object" &&
-    json &&
-    Object.getPrototypeOf(json) === Object.prototype
-  ) {
-    return <RenderObject json={json} />;
-  } else if (typeof json === "string") {
-    if (json?.startsWith("co_")) {
-      return (
-        <>
-          <a className="underline" href={`#/${json}`}>
-            {'"'}
-            {json}
-            {'"'}
-          </a>
-        </>
-      );
-    } else {
-      return <div className="truncate max-w-64 ml-1">{json}</div>;
-    }
-  } else {
-    return <div className="truncate max-w-64">{JSON.stringify(json)}</div>;
-  }
-}

package.json

@@ -6,6 +6,7 @@
     "packages/*",
     "examples/*"
   ],
+  "packageManager": "pnpm@9.1.4",
   "devDependencies": {
     "@changesets/cli": "^2.27.3",
     "husky": "^9.0.11",

packages/cojson-storage-indexeddb/CHANGELOG.md

@@ -1,5 +1,26 @@
 # cojson-storage-indexeddb
 
+## 0.7.17
+
+### Patch Changes
+
+-   Updated dependencies
+    -   cojson@0.7.17
+
+## 0.7.14
+
+### Patch Changes
+
+-   Updated dependencies
+    -   cojson@0.7.14
+
+## 0.7.11
+
+### Patch Changes
+
+-   Updated dependencies
+    -   cojson@0.7.11
+
 ## 0.7.10
 
 ### Patch Changes

packages/cojson-storage-indexeddb/package.json

@@ -1,14 +1,14 @@
 {
   "name": "cojson-storage-indexeddb",
-  "version": "0.7.10",
+  "version": "0.7.17",
   "main": "dist/index.js",
   "type": "module",
   "types": "src/index.ts",
   "license": "MIT",
   "dependencies": {
     "cojson": "workspace:*",
-    "typescript": "^5.1.6",
-    "isomorphic-streams": "https://github.com/sgwilym/isomorphic-streams.git#aa9394781bfc92f8d7c981be7daf8af4b4cd4fae"
+    "effect": "^3.1.5",
+    "typescript": "^5.1.6"
   },
   "devDependencies": {
     "@vitest/browser": "^0.34.1",

packages/cojson-storage-indexeddb/src/index.ts

@@ -6,14 +6,11 @@ import {
     CojsonInternalTypes,
     MAX_RECOMMENDED_TX_SIZE,
     AccountID,
+    IncomingSyncStream,
+    OutgoingSyncQueue,
 } from "cojson";
-import {
-    ReadableStream,
-    WritableStream,
-    ReadableStreamDefaultReader,
-    WritableStreamDefaultWriter,
-} from "isomorphic-streams";
 import { SyncPromise } from "./syncPromises.js";
+import { Effect, Queue, Stream } from "effect";
 
 type CoValueRow = {
     id: CojsonInternalTypes.RawCoID;
@@ -46,39 +43,35 @@ type SignatureAfterRow = {
 
 export class IDBStorage {
     db: IDBDatabase;
-    fromLocalNode!: ReadableStreamDefaultReader<SyncMessage>;
-    toLocalNode: WritableStreamDefaultWriter<SyncMessage>;
+    toLocalNode: OutgoingSyncQueue;
 
     constructor(
         db: IDBDatabase,
-        fromLocalNode: ReadableStream<SyncMessage>,
-        toLocalNode: WritableStream<SyncMessage>,
+        fromLocalNode: IncomingSyncStream,
+        toLocalNode: OutgoingSyncQueue,
     ) {
         this.db = db;
-        this.fromLocalNode = fromLocalNode.getReader();
-        this.toLocalNode = toLocalNode.getWriter();
+        this.toLocalNode = toLocalNode;
 
-        void (async () => {
-            let done = false;
-            while (!done) {
-                const result = await this.fromLocalNode.read();
-                done = result.done;
-
-                if (result.value) {
-                    // console.log(
-                    //     "IDB: handling msg",
-                    //     result.value.id,
-                    //     result.value.action
-                    // );
-                    await this.handleSyncMessage(result.value);
-                    // console.log(
-                    //     "IDB: handled msg",
-                    //     result.value.id,
-                    //     result.value.action
-                    // );
-                }
-            }
-        })();
+        void fromLocalNode.pipe(
+            Stream.runForEach((msg) =>
+                Effect.tryPromise({
+                    try: () => this.handleSyncMessage(msg),
+                    catch: (e) =>
+                        new Error(
+                            `Error reading from localNode, handling msg\n\n${JSON.stringify(
+                                msg,
+                                (k, v) =>
+                                    k === "changes" || k === "encryptedChanges"
+                                        ? v.slice(0, 20) + "..."
+                                        : v,
+                            )}`,
+                            { cause: e },
+                        ),
+                }),
+            ),
+            Effect.runPromise,
+        );
     }
 
     static async asPeer(
@@ -89,23 +82,30 @@ export class IDBStorage {
             localNodeName: "local",
         },
     ): Promise<Peer> {
-        const [localNodeAsPeer, storageAsPeer] = cojsonInternals.connectedPeers(
-            localNodeName,
-            "storage",
-            { peer1role: "client", peer2role: "server", trace },
-        );
+        return Effect.runPromise(
+            Effect.gen(function* () {
+                const [localNodeAsPeer, storageAsPeer] =
+                    yield* cojsonInternals.connectedPeers(
+                        localNodeName,
+                        "storage",
+                        { peer1role: "client", peer2role: "server", trace },
+                    );
 
-        await IDBStorage.open(
-            localNodeAsPeer.incoming,
-            localNodeAsPeer.outgoing,
-        );
+                yield* Effect.promise(() =>
+                    IDBStorage.open(
+                        localNodeAsPeer.incoming,
+                        localNodeAsPeer.outgoing,
+                    ),
+                );
 
-        return { ...storageAsPeer, priority: 100 };
+                return { ...storageAsPeer, priority: 100 };
+            }),
+        );
     }
 
     static async open(
-        fromLocalNode: ReadableStream<SyncMessage>,
-        toLocalNode: WritableStream<SyncMessage>,
+        fromLocalNode: IncomingSyncStream,
+        toLocalNode: OutgoingSyncQueue,
     ) {
         const dbPromise = new Promise<IDBDatabase>((resolve, reject) => {
             const request = indexedDB.open("jazz-storage", 4);
@@ -150,23 +150,6 @@ export class IDBStorage {
                         keyPath: ["ses", "idx"],
                     });
                 }
-                // if (ev.oldVersion !== 0 && ev.oldVersion <= 3) {
-                //     // fix embarrassing off-by-one error for transaction indices
-                //     console.log("Migration: fixing off-by-one error");
-                //     const transaction = (
-                //         ev.target as unknown as { transaction: IDBTransaction }
-                //     ).transaction;
-
-                //     const txsStore = transaction.objectStore("transactions");
-                //     const txs = await promised(txsStore.getAll());
-
-                //     for (const tx of txs) {
-                //         await promised(txsStore.delete([tx.ses, tx.idx]));
-                //         tx.idx -= 1;
-                //         await promised(txsStore.add(tx));
-                //     }
-                //     console.log("Migration: fixing off-by-one error - done");
-                // }
             };
         });
 
@@ -409,29 +392,35 @@ export class IDBStorage {
                             ),
                         ).then(() => {
                             // we're done with IndexedDB stuff here so can use native Promises again
-                            setTimeout(async () => {
-                                await this.toLocalNode.write({
-                                    action: "known",
-                                    ...ourKnown,
-                                    asDependencyOf,
-                                });
+                            setTimeout(() =>
+                                Effect.runPromise(
+                                    Effect.gen(this, function* () {
+                                        yield* Queue.offer(this.toLocalNode, {
+                                            action: "known",
+                                            ...ourKnown,
+                                            asDependencyOf,
+                                        });
 
-                                const nonEmptyNewContentPieces =
-                                    newContentPieces.filter(
-                                        (piece) =>
-                                            piece.header ||
-                                            Object.keys(piece.new).length > 0,
-                                    );
+                                        const nonEmptyNewContentPieces =
+                                            newContentPieces.filter(
+                                                (piece) =>
+                                                    piece.header ||
+                                                    Object.keys(piece.new)
+                                                        .length > 0,
+                                            );
 
-                                // console.log(theirKnown.id, nonEmptyNewContentPieces);
+                                        // console.log(theirKnown.id, nonEmptyNewContentPieces);
 
-                                for (const piece of nonEmptyNewContentPieces) {
-                                    await this.toLocalNode.write(piece);
-                                    await new Promise((resolve) =>
-                                        setTimeout(resolve, 0),
-                                    );
-                                }
-                            }, 0);
+                                        for (const piece of nonEmptyNewContentPieces) {
+                                            yield* Queue.offer(
+                                                this.toLocalNode,
+                                                piece,
+                                            );
+                                            yield* Effect.yieldNow();
+                                        }
+                                    }),
+                                ),
+                            );
 
                             return Promise.resolve();
                         });
@@ -456,13 +445,15 @@ export class IDBStorage {
                     const header = msg.header;
                     if (!header) {
                         console.error("Expected to be sent header first");
-                        void this.toLocalNode.write({
-                            action: "known",
-                            id: msg.id,
-                            header: false,
-                            sessions: {},
-                            isCorrection: true,
-                        });
+                        void Effect.runPromise(
+                            Queue.offer(this.toLocalNode, {
+                                action: "known",
+                                id: msg.id,
+                                header: false,
+                                sessions: {},
+                                isCorrection: true,
+                            }),
+                        );
                         throw new Error("Expected to be sent header first");
                     }
 
@@ -524,11 +515,13 @@ export class IDBStorage {
                         ),
                     ).then(() => {
                         if (invalidAssumptions) {
-                            void this.toLocalNode.write({
-                                action: "known",
-                                ...ourKnown,
-                                isCorrection: invalidAssumptions,
-                            });
+                            void Effect.runPromise(
+                                Queue.offer(this.toLocalNode, {
+                                    action: "known",
+                                    ...ourKnown,
+                                    isCorrection: invalidAssumptions,
+                                }),
+                            );
                         }
                     });
                 });

packages/cojson-storage-sqlite/CHANGELOG.md

@@ -1,5 +1,26 @@
 # cojson-storage-sqlite
 
+## 0.7.17
+
+### Patch Changes
+
+-   Updated dependencies
+    -   cojson@0.7.17
+
+## 0.7.14
+
+### Patch Changes
+
+-   Updated dependencies
+    -   cojson@0.7.14
+
+## 0.7.11
+
+### Patch Changes
+
+-   Updated dependencies
+    -   cojson@0.7.11
+
 ## 0.7.10
 
 ### Patch Changes

packages/cojson-storage-sqlite/package.json

@@ -1,15 +1,15 @@
 {
   "name": "cojson-storage-sqlite",
   "type": "module",
-  "version": "0.7.10",
+  "version": "0.7.17",
   "main": "dist/index.js",
   "types": "src/index.ts",
   "license": "MIT",
   "dependencies": {
     "better-sqlite3": "^8.5.2",
     "cojson": "workspace:*",
-    "typescript": "^5.1.6",
-    "isomorphic-streams": "https://github.com/sgwilym/isomorphic-streams.git#aa9394781bfc92f8d7c981be7daf8af4b4cd4fae"
+    "effect": "^3.1.5",
+    "typescript": "^5.1.6"
   },
   "devDependencies": {
     "@types/better-sqlite3": "^7.6.4"

packages/cojson-storage-sqlite/src/index.ts

@@ -6,15 +6,12 @@ import {
     SessionID,
     MAX_RECOMMENDED_TX_SIZE,
     AccountID,
+    IncomingSyncStream,
+    OutgoingSyncQueue,
 } from "cojson";
-import {
-    ReadableStream,
-    WritableStream,
-    ReadableStreamDefaultReader,
-    WritableStreamDefaultWriter,
-} from "isomorphic-streams";
 
 import Database, { Database as DatabaseT } from "better-sqlite3";
+import { Effect, Queue, Stream } from "effect";
 
 type CoValueRow = {
     id: CojsonInternalTypes.RawCoID;
@@ -46,30 +43,36 @@ type SignatureAfterRow = {
 };
 
 export class SQLiteStorage {
-    fromLocalNode!: ReadableStreamDefaultReader<SyncMessage>;
-    toLocalNode: WritableStreamDefaultWriter<SyncMessage>;
+    toLocalNode: OutgoingSyncQueue;
     db: DatabaseT;
 
     constructor(
         db: DatabaseT,
-        fromLocalNode: ReadableStream<SyncMessage>,
-        toLocalNode: WritableStream<SyncMessage>,
+        fromLocalNode: IncomingSyncStream,
+        toLocalNode: OutgoingSyncQueue,
     ) {
         this.db = db;
-        this.fromLocalNode = fromLocalNode.getReader();
-        this.toLocalNode = toLocalNode.getWriter();
+        this.toLocalNode = toLocalNode;
 
-        void (async () => {
-            let done = false;
-            while (!done) {
-                const result = await this.fromLocalNode.read();
-                done = result.done;
-
-                if (result.value) {
-                    await this.handleSyncMessage(result.value);
-                }
-            }
-        })();
+        void fromLocalNode.pipe(
+            Stream.runForEach((msg) =>
+                Effect.tryPromise({
+                    try: () => this.handleSyncMessage(msg),
+                    catch: (e) =>
+                        new Error(
+                            `Error reading from localNode, handling msg\n\n${JSON.stringify(
+                                msg,
+                                (k, v) =>
+                                    k === "changes" || k === "encryptedChanges"
+                                        ? v.slice(0, 20) + "..."
+                                        : v,
+                            )}`,
+                            { cause: e },
+                        ),
+                }),
+            ),
+            Effect.runPromise,
+        );
     }
 
     static async asPeer({
@@ -81,25 +84,32 @@ export class SQLiteStorage {
         trace?: boolean;
         localNodeName?: string;
     }): Promise<Peer> {
-        const [localNodeAsPeer, storageAsPeer] = cojsonInternals.connectedPeers(
-            localNodeName,
-            "storage",
-            { peer1role: "client", peer2role: "server", trace },
-        );
+        return Effect.runPromise(
+            Effect.gen(function* () {
+                const [localNodeAsPeer, storageAsPeer] =
+                    yield* cojsonInternals.connectedPeers(
+                        localNodeName,
+                        "storage",
+                        { peer1role: "client", peer2role: "server", trace },
+                    );
 
-        await SQLiteStorage.open(
-            filename,
-            localNodeAsPeer.incoming,
-            localNodeAsPeer.outgoing,
-        );
+                yield* Effect.promise(() =>
+                    SQLiteStorage.open(
+                        filename,
+                        localNodeAsPeer.incoming,
+                        localNodeAsPeer.outgoing,
+                    ),
+                );
 
-        return { ...storageAsPeer, priority: 100 };
+                return { ...storageAsPeer, priority: 100 };
+            }),
+        );
     }
 
     static async open(
         filename: string,
-        fromLocalNode: ReadableStream<SyncMessage>,
-        toLocalNode: WritableStream<SyncMessage>,
+        fromLocalNode: IncomingSyncStream,
+        toLocalNode: OutgoingSyncQueue,
     ) {
         const db = Database(filename);
         db.pragma("journal_mode = WAL");
@@ -431,11 +441,13 @@ export class SQLiteStorage {
             );
         }
 
-        await this.toLocalNode.write({
-            action: "known",
-            ...ourKnown,
-            asDependencyOf,
-        });
+        await Effect.runPromise(
+            Queue.offer(this.toLocalNode, {
+                action: "known",
+                ...ourKnown,
+                asDependencyOf,
+            }),
+        );
 
         const nonEmptyNewContentPieces = newContentPieces.filter(
             (piece) => piece.header || Object.keys(piece.new).length > 0,
@@ -444,7 +456,7 @@ export class SQLiteStorage {
         // console.log(theirKnown.id, nonEmptyNewContentPieces);
 
         for (const piece of nonEmptyNewContentPieces) {
-            await this.toLocalNode.write(piece);
+            await Effect.runPromise(Queue.offer(this.toLocalNode, piece));
             await new Promise((resolve) => setTimeout(resolve, 0));
         }
     }
@@ -466,13 +478,15 @@ export class SQLiteStorage {
             const header = msg.header;
             if (!header) {
                 console.error("Expected to be sent header first");
-                await this.toLocalNode.write({
-                    action: "known",
-                    id: msg.id,
-                    header: false,
-                    sessions: {},
-                    isCorrection: true,
-                });
+                await Effect.runPromise(
+                    Queue.offer(this.toLocalNode, {
+                        action: "known",
+                        id: msg.id,
+                        header: false,
+                        sessions: {},
+                        isCorrection: true,
+                    }),
+                );
                 return;
             }
 
@@ -604,11 +618,13 @@ export class SQLiteStorage {
         })();
 
         if (invalidAssumptions) {
-            await this.toLocalNode.write({
-                action: "known",
-                ...ourKnown,
-                isCorrection: invalidAssumptions,
-            });
+            await Effect.runPromise(
+                Queue.offer(this.toLocalNode, {
+                    action: "known",
+                    ...ourKnown,
+                    isCorrection: invalidAssumptions,
+                }),
+            );
         }
     }
 

packages/cojson-transport-nodejs-ws/src/index.ts

@@ -1,94 +0,0 @@
-import { WebSocket } from "ws";
-import { WritableStream, ReadableStream } from "isomorphic-streams";
-
-export function websocketReadableStream<T>(ws: WebSocket) {
-    ws.binaryType = "arraybuffer";
-
-    return new ReadableStream<T>({
-        start(controller) {
-            ws.addEventListener("message", (event) => {
-                if (typeof event.data !== "string")
-                    return console.warn(
-                        "Got non-string message from client",
-                        event.data,
-                    );
-                const msg = JSON.parse(event.data);
-                if (msg.type === "ping") {
-                    // console.debug(
-                    //     "Got ping from",
-                    //     msg.dc,
-                    //     "latency",
-                    //     Date.now() - msg.time,
-                    //     "ms"
-                    // );
-                    return;
-                }
-                controller.enqueue(msg);
-            });
-            ws.addEventListener("close", () => {
-                try {
-                    controller.close();
-                } catch (ignore) {
-                    // will throw if already closed, with no way to check before-hand
-                }
-            });
-            ws.addEventListener("error", () =>
-                controller.error(new Error("The WebSocket errored!")),
-            );
-        },
-
-        cancel() {
-            ws.close();
-        },
-    });
-}
-
-export function websocketWritableStream<T>(ws: WebSocket) {
-    return new WritableStream<T>({
-        start(controller) {
-            ws.addEventListener("close", () =>
-                controller.error(
-                    new Error("The WebSocket closed unexpectedly!"),
-                ),
-            );
-            ws.addEventListener("error", () =>
-                controller.error(new Error("The WebSocket errored!")),
-            );
-
-            if (ws.readyState === WebSocket.OPEN) {
-                return;
-            }
-
-            return new Promise((resolve) =>
-                ws.addEventListener("open", resolve, { once: true }),
-            );
-        },
-
-        write(chunk) {
-            ws.send(JSON.stringify(chunk));
-            // Return immediately, since the web socket gives us no easy way to tell
-            // when the write completes.
-        },
-
-        close() {
-            return closeWS(1000);
-        },
-
-        abort(reason) {
-            return closeWS(4000, reason && reason.message);
-        },
-    });
-
-    function closeWS(code: number, reasonString?: string) {
-        return new Promise<void>((resolve, reject) => {
-            ws.onclose = (e) => {
-                if (e.wasClean) {
-                    resolve();
-                } else {
-                    reject(new Error("The connection was not closed cleanly"));
-                }
-            };
-            ws.close(code, reasonString);
-        });
-    }
-}

packages/cojson-transport-ws/CHANGELOG.md

@@ -1,5 +1,26 @@
 # cojson-transport-nodejs-ws
 
+## 0.7.17
+
+### Patch Changes
+
+-   Updated dependencies
+    -   cojson@0.7.17
+
+## 0.7.14
+
+### Patch Changes
+
+-   Updated dependencies
+    -   cojson@0.7.14
+
+## 0.7.11
+
+### Patch Changes
+
+-   Updated dependencies
+    -   cojson@0.7.11
+
 ## 0.7.10
 
 ### Patch Changes

packages/cojson-transport-ws/package.json

@@ -1,15 +1,14 @@
 {
-  "name": "cojson-transport-nodejs-ws",
+  "name": "cojson-transport-ws",
   "type": "module",
-  "version": "0.7.10",
+  "version": "0.7.17",
   "main": "dist/index.js",
   "types": "src/index.ts",
   "license": "MIT",
   "dependencies": {
     "cojson": "workspace:*",
-    "typescript": "^5.1.6",
-    "ws": "^8.14.2",
-    "isomorphic-streams": "https://github.com/sgwilym/isomorphic-streams.git#aa9394781bfc92f8d7c981be7daf8af4b4cd4fae"
+    "effect": "^3.1.5",
+    "typescript": "^5.1.6"
   },
   "scripts": {
     "dev": "tsc --watch --sourceMap --outDir dist",

packages/cojson-transport-ws/src/index.ts

@@ -0,0 +1,108 @@
+import { DisconnectedError, Peer, PingTimeoutError, SyncMessage } from "cojson";
+import { Either, Stream, Queue, Effect, Exit } from "effect";
+
+interface AnyWebSocket {
+    addEventListener(
+        type: "close",
+        listener: (event: { code: number; reason: string }) => void,
+    ): void;
+    addEventListener(
+        type: "message",
+        listener: (event: { data: string | unknown }) => void,
+    ): void;
+    addEventListener(type: "open", listener: () => void): void;
+    close(): void;
+    send(data: string): void;
+}
+
+export function createWebSocketPeer(options: {
+    id: string;
+    websocket: AnyWebSocket;
+    role: Peer["role"];
+}): Effect.Effect<Peer> {
+    return Effect.gen(function* () {
+        const ws = options.websocket;
+
+        const incoming =
+            yield* Queue.unbounded<
+                Either.Either<SyncMessage, DisconnectedError | PingTimeoutError>
+            >();
+        const outgoing = yield* Queue.unbounded<SyncMessage>();
+
+        ws.addEventListener("close", (event) => {
+            void Effect.runPromiseExit(
+                Queue.offer(
+                    incoming,
+                    Either.left(
+                        new DisconnectedError(`${event.code}: ${event.reason}`),
+                    ),
+                ),
+            ).then((e) => {
+                if (Exit.isFailure(e) && !Exit.isInterrupted(e)) {
+                    console.warn("Failed closing ws", e);
+                }
+            });
+        });
+
+        let pingTimeout: ReturnType<typeof setTimeout> | undefined;
+
+        ws.addEventListener("message", (event) => {
+            const msg = JSON.parse(event.data as string);
+
+            if (pingTimeout) {
+                clearTimeout(pingTimeout);
+            }
+
+            pingTimeout = setTimeout(() => {
+                console.debug("Ping timeout");
+                void Effect.runPromise(
+                    Queue.offer(incoming, Either.left(new PingTimeoutError())),
+                );
+                try {
+                    ws.close();
+                } catch (e) {
+                    console.error(
+                        "Error while trying to close ws on ping timeout",
+                        e,
+                    );
+                }
+            }, 2500);
+
+            if (msg.type === "ping") {
+                // eslint-disable-next-line @typescript-eslint/no-explicit-any
+                (globalThis as any).jazzPings =
+                    // eslint-disable-next-line @typescript-eslint/no-explicit-any
+                    (globalThis as any).jazzPings || [];
+                // eslint-disable-next-line @typescript-eslint/no-explicit-any
+                (globalThis as any).jazzPings.push({
+                    received: Date.now(),
+                    sent: msg.time,
+                    dc: msg.dc,
+                });
+                return;
+            } else {
+                void Effect.runPromise(
+                    Queue.offer(incoming, Either.right(msg)),
+                );
+            }
+        });
+
+        ws.addEventListener("open", () => {
+            void Stream.fromQueue(outgoing).pipe(
+                Stream.runForEach((msg) =>
+                    Effect.sync(() => ws.send(JSON.stringify(msg))),
+                ),
+                Effect.runPromise,
+            );
+        });
+
+        return {
+            id: options.id,
+            incoming: Stream.fromQueue(incoming, { shutdown: true }).pipe(
+                Stream.mapEffect((either) => either),
+            ),
+            outgoing,
+            role: options.role,
+        };
+    });
+}

packages/cojson/CHANGELOG.md

@@ -1,5 +1,23 @@
 # cojson
 
+## 0.7.17
+
+### Patch Changes
+
+-   Fix bugs in new storage interface
+
+## 0.7.14
+
+### Patch Changes
+
+-   Use Effect Queues and Streams instead of custom queue implementation
+
+## 0.7.11
+
+### Patch Changes
+
+-   Fix webpack import of node:crypto module
+
 ## 0.7.10
 
 ### Patch Changes

packages/cojson/package.json

@@ -5,7 +5,7 @@
     "types": "src/index.ts",
     "type": "module",
     "license": "MIT",
-    "version": "0.7.10",
+    "version": "0.7.17",
     "devDependencies": {
         "@types/jest": "^29.5.3",
         "@typescript-eslint/eslint-plugin": "^6.2.1",
@@ -23,8 +23,7 @@
         "@noble/hashes": "^1.4.0",
         "@scure/base": "^1.1.1",
         "effect": "^3.1.5",
-        "hash-wasm": "^4.9.0",
-        "isomorphic-streams": "https://github.com/sgwilym/isomorphic-streams.git#aa9394781bfc92f8d7c981be7daf8af4b4cd4fae"
+        "hash-wasm": "^4.9.0"
     },
     "scripts": {
         "dev": "tsc --watch --sourceMap --outDir dist",

packages/cojson/src/crypto/WasmCrypto.ts

@@ -44,11 +44,13 @@ export class WasmCrypto extends CryptoProvider<Uint8Array> {
                 if ("crypto" in globalThis) {
                     resolve();
                 } else {
-                    return import("node:crypto").then(({ webcrypto }) => {
-                        // eslint-disable-next-line @typescript-eslint/no-explicit-any
-                        (globalThis as any).crypto = webcrypto;
-                        resolve();
-                    });
+                    return import(/*webpackIgnore: true*/ "node:crypto").then(
+                        ({ webcrypto }) => {
+                            // eslint-disable-next-line @typescript-eslint/no-explicit-any
+                            (globalThis as any).crypto = webcrypto;
+                            resolve();
+                        },
+                    );
                 }
             }),
         ]).then(([blake3instance]) => new WasmCrypto(blake3instance));

packages/cojson/src/index.ts

@@ -41,7 +41,13 @@ import type {
     BinaryCoStreamMeta,
 } from "./coValues/coStream.js";
 import type { JsonValue } from "./jsonValue.js";
-import type { SyncMessage, Peer } from "./sync.js";
+import type {
+    SyncMessage,
+    Peer,
+    IncomingSyncStream,
+    OutgoingSyncQueue,
+} from "./sync.js";
+import { DisconnectedError, PingTimeoutError } from "./sync.js";
 import type { AgentSecret } from "./crypto/crypto.js";
 import type {
     AccountID,
@@ -117,9 +123,19 @@ export {
     SyncMessage,
     isRawCoID,
     LSMStorage,
+    DisconnectedError,
+    PingTimeoutError,
 };
 
-export type { Value, FileSystem, FSErr, BlockFilename, WalFilename };
+export type {
+    Value,
+    FileSystem,
+    FSErr,
+    BlockFilename,
+    WalFilename,
+    IncomingSyncStream,
+    OutgoingSyncQueue,
+};
 
 // eslint-disable-next-line @typescript-eslint/no-namespace
 export namespace CojsonInternalTypes {

packages/cojson/src/storage/FileSystem.ts

@@ -118,20 +118,21 @@ export function writeBlock<WH, RH, FS extends FileSystem<WH, RH>>(
         const headerBytes = textEncoder.encode(JSON.stringify(blockHeader));
         yield* $(fs.append(file, headerBytes));
 
-        console.log(
-            "full file",
-            yield* $(
-                fs.read(file as unknown as RH, 0, offset + headerBytes.length),
-            ),
-        );
+        // console.log(
+        //     "full file",
+        //     yield* $(
+        //         fs.read(file as unknown as RH, 0, offset + headerBytes.length),
+        //     ),
+        // );
 
         const filename: BlockFilename = `${hash.digest()}-L${level}-H${
             headerBytes.length
         }.jsonl`;
-        console.log("renaming to" + filename);
+        // console.log("renaming to" + filename);
         yield* $(fs.closeAndRename(file, filename));
 
-        console.log("Wrote block", filename, blockHeader);
+        // console.log("Wrote block", filename, blockHeader);
+        // console.log("IDs in block", blockHeader.map(e => e.id));
     });
 }
 

packages/cojson/src/storage/chunksAndKnownStates.ts

@@ -126,10 +126,12 @@ export function mergeChunks(
                 } else {
                     const lastNewEntry = newEntries[newEntries.length - 1]!;
                     lastNewEntry.transactions.push(...entry.transactions);
+                    lastNewEntry.lastSignature = entry.lastSignature;
 
                     bytesSinceLastSignature += entry.transactions.length;
                 }
             }
+            newSessions[sessionID] = newEntries;
         } else {
             return Either.right("nonContigous" as const);
         }

packages/cojson/src/storage/index.ts

@@ -1,18 +1,13 @@
-import {
-    ReadableStream,
-    WritableStream,
-    ReadableStreamDefaultReader,
-    WritableStreamDefaultWriter,
-} from "isomorphic-streams";
-import { Effect, Either, SynchronizedRef } from "effect";
+import { Effect, Either, Queue, Stream, SynchronizedRef } from "effect";
 import { RawCoID } from "../ids.js";
 import { CoValueHeader, Transaction } from "../coValueCore.js";
 import { Signature } from "../crypto/crypto.js";
 import {
     CoValueKnownState,
+    IncomingSyncStream,
     NewContentMessage,
+    OutgoingSyncQueue,
     Peer,
-    SyncMessage,
 } from "../sync.js";
 import { CoID, RawCoValue } from "../index.js";
 import { connectedPeers } from "../streamUtils.js";
@@ -47,9 +42,6 @@ export type CoValueChunk = {
 };
 
 export class LSMStorage<WH, RH, FS extends FileSystem<WH, RH>> {
-    fromLocalNode!: ReadableStreamDefaultReader<SyncMessage>;
-    toLocalNode: WritableStreamDefaultWriter<SyncMessage>;
-    fs: FS;
     currentWal: SynchronizedRef.SynchronizedRef<WH | undefined>;
     coValues: SynchronizedRef.SynchronizedRef<{
         [id: RawCoID]: CoValueChunk | undefined;
@@ -61,44 +53,28 @@ export class LSMStorage<WH, RH, FS extends FileSystem<WH, RH>> {
     >();
 
     constructor(
-        fs: FS,
-        fromLocalNode: ReadableStream<SyncMessage>,
-        toLocalNode: WritableStream<SyncMessage>,
+        public fs: FS,
+        public fromLocalNode: IncomingSyncStream,
+        public toLocalNode: OutgoingSyncQueue,
     ) {
-        this.fs = fs;
-        this.fromLocalNode = fromLocalNode.getReader();
-        this.toLocalNode = toLocalNode.getWriter();
         this.coValues = SynchronizedRef.unsafeMake({});
         this.currentWal = SynchronizedRef.unsafeMake<WH | undefined>(undefined);
 
-        void Effect.runPromise(
-            Effect.gen(this, function* () {
-                let done = false;
-                while (!done) {
-                    const result = yield* Effect.promise(() =>
-                        this.fromLocalNode.read(),
-                    );
-                    done = result.done;
-
-                    if (result.value) {
-                        if (result.value.action === "done") {
-                            continue;
-                        }
-
-                        if (result.value.action === "content") {
-                            yield* this.handleNewContent(result.value);
-                        } else {
-                            yield* this.sendNewContent(
-                                result.value.id,
-                                result.value,
-                                undefined,
-                            );
-                        }
+        void this.fromLocalNode.pipe(
+            Stream.runForEach((msg) =>
+                Effect.gen(this, function* () {
+                    if (msg.action === "done") {
+                        return;
                     }
-                }
 
-                return;
-            }),
+                    if (msg.action === "content") {
+                        yield* this.handleNewContent(msg);
+                    } else {
+                        yield* this.sendNewContent(msg.id, msg, undefined);
+                    }
+                }),
+            ),
+            Effect.runPromise,
         );
 
         setTimeout(() => this.compact(), 20000);
@@ -132,15 +108,13 @@ export class LSMStorage<WH, RH, FS extends FileSystem<WH, RH>> {
             }
 
             if (!coValue) {
-                yield* Effect.promise(() =>
-                    this.toLocalNode.write({
-                        id: id,
-                        action: "known",
-                        header: false,
-                        sessions: {},
-                        asDependencyOf,
-                    }),
-                );
+                yield* Queue.offer(this.toLocalNode, {
+                    id: id,
+                    action: "known",
+                    header: false,
+                    sessions: {},
+                    asDependencyOf,
+                });
 
                 return coValues;
             }
@@ -195,17 +169,15 @@ export class LSMStorage<WH, RH, FS extends FileSystem<WH, RH>> {
 
             const ourKnown: CoValueKnownState = chunkToKnownState(id, coValue);
 
-            yield* Effect.promise(() =>
-                this.toLocalNode.write({
-                    action: "known",
-                    ...ourKnown,
-                    asDependencyOf,
-                }),
-            );
+            yield* Queue.offer(this.toLocalNode, {
+                action: "known",
+                ...ourKnown,
+                asDependencyOf,
+            });
 
             for (const message of newContentMessages) {
                 if (Object.keys(message.new).length === 0) continue;
-                yield* Effect.promise(() => this.toLocalNode.write(message));
+                yield* Queue.offer(this.toLocalNode, message);
             }
 
             return { ...coValues, [id]: coValue };
@@ -260,7 +232,7 @@ export class LSMStorage<WH, RH, FS extends FileSystem<WH, RH>> {
 
                 if (!coValue) {
                     if (newContent.header) {
-                        console.log("Creating in WAL", newContent.id);
+                        // console.log("Creating in WAL", newContent.id);
                         yield* this.withWAL((wal) =>
                             writeToWal(
                                 wal,
@@ -286,7 +258,7 @@ export class LSMStorage<WH, RH, FS extends FileSystem<WH, RH>> {
                         //         })
                         //     )
                         // );
-                        console.warn(
+                        yield* Effect.logWarning(
                             "Incontiguous incoming update for " + newContent.id,
                         );
                         return coValues;
@@ -296,6 +268,23 @@ export class LSMStorage<WH, RH, FS extends FileSystem<WH, RH>> {
                     if (Either.isRight(merged)) {
                         yield* Effect.logWarning(
                             "Non-contigous new content for " + newContent.id,
+                            Object.entries(coValue.sessionEntries).map(
+                                ([session, entries]) =>
+                                    entries.map((entry) => ({
+                                        session: session,
+                                        after: entry.after,
+                                        length: entry.transactions.length,
+                                    })),
+                            ),
+                            Object.entries(
+                                newContentAsChunk.sessionEntries,
+                            ).map(([session, entries]) =>
+                                entries.map((entry) => ({
+                                    session: session,
+                                    after: entry.after,
+                                    length: entry.transactions.length,
+                                })),
+                            ),
                         );
 
                         // yield* Effect.promise(() =>
@@ -308,7 +297,7 @@ export class LSMStorage<WH, RH, FS extends FileSystem<WH, RH>> {
 
                         return coValues;
                     } else {
-                        console.log("Appending to WAL", newContent.id);
+                        // console.log("Appending to WAL", newContent.id);
                         yield* this.withWAL((wal) =>
                             writeToWal(
                                 wal,
@@ -344,6 +333,8 @@ export class LSMStorage<WH, RH, FS extends FileSystem<WH, RH>> {
 
                 const { handle, size } = yield* fs.openToRead(blockFile);
 
+                // console.log("Attempting to load", id, blockFile);
+
                 if (!cachedHeader) {
                     cachedHeader = {};
                     const header = yield* readHeader(
@@ -363,6 +354,8 @@ export class LSMStorage<WH, RH, FS extends FileSystem<WH, RH>> {
                 }
                 const headerEntry = cachedHeader[id];
 
+                // console.log("Header entry", id, headerEntry);
+
                 let result;
                 if (headerEntry) {
                     result = yield* readChunk(handle, headerEntry, fs);
@@ -389,7 +382,7 @@ export class LSMStorage<WH, RH, FS extends FileSystem<WH, RH>> {
 
                 const coValues = new Map<RawCoID, CoValueChunk>();
 
-                console.log("Compacting WAL files", walFiles);
+                yield* Effect.log("Compacting WAL files", walFiles);
                 if (walFiles.length === 0) return;
 
                 yield* SynchronizedRef.updateEffect(this.currentWal, (wal) =>
@@ -402,7 +395,7 @@ export class LSMStorage<WH, RH, FS extends FileSystem<WH, RH>> {
                 );
 
                 for (const fileName of walFiles) {
-                    const { handle, size } =
+                    const { handle, size }: { handle: RH; size: number } =
                         yield* this.fs.openToRead(fileName);
                     if (size === 0) {
                         yield* this.fs.close(handle);
@@ -422,7 +415,7 @@ export class LSMStorage<WH, RH, FS extends FileSystem<WH, RH>> {
                         if (existingChunk) {
                             const merged = mergeChunks(existingChunk, chunk);
                             if (Either.isRight(merged)) {
-                                console.warn(
+                                yield* Effect.logWarning(
                                     "Non-contigous chunks in " +
                                         chunk.id +
                                         ", " +
@@ -452,7 +445,7 @@ export class LSMStorage<WH, RH, FS extends FileSystem<WH, RH>> {
         setTimeout(() => this.compact(), 5000);
     }
 
-    static asPeer<WH, RH, FS extends FileSystem<WH, RH>>({
+    static async asPeer<WH, RH, FS extends FileSystem<WH, RH>>({
         fs,
         trace,
         localNodeName = "local",
@@ -460,15 +453,13 @@ export class LSMStorage<WH, RH, FS extends FileSystem<WH, RH>> {
         fs: FS;
         trace?: boolean;
         localNodeName?: string;
-    }): Peer {
-        const [localNodeAsPeer, storageAsPeer] = connectedPeers(
-            localNodeName,
-            "storage",
-            {
+    }): Promise<Peer> {
+        const [localNodeAsPeer, storageAsPeer] = await Effect.runPromise(
+            connectedPeers(localNodeName, "storage", {
                 peer1role: "client",
                 peer2role: "server",
                 trace,
-            },
+            }),
         );
 
         new LSMStorage(fs, localNodeAsPeer.incoming, localNodeAsPeer.outgoing);

packages/cojson/src/streamUtils.ts

@@ -1,8 +1,4 @@
-import {
-    ReadableStream,
-    TransformStream,
-    WritableStream,
-} from "isomorphic-streams";
+import { Console, Effect, Queue, Stream } from "effect";
 import { Peer, PeerID, SyncMessage } from "./sync.js";
 
 export function connectedPeers(
@@ -17,160 +13,54 @@ export function connectedPeers(
         peer1role?: Peer["role"];
         peer2role?: Peer["role"];
     } = {},
-): [Peer, Peer] {
-    const [inRx1, inTx1] = newStreamPair<SyncMessage>(peer1id + "_in");
-    const [outRx1, outTx1] = newStreamPair<SyncMessage>(peer1id + "_out");
+): Effect.Effect<[Peer, Peer]> {
+    return Effect.gen(function* () {
+        const [from1to2Rx, from1to2Tx] = yield* newQueuePair(
+            trace ? { traceAs: `${peer1id} -> ${peer2id}` } : undefined,
+        );
+        const [from2to1Rx, from2to1Tx] = yield* newQueuePair(
+            trace ? { traceAs: `${peer2id} -> ${peer1id}` } : undefined,
+        );
 
-    const [inRx2, inTx2] = newStreamPair<SyncMessage>(peer2id + "_in");
-    const [outRx2, outTx2] = newStreamPair<SyncMessage>(peer2id + "_out");
+        const peer2AsPeer: Peer = {
+            id: peer2id,
+            incoming: from2to1Rx,
+            outgoing: from1to2Tx,
+            role: peer2role,
+        };
 
-    void outRx2
-        .pipeThrough(
-            new TransformStream({
-                transform(
-                    chunk: SyncMessage,
-                    controller: { enqueue: (msg: SyncMessage) => void },
-                ) {
-                    trace &&
-                        console.debug(
-                            `${peer2id} -> ${peer1id}`,
-                            JSON.stringify(
-                                chunk,
-                                (k, v) =>
-                                    k === "changes" || k === "encryptedChanges"
-                                        ? v.slice(0, 20) + "..."
-                                        : v,
-                                2,
-                            ),
-                        );
-                    controller.enqueue(chunk);
-                },
-            }),
-        )
-        .pipeTo(inTx1);
+        const peer1AsPeer: Peer = {
+            id: peer1id,
+            incoming: from1to2Rx,
+            outgoing: from2to1Tx,
+            role: peer1role,
+        };
 
-    void outRx1
-        .pipeThrough(
-            new TransformStream({
-                transform(
-                    chunk: SyncMessage,
-                    controller: { enqueue: (msg: SyncMessage) => void },
-                ) {
-                    trace &&
-                        console.debug(
-                            `${peer1id} -> ${peer2id}`,
-                            JSON.stringify(
-                                chunk,
-                                (k, v) =>
-                                    k === "changes" || k === "encryptedChanges"
-                                        ? v.slice(0, 20) + "..."
-                                        : v,
-                                2,
-                            ),
-                        );
-                    controller.enqueue(chunk);
-                },
-            }),
-        )
-        .pipeTo(inTx2);
-
-    const peer2AsPeer: Peer = {
-        id: peer2id,
-        incoming: inRx1,
-        outgoing: outTx1,
-        role: peer2role,
-    };
-
-    const peer1AsPeer: Peer = {
-        id: peer1id,
-        incoming: inRx2,
-        outgoing: outTx2,
-        role: peer1role,
-    };
-
-    return [peer1AsPeer, peer2AsPeer];
+        return [peer1AsPeer, peer2AsPeer];
+    });
 }
 
-export function newStreamPair<T>(
-    pairName?: string,
-): [ReadableStream<T>, WritableStream<T>] {
-    let queueLength = 0;
-    let readerClosed = false;
+export function newQueuePair(
+    options: { traceAs?: string } = {},
+): Effect.Effect<[Stream.Stream<SyncMessage>, Queue.Enqueue<SyncMessage>]> {
+    return Effect.gen(function* () {
+        const queue = yield* Queue.unbounded<SyncMessage>();
 
-    let resolveEnqueue: (enqueue: (item: T) => void) => void;
-    const enqueuePromise = new Promise<(item: T) => void>((resolve) => {
-        resolveEnqueue = resolve;
-    });
-
-    let resolveClose: (close: () => void) => void;
-    const closePromise = new Promise<() => void>((resolve) => {
-        resolveClose = resolve;
-    });
-
-    let queueWasOverflowing = false;
-
-    function maybeReportQueueLength() {
-        if (queueLength >= 100) {
-            queueWasOverflowing = true;
-            if (queueLength % 100 === 0) {
-                console.warn(pairName, "overflowing queue length", queueLength);
-            }
+        if (options.traceAs) {
+            return [Stream.fromQueue(queue).pipe(Stream.tap((msg) => Console.debug(
+                options.traceAs,
+                JSON.stringify(
+                    msg,
+                    (k, v) =>
+                        k === "changes" ||
+                        k === "encryptedChanges"
+                            ? v.slice(0, 20) + "..."
+                            : v,
+                    2,
+                ),
+            ))), queue];
         } else {
-            if (queueWasOverflowing) {
-                console.debug(pairName, "ok queue length", queueLength);
-                queueWasOverflowing = false;
-            }
+            return [Stream.fromQueue(queue), queue];
         }
-    }
-
-    const readable = new ReadableStream<T>({
-        async start(controller) {
-            resolveEnqueue(controller.enqueue.bind(controller));
-            resolveClose(controller.close.bind(controller));
-        },
-
-        cancel(_reason) {
-            console.log("Manually closing reader");
-            readerClosed = true;
-        },
-    }).pipeThrough(
-        // eslint-disable-next-line @typescript-eslint/no-explicit-any
-        new TransformStream<any, any>({
-            transform(
-                chunk: SyncMessage,
-                controller: { enqueue: (msg: SyncMessage) => void },
-            ) {
-                queueLength -= 1;
-                maybeReportQueueLength();
-                controller.enqueue(chunk);
-            },
-        }),
-    ) as ReadableStream<T>;
-
-    let lastWritePromise = Promise.resolve();
-
-    const writable = new WritableStream<T>({
-        async write(chunk) {
-            queueLength += 1;
-            maybeReportQueueLength();
-            const enqueue = await enqueuePromise;
-            if (readerClosed) {
-                throw new Error("Reader closed");
-            } else {
-                // make sure write resolves before corresponding read, but make sure writes are still in order
-                await lastWritePromise;
-                lastWritePromise = new Promise((resolve) => {
-                    enqueue(chunk);
-                    resolve();
-                });
-            }
-        },
-        async abort(reason) {
-            console.debug("Manually closing writer", reason);
-            const close = await closePromise;
-            close();
-        },
     });
-
-    return [readable, writable];
 }

packages/cojson/src/sync.ts

@@ -1,13 +1,9 @@
 import { Signature } from "./crypto/crypto.js";
 import { CoValueHeader, Transaction } from "./coValueCore.js";
 import { CoValueCore } from "./coValueCore.js";
-import { LocalNode } from "./localNode.js";
-import {
-    ReadableStream,
-    WritableStream,
-    WritableStreamDefaultWriter,
-} from "isomorphic-streams";
+import { LocalNode, newLoadingState } from "./localNode.js";
 import { RawCoID, SessionID } from "./ids.js";
+import { Effect, Queue, Stream } from "effect";
 
 export type CoValueKnownState = {
     id: RawCoID;
@@ -60,10 +56,27 @@ export type DoneMessage = {
 
 export type PeerID = string;
 
+export class DisconnectedError extends Error {
+    readonly _tag = "DisconnectedError";
+    constructor(public message: string) {
+        super(message);
+    }
+}
+
+export class PingTimeoutError extends Error {
+    readonly _tag = "PingTimeoutError";
+}
+
+export type IncomingSyncStream = Stream.Stream<
+    SyncMessage,
+    DisconnectedError | PingTimeoutError
+>;
+export type OutgoingSyncQueue = Queue.Enqueue<SyncMessage>;
+
 export interface Peer {
     id: PeerID;
-    incoming: ReadableStream<SyncMessage>;
-    outgoing: WritableStream<SyncMessage>;
+    incoming: IncomingSyncStream;
+    outgoing: OutgoingSyncQueue;
     role: "peer" | "server" | "client";
     delayOnError?: number;
     priority?: number;
@@ -73,8 +86,8 @@ export interface PeerState {
     id: PeerID;
     optimisticKnownStates: { [id: RawCoID]: CoValueKnownState };
     toldKnownState: Set<RawCoID>;
-    incoming: ReadableStream<SyncMessage>;
-    outgoing: WritableStreamDefaultWriter<SyncMessage>;
+    incoming: IncomingSyncStream;
+    outgoing: OutgoingSyncQueue;
     role: "peer" | "server" | "client";
     delayOnError?: number;
     priority?: number;
@@ -127,25 +140,24 @@ export class SyncManager {
         });
     }
 
-    async loadFromPeers(id: RawCoID, excludePeer?: PeerID) {
-        for (const peer of this.peersInPriorityOrder()) {
-            if (peer.id === excludePeer) {
-                continue;
-            }
-            if (peer.role !== "server") {
-                continue;
-            }
+    async loadFromPeers(id: RawCoID, forPeer?: PeerID) {
+        const eligiblePeers = this.peersInPriorityOrder().filter(
+            (peer) => peer.id !== forPeer && peer.role === "server",
+        );
+
+        for (const peer of eligiblePeers) {
             // console.log("loading", id, "from", peer.id);
-            peer.outgoing
-                .write({
+            Effect.runPromise(
+                Queue.offer(peer.outgoing, {
                     action: "load",
                     id: id,
                     header: false,
                     sessions: {},
-                })
-                .catch((e) => {
-                    console.error("Error writing to peer", e);
-                });
+                }),
+            ).catch((e) => {
+                console.error("Error writing to peer", e);
+            });
+
             const coValueEntry = this.local.coValues[id];
             if (coValueEntry?.state !== "loading") {
                 continue;
@@ -297,7 +309,9 @@ export class SyncManager {
                 let lastYield = performance.now();
                 for (const [_i, piece] of newContentPieces.entries()) {
                     // console.log(
-                    //     `${id} -> ${peer.id}: Sending content piece ${i + 1}/${newContentPieces.length} header: ${!!piece.header}`,
+                    //     `${id} -> ${peer.id}: Sending content piece ${i + 1}/${
+                    //         newContentPieces.length
+                    //     } header: ${!!piece.header}`,
                     //     // Object.values(piece.new).map((s) => s.newTransactions)
                     // );
                     await this.trySendToPeer(peer, piece);
@@ -328,7 +342,7 @@ export class SyncManager {
             id: peer.id,
             optimisticKnownStates: {},
             incoming: peer.incoming,
-            outgoing: peer.outgoing.getWriter(),
+            outgoing: peer.outgoing,
             toldKnownState: new Set(),
             role: peer.role,
             delayOnError: peer.delayOnError,
@@ -354,91 +368,55 @@ export class SyncManager {
             void initialSync();
         }
 
-        const readIncoming = async () => {
-            try {
-                for await (const msg of peerState.incoming) {
-                    try {
-                        // await this.handleSyncMessage(msg, peerState);
-                        this.handleSyncMessage(msg, peerState).catch((e) => {
-                            console.error(
-                                new Date(),
-                                `Error reading from peer ${peer.id}, handling msg`,
-                                JSON.stringify(msg, (k, v) =>
-                                    k === "changes" || k === "encryptedChanges"
-                                        ? v.slice(0, 20) + "..."
-                                        : v,
-                                ),
-                                e,
-                            );
-                        });
-                        // await new Promise<void>((resolve) => {
-                        //     setTimeout(resolve, 0);
-                        // });
-                    } catch (e) {
-                        console.error(
-                            new Date(),
-                            `Error reading from peer ${peer.id}, handling msg`,
-                            JSON.stringify(msg, (k, v) =>
-                                k === "changes" || k === "encryptedChanges"
-                                    ? v.slice(0, 20) + "..."
-                                    : v,
+        void Effect.runPromise(
+            peerState.incoming.pipe(
+                Stream.ensuring(
+                    Effect.sync(() => {
+                        console.log("Peer disconnected:", peer.id);
+                        delete this.peers[peer.id];
+                    }),
+                ),
+                Stream.runForEach((msg) =>
+                    Effect.tryPromise({
+                        try: () => this.handleSyncMessage(msg, peerState),
+                        catch: (e) =>
+                            new Error(
+                                `Error reading from peer ${
+                                    peer.id
+                                }, handling msg\n\n${JSON.stringify(
+                                    msg,
+                                    (k, v) =>
+                                        k === "changes" ||
+                                        k === "encryptedChanges"
+                                            ? v.slice(0, 20) + "..."
+                                            : v,
+                                )}`,
+                                { cause: e },
                             ),
-                            e,
-                        );
-                        if (peerState.delayOnError) {
-                            await new Promise<void>((resolve) => {
-                                setTimeout(resolve, peerState.delayOnError);
-                            });
-                        }
-                    }
-                }
-            } catch (e) {
-                console.error(`Error reading from peer ${peer.id}`, e);
-            }
-
-            console.log("Peer disconnected:", peer.id);
-            delete this.peers[peer.id];
-        };
-
-        void readIncoming();
+                    }).pipe(
+                        Effect.timeoutFail({
+                            duration: 10000,
+                            onTimeout: () =>
+                                new Error("Took >10s to process message"),
+                        }),
+                    ),
+                ),
+                Effect.catchAll((e) =>
+                    Effect.logError(
+                        "Error in peer",
+                        peer.id,
+                        e.message,
+                        typeof e.cause === "object" &&
+                            e.cause instanceof Error &&
+                            e.cause.message,
+                    ),
+                ),
+            ),
+        );
     }
 
     trySendToPeer(peer: PeerState, msg: SyncMessage) {
-        if (!this.peers[peer.id]) {
-            // already disconnected, return to drain potential queue
-            return Promise.resolve();
-        }
-
-        return new Promise<void>((resolve) => {
-            const start = Date.now();
-            peer.outgoing
-                .write(msg)
-                .then(() => {
-                    const end = Date.now();
-                    if (end - start > 1000) {
-                        // console.error(
-                        //     new Error(
-                        //         `Writing to peer "${peer.id}" took ${
-                        //             Math.round((Date.now() - start) / 100) / 10
-                        //         }s - this should never happen as write should resolve quickly or error`
-                        //     )
-                        // );
-                    } else {
-                        resolve();
-                    }
-                })
-                .catch((e) => {
-                    console.error(
-                        new Error(
-                            `Error writing to peer ${peer.id}, disconnecting`,
-                            {
-                                cause: e,
-                            },
-                        ),
-                    );
-                    delete this.peers[peer.id];
-                });
-        });
+        return Effect.runPromise(Queue.offer(peer.outgoing, msg));
     }
 
     async handleLoad(msg: LoadMessage, peer: PeerState) {
@@ -447,21 +425,50 @@ export class SyncManager {
 
         if (!entry) {
             // console.log(`Loading ${msg.id} from all peers except ${peer.id}`);
-            this.local
-                .loadCoValueCore(msg.id, {
-                    dontLoadFrom: peer.id,
-                    dontWaitFor: peer.id,
-                })
-                .catch((e) => {
-                    console.error("Error loading coValue in handleLoad", e);
-                });
+
+            // special case: we should be able to solve this much more neatly
+            // with an explicit state machine in the future
+            const eligiblePeers = this.peersInPriorityOrder().filter(
+                (other) => other.id !== peer.id && peer.role === "server",
+            );
+            if (eligiblePeers.length === 0) {
+                if (msg.header || Object.keys(msg.sessions).length > 0) {
+                    this.local.coValues[msg.id] = newLoadingState(
+                        new Set([peer.id]),
+                    );
+                    this.trySendToPeer(peer, {
+                        action: "known",
+                        id: msg.id,
+                        header: false,
+                        sessions: {},
+                    }).catch((e) => {
+                        console.error("Error sending known state back", e);
+                    });
+                }
+                return;
+            } else {
+                this.local
+                    .loadCoValueCore(msg.id, {
+                        dontLoadFrom: peer.id,
+                        dontWaitFor: peer.id,
+                    })
+                    .catch((e) => {
+                        console.error("Error loading coValue in handleLoad", e);
+                    });
+            }
 
             entry = this.local.coValues[msg.id]!;
         }
 
         if (entry.state === "loading") {
+            console.log(
+                "Waiting for loaded",
+                msg.id,
+                "after message from",
+                peer.id,
+            );
             const loaded = await entry.done;
-
+            console.log("Loaded", msg.id, loaded);
             if (loaded === "unavailable") {
                 peer.optimisticKnownStates[msg.id] = knownStateIn(msg);
                 peer.toldKnownState.add(msg.id);
@@ -508,7 +515,7 @@ export class SyncManager {
                 }
             } else {
                 throw new Error(
-                    "Expected coValue entry to be created, missing subscribe?",
+                    `Expected coValue entry for ${msg.id} to be created on known state, missing subscribe?`,
                 );
             }
         }
@@ -549,7 +556,7 @@ export class SyncManager {
 
         if (!entry) {
             throw new Error(
-                "Expected coValue entry to be created, missing subscribe?",
+                `Expected coValue entry for ${msg.id} to be created on new content, missing subscribe?`,
             );
         }
 

packages/cojson/src/tests/account.test.ts

@@ -3,6 +3,7 @@ import { newRandomSessionID } from "../coValueCore.js";
 import { LocalNode } from "../localNode.js";
 import { connectedPeers } from "../streamUtils.js";
 import { WasmCrypto } from "../crypto/WasmCrypto.js";
+import { Effect } from "effect";
 
 const Crypto = await WasmCrypto.create();
 
@@ -52,11 +53,13 @@ test("Can create account with one node, and then load it on another", async () =
     map.set("foo", "bar", "private");
     expect(map.get("foo")).toEqual("bar");
 
-    const [node1asPeer, node2asPeer] = connectedPeers("node1", "node2", {
+    const [node1asPeer, node2asPeer] = await Effect.runPromise(connectedPeers("node1", "node2", {
         trace: true,
         peer1role: "server",
         peer2role: "client",
-    });
+    }));
+
+    console.log("After connected peers")
 
     node.syncManager.addPeer(node2asPeer);
 

packages/jazz-browser-media-images/CHANGELOG.md

@@ -1,5 +1,51 @@
 # jazz-browser-media-images
 
+## 0.7.17
+
+### Patch Changes
+
+-   jazz-browser@0.7.17
+-   jazz-tools@0.7.17
+
+## 0.7.16
+
+### Patch Changes
+
+-   Updated dependencies
+    -   jazz-tools@0.7.16
+    -   jazz-browser@0.7.16
+
+## 0.7.14
+
+### Patch Changes
+
+-   Updated dependencies
+    -   jazz-tools@0.7.14
+    -   jazz-browser@0.7.14
+
+## 0.7.13
+
+### Patch Changes
+
+-   Updated dependencies
+    -   jazz-tools@0.7.13
+    -   jazz-browser@0.7.13
+
+## 0.7.12
+
+### Patch Changes
+
+-   Updated dependencies
+    -   jazz-tools@0.7.12
+    -   jazz-browser@0.7.12
+
+## 0.7.11
+
+### Patch Changes
+
+-   jazz-browser@0.7.11
+-   jazz-tools@0.7.11
+
 ## 0.7.10
 
 ### Patch Changes

packages/jazz-browser-media-images/package.json

@@ -1,6 +1,6 @@
 {
   "name": "jazz-browser-media-images",
-  "version": "0.7.10",
+  "version": "0.7.17",
   "type": "module",
   "main": "dist/index.js",
   "types": "src/index.ts",

packages/jazz-browser/CHANGELOG.md

@@ -1,5 +1,55 @@
 # jazz-browser
 
+## 0.7.17
+
+### Patch Changes
+
+-   Updated dependencies
+    -   cojson@0.7.17
+    -   cojson-storage-indexeddb@0.7.17
+    -   cojson-transport-ws@0.7.17
+    -   jazz-tools@0.7.17
+
+## 0.7.16
+
+### Patch Changes
+
+-   Updated dependencies
+    -   jazz-tools@0.7.16
+
+## 0.7.14
+
+### Patch Changes
+
+-   Updated dependencies
+    -   cojson@0.7.14
+    -   jazz-tools@0.7.14
+    -   cojson-storage-indexeddb@0.7.14
+    -   cojson-transport-ws@0.7.14
+
+## 0.7.13
+
+### Patch Changes
+
+-   Updated dependencies
+    -   jazz-tools@0.7.13
+
+## 0.7.12
+
+### Patch Changes
+
+-   Updated dependencies
+    -   jazz-tools@0.7.12
+
+## 0.7.11
+
+### Patch Changes
+
+-   Updated dependencies
+    -   cojson@0.7.11
+    -   cojson-storage-indexeddb@0.7.11
+    -   jazz-tools@0.7.11
+
 ## 0.7.10
 
 ### Patch Changes

packages/jazz-browser/package.json

@@ -1,6 +1,6 @@
 {
   "name": "jazz-browser",
-  "version": "0.7.10",
+  "version": "0.7.17",
   "type": "module",
   "main": "dist/index.js",
   "types": "src/index.ts",
@@ -9,8 +9,8 @@
     "@scure/bip39": "^1.3.0",
     "cojson": "workspace:*",
     "cojson-storage-indexeddb": "workspace:*",
+    "cojson-transport-ws": "workspace:*",
     "effect": "^3.1.5",
-    "isomorphic-streams": "https://github.com/sgwilym/isomorphic-streams.git#aa9394781bfc92f8d7c981be7daf8af4b4cd4fae",
     "jazz-tools": "workspace:*",
     "typescript": "^5.1.6"
   },

packages/jazz-browser/src/OPFSFilesystem.ts

@@ -309,7 +309,7 @@ const opfsWorkerJSSrc = `
             postMessage({requestId: event.data.requestId, data: buffer, result: "done"});
         } else if (event.data.type === "close") {
             const handle = handlesByRequest.get(event.data.handle);
-            console.log("Closing handle", filenamesForHandles.get(handle), event.data.handle, handle);
+            // console.log("Closing handle", filenamesForHandles.get(handle), event.data.handle, handle);
             handle.flush();
             handle.close();
             handlesByRequest.delete(handle);

packages/jazz-browser/src/index.ts

@@ -1,11 +1,8 @@
-import { ReadableStream, WritableStream } from "isomorphic-streams";
 import {
     CoValue,
     ID,
-    Peer,
     AgentID,
     SessionID,
-    SyncMessage,
     cojsonInternals,
     InviteSecret,
     Account,
@@ -13,10 +10,15 @@ import {
     WasmCrypto,
     CryptoProvider,
 } from "jazz-tools";
-import { AccountID, LSMStorage } from "cojson";
+import {
+    AccountID,
+    LSMStorage,
+} from "cojson";
 import { AuthProvider } from "./auth/auth.js";
 import { OPFSFilesystem } from "./OPFSFilesystem.js";
 import { IDBStorage } from "cojson-storage-indexeddb";
+import { Effect, Queue } from "effect";
+import { createWebSocketPeer } from "cojson-transport-ws";
 export * from "./auth/auth.js";
 
 /** @category Context Creation */
@@ -29,7 +31,7 @@ export type BrowserContext<Acc extends Account> = {
 /** @category Context Creation */
 export async function createJazzBrowserContext<Acc extends Account>({
     auth,
-    peer,
+    peer: peerAddr,
     reconnectionTimeout: initialReconnectionTimeout = 500,
     storage = "indexedDB",
     crypto: customCrypto,
@@ -43,7 +45,13 @@ export async function createJazzBrowserContext<Acc extends Account>({
     const crypto = customCrypto || (await WasmCrypto.create());
     let sessionDone: () => void;
 
-    const firstWsPeer = createWebSocketPeer(peer);
+    const firstWsPeer = await Effect.runPromise(
+        createWebSocketPeer({
+            websocket: new WebSocket(peerAddr),
+            id: peerAddr + "@" + new Date().toISOString(),
+            role: "server",
+        }),
+    );
     let shouldTryToReconnect = true;
 
     let currentReconnectionTimeout = initialReconnectionTimeout;
@@ -77,7 +85,7 @@ export async function createJazzBrowserContext<Acc extends Account>({
         while (shouldTryToReconnect) {
             if (
                 Object.keys(me._raw.core.node.syncManager.peers).some(
-                    (peerId) => peerId.includes(peer),
+                    (peerId) => peerId.includes(peerAddr),
                 )
             ) {
                 // TODO: this might drain battery, use listeners instead
@@ -107,7 +115,13 @@ export async function createJazzBrowserContext<Acc extends Account>({
                 });
 
                 me._raw.core.node.syncManager.addPeer(
-                    createWebSocketPeer(peer),
+                    await Effect.runPromise(
+                        createWebSocketPeer({
+                            websocket: new WebSocket(peerAddr),
+                            id: peerAddr + "@" + new Date().toISOString(),
+                            role: "server",
+                        }),
+                    ),
                 );
             }
         }
@@ -124,9 +138,7 @@ export async function createJazzBrowserContext<Acc extends Account>({
             for (const peer of Object.values(
                 me._raw.core.node.syncManager.peers,
             )) {
-                peer.outgoing
-                    .close()
-                    .catch((e) => console.error("Error while closing peer", e));
+                void Effect.runPromise(Queue.shutdown(peer.outgoing));
             }
             sessionDone?.();
         },
@@ -207,140 +219,6 @@ export function getSessionHandleFor(
     };
 }
 
-function websocketReadableStream<T>(ws: WebSocket) {
-    ws.binaryType = "arraybuffer";
-
-    return new ReadableStream<T>({
-        start(controller) {
-            let pingTimeout: ReturnType<typeof setTimeout> | undefined;
-
-            ws.onmessage = (event) => {
-                const msg = JSON.parse(event.data);
-
-                if (pingTimeout) {
-                    clearTimeout(pingTimeout);
-                }
-
-                pingTimeout = setTimeout(() => {
-                    console.debug("Ping timeout");
-                    try {
-                        controller.close();
-                        ws.close();
-                    } catch (e) {
-                        console.error(
-                            "Error while trying to close ws on ping timeout",
-                            e,
-                        );
-                    }
-                }, 2500);
-
-                if (msg.type === "ping") {
-                    // eslint-disable-next-line @typescript-eslint/no-explicit-any
-                    (window as any).jazzPings = (window as any).jazzPings || [];
-                    // eslint-disable-next-line @typescript-eslint/no-explicit-any
-                    (window as any).jazzPings.push({
-                        received: Date.now(),
-                        sent: msg.time,
-                        dc: msg.dc,
-                    });
-                    return;
-                }
-                controller.enqueue(msg);
-            };
-            const closeListener = () => {
-                controller.close();
-                clearTimeout(pingTimeout);
-            };
-            ws.addEventListener("close", closeListener);
-            ws.addEventListener("error", () => {
-                controller.error(new Error("The WebSocket errored!"));
-                ws.removeEventListener("close", closeListener);
-            });
-        },
-
-        cancel() {
-            ws.close();
-        },
-    });
-}
-
-export function createWebSocketPeer(syncAddress: string): Peer {
-    const ws = new WebSocket(syncAddress);
-
-    const incoming = websocketReadableStream<SyncMessage>(ws);
-    const outgoing = websocketWritableStream<SyncMessage>(ws);
-
-    return {
-        id: syncAddress + "@" + new Date().toISOString(),
-        incoming,
-        outgoing,
-        role: "server",
-    };
-}
-
-function websocketWritableStream<T>(ws: WebSocket) {
-    const initialQueue = [] as T[];
-    let isOpen = false;
-
-    return new WritableStream<T>({
-        start(controller) {
-            ws.addEventListener("error", (event) => {
-                controller.error(
-                    new Error("The WebSocket errored!" + JSON.stringify(event)),
-                );
-            });
-            ws.addEventListener("close", () => {
-                controller.error(
-                    new Error("The server closed the connection unexpectedly!"),
-                );
-            });
-            ws.addEventListener("open", () => {
-                for (const item of initialQueue) {
-                    ws.send(JSON.stringify(item));
-                }
-                isOpen = true;
-            });
-        },
-
-        async write(chunk) {
-            if (isOpen) {
-                ws.send(JSON.stringify(chunk));
-                // Return immediately, since the web socket gives us no easy way to tell
-                // when the write completes.
-            } else {
-                initialQueue.push(chunk);
-            }
-        },
-
-        close() {
-            return closeWS(1000);
-        },
-
-        abort(reason) {
-            return closeWS(4000, reason && reason.message);
-        },
-    });
-
-    function closeWS(code: number, reasonString?: string) {
-        return new Promise<void>((resolve, reject) => {
-            ws.addEventListener(
-                "close",
-                (e) => {
-                    if (e.wasClean) {
-                        resolve();
-                    } else {
-                        reject(
-                            new Error("The connection was not closed cleanly"),
-                        );
-                    }
-                },
-                { once: true },
-            );
-            ws.close(code, reasonString);
-        });
-    }
-}
-
 /** @category Invite Links */
 export function createInviteLink<C extends CoValue>(
     value: C,

packages/jazz-nodejs/CHANGELOG.md

@@ -1,5 +1,53 @@
 # jazz-autosub
 
+## 0.7.17
+
+### Patch Changes
+
+-   Updated dependencies
+    -   cojson@0.7.17
+    -   cojson-transport-ws@0.7.17
+    -   jazz-tools@0.7.17
+
+## 0.7.16
+
+### Patch Changes
+
+-   Updated dependencies
+    -   jazz-tools@0.7.16
+
+## 0.7.14
+
+### Patch Changes
+
+-   Updated dependencies
+    -   cojson@0.7.14
+    -   jazz-tools@0.7.14
+    -   cojson-transport-ws@0.7.14
+
+## 0.7.13
+
+### Patch Changes
+
+-   Updated dependencies
+    -   jazz-tools@0.7.13
+
+## 0.7.12
+
+### Patch Changes
+
+-   Updated dependencies
+    -   jazz-tools@0.7.12
+
+## 0.7.11
+
+### Patch Changes
+
+-   Updated dependencies
+    -   cojson@0.7.11
+    -   cojson-transport-nodejs-ws@0.7.11
+    -   jazz-tools@0.7.11
+
 ## 0.7.10
 
 ### Patch Changes

packages/jazz-nodejs/package.json

@@ -5,10 +5,11 @@
     "types": "src/index.ts",
     "type": "module",
     "license": "MIT",
-    "version": "0.7.10",
+    "version": "0.7.17",
     "dependencies": {
         "cojson": "workspace:*",
-        "cojson-transport-nodejs-ws": "workspace:*",
+        "cojson-transport-ws": "workspace:*",
+        "effect": "^3.1.5",
         "jazz-tools": "workspace:*",
         "ws": "^8.14.2"
     },

packages/jazz-nodejs/src/index.ts

@@ -1,11 +1,8 @@
-import {
-    websocketReadableStream,
-    websocketWritableStream,
-} from "cojson-transport-nodejs-ws";
-import { WebSocket } from "ws";
-
 import { AgentSecret, Peer, SessionID, WasmCrypto } from "cojson";
+import { createWebSocketPeer } from "cojson-transport-ws";
 import { Account, CoValueClass, ID } from "jazz-tools";
+import { Effect } from "effect";
+import { WebSocket } from "ws";
 
 /** @category Context Creation */
 export async function startWorker<Acc extends Account>({
@@ -21,14 +18,13 @@ export async function startWorker<Acc extends Account>({
     syncServer?: string;
     accountSchema?: CoValueClass<Acc> & typeof Account;
 }): Promise<{ worker: Acc }> {
-    const ws = new WebSocket(peer);
-
-    const wsPeer: Peer = {
-        id: "upstream",
-        role: "server",
-        incoming: websocketReadableStream(ws),
-        outgoing: websocketWritableStream(ws),
-    };
+    const wsPeer: Peer = await Effect.runPromise(
+        createWebSocketPeer({
+            id: "upstream",
+            websocket: new WebSocket(peer),
+            role: "server",
+        }),
+    );
 
     if (!accountID) {
         throw new Error("No accountID provided");
@@ -52,17 +48,17 @@ export async function startWorker<Acc extends Account>({
         crypto: await WasmCrypto.create(),
     });
 
-    setInterval(() => {
+    setInterval(async () => {
         if (!worker._raw.core.node.syncManager.peers["upstream"]) {
             console.log(new Date(), "Reconnecting to upstream " + peer);
-            const ws = new WebSocket(peer);
 
-            const wsPeer: Peer = {
-                id: "upstream",
-                role: "server",
-                incoming: websocketReadableStream(ws),
-                outgoing: websocketWritableStream(ws),
-            };
+            const wsPeer: Peer = await Effect.runPromise(
+                createWebSocketPeer({
+                    id: "upstream",
+                    websocket: new WebSocket(peer),
+                    role: "server",
+                }),
+            );
 
             worker._raw.core.node.syncManager.addPeer(wsPeer);
         }

packages/jazz-react/CHANGELOG.md

@@ -1,5 +1,62 @@
 # jazz-react
 
+## 0.7.17
+
+### Patch Changes
+
+-   Updated dependencies
+    -   cojson@0.7.17
+    -   jazz-browser@0.7.17
+    -   jazz-tools@0.7.17
+
+## 0.7.16
+
+### Patch Changes
+
+-   Updated dependencies
+    -   jazz-tools@0.7.16
+    -   jazz-browser@0.7.16
+
+## 0.7.15
+
+### Patch Changes
+
+-   Provide current res in ProgressiveImg
+
+## 0.7.14
+
+### Patch Changes
+
+-   Updated dependencies
+    -   cojson@0.7.14
+    -   jazz-tools@0.7.14
+    -   jazz-browser@0.7.14
+
+## 0.7.13
+
+### Patch Changes
+
+-   Updated dependencies
+    -   jazz-tools@0.7.13
+    -   jazz-browser@0.7.13
+
+## 0.7.12
+
+### Patch Changes
+
+-   Updated dependencies
+    -   jazz-tools@0.7.12
+    -   jazz-browser@0.7.12
+
+## 0.7.11
+
+### Patch Changes
+
+-   Updated dependencies
+    -   cojson@0.7.11
+    -   jazz-browser@0.7.11
+    -   jazz-tools@0.7.11
+
 ## 0.7.10
 
 ### Patch Changes

packages/jazz-react/package.json

@@ -1,6 +1,6 @@
 {
   "name": "jazz-react",
-  "version": "0.7.10",
+  "version": "0.7.17",
   "type": "module",
   "main": "dist/index.js",
   "types": "src/index.ts",

packages/jazz-react/src/auth/DemoAuth.tsx

@@ -104,7 +104,7 @@ const DemoAuthBasicUI = ({
     signUp: (username: string) => void;
 }) => {
     const [username, setUsername] = useState<string>("");
-    const darkMode = window.matchMedia("(prefers-color-scheme: dark)").matches;
+    const darkMode = typeof window !== 'undefined' ? window.matchMedia("(prefers-color-scheme: dark)").matches : false;
 
     return (
         <div

packages/jazz-react/src/media.tsx

@@ -9,7 +9,10 @@ export function useProgressiveImg({
     image: ImageDefinition | null | undefined;
     maxWidth?: number;
 }) {
-    const [src, setSrc] = useState<string | undefined>(undefined);
+    const [current, setCurrent] = useState<
+        | { src?: string; res?: `${number}x${number}` | "placeholder" }
+        | undefined
+    >(undefined);
 
     useEffect(() => {
         let lastHighestRes: string | undefined;
@@ -22,21 +25,28 @@ export function useProgressiveImg({
                     const blob = highestRes.stream.toBlob();
                     if (blob) {
                         const blobURI = URL.createObjectURL(blob);
-                        setSrc(blobURI);
+                        setCurrent({ src: blobURI, res: highestRes.res });
                         return () => {
                             setTimeout(() => URL.revokeObjectURL(blobURI), 200);
                         };
                     }
                 }
             } else {
-                setSrc(update?.placeholderDataURL);
+                setCurrent({
+                    src: update?.placeholderDataURL,
+                    res: "placeholder",
+                });
             }
         });
 
         return unsub;
     }, [image?.id, maxWidth]);
 
-    return { src, originalSize: image?.originalSize };
+    return {
+        src: current?.src,
+        res: current?.res,
+        originalSize: image?.originalSize,
+    };
 }
 
 /** @category Media */
@@ -47,6 +57,7 @@ export function ProgressiveImg({
 }: {
     children: (result: {
         src: string | undefined;
+        res: `${number}x${number}` | "placeholder" | undefined;
         originalSize: readonly [number, number] | undefined;
     }) => React.ReactNode;
     image: ImageDefinition | null | undefined;

packages/jazz-run/CHANGELOG.md

@@ -1,5 +1,53 @@
 # jazz-autosub
 
+## 0.7.17
+
+### Patch Changes
+
+-   Updated dependencies
+    -   cojson@0.7.17
+    -   cojson-transport-ws@0.7.17
+    -   jazz-tools@0.7.17
+
+## 0.7.16
+
+### Patch Changes
+
+-   Updated dependencies
+    -   jazz-tools@0.7.16
+
+## 0.7.14
+
+### Patch Changes
+
+-   Updated dependencies
+    -   cojson@0.7.14
+    -   jazz-tools@0.7.14
+    -   cojson-transport-ws@0.7.14
+
+## 0.7.13
+
+### Patch Changes
+
+-   Updated dependencies
+    -   jazz-tools@0.7.13
+
+## 0.7.12
+
+### Patch Changes
+
+-   Updated dependencies
+    -   jazz-tools@0.7.12
+
+## 0.7.11
+
+### Patch Changes
+
+-   Updated dependencies
+    -   cojson@0.7.11
+    -   cojson-transport-nodejs-ws@0.7.11
+    -   jazz-tools@0.7.11
+
 ## 0.7.10
 
 ### Patch Changes

packages/jazz-run/package.json

@@ -3,7 +3,7 @@
     "bin": "./dist/index.js",
     "type": "module",
     "license": "MIT",
-    "version": "0.7.10",
+    "version": "0.7.17",
     "scripts": {
         "lint": "eslint . --ext ts,tsx",
         "format": "prettier --write './src/**/*.{ts,tsx}'",
@@ -15,7 +15,7 @@
         "@effect/platform-node": "^0.49.2",
         "@effect/schema": "^0.66.16",
         "cojson": "workspace:*",
-        "cojson-transport-nodejs-ws": "workspace:*",
+        "cojson-transport-ws": "workspace:*",
         "effect": "^3.1.5",
         "fast-check": "^3.17.2",
         "jazz-tools": "workspace:*",

packages/jazz-run/src/index.ts

@@ -2,11 +2,8 @@
 import { Command, Options } from "@effect/cli";
 import { NodeContext, NodeRuntime } from "@effect/platform-node";
 import { Console, Effect } from "effect";
-import {
-    websocketReadableStream,
-    websocketWritableStream,
-} from "cojson-transport-nodejs-ws";
-import { WebSocket } from "ws";
+import { createWebSocketPeer } from "cojson-transport-ws";
+import { WebSocket } from "ws"
 import {
     Account,
     WasmCrypto,
@@ -24,23 +21,20 @@ const peer = Options.text("peer")
 const accountCreate = Command.make(
     "create",
     { name, peer },
-    ({ name, peer }) => {
+    ({ name, peer: peerAddr }) => {
         return Effect.gen(function* () {
-            const ws = new WebSocket(peer);
-
             const crypto = yield* Effect.promise(() => WasmCrypto.create());
 
+            const peer = yield* createWebSocketPeer({
+                id: "upstream",
+                websocket: new WebSocket(peerAddr),
+                role: "server",
+            });
+
             const account: Account = yield* Effect.promise(async () =>
                 Account.create({
                     creationProps: { name },
-                    peersToLoadFrom: [
-                        {
-                            id: "upstream",
-                            role: "server",
-                            incoming: websocketReadableStream(ws),
-                            outgoing: websocketWritableStream(ws),
-                        },
-                    ],
+                    peersToLoadFrom: [peer],
                     crypto,
                 }),
             );
@@ -59,7 +53,11 @@ const accountCreate = Command.make(
                 ),
             );
 
-            const ws2 = new WebSocket(peer);
+            const peer2 = yield* createWebSocketPeer({
+                id: "upstream2",
+                websocket: new WebSocket(peerAddr),
+                role: "server",
+            });
 
             yield* Effect.promise(async () =>
                 Account.become({
@@ -68,14 +66,7 @@ const accountCreate = Command.make(
                     sessionID: cojsonInternals.newRandomSessionID(
                         account.id as unknown as AccountID,
                     ),
-                    peersToLoadFrom: [
-                        {
-                            id: "upstream",
-                            role: "server",
-                            incoming: websocketReadableStream(ws2),
-                            outgoing: websocketWritableStream(ws2),
-                        },
-                    ],
+                    peersToLoadFrom: [peer2],
                     crypto,
                 }),
             );

packages/jazz-tools/CHANGELOG.md

@@ -1,5 +1,46 @@
 # jazz-autosub
 
+## 0.7.17
+
+### Patch Changes
+
+-   Updated dependencies
+    -   cojson@0.7.17
+
+## 0.7.16
+
+### Patch Changes
+
+-   Fix: allow null in encoded fields
+
+## 0.7.14
+
+### Patch Changes
+
+-   Use Effect Queues and Streams instead of custom queue implementation
+-   Updated dependencies
+    -   cojson@0.7.14
+
+## 0.7.13
+
+### Patch Changes
+
+-   Fix CoList.toJSON()
+
+## 0.7.12
+
+### Patch Changes
+
+-   Fix: toJSON infinitely recurses on circular CoValue structures
+
+## 0.7.11
+
+### Patch Changes
+
+-   Updated dependencies
+    -   cojson@0.7.11
+    -   cojson-transport-nodejs-ws@0.7.11
+
 ## 0.7.10
 
 ### Patch Changes

packages/jazz-tools/package.json

@@ -5,11 +5,10 @@
     "types": "./src/index.ts",
     "type": "module",
     "license": "MIT",
-    "version": "0.7.10",
+    "version": "0.7.17",
     "dependencies": {
         "@effect/schema": "^0.66.16",
         "cojson": "workspace:*",
-        "cojson-transport-nodejs-ws": "workspace:*",
         "effect": "^3.1.5",
         "fast-check": "^3.17.2"
     },

packages/jazz-tools/src/coValues/account.ts

@@ -12,7 +12,7 @@ import type {
     SessionID,
 } from "cojson";
 import { Context, Effect, Stream } from "effect";
-import type {
+import {
     CoMap,
     CoValue,
     CoValueClass,
@@ -61,9 +61,11 @@ export class Account extends CoValueBase implements CoValue {
                 ref: () => Profile,
                 optional: false,
             } satisfies RefEncoded<Profile>,
-            root: "json" satisfies Schema,
-            // eslint-disable-next-line @typescript-eslint/no-explicit-any
-        } as any;
+            root: {
+                ref: () => CoMap,
+                optional: true,
+            } satisfies RefEncoded<CoMap>,
+        };
     }
 
     get _owner(): Account {
@@ -214,7 +216,7 @@ export class Account extends CoValueBase implements CoValue {
         return this.fromNode(node) as A;
     }
 
-    static createAs<A extends Account>(
+    static async createAs<A extends Account>(
         this: CoValueClass<A> & typeof Account,
         as: Account,
         options: {
@@ -222,11 +224,11 @@ export class Account extends CoValueBase implements CoValue {
         },
     ) {
         // TODO: is there a cleaner way to do this?
-        const connectedPeers = cojsonInternals.connectedPeers(
+        const connectedPeers = await Effect.runPromise(cojsonInternals.connectedPeers(
             "creatingAccount",
             "createdAccount",
             { peer1role: "server", peer2role: "client" },
-        );
+        ));
 
         as._raw.core.node.syncManager.addPeer(connectedPeers[1]);
 

packages/jazz-tools/src/coValues/coList.ts

@@ -300,7 +300,8 @@ export class CoList<Item = any> extends Array<Item> implements CoValue {
         return deleted;
     }
 
-    toJSON() {
+    // eslint-disable-next-line @typescript-eslint/no-explicit-any
+    toJSON(_key?: string, seenAbove?: ID<CoValue>[]): any[] {
         const itemDescriptor = this._schema[ItemsSym] as Schema;
         if (itemDescriptor === "json") {
             return this._raw.asArray();
@@ -309,7 +310,14 @@ export class CoList<Item = any> extends Array<Item> implements CoValue {
                 .asArray()
                 .map((e) => encodeSync(itemDescriptor.encoded)(e));
         } else if (isRefEncoded(itemDescriptor)) {
-            return this.map((item) => (item as unknown as CoValue)?.toJSON());
+            return this.map((item, idx) =>
+                seenAbove?.includes((item as CoValue)?.id)
+                    ? { _circular: (item as CoValue).id }
+                    : (item as unknown as CoValue)?.toJSON(idx + "", [
+                          ...(seenAbove || []),
+                          this.id,
+                      ]),
+            );
         } else {
             return [];
         }

packages/jazz-tools/src/coValues/coMap.ts

@@ -254,7 +254,8 @@ export class CoMap extends CoValueBase implements CoValue {
         return instance;
     }
 
-    toJSON() {
+    // eslint-disable-next-line @typescript-eslint/no-explicit-any
+    toJSON(_key?: string, seenAbove?: ID<CoValue>[]): any[] {
         const jsonedFields = this._raw.keys().map((key) => {
             const tKey = key as CoKeys<this>;
             const descriptor = (this._schema[tKey] ||
@@ -264,7 +265,15 @@ export class CoMap extends CoValueBase implements CoValue {
                 return [key, this._raw.get(key)];
             } else if (isRefEncoded(descriptor)) {
                 // eslint-disable-next-line @typescript-eslint/no-explicit-any
-                const jsonedRef = (this as any)[tKey]?.toJSON();
+                if (seenAbove?.includes((this as any)[tKey]?.id)) {
+                    // eslint-disable-next-line @typescript-eslint/no-explicit-any
+                    return [key, { _circular: (this as any)[tKey]?.id }];
+                }
+                // eslint-disable-next-line @typescript-eslint/no-explicit-any
+                const jsonedRef = (this as any)[tKey]?.toJSON(tKey, [
+                    ...(seenAbove || []),
+                    this.id,
+                ]);
                 return [key, jsonedRef];
             } else {
                 return [key, undefined];
@@ -530,6 +539,7 @@ const CoMapProxyHandler: ProxyHandler<CoMap> = {
         if (
             (typeof key === "string" || ItemsSym) &&
             typeof value === "object" &&
+            value !== null &&
             SchemaInit in value
         ) {
             (target.constructor as typeof CoMap)._schema ||= {};

packages/jazz-tools/src/coValues/interfaces.ts

@@ -40,7 +40,7 @@ export interface CoValue {
     readonly _loadedAs: Account;
     /** @category Stringifying & Inspection */
     // eslint-disable-next-line @typescript-eslint/no-explicit-any
-    toJSON(): any[] | object;
+    toJSON(key?: string, seenAbove?: ID<CoValue>[]): any[] | object | string;
     /** @category Stringifying & Inspection */
     // eslint-disable-next-line @typescript-eslint/no-explicit-any
     [inspect](): any;
@@ -108,7 +108,7 @@ export class CoValueBase implements CoValue {
     }
 
     // eslint-disable-next-line @typescript-eslint/no-explicit-any
-    toJSON(): object | any[] {
+    toJSON(): object | any[] | string {
         return {
             id: this.id,
             type: this._type,

packages/jazz-tools/src/tests/coList.test.ts

@@ -157,11 +157,11 @@ describe("CoList resolution", async () => {
     test("Loading and availability", async () => {
         const { me, list } = await initNodeAndList();
 
-        const [initialAsPeer, secondPeer] = connectedPeers(
+        const [initialAsPeer, secondPeer] = await Effect.runPromise(connectedPeers(
             "initial",
             "second",
             { peer1role: "server", peer2role: "client" },
-        );
+        ));
         if (!isControlledAccount(me)) {
             throw "me is not a controlled account";
         }
@@ -216,11 +216,11 @@ describe("CoList resolution", async () => {
     test("Subscription & auto-resolution", async () => {
         const { me, list } = await initNodeAndList();
 
-        const [initialAsPeer, secondPeer] = connectedPeers(
+        const [initialAsPeer, secondPeer] = await Effect.runPromise(connectedPeers(
             "initial",
             "second",
             { peer1role: "server", peer2role: "client" },
-        );
+        ));
         if (!isControlledAccount(me)) {
             throw "me is not a controlled account";
         }

packages/jazz-tools/src/tests/coMap.test.ts

@@ -10,6 +10,7 @@ import {
     WasmCrypto,
     isControlledAccount,
 } from "../index.js";
+import { Schema } from "@effect/schema";
 
 const Crypto = await WasmCrypto.create();
 
@@ -24,6 +25,7 @@ describe("Simple CoMap operations", async () => {
         _height = co.number;
         birthday = co.encoded(Encoders.Date);
         name? = co.string;
+        nullable = co.encoded(Schema.NullOr(Schema.String));
 
         get roughColor() {
             return this.color + "ish";
@@ -39,6 +41,7 @@ describe("Simple CoMap operations", async () => {
             color: "red",
             _height: 10,
             birthday: birthday,
+            nullable: null,
         },
         { owner: me },
     );
@@ -49,7 +52,12 @@ describe("Simple CoMap operations", async () => {
         expect(map._height).toEqual(10);
         expect(map.birthday).toEqual(birthday);
         expect(map._raw.get("birthday")).toEqual(birthday.toISOString());
-        expect(Object.keys(map)).toEqual(["color", "_height", "birthday"]);
+        expect(Object.keys(map)).toEqual([
+            "color",
+            "_height",
+            "birthday",
+            "nullable",
+        ]);
     });
 
     test("Construction with too many things provided", () => {
@@ -84,6 +92,9 @@ describe("Simple CoMap operations", async () => {
             expect(map._height).toEqual(20);
             expect(map._raw.get("_height")).toEqual(20);
 
+            map.nullable = "not null";
+            map.nullable = null;
+
             map.name = "Secret name";
             expect(map.name).toEqual("Secret name");
             map.name = undefined;
@@ -253,10 +264,11 @@ describe("CoMap resolution", async () => {
 
     test("Loading and availability", async () => {
         const { me, map } = await initNodeAndMap();
-        const [initialAsPeer, secondPeer] = connectedPeers(
-            "initial",
-            "second",
-            { peer1role: "server", peer2role: "client" },
+        const [initialAsPeer, secondPeer] = await Effect.runPromise(
+            connectedPeers("initial", "second", {
+                peer1role: "server",
+                peer2role: "client",
+            }),
         );
         if (!isControlledAccount(me)) {
             throw "me is not a controlled account";
@@ -323,10 +335,11 @@ describe("CoMap resolution", async () => {
     test("Subscription & auto-resolution", async () => {
         const { me, map } = await initNodeAndMap();
 
-        const [initialAsPeer, secondAsPeer] = connectedPeers(
-            "initial",
-            "second",
-            { peer1role: "server", peer2role: "client" },
+        const [initialAsPeer, secondAsPeer] = await Effect.runPromise(
+            connectedPeers("initial", "second", {
+                peer1role: "server",
+                peer2role: "client",
+            }),
         );
         if (!isControlledAccount(me)) {
             throw "me is not a controlled account";

packages/jazz-tools/src/tests/coStream.test.ts

@@ -83,10 +83,11 @@ describe("CoStream resolution", async () => {
 
     test("Loading and availability", async () => {
         const { me, stream } = await initNodeAndStream();
-        const [initialAsPeer, secondPeer] = connectedPeers(
-            "initial",
-            "second",
-            { peer1role: "server", peer2role: "client" },
+        const [initialAsPeer, secondPeer] = await Effect.runPromise(
+            connectedPeers("initial", "second", {
+                peer1role: "server",
+                peer2role: "client",
+            }),
         );
         if (!isControlledAccount(me)) {
             throw "me is not a controlled account";
@@ -175,10 +176,11 @@ describe("CoStream resolution", async () => {
     test("Subscription & auto-resolution", async () => {
         const { me, stream } = await initNodeAndStream();
 
-        const [initialAsPeer, secondAsPeer] = connectedPeers(
-            "initial",
-            "second",
-            { peer1role: "server", peer2role: "client" },
+        const [initialAsPeer, secondAsPeer] = await Effect.runPromise(
+            connectedPeers("initial", "second", {
+                peer1role: "server",
+                peer2role: "client",
+            }),
         );
         me._raw.core.node.syncManager.addPeer(secondAsPeer);
         if (!isControlledAccount(me)) {
@@ -325,10 +327,11 @@ describe("BinaryCoStream loading & Subscription", async () => {
 
     test("Loading and availability", async () => {
         const { me, stream } = await initNodeAndStream();
-        const [initialAsPeer, secondAsPeer] = connectedPeers(
-            "initial",
-            "second",
-            { peer1role: "server", peer2role: "client" },
+        const [initialAsPeer, secondAsPeer] = await Effect.runPromise(
+            connectedPeers("initial", "second", {
+                peer1role: "server",
+                peer2role: "client",
+            }),
         );
         if (!isControlledAccount(me)) {
             throw "me is not a controlled account";
@@ -357,30 +360,32 @@ describe("BinaryCoStream loading & Subscription", async () => {
     });
 
     test("Subscription", async () => {
-        const { me } = await initNodeAndStream();
-
-        const stream = BinaryCoStream.create({ owner: me });
-
-        const [initialAsPeer, secondAsPeer] = connectedPeers(
-            "initial",
-            "second",
-            { peer1role: "server", peer2role: "client" },
-        );
-        me._raw.core.node.syncManager.addPeer(secondAsPeer);
-        if (!isControlledAccount(me)) {
-            throw "me is not a controlled account";
-        }
-        const meOnSecondPeer = await Account.become({
-            accountID: me.id,
-            accountSecret: me._raw.agentSecret,
-            peersToLoadFrom: [initialAsPeer],
-            // eslint-disable-next-line @typescript-eslint/no-explicit-any
-            sessionID: newRandomSessionID(me.id as any),
-            crypto: Crypto,
-        });
-
         await Effect.runPromise(
             Effect.gen(function* ($) {
+                const { me } = yield* Effect.promise(() => initNodeAndStream());
+
+                const stream = BinaryCoStream.create({ owner: me });
+
+                const [initialAsPeer, secondAsPeer] = yield* connectedPeers(
+                    "initial",
+                    "second",
+                    { peer1role: "server", peer2role: "client" },
+                );
+                me._raw.core.node.syncManager.addPeer(secondAsPeer);
+                if (!isControlledAccount(me)) {
+                    throw "me is not a controlled account";
+                }
+                const meOnSecondPeer = yield* Effect.promise(() =>
+                    Account.become({
+                        accountID: me.id,
+                        accountSecret: me._raw.agentSecret,
+                        peersToLoadFrom: [initialAsPeer],
+                        // eslint-disable-next-line @typescript-eslint/no-explicit-any
+                        sessionID: newRandomSessionID(me.id as any),
+                        crypto: Crypto,
+                    }),
+                );
+
                 const queue = yield* $(Queue.unbounded<BinaryCoStream>());
 
                 BinaryCoStream.subscribe(

packages/jazz-tools/src/tests/deepLoading.test.ts

@@ -14,6 +14,7 @@ import {
     ID,
 } from "../index.js";
 import { newRandomSessionID } from "cojson/src/coValueCore.js";
+import { Effect } from "effect";
 
 class TestMap extends CoMap {
     list = co.ref(TestList);
@@ -38,10 +39,10 @@ describe("Deep loading with depth arg", async () => {
         crypto: Crypto,
     });
 
-    const [initialAsPeer, secondPeer] = connectedPeers("initial", "second", {
+    const [initialAsPeer, secondPeer] = await Effect.runPromise(connectedPeers("initial", "second", {
         peer1role: "server",
         peer2role: "client",
-    });
+    }));
     if (!isControlledAccount(me)) {
         throw "me is not a controlled account";
     }
@@ -137,9 +138,8 @@ describe("Deep loading with depth arg", async () => {
             throw new Error("map4 is undefined");
         }
         expect(map4.list[0]?.stream).not.toBe(null);
-        // TODO: we should expect null here, but apparently we don't even have the id/ref?
-        expect(map4.list[0]?.stream?.[me.id]?.value).not.toBeDefined();
-        expect(map4.list[0]?.stream?.byMe?.value).not.toBeDefined();
+        expect(map4.list[0]?.stream?.[me.id]?.value).toBe(null);
+        expect(map4.list[0]?.stream?.byMe?.value).toBe(null);
 
         const map5 = await TestMap.load(map.id, meOnSecondPeer, {
             list: [{ stream: [{}] }],
@@ -252,10 +252,10 @@ test("Deep loading a record-like coMap", async () => {
         crypto: Crypto,
     });
 
-    const [initialAsPeer, secondPeer] = connectedPeers("initial", "second", {
+    const [initialAsPeer, secondPeer] = await Effect.runPromise(connectedPeers("initial", "second", {
         peer1role: "server",
         peer2role: "client",
-    });
+    }));
     if (!isControlledAccount(me)) {
         throw "me is not a controlled account";
     }