import { createReadStream, createWriteStream } from 'node:fs' 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' import { openZipFileEntries } from '../utils/zip-reader' import { upsertMediaMetadataCategory } from '../utils/media-library' import { getUploadRoot } from '../utils/upload-root' import { getPostgresClient } from './postgres-client' import { createAdminPost } from './content-repository' const UPLOAD_BASE_URL = '/uploads' const MAX_IMPORT_POSTS = 1000 const MAX_IMPORT_ZIP_ENTRY_COUNT = 5000 const MAX_IMPORT_ENTRY_BYTES = 300 * 1024 * 1024 const MAX_IMPORT_TOTAL_BYTES = 2 * 1024 * 1024 * 1024 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() /** * 파일명에 안전한 문자열로 정리한다. * @param {string} value - 원본 문자열 * @returns {string} 정리된 문자열 */ const sanitizeFilenameSegment = (value) => String(value || '') .trim() .replace(/[\\/:*?"<>|]+/g, '-') .replace(/\s+/g, '-') .replace(/-+/g, '-') .replace(/^-|-$/g, '') .slice(0, 80) || 'asset' /** * Import 작업 ZIP 저장 경로를 만든다. * @param {string} originalName - 원본 파일명 * @returns {Promise} 저장 경로 */ 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 - 읽기 스트림 * @returns {Promise} 해시 문자열 */ const createStreamHash = async (stream) => { const hash = createHash('sha256') for await (const chunk of stream) { hash.update(chunk) } return hash.digest('hex') } /** * 슬러그에 안전한 문자열로 정리한다. * @param {string} value - 원본 슬러그 * @returns {string} 정리된 슬러그 */ const sanitizeSlug = (value) => String(value || '') .trim() .toLowerCase() .normalize('NFC') .replace(/[^a-z0-9가-힣]+/g, '-') .replace(/-+/g, '-') .replace(/^-|-$/g, '') || 'imported-post' /** * YAML 문자열 따옴표를 해제한다. * @param {string} value - YAML 값 * @returns {string} 문자열 값 */ const unquoteYamlString = (value) => { const trimmed = String(value || '').trim() if (!trimmed) { return '' } if (trimmed.startsWith('"') && trimmed.endsWith('"')) { return trimmed .slice(1, -1) .replace(/\\"/g, '"') .replace(/\\\\/g, '\\') } return trimmed } /** * YAML 배열 값을 파싱한다. * @param {string} value - YAML 배열 문자열 * @returns {Array} 문자열 배열 */ const parseYamlArray = (value) => { const trimmed = String(value || '').trim() if (!trimmed || trimmed === '[]') { return [] } if (!trimmed.startsWith('[') || !trimmed.endsWith(']')) { return [] } const inner = trimmed.slice(1, -1).trim() if (!inner) { return [] } const values = [] let current = '' let inQuote = false let escaped = false for (const char of inner) { if (escaped) { current += char escaped = false continue } if (char === '\\') { current += char escaped = true continue } if (char === '"') { current += char inQuote = !inQuote continue } if (char === ',' && !inQuote) { values.push(unquoteYamlString(current)) current = '' continue } current += char } if (current.trim()) { values.push(unquoteYamlString(current)) } return values.map((item) => item.trim()).filter(Boolean) } /** * YAML 단일 값을 파싱한다. * @param {string} value - YAML 값 * @returns {unknown} 파싱된 값 */ const parseYamlValue = (value) => { const trimmed = String(value || '').trim() if (trimmed === 'null') { return null } if (trimmed === 'true') { return true } if (trimmed === 'false') { return false } if (trimmed.startsWith('[')) { return parseYamlArray(trimmed) } return unquoteYamlString(trimmed) } /** * Markdown frontmatter 줄을 키와 값으로 나눈다. * @param {string} line - frontmatter 한 줄 * @returns {{ key: string, value: string } | null} 키와 값 */ const parseFrontmatterLine = (line) => { const match = String(line || '').match(/^([A-Za-z0-9_-]+):\s*(.*)$/) if (!match) { return null } return { key: match[1].trim(), value: match[2] } } /** * YAML 블록 배열 값을 수집한다. * @param {Array} lines - frontmatter 줄 목록 * @param {number} startIndex - 시작 인덱스 * @returns {{ values: Array, nextIndex: number }} 수집 결과 */ const collectYamlBlockArray = (lines, startIndex) => { const values = [] let index = startIndex while (index < lines.length) { const line = lines[index] const itemMatch = String(line || '').match(/^\s*-\s*(.*)$/) if (!itemMatch) { break } const value = unquoteYamlString(itemMatch[1]) if (value) { values.push(value) } index += 1 } return { values, nextIndex: index } } /** * Markdown frontmatter와 본문을 분리한다. * @param {string} markdown - Markdown 문서 * @returns {{ frontmatter: Object, content: string }} 분리 결과 */ const parseMarkdownDocument = (markdown) => { const normalized = String(markdown || '').replace(/^\uFEFF/, '') if (!normalized.startsWith('---\n')) { return { frontmatter: {}, content: normalized } } const endIndex = normalized.indexOf('\n---', 4) if (endIndex < 0) { return { frontmatter: {}, content: normalized } } const frontmatterText = normalized.slice(4, endIndex) const content = normalized.slice(endIndex + 4).replace(/^\n/, '') const frontmatter = {} const lines = frontmatterText.split('\n') for (let index = 0; index < lines.length; index += 1) { const parsedLine = parseFrontmatterLine(lines[index]) if (!parsedLine) { continue } const { key, value } = parsedLine const trimmedValue = String(value || '').trim() if (!trimmedValue && lines[index + 1]?.match(/^\s*-\s*/)) { const blockArray = collectYamlBlockArray(lines, index + 1) frontmatter[key] = blockArray.values index = blockArray.nextIndex - 1 continue } frontmatter[key] = parseYamlValue(value) } return { frontmatter, content } } /** * ZIP 엔트리를 경로 기준 Map으로 만든다. * @param {Array<{ path: string }>} entries - ZIP 엔트리 * @returns {Map} 엔트리 맵 */ const createZipEntryMap = (entries) => new Map(entries.map((entry) => [entry.path, entry])) /** * Markdown 파일의 상위 폴더를 조회한다. * @param {string} path - ZIP 내부 경로 * @returns {string} 상위 폴더 */ const getPostFolder = (path) => { const parts = String(path || '').split('/').filter(Boolean) parts.pop() return parts.join('/') } /** * 자산 경로를 Markdown 파일 기준 ZIP 엔트리 경로로 해석한다. * @param {string} postFolder - 게시물 폴더 * @param {string} assetPath - Markdown 안의 자산 경로 * @returns {string} ZIP 엔트리 경로 */ const resolveAssetEntryPath = (postFolder, assetPath) => { const cleaned = String(assetPath || '') .split(/[?#]/)[0] .replace(/^\.\/+/, '') .replace(/^\/+/, '') if (!cleaned) { return '' } const base = postFolder ? `${postFolder}/${cleaned}` : cleaned return base .split('/') .filter((part) => part && part !== '.' && part !== '..') .join('/') } /** * 저장할 고유 파일명을 고른다. * @param {string} directoryPath - 저장 디렉터리 * @param {string} originalName - 원본 파일명 * @returns {Promise<{ fileName: string, filePath: string }>} 저장 파일명과 경로 */ const pickUniqueDiskFileName = async (directoryPath, originalName) => { const extension = extname(originalName || '') || '.bin' const stem = sanitizeFilenameSegment(String(originalName || '').replace(/\.[^.]+$/g, '')) || 'asset' let suffix = 1 while (suffix < 10000) { const fileName = suffix === 1 ? `${stem}${extension}` : `${stem}-${suffix}${extension}` const filePath = join(directoryPath, fileName) try { await stat(filePath) suffix += 1 } catch { return { fileName, filePath } } } throw new Error('IMPORT_ASSET_FILENAME_FAILED') } /** * 이미 저장된 같은 자산 파일을 찾는다. * @param {string} directoryPath - 저장 디렉터리 * @param {string} originalName - 원본 파일명 * @param {{ uncompressedSize: number, createReadStream: () => NodeJS.ReadableStream }} entry - ZIP 엔트리 * @returns {Promise<{ fileName: string, filePath: string }|null>} 재사용 파일 */ const findReusableDiskAsset = async (directoryPath, originalName, entry) => { const extension = extname(originalName || '') || '.bin' const stem = sanitizeFilenameSegment(String(originalName || '').replace(/\.[^.]+$/g, '')) || 'asset' const entrySize = Number(entry.uncompressedSize || 0) let entryHash = '' for (let suffix = 1; suffix < 10000; suffix += 1) { const fileName = suffix === 1 ? `${stem}${extension}` : `${stem}-${suffix}${extension}` const filePath = join(directoryPath, fileName) let fileStat = null try { fileStat = await stat(filePath) } catch { return null } if (entrySize > 0 && Number(fileStat.size || 0) !== entrySize) { continue } entryHash ||= await createStreamHash(entry.createReadStream()) if (await createStreamHash(createReadStream(filePath)) === entryHash) { return { fileName, filePath } } } return null } /** * Import 자산을 업로드 폴더에 저장한다. * @param {Object} input - 저장 입력 * @param {Map} input.entryMap - ZIP 엔트리 맵 * @param {string} input.postFolder - 게시물 폴더 * @param {Set} input.assetPaths - 자산 경로 목록 * @param {Map} input.savedAssetUrls - ZIP 엔트리별 저장 URL 캐시 * @returns {Promise<{ replacements: Map, missingAssets: Array, savedCount: number, reusedCount: number }>} 저장 결과 */ const saveImportAssets = async ({ entryMap, postFolder, assetPaths, savedAssetUrls }) => { const now = new Date() const year = String(now.getFullYear()) const month = String(now.getMonth() + 1).padStart(2, '0') const directoryPath = join(getUploadRoot(), 'posts', year, month) const replacements = new Map() const missingAssets = [] let savedCount = 0 let reusedCount = 0 await mkdir(directoryPath, { recursive: true }) for (const assetPath of assetPaths) { const entryPath = resolveAssetEntryPath(postFolder, assetPath) const entry = entryMap.get(entryPath) if (!entry) { missingAssets.push(assetPath) continue } const cachedUrl = savedAssetUrls.get(entryPath) if (cachedUrl) { replacements.set(assetPath, cachedUrl) replacements.set(assetPath.replace(/^\.\//, ''), cachedUrl) replacements.set(`./${assetPath.replace(/^\.\//, '')}`, cachedUrl) reusedCount += 1 continue } const reusableAsset = await findReusableDiskAsset(directoryPath, basename(entryPath), entry) if (reusableAsset) { const publicUrl = `${UPLOAD_BASE_URL}/posts/${year}/${month}/${reusableAsset.fileName}` savedAssetUrls.set(entryPath, publicUrl) replacements.set(assetPath, publicUrl) replacements.set(assetPath.replace(/^\.\//, ''), publicUrl) replacements.set(`./${assetPath.replace(/^\.\//, '')}`, publicUrl) reusedCount += 1 continue } const { fileName, filePath } = await pickUniqueDiskFileName(directoryPath, basename(entryPath)) try { await pipeline( entry.createReadStream(), createWriteStream(filePath, { flags: 'wx' }) ) } catch (error) { await rm(filePath, { force: true }) throw error } const publicUrl = `${UPLOAD_BASE_URL}/posts/${year}/${month}/${fileName}` await upsertMediaMetadataCategory(publicUrl, '미분류', { origin: 'import' }) savedAssetUrls.set(entryPath, publicUrl) replacements.set(assetPath, publicUrl) replacements.set(assetPath.replace(/^\.\//, ''), publicUrl) replacements.set(`./${assetPath.replace(/^\.\//, '')}`, publicUrl) savedCount += 1 } return { replacements, missingAssets, savedCount, reusedCount } } /** * Markdown 안의 로컬 자산 경로를 새 업로드 URL로 교체한다. * @param {string} content - 원본 본문 * @param {Map} replacements - 경로 교체 맵 * @returns {string} 교체된 본문 */ const replaceAssetPaths = (content, replacements) => { let next = String(content || '') const entries = [...replacements.entries()] .sort((a, b) => b[0].length - a[0].length) for (const [source, target] of entries) { next = next.split(source).join(target) } return next } /** * Import 대상 자산 경로를 수집한다. * @param {Object} input - 수집 입력 * @param {Object} input.frontmatter - frontmatter * @param {string} input.content - 본문 * @returns {Set} 자산 경로 목록 */ const collectImportAssetPaths = ({ frontmatter, content }) => { const paths = new Set() const localAssetPattern = /(?:\.\/)?(?:images|files)\/[^\s"'<>)]*/g for (const match of String(content || '').match(localAssetPattern) || []) { paths.add(match) } for (const key of ['featured_image', 'og_image']) { const value = frontmatter[key] if (typeof value === 'string' && /^(?:\.\/)?(?:images|files)\//.test(value)) { paths.add(value) } } return paths } /** * 게시물 슬러그가 이미 존재하는지 확인한다. * @param {string} slug - 게시물 슬러그 * @returns {Promise} 존재 여부 */ const postSlugExists = async (slug) => { const sql = getPostgresClient() if (!sql) { return false } const rows = await sql` SELECT 1 FROM posts WHERE slug = ${slug} LIMIT 1 ` return rows.length > 0 } /** * Import 게시물 슬러그를 정리한다. * @param {string} baseSlug - 기준 슬러그 * @param {Set} reservedSlugs - 이번 Import에서 예약된 슬러그 * @returns {{ slug: string, duplicateInZip: boolean }} 정리된 슬러그 */ const reserveImportPostSlug = (baseSlug, reservedSlugs) => { const slug = sanitizeSlug(baseSlug) const duplicateInZip = reservedSlugs.has(slug) reservedSlugs.add(slug) return { slug, duplicateInZip } } /** * 게시물 상태를 Import 가능한 값으로 정리한다. * @param {unknown} value - 상태 값 * @returns {'published'|'draft'|'members'|'private'} 게시물 상태 */ const normalizePostStatus = (value) => { const status = String(value || '').trim() if (['published', 'draft', 'members', 'private'].includes(status)) { return status } return 'draft' } /** * frontmatter 이미지 값을 Import 후 URL로 정리한다. * @param {unknown} value - frontmatter 이미지 값 * @param {Map} replacements - 자산 교체 맵 * @returns {string|null} 저장할 이미지 URL */ const resolveImportedImageUrl = (value, replacements) => { if (typeof value !== 'string' || !value.trim()) { return null } const trimmed = value.trim() const replaced = replacements.get(trimmed) if (replaced) { return replaced } if (/^(?:\.\/)?(?:images|files)\//.test(trimmed)) { return null } return trimmed } /** * ISO 날짜 문자열을 정리한다. * @param {unknown} value - 날짜 값 * @returns {string|null} ISO 문자열 */ const normalizeIsoDate = (value) => { if (!value) { return null } const date = new Date(String(value)) if (Number.isNaN(date.getTime())) { return null } return date.toISOString() } /** * ZIP 엔트리에서 Markdown 게시물 목록을 만든다. * @param {Array<{ path: string, uncompressedSize: number, readBuffer: (maxBytes?: number) => Promise }>} entries - ZIP 엔트리 * @returns {Promise>} Markdown 문서 목록 */ const collectMarkdownPosts = async (entries) => { const markdownEntries = entries.filter((entry) => MARKDOWN_EXTENSION_PATTERN.test(entry.path)) if (markdownEntries.length > MAX_IMPORT_POSTS) { throw new Error('IMPORT_POST_LIMIT_EXCEEDED') } const markdownPosts = [] let totalMarkdownBytes = 0 for (const entry of markdownEntries) { totalMarkdownBytes += Number(entry.uncompressedSize || 0) if (totalMarkdownBytes > MAX_IMPORT_TOTAL_MARKDOWN_BYTES) { throw new Error('IMPORT_MARKDOWN_SIZE_LIMIT_EXCEEDED') } const data = await entry.readBuffer(MAX_IMPORT_MARKDOWN_BYTES) markdownPosts.push({ path: entry.path, postFolder: getPostFolder(entry.path), ...parseMarkdownDocument(data.toString('utf8')) }) } return markdownPosts } /** * Export ZIP을 게시물로 가져온다. * @param {Object} input - Import 입력 * @param {string} input.zipFilePath - ZIP 파일 경로 * @param {string} input.authorId - 작성자 ID * @param {(posts: Array) => Promise|void} [input.onPlanned] - 게시물 계획 콜백 * @param {(item: Object) => Promise|void} [input.onItemStart] - 게시물 처리 시작 콜백 * @param {(item: Object) => Promise|void} [input.onItemComplete] - 게시물 처리 완료 콜백 * @returns {Promise<{ importedCount: number, skippedCount: number, failedCount: number, assetCount: number, reusedAssetCount: number, warningCount: number, warnings: Array, posts: Array, skippedPosts: Array, failedPosts: Array }>} Import 결과 */ 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, maxTotalBytes: MAX_IMPORT_TOTAL_BYTES, maxCompressionRatio: MAX_IMPORT_COMPRESSION_RATIO }) const entryMap = createZipEntryMap(entries) const markdownPosts = await collectMarkdownPosts(entries) if (!markdownPosts.length) { 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 (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, slug, 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 } if (await postSlugExists(slug)) { skippedPosts.push({ title, slug, reason: 'existing-slug' }) await onItemComplete?.({ itemIndex, status: IMPORT_ITEM_STATUS.SKIPPED, progress: 100, message: '이미 같은 slug의 게시물이 있어 건너뜀' }) continue } const assetPaths = collectImportAssetPaths({ frontmatter, content }) const { replacements, missingAssets, savedCount, reusedCount } = await saveImportAssets({ entryMap, postFolder, assetPaths, savedAssetUrls }) importedAssetCount += savedCount reusedAssetCount += reusedCount missingAssets.forEach((assetPath) => { 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) const publishedAt = status === 'published' || status === 'members' ? normalizeIsoDate(frontmatter.published_at) || new Date().toISOString() : normalizeIsoDate(frontmatter.published_at) let post try { post = await createAdminPost({ title, slug, content: replaceAssetPaths(content, replacements), excerpt: String(frontmatter.excerpt || ''), featuredImage, showFeaturedImage: Boolean(frontmatter.show_featured_image), isFeatured: false, seoTitle: String(frontmatter.seo_title || ''), seoDescription: String(frontmatter.seo_description || ''), canonicalUrl: String(frontmatter.canonical_url || ''), noindex: Boolean(frontmatter.noindex), ogImage, status, publishedAt, tags: Array.isArray(frontmatter.tags) ? frontmatter.tags : [] }, authorId) } catch (error) { if (error?.code === '23505') { skippedPosts.push({ title, 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, warnings: warnings.slice(0, 20), posts: importedPosts.map((post) => ({ id: post.id, title: post.title, slug: post.slug })), skippedPosts: skippedPosts.slice(0, 20), failedPosts: failedPosts.slice(0, 20) } } /** * Import 작업 단일 행을 조회한다. * @param {string} jobId - Import 작업 ID * @returns {Promise} 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} 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} 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} 생성된 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>} 실행 가능한 작업 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} items - 게시물 계획 * @returns {Promise} */ 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} */ 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} */ 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} */ 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} */ 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} */ 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} */ 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} */ export const queuePendingPostImportJobs = async () => { const jobIds = await listQueuedPostImportJobIds() jobIds.forEach(queuePostImportJobRun) } /** * 게시물 Import 작업과 업로드 ZIP을 삭제한다. * @param {string} jobId - Import 작업 ID * @returns {Promise} 삭제 여부 */ 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 }