UHD_Media_Tool/apps/server/test/lifecycle/lifecycle-events.test.ts

import { type EventType, isEventType, LIFECYCLE_EVENTS, parseEventPayload } from "@umt/contracts";
import { OutboxRelay } from "@umt/tasks";
import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
import { collectOutboxRoutes } from "../../src/context.ts";
import { HookRegistry } from "../../src/hooks/index.ts";
import {
  emitDownloadLifecycle,
  emitScrapeFinished,
} from "../../src/modules/acquisition-organize/services/lifecycle-events.ts";
import { emitRecord } from "../../src/modules/acquisition-organize/services/organize.ts";
import { completeSubscription } from "../../src/modules/acquisition-organize/services/subscriptions.ts";
import { SERVER_MODULES } from "../../src/modules/index.ts";
import {
  emitMetadataRefreshed,
  emitSiteReachability,
} from "../../src/modules/sources-identity/service/lifecycle-events.ts";
import type { ServerContext, ServerModule } from "../../src/modules/types.ts";
import { createHarness, type Harness, pgAvailable, seedBase, seedEntity } from "../acquisition-organize/harness.ts";

// 生命周期广播事件(ADR-006):各业务节点在事务内写 outbox,payload 过 contracts schema,幂等键防重复。

const hasPg = await pgAvailable();

