From d54ecaf64fee49a201a617b611684ab2e10d8dc9 Mon Sep 17 00:00:00 2001 From: James Long Date: Thu, 19 Mar 2026 14:34:35 -0400 Subject: [PATCH] feat(core): initial implementation of syncing --- .../20260318163842_events/migration.sql | 13 + .../20260318163842_events/snapshot.json | 1325 +++++++++++++++++ packages/opencode/src/bus/bus-event.ts | 3 - packages/opencode/src/cli/cmd/github.ts | 6 +- packages/opencode/src/id/id.ts | 1 + packages/opencode/src/installation/index.ts | 4 + packages/opencode/src/plugin/index.ts | 3 +- packages/opencode/src/server/projectors.ts | 6 + .../opencode/src/server/routes/session.ts | 6 +- packages/opencode/src/server/server.ts | 3 + packages/opencode/src/session/index.ts | 292 ++-- packages/opencode/src/session/message-v2.ts | 48 +- packages/opencode/src/session/projectors.ts | 135 ++ packages/opencode/src/session/revert.ts | 12 +- packages/opencode/src/share/share-next.ts | 25 +- packages/opencode/src/storage/db.ts | 48 +- packages/opencode/src/sync/event.sql.ts | 16 + packages/opencode/src/sync/index.ts | 159 ++ packages/opencode/src/test-projection.ts | 83 ++ packages/opencode/src/test-replicate.ts | 82 + .../opencode/test/session/session.test.ts | 4 +- packages/opencode/test/skill/skill.test.ts | 1 + packages/opencode/test/sync/index.test.ts | 162 ++ 23 files changed, 2177 insertions(+), 260 deletions(-) create mode 100644 packages/opencode/migration/20260318163842_events/migration.sql create mode 100644 packages/opencode/migration/20260318163842_events/snapshot.json create mode 100644 packages/opencode/src/server/projectors.ts create mode 100644 packages/opencode/src/session/projectors.ts create mode 100644 packages/opencode/src/sync/event.sql.ts create mode 100644 packages/opencode/src/sync/index.ts create mode 100644 packages/opencode/src/test-projection.ts create mode 100644 packages/opencode/src/test-replicate.ts create mode 100644 packages/opencode/test/sync/index.test.ts diff --git a/packages/opencode/migration/20260318163842_events/migration.sql b/packages/opencode/migration/20260318163842_events/migration.sql new file mode 100644 index 0000000000..ac2fe89f97 --- /dev/null +++ b/packages/opencode/migration/20260318163842_events/migration.sql @@ -0,0 +1,13 @@ +CREATE TABLE `event_sequence` ( + `aggregate_id` text PRIMARY KEY, + `seq` integer NOT NULL +); +--> statement-breakpoint +CREATE TABLE `event` ( + `id` text PRIMARY KEY, + `aggregate_id` text NOT NULL, + `seq` integer NOT NULL, + `name` text NOT NULL, + `data` text NOT NULL, + CONSTRAINT `fk_event_aggregate_id_event_sequence_aggregate_id_fk` FOREIGN KEY (`aggregate_id`) REFERENCES `event_sequence`(`aggregate_id`) ON DELETE CASCADE +); diff --git a/packages/opencode/migration/20260318163842_events/snapshot.json b/packages/opencode/migration/20260318163842_events/snapshot.json new file mode 100644 index 0000000000..93bb66bef4 --- /dev/null +++ b/packages/opencode/migration/20260318163842_events/snapshot.json @@ -0,0 +1,1325 @@ +{ + "version": "7", + "dialect": "sqlite", + "id": "64c94495-3f7d-4169-a078-2df22ca80057", + "prevIds": [ + "fb311f30-9948-4131-b15c-7d308478a878" + ], + "ddl": [ + { + "name": "account_state", + "entityType": "tables" + }, + { + "name": "account", + "entityType": "tables" + }, + { + "name": "control_account", + "entityType": "tables" + }, + { + "name": "workspace", + "entityType": "tables" + }, + { + "name": "project", + "entityType": "tables" + }, + { + "name": "message", + "entityType": "tables" + }, + { + "name": "part", + "entityType": "tables" + }, + { + "name": "permission", + "entityType": "tables" + }, + { + "name": "session", + "entityType": "tables" + }, + { + "name": "todo", + "entityType": "tables" + }, + { + "name": "session_share", + "entityType": "tables" + }, + { + "name": "event_sequence", + "entityType": "tables" + }, + { + "name": "event", + "entityType": "tables" + }, + { + "type": "integer", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "id", + "entityType": "columns", + "table": "account_state" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "active_account_id", + "entityType": "columns", + "table": "account_state" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "active_org_id", + "entityType": "columns", + "table": "account_state" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "id", + "entityType": "columns", + "table": "account" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "email", + "entityType": "columns", + "table": "account" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "url", + "entityType": "columns", + "table": "account" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "access_token", + "entityType": "columns", + "table": "account" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "refresh_token", + "entityType": "columns", + "table": "account" + }, + { + "type": "integer", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "token_expiry", + "entityType": "columns", + "table": "account" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "time_created", + "entityType": "columns", + "table": "account" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "time_updated", + "entityType": "columns", + "table": "account" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "email", + "entityType": "columns", + "table": "control_account" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "url", + "entityType": "columns", + "table": "control_account" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "access_token", + "entityType": "columns", + "table": "control_account" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "refresh_token", + "entityType": "columns", + "table": "control_account" + }, + { + "type": "integer", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "token_expiry", + "entityType": "columns", + "table": "control_account" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "active", + "entityType": "columns", + "table": "control_account" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "time_created", + "entityType": "columns", + "table": "control_account" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "time_updated", + "entityType": "columns", + "table": "control_account" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "id", + "entityType": "columns", + "table": "workspace" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "type", + "entityType": "columns", + "table": "workspace" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "branch", + "entityType": "columns", + "table": "workspace" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "name", + "entityType": "columns", + "table": "workspace" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "directory", + "entityType": "columns", + "table": "workspace" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "extra", + "entityType": "columns", + "table": "workspace" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "project_id", + "entityType": "columns", + "table": "workspace" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "id", + "entityType": "columns", + "table": "project" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "worktree", + "entityType": "columns", + "table": "project" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "vcs", + "entityType": "columns", + "table": "project" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "name", + "entityType": "columns", + "table": "project" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "icon_url", + "entityType": "columns", + "table": "project" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "icon_color", + "entityType": "columns", + "table": "project" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "time_created", + "entityType": "columns", + "table": "project" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "time_updated", + "entityType": "columns", + "table": "project" + }, + { + "type": "integer", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "time_initialized", + "entityType": "columns", + "table": "project" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "sandboxes", + "entityType": "columns", + "table": "project" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "commands", + "entityType": "columns", + "table": "project" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "id", + "entityType": "columns", + "table": "message" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "session_id", + "entityType": "columns", + "table": "message" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "time_created", + "entityType": "columns", + "table": "message" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "time_updated", + "entityType": "columns", + "table": "message" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "data", + "entityType": "columns", + "table": "message" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "id", + "entityType": "columns", + "table": "part" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "message_id", + "entityType": "columns", + "table": "part" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "session_id", + "entityType": "columns", + "table": "part" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "time_created", + "entityType": "columns", + "table": "part" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "time_updated", + "entityType": "columns", + "table": "part" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "data", + "entityType": "columns", + "table": "part" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "project_id", + "entityType": "columns", + "table": "permission" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "time_created", + "entityType": "columns", + "table": "permission" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "time_updated", + "entityType": "columns", + "table": "permission" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "data", + "entityType": "columns", + "table": "permission" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "id", + "entityType": "columns", + "table": "session" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "project_id", + "entityType": "columns", + "table": "session" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "workspace_id", + "entityType": "columns", + "table": "session" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "parent_id", + "entityType": "columns", + "table": "session" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "slug", + "entityType": "columns", + "table": "session" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "directory", + "entityType": "columns", + "table": "session" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "title", + "entityType": "columns", + "table": "session" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "version", + "entityType": "columns", + "table": "session" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "share_url", + "entityType": "columns", + "table": "session" + }, + { + "type": "integer", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "summary_additions", + "entityType": "columns", + "table": "session" + }, + { + "type": "integer", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "summary_deletions", + "entityType": "columns", + "table": "session" + }, + { + "type": "integer", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "summary_files", + "entityType": "columns", + "table": "session" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "summary_diffs", + "entityType": "columns", + "table": "session" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "revert", + "entityType": "columns", + "table": "session" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "permission", + "entityType": "columns", + "table": "session" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "time_created", + "entityType": "columns", + "table": "session" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "time_updated", + "entityType": "columns", + "table": "session" + }, + { + "type": "integer", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "time_compacting", + "entityType": "columns", + "table": "session" + }, + { + "type": "integer", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "time_archived", + "entityType": "columns", + "table": "session" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "session_id", + "entityType": "columns", + "table": "todo" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "content", + "entityType": "columns", + "table": "todo" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "status", + "entityType": "columns", + "table": "todo" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "priority", + "entityType": "columns", + "table": "todo" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "position", + "entityType": "columns", + "table": "todo" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "time_created", + "entityType": "columns", + "table": "todo" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "time_updated", + "entityType": "columns", + "table": "todo" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "session_id", + "entityType": "columns", + "table": "session_share" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "id", + "entityType": "columns", + "table": "session_share" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "secret", + "entityType": "columns", + "table": "session_share" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "url", + "entityType": "columns", + "table": "session_share" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "time_created", + "entityType": "columns", + "table": "session_share" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "time_updated", + "entityType": "columns", + "table": "session_share" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "aggregate_id", + "entityType": "columns", + "table": "event_sequence" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "seq", + "entityType": "columns", + "table": "event_sequence" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "id", + "entityType": "columns", + "table": "event" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "aggregate_id", + "entityType": "columns", + "table": "event" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "seq", + "entityType": "columns", + "table": "event" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "name", + "entityType": "columns", + "table": "event" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "data", + "entityType": "columns", + "table": "event" + }, + { + "columns": [ + "active_account_id" + ], + "tableTo": "account", + "columnsTo": [ + "id" + ], + "onUpdate": "NO ACTION", + "onDelete": "SET NULL", + "nameExplicit": false, + "name": "fk_account_state_active_account_id_account_id_fk", + "entityType": "fks", + "table": "account_state" + }, + { + "columns": [ + "project_id" + ], + "tableTo": "project", + "columnsTo": [ + "id" + ], + "onUpdate": "NO ACTION", + "onDelete": "CASCADE", + "nameExplicit": false, + "name": "fk_workspace_project_id_project_id_fk", + "entityType": "fks", + "table": "workspace" + }, + { + "columns": [ + "session_id" + ], + "tableTo": "session", + "columnsTo": [ + "id" + ], + "onUpdate": "NO ACTION", + "onDelete": "CASCADE", + "nameExplicit": false, + "name": "fk_message_session_id_session_id_fk", + "entityType": "fks", + "table": "message" + }, + { + "columns": [ + "message_id" + ], + "tableTo": "message", + "columnsTo": [ + "id" + ], + "onUpdate": "NO ACTION", + "onDelete": "CASCADE", + "nameExplicit": false, + "name": "fk_part_message_id_message_id_fk", + "entityType": "fks", + "table": "part" + }, + { + "columns": [ + "project_id" + ], + "tableTo": "project", + "columnsTo": [ + "id" + ], + "onUpdate": "NO ACTION", + "onDelete": "CASCADE", + "nameExplicit": false, + "name": "fk_permission_project_id_project_id_fk", + "entityType": "fks", + "table": "permission" + }, + { + "columns": [ + "project_id" + ], + "tableTo": "project", + "columnsTo": [ + "id" + ], + "onUpdate": "NO ACTION", + "onDelete": "CASCADE", + "nameExplicit": false, + "name": "fk_session_project_id_project_id_fk", + "entityType": "fks", + "table": "session" + }, + { + "columns": [ + "session_id" + ], + "tableTo": "session", + "columnsTo": [ + "id" + ], + "onUpdate": "NO ACTION", + "onDelete": "CASCADE", + "nameExplicit": false, + "name": "fk_todo_session_id_session_id_fk", + "entityType": "fks", + "table": "todo" + }, + { + "columns": [ + "session_id" + ], + "tableTo": "session", + "columnsTo": [ + "id" + ], + "onUpdate": "NO ACTION", + "onDelete": "CASCADE", + "nameExplicit": false, + "name": "fk_session_share_session_id_session_id_fk", + "entityType": "fks", + "table": "session_share" + }, + { + "columns": [ + "aggregate_id" + ], + "tableTo": "event_sequence", + "columnsTo": [ + "aggregate_id" + ], + "onUpdate": "NO ACTION", + "onDelete": "CASCADE", + "nameExplicit": false, + "name": "fk_event_aggregate_id_event_sequence_aggregate_id_fk", + "entityType": "fks", + "table": "event" + }, + { + "columns": [ + "email", + "url" + ], + "nameExplicit": false, + "name": "control_account_pk", + "entityType": "pks", + "table": "control_account" + }, + { + "columns": [ + "session_id", + "position" + ], + "nameExplicit": false, + "name": "todo_pk", + "entityType": "pks", + "table": "todo" + }, + { + "columns": [ + "id" + ], + "nameExplicit": false, + "name": "account_state_pk", + "table": "account_state", + "entityType": "pks" + }, + { + "columns": [ + "id" + ], + "nameExplicit": false, + "name": "account_pk", + "table": "account", + "entityType": "pks" + }, + { + "columns": [ + "id" + ], + "nameExplicit": false, + "name": "workspace_pk", + "table": "workspace", + "entityType": "pks" + }, + { + "columns": [ + "id" + ], + "nameExplicit": false, + "name": "project_pk", + "table": "project", + "entityType": "pks" + }, + { + "columns": [ + "id" + ], + "nameExplicit": false, + "name": "message_pk", + "table": "message", + "entityType": "pks" + }, + { + "columns": [ + "id" + ], + "nameExplicit": false, + "name": "part_pk", + "table": "part", + "entityType": "pks" + }, + { + "columns": [ + "project_id" + ], + "nameExplicit": false, + "name": "permission_pk", + "table": "permission", + "entityType": "pks" + }, + { + "columns": [ + "id" + ], + "nameExplicit": false, + "name": "session_pk", + "table": "session", + "entityType": "pks" + }, + { + "columns": [ + "session_id" + ], + "nameExplicit": false, + "name": "session_share_pk", + "table": "session_share", + "entityType": "pks" + }, + { + "columns": [ + "aggregate_id" + ], + "nameExplicit": false, + "name": "event_sequence_pk", + "table": "event_sequence", + "entityType": "pks" + }, + { + "columns": [ + "id" + ], + "nameExplicit": false, + "name": "event_pk", + "table": "event", + "entityType": "pks" + }, + { + "columns": [ + { + "value": "session_id", + "isExpression": false + } + ], + "isUnique": false, + "where": null, + "origin": "manual", + "name": "message_session_idx", + "entityType": "indexes", + "table": "message" + }, + { + "columns": [ + { + "value": "message_id", + "isExpression": false + } + ], + "isUnique": false, + "where": null, + "origin": "manual", + "name": "part_message_idx", + "entityType": "indexes", + "table": "part" + }, + { + "columns": [ + { + "value": "session_id", + "isExpression": false + } + ], + "isUnique": false, + "where": null, + "origin": "manual", + "name": "part_session_idx", + "entityType": "indexes", + "table": "part" + }, + { + "columns": [ + { + "value": "project_id", + "isExpression": false + } + ], + "isUnique": false, + "where": null, + "origin": "manual", + "name": "session_project_idx", + "entityType": "indexes", + "table": "session" + }, + { + "columns": [ + { + "value": "workspace_id", + "isExpression": false + } + ], + "isUnique": false, + "where": null, + "origin": "manual", + "name": "session_workspace_idx", + "entityType": "indexes", + "table": "session" + }, + { + "columns": [ + { + "value": "parent_id", + "isExpression": false + } + ], + "isUnique": false, + "where": null, + "origin": "manual", + "name": "session_parent_idx", + "entityType": "indexes", + "table": "session" + }, + { + "columns": [ + { + "value": "session_id", + "isExpression": false + } + ], + "isUnique": false, + "where": null, + "origin": "manual", + "name": "todo_session_idx", + "entityType": "indexes", + "table": "todo" + } + ], + "renames": [] +} \ No newline at end of file diff --git a/packages/opencode/src/bus/bus-event.ts b/packages/opencode/src/bus/bus-event.ts index 7fe13833c8..d97922290e 100644 --- a/packages/opencode/src/bus/bus-event.ts +++ b/packages/opencode/src/bus/bus-event.ts @@ -1,10 +1,7 @@ import z from "zod" import type { ZodType } from "zod" -import { Log } from "../util/log" export namespace BusEvent { - const log = Log.create({ service: "event" }) - export type Definition = ReturnType const registry = new Map() diff --git a/packages/opencode/src/cli/cmd/github.ts b/packages/opencode/src/cli/cmd/github.ts index edd9d75610..ed3f482353 100644 --- a/packages/opencode/src/cli/cmd/github.ts +++ b/packages/opencode/src/cli/cmd/github.ts @@ -890,10 +890,10 @@ export const GithubRunCommand = cmd({ } let text = "" - Bus.subscribe(MessageV2.Event.PartUpdated, async (evt) => { - if (evt.properties.part.sessionID !== session.id) return + Bus.subscribe(MessageV2.Event.PartUpdated, (evt) => { + if (evt.properties.data.part.sessionID !== session.id) return //if (evt.properties.part.messageID === messageID) return - const part = evt.properties.part + const part = evt.properties.data.part if (part.type === "tool" && part.state.status === "completed") { const [tool, color] = TOOL[part.tool] ?? [part.tool, UI.Style.TEXT_INFO_BOLD] diff --git a/packages/opencode/src/id/id.ts b/packages/opencode/src/id/id.ts index 6673297cbf..9e324962bf 100644 --- a/packages/opencode/src/id/id.ts +++ b/packages/opencode/src/id/id.ts @@ -3,6 +3,7 @@ import { randomBytes } from "crypto" export namespace Identifier { const prefixes = { + event: "evt", session: "ses", message: "msg", permission: "per", diff --git a/packages/opencode/src/installation/index.ts b/packages/opencode/src/installation/index.ts index 3551c861e4..e678e60023 100644 --- a/packages/opencode/src/installation/index.ts +++ b/packages/opencode/src/installation/index.ts @@ -72,6 +72,10 @@ export namespace Installation { return CHANNEL === "local" } + export function isTesting() { + return process.env.NODE_ENV === "test" + } + export class UpgradeFailedError extends Schema.TaggedErrorClass()("UpgradeFailedError", { stderr: Schema.String, }) {} diff --git a/packages/opencode/src/plugin/index.ts b/packages/opencode/src/plugin/index.ts index 57dcff8f67..9e00802f98 100644 --- a/packages/opencode/src/plugin/index.ts +++ b/packages/opencode/src/plugin/index.ts @@ -3,7 +3,6 @@ import { Config } from "../config/config" import { Bus } from "../bus" import { Log } from "../util/log" import { createOpencodeClient } from "@opencode-ai/sdk" -import { Server } from "../server/server" import { BunProc } from "../bun" import { Flag } from "../flag/flag" import { CodexAuthPlugin } from "./codex" @@ -57,6 +56,8 @@ export namespace Plugin { const hooks: Hooks[] = [] yield* Effect.promise(async () => { + const { Server } = await import("../server/server") + const client = createOpencodeClient({ baseUrl: "http://localhost:4096", directory: ctx.directory, diff --git a/packages/opencode/src/server/projectors.ts b/packages/opencode/src/server/projectors.ts new file mode 100644 index 0000000000..322c0166b6 --- /dev/null +++ b/packages/opencode/src/server/projectors.ts @@ -0,0 +1,6 @@ +import sessionProjectors from "../session/projectors" +import { SyncEvent } from "../sync" + +export function initProjectors() { + SyncEvent.init(sessionProjectors) +} diff --git a/packages/opencode/src/server/routes/session.ts b/packages/opencode/src/server/routes/session.ts index abc820c2af..97f83e3790 100644 --- a/packages/opencode/src/server/routes/session.ts +++ b/packages/opencode/src/server/routes/session.ts @@ -279,14 +279,14 @@ export const SessionRoutes = lazy(() => const sessionID = c.req.valid("param").sessionID const updates = c.req.valid("json") - let session = await Session.get(sessionID) if (updates.title !== undefined) { - session = await Session.setTitle({ sessionID, title: updates.title }) + await Session.setTitle({ sessionID, title: updates.title }) } if (updates.time?.archived !== undefined) { - session = await Session.setArchived({ sessionID, time: updates.time.archived }) + await Session.setArchived({ sessionID, time: updates.time.archived }) } + const session = await Session.get(sessionID) return c.json(session) }, ) diff --git a/packages/opencode/src/server/server.ts b/packages/opencode/src/server/server.ts index 7ead4df8a3..bd03844267 100644 --- a/packages/opencode/src/server/server.ts +++ b/packages/opencode/src/server/server.ts @@ -43,10 +43,13 @@ import { PermissionRoutes } from "./routes/permission" import { GlobalRoutes } from "./routes/global" import { MDNS } from "./mdns" import { lazy } from "@/util/lazy" +import { initProjectors } from "./projectors" // @ts-ignore This global is needed to prevent ai-sdk from logging warnings to stdout https://github.com/vercel/ai/blob/2dc67e0ef538307f21368db32d5a12345d98831b/packages/ai/src/logger/log-warnings.ts#L85 globalThis.AI_SDK_LOG_WARNINGS = false +initProjectors() + export namespace Server { const log = Log.create({ service: "server" }) diff --git a/packages/opencode/src/session/index.ts b/packages/opencode/src/session/index.ts index f2d436ff10..d2aa4deaa8 100644 --- a/packages/opencode/src/session/index.ts +++ b/packages/opencode/src/session/index.ts @@ -9,9 +9,10 @@ import { Config } from "../config/config" import { Flag } from "../flag/flag" import { Installation } from "../installation" -import { Database, NotFoundError, eq, and, or, gte, isNull, desc, like, inArray, lt } from "../storage/db" +import { Database, NotFoundError, eq, and, gte, isNull, desc, like, inArray, lt } from "../storage/db" +import { SyncEvent } from "../sync" import type { SQL } from "../storage/db" -import { SessionTable, MessageTable, PartTable } from "./session.sql" +import { SessionTable } from "./session.sql" import { ProjectTable } from "../project/project.sql" import { Storage } from "@/storage/storage" import { Log } from "../util/log" @@ -182,24 +183,44 @@ export namespace Session { export type GlobalInfo = z.output export const Event = { - Created: BusEvent.define( - "session.created", - z.object({ + Created: SyncEvent.define({ + type: "session.created", + version: "v1", + aggregate: "sessionID", + schema: z.object({ + sessionID: z.string(), info: Info, }), - ), - Updated: BusEvent.define( - "session.updated", - z.object({ + }), + Shared: SyncEvent.define({ + type: "session.shared", + version: "v1", + aggregate: "sessionID", + schema: z.object({ + sessionID: z.string(), + url: z.string().optional(), + }), + }), + Updated: SyncEvent.define({ + type: "session.updated", + version: "v1", + aggregate: "sessionID", + schema: z.object({ + sessionID: z.string(), + info: Info.partial().extend({ + time: Info.shape.time.partial().optional(), + }), + }), + }), + Deleted: SyncEvent.define({ + type: "session.deleted", + version: "v1", + aggregate: "sessionID", + schema: z.object({ + sessionID: z.string(), info: Info, }), - ), - Deleted: BusEvent.define( - "session.deleted", - z.object({ - info: Info, - }), - ), + }), Diff: BusEvent.define( "session.diff", z.object({ @@ -280,18 +301,8 @@ export namespace Session { ) export const touch = fn(SessionID.zod, async (sessionID) => { - const now = Date.now() - Database.use((db) => { - const row = db - .update(SessionTable) - .set({ time_updated: now }) - .where(eq(SessionTable.id, sessionID)) - .returning() - .get() - if (!row) throw new NotFoundError({ message: `Session not found: ${sessionID}` }) - const info = fromRow(row) - Database.effect(() => Bus.publish(Event.Updated, { info })) - }) + const time = Date.now() + SyncEvent.run(Event.Updated, { sessionID, info: { time: { updated: time } } }) }) export async function createNext(input: { @@ -318,22 +329,18 @@ export namespace Session { }, } log.info("created", result) - Database.use((db) => { - db.insert(SessionTable).values(toRow(result)).run() - Database.effect(() => - Bus.publish(Event.Created, { - info: result, - }), - ) - }) + + SyncEvent.run(Event.Created, { sessionID: result.id, info: result }) + const cfg = await Config.get() if (!result.parentID && (Flag.OPENCODE_AUTO_SHARE || cfg.share === "auto")) share(result.id).catch(() => { // Silently ignore sharing errors during session creation }) - Bus.publish(Event.Updated, { - info: result, - }) + // Bus.publish(Event.Updated, { + // id: result.id, + // info: result, + // }) return result } @@ -357,12 +364,9 @@ export namespace Session { } const { ShareNext } = await import("@/share/share-next") const share = await ShareNext.create(id) - Database.use((db) => { - const row = db.update(SessionTable).set({ share_url: share.url }).where(eq(SessionTable.id, id)).returning().get() - if (!row) throw new NotFoundError({ message: `Session not found: ${id}` }) - const info = fromRow(row) - Database.effect(() => Bus.publish(Event.Updated, { info })) - }) + + SyncEvent.run(Event.Shared, { sessionID: id, url: share.url }) + return share }) @@ -370,12 +374,8 @@ export namespace Session { // Use ShareNext to remove the share (same as share function uses ShareNext to create) const { ShareNext } = await import("@/share/share-next") await ShareNext.remove(id) - Database.use((db) => { - const row = db.update(SessionTable).set({ share_url: null }).where(eq(SessionTable.id, id)).returning().get() - if (!row) throw new NotFoundError({ message: `Session not found: ${id}` }) - const info = fromRow(row) - Database.effect(() => Bus.publish(Event.Updated, { info })) - }) + + SyncEvent.run(Event.Shared, { sessionID: id, url: undefined }) }) export const setTitle = fn( @@ -384,18 +384,7 @@ export namespace Session { title: z.string(), }), async (input) => { - return Database.use((db) => { - const row = db - .update(SessionTable) - .set({ title: input.title }) - .where(eq(SessionTable.id, input.sessionID)) - .returning() - .get() - if (!row) throw new NotFoundError({ message: `Session not found: ${input.sessionID}` }) - const info = fromRow(row) - Database.effect(() => Bus.publish(Event.Updated, { info })) - return info - }) + SyncEvent.run(Event.Updated, { sessionID: input.sessionID, info: { title: input.title } }) }, ) @@ -405,18 +394,7 @@ export namespace Session { time: z.number().optional(), }), async (input) => { - return Database.use((db) => { - const row = db - .update(SessionTable) - .set({ time_archived: input.time }) - .where(eq(SessionTable.id, input.sessionID)) - .returning() - .get() - if (!row) throw new NotFoundError({ message: `Session not found: ${input.sessionID}` }) - const info = fromRow(row) - Database.effect(() => Bus.publish(Event.Updated, { info })) - return info - }) + SyncEvent.run(Event.Updated, { sessionID: input.sessionID, info: { time: { archived: input.time } } }) }, ) @@ -426,17 +404,9 @@ export namespace Session { permission: Permission.Ruleset, }), async (input) => { - return Database.use((db) => { - const row = db - .update(SessionTable) - .set({ permission: input.permission, time_updated: Date.now() }) - .where(eq(SessionTable.id, input.sessionID)) - .returning() - .get() - if (!row) throw new NotFoundError({ message: `Session not found: ${input.sessionID}` }) - const info = fromRow(row) - Database.effect(() => Bus.publish(Event.Updated, { info })) - return info + SyncEvent.run(Event.Updated, { + sessionID: input.sessionID, + info: { permission: input.permission, time: { updated: Date.now() } }, }) }, ) @@ -448,42 +418,24 @@ export namespace Session { summary: Info.shape.summary, }), async (input) => { - return Database.use((db) => { - const row = db - .update(SessionTable) - .set({ - revert: input.revert ?? null, - summary_additions: input.summary?.additions, - summary_deletions: input.summary?.deletions, - summary_files: input.summary?.files, - time_updated: Date.now(), - }) - .where(eq(SessionTable.id, input.sessionID)) - .returning() - .get() - if (!row) throw new NotFoundError({ message: `Session not found: ${input.sessionID}` }) - const info = fromRow(row) - Database.effect(() => Bus.publish(Event.Updated, { info })) - return info + SyncEvent.run(Event.Updated, { + sessionID: input.sessionID, + info: { + summary: input.summary, + time: { updated: Date.now() }, + revert: input.revert, + }, }) }, ) export const clearRevert = fn(SessionID.zod, async (sessionID) => { - return Database.use((db) => { - const row = db - .update(SessionTable) - .set({ - revert: null, - time_updated: Date.now(), - }) - .where(eq(SessionTable.id, sessionID)) - .returning() - .get() - if (!row) throw new NotFoundError({ message: `Session not found: ${sessionID}` }) - const info = fromRow(row) - Database.effect(() => Bus.publish(Event.Updated, { info })) - return info + SyncEvent.run(Event.Updated, { + sessionID, + info: { + time: { updated: Date.now() }, + revert: undefined, + }, }) }) @@ -493,22 +445,12 @@ export namespace Session { summary: Info.shape.summary, }), async (input) => { - return Database.use((db) => { - const row = db - .update(SessionTable) - .set({ - summary_additions: input.summary?.additions, - summary_deletions: input.summary?.deletions, - summary_files: input.summary?.files, - time_updated: Date.now(), - }) - .where(eq(SessionTable.id, input.sessionID)) - .returning() - .get() - if (!row) throw new NotFoundError({ message: `Session not found: ${input.sessionID}` }) - const info = fromRow(row) - Database.effect(() => Bus.publish(Event.Updated, { info })) - return info + SyncEvent.run(Event.Updated, { + sessionID: input.sessionID, + info: { + time: { updated: Date.now() }, + summary: input.summary, + }, }) }, ) @@ -662,46 +604,25 @@ export namespace Session { }) export const remove = fn(SessionID.zod, async (sessionID) => { - const project = Instance.project try { const session = await get(sessionID) for (const child of await children(sessionID)) { await remove(child.id) } await unshare(sessionID).catch(() => {}) - // CASCADE delete handles messages and parts automatically - Database.use((db) => { - db.delete(SessionTable).where(eq(SessionTable.id, sessionID)).run() - Database.effect(() => - Bus.publish(Event.Deleted, { - info: session, - }), - ) - }) + + SyncEvent.run(Event.Deleted, { sessionID, info: session }) } catch (e) { log.error(e) } }) export const updateMessage = fn(MessageV2.Info, async (msg) => { - const time_created = msg.time.created - const { id, sessionID, ...data } = msg - Database.use((db) => { - db.insert(MessageTable) - .values({ - id, - session_id: sessionID, - time_created, - data, - }) - .onConflictDoUpdate({ target: MessageTable.id, set: { data } }) - .run() - Database.effect(() => - Bus.publish(MessageV2.Event.Updated, { - info: msg, - }), - ) + SyncEvent.run(MessageV2.Event.Updated, { + sessionID: msg.sessionID, + info: msg, }) + return msg }) @@ -711,17 +632,9 @@ export namespace Session { messageID: MessageID.zod, }), async (input) => { - // CASCADE delete handles parts automatically - Database.use((db) => { - db.delete(MessageTable) - .where(and(eq(MessageTable.id, input.messageID), eq(MessageTable.session_id, input.sessionID))) - .run() - Database.effect(() => - Bus.publish(MessageV2.Event.Removed, { - sessionID: input.sessionID, - messageID: input.messageID, - }), - ) + SyncEvent.run(MessageV2.Event.Removed, { + sessionID: input.sessionID, + messageID: input.messageID, }) return input.messageID }, @@ -734,17 +647,10 @@ export namespace Session { partID: PartID.zod, }), async (input) => { - Database.use((db) => { - db.delete(PartTable) - .where(and(eq(PartTable.id, input.partID), eq(PartTable.session_id, input.sessionID))) - .run() - Database.effect(() => - Bus.publish(MessageV2.Event.PartRemoved, { - sessionID: input.sessionID, - messageID: input.messageID, - partID: input.partID, - }), - ) + SyncEvent.run(MessageV2.Event.PartRemoved, { + sessionID: input.sessionID, + messageID: input.messageID, + partID: input.partID, }) return input.partID }, @@ -753,24 +659,10 @@ export namespace Session { const UpdatePartInput = MessageV2.Part export const updatePart = fn(UpdatePartInput, async (part) => { - const { id, messageID, sessionID, ...data } = part - const time = Date.now() - Database.use((db) => { - db.insert(PartTable) - .values({ - id, - message_id: messageID, - session_id: sessionID, - time_created: time, - data, - }) - .onConflictDoUpdate({ target: PartTable.id, set: { data } }) - .run() - Database.effect(() => - Bus.publish(MessageV2.Event.PartUpdated, { - part: structuredClone(part), - }), - ) + SyncEvent.run(MessageV2.Event.PartUpdated, { + sessionID: part.sessionID, + part: structuredClone(part), + time: Date.now(), }) return part }) diff --git a/packages/opencode/src/session/message-v2.ts b/packages/opencode/src/session/message-v2.ts index f1335f6f21..0efd91a1ca 100644 --- a/packages/opencode/src/session/message-v2.ts +++ b/packages/opencode/src/session/message-v2.ts @@ -6,8 +6,9 @@ import { APICallError, convertToModelMessages, LoadAPIKeyError, type ModelMessag import { LSP } from "../lsp" import { Snapshot } from "@/snapshot" import { fn } from "@/util/fn" -import { Database, NotFoundError, and, desc, eq, inArray, lt, or } from "@/storage/db" -import { MessageTable, PartTable, SessionTable } from "./session.sql" +import { Database, eq, desc, inArray } from "@/storage/db" +import { SyncEvent } from "../sync" +import { MessageTable, PartTable } from "./session.sql" import { ProviderTransform } from "@/provider/transform" import { STATUS_CODES } from "http" import { Storage } from "@/storage/storage" @@ -449,25 +450,34 @@ export namespace MessageV2 { export type Info = z.infer export const Event = { - Updated: BusEvent.define( - "message.updated", - z.object({ + Updated: SyncEvent.define({ + type: "message.updated", + version: "v1", + aggregate: "sessionID", + schema: z.object({ + sessionID: z.string(), info: Info, }), - ), - Removed: BusEvent.define( - "message.removed", - z.object({ + }), + Removed: SyncEvent.define({ + type: "message.removed", + version: "v1", + aggregate: "sessionID", + schema: z.object({ sessionID: SessionID.zod, messageID: MessageID.zod, }), - ), - PartUpdated: BusEvent.define( - "message.part.updated", - z.object({ + }), + PartUpdated: SyncEvent.define({ + type: "message.part.updated", + version: "v1", + aggregate: "sessionID", + schema: z.object({ + sessionID: z.string(), part: Part, + time: z.number(), }), - ), + }), PartDelta: BusEvent.define( "message.part.delta", z.object({ @@ -478,14 +488,16 @@ export namespace MessageV2 { delta: z.string(), }), ), - PartRemoved: BusEvent.define( - "message.part.removed", - z.object({ + PartRemoved: SyncEvent.define({ + type: "message.part.removed", + version: "v1", + aggregate: "sessionID", + schema: z.object({ sessionID: SessionID.zod, messageID: MessageID.zod, partID: PartID.zod, }), - ), + }), } export const WithParts = z.object({ diff --git a/packages/opencode/src/session/projectors.ts b/packages/opencode/src/session/projectors.ts new file mode 100644 index 0000000000..1dcffebbf5 --- /dev/null +++ b/packages/opencode/src/session/projectors.ts @@ -0,0 +1,135 @@ +import { NotFoundError, eq, and } from "../storage/db" +import { SyncEvent } from "@/sync" +import { Session } from "./index" +import { MessageV2 } from "./message-v2" +import { SessionTable, MessageTable, PartTable } from "./session.sql" +import { ProjectTable } from "../project/project.sql" + +export type DeepPartial = { + [K in keyof T]?: T[K] extends object ? DeepPartial : T[K] +} + +function grab( + obj: T, + field1: K1, + cb?: (val: NonNullable) => X, +): X | undefined { + if (obj == undefined || !(field1 in obj)) return undefined + + const val = obj[field1] + if (val && typeof val === "object" && cb) { + return cb(val) + } + return (val === undefined ? null : val) as X | undefined +} + +export function toPartialRow(info: DeepPartial) { + const obj = { + id: grab(info, "id"), + project_id: grab(info, "projectID"), + workspace_id: grab(info, "workspaceID"), + parent_id: grab(info, "parentID"), + slug: grab(info, "slug"), + directory: grab(info, "directory"), + title: grab(info, "title"), + version: grab(info, "version"), + share_url: grab(info, "share", (v) => grab(v, "url")), + summary_additions: grab(info, "summary", (v) => grab(v, "additions")), + summary_deletions: grab(info, "summary", (v) => grab(v, "deletions")), + summary_files: grab(info, "summary", (v) => grab(v, "files")), + summary_diffs: grab(info, "summary", (v) => grab(v, "diffs")), + revert: grab(info, "revert"), + permission: grab(info, "permission"), + time_created: grab(info, "time", (v) => grab(v, "created")), + time_updated: grab(info, "time", (v) => grab(v, "updated")), + time_compacting: grab(info, "time", (v) => grab(v, "compacting")), + time_archived: grab(info, "time", (v) => grab(v, "archived")), + } + + return Object.fromEntries(Object.entries(obj).filter(([_, val]) => val !== undefined)) +} + +export default [ + SyncEvent.project(Session.Event.Created, (db, data) => { + const existing = db + .select({ id: ProjectTable.id }) + .from(ProjectTable) + .where(eq(ProjectTable.id, data.info.projectID)) + .get() + if (!existing) { + // Create a (temporary) project to make this work. In the future + // we should separate sessions and projects + db.insert(ProjectTable) + .values({ + id: data.info.projectID, + worktree: data.info.directory, + sandboxes: [], + }) + .run() + } + + db.insert(SessionTable).values(Session.toRow(data.info)).run() + }), + + SyncEvent.project(Session.Event.Updated, (db, data) => { + const info = data.info + const row = db.update(SessionTable).set(toPartialRow(info)).where(eq(SessionTable.id, data.sessionID)).returning().get() + if (!row) throw new NotFoundError({ message: `Session not found: ${data.sessionID}` }) + }), + + SyncEvent.project(Session.Event.Shared, (db, data) => { + const row = db + .update(SessionTable) + .set({ share_url: data.url }) + .where(eq(SessionTable.id, data.sessionID)) + .returning() + .get() + if (!row) throw new NotFoundError({ message: `Session not found: ${data.sessionID}` }) + }), + + SyncEvent.project(Session.Event.Deleted, (db, data) => { + db.delete(SessionTable).where(eq(SessionTable.id, data.sessionID)).run() + }), + + SyncEvent.project(MessageV2.Event.Updated, (db, data) => { + const time_created = data.info.time.created + const { id, sessionID, ...rest } = data.info + + db.insert(MessageTable) + .values({ + id, + session_id: sessionID, + time_created, + data: rest, + }) + .onConflictDoUpdate({ target: MessageTable.id, set: { data: rest } }) + .run() + }), + + SyncEvent.project(MessageV2.Event.Removed, (db, data) => { + db.delete(MessageTable) + .where(and(eq(MessageTable.id, data.messageID), eq(MessageTable.session_id, data.sessionID))) + .run() + }), + + SyncEvent.project(MessageV2.Event.PartRemoved, (db, data) => { + db.delete(PartTable) + .where(and(eq(PartTable.id, data.partID), eq(PartTable.session_id, data.sessionID))) + .run() + }), + + SyncEvent.project(MessageV2.Event.PartUpdated, (db, data) => { + const { id, messageID, sessionID, ...rest } = data.part + + db.insert(PartTable) + .values({ + id, + message_id: messageID, + session_id: sessionID, + time_created: data.time, + data: rest, + }) + .onConflictDoUpdate({ target: PartTable.id, set: { data: rest } }) + .run() + }), +] diff --git a/packages/opencode/src/session/revert.ts b/packages/opencode/src/session/revert.ts index c5c9edbbdf..6df8b3d53f 100644 --- a/packages/opencode/src/session/revert.ts +++ b/packages/opencode/src/session/revert.ts @@ -4,8 +4,7 @@ import { Snapshot } from "../snapshot" import { MessageV2 } from "./message-v2" import { Session } from "." import { Log } from "../util/log" -import { Database, eq } from "../storage/db" -import { MessageTable, PartTable } from "./session.sql" +import { SyncEvent } from "../sync" import { Storage } from "@/storage/storage" import { Bus } from "../bus" import { SessionPrompt } from "./prompt" @@ -113,8 +112,10 @@ export namespace SessionRevert { remove.push(msg) } for (const msg of remove) { - Database.use((db) => db.delete(MessageTable).where(eq(MessageTable.id, msg.info.id)).run()) - await Bus.publish(MessageV2.Event.Removed, { sessionID: sessionID, messageID: msg.info.id }) + SyncEvent.run(MessageV2.Event.Removed, { + sessionID: sessionID, + messageID: msg.info.id, + }) } if (session.revert.partID && target) { const partID = session.revert.partID @@ -124,8 +125,7 @@ export namespace SessionRevert { const removeParts = target.parts.slice(removeStart) target.parts = preserveParts for (const part of removeParts) { - Database.use((db) => db.delete(PartTable).where(eq(PartTable.id, part.id)).run()) - await Bus.publish(MessageV2.Event.PartRemoved, { + SyncEvent.run(MessageV2.Event.PartRemoved, { sessionID: sessionID, messageID: target.info.id, partID: part.id, diff --git a/packages/opencode/src/share/share-next.ts b/packages/opencode/src/share/share-next.ts index e331e8fc6a..c40f8e3628 100644 --- a/packages/opencode/src/share/share-next.ts +++ b/packages/opencode/src/share/share-next.ts @@ -66,38 +66,37 @@ export namespace ShareNext { export async function init() { if (disabled) return Bus.subscribe(Session.Event.Updated, async (evt) => { - await sync(evt.properties.info.id, [ + const session = await Session.get(evt.properties.data.sessionID) + + await sync(session.id, [ { type: "session", - data: evt.properties.info, + data: session, }, ]) }) Bus.subscribe(MessageV2.Event.Updated, async (evt) => { - await sync(evt.properties.info.sessionID, [ + const info = evt.properties.data.info + await sync(info.sessionID, [ { type: "message", - data: evt.properties.info, + data: evt.properties.data.info, }, ]) - if (evt.properties.info.role === "user") { - await sync(evt.properties.info.sessionID, [ + if (info.role === "user") { + await sync(info.sessionID, [ { type: "model", - data: [ - await Provider.getModel(evt.properties.info.model.providerID, evt.properties.info.model.modelID).then( - (m) => m, - ), - ], + data: [await Provider.getModel(info.model.providerID, info.model.modelID).then((m) => m)], }, ]) } }) Bus.subscribe(MessageV2.Event.PartUpdated, async (evt) => { - await sync(evt.properties.part.sessionID, [ + await sync(evt.properties.data.part.sessionID, [ { type: "part", - data: evt.properties.part, + data: evt.properties.data.part, }, ]) }) diff --git a/packages/opencode/src/storage/db.ts b/packages/opencode/src/storage/db.ts index 1bb8c1a69b..bd2411fecc 100644 --- a/packages/opencode/src/storage/db.ts +++ b/packages/opencode/src/storage/db.ts @@ -27,16 +27,24 @@ export const NotFoundError = NamedError.create( const log = Log.create({ service: "db" }) export namespace Database { - export const Path = iife(() => { + export function getChannelPath() { if (Flag.OPENCODE_DB) { if (path.isAbsolute(Flag.OPENCODE_DB)) return Flag.OPENCODE_DB return path.join(Global.Path.data, Flag.OPENCODE_DB) } - const channel = Installation.CHANNEL - if (["latest", "beta"].includes(channel) || Flag.OPENCODE_DISABLE_CHANNEL_DB) - return path.join(Global.Path.data, "opencode.db") - const safe = channel.replace(/[^a-zA-Z0-9._-]/g, "-") - return path.join(Global.Path.data, `opencode-${safe}.db`) + + // const channel = Installation.CHANNEL + // if (["latest", "beta"].includes(channel) || Flag.OPENCODE_DISABLE_CHANNEL_DB) + // return path.join(Global.Path.data, "opencode.db") + // const safe = channel.replace(/[^a-zA-Z0-9._-]/g, "-") + // return path.join(Global.Path.data, `opencode-${safe}.db`) + } + + export const Path = iife(() => { + if (Installation.isTesting()) { + return ":memory:" + } + return getChannelPath() }) export type Transaction = SQLiteTransaction<"sync", void> @@ -145,19 +153,37 @@ export namespace Database { } } - export function transaction(callback: (tx: TxOrDb) => T): T { + type NotPromise = T extends Promise ? never : T + + export function _transaction( + callback: (tx: TxOrDb) => NotPromise, + options?: { + behavior?: "deferred" | "immediate" | "exclusive" + }, + ): NotPromise { try { return callback(ctx.use().tx) } catch (err) { if (err instanceof Context.NotFound) { const effects: (() => void | Promise)[] = [] - const result = (Client().transaction as any)((tx: TxOrDb) => { - return ctx.provide({ tx, effects }, () => callback(tx)) - }) + const result = Client().transaction( + (tx: TxOrDb) => { + return ctx.provide({ tx, effects }, () => callback(tx)) + }, + { behavior: options?.behavior }, + ) for (const effect of effects) effect() - return result + return result as NotPromise } throw err } } + + export function transaction(callback: (tx: TxOrDb) => NotPromise) { + return _transaction(callback) + } + + export function immediateTransaction(callback: (tx: TxOrDb) => NotPromise) { + return _transaction(callback, { behavior: "immediate" }) + } } diff --git a/packages/opencode/src/sync/event.sql.ts b/packages/opencode/src/sync/event.sql.ts new file mode 100644 index 0000000000..d17dabd72d --- /dev/null +++ b/packages/opencode/src/sync/event.sql.ts @@ -0,0 +1,16 @@ +import { sqliteTable, text, integer } from "drizzle-orm/sqlite-core" + +export const EventSequenceTable = sqliteTable("event_sequence", { + aggregate_id: text().notNull().primaryKey(), + seq: integer().notNull(), +}) + +export const EventTable = sqliteTable("event", { + id: text().primaryKey(), + aggregate_id: text() + .notNull() + .references(() => EventSequenceTable.aggregate_id, { onDelete: "cascade" }), + seq: integer().notNull(), + name: text().notNull(), + data: text({ mode: "json" }).$type>().notNull(), +}) diff --git a/packages/opencode/src/sync/index.ts b/packages/opencode/src/sync/index.ts new file mode 100644 index 0000000000..b4b6a4a049 --- /dev/null +++ b/packages/opencode/src/sync/index.ts @@ -0,0 +1,159 @@ +import z from "zod" +import type { ZodObject } from "zod" +import { Identifier } from "@/id/id" +import { BusEvent } from "@/bus/bus-event" +import { Database, eq } from "@/storage/db" +import { Bus } from "@/bus" +import { EventSequenceTable, EventTable } from "./event.sql" + +export namespace SyncEvent { + export type Definition = { + type: string + properties: ZodObject<{ id: z.ZodString; seq: z.ZodNumber; aggregateID: z.ZodString; data: z.ZodObject }> + version: string + aggregate: string + } + + export type Event = { + id: string + seq: number + aggregateID: string + data: z.infer["data"] + } + + export type SerializedEvent = Event & { type: string } + + type ProjectorFunc = (db: Database.TxOrDb, data: unknown) => void + + const registry = new Map() + let projectors: Map | undefined + + export function init(pjs: Array<[Definition, ProjectorFunc]>) { + projectors = new Map(pjs) + } + + export function versionedName(type: string, version?: string) { + return version ? `${type}.${version}` : type + } + + export function define< + Type extends string, + Version extends string, + Agg extends string, + Schema extends ZodObject>, + >(input: { type: Type; version: Version; aggregate: Agg; schema: Schema }) { + const def = { + type: input.type, + properties: z.object({ + id: Identifier.schema("event"), + seq: z.number(), + aggregateID: z.string(), + data: input.schema, + }), + version: input.version, + aggregate: input.aggregate, + } + + registry.set(versionedName(def.type, def.version), def) + BusEvent.define(versionedName(def.type, def.version), def.properties) + + return def + } + + export function project( + def: Def, + func: (db: Database.TxOrDb, data: Event["data"]) => void, + ): [Definition, ProjectorFunc] { + return [def, func as ProjectorFunc] + } + + function process(def: Def, input: Event) { + if (projectors == null) { + throw new Error("No projectors available. Call `SyncEvent.init` to install projectors") + } + + const projector = projectors.get(def) + if (!projector) { + throw new Error(`Projector not found for event: ${def.type}`) + } + + // idempotent: need to ignore any events already logged + + Database.transaction((tx) => { + projector(tx, input.data) + tx.insert(EventSequenceTable) + .values({ + aggregate_id: input.aggregateID, + seq: input.seq, + }) + .onConflictDoUpdate({ + target: EventSequenceTable.aggregate_id, + set: { seq: input.seq }, + }) + .run() + tx.insert(EventTable) + .values({ + id: input.id, + seq: input.seq, + aggregate_id: input.aggregateID, + name: versionedName(def.type, def.version), + data: input.data as Record, + }) + .run() + }) + } + + // TODO: + // + // * Support applying multiple events at one time. One transaction, + // and it validets all the sequence ids + // * when loading events from db, apply zod validation to ensure shape + + export function replay(event: SerializedEvent) { + const def = registry.get(event.type) + if (!def) { + throw new Error(`Unknown event type: ${event.type}`) + } + + const row = Database.use((db) => + db + .select({ seq: EventSequenceTable.seq }) + .from(EventSequenceTable) + .where(eq(EventSequenceTable.aggregate_id, event.aggregateID)) + .get(), + ) + + const expected = row ? row.seq + 1 : 0 + if (event.seq !== expected) { + throw new Error(`Sequence mismatch for aggregate "${event.aggregateID}": expected ${expected}, got ${event.seq}`) + } + + process(def, event) + } + + export function run(def: Def, data: Event["data"]) { + const agg = (data as Record)[def.aggregate] + // This should never happen: we've enforced it via typescript in + // the definition + if (agg == null) { + throw new Error(`SyncEvent: "${def.aggregate}" required but not found: ${JSON.stringify(data)}`) + } + + Database.immediateTransaction((tx) => { + const id = Identifier.ascending("workspace") + const row = tx + .select({ seq: EventSequenceTable.seq }) + .from(EventSequenceTable) + .where(eq(EventSequenceTable.aggregate_id, agg)) + .get() + const seq = row?.seq != null ? row.seq + 1 : 0 + + process(def, { id, seq, aggregateID: agg, data }) + + Database.effect(() => { + const versionedDef = { ...def, type: versionedName(def.type, def.version) } + Bus.publish(versionedDef, { id, seq, aggregateID: agg, data } as z.output) + }) + }) + } +} diff --git a/packages/opencode/src/test-projection.ts b/packages/opencode/src/test-projection.ts new file mode 100644 index 0000000000..8d3629ca2a --- /dev/null +++ b/packages/opencode/src/test-projection.ts @@ -0,0 +1,83 @@ +import os from "os" +import path from "path" +import fs from "fs/promises" + +// Set XDG env vars BEFORE any src/ imports to isolate from real data +const dir = path.join(os.tmpdir(), "opencode-test-projection-" + process.pid) +await fs.mkdir(dir, { recursive: true }) +process.env["XDG_DATA_HOME"] = path.join(dir, "share") +process.env["XDG_CACHE_HOME"] = path.join(dir, "cache") +process.env["XDG_CONFIG_HOME"] = path.join(dir, "config") +process.env["XDG_STATE_HOME"] = path.join(dir, "state") + +// Write the cache version file +const cache = path.join(dir, "cache", "opencode") +await fs.mkdir(cache, { recursive: true }) +await fs.writeFile(path.join(cache, "version"), "14") + +// Now safe to import src/ +const { Log } = await import("@/util/log") +Log.init({ print: true, dev: true, level: "DEBUG" }) + +const { Instance } = await import("@/project/instance") +const { Database } = await import("@/storage/db") +const { GlobalBus } = await import("@/bus/global") +const { Bus } = await import("@/bus") +const { Session } = await import("@/session") +const { Server } = await import("@/server/server") +const { SessionPrompt } = await import("@/session/prompt") + +async function wait(ms: number) { + return new Promise((resolve) => { + setTimeout(resolve, ms) + }) +} + +async function run() { + console.log("project id:", Instance.project.id) + + // start the server + const server = Server.listen({ + port: 0, + hostname: "127.0.0.1", + }) + console.log("server listening on:", server.url.toString()) + console.log("SSE endpoint:", `${server.url}event`) + + const base = server.url.toString().replace(/\/$/, "") + console.log("\nServer running. Try:") + console.log(` curl -N ${base}/event`) + console.log("\nPress Ctrl+C to stop.\n") + + while (1) { + await wait(5000) + + const session = await Session.create({ + title: "test session", + }) + console.log("created session:", session.id, session.title) + + // send messages to the session + async function prompt(text: string) { + console.log(`\n--- sending: "${text}" ---`) + await SessionPrompt.prompt({ + sessionID: session.id, + parts: [{ type: "text", text }], + }) + console.log(`--- done: "${text}" ---`) + } + + await prompt("What is 2 + 2?") + await wait(2500) + await prompt("Now multiply that by 10") + await wait(2500) + await prompt("Summarize what we've discussed") + } + + await new Promise(() => {}) +} + +await Instance.provide({ + directory: "~/tmp/project-test7", + fn: run, +}) diff --git a/packages/opencode/src/test-replicate.ts b/packages/opencode/src/test-replicate.ts new file mode 100644 index 0000000000..87c68f0c33 --- /dev/null +++ b/packages/opencode/src/test-replicate.ts @@ -0,0 +1,82 @@ +import os from "os" +import path from "path" +import fs from "fs/promises" + +// Set XDG env vars BEFORE any src/ imports to isolate from real data +// const dir = path.join("/Users/james/tmp/opencode-test-replicate") +const dir = path.join(os.tmpdir(), "opencode-test-projection-" + process.pid) +await fs.mkdir(dir, { recursive: true }) +console.log(dir) +process.env["XDG_DATA_HOME"] = path.join(dir, "share") +process.env["XDG_CACHE_HOME"] = path.join(dir, "cache") +process.env["XDG_CONFIG_HOME"] = path.join(dir, "config") +process.env["XDG_STATE_HOME"] = path.join(dir, "state") + +// Write the cache version file +const cache = path.join(dir, "cache", "opencode") +await fs.mkdir(cache, { recursive: true }) +await fs.writeFile(path.join(cache, "version"), "14") + +// Now safe to import src/ +const { Log } = await import("@/util/log") +Log.init({ print: true, dev: true, level: "DEBUG" }) + +const { Database } = await import("@/storage/db") +const { SyncEvent } = await import("@/sync") +const { parseSSE } = await import("@/control-plane/sse") + +const url = process.argv[2] || "http://127.0.0.1:4096/global/event" +const ac = new AbortController() + +process.on("SIGINT", () => ac.abort()) +process.on("SIGTERM", () => ac.abort()) + +async function run() { + const res = await fetch(url, { + headers: { accept: "text/event-stream" }, + signal: ac.signal, + }) + + if (!res.ok) { + console.error("failed to connect:", res.status, await res.text()) + process.exit(1) + } + + if (!res.body) { + console.error("no response body") + process.exit(1) + } + + console.log("connected, listening for events...\n") + const { default: sessionProjectors } = await import("@/session/projectors") + SyncEvent.init(sessionProjectors) + + Database.Client() + + await parseSSE(res.body, ac.signal, (event: any) => { + // console.log("[sse]", JSON.stringify(event, null, 2)) + const payload = event.payload + if (payload.type && payload.properties && payload.properties.data) { + try { + SyncEvent.replay({ + id: payload.properties.id, + type: payload.type, + seq: payload.properties.seq, + aggregateID: payload.properties.aggregateId, + data: payload.properties.data, + }) + + // console.log("[apply] ok:", event.type) + // console.log("db path", Database.Path) + } catch (err) { + console.error("[apply] error:", err) + } + } + }) + + console.log("\ndisconnected") + Database.close() + await fs.rm(dir, { recursive: true, force: true }) +} + +run() diff --git a/packages/opencode/test/session/session.test.ts b/packages/opencode/test/session/session.test.ts index 2332586223..742886dad1 100644 --- a/packages/opencode/test/session/session.test.ts +++ b/packages/opencode/test/session/session.test.ts @@ -20,7 +20,7 @@ describe("session.started event", () => { const unsub = Bus.subscribe(Session.Event.Created, (event) => { eventReceived = true - receivedInfo = event.properties.info as Session.Info + receivedInfo = event.properties.data.info as Session.Info }) const session = await Session.create({}) @@ -95,7 +95,7 @@ describe("step-finish token propagation via Bus event", () => { let received: MessageV2.Part | undefined const unsub = Bus.subscribe(MessageV2.Event.PartUpdated, (event) => { - received = event.properties.part + received = event.properties.data.part }) const tokens = { diff --git a/packages/opencode/test/skill/skill.test.ts b/packages/opencode/test/skill/skill.test.ts index 12e16f86a1..26df70b62c 100644 --- a/packages/opencode/test/skill/skill.test.ts +++ b/packages/opencode/test/skill/skill.test.ts @@ -50,6 +50,7 @@ Instructions here. directory: tmp.path, fn: async () => { const skills = await Skill.all() + console.log(skills) expect(skills.length).toBe(1) const testSkill = skills.find((s) => s.name === "test-skill") expect(testSkill).toBeDefined() diff --git a/packages/opencode/test/sync/index.test.ts b/packages/opencode/test/sync/index.test.ts new file mode 100644 index 0000000000..8b11202332 --- /dev/null +++ b/packages/opencode/test/sync/index.test.ts @@ -0,0 +1,162 @@ +import { describe, test, expect, beforeEach } from "bun:test" +import { tmpdir } from "../fixture/fixture" +import z from "zod" +import { Bus } from "../../src/bus" +import { Instance } from "../../src/project/instance" +import { SyncEvent } from "../../src/sync" +import { Database } from "../../src/storage/db" +import { EventTable } from "../../src/sync/event.sql" +import { Identifier } from "../../src/id/id" + +beforeEach(() => { + Database.Client.reset() +}) + +function withInstance(fn: () => void | Promise) { + return async () => { + await using tmp = await tmpdir() + + await Instance.provide({ + directory: tmp.path, + fn: async () => { + await fn() + }, + }) + } +} + +describe("SyncEvent", () => { + const Created = SyncEvent.define({ + type: "item.created", + version: "v1", + aggregate: "id", + schema: z.object({ id: z.string(), name: z.string() }), + }) + const Sent = SyncEvent.define({ + type: "item.sent", + version: "v1", + aggregate: "item_id", + schema: z.object({ item_id: z.string(), to: z.string() }), + }) + + SyncEvent.init([SyncEvent.project(Created, () => {}), SyncEvent.project(Sent, () => {})]) + + describe("run", () => { + test( + "inserts event row", + withInstance(() => { + SyncEvent.run(Created, { id: "msg_1", name: "first" }) + const rows = Database.use((db) => db.select().from(EventTable).all()) + expect(rows).toHaveLength(1) + expect(rows[0].name).toBe("item.created.v1") + expect(rows[0].aggregate_id).toBe("msg_1") + }), + ) + + test( + "increments seq per aggregate", + withInstance(() => { + SyncEvent.run(Created, { id: "msg_1", name: "first" }) + SyncEvent.run(Created, { id: "msg_1", name: "second" }) + const rows = Database.use((db) => db.select().from(EventTable).all()) + expect(rows).toHaveLength(2) + expect(rows[1].seq).toBe(rows[0].seq + 1) + }), + ) + + test( + "uses custom aggregate field from agg()", + withInstance(() => { + SyncEvent.run(Sent, { item_id: "msg_1", to: "james" }) + const rows = Database.use((db) => db.select().from(EventTable).all()) + expect(rows).toHaveLength(1) + expect(rows[0].aggregate_id).toBe("msg_1") + }), + ) + + test( + "emits events", + withInstance(async () => { + const events: Array<{ + type: string + properties: { seq: number; aggregateID: string; data: { id: string; name: string } } + }> = [] + const unsub = Bus.subscribeAll((event) => events.push(event)) + + SyncEvent.run(Created, { id: "msg_1", name: "test" }) + + expect(events).toHaveLength(1) + expect(events[0]).toEqual({ + type: "item.created.v1", + properties: { + seq: 0, + aggregateID: "msg_1", + data: { + id: "msg_1", + name: "test", + }, + }, + }) + + unsub() + }), + ) + }) + + describe("replay", () => { + test( + "inserts event from external payload", + withInstance(() => { + const id = Identifier.descending("message") + SyncEvent.replay({ + id: "evt_1", + type: "item.created.v1", + seq: 0, + aggregateID: id, + data: { id, name: "replayed" }, + }) + const rows = Database.use((db) => db.select().from(EventTable).all()) + expect(rows).toHaveLength(1) + expect(rows[0].aggregate_id).toBe(id) + }), + ) + + test( + "throws on sequence mismatch", + withInstance(() => { + const id = Identifier.descending("message") + SyncEvent.replay({ + id: "evt_1", + type: "item.created.v1", + seq: 0, + aggregateID: id, + data: { id, name: "first" }, + }) + expect(() => + SyncEvent.replay({ + id: "evt_1", + type: "item.created.v1", + seq: 5, + aggregateID: id, + data: { id, name: "bad" }, + }), + ).toThrow(/Sequence mismatch/) + }), + ) + + test( + "throws on unknown event type", + withInstance(() => { + expect(() => + SyncEvent.replay({ + id: "evt_1", + type: "unknown.event.1", + seq: 0, + aggregateID: "x", + data: {}, + }), + ).toThrow(/Unknown event type/) + }), + ) + }) +})