iloom-flatten/migrations/20260604_200000_data_integrity_constraints.sql

-- ============================================================================
-- 数据完整性强约束 - 确保财务与核心数据100%正确
-- ============================================================================
-- 目标：系统错误率 ≤ 0.01% (万分之一)
-- 范围：production_plans, inventory_records, payments, accounts_payable
-- ============================================================================

-- ============================================================================
-- 1. 创建数据校验函数
-- ============================================================================

-- 1.1 验证计划数量一致性（completed_quantity = SUM(inventory_records.quantity)）
CREATE OR REPLACE FUNCTION validate_plan_quantity_consistency()
RETURNS TRIGGER LANGUAGE plpgsql AS $$
DECLARE
  v_inventory_total NUMERIC;
  v_tolerance NUMERIC := 0.01; -- 允许0.01米的浮点误差
BEGIN
  -- 仅当 completed_quantity 被更新时触发
  IF OLD.completed_quantity IS DISTINCT FROM NEW.completed_quantity THEN
    SELECT COALESCE(SUM(quantity), 0) INTO v_inventory_total
    FROM inventory_records
    WHERE plan_id = NEW.id;
    
    IF ABS(NEW.completed_quantity - v_inventory_total) > v_tolerance THEN
      RAISE EXCEPTION 
        '数据完整性错误：计划 % 的完成数量(%)与入库记录总和(%)不一致，差异: %',
        NEW.plan_code, NEW.completed_quantity, v_inventory_total, 
        NEW.completed_quantity - v_inventory_total;
    END IF;
  END IF;
  
  RETURN NEW;
END;
$$;

-- 1.2 验证入库记录不能为负数
CREATE OR REPLACE FUNCTION validate_inventory_record_positive()
RETURNS TRIGGER LANGUAGE plpgsql AS $$
BEGIN
  IF NEW.quantity < 0 THEN
    RAISE EXCEPTION '入库数量不能为负数: %', NEW.quantity;
  END IF;
  
  IF NEW.rolls < 0 THEN
    RAISE EXCEPTION '入库匹数不能为负数: %', NEW.rolls;
  END IF;
  
  IF NEW.price_per_meter < 0 THEN
    RAISE EXCEPTION '单价不能为负数: %', NEW.price_per_meter;
  END IF;
  
  RETURN NEW;
END;
$$;

-- 1.3 验证付款金额一致性
CREATE OR REPLACE FUNCTION validate_payment_amount()
RETURNS TRIGGER LANGUAGE plpgsql AS $$
DECLARE
  v_expected_amount NUMERIC;
  v_tolerance NUMERIC := 0.01;
BEGIN
  -- 计算预期金额 = 数量 × 单价
  v_expected_amount := NEW.quantity * NEW.price_per_meter;
  
  IF ABS(NEW.amount - v_expected_amount) > v_tolerance THEN
    RAISE EXCEPTION 
      '付款金额不一致：期望 %(数量 % × 单价 %)，实际 %',
      v_expected_amount, NEW.quantity, NEW.price_per_meter, NEW.amount;
  END IF;
  
  RETURN NEW;
END;
$$;

-- 1.4 验证应付账款汇总一致性
CREATE OR REPLACE FUNCTION validate_accounts_payable_totals()
RETURNS TRIGGER LANGUAGE plpgsql AS $$
DECLARE
  v_items_total NUMERIC;
  v_items_paid NUMERIC;
  v_tolerance NUMERIC := 0.01;
BEGIN
  -- 从明细表重新计算
  SELECT 
    COALESCE(SUM(amount), 0),
    COALESCE(SUM(CASE WHEN status = 'paid' THEN amount ELSE 0 END), 0)
  INTO v_items_total, v_items_paid
  FROM accounts_payable_items
  WHERE accounts_payable_id = NEW.id;
  
  -- 验证总金额
  IF ABS(NEW.total_amount - v_items_total) > v_tolerance THEN
    RAISE EXCEPTION 
      '应付账款总金额不一致：表头 %, 明细合计 %',
      NEW.total_amount, v_items_total;
  END IF;
  
  -- 验证已付金额
  IF ABS(NEW.paid_amount - v_items_paid) > v_tolerance THEN
    RAISE EXCEPTION 
      '应付账款已付金额不一致：表头 %, 明细合计 %',
      NEW.paid_amount, v_items_paid;
  END IF;
  
  -- 验证未付金额
  IF ABS(NEW.unpaid_amount - (v_items_total - v_items_paid)) > v_tolerance THEN
    RAISE EXCEPTION 
      '应付账款未付金额不一致：表头 %, 计算值 %',
      NEW.unpaid_amount, (v_items_total - v_items_paid);
  END IF;
  
  RETURN NEW;
