- migrations 0002-0012 split by domain (foundation/instances/content/identity/ sources/assets/policy/execution/agent/ops/organization): 41 tables, uuid pk, created_at/updated_at + touch trigger, text+CHECK status sets, pgcrypto/pg_trgm - key constraints: source_release(site_id, site_release_id), observation(feed_id, guid), external_identity accepted partial unique, download_task(instance_id, torrent_hash), one active acquisition_plan per slot, one active path_binding per entity/season, asset_location root+path uniqueness, organization_record change_seq/change_xid cursor - secret table (ciphertext/nonce/kdf) + secret_ref columns; outbox dispatch fields (ADR-002) - src/types.ts generated from the migrated schema (scripts/gen-types.ts), drift-checked in tests; int8 -> number, date -> string parsers; migrator accepts migrationTableSchema - contracts: Row/Create/Update schemas per domain, public organization-record API (changeSeq cursor), GlobalSettings, CapabilityDescriptor + 14 capability IO schemas, typed SSE events, Api endpoint table - tests: migration up/down round trip on real PG in an isolated schema, constraint checks, contracts<->db column/enum sync, positive/negative parse for every exported schema Co-Authored-By: Claude Fable 5.1 <[email protected]>
148 lines
6.7 KiB
TypeScript
148 lines
6.7 KiB
TypeScript
import type { Kysely } from "kysely";
|
|
import { dropTables, entityTable, execDdl, oneOf } from "../src/migration-helpers.ts";
|
|
|
|
// 执行域:task(业务工作单/时间线投影)、acquisition_plan、download_task、asset_operation(REQ §8.4、§11)
|
|
// 引擎(OpenWorkflow)只管过程推进,业务状态以本域表为准(ADR-002)。
|
|
// 一个 Slot 同时只允许一个有效预留;download_task 以 (instance_id, torrent_hash) 唯一;
|
|
// asset_operation 是副作用账本,operation_id 由服务端按语义生成且唯一。
|
|
|
|
// biome-ignore lint/suspicious/noExplicitAny: 迁移不依赖应用表类型
|
|
export async function up(db: Kysely<any>): Promise<void> {
|
|
await execDdl(
|
|
db,
|
|
`
|
|
${entityTable(
|
|
"task",
|
|
`
|
|
kind text not null check (kind ~ '^[a-z][A-Za-z0-9]*$'),
|
|
handler_version integer not null default 1,
|
|
idempotency_key text not null,
|
|
status text not null default 'pending' ${oneOf("status", ["pending", "running", "waiting", "succeeded", "failed", "cancelled", "dead"])},
|
|
priority integer not null default 100,
|
|
input jsonb not null default '{}'::jsonb,
|
|
output jsonb,
|
|
subject_type text,
|
|
subject_id uuid,
|
|
parent_task_id uuid references task (id) on delete set null,
|
|
engine_run_id text,
|
|
attempts integer not null default 0,
|
|
max_attempts integer not null default 5,
|
|
next_run_at timestamptz not null default now(),
|
|
wait_reason text,
|
|
cancel_requested_at timestamptz,
|
|
cancel_reason text,
|
|
last_error jsonb,
|
|
started_at timestamptz,
|
|
finished_at timestamptz,
|
|
triggered_by text,
|
|
constraint task_idempotency_key unique (kind, idempotency_key)`,
|
|
)}
|
|
create index task_due_idx on task (status, next_run_at) where status in ('pending', 'waiting');
|
|
create index task_subject_idx on task (subject_type, subject_id) where subject_id is not null;
|
|
create index task_parent_idx on task (parent_task_id) where parent_task_id is not null;
|
|
create index task_created_idx on task (created_at desc);
|
|
|
|
${entityTable(
|
|
"acquisition_plan",
|
|
`
|
|
slot_id uuid not null references acquisition_slot (id) on delete restrict,
|
|
source_release_id uuid not null references source_release (id) on delete restrict,
|
|
task_id uuid references task (id) on delete set null,
|
|
purpose text not null default 'fill' ${oneOf("purpose", ["fill", "upgrade", "replace_source"])},
|
|
status text not null default 'draft' ${oneOf("status", ["draft", "reserved", "submitted", "completed", "cancelled", "failed", "expired"])},
|
|
priority text not null default 'tracking' ${oneOf("priority", ["immediate", "tracking", "new_work", "upgrade"])},
|
|
evaluation jsonb not null default '{}'::jsonb,
|
|
reserved_bytes bigint not null default 0,
|
|
budget_reservation jsonb not null default '{}'::jsonb,
|
|
payload_hash text not null,
|
|
expected_revision integer not null default 1,
|
|
idempotency_key text not null,
|
|
expires_at timestamptz,
|
|
status_reason text,
|
|
constraint acquisition_plan_idempotency_key unique (idempotency_key)`,
|
|
)}
|
|
create unique index acquisition_plan_slot_active_key on acquisition_plan (slot_id) where status in ('reserved', 'submitted');
|
|
create index acquisition_plan_status_idx on acquisition_plan (status);
|
|
create index acquisition_plan_release_idx on acquisition_plan (source_release_id);
|
|
|
|
${entityTable(
|
|
"download_task",
|
|
`
|
|
instance_id uuid not null references downloader_instance (id) on delete restrict,
|
|
torrent_hash text not null check (torrent_hash ~ '^[0-9a-f]{40}$|^[0-9a-f]{64}$'),
|
|
infohash_v1 text,
|
|
infohash_v2 text,
|
|
acquisition_plan_id uuid references acquisition_plan (id) on delete set null,
|
|
source_release_id uuid references source_release (id) on delete set null,
|
|
name text not null,
|
|
save_path text,
|
|
content_path text,
|
|
category text,
|
|
tags text[] not null default '{}'::text[],
|
|
managed boolean not null default true,
|
|
origin text not null default 'submitted' ${oneOf("origin", ["submitted", "adopted"])},
|
|
status text not null default 'queued' ${oneOf("status", ["queued", "submitting", "downloading", "paused", "stalled", "completed", "importing", "imported", "failed", "removed", "missing"])},
|
|
pause_reason text ${oneOf("pause_reason", ["user", "threshold", "policy", "agent"])},
|
|
raw_state text,
|
|
progress double precision not null default 0 check (progress >= 0 and progress <= 1),
|
|
size_bytes bigint,
|
|
downloaded_bytes bigint not null default 0,
|
|
remaining_bytes bigint,
|
|
ratio double precision,
|
|
seeding_time_sec bigint,
|
|
completed_at timestamptz,
|
|
last_progress_at timestamptz,
|
|
no_progress_since timestamptz,
|
|
last_synced_at timestamptz,
|
|
next_check_at timestamptz,
|
|
removed_at timestamptz,
|
|
file_list jsonb,
|
|
last_error jsonb,
|
|
constraint download_task_instance_hash_key unique (instance_id, torrent_hash)`,
|
|
)}
|
|
create index download_task_status_idx on download_task (status);
|
|
create index download_task_next_check_idx on download_task (next_check_at) where next_check_at is not null;
|
|
create index download_task_plan_idx on download_task (acquisition_plan_id) where acquisition_plan_id is not null;
|
|
|
|
${entityTable(
|
|
"asset_operation",
|
|
`
|
|
operation_id text not null,
|
|
task_id uuid references task (id) on delete set null,
|
|
plan_fingerprint text,
|
|
ordinal integer not null default 0,
|
|
phase text not null ${oneOf("phase", ["decision", "prepare", "transfer", "verify", "commit", "cleanup", "compensate"])},
|
|
kind text not null ${oneOf("kind", ["copy", "move", "rename", "mkdir", "hardlink", "delete_source", "delete_target", "verify", "register", "qb_pause", "qb_resume", "qb_remove", "qb_set_tags", "other"])},
|
|
status text not null default 'planned' ${oneOf("status", ["planned", "started", "succeeded", "failed", "skipped", "compensated"])},
|
|
intent jsonb not null default '{}'::jsonb,
|
|
attempt_token uuid,
|
|
attempts integer not null default 0,
|
|
result jsonb,
|
|
error jsonb,
|
|
media_asset_id uuid references media_asset (id) on delete set null,
|
|
download_task_id uuid references download_task (id) on delete set null,
|
|
started_at timestamptz,
|
|
finished_at timestamptz,
|
|
constraint asset_operation_operation_id_key unique (operation_id)`,
|
|
)}
|
|
create index asset_operation_task_idx on asset_operation (task_id, ordinal) where task_id is not null;
|
|
create index asset_operation_unfinished_idx on asset_operation (updated_at) where status in ('planned', 'started');
|
|
|
|
alter table media_asset add column download_task_id uuid references download_task (id) on delete set null;
|
|
alter table acquisition_slot add column reserved_plan_id uuid references acquisition_plan (id) on delete set null;
|
|
`,
|
|
);
|
|
}
|
|
|
|
// biome-ignore lint/suspicious/noExplicitAny: 同上
|
|
export async function down(db: Kysely<any>): Promise<void> {
|
|
await execDdl(
|
|
db,
|
|
`
|
|
alter table acquisition_slot drop column if exists reserved_plan_id;
|
|
alter table media_asset drop column if exists download_task_id;
|
|
${dropTables(["asset_operation", "download_task", "acquisition_plan", "task"])}
|
|
`,
|
|
);
|
|
}
|