[OGUI-1747] Create simple client p2p connection (#3021)

* implements fetching to other clients via P2P connection and handles listnening on peer-end.

Author: OmegaCreations

Tokenization/backend/proto/wrapper.proto

@@ -25,6 +25,12 @@ service CentralSystem {
     rpc ClientStream(stream Payload) returns (stream Payload);
 }
 
+// Peer2Peer service handling HTTP-like requests between wrapper clients
+service Peer2Peer {
+  rpc Fetch (HttpLikeRequest) returns (HttpLikeResponse);
+}
+
+
 // ======================================
 //                MESSAGES
 // ======================================
@@ -51,14 +57,37 @@ message Payload {
     }
 }
 
+// http method enum
+enum HttpMethod {
+  HTTP_METHOD_UNSPECIFIED = 0;
+  GET = 1; 
+  POST = 2; 
+  PUT = 3; 
+  PATCH = 4; 
+  DELETE = 5; 
+  HEAD = 6; 
+  OPTIONS = 7;
+}
+
+message HttpLikeRequest {
+  HttpMethod method = 1;    // GET/POST/...
+  string path = 2;  // request path e.g. "/orders/add"
+  map<string,string> headers = 3;   // "content-type": "application/json"
+  bytes body = 4;   // body (e.g. JSON)
+}
+
+message HttpLikeResponse {
+  int32 status = 1;
+  map<string,string> headers = 2;
+  bytes body = 3;
+}
+
+
 // ======================================
 //                 ENUMS
 // ======================================
 
 enum MessageEvent {
-    // Unspecified event type
-    MESSAGE_EVENT_UNSPECIFIED = 0;
-
     // Default value, represents an empty event
     MESSAGE_EVENT_EMPTY = 0;
 

Tokenization/backend/wrapper/src/client/commands/newToken/newToken.handler.ts

@@ -46,7 +46,7 @@ export class NewTokenHandler implements CommandHandler<NewTokenCommand> {
 
     for (const dir of directions) {
       let conn = this.manager.getConnectionByAddress(targetAddress, dir);
-      conn ??= this.manager.createNewConnection(targetAddress, dir, token);
+      conn ??= await this.manager.createNewConnection(targetAddress, dir, token);
       conn.token = token;
     }
   }

Tokenization/backend/wrapper/src/client/connection/Connection.ts

@@ -13,7 +13,9 @@
  */
 
 import type { ConnectionDirection } from '../../models/message.model';
+import type { ConnectionHeaders, FetchOptions, FetchResponse } from '../../models/connection.model';
 import { ConnectionStatus } from '../../models/connection.model';
+import * as grpc from '@grpc/grpc-js';
 
 /**
  * @description This class represents a connection to a target client and manages sending messages to it.
@@ -22,6 +24,7 @@ export class Connection {
   private _token: string;
   private _targetAddress: string;
   private _status: ConnectionStatus;
+  private _peerClient: any;
   public direction: ConnectionDirection;
 
   /**
@@ -31,9 +34,10 @@ export class Connection {
    * @param targetAddress - The unique address of the target client.
    * @param direction - The direction of the connection (e.g., sending or receiving).
    */
-  constructor(token: string, targetAddress: string, direction: ConnectionDirection) {
+  constructor(token: string, targetAddress: string, direction: ConnectionDirection, peerCtor: any) {
     this._token = token;
     this._targetAddress = targetAddress;
+    this._peerClient = new peerCtor(targetAddress, grpc.credentials.createInsecure());
     this.direction = direction;
 
     this._status = ConnectionStatus.CONNECTED;
@@ -72,11 +76,63 @@ export class Connection {
     return this._status;
   }
 
+  /**
+   * Sets the status of this connection.
+   * @param status The new status of the connection.
+   */
+  public set status(status: ConnectionStatus) {
+    this._status = status;
+  }
+
   /**
    * Returns target address for this Connection object
    * @returns Target address
    */
   public get targetAddress(): string {
     return this._targetAddress;
   }
+
+  /**
+   * "HTTP-like" fetch via gRPC protocol
+   * @returns Promise with peer's response
+   */
+  public fetch(options: FetchOptions = {}): Promise<FetchResponse> {
+    if (!this._peerClient) {
+      return Promise.reject(new Error(`Peer client not attached for ${this.targetAddress}`));
+    }
+
+    // Build a request object
+    const method = (options.method ?? 'POST').toUpperCase();
+    const path = options.path ?? '/';
+    const headers: ConnectionHeaders = { ...(options.headers ?? {}) };
+
+    let bodyBuf: Buffer = Buffer.alloc(0);
+    const b = options.body;
+    if (b != null) {
+      if (Buffer.isBuffer(b)) bodyBuf = b;
+      else if (b instanceof Uint8Array) bodyBuf = Buffer.from(b);
+      else if (typeof b === 'string') bodyBuf = Buffer.from(b, 'utf8');
+      else return Promise.reject(new Error('Body must be a string/Buffer/Uint8Array'));
+    }
+
+    const req = { method, path, headers, body: bodyBuf };
+
+    // Return promise with response
+    return new Promise<FetchResponse>((resolve, reject) => {
+      this._peerClient.Fetch(req, (err: any, resp: any) => {
+        if (err) return reject(err);
+
+        const resBody = resp?.body ? Buffer.from(resp.body) : Buffer.alloc(0);
+        const fetchResponse: FetchResponse = {
+          status: Number(resp?.status ?? 200),
+          headers: resp?.headers ?? {},
+          body: resBody,
+          text: async () => resBody.toString('utf8'),
+          json: async () => JSON.parse(resBody.toString('utf8')),
+        };
+
+        resolve(fetchResponse);
+      });
+    });
+  }
 }

Tokenization/backend/wrapper/src/client/connectionManager/ConnectionManager.ts

@@ -21,10 +21,9 @@ import { LogManager } from '@aliceo2/web-ui';
 import type { Command, CommandHandler } from 'models/commands.model';
 import type { DuplexMessageEvent } from '../../models/message.model';
 import { ConnectionDirection } from '../../models/message.model';
+import { ConnectionStatus } from '../../models/connection.model';
+import { peerListener } from '../../utils/connection/peerListener';
 
-/**
- * @description Manages all the connection between clients and central system.
- */
 /**
  * Manages the lifecycle and connection logic for a gRPC client communicating with the central system.
  *
@@ -45,6 +44,10 @@ export class ConnectionManager {
   private _centralConnection: CentralConnection;
   private _sendingConnections = new Map<string, Connection>();
   private _receivingConnections = new Map<string, Connection>();
+  private _wrapper: any;
+  private _peerCtor: any;
+  private _peerServer: grpc.Server | undefined;
+  private _baseAPIPath: string = '';
 
   /**
    * Initializes a new instance of the ConnectionManager class.
@@ -64,9 +67,10 @@ export class ConnectionManager {
     });
 
     const proto = grpc.loadPackageDefinition(packageDef) as any;
-    const wrapper = proto.webui.tokenization;
+    this._wrapper = proto.webui.tokenization;
+    this._peerCtor = this._wrapper.Peer2Peer;
 
-    const client = new wrapper.CentralSystem(centralAddress, grpc.credentials.createInsecure());
+    const client = new this._wrapper.CentralSystem(centralAddress, grpc.credentials.createInsecure());
 
     // Event dispatcher for central system events
     this._centralDispatcher = new CentralCommandDispatcher();
@@ -109,13 +113,15 @@ export class ConnectionManager {
    * @param token Optional token for connection
    */
   createNewConnection(address: string, direction: ConnectionDirection, token?: string) {
-    const conn = new Connection(token ?? '', address, direction);
+    const conn = new Connection(token ?? '', address, direction, this._peerCtor);
 
     if (direction === ConnectionDirection.RECEIVING) {
       this._receivingConnections.set(address, conn);
     } else {
       this._sendingConnections.set(address, conn);
     }
+    conn.status = ConnectionStatus.CONNECTED;
+    this._logger.infoMessage(`Connection with ${address} has been estabilished. Status: ${conn.status}`);
 
     return conn;
   }
@@ -150,4 +156,26 @@ export class ConnectionManager {
       receiving: [...this._receivingConnections.values()],
     };
   }
+
+  /** Starts a listener server for p2p connections */
+  public async listenForPeers(port: number, baseAPIPath?: string): Promise<void> {
+    if (baseAPIPath) this._baseAPIPath = baseAPIPath;
+
+    if (this._peerServer) {
+      this._peerServer.forceShutdown();
+      this._peerServer = undefined;
+    }
+
+    this._peerServer = new grpc.Server();
+    this._peerServer.addService(this._wrapper.Peer2Peer.service, {
+      Fetch: async (call: grpc.ServerUnaryCall<any, any>, callback: grpc.sendUnaryData<any>) =>
+        peerListener(call, callback, this._logger, this._receivingConnections, this._peerCtor, this._baseAPIPath),
+    });
+
+    await new Promise<void>((resolve, reject) => {
+      this._peerServer?.bindAsync(`localhost:${port}`, grpc.ServerCredentials.createInsecure(), (err) => (err ? reject(err) : resolve()));
+    });
+
+    this._logger.infoMessage(`Peer server listening on localhost:${port}`);
+  }
 }

