UHD_Media_Tool/packages/db/migrations/0012_organization.ts

import type { Kysely } from "kysely";
import { dropTables, entityTable, execDdl, oneOf } from "../src/migration-helpers.ts";

// 整理域:organization_record(批次)+ organization_file_result(逐文件)(REQ §12.4)
// change_seq:任何 update(含其文件明细变化)都会刷新为新的序列值,用于稳定游标增量读取;
// change_xid:最后一次变更的事务号。读游标时只返回 change_xid < pg_snapshot_xmin(pg_current_snapshot()) 的行,
// 避免"序号小但提交晚"的事务被游标跳过。

const RECORD_STATUS = ["queued", "running", "succeeded", "partial", "failed", "cancelled"] as const;
const ACTIONS = ["copy", "move", "rename", "relocate", "hardlink"] as const;

// biome-ignore lint/suspicious/noExplicitAny: 迁移不依赖应用表类型
export async function up(db: Kysely<any>): Promise<void> {
  await execDdl(
    db,
    `
create sequence organization_record_change_seq as bigint;

${entityTable(
  "organization_record",
  `
  task_id uuid references task (id) on delete set null,
  trigger text not null ${oneOf("trigger", ["auto", "manual", "agent"])},
  actor_type text not null ${oneOf("actor_type", ["user", "agent", "system", "api_key"])},
  actor_id text,
  media_entity_id uuid references media_entity (id) on delete set null,
  content_unit_id uuid references content_unit (id) on delete set null,
  download_task_id uuid references download_task (id) on delete set null,
  source_downloader_instance_id uuid references downloader_instance (id) on delete set null,
  source_path_view text not null default 'local' ${oneOf("source_path_view", ["downloader", "local", "library"])},
  source_path text,
  target_library_root_id uuid references library_root (id) on delete set null,
  target_path text,
  action text not null ${oneOf("action", ACTIONS)},
  status text not null default 'queued' ${oneOf("status", RECORD_STATUS)},
  reason text,
  error jsonb,
  plan jsonb not null default '{}'::jsonb,
  plan_fingerprint text,
  file_total integer not null default 0,
  file_succeeded integer not null default 0,
  file_failed integer not null default 0,
  file_skipped integer not null default 0,
  started_at timestamptz,
  finished_at timestamptz,
  change_seq bigint not null default nextval('organization_record_change_seq'),
  change_xid xid8 not null default pg_current_xact_id()`,
)}
alter sequence organization_record_change_seq owned by organization_record.change_seq;
create unique index organization_record_change_seq_key on organization_record (change_seq);
create index organization_record_entity_idx on organization_record (media_entity_id) where media_entity_id is not null;
create index organization_record_task_idx on organization_record (task_id) where task_id is not null;
create index organization_record_status_idx on organization_record (status, created_at desc);

create function organization_record_bump_change_seq() returns trigger language plpgsql as $$
begin
  new.change_seq := nextval('organization_record_change_seq');
  new.change_xid := pg_current_xact_id();
  return new;
end $$;
create trigger organization_record_bump_change_seq before update on organization_record
  for each row execute function organization_record_bump_change_seq();

${entityTable(
  "organization_file_result",
  `
  organization_record_id uuid not null references organization_record (id) on delete cascade,
  ordinal integer not null default 0,
  asset_operation_id uuid references asset_operation (id) on delete set null,
  media_asset_id uuid references media_asset (id) on delete set null,
  content_unit_id uuid references content_unit (id) on delete set null,
  role text not null default 'main' ${oneOf("role", ["main", "subtitle", "audio", "bluray_dir", "nfo", "image", "other"])},
  mode text not null ${oneOf("mode", ACTIONS)},
  source_path text not null,
  source_size_bytes bigint,
  target_library_root_id uuid references library_root (id) on delete set null,
  target_path text,
  status text not null default 'queued' ${oneOf("status", ["queued", "running", "succeeded", "failed", "skipped", "cancelled"])},
  skip_reason text,
  error jsonb,
  checksum text,
  started_at timestamptz,
  finished_at timestamptz`,
)}
create index organization_file_result_record_idx on organization_file_result (organization_record_id, ordinal);

-- 文件明细变化时刷新父记录 change_seq,使增量读取能拿到"既有记录状态更新"
create function organization_file_result_touch_record() returns trigger language plpgsql as $$
begin
  update organization_record set updated_at = now()
   where id = coalesce(new.organization_record_id, old.organization_record_id);
  return null;
end $$;
create trigger organization_file_result_touch_record after insert or update or delete on organization_file_result
  for each row execute function organization_file_result_touch_record();

alter table asset_operation add column organization_record_id uuid references organization_record (id) on delete set null;
create index asset_operation_record_idx on asset_operation (organization_record_id) where organization_record_id is not null;
`,
  );
}

// biome-ignore lint/suspicious/noExplicitAny: 同上
export async function down(db: Kysely<any>): Promise<void> {
  await execDdl(
    db,
    `
drop index if exists asset_operation_record_idx;
alter table asset_operation drop column if exists organization_record_id;
${dropTables(["organization_file_result", "organization_record"])}
drop function if exists organization_file_result_touch_record();
drop function if exists organization_record_bump_change_seq();
-- organization_record_change_seq 由 change_seq 列拥有,随表一起删除
`,
  );
}