mail-yubao/tests/integration/pipeline-fixtures.test.ts

import { randomUUID } from "crypto";
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import { prisma } from "@/services/db";
import { ConfirmImport } from "@/services/import/confirm";
import { drainFetchedMails } from "@/services/parse/drain-fetched";
import { getMockTargetAPI } from "@/services/cc/mock-target-api";
import { ImapPoller } from "@/services/imap/poller";
import { getMockMailServer } from "@/services/imap/mock-mail-server";
import { loadEmailFixture, loadAllEmailFixtures } from "@/services/test/email-fixture-library";
import { enableTestModeEnv, disableTestModeEnv } from "@/services/test/test-mode-env";
import {
  deleteTestMailsByRunId,
  ingestFixtureAndParse,
  ingestFixtureFetched,
  loadParsedMail,
  pingTestDatabase,
} from "@/services/test/db-harness";
import type { ShipmentRow } from "@/types/mail";

function uniqueContainer(): string {
  return `TPZU${String(Date.now()).slice(-7)}`;
}

async function confirmForecast(
  mailId: bigint,
  version: number,
  indexes: number[],
  containerNo?: string,
) {
  return ConfirmImport.execute({
    mailId,
    actor: "p3-test",
    version,
    idempotencyKey: randomUUID(),
    F_TransMode: 0,
    F_OperationType: 0,
    containerHeader: containerNo ? { F_ContainerNo: containerNo } : {},
    selectedRowIndexes: indexes,
    ackUnmappedChannels: true,
    forceSkipConflict: true,
  });
}

const RUN_ID = `r${Date.now().toString(36)}`;