Tokenization/backend/wrapper/src/client/gRPCWrapper.ts

@@ -14,9 +14,9 @@
 
 import { ConnectionManager } from './connectionManager/ConnectionManager';
 import { RevokeTokenHandler } from './commands/revokeToken/revokeToken.handler';
-import { DuplexMessageEvent } from '../models/message.model';
-import type { Connection } from './connection/Connection';
+import { ConnectionDirection, DuplexMessageEvent } from '../models/message.model';
 import { NewTokenHandler } from './commands/newToken/newToken.handler';
+import type { Connection } from './connection/Connection';
 
 /**
  * @description Wrapper class for managing secure gRPC wrapper.
@@ -56,12 +56,37 @@ export class gRPCWrapper {
   }
 
   /**
-   * Starts the Connection Manager stream connection with Central System
+   * Connects to the central system using the underlying ConnectionManager.
+   *
+   * @remarks
+   * This method starts the duplex stream connection with the central gRPC server.
    */
   public connectToCentralSystem() {
     this._connectionManager.connectToCentralSystem();
   }
 
+  /**
+   * Establishes a new connection to a target client.
+   *
+   * @param address - The target address of the client.
+   * @param token - Optional authentication token for the connection.
+   *
+   * @returns A promise that resolves to the newly created connection ready to use for fetching data.
+   */
+  public async connectToClient(address: string, token?: string): Promise<Connection> {
+    return this._connectionManager.createNewConnection(address, ConnectionDirection.SENDING, token ?? '');
+  }
+
+  /**
+   * Starts a listener server for p2p connections.
+   * @param port The port number to bind the p2p server to.
+   * @param baseAPIPath Optional base API path to forward requests to e.g. '/api'.
+   * @returns A promise that resolves when the p2p listener server is started.
+   */
+  public async listenForPeers(port: number, baseAPIPath?: string): Promise<void> {
+    return this._connectionManager.listenForPeers(port, baseAPIPath);
+  }
+
   /**
    * Returns all saved connections.
    *

Tokenization/backend/wrapper/src/models/connection.model.ts

@@ -44,3 +44,35 @@ export enum ConnectionStatus {
   // The connection is refreshing its authentication token
   TOKEN_REFRESH = 'TOKEN_REFRESH',
 }
+
+export type ConnectionHeaders = Record<string, string>;
+
+export type FetchOptions = {
+  method?: string;
+  path?: string;
+  headers?: ConnectionHeaders;
+  body?: string | Buffer | Uint8Array | null;
+};
+
+export type FetchResponse = {
+  status: number;
+  headers: ConnectionHeaders;
+  body: Buffer;
+  text: () => Promise<string>;
+  json: () => Promise<any>;
+};
+
+export type HttpLikeRequest = {
+  method: string;
+  path: string;
+  headers: Headers;
+  body: Buffer;
+  correlation_id?: string;
+  sequence_number?: number;
+};
+
+export type HttpLikeResponse = {
+  status: number;
+  headers: Headers;
+  body: Buffer;
+};

Tokenization/backend/wrapper/src/models/message.model.ts

@@ -23,7 +23,6 @@
  * @property MESSAGE_EVENT_REVOKE_TOKEN: Event for revoking an existing token.
  */
 export enum DuplexMessageEvent {
-  MESSAGE_EVENT_UNSPECIFIED = 'MESSAGE_EVENT_UNSPECIFIED',
   MESSAGE_EVENT_EMPTY = 'MESSAGE_EVENT_EMPTY',
   MESSAGE_EVENT_NEW_TOKEN = 'MESSAGE_EVENT_NEW_TOKEN',
   MESSAGE_EVENT_REVOKE_TOKEN = 'MESSAGE_EVENT_REVOKE_TOKEN',

Tokenization/backend/wrapper/src/test/client/commands/newToken.test.ts

@@ -18,6 +18,9 @@ import { Connection } from '../../../client/connection/Connection';
 import { ConnectionManager } from '../../../client/connectionManager/ConnectionManager';
 import { Command } from 'models/commands.model';
 import { ConnectionDirection, DuplexMessageEvent } from '../../../models/message.model';
+import * as grpc from '@grpc/grpc-js';
+import * as protoLoader from '@grpc/proto-loader';
+import path from 'path';
 
 /**
  * Helper to create a new token command with given address, direction, and token.
@@ -36,6 +39,19 @@ function createEventMessage(targetAddress: string, connectionDirection: Connecti
 describe('NewTokenHandler', () => {
   let manager: ConnectionManager;
 
+  const protoPath = path.join(__dirname, '..', '..', '..', '..', '..', 'proto', 'wrapper.proto');
+  const packageDef = protoLoader.loadSync(protoPath, {
+    keepCase: true,
+    longs: String,
+    enums: String,
+    defaults: true,
+    oneofs: true,
+  });
+
+  const proto = grpc.loadPackageDefinition(packageDef) as any;
+  const wrapper = proto.webui.tokenization;
+  const peerCtor = wrapper.Peer2Peer;
+
   beforeEach(() => {
     manager = {
       sendingConnections: new Map<string, Connection>(),
@@ -49,7 +65,7 @@ describe('NewTokenHandler', () => {
         return undefined;
       }),
       createNewConnection: jest.fn(function (this: any, address: string, dir: ConnectionDirection, token: string) {
-        const conn = new Connection(token, address, dir);
+        const conn = new Connection(token, address, dir, peerCtor);
         if (dir === ConnectionDirection.SENDING) {
           this.sendingConnections.set(address, conn);
         } else {
@@ -62,7 +78,7 @@ describe('NewTokenHandler', () => {
 
   it('should update token on existing SENDING connection', async () => {
     const targetAddress = 'peer-123';
-    const conn = new Connection('old-token', targetAddress, ConnectionDirection.SENDING);
+    const conn = new Connection('old-token', targetAddress, ConnectionDirection.SENDING, peerCtor);
     (manager as any).sendingConnections.set(targetAddress, conn);
 
     const handler = new NewTokenHandler(manager);