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): Promise { 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): Promise { 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 列拥有,随表一起删除 `, ); }