feat: websocket server proxying to provider websocket

apps/api/examples/lease-flow.ts

@@ -2,7 +2,7 @@
  * This script demonstrates how to create a deployment with a lease using the API and an API key.
  *
  * The script follows these steps:
- * 1. Creates a certificate for secure communication
+ * 1. Checks if API key env var is set
  * 2. Creates a deployment using the provided SDL file
  * 3. Waits for and collects bids from providers
  * 4. Creates a lease using the first received bid
@@ -23,6 +23,7 @@ import { config } from "@dotenvx/dotenvx";
 import axios from "axios";
 import * as fs from "node:fs";
 import * as path from "node:path";
+import WebSocket from "ws";
 
 // Load environment variables from .env.local in the script directory
 const envPath = path.resolve(__dirname, ".env.local");
@@ -68,7 +69,7 @@ async function waitForBids(dseq: string, apiKey: string, maxAttempts = 10): Prom
 
 /**
  * This script is used to create a lease for a deployment using an api key.
- * It creates a certificate, creates a deployment, waits for bids, creates a lease, and then closes the deployment.
+ * It creates a deployment, waits for bids, creates a lease, and then closes the deployment.
  */
 async function main() {
   try {
@@ -78,22 +79,7 @@ async function main() {
       throw new Error("API_KEY environment variable is required");
     }
 
-    // 2. Create certificate
-    console.log("Creating certificate...");
-    const certResponse = await api.post(
-      "/v1/certificates",
-      {},
-      {
-        headers: {
-          "x-api-key": apiKey
-        }
-      }
-    );
-
-    const { certPem, encryptedKey } = certResponse.data.data;
-    console.log("Certificate created successfully");
-
-    // 3. Create deployment
+    // 2. Create deployment
     console.log("Creating deployment...");
     const deployResponse = await api.post(
       "/v1/deployments",
@@ -113,7 +99,7 @@ async function main() {
     const { dseq, manifest } = deployResponse.data.data;
     console.log(`Deployment created with dseq: ${dseq}`);
 
-    // 4. Wait for and get bids
+    // 3. Wait for and get bids
     console.log("Waiting for bids...");
     const bids = await waitForBids(dseq, apiKey);
     console.log(`Received ${bids.length} bids`);
@@ -126,23 +112,21 @@ async function main() {
       throw new Error(`No bid found from provider ${targetProvider}`);
     }
 
+    const { provider, gseq, oseq } = selectedBid.bid.bid_id;
+
     const body = {
       manifest,
-      certificate: {
-        certPem,
-        keyPem: encryptedKey
-      },
       leases: [
         {
           dseq,
-          gseq: selectedBid.bid.bid_id.gseq,
-          oseq: selectedBid.bid.bid_id.oseq,
-          provider: selectedBid.bid.bid_id.provider
+          gseq,
+          oseq,
+          provider
         }
       ]
     };
 
-    // 5. Create lease and send manifest
+    // 4. Create lease and send manifest
     console.log("Creating lease and sending manifest...");
     const leaseResponse = await api.post("/v1/leases", body, {
       headers: {
@@ -155,7 +139,7 @@ async function main() {
     }
     console.log("Lease created successfully", JSON.stringify(leaseResponse.data.data, null, 2));
 
-    // 6. Deposit into deployment
+    // 5. Deposit into deployment
     console.log("Depositing into deployment...");
     const depositResponse = await api.post(
       `/v1/deposit-deployment`,
@@ -182,11 +166,7 @@ async function main() {
       `/v1/deployments/${dseq}`,
       {
         data: {
-          sdl: updatedYml,
-          certificate: {
-            certPem,
-            keyPem: encryptedKey
-          }
+          sdl: updatedYml
         }
       },
       {
@@ -201,7 +181,7 @@ async function main() {
     }
     console.log("Deployment updated successfully");
 
-    // 7. Get the deployment details
+    // 6. Get the deployment details
     console.log("Getting deployment details...");
     const deploymentResponse = await api.get(`/v1/deployments/${dseq}`, {
       headers: {
@@ -211,6 +191,39 @@ async function main() {
 
     console.log("Deployment details:", JSON.stringify(deploymentResponse.data.data, null, 2));
 
+    // 7. Stream logs from provider
+    const providerResponse = await api.get(`/v1/providers/${provider}`, {
+      headers: {
+        "x-api-key": apiKey
+      }
+    });
+    const { hostUri } = providerResponse.data;
+
+    const websocket = new WebSocket(`${API_URL}/v1/ws`, {
+      headers: {
+        "x-api-key": apiKey
+      }
+    });
+
+    websocket.on("message", message => {
+      console.log("WebSocket message received:", message.toString());
+    });
+
+    websocket.on("open", () => {
+      console.log("WebSocket connected, sending message to stream logs");
+      websocket.send(
+        JSON.stringify({
+          type: "websocket",
+          providerAddress: provider,
+          url: `${hostUri}/lease/${dseq}/${gseq}/${oseq}/logs`,
+          chainNetwork: "sandbox"
+        })
+      );
+    });
+
+    // wait for 5 seconds before closing the deployment
+    await new Promise(resolve => setTimeout(resolve, 5000));
+
     // 8. Close deployment
     console.log("Closing deployment...");
     const closeResponse = await api.delete(`/v1/deployments/${dseq}`, {

apps/api/package.json

@@ -42,6 +42,7 @@
     "@akashnetwork/database": "*",
     "@akashnetwork/env-loader": "*",
     "@akashnetwork/http-sdk": "*",
+    "@akashnetwork/jwt": "*",
     "@akashnetwork/logging": "*",
     "@akashnetwork/net": "*",
     "@akashnetwork/react-query-sdk": "*",
@@ -57,6 +58,7 @@
     "@cosmjs/tendermint-rpc": "^0.32.4",
     "@dotenvx/dotenvx": "^1.9.0",
     "@hono/node-server": "1.13.7",
+    "@hono/node-ws": "^1.2.0",
     "@hono/otel": "~0.4.0",
     "@hono/swagger-ui": "0.4.1",
     "@hono/zod-openapi": "0.18.4",
@@ -92,6 +94,7 @@
     "markdown-to-txt": "^2.0.1",
     "memory-cache": "^0.2.0",
     "murmurhash": "^2.0.1",
+    "node-forge": "^1.3.1",
     "pg": "^8.12.0",
     "pg-boss": "^10.3.2",
     "pg-hstore": "^2.3.4",
@@ -125,9 +128,11 @@
     "@types/memory-cache": "^0.2.2",
     "@types/node": "^22.13.11",
     "@types/node-fetch": "^2.6.2",
+    "@types/node-forge": "^1.3.11",
     "@types/pg": "^8.11.6",
     "@types/semver": "^7.5.2",
     "@types/uuid": "^8.3.1",
+    "@types/ws": "^8.18.1",
     "@typescript-eslint/eslint-plugin": "^7.12.0",
     "alias-hq": "^5.1.6",
     "copy-webpack-plugin": "^12.0.2",
@@ -151,6 +156,7 @@
     "ts-loader": "^9.5.2",
     "type-fest": "^4.26.1",
     "typescript": "~5.8.2",
+    "wait-for-expect": "^4.0.0",
     "webpack": "^5.91.0",
     "webpack-cli": "4.10.0",
     "webpack-node-externals": "^3.0.0"

apps/api/src/app.ts

@@ -4,11 +4,13 @@ import "./app/providers/jobs.provider";
 import { LoggerService } from "@akashnetwork/logging";
 import { HttpLoggerIntercepter } from "@akashnetwork/logging/hono";
 import { serve } from "@hono/node-server";
+import { createNodeWebSocket } from "@hono/node-ws";
 import { otel } from "@hono/otel";
 import { swaggerUI } from "@hono/swagger-ui";
 import { Hono } from "hono";
 import { cors } from "hono/cors";
 import once from "lodash/once";
+import type { AddressInfo } from "net";
 import { container } from "tsyringe";
 
 import { AuthInterceptor } from "@src/auth/services/auth.interceptor";
@@ -39,6 +41,7 @@ import { userRouter } from "./routers/userRouter";
 import { web3IndexRouter } from "./routers/web3indexRouter";
 import { env } from "./utils/env";
 import { bytesToHumanReadableSize } from "./utils/files";
+import { initLeaseWebsocketRoute } from "./websocket/routes/websocket/websocket.router";
 import { addressRouter } from "./address";
 import { sendVerificationEmailRouter } from "./auth";
 import {
@@ -163,6 +166,9 @@ for (const handler of openApiHonoHandlers) {
   appHono.route("/", handler);
 }
 
+const { injectWebSocket, upgradeWebSocket } = createNodeWebSocket({ app: appHono });
+initLeaseWebsocketRoute(appHono, upgradeWebSocket);
+
 appHono.route("/", notificationsApiProxy);
 
 appHono.route("/", healthzRouter);
@@ -199,22 +205,31 @@ const appLogger = LoggerService.forContext("APP");
  * Start scheduler
  * Start server
  */
-export async function initApp() {
+export async function initApp(port: number = Number(PORT)): Promise<AppServer> {
   try {
     await Promise.all([initDb(), ...container.resolveAll(APP_INITIALIZER).map(initializer => initializer[ON_APP_START]())]);
     startScheduler();
 
-    appLogger.info({ event: "SERVER_STARTING", url: `http://localhost:${PORT}`, NODE_OPTIONS: process.env.NODE_OPTIONS });
+    appLogger.info({ event: "SERVER_STARTING", url: `http://localhost:${port}`, NODE_OPTIONS: process.env.NODE_OPTIONS });
     const server = serve({
       fetch: appHono.fetch,
-      port: typeof PORT === "string" ? parseInt(PORT, 10) : PORT
+      port: typeof port === "string" ? parseInt(port, 10) : port
     });
+    injectWebSocket(server);
     const shutdown = once(() => shutdownServer(server, appLogger, container.dispose.bind(container)));
 
     process.on("SIGTERM", shutdown);
     process.on("SIGINT", shutdown);
+
+    return {
+      host: `http://localhost:${(server.address() as AddressInfo).port}`,
+      async close() {
+        await shutdown();
+      }
+    };
   } catch (error) {
     appLogger.error({ event: "APP_INIT_ERROR", error });
+    throw error;
   }
 }
 
@@ -239,3 +254,8 @@ export async function initDb() {
 }
 
 export { appHono as app };
+
+export interface AppServer {
+  host: string;
+  close(): Promise<void>;
+}

apps/api/src/billing/lib/wallet/wallet.ts

@@ -46,4 +46,8 @@ export class Wallet implements OfflineDirectSigner {
   async getMnemonic() {
     return (await this.instanceAsPromised).mnemonic;
   }
+
+  async getInstance() {
+    return await this.instanceAsPromised;
+  }
 }

apps/api/src/billing/services/wallet-initializer/wallet-initializer.service.spec.ts

@@ -7,6 +7,7 @@ import { DomainEventsService } from "@src/core/services/domain-events/domain-eve
 import type { FeatureFlagValue } from "@src/core/services/feature-flags/feature-flags";
 import { FeatureFlags } from "@src/core/services/feature-flags/feature-flags";
 import { FeatureFlagsService } from "@src/core/services/feature-flags/feature-flags.service";
+import { JwtTokenService } from "@src/provider/services/jwt-token/jwt-token.service";
 import { UserWalletRepository } from "../../repositories/user-wallet/user-wallet.repository";
 import { ManagedUserWalletService } from "../managed-user-wallet/managed-user-wallet.service";
 import { WalletInitializerService } from "./wallet-initializer.service";
@@ -150,6 +151,12 @@ describe(WalletInitializerService.name, () => {
         isEnabled: jest.fn(flag => !!input?.enabledFeatures?.includes(flag))
       })
     );
+    di.registerInstance(
+      JwtTokenService,
+      mock<JwtTokenService>({
+        generateJwtToken: jest.fn().mockResolvedValue("mock-jwt-token")
+      })
+    );
 
     container.clearInstances();
 

apps/api/src/core/lib/telemetry.ts

@@ -0,0 +1,12 @@
+import type { Span } from "@opentelemetry/api";
+import { context } from "@opentelemetry/api";
+import { trace } from "@opentelemetry/api";
+
+export function traceActiveSpan<T extends (span: Span) => any>(name: string, callback: T): ReturnType<T> {
+  return trace.getTracer("default").startActiveSpan(name, callback);
+}
+
+export function propagateTracingContext<T extends (...args: any[]) => any>(callback: T): T {
+  const currentContext = context.active();
+  return ((...args) => context.with(currentContext, () => callback(...args))) as T;
+}

apps/api/src/deployment/http-schemas/deployment.schema.ts

@@ -101,11 +101,7 @@ export const DepositDeploymentResponseSchema = z.object({
 
 export const UpdateDeploymentRequestSchema = z.object({
   data: z.object({
-    sdl: z.string(),
-    certificate: z.object({
-      certPem: z.string(),
-      keyPem: z.string()
-    })
+    sdl: z.string()
   })
 });
 

apps/api/src/deployment/http-schemas/lease.schema.ts

@@ -2,10 +2,6 @@ import { z } from "zod";
 
 export const CreateLeaseRequestSchema = z.object({
   manifest: z.string(),
-  certificate: z.object({
-    certPem: z.string(),
-    keyPem: z.string()
-  }),
   leases: z.array(
     z.object({
       dseq: z.string(),