feat(livekit): add livekit webhook receiver

client/web/src/utils/global-state-helper.ts

@@ -1,5 +1,5 @@
 import { useEffect } from 'react';
-import { AppStore, AppState, AppSocket, useMemoizedFn } from 'tailchat-shared';
+import { AppStore, AppState, AppSocket, useEvent } from 'tailchat-shared';
 
 let _store: AppStore;
 export function setGlobalStore(store: AppStore) {
@@ -31,7 +31,7 @@ export function useGlobalSocketEvent<T>(
   eventName: string,
   callback: (data: T) => void
 ) {
-  const fn = useMemoizedFn(callback);
+  const fn = useEvent(callback);
 
   useEffect(() => {
     if (_socket) {
@@ -45,3 +45,16 @@ export function useGlobalSocketEvent<T>(
     };
   }, []);
 }
+
+export async function emitGlobalSocketEvent(
+  eventName: string,
+  eventData?: unknown
+): Promise<unknown> {
+  if (!_socket) {
+    throw new Error('socket not inited');
+  }
+
+  const res = await _socket.request(eventName, eventData);
+
+  return res;
+}

pnpm-lock.yaml

@@ -1941,6 +1941,12 @@ importers:
 
   server/plugins/com.msgbyte.livekit:
     dependencies:
+      dotenv:
+        specifier: ^16.3.1
+        version: 16.3.1
+      express:
+        specifier: ^4.18.2
+        version: 4.18.2
       livekit-server-sdk:
         specifier: ^1.2.5
         version: 1.2.5
@@ -1951,12 +1957,18 @@ importers:
         specifier: '*'
         version: link:../../packages/sdk
     devDependencies:
+      '@types/express':
+        specifier: ^4.17.15
+        version: 4.17.17
       '@types/react':
         specifier: 18.0.20
         version: 18.0.20
       mini-star:
         specifier: '*'
         version: 1.3.1
+      ts-node:
+        specifier: 10.9.1
+        version: 10.9.1(@types/node@18.11.9)(typescript@4.9.4)
 
   server/plugins/com.msgbyte.livekit/web/plugins/com.msgbyte.livekit:
     dependencies:
@@ -11964,6 +11976,7 @@ packages:
       '@types/node': 18.11.9
       '@types/qs': 6.9.7
       '@types/range-parser': 1.2.4
+    dev: true
 
   /@types/express-serve-static-core@4.17.33:
     resolution: {integrity: sha512-TPBqmR/HRYI3eC2E5hmiivIzv+bidAfXofM+sbonAGvyDhySGw9/PQZFt2BLOrjUUR++4eJVpx6KnLQK1Fk9tA==}
@@ -11979,6 +11992,7 @@ packages:
       '@types/express-serve-static-core': 4.17.31
       '@types/qs': 6.9.7
       '@types/serve-static': 1.15.0
+    dev: true
 
   /@types/express@4.17.17:
     resolution: {integrity: sha512-Q4FmmuLGBG58btUnfS1c1r/NQdlp3DMfGDGig8WhfpA2YRUtEkxAjkZb0yvplJGYdF1fsQ81iMDcH24sSCNC/Q==}
@@ -12255,6 +12269,7 @@ packages:
 
   /@types/mime@3.0.1:
     resolution: {integrity: sha512-Y4XFY5VJAuw0FgAqPNd6NNoV44jbq9Bz2L7Rh/J6jLTiHBSBJa9fxqQIvkIld4GsoDOcCbvzOUAbLPsSKKg+uA==}
+    dev: true
 
   /@types/min-document@2.19.0:
     resolution: {integrity: sha512-lsYeSW1zfNqHTL1RuaOgfAhoiOWV1RAQDKT0BZ26z4Faz8llVIj1r1ablUo5QY6yzHMketuvu4+N0sv0eZpXTg==}
@@ -12582,6 +12597,7 @@ packages:
     dependencies:
       '@types/mime': 3.0.1
       '@types/node': 18.11.9
+    dev: true
 
   /@types/serve-static@1.15.1:
     resolution: {integrity: sha512-NUo5XNiAdULrJENtJXZZ3fHtfMolzZwczzBbnAeBbqBwG+LaG6YaJtuwzwGSQZ2wsCrxjEhNNjAkKigy3n8teQ==}
@@ -17443,6 +17459,11 @@ packages:
     engines: {node: '>=12'}
     dev: false
 
+  /dotenv@16.3.1:
+    resolution: {integrity: sha512-IPzF4w4/Rd94bA9imS68tZBaYyBWSCE47V1RGuMrB94iyTOIEwRmVL2x/4An+6mETpLrKJ5hQkB8W4kFAadeIQ==}
+    engines: {node: '>=12'}
+    dev: false
+
   /dotenv@8.6.0:
     resolution: {integrity: sha512-IrPdXQsk2BbzvCBGBOTmmSH5SodmqZNt4ERAZDmW4CT+tL8VtvinqywuANaFu4bOMWki16nqf0e4oC0QIaDr/g==}
     engines: {node: '>=10'}
@@ -27020,7 +27041,7 @@ packages:
       '@probot/get-private-key': 1.1.1
       '@probot/octokit-plugin-config': 1.1.6(@octokit/core@3.6.0)
       '@probot/pino': 2.3.5
-      '@types/express': 4.17.15
+      '@types/express': 4.17.17
       '@types/ioredis': 4.28.10
       '@types/pino': 6.3.12
       '@types/pino-http': 5.8.1

server/plugins/com.msgbyte.livekit/package.json

@@ -7,14 +7,19 @@
   "license": "MIT",
   "private": true,
   "scripts": {
+    "dev:webhook": "ts-node ./webhook/index.ts",
     "build:web": "ministar buildPlugin all",
     "build:web:watch": "ministar watchPlugin all"
   },
   "devDependencies": {
+    "@types/express": "^4.17.15",
     "@types/react": "18.0.20",
-    "mini-star": "*"
+    "mini-star": "*",
+    "ts-node": "10.9.1"
   },
   "dependencies": {
+    "dotenv": "^16.3.1",
+    "express": "^4.18.2",
     "livekit-server-sdk": "^1.2.5",
     "long": "^5.2.3",
     "tailchat-server-sdk": "*"

server/plugins/com.msgbyte.livekit/services/livekit.service.dev.ts

@@ -1,7 +1,11 @@
 import type { TcContext } from 'tailchat-server-sdk';
 import { TcService, TcDbService } from 'tailchat-server-sdk';
 import type { LivekitDocument, LivekitModel } from '../models/livekit';
-import { AccessToken, RoomServiceClient } from 'livekit-server-sdk';
+import {
+  AccessToken,
+  RoomServiceClient,
+  WebhookEvent,
+} from 'livekit-server-sdk';
 
 /**
  * livekit
@@ -68,6 +72,7 @@ class LivekitService extends TcService {
         roomName: 'string',
       },
     });
+    this.registerAction('webhook', this.webhook);
   }
 
   async url(ctx: TcContext) {
@@ -120,6 +125,34 @@ class LivekitService extends TcService {
       return [];
     }
   }
+
+  async webhook(ctx: TcContext) {
+    const payload = ctx.params as WebhookEvent;
+
+    if (payload.event === 'participant_joined') {
+      const room = payload.room;
+      const [groupId, panelId] = room.name.split('#');
+
+      this.roomcastNotify(ctx, groupId, 'participantJoined', {
+        groupId,
+        panelId,
+        participant: payload.participant,
+      });
+
+      return;
+    } else if (payload.event === 'participant_left') {
+      const room = payload.room;
+      const [groupId, panelId] = room.name.split('#');
+
+      this.roomcastNotify(ctx, groupId, 'participantLeft', {
+        groupId,
+        panelId,
+        participant: payload.participant,
+      });
+
+      return;
+    }
+  }
 }
 
 export default LivekitService;

server/plugins/com.msgbyte.livekit/web/plugins/com.msgbyte.livekit/src/group/LivekitPanelBadge.tsx

@@ -1,5 +1,6 @@
+import { useGlobalSocketEvent, useWatch } from '@capital/common';
 import { Avatar, Tooltip, UserAvatar, UserName } from '@capital/component';
-import React, { useEffect } from 'react';
+import React, { useEffect, useState } from 'react';
 import { useRoomParticipants } from '../utils/useRoomParticipants';
 
 export const LivekitPanelBadge: React.FC<{
@@ -8,11 +9,58 @@ export const LivekitPanelBadge: React.FC<{
 }> = React.memo((props) => {
   const roomName = `${props.groupId}#${props.panelId}`;
   const { participants, fetchParticipants } = useRoomParticipants(roomName);
+  const [displayParticipants, setDisplayParticipants] = useState<
+    {
+      sid: string;
+      identity: string;
+    }[]
+  >([]);
+
+  useWatch([participants.length], () => {
+    setDisplayParticipants(participants);
+  });
 
   useEffect(() => {
     fetchParticipants();
   }, []);
 
+  useGlobalSocketEvent(
+    'plugin:com.msgbyte.livekit.participantJoined',
+    (payload: any) => {
+      if (
+        payload.groupId === props.groupId &&
+        payload.panelId === props.panelId &&
+        payload.participant
+      ) {
+        setDisplayParticipants((state) => [...state, payload.participant]);
+      }
+    }
+  );
+
+  useGlobalSocketEvent(
+    'plugin:com.msgbyte.livekit.participantLeft',
+    (payload: any) => {
+      if (
+        payload.groupId === props.groupId &&
+        payload.panelId === props.panelId &&
+        payload.participant
+      ) {
+        setDisplayParticipants((state) => {
+          const index = state.findIndex(
+            (item) => item.sid === payload.participant.sid
+          );
+          if (index >= 0) {
+            const fin = [...state];
+            fin.splice(index, 1);
+            return fin;
+          } else {
+            return [...state];
+          }
+        });
+      }
+    }
+  );
+
   return (
     <Avatar.Group
       maxCount={4}
@@ -20,7 +68,7 @@ export const LivekitPanelBadge: React.FC<{
       style={{ verticalAlign: 'middle' }}
       maxStyle={{ color: '#f56a00', backgroundColor: '#fde3cf' }}
     >
-      {participants.map((info, i) => (
+      {displayParticipants.map((info, i) => (
         <Tooltip
           key={`${info.sid}#${i}`}
           title={<UserName userId={info.identity} />}

server/plugins/com.msgbyte.livekit/webhook/README.md

@@ -0,0 +1 @@
+## Receive webhook from livekit and send to `livekit.service.js`;

server/plugins/com.msgbyte.livekit/webhook/broker.ts

@@ -0,0 +1,28 @@
+import { TcBroker, SYSTEM_USERID } from 'tailchat-server-sdk';
+import brokerConfig from '../../../moleculer.config';
+
+const transporter = process.env.TRANSPORTER;
+export const broker = new TcBroker({
+  ...brokerConfig,
+  metrics: false,
+  logger: false,
+  transporter,
+});
+
+broker.start().then(() => {
+  console.log('Connnected to Tailchat network, TRANSPORTER: ', transporter);
+});
+
+export function callBrokerAction<T>(
+  actionName: string,
+  params: any,
+  opts?: Record<string, any>
+): Promise<T> {
+  return broker.call(actionName, params, {
+    ...opts,
+    meta: {
+      ...opts?.meta,
+      userId: SYSTEM_USERID,
+    },
+  });
+}

server/plugins/com.msgbyte.livekit/webhook/index.ts

@@ -0,0 +1,40 @@
+import dotenv from 'dotenv';
+import path from 'path';
+dotenv.config({ path: path.resolve(__dirname, '../../../.env') });
+import { WebhookReceiver } from 'livekit-server-sdk';
+import express from 'express';
+import { callBrokerAction } from './broker';
+const app = express();
+
+console.log(path.resolve(__dirname, '../../../.env'));
+
+if (!process.env.LIVEKIT_API_KEY || !process.env.LIVEKIT_API_SECRET) {
+  console.error(
+    'Required env LIVEKIT_API_KEY and LIVEKIT_API_SECRET from livekit'
+  );
+  process.exit(1);
+}
+
+const port = process.env.LIVEKIT_WEBHOOK_PORT || 11008;
+
+app.use(express.raw({ type: 'application/webhook+json' }));
+
+const receiver = new WebhookReceiver(
+  process.env.LIVEKIT_API_KEY,
+  process.env.LIVEKIT_API_SECRET
+);
+
+app.post('/livekit/webhook', (req, res) => {
+  // event is a WebhookEvent object
+  try {
+    const event = receiver.receive(req.body, req.get('Authorization'));
+
+    callBrokerAction('plugin:com.msgbyte.livekit.webhook', event);
+  } finally {
+    res.json({ result: true });
+  }
+});
+
+app.listen(port, () => {
+  console.log(`Livekit Webhook Server is running on port ${port}`);
+});