import { BadRequestException, ForbiddenException, Injectable, NotFoundException, } from '@nestjs/common'; import { ConversionJob, Page, User } from '@prisma/client'; import { ConversionJobView, ImportVaultOptions, MAX_LABEL_DEPTH, MAX_UPLOAD_PARSE_BYTES, editorSchema, importVaultOptionsSchema, markdownToDoc, nodeDepth, } from '@dorfteich/shared'; import { Node } from 'prosemirror-model'; import { PinoLogger } from 'nestjs-pino'; import { FilesService } from '../files/files.service'; import { LabelsService } from '../labels/labels.service'; import { PagesService } from '../pages/pages.service'; import { docToState, emptyPageState } from '../pages/yjs-content'; import { PrismaService } from '../prisma/prisma.service'; import { parseClassifiedMarkdown } from './classified-markdown'; import { ImportProcessor } from './import.constants'; import { ASSET_PLACEHOLDER_PREFIX, VaultError, VaultImportPlan, parseVaultZip, planVaultImport, } from './obsidian-vault'; import { ConversionError, conversionInputOf, PandocConverter } from './pandoc.converter'; import { ConversionJobService } from './conversion-job.service'; /** Source format per accepted upload extension (ADR 0009). `md` is our own * in-process format (no pandoc); the rest go through the sidecar. */ const IMPORT_FORMATS: Readonly> = { docx: 'docx', odt: 'odt', md: 'md', markdown: 'md', }; /** Job `kind` per source format, so the worker can route the job to the import * pipeline (a plain byte→byte conversion has a different kind, #62/#65). */ function importKind(format: string): string { return `import_${format}`; } /** * Writer format for the structural pass. GFM matches our editor schema * (tables, task lists, strikethrough); `-implicit_figures` keeps images as * inline `![alt](src)` instead of wrapping them in a figure with a duplicated * caption; `-raw_html` stops pandoc emitting raw HTML for constructs it can't * represent (our Markdown parser runs with `html:false` and would drop it * anyway — per ADR 0009 we preserve structure, not layout). */ const IMPORT_MARKDOWN_FORMAT = 'gfm-implicit_figures-raw_html'; /** * The two-pass conversion at the heart of import, extracted so tests can run it * over the fixture corpus against a real sidecar. pandoc-server is stateless and * hands back a document's media no other way, so we first inline every image as * a `data:` URI (`source → html`, embed-resources), then produce clean * structural Markdown that still carries those data URIs inline at the right * positions (`html → gfm`, no wrapping). */ export async function convertImportedDocument( converter: PandocConverter, sourceFormat: string, input: Buffer, ): Promise { const html = await converter.convert({ from: sourceFormat, to: 'html', input, standalone: false, embedResources: true, }); const md = await converter.convert({ from: 'html', to: IMPORT_MARKDOWN_FORMAT, input: html.output, standalone: false, wrap: 'none', }); return md.output.toString('utf8'); } /** Minimal shape of a ProseMirror document as JSON — enough to walk it for * image nodes and the leading heading without pulling in prosemirror types. */ interface PmNode { type: string; attrs?: Record; content?: PmNode[]; text?: string; marks?: unknown[]; } interface DecodedImage { buffer: Buffer; extension: string; } /** * Imports `.docx`/`.odt` documents as new pages (ADR 0009, issue #63). Enqueue * is synchronous and cheap (validate + persist a job on the #62 queue); the * heavy conversion runs out of band in {@link run}, invoked by the shared * {@link ConversionWorker} for import-kind jobs so it inherits the queue's * locking, retry, and restart-survival. * * Pipeline: the pandoc sidecar is stateless and will not hand back a * document's embedded media any other way, so we convert in two passes — * `docx/odt → html` with `embed-resources` inlines every image as a `data:` * URI, then `html → gfm` produces clean structural Markdown with those data * URIs still inline at the right positions. We parse that to an editor * document, store each embedded image as a pond file (with quota accounting) * and rewrite its reference to the file id, derive the title from a leading * top-level heading (else the file name), and create the page from the * resulting Yjs state. */ @Injectable() export class ImportService implements ImportProcessor { constructor( private readonly prisma: PrismaService, private readonly jobs: ConversionJobService, private readonly converter: PandocConverter, private readonly files: FilesService, private readonly pages: PagesService, private readonly labels: LabelsService, private readonly logger: PinoLogger, ) { this.logger.setContext(ImportService.name); } /** * Validate and start an import. Markdown needs no conversion, so it is * imported in-process and the returned view is already `succeeded` with the * created `resultPageId` (issue #64 — "Markdown imports directly, no job"). * `.docx`/`.odt` enqueue a job (the returned id is polled via `GET /jobs/:id` * until it reports `resultPageId`). */ async enqueue( user: User, pondId: string, file: { buffer: Buffer; originalname: string }, ): Promise { const extension = fileExtension(file.originalname); const format = extension ? IMPORT_FORMATS[extension] : undefined; if (!format) { throw new BadRequestException({ code: 'import_unsupported_format' }); } if (format === 'md') return this.importMarkdown(user, pondId, file); const job = await this.jobs.enqueue({ ownerId: user.id, pondId, kind: importKind(format), from: format, to: 'page', input: file.buffer, sourceName: file.originalname, standalone: false, }); return this.jobs.viewOf(job); } /** * Import a Markdown file synchronously (no conversion sidecar, no job): parse, * store any embedded images, create the page, and return a `succeeded` view so * the client can navigate straight to the new page. A conversion-level failure * (a pond out of storage) surfaces as an HTTP error rather than a job status. */ private async importMarkdown( user: User, pondId: string, file: { buffer: Buffer; originalname: string }, ): Promise { const now = new Date().toISOString(); try { const page = await this.createPageFromMarkdown( user, pondId, file.buffer.toString('utf8'), file.originalname, ); return { id: page.id, status: 'succeeded', kind: 'import_md', sourceFormat: 'md', targetFormat: 'page', errorCode: null, resultPageId: page.id, expiresAt: null, createdAt: now, updatedAt: now, }; } catch (error) { throw conversionErrorToHttp(error); } } /** * Validate and enqueue an Obsidian vault import (issue #117). The ZIP is * parsed once here for fast feedback (an invalid archive 400s instead of * failing a job later); the worker re-parses when the job runs. The mount * parent is validated up front too, and again at run time. */ async enqueueVault( user: User, pondId: string, file: { buffer: Buffer; originalname: string }, options: ImportVaultOptions, ): Promise { if (fileExtension(file.originalname) !== 'zip') { throw new BadRequestException({ code: 'import_unsupported_format' }); } try { parseVaultZip(new Uint8Array(file.buffer)); } catch (error) { if (error instanceof VaultError) { throw new BadRequestException({ code: error.code }); } throw error; } if (options.parentPageId) await this.requireLivePageInPond(pondId, options.parentPageId); for (const labelId of options.labelIds) { const label = await this.prisma.label.findFirst({ where: { id: labelId, pondId } }); if (!label) throw new NotFoundException(); } const job = await this.jobs.enqueue({ ownerId: user.id, pondId, kind: 'import_vault', from: 'zip', to: 'pages', input: file.buffer, sourceName: file.originalname, standalone: false, options, // The 25 MiB default guards the pandoc sidecar; a vault never goes there. maxInputBytes: MAX_UPLOAD_PARSE_BYTES, }); return this.jobs.viewOf(job); } /** * Run one import job to completion (called by the worker). Throws a * {@link ConversionError} on failure so the worker applies the queue's * retry/fail policy; on success it records the created page on the job. * Media stored during a failed attempt is rolled back so a retry does not * leak files or double-count quota. */ async run(job: ConversionJob): Promise { if (!job.pondId) { throw new ConversionError('conversion_failed', false, 'import job has no pond'); } const user = await this.prisma.user.findUnique({ where: { id: job.ownerId } }); if (!user) { throw new ConversionError('conversion_failed', false, 'import job owner is gone'); } if (job.kind === 'import_vault') { return this.runVault(job, user, job.pondId); } const rawMarkdown = await convertImportedDocument( this.converter, job.sourceFormat, Buffer.from(conversionInputOf(job)), ); const page = await this.createPageFromMarkdown(user, job.pondId, rawMarkdown, job.sourceName); await this.prisma.conversionJob.update({ where: { id: job.id }, data: { status: 'SUCCEEDED', resultPageId: page.id, errorCode: null }, }); this.logger.info( { jobId: job.id, pageId: page.id, pondId: job.pondId }, 'audit: document imported', ); } /** * Run a vault import job (issue #117): transform (issue #116) → containers * top-down → notes → assets → labels, ALL-OR-NOTHING. On any failure every * created page is hard-deleted (children before parents) and every stored * file removed (quota restored), so the worker's retry policy is safe and a * failed import leaves no trace. */ private async runVault(job: ConversionJob, user: User, pondId: string): Promise { const options = importVaultOptionsSchema.parse(job.options ?? {}); // Re-validate the mount parent (it may have been trashed since enqueue) // and compute its depth for the folder budget. let mountDepth = 0; if (options.parentPageId) { const live = await this.prisma.page.findFirst({ where: { id: options.parentPageId, pondId, deletedAt: null }, select: { id: true }, }); if (!live) { throw new ConversionError('conversion_failed', false, 'vault mount parent is gone'); } const tree = await this.prisma.page.findMany({ where: { pondId, deletedAt: null }, select: { id: true, parentId: true }, }); mountDepth = nodeDepth(tree, options.parentPageId); } // Slugs are unique across live AND trashed pages (the unique index does // not care about deletedAt), so the reservation set includes everything. const existing = await this.prisma.page.findMany({ where: { pondId }, select: { slug: true }, }); let plan: VaultImportPlan; let assetBytes: Map; try { const zip = new Uint8Array(conversionInputOf(job)); plan = planVaultImport(zip, { frontmatterMode: options.frontmatterMode, existingSlugs: new Set(existing.map((row) => row.slug)), mountDepth, }); const vault = parseVaultZip(zip); assetBytes = new Map( vault.assets.map((asset) => [asset.path, { data: asset.data, name: asset.name }]), ); } catch (error) { if (error instanceof VaultError) { throw new ConversionError(error.code, false, error.message); } throw error; } const createdPageIds: string[] = []; const storedFileIds: string[] = []; try { // Container pages top-down (parents exist before their children). const pageIdByKey = new Map(); const parentOf = (key: string | null): string | null => key ? pageIdByKey.get(key)! : (options.parentPageId ?? null); for (const container of plan.containers) { const page = await this.pages.createWithState( user, pondId, container.title, emptyPageState(), parentOf(container.parentKey), container.slug, ); createdPageIds.push(page.id); pageIdByKey.set(container.key, page.id); } // Referenced assets, uploaded once; the attachment row links to the // first page that references the asset (done per note below). const fileIdByAsset = new Map(); for (const path of plan.referencedAssets) { const asset = assetBytes.get(path); if (!asset) continue; const view = await this.uploadVaultAsset(user, pondId, asset, storedFileIds); if (view) fileIdByAsset.set(path, view); } // Notes: replace asset placeholders, parse, create, link, label. const labelIdByPath = new Map(); const linkedAssets = new Set(); for (const note of plan.notes) { const placeholder = new RegExp( `!\\[([^\\]]*)\\]\\(${ASSET_PLACEHOLDER_PREFIX}([^)\\s]+)\\)`, 'g', ); const markdown = note.markdown.replaceAll( placeholder, (_whole, alt: string, path: string) => { const fileId = fileIdByAsset.get(path); return fileId ? `![${alt}](${fileId})` : ''; }, ); const json = markdownToDoc(markdown).toJSON() as unknown as PmNode; const state = docToState(Node.fromJSON(editorSchema, json)); const page = await this.pages.createWithState( user, pondId, note.title, state, parentOf(note.parentKey), note.slug, ); createdPageIds.push(page.id); const noteAssets = [...note.imageAssets, ...note.attachmentAssets] .filter((path) => !linkedAssets.has(path)) .map((path) => fileIdByAsset.get(path)) .filter((id): id is string => Boolean(id)); for (const path of [...note.imageAssets, ...note.attachmentAssets]) { linkedAssets.add(path); } await this.files.linkAttachmentsToPage(noteAssets, page.id); for (const tag of note.tags) { const labelId = await this.ensureLabelPath(user, pondId, tag, labelIdByPath); if (labelId) await this.labels.assign(user, page.id, labelId); } for (const labelId of options.labelIds) { await this.labels.assign(user, page.id, labelId); } } await this.prisma.conversionJob.update({ where: { id: job.id }, data: { status: 'SUCCEEDED', resultPageId: options.parentPageId ?? null, errorCode: null, }, }); this.logger.info( { jobId: job.id, pondId, pages: createdPageIds.length, files: storedFileIds.length }, 'audit: vault imported', ); } catch (error) { await this.rollbackPages(createdPageIds); await this.rollbackMedia(user, storedFileIds); if (error instanceof ForbiddenException && errorCode(error) === 'quota_exceeded') { throw new ConversionError('quota_exceeded', false, 'pond storage exhausted during import'); } if (error instanceof ConversionError || error instanceof VaultError) { throw error instanceof VaultError ? new ConversionError(error.code, false, error.message) : error; } throw new ConversionError( 'conversion_failed', true, error instanceof Error ? error.message : 'vault import failed', ); } } /** Uploads one vault asset as a pond file; a type the instance rejects is * dropped (structure over layout), an exhausted pond fails the import. */ private async uploadVaultAsset( user: User, pondId: string, asset: { data: Uint8Array; name: string }, storedFileIds: string[], ): Promise { try { const view = await this.files.upload(user, pondId, { buffer: Buffer.from(asset.data), size: asset.data.length, originalname: asset.name, }); storedFileIds.push(view.id); return view.id; } catch (error) { if (error instanceof ForbiddenException && errorCode(error) === 'quota_exceeded') { throw error; } this.logger.warn({ pondId, code: errorCode(error) }, 'vault import: dropped an asset'); return null; } } /** * Find-or-create the label chain for a nested tag (`a/b` → label `a` with * child `b`), returning the deepest label's id. Creation goes through * LabelsService so locking and permission-cache invalidation apply; chains * deeper than the label limit are clamped to it. */ private async ensureLabelPath( user: User, pondId: string, tag: string[], cache: Map, ): Promise { const path = tag.slice(0, MAX_LABEL_DEPTH); let parentId: string | null = null; let key = ''; for (const name of path) { key = key ? `${key}/${name}` : name; const cached = cache.get(key); if (cached) { parentId = cached; continue; } const found: { id: string } | null = await this.prisma.label.findFirst({ where: { pondId, parentId, name }, select: { id: true }, }); const id: string = found ? found.id : (await this.labels.create(user, pondId, { name, parentId })).id; cache.set(key, id); parentId = id; } return parentId; } /** Hard-deletes the pages a failed vault import created — children before * parents (creation order reversed), attachments already gone via the media * rollback, everything else cascades. */ private async rollbackPages(pageIds: string[]): Promise { for (const id of [...pageIds].reverse()) { try { await this.prisma.page.delete({ where: { id } }); } catch (error) { this.logger.warn( { pageId: id, error: error instanceof Error ? error.message : String(error) }, 'vault import: page rollback failed', ); } } } private async requireLivePageInPond(pondId: string, pageId: string): Promise { const page = await this.prisma.page.findFirst({ where: { id: pageId, pondId, deletedAt: null }, select: { id: true }, }); if (!page) throw new NotFoundException(); } /** * The shared tail of every import: store embedded images, parse the Markdown * into an editor document, and create a page from it. Media stored during a * failed attempt is rolled back so a retry (or the caller's error) leaves no * orphaned files or double-counted quota, and no half-created page. */ private async createPageFromMarkdown( user: User, pondId: string, rawMarkdown: string, sourceName: string | null, ): Promise { // Media is stored before the page exists (the page's state references the // file ids); track what we create so a later failure can be rolled back. const storedFileIds: string[] = []; try { // Our own classified export wraps the content in frontmatter + marking // lines (#210) — strip them and carry the level into the new page, so // a round-trip neither duplicates the marking nor loses it. const { markdown: unwrapped, classification } = parseClassifiedMarkdown(rawMarkdown); const markdown = await this.storeEmbeddedImages(unwrapped, user, pondId, storedFileIds); const json = markdownToDoc(markdown).toJSON() as unknown as PmNode; const { title, doc } = this.splitTitle(json, sourceName); const state = docToState(Node.fromJSON(editorSchema, doc)); const page = await this.pages.createWithState( user, pondId, title, state, null, undefined, classification, ); await this.files.linkAttachmentsToPage(storedFileIds, page.id); return page; } catch (error) { await this.rollbackMedia(user, storedFileIds); throw error; } } /** * Store each embedded image (an inline `data:` URI, from the embed-resources * pass) as a pond file and rewrite the Markdown reference to the stored file * id. Done on the Markdown text, before parsing, because the editor's parser * only admits a whitelist of `data:` image types (png/jpeg/gif/webp) and * would leave any other inline as a huge literal base64 string. base64 has no * `)`, so the image regex is unambiguous. An image whose bytes the upload * pipeline rejects (e.g. a vector `image/x-emf` Word can embed) is dropped * rather than failing the whole import — structure over layout (ADR 0009); a * pond that runs out of storage fails the import. */ private async storeEmbeddedImages( markdown: string, user: User, pondId: string, storedFileIds: string[], ): Promise { const image = /!\[([^\]]*)\]\((data:[^)\s]+)\)/g; const uris = new Set(); for (const match of markdown.matchAll(image)) uris.add(match[2]!); if (uris.size === 0) return markdown; // Dedup identical images so a document that repeats one stores it once. const resolved = new Map(); let index = 0; for (const uri of uris) { resolved.set(uri, await this.storeImage(uri, user, pondId, ++index, storedFileIds)); } return markdown.replace(image, (_whole, alt: string, uri: string) => { const fileId = resolved.get(uri); return fileId ? `![${alt}](${fileId})` : ''; }); } private async storeImage( uri: string, user: User, pondId: string, index: number, storedFileIds: string[], ): Promise { const decoded = decodeDataUri(uri); if (!decoded) return null; try { const view = await this.files.upload(user, pondId, { buffer: decoded.buffer, size: decoded.buffer.length, originalname: `import-${index}.${decoded.extension}`, }); storedFileIds.push(view.id); return view.id; } catch (error) { if (error instanceof ForbiddenException && errorCode(error) === 'quota_exceeded') { throw new ConversionError('quota_exceeded', false, 'pond storage exhausted during import'); } // Unsupported image type / rejected bytes — drop this one, keep going. this.logger.warn({ pondId, code: errorCode(error) }, 'import: dropped an image'); return null; } } /** * Use a leading top-level heading as the page title (and remove it from the * body so it is not duplicated), else fall back to the upload file name * without its extension, else a generic title. */ private splitTitle(root: PmNode, sourceName: string | null): { title: string; doc: PmNode } { const content = root.content ?? []; const first = content[0]; if (first && first.type === 'heading') { const headingText = textOf(first).trim(); if (headingText) { return { title: headingText, doc: { ...root, content: content.slice(1) } }; } } const fallback = baseName(sourceName) || 'Imported document'; return { title: fallback, doc: root }; } private async rollbackMedia(user: User, fileIds: string[]): Promise { for (const id of fileIds) { try { await this.files.remove(user, id); } catch (error) { this.logger.warn({ fileId: id, code: errorCode(error) }, 'import: media rollback failed'); } } } } /** Concatenated text of a node's inline content (headings have no nesting). */ function textOf(node: PmNode): string { if (typeof node.text === 'string') return node.text; return (node.content ?? []).map(textOf).join(''); } const DATA_URI = /^data:(image\/[a-zA-Z0-9.+-]+);base64,(.+)$/; const EXTENSION_BY_MIME: Readonly> = { 'image/png': 'png', 'image/jpeg': 'jpg', 'image/gif': 'gif', 'image/webp': 'webp', 'image/svg+xml': 'svg', }; function decodeDataUri(value: string): DecodedImage | null { const match = DATA_URI.exec(value); if (!match) return null; const mime = match[1]!; const buffer = Buffer.from(match[2]!, 'base64'); if (buffer.length === 0) return null; return { buffer, extension: EXTENSION_BY_MIME[mime] ?? 'bin' }; } /** Lowercased extension without the dot, or '' when there is none. */ function fileExtension(fileName: string): string { const dot = fileName.lastIndexOf('.'); return dot >= 0 ? fileName.slice(dot + 1).toLowerCase() : ''; } /** File name without its extension. */ function baseName(fileName: string | null): string { if (!fileName) return ''; const dot = fileName.lastIndexOf('.'); return (dot > 0 ? fileName.slice(0, dot) : fileName).trim(); } /** The `code` from a Nest HttpException response body, if any. */ function errorCode(error: unknown): string | undefined { if (error instanceof BadRequestException || error instanceof ForbiddenException) { const body = error.getResponse(); if (body && typeof body === 'object' && 'code' in body) { const code = (body as { code?: unknown }).code; return typeof code === 'string' ? code : undefined; } } return undefined; } /** * Turn a conversion-level failure from the shared pipeline into an HTTP error * for the synchronous Markdown path (there is no job to carry the code). An * exhausted pond is a 403 `quota_exceeded`; any other conversion error is a 400 * with its code. A Nest HttpException (e.g. the pond 404) passes through. */ function conversionErrorToHttp(error: unknown): unknown { if (error instanceof ConversionError) { if (error.code === 'quota_exceeded') { return new ForbiddenException({ code: 'quota_exceeded' }); } return new BadRequestException({ code: error.code }); } return error; }