feat(fulfillment): 建立正式履约影子模型

apps/backend/scripts/audit-fulfillment-state.ts

@@ -93,11 +93,100 @@ async function loadRecentDuplicateAttempts(): Promise<CountRow[]> {
   return result.rows
 }
 
+async function loadShadowModelDiffs(): Promise<CountRow[]> {
+  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<void> {
-  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<void> {
     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

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 后新旧事件都自动关联';

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('履约事件未关联影子记录'))
+})

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])
       }

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/`。

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` 影子模型，从现有读模型回填，并由订单项、任务、工单触发器持续双写。
+- 已完成：审计脚本增加影子履约、尝试、开放异常和事件关联差异检查；当前读路径仍保持原表，未切换。
+- 已完成：修正统一视图“明确渠道仍被判定为缺渠道”的旧口径，已有渠道直接视为已路由。
+- 待继续：观察生产影子差异、切换读路径、停止旧字段写入、独立售后队列和生产存量修复。
 
 ### 历史数据边界