diff --git a/apps/api/src/modules/import/import.service.ts b/apps/api/src/modules/import/import.service.ts index 1eeaf9c..b55deba 100644 --- a/apps/api/src/modules/import/import.service.ts +++ b/apps/api/src/modules/import/import.service.ts @@ -34,6 +34,7 @@ import { MAPPING_VERSION, PARSER_VERSION, SANITIZER_VERSION, + type AdapterEvent, } from '@sf/adapter-wp' import { importJobs, sites, sourceConnections, sourceSnapshots } from '../../db/schema.js' import type * as schema from '../../db/schema.js' @@ -51,6 +52,7 @@ export type ImportErrorCode = | 'SNAPSHOT_NOT_FOUND' | 'NO_WXR_SNAPSHOT' | 'PIPELINE_FAILED' + | 'REPORT_WRITE_FAILED' /** 导入失败(可预期错误):CLI 以 code 区分呈现;不裹程序 bug(那些原样上抛) */ export class ImportError extends Error { @@ -83,6 +85,11 @@ export interface SnapshotRef { /** 管道执行上下文:Task 6 真实管道的全部输入(站点 locale 接缝见前置事实 8) */ export interface ImportJobContext { + /** + * 写入器自行管理事务:ctx.db 为直连(非事务代理),需要原子性的多语句写入用 + * db.transaction 自行包裹;单语句 upsert(on conflict do nothing/update)本身原子, + * 无需显式事务。 + */ db: ImportDb jobId: string siteId: string @@ -100,6 +107,13 @@ export interface ImportJobContext { * ctx.report.issues.push(...))。合并规则见 {@link ImportPipeline}——返回值优先。 */ report: ImportPipelineOutcome + /** + * 适配器事件流(runWxrAdapter 产物:row/unsupported/error/term 按文档序交错)。 + * Task 2 加法扩展:测试经 {@link CreateImportJobInput.events} 手构注入;Task 6 编排 + * 时改为从 wxr 快照构造真流(runWxrAdapter(storagePath 流))——多个写入器共享同流 + * 时的重迭代策略(每写入器一新生成器等)由 Task 6 决定,本字段只承诺"可迭代一次"。 + */ + events: AsyncIterable } /** @@ -130,6 +144,12 @@ export interface CreateImportJobInput { connectionName?: string /** 报告根(落 //reports/import.json);缺省 repoRoot/artifacts/import */ reportRoot?: string + /** + * 适配器事件流(ctx.events 的注入位):Task 2 测试手构注入;缺省空流(空管道下无消费 + * 方)。Task 6 编排时由此改为从 wxr 快照构造 runWxrAdapter 真流(前置事实 8:mapConfig + * .locale 必须传 site.default_locale)。 + */ + events?: AsyncIterable /** 管道钩子(测试/Task 6 注入);缺省空管道 */ pipeline?: ImportPipeline } @@ -344,7 +364,7 @@ export async function createImportJob( // 阶段 5:管道(本任务缺省空管道;Task 6 接真实写入器)——抛错 → failed 终态。 // 产出汇合规则(ImportPipeline JSDoc 钉死):返回值 counts 覆盖累加器同名键、issues 追加在后。 - let failed: { code: string; summary: string } | null = null + let failed: { code: ImportErrorCode; summary: string } | null = null const ctx: ImportJobContext = { db, jobId, @@ -357,6 +377,7 @@ export async function createImportJob( manualUrlListSnapshot, versions: VERSIONS, report: { counts: {}, issues: [] }, + events: input.events ?? emptyEvents(), } let outcome: ImportPipelineOutcome | null = null try { @@ -497,3 +518,7 @@ function slimRef(ref: SnapshotRef): { async function emptyPipeline(_ctx: ImportJobContext): Promise { return { counts: {}, issues: [] } } + +/** 缺省空事件流(input.events 缺席时的 ctx.events 占位;每次调用产新迭代器) */ +async function* emptyEvents(): AsyncGenerator {} + diff --git a/apps/api/src/modules/import/term-writer.ts b/apps/api/src/modules/import/term-writer.ts new file mode 100644 index 0000000..731f17e --- /dev/null +++ b/apps/api/src/modules/import/term-writer.ts @@ -0,0 +1,311 @@ +// term 物化写入器(Plan 3 Task 2,spec §5.4/§7.5、前置事实 1/9、phase2-smoke 移交项①): +// 消费 ctx.events(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 的 +// wp:term 表可整域缺席 category/post_tag,148+659 条引用只能靠引用侧自建):miss 时 +// 自动建 term(source_id=`auto:`、slug=nicename)。 +// +// 单遍流式(Task 决策):term 事件即建、row 到达时 ensure-term + 关联;内存只留已见 +// (taxonomy, slug) 索引(不囤 row)。WXR 文档序 items 先于 wp:term(mini 与真实导出同 +// 序)——row 先到时按 nicename 先自建,对应 wp:term 事件后到时**收编**:就地 UPDATE +// source_id/name 扶正为权威值(content_terms 挂的是 term id,不受收编影响)——两种事件 +// 顺序终态一致(测试钉死顺序无关性)。 +// +// name 的自建回退(如实声明):UniversalRow.taxonomyRefs 只携带 (taxonomy, +// sourceTermId=nicename),不携带 文本 value(契约冻结,升版须评审)——自建 +// term 的 name 暂用 nicename;显示名回填留待后续契约决策。 +// +// is_primary(WP 单主分类近似,钉死):WXR 不导出 Yoast 主分类标记,按 refs 文档序 +// **每 (entry, taxonomy) 首个引用**记 true 余 false("每 entry 每 taxonomy 至多一个 +// primary"由本写入器保证——库层刻意无部分唯一索引)。 +// +// 事务边界:所有 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 { 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' + +/** 自建 term 的 source_id 前缀(与 wp:term 的 term_id 字符串源身份空间隔离) */ +const AUTO_SOURCE_ID_PREFIX = 'auto:' + +/** taxonomy key → (object_type, public_base_path) 钉死常量(前置事实 9:WP 默认路径) */ +export interface TermTaxonomyDefaults { + objectType: string + publicBasePath: string +} + +export const TERM_TAXONOMY_DEFAULTS: Readonly> = Object.freeze({ + category: Object.freeze({ objectType: 'post', publicBasePath: '/category/' }), + post_tag: Object.freeze({ objectType: 'post', publicBasePath: '/tag/' }), + product_cat: Object.freeze({ objectType: 'product', publicBasePath: '/product-category/' }), + product_tag: Object.freeze({ objectType: 'product', publicBasePath: '/product-tag/' }), +}) + +/** + * 表外 taxonomy 的兜底映射(真实数据存在:nav_menu/elementor_library_type 有 wp:term + * 事件但不在常量表):object_type 'post'、public_base_path `//`——仅记录不消费 + * (路由生成不读表外 taxonomy 的路径),并逐 key 记 issue(不静默)。 + */ +function taxonomyDefaultsFor(key: string): TermTaxonomyDefaults { + if (Object.hasOwn(TERM_TAXONOMY_DEFAULTS, key)) return TERM_TAXONOMY_DEFAULTS[key]! + return { objectType: 'post', publicBasePath: `/${key}/` } +} + +/** 已见 term 的内存索引项(countedAuto:是否已计入 autoCreatedTerms——收编时回退计数) */ +interface SeenTerm { + id: string + sourceId: string + countedAuto: boolean +} + +/** + * term 物化管道写入器。产出 counts: + * - termsProcessed:处理的 wp:term 事件数; + * - autoCreatedTerms:**终态**自建数(引用侧自建且未被 wp:term 收编的行数——瞬态自建 + * 被收编即回退,不虚报"wp:term 表缺失"); + * - contentTermsCreated:新建的 (entry, term) 关联数(幂等重跑为 0)。 + * issues:无 entry 可关联的 row 分类引用、表外 taxonomy 兜底、缺 term_id 的 wp:term + * 事件、收编撞唯一约束——全部显式报告(§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() + + 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, 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 found = await db + .select({ id: terms.id }) + .from(terms) + .where( + and( + eq(terms.siteId, ctx.siteId), + eq(terms.taxonomyId, args.taxonomyId), + eq(terms.sourceId, args.sourceId), + ), + ) + .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 + } + 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 (isAuto(seen.sourceId) && realSourceId !== null && seen.sourceId !== realSourceId) { + await adoptTerm(seen, realSourceId, term.name, term.taxonomy) + } + return // 已物化(同流重复事件/重跑既有行):幂等跳过 + } + 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, 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 + } + + /** 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++ + } + } + + for await (const ev of ctx.events) { + if (ev.kind === 'term') { + await handleTermEvent(ev.term) + } else if (ev.kind === 'row') { + await handleRow(ev.row) + } + // unsupported/error 事件:无分类信息,本写入器不消费(Task 6 各写入器各取所需) + } + return { counts, issues } + } +} diff --git a/tests/integration/import-terms.test.ts b/tests/integration/import-terms.test.ts new file mode 100644 index 0000000..41fd8df --- /dev/null +++ b/tests/integration/import-terms.test.ts @@ -0,0 +1,465 @@ +// Plan 3 Task 2 失败测试(先于实现):term 物化(含自建缺失)。 +// 覆盖契约(对照计划 Task 2 与前置事实 1/9、phase2-smoke 移交项①): +// 1. mini.xml 全链:wp:term 事件 → taxonomies(key→object_type 映射 + public_base_path +// 前置事实 9 常量)+ terms(source_id=term_id 字符串)+ content_terms(entry↔term, +// is_primary 近似)。mini 文档序 items 先于 wp:term——row 先到时按 nicename 自建、 +// wp:term 后到**收编**(source_id/name 就地扶正,无 auto: 残留),终态与 term 先行序 +// 一致(顺序无关性); +// 2. 自建缺失(phase2-smoke:真实 WXR 的 wp:term 表缺 category/post_tag 整域):row 的 +// taxonomyRefs (domain, nicename) 无对应 term → 自动建(source_id=auto:、 +// slug=nicename、name=nicename——UniversalRow 不携带 文本,实现层 JSDoc 注明); +// 同 nicename 多次引用只建一条(应用层索引 + UNIQUE(site,taxonomy,source_id) 兜底); +// 3. content_terms:is_primary 每 (entry, taxonomy) 首个 true 余 false(WP 单主分类近似); +// row 无对应 content entry → term 仍物化、关联跳过 + issue(不静默丢弃); +// 4. counts:termsProcessed / autoCreatedTerms / contentTermsCreated;正常路径 issues 空; +// 5. 幂等:同流重跑(换幂等键新 job)taxonomies/terms/content_terms 均不增行。 +// 运行于 globalSetup 临时库(ADR-015);报告根用 os.tmpdir 临时目录,finally 清理。 +// content entry 由测试内 seed 模拟 Task 3 内容写入器的最小落库足迹——source_records. +// content_entry_id 是 term 关联的消费接缝(Task 6 编排时由内容写入器保证存在)。 +import { createReadStream } from 'node:fs' +import { mkdtemp, readFile, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join, resolve } from 'node:path' +import { fileURLToPath } from 'node:url' +import { describe, expect, it } from 'vitest' +import { runWxrAdapter, type AdapterEvent } from '@sf/adapter-wp' +import type { UniversalRow } from '@sf/contracts' +import type { ImportJobContext, ImportPipeline } from '../../apps/api/src/modules/import/import.service.js' +import { createImportJob } from '../../apps/api/src/modules/import/import.service.js' +import { createTermWriter } from '../../apps/api/src/modules/import/term-writer.js' +import { registerSource } from '../../apps/api/src/modules/source/register.service.js' +import { rand, withImmutableCleanup, withTestDb, type Db } from './helpers.js' + +const FIXTURES_ROOT = fileURLToPath(new URL('../fixtures', import.meta.url)) +const WXR_FIXTURE = resolve(FIXTURES_ROOT, 'wxr', 'mini.xml') + +/** 手构事件流的 AsyncIterable 包装(每次调用产新迭代器——重跑各流独立) */ +async function* eventsOf(events: AdapterEvent[]): AsyncGenerator { + yield* events +} + +/** db.execute 行对象规整:postgres.js 行对象的原型差异会让 toStrictEqual 报"无视觉差异"失败——浅拷贝到普通对象 */ +function plain(rows: any[]): any[] { + return rows.map((r) => ({ ...r })) +} + +/** 手构 row 事件(全必填字段的最小 UniversalRow) */ +function rowEvent(spec: { + sourceType?: string + sourceId: string + taxonomyRefs: Array<{ taxonomy: string; sourceTermId: string }> +}): AdapterEvent { + const row: UniversalRow = { + sourceType: spec.sourceType ?? 'wp:post', + sourceId: spec.sourceId, + status: 'publish', + locale: 'en', + taxonomyRefs: spec.taxonomyRefs, + attachmentRefs: [], + seo: {}, + knownFields: {}, + extras: {}, + rawLocation: `wxr://item/${spec.sourceId}`, + } + return { kind: 'row', row, issues: [], sanitizeDiffs: [], changed: false } +} + +/** 手构 wp:term 事件 */ +function termEvent(spec: { termId: number | null; taxonomy: string; slug: string; name: string }): AdapterEvent { + return { + kind: 'term', + term: { + rawLocation: `wxr://term/${spec.slug}`, + termId: spec.termId, + taxonomy: spec.taxonomy, + slug: spec.slug, + name: spec.name, + }, + } +} + +/** + * 模拟 Task 3 内容写入器的最小落库足迹:content_entries + source_records(含 + * content_entry_id 接缝列)。幂等(identity 命中即复用)——重跑场景同流再 seed 不增行。 + */ +async function seedContentEntry( + ctx: ImportJobContext, + spec: { sourceType: string; sourceId: string; typeKey: string; locale?: string }, +): Promise { + const locale = spec.locale ?? 'en' + const found = (await ctx.db.execute(`select content_entry_id from source_records + where site_id = '${ctx.siteId}' and source_connection_id = '${ctx.connectionId}' + and source_type = '${spec.sourceType}' and source_id = '${spec.sourceId}' and locale = '${locale}' + and content_entry_id is not null + order by created_at desc limit 1`)) as any + if (found.length > 0) return + const entry = (await ctx.db.execute(`insert into content_entries (id, site_id, type_key, lifecycle_status) + values (gen_random_uuid(), '${ctx.siteId}', '${spec.typeKey}', 'active') returning id`)) as any + await ctx.db.execute(`insert into source_records (id, site_id, source_connection_id, source_snapshot_id, import_job_id, + source_type, source_id, locale, raw_location, parse_status, content_entry_id) + values (gen_random_uuid(), '${ctx.siteId}', '${ctx.connectionId}', '${ctx.wxrSnapshot.id}', '${ctx.jobId}', + '${spec.sourceType}', '${spec.sourceId}', '${locale}', 'wxr://item/${spec.sourceId}', 'mapped', '${entry[0].id}')`) +} + +/** 测试管道:先 seed(模拟内容写入器足迹)再跑 term 写入器——Task 6 编排的雏形 */ +function seededTermWriter(seed: (ctx: ImportJobContext) => Promise): ImportPipeline { + return async (ctx) => { + await seed(ctx) + return createTermWriter()(ctx) + } +} + +/** 本文件写入的行清场:FK 反序(terms 自引用单语句全删安全;快照不可变走触发器开关) */ +async function cleanupSiteRows(db: Db, siteId: string): Promise { + await db.execute(`delete from content_terms where site_id = '${siteId}'`) + await db.execute(`delete from terms where site_id = '${siteId}'`) + await db.execute(`delete from taxonomies where site_id = '${siteId}'`) + await db.execute(`delete from source_records where site_id = '${siteId}'`) + await db.execute(`delete from content_entries where site_id = '${siteId}'`) + await db.execute(`delete from import_jobs where site_id = '${siteId}'`) + await withImmutableCleanup(db, 'source_snapshots', 'source_snapshots_immutable', `site_id = '${siteId}'`) + await db.execute(`delete from source_connections where site_id = '${siteId}'`) + await db.execute(`delete from sites where id = '${siteId}'`) +} + +/** mini 三内容行的 seed(前置事实 11 的 typeKey 映射:wp:post→news_article、wp:page→basic_page、wp:product→product) */ +async function seedMiniEntries(ctx: ImportJobContext): Promise { + await seedContentEntry(ctx, { sourceType: 'wp:post', sourceId: '101', typeKey: 'news_article' }) + await seedContentEntry(ctx, { sourceType: 'wp:product', sourceId: '111', typeKey: 'product' }) + await seedContentEntry(ctx, { sourceType: 'wp:page', sourceId: '121', typeKey: 'basic_page' }) +} + +describe('term 物化(含自建缺失)', () => { + it('mini.xml 全链:taxonomies 常量映射 + terms(source_id=term_id) + content_terms(is_primary) + 收编(row 先到 wp:term 后到)+ counts/issues', async () => { + await withTestDb(async (db) => { + const work = await mkdtemp(join(tmpdir(), 'sf-terms-')) + 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 result = await createImportJob(db, { + siteKey: key, + snapshotId: wxr.id, + reportRoot: join(work, 'artifacts', 'import'), + events: runWxrAdapter(createReadStream(WXR_FIXTURE), { mapConfig: { locale: 'en' } }), + pipeline: seededTermWriter(seedMiniEntries), + }) + expect(result.reused).toBe(false) + expect(result.status).toBe('succeeded') + + // taxonomies:key→object_type 映射 + 前置事实 9 常量(public_base_path 仅记录) + const tax = (await db.execute(`select key, object_type, public_base_path from taxonomies + where site_id = '${reg.siteId}' order by key`)) as any + expect(plain(tax)).toStrictEqual([ + { key: 'category', object_type: 'post', public_base_path: '/category/' }, + { key: 'product_cat', object_type: 'product', public_base_path: '/product-category/' }, + ]) + + // terms:source_id=term_id 字符串;mini 文档序 items 先于 wp:term——post 101 的 + // category/company 引用先到(自建 auto:company),wp:term 31 后到收编扶正 + const termsRows = (await db.execute(`select t.source_id, t.slug, t.name, x.key as taxonomy + from terms t join taxonomies x on x.id = t.taxonomy_id + where t.site_id = '${reg.siteId}' order by x.key`)) as any + expect(plain(termsRows)).toStrictEqual([ + { source_id: '31', slug: 'company', name: 'Company', taxonomy: 'category' }, + { source_id: '32', slug: 'pipe-fittings', name: 'Pipe Fittings', taxonomy: 'product_cat' }, + ]) + const autoLeft = (await db.execute(`select count(*)::int as n from terms + where site_id = '${reg.siteId}' and source_id like 'auto:%'`)) as any + expect(autoLeft[0].n).toBe(0) + + // content_terms:post 101 ↔ category/company(单引用即首引用,is_primary=true) + const ct = (await db.execute(`select content_entry_id, term_id, is_primary from content_terms + where site_id = '${reg.siteId}'`)) as any + expect(ct).toHaveLength(1) + const entry101 = (await db.execute(`select content_entry_id from source_records + where site_id = '${reg.siteId}' and source_type = 'wp:post' and source_id = '101'`)) as any + const term31 = (await db.execute(`select id from terms + where site_id = '${reg.siteId}' and source_id = '31'`)) as any + expect(ct[0].content_entry_id).toBe(entry101[0].content_entry_id) + expect(ct[0].term_id).toBe(term31[0].id) + expect(ct[0].is_primary).toBe(true) + + // 报告:counts 三键(自建终态 0——瞬态自建已被收编)+ issues 空(正常路径) + const report = JSON.parse( + await readFile(join(work, 'artifacts', 'import', result.jobId, 'reports', 'import.json'), 'utf8'), + ) as { status: string; counts: Record; issues: string[] } + expect(report.status).toBe('succeeded') + expect(report.counts).toStrictEqual({ + termsProcessed: 2, + autoCreatedTerms: 0, + contentTermsCreated: 1, + }) + expect(report.issues).toStrictEqual([]) + } finally { + await cleanupSiteRows(db, reg.siteId) + } + } finally { + await rm(work, { recursive: true, force: true }) + } + }) + }) + + it('自建缺失 term(手构缺表场景):auto: 前缀 + 同 nicename 单条 + is_primary 规则 + 无 entry 跳过关联 + 未知 taxonomy 兜底', async () => { + await withTestDb(async (db) => { + const work = await mkdtemp(join(tmpdir(), 'sf-terms-')) + 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')! + // 缺表场景:category/post_tag 无任何 wp:term 事件(照真实 hesupipe WXR); + // nav_menu 是表外 taxonomy(wp:term 有、常量表无——兜底映射钉行为) + const events: AdapterEvent[] = [ + rowEvent({ sourceId: '501', taxonomyRefs: [{ taxonomy: 'category', sourceTermId: 'company-news' }] }), + rowEvent({ sourceId: '502', taxonomyRefs: [{ taxonomy: 'category', sourceTermId: 'company-news' }] }), + rowEvent({ + sourceId: '503', + taxonomyRefs: [ + { taxonomy: 'category', sourceTermId: 'news-a' }, + { taxonomy: 'category', sourceTermId: 'news-b' }, + ], + }), + rowEvent({ + sourceType: 'wp:product', + sourceId: '504', + taxonomyRefs: [ + { taxonomy: 'product_cat', sourceTermId: 'pc-1' }, + { taxonomy: 'product_tag', sourceTermId: 'pt-1' }, + ], + }), + // 505 故意不 seed entry:term 仍物化、关联跳过 + issue + rowEvent({ sourceId: '505', taxonomyRefs: [{ taxonomy: 'post_tag', sourceTermId: 'orphan-tag' }] }), + termEvent({ termId: 77, taxonomy: 'nav_menu', slug: 'mainmenu', name: 'Main Menu' }), + ] + 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' }) + await seedContentEntry(ctx, { sourceType: 'wp:post', sourceId: '502', typeKey: 'news_article' }) + await seedContentEntry(ctx, { sourceType: 'wp:post', sourceId: '503', typeKey: 'news_article' }) + await seedContentEntry(ctx, { sourceType: 'wp:product', sourceId: '504', typeKey: 'product' }) + }), + }) + expect(result.status).toBe('succeeded_with_issues') + + // 自建 terms:source_id=auto:、slug=nicename、name=nicename; + // company-news 两处引用只建一条(应用层索引;UNIQUE(site,taxonomy,source_id) 兜底) + const autoTerms = (await db.execute(`select t.source_id, t.slug, t.name, x.key as taxonomy + from terms t join taxonomies x on x.id = t.taxonomy_id + where t.site_id = '${reg.siteId}' and t.source_id like 'auto:%' order by t.source_id`)) as any + expect(plain(autoTerms)).toStrictEqual([ + { source_id: 'auto:company-news', slug: 'company-news', name: 'company-news', taxonomy: 'category' }, + { source_id: 'auto:news-a', slug: 'news-a', name: 'news-a', taxonomy: 'category' }, + { source_id: 'auto:news-b', slug: 'news-b', name: 'news-b', taxonomy: 'category' }, + { source_id: 'auto:orphan-tag', slug: 'orphan-tag', name: 'orphan-tag', taxonomy: 'post_tag' }, + { source_id: 'auto:pc-1', slug: 'pc-1', name: 'pc-1', taxonomy: 'product_cat' }, + { source_id: 'auto:pt-1', slug: 'pt-1', name: 'pt-1', taxonomy: 'product_tag' }, + ]) + + // 未知 taxonomy 兜底(非静默):nav_menu → object_type 'post'、/nav_menu/ + issue + const navTax = (await db.execute(`select object_type, public_base_path from taxonomies + where site_id = '${reg.siteId}' and key = 'nav_menu'`)) as any + expect(plain(navTax)).toStrictEqual([{ object_type: 'post', public_base_path: '/nav_menu/' }]) + const navTerm = (await db.execute(`select t.source_id, t.name from terms t + join taxonomies x on x.id = t.taxonomy_id + where t.site_id = '${reg.siteId}' and x.key = 'nav_menu'`)) as any + expect(plain(navTerm)).toStrictEqual([{ source_id: '77', name: 'Main Menu' }]) + + // content_terms:is_primary 每 (entry, taxonomy) 首个 true 余 false; + // 505 无 entry → 不关联(term 已物化) + const ct = (await db.execute(`select r.source_id as row_id, x.key as taxonomy, 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 + join taxonomies x on x.id = t.taxonomy_id + where c.site_id = '${reg.siteId}' order by r.source_id, t.slug`)) as any + expect(plain(ct)).toStrictEqual([ + { row_id: '501', taxonomy: 'category', slug: 'company-news', is_primary: true }, + { row_id: '502', taxonomy: 'category', slug: 'company-news', is_primary: true }, + { row_id: '503', taxonomy: 'category', slug: 'news-a', is_primary: true }, + { row_id: '503', taxonomy: 'category', slug: 'news-b', is_primary: false }, + { row_id: '504', taxonomy: 'product_cat', slug: 'pc-1', is_primary: true }, + { row_id: '504', taxonomy: 'product_tag', slug: 'pt-1', is_primary: true }, + ]) + + // 报告:自建 6(orphan-tag 计入——物化不依赖 entry);两条 issue(505 无 entry + nav_menu 兜底) + const report = JSON.parse( + await readFile(join(work, 'artifacts', 'import', result.jobId, 'reports', 'import.json'), 'utf8'), + ) as { status: string; counts: Record; issues: string[] } + expect(report.counts).toStrictEqual({ + termsProcessed: 1, + autoCreatedTerms: 6, + contentTermsCreated: 6, + }) + expect(report.issues).toHaveLength(2) + expect(report.issues.some((i) => i.includes('505') && i.includes('content entry'))).toBe(true) + expect(report.issues.some((i) => i.includes('nav_menu'))).toBe(true) + } finally { + await cleanupSiteRows(db, reg.siteId) + } + } finally { + await rm(work, { recursive: true, force: true }) + } + }) + }) + + it('顺序无关性(term 先行序):wp:term 先到 → row 后到直连既有 term,零自建、终态与 row 先行序一致', async () => { + await withTestDb(async (db) => { + const work = await mkdtemp(join(tmpdir(), 'sf-terms-')) + 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: 88, taxonomy: 'product_tag', slug: 'special', name: 'Special' }), + rowEvent({ sourceId: '101', taxonomyRefs: [{ taxonomy: 'category', sourceTermId: 'company' }] }), + rowEvent({ + sourceType: 'wp:product', + sourceId: '111', + taxonomyRefs: [{ taxonomy: 'product_tag', sourceTermId: 'special' }], + }), + ] + 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: '101', typeKey: 'news_article' }) + await seedContentEntry(ctx, { sourceType: 'wp:product', sourceId: '111', typeKey: 'product' }) + }), + }) + expect(result.status).toBe('succeeded') + + // 与 mini 序(row 先到 + 收编)同一终态:source_id=term_id、name=wp:term 权威名 + const termsRows = (await db.execute(`select t.source_id, t.slug, t.name, x.key as taxonomy + from terms t join taxonomies x on x.id = t.taxonomy_id + where t.site_id = '${reg.siteId}' order by x.key`)) as any + expect(plain(termsRows)).toStrictEqual([ + { source_id: '31', slug: 'company', name: 'Company', taxonomy: 'category' }, + { source_id: '88', slug: 'special', name: 'Special', taxonomy: 'product_tag' }, + ]) + const taxRows = (await db.execute(`select key, object_type, public_base_path from taxonomies + where site_id = '${reg.siteId}' order by key`)) as any + expect(plain(taxRows)).toStrictEqual([ + { key: 'category', object_type: 'post', public_base_path: '/category/' }, + { key: 'product_tag', object_type: 'product', public_base_path: '/product-tag/' }, + ]) + + // term 先行序全程零自建;两条关联均挂真实 term + const report = JSON.parse( + await readFile(join(work, 'artifacts', 'import', result.jobId, 'reports', 'import.json'), 'utf8'), + ) as { status: string; counts: Record; issues: string[] } + expect(report.counts).toStrictEqual({ + termsProcessed: 2, + autoCreatedTerms: 0, + contentTermsCreated: 2, + }) + expect(report.issues).toStrictEqual([]) + 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}' order by r.source_id`)) as any + expect(plain(ct)).toStrictEqual([ + { row_id: '101', term_source_id: '31', is_primary: true }, + { row_id: '111', term_source_id: '88', is_primary: true }, + ]) + } finally { + await cleanupSiteRows(db, reg.siteId) + } + } finally { + await rm(work, { recursive: true, force: true }) + } + }) + }) + + it('幂等重跑(换幂等键新 job 同流):taxonomies/terms/content_terms 均不增行、关联零新增', async () => { + await withTestDb(async (db) => { + const work = await mkdtemp(join(tmpdir(), 'sf-terms-')) + 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 base = { + siteKey: key, + snapshotId: wxr.id, + reportRoot: join(work, 'artifacts', 'import'), + pipeline: seededTermWriter(seedMiniEntries), + } + const first = await createImportJob(db, { + ...base, + events: runWxrAdapter(createReadStream(WXR_FIXTURE), { mapConfig: { locale: 'en' } }), + }) + expect(first.status).toBe('succeeded') + const before = (await db.execute(`select + (select count(*)::int from taxonomies where site_id = '${reg.siteId}') as taxonomies, + (select count(*)::int from terms where site_id = '${reg.siteId}') as terms, + (select count(*)::int from content_terms where site_id = '${reg.siteId}') as content_terms`)) as any + expect(plain(before)[0]).toStrictEqual({ taxonomies: 2, terms: 2, content_terms: 1 }) + + // 同流重跑:显式幂等键(默认键会复用既有 job 不重跑——那是 Task 1 语义,此处钉写入器幂等) + const again = await createImportJob(db, { + ...base, + idempotencyKey: `sfterm-rerun-${rand()}`, + events: runWxrAdapter(createReadStream(WXR_FIXTURE), { mapConfig: { locale: 'en' } }), + }) + expect(again.reused).toBe(false) + expect(again.status).toBe('succeeded') + const after = (await db.execute(`select + (select count(*)::int from taxonomies where site_id = '${reg.siteId}') as taxonomies, + (select count(*)::int from terms where site_id = '${reg.siteId}') as terms, + (select count(*)::int from content_terms where site_id = '${reg.siteId}') as content_terms`)) as any + expect(plain(after)[0]).toStrictEqual({ taxonomies: 2, terms: 2, content_terms: 1 }) + + // 重跑计数:事件仍处理(termsProcessed=2),但物化/关联零新增 + const report = JSON.parse( + await readFile(join(work, 'artifacts', 'import', again.jobId, 'reports', 'import.json'), 'utf8'), + ) as { counts: Record; issues: string[] } + expect(report.counts).toStrictEqual({ + termsProcessed: 2, + autoCreatedTerms: 0, + contentTermsCreated: 0, + }) + expect(report.issues).toStrictEqual([]) + } finally { + await cleanupSiteRows(db, reg.siteId) + } + } finally { + await rm(work, { recursive: true, force: true }) + } + }) + }) +})