게시물 가져오기 작업 센터 추가
This commit is contained in:
22
server/plugins/post-import-worker.js
Normal file
22
server/plugins/post-import-worker.js
Normal file
@@ -0,0 +1,22 @@
|
||||
import { Cron } from 'croner'
|
||||
import { queuePendingPostImportJobs } from '../repositories/post-import-repository'
|
||||
|
||||
const queuePostImportWork = async () => {
|
||||
try {
|
||||
await queuePendingPostImportJobs()
|
||||
} catch (error) {
|
||||
console.error('[post-import] 작업 복구 확인 실패', error)
|
||||
}
|
||||
}
|
||||
|
||||
export default defineNitroPlugin(() => {
|
||||
if (process.env.NUXT_POST_IMPORT_WORKER === '0' || process.env.PLAYWRIGHT_TEST === '1') {
|
||||
return
|
||||
}
|
||||
|
||||
queuePostImportWork()
|
||||
|
||||
new Cron('*/15 * * * * *', {
|
||||
protect: true
|
||||
}, queuePostImportWork)
|
||||
})
|
||||
@@ -1,5 +1,5 @@
|
||||
import { createReadStream, createWriteStream } from 'node:fs'
|
||||
import { createHash } from 'node:crypto'
|
||||
import { createHash, randomUUID } from 'node:crypto'
|
||||
import { mkdir, rm, stat } from 'node:fs/promises'
|
||||
import { basename, extname, join } from 'node:path'
|
||||
import { pipeline } from 'node:stream/promises'
|
||||
@@ -18,6 +18,21 @@ const MAX_IMPORT_COMPRESSION_RATIO = 250
|
||||
const MAX_IMPORT_MARKDOWN_BYTES = 5 * 1024 * 1024
|
||||
const MAX_IMPORT_TOTAL_MARKDOWN_BYTES = 50 * 1024 * 1024
|
||||
const MARKDOWN_EXTENSION_PATTERN = /\.md$/i
|
||||
const IMPORT_JOB_ROOT_NAME = '_import-jobs'
|
||||
const IMPORT_STATUS = {
|
||||
QUEUED: 'queued',
|
||||
PROCESSING: 'processing',
|
||||
READY: 'ready',
|
||||
FAILED: 'failed'
|
||||
}
|
||||
const IMPORT_ITEM_STATUS = {
|
||||
PENDING: 'pending',
|
||||
PROCESSING: 'processing',
|
||||
READY: 'ready',
|
||||
SKIPPED: 'skipped',
|
||||
FAILED: 'failed'
|
||||
}
|
||||
const runningPostImportJobIds = new Set()
|
||||
|
||||
/**
|
||||
* 파일명에 안전한 문자열로 정리한다.
|
||||
@@ -32,6 +47,69 @@ const sanitizeFilenameSegment = (value) => String(value || '')
|
||||
.replace(/^-|-$/g, '')
|
||||
.slice(0, 80) || 'asset'
|
||||
|
||||
/**
|
||||
* Import 작업 ZIP 저장 경로를 만든다.
|
||||
* @param {string} originalName - 원본 파일명
|
||||
* @returns {Promise<string>} 저장 경로
|
||||
*/
|
||||
export const createPostImportJobZipPath = async (originalName) => {
|
||||
const directoryPath = join(getUploadRoot(), IMPORT_JOB_ROOT_NAME)
|
||||
const fileName = `${Date.now()}-${randomUUID()}-${sanitizeFilenameSegment(originalName || 'import.zip')}`
|
||||
|
||||
await mkdir(directoryPath, { recursive: true })
|
||||
|
||||
return join(directoryPath, fileName.toLowerCase().endsWith('.zip') ? fileName : `${fileName}.zip`)
|
||||
}
|
||||
|
||||
/**
|
||||
* Import 작업 행을 응답 구조로 변환한다.
|
||||
* @param {Object} row - DB 행
|
||||
* @returns {Object} Import 작업
|
||||
*/
|
||||
const mapPostImportJobRow = (row) => ({
|
||||
id: row.id,
|
||||
requestedBy: row.requested_by || null,
|
||||
requestedEmail: row.requested_email || '',
|
||||
originalFileName: row.original_file_name || '',
|
||||
status: row.status,
|
||||
totalCount: Number(row.total_count || 0),
|
||||
processedCount: Number(row.processed_count || 0),
|
||||
importedCount: Number(row.imported_count || 0),
|
||||
skippedCount: Number(row.skipped_count || 0),
|
||||
failedCount: Number(row.failed_count || 0),
|
||||
assetCount: Number(row.asset_count || 0),
|
||||
reusedAssetCount: Number(row.reused_asset_count || 0),
|
||||
warningCount: Number(row.warning_count || 0),
|
||||
message: row.message || '',
|
||||
errorDetail: row.error_detail || '',
|
||||
createdAt: row.created_at.toISOString(),
|
||||
updatedAt: row.updated_at.toISOString(),
|
||||
startedAt: row.started_at ? row.started_at.toISOString() : null,
|
||||
completedAt: row.completed_at ? row.completed_at.toISOString() : null,
|
||||
items: row.items || []
|
||||
})
|
||||
|
||||
/**
|
||||
* Import 게시물 행을 응답 구조로 변환한다.
|
||||
* @param {Object} row - DB 행
|
||||
* @returns {Object} Import 게시물
|
||||
*/
|
||||
const mapPostImportItemRow = (row) => ({
|
||||
id: row.id,
|
||||
jobId: row.job_id,
|
||||
itemIndex: Number(row.item_index || 0),
|
||||
title: row.title || '',
|
||||
slug: row.slug || '',
|
||||
sourcePath: row.source_path || '',
|
||||
status: row.status,
|
||||
progress: Number(row.progress || 0),
|
||||
message: row.message || '',
|
||||
postId: row.post_id || null,
|
||||
createdAt: row.created_at.toISOString(),
|
||||
updatedAt: row.updated_at.toISOString(),
|
||||
completedAt: row.completed_at ? row.completed_at.toISOString() : null
|
||||
})
|
||||
|
||||
/**
|
||||
* 읽기 스트림의 SHA-256 해시를 만든다.
|
||||
* @param {NodeJS.ReadableStream} stream - 읽기 스트림
|
||||
@@ -649,10 +727,21 @@ const collectMarkdownPosts = async (entries) => {
|
||||
|
||||
/**
|
||||
* Export ZIP을 게시물로 가져온다.
|
||||
* @param {{ zipFilePath: string, authorId: string }} input - Import 입력
|
||||
* @returns {Promise<{ importedCount: number, skippedCount: number, assetCount: number, reusedAssetCount: number, warningCount: number, warnings: Array<string>, posts: Array<Object>, skippedPosts: Array<Object> }>} Import 결과
|
||||
* @param {Object} input - Import 입력
|
||||
* @param {string} input.zipFilePath - ZIP 파일 경로
|
||||
* @param {string} input.authorId - 작성자 ID
|
||||
* @param {(posts: Array<Object>) => Promise<void>|void} [input.onPlanned] - 게시물 계획 콜백
|
||||
* @param {(item: Object) => Promise<void>|void} [input.onItemStart] - 게시물 처리 시작 콜백
|
||||
* @param {(item: Object) => Promise<void>|void} [input.onItemComplete] - 게시물 처리 완료 콜백
|
||||
* @returns {Promise<{ importedCount: number, skippedCount: number, failedCount: number, assetCount: number, reusedAssetCount: number, warningCount: number, warnings: Array<string>, posts: Array<Object>, skippedPosts: Array<Object>, failedPosts: Array<Object> }>} Import 결과
|
||||
*/
|
||||
export const importPostsFromExportZip = async ({ zipFilePath, authorId }) => {
|
||||
export const importPostsFromExportZip = async ({
|
||||
zipFilePath,
|
||||
authorId,
|
||||
onPlanned = null,
|
||||
onItemStart = null,
|
||||
onItemComplete = null
|
||||
}) => {
|
||||
const entries = await openZipFileEntries(zipFilePath, {
|
||||
maxEntryCount: MAX_IMPORT_ZIP_ENTRY_COUNT,
|
||||
maxEntryBytes: MAX_IMPORT_ENTRY_BYTES,
|
||||
@@ -666,19 +755,41 @@ export const importPostsFromExportZip = async ({ zipFilePath, authorId }) => {
|
||||
throw new Error('IMPORT_MARKDOWN_NOT_FOUND')
|
||||
}
|
||||
|
||||
await onPlanned?.(markdownPosts.map((markdownPost, index) => {
|
||||
const title = String(markdownPost.frontmatter.title || basename(markdownPost.path).replace(MARKDOWN_EXTENSION_PATTERN, '') || 'Imported Post').trim()
|
||||
const slug = sanitizeSlug(markdownPost.frontmatter.slug || title)
|
||||
|
||||
return {
|
||||
itemIndex: index + 1,
|
||||
title,
|
||||
slug,
|
||||
sourcePath: markdownPost.path
|
||||
}
|
||||
}))
|
||||
|
||||
const reservedSlugs = new Set()
|
||||
const savedAssetUrls = new Map()
|
||||
const importedPosts = []
|
||||
const skippedPosts = []
|
||||
const failedPosts = []
|
||||
const warnings = []
|
||||
let importedAssetCount = 0
|
||||
let reusedAssetCount = 0
|
||||
|
||||
for (const markdownPost of markdownPosts) {
|
||||
for (let index = 0; index < markdownPosts.length; index += 1) {
|
||||
const markdownPost = markdownPosts[index]
|
||||
const itemIndex = index + 1
|
||||
const { frontmatter, content, postFolder } = markdownPost
|
||||
const title = String(frontmatter.title || basename(markdownPost.path).replace(MARKDOWN_EXTENSION_PATTERN, '') || 'Imported Post').trim()
|
||||
const { slug, duplicateInZip } = reserveImportPostSlug(frontmatter.slug || title, reservedSlugs)
|
||||
|
||||
await onItemStart?.({
|
||||
itemIndex,
|
||||
title,
|
||||
slug,
|
||||
sourcePath: markdownPost.path
|
||||
})
|
||||
|
||||
if (duplicateInZip) {
|
||||
skippedPosts.push({
|
||||
title,
|
||||
@@ -686,6 +797,12 @@ export const importPostsFromExportZip = async ({ zipFilePath, authorId }) => {
|
||||
reason: 'duplicate-in-zip'
|
||||
})
|
||||
warnings.push(`${markdownPost.path}: ZIP 안에 같은 slug가 있어 건너뛰었습니다. (${slug})`)
|
||||
await onItemComplete?.({
|
||||
itemIndex,
|
||||
status: IMPORT_ITEM_STATUS.SKIPPED,
|
||||
progress: 100,
|
||||
message: 'ZIP 안에 같은 slug가 있어 건너뜀'
|
||||
})
|
||||
continue
|
||||
}
|
||||
|
||||
@@ -695,6 +812,12 @@ export const importPostsFromExportZip = async ({ zipFilePath, authorId }) => {
|
||||
slug,
|
||||
reason: 'existing-slug'
|
||||
})
|
||||
await onItemComplete?.({
|
||||
itemIndex,
|
||||
status: IMPORT_ITEM_STATUS.SKIPPED,
|
||||
progress: 100,
|
||||
message: '이미 같은 slug의 게시물이 있어 건너뜀'
|
||||
})
|
||||
continue
|
||||
}
|
||||
|
||||
@@ -716,6 +839,21 @@ export const importPostsFromExportZip = async ({ zipFilePath, authorId }) => {
|
||||
warnings.push(`${markdownPost.path}: ZIP 내부 자산을 찾지 못했습니다. (${assetPath})`)
|
||||
})
|
||||
|
||||
if (missingAssets.length > 0) {
|
||||
failedPosts.push({
|
||||
title,
|
||||
slug,
|
||||
reason: 'missing-assets'
|
||||
})
|
||||
await onItemComplete?.({
|
||||
itemIndex,
|
||||
status: IMPORT_ITEM_STATUS.FAILED,
|
||||
progress: 100,
|
||||
message: `ZIP 내부 자산 ${missingAssets.length}개 누락`
|
||||
})
|
||||
continue
|
||||
}
|
||||
|
||||
const featuredImage = resolveImportedImageUrl(frontmatter.featured_image, replacements)
|
||||
const ogImage = resolveImportedImageUrl(frontmatter.og_image, replacements)
|
||||
const status = normalizePostStatus(frontmatter.status)
|
||||
@@ -750,18 +888,38 @@ export const importPostsFromExportZip = async ({ zipFilePath, authorId }) => {
|
||||
slug,
|
||||
reason: 'existing-slug'
|
||||
})
|
||||
await onItemComplete?.({
|
||||
itemIndex,
|
||||
status: IMPORT_ITEM_STATUS.SKIPPED,
|
||||
progress: 100,
|
||||
message: '이미 같은 slug의 게시물이 있어 건너뜀'
|
||||
})
|
||||
continue
|
||||
}
|
||||
|
||||
await onItemComplete?.({
|
||||
itemIndex,
|
||||
status: IMPORT_ITEM_STATUS.FAILED,
|
||||
progress: 100,
|
||||
message: error instanceof Error ? error.message : '게시물 저장 실패'
|
||||
})
|
||||
throw error
|
||||
}
|
||||
|
||||
importedPosts.push(post)
|
||||
await onItemComplete?.({
|
||||
itemIndex,
|
||||
status: IMPORT_ITEM_STATUS.READY,
|
||||
progress: 100,
|
||||
message: '완료',
|
||||
postId: post.id
|
||||
})
|
||||
}
|
||||
|
||||
return {
|
||||
importedCount: importedPosts.length,
|
||||
skippedCount: skippedPosts.length,
|
||||
failedCount: failedPosts.length,
|
||||
assetCount: importedAssetCount,
|
||||
reusedAssetCount,
|
||||
warningCount: warnings.length,
|
||||
@@ -771,6 +929,527 @@ export const importPostsFromExportZip = async ({ zipFilePath, authorId }) => {
|
||||
title: post.title,
|
||||
slug: post.slug
|
||||
})),
|
||||
skippedPosts: skippedPosts.slice(0, 20)
|
||||
skippedPosts: skippedPosts.slice(0, 20),
|
||||
failedPosts: failedPosts.slice(0, 20)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Import 작업 단일 행을 조회한다.
|
||||
* @param {string} jobId - Import 작업 ID
|
||||
* @returns {Promise<Object|null>} DB 행
|
||||
*/
|
||||
const getPostImportJobRowById = async (jobId) => {
|
||||
const sql = getPostgresClient()
|
||||
|
||||
if (!sql) {
|
||||
return null
|
||||
}
|
||||
|
||||
const rows = await sql`
|
||||
SELECT *
|
||||
FROM post_import_jobs
|
||||
WHERE id = ${jobId}
|
||||
LIMIT 1
|
||||
`
|
||||
|
||||
return rows[0] || null
|
||||
}
|
||||
|
||||
/**
|
||||
* Import 작업 목록을 조회한다.
|
||||
* @returns {Promise<Array>} Import 작업 목록
|
||||
*/
|
||||
export const listPostImportJobs = async () => {
|
||||
const sql = getPostgresClient()
|
||||
|
||||
if (!sql) {
|
||||
return []
|
||||
}
|
||||
|
||||
const jobRows = await sql`
|
||||
SELECT *
|
||||
FROM post_import_jobs
|
||||
ORDER BY created_at DESC
|
||||
LIMIT 20
|
||||
`
|
||||
|
||||
if (jobRows.length === 0) {
|
||||
return []
|
||||
}
|
||||
|
||||
const jobIds = jobRows.map((row) => row.id)
|
||||
const itemRows = await sql`
|
||||
SELECT *
|
||||
FROM post_import_items
|
||||
WHERE job_id = ANY(${jobIds})
|
||||
ORDER BY job_id, item_index ASC
|
||||
`
|
||||
const itemsByJobId = itemRows.reduce((groups, row) => {
|
||||
const key = row.job_id
|
||||
groups[key] = groups[key] || []
|
||||
groups[key].push(mapPostImportItemRow(row))
|
||||
return groups
|
||||
}, {})
|
||||
|
||||
return jobRows.map((row) => mapPostImportJobRow({
|
||||
...row,
|
||||
items: itemsByJobId[row.id] || []
|
||||
}))
|
||||
}
|
||||
|
||||
/**
|
||||
* Import 작업을 조회한다.
|
||||
* @param {string} jobId - Import 작업 ID
|
||||
* @returns {Promise<Object|null>} Import 작업
|
||||
*/
|
||||
export const getPostImportJobById = async (jobId) => {
|
||||
const sql = getPostgresClient()
|
||||
|
||||
if (!sql) {
|
||||
return null
|
||||
}
|
||||
|
||||
const jobRows = await sql`
|
||||
SELECT *
|
||||
FROM post_import_jobs
|
||||
WHERE id = ${jobId}
|
||||
LIMIT 1
|
||||
`
|
||||
|
||||
if (jobRows.length === 0) {
|
||||
return null
|
||||
}
|
||||
|
||||
const itemRows = await sql`
|
||||
SELECT *
|
||||
FROM post_import_items
|
||||
WHERE job_id = ${jobId}
|
||||
ORDER BY item_index ASC
|
||||
`
|
||||
|
||||
return mapPostImportJobRow({
|
||||
...jobRows[0],
|
||||
items: itemRows.map(mapPostImportItemRow)
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Import 작업을 생성한다.
|
||||
* @param {Object} input - 생성 입력
|
||||
* @param {string} input.requestedBy - 요청 관리자 ID
|
||||
* @param {string} input.requestedEmail - 요청 관리자 이메일
|
||||
* @param {string} input.originalFileName - 원본 파일명
|
||||
* @param {string} input.zipFilePath - ZIP 파일 경로
|
||||
* @returns {Promise<Object>} 생성된 Import 작업
|
||||
*/
|
||||
export const createPostImportJob = async (input) => {
|
||||
const sql = getPostgresClient()
|
||||
|
||||
if (!sql) {
|
||||
throw new Error('DATABASE_REQUIRED')
|
||||
}
|
||||
|
||||
const rows = await sql`
|
||||
INSERT INTO post_import_jobs (
|
||||
requested_by,
|
||||
requested_email,
|
||||
original_file_name,
|
||||
zip_file_path,
|
||||
status,
|
||||
message
|
||||
)
|
||||
VALUES (
|
||||
${input.requestedBy},
|
||||
${input.requestedEmail || ''},
|
||||
${input.originalFileName || ''},
|
||||
${input.zipFilePath},
|
||||
${IMPORT_STATUS.QUEUED},
|
||||
${'Import 작업이 대기열에 등록되었습니다.'}
|
||||
)
|
||||
RETURNING *
|
||||
`
|
||||
|
||||
return mapPostImportJobRow({
|
||||
...rows[0],
|
||||
items: []
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* 실행하거나 재개할 Import 작업 ID 목록을 조회한다.
|
||||
* @returns {Promise<Array<string>>} 실행 가능한 작업 ID 목록
|
||||
*/
|
||||
export const listQueuedPostImportJobIds = async () => {
|
||||
const sql = getPostgresClient()
|
||||
|
||||
if (!sql) {
|
||||
return []
|
||||
}
|
||||
|
||||
const rows = await sql`
|
||||
SELECT id
|
||||
FROM post_import_jobs
|
||||
WHERE status IN (${IMPORT_STATUS.QUEUED}, ${IMPORT_STATUS.PROCESSING})
|
||||
ORDER BY created_at ASC
|
||||
LIMIT 1
|
||||
`
|
||||
|
||||
return rows.map((row) => row.id)
|
||||
}
|
||||
|
||||
/**
|
||||
* Import 작업 게시물 행을 계획 상태로 저장한다.
|
||||
* @param {import('postgres').Sql} sql - SQL 클라이언트
|
||||
* @param {string} jobId - Import 작업 ID
|
||||
* @param {Array<Object>} items - 게시물 계획
|
||||
* @returns {Promise<void>}
|
||||
*/
|
||||
const upsertPostImportItems = async (sql, jobId, items) => {
|
||||
for (const item of items) {
|
||||
await sql`
|
||||
INSERT INTO post_import_items (
|
||||
job_id,
|
||||
item_index,
|
||||
title,
|
||||
slug,
|
||||
source_path,
|
||||
status,
|
||||
progress,
|
||||
message
|
||||
)
|
||||
VALUES (
|
||||
${jobId},
|
||||
${item.itemIndex},
|
||||
${item.title || ''},
|
||||
${item.slug || ''},
|
||||
${item.sourcePath || ''},
|
||||
${IMPORT_ITEM_STATUS.PENDING},
|
||||
0,
|
||||
${'대기 중'}
|
||||
)
|
||||
ON CONFLICT (job_id, item_index)
|
||||
DO UPDATE SET
|
||||
title = EXCLUDED.title,
|
||||
slug = EXCLUDED.slug,
|
||||
source_path = EXCLUDED.source_path,
|
||||
status = EXCLUDED.status,
|
||||
progress = EXCLUDED.progress,
|
||||
message = EXCLUDED.message,
|
||||
post_id = NULL,
|
||||
completed_at = NULL,
|
||||
updated_at = now()
|
||||
`
|
||||
}
|
||||
|
||||
await sql`
|
||||
UPDATE post_import_jobs
|
||||
SET total_count = ${items.length},
|
||||
processed_count = 0,
|
||||
imported_count = 0,
|
||||
skipped_count = 0,
|
||||
failed_count = 0,
|
||||
asset_count = 0,
|
||||
reused_asset_count = 0,
|
||||
warning_count = 0,
|
||||
message = ${items.length > 0 ? 'Import할 게시물 목록을 준비했습니다.' : 'Import할 게시물이 없습니다.'},
|
||||
updated_at = now()
|
||||
WHERE id = ${jobId}
|
||||
`
|
||||
}
|
||||
|
||||
/**
|
||||
* Import 작업을 처리 중으로 표시한다.
|
||||
* @param {import('postgres').Sql} sql - SQL 클라이언트
|
||||
* @param {string} jobId - Import 작업 ID
|
||||
* @returns {Promise<void>}
|
||||
*/
|
||||
const markPostImportJobProcessing = async (sql, jobId) => {
|
||||
await sql`
|
||||
UPDATE post_import_jobs
|
||||
SET status = ${IMPORT_STATUS.PROCESSING},
|
||||
started_at = COALESCE(started_at, now()),
|
||||
completed_at = NULL,
|
||||
error_detail = '',
|
||||
message = ${'Import 작업을 준비하는 중입니다.'},
|
||||
updated_at = now()
|
||||
WHERE id = ${jobId}
|
||||
`
|
||||
}
|
||||
|
||||
/**
|
||||
* Import 게시물 상태를 갱신한다.
|
||||
* @param {import('postgres').Sql} sql - SQL 클라이언트
|
||||
* @param {Object} input - 갱신 입력
|
||||
* @returns {Promise<void>}
|
||||
*/
|
||||
const updatePostImportItem = async (sql, input) => {
|
||||
await sql`
|
||||
UPDATE post_import_items
|
||||
SET status = ${input.status},
|
||||
progress = ${Math.min(Math.max(Number(input.progress || 0), 0), 100)},
|
||||
message = ${input.message || ''},
|
||||
post_id = ${input.postId || null},
|
||||
completed_at = CASE
|
||||
WHEN ${input.status} IN (${IMPORT_ITEM_STATUS.READY}, ${IMPORT_ITEM_STATUS.SKIPPED}, ${IMPORT_ITEM_STATUS.FAILED}) THEN now()
|
||||
ELSE completed_at
|
||||
END,
|
||||
updated_at = now()
|
||||
WHERE job_id = ${input.jobId}
|
||||
AND item_index = ${input.itemIndex}
|
||||
`
|
||||
}
|
||||
|
||||
/**
|
||||
* Import 작업 집계와 메시지를 갱신한다.
|
||||
* @param {import('postgres').Sql} sql - SQL 클라이언트
|
||||
* @param {Object} input - 갱신 입력
|
||||
* @returns {Promise<void>}
|
||||
*/
|
||||
const updatePostImportJobAggregate = async (sql, input) => {
|
||||
const rows = await sql`
|
||||
SELECT
|
||||
COUNT(*) FILTER (WHERE status IN (${IMPORT_ITEM_STATUS.READY}, ${IMPORT_ITEM_STATUS.SKIPPED}, ${IMPORT_ITEM_STATUS.FAILED})) AS processed_count,
|
||||
COUNT(*) FILTER (WHERE status = ${IMPORT_ITEM_STATUS.READY}) AS imported_count,
|
||||
COUNT(*) FILTER (WHERE status = ${IMPORT_ITEM_STATUS.SKIPPED}) AS skipped_count,
|
||||
COUNT(*) FILTER (WHERE status = ${IMPORT_ITEM_STATUS.FAILED}) AS failed_count
|
||||
FROM post_import_items
|
||||
WHERE job_id = ${input.jobId}
|
||||
`
|
||||
const row = rows[0] || {}
|
||||
|
||||
await sql`
|
||||
UPDATE post_import_jobs
|
||||
SET processed_count = ${Number(row.processed_count || 0)},
|
||||
imported_count = ${Number(row.imported_count || 0)},
|
||||
skipped_count = ${Number(row.skipped_count || 0)},
|
||||
failed_count = ${Number(row.failed_count || 0)},
|
||||
asset_count = ${Number(input.assetCount || 0)},
|
||||
reused_asset_count = ${Number(input.reusedAssetCount || 0)},
|
||||
warning_count = ${Number(input.warningCount || 0)},
|
||||
message = ${input.message || ''},
|
||||
updated_at = now()
|
||||
WHERE id = ${input.jobId}
|
||||
`
|
||||
}
|
||||
|
||||
/**
|
||||
* Import 작업을 완료 처리한다.
|
||||
* @param {import('postgres').Sql} sql - SQL 클라이언트
|
||||
* @param {string} jobId - Import 작업 ID
|
||||
* @param {Object} result - Import 결과
|
||||
* @returns {Promise<void>}
|
||||
*/
|
||||
const markPostImportJobReady = async (sql, jobId, result) => {
|
||||
await updatePostImportJobAggregate(sql, {
|
||||
jobId,
|
||||
assetCount: result.assetCount,
|
||||
reusedAssetCount: result.reusedAssetCount,
|
||||
warningCount: result.warningCount,
|
||||
message: `완료: 게시물 ${result.importedCount}개, 건너뜀 ${result.skippedCount}개`
|
||||
})
|
||||
await sql`
|
||||
UPDATE post_import_jobs
|
||||
SET status = ${IMPORT_STATUS.READY},
|
||||
completed_at = now(),
|
||||
updated_at = now()
|
||||
WHERE id = ${jobId}
|
||||
`
|
||||
}
|
||||
|
||||
/**
|
||||
* Import 작업을 실패 처리한다.
|
||||
* @param {import('postgres').Sql} sql - SQL 클라이언트
|
||||
* @param {string} jobId - Import 작업 ID
|
||||
* @param {unknown} error - 실패 원인
|
||||
* @returns {Promise<void>}
|
||||
*/
|
||||
const markPostImportJobFailed = async (sql, jobId, error) => {
|
||||
const message = error instanceof Error ? error.message : '게시물 Import 실패'
|
||||
|
||||
await sql`
|
||||
UPDATE post_import_jobs
|
||||
SET status = ${IMPORT_STATUS.FAILED},
|
||||
message = ${'게시물 Import를 완료하지 못했습니다.'},
|
||||
error_detail = ${message},
|
||||
completed_at = now(),
|
||||
updated_at = now()
|
||||
WHERE id = ${jobId}
|
||||
`
|
||||
}
|
||||
|
||||
/**
|
||||
* 게시물 Import 작업을 실제로 실행한다.
|
||||
* @param {string} jobId - Import 작업 ID
|
||||
* @returns {Promise<void>}
|
||||
*/
|
||||
export const runPostImportJob = async (jobId) => {
|
||||
const sql = getPostgresClient()
|
||||
|
||||
if (!sql || runningPostImportJobIds.has(jobId)) {
|
||||
return
|
||||
}
|
||||
|
||||
runningPostImportJobIds.add(jobId)
|
||||
let reservedSql = null
|
||||
let advisoryLockAcquired = false
|
||||
const advisoryLockNamespace = 'sori-post-import'
|
||||
|
||||
try {
|
||||
reservedSql = await sql.reserve()
|
||||
const lockRows = await reservedSql`
|
||||
SELECT pg_try_advisory_lock(
|
||||
hashtext(${advisoryLockNamespace}),
|
||||
hashtext(${jobId})
|
||||
) AS acquired
|
||||
`
|
||||
advisoryLockAcquired = Boolean(lockRows[0]?.acquired)
|
||||
|
||||
if (!advisoryLockAcquired) {
|
||||
return
|
||||
}
|
||||
|
||||
const job = await getPostImportJobRowById(jobId)
|
||||
|
||||
if (!job || ![IMPORT_STATUS.QUEUED, IMPORT_STATUS.PROCESSING].includes(job.status)) {
|
||||
return
|
||||
}
|
||||
|
||||
await markPostImportJobProcessing(sql, jobId)
|
||||
|
||||
let assetCount = 0
|
||||
let reusedAssetCount = 0
|
||||
let warningCount = 0
|
||||
const result = await importPostsFromExportZip({
|
||||
zipFilePath: job.zip_file_path,
|
||||
authorId: job.requested_by,
|
||||
onPlanned: async (items) => {
|
||||
await upsertPostImportItems(sql, jobId, items)
|
||||
},
|
||||
onItemStart: async (item) => {
|
||||
await updatePostImportItem(sql, {
|
||||
jobId,
|
||||
itemIndex: item.itemIndex,
|
||||
status: IMPORT_ITEM_STATUS.PROCESSING,
|
||||
progress: 35,
|
||||
message: '처리 중'
|
||||
})
|
||||
await updatePostImportJobAggregate(sql, {
|
||||
jobId,
|
||||
assetCount,
|
||||
reusedAssetCount,
|
||||
warningCount,
|
||||
message: `${item.title || item.sourcePath} 처리 중`
|
||||
})
|
||||
},
|
||||
onItemComplete: async (item) => {
|
||||
await updatePostImportItem(sql, {
|
||||
jobId,
|
||||
itemIndex: item.itemIndex,
|
||||
status: item.status,
|
||||
progress: item.progress,
|
||||
message: item.message,
|
||||
postId: item.postId
|
||||
})
|
||||
await updatePostImportJobAggregate(sql, {
|
||||
jobId,
|
||||
assetCount,
|
||||
reusedAssetCount,
|
||||
warningCount,
|
||||
message: item.message || '처리 상태 갱신'
|
||||
})
|
||||
}
|
||||
})
|
||||
assetCount = result.assetCount
|
||||
reusedAssetCount = result.reusedAssetCount
|
||||
warningCount = result.warningCount
|
||||
|
||||
await markPostImportJobReady(sql, jobId, result)
|
||||
} catch (error) {
|
||||
if (advisoryLockAcquired) {
|
||||
await markPostImportJobFailed(sql, jobId, error)
|
||||
} else {
|
||||
console.error('게시물 Import 작업 잠금 획득 실패', error)
|
||||
}
|
||||
} finally {
|
||||
if (reservedSql && advisoryLockAcquired) {
|
||||
await reservedSql`
|
||||
SELECT pg_advisory_unlock(
|
||||
hashtext(${advisoryLockNamespace}),
|
||||
hashtext(${jobId})
|
||||
)
|
||||
`.catch((error) => {
|
||||
console.error('게시물 Import 작업 잠금 해제 실패', error)
|
||||
})
|
||||
}
|
||||
reservedSql?.release()
|
||||
runningPostImportJobIds.delete(jobId)
|
||||
if (advisoryLockAcquired) {
|
||||
setTimeout(() => {
|
||||
queuePendingPostImportJobs().catch((error) => {
|
||||
console.error('후속 게시물 Import 작업 실행 실패', error)
|
||||
})
|
||||
}, 0)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Import 작업을 백그라운드 실행 대기열에 올린다.
|
||||
* @param {string} jobId - Import 작업 ID
|
||||
* @returns {void}
|
||||
*/
|
||||
export const queuePostImportJobRun = (jobId) => {
|
||||
if (!jobId || runningPostImportJobIds.has(jobId)) {
|
||||
return
|
||||
}
|
||||
|
||||
setTimeout(() => {
|
||||
runPostImportJob(jobId).catch((error) => {
|
||||
console.error('게시물 Import 작업 실행 실패', error)
|
||||
})
|
||||
}, 0)
|
||||
}
|
||||
|
||||
/**
|
||||
* 대기 중이거나 중단된 Import 작업을 백그라운드 실행 대기열에 올린다.
|
||||
* @returns {Promise<void>}
|
||||
*/
|
||||
export const queuePendingPostImportJobs = async () => {
|
||||
const jobIds = await listQueuedPostImportJobIds()
|
||||
jobIds.forEach(queuePostImportJobRun)
|
||||
}
|
||||
|
||||
/**
|
||||
* 게시물 Import 작업과 업로드 ZIP을 삭제한다.
|
||||
* @param {string} jobId - Import 작업 ID
|
||||
* @returns {Promise<boolean>} 삭제 여부
|
||||
*/
|
||||
export const deletePostImportJob = async (jobId) => {
|
||||
const sql = getPostgresClient()
|
||||
|
||||
if (!sql) {
|
||||
return false
|
||||
}
|
||||
|
||||
const job = await getPostImportJobRowById(jobId)
|
||||
|
||||
if (!job) {
|
||||
return false
|
||||
}
|
||||
|
||||
if ([IMPORT_STATUS.QUEUED, IMPORT_STATUS.PROCESSING].includes(job.status)) {
|
||||
return false
|
||||
}
|
||||
|
||||
if (job.zip_file_path) {
|
||||
await rm(job.zip_file_path, { force: true })
|
||||
}
|
||||
|
||||
await sql`
|
||||
DELETE FROM post_import_jobs
|
||||
WHERE id = ${jobId}
|
||||
`
|
||||
|
||||
return true
|
||||
}
|
||||
|
||||
19
server/routes/admin/api/posts/import-jobs.get.js
Normal file
19
server/routes/admin/api/posts/import-jobs.get.js
Normal file
@@ -0,0 +1,19 @@
|
||||
import { requireAdminSession } from '../../../../utils/admin-auth'
|
||||
import {
|
||||
listPostImportJobs,
|
||||
queuePendingPostImportJobs
|
||||
} from '../../../../repositories/post-import-repository'
|
||||
|
||||
/**
|
||||
* 관리자 게시물 Import 작업 목록 API
|
||||
* @param {import('h3').H3Event} event - 요청 이벤트
|
||||
* @returns {Promise<Array>} Import 작업 목록
|
||||
*/
|
||||
export default defineEventHandler((event) => {
|
||||
requireAdminSession(event)
|
||||
queuePendingPostImportJobs().catch((error) => {
|
||||
console.error('대기 중인 게시물 Import 작업 실행 실패', error)
|
||||
})
|
||||
|
||||
return listPostImportJobs()
|
||||
})
|
||||
32
server/routes/admin/api/posts/import-jobs/[jobId].delete.js
Normal file
32
server/routes/admin/api/posts/import-jobs/[jobId].delete.js
Normal file
@@ -0,0 +1,32 @@
|
||||
import { createError, getRouterParam } from 'h3'
|
||||
import { requireAdminSession } from '../../../../../utils/admin-auth'
|
||||
import { deletePostImportJob } from '../../../../../repositories/post-import-repository'
|
||||
|
||||
/**
|
||||
* 관리자 게시물 Import 작업 삭제 API
|
||||
* @param {import('h3').H3Event} event - 요청 이벤트
|
||||
* @returns {Promise<{ ok: boolean }>} 삭제 결과
|
||||
*/
|
||||
export default defineEventHandler(async (event) => {
|
||||
requireAdminSession(event)
|
||||
|
||||
const jobId = getRouterParam(event, 'jobId')
|
||||
|
||||
if (!jobId) {
|
||||
throw createError({
|
||||
statusCode: 400,
|
||||
statusMessage: 'Import 작업 ID가 필요합니다.'
|
||||
})
|
||||
}
|
||||
|
||||
const deleted = await deletePostImportJob(jobId)
|
||||
|
||||
if (!deleted) {
|
||||
throw createError({
|
||||
statusCode: 404,
|
||||
statusMessage: 'Import 작업을 찾을 수 없습니다.'
|
||||
})
|
||||
}
|
||||
|
||||
return { ok: true }
|
||||
})
|
||||
@@ -1,9 +1,11 @@
|
||||
import { mkdtemp, rm } from 'node:fs/promises'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { rm } from 'node:fs/promises'
|
||||
import { createError } from 'h3'
|
||||
import { requireAdminSession } from '../../../../utils/admin-auth'
|
||||
import { importPostsFromExportZip } from '../../../../repositories/post-import-repository'
|
||||
import {
|
||||
createPostImportJob,
|
||||
createPostImportJobZipPath,
|
||||
queuePostImportJobRun
|
||||
} from '../../../../repositories/post-import-repository'
|
||||
import { saveMultipartFileToDisk } from '../../../../utils/multipart-file-upload'
|
||||
|
||||
const MAX_IMPORT_ZIP_BYTES = 300 * 1024 * 1024
|
||||
@@ -11,12 +13,12 @@ const MAX_IMPORT_ZIP_BYTES = 300 * 1024 * 1024
|
||||
/**
|
||||
* 관리자 게시물 Import API
|
||||
* @param {import('h3').H3Event} event - 요청 이벤트
|
||||
* @returns {Promise<{ importedCount: number, skippedCount: number, assetCount: number, reusedAssetCount: number, warningCount: number, warnings: Array<string>, posts: Array<Object>, skippedPosts: Array<Object> }>} Import 결과
|
||||
* @returns {Promise<Object>} 생성된 Import 작업
|
||||
*/
|
||||
export default defineEventHandler(async (event) => {
|
||||
const adminSession = requireAdminSession(event)
|
||||
const tempDirectory = await mkdtemp(join(tmpdir(), 'sori-studio-import-'))
|
||||
const zipFilePath = join(tempDirectory, 'upload.zip')
|
||||
const zipFilePath = await createPostImportJobZipPath('upload.zip')
|
||||
let shouldRemoveZip = true
|
||||
|
||||
try {
|
||||
const uploaded = await saveMultipartFileToDisk(event, {
|
||||
@@ -39,10 +41,16 @@ export default defineEventHandler(async (event) => {
|
||||
})
|
||||
}
|
||||
|
||||
return await importPostsFromExportZip({
|
||||
const job = await createPostImportJob({
|
||||
zipFilePath,
|
||||
authorId: adminSession.userId
|
||||
requestedBy: adminSession.userId,
|
||||
requestedEmail: adminSession.email,
|
||||
originalFileName: uploaded.filename || 'upload.zip'
|
||||
})
|
||||
shouldRemoveZip = false
|
||||
queuePostImportJobRun(job.id)
|
||||
|
||||
return job
|
||||
} catch (error) {
|
||||
if (error?.statusCode) {
|
||||
throw error
|
||||
@@ -109,12 +117,11 @@ export default defineEventHandler(async (event) => {
|
||||
|
||||
throw createError({
|
||||
statusCode: 500,
|
||||
message: '게시물 Import를 완료하지 못했습니다.'
|
||||
message: '게시물 Import 작업을 시작하지 못했습니다.'
|
||||
})
|
||||
} finally {
|
||||
await rm(tempDirectory, {
|
||||
recursive: true,
|
||||
force: true
|
||||
})
|
||||
if (shouldRemoveZip) {
|
||||
await rm(zipFilePath, { force: true })
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
@@ -82,7 +82,7 @@ const resolveUploadFilePath = (pathname) => {
|
||||
|
||||
const relativeUrlPath = decodedPath.replace(/^\/uploads\/?/g, '')
|
||||
|
||||
if (!relativeUrlPath || relativeUrlPath.includes('\0')) {
|
||||
if (!relativeUrlPath || relativeUrlPath.includes('\0') || relativeUrlPath === '_import-jobs' || relativeUrlPath.startsWith('_import-jobs/')) {
|
||||
throw createError({
|
||||
statusCode: 404,
|
||||
message: '업로드 파일을 찾을 수 없습니다.'
|
||||
|
||||
Reference in New Issue
Block a user