diff --git a/.changeset/nice-mayflies-shop.md b/.changeset/nice-mayflies-shop.md new file mode 100644 index 00000000..dd0cc1e6 --- /dev/null +++ b/.changeset/nice-mayflies-shop.md @@ -0,0 +1,5 @@ +--- +"partysync": patch +--- + +PartySync, phase 1 diff --git a/bun.lockb b/bun.lockb index 69d73401..f6b7b74a 100755 Binary files a/bun.lockb and b/bun.lockb differ diff --git a/fixtures/todo-sync/README.md b/fixtures/todo-sync/README.md new file mode 100644 index 00000000..98692c39 --- /dev/null +++ b/fixtures/todo-sync/README.md @@ -0,0 +1,3 @@ +## todo-sync + +A fixture showing how to sync data from a Durable Object to a client with PartySync. diff --git a/fixtures/todo-sync/package.json b/fixtures/todo-sync/package.json new file mode 100644 index 00000000..353618e1 --- /dev/null +++ b/fixtures/todo-sync/package.json @@ -0,0 +1,19 @@ +{ + "name": "@partyserver/fixture-todo-sync", + "private": true, + "version": "0.0.0", + "scripts": { + "start": "wrangler dev" + }, + "dependencies": { + "@types/react": "^19.0.2", + "@types/react-dom": "^19.0.2", + "partyserver": "workspace:*", + "partysocket": "^1.0.3", + "react": "^19.0.0", + "react-dom": "^19.0.0" + }, + "devDependencies": { + "@cloudflare/workers-types": "^4.20241224.0" + } +} diff --git a/fixtures/todo-sync/public/index.html b/fixtures/todo-sync/public/index.html new file mode 100644 index 00000000..f5a6ad74 --- /dev/null +++ b/fixtures/todo-sync/public/index.html @@ -0,0 +1,27 @@ + + + + + + + Todo Sync + + + + + + + + + + + + + +
Loading...
+ + + diff --git a/fixtures/todo-sync/public/normalize.css b/fixtures/todo-sync/public/normalize.css new file mode 100644 index 00000000..7f612e4b --- /dev/null +++ b/fixtures/todo-sync/public/normalize.css @@ -0,0 +1,351 @@ +/*! normalize.css v8.0.1 | MIT License | github.com/necolas/normalize.css */ + +/* Document + ========================================================================== */ + +/** + * 1. Correct the line height in all browsers. + * 2. Prevent adjustments of font size after orientation changes in iOS. + */ + +html { + line-height: 1.15; /* 1 */ + -webkit-text-size-adjust: 100%; /* 2 */ +} + +/* Sections + ========================================================================== */ + +/** + * Remove the margin in all browsers. + */ + +body { + margin: 0; +} + +/** + * Render the `main` element consistently in IE. + */ + +main { + display: block; +} + +/** + * Correct the font size and margin on `h1` elements within `section` and + * `article` contexts in Chrome, Firefox, and Safari. + */ + +h1 { + font-size: 2em; + margin: 0.67em 0; +} + +/* Grouping content + ========================================================================== */ + +/** + * 1. Add the correct box sizing in Firefox. + * 2. Show the overflow in Edge and IE. + */ + +hr { + box-sizing: content-box; /* 1 */ + height: 0; /* 1 */ + overflow: visible; /* 2 */ +} + +/** + * 1. Correct the inheritance and scaling of font size in all browsers. + * 2. Correct the odd `em` font sizing in all browsers. + */ + +pre { + font-family: monospace; /* 1 */ + font-size: 1em; /* 2 */ +} + +/* Text-level semantics + ========================================================================== */ + +/** + * Remove the gray background on active links in IE 10. + */ + +a { + background-color: transparent; +} + +/** + * 1. Remove the bottom border in Chrome 57- + * 2. Add the correct text decoration in Chrome, Edge, IE, Opera, and Safari. + */ + +abbr[title] { + border-bottom: none; /* 1 */ + text-decoration: underline; /* 2 */ + text-decoration: underline dotted; /* 2 */ +} + +/** + * Add the correct font weight in Chrome, Edge, and Safari. + */ + +b, +strong { + font-weight: bolder; +} + +/** + * 1. Correct the inheritance and scaling of font size in all browsers. + * 2. Correct the odd `em` font sizing in all browsers. + */ + +code, +kbd, +samp { + font-family: monospace; /* 1 */ + font-size: 1em; /* 2 */ +} + +/** + * Add the correct font size in all browsers. + */ + +small { + font-size: 80%; +} + +/** + * Prevent `sub` and `sup` elements from affecting the line height in + * all browsers. + */ + +sub, +sup { + font-size: 75%; + line-height: 0; + position: relative; + vertical-align: baseline; +} + +sub { + bottom: -0.25em; +} + +sup { + top: -0.5em; +} + +/* Embedded content + ========================================================================== */ + +/** + * Remove the border on images inside links in IE 10. + */ + +img { + border-style: none; +} + +/* Forms + ========================================================================== */ + +/** + * 1. Change the font styles in all browsers. + * 2. Remove the margin in Firefox and Safari. + */ + +button, +input, +optgroup, +select, +textarea { + font-family: inherit; /* 1 */ + font-size: 100%; /* 1 */ + line-height: 1.15; /* 1 */ + margin: 0; /* 2 */ +} + +/** + * Show the overflow in IE. + * 1. Show the overflow in Edge. + */ + +button, +input { + /* 1 */ + overflow: visible; +} + +/** + * Remove the inheritance of text transform in Edge, Firefox, and IE. + * 1. Remove the inheritance of text transform in Firefox. + */ + +button, +select { + /* 1 */ + text-transform: none; +} + +/** + * Correct the inability to style clickable types in iOS and Safari. + */ + +button, +[type="button"], +[type="reset"], +[type="submit"] { + -webkit-appearance: button; +} + +/** + * Remove the inner border and padding in Firefox. + */ + +button::-moz-focus-inner, +[type="button"]::-moz-focus-inner, +[type="reset"]::-moz-focus-inner, +[type="submit"]::-moz-focus-inner { + border-style: none; + padding: 0; +} + +/** + * Restore the focus styles unset by the previous rule. + */ + +button:-moz-focusring, +[type="button"]:-moz-focusring, +[type="reset"]:-moz-focusring, +[type="submit"]:-moz-focusring { + outline: 1px dotted ButtonText; +} + +/** + * Correct the padding in Firefox. + */ + +fieldset { + padding: 0.35em 0.75em 0.625em; +} + +/** + * 1. Correct the text wrapping in Edge and IE. + * 2. Correct the color inheritance from `fieldset` elements in IE. + * 3. Remove the padding so developers are not caught out when they zero out + * `fieldset` elements in all browsers. + */ + +legend { + box-sizing: border-box; /* 1 */ + color: inherit; /* 2 */ + display: table; /* 1 */ + max-width: 100%; /* 1 */ + padding: 0; /* 3 */ + white-space: normal; /* 1 */ +} + +/** + * Add the correct vertical alignment in Chrome, Firefox, and Opera. + */ + +progress { + vertical-align: baseline; +} + +/** + * Remove the default vertical scrollbar in IE 10+. + */ + +textarea { + overflow: auto; +} + +/** + * 1. Add the correct box sizing in IE 10. + * 2. Remove the padding in IE 10. + */ + +[type="checkbox"], +[type="radio"] { + box-sizing: border-box; /* 1 */ + padding: 0; /* 2 */ +} + +/** + * Correct the cursor style of increment and decrement buttons in Chrome. + */ + +[type="number"]::-webkit-inner-spin-button, +[type="number"]::-webkit-outer-spin-button { + height: auto; +} + +/** + * 1. Correct the odd appearance in Chrome and Safari. + * 2. Correct the outline style in Safari. + */ + +[type="search"] { + -webkit-appearance: textfield; /* 1 */ + outline-offset: -2px; /* 2 */ +} + +/** + * Remove the inner padding in Chrome and Safari on macOS. + */ + +[type="search"]::-webkit-search-decoration { + -webkit-appearance: none; +} + +/** + * 1. Correct the inability to style clickable types in iOS and Safari. + * 2. Change font properties to `inherit` in Safari. + */ + +::-webkit-file-upload-button { + -webkit-appearance: button; /* 1 */ + font: inherit; /* 2 */ +} + +/* Interactive + ========================================================================== */ + +/* + * Add the correct display in Edge, IE 10+, and Firefox. + */ + +details { + display: block; +} + +/* + * Add the correct display in all browsers. + */ + +summary { + display: list-item; +} + +/* Misc + ========================================================================== */ + +/** + * Add the correct display in IE 10+. + */ + +template { + display: none; +} + +/** + * Add the correct display in IE 10. + */ + +[hidden] { + display: none; +} diff --git a/fixtures/todo-sync/src/client/index.tsx b/fixtures/todo-sync/src/client/index.tsx new file mode 100644 index 00000000..e3a37bf3 --- /dev/null +++ b/fixtures/todo-sync/src/client/index.tsx @@ -0,0 +1,128 @@ +import "./styles.css"; + +import { useState } from "react"; +import { createRoot } from "react-dom/client"; +import { useSync } from "partysync/react"; + +import type { TodoRecord, TodoRpc } from "../shared"; + +function App() { + const [newTodo, setNewTodo] = useState(""); + const [todos, mutate] = useSync( + "todos", + (todos, request): TodoRecord[] => { + switch (request.type) { + case "create": { + const { id, text, completed } = request.payload; + return [ + ...todos, + [id, text, completed, Date.now(), Date.now(), null] + ]; + } + case "update": { + const { id, text, completed } = request.payload; + return todos.map((todo) => + todo[0] === id + ? [id, text, completed, todo[3], Date.now(), null] + : todo + ); + } + case "delete": + return todos.filter((todo) => todo[0] !== request.payload.id); + default: + console.error("unknown request", request); + return todos; + } + } + ); + + const addTodo = (e: React.FormEvent) => { + e.preventDefault(); + if (newTodo.trim()) { + setNewTodo(""); + mutate({ + type: "create", + payload: { + id: crypto.randomUUID(), + text: newTodo.trim(), + completed: 0 + } + }); + } + }; + + const toggleTodo = (id: string) => { + const todo = todos.find((todo) => todo[0] === id); + if (!todo) return; + mutate({ + type: "update", + payload: { + id, + text: todo[1], + completed: todo[2] === 0 ? 1 : 0 + } + }); + }; + + const deleteTodo = (id: string) => { + const todo = todos.find((todo) => todo[0] === id); + if (!todo) return; + mutate({ + type: "delete", + payload: { id } + }); + }; + + return ( +
+

Todo App

+ +
+ setNewTodo(e.target.value)} + placeholder="Add a new todo" + className="flex-1 px-3 py-2 border rounded" + /> + +
+ + +
+ ); +} +createRoot(document.getElementById("root")!).render(); diff --git a/fixtures/todo-sync/src/client/styles.css b/fixtures/todo-sync/src/client/styles.css new file mode 100644 index 00000000..f8efd9d6 --- /dev/null +++ b/fixtures/todo-sync/src/client/styles.css @@ -0,0 +1,7 @@ +/* + We've already included normalize.css. + + But we'd like a modern looking boilerplate. + Clean type, sans-serif, and a nice color palette. + +*/ diff --git a/fixtures/todo-sync/src/client/tsconfig.json b/fixtures/todo-sync/src/client/tsconfig.json new file mode 100644 index 00000000..4e540af9 --- /dev/null +++ b/fixtures/todo-sync/src/client/tsconfig.json @@ -0,0 +1,6 @@ +{ + "extends": "../../../../tsconfig.base.json", + "compilerOptions": { + "lib": ["DOM", "ESNext"] + } +} diff --git a/fixtures/todo-sync/src/server/index.ts b/fixtures/todo-sync/src/server/index.ts new file mode 100644 index 00000000..65641d44 --- /dev/null +++ b/fixtures/todo-sync/src/server/index.ts @@ -0,0 +1,97 @@ +import { routePartykitRequest } from "partyserver"; +import { SyncServer } from "partysync/server"; + +import type { TodoRecord, TodoRpc } from "../shared"; +import type { RpcRequest } from "partysync"; + +type Env = { + ToDos: DurableObjectNamespace; +}; + +function assert(condition: unknown, message: string): asserts condition { + if (!condition) { + throw new Error(message); + } +} + +export class ToDos extends SyncServer { + sql(sql: string, ...values: (string | number | null)[]) { + if ( + ["insert", "update", "delete"].includes( + sql.slice(0, sql.indexOf(" ")).toLowerCase() + ) + ) { + // set alarm to delete expired todos + this.ctx.storage.setAlarm(Date.now() + 24 * 60 * 60 * 1000); + } + return this.ctx.storage.sql.exec(sql, ...values); + } + constructor(state: DurableObjectState, env: Env) { + super(state, env); + this.sql(`CREATE TABLE IF NOT EXISTS todos ( + id TEXT PRIMARY KEY NOT NULL UNIQUE, + text TEXT NOT NULL, + completed INTEGER NOT NULL, + created_at INTEGER NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at INTEGER NOT NULL DEFAULT CURRENT_TIMESTAMP, + deleted_at INTEGER DEFAULT NULL + )`); + } + handleRpcRequest(rpc: RpcRequest): TodoRecord[] { + switch (rpc.request.type) { + case "create": { + const { id, text, completed } = rpc.request.payload; + const result = [ + ...this.sql( + "INSERT INTO todos (id, text, completed) VALUES (?, ?, ?) RETURNING *", + id || crypto.randomUUID(), + text, + completed + ).raw() + ]; + return result as TodoRecord[]; + } + case "update": { + const { id, text, completed } = rpc.request.payload; + + const result = [ + ...this.sql( + "UPDATE todos SET text = ?, completed = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ? RETURNING *", + text, + completed, + id + ).raw() + ]; + + return result as TodoRecord[]; + } + case "delete": { + const { id } = rpc.request.payload; + assert(id, "id is required"); + const result = [ + ...this.sql( + "UPDATE todos SET deleted_at = CURRENT_TIMESTAMP WHERE id = ? RETURNING *", + id + ).raw() + ]; + return result as TodoRecord[]; + } + } + } + async alarm() { + // delete any todos that have been deleted more than 24 hours ago + this.sql( + "DELETE FROM todos WHERE deleted_at < ?", + Date.now() - 24 * 60 * 60 * 1000 + ); + } +} + +export default { + async fetch(request: Request, env: Env): Promise { + return ( + (await routePartykitRequest(request, env)) || + new Response("Not Found", { status: 404 }) + ); + } +} satisfies ExportedHandler; diff --git a/fixtures/todo-sync/src/server/tsconfig.json b/fixtures/todo-sync/src/server/tsconfig.json new file mode 100644 index 00000000..b9b9d198 --- /dev/null +++ b/fixtures/todo-sync/src/server/tsconfig.json @@ -0,0 +1,6 @@ +{ + "extends": "../../../../tsconfig.base.json", + "compilerOptions": { + "types": ["@cloudflare/workers-types"] + } +} diff --git a/fixtures/todo-sync/src/shared.ts b/fixtures/todo-sync/src/shared.ts new file mode 100644 index 00000000..27c9db26 --- /dev/null +++ b/fixtures/todo-sync/src/shared.ts @@ -0,0 +1,32 @@ +export type TodoRecord = [ + id: string, + text: string, + completed: 0 | 1, + created_at: number, + updated_at: number, + deleted_at: number | null +]; + +export type TodoRpc = + | { + type: "create"; + payload: { + id: string; + text: string; + completed: 0 | 1; + }; + } + | { + type: "update"; + payload: { + id: string; + text: string; + completed: 0 | 1; + }; + } + | { + type: "delete"; + payload: { + id: string; + }; + }; diff --git a/fixtures/todo-sync/wrangler.toml b/fixtures/todo-sync/wrangler.toml new file mode 100644 index 00000000..d2c89d5b --- /dev/null +++ b/fixtures/todo-sync/wrangler.toml @@ -0,0 +1,17 @@ +name = "partyserver-fixture-globe" +main = "src/server/index.ts" +compatibility_date = "2024-04-19" + +assets = {directory = "./public"} + +[build] +command = "npx esbuild src/client/index.tsx --bundle --outdir=public/dist --splitting --sourcemap --format=esm" + + +[[durable_objects.bindings]] +name = "Todos" +class_name = "ToDos" + +[[migrations]] +tag = "v1" # Should be unique for each entry +new_sqlite_classes = ["ToDos"] diff --git a/package.json b/package.json index 8c90cfd2..3883de7f 100644 --- a/package.json +++ b/package.json @@ -4,7 +4,7 @@ "private": true, "description": "Everything's better with friends", "scripts": { - "build": "npm run build -w partyserver -w y-partyserver -w partysub", + "build": "npm run build -w partyserver -w y-partyserver -w partysub -w partysync", "check": "concurrently \"npm run check:repo\" \"npm run check:format\" \"npm run check:lint\" \"npm run check:type\" \"npm run check:test\" --kill-others-on-fail", "check:format": "prettier . --check --ignore-unknown", "check:lint": "biome check", diff --git a/packages/partysync/README.md b/packages/partysync/README.md new file mode 100644 index 00000000..ff50a496 --- /dev/null +++ b/packages/partysync/README.md @@ -0,0 +1,106 @@ +## partysync + +An experimental library to synchronise state from a Durable Object to a client. + +> [!WARNING] +> WIP, API and design is subject to change. + +See the [Todo fixture](/fixtures/todo-sync/) for a fully working example. + +## Usage + +First, Setup your server. + +```ts +// server.ts +import { SyncServer } from "partysync"; + +// define your mutations +type Mutations = + | { + type: "create"; + payload: { + id: string; + text: string; + completed: 0 | 1; // to match sqlite's idea of a boolean + }; + } + | { + type: "update"; + // ... etc + }; + +// define the shape of the records that are stored in the Durable Object database +type RecordType = [ + // always add id + string, // id + string, // text + boolean, // completed + // always add created_at, updated_at, deleted_at + number, // created_at + number, // updated_at + number | null // deleted_at +]; + +// note: we do soft deletes to be able to sync deleted records to the client + +export class MyServer extends SyncServer { + constructor(state: DurableObjectState, env: Env) { + super(state, env); + // setup a database table for your records + this.sql(`CREATE TABLE IF NOT EXISTS todos ( + id TEXT PRIMARY KEY NOT NULL UNIQUE, + text TEXT NOT NULL, + completed INTEGER NOT NULL, + created_at INTEGER NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at INTEGER NOT NULL DEFAULT CURRENT_TIMESTAMP, + deleted_at INTEGER DEFAULT NULL + )`); + } + // setup a handler for rpc requests + handleRpcRequest(rpc) { + switch (rpc.type) { + case "create": { + const { id, text, completed } = rpc.payload; + return this.ctx.storage.sql.exec( + "INSERT INTO todos (id, text, completed) VALUES (?, ?, ?) RETURNING *", + id, + text, + completed + ); + } + // etc + } + } +} +``` + +Then, setup your client. + +```tsx +// client.tsx +import { useSync } from "partysync/react"; + +// in your component... +const [todos, mutate] = useSync( + "todos", + (todos, mutation) => { + // optionally do an optimistic update + switch (mutation.type) { + case "create": { + const { id, text, completed } = mutation.payload; + return [...todos, [id, text, completed, Date.now(), Date.now(), null]]; + } + // ... etc + } + } +); + +// call the mutation whenever +function onClick() { + mutate({ + type: "create", + payload: { id: "1", text: "hello", completed: 0 } + }); +} +``` diff --git a/packages/partysync/package.json b/packages/partysync/package.json new file mode 100644 index 00000000..76ad8950 --- /dev/null +++ b/packages/partysync/package.json @@ -0,0 +1,55 @@ +{ + "name": "partysync", + "version": "0.0.1", + "repository": { + "type": "git", + "url": "git://github.com/threepointone/partyserver.git" + }, + "type": "module", + "exports": { + ".": { + "types": "./dist/index.d.ts", + "require": "./dist/index.js", + "import": "./dist/index.js" + }, + "./client": { + "types": "./dist/client/index.d.ts", + "require": "./dist/client/index.js", + "import": "./dist/client/index.js" + }, + "./server": { + "types": "./dist/server/index.d.ts", + "require": "./dist/server/index.js", + "import": "./dist/server/index.js" + }, + "./react": { + "types": "./dist/react/index.d.ts", + "require": "./dist/react/index.js", + "import": "./dist/react/index.js" + } + }, + "scripts": { + "build": "bun scripts/build.ts" + }, + "files": [ + "dist", + "README.md" + ], + "keywords": [ + "sync", + "local-first", + "durable-objects" + ], + "author": "Sunil Pai ", + "license": "ISC", + "description": "", + "dependencies": {}, + "peerDependencies": { + "@cloudflare/workers-types": "^4.20240729.0", + "partyserver": "workspace:*" + }, + "devDependencies": { + "@cloudflare/workers-types": "^4.20241224.0", + "partyserver": "workspace:*" + } +} diff --git a/packages/partysync/scripts/build.ts b/packages/partysync/scripts/build.ts new file mode 100644 index 00000000..073a77a1 --- /dev/null +++ b/packages/partysync/scripts/build.ts @@ -0,0 +1,22 @@ +import { $ } from "bun"; +import { build } from "tsup"; + +await build({ + entry: [ + "src/index.ts", + "src/server/index.ts", + "src/client/index.ts", + "src/react/index.tsx" + ], + splitting: true, + sourcemap: true, + clean: true, + external: ["cloudflare:workers", "partyserver", "react"], + format: "esm", + dts: true +}); + +// then run prettier on the generated .d.ts files +await $`prettier --write ./dist/**/*.d.ts`; + +process.exit(0); diff --git a/packages/partysync/scripts/tsconfig.json b/packages/partysync/scripts/tsconfig.json new file mode 100644 index 00000000..f76546d0 --- /dev/null +++ b/packages/partysync/scripts/tsconfig.json @@ -0,0 +1,6 @@ +{ + "extends": "../../../tsconfig.base.json", + "compilerOptions": { + "types": ["bun"] + } +} diff --git a/packages/partysync/src/client/index.ts b/packages/partysync/src/client/index.ts new file mode 100644 index 00000000..051aaa72 --- /dev/null +++ b/packages/partysync/src/client/index.ts @@ -0,0 +1 @@ +export class SyncClient {} diff --git a/packages/partysync/src/client/tsconfig.json b/packages/partysync/src/client/tsconfig.json new file mode 100644 index 00000000..c382b56f --- /dev/null +++ b/packages/partysync/src/client/tsconfig.json @@ -0,0 +1,6 @@ +{ + "extends": "../../../../tsconfig.base.json", + "compilerOptions": { + "lib": ["DOM"] + } +} diff --git a/packages/partysync/src/index.ts b/packages/partysync/src/index.ts new file mode 100644 index 00000000..fd0dfe35 --- /dev/null +++ b/packages/partysync/src/index.ts @@ -0,0 +1,51 @@ +export type RpcRequest = { + id: string; + channel: string; + rpc: true; + request: T; +}; + +export type RpcResponse = + | { + id: string; + channel: string; + rpc: true; + type: "success"; + result: RecordType[]; + } + | { + id: string; + channel: string; + rpc: true; + type: "error"; + error: string[]; + } + | { + rpc: true; + type: "exception"; + exception: string[]; + }; + +export type BroadcastMessage = { + broadcast: true; + channel: string; +} & ( + | { + type: "update"; + payload: T[]; + } + | { + type: "delete-all"; + } +); + +export type SyncRequest = { + channel: string; + sync: true; +}; + +export type SyncResponse = { + channel: string; + sync: true; + payload: T[]; +}; diff --git a/packages/partysync/src/react/index.tsx b/packages/partysync/src/react/index.tsx new file mode 100644 index 00000000..79d2e740 --- /dev/null +++ b/packages/partysync/src/react/index.tsx @@ -0,0 +1,222 @@ +import { startTransition, useEffect, useOptimistic, useState } from "react"; +import { usePartySocket } from "partysocket/react"; + +import type { + BroadcastMessage, + RpcRequest, + RpcResponse, + SyncRequest +} from ".."; +import type { PartySocket } from "partysocket"; + +// we keep the actual cache external to the class +// so it can be reused across instances/rerenders + +const rpcCaches = new Map< + string, + Map> +>(); + +class RPC { + private rpcCache: Map>; + private controller = new AbortController(); + constructor( + private channel: string, + private socket: PartySocket + ) { + const cache = rpcCaches.get(channel); + if (!cache) { + rpcCaches.set(channel, new Map()); + } + this.rpcCache = rpcCaches.get(channel)!; + this.socket.addEventListener( + "message", + (event) => { + const message = JSON.parse(event.data) as RpcResponse; + if ( + (message.type === "success" || message.type === "error") && + message.channel === this.channel && + message.rpc === true + ) { + this.resolve(message); + } + }, + { signal: this.controller.signal } + ); + } + + private rpc(id: string, timeout = 10000) { + const resolver = Promise.withResolvers(); + this.rpcCache.set(id, resolver); + setTimeout(() => { + this.rpcCache.delete(id); + resolver.reject(new Error(`RPC call ${id} timed out`)); + }, timeout); + return resolver.promise; + } + + public async call(request: Mutation, timeout = 10000): Promise { + const id: string = crypto.randomUUID(); + this.socket.send( + JSON.stringify({ + id, + channel: this.channel, + rpc: true, + request + } satisfies RpcRequest) + ); + return this.rpc(id, timeout) as Promise; + } + + private async resolve(response: RpcResponse) { + if (response.type === "exception") { + throw new Error(response.exception.join("\n")); + } + const resolver = this.rpcCache.get(response.id); + if (!resolver) { + console.warn(`No resolver found for id: ${response.id}`); + return; + } + if (response.type === "success") { + resolver.resolve(response.result); + } else { + resolver.reject(new Error(response.error.join("\n"))); + } + } + + destroy() { + // this.rpcCache.clear(); + // cancel the signal + this.controller.abort(); + } +} + +export function useSync( + key: string, + mutate: (currentState: RecordType[], request: Mutation) => RecordType[] +): [RecordType[], (request: Mutation) => void] { + const [data, setData] = useState([] as RecordType[]); + + const socket = usePartySocket({ + party: key, + room: "default" + }); + + const [rpc] = useState>( + () => new RPC(key, socket) + ); + + useEffect(() => { + return () => { + rpc.destroy(); + }; + }, [rpc]); + + useEffect(() => { + // do initial sync + const controller = new AbortController(); + socket.send( + JSON.stringify({ + channel: key, + sync: true + } satisfies SyncRequest) + ); + socket.addEventListener( + "message", + (event) => { + const message = JSON.parse(event.data); + if (message.channel === key && message.sync === true) { + // this is all the data for initial sync + setData(message.payload); + } + }, + { signal: controller.signal } + ); + return () => { + controller.abort(); + }; + }, [socket, key]); + + useEffect(() => { + function handleMessage(event: MessageEvent) { + const message = JSON.parse(event.data) as BroadcastMessage; + if (message.broadcast === true && message.channel === key) { + if (message.type === "update") { + setData((records) => { + const updates = message.payload; + const updatedRecords = [...records]; + for (const update of updates) { + if (update.at(-1) == null) { + // doesn't have deleted_at, so it's not a delete + const index = updatedRecords.findIndex( + (r) => r[0] === update[0] + ); + if (index !== -1) { + // update the record + updatedRecords.splice(index, 1, update); + } else { + // add the record + updatedRecords.push(update); + } + } else { + // this is a delete + const index = updatedRecords.findIndex( + (r) => r[0] === update[0] + ); + if (index !== -1) { + updatedRecords.splice(index, 1); + } + } + } + return updatedRecords; + }); + } else if (message.type === "delete-all") { + setData([]); + } + } + } + + socket.addEventListener("message", handleMessage); + + return () => { + socket.removeEventListener("message", handleMessage); + }; + }, [socket, key]); + + const [value, setValue] = useOptimistic( + data, // todo: this should be synced from the server + (currentState, request) => { + return mutate(currentState, request); + } + ); + return [ + value, + (request) => { + rpc.call(request).then((result) => { + if (result.length === 0) { + return; + } + // let changed = false; + const newData = [...data]; + for (const record of result) { + // if record is in data, update it + + const index = data.findIndex((item) => item[0] === record[0]); + if (index !== -1) { + newData.splice(index, 1, record); + // changed = true; + } + // if record is not in data, add it + else if (index === -1) { + newData.push(record); + // changed = true; + } + } + setData(newData); + }); + startTransition(() => { + setValue(request); + }); + } + ]; +} diff --git a/packages/partysync/src/react/tsconfig.json b/packages/partysync/src/react/tsconfig.json new file mode 100644 index 00000000..c382b56f --- /dev/null +++ b/packages/partysync/src/react/tsconfig.json @@ -0,0 +1,6 @@ +{ + "extends": "../../../../tsconfig.base.json", + "compilerOptions": { + "lib": ["DOM"] + } +} diff --git a/packages/partysync/src/server/index.ts b/packages/partysync/src/server/index.ts new file mode 100644 index 00000000..a60b933a --- /dev/null +++ b/packages/partysync/src/server/index.ts @@ -0,0 +1,106 @@ +import { Server } from "partyserver"; + +import type { + BroadcastMessage, + RpcRequest, + RpcResponse, + SyncRequest, + SyncResponse +} from "../index.ts"; +import type { Connection, WSMessage } from "partyserver"; + +export class SyncServer< + Env, + RecordType extends unknown[], + Mutations extends { type: string; payload: unknown } +> extends Server { + handleRpcRequest(rpc: RpcRequest): RecordType[] { + throw new Error("Not implemented"); + } + + // todo: store returns on a queue to flush back to client + async onMessage(connection: Connection, message: WSMessage): Promise { + if (typeof message !== "string") { + console.error("Received non-string message"); + return; + } + let json: RpcRequest | SyncRequest; + + try { + json = JSON.parse(message); + } catch (err) { + connection.send( + JSON.stringify({ + type: "exception", + rpc: true, + exception: [`Failed to parse message: ${(err as Error).message}`] + } satisfies RpcResponse) + ); + return; + } + + if ("sync" in json && json.sync) { + connection.send( + JSON.stringify({ + sync: true, + channel: json.channel, + payload: [ + ...this.ctx.storage.sql + .exec(`SELECT * FROM ${json.channel} WHERE deleted_at IS NULL`) + .raw() + ] as RecordType[] + } satisfies SyncResponse) + ); + return; + } + + function replySuccess(json: RpcRequest, message: RecordType[]) { + connection.send( + JSON.stringify({ + type: "success", + rpc: true, + channel: json.channel, + id: json.id, + result: message + } satisfies RpcResponse) + ); + } + + function replyError(json: RpcRequest, message: string[]) { + connection.send( + JSON.stringify({ + type: "error", + rpc: true, + channel: json.channel, + id: json.id, + error: message + } satisfies RpcResponse) + ); + } + + const broadcastUpdate = (records: RecordType[]) => { + this.broadcast( + JSON.stringify({ + broadcast: true, + type: "update", + channel: json.channel, + payload: records + } satisfies BroadcastMessage), + [connection.id] + ); + }; + + try { + const result = this.handleRpcRequest(json as RpcRequest); + + replySuccess(json as RpcRequest, result); + broadcastUpdate(result); + return; + } catch (err) { + replyError(json as RpcRequest, [(err as Error).message]); + return; + } + } + + async onConnect(connection: Connection) {} +} diff --git a/packages/partysync/src/server/tsconfig.json b/packages/partysync/src/server/tsconfig.json new file mode 100644 index 00000000..b9b9d198 --- /dev/null +++ b/packages/partysync/src/server/tsconfig.json @@ -0,0 +1,6 @@ +{ + "extends": "../../../../tsconfig.base.json", + "compilerOptions": { + "types": ["@cloudflare/workers-types"] + } +}