From bd7dd4efccbe7473f4cc7fa8f1dc7c3c1f9dc4d7 Mon Sep 17 00:00:00 2001 From: yml2213 Date: Thu, 5 Nov 2026 21:07:11 +0800 Subject: [PATCH] =?UTF-8?q?feat(fulfillment):=20=E5=BB=BA=E7=AB=8B?= =?UTF-8?q?=E6=AD=A3=E5=BC=8F=E5=B1=A5=E7=BA=A6=E5=BD=B1=E5=AD=90=E6=A8=A1?= =?UTF-8?q?=E5=9E=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../scripts/audit-fulfillment-state.ts | 100 ++- .../141_fulfillment_shadow_model.sql | 810 ++++++++++++++++++ .../fulfillment-shadow-model.test.ts | 46 + ...fillment-state-service.integration.test.ts | 150 ++++ docs/README.md | 2 +- docs/履约系统架构复盘与优化方案-2026-11-05.md | 7 +- 6 files changed, 1110 insertions(+), 5 deletions(-) create mode 100644 apps/backend/src/db/migrations/141_fulfillment_shadow_model.sql create mode 100644 apps/backend/src/services/fulfillment/fulfillment-shadow-model.test.ts diff --git a/apps/backend/scripts/audit-fulfillment-state.ts b/apps/backend/scripts/audit-fulfillment-state.ts index 2b25056b..f624404c 100644 --- a/apps/backend/scripts/audit-fulfillment-state.ts +++ b/apps/backend/scripts/audit-fulfillment-state.ts @@ -93,11 +93,100 @@ async function loadRecentDuplicateAttempts(): Promise { return result.rows } +async function loadShadowModelDiffs(): Promise { + const result = await query<{ + missing_fulfillments: number + fulfillment_state_diffs: number + missing_attempts: number + missing_open_issues: number + unlinked_events: number + }>( + ` + SELECT + COUNT(*) FILTER (WHERE fulfillment.id IS NULL)::int AS missing_fulfillments, + COUNT(*) FILTER ( + WHERE fulfillment.id IS NOT NULL + AND ROW( + fulfillment.route_status, + fulfillment.channel, + fulfillment.carrier_status, + fulfillment.current_attempt_type, + fulfillment.current_attempt_source_id, + fulfillment.issue_code, + fulfillment.issue_category + ) IS DISTINCT FROM ROW( + CASE WHEN state.route_status = 'configured' THEN 'configured' ELSE 'unrouted' END, + state.intended_channel, + state.carrier_status, + state.current_attempt_type, + NULLIF(state.current_attempt_id, 0), + state.issue_code, + state.issue_category + ) + )::int AS fulfillment_state_diffs, + ( + SELECT COUNT(*)::int + FROM ( + SELECT task.order_item_id + FROM fulfillment_tasks task + JOIN fulfillments fulfillment ON fulfillment.order_item_id = task.order_item_id + WHERE NOT EXISTS ( + SELECT 1 + FROM fulfillment_attempts attempt + WHERE attempt.attempt_type = 'task' + AND attempt.source_id = task.id + ) + UNION ALL + SELECT work_order.order_item_id + FROM work_orders work_order + JOIN fulfillments fulfillment ON fulfillment.order_item_id = work_order.order_item_id + WHERE work_order.order_item_id IS NOT NULL + AND NOT EXISTS ( + SELECT 1 + FROM fulfillment_attempts attempt + WHERE attempt.attempt_type = 'work_order' + AND attempt.source_id = work_order.id + ) + ) missing + )::int AS missing_attempts, + COUNT(*) FILTER ( + WHERE state.issue_code <> '' + AND NOT EXISTS ( + SELECT 1 + FROM fulfillment_issues issue + WHERE issue.fulfillment_id = fulfillment.id + AND issue.issue_code = state.issue_code + AND issue.status <> 'resolved' + ) + )::int AS missing_open_issues, + ( + SELECT COUNT(*)::int + FROM fulfillment_state_events event + JOIN v_fulfillment_item_state active_state + ON active_state.order_item_id = event.order_item_id + WHERE event.fulfillment_id IS NULL + ) AS unlinked_events + FROM v_fulfillment_item_state state + LEFT JOIN fulfillments fulfillment + ON fulfillment.order_item_id = state.order_item_id + `, + ) + const row = result.rows[0] + return [ + { label: '影子履约缺失', count: Number(row?.missing_fulfillments || 0) }, + { label: '影子履约状态差异', count: Number(row?.fulfillment_state_diffs || 0) }, + { label: '影子执行尝试缺失', count: Number(row?.missing_attempts || 0) }, + { label: '影子开放异常缺失', count: Number(row?.missing_open_issues || 0) }, + { label: '履约事件未关联影子记录', count: Number(row?.unlinked_events || 0) }, + ] +} + async function main(): Promise { - const [counts, issues, duplicates] = await Promise.all([ + const [counts, issues, duplicates, shadowDiffs] = await Promise.all([ loadCounts(), loadIssueBreakdown(), loadRecentDuplicateAttempts(), + loadShadowModelDiffs(), ]) console.info('[fulfillment-audit] 总览') @@ -119,13 +208,20 @@ async function main(): Promise { console.info(` ${row.label}: ${row.count}`) } + console.info('[fulfillment-audit] 影子模型差异') + for (const row of shadowDiffs) { + console.info(` ${row.label}: ${row.count}`) + } + const critical = new Map(counts.map((row) => [row.label, Number(row.count || 0)])) const duplicateCount = Number(duplicates[0]?.count || 0) + const shadowDiffCount = shadowDiffs.reduce((sum, row) => sum + Number(row.count || 0), 0) const hasCritical = (critical.get('状态不一致') || 0) > 0 || (critical.get('售后关闭仍进入履约队列') || 0) > 0 || (critical.get('已分配但无有效载体') || 0) > 0 || - duplicateCount > 0 + duplicateCount > 0 || + shadowDiffCount > 0 if (strict && hasCritical) { process.exitCode = 1 diff --git a/apps/backend/src/db/migrations/141_fulfillment_shadow_model.sql b/apps/backend/src/db/migrations/141_fulfillment_shadow_model.sql new file mode 100644 index 00000000..a5539320 --- /dev/null +++ b/apps/backend/src/db/migrations/141_fulfillment_shadow_model.sql @@ -0,0 +1,810 @@ +-- 141_fulfillment_shadow_model.sql +-- +-- 正式履约影子模型： +-- 1. fulfillments 保存订单项当前履约事实； +-- 2. fulfillment_attempts 保存任务/工单的执行尝试； +-- 3. fulfillment_issues 保存当前结构化异常，并保留已解决历史； +-- 4. 现有 order_items / fulfillment_tasks / work_orders 继续作为真源， +-- 通过触发器双写影子表，读路径暂不切换。 + +-- 原视图把 route_status 错写成“内部 SKU 是否有配置”，导致已经明确指定渠道的 +-- 订单项仍被标记为 missing_channel。保留原视图为 raw 层，外层按“已有渠道即已路由” +-- 修正路由口径，原视图名和字段保持不变。 +ALTER VIEW v_fulfillment_item_state RENAME TO v_fulfillment_item_state_raw; + +CREATE OR REPLACE VIEW v_fulfillment_item_state AS +SELECT + raw.order_item_id, + raw.order_id, + raw.local_order_id, + raw.provider, + raw.platform, + raw.shop_id, + raw.shop_name, + raw.platform_order_id, + raw.pay_status, + raw.order_status, + raw.fulfillment_status, + raw.internal_sku, + raw.intended_channel, + raw.stored_channel, + raw.configured_channel, + CASE + WHEN COALESCE(raw.route_status, '') = 'missing_internal_sku' THEN 'missing_internal_sku' + WHEN raw.intended_channel IS NOT NULL THEN 'configured' + ELSE 'missing_channel' + END AS route_status, + raw.carrier_status, + raw.current_attempt_type, + raw.current_attempt_id, + raw.current_attempt_status, + raw.current_attempt_result_code, + raw.latest_task_id, + raw.latest_task_executor_key, + raw.latest_task_status, + raw.latest_task_result_code, + raw.latest_work_order_id, + raw.latest_work_order_status, + CASE + WHEN raw.route_status = 'missing_channel' + AND raw.intended_channel IS NOT NULL + THEN '' + ELSE raw.issue_code + END AS issue_code, + CASE + WHEN raw.route_status = 'missing_channel' + AND raw.intended_channel IS NOT NULL + THEN '' + ELSE raw.issue_category + END AS issue_category, + CASE + WHEN raw.route_status = 'missing_channel' + AND raw.intended_channel IS NOT NULL + THEN FALSE + ELSE raw.retryable + END AS retryable, + CASE + WHEN raw.route_status = 'missing_channel' + AND raw.intended_channel IS NOT NULL + THEN FALSE + ELSE raw.auto_retryable + END AS auto_retryable, + CASE + WHEN raw.route_status = 'missing_channel' + AND raw.intended_channel IS NOT NULL + THEN '' + ELSE raw.owner_role + END AS owner_role, + raw.fulfillment_locked, + raw.lock_reason, + raw.fulfillment_pending_reason, + raw.fulfillment_pending_detail, + raw.fulfillment_source, + raw.sku_code, + raw.sku_name, + raw.quantity, + raw.revenue_amount_fen, + raw.created_at, + raw.updated_at, + raw.age_minutes +FROM v_fulfillment_item_state_raw raw; + +COMMENT ON VIEW v_fulfillment_item_state IS + '履约统一读模型：每订单项一行；明确渠道即视为已路由，渠道配置缺失只作为异常明细'; + +CREATE TABLE IF NOT EXISTS fulfillments ( + id BIGSERIAL PRIMARY KEY, + order_item_id BIGINT NOT NULL UNIQUE REFERENCES order_items(id) ON DELETE CASCADE, + order_id BIGINT NOT NULL REFERENCES orders(id) ON DELETE CASCADE, + internal_sku TEXT NOT NULL DEFAULT '', + route_status TEXT NOT NULL DEFAULT 'unrouted' + CHECK (route_status IN ('unrouted', 'configured')), + channel TEXT + CHECK ( + channel IS NULL + OR channel IN ('work_order', 'affiliate_dash', 'third_party_voucher', 'manual', 'on_account') + ), + carrier_status TEXT NOT NULL DEFAULT 'missing' + CHECK (carrier_status IN ('not_required', 'missing', 'creating', 'ready', 'failed', 'terminal')), + current_attempt_type TEXT NOT NULL DEFAULT '' + CHECK (current_attempt_type IN ('', 'task', 'work_order')), + current_attempt_source_id BIGINT, + current_attempt_status TEXT NOT NULL DEFAULT '', + lock_reason TEXT NOT NULL DEFAULT '', + issue_code TEXT NOT NULL DEFAULT '', + issue_category TEXT NOT NULL DEFAULT '', + retryable BOOLEAN NOT NULL DEFAULT FALSE, + owner_role TEXT NOT NULL DEFAULT '', + created_at TIMESTAMPTZ NOT NULL, + updated_at TIMESTAMPTZ NOT NULL, + CHECK ( + (current_attempt_type = '' AND current_attempt_source_id IS NULL) + OR (current_attempt_type <> '' AND current_attempt_source_id IS NOT NULL) + ) +); + +CREATE INDEX IF NOT EXISTS idx_fulfillments_order + ON fulfillments (order_id, updated_at DESC, id DESC); +CREATE INDEX IF NOT EXISTS idx_fulfillments_issue + ON fulfillments (issue_category, carrier_status, updated_at DESC) + WHERE issue_code <> ''; +CREATE INDEX IF NOT EXISTS idx_fulfillments_channel + ON fulfillments (channel, carrier_status, updated_at DESC); + +COMMENT ON TABLE fulfillments IS + '履约主记录：每订单项一行，拆分路由、渠道、载体、当前尝试和当前异常摘要'; + +CREATE TABLE IF NOT EXISTS fulfillment_attempts ( + id BIGSERIAL PRIMARY KEY, + fulfillment_id BIGINT NOT NULL REFERENCES fulfillments(id) ON DELETE CASCADE, + unit_index INTEGER NOT NULL DEFAULT 1, + attempt_no INTEGER NOT NULL, + attempt_type TEXT NOT NULL CHECK (attempt_type IN ('task', 'work_order')), + source_id BIGINT NOT NULL, + source_attempt_no INTEGER NOT NULL DEFAULT 1, + executor_key TEXT NOT NULL DEFAULT '', + status TEXT NOT NULL DEFAULT '', + result_code TEXT NOT NULL DEFAULT '', + result_message TEXT NOT NULL DEFAULT '', + started_at TIMESTAMPTZ NOT NULL, + finished_at TIMESTAMPTZ, + is_current BOOLEAN NOT NULL DEFAULT FALSE, + context_json JSONB NOT NULL DEFAULT '{}'::jsonb, + created_at TIMESTAMPTZ NOT NULL, + updated_at TIMESTAMPTZ NOT NULL, + UNIQUE (attempt_type, source_id), + UNIQUE (fulfillment_id, unit_index, attempt_no), + CHECK (attempt_no > 0), + CHECK (unit_index > 0) +); + +CREATE UNIQUE INDEX IF NOT EXISTS uq_fulfillment_attempts_current_unit + ON fulfillment_attempts (fulfillment_id, unit_index) + WHERE is_current = TRUE; +CREATE INDEX IF NOT EXISTS idx_fulfillment_attempts_history + ON fulfillment_attempts (fulfillment_id, unit_index, attempt_no DESC); +CREATE INDEX IF NOT EXISTS idx_fulfillment_attempts_status + ON fulfillment_attempts (attempt_type, status, updated_at DESC); + +COMMENT ON TABLE fulfillment_attempts IS + '履约执行尝试：任务和接单工单的追加式历史，同一订单项单元仅一个当前尝试'; + +CREATE TABLE IF NOT EXISTS fulfillment_issues ( + id BIGSERIAL PRIMARY KEY, + fulfillment_id BIGINT NOT NULL REFERENCES fulfillments(id) ON DELETE CASCADE, + issue_code TEXT NOT NULL, + issue_category TEXT NOT NULL DEFAULT '', + severity TEXT NOT NULL DEFAULT 'warning' + CHECK (severity IN ('info', 'warning', 'error', 'critical')), + source TEXT NOT NULL DEFAULT 'read_model', + retryable BOOLEAN NOT NULL DEFAULT FALSE, + auto_retryable BOOLEAN NOT NULL DEFAULT FALSE, + owner_role TEXT NOT NULL DEFAULT '', + status TEXT NOT NULL DEFAULT 'open' + CHECK (status IN ('open', 'acknowledged', 'waiting', 'resolved')), + detail_json JSONB NOT NULL DEFAULT '{}'::jsonb, + first_seen_at TIMESTAMPTZ NOT NULL, + last_seen_at TIMESTAMPTZ NOT NULL, + resolved_at TIMESTAMPTZ, + UNIQUE (fulfillment_id, issue_code) +); + +CREATE INDEX IF NOT EXISTS idx_fulfillment_issues_queue + ON fulfillment_issues (status, issue_category, last_seen_at DESC) + WHERE status <> 'resolved'; +CREATE INDEX IF NOT EXISTS idx_fulfillment_issues_code + ON fulfillment_issues (issue_code, status, last_seen_at DESC); + +COMMENT ON TABLE fulfillment_issues IS + '履约结构化异常：当前开放异常按 issue_code 唯一，解决后保留历史'; + +ALTER TABLE fulfillment_state_events + ADD COLUMN IF NOT EXISTS fulfillment_id BIGINT; + +CREATE INDEX IF NOT EXISTS idx_fulfillment_state_events_fulfillment + ON fulfillment_state_events (fulfillment_id, id DESC) + WHERE fulfillment_id IS NOT NULL; + +CREATE OR REPLACE FUNCTION sync_fulfillment_shadow(p_order_item_id BIGINT) +RETURNS VOID +LANGUAGE plpgsql +AS $$ +DECLARE + item_record RECORD; + state_record RECORD; + fulfillment_id_value BIGINT; +BEGIN + SELECT + oi.id, + oi.order_id, + resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code) AS internal_sku, + oi.created_at + INTO item_record + FROM order_items oi + WHERE oi.id = p_order_item_id; + + IF NOT FOUND THEN + RETURN; + END IF; + + SELECT state.* + INTO state_record + FROM v_fulfillment_item_state state + WHERE state.order_item_id = item_record.id; + + IF FOUND THEN + INSERT INTO fulfillments ( + order_item_id, + order_id, + internal_sku, + route_status, + channel, + carrier_status, + current_attempt_type, + current_attempt_source_id, + current_attempt_status, + lock_reason, + issue_code, + issue_category, + retryable, + owner_role, + created_at, + updated_at + ) VALUES ( + item_record.id, + item_record.order_id, + COALESCE(state_record.internal_sku, ''), + CASE + WHEN state_record.route_status = 'configured' THEN 'configured' + ELSE 'unrouted' + END, + state_record.intended_channel, + state_record.carrier_status, + state_record.current_attempt_type, + NULLIF(state_record.current_attempt_id, 0), + state_record.current_attempt_status, + state_record.lock_reason, + state_record.issue_code, + state_record.issue_category, + state_record.retryable, + state_record.owner_role, + item_record.created_at, + NOW() + ) + ON CONFLICT (order_item_id) DO UPDATE SET + order_id = EXCLUDED.order_id, + internal_sku = EXCLUDED.internal_sku, + route_status = EXCLUDED.route_status, + channel = EXCLUDED.channel, + carrier_status = EXCLUDED.carrier_status, + current_attempt_type = EXCLUDED.current_attempt_type, + current_attempt_source_id = EXCLUDED.current_attempt_source_id, + current_attempt_status = EXCLUDED.current_attempt_status, + lock_reason = EXCLUDED.lock_reason, + issue_code = EXCLUDED.issue_code, + issue_category = EXCLUDED.issue_category, + retryable = EXCLUDED.retryable, + owner_role = EXCLUDED.owner_role, + updated_at = NOW() + RETURNING id INTO fulfillment_id_value; + + IF state_record.issue_code <> '' THEN + INSERT INTO fulfillment_issues ( + fulfillment_id, + issue_code, + issue_category, + severity, + source, + retryable, + auto_retryable, + owner_role, + status, + detail_json, + first_seen_at, + last_seen_at, + resolved_at + ) VALUES ( + fulfillment_id_value, + state_record.issue_code, + state_record.issue_category, + CASE + WHEN state_record.issue_category IN ('after_sales', 'system') THEN 'error' + WHEN state_record.issue_category = 'carrier' THEN 'warning' + ELSE 'info' + END, + 'read_model', + state_record.retryable, + state_record.auto_retryable, + state_record.owner_role, + 'open', + COALESCE(state_record.fulfillment_pending_detail, '{}'::jsonb), + state_record.created_at, + NOW(), + NULL + ) + ON CONFLICT (fulfillment_id, issue_code) DO UPDATE SET + issue_category = EXCLUDED.issue_category, + severity = EXCLUDED.severity, + retryable = EXCLUDED.retryable, + auto_retryable = EXCLUDED.auto_retryable, + owner_role = EXCLUDED.owner_role, + status = 'open', + detail_json = EXCLUDED.detail_json, + last_seen_at = NOW(), + resolved_at = NULL; + + UPDATE fulfillment_issues + SET status = 'resolved', + resolved_at = NOW() + WHERE fulfillment_id = fulfillment_id_value + AND issue_code <> state_record.issue_code + AND status <> 'resolved'; + ELSE + UPDATE fulfillment_issues + SET status = 'resolved', + resolved_at = NOW() + WHERE fulfillment_id = fulfillment_id_value + AND status <> 'resolved'; + END IF; + ELSE + INSERT INTO fulfillments ( + order_item_id, + order_id, + internal_sku, + route_status, + channel, + carrier_status, + current_attempt_type, + current_attempt_source_id, + current_attempt_status, + lock_reason, + issue_code, + issue_category, + retryable, + owner_role, + created_at, + updated_at + ) VALUES ( + item_record.id, + item_record.order_id, + COALESCE(item_record.internal_sku, ''), + CASE WHEN COALESCE(item_record.internal_sku, '') = '' THEN 'unrouted' ELSE 'configured' END, + NULL, + 'terminal', + '', + NULL, + '', + '', + '', + '', + FALSE, + '', + item_record.created_at, + NOW() + ) + ON CONFLICT (order_item_id) DO UPDATE SET + route_status = EXCLUDED.route_status, + channel = NULL, + carrier_status = 'terminal', + current_attempt_type = '', + current_attempt_source_id = NULL, + current_attempt_status = '', + lock_reason = '', + issue_code = '', + issue_category = '', + retryable = FALSE, + owner_role = '', + updated_at = NOW() + RETURNING id INTO fulfillment_id_value; + + UPDATE fulfillment_issues + SET status = 'resolved', + resolved_at = NOW() + WHERE fulfillment_id = fulfillment_id_value + AND status <> 'resolved'; + END IF; +END; +$$; + +-- 先为当前活跃订单项建立履约主记录和结构化异常。 +SELECT sync_fulfillment_shadow(state.order_item_id) +FROM v_fulfillment_item_state state; + +WITH source_attempts AS ( + SELECT + task.order_item_id, + task.unit_index, + task.created_at AS occurred_at, + 'task'::text AS attempt_type, + task.id AS source_id, + task.attempt_no AS source_attempt_no, + task.executor_key, + task.task_status AS status, + task.result_code, + task.result_message, + task.created_at AS started_at, + CASE + WHEN task.task_status IN ( + 'closed', 'expired', 'failed', 'manual_review', 'retry_pending', 'cancelled' + ) THEN task.updated_at + ELSE NULL + END AS finished_at, + task.is_current, + task.context_json, + task.created_at, + task.updated_at + FROM fulfillment_tasks task + JOIN fulfillments fulfillment ON fulfillment.order_item_id = task.order_item_id + + UNION ALL + + SELECT + work_order.order_item_id, + 1, + work_order.created_at, + 'work_order'::text, + work_order.id, + 1, + 'work_order'::text, + work_order.status, + ''::text, + work_order.problem_note, + work_order.created_at, + CASE + WHEN work_order.status IN ('accepted', 'cancelled', 'closed') THEN work_order.updated_at + ELSE NULL + END, + work_order.status <> 'cancelled', + jsonb_build_object( + 'workOrderNo', work_order.work_order_no, + 'productName', work_order.product_name + ), + work_order.created_at, + work_order.updated_at + FROM work_orders work_order + JOIN fulfillments fulfillment ON fulfillment.order_item_id = work_order.order_item_id + WHERE work_order.order_item_id IS NOT NULL +), +ranked AS ( + SELECT + source_attempts.*, + ROW_NUMBER() OVER ( + PARTITION BY order_item_id, unit_index + ORDER BY occurred_at, attempt_type, source_id + )::int AS attempt_no + FROM source_attempts +) +INSERT INTO fulfillment_attempts ( + fulfillment_id, + unit_index, + attempt_no, + attempt_type, + source_id, + source_attempt_no, + executor_key, + status, + result_code, + result_message, + started_at, + finished_at, + is_current, + context_json, + created_at, + updated_at +) +SELECT + fulfillment.id, + ranked.unit_index, + ranked.attempt_no, + ranked.attempt_type, + ranked.source_id, + ranked.source_attempt_no, + ranked.executor_key, + ranked.status, + ranked.result_code, + ranked.result_message, + ranked.started_at, + ranked.finished_at, + CASE + WHEN ranked.attempt_type = fulfillment.current_attempt_type + AND ranked.source_id = fulfillment.current_attempt_source_id + THEN TRUE + ELSE FALSE + END, + ranked.context_json, + ranked.created_at, + ranked.updated_at +FROM ranked +JOIN fulfillments fulfillment ON fulfillment.order_item_id = ranked.order_item_id +ON CONFLICT (attempt_type, source_id) DO NOTHING; + +UPDATE fulfillment_state_events event +SET fulfillment_id = fulfillment.id +FROM fulfillments fulfillment +WHERE fulfillment.order_item_id = event.order_item_id + AND event.fulfillment_id IS NULL; + +DO $$ +BEGIN + IF NOT EXISTS ( + SELECT 1 + FROM pg_constraint + WHERE conname = 'fk_fulfillment_state_events_fulfillment' + ) THEN + ALTER TABLE fulfillment_state_events + ADD CONSTRAINT fk_fulfillment_state_events_fulfillment + FOREIGN KEY (fulfillment_id) REFERENCES fulfillments(id) ON DELETE SET NULL; + END IF; +END $$; + +CREATE OR REPLACE FUNCTION assign_fulfillment_event_shadow() +RETURNS TRIGGER +LANGUAGE plpgsql +AS $$ +BEGIN + IF NEW.fulfillment_id IS NULL THEN + SELECT fulfillment.id + INTO NEW.fulfillment_id + FROM fulfillments fulfillment + WHERE fulfillment.order_item_id = NEW.order_item_id; + END IF; + RETURN NEW; +END; +$$; + +DROP TRIGGER IF EXISTS trg_fulfillment_state_events_shadow ON fulfillment_state_events; +CREATE TRIGGER trg_fulfillment_state_events_shadow +BEFORE INSERT ON fulfillment_state_events +FOR EACH ROW +EXECUTE FUNCTION assign_fulfillment_event_shadow(); + +CREATE OR REPLACE FUNCTION sync_fulfillment_shadow_from_order_item() +RETURNS TRIGGER +LANGUAGE plpgsql +AS $$ +BEGIN + PERFORM sync_fulfillment_shadow(NEW.id); + RETURN NEW; +END; +$$; + +DROP TRIGGER IF EXISTS trg_order_items_fulfillment_shadow ON order_items; +CREATE TRIGGER trg_order_items_fulfillment_shadow +AFTER INSERT OR UPDATE OF + fulfillment_status, + fulfillment_channel, + fulfillment_source, + fulfillment_locked, + fulfillment_pending_reason, + fulfillment_pending_detail, + fulfillment_reason, + manual_sku_key, + sku_code, + sku_name +ON order_items +FOR EACH ROW +EXECUTE FUNCTION sync_fulfillment_shadow_from_order_item(); + +CREATE OR REPLACE FUNCTION sync_fulfillment_task_attempt() +RETURNS TRIGGER +LANGUAGE plpgsql +AS $$ +DECLARE + fulfillment_id_value BIGINT; + existing_attempt_id BIGINT; + next_attempt_no INTEGER; +BEGIN + IF TG_OP = 'DELETE' THEN + DELETE FROM fulfillment_attempts + WHERE attempt_type = 'task' AND source_id = OLD.id; + PERFORM sync_fulfillment_shadow(OLD.order_item_id); + RETURN OLD; + END IF; + + PERFORM sync_fulfillment_shadow(NEW.order_item_id); + SELECT fulfillment.id + INTO fulfillment_id_value + FROM fulfillments fulfillment + WHERE fulfillment.order_item_id = NEW.order_item_id; + IF fulfillment_id_value IS NULL THEN + RETURN NEW; + END IF; + + SELECT attempt.id + INTO existing_attempt_id + FROM fulfillment_attempts attempt + WHERE attempt.attempt_type = 'task' + AND attempt.source_id = NEW.id; + + IF existing_attempt_id IS NULL THEN + SELECT COALESCE(MAX(attempt.attempt_no), 0) + 1 + INTO next_attempt_no + FROM fulfillment_attempts attempt + WHERE attempt.fulfillment_id = fulfillment_id_value + AND attempt.unit_index = NEW.unit_index; + END IF; + + UPDATE fulfillment_attempts + SET is_current = FALSE, + updated_at = NOW() + WHERE fulfillment_id = fulfillment_id_value + AND unit_index = NEW.unit_index + AND is_current = TRUE + AND NOT (attempt_type = 'task' AND source_id = NEW.id); + + INSERT INTO fulfillment_attempts ( + fulfillment_id, + unit_index, + attempt_no, + attempt_type, + source_id, + source_attempt_no, + executor_key, + status, + result_code, + result_message, + started_at, + finished_at, + is_current, + context_json, + created_at, + updated_at + ) VALUES ( + fulfillment_id_value, + NEW.unit_index, + COALESCE(next_attempt_no, 1), + 'task', + NEW.id, + NEW.attempt_no, + NEW.executor_key, + NEW.task_status, + NEW.result_code, + NEW.result_message, + NEW.created_at, + CASE + WHEN NEW.task_status IN ( + 'closed', 'expired', 'failed', 'manual_review', 'retry_pending', 'cancelled' + ) THEN NEW.updated_at + ELSE NULL + END, + NEW.is_current, + NEW.context_json, + NEW.created_at, + NEW.updated_at + ) + ON CONFLICT (attempt_type, source_id) DO UPDATE SET + executor_key = EXCLUDED.executor_key, + status = EXCLUDED.status, + result_code = EXCLUDED.result_code, + result_message = EXCLUDED.result_message, + finished_at = EXCLUDED.finished_at, + is_current = EXCLUDED.is_current, + context_json = EXCLUDED.context_json, + updated_at = EXCLUDED.updated_at; + + PERFORM sync_fulfillment_shadow(NEW.order_item_id); + RETURN NEW; +END; +$$; + +DROP TRIGGER IF EXISTS trg_fulfillment_tasks_shadow ON fulfillment_tasks; +CREATE TRIGGER trg_fulfillment_tasks_shadow +AFTER INSERT OR UPDATE OR DELETE ON fulfillment_tasks +FOR EACH ROW +EXECUTE FUNCTION sync_fulfillment_task_attempt(); + +CREATE OR REPLACE FUNCTION sync_fulfillment_work_order_attempt() +RETURNS TRIGGER +LANGUAGE plpgsql +AS $$ +DECLARE + fulfillment_id_value BIGINT; + existing_attempt_id BIGINT; + next_attempt_no INTEGER; +BEGIN + IF TG_OP = 'DELETE' THEN + DELETE FROM fulfillment_attempts + WHERE attempt_type = 'work_order' AND source_id = OLD.id; + IF OLD.order_item_id IS NOT NULL THEN + PERFORM sync_fulfillment_shadow(OLD.order_item_id); + END IF; + RETURN OLD; + END IF; + + IF NEW.order_item_id IS NULL THEN + RETURN NEW; + END IF; + + PERFORM sync_fulfillment_shadow(NEW.order_item_id); + SELECT fulfillment.id + INTO fulfillment_id_value + FROM fulfillments fulfillment + WHERE fulfillment.order_item_id = NEW.order_item_id; + IF fulfillment_id_value IS NULL THEN + RETURN NEW; + END IF; + + SELECT attempt.id + INTO existing_attempt_id + FROM fulfillment_attempts attempt + WHERE attempt.attempt_type = 'work_order' + AND attempt.source_id = NEW.id; + + IF existing_attempt_id IS NULL THEN + SELECT COALESCE(MAX(attempt.attempt_no), 0) + 1 + INTO next_attempt_no + FROM fulfillment_attempts attempt + WHERE attempt.fulfillment_id = fulfillment_id_value + AND attempt.unit_index = 1; + END IF; + + UPDATE fulfillment_attempts + SET is_current = FALSE, + updated_at = NOW() + WHERE fulfillment_id = fulfillment_id_value + AND unit_index = 1 + AND is_current = TRUE + AND NOT (attempt_type = 'work_order' AND source_id = NEW.id); + + INSERT INTO fulfillment_attempts ( + fulfillment_id, + unit_index, + attempt_no, + attempt_type, + source_id, + source_attempt_no, + executor_key, + status, + result_code, + result_message, + started_at, + finished_at, + is_current, + context_json, + created_at, + updated_at + ) VALUES ( + fulfillment_id_value, + 1, + COALESCE(next_attempt_no, 1), + 'work_order', + NEW.id, + 1, + 'work_order', + NEW.status, + '', + NEW.problem_note, + NEW.created_at, + CASE + WHEN NEW.status IN ('accepted', 'cancelled', 'closed') THEN NEW.updated_at + ELSE NULL + END, + NEW.status <> 'cancelled', + jsonb_build_object( + 'workOrderNo', NEW.work_order_no, + 'productName', NEW.product_name + ), + NEW.created_at, + NEW.updated_at + ) + ON CONFLICT (attempt_type, source_id) DO UPDATE SET + status = EXCLUDED.status, + result_message = EXCLUDED.result_message, + finished_at = EXCLUDED.finished_at, + is_current = EXCLUDED.is_current, + context_json = EXCLUDED.context_json, + updated_at = EXCLUDED.updated_at; + + PERFORM sync_fulfillment_shadow(NEW.order_item_id); + RETURN NEW; +END; +$$; + +DROP TRIGGER IF EXISTS trg_work_orders_fulfillment_shadow ON work_orders; +CREATE TRIGGER trg_work_orders_fulfillment_shadow +AFTER INSERT OR UPDATE OR DELETE ON work_orders +FOR EACH ROW +EXECUTE FUNCTION sync_fulfillment_work_order_attempt(); + +COMMENT ON COLUMN fulfillment_state_events.fulfillment_id IS + '正式履约影子记录；迁移 141 后新旧事件都自动关联'; diff --git a/apps/backend/src/services/fulfillment/fulfillment-shadow-model.test.ts b/apps/backend/src/services/fulfillment/fulfillment-shadow-model.test.ts new file mode 100644 index 00000000..a0484cb8 --- /dev/null +++ b/apps/backend/src/services/fulfillment/fulfillment-shadow-model.test.ts @@ -0,0 +1,46 @@ +import assert from 'node:assert/strict' +import { readFileSync } from 'node:fs' +import test from 'node:test' + +const migration = readFileSync( + new URL('../../db/migrations/141_fulfillment_shadow_model.sql', import.meta.url), + 'utf8', +) +const auditScript = readFileSync( + new URL('../../../scripts/audit-fulfillment-state.ts', import.meta.url), + 'utf8', +) + +test('正式履约影子模型包含履约、尝试和异常三张表', () => { + assert.ok(migration.includes('CREATE TABLE IF NOT EXISTS fulfillments')) + assert.ok(migration.includes('CREATE TABLE IF NOT EXISTS fulfillment_attempts')) + assert.ok(migration.includes('CREATE TABLE IF NOT EXISTS fulfillment_issues')) + assert.ok(migration.includes('uq_fulfillment_attempts_current_unit')) + assert.ok(migration.includes('UNIQUE (fulfillment_id, issue_code)')) +}) + +test('影子模型从现有真源回填并持续双写', () => { + assert.ok(migration.includes('SELECT sync_fulfillment_shadow(state.order_item_id)')) + assert.ok(migration.includes('trg_order_items_fulfillment_shadow')) + assert.ok(migration.includes('trg_fulfillment_tasks_shadow')) + assert.ok(migration.includes('trg_work_orders_fulfillment_shadow')) + assert.ok(migration.includes('assign_fulfillment_event_shadow')) +}) + +test('明确渠道即视为已路由，不用渠道配置缺失覆盖已选渠道', () => { + assert.ok(migration.includes('v_fulfillment_item_state_raw')) + assert.ok(migration.includes("WHEN raw.intended_channel IS NOT NULL THEN 'configured'")) + assert.ok( + migration.includes( + "WHEN raw.route_status = 'missing_channel'\n AND raw.intended_channel IS NOT NULL\n THEN ''", + ), + ) +}) + +test('审计脚本核对影子表、执行尝试、开放异常和事件关联', () => { + assert.ok(auditScript.includes('影子履约缺失')) + assert.ok(auditScript.includes('影子履约状态差异')) + assert.ok(auditScript.includes('影子执行尝试缺失')) + assert.ok(auditScript.includes('影子开放异常缺失')) + assert.ok(auditScript.includes('履约事件未关联影子记录')) +}) diff --git a/apps/backend/src/services/fulfillment/fulfillment-state-service.integration.test.ts b/apps/backend/src/services/fulfillment/fulfillment-state-service.integration.test.ts index 3707b316..fda1f1f3 100644 --- a/apps/backend/src/services/fulfillment/fulfillment-state-service.integration.test.ts +++ b/apps/backend/src/services/fulfillment/fulfillment-state-service.integration.test.ts @@ -27,6 +27,7 @@ test( let orderItemId = 0 let manualConfigId = 0 let productId = 0 + let profileId = 0 try { const order = await query<{ id: number }>( @@ -51,6 +52,19 @@ test( ) orderItemId = Number(item.rows[0]!.id) + const initialShadow = await query<{ + route_status: string + carrier_status: string + issue_code: string + }>( + `SELECT route_status, carrier_status, issue_code + FROM fulfillments + WHERE order_item_id = $1`, + [orderItemId], + ) + assert.equal(initialShadow.rows[0]?.route_status, 'unrouted') + assert.equal(initialShadow.rows[0]?.issue_code, 'missing_channel') + const pendingDetail = { internalSku: `状态事件测试-${suffix}`, channel: 'manual', @@ -146,6 +160,33 @@ test( assert.equal(events[0]?.previous_channel, null) assert.equal(events[0]?.next_channel, 'third_party_voucher') + const assignedShadow = await query<{ + route_status: string + channel: string + carrier_status: string + issue_code: string + event_count: number + }>( + `SELECT + f.route_status, + f.channel, + f.carrier_status, + f.issue_code, + ( + SELECT count(*)::int + FROM fulfillment_state_events event + WHERE event.fulfillment_id = f.id + ) AS event_count + FROM fulfillments f + WHERE f.order_item_id = $1`, + [orderItemId], + ) + assert.equal(assignedShadow.rows[0]?.route_status, 'configured') + assert.equal(assignedShadow.rows[0]?.channel, 'third_party_voucher') + assert.equal(assignedShadow.rows[0]?.carrier_status, 'not_required') + assert.equal(assignedShadow.rows[0]?.issue_code, '') + assert.equal(Number(assignedShadow.rows[0]?.event_count), 3) + const product = await query<{ id: number }>( `INSERT INTO product_catalog ( canonical_sku, canonical_name, category, unit_name, status, @@ -193,6 +234,112 @@ test( assert.equal(Number(configured[0]?.fulfillment_config_id), manualConfigId) assert.equal(Number(configured[0]?.canonical_product_id), productId) + const profile = await query<{ id: number }>( + `INSERT INTO fulfillment_profiles ( + profile_key, name, executor_key, requires_claim, auto_dispatch, + inventory_strategy, config_json, created_at, updated_at + ) VALUES ($1, '状态事件任务', 'affiliate_dash', TRUE, FALSE, + 'external_platform', '{}'::jsonb, NOW(), NOW()) + RETURNING id`, + [`ff-state-profile-${suffix}`], + ) + profileId = Number(profile.rows[0]!.id) + await query( + `INSERT INTO fulfillment_tasks ( + order_id, order_item_id, unit_index, task_no, provider, platform, + shop_id, shop_name, platform_order_id, profile_id, executor_key, + task_status, attempt_no, is_current, created_at, updated_at + ) + SELECT + item.order_id, item.id, 1, $2, source.provider, source.platform, + source.shop_id, source.shop_name, source.platform_order_id, $3, 'affiliate_dash', + 'paid', 1, TRUE, NOW(), NOW() + FROM order_items item + JOIN orders source ON source.id = item.order_id + WHERE item.id = $1`, + [orderItemId, `FF-SHADOW-TASK-${suffix}`, profileId], + ) + const attempts = await query<{ + attempt_type: string + unit_index: number + attempt_no: number + source_id: number + is_current: boolean + }>( + `SELECT attempt_type, unit_index, attempt_no, source_id, is_current + FROM fulfillment_attempts + WHERE fulfillment_id = ( + SELECT id FROM fulfillments WHERE order_item_id = $1 + ) + ORDER BY id`, + [orderItemId], + ) + assert.equal(attempts.rows.length, 1) + assert.equal(attempts.rows[0]?.attempt_type, 'task') + assert.equal(attempts.rows[0]?.unit_index, 1) + assert.equal(attempts.rows[0]?.attempt_no, 1) + assert.equal(attempts.rows[0]?.is_current, true) + + const shadowDiff = await query<{ + fulfillment_state_diffs: number + missing_attempts: number + missing_open_issues: number + unlinked_events: number + }>( + `SELECT + COUNT(*) FILTER ( + WHERE ROW( + fulfillment.route_status, + fulfillment.channel, + fulfillment.carrier_status, + fulfillment.current_attempt_type, + fulfillment.current_attempt_source_id, + fulfillment.issue_code, + fulfillment.issue_category + ) IS DISTINCT FROM ROW( + CASE WHEN state.route_status = 'configured' THEN 'configured' ELSE 'unrouted' END, + state.intended_channel, + state.carrier_status, + state.current_attempt_type, + NULLIF(state.current_attempt_id, 0), + state.issue_code, + state.issue_category + ) + )::int AS fulfillment_state_diffs, + ( + SELECT COUNT(*)::int + FROM fulfillment_tasks task + WHERE task.order_item_id = $1 + AND NOT EXISTS ( + SELECT 1 FROM fulfillment_attempts attempt + WHERE attempt.attempt_type = 'task' AND attempt.source_id = task.id + ) + ) AS missing_attempts, + COUNT(*) FILTER ( + WHERE state.issue_code <> '' + AND NOT EXISTS ( + SELECT 1 FROM fulfillment_issues issue + WHERE issue.fulfillment_id = fulfillment.id + AND issue.issue_code = state.issue_code + AND issue.status <> 'resolved' + ) + )::int AS missing_open_issues, + ( + SELECT COUNT(*)::int + FROM fulfillment_state_events event + WHERE event.order_item_id = $1 + AND event.fulfillment_id IS NULL + ) AS unlinked_events + FROM v_fulfillment_item_state state + JOIN fulfillments fulfillment ON fulfillment.order_item_id = state.order_item_id + WHERE state.order_item_id = $1`, + [orderItemId], + ) + assert.equal(Number(shadowDiff.rows[0]?.fulfillment_state_diffs), 0) + assert.equal(Number(shadowDiff.rows[0]?.missing_attempts), 0) + assert.equal(Number(shadowDiff.rows[0]?.missing_open_issues), 0) + assert.equal(Number(shadowDiff.rows[0]?.unlinked_events), 0) + await assert.rejects( query(`UPDATE fulfillment_state_events SET reason = 'tampered' WHERE order_item_id = $1`, [ orderItemId, @@ -209,6 +356,9 @@ test( if (productId > 0) { await query(`DELETE FROM product_catalog WHERE id = $1`, [productId]) } + if (profileId > 0) { + await query(`DELETE FROM fulfillment_profiles WHERE id = $1`, [profileId]) + } if (orderId > 0) { await query(`DELETE FROM orders WHERE id = $1`, [orderId]) } diff --git a/docs/README.md b/docs/README.md index 77fff2ff..1076b9a8 100644 --- a/docs/README.md +++ b/docs/README.md @@ -1,6 +1,6 @@ # order_site 文档索引 -> 最后补充：2026-11-05，新增《履约系统架构复盘与优化方案》；生产代码基线 `e8b76b3b` / 迁移 `136`，开发分支 `dev_11.5_2` 已推进到迁移 `140`（履约统一状态写入口与追加事件）。 +> 最后补充：2026-11-05，新增《履约系统架构复盘与优化方案》；生产代码基线 `e8b76b3b` / 迁移 `136`，开发分支 `dev_11.5_2` 已推进到迁移 `141`（履约统一状态、追加事件与正式影子模型）。 > 数据基线：生产 dump `backups/postgres/order_site-20261022-092601.dump`（数据 2026-07-10 ~ 2026-10-22）；历史全量数字仅作基线，**上线验收以 [财务上线初始化与验收-2026-10-26.md](财务上线初始化与验收-2026-10-26.md) 为准**。 > 本文是 `docs/` 的唯一入口。**整体财务流程看 [财务流程总览-2026-11-01.md](财务流程总览-2026-11-01.md)**；数据模型看 [成本中心与履约模型.md](成本中心与履约模型.md)，接口看 [成本中心API契约.md](成本中心API契约.md)。 > 新增文档必须在下表登记状态；过期文档不再直接改写结论，统一移入 `docs/archive/`。 diff --git a/docs/履约系统架构复盘与优化方案-2026-11-05.md b/docs/履约系统架构复盘与优化方案-2026-11-05.md index 3c77f5a1..c9cf399d 100644 --- a/docs/履约系统架构复盘与优化方案-2026-11-05.md +++ b/docs/履约系统架构复盘与优化方案-2026-11-05.md @@ -10,7 +10,7 @@ status: draft > 状态：**方案**，尚未实施。 > 最后校准：2026-11-05。 > 代码基线：`e8b76b3b` + `dev_11.5_2` 履约架构改造。 -> 数据库基线：生产迁移 `136`；开发分支迁移 `140`。 +> 数据库基线：生产迁移 `136`；开发分支迁移 `141`。 > 生产核对方式：只读 SQL、后台页面、现行文档与源码交叉检查。 ## 实施进度 @@ -31,7 +31,10 @@ status: draft - 已完成：人工履约配置、标准商品关联、自动发货/接单平台商品回填、模板重置和调度补挂统一通过状态映射接口写入，并校验并发期间的原商品值。 - 已完成：任务创建先锁订单项并复用已有活跃尝试，避免并发调度重复插入同一当前尝试。 - 已完成：新增真实 PostgreSQL 状态事件集成测试；全新库 `001 → 140` 迁移和既有履约路由集成测试通过。 -- 待继续：正式 `fulfillments / attempts / issues` 表、独立售后队列和生产存量修复。 +- 已完成：迁移 `141` 建立 `fulfillments / fulfillment_attempts / fulfillment_issues` 影子模型，从现有读模型回填，并由订单项、任务、工单触发器持续双写。 +- 已完成：审计脚本增加影子履约、尝试、开放异常和事件关联差异检查；当前读路径仍保持原表，未切换。 +- 已完成：修正统一视图“明确渠道仍被判定为缺渠道”的旧口径，已有渠道直接视为已路由。 +- 待继续：观察生产影子差异、切换读路径、停止旧字段写入、独立售后队列和生产存量修复。 ### 历史数据边界