dorfteich/apps/api/src/import-export/import.service.ts

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 { 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<Record<string, string>> = {
  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<string> {
  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<string, unknown>;
  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<ConversionJobView> {
    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<ConversionJobView> {
    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<ConversionJobView> {
    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<void> {
    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<void> {
    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<string, { data: Uint8Array; name: string }>;
    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<string, string>();
      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<string, string>();
      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<string, string>();
      const linkedAssets = new Set<string>();
      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<string | null> {
    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<string, string>,
  ): Promise<string | null> {
    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<void> {
    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<void> {
    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<Page> {
    // 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 {
      const markdown = await this.storeEmbeddedImages(rawMarkdown, 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);
      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<string> {
    const image = /!\[([^\]]*)\]\((data:[^)\s]+)\)/g;
    const uris = new Set<string>();
    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<string, string | null>();
    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<string | null> {
    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<void> {
    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<Record<string, string>> = {
  '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;
}