wip: fixing up vaults agent handlers

Author: tegefaulkes

package-lock.json

@@ -70,7 +70,7 @@
     "@matrixai/resources": "^1.1.5",
     "@matrixai/timer": "^1.1.1",
     "@matrixai/workers": "^1.3.7",
-    "@matrixai/quic": "^0.0.13",
+    "@matrixai/quic": "^0.0.14",
     "@peculiar/asn1-pkcs8": "^2.3.0",
     "@peculiar/asn1-schema": "^2.3.0",
     "@peculiar/asn1-x509": "^2.3.0",

package.json

@@ -12,7 +12,12 @@ import type {
   VaultsGitPackGetMessage,
   VaultsScanMessage,
 } from './types';
-import { DuplexCaller, ServerCaller, UnaryCaller } from '../../rpc/callers';
+import {
+  DuplexCaller,
+  RawCaller,
+  ServerCaller,
+  UnaryCaller,
+} from '../../rpc/callers';
 
 const nodesClaimsGet = new ServerCaller<
   AgentRPCRequestParams<ClaimIdMessage>,
@@ -39,10 +44,7 @@ const notificationsSend = new UnaryCaller<
   AgentRPCResponseResult
 >();
 
-const vaultsGitInfoGet = new ServerCaller<
-  AgentRPCRequestParams<VaultsGitInfoGetMessage>,
-  AgentRPCResponseResult<VaultInfo | GitPackMessage>
->();
+const vaultsGitInfoGet = new RawCaller();
 
 const vaultsGitPackGet = new ServerCaller<
   AgentRPCRequestParams<VaultsGitPackGetMessage>,

src/agent/handlers/clientManifest.ts

@@ -41,8 +41,8 @@ const serverManifest = (container: {
     nodesCrossSignClaim: new NodesCrossSignClaimHandler(container),
     nodesHolePunchMessageSend: new NodesHolePunchMessageSendHandler(container),
     notificationsSend: new NotificationsSendHandler(container),
-    VaultsGitInfoGet: new VaultsGitInfoGetHandler(container),
-    VaultsGitPackGet: new VaultsGitPackGetHandler(container),
+    vaultsGitInfoGet: new VaultsGitInfoGetHandler(container),
+    vaultsGitPackGet: new VaultsGitPackGetHandler(container),
     vaultsScan: new VaultsScanHandler(container),
   };
 };

src/agent/handlers/serverManifest.ts

@@ -6,59 +6,56 @@ import type { ACL } from '../../acl';
 import type Logger from '@matrixai/logger';
 import type { VaultsGitInfoGetMessage } from './types';
 import type { VaultAction } from '../../vaults/types';
+import type { JSONRPCRequest } from '@/rpc/types';
+import type { ContextTimed } from '@matrixai/contexts';
+import type { JSONValue } from '@/types';
+import { ReadableStream } from 'stream/web';
 import * as agentErrors from '../errors';
 import * as vaultsUtils from '../../vaults/utils';
 import * as vaultsErrors from '../../vaults/errors';
-import { ServerHandler } from '../../rpc/handlers';
+import { RawHandler } from '../../rpc/handlers';
 import { validateSync } from '../../validation';
-import { matchSync } from '../../utils';
+import { matchSync, never } from '../../utils';
 import * as validationUtils from '../../validation/utils';
 import * as nodesUtils from '../../nodes/utils';
 import * as agentUtils from '../utils';
+import * as utils from '../../utils';
 
