feat: 增加服务端与agora后台的数据通信处理

server/packages/sdk/src/db/typegoose.ts

@@ -4,6 +4,6 @@ export {
   modelOptions,
   Severity,
 } from '@typegoose/typegoose';
-export type { DocumentType, Ref } from '@typegoose/typegoose';
+export type { DocumentType, Ref, ReturnModelType } from '@typegoose/typegoose';
 export { TimeStamps } from '@typegoose/typegoose/lib/defaultClasses';
 export type { Base } from '@typegoose/typegoose/lib/defaultClasses';

server/plugins/com.msgbyte.agora/models/agora-meeting.ts

@@ -0,0 +1,55 @@
+import { db } from 'tailchat-server-sdk';
+const { getModelForClass, prop, modelOptions, TimeStamps } = db;
+
+@modelOptions({
+  options: {
+    customName: 'p_agora_meeting',
+  },
+})
+export class AgoraMeeting extends TimeStamps implements db.Base {
+  _id: db.Types.ObjectId;
+  id: string;
+
+  @prop()
+  converseId: string;
+
+  @prop()
+  channelName: string;
+
+  @prop()
+  active: boolean;
+
+  /**
+   * 参会人
+   */
+  @prop({
+    default: [],
+  })
+  members: string[];
+
+  /**
+   * 结束时间
+   */
+  @prop()
+  endAt?: Date;
+
+  static async findLastestMeetingByConverseId(
+    this: db.ReturnModelType<typeof AgoraMeeting>,
+    converseId: string
+  ) {
+    return this.findOne({
+      converseId,
+      active: true,
+    }).sort({
+      _id: -1,
+    });
+  }
+}
+
+export type AgoraMeetingDocument = db.DocumentType<AgoraMeeting>;
+
+const model = getModelForClass(AgoraMeeting);
+
+export type AgoraMeetingModel = typeof model;
+
+export default model;

server/plugins/com.msgbyte.agora/models/agora.ts

@@ -1,20 +0,0 @@
-import { db } from 'tailchat-server-sdk';
-const { getModelForClass, prop, modelOptions, TimeStamps } = db;
-
-@modelOptions({
-  options: {
-    customName: 'p_agora',
-  },
-})
-export class Agora extends TimeStamps implements db.Base {
-  _id: db.Types.ObjectId;
-  id: string;
-}
-
-export type AgoraDocument = db.DocumentType<Agora>;
-
-const model = getModelForClass(Agora);
-
-export type AgoraModel = typeof model;
-
-export default model;

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

@@ -1,8 +1,12 @@
 import { DataNotFoundError, TcContext } from 'tailchat-server-sdk';
-import { TcService, TcDbService } from 'tailchat-server-sdk';
+import { TcService, TcDbService, db } from 'tailchat-server-sdk';
-import type { AgoraDocument, AgoraModel } from '../models/agora';
+import type {
+  AgoraMeetingDocument,
+  AgoraMeetingModel,
+} from '../models/agora-meeting';
 import { RtcTokenBuilder, Role as RtcRole } from './utils/RtcTokenBuilder2';
 import got from 'got';
