koc-loop/lib/wecom-group-push-service.ts

import type { DatabaseClient } from "./database";
import { getRuntimeEnv } from "./runtime-env";
import {
  createGroupMsgTemplate,
  listCustomerGroupChats,
  WecomClientError,
  type WecomConfig,
} from "./wecom-client";

type FetchLike = typeof fetch;

export const WECOM_GROUP_BATCH_LIMIT = 2000;
const TEXT_MAX_BYTES = 4000;
const LINK_TITLE_MAX_BYTES = 128;
const LINK_DESC_MAX_BYTES = 512;
const LINK_URL_MAX_BYTES = 2048;

export type TaskPushRow = {
  id: string;
  name: string;
  brand: string;
  quantity: number;
  due_at: string;
  task_type: string;
  platform: string;
  content_format: string;
  share_token: string | null;
};

export type TaskPushMessage = {
  text: string;
  link: { title: string; desc: string; url: string } | null;
};

export type GroupChatRow = {
  chat_id: string;
  name: string;
  owner_user_id: string;
  member_count: number;
  status: number;
};

export type GroupPushResult = {
  sender: string;
  msgid: string;
  chatCount: number;
  failList: string[];
};

export type GroupPushRecord = {
  id: string;
  taskId: string;
  msgid: string;
  sender: string;
  chatIds: string[];
  failList: string[];
  createdAt: string;
};

function bindingValue(value: unknown) {
  return String(value ?? "").trim();
}

export function trimToBytes(value: string, maxBytes: number) {
  let output = "";
  let used = 0;
  for (const char of Array.from(value)) {
    const size = Buffer.byteLength(char, "utf8");
    if (used + size > maxBytes) break;
    output += char;
    used += size;
  }
  return output;
}

export function buildTaskGroupPushMessage(
  task: TaskPushRow,
  portalUrl: string,
): TaskPushMessage {
  const portal = portalUrl.trim().replace(/\/$/, "");
  const shareToken = bindingValue(task.share_token);
  const claimUrl = portal && shareToken ? `${portal}/?task=${shareToken}` : "";
  const formatLabel = task.content_format === "video" ? "视频" : "图文";
  const lines = [
    `【新任务】${task.name}`,
    `品牌：${task.brand}｜数量：${task.quantity} 份｜截止：${task.due_at}`,
    `平台：${task.platform}｜形式：${formatLabel}`,
    claimUrl ? `领取链接：${claimUrl}` : "领取链接生成失败，请联系管理员",
  ];
  return {
    text: trimToBytes(lines.join("\n"), TEXT_MAX_BYTES),
    link: claimUrl
      ? {
          title: trimToBytes(task.name, LINK_TITLE_MAX_BYTES),
          desc: trimToBytes(
            `品牌 ${task.brand} · ${task.quantity} 份 · 截止 ${task.due_at}`,
            LINK_DESC_MAX_BYTES,
          ),
          url: claimUrl.slice(0, LINK_URL_MAX_BYTES),
        }
      : null,
  };
}

export function groupChatsByOwner(chats: GroupChatRow[]) {
  const grouped = new Map<string, string[]>();
  for (const chat of chats) {
    const owner = bindingValue(chat.owner_user_id);
    if (!owner) continue;
    grouped.set(owner, [...(grouped.get(owner) ?? []), chat.chat_id]);
  }
  return [...grouped.entries()].map(([ownerUserId, chatIds]) => ({
    ownerUserId,
    chatIds,
  }));
}

export async function syncGroupChats(
  db: DatabaseClient,
  config: WecomConfig,
  fetchImpl: FetchLike = fetch,
): Promise<GroupChatRow[]> {
  if (!(config.corpId && config.agentId && config.secret)) {
    throw new WecomClientError(
      "企业微信自建应用未配置（WECOM_CORP_ID / WECOM_AGENT_ID / WECOM_SECRET）",
      400,
    );
  }
  const groups = await listCustomerGroupChats(config, fetchImpl);
  const syncedAt = new Date().toISOString();
  await db.prepare("DELETE FROM wecom_group_chats").run();
  const inserts = groups.map((group) =>
    db
      .prepare(
        `INSERT INTO wecom_group_chats
          (chat_id, name, owner_user_id, member_count, status, synced_at)
         VALUES (?, ?, ?, ?, ?, ?)`,
      )
      .bind(
        group.chatId,
        group.name,
        group.ownerUserId,
        group.memberCount,
        group.status,
        syncedAt,
      ),
  );
  if (inserts.length > 0) await db.batch(inserts);
  return groups.map((group) => ({
    chat_id: group.chatId,
    name: group.name,
    owner_user_id: group.ownerUserId,
    member_count: group.memberCount,
    status: group.status,
  }));
}

export async function listGroupChatRows(db: DatabaseClient) {
  const result = await db
    .prepare(
      `SELECT chat_id, name, owner_user_id, member_count, status
       FROM wecom_group_chats
       ORDER BY member_count DESC`,
    )
    .all<GroupChatRow>();
  return result.results;
}