END;
$$;

-- ============================================================================
-- 2. 创建触发器
-- ============================================================================

-- 2.1 计划数量一致性触发器
DROP TRIGGER IF EXISTS trg_validate_plan_quantity ON production_plans;
CREATE TRIGGER trg_validate_plan_quantity
  BEFORE UPDATE ON production_plans
  FOR EACH ROW
  WHEN (OLD.completed_quantity IS DISTINCT FROM NEW.completed_quantity)
  EXECUTE FUNCTION validate_plan_quantity_consistency();

-- 2.2 入库记录正数验证触发器
DROP TRIGGER IF EXISTS trg_validate_inventory_positive ON inventory_records;
CREATE TRIGGER trg_validate_inventory_positive
  BEFORE INSERT OR UPDATE ON inventory_records
  FOR EACH ROW
  EXECUTE FUNCTION validate_inventory_record_positive();

-- 2.3 付款金额验证触发器
DROP TRIGGER IF EXISTS trg_validate_payment ON payments;
CREATE TRIGGER trg_validate_payment
  BEFORE INSERT OR UPDATE ON payments
  FOR EACH ROW
  EXECUTE FUNCTION validate_payment_amount();

-- 2.4 应付账款汇总验证触发器
DROP TRIGGER IF EXISTS trg_validate_accounts_payable ON accounts_payable;
CREATE TRIGGER trg_validate_accounts_payable
  BEFORE INSERT OR UPDATE ON accounts_payable
  FOR EACH ROW
  EXECUTE FUNCTION validate_accounts_payable_totals();

-- ============================================================================
-- 3. 创建审计日志表（记录所有关键数据变更）
-- ============================================================================

CREATE TABLE IF NOT EXISTS data_audit_logs (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  table_name TEXT NOT NULL,
  record_id UUID NOT NULL,
  operation TEXT NOT NULL CHECK (operation IN ('INSERT', 'UPDATE', 'DELETE')),
  old_values JSONB,
  new_values JSONB,
  changed_by UUID REFERENCES auth.users(id),
  changed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
  validation_passed BOOLEAN NOT NULL DEFAULT TRUE,
  error_message TEXT
);

-- 创建索引加速查询
CREATE INDEX IF NOT EXISTS idx_audit_logs_table_record ON data_audit_logs(table_name, record_id);
CREATE INDEX IF NOT EXISTS idx_audit_logs_changed_at ON data_audit_logs(changed_at DESC);
CREATE INDEX IF NOT EXISTS idx_audit_logs_validation ON data_audit_logs(validation_passed) WHERE validation_passed = FALSE;

-- 启用 RLS
ALTER TABLE data_audit_logs ENABLE ROW LEVEL SECURITY;

-- 审计日志只读策略（仅管理员可查看）
CREATE POLICY admin_select_audit_logs ON data_audit_logs
  FOR SELECT USING (true); -- 暂时开放，后续可改为角色检查

CREATE POLICY system_insert_audit_logs ON data_audit_logs
  FOR INSERT WITH CHECK (true); -- 系统自动写入

-- ============================================================================
-- 4. 创建审计触发器函数
-- ============================================================================

CREATE OR REPLACE FUNCTION log_data_audit()
RETURNS TRIGGER LANGUAGE plpgsql AS $$
DECLARE
  v_old_values JSONB;
  v_new_values JSONB;
  v_operation TEXT;
BEGIN
  v_operation := TG_OP;
  
  IF TG_OP = 'DELETE' THEN
    v_old_values := to_jsonb(OLD);
    v_new_values := NULL;
  ELSIF TG_OP = 'UPDATE' THEN
    v_old_values := to_jsonb(OLD);
    v_new_values := to_jsonb(NEW);
  ELSE -- INSERT
    v_old_values := NULL;
    v_new_values := to_jsonb(NEW);
  END IF;
  
  INSERT INTO data_audit_logs (
    table_name, record_id, operation, old_values, new_values, changed_by
  ) VALUES (
    TG_TABLE_NAME,
    COALESCE(NEW.id, OLD.id),
    v_operation,
    v_old_values,
    v_new_values,
    auth.uid()
  );
  
  IF TG_OP = 'DELETE' THEN
    RETURN OLD;
  ELSE
    RETURN NEW;
  END IF;
END;
$$;

-- ============================================================================
-- 5. 为核心表添加审计触发器
-- ============================================================================

-- 5.1 production_plans 审计
DROP TRIGGER IF EXISTS trg_audit_production_plans ON production_plans;
CREATE TRIGGER trg_audit_production_plans
  AFTER INSERT OR UPDATE OR DELETE ON production_plans
  FOR EACH ROW
  EXECUTE FUNCTION log_data_audit();

