hn-hlw-service/sessionArchive/session-archive-auth/weCom/index.js

const sessionArchiveAuth = require("../index");
const aiSession = require("../../ai-session/index");
const crypto = require("crypto");
const api = require("../../../api");
const common = require("../../../common");
let db = null;
exports.main = async (event, mongodb) => {
  db = mongodb;
  switch (event.type) {
    case "getSessionArchiveUserList":
      return await exports.getSessionArchiveUserList(event);
    case "chatDataList":
      return await exports.chatDataList(event);
    case "searchChatRecord":
      return await exports.searchChatRecord(event);
    case "batchStorageSession":
      return await exports.batchStorageSession(event);
    case "setCorpPublicKey":
      return await exports.setCorpPublicKey(event);
  }
};

/**
 *
 * @param {*} event
 * @returns
 * cursor
 * has_more
 * corpId
 */
exports.getSessionArchiveUserList = async (event) => {
  let { corpId, cursor, limit } = event;
  let query = {
    cursor,
    limit,
    corpId,
    type: "getSessionAuthUserList",
  };
  return await api.getWecomApi(query);
};

// 获取会话列表
exports.chatDataList = async (event) => {
  let { corpId } = event;
  const res = await aiSession.main(
    {
      type: "getSessionArchive",
      ...event,
    },
    db
  );
  if (res.errcode === 0) {
    // 筛选出单聊的数据
    // customerUserId
    // memberUserId
    let msgList = res.msg_list
      .filter((i) => i.receiver_list.length === 1)
      .map((item) => {
        let obj = {
          msgid: item.msgid,
          corpId,
          senderUserId: item.sender.id,
          senderType: item.sender.type,
          sendTime: item.send_time,
          msgtype: item.msgtype,
          encryptedSecretKey: item.service_encrypt_info.encrypted_secret_key,
          publicKeyVersion: item.service_encrypt_info.public_key_ver,
          receiverUserId: item.receiver_list[0].id,
          receiverType: item.receiver_list[0].type,
        };
        if (item.sender.type === 1) {
          obj.memberUserId = item.sender.id;
          obj.customerUserId = item.receiver_list[0].id;
        } else if (item.sender.type === 2) {
          obj.customerUserId = item.sender.id;
          obj.memberUserId = item.receiver_list[0].id;
        }
        return obj;
      });
    return {
      success: true,
      message: "获取成功",
      data: msgList,
      next_cursor: res.next_cursor,
      has_more: res.has_more,
    };
  } else {
    return {
      success: false,
      message: res.errmsg,
      has_more: false,
      next_cursor: "",
      data: [],
    };
  }
};

exports.setCorpPublicKey = async (context) => {
  let { corpId, accesstoken } = context;
  if (!accesstoken) {
    const res = await token.getToken({ corpId });
    accesstoken = res.accessToken;
  }

  // 获取会话归档数据
  const { data } = await sessionArchiveAuth.main(
    {
      corpId,
      type: "getCorpSessionArchive",
    },
    db
  );
  const { publicKeys = [] } = data;

  // 生成 RSA 密钥对
  const { publicKey, privateKey } = crypto.generateKeyPairSync("rsa", {
    modulusLength: 2048,
    publicKeyEncoding: {
      type: "spki",
      format: "pem",
    },
    privateKeyEncoding: {
      type: "pkcs8",
      format: "pem",
    },
  });

  // 获取新版本号
  let publicKeyVersion =
    publicKeys.length !== 0
      ? publicKeys[publicKeys.length - 1].publicKeyVersion
      : 1;
  // 调用接口设置公钥
  const setPublicKeyRes = await api.getWecomApi({
    publicKeyVersion,
    accesstoken,
    publicKey,
    type: "setSessoinPublicKey",
  });

  if (!setPublicKeyRes.success) {
    return setPublicKeyRes;
  }
  const keyItem = {
    publicKeyVersion: setPublicKeyRes.publicKeyVersion,
    publicKey,
    privateKey,
  };
  publicKeys.push(keyItem);
  // 更新 publicKeys
  try {
    const updateRes = await db.collection("corp-session-archive").updateOne(
      { corpId }, // 查询条件
      { $set: { publicKeys } } // 更新操作
    );
    // 如果更新操作没有修改任何文档，返回失败
    if (updateRes.modifiedCount === 0) {
      return { success: false, message: "更新公钥失败" };
    }
    return {
      success: true,
      message: "设置key成功",
    };
  } catch (error) {
    console.error("更新公钥失败:", error);
    return { success: false, message: "设置key失败" };
  }
};

