refactor: move redis p2p logic to generic driver

Author: NathanFlurry

CLAUDE.md

@@ -0,0 +1,41 @@
+# Actor-Core Development Guide
+
+## Build Commands
+
+- **Build:** `yarn build` - Production build
+- **Dev:** `yarn dev` - Watch mode for development
+- **Format:** `yarn fmt` - Format code with Biome
+- **Type Check:** `yarn check-types` - Verify TypeScript types
+
+## Code Style Guidelines
+
+- **Formatting:** Uses Biome for consistent formatting
+- **Imports:** Organized imports enforced, unused imports warned
+- **TypeScript:** Strict mode enabled, target ESNext
+- **Naming:** 
+  - camelCase for variables, functions
+  - PascalCase for classes, interfaces, types
+  - UPPER_CASE for constants
+  - Use `#` prefix for private class members (not `private` keyword)
+- **Error Handling:** 
+  - Extend from `ActorError` base class
+  - Use `UserError` for client-safe errors
+  - Use `InternalError` for internal errors
+
+## Project Structure
+
+- Monorepo with Yarn workspaces and Turborepo
+- Core code in `packages/actor-core/`
+- Platform implementations in `packages/platforms/`
+- Driver implementations in `packages/drivers/`
+
+## Development Notes
+
+- Prefer classes over factory functions
+- Use zod for runtime type validation
+- Use `assertUnreachable(x: never)` for exhaustive type checking in switch statements
+- Follow existing patterns for P2P networking
+- Add proper JSDoc comments for public APIs
+- Ensure proper error handling with descriptive messages
+- Verify type safety with `yarn check-types` before committing
+

package.json

@@ -5,6 +5,7 @@
   "workspaces": [
     "packages/*",
     "packages/platforms/*",
+    "packages/drivers/*",
     "packages/misc/*",
     "examples/*",
     "examples/*/platforms/*"

packages/actor-core/package.json

@@ -41,91 +41,60 @@
         "default": "./dist/common/log.cjs"
       }
     },
