UHD_Media_Tool/apps/server/test/sources-identity/resolve.test.ts

import { CapabilityRegistry } from "@umt/agent";
import { createMockTmdb, type MockTmdb, TMDB_FIXTURES, type TmdbFixture } from "@umt/testkit";
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import { refreshMetadata } from "../../src/modules/sources-identity/service/metadata.ts";
import { resolveRelease } from "../../src/modules/sources-identity/service/resolve.ts";
import { createHarness, type Harness, json, pgAvailable, seedSite } from "./harness.ts";

const up = await pgAvailable();

const twin = (id: number): TmdbFixture => ({
  type: "movie",
  id,
  original_title: "Twins",
  original_language: "en",
  origin_country: ["US"],
  date: "1988-12-09",
  titles: { "zh-CN": "龙兄鼠弟", "en-US": "Twins" },
  popularity: 10 + id / 1000,
  alternative_titles: [],
  external_ids: {},
});

describe.skipIf(!up)("resolveObservation / 身份与元数据", () => {
  let h: Harness;
  let tmdb: MockTmdb;
  let siteId: string;
  let n = 0;

  beforeAll(async () => {
    tmdb = await createMockTmdb({ fixtures: [...TMDB_FIXTURES, twin(111), twin(112)] });
    h = await createHarness("resolve", { tmdbUrl: tmdb.url });
    siteId = (await seedSite(h, { baseUrl: "https://audiences.example/" })).site.id;
  });
  afterAll(async () => {
    await h?.close();
    await tmdb?.close();
  });

  const release = async (title: string, extra: { subtitle?: string; category?: string } = {}) =>
    h.db
      .insertInto("source_release")
      .values({
        site_id: siteId,
        site_release_id: String(700000 + ++n),
        title,
        subtitle: extra.subtitle ?? null,
        category_raw: extra.category ?? null,
      })
      .returningAll()
      .executeTakeFirstOrThrow();

  let oppId: string;

  it("auto:按 TMDB 建本地作品 + accepted 外部身份 + 别名,发 release.resolved", async () => {
    const r = await release("Oppenheimer.2023.2160p.UHD.BluRay.REMUX.HDR.HEVC-FGT");
    const res = await resolveRelease(h.rt, { sourceReleaseId: r.id });
    expect(res.outcome).toBe("auto");
    oppId = res.mediaEntityId!;
    const e = await h.db.selectFrom("media_entity").selectAll().where("id", "=", oppId).executeTakeFirstOrThrow();
    expect(e).toMatchObject({ kind: "movie", year: 2023, status: "active" });
    const ids = await h.db.selectFrom("external_identity").selectAll().where("media_entity_id", "=", oppId).execute();
    expect(ids.map((i) => `${i.provider}:${i.entity_type}:${i.external_id}:${i.status}`).sort()).toEqual([
      "imdb:movie:tt15398776:accepted",
      "tmdb:movie:872585:accepted",
    ]);
    const revs = await h.db.selectFrom("identity_revision").selectAll().execute();
    expect(revs.length).toBeGreaterThanOrEqual(2);
    const aliases = await h.db
      .selectFrom("title_alias")
      .select(["alias"])
      .where("media_entity_id", "=", oppId)
      .execute();
    expect(aliases.map((a) => a.alias)).toEqual(expect.arrayContaining(["奥本海默", "Oppenheimer"]));
    // 同一文本只登记一次;建档即带 TMDB 字段(QA U9)
    expect(new Set(aliases.map((a) => a.alias)).size, JSON.stringify(aliases)).toBe(aliases.length);
    const created = await h.db.selectFrom("media_entity").selectAll().where("id", "=", oppId).executeTakeFirstOrThrow();
    expect((created.metadata as Record<string, { source: string }>).overview?.source).toBe("tmdb");
    const rel = await h.db.selectFrom("source_release").selectAll().where("id", "=", r.id).executeTakeFirstOrThrow();
    expect(rel).toMatchObject({ resolution_status: "resolved", media_entity_id: oppId, parser_version: "1" });
    expect((rel.parsed as { year: number }).year).toBe(2023);
    const ev = (await h.outbox("release.resolved")).find((x) => x.payload.sourceReleaseId === r.id);
    expect(ev?.payload).toMatchObject({ mediaEntityId: oppId, evidenceLevel: "rule_confirmed", contentUnitIds: [] });
    // 幂等:证据未变的重复识别直接短路
    expect((await resolveRelease(h.rt, { sourceReleaseId: r.id })).outcome).toBe("skipped");
  });

  it("同一作品的其它资源经本地召回复用同一实体,不重复建档", async () => {
    const r = await release("奥本海默.Oppenheimer.2023.1080p.WEB-DL.H264-GRP");
    const res = await resolveRelease(h.rt, { sourceReleaseId: r.id });
    expect(res).toMatchObject({ outcome: "auto", mediaEntityId: oppId });
    const count = await h.db
      .selectFrom("media_entity")
      .select((eb) => eb.fn.countAll<number>().as("c"))
      .executeTakeFirstOrThrow();
    expect(Number(count.c)).toBe(1);
  });

  it("剧集:建 series 作品与临时季/集单元", async () => {
    const r = await release("The.Bear.S03E02.2160p.DSNP.WEB-DL.DDP5.1.H.265-FLUX", { subtitle: "熊家餐馆 第三季" });
    const res = await resolveRelease(h.rt, { sourceReleaseId: r.id });
    expect(res.outcome).toBe("auto");
    const units = await h.db
      .selectFrom("content_unit")
      .selectAll()
      .where("media_entity_id", "=", res.mediaEntityId!)
      .orderBy("kind")
      .execute();
    expect(units.map((u) => `${u.kind}:${u.season_number}:${u.episode_number}:${u.status}`)).toEqual([
      "episode:3:2:provisional",
      "season:3:null:provisional",
    ]);
    expect(res.contentUnitIds).toEqual([units[0]!.id]);
    const e = await h.db
      .selectFrom("media_entity")
      .select(["kind"])
      .where("id", "=", res.mediaEntityId!)
      .executeTakeFirstOrThrow();
    expect(e.kind).toBe("series");
  });

  let twinsRelease: string;
  it("ambiguous:进待识别池 + 待处理事项,不建正式作品", async () => {
    const before = await h.db.selectFrom("media_entity").select("id").execute();
    const r = await release("Twins.1988.1080p.BluRay.x264-GRP");
    twinsRelease = r.id;
    const res = await resolveRelease(h.rt, { sourceReleaseId: r.id });
    expect(res.outcome).toBe("ambiguous");
    expect((await h.db.selectFrom("media_entity").select("id").execute()).length).toBe(before.length);
    const pool = await h.db
      .selectFrom("identity_candidate")
      .selectAll()
      .where("source_release_id", "=", r.id)
      .executeTakeFirstOrThrow();
    expect(pool).toMatchObject({ status: "pending", reason: "ambiguous", attempts: 1 });
    expect((pool.candidates as unknown[]).length).toBe(2);
    expect(pool.next_review_at!.getTime()).toBeGreaterThan(Date.now());
    const d = await h.db.selectFrom("decision").selectAll().where("id", "=", res.decisionId!).executeTakeFirstOrThrow();
    expect(d).toMatchObject({ kind: "identity_ambiguous", status: "open", subject_id: r.id });
    expect((d.options as { key: string }[]).map((o) => o.key)).toEqual([
      "c0",
      "c1",
      "manual_tmdb",
      "local_entity",
      "dismiss",
    ]);
    expect((await h.outbox("decision.created")).some((e) => e.payload.decisionId === d.id)).toBe(true);
    const rel = await h.db.selectFrom("source_release").selectAll().where("id", "=", r.id).executeTakeFirstOrThrow();
    expect(rel.resolution_status).toBe("ambiguous");
    expect(rel.media_entity_id).toBeNull();
    const attempts = await h.db
      .selectFrom("resolution_attempt")
      .selectAll()
      .where("source_release_id", "=", r.id)
      .execute();
    expect(attempts.map((a) => a.outcome)).toEqual(["ambiguous"]);
    // 同一证据再次复查:计数递增(退避加长),不重复开待处理
    await resolveRelease(h.rt, { sourceReleaseId: r.id, force: true });
    const pool2 = await h.db
      .selectFrom("identity_candidate")
      .selectAll()
      .where("source_release_id", "=", r.id)
      .executeTakeFirstOrThrow();
    expect(pool2.attempts).toBe(2);
    expect(pool2.next_review_at!.getTime()).toBeGreaterThan(pool.next_review_at!.getTime());
    expect(
      await h.db.selectFrom("decision").select("id").where("dedupe_key", "=", `identity:${r.id}`).execute(),
    ).toHaveLength(1);
  });

  it("待识别池可按原始标题检索;经待处理中心选候选 = 人工指定并加锁", async () => {
    const list = await json(await h.req("/identity-pool?q=Twins"));
    expect(list.status).toBe(200);
    expect(list.body.data.items.map((i: { sourceReleaseId: string }) => i.sourceReleaseId)).toContain(twinsRelease);
    const pool = list.body.data.items.find((i: { sourceReleaseId: string }) => i.sourceReleaseId === twinsRelease);
    const res = await json(await h.req(`/decisions/${pool.decisionId}/resolve`, { body: { optionKey: "c1" } }));
    expect(res.status).toBe(200);
    expect(res.body.data.status).toBe("resolved");
    const rel = await h.db
      .selectFrom("source_release")
      .selectAll()
      .where("id", "=", twinsRelease)
      .executeTakeFirstOrThrow();
    expect(rel.resolution_locked).toBe(true);
    const chosen = pool.candidates[1];
    const ident = await h.db
      .selectFrom("external_identity")
      .selectAll()
      .where("media_entity_id", "=", rel.media_entity_id!)
      .where("provider", "=", "tmdb")
      .executeTakeFirstOrThrow();
    expect(ident).toMatchObject({
      external_id: chosen.externalId,
      evidence_level: "manual",
      locked: true,
      status: "accepted",
    });
    const p = await h.db
      .selectFrom("identity_candidate")
      .selectAll()
      .where("source_release_id", "=", twinsRelease)
      .executeTakeFirstOrThrow();
    expect(p.status).toBe("resolved");
    const ev = (await h.outbox("release.resolved")).find((x) => x.payload.sourceReleaseId === twinsRelease);
    expect(ev?.payload.evidenceLevel).toBe("manual");
    // 人工锁:自动识别不再改写
    const again = await resolveRelease(h.rt, { sourceReleaseId: twinsRelease, force: true });
    expect(again.outcome).toBe("skipped_locked");
    const after = await h.db
      .selectFrom("source_release")
      .select(["media_entity_id"])
      .where("id", "=", twinsRelease)
      .executeTakeFirstOrThrow();
    expect(after.media_entity_id).toBe(rel.media_entity_id);
  });

  it("no_match 进池并开未识别事项;手动指定 TMDB id / 选本地作品即人工指定加锁;TMDB 失败记为 lookup_failed", async () => {
    const r1 = await release("Zzqxv.Nonexistent.2019.1080p.WEB-DL-NOPE");
    const a = await resolveRelease(h.rt, { sourceReleaseId: r1.id });
    expect(a.outcome).toBe("no_match");
    expect(a.decisionId).not.toBeNull();
    const rel = await h.db
      .selectFrom("source_release")
      .select(["resolution_status"])
      .where("id", "=", r1.id)
      .executeTakeFirstOrThrow();
    expect(rel.resolution_status).toBe("unidentifiable");
    const d = await h.db.selectFrom("decision").selectAll().where("id", "=", a.decisionId!).executeTakeFirstOrThrow();
    expect(d).toMatchObject({ kind: "identity_unresolved", status: "open", subject_id: r1.id });
    expect((d.options as { key: string }[]).map((o) => o.key)).toEqual(["manual_tmdb", "local_entity", "dismiss"]);
    const pool1 = await h.db
      .selectFrom("identity_candidate")
      .select(["decision_id"])
      .where("source_release_id", "=", r1.id)
      .executeTakeFirstOrThrow();
    expect(pool1.decision_id).toBe(d.id);
    // 缺少输入 → 400;给 TMDB id → 人工指定加锁,事项关闭
    expect((await json(await h.req(`/decisions/${d.id}/resolve`, { body: { optionKey: "manual_tmdb" } }))).status).toBe(
      400,
    );
    const fx = TMDB_FIXTURES.find((f) => f.type === "movie")!;
    const ok = await json(
      await h.req(`/decisions/${d.id}/resolve`, {
        body: { optionKey: "manual_tmdb", identity: { tmdb: { type: "movie", id: String(fx.id) } } },
      }),
    );
    expect(ok.status).toBe(200);
    expect(ok.body.data.status).toBe("resolved");
    const rel1 = await h.db.selectFrom("source_release").selectAll().where("id", "=", r1.id).executeTakeFirstOrThrow();
    expect(rel1).toMatchObject({ resolution_status: "resolved", resolution_locked: true });
    const ident = await h.db
      .selectFrom("external_identity")
      .selectAll()
      .where("media_entity_id", "=", rel1.media_entity_id!)
      .where("provider", "=", "tmdb")
      .executeTakeFirstOrThrow();
    expect(ident.external_id).toBe(String(fx.id));

    // 选本地作品
    const r3 = await release("Xyzzy.Plugh.2018.1080p.WEB-DL-NOPE");
    const c = await resolveRelease(h.rt, { sourceReleaseId: r3.id });
    expect(c.outcome).toBe("no_match");
    const local = await json(
      await h.req(`/decisions/${c.decisionId}/resolve`, {
        body: { optionKey: "local_entity", identity: { mediaEntityId: oppId } },
      }),
    );
    expect(local.status).toBe(200);
    const rel3 = await h.db.selectFrom("source_release").selectAll().where("id", "=", r3.id).executeTakeFirstOrThrow();
    expect(rel3).toMatchObject({ media_entity_id: oppId, resolution_locked: true });

    tmdb.fault(/\/search\//, { status: 500, times: 20 });
    const r2 = await release("Qwerty.Unknown.2020.1080p.WEB-DL-NOPE");
    const b = await resolveRelease(h.rt, { sourceReleaseId: r2.id });
    expect(b.outcome).toBe("lookup_failed");
    const pool = await h.db
      .selectFrom("identity_candidate")
      .selectAll()
      .where("source_release_id", "=", r2.id)
      .executeTakeFirstOrThrow();
    expect(pool.reason).toBe("lookup_failed");
    // 查询失败的复查间隔短于无结果
    expect(pool.next_review_at!.getTime() - Date.now()).toBeLessThan(20 * 60_000);
    tmdb.clearFaults();
  });

  it("自动化关闭:证据足够也只提议(proposed),进池待确认", async () => {
    await h.db
      .insertInto("settings")
      .values({ key: "automation", value: JSON.stringify({ enabled: false }) } as never)
      .execute();
    const r = await release("Dune.Part.Two.2024.2160p.UHD.BluRay.REMUX-FraMeSToR", { subtitle: "沙丘2" });
    const res = await resolveRelease(h.rt, { sourceReleaseId: r.id });
    expect(res.outcome).toBe("proposed");
    expect(res.decisionId).not.toBeNull();
    await h.db.deleteFrom("settings").where("key", "=", "automation").execute();
  });

  it("refreshMetadata:字段级刷新;用户锁定字段不被覆盖,解锁后回到来源值", async () => {
    const r1 = await refreshMetadata(h.rt, oppId);
    expect(r1.status).toBe("refreshed");
    const fields = await h.db.selectFrom("metadata_field").selectAll().where("media_entity_id", "=", oppId).execute();
    expect(fields.map((f) => f.field)).toEqual(expect.arrayContaining(["title", "overview", "year"]));
    let e = await h.db.selectFrom("media_entity").selectAll().where("id", "=", oppId).executeTakeFirstOrThrow();
    expect(e.title).toBe("奥本海默");
    expect((e.metadata as Record<string, { source: string }>).overview?.source).toBe("tmdb");

    const put = await json(
      await h.req(`/media-entities/${oppId}/fields/title`, {
        method: "PUT",
        body: { value: "奥本海默(IMAX)", reason: "偏好" },
      }),
    );
    expect(put.status).toBe(200);
    expect(put.body.data.lockedFields).toContain("title");
    const r2 = await refreshMetadata(h.rt, oppId);
    expect(r2.lockedFields).toContain("title");
    e = await h.db.selectFrom("media_entity").selectAll().where("id", "=", oppId).executeTakeFirstOrThrow();
    expect(e.title).toBe("奥本海默(IMAX)");
    expect(e.field_locks).toContain("title");
    expect((e.metadata as Record<string, { source: string; locked: boolean }>).title).toMatchObject({
      source: "user",
      locked: true,
    });

    const del = await json(await h.req(`/media-entities/${oppId}/fields/title`, { method: "DELETE" }));
    expect(del.status).toBe(200);
    e = await h.db.selectFrom("media_entity").selectAll().where("id", "=", oppId).executeTakeFirstOrThrow();
    expect(e.title).toBe("奥本海默");
    expect(e.field_locks).not.toContain("title");
  });

  it("人工纠错映射:旧映射退役留修订,新映射 accepted + 锁;自动匹配不改回", async () => {
    const res = await json(
      await h.req(`/media-entities/${oppId}/identities`, {
        body: { provider: "tmdb", entityType: "movie", externalId: "693134", reason: "误匹配" },
      }),
    );
    expect(res.status).toBe(200);
    expect(res.body.data.changed).toBe(true);
    const ids = await h.db
      .selectFrom("external_identity")
      .selectAll()
      .where("media_entity_id", "=", oppId)
      .where("provider", "=", "tmdb")
      .execute();
    const old = ids.find((i) => i.external_id === "872585")!;
    expect(old).toMatchObject({ status: "retired", locked: true });
    expect(ids.find((i) => i.external_id === "693134")).toMatchObject({
      status: "accepted",
      locked: true,
      evidence_level: "manual",
    });
    const revs = await json(await h.req(`/external-identities/${old.id}/revisions`));
    expect(revs.body.data.items.map((r: { action: string }) => r.action)).toEqual(["created", "status_changed"]);

    // 新来一条奥本海默资源:TMDB 872585 不再自动绑回该作品
    const r = await release("Oppenheimer.2023.1080p.BluRay.x264-OTHER");
    await resolveRelease(h.rt, { sourceReleaseId: r.id });
    const accepted = await h.db
      .selectFrom("external_identity")
      .select(["media_entity_id"])
      .where("provider", "=", "tmdb")
      .where("external_id", "=", "872585")
      .where("status", "=", "accepted")
      .execute();
    expect(accepted.every((a) => a.media_entity_id !== oppId)).toBe(true);
    const still = await h.db
      .selectFrom("external_identity")
      .select(["external_id"])
      .where("media_entity_id", "=", oppId)
      .where("provider", "=", "tmdb")
      .where("status", "=", "accepted")
      .executeTakeFirstOrThrow();
    expect(still.external_id).toBe("693134");

    // 目标身份已属于其它作品 → 409
    const bear = await h.db
      .selectFrom("external_identity")
      .select(["media_entity_id"])
      .where("external_id", "=", "136315")
      .executeTakeFirstOrThrow();
    const conflict = await json(
      await h.req(`/media-entities/${oppId}/identities`, {
        body: { provider: "tmdb", entityType: "tv", externalId: "136315" },
      }),
    );
    expect(conflict.status).toBe(409);
    expect(bear.media_entity_id).not.toBe(oppId);
  });

  it("合并重复作品:旧 ID 跳转,来源资源与别名迁到目标", async () => {
    const dup = await json(
      await h.req("/media-entities", { body: { kind: "movie", title: "奥本海默 重复", year: 2023 } }),
    );
    expect(dup.status).toBe(200);
    const dupId = dup.body.data.id;
    const r = await release("Dup.Title.2023.1080p-X");
    await h.db.updateTable("source_release").set({ media_entity_id: dupId }).where("id", "=", r.id).execute();
    const m = await json(await h.req(`/media-entities/${dupId}/merge`, { body: { intoId: oppId, reason: "重复" } }));
    expect(m.status).toBe(200);
    expect(m.body.data.id).toBe(oppId);
    const detail = await json(await h.req(`/media-entities/${dupId}/detail`));
    expect(detail.body.data.entity.id).toBe(oppId);
    expect(detail.body.data.redirectedFrom).toBe(dupId);
    expect(detail.body.data.aliases.map((a: { alias: string }) => a.alias)).toContain("奥本海默 重复");
    const moved = await h.db
      .selectFrom("source_release")
      .select(["media_entity_id"])
      .where("id", "=", r.id)
      .executeTakeFirstOrThrow();
    expect(moved.media_entity_id).toBe(oppId);
    const redirect = await h.db
      .selectFrom("identity_redirect")
      .selectAll()
      .where("from_entity_id", "=", dupId)
      .executeTakeFirstOrThrow();
    expect(redirect.to_entity_id).toBe(oppId);
    // 再合并一次 → 409
    expect((await h.req(`/media-entities/${dupId}/merge`, { body: { intoId: oppId } })).status).toBe(409);
  });

  it("待处理中心聚合与列表", async () => {
    const s = await json(await h.req("/decisions/summary"));
    expect(s.status).toBe(200);
    expect(s.body.data.total).toBeGreaterThanOrEqual(1);
    expect(s.body.data.groups.some((g: { kind: string }) => g.kind === "identity_ambiguous")).toBe(true);
    const l = await json(await h.req("/decisions?kind=identity_ambiguous"));
    expect(
      l.body.data.items.every((d: { status: string }) => ["open", "investigating", "awaiting_user"].includes(d.status)),
    ).toBe(true);
    // 已关闭的事项不可再处理
    const closed = await h.db
      .selectFrom("decision")
      .select(["id"])
      .where("status", "=", "resolved")
      .executeTakeFirstOrThrow();
    expect((await h.req(`/decisions/${closed.id}/resolve`, { body: { dismiss: true } })).status).toBe(409);
  });

  it("能力:catalog.lookup / release.search 可调用;identity.resolve(propose)未授权被拒", async () => {
    const reg = new CapabilityRegistry({});
    h.module.registerCapabilities!(reg, h.ctx);
    const c = { actorId: "t", taskId: "t", scopeIds: [], signal: new AbortController().signal };
    const look = (await reg.invoke("catalog.lookup", { query: "奥本海默", includeRemote: true }, c)) as {
      local: { entity: { id: string } }[];
      remote: { externalId: string }[];
    };
    expect(look.local.map((l) => l.entity.id)).toContain(oppId);
    expect(look.remote.map((r) => r.externalId)).toContain("872585");
    const search = (await reg.invoke("release.search", { query: "Oppenheimer", allowLiveSearch: false }, c)) as {
      items: unknown[];
    };
    expect(search.items.length).toBeGreaterThan(0);
    await expect(reg.invoke("identity.resolve", { sourceReleaseId: twinsRelease }, c)).rejects.toThrow();
  });
});