describe.skipIf(!hasPg)("生命周期广播事件(真实 PG)", () => {
  let h: Harness;
  let base: Awaited<ReturnType<typeof seedBase>>;

  /** 某类事件的全部行;顺带确认 payload 过 schema */
  const events = async (type: EventType) => {
    const rows = await h.outbox(type);
    for (const r of rows) expect(parseEventPayload(type, r.payload), `${type} payload`).not.toBeNull();
    return rows;
  };

  beforeAll(async () => {
    h = await createHarness("lifecycle");
    base = await seedBase(h.db, {
      libraryPath: "/tmp/umt-lc-lib",
      cachePath: "/tmp/umt-lc-dl",
      qbUrl: "http://127.0.0.1:9",
    });
  }, 60_000);
  afterAll(async () => {
    await h?.close();
  });

  it("清单里的事件都是已登记类型", () => {
    for (const e of LIFECYCLE_EVENTS) expect(isEventType(e.type)).toBe(true);
  });

  it("订阅:已建(并发作品进入关注)/ 已改 / 暂停也算已改 / 删除发已取消", async () => {
    const e = await seedEntity(h.db, { kind: "movie", title: "订阅事件片", year: 2024 });
    const created = await h.api("POST", "/subscriptions", { mediaEntityId: e, mediaKind: "movie" });
    expect(created.status).toBe(201);
    const id = created.body.data.id as string;
    expect((await events("subscription.created")).filter((r) => r.payload.subscriptionId === id)).toEqual([
      expect.objectContaining({
        payload: { subscriptionId: id, mediaEntityId: e, mediaKind: "movie", origin: "user", status: "active" },
        idempotency_key: `subscription.created/${id}`,
      }),
    ]);
    expect((await events("media.watched")).find((r) => r.payload.mediaEntityId === e)).toMatchObject({
      payload: { mediaEntityId: e, reason: "subscription" },
      idempotency_key: `media.watched/${e}/subscription/${id}`,
    });

    expect((await h.api("PATCH", `/subscriptions/${id}`, { priority: "new_work" })).status).toBe(200);
    expect((await h.api("POST", `/subscriptions/${id}/pause`)).status).toBe(200);
    const updated = (await events("subscription.updated")).filter((r) => r.payload.subscriptionId === id);
    expect(updated.map((r) => [r.payload.changes, r.payload.status])).toEqual([
      [["priority"], "active"],
      [["status"], "paused"],
    ]);

    expect((await h.api("POST", "/subscriptions/batch", { ids: [id], action: "delete" })).status).toBe(200);
    expect((await events("subscription.cancelled")).filter((r) => r.payload.subscriptionId === id)).toEqual([
      expect.objectContaining({ payload: expect.objectContaining({ reason: "user_deleted" }) }),
    ]);
  });

  it("订阅完成:发已完成(按历史行去重);钩子 subscription.completion.check 拒绝时不完成、不发事件", async () => {
    const e = await seedEntity(h.db, { kind: "movie", title: "完成判定片", year: 2023 });
    const sub = await h.db
      .insertInto("subscription")
      .values({ media_entity_id: e, media_kind: "movie", status: "active" })
      .returningAll()
      .executeTakeFirstOrThrow();

    const hooks = new HookRegistry();
    const seen: unknown[] = [];
    hooks.forModule("test").intercept("subscription.completion.check", (c) => {
      seen.push(c);
      return "还要等导演剪辑版";
    });
    h.ctx.hooks = hooks;
    try {
      const done = await h.db.transaction().execute((trx) => completeSubscription(h.d, trx, sub, "slot_satisfied"));
      expect(done).toBe(false);
    } finally {
      h.ctx.hooks = undefined;
    }
    expect(seen).toEqual([
      { subscriptionId: sub.id, mediaEntityId: e, mediaKind: "movie", reason: "slot_satisfied", progress: {} },
    ]);
    const still = await h.db.selectFrom("subscription").select("status").where("id", "=", sub.id).executeTakeFirst();
    expect(still?.status).toBe("active");
    expect((await events("subscription.completed")).filter((r) => r.payload.subscriptionId === sub.id)).toEqual([]);

    expect(await h.db.transaction().execute((trx) => completeSubscription(h.d, trx, sub, "slot_satisfied"))).toBe(true);
    // 已完成的再判一次不会重复发
    expect(await h.db.transaction().execute((trx) => completeSubscription(h.d, trx, sub, "slot_satisfied"))).toBe(
      false,
    );
    const completed = (await events("subscription.completed")).filter((r) => r.payload.subscriptionId === sub.id);
    expect(completed).toHaveLength(1);
    expect(completed[0]!.payload.reason).toBe("slot_satisfied");
    expect(completed[0]!.idempotency_key).toMatch(new RegExp(`^subscription\\.completed/${sub.id}/`));
  });

  it("整理结束:终态才发 organization.finished,同一次状态变化只发一次;进行中不发", async () => {
    const e = await seedEntity(h.db, { kind: "movie", title: "整理事件片", year: 2022 });
    const rec = await h.db
      .insertInto("organization_record")
      .values({ trigger: "auto", actor_type: "system", action: "copy", status: "running", media_entity_id: e })
      .returning("id")
      .executeTakeFirstOrThrow();
    await emitRecord(h.d, h.db, rec.id);
    expect((await events("organization.finished")).filter((r) => r.payload.recordId === rec.id)).toEqual([]);

    await h.db.updateTable("organization_record").set({ status: "succeeded" }).where("id", "=", rec.id).execute();
    await emitRecord(h.d, h.db, rec.id);
    await emitRecord(h.d, h.db, rec.id); // 重放:同一 change_seq
    const fin = (await events("organization.finished")).filter((r) => r.payload.recordId === rec.id);
    expect(fin).toHaveLength(1);
    expect(fin[0]!.payload).toMatchObject({
      outcome: "succeeded",
      action: "copy",
      trigger: "auto",
      mediaEntityId: e,
      downloadTaskId: null,
    });
    expect(fin[0]!.idempotency_key).toBe(`organization.finished/${rec.id}/${fin[0]!.payload.changeSeq}`);
    // organization.changed 照旧每次都发
    expect((await events("organization.changed")).filter((r) => r.payload.recordId === rec.id)).toHaveLength(3);

    // 重新整理后失败:新的状态变化再发一条
    await h.db
      .updateTable("organization_record")
      .set({ status: "failed", reason: "target_exists" })
      .where("id", "=", rec.id)
      .execute();
    await emitRecord(h.d, h.db, rec.id);
    const again = (await events("organization.finished")).filter((r) => r.payload.recordId === rec.id);
    expect(again.map((r) => [r.payload.outcome, r.payload.reason])).toEqual([
      ["succeeded", null],
      ["failed", "target_exists"],
    ]);
  });

  it("下载:完成 / 失败 / 移除各发一次(按任务行代数去重),其它状态不发", async () => {
    const dt = await h.db
      .insertInto("download_task")
      .values({ instance_id: base.instanceId, torrent_hash: "a".repeat(40), name: "Some.Movie.2024.mkv" })
      .returning("id")
      .executeTakeFirstOrThrow();
    await emitDownloadLifecycle(h.d, h.db, dt.id, "queued", "downloading");
    await emitDownloadLifecycle(h.d, h.db, dt.id, "downloading", "completed");
    await emitDownloadLifecycle(h.d, h.db, dt.id, "downloading", "completed");
    await emitDownloadLifecycle(h.d, h.db, dt.id, "completed", "removed");
    const done = (await events("download.completed")).filter((r) => r.payload.downloadTaskId === dt.id);
    expect(done).toEqual([
      expect.objectContaining({
        payload: {
          downloadTaskId: dt.id,
          planId: null,
          mediaEntityId: null,
          name: "Some.Movie.2024.mkv",
          generation: 0,
        },
        idempotency_key: `download.completed/${dt.id}/g0`,
      }),
    ]);
    expect((await events("download.removed")).filter((r) => r.payload.downloadTaskId === dt.id)).toEqual([
      expect.objectContaining({ payload: expect.objectContaining({ from: "completed" }) }),
    ]);
    // 复活后(代数 +1)再完成会再发
    await h.db.updateTable("download_task").set({ generation: 1 }).where("id", "=", dt.id).execute();
    await emitDownloadLifecycle(h.d, h.db, dt.id, "downloading", "completed");
    await emitDownloadLifecycle(h.d, h.db, dt.id, "downloading", "failed");
    expect((await events("download.completed")).filter((r) => r.payload.downloadTaskId === dt.id)).toHaveLength(2);
    expect((await events("download.failed")).filter((r) => r.payload.downloadTaskId === dt.id)).toHaveLength(1);
  });

  it("刮削:失败发 scrape.failed,成功发 scrape.completed", async () => {
    const lib = await h.db.selectFrom("library_root").select("id").executeTakeFirstOrThrow();
    const mk = async (status: "failed" | "succeeded") =>
      h.db
        .insertInto("scrape_record")
        .values({
          root_id: lib.id,
          root_path: "/tmp/umt-lc-lib",
          status,
          reason: status === "failed" ? "scrape_no_metadata" : null,
          written: status === "failed" ? 0 : 3,
          finished_at: new Date(),
        })
        .returningAll()
        .executeTakeFirstOrThrow();
    const bad = await mk("failed");
    const ok = await mk("succeeded");
    await emitScrapeFinished(h.d, bad);
    await emitScrapeFinished(h.d, bad);
    await emitScrapeFinished(h.d, ok);
    expect((await events("scrape.failed")).map((r) => r.payload)).toEqual([
      expect.objectContaining({ scrapeRecordId: bad.id, reason: "scrape_no_metadata", status: "failed" }),
    ]);
    expect((await events("scrape.completed")).map((r) => r.payload)).toEqual([
      expect.objectContaining({ scrapeRecordId: ok.id, written: 3, status: "succeeded" }),
    ]);
  });

  it("站点不可达 / 恢复、元数据已刷新:payload 合法,带键的按键去重", async () => {
    await emitSiteReachability(h.ctx, h.db, base.siteId, {
      reachable: false,
      reason: "auth_expired",
      message: "登录失效",
    });
    await emitSiteReachability(
      h.ctx,
      h.db,
      base.siteId,
      { reachable: true, reason: "health_check" },
      "site.recovered/x",
    );
    await emitSiteReachability(
      h.ctx,
      h.db,
      base.siteId,
      { reachable: true, reason: "health_check" },
      "site.recovered/x",
    );
    expect((await events("site.unreachable")).map((r) => r.payload)).toEqual([
      { siteId: base.siteId, reason: "auth_expired", message: "登录失效" },
    ]);
    expect(await events("site.recovered")).toHaveLength(1);

    const e = await seedEntity(h.db, { kind: "series", title: "刷新剧", year: 2021 });
    for (let i = 0; i < 2; i++)
      await emitMetadataRefreshed(h.ctx, h.db, {
        mediaEntityId: e,
        full: true,
        changedFields: ["overview"],
        runId: "run-1",
      });
    expect((await events("media.metadata_refreshed")).map((r) => r.payload)).toEqual([
      { mediaEntityId: e, full: true, changedFields: ["overview"] },
    ]);
  });

  it("补字幕订阅整理结束:中继投递时补字幕与其它订阅方都收到,入库类整理被认作补字幕候选", async () => {
    await h.db.deleteFrom("outbox_event").execute();
    const e = await seedEntity(h.db, { kind: "movie", title: "字幕订阅片", year: 2020 });
    const recs = [];
    for (const action of ["copy", "trash"] as const) {
      const r = await h.db
        .insertInto("organization_record")
        .values({ trigger: "auto", actor_type: "system", action, status: "succeeded", media_entity_id: e })
        .returning("id")
        .executeTakeFirstOrThrow();
      await emitRecord(h.d, h.db, r.id);
      recs.push(r.id);
    }
    const debug = vi.fn();
    const ctx = { ...h.ctx, logger: { debug, warn: vi.fn() } } as unknown as ServerContext;
    const other: ServerModule = {
      name: "test-other",
      registerRoutes() {},
      registerTasks() {},
      outboxRoutes: () => ({
        "organization.finished": (ev) => ({
          task: "otherTask",
          input: { id: (ev.payload as { recordId: string }).recordId },
        }),
      }),
    };
    const enqueued: { task: string; key: string }[] = [];
    const relay = new OutboxRelay({
      db: h.db,
      engine: {
        enqueue: async (task, _input, o) => {
          enqueued.push({ task, key: o.idempotencyKey });
          return { runId: o.idempotencyKey, deduplicated: false };
        },
      },
      routes: collectOutboxRoutes([...SERVER_MODULES, other], ctx),
    });
    await relay.pollOnce();
    // 补字幕只认复制 / 移动 / 原地入库(进回收站不算),自动补字幕还没做,不建任务
    expect(debug.mock.calls.map((c) => c[0])).toEqual([{ recordId: recs[0], mediaEntityId: e }]);
    expect(enqueued.map((x) => x.task)).toEqual(["otherTask", "otherTask"]);
    const pending = await h.db
      .selectFrom("outbox_event")
      .select(["type", "status"])
      .where("type", "=", "organization.finished")
      .execute();
    expect(pending.map((p) => p.status)).toEqual(["dispatched", "dispatched"]);
  });

  it("payload 不合法直接抛错,不写 outbox", async () => {
    await expect(h.ctx.outbox.emit(h.db, "organization.finished", { recordId: "x" } as never)).rejects.toThrow(
      /不符合 contracts/,
    );
  });
});