diff --git a/apps/api/src/modules/import/term-writer.ts b/apps/api/src/modules/import/term-writer.ts index 731f17e..39e8a61 100644 --- a/apps/api/src/modules/import/term-writer.ts +++ b/apps/api/src/modules/import/term-writer.ts @@ -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:）并记 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)，不携带 文本 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 + /** 流终取回累计产出（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() - const slugIndex = new Map() - const flaggedUnknownTaxonomies = new Set() +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() + const slugIndex = new Map() + const flaggedUnknownTaxonomies = new Set() - 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 { - 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 { + 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 { - 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 - }): Promise { - 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 { - 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 { - 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 { + 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 + }): Promise { + 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 { + 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 { + 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} 落位`, + ) + // 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} 落位`, + ) + } + 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 { + 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 { + 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 不携带 + // 文本 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} 落位`, - ) - } - 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 { - 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 { - 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 不携带 - // 文本 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() - 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() + 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() } } diff --git a/tests/integration/import-terms.test.ts b/tests/integration/import-terms.test.ts index 41fd8df..3bba6fa 100644 --- a/tests/integration/import-terms.test.ts +++ b/tests/integration/import-terms.test.ts @@ -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; 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; 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 }) + } + }) + }) })