alibb-server/backend/src/services/ingest.service.ts

import { getPrisma } from '../lib/prisma.js';
import { batchInsertMessages } from '../utils/batchInsert.js';
import { maxSendTime } from '../utils/time.js';

type Payload = {
  inquiry: any;
  customer: any | null;
  messages: { messageId: string; sendTime: string; sender?: string; direction: string; content: string; translation?: string }[];
};

export async function ingest(p: Payload) {
  const prisma = getPrisma();
  const { inquiry: iq, customer: cu, messages } = p;
  const newLast = maxSendTime(messages);

  // 按 customerId 去重:同客户只存 1 条代表 inquiry(同客户询盘详情/聊天一致)。
  // 后续同客户询盘 → 追加到代表的 allInquiries,消息归代表,不新建 inquiry。
  if (cu?.customerId) {
    // 代表 = 该客户最早的询盘(createTime 升序第一);同客户后续询盘追加到 allInquiries
    const rep = await prisma.inquiry.findFirst({
      where: { customerId: cu.customerId },
      orderBy: { createTime: 'asc' },
    });
    if (rep && rep.inquiryId !== iq.inquiryId) {
      const allList: any[] = rep.allInquiries
        ? JSON.parse(rep.allInquiries)
        : [{ inquiryId: rep.inquiryId, createTime: rep.createTime.toISOString(), status: rep.status, msgPreview: rep.msgPreview }];
      if (!allList.find((x: any) => x.inquiryId === iq.inquiryId)) {
        allList.push({ inquiryId: iq.inquiryId, createTime: iq.createTime, status: iq.status, msgPreview: iq.msgPreview });
      }
      // 代表 createTime = 该客户所有会话(询盘)里最早的创建时间
      const allTimes = allList.map((x: any) => new Date(x.createTime).getTime()).filter((t: number) => !isNaN(t));
      const earliest = allTimes.length ? new Date(Math.min(...allTimes)) : new Date(iq.createTime);
      await prisma.$transaction(async (tx) => {
        if (cu) {
          const data = mapCustomer(cu);
          await tx.customer.upsert({ where: { customerId: cu.customerId }, create: data, update: skipNulls(data) });
        }
        await tx.inquiry.update({
          where: { inquiryId: rep.inquiryId },
          data: {
            allInquiries: JSON.stringify(allList),
            createTime: earliest,
            updateTime: new Date(iq.updateTime),
            status: iq.status ?? undefined,
            msgPreview: iq.msgPreview ?? undefined,
          },
        });
      });
      const { failedBatches } = await batchInsertMessages(rep.inquiryId, messages);
      const total = await prisma.chatMessage.count({ where: { inquiryId: rep.inquiryId } });
      await prisma.inquiry.update({
        where: { inquiryId: rep.inquiryId },
        data: { messagesCount: total, lastSendTime: newLast, hasFailedBatches: failedBatches > 0 },
      });
      return { ok: true, inserted: total, failedBatches, earlyReturn: false };
    }
  }

  // 新客户(或同 inquiryId 代表重抓):customer/inquiry 每次 upsert(保证行为/字段最新);
  // 消息只在「有变化」时才 batchInsert(增量,skipDuplicates 去重),没变则跳过省一次批量插入。
  const existing = await prisma.inquiry.findUnique({ where: { inquiryId: iq.inquiryId } });
  const msgsChanged = !existing || existing.hasFailedBatches
    || existing.messagesCount !== messages.length
    || !(newLast && existing.lastSendTime && existing.lastSendTime.getTime() === newLast.getTime());

  // 建/更新代表 inquiry + customer(每次,保证最新)
  await prisma.$transaction(async (tx) => {
    if (cu) {
      const data = mapCustomer(cu);
      await tx.customer.upsert({ where: { customerId: cu.customerId }, create: data, update: skipNulls(data) });
    }
    await tx.inquiry.upsert({
      where: { inquiryId: iq.inquiryId },
      create: {
        inquiryId: iq.inquiryId, customerId: cu?.customerId ?? null, manager: iq.manager,
        createTime: new Date(iq.createTime), updateTime: new Date(iq.updateTime),
        status: iq.status, msgPreview: iq.msgPreview, isFlagged: iq.isFlagged,
      },
      update: {
        customerId: cu?.customerId ?? null, manager: iq.manager,
        createTime: new Date(iq.createTime), updateTime: new Date(iq.updateTime),
        status: iq.status, msgPreview: iq.msgPreview, isFlagged: iq.isFlagged,
      },
    });
  });

  // 消息没变 → 跳过 batchInsert(customer/inquiry 已更新 above);变了才插入 + 更新计数
  if (!msgsChanged) {
    return { ok: true, inserted: existing!.messagesCount, failedBatches: 0, earlyReturn: true };
  }

  const { failedBatches } = await batchInsertMessages(iq.inquiryId, messages);
  const total = await prisma.chatMessage.count({ where: { inquiryId: iq.inquiryId } });
  await prisma.inquiry.update({
    where: { inquiryId: iq.inquiryId },
    data: { messagesCount: total, lastSendTime: newLast, hasFailedBatches: failedBatches > 0 },
  });
  return { ok: true, inserted: total, failedBatches, earlyReturn: false };
}

function mapCustomer(c: any) {
  // 空串归一化为 null:避免空串占据 NOT NULL 语义、影响 stats 的 `country IS NOT NULL` 过滤等
  const s = (v: any) => (v == null || v === '' ? null : v);
  return {
    customerId: c.customerId, name: s(c.name), country: s(c.country), email: s(c.email), phone: s(c.phone), landline: s(c.landline), businessType: s(c.businessType), salesPlatform: s(c.salesPlatform), website: s(c.website), company: s(c.company),
    registerTime: c.registerTime ? new Date(c.registerTime) : null, avatarUrl: s(c.avatarUrl),
    buyerTags: s(c.buyerTags), buyerFeatures: s(c.buyerFeatures),
    bProductViews: c.behavior?.productViews ?? null, bValidInquiries: c.behavior?.validInquiries ?? null,
    bValidRfq: c.behavior?.validRfq ?? null, bLoginDays: c.behavior?.loginDays ?? null,
    bSpamInquiries: c.behavior?.spamInquiries ?? null, bBlacklist: c.behavior?.blacklistCount ?? null, bTopIndustry: c.behavior?.topIndustry ?? null, recentProductsImages: Array.isArray(c.recentInquiryProducts) && c.recentInquiryProducts.length ? JSON.stringify(c.recentInquiryProducts) : null, storeInquiryProduct: s(c.storeInquiryProduct), storeInquiryQty: s(c.storeInquiryQty), storeVisitProducts: Array.isArray(c.storeVisitProducts) && c.storeVisitProducts.length ? JSON.stringify(c.storeVisitProducts) : null,
  };
}

// update 时跳过 null 字段(避免覆盖已落库的有效值,如上次抓到 behavior 这次没抓到)
function skipNulls(data: Record<string, any>) {
  const out: Record<string, any> = {};
  for (const [k, v] of Object.entries(data)) if (v != null) out[k] = v;
  return out;
}