-class VaultsGitInfoGetHandler extends ServerHandler<
-  {
-    db: DB;
-    vaultManager: VaultManager;
-    acl: ACL;
-    logger: Logger;
-  },
-  AgentRPCRequestParams<VaultsGitInfoGetMessage>,
-  AgentRPCResponseResult<VaultInfo | GitPackMessage>
-> {
-  public async *handle(
-    input: AgentRPCRequestParams<VaultsGitInfoGetMessage>,
-    _cancel,
-    meta,
-  ): AsyncGenerator<VaultInfo | GitPackMessage> {
+class VaultsGitInfoGetHandler extends RawHandler<{
+  db: DB;
+  vaultManager: VaultManager;
+  acl: ACL;
+  logger: Logger;
+}> {
+  public async handle(
+    input: [JSONRPCRequest, ReadableStream<Uint8Array>],
+    cancel: (reason?: any) => void,
+    meta: Record<string, JSONValue> | undefined,
+    ctx: ContextTimed,
+  ): Promise<[JSONValue, ReadableStream<Uint8Array>]> {
     const { db, vaultManager, acl } = this.container;
-    yield* db.withTransactionG(async function* (
-      tran,
-    ): AsyncGenerator<VaultInfo | GitPackMessage> {
+    const [headerMessage, inputStream] = input;
+    const params = headerMessage.params;
+    if (params == null || !utils.isObject(params)) never();
+    if (
+      !('vaultNameOrId' in params) ||
+      typeof params.vaultNameOrId != 'string'
+    ) {
+      never();
+    }
+    if (!('action' in params) || typeof params.action != 'string') never();
+    const vaultNameOrId = params.vaultNameOrId;
+    const actionType = validationUtils.parseVaultAction(params.action);
+    const data = await db.withTransactionF(async (tran) => {
       const vaultIdFromName = await vaultManager.getVaultId(
-        input.vaultNameOrId,
+        vaultNameOrId,
         tran,
       );
       const vaultId =
-        vaultIdFromName ?? vaultsUtils.decodeVaultId(input.vaultNameOrId);
+        vaultIdFromName ?? vaultsUtils.decodeVaultId(vaultNameOrId);
       if (vaultId == null) {
         throw new vaultsErrors.ErrorVaultsVaultUndefined();
       }
-      const {
-        actionType,
-      }: {
-        actionType: VaultAction;
-      } = validateSync(
-        (keyPath, value) => {
-          return matchSync(keyPath)(
-            [['actionType'], () => validationUtils.parseVaultAction(value)],
-            () => value,
-          );
-        },
-        {
-          actionType: input.action,
-        },
-      );
       const vaultName = (await vaultManager.getVaultMeta(vaultId, tran))
         ?.vaultName;
       if (vaultName == null) {
@@ -85,20 +82,35 @@ class VaultsGitInfoGetHandler extends ServerHandler<
         );
       }
 
-      yield {
-        vaultName: vaultName,
-        vaultIdEncoded: vaultsUtils.encodeVaultId(vaultId),
+      return {
+        vaultId,
+        vaultName,
       };
-      for await (const byte of vaultManager.handleInfoRequest(vaultId, tran)) {
-        if (byte !== null) {
-          yield {
-            chunk: byte.toString('binary'),
-          };
-        } else {
-          return;
+    });
+
+    // TODO: Needs to handle cancellation
+    const stream = new ReadableStream({
+      start: async (controller) => {
+        for await (const buffer of vaultManager.handleInfoRequest(
+          data.vaultId,
+        )) {
+          if (buffer != null) {
+            controller.enqueue(buffer);
+          } else {
+            break;
+          }
         }
-      }
+        controller.close();
+      },
     });
+
+    return [
+      {
+        vaultName: data.vaultName,
+        vaultIdEncoded: vaultsUtils.encodeVaultId(data.vaultId),
+      },
+      stream,
+    ];
   }
 }
 

src/agent/handlers/vaultsGitInfoGet.ts

@@ -24,6 +24,7 @@ import * as rpcUtils from '../rpc/utils';
 import * as keysUtils from '../keys/utils';
 import * as nodesUtils from '../nodes/utils';
 import { never } from '../utils';
+import * as utils from '../utils';
 
 /**
  * Encapsulates the unidirectional client-side connection of one node to another.
@@ -93,9 +94,9 @@ class NodeConnection<M extends ClientManifest> extends EventTarget {
       targetHostname,
       crypto,
       tlsConfig,
+      manifest,
       quicConfig = {},
       quicSocket,
-      manifest,
       logger = new Logger(this.name),
     }: {
       handleStream: (stream: RPCStream<Uint8Array, Uint8Array>) => void;
@@ -105,9 +106,9 @@ class NodeConnection<M extends ClientManifest> extends EventTarget {
       targetHostname?: Hostname;
       crypto: ClientCrypto;
       tlsConfig: TLSConfig;
+      manifest: M;
       quicConfig?: QuicConfig;
       quicSocket?: QUICSocket;
-      manifest: M;
       logger?: Logger;
     },
     @context ctx: ContextTimed,
@@ -140,6 +141,8 @@ class NodeConnection<M extends ClientManifest> extends EventTarget {
         crypto: {
           ops: crypto,
         },
+        reasonToCode: utils.reasonToCode,
+        codeToReason: utils.codeToReason,
         logger: logger.getChild(QUICClient.name),
       },
       ctx,

src/nodes/NodeConnection.ts

@@ -568,7 +568,7 @@ class RPCServer extends EventTarget {
       let handlerResult: [JSONValue | undefined, ReadableStream<Uint8Array>];
       const headerWriter = rpcStream.writable.getWriter();
       try {
-        handlerResult = handler(
+        handlerResult = await handler(
           [headerMessage.value, inputStream],
           rpcStream.cancel,
           rpcStream.meta,

src/rpc/RPCServer.ts

@@ -29,7 +29,7 @@ abstract class RawHandler<
     cancel: (reason?: any) => void,
     meta: Record<string, JSONValue> | undefined,
     ctx: ContextTimed,
-  ): [JSONValue, ReadableStream<Uint8Array>];
+  ): Promise<[JSONValue, ReadableStream<Uint8Array>]>;
 }
 
 abstract class DuplexHandler<

src/rpc/handlers.ts

@@ -429,12 +429,20 @@ function lexiUnpackBuffer(b: Buffer): number {
   return lexi.unpack([...b]);
 }
 
+// TODO: remove this, quick hack to allow errors to jump the network
+const codeMap = new Map<number, any>();
+let code = 1;
+
 const reasonToCode = (_type: 'recv' | 'send', _reason?: any): number => {
-  return 0;
+  codeMap.set(code, _reason);
+  const returnCode = code;
+  code++;
+  return returnCode;
 };
 
 const codeToReason = (type: 'recv' | 'send', code: number): any => {
-  return Error(`${type} ${code}`);
+  const asd = codeMap.get(code);
+  return asd;
 };
 
 export {