-    "./platform": {
-      "import": {
-        "types": "./dist/platform.d.ts",
-        "default": "./dist/platform.js"
-      },
-      "require": {
-        "types": "./dist/platform.d.cts",
-        "default": "./dist/platform.cjs"
-      }
-    },
-    "./actor/generic": {
-      "import": {
-        "types": "./dist/actor/generic/mod.d.ts",
-        "default": "./dist/actor/generic/mod.js"
-      },
-      "require": {
-        "types": "./dist/actor/generic/mod.d.cts",
-        "default": "./dist/actor/generic/mod.cjs"
-      }
-    },
-    "./actor/protocol/serde": {
-      "import": {
-        "types": "./dist/actor/protocol/serde.d.ts",
-        "default": "./dist/actor/protocol/serde.js"
-      },
-      "require": {
-        "types": "./dist/actor/protocol/serde.d.cts",
-        "default": "./dist/actor/protocol/serde.cjs"
-      }
-    },
-    "./actor/protocol/message": {
+    "./actor/errors": {
       "import": {
-        "types": "./dist/actor/protocol/message/mod.d.ts",
-        "default": "./dist/actor/protocol/message/mod.js"
+        "types": "./dist/actor/errors.d.ts",
+        "default": "./dist/actor/errors.js"
       },
       "require": {
-        "types": "./dist/actor/protocol/message/mod.d.cts",
-        "default": "./dist/actor/protocol/message/mod.cjs"
+        "types": "./dist/actor/errors.d.cts",
+        "default": "./dist/actor/errors.cjs"
       }
     },
-    "./actor/protocol/message/to_server": {
+    "./utils": {
       "import": {
-        "types": "./dist/actor/protocol/message/to_server.d.ts",
-        "default": "./dist/actor/protocol/message/to_server.js"
+        "types": "./dist/utils.d.ts",
+        "default": "./dist/utils.js"
       },
       "require": {
-        "types": "./dist/actor/protocol/message/to_server.d.cts",
-        "default": "./dist/actor/protocol/message/to_server.cjs"
+        "types": "./dist/utils.d.cts",
+        "default": "./dist/utils.cjs"
       }
     },
-    "./actor/protocol/message/to_client": {
+    "./driver-helpers": {
       "import": {
-        "types": "./dist/actor/protocol/message/to_client.d.ts",
-        "default": "./dist/actor/protocol/message/to_client.js"
+        "types": "./dist/driver-helpers.d.ts",
+        "default": "./dist/driver-helpers.js"
       },
       "require": {
-        "types": "./dist/actor/protocol/message/to_client.d.cts",
-        "default": "./dist/actor/protocol/message/to_client.cjs"
+        "types": "./dist/driver-helpers.d.cts",
+        "default": "./dist/driver-helpers.cjs"
       }
     },
-    "./actor/errors": {
+    "./topologies/p2p": {
       "import": {
-        "types": "./dist/actor/errors.d.ts",
-        "default": "./dist/actor/errors.js"
+        "types": "./dist/topologies/p2p/mod.d.ts",
+        "default": "./dist/topologies/p2p/mod.js"
       },
       "require": {
-        "types": "./dist/actor/errors.d.cts",
-        "default": "./dist/actor/errors.cjs"
+        "types": "./dist/topologies/p2p/mod.d.cts",
+        "default": "./dist/topologies/p2p/mod.cjs"
       }
     },
-    "./manager/protocol": {
+    "./topologies/isolated": {
       "import": {
-        "types": "./dist/manager/protocol/mod.d.ts",
-        "default": "./dist/manager/protocol/mod.js"
+        "types": "./dist/topologies/isolated/mod.d.ts",
+        "default": "./dist/topologies/isolated/mod.js"
       },
       "require": {
-        "types": "./dist/manager/protocol/mod.d.cts",
-        "default": "./dist/manager/protocol/mod.cjs"
+        "types": "./dist/topologies/isolated/mod.d.cts",
+        "default": "./dist/topologies/isolated/mod.cjs"
       }
     }
   },
   "sideEffects": false,
   "scripts": {
-    "build": "tsup src/mod.ts src/client/mod.ts src/common/log.ts src/platform.ts src/actor/generic/mod.ts src/actor/protocol/serde.ts src/actor/protocol/message/mod.ts src/actor/protocol/message/to_server.ts src/actor/protocol/message/to_client.ts src/actor/errors.ts src/manager/protocol/mod.ts",
-    "dev": "yarn build --watch",
+    "build": "tsup src/mod.ts src/client/mod.ts src/common/log.ts src/actor/errors.ts src/topologies/p2p/mod.ts src/topologies/isolated/mod.ts src/utils.ts src/driver-helpers.ts",
     "check-types": "tsc --noEmit"
   },
   "dependencies": {

packages/actor-core/src/actor/runtime/actor.ts

@@ -11,7 +11,8 @@ import type { WSContext, WSEvents } from "hono/ws";
 import onChange from "on-change";
 import { type ActorConfig, mergeActorConfig } from "./actor_config";
 import { Connection, type ConnectionId } from "./connection";
-import type { ActorDriver, ConnectionDriver } from "./driver";
+import type { ActorDriver, ConnectionDrivers } from "./driver";
+import type { ConnectionDriver } from "./driver";
 import * as errors from "../errors";
 import { parseMessage, processMessage } from "../protocol/message/mod";
 import { instanceLogger, logger } from "./log";
@@ -153,7 +154,8 @@ export abstract class Actor<
 
 	#backgroundPromises: Promise<void>[] = [];
 	#config: ActorConfig;
-	#driver!: ActorDriver;
+	#connectionDrivers!: ConnectionDrivers;
+	#actorDriver!: ActorDriver;
 	#actorId!: string;
 	#tags!: ActorTags;
 	#region!: string;
@@ -186,16 +188,18 @@ export abstract class Actor<
 	}
 
 	async __start(
-		driver: ActorDriver,
+		connectionDrivers: ConnectionDrivers,
+		actorDriver: ActorDriver,
 		actorId: string,
 		tags: ActorTags,
 		region: string,
 	) {
-		this.#driver = driver;
+		this.#connectionDrivers = connectionDrivers;
+		this.#actorDriver = actorDriver;
 		this.#actorId = actorId;
 		this.#tags = tags;
 		this.#region = region;
-		this.#schedule = new Schedule(this, driver);
+		this.#schedule = new Schedule(this, actorDriver);
 
 		// Initialize server
 		//
@@ -270,7 +274,7 @@ export abstract class Actor<
 					this.#persistChanged = false;
 
 					// Write to KV
-					await this.#driver.kvPut(KEYS.STATE.DATA, this.#persistRaw);
+					await this.#actorDriver.kvPut(this.#actorId, KEYS.STATE.DATA, this.#persistRaw);
 
 					logger().debug("persist saved");
 				});
@@ -339,7 +343,7 @@ export abstract class Actor<
 	async #initialize() {
 		// Read initial state
 		const [[_i, initialized], [_s, persistData]] =
-			(await this.#driver.kvGetBatch([
+			(await this.#actorDriver.kvGetBatch(this.#actorId, [
 				KEYS.STATE.INITIALIZED,
 				KEYS.STATE.DATA,
 			])) as [
@@ -400,7 +404,7 @@ export abstract class Actor<
 
 			// Update state
 			logger().debug("writing state");
-			await this.#driver.kvPutBatch([
+			await this.#actorDriver.kvPutBatch(this.#actorId, [
 				[KEYS.STATE.INITIALIZED, true],
 				[KEYS.STATE.DATA, persist],
 			]);
@@ -473,7 +477,7 @@ export abstract class Actor<
 
 	__getConnectionDriver(driverId: string): ConnectionDriver {
 		// Get driver
-		const driver = this.#driver.connectionDrivers[driverId];
+		const driver = this.#connectionDrivers[driverId];
 		if (!driver) throw new Error(`No connection driver: ${driverId}`);
 		return driver;
 	}

packages/actor-core/src/actor/runtime/actor_router.ts

@@ -77,7 +77,7 @@ export interface ActorRouterHandler {
 export function createActorRouter(
 	config: BaseConfig,
 	handler: ActorRouterHandler,
-) {
+): Hono {
 	const app = new Hono();
 
 	app.get("/", (c) => {

packages/actor-core/src/actor/runtime/config.ts

@@ -1,12 +1,69 @@
+import { P2PDriver } from "@/driver-helpers";
+import type { AnyActorConstructor } from "./actor";
+import { ActorDriver, ManagerDriver } from "./driver";
+import type { Hono, Context as HonoContext, Handler as HonoHandler } from "hono";
+
 export const DEFAULT_ROUTER_MAX_CONNECTION_PARAMETER_SIZE = 8_192;
 export const DEFAULT_ROUTER_MAX_INCOMING_MESSAGE_SIZE = 65_536;
 
+export const DEFAULT_ACTOR_PEER_LEASE_DURATION = 3000;
+export const DEFAULT_ACTOR_PEER_RENEW_LEASE_GRACE = 1500;
+export const DEFAULT_ACTOR_PEER_CHECK_LEASE_INTERVAL = 1000;
+export const DEFAULT_ACTOR_PEER_CHECK_LEASE_JITTER = 500;
+export const DEFAULT_ACTOR_PEER_MESSAGE_ACK_TIMEOUT = 1000;
+
+export type Topology = "single" | "isolated" | "p2p";
+
 /** Base config used for the actor config across all platforms. */
 export interface BaseConfig {
+	topology: Topology;
+	actors: Record<string, AnyActorConstructor>;
+	drivers: {
+		manager: ManagerDriver;
+		actor: ActorDriver;
+		p2p?: P2PDriver;
+	};
 	router?: {
+		// This is dynamic since NodeJS requires a reference to the app to initialize WebSockets
+		getUpgradeWebSocket?: (
+			app: Hono,
+		) => (createEvents: (c: HonoContext) => any) => HonoHandler;
+
 		/** This goes in the URL so it needs to be short. */
 		maxConnectionParametersSize?: number;
 
 		maxIncomingMessageSize?: number;
 	};
+	actorPeer?: {
+		/**
+		 * How long the actor leader holds a lease for.
+		 *
+		 * Milliseconds
+		 **/
+		leaseDuration?: number;
+		/**
+		 * How long before the lease will expire to issue the renew command.
+		 *
+		 * Milliseconds
+		 */
+		renewLeaseGrace?: number;
+		/**
+		 * How frequently the followers check if the leader is still active.
+		 *
+		 * Milliseconds
+		 */
+		checkLeaseInterval?: number;
+		/**
+		 * Positive jitter for check lease interval
+		 *
+		 * Milliseconds
+		 */
+		checkLeaseJitter?: number;
+		/**
+		 * How long to wait for a message ack.
+		 *
+		 * Milliseconds
+		 */
+		messageAckTimeout?: number;
+	};
 }

packages/actor-core/src/actor/runtime/driver.ts

@@ -1,32 +1,52 @@
-import { CachedSerializer } from "../protocol/serde";
+import type { ActorTags, Connection } from "./mod";
 import type * as messageToClient from "@/actor/protocol/message/to_client";
-import { AnyActor } from "./actor";
-import { Connection } from "./connection";
-
-export interface LoadOutput {
-	actor: {
-		id: string;
-		tags: Record<string, string>;
-		createdAt: Date;
-	};
-	region: string;
+import type { CachedSerializer } from "@/actor/protocol/serde";
+import type { AnyActor } from "./actor";
+
+export type ConnectionDrivers = Record<string, ConnectionDriver>;
+
+export interface GetForIdInput {
+	origin: string;
+	actorId: string;
 }
 
-export interface ActorDriver {
-	connectionDrivers: Record<string, ConnectionDriver>;
+export interface GetWithTagsInput {
+	origin: string;
+	tags: ActorTags;
+}
+
+export interface CreateActorInput {
+	origin: string;
+	region?: string;
+	tags: ActorTags;
+}
 
+export interface GetActorOutput {
+	endpoint: string;
+}
+
+export interface ManagerDriver {
+	getForId(input: GetForIdInput): Promise<GetActorOutput>;
+	getWithTags(input: GetWithTagsInput): Promise<GetActorOutput | undefined>;
+	createActor(input: CreateActorInput): Promise<GetActorOutput>;
+}
+
+export type KvKey = unknown[];
+export type KvValue = unknown;
+
+export interface ActorDriver {
 	//load(): Promise<LoadOutput>;
 
 	// HACK: Clean these up
-	kvGet(key: any): Promise<any>;
-	kvGetBatch(key: any[]): Promise<[any, any][]>;
-	kvPut(key: any, value: any): Promise<void>;
-	kvPutBatch(key: [any, any][]): Promise<void>;
-	kvDelete(key: any): Promise<void>;
-	kvDeleteBatch(key: any[]): Promise<void>;
+	kvGet(actorId: string, key: KvKey): Promise<KvValue | undefined>;
+	kvGetBatch(actorId: string, key: KvKey[]): Promise<(KvValue | undefined)[]>;
+	kvPut(actorId: string, key: KvKey, value: KvValue): Promise<void>;
+	kvPutBatch(actorId: string, key: [KvKey, KvValue][]): Promise<void>;
+	kvDelete(actorId: string, key: KvKey): Promise<void>;
+	kvDeleteBatch(actorId: string, key: KvKey[]): Promise<void>;
 
 	// Schedule
-	setAlarm(timestamp: number): Promise<void>;
+	setAlarm(actorId: string, timestamp: number): Promise<void>;
 
 	// TODO:
 	//destroy(): Promise<void>;