fix: fixing vaults handlers

[ci skip]

Author: tegefaulkes

src/agent/handlers/clientManifest.ts

@@ -7,8 +7,6 @@ import type {
   NodeAddressMessage,
   NodeIdMessage,
   SignedNotificationEncoded,
-  VaultInfo,
-  VaultsGitInfoGetMessage,
   VaultsGitPackGetMessage,
   VaultsScanMessage,
 } from './types';

src/agent/handlers/types.ts

@@ -55,4 +55,6 @@ export type GitPackMessage = {
 
 export type VaultsGitPackGetMessage = {
   body: string;
+  nameOrId: VaultIdEncoded | VaultName;
+  vaultAction: VaultAction;
 };

src/agent/handlers/vaultsGitInfoGet.ts

@@ -1,21 +1,16 @@
-import type { GitPackMessage, VaultInfo } from './types';
-import type { AgentRPCRequestParams, AgentRPCResponseResult } from '../types';
 import type { DB } from '@matrixai/db';
 import type { VaultManager } from '../../vaults';
 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 { JSONRPCRequest } from '../../rpc/types';
 import type { ContextTimed } from '@matrixai/contexts';
-import type { JSONValue } from '@/types';
+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 { RawHandler } from '../../rpc/handlers';
-import { validateSync } from '../../validation';
-import { matchSync, never } from '../../utils';
+import { never } from '../../utils';
 import * as validationUtils from '../../validation/utils';
 import * as nodesUtils from '../../nodes/utils';
 import * as agentUtils from '../utils';
@@ -29,12 +24,18 @@ class VaultsGitInfoGetHandler extends RawHandler<{
 }> {
   public async handle(
     input: [JSONRPCRequest, ReadableStream<Uint8Array>],
-    cancel: (reason?: any) => void,
+    _cancel,
     meta: Record<string, JSONValue> | undefined,
-    ctx: ContextTimed,
+    _ctx: ContextTimed, // TODO: use
   ): Promise<[JSONValue, ReadableStream<Uint8Array>]> {
     const { db, vaultManager, acl } = this.container;
     const [headerMessage, inputStream] = input;
+    const readableProm = (async () => {
+      for await (const _ of inputStream) {
+        // Input stream is not used here, wait for finish.
+        // It should be closed by the caller immediately
+      }
+    })();
     const params = headerMessage.params;
     if (params == null || !utils.isObject(params)) never();
     if (
@@ -103,7 +104,7 @@ class VaultsGitInfoGetHandler extends RawHandler<{
         controller.close();
       },
     });
-
+    await readableProm;
     return [
       {
         vaultName: data.vaultName,

src/agent/handlers/vaultsGitPackGet.ts

@@ -14,6 +14,7 @@ import { matchSync } from '../../utils';
 import * as validationUtils from '../../validation/utils';
 import { ServerHandler } from '../../rpc/handlers';
 
+// TODO: This needs to be a raw handler
 class VaultsGitPackGetHandler extends ServerHandler<
   {
     vaultManager: VaultManager;
@@ -34,7 +35,7 @@ class VaultsGitPackGetHandler extends ServerHandler<
       throw new agentErrors.ErrorAgentNodeIdMissing();
     }
     const nodeIdEncoded = nodesUtils.encodeNodeId(requestingNodeId);
-    const nameOrId = meta.get('vaultNameOrId').pop()!.toString();
+    const nameOrId = input.nameOrId;
     yield* db.withTransactionG(async function* (
       tran,
     ): AsyncGenerator<AgentRPCResponseResult<GitPackMessage>> {
@@ -58,7 +59,7 @@ class VaultsGitPackGetHandler extends ServerHandler<
           );
         },
         {
-          actionType: meta.get('vaultAction').pop()!.toString(),
+          actionType: input.vaultAction,
         },
       );
       // Checking permissions

src/nodes/NodeConnection.ts

@@ -66,7 +66,7 @@ class NodeConnection<M extends ClientManifest> extends EventTarget {
       targetHostname,
       tlsConfig,
       connectionKeepAliveIntervalTime,
-      connectionMaxIdleTimeout,
+      connectionMaxIdleTimeout = 60_000,
       quicSocket,
       manifest,
       logger,
@@ -98,7 +98,7 @@ class NodeConnection<M extends ClientManifest> extends EventTarget {
       tlsConfig,
       manifest,
       connectionKeepAliveIntervalTime,
-      connectionMaxIdleTimeout,
+      connectionMaxIdleTimeout = 60_000,
       quicSocket,
       logger = new Logger(this.name),
     }: {

src/nodes/types.ts

@@ -26,7 +26,6 @@ type NodeData = {
 
 type SeedNodes = Record<NodeIdEncoded, NodeAddress>;
 
-
 export type {
   NodeId,
   NodeIdString,

src/rpc/RPCServer.ts

@@ -240,7 +240,7 @@ class RPCServer extends EventTarget {
     handler: DuplexHandlerImplementation<I, O>,
     timeout: number | undefined,
   ): void {
-    const rawSteamHandler: RawHandlerImplementation = (
+    const rawSteamHandler: RawHandlerImplementation = async (
       [header, input],
       cancel,
       meta,

src/rpc/types.ts

@@ -164,7 +164,7 @@ type HandlerImplementation<I, O> = (
 
 type RawHandlerImplementation = HandlerImplementation<
   [JSONRPCRequest, ReadableStream<Uint8Array>],
-  [JSONValue | undefined, ReadableStream<Uint8Array>]
+  Promise<[JSONValue | undefined, ReadableStream<Uint8Array>]>
 >;
 
 type DuplexHandlerImplementation<

src/vaults/VaultInternal.ts

@@ -17,6 +17,7 @@ import type { NodeId, NodeIdEncoded } from '../ids/types';
 import type NodeConnectionManager from '../nodes/NodeConnectionManager';
 import type RPCClient from '../rpc/RPCClient';
 import type { clientManifest as agentClientManifest } from '../agent/handlers/clientManifest';
+import type { POJO } from '../types';
 import path from 'path';
 import git from 'isomorphic-git';
 import Logger from '@matrixai/logger';
@@ -26,11 +27,11 @@ import {
 } from '@matrixai/async-init/dist/CreateDestroyStartStop';
 import { withF, withG } from '@matrixai/resources';
 import { RWLockWriter } from '@matrixai/async-locks';
-import * as utils from '@/utils';
-import * as validationUtils from '@/validation/utils';
 import * as vaultsErrors from './errors';
 import * as vaultsUtils from './utils';
 import { tagLast } from './types';
+import * as validationUtils from '../validation/utils';
+import * as utils from '../utils';
 import * as nodesUtils from '../nodes/utils';
 import { never } from '../utils/utils';
 
@@ -139,7 +140,7 @@ class VaultInternal {
 
     const vaultIdEncoded = vaultsUtils.encodeVaultId(vaultId);
     logger.info(`Cloning ${this.name} - ${vaultIdEncoded}`);
-    const vault = new VaultInternal({
+    const vault = new this({
       vaultId,
       db,
       vaultsDbPath,
@@ -766,13 +767,11 @@ class VaultInternal {
       typeof vaultNameOrId === 'string'
         ? vaultNameOrId
         : vaultsUtils.encodeVaultId(vaultNameOrId);
-    console.log('a');
     const response = await client.methods.vaultsGitInfoGet({
       vaultNameOrId: vaultNameOrId_,
       action: vaultAction,
     });
-    console.log('a');
-    console.log(response.meta);
+    await response.writable.close();
 
     const result = response.meta?.result;
     if (result == null || !utils.isObject(result)) never();
@@ -793,7 +792,6 @@ class VaultInternal {
     for await (const chunk of response.readable) {
       infoResponse.push(chunk);
     }
-    // TODO: complete
     return [
       async function ({
         url,
@@ -818,17 +816,13 @@ class VaultInternal {
           };
         } else if (method === 'POST') {
           const responseBuffers: Array<Uint8Array> = [];
-          const stream = client.methods.vaultsGitPackGet(metadata);
-          const chunk = new vaultsPB.PackChunk();
-          // Body is usually an async generator but in the cases we are using,
-          // only the first value is used
-          chunk.setChunk(body[0]);
-          // Tell the server what commit we need
-          await stream.write(chunk);
-          let packResponse = (await stream.read()).value;
-          while (packResponse != null) {
-            responseBuffers.push(packResponse.getChunk_asU8());
-            packResponse = (await stream.read()).value;
+          const stream = await client.methods.vaultsGitPackGet({
+            body: body[0].toString('binary'),
+            nameOrId: result.vaultIdEncoded as string,
+            vaultAction,
+          });
+          for await (const value of stream) {
+            responseBuffers.push(Buffer.from(value.chunk, 'binary'));
           }
           return {
             url: url,

src/vaults/VaultManager.ts

@@ -796,11 +796,11 @@ class VaultManager {
     tran?: DBTransaction,
   ): AsyncGenerator<Buffer> {
     if (tran == null) {
-      return this.db.withTransactionF(async (tran) =>
-        this.handleInfoRequest(vaultId, tran),
-      );
+      const handleInfoRequest = (tran) => this.handleInfoRequest(vaultId, tran);
+      return yield* this.db.withTransactionG(async function* (tran) {
+        return yield* handleInfoRequest(tran);
+      });
     }
-
     const efs = this.efs;
     const vault = await this.getVault(vaultId, tran);
     return yield* withG(
@@ -986,7 +986,6 @@ class VaultManager {
     if (tran == null) {
       return this.db.withTransactionF((tran) => this.getVault(vaultId, tran));
     }
-
     const vaultIdString = vaultId.toString() as VaultIdString;
     // 1. get the vault, if it exists then return that
     const vault = this.vaultMap.get(vaultIdString);
@@ -1035,7 +1034,6 @@ class VaultManager {
         return [vaultId.toString(), RWLockWriter, 'read'];
       },
     );
-
     // Running the function with locking
     return await this.vaultLocks.withF(...vaultLocks, async () => {
       // Getting the vaults while locked
@@ -1044,7 +1042,7 @@ class VaultManager {
           return await this.getVault(vaultId, tran);
         }),
       );
-      return f(...vaults);
+      return await f(...vaults);
     });
   }