重构:统一商品渠道投影并收敛权威文档

This commit is contained in:
yml2213
2026-11-08 19:44:27 +08:00
parent 0a595acfa9
commit 1634502363
67 changed files with 2147 additions and 444 deletions
@@ -0,0 +1,426 @@
-- 154_internal_sku_registry_and_routes.sql
--
-- 统一内部 SKU 身份与履约路由:
-- 1. internal_skus 成为内部 SKU 的稳定身份;
-- 2. sku_channel_routes 成为主渠道/兜底渠道的唯一结构化表达;
-- 3. 兼容期由旧 internal_sku_channels 通过触发器同步;
-- 4. product_catalog 增加 source_kind 与 internal_sku_id,不再依赖前缀判断业务来源。
CREATE TABLE internal_skus (
id BIGSERIAL PRIMARY KEY,
internal_sku_key TEXT NOT NULL,
display_name TEXT NOT NULL DEFAULT '',
status TEXT NOT NULL DEFAULT 'active',
merged_into_id BIGINT REFERENCES internal_skus(id) ON DELETE SET NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
CONSTRAINT internal_skus_key_not_blank CHECK (BTRIM(internal_sku_key) <> ''),
CONSTRAINT internal_skus_status_check CHECK (status IN ('active', 'disabled', 'merged'))
);
CREATE UNIQUE INDEX uq_internal_skus_key
ON internal_skus(internal_sku_key);
COMMENT ON TABLE internal_skus IS
'中文内部 SKU 稳定身份;所有路由、配置和商品关联最终通过 internal_sku_id 引用';
COMMENT ON COLUMN internal_skus.internal_sku_key IS
'NFKC、去空白、转小写后的稳定键';
WITH source_rows AS (
SELECT internal_sku AS raw_value, updated_at FROM internal_sku_channels
UNION ALL
SELECT internal_sku AS raw_value, updated_at FROM internal_sku_mappings
UNION ALL
SELECT sku_key AS raw_value, updated_at FROM manual_fulfillment_configs
UNION ALL
SELECT sku_key AS raw_value, updated_at FROM manual_sku_registry
UNION ALL
SELECT canonical_name AS raw_value, updated_at FROM product_catalog
),
normalized AS (
SELECT
canonicalize_internal_sku(raw_value) AS internal_sku_key,
BTRIM(raw_value) AS display_name,
updated_at
FROM source_rows
WHERE canonicalize_internal_sku(raw_value) <> ''
),
picked AS (
SELECT DISTINCT ON (internal_sku_key)
internal_sku_key,
display_name,
updated_at
FROM normalized
ORDER BY internal_sku_key, updated_at DESC NULLS LAST, display_name
)
INSERT INTO internal_skus (
internal_sku_key,
display_name,
status,
created_at,
updated_at
)
SELECT
internal_sku_key,
display_name,
'active',
COALESCE(updated_at, NOW()),
COALESCE(updated_at, NOW())
FROM picked
ON CONFLICT (internal_sku_key) DO UPDATE SET
display_name = CASE
WHEN internal_skus.display_name = '' THEN EXCLUDED.display_name
ELSE internal_skus.display_name
END,
updated_at = GREATEST(internal_skus.updated_at, EXCLUDED.updated_at);
ALTER TABLE internal_sku_mappings
ADD COLUMN internal_sku_id BIGINT REFERENCES internal_skus(id) ON DELETE SET NULL;
ALTER TABLE manual_fulfillment_configs
ADD COLUMN internal_sku_id BIGINT REFERENCES internal_skus(id) ON DELETE SET NULL;
ALTER TABLE product_catalog
ADD COLUMN internal_sku_id BIGINT REFERENCES internal_skus(id) ON DELETE SET NULL,
ADD COLUMN source_kind TEXT;
UPDATE internal_sku_mappings mapping
SET internal_sku_id = sku.id
FROM internal_skus sku
WHERE sku.internal_sku_key = canonicalize_internal_sku(mapping.internal_sku);
UPDATE manual_fulfillment_configs config
SET internal_sku_id = sku.id
FROM internal_skus sku
WHERE sku.internal_sku_key = canonicalize_internal_sku(config.sku_key);
UPDATE product_catalog product
SET source_kind = CASE
WHEN product.canonical_sku LIKE 'aff:%' THEN 'affiliate_dash'
WHEN product.canonical_sku LIKE 'tpl:%' OR product.canonical_sku LIKE 'wo:%' THEN 'work_order'
WHEN product.canonical_sku LIKE 'tpv:%' THEN 'third_party_voucher'
WHEN product.canonical_sku LIKE 'man:%' THEN 'manual'
ELSE 'unclassified'
END;
UPDATE product_catalog product
SET internal_sku_id = sku.id
FROM internal_skus sku
WHERE sku.internal_sku_key = canonicalize_internal_sku(
CASE
WHEN product.canonical_sku LIKE 'man:%' THEN substring(product.canonical_sku from 5)
WHEN product.canonical_sku LIKE 'tpv:%' THEN substring(product.canonical_sku from 5)
ELSE NULL
END
);
ALTER TABLE product_catalog
ALTER COLUMN source_kind SET DEFAULT 'unclassified',
ALTER COLUMN source_kind SET NOT NULL;
CREATE INDEX idx_product_catalog_internal_sku
ON product_catalog(internal_sku_id)
WHERE internal_sku_id IS NOT NULL;
CREATE INDEX idx_product_catalog_source_kind
ON product_catalog(source_kind, status);
CREATE INDEX idx_internal_sku_mappings_internal_sku
ON internal_sku_mappings(internal_sku_id)
WHERE internal_sku_id IS NOT NULL;
CREATE INDEX idx_manual_configs_internal_sku
ON manual_fulfillment_configs(internal_sku_id)
WHERE internal_sku_id IS NOT NULL;
CREATE TABLE sku_channel_routes (
id BIGSERIAL PRIMARY KEY,
internal_sku_id BIGINT NOT NULL REFERENCES internal_skus(id) ON DELETE CASCADE,
channel TEXT NOT NULL,
route_role TEXT NOT NULL DEFAULT 'fallback',
priority INTEGER NOT NULL DEFAULT 100,
enabled BOOLEAN NOT NULL DEFAULT TRUE,
canonical_product_id BIGINT REFERENCES product_catalog(id) ON DELETE SET NULL,
carrier_config_json JSONB NOT NULL DEFAULT '{}'::jsonb,
source TEXT NOT NULL DEFAULT 'legacy_sync',
note TEXT NOT NULL DEFAULT '',
created_by_admin_user_id BIGINT REFERENCES admin_users(id) ON DELETE SET NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
CONSTRAINT sku_channel_routes_channel_check
CHECK (channel IN ('work_order', 'affiliate_dash', 'third_party_voucher', 'manual', 'on_account')),
CONSTRAINT sku_channel_routes_role_check
CHECK (route_role IN ('primary', 'fallback')),
CONSTRAINT sku_channel_routes_priority_check
CHECK (priority >= 0)
);
CREATE UNIQUE INDEX uq_sku_channel_routes_channel
ON sku_channel_routes(internal_sku_id, channel);
CREATE UNIQUE INDEX uq_sku_channel_routes_primary
ON sku_channel_routes(internal_sku_id)
WHERE enabled = TRUE AND route_role = 'primary';
CREATE INDEX idx_sku_channel_routes_lookup
ON sku_channel_routes(channel, enabled, priority)
WHERE enabled = TRUE;
COMMENT ON TABLE sku_channel_routes IS
'内部 SKU 履约路由:一个主渠道加若干显式兜底渠道';
COMMENT ON COLUMN sku_channel_routes.carrier_config_json IS
'渠道载体参数,例如 workRuleKey / affiliateSku / manualConfigId / voucherSku';
WITH legacy_rows AS (
SELECT
sku.id AS internal_sku_id,
channel.channel,
channel.priority,
channel.is_fallback,
channel.enabled,
channel.canonical_product_id,
channel.config_json,
channel.source,
channel.note,
channel.created_by_admin_user_id,
channel.created_at,
channel.updated_at,
ROW_NUMBER() OVER (
PARTITION BY sku.id, channel.channel
ORDER BY channel.is_fallback, channel.priority, channel.id
) AS channel_rank
FROM internal_sku_channels channel
JOIN internal_skus sku
ON sku.internal_sku_key = canonicalize_internal_sku(channel.internal_sku)
WHERE channel.enabled = TRUE
),
deduplicated AS (
SELECT * FROM legacy_rows WHERE channel_rank = 1
),
ranked AS (
SELECT
*,
ROW_NUMBER() OVER (
PARTITION BY internal_sku_id
ORDER BY is_fallback, priority, channel
) AS overall_rank
FROM deduplicated
)
INSERT INTO sku_channel_routes (
internal_sku_id,
channel,
route_role,
priority,
enabled,
canonical_product_id,
carrier_config_json,
source,
note,
created_by_admin_user_id,
created_at,
updated_at
)
SELECT
internal_sku_id,
channel,
CASE WHEN overall_rank = 1 THEN 'primary' ELSE 'fallback' END,
CASE WHEN overall_rank = 1 THEN priority ELSE GREATEST(priority, 90) END,
TRUE,
canonical_product_id,
COALESCE(config_json, '{}'::jsonb),
COALESCE(NULLIF(source, ''), 'legacy_sync'),
COALESCE(note, ''),
created_by_admin_user_id,
COALESCE(created_at, NOW()),
COALESCE(updated_at, NOW())
FROM ranked
ON CONFLICT (internal_sku_id, channel) DO NOTHING;
CREATE OR REPLACE FUNCTION ensure_canonical_internal_sku(p_raw TEXT)
RETURNS BIGINT
LANGUAGE plpgsql
AS $$
DECLARE
v_key TEXT := canonicalize_internal_sku(p_raw);
v_display_name TEXT := BTRIM(COALESCE(p_raw, ''));
v_id BIGINT;
BEGIN
IF v_key = '' THEN
RETURN NULL;
END IF;
INSERT INTO internal_skus (internal_sku_key, display_name, status, created_at, updated_at)
VALUES (v_key, v_display_name, 'active', NOW(), NOW())
ON CONFLICT (internal_sku_key) DO UPDATE SET
display_name = CASE
WHEN internal_skus.display_name = '' THEN EXCLUDED.display_name
ELSE internal_skus.display_name
END,
updated_at = GREATEST(internal_skus.updated_at, EXCLUDED.updated_at)
RETURNING id INTO v_id;
RETURN v_id;
END;
$$;
CREATE OR REPLACE FUNCTION sync_sku_channel_routes_for_key(p_raw TEXT)
RETURNS VOID
LANGUAGE plpgsql
AS $$
DECLARE
v_sku_id BIGINT;
BEGIN
SELECT id INTO v_sku_id
FROM internal_skus
WHERE internal_sku_key = canonicalize_internal_sku(p_raw)
LIMIT 1;
IF v_sku_id IS NULL THEN
RETURN;
END IF;
DELETE FROM sku_channel_routes WHERE internal_sku_id = v_sku_id;
WITH legacy_rows AS (
SELECT
channel.channel,
channel.priority,
channel.is_fallback,
channel.canonical_product_id,
channel.config_json,
channel.source,
channel.note,
channel.created_by_admin_user_id,
channel.created_at,
channel.updated_at,
ROW_NUMBER() OVER (
PARTITION BY channel.channel
ORDER BY channel.is_fallback, channel.priority, channel.id
) AS channel_rank
FROM internal_sku_channels channel
WHERE channel.enabled = TRUE
AND canonicalize_internal_sku(channel.internal_sku)
= canonicalize_internal_sku(p_raw)
),
ranked AS (
SELECT
*,
ROW_NUMBER() OVER (
ORDER BY is_fallback, priority, channel
) AS overall_rank
FROM legacy_rows
WHERE channel_rank = 1
)
INSERT INTO sku_channel_routes (
internal_sku_id,
channel,
route_role,
priority,
enabled,
canonical_product_id,
carrier_config_json,
source,
note,
created_by_admin_user_id,
created_at,
updated_at
)
SELECT
v_sku_id,
channel,
CASE WHEN overall_rank = 1 THEN 'primary' ELSE 'fallback' END,
CASE WHEN overall_rank = 1 THEN priority ELSE GREATEST(priority, 90) END,
TRUE,
canonical_product_id,
COALESCE(config_json, '{}'::jsonb),
COALESCE(NULLIF(source, ''), 'legacy_sync'),
COALESCE(note, ''),
created_by_admin_user_id,
COALESCE(created_at, NOW()),
COALESCE(updated_at, NOW())
FROM ranked;
END;
$$;
CREATE OR REPLACE FUNCTION sync_sku_channel_routes_trigger()
RETURNS TRIGGER
LANGUAGE plpgsql
AS $$
BEGIN
IF TG_OP <> 'INSERT' THEN
PERFORM sync_sku_channel_routes_for_key(OLD.internal_sku);
END IF;
IF TG_OP <> 'DELETE' THEN
PERFORM ensure_canonical_internal_sku(NEW.internal_sku);
PERFORM sync_sku_channel_routes_for_key(NEW.internal_sku);
END IF;
RETURN COALESCE(NEW, OLD);
END;
$$;
CREATE TRIGGER trg_sync_sku_channel_routes
AFTER INSERT OR UPDATE OR DELETE ON internal_sku_channels
FOR EACH ROW
EXECUTE FUNCTION sync_sku_channel_routes_trigger();
CREATE OR REPLACE FUNCTION set_internal_sku_id_from_text()
RETURNS TRIGGER
LANGUAGE plpgsql
AS $$
BEGIN
NEW.internal_sku_id := ensure_canonical_internal_sku(NEW.internal_sku);
RETURN NEW;
END;
$$;
CREATE TRIGGER trg_internal_sku_mapping_identity
BEFORE INSERT OR UPDATE OF internal_sku ON internal_sku_mappings
FOR EACH ROW
EXECUTE FUNCTION set_internal_sku_id_from_text();
CREATE OR REPLACE FUNCTION set_manual_config_internal_sku_id()
RETURNS TRIGGER
LANGUAGE plpgsql
AS $$
BEGIN
NEW.internal_sku_id := ensure_canonical_internal_sku(NEW.sku_key);
RETURN NEW;
END;
$$;
CREATE TRIGGER trg_manual_config_internal_sku_identity
BEFORE INSERT OR UPDATE OF sku_key ON manual_fulfillment_configs
FOR EACH ROW
EXECUTE FUNCTION set_manual_config_internal_sku_id();
CREATE OR REPLACE FUNCTION normalize_product_identity()
RETURNS TRIGGER
LANGUAGE plpgsql
AS $$
BEGIN
NEW.source_kind := COALESCE(
NULLIF(NEW.source_kind, ''),
CASE
WHEN NEW.canonical_sku LIKE 'aff:%' THEN 'affiliate_dash'
WHEN NEW.canonical_sku LIKE 'tpl:%' OR NEW.canonical_sku LIKE 'wo:%' THEN 'work_order'
WHEN NEW.canonical_sku LIKE 'tpv:%' THEN 'third_party_voucher'
WHEN NEW.canonical_sku LIKE 'man:%' THEN 'manual'
ELSE 'unclassified'
END
);
IF NEW.internal_sku_id IS NULL THEN
SELECT sku.id INTO NEW.internal_sku_id
FROM internal_skus sku
WHERE sku.internal_sku_key = canonicalize_internal_sku(
CASE
WHEN NEW.canonical_sku LIKE 'man:%' THEN substring(NEW.canonical_sku from 5)
WHEN NEW.canonical_sku LIKE 'tpv:%' THEN substring(NEW.canonical_sku from 5)
ELSE NULL
END
)
LIMIT 1;
END IF;
RETURN NEW;
END;
$$;
CREATE TRIGGER trg_normalize_product_identity
BEFORE INSERT OR UPDATE OF canonical_sku, canonical_name, source_kind, internal_sku_id
ON product_catalog
FOR EACH ROW
EXECUTE FUNCTION normalize_product_identity();
@@ -0,0 +1,96 @@
-- 155_product_current_projection.sql
--
-- 统一商品当前渠道投影,并清理已被其他渠道接管且没有人工兜底配置的旧 man 商品。
UPDATE sku_channel_routes route
SET canonical_product_id = product.id,
updated_at = NOW()
FROM product_catalog product
WHERE route.channel = 'work_order'
AND route.canonical_product_id IS NULL
AND product.status = 'active'
AND product.canonical_sku = 'tpl:' || COALESCE(route.carrier_config_json->>'workRuleKey', '');
CREATE OR REPLACE VIEW v_product_current AS
SELECT
product.*,
sku.internal_sku_key,
COALESCE(
primary_route.channel,
CASE product.source_kind
WHEN 'affiliate_dash' THEN 'affiliate_dash'
WHEN 'work_order' THEN 'work_order'
WHEN 'third_party_voucher' THEN 'third_party_voucher'
WHEN 'manual' THEN 'manual'
ELSE NULL
END
) AS active_primary_channel,
COALESCE(fallback_routes.channels, ARRAY[]::text[]) AS fallback_channels,
manual_config.id AS manual_config_id,
(manual_config.id IS NOT NULL) AS has_active_manual_config
FROM product_catalog product
LEFT JOIN internal_skus sku ON sku.id = product.internal_sku_id
LEFT JOIN LATERAL (
SELECT route.channel
FROM sku_channel_routes route
WHERE route.internal_sku_id = product.internal_sku_id
AND route.enabled = TRUE
AND route.route_role = 'primary'
ORDER BY route.priority, route.channel
LIMIT 1
) primary_route ON TRUE
LEFT JOIN LATERAL (
SELECT ARRAY_AGG(route.channel ORDER BY route.priority, route.channel) AS channels
FROM sku_channel_routes route
WHERE route.internal_sku_id = product.internal_sku_id
AND route.enabled = TRUE
AND route.route_role = 'fallback'
) fallback_routes ON TRUE
LEFT JOIN LATERAL (
SELECT config.id
FROM manual_fulfillment_configs config
WHERE config.enabled = TRUE
AND (
config.canonical_product_id = product.id
OR (
sku.internal_sku_key IS NOT NULL
AND canonicalize_internal_sku(config.sku_key) = sku.internal_sku_key
)
)
ORDER BY config.id
LIMIT 1
) manual_config ON TRUE;
COMMENT ON VIEW v_product_current IS
'标准商品统一投影:来源、当前主渠道、兜底渠道和人工配置状态同源输出';
WITH candidates AS (
SELECT product.id
FROM v_product_current product
WHERE product.status = 'active'
AND product.canonical_sku LIKE 'man:%'
AND product.active_primary_channel IS NOT NULL
AND product.active_primary_channel <> 'manual'
AND product.has_active_manual_config = FALSE
AND EXISTS (
SELECT 1
FROM sku_channel_routes route
WHERE route.internal_sku_id = product.internal_sku_id
AND route.enabled = TRUE
AND route.route_role = 'primary'
AND route.channel <> 'manual'
)
)
UPDATE product_catalog product
SET status = 'archived',
attributes_json = COALESCE(product.attributes_json, '{}'::jsonb) || jsonb_build_object(
'legacyManualArchived', TRUE,
'legacyManualArchiveReason', 'superseded_by_current_primary_channel',
'legacyManualArchivedAt', NOW()
),
updated_at = NOW()
FROM candidates
WHERE product.id = candidates.id;
COMMENT ON COLUMN product_catalog.source_kind IS
'标准商品来源类型;来源不等于当前履约渠道,当前渠道统一读取 v_product_current';
@@ -0,0 +1,53 @@
-- 156_fix_product_source_kind_default.sql
--
-- source_kind 的列默认值会让 BEFORE INSERT 触发器看到“非空但未分类”,
-- 从而跳过前缀回填。这里保留默认值兼容直接写入,同时让已知前缀始终覆盖
-- unclassified,并修复兼容期内已经写入的误判行。
CREATE OR REPLACE FUNCTION normalize_product_identity()
RETURNS TRIGGER
LANGUAGE plpgsql
AS $$
BEGIN
NEW.source_kind := CASE
WHEN NEW.canonical_sku LIKE 'aff:%' THEN 'affiliate_dash'
WHEN NEW.canonical_sku LIKE 'tpl:%' OR NEW.canonical_sku LIKE 'wo:%' THEN 'work_order'
WHEN NEW.canonical_sku LIKE 'tpv:%' THEN 'third_party_voucher'
WHEN NEW.canonical_sku LIKE 'man:%' THEN 'manual'
ELSE COALESCE(NULLIF(NEW.source_kind, ''), 'unclassified')
END;
IF NEW.internal_sku_id IS NULL THEN
SELECT sku.id INTO NEW.internal_sku_id
FROM internal_skus sku
WHERE sku.internal_sku_key = canonicalize_internal_sku(
CASE
WHEN NEW.canonical_sku LIKE 'man:%' THEN substring(NEW.canonical_sku from 5)
WHEN NEW.canonical_sku LIKE 'tpv:%' THEN substring(NEW.canonical_sku from 5)
ELSE NULL
END
)
LIMIT 1;
END IF;
RETURN NEW;
END;
$$;
UPDATE product_catalog
SET source_kind = CASE
WHEN canonical_sku LIKE 'aff:%' THEN 'affiliate_dash'
WHEN canonical_sku LIKE 'tpl:%' OR canonical_sku LIKE 'wo:%' THEN 'work_order'
WHEN canonical_sku LIKE 'tpv:%' THEN 'third_party_voucher'
WHEN canonical_sku LIKE 'man:%' THEN 'manual'
ELSE source_kind
END,
updated_at = NOW()
WHERE source_kind = 'unclassified'
AND (
canonical_sku LIKE 'aff:%'
OR canonical_sku LIKE 'tpl:%'
OR canonical_sku LIKE 'wo:%'
OR canonical_sku LIKE 'tpv:%'
OR canonical_sku LIKE 'man:%'
);
@@ -0,0 +1,105 @@
-- 157_preserve_fallback_only_routes.sql
--
-- 兼容同步不能把“只有兜底渠道”的旧配置自动升级成主渠道。
-- 只有旧表中明确 is_fallback = FALSE 的渠道才允许竞争 primary;
-- 如果某个内部 SKU 没有主渠道,则所有启用路由都保持 fallback。
CREATE OR REPLACE FUNCTION sync_sku_channel_routes_for_key(p_raw TEXT)
RETURNS VOID
LANGUAGE plpgsql
AS $$
DECLARE
v_sku_id BIGINT;
BEGIN
SELECT id INTO v_sku_id
FROM internal_skus
WHERE internal_sku_key = canonicalize_internal_sku(p_raw)
LIMIT 1;
IF v_sku_id IS NULL THEN
RETURN;
END IF;
DELETE FROM sku_channel_routes WHERE internal_sku_id = v_sku_id;
WITH legacy_rows AS (
SELECT
channel.channel,
channel.priority,
channel.is_fallback,
channel.canonical_product_id,
channel.config_json,
channel.source,
channel.note,
channel.created_by_admin_user_id,
channel.created_at,
channel.updated_at,
ROW_NUMBER() OVER (
PARTITION BY channel.channel
ORDER BY channel.priority, channel.id
) AS channel_rank
FROM internal_sku_channels channel
WHERE channel.enabled = TRUE
AND canonicalize_internal_sku(channel.internal_sku)
= canonicalize_internal_sku(p_raw)
),
ranked AS (
SELECT
*,
ROW_NUMBER() OVER (
PARTITION BY is_fallback
ORDER BY priority, channel
) AS role_rank
FROM legacy_rows
WHERE channel_rank = 1
)
INSERT INTO sku_channel_routes (
internal_sku_id,
channel,
route_role,
priority,
enabled,
canonical_product_id,
carrier_config_json,
source,
note,
created_by_admin_user_id,
created_at,
updated_at
)
SELECT
v_sku_id,
channel,
CASE
WHEN is_fallback = FALSE AND role_rank = 1 THEN 'primary'
ELSE 'fallback'
END,
CASE
WHEN is_fallback = FALSE AND role_rank = 1 THEN priority
ELSE GREATEST(priority, 90)
END,
TRUE,
canonical_product_id,
COALESCE(config_json, '{}'::jsonb),
COALESCE(NULLIF(source, ''), 'legacy_sync'),
COALESCE(note, ''),
created_by_admin_user_id,
COALESCE(created_at, NOW()),
COALESCE(updated_at, NOW())
FROM ranked;
END;
$$;
DO $$
DECLARE
v_internal_sku TEXT;
BEGIN
FOR v_internal_sku IN
SELECT DISTINCT internal_sku
FROM internal_sku_channels
WHERE enabled = TRUE
LOOP
PERFORM sync_sku_channel_routes_for_key(v_internal_sku);
END LOOP;
END;
$$;
@@ -0,0 +1,57 @@
-- 158_manual_source_is_not_manual_route.sql
--
-- man: 只表示商品来源,不能证明当前由人工履约。
-- 兼容期内,自动发货 / 接单平台 / 第三方卡券这类直连来源允许暂时由 source_kind
-- 推导当前渠道;人工客服必须有真实启用的 primary route,否则就是“未路由”。
CREATE OR REPLACE VIEW v_product_current AS
SELECT
product.*,
sku.internal_sku_key,
COALESCE(
primary_route.channel,
CASE product.source_kind
WHEN 'affiliate_dash' THEN 'affiliate_dash'
WHEN 'work_order' THEN 'work_order'
WHEN 'third_party_voucher' THEN 'third_party_voucher'
ELSE NULL
END
) AS active_primary_channel,
COALESCE(fallback_routes.channels, ARRAY[]::text[]) AS fallback_channels,
manual_config.id AS manual_config_id,
(manual_config.id IS NOT NULL) AS has_active_manual_config
FROM product_catalog product
LEFT JOIN internal_skus sku ON sku.id = product.internal_sku_id
LEFT JOIN LATERAL (
SELECT route.channel
FROM sku_channel_routes route
WHERE route.internal_sku_id = product.internal_sku_id
AND route.enabled = TRUE
AND route.route_role = 'primary'
ORDER BY route.priority, route.channel
LIMIT 1
) primary_route ON TRUE
LEFT JOIN LATERAL (
SELECT ARRAY_AGG(route.channel ORDER BY route.priority, route.channel) AS channels
FROM sku_channel_routes route
WHERE route.internal_sku_id = product.internal_sku_id
AND route.enabled = TRUE
AND route.route_role = 'fallback'
) fallback_routes ON TRUE
LEFT JOIN LATERAL (
SELECT config.id
FROM manual_fulfillment_configs config
WHERE config.enabled = TRUE
AND (
config.canonical_product_id = product.id
OR (
sku.internal_sku_key IS NOT NULL
AND canonicalize_internal_sku(config.sku_key) = sku.internal_sku_key
)
)
ORDER BY config.id
LIMIT 1
) manual_config ON TRUE;
COMMENT ON VIEW v_product_current IS
'标准商品统一投影:来源不等于当前渠道;manual 必须有真实主路由,直连来源兼容期允许回退 source_kind';
@@ -21,6 +21,18 @@ import type {
// 履约方式 / 商品来源 / 成本模型
// ---------------------------------------------------------------------------
/** 标准商品目录允许使用的当前渠道筛选,包含“未路由”这个投影状态。 */
export const PRODUCT_CURRENT_CHANNEL_FILTERS = [
'affiliate_dash',
'work_order',
'third_party_voucher',
'manual',
'on_account',
'unrouted',
] as const
export type ProductCurrentChannelFilter = (typeof PRODUCT_CURRENT_CHANNEL_FILTERS)[number]
/**
* 由 canonical_sku 前缀推导商品来源。
* 返回空串表示未分类;未分类商品不允许反推履约方式。
@@ -43,6 +55,20 @@ export function productSourceFilterSql(paramIndex: number, column = 'p.canonical
)`
}
/** 当前渠道筛选:标准商品页签的唯一筛选口径。 */
export function productChannelFilterSql(
paramIndex: number,
column = 'p.active_primary_channel',
): string {
return `(
($${paramIndex} = 'unrouted' AND ${column} IS NULL)
OR (
$${paramIndex} <> 'unrouted'
AND ${column} = $${paramIndex}
)
)`
}
/**
* 成本模型是否适用于该履约方式。
*
@@ -95,7 +121,8 @@ export function mapProduct(row: ProductRow) {
canonicalSku: row.canonical_sku,
canonicalName: row.canonical_name,
category: row.category || '',
source: productSourceOf(row.canonical_sku),
source: String(row.source_kind || productSourceOf(row.canonical_sku)),
sourceKind: String(row.source_kind || productSourceOf(row.canonical_sku)),
rootCategoryId: row.root_category_id ? Number(row.root_category_id) : null,
rootCategoryName: row.root_category_name || row.category || '',
subCategoryId: row.sub_category_id ? Number(row.sub_category_id) : null,
@@ -122,6 +149,7 @@ export function mapProduct(row: ProductRow) {
? null
: row.template_sharing_enabled === true,
activePrimaryChannel: row.active_primary_channel || null,
fallbackChannels: Array.isArray(row.fallback_channels) ? row.fallback_channels : [],
affiliateTaskCount: Number(row.affiliate_task_count || 0),
affiliateLastFulfilledAt: row.affiliate_last_fulfilled_at || null,
createdAt: row.created_at,
@@ -2,20 +2,26 @@
* 成本中心总览与选项。
*
* 口径(2026-10-23 重写):
* 1. 商品来源统计按 canonical_sku 前缀分组,不再读 product_catalog.carrier_type;
* 1. 商品统计按 v_product_current.active_primary_channel 分组;
* 2. 成本项统计按 cost_model 分组,不再读 cost_items.carrier_type。
*/
import { query } from '../../db/client.js'
import { productSourceFilterSql } from './cost-center-mappers.js'
import { PROVIDERS, productSourceOf, toNumber } from './cost-center-shared.js'
import { productChannelFilterSql } from './cost-center-mappers.js'
import { PROVIDERS, toNumber } from './cost-center-shared.js'
export type CostCenterScope = 'affiliate_dash' | 'work_order' | 'manual'
export type CostCenterScope =
| 'affiliate_dash'
| 'work_order'
| 'third_party_voucher'
| 'manual'
| 'on_account'
| 'unrouted'
/**
* 成本中心总览。
*
* scope 为空表示「全部来源」;传入具体来源时,商品/映射/成本项/版本/异常/未映射
* 六项统计都按该来源收敛,页面顶部数字才会和下面的列表对得上。
* scope 为空表示「全部渠道」;传入具体渠道时,商品/映射/成本项/版本/异常/未映射
* 六项统计都按该渠道收敛,页面顶部数字才会和下面的列表对得上。
*
* 来源与成本模型的对应关系:
* affiliate_dash → affiliate_purchase
@@ -36,9 +42,9 @@ export async function getAdminCostCenterOverview(params: { scope?: string } = {}
`
WITH scoped_products AS (
SELECT p.id
FROM product_catalog p
FROM v_product_current p
WHERE p.status = 'active'
AND ($1::text IS NULL OR ${productSourceFilterSql(1)})
AND ($1::text IS NULL OR ${productChannelFilterSql(1)})
),
scoped_cost_models AS (
SELECT CASE $1
@@ -46,6 +52,7 @@ export async function getAdminCostCenterOverview(params: { scope?: string } = {}
WHEN 'work_order' THEN 'work_order_reward'
WHEN 'third_party_voucher' THEN 'third_party_voucher_cost'
WHEN 'manual' THEN 'manual_actual'
WHEN 'on_account' THEN 'manual_actual'
END AS cost_model
)
SELECT
@@ -93,22 +100,12 @@ export async function getAdminCostCenterOverview(params: { scope?: string } = {}
)
const row = result.rows[0]
const [productSourceResult, costModelResult] = await Promise.all([
query<{ canonical_sku: string; active_primary_channel: string | null; total: number }>(
`SELECT p.canonical_sku,
(
SELECT c.channel
FROM internal_sku_channels c
WHERE c.enabled = TRUE
AND c.is_fallback = FALSE
AND canonicalize_internal_sku(c.internal_sku)
= canonicalize_internal_sku(p.canonical_name)
ORDER BY c.priority, c.channel
LIMIT 1
) AS active_primary_channel,
query<{ channel: string | null; total: number }>(
`SELECT p.active_primary_channel AS channel,
COUNT(*)::int AS total
FROM product_catalog p
FROM v_product_current p
WHERE p.status = 'active'
GROUP BY p.canonical_sku, p.canonical_name`,
GROUP BY p.active_primary_channel`,
),
query<{ cost_model: string; total: number }>(
`SELECT cost_model, COUNT(*)::int AS total FROM cost_items WHERE status = 'active' GROUP BY cost_model`,
@@ -134,6 +131,8 @@ const COST_CENTER_SCOPES = new Set([
'work_order',
'third_party_voucher',
'manual',
'on_account',
'unrouted',
])
function normalizeScope(raw: unknown): CostCenterScope | null {
@@ -141,27 +140,18 @@ function normalizeScope(raw: unknown): CostCenterScope | null {
return COST_CENTER_SCOPES.has(value) ? (value as CostCenterScope) : null
}
/** 商品来源计数:按 canonical_sku 前缀归类,未分类单独成桶。 */
function productSourceCountsOf(
rows: Array<{ canonical_sku: string; active_primary_channel: string | null; total: number }>,
) {
/** 商品当前渠道计数:总览和列表共用 v_product_current。 */
function productSourceCountsOf(rows: Array<{ channel: string | null; total: number }>) {
const counts: Record<string, number> = {
affiliate_dash: 0,
work_order: 0,
third_party_voucher: 0,
manual: 0,
uncategorized: 0,
on_account: 0,
unrouted: 0,
}
for (const row of rows) {
const source = productSourceOf(row.canonical_sku)
// man: 只是旧来源;已有其他主渠道的商品不再计入当前人工客服。
if (
source === 'manual' &&
row.active_primary_channel &&
row.active_primary_channel !== 'manual'
) {
continue
}
const key = source || 'uncategorized'
const key = row.channel || 'unrouted'
counts[key] = (counts[key] || 0) + toNumber(row.total)
}
return counts
@@ -183,10 +173,12 @@ export async function getAdminCostCenterOptions() {
canonical_sku: string
canonical_name: string
unit_name: string
source_kind: string
active_primary_channel: string | null
}>(
`
SELECT id, canonical_sku, canonical_name, unit_name
FROM product_catalog
SELECT id, canonical_sku, canonical_name, unit_name, source_kind, active_primary_channel
FROM v_product_current
WHERE status = 'active'
ORDER BY canonical_name, id
LIMIT 5000
@@ -225,7 +217,8 @@ export async function getAdminCostCenterOptions() {
canonicalSku: row.canonical_sku,
canonicalName: row.canonical_name,
unitName: row.unit_name,
source: productSourceOf(row.canonical_sku),
source: row.source_kind,
activePrimaryChannel: row.active_primary_channel,
})),
costItems: costItemsResult.rows.map((row) => ({
id: Number(row.id),
@@ -29,7 +29,10 @@ export type ProductRow = {
template_pricing_mode?: string | null
template_enabled?: boolean | null
template_sharing_enabled?: boolean | null
source_kind?: string | null
active_primary_channel?: string | null
fallback_channels?: string[] | null
manual_config_id?: number | null
affiliate_task_count?: number
affiliate_last_fulfilled_at?: string | null
}
@@ -1,4 +1,5 @@
import assert from 'node:assert/strict'
import { readFile } from 'node:fs/promises'
import test from 'node:test'
import { costModelAppliesTo } from './cost-center-mappers.js'
@@ -9,7 +10,7 @@ import {
} from './cost-center-shared.js'
import { calculateCostTotal, validateCostVersionScope } from './cost-resolution-service.js'
import { createAdminProduct, resolveMappingProductId } from './product-catalog-service.js'
import { productSourceFilterSql } from './cost-center-mappers.js'
import { productChannelFilterSql, productSourceFilterSql } from './cost-center-mappers.js'
import {
buildTemplateMappingPlan,
buildTemplateProductSku,
@@ -289,6 +290,10 @@ test('成本模型与商品来源必须覆盖第三方卡券渠道', () => {
assert.equal(costModelAppliesTo('third_party_voucher_cost', 'manual'), false)
assert.ok(productSourceFilterSql(1).includes("'third_party_voucher'"))
assert.ok(productSourceFilterSql(1).includes("'tpv:%'"))
const channelSql = productChannelFilterSql(1)
assert.ok(channelSql.includes('active_primary_channel'))
assert.ok(channelSql.includes("$1 = 'unrouted'"))
assert.ok(!channelSql.includes('canonical_sku'))
})
test('第三方卡券删除:无历史引用硬删除,有历史引用自动归档并停用规则', async () => {
@@ -310,11 +315,15 @@ test('第三方卡券删除:无历史引用硬删除,有历史引用自动
assert.ok(routeSource.includes("'/cost-center/voucher-products/:productId'"))
})
test('人工客服筛选保留所有可维护的 man 商品和显式兜底配置', () => {
const sql = productSourceFilterSql(1)
assert.ok(sql.includes(`$1 = 'manual'`))
assert.ok(sql.includes("LIKE 'man:%'"))
assert.ok(!sql.includes('live_channel'))
test('人工客服页签只按当前主渠道筛选,不能再按 man 前缀放回人工页', async () => {
const source = await readFile(new URL('./product-catalog-service.ts', import.meta.url), 'utf8')
const sql = productChannelFilterSql(1)
assert.ok(sql.includes('active_primary_channel'))
assert.ok(sql.includes("$1 <> 'unrouted'"))
assert.ok(sql.includes('active_primary_channel = $1'))
assert.ok(!sql.includes('canonical_sku'))
assert.ok(source.includes('FROM v_product_current p'))
assert.ok(source.includes('productChannelFilterSql'))
})
test('成本中心总览异常统计必须走异常表关联,不能全表扫描订单项子查询', async () => {
@@ -325,7 +334,21 @@ test('成本中心总览异常统计必须走异常表关联,不能全表扫
)
assert.ok(source.includes('JOIN order_items oi ON oi.id = ce.order_item_id'))
assert.ok(!source.includes('ce.order_item_id IN ('))
assert.ok(source.includes('productSourceFilterSql(1)'))
assert.ok(source.includes('productChannelFilterSql(1)'))
})
test('运行期停止写入已废弃的 product_channel_capabilities', async () => {
const files = await Promise.all(
[
'./voucher-product-service.ts',
'./product-normalize-service.ts',
'./product-catalog-service.ts',
'../fulfillment/channel-assign-service.ts',
].map((path) => readFile(new URL(path, import.meta.url), 'utf8')),
)
for (const source of files) {
assert.ok(!source.includes('product_channel_capabilities'))
}
})
test('tpv: 商品来源归为第三方卡券', async () => {
@@ -27,7 +27,6 @@ import { orderItemBasisDates, refreshFactsForDates } from '../finance/refresh-fa
import { syncAdminWorkProductRulePrice } from '../worker-platform/admin-worker-catalog-service.js'
import {
AFFILIATE_TASK_AGG_JOIN,
FULFILLMENT_CHANNELS,
MAPPING_CONFIDENCES,
MAPPING_SOURCES,
PRODUCT_CATEGORY_JOINS,
@@ -52,7 +51,8 @@ import {
normalizeRequiredText,
normalizeStatus,
parseJsonObject,
productSourceFilterSql,
productChannelFilterSql,
PRODUCT_CURRENT_CHANNEL_FILTERS,
requiredRow,
resolveMappingNameKey,
text,
@@ -64,13 +64,8 @@ import {
type WorkCategoryRow,
} from './cost-center-shared.js'
/** 商品来源的筛选取值:与 canonical_sku 前缀一一对应。 */
const PRODUCT_SOURCE_FILTERS = new Set([
'affiliate_dash',
'work_order',
'third_party_voucher',
'manual',
])
/** 标准商品页签的当前渠道筛选取值。 */
const PRODUCT_CHANNEL_FILTERS = new Set<string>(PRODUCT_CURRENT_CHANNEL_FILTERS)
/**
* 校验新建商品的 SKU 前缀。
@@ -108,17 +103,19 @@ export async function listAdminProducts(payload: JsonObject = {}) {
assertOneOf(status, PRODUCT_STATUSES, '商品状态无效')
baseParams.push(status)
baseClauses.push(`p.status = $${baseParams.length}`)
// carrierType 参数保留,语义改为「按 canonical_sku 前缀筛选商品来源」。
const source = text(payload.carrierType ?? payload.carrier_type ?? payload.source)
if (source) {
assertOneOf(source, PRODUCT_SOURCE_FILTERS, '商品来源无效')
baseParams.push(source)
baseClauses.push(productSourceFilterSql(baseParams.length))
// carrierType 参数仅为兼容旧前端,语义统一为「按当前主渠道筛选」。
const channel = text(
payload.channel ?? payload.currentChannel ?? payload.carrierType ?? payload.carrier_type,
)
if (channel) {
assertOneOf(channel, PRODUCT_CHANNEL_FILTERS, '商品渠道无效')
baseParams.push(channel)
baseClauses.push(productChannelFilterSql(baseParams.length))
}
// 人工/接单平台商品不需要 affiliate 任务统计:跳过 30 天 fulfillment_tasks 聚合。
// 只有「全部」和「自动发货」Tab 才需要这条聚合,避免小库 IO 被无关数据拖慢。
const needsAffiliateTaskStats = !source || source === 'affiliate_dash'
const needsAffiliateTaskStats = !channel || channel === 'affiliate_dash'
const affiliateTaskJoinSql = needsAffiliateTaskStats ? AFFILIATE_TASK_AGG_JOIN : ''
const affiliateColumnsSql = needsAffiliateTaskStats
? 'COALESCE(aff.task_count, 0) AS affiliate_task_count, aff.last_fulfilled_at AS affiliate_last_fulfilled_at'
@@ -146,7 +143,7 @@ export async function listAdminProducts(payload: JsonObject = {}) {
const [countResult, result, categoryStatResult, categoryResult] = await Promise.all([
query<{ total: number }>(
`SELECT COUNT(*)::int AS total FROM product_catalog p ${PRODUCT_CATEGORY_JOINS} ${where}`,
`SELECT COUNT(*)::int AS total FROM v_product_current p ${PRODUCT_CATEGORY_JOINS} ${where}`,
listParams,
),
query<ProductRow>(
@@ -156,16 +153,6 @@ export async function listAdminProducts(payload: JsonObject = {}) {
tpl_rule.pricing_mode AS template_pricing_mode,
tpl_rule.enabled AS template_enabled,
tpl_rule.sharing_enabled AS template_sharing_enabled,
(
SELECT c.channel
FROM internal_sku_channels c
WHERE c.enabled = TRUE
AND c.is_fallback = FALSE
AND canonicalize_internal_sku(c.internal_sku)
= canonicalize_internal_sku(p.canonical_name)
ORDER BY c.priority, c.channel
LIMIT 1
) AS active_primary_channel,
CASE WHEN c.parent_id IS NOT NULL THEN c.id END AS sub_category_id,
CASE WHEN c.parent_id IS NOT NULL THEN c.name ELSE '' END AS sub_category_name,
root.id AS root_category_id,
@@ -228,7 +215,7 @@ export async function listAdminProducts(payload: JsonObject = {}) {
)
) AS active_cost_effective_from,
${affiliateColumnsSql}
FROM product_catalog p
FROM v_product_current p
${PRODUCT_CATEGORY_JOINS}
LEFT JOIN work_product_rules tpl_rule
ON p.canonical_sku LIKE 'tpl:%'
@@ -252,7 +239,7 @@ export async function listAdminProducts(payload: JsonObject = {}) {
query<ProductCategoryStatRow>(
`
SELECT c.id AS category_id, c.name, c.parent_id, COUNT(*)::int AS total
FROM product_catalog p
FROM v_product_current p
${PRODUCT_CATEGORY_JOINS}
${baseWhere}
GROUP BY c.id, c.name, c.parent_id
@@ -512,16 +499,6 @@ export async function deleteAdminVoucherProduct(productId: number | string) {
`,
[id, internalSku, nowIso()],
)
await client.query(
`
UPDATE product_channel_capabilities
SET enabled = FALSE,
updated_at = $2
WHERE product_id = $1 AND channel = 'third_party_voucher'
`,
[id, nowIso()],
)
if (mappingCount > 0 || snapshotCount > 0 || orderItemCount > 0) {
await client.query(
`
@@ -566,7 +543,6 @@ export async function deleteAdminVoucherProduct(productId: number | string) {
`DELETE FROM cost_versions WHERE canonical_product_id = $1`,
[id],
)
await client.query(`DELETE FROM product_channel_capabilities WHERE product_id = $1`, [id])
await client.query(
`
DELETE FROM internal_sku_channels
@@ -627,12 +603,14 @@ export async function listAdminChannelMappings(payload: JsonObject = {}) {
params.push(canonicalProductId)
clauses.push(`m.canonical_product_id = $${params.length}`)
}
// carrierType 参数保留,语义改为「按 canonical_sku 前缀筛选商品来源」。
const source = text(payload.carrierType ?? payload.carrier_type ?? payload.source)
if (source) {
assertOneOf(source, PRODUCT_SOURCE_FILTERS, '商品来源无效')
params.push(source)
clauses.push(productSourceFilterSql(params.length))
// carrierType 仅为旧客户端兼容,筛选口径统一是商品当前主渠道。
const channel = text(
payload.channel ?? payload.currentChannel ?? payload.carrierType ?? payload.carrier_type,
)
if (channel) {
assertOneOf(channel, PRODUCT_CHANNEL_FILTERS, '商品渠道无效')
params.push(channel)
clauses.push(productChannelFilterSql(params.length))
}
if (payload.enabled !== undefined && payload.enabled !== '') {
params.push(normalizeBoolean(payload.enabled))
@@ -652,7 +630,7 @@ export async function listAdminChannelMappings(payload: JsonObject = {}) {
p.unit_name,
p.status
FROM channel_product_mappings m
JOIN product_catalog p ON p.id = m.canonical_product_id
JOIN v_product_current p ON p.id = m.canonical_product_id
${where}
GROUP BY
p.id,
@@ -672,7 +650,7 @@ export async function listAdminChannelMappings(payload: JsonObject = {}) {
COUNT(DISTINCT p.id)::int AS total_count,
COUNT(*)::int AS mapping_total
FROM channel_product_mappings m
JOIN product_catalog p ON p.id = m.canonical_product_id
JOIN v_product_current p ON p.id = m.canonical_product_id
${where}
`,
params,
@@ -698,7 +676,7 @@ export async function listAdminChannelMappings(payload: JsonObject = {}) {
p.canonical_sku,
p.canonical_name
FROM channel_product_mappings m
JOIN product_catalog p ON p.id = m.canonical_product_id
JOIN v_product_current p ON p.id = m.canonical_product_id
${where}
${where ? 'AND' : 'WHERE'} m.canonical_product_id = ANY($${mappingParams.length}::bigint[])
ORDER BY
@@ -748,16 +726,16 @@ export async function listAdminChannelMappings(payload: JsonObject = {}) {
* 「全部」时并行跑四种模式再求和,避免一次全表 hash join。
*/
export async function getAdminChannelMappingDiagnostics(payload: JsonObject = {}) {
const source = text(payload.carrierType ?? payload.carrier_type ?? payload.source)
if (source) assertOneOf(source, PRODUCT_SOURCE_FILTERS, '商品来源无效')
const channel = text(payload.channel ?? payload.channel)
if (channel) assertOneOf(channel, FULFILLMENT_CHANNELS, '履约方式无效')
const channel = text(
payload.channel ?? payload.currentChannel ?? payload.carrierType ?? payload.carrier_type,
)
if (channel) assertOneOf(channel, PRODUCT_CHANNEL_FILTERS, '商品渠道无效')
const productParams: unknown[] = []
let sourceSql = 'TRUE'
if (source) {
productParams.push(source)
sourceSql = productSourceFilterSql(productParams.length)
let productChannelSql = 'TRUE'
if (channel) {
productParams.push(channel)
productChannelSql = productChannelFilterSql(productParams.length)
}
const counts = await query<{
product_total: number
@@ -767,20 +745,20 @@ export async function getAdminChannelMappingDiagnostics(payload: JsonObject = {}
`
SELECT
(
SELECT COUNT(*)::int FROM product_catalog p
WHERE p.status = 'active' AND ${sourceSql}
SELECT COUNT(*)::int FROM v_product_current p
WHERE p.status = 'active' AND ${productChannelSql}
) AS product_total,
(
SELECT COUNT(*)::int
FROM channel_product_mappings m
JOIN product_catalog p ON p.id = m.canonical_product_id
WHERE m.enabled = TRUE AND ${sourceSql}
JOIN v_product_current p ON p.id = m.canonical_product_id
WHERE m.enabled = TRUE AND ${productChannelSql}
) AS mapping_total,
(
SELECT COUNT(DISTINCT m.canonical_product_id)::int
FROM channel_product_mappings m
JOIN product_catalog p ON p.id = m.canonical_product_id
WHERE m.enabled = TRUE AND ${sourceSql}
JOIN v_product_current p ON p.id = m.canonical_product_id
WHERE m.enabled = TRUE AND ${productChannelSql}
) AS mapping_product_total
`,
productParams,
@@ -816,9 +794,12 @@ export async function getAdminChannelMappingDiagnostics(payload: JsonObject = {}
}
}
const modes = channel
? [channel]
: ['work_order', 'affiliate_dash', 'third_party_voucher', 'manual', 'on_account']
const modes =
channel === 'unrouted'
? []
: channel
? [channel]
: ['work_order', 'affiliate_dash', 'third_party_voucher', 'manual', 'on_account']
// 顺序跑:并行会同时扫 cost_exceptions,实测 4 路并发比串行更慢(I/O 争抢)。
const aggregates: Array<{ groupTotal: number; itemTotal: number }> = []
for (const mode of modes) {
@@ -849,8 +830,14 @@ export async function listAdminUnmappedMappingGroups(payload: JsonObject = {}) {
const params: unknown[] = []
const clauses = [`ce.reason = 'product_mapping_missing'`, `ce.status = 'open'`]
const channel = text(payload.channel ?? payload.channel)
if (channel) assertOneOf(channel, FULFILLMENT_CHANNELS, '履约方式无效')
const channel = text(payload.channel ?? payload.currentChannel)
if (channel) assertOneOf(channel, PRODUCT_CHANNEL_FILTERS, '商品渠道无效')
if (channel === 'unrouted') {
return {
items: [],
pagination: { page, pageSize, total: 0, itemTotal: 0 },
}
}
const provider = text(payload.provider)
if (provider) {
params.push(normalizeProviderKey(provider))
@@ -210,15 +210,6 @@ async function ensureVoucherProductForItem(item: ItemRow): Promise<number | null
onlyWhenCanonicalProductDifferent: true,
})
await query(
`
INSERT INTO product_channel_capabilities (product_id, channel, priority, is_fallback, enabled, source)
VALUES ($1, 'third_party_voucher', 10, FALSE, TRUE, 'auto')
ON CONFLICT (product_id, channel) DO NOTHING
`,
[productId],
)
await upsertInternalSkuChannel({
internalSku,
channel: 'third_party_voucher',
@@ -89,21 +89,6 @@ export async function upsertAdminVoucherProduct(payload: JsonObject, session: Ad
throw createHttpError('卡券标准商品保存失败', { statusCode: 500 })
}
await client.query(
`
INSERT INTO product_channel_capabilities (
product_id, channel, priority, is_fallback, enabled, source, created_at, updated_at
) VALUES ($1, 'third_party_voucher', 10, FALSE, TRUE, 'voucher', $2, $2)
ON CONFLICT (product_id, channel) DO UPDATE SET
priority = 10,
is_fallback = FALSE,
enabled = TRUE,
source = 'voucher',
updated_at = EXCLUDED.updated_at
`,
[productId, now],
)
await upsertInternalSkuChannelWithClient(client, {
internalSku: skuKey,
channel: 'third_party_voucher',
@@ -2146,15 +2146,6 @@ async function ensureDraftProductForChannel(
const productId = Number(product?.id || 0)
if (!productId || product?.status !== 'active') return null
await query(
`
INSERT INTO product_channel_capabilities (
product_id, channel, priority, is_fallback, enabled, source
) VALUES ($1, $2, 10, FALSE, TRUE, 'auto')
ON CONFLICT (product_id, channel) DO NOTHING
`,
[productId, channel],
)
return productId
}
@@ -0,0 +1,145 @@
/**
* 内部 SKU 与渠道路由兼容层集成测试(需要真实 PostgreSQL)。
*
* INTERNAL_SKU_ROUTING_TEST_DATABASE_URL=postgres://... \
* npx tsx --test src/services/fulfillment/internal-sku-routing.integration.test.ts
*/
import assert from 'node:assert/strict'
import test from 'node:test'
const TEST_DATABASE_URL = String(process.env.INTERNAL_SKU_ROUTING_TEST_DATABASE_URL || '').trim()
test(
'旧 internal_sku_channels 写入会同步路由,商品来源与主渠道投影保持分离',
{ skip: !TEST_DATABASE_URL ? '未设置 INTERNAL_SKU_ROUTING_TEST_DATABASE_URL' : false },
async () => {
process.env.DATABASE_URL = TEST_DATABASE_URL
const { closeDb, query } = await import('../../db/client.js')
const suffix = `${Date.now()}_${Math.floor(Math.random() * 1_000_000)}`
const internalSku = `路由集成测试-${suffix}`
const canonicalSku = `man:${internalSku}`
let internalSkuId = 0
let productId = 0
try {
const createdSku = await query<{ id: string }>(
'SELECT ensure_canonical_internal_sku($1) AS id',
[internalSku],
)
internalSkuId = Number(createdSku.rows[0]?.id || 0)
assert.ok(internalSkuId > 0)
const createdProduct = await query<{ id: string; source_kind: string }>(
`
INSERT INTO product_catalog (
canonical_sku, canonical_name, category, unit_name, status, attributes_json,
created_at, updated_at
) VALUES ($1, $2, '集成测试', '件', 'active', '{}'::jsonb, NOW(), NOW())
RETURNING id, source_kind
`,
[canonicalSku, internalSku],
)
productId = Number(createdProduct.rows[0]?.id || 0)
assert.ok(productId > 0)
assert.equal(createdProduct.rows[0]?.source_kind, 'manual')
const beforeRoute = await query<{ active_primary_channel: string | null }>(
'SELECT active_primary_channel FROM v_product_current WHERE id = $1',
[productId],
)
assert.equal(
beforeRoute.rows[0]?.active_primary_channel,
null,
'man: 来源没有主路由时必须显示未路由',
)
await query(
`
INSERT INTO internal_sku_channels (
internal_sku, channel, priority, is_fallback, enabled, canonical_product_id, source
) VALUES ($1, 'manual', 10, FALSE, TRUE, $2, 'integration_test')
`,
[internalSku, productId],
)
const primary = await query<{ route_role: string; channel: string }>(
`
SELECT route_role, channel
FROM sku_channel_routes
WHERE internal_sku_id = $1 AND enabled = TRUE
`,
[internalSkuId],
)
assert.deepEqual(primary.rows, [{ route_role: 'primary', channel: 'manual' }])
const projection = await query<{
source_kind: string
active_primary_channel: string
internal_sku_id: string
}>(
`
SELECT source_kind, active_primary_channel, internal_sku_id
FROM v_product_current
WHERE id = $1
`,
[productId],
)
assert.equal(projection.rows[0]?.source_kind, 'manual')
assert.equal(projection.rows[0]?.active_primary_channel, 'manual')
assert.equal(Number(projection.rows[0]?.internal_sku_id), internalSkuId)
const { getAdminCostCenterOverview } =
await import('../cost-center/cost-center-overview-service.js')
const { listAdminProducts } = await import('../cost-center/product-catalog-service.js')
const overview = await getAdminCostCenterOverview({ scope: 'manual' })
const products = await listAdminProducts({ channel: 'manual', page: 1, pageSize: 100 })
assert.equal(products.pagination.total, overview.productCount)
assert.ok(products.items.every((product) => product.activePrimaryChannel === 'manual'))
await query(
`
UPDATE internal_sku_channels
SET is_fallback = TRUE, updated_at = NOW()
WHERE internal_sku = $1 AND channel = 'manual'
`,
[internalSku],
)
const fallback = await query<{ route_role: string }>(
'SELECT route_role FROM sku_channel_routes WHERE internal_sku_id = $1',
[internalSkuId],
)
assert.deepEqual(fallback.rows, [{ route_role: 'fallback' }])
await query(
`
UPDATE internal_sku_channels
SET is_fallback = FALSE, updated_at = NOW()
WHERE internal_sku = $1 AND channel = 'manual'
`,
[internalSku],
)
await assert.rejects(
() =>
query(
`
INSERT INTO sku_channel_routes (
internal_sku_id, channel, route_role, priority, enabled
) VALUES ($1, 'work_order', 'primary', 1, TRUE)
`,
[internalSkuId],
),
/uq_sku_channel_routes_primary|duplicate key/,
)
} finally {
if (productId > 0) {
await query('DELETE FROM product_catalog WHERE id = $1', [productId])
}
if (internalSkuId > 0) {
await query('DELETE FROM internal_sku_channels WHERE internal_sku = $1', [internalSku])
await query('DELETE FROM internal_skus WHERE id = $1', [internalSkuId])
}
await closeDb()
}
},
)
@@ -93,7 +93,7 @@ function buildRuleEntries(
})
}
// 只配了 sellerId 的"空条件结构化"(旧实现会整店命中)不编译:属于历史脏配置,
// 新模型一律要求显式判定维度(见 docs/接单模板匹配引擎重构设计.md §2.3)。
// 新模型一律要求显式判定维度(见 docs/archive/2026-11-08/接单模板匹配引擎重构设计.md §2.3)。
if (structuredAll.length > 0) {
entries.push({
key: `legacy:rule:${rule.id}:structured`,
@@ -5,7 +5,7 @@
* - 引擎:条件打分 + 决策(并列最高=ambiguous,不猜)+ 诊断(差在哪一维);
* - 价格/数量/拼单:`pricing.ts` 唯一实现。
*
* 设计见 docs/接单模板匹配引擎重构设计.md。
* 设计见 docs/archive/2026-11-08/接单模板匹配引擎重构设计.md。
*/
export type {
ProductMatchCandidate,