fix(import): term 边缘欠报闭合（null-id 双序等价/同 slug 双 id 报告）；handleEvent 分派形状
审查修复：
- null-id wp:term 双序等价：seen 分支同样报缺 term_id issue；自建行 name-only
收编（UPDATE name 为权威名、source_id 不动），term 先行/row 先行终态 name
与 issue 双序一致（测试钉死）；真实 id 行不覆盖（first-wins）
- 同 slug 双 term_id：镜像条件显式报告（保留先到、跳过后到 + rawLocation），
文档序 first-wins 策略 JSDoc 钉死（测试含关联绑保留行）
- 冲突重查未命中由 found[0]! 改显式 throw（taxonomy/term 两处，含键上下文）
- adoptTerm catch 注明 ctx.db 直连约定（事务内 25P02 连锁风险——Task 6 包
事务时须重估）；收编撞唯一约束路径补测试（自建行原样保留 + issue）
- 重构：内部件抽为 createTermEventHandler({handleEvent, finish}) 可分派形状
（Task 6 单消费方拉流分派、无 tee 的编排意图），createTermWriter 为自消费
整流便捷壳
Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
浅念
@@ -1,5 +1,5 @@
|
||||
// term 物化写入器（Plan 3 Task 2，spec §5.4/§7.5、前置事实 1/9、phase2-smoke 移交项①）：
|
||||
// 消费 ctx.events（AdapterEvent 流）把分类域落库——**双源物化**：
|
||||
// 消费适配器事件流（AdapterEvent）把分类域落库——**双源物化**：
|
||||
// 1. wp:term 事件（表数据旁路直通）：taxonomies 行（key→object_type 映射 +
|
||||
// public_base_path 前置事实 9 常量）+ terms 行（source_id = term_id 字符串）；
|
||||
// 2. row 的 taxonomyRefs（(domain, nicename) 锚点——phase2-smoke 实证：真实 WXR 的
|
||||
@@ -12,6 +12,15 @@
|
||||
// source_id/name 扶正为权威值（content_terms 挂的是 term id，不受收编影响）——两种事件
|
||||
// 顺序终态一致（测试钉死顺序无关性）。
|
||||
//
|
||||
// 缺陷身份的两条兜底策略（审查钉死）：
|
||||
// - **null-id wp:term 双序等价**：缺 term_id 的 wp:term 事件照建行（source_id=
|
||||
// auto:<slug>）并记 issue；若同 slug 行已见（row 先到自建），只做 **name-only 收编**
|
||||
// （UPDATE name 为 wp:term 权威名，source_id 不动——无权威值可扶正）。name 收编仅
|
||||
// 施于自建行（name=nicename 回退值）；真实 id 行的 name 已来自更完整的 wp:term 事件，
|
||||
// 按文档序 first-wins 不覆盖。两种顺序终态 name 一致、issue 均在（测试钉死）。
|
||||
// - **同 slug 双 term_id**：同一 slug 先后带两个不同 term_id 到达时**文档序
|
||||
// first-wins**——保留先到行的 source_id，后到跳过并记 issue（不静默）。
|
||||
//
|
||||
// name 的自建回退（如实声明）：UniversalRow.taxonomyRefs 只携带 (taxonomy,
|
||||
// sourceTermId=nicename)，不携带 <category> 文本 value（契约冻结，升版须评审）——自建
|
||||
// term 的 name 暂用 nicename；显示名回填留待后续契约决策。
|
||||
@@ -23,11 +32,11 @@
|
||||
// 事务边界：所有 upsert 均为单语句原子操作（insert ... on conflict do nothing +
|
||||
// returning / 单行 update），无跨语句不变量需要显式事务（约定见 ImportJobContext.db）。
|
||||
import { and, desc, eq, isNotNull } from 'drizzle-orm'
|
||||
import type { WxrTerm } from '@sf/adapter-wp'
|
||||
import type { AdapterEvent, WxrTerm } from '@sf/adapter-wp'
|
||||
import type { UniversalRow } from '@sf/contracts'
|
||||
import { contentTerms, sourceRecords, taxonomies, terms } from '../../db/schema.js'
|
||||
import { isUniqueViolation } from '../source/register.service.js'
|
||||
import type { ImportPipeline } from './import.service.js'
|
||||
import type { ImportDb, ImportPipeline, ImportPipelineOutcome } from './import.service.js'
|
||||
|
||||
/** 自建 term 的 source_id 前缀（与 wp:term 的 term_id 字符串源身份空间隔离） */
|
||||
const AUTO_SOURCE_ID_PREFIX = 'auto:'
|
||||
@@ -63,249 +72,325 @@ interface SeenTerm {
|
||||
}
|
||||
|
||||
/**
|
||||
* term 物化管道写入器。产出 counts：
|
||||
* 可分派的 term 写入器句柄（Task 6 编排形状）：编排器单消费方 for-await 拉一遍事件流、
|
||||
* 逐事件分派给各写入器（content/media/route 同形状）——不经 tee、不重迭代。本任务
|
||||
* {@link createTermWriter} 是"自消费整流"的便捷壳；两态（counts/issues）经
|
||||
* {@link TermEventHandler.finish} 取回。句柄非可重入：一个句柄只服务一遍流。
|
||||
*/
|
||||
export interface TermEventHandler {
|
||||
/** 单事件分派（unsupported/error 为 no-op——无分类信息，本写入器不消费） */
|
||||
handleEvent(ev: AdapterEvent): Promise<void>
|
||||
/** 流终取回累计产出（counts/issues；调用后句柄弃用） */
|
||||
finish(): ImportPipelineOutcome
|
||||
}
|
||||
|
||||
/**
|
||||
* term 物化的事件分派句柄。产出 counts：
|
||||
* - termsProcessed：处理的 wp:term 事件数；
|
||||
* - autoCreatedTerms：**终态**自建数（引用侧自建且未被 wp:term 收编的行数——瞬态自建
|
||||
* 被收编即回退，不虚报"wp:term 表缺失"）；
|
||||
* - contentTermsCreated：新建的 (entry, term) 关联数（幂等重跑为 0）。
|
||||
* issues：无 entry 可关联的 row 分类引用、表外 taxonomy 兜底、缺 term_id 的 wp:term
|
||||
* 事件、收编撞唯一约束——全部显式报告（§7.6 禁静默）。
|
||||
* 事件、同 slug 双 term_id、收编撞唯一约束——全部显式报告（§7.6 禁静默）。
|
||||
*/
|
||||
export function createTermWriter(): ImportPipeline {
|
||||
return async (ctx) => {
|
||||
const db = ctx.db
|
||||
const counts = { termsProcessed: 0, autoCreatedTerms: 0, contentTermsCreated: 0 }
|
||||
const issues: string[] = []
|
||||
const taxonomyIds = new Map<string, string>()
|
||||
const slugIndex = new Map<string, SeenTerm>()
|
||||
const flaggedUnknownTaxonomies = new Set<string>()
|
||||
export function createTermEventHandler(identity: {
|
||||
db: ImportDb
|
||||
siteId: string
|
||||
/** entry 解析（source_records 稳定身份查询）用的连接锚点 */
|
||||
connectionId: string
|
||||
}): TermEventHandler {
|
||||
const db = identity.db
|
||||
const siteId = identity.siteId
|
||||
const counts = { termsProcessed: 0, autoCreatedTerms: 0, contentTermsCreated: 0 }
|
||||
const issues: string[] = []
|
||||
const taxonomyIds = new Map<string, string>()
|
||||
const slugIndex = new Map<string, SeenTerm>()
|
||||
const flaggedUnknownTaxonomies = new Set<string>()
|
||||
|
||||
const idxKey = (taxonomyKey: string, slug: string): string => `${taxonomyKey}\u0000${slug}`
|
||||
const isAuto = (sourceId: string): boolean => sourceId.startsWith(AUTO_SOURCE_ID_PREFIX)
|
||||
const idxKey = (taxonomyKey: string, slug: string): string => `${taxonomyKey}\u0000${slug}`
|
||||
const isAuto = (sourceId: string): boolean => sourceId.startsWith(AUTO_SOURCE_ID_PREFIX)
|
||||
|
||||
/** taxonomy 行 ensure（UNIQUE(site,key,object_type) 幂等；常量表外 key 记 issue） */
|
||||
async function ensureTaxonomy(key: string): Promise<string> {
|
||||
const cached = taxonomyIds.get(key)
|
||||
if (cached !== undefined) return cached
|
||||
const defaults = taxonomyDefaultsFor(key)
|
||||
if (!flaggedUnknownTaxonomies.has(key) && !Object.hasOwn(TERM_TAXONOMY_DEFAULTS, key)) {
|
||||
flaggedUnknownTaxonomies.add(key)
|
||||
issues.push(
|
||||
`未知 taxonomy key：${JSON.stringify(key)}（不在前置事实 9 常量表）——object_type 兜底 '${defaults.objectType}'、public_base_path 兜底 '${defaults.publicBasePath}'（仅记录，路由不消费）`,
|
||||
)
|
||||
}
|
||||
const inserted = await db
|
||||
.insert(taxonomies)
|
||||
.values({ siteId: ctx.siteId, key, objectType: defaults.objectType, publicBasePath: defaults.publicBasePath })
|
||||
.onConflictDoNothing()
|
||||
.returning({ id: taxonomies.id })
|
||||
let id: string
|
||||
if (inserted.length > 0) {
|
||||
id = inserted[0]!.id
|
||||
} else {
|
||||
// 撞唯一三元组（重跑/既有行）：重查复用——插入冲突即行存在，必命中
|
||||
const found = await db
|
||||
.select({ id: taxonomies.id })
|
||||
.from(taxonomies)
|
||||
.where(
|
||||
and(
|
||||
eq(taxonomies.siteId, ctx.siteId),
|
||||
eq(taxonomies.key, key),
|
||||
eq(taxonomies.objectType, defaults.objectType),
|
||||
),
|
||||
)
|
||||
.limit(1)
|
||||
id = found[0]!.id
|
||||
}
|
||||
taxonomyIds.set(key, id)
|
||||
return id
|
||||
/** taxonomy 行 ensure（UNIQUE(site,key,object_type) 幂等；常量表外 key 记 issue） */
|
||||
async function ensureTaxonomy(key: string): Promise<string> {
|
||||
const cached = taxonomyIds.get(key)
|
||||
if (cached !== undefined) return cached
|
||||
const defaults = taxonomyDefaultsFor(key)
|
||||
if (!flaggedUnknownTaxonomies.has(key) && !Object.hasOwn(TERM_TAXONOMY_DEFAULTS, key)) {
|
||||
flaggedUnknownTaxonomies.add(key)
|
||||
issues.push(
|
||||
`未知 taxonomy key：${JSON.stringify(key)}（不在前置事实 9 常量表）——object_type 兜底 '${defaults.objectType}'、public_base_path 兜底 '${defaults.publicBasePath}'（仅记录，路由不消费）`,
|
||||
)
|
||||
}
|
||||
|
||||
/** 按 (taxonomy, slug) 查既有 term（重跑/跨管道持久化行；slug 无唯一索引，取文档序首行） */
|
||||
async function findTermBySlug(taxonomyId: string, slug: string): Promise<SeenTerm | null> {
|
||||
const rows = await db
|
||||
.select({ id: terms.id, sourceId: terms.sourceId })
|
||||
.from(terms)
|
||||
.where(and(eq(terms.siteId, ctx.siteId), eq(terms.taxonomyId, taxonomyId), eq(terms.slug, slug)))
|
||||
.orderBy(terms.sourceId)
|
||||
.limit(1)
|
||||
if (rows.length === 0) return null
|
||||
return { id: rows[0]!.id, sourceId: rows[0]!.sourceId, countedAuto: false }
|
||||
}
|
||||
|
||||
/** term 行插入（UNIQUE(site,taxonomy,source_id) 兜底：冲突即重查复用——同键必命中） */
|
||||
async function insertTerm(args: {
|
||||
taxonomyId: string
|
||||
sourceId: string
|
||||
slug: string
|
||||
name: string
|
||||
data: Record<string, unknown>
|
||||
}): Promise<string> {
|
||||
const inserted = await db
|
||||
.insert(terms)
|
||||
.values({
|
||||
siteId: ctx.siteId,
|
||||
taxonomyId: args.taxonomyId,
|
||||
sourceId: args.sourceId,
|
||||
slug: args.slug,
|
||||
name: args.name,
|
||||
data: args.data,
|
||||
})
|
||||
.onConflictDoNothing()
|
||||
.returning({ id: terms.id })
|
||||
if (inserted.length > 0) return inserted[0]!.id
|
||||
const inserted = await db
|
||||
.insert(taxonomies)
|
||||
.values({ siteId, key, objectType: defaults.objectType, publicBasePath: defaults.publicBasePath })
|
||||
.onConflictDoNothing()
|
||||
.returning({ id: taxonomies.id })
|
||||
let id: string
|
||||
if (inserted.length > 0) {
|
||||
id = inserted[0]!.id
|
||||
} else {
|
||||
// 撞唯一三元组（重跑/既有行）：重查复用——插入冲突即行存在，必命中
|
||||
const found = await db
|
||||
.select({ id: terms.id })
|
||||
.from(terms)
|
||||
.select({ id: taxonomies.id })
|
||||
.from(taxonomies)
|
||||
.where(
|
||||
and(
|
||||
eq(terms.siteId, ctx.siteId),
|
||||
eq(terms.taxonomyId, args.taxonomyId),
|
||||
eq(terms.sourceId, args.sourceId),
|
||||
eq(taxonomies.siteId, siteId),
|
||||
eq(taxonomies.key, key),
|
||||
eq(taxonomies.objectType, defaults.objectType),
|
||||
),
|
||||
)
|
||||
.limit(1)
|
||||
return found[0]!.id
|
||||
}
|
||||
|
||||
/** 收编：自建行就地 UPDATE 为 wp:term 权威身份（source_id/name）；撞唯一约束则保留自建行并记 issue */
|
||||
async function adoptTerm(seen: SeenTerm, realSourceId: string, name: string, taxonomyKey: string): Promise<void> {
|
||||
try {
|
||||
await db.update(terms).set({ sourceId: realSourceId, name }).where(eq(terms.id, seen.id))
|
||||
} catch (err) {
|
||||
if (isUniqueViolation(err)) {
|
||||
issues.push(
|
||||
`term 收编撞唯一约束：taxonomy=${taxonomyKey} source_id=${realSourceId} 已被他行占用——自建行保留（source_id=${seen.sourceId}）`,
|
||||
)
|
||||
return
|
||||
}
|
||||
throw err
|
||||
const row = found[0]
|
||||
if (row === undefined) {
|
||||
throw new Error(
|
||||
`taxonomy 冲突重查未命中（不可能路径：taxonomies_uq 冲突即行存在）——key=${JSON.stringify(key)} objectType=${defaults.objectType}`,
|
||||
)
|
||||
}
|
||||
if (seen.countedAuto) counts.autoCreatedTerms--
|
||||
seen.sourceId = realSourceId
|
||||
seen.countedAuto = false
|
||||
id = row.id
|
||||
}
|
||||
taxonomyIds.set(key, id)
|
||||
return id
|
||||
}
|
||||
|
||||
/** wp:term 事件：建 taxonomy/term；同 slug 已有自建行（row 先到）→ 收编 */
|
||||
async function handleTermEvent(term: WxrTerm): Promise<void> {
|
||||
counts.termsProcessed++
|
||||
const taxonomyId = await ensureTaxonomy(term.taxonomy)
|
||||
const realSourceId = term.termId === null ? null : String(term.termId)
|
||||
const key = idxKey(term.taxonomy, term.slug)
|
||||
/** 按 (taxonomy, slug) 查既有 term（重跑/跨管道持久化行；slug 无唯一索引，取文档序首行） */
|
||||
async function findTermBySlug(taxonomyId: string, slug: string): Promise<SeenTerm | null> {
|
||||
const rows = await db
|
||||
.select({ id: terms.id, sourceId: terms.sourceId })
|
||||
.from(terms)
|
||||
.where(and(eq(terms.siteId, siteId), eq(terms.taxonomyId, taxonomyId), eq(terms.slug, slug)))
|
||||
.orderBy(terms.sourceId)
|
||||
.limit(1)
|
||||
const row = rows[0]
|
||||
if (row === undefined) return null
|
||||
return { id: row.id, sourceId: row.sourceId, countedAuto: false }
|
||||
}
|
||||
|
||||
/** term 行插入（UNIQUE(site,taxonomy,source_id) 兜底：冲突即重查复用——同键必命中） */
|
||||
async function insertTerm(args: {
|
||||
taxonomyId: string
|
||||
sourceId: string
|
||||
slug: string
|
||||
name: string
|
||||
data: Record<string, unknown>
|
||||
}): Promise<string> {
|
||||
const inserted = await db
|
||||
.insert(terms)
|
||||
.values({
|
||||
siteId,
|
||||
taxonomyId: args.taxonomyId,
|
||||
sourceId: args.sourceId,
|
||||
slug: args.slug,
|
||||
name: args.name,
|
||||
data: args.data,
|
||||
})
|
||||
.onConflictDoNothing()
|
||||
.returning({ id: terms.id })
|
||||
if (inserted.length > 0) return inserted[0]!.id
|
||||
const found = await db
|
||||
.select({ id: terms.id })
|
||||
.from(terms)
|
||||
.where(
|
||||
and(
|
||||
eq(terms.siteId, siteId),
|
||||
eq(terms.taxonomyId, args.taxonomyId),
|
||||
eq(terms.sourceId, args.sourceId),
|
||||
),
|
||||
)
|
||||
.limit(1)
|
||||
const row = found[0]
|
||||
if (row === undefined) {
|
||||
throw new Error(
|
||||
`term 冲突重查未命中（不可能路径：terms_identity_uq 冲突即行存在）——taxonomyId=${args.taxonomyId} sourceId=${args.sourceId}`,
|
||||
)
|
||||
}
|
||||
return row.id
|
||||
}
|
||||
|
||||
/** 收编：自建行就地 UPDATE 为 wp:term 权威身份（source_id/name）；撞唯一约束则保留自建行并记 issue */
|
||||
async function adoptTerm(seen: SeenTerm, realSourceId: string, name: string, taxonomyKey: string): Promise<void> {
|
||||
try {
|
||||
await db.update(terms).set({ sourceId: realSourceId, name }).where(eq(terms.id, seen.id))
|
||||
} catch (err) {
|
||||
if (isUniqueViolation(err)) {
|
||||
// "记 issue 继续"依赖 ctx.db 直连约定（ImportJobContext.db JSDoc）：单语句失败
|
||||
// 不污染连接、后续语句照常。若 Task 6 把写入器包进 db.transaction，本 catch 须
|
||||
// 重估——事务内语句失败即整体中止（25P02），续查/续写都会被连锁拒绝。
|
||||
issues.push(
|
||||
`term 收编撞唯一约束：taxonomy=${taxonomyKey} source_id=${realSourceId} 已被他行占用——自建行保留（source_id=${seen.sourceId}）`,
|
||||
)
|
||||
return
|
||||
}
|
||||
throw err
|
||||
}
|
||||
if (seen.countedAuto) counts.autoCreatedTerms--
|
||||
seen.sourceId = realSourceId
|
||||
seen.countedAuto = false
|
||||
}
|
||||
|
||||
/** wp:term 事件：建 taxonomy/term；同 slug 已有自建行（row 先到）→ 收编 */
|
||||
async function handleTermEvent(term: WxrTerm): Promise<void> {
|
||||
counts.termsProcessed++
|
||||
const taxonomyId = await ensureTaxonomy(term.taxonomy)
|
||||
const realSourceId = term.termId === null ? null : String(term.termId)
|
||||
const key = idxKey(term.taxonomy, term.slug)
|
||||
let seen = slugIndex.get(key) ?? null
|
||||
if (seen === null) {
|
||||
const dbTerm = await findTermBySlug(taxonomyId, term.slug)
|
||||
if (dbTerm !== null) {
|
||||
slugIndex.set(key, dbTerm)
|
||||
seen = dbTerm
|
||||
}
|
||||
}
|
||||
if (seen !== null) {
|
||||
if (realSourceId === null) {
|
||||
// 缺 term_id 的 wp:term 事件（与下方建行分支同一措辞，两分支都报告）
|
||||
issues.push(
|
||||
`wp:term 事件缺 term_id（${term.rawLocation}，taxonomy=${term.taxonomy} slug=${term.slug}）——source_id 以 ${AUTO_SOURCE_ID_PREFIX}<slug> 落位`,
|
||||
)
|
||||
// name-only 收编：自建行的 name（nicename 回退值）扶正为 wp:term 权威名；
|
||||
// source_id 不动（无权威值）。真实 id 行不覆盖（先到的完整事件已给权威名，
|
||||
// first-wins 与同 slug 双 id 策略一致）；name 为空串的源数据不具权威性，跳过
|
||||
if (isAuto(seen.sourceId) && term.name !== '') {
|
||||
await db.update(terms).set({ name: term.name }).where(eq(terms.id, seen.id))
|
||||
}
|
||||
return
|
||||
}
|
||||
if (isAuto(seen.sourceId)) {
|
||||
if (seen.sourceId !== realSourceId) {
|
||||
await adoptTerm(seen, realSourceId, term.name, term.taxonomy)
|
||||
}
|
||||
return // 同 id 重复事件：幂等跳过
|
||||
}
|
||||
// 真实 id 行 + 不同 term_id：同 slug 双 term_id——文档序 first-wins（保留先到、后到跳过并报告）
|
||||
if (seen.sourceId !== realSourceId) {
|
||||
issues.push(`同 slug 双 term_id：保留 ${seen.sourceId}、跳过 ${realSourceId}（${term.rawLocation}）`)
|
||||
}
|
||||
return // 同 id 重复事件：幂等跳过
|
||||
}
|
||||
if (realSourceId === null) {
|
||||
issues.push(
|
||||
`wp:term 事件缺 term_id（${term.rawLocation}，taxonomy=${term.taxonomy} slug=${term.slug}）——source_id 以 ${AUTO_SOURCE_ID_PREFIX}<slug> 落位`,
|
||||
)
|
||||
}
|
||||
const sourceId = realSourceId ?? `${AUTO_SOURCE_ID_PREFIX}${term.slug}`
|
||||
const id = await insertTerm({
|
||||
taxonomyId,
|
||||
sourceId,
|
||||
slug: term.slug,
|
||||
name: term.name,
|
||||
data:
|
||||
realSourceId === null
|
||||
? { origin: 'wp:term-no-id' }
|
||||
: { origin: 'wp:term', rawLocation: term.rawLocation },
|
||||
})
|
||||
slugIndex.set(key, { id, sourceId, countedAuto: false })
|
||||
}
|
||||
|
||||
/** row 的稳定身份 → content entry（source_records.content_entry_id 是内容写入器的接缝列） */
|
||||
async function resolveContentEntryId(row: UniversalRow): Promise<string | null> {
|
||||
const rows = await db
|
||||
.select({ contentEntryId: sourceRecords.contentEntryId })
|
||||
.from(sourceRecords)
|
||||
.where(
|
||||
and(
|
||||
eq(sourceRecords.siteId, siteId),
|
||||
eq(sourceRecords.sourceConnectionId, identity.connectionId),
|
||||
eq(sourceRecords.sourceType, row.sourceType),
|
||||
eq(sourceRecords.sourceId, row.sourceId),
|
||||
eq(sourceRecords.locale, row.locale),
|
||||
isNotNull(sourceRecords.contentEntryId),
|
||||
),
|
||||
)
|
||||
.orderBy(desc(sourceRecords.createdAt))
|
||||
.limit(1)
|
||||
return rows.length > 0 ? rows[0]!.contentEntryId : null
|
||||
}
|
||||
|
||||
/** row 事件：先收集 refs 建齐 term（含自建缺失），再关联 content_terms（顺序无关性） */
|
||||
async function handleRow(row: UniversalRow): Promise<void> {
|
||||
if (row.taxonomyRefs.length === 0) return
|
||||
const resolvedTermIds: string[] = []
|
||||
for (const ref of row.taxonomyRefs) {
|
||||
const taxonomyId = await ensureTaxonomy(ref.taxonomy)
|
||||
const key = idxKey(ref.taxonomy, ref.sourceTermId)
|
||||
let seen = slugIndex.get(key) ?? null
|
||||
if (seen === null) {
|
||||
const dbTerm = await findTermBySlug(taxonomyId, term.slug)
|
||||
const dbTerm = await findTermBySlug(taxonomyId, ref.sourceTermId)
|
||||
if (dbTerm !== null) {
|
||||
slugIndex.set(key, dbTerm)
|
||||
seen = dbTerm
|
||||
}
|
||||
}
|
||||
if (seen !== null) {
|
||||
if (isAuto(seen.sourceId) && realSourceId !== null && seen.sourceId !== realSourceId) {
|
||||
await adoptTerm(seen, realSourceId, term.name, term.taxonomy)
|
||||
}
|
||||
return // 已物化（同流重复事件/重跑既有行）：幂等跳过
|
||||
if (seen === null) {
|
||||
// 自建缺失 term（移交项①）：name 用 nicename——UniversalRow 不携带 <category>
|
||||
// 文本 value（契约冻结），显示名回填留待后续契约决策
|
||||
const sourceId = `${AUTO_SOURCE_ID_PREFIX}${ref.sourceTermId}`
|
||||
const id = await insertTerm({
|
||||
taxonomyId,
|
||||
sourceId,
|
||||
slug: ref.sourceTermId,
|
||||
name: ref.sourceTermId,
|
||||
data: { origin: 'auto', firstSeenRawLocation: row.rawLocation },
|
||||
})
|
||||
seen = { id, sourceId, countedAuto: true }
|
||||
slugIndex.set(key, seen)
|
||||
counts.autoCreatedTerms++
|
||||
}
|
||||
if (realSourceId === null) {
|
||||
issues.push(
|
||||
`wp:term 事件缺 term_id（${term.rawLocation}，taxonomy=${term.taxonomy} slug=${term.slug}）——source_id 以 ${AUTO_SOURCE_ID_PREFIX}<slug> 落位`,
|
||||
)
|
||||
}
|
||||
const sourceId = realSourceId ?? `${AUTO_SOURCE_ID_PREFIX}${term.slug}`
|
||||
const id = await insertTerm({
|
||||
taxonomyId,
|
||||
sourceId,
|
||||
slug: term.slug,
|
||||
name: term.name,
|
||||
data:
|
||||
realSourceId === null
|
||||
? { origin: 'wp:term-no-id' }
|
||||
: { origin: 'wp:term', rawLocation: term.rawLocation },
|
||||
})
|
||||
slugIndex.set(key, { id, sourceId, countedAuto: false })
|
||||
resolvedTermIds.push(seen.id)
|
||||
}
|
||||
|
||||
/** row 的稳定身份 → content entry（source_records.content_entry_id 是内容写入器的接缝列） */
|
||||
async function resolveContentEntryId(row: UniversalRow): Promise<string | null> {
|
||||
const rows = await db
|
||||
.select({ contentEntryId: sourceRecords.contentEntryId })
|
||||
.from(sourceRecords)
|
||||
.where(
|
||||
and(
|
||||
eq(sourceRecords.siteId, ctx.siteId),
|
||||
eq(sourceRecords.sourceConnectionId, ctx.connectionId),
|
||||
eq(sourceRecords.sourceType, row.sourceType),
|
||||
eq(sourceRecords.sourceId, row.sourceId),
|
||||
eq(sourceRecords.locale, row.locale),
|
||||
isNotNull(sourceRecords.contentEntryId),
|
||||
),
|
||||
)
|
||||
.orderBy(desc(sourceRecords.createdAt))
|
||||
.limit(1)
|
||||
return rows.length > 0 ? rows[0]!.contentEntryId : null
|
||||
const entryId = await resolveContentEntryId(row)
|
||||
if (entryId === null) {
|
||||
issues.push(
|
||||
`row 的分类引用无对应 content entry（${row.sourceType}/${row.sourceId}/${row.locale}，refs=${row.taxonomyRefs.length}）——term 已物化，content_terms 跳过`,
|
||||
)
|
||||
return
|
||||
}
|
||||
|
||||
/** row 事件：先收集 refs 建齐 term（含自建缺失），再关联 content_terms（顺序无关性） */
|
||||
async function handleRow(row: UniversalRow): Promise<void> {
|
||||
if (row.taxonomyRefs.length === 0) return
|
||||
const resolvedTermIds: string[] = []
|
||||
for (const ref of row.taxonomyRefs) {
|
||||
const taxonomyId = await ensureTaxonomy(ref.taxonomy)
|
||||
const key = idxKey(ref.taxonomy, ref.sourceTermId)
|
||||
let seen = slugIndex.get(key) ?? null
|
||||
if (seen === null) {
|
||||
const dbTerm = await findTermBySlug(taxonomyId, ref.sourceTermId)
|
||||
if (dbTerm !== null) {
|
||||
slugIndex.set(key, dbTerm)
|
||||
seen = dbTerm
|
||||
}
|
||||
}
|
||||
if (seen === null) {
|
||||
// 自建缺失 term（移交项①）：name 用 nicename——UniversalRow 不携带 <category>
|
||||
// 文本 value（契约冻结），显示名回填留待后续契约决策
|
||||
const sourceId = `${AUTO_SOURCE_ID_PREFIX}${ref.sourceTermId}`
|
||||
const id = await insertTerm({
|
||||
taxonomyId,
|
||||
sourceId,
|
||||
slug: ref.sourceTermId,
|
||||
name: ref.sourceTermId,
|
||||
data: { origin: 'auto', firstSeenRawLocation: row.rawLocation },
|
||||
})
|
||||
seen = { id, sourceId, countedAuto: true }
|
||||
slugIndex.set(key, seen)
|
||||
counts.autoCreatedTerms++
|
||||
}
|
||||
resolvedTermIds.push(seen.id)
|
||||
}
|
||||
const entryId = await resolveContentEntryId(row)
|
||||
if (entryId === null) {
|
||||
issues.push(
|
||||
`row 的分类引用无对应 content entry（${row.sourceType}/${row.sourceId}/${row.locale}，refs=${row.taxonomyRefs.length}）——term 已物化，content_terms 跳过`,
|
||||
)
|
||||
return
|
||||
}
|
||||
const primarySeen = new Set<string>()
|
||||
for (let i = 0; i < row.taxonomyRefs.length; i++) {
|
||||
const ref = row.taxonomyRefs[i]!
|
||||
const isPrimary = !primarySeen.has(ref.taxonomy)
|
||||
if (isPrimary) primarySeen.add(ref.taxonomy)
|
||||
const inserted = await db
|
||||
.insert(contentTerms)
|
||||
.values({ siteId: ctx.siteId, contentEntryId: entryId, termId: resolvedTermIds[i]!, isPrimary })
|
||||
.onConflictDoNothing()
|
||||
.returning({ contentEntryId: contentTerms.contentEntryId })
|
||||
if (inserted.length > 0) counts.contentTermsCreated++
|
||||
}
|
||||
const primarySeen = new Set<string>()
|
||||
for (let i = 0; i < row.taxonomyRefs.length; i++) {
|
||||
const ref = row.taxonomyRefs[i]!
|
||||
const isPrimary = !primarySeen.has(ref.taxonomy)
|
||||
if (isPrimary) primarySeen.add(ref.taxonomy)
|
||||
const inserted = await db
|
||||
.insert(contentTerms)
|
||||
.values({ siteId, contentEntryId: entryId, termId: resolvedTermIds[i]!, isPrimary })
|
||||
.onConflictDoNothing()
|
||||
.returning({ contentEntryId: contentTerms.contentEntryId })
|
||||
if (inserted.length > 0) counts.contentTermsCreated++
|
||||
}
|
||||
}
|
||||
|
||||
for await (const ev of ctx.events) {
|
||||
return {
|
||||
async handleEvent(ev) {
|
||||
if (ev.kind === 'term') {
|
||||
await handleTermEvent(ev.term)
|
||||
} else if (ev.kind === 'row') {
|
||||
await handleRow(ev.row)
|
||||
}
|
||||
// unsupported/error 事件：无分类信息，本写入器不消费（Task 6 各写入器各取所需）
|
||||
}
|
||||
return { counts, issues }
|
||||
},
|
||||
finish() {
|
||||
return { counts, issues }
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* term 物化管道写入器（{@link createTermEventHandler} 的自消费整流便捷壳）：单遍拉完
|
||||
* ctx.events、逐事件分派、流终返回 outcome。Task 6 编排时可改用句柄形状直接分派
|
||||
* （单消费方拉流分派给各写入器，无 tee）。
|
||||
*/
|
||||
export function createTermWriter(): ImportPipeline {
|
||||
return async (ctx) => {
|
||||
const handler = createTermEventHandler({
|
||||
db: ctx.db,
|
||||
siteId: ctx.siteId,
|
||||
connectionId: ctx.connectionId,
|
||||
})
|
||||
for await (const ev of ctx.events) {
|
||||
await handler.handleEvent(ev)
|
||||
}
|
||||
return handler.finish()
|
||||
}
|
||||
}
|
||||
|
||||
@@ -462,4 +462,218 @@ describe('term 物化（含自建缺失）', () => {
|
||||
}
|
||||
})
|
||||
})
|
||||
|
||||
it('null-id wp:term 双序等价：name-only 收编——term 先行/row 先行终态 name 与 issue 双序一致', async () => {
|
||||
await withTestDb(async (db) => {
|
||||
// 两序各起独立站点（终态对比不受他序残留影响）；work 根共用、子目录隔离
|
||||
const work = await mkdtemp(join(tmpdir(), 'sf-terms-nullid-'))
|
||||
const runScenario = async (order: 'term-first' | 'row-first') => {
|
||||
const dir = join(work, order)
|
||||
const key = `sfterm-${rand()}`
|
||||
const reg = await registerSource(db, {
|
||||
siteKey: key,
|
||||
wxrPath: WXR_FIXTURE,
|
||||
allowedRoots: [FIXTURES_ROOT],
|
||||
storageRoot: join(dir, 'storage'),
|
||||
reportRoot: join(dir, 'artifacts', 'register'),
|
||||
})
|
||||
try {
|
||||
const wxr = reg.snapshots.find((s) => s.kind === 'wxr')!
|
||||
const nullIdTerm = termEvent({
|
||||
termId: null,
|
||||
taxonomy: 'category',
|
||||
slug: 'company-news',
|
||||
name: 'Company News',
|
||||
})
|
||||
const row = rowEvent({
|
||||
sourceId: '501',
|
||||
taxonomyRefs: [{ taxonomy: 'category', sourceTermId: 'company-news' }],
|
||||
})
|
||||
const result = await createImportJob(db, {
|
||||
siteKey: key,
|
||||
snapshotId: wxr.id,
|
||||
reportRoot: join(dir, 'artifacts', 'import'),
|
||||
events: eventsOf(order === 'term-first' ? [nullIdTerm, row] : [row, nullIdTerm]),
|
||||
pipeline: seededTermWriter(async (ctx) => {
|
||||
await seedContentEntry(ctx, { sourceType: 'wp:post', sourceId: '501', typeKey: 'news_article' })
|
||||
}),
|
||||
})
|
||||
expect(result.status).toBe('succeeded_with_issues')
|
||||
const termsRows = (await db.execute(`select t.source_id, t.slug, t.name
|
||||
from terms t join taxonomies x on x.id = t.taxonomy_id
|
||||
where t.site_id = '${reg.siteId}'`)) as any
|
||||
const ct = (await db.execute(`select r.source_id as row_id, t.slug, c.is_primary
|
||||
from content_terms c
|
||||
join source_records r on r.content_entry_id = c.content_entry_id
|
||||
join terms t on t.id = c.term_id
|
||||
where c.site_id = '${reg.siteId}'`)) as any
|
||||
const report = JSON.parse(
|
||||
await readFile(join(dir, 'artifacts', 'import', result.jobId, 'reports', 'import.json'), 'utf8'),
|
||||
) as { issues: string[] }
|
||||
return { terms: plain(termsRows), ct: plain(ct), issues: report.issues }
|
||||
} finally {
|
||||
await cleanupSiteRows(db, reg.siteId)
|
||||
}
|
||||
}
|
||||
try {
|
||||
const termFirst = await runScenario('term-first')
|
||||
const rowFirst = await runScenario('row-first')
|
||||
// 双序终态一致：单行 auto:company-news、name 为 wp:term 权威名（name-only 收编）
|
||||
const expectedTerms = [
|
||||
{ source_id: 'auto:company-news', slug: 'company-news', name: 'Company News' },
|
||||
]
|
||||
expect(termFirst.terms).toStrictEqual(expectedTerms)
|
||||
expect(rowFirst.terms).toStrictEqual(expectedTerms)
|
||||
expect(termFirst.ct).toStrictEqual(rowFirst.ct)
|
||||
expect(termFirst.ct).toStrictEqual([
|
||||
{ row_id: '501', slug: 'company-news', is_primary: true },
|
||||
])
|
||||
// issue 双序均在（缺 term_id 的 wp:term 事件无论先后到都报告）
|
||||
expect(termFirst.issues).toHaveLength(1)
|
||||
expect(rowFirst.issues).toHaveLength(1)
|
||||
expect(termFirst.issues[0]).toMatch(/缺 term_id/)
|
||||
expect(rowFirst.issues[0]).toMatch(/缺 term_id/)
|
||||
} finally {
|
||||
await rm(work, { recursive: true, force: true })
|
||||
}
|
||||
})
|
||||
})
|
||||
|
||||
it('同 slug 双 term_id：文档序 first-wins——保留先到 id、后到跳过并报告，关联绑保留行、name 不被覆盖', async () => {
|
||||
await withTestDb(async (db) => {
|
||||
const work = await mkdtemp(join(tmpdir(), 'sf-terms-dualid-'))
|
||||
try {
|
||||
const key = `sfterm-${rand()}`
|
||||
const reg = await registerSource(db, {
|
||||
siteKey: key,
|
||||
wxrPath: WXR_FIXTURE,
|
||||
allowedRoots: [FIXTURES_ROOT],
|
||||
storageRoot: join(work, 'storage'),
|
||||
reportRoot: join(work, 'artifacts', 'register'),
|
||||
})
|
||||
try {
|
||||
const wxr = reg.snapshots.find((s) => s.kind === 'wxr')!
|
||||
const events: AdapterEvent[] = [
|
||||
termEvent({ termId: 31, taxonomy: 'category', slug: 'company', name: 'Company' }),
|
||||
termEvent({ termId: 99, taxonomy: 'category', slug: 'company', name: 'Company Dup' }),
|
||||
rowEvent({ sourceId: '501', taxonomyRefs: [{ taxonomy: 'category', sourceTermId: 'company' }] }),
|
||||
]
|
||||
const result = await createImportJob(db, {
|
||||
siteKey: key,
|
||||
snapshotId: wxr.id,
|
||||
reportRoot: join(work, 'artifacts', 'import'),
|
||||
events: eventsOf(events),
|
||||
pipeline: seededTermWriter(async (ctx) => {
|
||||
await seedContentEntry(ctx, { sourceType: 'wp:post', sourceId: '501', typeKey: 'news_article' })
|
||||
}),
|
||||
})
|
||||
expect(result.status).toBe('succeeded_with_issues')
|
||||
|
||||
// 单行：先到的 31 保留（name 不被后到 99 的事件覆盖——first-wins）
|
||||
const termsRows = (await db.execute(`select t.source_id, t.slug, t.name
|
||||
from terms t join taxonomies x on x.id = t.taxonomy_id
|
||||
where t.site_id = '${reg.siteId}'`)) as any
|
||||
expect(plain(termsRows)).toStrictEqual([
|
||||
{ source_id: '31', slug: 'company', name: 'Company' },
|
||||
])
|
||||
|
||||
// issue：保留/跳过双 id + rawLocation 均可审计
|
||||
const report = JSON.parse(
|
||||
await readFile(join(work, 'artifacts', 'import', result.jobId, 'reports', 'import.json'), 'utf8'),
|
||||
) as { counts: Record<string, number>; issues: string[] }
|
||||
expect(report.issues).toHaveLength(1)
|
||||
expect(report.issues[0]).toMatch(/同 slug 双 term_id：保留 31、跳过 99/)
|
||||
|
||||
// 关联绑保留行（31）；计数零自建
|
||||
const ct = (await db.execute(`select r.source_id as row_id, t.source_id as term_source_id, c.is_primary
|
||||
from content_terms c
|
||||
join source_records r on r.content_entry_id = c.content_entry_id
|
||||
join terms t on t.id = c.term_id
|
||||
where c.site_id = '${reg.siteId}'`)) as any
|
||||
expect(plain(ct)).toStrictEqual([
|
||||
{ row_id: '501', term_source_id: '31', is_primary: true },
|
||||
])
|
||||
expect(report.counts).toStrictEqual({
|
||||
termsProcessed: 2,
|
||||
autoCreatedTerms: 0,
|
||||
contentTermsCreated: 1,
|
||||
})
|
||||
} finally {
|
||||
await cleanupSiteRows(db, reg.siteId)
|
||||
}
|
||||
} finally {
|
||||
await rm(work, { recursive: true, force: true })
|
||||
}
|
||||
})
|
||||
})
|
||||
|
||||
it('收编撞唯一约束：source_id 已被他行占用 → 自建行原样保留 + issue（写入器唯一 try/catch 路径）', async () => {
|
||||
await withTestDb(async (db) => {
|
||||
const work = await mkdtemp(join(tmpdir(), 'sf-terms-adopt-'))
|
||||
try {
|
||||
const key = `sfterm-${rand()}`
|
||||
const reg = await registerSource(db, {
|
||||
siteKey: key,
|
||||
wxrPath: WXR_FIXTURE,
|
||||
allowedRoots: [FIXTURES_ROOT],
|
||||
storageRoot: join(work, 'storage'),
|
||||
reportRoot: join(work, 'artifacts', 'register'),
|
||||
})
|
||||
try {
|
||||
const wxr = reg.snapshots.find((s) => s.kind === 'wxr')!
|
||||
// 事件 1 占位 source_id=31（slug 不同→不自建命中）；事件 2 按 slug 自建
|
||||
// auto:company；事件 3 收编 UPDATE auto:company→31 撞 terms_identity_uq
|
||||
const events: AdapterEvent[] = [
|
||||
termEvent({ termId: 31, taxonomy: 'category', slug: 'company-alt', name: 'Alt' }),
|
||||
rowEvent({ sourceId: '501', taxonomyRefs: [{ taxonomy: 'category', sourceTermId: 'company' }] }),
|
||||
termEvent({ termId: 31, taxonomy: 'category', slug: 'company', name: 'Company' }),
|
||||
]
|
||||
const result = await createImportJob(db, {
|
||||
siteKey: key,
|
||||
snapshotId: wxr.id,
|
||||
reportRoot: join(work, 'artifacts', 'import'),
|
||||
events: eventsOf(events),
|
||||
pipeline: seededTermWriter(async (ctx) => {
|
||||
await seedContentEntry(ctx, { sourceType: 'wp:post', sourceId: '501', typeKey: 'news_article' })
|
||||
}),
|
||||
})
|
||||
expect(result.status).toBe('succeeded_with_issues')
|
||||
|
||||
// 两行并存：占位 31 原样；自建行整语句回退（source_id 与 name 都未被收编改动）
|
||||
const termsRows = (await db.execute(`select t.source_id, t.slug, t.name
|
||||
from terms t join taxonomies x on x.id = t.taxonomy_id
|
||||
where t.site_id = '${reg.siteId}' order by t.source_id`)) as any
|
||||
expect(plain(termsRows)).toStrictEqual([
|
||||
{ source_id: '31', slug: 'company-alt', name: 'Alt' },
|
||||
{ source_id: 'auto:company', slug: 'company', name: 'company' },
|
||||
])
|
||||
|
||||
const report = JSON.parse(
|
||||
await readFile(join(work, 'artifacts', 'import', result.jobId, 'reports', 'import.json'), 'utf8'),
|
||||
) as { counts: Record<string, number>; issues: string[] }
|
||||
expect(report.issues).toHaveLength(1)
|
||||
expect(report.issues[0]).toMatch(/收编撞唯一约束.*31/)
|
||||
|
||||
// 关联仍挂保留的自建行；自建计数不回退（收编失败，行仍是自建终态）
|
||||
const ct = (await db.execute(`select r.source_id as row_id, t.source_id as term_source_id, c.is_primary
|
||||
from content_terms c
|
||||
join source_records r on r.content_entry_id = c.content_entry_id
|
||||
join terms t on t.id = c.term_id
|
||||
where c.site_id = '${reg.siteId}'`)) as any
|
||||
expect(plain(ct)).toStrictEqual([
|
||||
{ row_id: '501', term_source_id: 'auto:company', is_primary: true },
|
||||
])
|
||||
expect(report.counts).toStrictEqual({
|
||||
termsProcessed: 2,
|
||||
autoCreatedTerms: 1,
|
||||
contentTermsCreated: 1,
|
||||
})
|
||||
} finally {
|
||||
await cleanupSiteRows(db, reg.siteId)
|
||||
}
|
||||
} finally {
|
||||
await rm(work, { recursive: true, force: true })
|
||||
}
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
Reference in New Issue
Block a user