-- 5.2 inventory_records 审计
DROP TRIGGER IF EXISTS trg_audit_inventory_records ON inventory_records;
CREATE TRIGGER trg_audit_inventory_records
  AFTER INSERT OR UPDATE OR DELETE ON inventory_records
  FOR EACH ROW
  EXECUTE FUNCTION log_data_audit();

-- 5.3 payments 审计
DROP TRIGGER IF EXISTS trg_audit_payments ON payments;
CREATE TRIGGER trg_audit_payments
  AFTER INSERT OR UPDATE OR DELETE ON payments
  FOR EACH ROW
  EXECUTE FUNCTION log_data_audit();

-- 5.4 accounts_payable 审计
DROP TRIGGER IF EXISTS trg_audit_accounts_payable ON accounts_payable;
CREATE TRIGGER trg_audit_accounts_payable
  AFTER INSERT OR UPDATE OR DELETE ON accounts_payable
  FOR EACH ROW
  EXECUTE FUNCTION log_data_audit();

-- 5.5 accounts_payable_items 审计
DROP TRIGGER IF EXISTS trg_audit_accounts_payable_items ON accounts_payable_items;
CREATE TRIGGER trg_audit_accounts_payable_items
  AFTER INSERT OR UPDATE OR DELETE ON accounts_payable_items
  FOR EACH ROW
  EXECUTE FUNCTION log_data_audit();

-- ============================================================================
-- 6. 创建数据一致性检查视图
-- ============================================================================

-- 6.1 计划数量一致性检查视图
CREATE OR REPLACE VIEW v_plan_quantity_check AS
SELECT 
  p.id as plan_id,
  p.plan_code,
  p.status,
  p.completed_quantity,
  COALESCE(SUM(ir.quantity), 0) as inventory_total,
  p.completed_quantity - COALESCE(SUM(ir.quantity), 0) as difference,
  CASE 
    WHEN ABS(p.completed_quantity - COALESCE(SUM(ir.quantity), 0)) <= 0.01 THEN '✓ 一致'
    ELSE '✗ 不一致'
  END as check_result
FROM production_plans p
LEFT JOIN inventory_records ir ON ir.plan_id = p.id
GROUP BY p.id, p.plan_code, p.status, p.completed_quantity;

-- 6.2 付款金额一致性检查视图
CREATE OR REPLACE VIEW v_payment_check AS
SELECT 
  id as payment_id,
  plan_id,
  quantity,
  price_per_meter,
  amount,
  quantity * price_per_meter as expected_amount,
  amount - (quantity * price_per_meter) as difference,
  CASE 
    WHEN ABS(amount - (quantity * price_per_meter)) <= 0.01 THEN '✓ 一致'
    ELSE '✗ 不一致'
  END as check_result
FROM payments;

-- 6.3 应付账款汇总一致性检查视图
CREATE OR REPLACE VIEW v_accounts_payable_check AS
SELECT 
  ap.id as ap_id,
  ap.plan_id,
  ap.total_amount,
  ap.paid_amount,
  ap.unpaid_amount,
  COALESCE(SUM(api.amount), 0) as items_total,
  COALESCE(SUM(CASE WHEN api.status = 'paid' THEN api.amount ELSE 0 END), 0) as items_paid,
  ap.total_amount - COALESCE(SUM(api.amount), 0) as total_diff,
  ap.paid_amount - COALESCE(SUM(CASE WHEN api.status = 'paid' THEN api.amount ELSE 0 END), 0) as paid_diff,
  CASE 
    WHEN ABS(ap.total_amount - COALESCE(SUM(api.amount), 0)) <= 0.01 
     AND ABS(ap.paid_amount - COALESCE(SUM(CASE WHEN api.status = 'paid' THEN api.amount ELSE 0 END), 0)) <= 0.01
    THEN '✓ 一致'
    ELSE '✗ 不一致'
  END as check_result
FROM accounts_payable ap
LEFT JOIN accounts_payable_items api ON api.accounts_payable_id = ap.id
GROUP BY ap.id, ap.plan_id, ap.total_amount, ap.paid_amount, ap.unpaid_amount;

-- ============================================================================
-- 7. 创建定期一致性检查函数（可由 Edge Function 调用）
-- ============================================================================