+import _ from 'lodash';
 
 // Reference: https://docs.agora.io/cn/metachat/rtc_channel_management_restfulapi#查询用户列表
 interface ChannelUserListRet {
@@ -29,7 +33,7 @@ interface ChannelUserListRet {
  */
 interface AgoraService
   extends TcService,
-    TcDbService<AgoraDocument, AgoraModel> {}
+    TcDbService<AgoraMeetingDocument, AgoraMeetingModel> {}
 class AgoraService extends TcService {
   get serviceName() {
     return 'plugin:com.msgbyte.agora';
@@ -75,7 +79,7 @@ class AgoraService extends TcService {
       );
       return;
     }
-    // this.registerLocalDb(require('../models/agora').default);
+    this.registerLocalDb(require('../models/agora-meeting').default);
 
     this.registerAction('generateToken', this.generateToken, {
       params: {
@@ -89,6 +93,15 @@ class AgoraService extends TcService {
         channelName: 'string',
       },
     });
+    this.registerAction('webhook', this.webhook, {
+      params: {
+        noticeId: 'string',
+        productId: 'number',
+        eventType: 'number',
+        notifyMs: 'number',
+        payload: 'any',
+      },
+    });
   }
 
   generateToken(
@@ -155,6 +168,92 @@ class AgoraService extends TcService {
     return data;
   }
 
+  /**
+   * agora服务的回调
+   * Reference: https://docs.agora.io/cn/live-streaming-premium-legacy/rtc_channel_event?platform=RESTful#101-channel-create
+   */
+  async webhook(
+    ctx: TcContext<{
+      noticeId: string;
+      productId: number;
+      eventType: number;
+      notifyMs: number;
+      payload: any;
+    }>
+  ) {
+    const { eventType, payload } = ctx.params;
+
+    if (eventType === 101) {
+      // 频道被创建
+      const { channelName } = payload;
+      const converseId = this.getConverseIdFromChannelName(channelName);
+
+      const meeting = await this.adapter.model.create({
+        channelName,
+        converseId,
+        active: true,
+      });
+      this.roomcastNotify(ctx, converseId, 'agoraChannelCreate', {
+        converseId,
+        meetingId: String(meeting._id),
+      });
+    } else if (eventType === 102) {
+      // 频道被销毁
+      const { channelName } = payload;
+      const converseId = this.getConverseIdFromChannelName(channelName);
+
+      const meeting = await this.adapter.model.findLastestMeetingByConverseId(
+        converseId
+      );
+      if (!meeting) {
+        return;
+      }
+
+      meeting.active = false;
+      await meeting.save();
+      this.roomcastNotify(ctx, converseId, 'agoraChannelDestroy', {
+        converseId,
+        meetingId: String(meeting._id),
+      });
+    } else if (eventType === 103) {
+      // 用户加入
+      const { channelName, uid: userId } = payload;
+      const converseId = this.getConverseIdFromChannelName(channelName);
+
+      const meeting = await this.adapter.model.findLastestMeetingByConverseId(
+        converseId
+      );
+      if (!meeting) {
+        return;
+      }
+      meeting.members = _.uniq([...meeting.members, userId]);
+      await meeting.save();
+
+      this.roomcastNotify(ctx, converseId, 'agoraBroadcasterJoin', {
+        converseId,
+        meetingId: String(meeting._id),
+        userId,
+      });
+    } else if (eventType === 104) {
+      // 用户离开
+      const { channelName, uid } = payload;
+      const converseId = this.getConverseIdFromChannelName(channelName);
+
+      const meeting = await this.adapter.model.findLastestMeetingByConverseId(
+        converseId
+      );
+      if (!meeting) {
+        return;
+      }
+
+      this.roomcastNotify(ctx, converseId, 'agoraBroadcasterLeave', {
+        converseId,
+        meetingId: String(meeting._id),
+        userId: uid,
+      });
+    }
+  }
+
   /**
    * 生成restful api需要的请求头
    */
@@ -168,6 +267,24 @@ class AgoraService extends TcService {
       'Content-Type': 'application/json',
     };
   }
+
+  /**
+   * NOTICE: 这里不用每次唯一的ChannelName是期望设计是会话维度的，即可以重复使用
+   */
+  private getConverseIdFromChannelName(channelName: string): string {
+    if (!channelName) {
+      this.logger.error('channel name invalid', channelName);
+      throw new Error('channel name invalid');
+    }
+
+    const [groupId, converseId] = channelName.split('|');
+    if (!db.Types.ObjectId.isValid(converseId)) {
+      this.logger.error('converseId invalid', converseId);
+      throw new Error('converseId invalid');
+    }
+
+    return converseId;
+  }
 }
 
 export default AgoraService;