describe("pipeline integration (DB + Mock IMAP/CC)", () => {
  let dbOk = false;

  beforeAll(async () => {
    enableTestModeEnv();
    dbOk = await pingTestDatabase();
  });

  afterAll(async () => {
    if (dbOk) await deleteTestMailsByRunId(RUN_ID);
    disableTestModeEnv();
  });

  it("ParsePipeline vs expected.json for pullable fixtures", async () => {
    if (!dbOk) return;
    const all = await loadAllEmailFixtures();
    for (const f of all) {
      if (f.expected.pull === false) continue;
      if (f.expected.idempotent_with) continue;
      const { mailId } = await ingestFixtureAndParse(f, RUN_ID);
      const mail = await loadParsedMail(mailId);
      const expectType = f.expected.pipeline_mail_type || f.expected.mail_type;
      expect(mail.mailType, f.id).toBe(expectType);
      expect(mail.status, f.id).toBe(f.expected.status);
      if (f.expected.last_error === "missing_columns") {
        expect(mail.lastError, f.id).toMatch(/^missing_columns/);
      } else {
        expect(mail.lastError, f.id).toBe(f.expected.last_error);
      }
      if (f.expected.status === "PARSE_FAILED" || f.expected.status === "REJECTED_VALIDATION") {
        const raw = mail.snapshotPath;
        expect(raw, `${f.id} keeps snapshot`).toBeTruthy();
      }
      expect(mail.traceId, `${f.id} has traceId`).toBeTruthy();
      const shipments = (mail.parseResult?.shipments as unknown as ShipmentRow[]) || [];
      if (f.expected.shipment_count != null && mail.parseResult) {
        expect(shipments.length, f.id).toBe(f.expected.shipment_count);
      }
    }
  }, 30_000);

  it("parse fail on one mail does not block drain of another", async () => {
    if (!dbOk) return;
    const bad = await loadEmailFixture("12-missing-required");
    const good = await loadEmailFixture("01-normal-forecast");
    const a = await ingestFixtureFetched(bad, RUN_ID, "drain-bad");
    const b = await ingestFixtureFetched(good, RUN_ID, "drain-good");
    await drainFetchedMails(20);
    const mailA = await loadParsedMail(a.mailId);
    const mailB = await loadParsedMail(b.mailId);
    expect(mailA.status).toBe("PARSE_FAILED");
    expect(mailA.lastError).toMatch(/missing_columns/);
    expect(mailB.status).toBe("PENDING_CONFIRM");
  });

  it("unknown format keyword miss stays IGNORED (not pulled into ops queue)", async () => {
    if (!dbOk) return;
    const f = await loadEmailFixture("16-unknown-format");
    expect(f.expected.status).toBe("IGNORED");
    expect(f.expected.last_error).toBe("SKIP_FILTER");
  });

  it("MockMailServer → parse → stage → MockTargetAPI SaveContainer payload matches source", async () => {
    if (!dbOk) return;
    const f = await loadEmailFixture("01-normal-forecast");
    const { mailId } = await ingestFixtureAndParse(f, `${RUN_ID}.cc`);
    const mail = await loadParsedMail(mailId);
    expect(mail.status).toBe("PENDING_CONFIRM");
    const shipments = (mail.parseResult?.shipments as unknown as ShipmentRow[]) || [];
    const indexes = shipments.filter((s) => s.row_status === "VALID").map((s) => s.row_index);
    const cn = uniqueContainer();
    getMockTargetAPI().reset();
    const result = await confirmForecast(mailId, mail.version, indexes, cn);
    expect(result.mailStatus).toBe("SUCCESS");
    const posts = getMockTargetAPI()
      .getRequests()
      .filter((r) => r.path === "/Container/SaveContainer");
    expect(posts.length).toBeGreaterThanOrEqual(1);
    const body = JSON.stringify(posts[0].body);
    expect(body).toContain(cn);
    expect(body).toContain("LAS1");
  });

  it("duplicate submit is idempotent (mailId+container+hash), no second CC write", async () => {
    if (!dbOk) return;
    const f = await loadEmailFixture("01-normal-forecast");
    const { mailId } = await ingestFixtureAndParse(f, `${RUN_ID}.dup`);
    const mail = await loadParsedMail(mailId);
    const shipments = (mail.parseResult?.shipments as unknown as ShipmentRow[]) || [];
    const indexes = shipments.filter((s) => s.row_status === "VALID").map((s) => s.row_index);
    const cn = uniqueContainer();
    getMockTargetAPI().reset();
    const first = await confirmForecast(mailId, mail.version, indexes, cn);
    expect(first.mailStatus).toBe("SUCCESS");
    const after = await prisma.mailMessage.findUniqueOrThrow({
      where: { id: mailId },
      select: { version: true, status: true },
    });
    getMockTargetAPI().reset();
    const second = await confirmForecast(mailId, after.version, indexes, cn);
    expect(second.mailStatus).toBe("SUCCESS");
    expect(
      getMockTargetAPI().getRequests().filter((r) => r.path === "/Container/SaveContainer"),
    ).toHaveLength(0);
  });

  it("submit 401 leaves mail FAILED, not SUCCESS", async () => {
    if (!dbOk) return;
    const f = await loadEmailFixture("01-normal-forecast");
    const { mailId } = await ingestFixtureAndParse(f, `${RUN_ID}.401`);
    const mail = await loadParsedMail(mailId);
    const shipments = (mail.parseResult?.shipments as unknown as ShipmentRow[]) || [];
    const indexes = shipments.filter((s) => s.row_status === "VALID").map((s) => s.row_index);
    getMockTargetAPI().reset();
    getMockTargetAPI().setFault("401", "/Container/SaveContainer");
    const result = await confirmForecast(
      mailId,
      mail.version,
      indexes,
      uniqueContainer(),
    );
    expect(result.mailStatus).toBe("FAILED");
    expect(result.imports[0].status).toBe("FAILED");
    expect(result.imports[0].lastError).toMatch(/登录|权限|CarrierCentral/);
    const comps = await prisma.importCompensation.findMany({
      where: { importId: result.imports[0].id },
    });
    expect(comps).toHaveLength(0);
  });

  it("submit timeout creates compensation for one-click retry", async () => {
    if (!dbOk) return;
    const f = await loadEmailFixture("01-normal-forecast");
    const { mailId } = await ingestFixtureAndParse(f, `${RUN_ID}.to`);
    const mail = await loadParsedMail(mailId);
    const shipments = (mail.parseResult?.shipments as unknown as ShipmentRow[]) || [];
    const indexes = shipments.filter((s) => s.row_status === "VALID").map((s) => s.row_index);
    getMockTargetAPI().reset();
    getMockTargetAPI().setFault("timeout", "/Container/SaveContainer");
    const result = await confirmForecast(
      mailId,
      mail.version,
      indexes,
      uniqueContainer(),
    );
    expect(result.mailStatus).toBe("FAILED");
    expect(result.imports[0].status).toBe("TIMEOUT_UNKNOWN");
    const comps = await prisma.importCompensation.findMany({
      where: { importId: result.imports[0].id, reason: "TIMEOUT_UNKNOWN" },
    });
    expect(comps.length).toBe(1);
    expect(comps[0].status).toBe("OPEN");
  });

  it("submit 500 creates CC_5XX compensation, mail not SUCCESS", async () => {
    if (!dbOk) return;
    const f = await loadEmailFixture("01-normal-forecast");
    const { mailId } = await ingestFixtureAndParse(f, `${RUN_ID}.500`);
    const mail = await loadParsedMail(mailId);
    const shipments = (mail.parseResult?.shipments as unknown as ShipmentRow[]) || [];
    const indexes = shipments.filter((s) => s.row_status === "VALID").map((s) => s.row_index);
    getMockTargetAPI().reset();
    getMockTargetAPI().setFault("500", "/Container/SaveContainer");
    const result = await confirmForecast(
      mailId,
      mail.version,
      indexes,
      uniqueContainer(),
    );
    expect(result.mailStatus).toBe("FAILED");
    expect(result.imports[0].status).toBe("FAILED");
    const comps = await prisma.importCompensation.findMany({
      where: { importId: result.imports[0].id, reason: "CC_5XX" },
    });
    expect(comps.length).toBe(1);
    expect(comps[0].status).toBe("OPEN");
    getMockTargetAPI().setFault("ok");
  });

  it("IMAP auth_fail: tick does not throw and records consecutiveFail", async () => {
    if (!dbOk) return;
    const statusPath = `${process.cwd().replace(/\\/g, "/")}/data/imap-runtime.json`;
    const fs = await import("fs/promises");
    const backup = await fs.readFile(statusPath, "utf8").catch(() => null);
    try {
      getMockMailServer().setFault("auth_fail");
      const poller = ImapPoller.create();
      await expect(poller.tick(1)).resolves.toMatchObject({ skipped: false });
      const { readImapRuntimeStatus } = await import(
        "@/services/imap/runtime-status"
      );
      const runtime = await readImapRuntimeStatus();
      expect(runtime.consecutiveFail).toBeGreaterThanOrEqual(1);
      expect(runtime.lastError).toMatch(/AUTH|鉴权|同步失败|authentication/i);
    } finally {
      getMockMailServer().setFault("none");
      if (backup != null) await fs.writeFile(statusPath, backup);
      else await fs.rm(statusPath, { force: true });
    }
  });
});