export async function listGroupPushes(
  db: DatabaseClient,
  taskId?: string,
): Promise<GroupPushRecord[]> {
  const filter = taskId ? "WHERE task_id = ?" : "";
  const statement = db
    .prepare(
      `SELECT id, task_id, msgid, sender, chat_ids, fail_list, created_at
       FROM wecom_group_pushes
       ${filter}
       ORDER BY created_at DESC
       LIMIT 50`,
    )
    .bind(...(taskId ? [taskId] : []));
  const result = await statement.all<{
    id: string;
    task_id: string;
    msgid: string;
    sender: string;
    chat_ids: string;
    fail_list: string;
    created_at: string;
  }>();
  return result.results.map((row) => ({
    id: row.id,
    taskId: row.task_id,
    msgid: row.msgid,
    sender: row.sender,
    chatIds: safeParseArray(row.chat_ids),
    failList: safeParseArray(row.fail_list),
    createdAt: row.created_at,
  }));
}

export async function pushTaskToGroupChats(
  db: DatabaseClient,
  params: { taskId: string; chatIds: unknown; text?: unknown },
  config: WecomConfig,
  fetchImpl: FetchLike = fetch,
): Promise<{ results: GroupPushResult[]; linkUrl: string }> {
  const taskId = bindingValue(params.taskId);
  if (!taskId) throw new WecomClientError("缺少 taskId", 400);

  const requested = [
    ...new Set(
      (Array.isArray(params.chatIds) ? params.chatIds : [])
        .map((id) => bindingValue(id))
        .filter(Boolean),
    ),
  ];
  if (requested.length === 0) {
    throw new WecomClientError("请选择要发送的客户群", 400);
  }

  const task = await db
    .prepare(
      `SELECT id, name, brand, quantity, due_at, task_type, platform,
              content_format, share_token
       FROM tasks WHERE id = ?`,
    )
    .bind(taskId)
    .first<TaskPushRow>();
  if (!task) throw new WecomClientError("任务不存在", 404);

  const chatRows: GroupChatRow[] = [];
  for (let index = 0; index < requested.length; index += 100) {
    const chunk = requested.slice(index, index + 100);
    const placeholders = chunk.map(() => "?").join(", ");
    const result = await db
      .prepare(
        `SELECT chat_id, name, owner_user_id, member_count, status
         FROM wecom_group_chats WHERE chat_id IN (${placeholders})`,
      )
      .bind(...chunk)
      .all<GroupChatRow>();
    chatRows.push(...result.results);
  }
  const byId = new Map(chatRows.map((row) => [row.chat_id, row]));
  const missing = requested.filter((id) => !byId.has(id));
  if (missing.length > 0) {
    throw new WecomClientError(
      `以下客户群未同步到平台，请先同步群列表：${missing.join("、")}`,
      400,
    );
  }
  const noOwner = requested
    .map((id) => byId.get(id))
    .filter((row) => row && !bindingValue(row.owner_user_id));
  if (noOwner.length > 0) {
    throw new WecomClientError(
      `群「${noOwner.map((row) => row?.name || row?.chat_id).join("、")}」缺少群主信息，请重新同步群列表`,
      400,
    );
  }

  const portalUrl = bindingValue(getRuntimeEnv().KOC_PORTAL_URL);
  const fallback = buildTaskGroupPushMessage(task, portalUrl);
  const text = bindingValue(params.text) || fallback.text;
  const link = fallback.link;
  if (!text && !link) {
    throw new WecomClientError("文本与图文附件不能同时为空", 400);
  }

  const results: GroupPushResult[] = [];
  const groups = requested.map((id) => byId.get(id)!);
  for (const { ownerUserId, chatIds } of groupChatsByOwner(groups)) {
    for (
      let index = 0;
      index < chatIds.length;
      index += WECOM_GROUP_BATCH_LIMIT
    ) {
      const batch = chatIds.slice(index, index + WECOM_GROUP_BATCH_LIMIT);
      const sent = await createGroupMsgTemplate(
        config,
        { sender: ownerUserId, chatIdList: batch, text, link: link ?? undefined },
        fetchImpl,
      );
      await db
        .prepare(
          `INSERT INTO wecom_group_pushes
            (id, task_id, msgid, sender, chat_ids, text_content, link_url, fail_list)
           VALUES (?, ?, ?, ?, ?, ?, ?, ?)`,
        )
        .bind(
          `wgp-${crypto.randomUUID().slice(0, 12)}`,
          taskId,
          sent.msgid,
          ownerUserId,
          JSON.stringify(batch),
          text,
          link?.url ?? "",
          JSON.stringify(sent.failList),
        )
        .run();
      results.push({
        sender: ownerUserId,
        msgid: sent.msgid,
        chatCount: batch.length,
        failList: sent.failList,
      });
    }
  }
  return { results, linkUrl: link?.url ?? "" };
}

function safeParseArray(value: string) {
  try {
    const parsed = JSON.parse(value);
    return Array.isArray(parsed) ? parsed.map((item) => String(item)) : [];
  } catch {
    return [];
  }
}