refactor: Replace "message" IPC pattern with RpcClient and RpcService

packages/extension/src/background.ts

@@ -1,6 +1,7 @@
+import { isJsonRpcResponse } from '@metamask/utils';
 import type { Json } from '@metamask/utils';
-import { KernelCommandMethod, isKernelCommandReply } from '@ocap/kernel';
-import type { KernelCommand } from '@ocap/kernel';
+import { kernelMethodSpecs } from '@ocap/kernel/rpc';
+import { RpcClient } from '@ocap/rpc-methods';
 import { ChromeRuntimeDuplexStream } from '@ocap/streams/browser';
 import { delay, Logger } from '@ocap/utils';
 
@@ -29,23 +30,23 @@ async function main(): Promise<void> {
     'offscreen',
   );
 
-  /**
-   * Send a command to the offscreen document.
-   *
-   * @param command - The command to send.
-   */
-  const sendClusterCommand = async (command: KernelCommand): Promise<void> => {
-    await offscreenStream.write(command);
+  const rpcClient = new RpcClient(
+    kernelMethodSpecs,
+    async (request) => {
+      await offscreenStream.write(request);
+    },
+    'background:',
+  );
+
+  const ping = async (): Promise<void> => {
+    const result = await rpcClient.call('ping', []);
+    logger.info(result);
   };
 
   // globalThis.kernel will exist due to dev-console.js in background-trusted-prelude.js
   Object.defineProperties(globalThis.kernel, {
     ping: {
-      value: async () =>
-        sendClusterCommand({
-          method: KernelCommandMethod.ping,
-          params: [],
-        }),
+      value: ping,
     },
     sendMessage: {
       value: async (message: Json) => await offscreenStream.write(message),
@@ -55,29 +56,16 @@ async function main(): Promise<void> {
 
   // With this we can click the extension action button to wake up the service worker.
   chrome.action.onClicked.addListener(() => {
-    sendClusterCommand({
-      method: KernelCommandMethod.ping,
-      params: [],
-    }).catch(logger.error);
+    ping().catch(logger.error);
   });
 
   // Handle replies from the offscreen document
   for await (const message of offscreenStream) {
-    if (!isKernelCommandReply(message)) {
+    if (!isJsonRpcResponse(message)) {
       logger.error('Background received unexpected message', message);
       continue;
     }
-
-    switch (message.method) {
-      case KernelCommandMethod.ping:
-        logger.info('Background received ping reply', message.params);
-        break;
-      default:
-        logger.error(
-          // @ts-expect-error Compile-time exhaustiveness check
-          `Background received unexpected command method: "${message.method.valueOf()}"`,
-        );
-    }
+    rpcClient.handleResponse(message.id as string, message);
   }
 
   throw new Error('Offscreen connection closed unexpectedly');

packages/extension/src/iframe.ts

@@ -1,10 +1,10 @@
-import { isVatCommand, VatSupervisor } from '@ocap/kernel';
-import type { VatCommand, VatCommandReply } from '@ocap/kernel';
+import { VatSupervisor } from '@ocap/kernel';
 import {
   MessagePortDuplexStream,
   receiveMessagePort,
 } from '@ocap/streams/browser';
-import { Logger } from '@ocap/utils';
+import type { JsonRpcMessage } from '@ocap/utils';
+import { isJsonRpcMessage, Logger } from '@ocap/utils';
 
 const logger = new Logger('iframe');
 
@@ -14,13 +14,13 @@ main().catch(logger.error);
  * The main function for the iframe.
  */
 async function main(): Promise<void> {
-  const commandStream = await receiveMessagePort(
+  const kernelStream = await receiveMessagePort(
     (listener) => addEventListener('message', listener),
     (listener) => removeEventListener('message', listener),
   ).then(async (port) =>
-    MessagePortDuplexStream.make<VatCommand, VatCommandReply>(
+    MessagePortDuplexStream.make<JsonRpcMessage, JsonRpcMessage>(
       port,
-      isVatCommand,
+      isJsonRpcMessage,
     ),
   );
 
@@ -30,7 +30,7 @@ async function main(): Promise<void> {
   // eslint-disable-next-line no-new
   new VatSupervisor({
     id: vatId,
-    commandStream,
+    kernelStream,
   });
 
   logger.info('VatSupervisor initialized with vatId:', vatId);

packages/extension/src/kernel-integration/VatWorkerClient.test.ts

@@ -10,15 +10,6 @@ import { describe, it, expect, beforeEach, vi } from 'vitest';
 import type { VatWorkerClientStream } from './VatWorkerClient.ts';
 import { ExtensionVatWorkerClient } from './VatWorkerClient.ts';
 
-vi.mock('@ocap/kernel', async () => ({
-  isVatCommandReply: vi.fn(() => true),
-  VatWorkerServiceCommandMethod: {
-    launch: 'launch',
-    terminate: 'terminate',
-    terminateAll: 'terminateAll',
-  },
-}));
-
 vi.mock('@ocap/streams/browser', async (importOriginal) => {
   // eslint-disable-next-line @typescript-eslint/no-shadow
   const { TestDuplexStream } = await import('@ocap/test-utils/streams');
@@ -109,15 +100,15 @@ describe('ExtensionVatWorkerClient', () => {
       await expect(resultP).rejects.toThrow('foo');
     });
 
-    it('calls logger.error when receiving an unexpected reply', async () => {
-      const errorSpy = vi.spyOn(clientLogger, 'error');
+    it('calls logger.debug when receiving an unexpected reply', async () => {
+      const debugSpy = vi.spyOn(clientLogger, 'debug');
       const unexpectedReply = makeNullReply('m9');
 
       await stream.receiveInput(unexpectedReply);
       await delay(10);
 
-      expect(errorSpy).toHaveBeenCalledOnce();
-      expect(errorSpy).toHaveBeenLastCalledWith(
+      expect(debugSpy).toHaveBeenCalledOnce();
+      expect(debugSpy).toHaveBeenLastCalledWith(
         'Received response with unexpected id "m9".',
       );
     });

packages/extension/src/kernel-integration/VatWorkerClient.ts

@@ -4,15 +4,8 @@ import type {
   JsonRpcRequest,
   JsonRpcResponse,
 } from '@metamask/utils';
-import { isVatCommandReply } from '@ocap/kernel';
-import type {
-  VatWorkerManager,
-  VatId,
-  VatConfig,
-  VatCommand,
-  VatCommandReply,
-} from '@ocap/kernel';
-import { vatWorkerService } from '@ocap/kernel/rpc';
+import type { VatWorkerManager, VatId, VatConfig } from '@ocap/kernel';
+import { vatWorkerServiceMethodSpecs } from '@ocap/kernel/rpc';
 import { RpcClient } from '@ocap/rpc-methods';
 import type { DuplexStream } from '@ocap/streams';
 import {
@@ -23,8 +16,8 @@ import type {
   PostMessageEnvelope,
   PostMessageTarget,
 } from '@ocap/streams/browser';
-import type { Logger } from '@ocap/utils';
-import { makeLogger, stringify } from '@ocap/utils';
+import type { JsonRpcMessage, Logger } from '@ocap/utils';
+import { isJsonRpcMessage, makeLogger, stringify } from '@ocap/utils';
 
 // Appears in the docs.
 // eslint-disable-next-line @typescript-eslint/no-unused-vars
@@ -40,7 +33,7 @@ export class ExtensionVatWorkerClient implements VatWorkerManager {
 
   readonly #stream: VatWorkerClientStream;
 
-  readonly #rpcClient: RpcClient<typeof vatWorkerService.methodSpecs>;
+  readonly #rpcClient: RpcClient<typeof vatWorkerServiceMethodSpecs>;
 
   readonly #portMap: Map<JsonRpcId, MessagePort | undefined>;
 
@@ -66,7 +59,7 @@ export class ExtensionVatWorkerClient implements VatWorkerManager {
     this.#portMap = new Map();
     this.#logger = logger ?? makeLogger('[vat worker client]');
     this.#rpcClient = new RpcClient(
-      vatWorkerService.methodSpecs,
+      vatWorkerServiceMethodSpecs,
       async (request) => {
         if (request.method === 'launch') {
           this.#portMap.set(request.id, undefined);
@@ -112,7 +105,7 @@ export class ExtensionVatWorkerClient implements VatWorkerManager {
   async launch(
     vatId: VatId,
     vatConfig: VatConfig,
-  ): Promise<DuplexStream<VatCommandReply, VatCommand>> {
+  ): Promise<DuplexStream<JsonRpcMessage, JsonRpcMessage>> {
     const [id] = await this.#rpcClient.callAndGetId('launch', {
       vatId,
       vatConfig,
@@ -124,9 +117,9 @@ export class ExtensionVatWorkerClient implements VatWorkerManager {
       );
     }
     this.#portMap.delete(id);
-    return await MessagePortDuplexStream.make<VatCommandReply, VatCommand>(
+    return await MessagePortDuplexStream.make<JsonRpcMessage, JsonRpcMessage>(
       port,
-      isVatCommandReply,
+      isJsonRpcMessage,
     );
   }
 

packages/extension/src/kernel-integration/VatWorkerServer.ts

@@ -9,7 +9,7 @@ import type {
 import { VatAlreadyExistsError, VatNotFoundError } from '@ocap/errors';
 import type { VatId, VatConfig } from '@ocap/kernel';
 import type { VatWorkerServiceMethod } from '@ocap/kernel/rpc';
-import { vatWorkerService } from '@ocap/kernel/rpc';
+import { vatWorkerServiceMethodSpecs } from '@ocap/kernel/rpc';
 import type { ExtractParams } from '@ocap/rpc-methods';
 import { PostMessageDuplexStream } from '@ocap/streams/browser';
 import type {
@@ -105,7 +105,7 @@ export class ExtensionVatWorkerService {
   }
 
   #assertHasMethod(method: string): asserts method is VatWorkerServiceMethod {
-    if (!hasProperty(vatWorkerService.methodSpecs, method)) {
+    if (!hasProperty(vatWorkerServiceMethodSpecs, method)) {
       throw rpcErrors.methodNotFound();
     }
   }
@@ -115,9 +115,9 @@ export class ExtensionVatWorkerService {
     params: unknown,
   ): asserts params is ExtractParams<
     Method,
-    typeof vatWorkerService.methodSpecs
+    typeof vatWorkerServiceMethodSpecs
   > {
-    vatWorkerService.methodSpecs[method].params.assert(params);
+    vatWorkerServiceMethodSpecs[method].params.assert(params);
   }
 
   async #handleMessage(event: MessageEvent<JsonRpcRequest>): Promise<void> {

packages/extension/src/kernel-integration/handlers/clear-state.ts

@@ -4,7 +4,7 @@ import type { Kernel } from '@ocap/kernel';
 import type { MethodSpec, Handler } from '@ocap/rpc-methods';
 import { EmptyJsonArray } from '@ocap/utils';
 
-export const clearStateSpec: MethodSpec<'clearState', Json[], null> = {
+export const clearStateSpec: MethodSpec<'clearState', Json[], Promise<null>> = {
   method: 'clearState',
   params: EmptyJsonArray,
   result: literal(null),
@@ -15,7 +15,7 @@ export type ClearStateHooks = { kernel: Pick<Kernel, 'reset'> };
 export const clearStateHandler: Handler<
   'clearState',
   Json[],
-  null,
+  Promise<null>,
   ClearStateHooks
 > = {
   ...clearStateSpec,

packages/extension/src/kernel-integration/handlers/collect-garbage.test.ts

@@ -0,0 +1,34 @@
+import type { Kernel } from '@ocap/kernel';
+import { describe, it, expect, vi, beforeEach } from 'vitest';
+
+import { collectGarbageHandler } from './collect-garbage.ts';
+
+describe('collectGarbageHandler', () => {
+  let mockKernel: Kernel;
+
+  beforeEach(() => {
+    mockKernel = {
+      collectGarbage: vi.fn(),
+    } as unknown as Kernel;
+  });
+
+  it('collects garbage', () => {
+    const result = collectGarbageHandler.implementation(
+      { kernel: mockKernel },
+      [],
+    );
+
+    expect(mockKernel.collectGarbage).toHaveBeenCalledTimes(1);
+    expect(result).toBeNull();
+  });
+
+  it('should propagate errors from collectGarbage', async () => {
+    const error = new Error('Collect garbage failed');
+    vi.mocked(mockKernel.collectGarbage).mockImplementationOnce(() => {
+      throw error;
+    });
+    expect(() =>
+      collectGarbageHandler.implementation({ kernel: mockKernel }, []),
+    ).toThrow(error);
+  });
+});

packages/extension/src/kernel-integration/handlers/collect-garbage.ts

@@ -23,7 +23,7 @@ export const collectGarbageHandler: Handler<
 > = {
   ...collectGarbageSpec,
   hooks: { kernel: true },
-  implementation: async ({ kernel }): Promise<null> => {
+  implementation: ({ kernel }) => {
     kernel.collectGarbage();
     return null;
   },

packages/extension/src/kernel-integration/handlers/execute-db-query.ts

@@ -4,7 +4,7 @@ import type { MethodSpec, Handler } from '@ocap/rpc-methods';
 export const executeDBQuerySpec: MethodSpec<
   'executeDBQuery',
   { sql: string },
-  Record<string, string>[]
+  Promise<Record<string, string>[]>
 > = {
   method: 'executeDBQuery',
   params: object({
@@ -20,7 +20,7 @@ export type ExecuteDBQueryHooks = {
 export const executeDBQueryHandler: Handler<
   'executeDBQuery',
   { sql: string },
-  Record<string, string>[],
+  Promise<Record<string, string>[]>,
   ExecuteDBQueryHooks
 > = {
   ...executeDBQuerySpec,

packages/extension/src/kernel-integration/handlers/get-status.test.ts

@@ -16,25 +16,22 @@ describe('getStatusHandler', () => {
     });
   });
 
-  it('should return vats status and cluster config', async () => {
+  it('should return vats status and cluster config', () => {
     vi.mocked(mockKernel.getVats).mockReturnValueOnce([]);
 
-    const result = await getStatusHandler.implementation(
-      { kernel: mockKernel },
-      [],
-    );
+    const result = getStatusHandler.implementation({ kernel: mockKernel }, []);
 
     expect(mockKernel.getVats).toHaveBeenCalledTimes(1);
     expect(result).toStrictEqual({ vats: [], clusterConfig: { foo: 'bar' } });
   });
 
-  it('should propagate errors from getVats', async () => {
+  it('should propagate errors from getVats', () => {
     const error = new Error('Status check failed');
     vi.mocked(mockKernel.getVats).mockImplementationOnce(() => {
       throw error;
     });
-    await expect(
+    expect(() =>
       getStatusHandler.implementation({ kernel: mockKernel }, []),
-    ).rejects.toThrow(error);
+    ).toThrow(error);
   });
 });

packages/extension/src/kernel-integration/handlers/get-status.ts

@@ -43,9 +43,7 @@ export const getStatusHandler: Handler<
 > = {
   ...getStatusSpec,
   hooks: { kernel: true },
-  implementation: async ({
-    kernel,
-  }: GetStatusHooks): Promise<KernelStatus> => ({
+  implementation: ({ kernel }: GetStatusHooks): KernelStatus => ({
     vats: kernel.getVats(),
     clusterConfig: kernel.clusterConfig,
   }),