feat: desktop sync, the app server mirrors every CodeWork machine

apps/server/package.json

@@ -18,6 +18,7 @@
 		"effect": "4.0.5"
 	},
 	"devDependencies": {
+		"@codeworksh/aikit": "workspace:*",
 		"@types/node": "catalog:",
 		"typescript": "catalog:",
 		"vite-plus": "catalog:"

apps/server/src/cli.ts

@@ -101,7 +101,7 @@ const found = Effect.gen(function* () {
 });
 
 /** The `codework` a server runs: the one chosen for it, else the one found. */
-const executableOf = (row: Servers.Row, foundOnce: typeof found) => {
+const executableOf = (row: Servers.LocalRow, foundOnce: typeof found) => {
 	const { executable } = row;
 	if (executable === null) return foundOnce;
 	const command = split(executable);
@@ -151,14 +151,15 @@ export const awaitConnection = Effect.gen(function* () {
 	);
 });
 
-/** Every server and how it is set up, with its executable and process as found now. */
+/** Every server and how it is set up, with a local one's executable and process as found now. */
 export const statuses = Effect.gen(function* () {
 	const servers = yield* Servers.Service;
 	const foundOnce = yield* Effect.cached(found);
 	return yield* Effect.forEach(
 		yield* servers.list,
 		(row) =>
 			Effect.gen(function* () {
+				if (row.kind === "remote") return row satisfies ServerStatus;
 				const home = servers.homeOf(row);
 				const [cli, server] = yield* Effect.all(
 					[executableOf(row, foundOnce), serverAt(home.path, row.isDefault)],
@@ -186,7 +187,7 @@ export const statuses = Effect.gen(function* () {
 
 // The CLI owns a server's lifecycle; the app only asks it, with the server's variables and in its
 // home, which a variable never overrides.
-export const service = Effect.fn("Cli.service")(function* (row: Servers.Row, action: CliAction) {
+export const service = Effect.fn("Cli.service")(function* (row: Servers.LocalRow, action: CliAction) {
 	const servers = yield* Servers.Service;
 	const cli = yield* executableOf(row, found);
 	if (cli === null || !("version" in cli))

apps/server/src/codework.ts

@@ -1,19 +1,14 @@
 import { Codework as Mirror } from "@codeworksh/widget/codework";
 import { Codework as Client } from "@codeworksh/widget/backend";
 import { NodeSocket } from "@effect/platform-node";
-import { Deferred, Duration, Effect, FileSystem, Layer, Schedule, Schema, Stream } from "effect";
+import { Deferred, Duration, Effect, FileSystem, Layer, Option, Schedule, Schema, Stream } from "effect";
 import { RpcClient, RpcSerialization } from "effect/rpc";
 import { Socket } from "effect/socket";
 /* @effect-diagnostics nodeBuiltinImport:off -- a home's registration is a file path. */
 import path from "node:path";
 import { CodeworkUnavailable, HostError } from "@codeworksh/contract";
 import { Servers } from "./servers.ts";
 
-export type ThreadInfo = Mirror.ThreadInfo;
-
-/** What an untitled thread is called until its first prompt names it. */
-export const UNTITLED = Mirror.UNTITLED;
-
 /** How a running server advertises itself: `<home>/service.json`, written by `codework serve`. */
 const Registration = Schema.fromJsonString(
 	Schema.Struct({ pid: Schema.Int, version: Schema.String, url: Schema.String, port: Schema.Int }),
@@ -62,37 +57,69 @@ const endOnHomeChange = (socket: Socket.Socket, changes: Stream.Stream<string>):
 	}),
 });
 
+/** A client over `socket`, which reconnects on its own. */
+const connect = <R>(socket: Layer.Layer<Socket.Socket, never, R>) =>
+	Layer.effect(Client, RpcClient.make(Mirror.Api, { flatten: true })).pipe(
+		Layer.provide(RpcClient.layerProtocolSocket({ retryTransientErrors: true })),
+		Layer.provide(socket.pipe(Layer.provide(NodeSocket.layerWebSocketConstructor))),
+		Layer.provide(RpcSerialization.layerNdjson),
+		Layer.orDie,
+	);
+
 /**
- * The app's one connection to a CodeWork server: the default server's. The registration is read on
- * every connect, so a server restarted on a new port, a new home or a new default is picked up
- * without restarting this app; until one is registered, connecting waits.
+ * The URL `registered` reads, waited for until there is one. The socket reads its URL while opening,
+ * which Effect runs uninterruptibly; waiting there for a server to register would otherwise keep the
+ * app server from ever shutting down.
  */
-export const layer = Layer.effect(Client, RpcClient.make(Mirror.Api, { flatten: true })).pipe(
-	Layer.provide(RpcClient.layerProtocolSocket({ retryTransientErrors: true })),
-	Layer.provide(
+const waitFor = (registered: Effect.Effect<Option.Option<{ readonly url: string }>>) =>
+	registered.pipe(
+		Effect.flatMap(Effect.fromOption),
+		Effect.map((found) => found.url),
+		Effect.retry(Schedule.spaced("1 second")),
+		Effect.orDie,
+		Effect.interruptible,
+	);
+
+/**
+ * The app's connection to the default server. The registration is read on every connect, so a server
+ * restarted on a new port, a new home or a new default is picked up without restarting this app;
+ * until one is registered, connecting waits.
+ */
+export const layer = connect(
+	Layer.effect(
+		Socket.Socket,
+		Effect.gen(function* () {
+			const fs = yield* FileSystem.FileSystem;
+			const servers = yield* Servers.Service;
+			const url = waitFor(
+				registration.pipe(
+					Effect.provideService(FileSystem.FileSystem, fs),
+					Effect.provideService(Servers.Service, servers),
+				),
+			);
+			return endOnHomeChange(yield* Socket.makeWebSocket(url), servers.homeChanges);
+		}),
+	),
+);
+
+/**
+ * A connection to one machine's server: a remote one at its URL, a local one at what its home
+ * registers, read on every connect as {@link layer} does.
+ */
+export const machine = (row: Servers.Row) =>
+	connect(
 		Layer.effect(
 			Socket.Socket,
 			Effect.gen(function* () {
+				if (row.kind === "remote") return yield* Socket.makeWebSocket(row.url);
 				const fs = yield* FileSystem.FileSystem;
-				const servers = yield* Servers.Service;
-				const url = registration.pipe(
-					Effect.flatMap(Effect.fromOption),
-					Effect.map((registered) => registered.url),
-					Effect.retry(Schedule.spaced("1 second")),
-					Effect.orDie,
-					Effect.provideService(FileSystem.FileSystem, fs),
-					Effect.provideService(Servers.Service, servers),
-					// The socket reads its URL while opening, which Effect runs uninterruptibly; waiting there
-					// for a server to register would otherwise keep the app server from ever shutting down.
-					Effect.interruptible,
+				const home = (yield* Servers.Service).homeOf(row).path;
+				return yield* Socket.makeWebSocket(
+					waitFor(registrationAt(home).pipe(Effect.provideService(FileSystem.FileSystem, fs))),
 				);
-				return endOnHomeChange(yield* Socket.makeWebSocket(url), servers.homeChanges);
 			}),
-		).pipe(Layer.provide(NodeSocket.layerWebSocketConstructor)),
-	),
-	Layer.provide(RpcSerialization.layerNdjson),
-	Layer.orDie,
-);
+		),
+	);
 
 export interface CallOptions {
 	/** How long the server has to answer; 5 seconds unless given. */