exports.searchChatRecord = async (event) => {
  try {
    let { corpId, cursor, keyword, limit } = event;
    return await api.getWecomApi({
      type: "getSessionSearchMsg",
      keyword,
      limit,
      corpId,
      cursor,
    });
  } catch (e) {
    return {
      success: false,
      message: e,
    };
  }
};

// 批量存储会话存档
exports.batchStorageSession = async ({ corpId, limit = 1000 }) => {
  const sessionRes = await sessionArchiveAuth.main(
    {
      corpId,
      type: "getCorpSessionArchive",
    },
    db
  );
  if (!sessionRes.success) {
    return { success: false, message: sessionRes.message };
  }
  let cursor = sessionRes.data?.next_cursor || "";
  const result = await storageAllSessions({ cursor, corpId, limit });
  if (result.success && result.cursor) {
    await db
      .collection("corp-session-archive")
      .updateOne(
        { corpId },
        { $set: { next_cursor: result.cursor } },
        { upsert: true }
      );
  }
  return result.success
    ? { success: true, message: "存储成功" }
    : { success: false, message: result.message };
};

// 存储会话存档
async function storageAllSessions({ cursor, corpId, limit }) {
  let hasMore = true;
  while (hasMore) {
    const { data, has_more, next_cursor } = await exports.chatDataList({
      cursor,
      corpId,
      limit,
    });
    if (!data || data.length === 0) break;
    await processChatData(corpId, data);
    cursor = next_cursor;
    hasMore = has_more;
  }
  return { success: true, cursor };
}

// 处理会话数据（分块存储 + 批量数据库操作）
async function processChatData(corpId, data) {
  const chunkSize = 500; // 每块处理的数据量
  for (let i = 0; i < data.length; i += chunkSize) {
    const chunk = data.slice(i, i + chunkSize);
    await Promise.all([
      batchInsertRecords(chunk),
      updateChatUsers(corpId, chunk),
    ]);
  }
}
// 批量插入会话记录
async function batchInsertRecords(records) {
  await db.collection("session-archive-record").insertMany({
    ...records,
    _id: common.generateRandomString(24),
  });
}
// 批量更新会话用户数据
async function updateChatUsers(corpId, records) {
  const uniqueUsers = Object.values(
    records
      .filter((item) => !(item.senderType === 1 && item.receiverType === 1))
      .reduce((acc, cur) => {
        if (
          !acc[cur.customerUserId] ||
          acc[cur.customerUserId].sendTime < cur.sendTime
        ) {
          acc[cur.customerUserId] = cur;
        }
        return acc;
      }, {})
  );
  const updates = uniqueUsers.map(async (item) => {
    const { customerUserId, sendTime, memberUserId } = item;
    const existing = await db.collection("session-archive-user").findOne({
      corpId,
      memberUserId,
      customerUserId,
    });
    if (!existing) {
      await db.collection("session-archive-user").insertOne({
        corpId,
        memberUserId,
        sendTime,
        customerUserId,
        _id: common.generateRandomString(24),
      });
    } else {
      await db
        .collection("session-archive-user")
        .updateOne(
          { corpId, memberUserId, customerUserId },
          { $set: { sendTime } }
        );
    }
  });
  await Promise.all(updates);
}