CREATE OR REPLACE FUNCTION run_data_integrity_check()
RETURNS TABLE (
  check_type TEXT,
  total_records BIGINT,
  inconsistent_records BIGINT,
  error_rate NUMERIC,
  status TEXT
) LANGUAGE plpgsql AS $$
BEGIN
  -- 检查计划数量一致性
  RETURN QUERY
  SELECT 
    'plan_quantity'::TEXT,
    COUNT(*)::BIGINT,
    SUM(CASE WHEN check_result = '✗ 不一致' THEN 1 ELSE 0 END)::BIGINT,
    ROUND(
      SUM(CASE WHEN check_result = '✗ 不一致' THEN 1 ELSE 0 END)::NUMERIC / 
      NULLIF(COUNT(*), 0) * 100, 4
    ),
    CASE 
      WHEN SUM(CASE WHEN check_result = '✗ 不一致' THEN 1 ELSE 0 END) = 0 THEN 'PASS'
      ELSE 'FAIL'
    END
  FROM v_plan_quantity_check;
  
  -- 检查付款金额一致性
  RETURN QUERY
  SELECT 
    'payment_amount'::TEXT,
    COUNT(*)::BIGINT,
    SUM(CASE WHEN check_result = '✗ 不一致' THEN 1 ELSE 0 END)::BIGINT,
    ROUND(
      SUM(CASE WHEN check_result = '✗ 不一致' THEN 1 ELSE 0 END)::NUMERIC / 
      NULLIF(COUNT(*), 0) * 100, 4
    ),
    CASE 
      WHEN SUM(CASE WHEN check_result = '✗ 不一致' THEN 1 ELSE 0 END) = 0 THEN 'PASS'
      ELSE 'FAIL'
    END
  FROM v_payment_check;
  
  -- 检查应付账款汇总一致性
  RETURN QUERY
  SELECT 
    'accounts_payable'::TEXT,
    COUNT(*)::BIGINT,
    SUM(CASE WHEN check_result = '✗ 不一致' THEN 1 ELSE 0 END)::BIGINT,
    ROUND(
      SUM(CASE WHEN check_result = '✗ 不一致' THEN 1 ELSE 0 END)::NUMERIC / 
      NULLIF(COUNT(*), 0) * 100, 4
    ),
    CASE 
      WHEN SUM(CASE WHEN check_result = '✗ 不一致' THEN 1 ELSE 0 END) = 0 THEN 'PASS'
      ELSE 'FAIL'
    END
  FROM v_accounts_payable_check;
END;
$$;

-- ============================================================================
-- 8. 添加 CHECK 约束（作为最后一道防线）
-- ============================================================================

-- 8.1 入库记录非负约束
ALTER TABLE inventory_records 
  DROP CONSTRAINT IF EXISTS chk_inventory_positive;
ALTER TABLE inventory_records 
  ADD CONSTRAINT chk_inventory_positive 
  CHECK (quantity >= 0 AND rolls >= 0 AND (price_per_meter IS NULL OR price_per_meter >= 0));

-- 8.2 付款金额非负约束
ALTER TABLE payments 
  DROP CONSTRAINT IF EXISTS chk_payment_positive;
ALTER TABLE payments 
  ADD CONSTRAINT chk_payment_positive 
  CHECK (amount >= 0 AND quantity >= 0 AND price_per_meter >= 0);

-- 8.3 应付账款金额逻辑约束
ALTER TABLE accounts_payable 
  DROP CONSTRAINT IF EXISTS chk_ap_amounts;
ALTER TABLE accounts_payable 
  ADD CONSTRAINT chk_ap_amounts 
  CHECK (
    total_amount >= 0 AND 
    paid_amount >= 0 AND 
    unpaid_amount >= 0 AND
    paid_amount <= total_amount AND
    unpaid_amount <= total_amount
  );

-- ============================================================================
-- 9. 创建错误率监控视图
-- ============================================================================

CREATE OR REPLACE VIEW v_error_rate_monitor AS
SELECT 
  DATE_TRUNC('day', changed_at) as check_date,
  table_name,
  COUNT(*) as total_operations,
  SUM(CASE WHEN validation_passed = FALSE THEN 1 ELSE 0 END) as failed_operations,
  ROUND(
    SUM(CASE WHEN validation_passed = FALSE THEN 1 ELSE 0 END)::NUMERIC / 
    NULLIF(COUNT(*), 0) * 10000, 2
  ) as error_rate_per_10k,
  CASE 
    WHEN SUM(CASE WHEN validation_passed = FALSE THEN 1 ELSE 0 END)::NUMERIC / 
         NULLIF(COUNT(*), 0) * 10000 <= 1 THEN '✓ 达标 (≤1/万)'
    ELSE '✗ 超标 (>1/万)'
  END as target_status
FROM data_audit_logs
WHERE changed_at >= NOW() - INTERVAL '30 days'
GROUP BY DATE_TRUNC('day', changed_at), table_name
ORDER BY check_date DESC, table_name;

-- ============================================================================
-- 结束
-- ============================================================================