diff --git a/ui/src/composables/use-migrate-task-runner.ts b/ui/src/composables/use-migrate-task-runner.ts index 67db5b7..a1082e3 100644 --- a/ui/src/composables/use-migrate-task-runner.ts +++ b/ui/src/composables/use-migrate-task-runner.ts @@ -39,6 +39,55 @@ export function useMigrateTaskRunner(taskGroups: Ref) { const failedTasks = computed(() => allTasks.value.filter((task) => task.status === 'failed')) const hasFailedTasks = computed(() => failedTasks.value.length > 0) + function taskKey(task: MigrateTaskItem) { + return `${task.type}:${task.id}` + } + + async function runTasksInDependencyOrder(tasks: MigrateTaskItem[]) { + const tasksToRun = tasks.filter((task) => task.status !== 'running') + const remaining = new Set(tasksToRun) + const taskByKey = new Map(allTasks.value.map((task) => [taskKey(task), task])) + + tasksToRun.forEach((task) => { + task.status = 'pending' + task.error = undefined + }) + + while (remaining.size > 0) { + const readyTasks = Array.from(remaining).filter((task) => + (task.dependsOn || []).every((dependencyKey) => { + const dependency = taskByKey.get(dependencyKey) + return !dependency || dependency.status === 'success' + }) + ) + + if (readyTasks.length === 0) { + const hasRunningDependency = Array.from(remaining).some((task) => + (task.dependsOn || []).some( + (dependencyKey) => taskByKey.get(dependencyKey)?.status === 'running' + ) + ) + if (hasRunningDependency) { + await taskQueue.drained() + continue + } + + remaining.forEach((task) => { + const unresolvedDependencies = (task.dependsOn || []).filter((dependencyKey) => { + const dependency = taskByKey.get(dependencyKey) + return dependency && dependency.status !== 'success' + }) + task.status = 'failed' + task.error = `依赖任务未成功：${unresolvedDependencies.join('、')}` + }) + break + } + + await Promise.all(readyTasks.map((task) => taskQueue.push(task))) + readyTasks.forEach((task) => remaining.delete(task)) + } + } + watch( taskGroups, (groups) => { @@ -49,10 +98,7 @@ export function useMigrateTaskRunner(taskGroups: Ref) { return } - task.status = 'pending' - task.error = undefined - importLoading.value = true - taskQueue.push(task) + void retryTask(task) } }) }) @@ -94,19 +140,7 @@ export function useMigrateTaskRunner(taskGroups: Ref) { isImportStarted.value = true beforeUnloadGuard.enable() - const tasksToRun = [...pendingTasks.value, ...failedTasks.value] - tasksToRun.forEach((task) => { - if (task.status !== 'running') { - task.status = 'pending' - task.error = undefined - } - - taskQueue.push(task).catch(() => { - // error is handled in asyncWorker - }) - }) - - await taskQueue.drained() + await runTasksInDependencyOrder([...pendingTasks.value, ...failedTasks.value]) importLoading.value = false beforeUnloadGuard.disable() @@ -128,11 +162,7 @@ export function useMigrateTaskRunner(taskGroups: Ref) { } importLoading.value = true - failedTasks.value.forEach((task) => { - task.retry() - }) - - await taskQueue.drained() + await runTasksInDependencyOrder([...failedTasks.value]) importLoading.value = false @@ -148,9 +178,8 @@ export function useMigrateTaskRunner(taskGroups: Ref) { } async function retryTask(task: MigrateTaskItem) { - task.retry() importLoading.value = true - await taskQueue.drained() + await runTasksInDependencyOrder([task]) importLoading.value = false } diff --git a/ui/src/composables/use-migrate-task.ts b/ui/src/composables/use-migrate-task.ts index 50d114c..90b6ec3 100644 --- a/ui/src/composables/use-migrate-task.ts +++ b/ui/src/composables/use-migrate-task.ts @@ -33,13 +33,15 @@ function createTaskItem( type: string, label: string, item: T, - run: () => Promise + run: () => Promise, + dependsOn?: string[] ): MigrateTaskItem { return { id, type, label, item, + ...(dependsOn?.length ? { dependsOn } : {}), status: 'pending', run, retry: () => {} @@ -244,6 +246,10 @@ export function useMigrateTask( ) ) } else { + const dependsOn = [`comment:${comment.spec.commentName}`] + if (comment.spec.quoteReply) { + dependsOn.push(`reply:${comment.spec.quoteReply}`) + } commentTasks.push( createTaskItem( comment.metadata?.name || 'unknown', @@ -253,7 +259,8 @@ export function useMigrateTask( () => coreApiClient.content.reply.createReply({ reply: comment as Reply - }) + }), + dependsOn ) ) } diff --git a/ui/src/modules/wordpress/use-wordpress-data-parser.ts b/ui/src/modules/wordpress/use-wordpress-data-parser.ts index fef04ba..f16c385 100644 --- a/ui/src/modules/wordpress/use-wordpress-data-parser.ts +++ b/ui/src/modules/wordpress/use-wordpress-data-parser.ts @@ -335,7 +335,7 @@ export function useWordPressDataParser(file: File): useWordPressDataParserReturn const createReply = ( reply: Comment, refType: 'Post' | 'SinglePage', - replyTarget: { commentName: string; quoteReply: string }, + replyTarget: { commentName: string; quoteReply?: string }, authorById: Map ): MigrateReply => { const fallbackAuthor = authorById.get(reply['wp:comment_user_id']) @@ -368,7 +368,7 @@ export function useWordPressDataParser(file: File): useWordPressDataParserReturn creationTime: new Date(reply['wp:comment_date']).toISOString(), hidden: false, commentName: replyTarget.commentName, - quoteReply: replyTarget.quoteReply + ...(replyTarget.quoteReply ? { quoteReply: replyTarget.quoteReply } : {}) }, status: {}, ownerRef: @@ -398,13 +398,13 @@ export function useWordPressDataParser(file: File): useWordPressDataParserReturn function resolveWordPressReplyTarget( comment: Comment, commentById: Map - ): { commentName: string; quoteReply: string } | undefined { + ): { commentName: string; quoteReply?: string } | undefined { if (comment['wp:comment_parent'] === 0) { return undefined } - const quoteReply = String(comment['wp:comment_parent']) - const directParent = commentById.get(quoteReply) + const directParentName = String(comment['wp:comment_parent']) + const directParent = commentById.get(directParentName) if (!directParent) { return undefined } @@ -415,7 +415,7 @@ export function useWordPressDataParser(file: File): useWordPressDataParserReturn while (rootComment['wp:comment_parent'] !== 0) { const currentId = String(rootComment['wp:comment_id']) if (visited.has(currentId)) { - break + return undefined } visited.add(currentId) @@ -427,9 +427,10 @@ export function useWordPressDataParser(file: File): useWordPressDataParserReturn rootComment = nextParent } + const commentName = String(rootComment['wp:comment_id']) return { - commentName: String(rootComment['wp:comment_id']), - quoteReply + commentName, + ...(directParentName !== commentName ? { quoteReply: directParentName } : {}) } } diff --git a/ui/src/types/index.ts b/ui/src/types/index.ts index e86f2fc..ac0482e 100644 --- a/ui/src/types/index.ts +++ b/ui/src/types/index.ts @@ -224,6 +224,7 @@ export interface MigrateTaskItem { type: string label: string item: T + dependsOn?: string[] status: MigrateTaskState error?: string run: () => Promise> diff --git a/ui/tests/use-migrate-task-runner.test.ts b/ui/tests/use-migrate-task-runner.test.ts new file mode 100644 index 0000000..ced9341 --- /dev/null +++ b/ui/tests/use-migrate-task-runner.test.ts @@ -0,0 +1,260 @@ +import { useMigrateTaskRunner } from '@/composables/use-migrate-task-runner' +import type { MigrateData, MigrateTaskItem } from '@/types' +import { Dialog } from '@halo-dev/components' +import { afterEach, beforeEach, describe, expect, it } from '@rstest/core' +import { createApp, defineComponent, ref } from 'vue' + +const originalDialogSuccess = Dialog.success +const originalDialogWarning = Dialog.warning +const mountedApps: Array<() => void> = [] + +function createTask(options: { + id: string + type: string + dependsOn?: string[] + run: () => Promise +}) { + return { + id: options.id, + type: options.type, + label: options.id, + item: {}, + status: 'pending', + dependsOn: options.dependsOn, + run: async () => { + await options.run() + return { data: {} } as any + }, + retry: () => {} + } as MigrateTaskItem +} + +async function runTasks(tasks: MigrateTaskItem[]) { + let runner: ReturnType | undefined + const app = createApp( + defineComponent({ + setup() { + runner = useMigrateTaskRunner( + ref([ + { + key: 'test', + label: 'Test', + tasks + } + ]) + ) + return () => null + } + }) + ) + app.mount(document.createElement('div')) + mountedApps.push(() => app.unmount()) + + await runner!.runImport({ data: { sourceProvider: 'halo' } as MigrateData }) + return runner! +} + +describe('useMigrateTaskRunner dependency scheduling', () => { + beforeEach(() => { + Dialog.success = (() => undefined) as typeof Dialog.success + Dialog.warning = (() => undefined) as typeof Dialog.warning + }) + + afterEach(() => { + mountedApps.splice(0).forEach((unmount) => unmount()) + Dialog.success = originalDialogSuccess + Dialog.warning = originalDialogWarning + }) + + it('waits for a parent comment while keeping unrelated tasks concurrent', async () => { + const events: string[] = [] + const reply = createTask({ + id: 'reply-1', + type: 'reply', + dependsOn: ['comment:comment-1'], + run: async () => { + events.push('reply:start') + } + }) + const comment = createTask({ + id: 'comment-1', + type: 'comment', + run: async () => { + events.push('comment:start') + await new Promise((resolve) => setTimeout(resolve, 20)) + events.push('comment:end') + } + }) + const tag = createTask({ + id: 'tag-1', + type: 'tag', + run: async () => { + events.push('tag:start') + } + }) + + await runTasks([reply, comment, tag]) + + expect(events.indexOf('tag:start')).toBeLessThan(events.indexOf('comment:end')) + expect(events.indexOf('reply:start')).toBeGreaterThan(events.indexOf('comment:end')) + }) + + it('runs an out-of-order nested reply chain one level at a time', async () => { + const events: string[] = [] + const tasks = [ + createTask({ + id: 'reply-2', + type: 'reply', + dependsOn: ['comment:comment-1', 'reply:reply-1'], + run: async () => { + events.push('reply-2') + } + }), + createTask({ + id: 'reply-1', + type: 'reply', + dependsOn: ['comment:comment-1'], + run: async () => { + events.push('reply-1') + } + }), + createTask({ + id: 'comment-1', + type: 'comment', + run: async () => { + events.push('comment-1') + } + }) + ] + + await runTasks(tasks) + + expect(events).toEqual(['comment-1', 'reply-1', 'reply-2']) + }) + + it('runs sibling replies concurrently after their parent succeeds', async () => { + let activeReplies = 0 + let maxActiveReplies = 0 + const createReply = (id: string) => + createTask({ + id, + type: 'reply', + dependsOn: ['comment:comment-1'], + run: async () => { + activeReplies++ + maxActiveReplies = Math.max(maxActiveReplies, activeReplies) + await new Promise((resolve) => setTimeout(resolve, 20)) + activeReplies-- + } + }) + const comment = createTask({ + id: 'comment-1', + type: 'comment', + run: async () => {} + }) + + await runTasks([createReply('reply-1'), createReply('reply-2'), comment]) + + expect(maxActiveReplies).toBe(2) + }) + + it('does not run a child when its parent task fails', async () => { + let childRuns = 0 + const comment = createTask({ + id: 'comment-1', + type: 'comment', + run: async () => { + throw new Error('comment failed') + } + }) + const reply = createTask({ + id: 'reply-1', + type: 'reply', + dependsOn: ['comment:comment-1'], + run: async () => { + childRuns++ + } + }) + + await runTasks([reply, comment]) + + expect(comment.status).toBe('failed') + expect(reply.status).toBe('failed') + expect(reply.error).toContain('comment:comment-1') + expect(childRuns).toBe(0) + }) + + it('retries failed parents before their blocked children', async () => { + const events: string[] = [] + let parentAttempts = 0 + const comment = createTask({ + id: 'comment-1', + type: 'comment', + run: async () => { + parentAttempts++ + events.push(`comment:${parentAttempts}`) + if (parentAttempts === 1) { + throw new Error('comment failed') + } + } + }) + const reply = createTask({ + id: 'reply-1', + type: 'reply', + dependsOn: ['comment:comment-1'], + run: async () => { + events.push('reply') + } + }) + + const runner = await runTasks([reply, comment]) + await runner.retryAll() + + expect(events).toEqual(['comment:1', 'comment:2', 'reply']) + expect(comment.status).toBe('success') + expect(reply.status).toBe('success') + }) + + it('allows dependencies that are outside the current migration data', async () => { + let runs = 0 + const reply = createTask({ + id: 'reply-1', + type: 'reply', + dependsOn: ['comment:already-imported'], + run: async () => { + runs++ + } + }) + + await runTasks([reply]) + + expect(reply.status).toBe('success') + expect(runs).toBe(1) + }) + + it('fails cyclic dependencies without running either task', async () => { + let runs = 0 + const first = createTask({ + id: 'reply-1', + type: 'reply', + dependsOn: ['reply:reply-2'], + run: async () => { + runs++ + } + }) + const second = createTask({ + id: 'reply-2', + type: 'reply', + dependsOn: ['reply:reply-1'], + run: async () => { + runs++ + } + }) + + await runTasks([first, second]) + + expect(first.status).toBe('failed') + expect(second.status).toBe('failed') + expect(runs).toBe(0) + }) +}) diff --git a/ui/tests/use-migrate-task.test.ts b/ui/tests/use-migrate-task.test.ts new file mode 100644 index 0000000..6d3d3ab --- /dev/null +++ b/ui/tests/use-migrate-task.test.ts @@ -0,0 +1,47 @@ +import { useMigrateTask } from '@/composables/use-migrate-task' +import type { MigrateData } from '@/types' +import { describe, expect, it } from '@rstest/core' + +describe('useMigrateTask', () => { + it('declares parent dependencies for direct and nested replies', () => { + const data = { + sourceProvider: 'wordpress', + comments: [ + { + kind: 'Comment', + refType: 'Post', + metadata: { name: 'comment-1' }, + spec: { owner: { kind: 'Email', name: 'a@example.com', displayName: 'A' } } + }, + { + kind: 'Reply', + refType: 'Post', + metadata: { name: 'reply-1' }, + spec: { + commentName: 'comment-1', + owner: { kind: 'Email', name: 'b@example.com', displayName: 'B' } + } + }, + { + kind: 'Reply', + refType: 'Post', + metadata: { name: 'reply-2' }, + spec: { + commentName: 'comment-1', + quoteReply: 'reply-1', + owner: { kind: 'Email', name: 'c@example.com', displayName: 'C' } + } + } + ] + } as MigrateData + + const tasks = useMigrateTask(data).flatMap((group) => group.tasks) + + expect(tasks.find((task) => task.id === 'comment-1')?.dependsOn).toBeUndefined() + expect(tasks.find((task) => task.id === 'reply-1')?.dependsOn).toEqual(['comment:comment-1']) + expect(tasks.find((task) => task.id === 'reply-2')?.dependsOn).toEqual([ + 'comment:comment-1', + 'reply:reply-1' + ]) + }) +}) diff --git a/ui/tests/wordpress-data-parser.test.ts b/ui/tests/wordpress-data-parser.test.ts index 75174d0..35e1812 100644 --- a/ui/tests/wordpress-data-parser.test.ts +++ b/ui/tests/wordpress-data-parser.test.ts @@ -260,10 +260,10 @@ describe('useWordPressDataParser', () => { expect(data.comments?.[1]).toMatchObject({ kind: 'Reply', spec: { - commentName: '501', - quoteReply: '501' + commentName: '501' } }) + expect(data.comments?.[1]?.spec.quoteReply).toBeUndefined() expect(data.tags?.[0]).toMatchObject({ metadata: { name: '5' }, spec: { displayName: 'Tips', slug: 'tips' } @@ -331,11 +331,11 @@ describe('useWordPressDataParser', () => { spec: expect.objectContaining({ raw: '😂', content: '😂', - commentName: '501', - quoteReply: '501' + commentName: '501' }) }) ]) + expect(data.comments?.[1]?.spec.quoteReply).toBeUndefined() }) it('derives attachment path from attachment url when metadata is sparse', async () => { @@ -417,10 +417,10 @@ describe('useWordPressDataParser', () => { expect(data.comments?.[1]).toMatchObject({ kind: 'Reply', spec: { - commentName: '501', - quoteReply: '501' + commentName: '501' } }) + expect(data.comments?.[1]?.spec.quoteReply).toBeUndefined() expect(data.comments?.[2]).toMatchObject({ kind: 'Reply', spec: { @@ -429,4 +429,57 @@ describe('useWordPressDataParser', () => { } }) }) + + it('promotes an orphaned reply to a comment and keeps its children attached', async () => { + const wxrWithOrphanedReply = createWordPressWxr() + .replace( + '501', + '999' + ) + .replace( + '\n ', + ` + + 503 + + + + + + + 1 + 502 + 0 + + ` + ) + + const data = await useWordPressDataParser(createWordPressWxrFile(wxrWithOrphanedReply)).parse() + + expect(data.comments?.map((item) => [item.metadata.name, item.kind])).toEqual([ + ['501', 'Comment'], + ['502', 'Comment'], + ['503', 'Reply'] + ]) + expect(data.comments?.[2]).toMatchObject({ + spec: { + commentName: '502' + } + }) + expect(data.comments?.[2]?.spec.quoteReply).toBeUndefined() + }) + + it('promotes cyclic WordPress comments instead of creating invalid reply references', async () => { + const wxrWithCyclicComments = createWordPressWxr().replace( + '0', + '502' + ) + + const data = await useWordPressDataParser(createWordPressWxrFile(wxrWithCyclicComments)).parse() + + expect(data.comments?.map((item) => [item.metadata.name, item.kind])).toEqual([ + ['501', 'Comment'], + ['502', 'Comment'] + ]) + }) })