-- 通用货品台账系统 PostgreSQL schema v1
-- 设计范围：合同/采购订单/渠道分配/样品库存/出入库/签收/核对快照/MCP审计

CREATE EXTENSION IF NOT EXISTS pgcrypto;

-- ========== 基础主数据 ==========

CREATE TABLE IF NOT EXISTS parties (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    party_code TEXT UNIQUE,
    display_name TEXT NOT NULL,
    legal_name TEXT,
    short_name TEXT,
    party_type TEXT NOT NULL CHECK (party_type IN (
        'own_company', 'supplier', 'customer', 'channel', 'warehouse', 'person', 'platform', 'other'
    )),
    contact_masked TEXT,
    notes TEXT,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE TABLE IF NOT EXISTS projects (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    project_code TEXT UNIQUE,
    name TEXT NOT NULL,
    ip_name TEXT,
    owner_party_id UUID REFERENCES parties(id),
    status TEXT NOT NULL DEFAULT 'active' CHECK (status IN ('active', 'paused', 'archived')),
    notes TEXT,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE TABLE IF NOT EXISTS products (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    sku_code TEXT UNIQUE,
    standard_name TEXT NOT NULL,
    raw_name_default TEXT,
    category TEXT,
    spec TEXT,
    default_unit TEXT,
    barcode_69 TEXT,
    guobo_code TEXT,
    status TEXT NOT NULL DEFAULT 'active' CHECK (status IN ('draft', 'active', 'inactive')),
    notes TEXT,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE TABLE IF NOT EXISTS source_files (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    original_filename TEXT NOT NULL,
    stored_path TEXT,
    source_cache_path TEXT,
    sha256 TEXT,
    byte_size BIGINT,
    mime_type TEXT,
    material_type TEXT CHECK (material_type IN (
        'contract', 'purchase_order', 'supplemental_demand', 'shipment_signoff',
        'ledger', 'extraction_csv', 'image', 'other'
    )),
    received_at TIMESTAMPTZ,
    extraction_status TEXT NOT NULL DEFAULT 'pending' CHECK (extraction_status IN (
        'pending', 'parsed', 'failed', 'skipped', 'needs_review'
    )),
    sensitivity_level TEXT NOT NULL DEFAULT 'internal' CHECK (sensitivity_level IN (
        'public', 'internal', 'sensitive', 'restricted'
    )),
    notes TEXT,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    UNIQUE (sha256)
);

CREATE TABLE IF NOT EXISTS product_aliases (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    product_id UUID REFERENCES products(id) ON DELETE CASCADE,
    alias_name TEXT NOT NULL,
    alias_type TEXT NOT NULL DEFAULT 'raw_name' CHECK (alias_type IN (
        'raw_name', 'spoken_name', 'sheet_name', 'old_name', 'barcode_name', 'other'
    )),
    source_file_id UUID REFERENCES source_files(id),
    confidence NUMERIC(5,4),
    verification_status TEXT NOT NULL DEFAULT 'pending' CHECK (verification_status IN (
        'pending', 'confirmed', 'rejected', 'conflict'
    )),
    notes TEXT,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    UNIQUE (product_id, alias_name, alias_type)
);

-- ========== 单据与订单 ==========

CREATE TABLE IF NOT EXISTS business_documents (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    document_code TEXT UNIQUE,
    document_type TEXT NOT NULL CHECK (document_type IN (
        'supplier_contract', 'cooperation_agreement', 'upstream_purchase_order',
        'supplemental_demand', 'shipment_signoff', 'internal_ledger', 'other'
    )),
    source_file_id UUID REFERENCES source_files(id),
    project_id UUID REFERENCES projects(id),
    own_party_id UUID REFERENCES parties(id),
    counterparty_id UUID REFERENCES parties(id),
    document_no TEXT,
    document_date DATE,
    title TEXT,
    status TEXT NOT NULL DEFAULT 'draft' CHECK (status IN ('draft', 'confirmed', 'superseded', 'cancelled')),
    raw_payload JSONB NOT NULL DEFAULT '{}'::jsonb,
    notes TEXT,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE TABLE IF NOT EXISTS purchase_orders (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    order_code TEXT UNIQUE,
    document_id UUID REFERENCES business_documents(id),
    project_id UUID REFERENCES projects(id),
    order_side TEXT NOT NULL CHECK (order_side IN (
        'downstream_supplier_purchase', 'upstream_customer_order',
        'internal_transfer', 'supplemental_demand'
    )),
    order_type TEXT NOT NULL DEFAULT 'formal_order' CHECK (order_type IN (
        'formal_order', 'contract_order', 'no_document_supplement', 'replenishment', 'adjustment'
    )),
    buyer_party_id UUID REFERENCES parties(id),
    seller_party_id UUID REFERENCES parties(id),
    order_no TEXT,
    order_date DATE,
    settlement_method TEXT,
    expected_arrival TEXT,
    status TEXT NOT NULL DEFAULT 'draft' CHECK (status IN ('draft', 'confirmed', 'cancelled', 'superseded')),
    raw_payload JSONB NOT NULL DEFAULT '{}'::jsonb,
    notes TEXT,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE TABLE IF NOT EXISTS order_lines (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    order_id UUID NOT NULL REFERENCES purchase_orders(id) ON DELETE CASCADE,
    line_no INTEGER NOT NULL DEFAULT 1,
    product_id UUID REFERENCES products(id),
    product_name_raw TEXT NOT NULL,
    product_name_standard_draft TEXT,
    quantity NUMERIC(18,4) NOT NULL CHECK (quantity >= 0),
    unit TEXT,
    unit_price NUMERIC(18,4),
    amount NUMERIC(18,2),
    material TEXT,
    evidence_note TEXT,
    verification_status TEXT NOT NULL DEFAULT 'pending' CHECK (verification_status IN (
        'pending', 'confirmed', 'conflict', 'rejected'
    )),
    raw_payload JSONB NOT NULL DEFAULT '{}'::jsonb,
    notes TEXT,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    UNIQUE (order_id, line_no)
);

CREATE TABLE IF NOT EXISTS channel_allocations (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    order_line_id UUID NOT NULL REFERENCES order_lines(id) ON DELETE CASCADE,
    channel_party_id UUID REFERENCES parties(id),
    raw_channel TEXT,
    standard_channel TEXT NOT NULL,
    original_quantity NUMERIC(18,4) NOT NULL DEFAULT 0,
    adjustment_quantity NUMERIC(18,4) NOT NULL DEFAULT 0,
    adjusted_quantity NUMERIC(18,4) GENERATED ALWAYS AS (original_quantity + adjustment_quantity) STORED,
    adjustment_reason TEXT,
    evidence_note TEXT,
    notes TEXT,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

-- ========== 库存与出入库 ==========

CREATE TABLE IF NOT EXISTS inventory_accounts (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    account_code TEXT UNIQUE,
    owner_party_id UUID NOT NULL REFERENCES parties(id),
    project_id UUID REFERENCES projects(id),
    account_type TEXT NOT NULL DEFAULT 'sample_stock' CHECK (account_type IN (
        'sample_stock', 'sellable_stock', 'gift_stock', 'returnable_claim', 'other'
    )),
    warehouse_party_id UUID REFERENCES parties(id),
    status TEXT NOT NULL DEFAULT 'active' CHECK (status IN ('active', 'closed')),
    notes TEXT,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE TABLE IF NOT EXISTS stock_items (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    stock_item_code TEXT UNIQUE,
    inventory_account_id UUID NOT NULL REFERENCES inventory_accounts(id) ON DELETE CASCADE,
    product_id UUID REFERENCES products(id),
    display_name TEXT NOT NULL,
    spec TEXT,
    unit TEXT,
    unit_cost_tax_included NUMERIC(18,4),
    source_policy TEXT,
    status TEXT NOT NULL DEFAULT 'active' CHECK (status IN ('active', 'inactive', 'draft')),
    notes TEXT,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE TABLE IF NOT EXISTS stock_movements (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    movement_code TEXT UNIQUE,
    stock_item_id UUID NOT NULL REFERENCES stock_items(id) ON DELETE CASCADE,
    movement_date DATE NOT NULL,
    direction TEXT NOT NULL CHECK (direction IN ('in', 'out')),
    movement_type TEXT NOT NULL CHECK (movement_type IN (
        'purchase_in', 'self_stock_in', 'shipment_out', 'claim_out', 'return_in',
        'gift_out', 'transfer_out', 'transfer_in', 'adjustment', 'scrap_out'
    )),
    quantity NUMERIC(18,4) NOT NULL CHECK (quantity >= 0),
    unit TEXT,
    handler_party_id UUID REFERENCES parties(id),
    counterparty_id UUID REFERENCES parties(id),
    purpose TEXT,
    returnable BOOLEAN,
    unit_cost_tax_included NUMERIC(18,4),
    source_document_id UUID REFERENCES business_documents(id),
    evidence_file_id UUID REFERENCES source_files(id),
    notes TEXT,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

-- ========== 发货签收 ==========

CREATE TABLE IF NOT EXISTS shipment_signoffs (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    signoff_code TEXT UNIQUE,
    project_id UUID REFERENCES projects(id),
    shipper_party_id UUID REFERENCES parties(id),
    receiver_party_id UUID REFERENCES parties(id),
    ship_date DATE,
    purpose TEXT,
    status TEXT NOT NULL DEFAULT 'draft' CHECK (status IN ('draft', 'sent', 'signed', 'cancelled')),
    source_document_id UUID REFERENCES business_documents(id),
    notes TEXT,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE TABLE IF NOT EXISTS shipment_signoff_lines (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    signoff_id UUID NOT NULL REFERENCES shipment_signoffs(id) ON DELETE CASCADE,
    line_no INTEGER NOT NULL DEFAULT 1,
    product_id UUID REFERENCES products(id),
    product_name TEXT NOT NULL,
    quantity NUMERIC(18,4) NOT NULL CHECK (quantity >= 0),
    unit TEXT,
    stock_movement_id UUID REFERENCES stock_movements(id),
    sign_status TEXT NOT NULL DEFAULT 'pending' CHECK (sign_status IN ('pending', 'signed', 'rejected')),
    notes TEXT,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    UNIQUE (signoff_id, line_no)
);

-- ========== 计算快照 ==========

CREATE TABLE IF NOT EXISTS reconciliation_runs (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    run_code TEXT UNIQUE,
    project_id UUID REFERENCES projects(id),
    reconciliation_type TEXT NOT NULL CHECK (reconciliation_type IN (
        'supplier_vs_upstream_demand', 'channel_quantity_check', 'stock_snapshot', 'other'
    )),
    policy_version TEXT,
    input_scope JSONB NOT NULL DEFAULT '{}'::jsonb,
    run_status TEXT NOT NULL DEFAULT 'draft' CHECK (run_status IN ('draft', 'completed', 'failed', 'superseded')),
    run_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    notes TEXT,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE TABLE IF NOT EXISTS reconciliation_lines (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    run_id UUID NOT NULL REFERENCES reconciliation_runs(id) ON DELETE CASCADE,
    product_id UUID REFERENCES products(id),
    product_name_standard_draft TEXT NOT NULL,
    supplier_purchase_quantity_total NUMERIC(18,4) NOT NULL DEFAULT 0,
    upstream_demand_quantity_total NUMERIC(18,4) NOT NULL DEFAULT 0,
    formal_order_quantity_total NUMERIC(18,4) NOT NULL DEFAULT 0,
    supplement_demand_quantity_total NUMERIC(18,4) NOT NULL DEFAULT 0,
    difference_supplier_minus_demand NUMERIC(18,4) NOT NULL DEFAULT 0,
    self_stock_candidate_qty NUMERIC(18,4) NOT NULL DEFAULT 0,
    shortage_qty NUMERIC(18,4) NOT NULL DEFAULT 0,
    difference_status TEXT,
    related_supplier_sources TEXT,
    related_upstream_sources TEXT,
    raw_payload JSONB NOT NULL DEFAULT '{}'::jsonb,
    notes TEXT,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

-- ========== 导入、校验与机器人写入审计 ==========

CREATE TABLE IF NOT EXISTS import_batches (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    batch_code TEXT UNIQUE,
    source_file_id UUID REFERENCES source_files(id),
    import_type TEXT NOT NULL,
    profile_name TEXT,
    status TEXT NOT NULL DEFAULT 'pending' CHECK (status IN ('pending', 'running', 'completed', 'failed', 'needs_review')),
    started_at TIMESTAMPTZ,
    finished_at TIMESTAMPTZ,
    summary JSONB NOT NULL DEFAULT '{}'::jsonb,
    notes TEXT,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

-- Raw archive row table: keeps every source Excel/CSV row as JSONB so no source field is lost
-- even before a column is promoted into a normalized business table.
CREATE TABLE IF NOT EXISTS raw_import_rows (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    import_batch_id UUID REFERENCES import_batches(id),
    source_label TEXT NOT NULL,
    source_file_path TEXT NOT NULL,
    sheet_or_table TEXT NOT NULL,
    row_number INTEGER NOT NULL,
    row_payload JSONB NOT NULL,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE TABLE IF NOT EXISTS validation_issues (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    issue_type TEXT NOT NULL CHECK (issue_type IN (
        'missing_field', 'conflict', 'quantity_mismatch', 'channel_mismatch',
        'duplicate_file', 'unknown_product', 'unknown_party', 'negative_stock', 'other'
    )),
    severity TEXT NOT NULL DEFAULT 'warning' CHECK (severity IN ('info', 'warning', 'error')),
    related_table TEXT,
    related_id UUID,
    field_name TEXT,
    message TEXT NOT NULL,
    status TEXT NOT NULL DEFAULT 'open' CHECK (status IN ('open', 'resolved', 'ignored')),
    source_file_id UUID REFERENCES source_files(id),
    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    resolved_at TIMESTAMPTZ
);

CREATE TABLE IF NOT EXISTS agent_write_audit (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    agent_profile TEXT,
    tool_name TEXT,
    operation TEXT NOT NULL CHECK (operation IN ('create', 'update', 'import', 'compute', 'delete_request')),
    target_table TEXT,
    target_id UUID,
    request_payload JSONB NOT NULL DEFAULT '{}'::jsonb,
    result_status TEXT NOT NULL CHECK (result_status IN ('success', 'failed', 'needs_review')),
    result_message TEXT,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

-- ========== 常用索引 ==========

CREATE INDEX IF NOT EXISTS idx_parties_display_name ON parties(display_name);
CREATE INDEX IF NOT EXISTS idx_projects_ip_name ON projects(ip_name);
CREATE INDEX IF NOT EXISTS idx_products_standard_name ON products(standard_name);
CREATE INDEX IF NOT EXISTS idx_product_aliases_alias_name ON product_aliases(alias_name);
CREATE INDEX IF NOT EXISTS idx_source_files_sha256 ON source_files(sha256);
CREATE INDEX IF NOT EXISTS idx_business_documents_project ON business_documents(project_id);
CREATE INDEX IF NOT EXISTS idx_purchase_orders_project_side ON purchase_orders(project_id, order_side);
CREATE INDEX IF NOT EXISTS idx_purchase_orders_date ON purchase_orders(order_date);
CREATE INDEX IF NOT EXISTS idx_order_lines_order ON order_lines(order_id);
CREATE INDEX IF NOT EXISTS idx_order_lines_product ON order_lines(product_id);
CREATE INDEX IF NOT EXISTS idx_channel_allocations_line ON channel_allocations(order_line_id);
CREATE INDEX IF NOT EXISTS idx_inventory_accounts_owner_project ON inventory_accounts(owner_party_id, project_id);
CREATE INDEX IF NOT EXISTS idx_stock_items_account ON stock_items(inventory_account_id);
CREATE INDEX IF NOT EXISTS idx_stock_movements_item_date ON stock_movements(stock_item_id, movement_date);
CREATE INDEX IF NOT EXISTS idx_validation_issues_status ON validation_issues(status, severity);
CREATE INDEX IF NOT EXISTS idx_raw_import_rows_source ON raw_import_rows(source_label, sheet_or_table);

-- ========== 计算视图 ==========

CREATE OR REPLACE VIEW v_order_line_channel_totals AS
SELECT
    ol.id AS order_line_id,
    ol.order_id,
    ol.product_id,
    COALESCE(ol.product_name_standard_draft, ol.product_name_raw) AS product_name,
    ol.quantity AS order_quantity,
    COALESCE(SUM(ca.adjusted_quantity), 0) AS allocated_quantity,
    ol.quantity - COALESCE(SUM(ca.adjusted_quantity), 0) AS allocation_diff
FROM order_lines ol
LEFT JOIN channel_allocations ca ON ca.order_line_id = ol.id
GROUP BY ol.id, ol.order_id, ol.product_id, ol.product_name_standard_draft, ol.product_name_raw, ol.quantity;

CREATE OR REPLACE VIEW v_current_stock AS
SELECT
    si.id AS stock_item_id,
    si.stock_item_code,
    ia.owner_party_id,
    ia.project_id,
    si.product_id,
    si.display_name,
    si.unit,
    SUM(CASE WHEN sm.direction = 'in' THEN sm.quantity ELSE -sm.quantity END) AS current_quantity,
    SUM(CASE WHEN sm.direction = 'in' THEN sm.quantity ELSE -sm.quantity END) * MAX(si.unit_cost_tax_included) AS current_cost_estimate
FROM stock_items si
JOIN inventory_accounts ia ON ia.id = si.inventory_account_id
LEFT JOIN stock_movements sm ON sm.stock_item_id = si.id
GROUP BY si.id, si.stock_item_code, ia.owner_party_id, ia.project_id, si.product_id, si.display_name, si.unit;

CREATE OR REPLACE VIEW v_supplier_purchase_totals AS
SELECT
    po.project_id,
    ol.product_id,
    COALESCE(ol.product_name_standard_draft, ol.product_name_raw) AS product_name,
    SUM(ol.quantity) AS supplier_purchase_quantity_total,
    SUM(COALESCE(ol.amount, ol.quantity * COALESCE(ol.unit_price, 0))) AS supplier_purchase_amount_total
FROM purchase_orders po
JOIN order_lines ol ON ol.order_id = po.id
WHERE po.order_side = 'downstream_supplier_purchase'
GROUP BY po.project_id, ol.product_id, COALESCE(ol.product_name_standard_draft, ol.product_name_raw);

CREATE OR REPLACE VIEW v_upstream_demand_totals AS
SELECT
    po.project_id,
    ol.product_id,
    COALESCE(ol.product_name_standard_draft, ol.product_name_raw) AS product_name,
    SUM(ol.quantity) AS upstream_demand_quantity_total,
    SUM(CASE WHEN po.order_type = 'no_document_supplement' THEN ol.quantity ELSE 0 END) AS supplement_demand_quantity_total,
    SUM(CASE WHEN po.order_type <> 'no_document_supplement' THEN ol.quantity ELSE 0 END) AS formal_order_quantity_total
FROM purchase_orders po
JOIN order_lines ol ON ol.order_id = po.id
WHERE po.order_side IN ('upstream_customer_order', 'supplemental_demand')
GROUP BY po.project_id, ol.product_id, COALESCE(ol.product_name_standard_draft, ol.product_name_raw);

CREATE OR REPLACE VIEW v_supplier_vs_upstream_reconciliation_live AS
SELECT
    COALESCE(s.project_id, u.project_id) AS project_id,
    COALESCE(s.product_id, u.product_id) AS product_id,
    COALESCE(s.product_name, u.product_name) AS product_name,
    COALESCE(s.supplier_purchase_quantity_total, 0) AS supplier_purchase_quantity_total,
    COALESCE(u.upstream_demand_quantity_total, 0) AS upstream_demand_quantity_total,
    COALESCE(u.formal_order_quantity_total, 0) AS formal_order_quantity_total,
    COALESCE(u.supplement_demand_quantity_total, 0) AS supplement_demand_quantity_total,
    COALESCE(s.supplier_purchase_quantity_total, 0) - COALESCE(u.upstream_demand_quantity_total, 0) AS difference_supplier_minus_demand,
    GREATEST(COALESCE(s.supplier_purchase_quantity_total, 0) - COALESCE(u.upstream_demand_quantity_total, 0), 0) AS self_stock_candidate_qty,
    GREATEST(COALESCE(u.upstream_demand_quantity_total, 0) - COALESCE(s.supplier_purchase_quantity_total, 0), 0) AS shortage_qty
FROM v_supplier_purchase_totals s
FULL OUTER JOIN v_upstream_demand_totals u
    ON s.project_id = u.project_id
   AND COALESCE(s.product_id::TEXT, s.product_name) = COALESCE(u.product_id::TEXT, u.product_name);
