feat: bridge I/O backend

.env.sample

@@ -6,3 +6,7 @@ DB_CONNECTION_STRING=<DB_CONNECTION_STRING>
 PUBLIC_HOST=<PUBLIC_HOST>
 OSC_ACCESS_TOKEN=<OSC_ACCESS_TOKEN>
 ICE_SERVERS=turn:username:password@turn.example.com,stun:stun.example.com
+WHIP_GATEWAY_URL=<WHIP_GATEWAY_URL>
+WHIP_GATEWAY_API_KEY=<WHIP_GATEWAY_API_KEY>
+WHEP_GATEWAY_URL=<WHEP_GATEWAY_URL>
+WHEP_GATEWAY_API_KEY=<WHEP_GATEWAY_API_KEY>

src/api.test.ts

@@ -1,6 +1,7 @@
 import api from './api';
 import { CoreFunctions } from './api_productions_core_functions';
 import { ConnectionQueue } from './connection_queue';
+import { UserSession } from './models';
 
 jest.mock('./db/interface', () => ({
   getIngests: jest.fn().mockResolvedValue([]),
@@ -34,6 +35,18 @@
   getIngests: jest.fn().mockResolvedValue([]),
   updateIngest: jest.fn().mockResolvedValue(undefined),
   deleteIngest: jest.fn().mockResolvedValue(true),
+  addTransmitter: jest.fn().mockResolvedValue({}),
+  getTransmitter: jest.fn().mockResolvedValue(undefined),
+  getTransmitters: jest.fn().mockResolvedValue([]),
+  getTransmittersLength: jest.fn().mockResolvedValue(0),
+  updateTransmitter: jest.fn().mockResolvedValue(undefined),
+  deleteTransmitter: jest.fn().mockResolvedValue(true),
+  addReceiver: jest.fn().mockResolvedValue({}),
+  getReceiver: jest.fn().mockResolvedValue(undefined),
+  getReceivers: jest.fn().mockResolvedValue([]),
+  getReceiversLength: jest.fn().mockResolvedValue(0),
+  updateReceiver: jest.fn().mockResolvedValue(undefined),
+  deleteReceiver: jest.fn().mockResolvedValue(true),
   saveUserSession: jest.fn().mockResolvedValue(undefined),
   getSession: jest.fn().mockResolvedValue(null),
   deleteUserSession: jest.fn().mockResolvedValue(true),
@@ -66,12 +79,12 @@
   on: jest.fn(),
   once: jest.fn(),
   emit: jest.fn()
 } as any;
 
 const mockIngestManager = {
   load: jest.fn().mockResolvedValue(undefined),
   startPolling: jest.fn()
 } as any;
 
 describe('api', () => {
   it('responds with hello, world!', async () => {
@@ -86,7 +99,9 @@
       coreFunctions: new CoreFunctions(
         mockProductionManager,
         new ConnectionQueue()
-      )
+      ),
+      whipGatewayUrl: '',
+      whepGatewayUrl: ''
     });
     const response = await server.inject({
       method: 'GET',

src/api.ts

@@ -12,6 +12,8 @@ import apiReAuth from './api_re_auth';
 import apiShare from './api_share';
 import apiWhip, { ApiWhipOptions } from './api_whip';
 import apiWhep, { ApiWhepOptions } from './api_whep';
+import apiBridgeTx, { ApiBridgeTxOptions } from './api_bridge_tx';
+import apiBridgeRx, { ApiBridgeRxOptions } from './api_bridge_rx';
 import { DbManager } from './db/interface';
 import { IngestManager } from './ingest_manager';
 import { ProductionManager } from './production_manager';
@@ -55,12 +57,18 @@ export interface ApiGeneralOptions {
   dbManager: DbManager;
   productionManager: ProductionManager;
   ingestManager: IngestManager;
+  whipGatewayUrl?: string;
+  whipGatewayApiKey?: string;
+  whepGatewayUrl?: string;
+  whepGatewayApiKey?: string;
 }
 
 export type ApiOptions = ApiGeneralOptions &
   ApiProductionsOptions &
   ApiWhipOptions &
-  ApiWhepOptions;
+  ApiWhepOptions &
+  ApiBridgeTxOptions &
+  ApiBridgeRxOptions;
 
 export default async (opts: ApiOptions) => {
   const api = fastify({
@@ -96,6 +104,31 @@ export default async (opts: ApiOptions) => {
   });
 
   api.register(healthcheck, { title: opts.title });
+
+  // Bridge configuration endpoint
+  const BridgeConfig = Type.Object({
+    whipGatewayEnabled: Type.Boolean(),
+    whepGatewayEnabled: Type.Boolean()
+  });
+
+  api.get<{ Reply: Static<typeof BridgeConfig> }>(
+    '/api/v1/bridge/config',
+    {
+      schema: {
+        description: 'Get bridge gateway configuration',
+        response: {
+          200: BridgeConfig
+        }
+      }
+    },
+    async (_, reply) => {
+      reply.send({
+        whipGatewayEnabled: !!opts.whipGatewayUrl,
+        whepGatewayEnabled: !!opts.whepGatewayUrl
+      });
+    }
+  );
+
   // register other API routes here
   api.register(getApiProductions(), {
     prefix: 'api/v1',
@@ -114,7 +147,8 @@ export default async (opts: ApiOptions) => {
     coreFunctions: opts.coreFunctions,
     productionManager: opts.productionManager,
     dbManager: opts.dbManager,
-    whipAuthKey: opts.whipAuthKey
+    whipAuthKey: opts.whipAuthKey,
+    whipGatewayUrl: opts.whipGatewayUrl
   });
   api.register(apiWhep, {
     prefix: 'api/v1',
@@ -123,11 +157,30 @@ export default async (opts: ApiOptions) => {
     smbServerBaseUrl: opts.smbServerBaseUrl,
     coreFunctions: opts.coreFunctions,
     productionManager: opts.productionManager,
-    dbManager: opts.dbManager
+    dbManager: opts.dbManager,
+    whepGatewayUrl: opts.whepGatewayUrl
   });
   api.register(apiShare, { publicHost: opts.publicHost, prefix: 'api/v1' });
   api.register(apiReAuth, { prefix: 'api/v1' });
 
+  // Register bridge IO endpoints (only if gateways are configured)
+  if (opts.whipGatewayUrl) {
+    api.register(apiBridgeTx, {
+      prefix: 'api/v1',
+      dbManager: opts.dbManager,
+      whipGatewayUrl: opts.whipGatewayUrl,
+      whipGatewayApiKey: opts.whipGatewayApiKey
+    });
+  }
+  if (opts.whepGatewayUrl) {
+    api.register(apiBridgeRx, {
+      prefix: 'api/v1',
+      dbManager: opts.dbManager,
+      whepGatewayUrl: opts.whepGatewayUrl,
+      whepGatewayApiKey: opts.whepGatewayApiKey
+    });
+  }
+
   api.all('/whip/:productionId/:lineId', async (request, reply) => {
     if (request.method !== 'POST' && request.method !== 'OPTIONS') {
       return reply