fix(mail): Duplikate auch bei gerade laufenden Importen erkennen

Wurde dieselbe Mail zweimal weitergeleitet, erkannte der Import-Assistent das Duplikat nicht, solange Paperless das erste Dokument noch verarbeitete – geprüft wurde nur die Checksumme in Paperless. - checkDuplicate() berücksichtigt zusätzlich offene Paperless-Tasks von Anhängen mit gleicher Checksumme (gleicher Seitenbereich) - Assistent prüft erneut vor dem Abrufen der Belegnummern und vor dem Import; Duplikate werden ignoriert, reservierte Nummern freigegeben - executeImport prüft serverseitig vor Upload und Festlegen der Nummer, sperrt laufende Importe gegen parallele Doppelimporte und meldet 409 Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
2026-09-25 10:29:43 +02:00
1 parent 3cea3fb239
commit 534b313054
@@ -95,15 +95,24 @@ export class EmailImportController {
    return { success: true };
  }

-  // --- Split Checksum Check ---
+  // --- Duplikatprüfung ---
+  @Post('attachments/:attachmentId/check-duplicate')
+  @RequirePermissions(Permission.VIEW_MAIL)
+  async checkDuplicate(@Param('attachmentId') attachmentId: string) {
+    const isDuplicate = await this.importService.checkDuplicate(
+      parseInt(attachmentId, 10),
+    );
+    return { isDuplicate };
+  }
+
  @Post('attachments/:attachmentId/check-split-checksum')
  @RequirePermissions(Permission.VIEW_MAIL)
  async checkSplitChecksum(
-    @Param('attachmentId') attachmentId: number,
+    @Param('attachmentId') attachmentId: string,
    @Body() body: { pages: { start: number; end: number } },
  ) {
-    const isDuplicate = await this.importService.checkSplitChecksum(
-      attachmentId,
+    const isDuplicate = await this.importService.checkDuplicate(
+      parseInt(attachmentId, 10),
      body.pages,
    );
    return { isDuplicate };
@@ -187,6 +196,8 @@ export class EmailImportController {
      return result;
    } catch (err: any) {
      this.logger.error(`Error executing import: ${err.message}`);
+      // Gezielte Fehler (z.B. 409 bei Duplikaten) unverändert weiterreichen
+      if (err instanceof HttpException) throw err;
      throw new HttpException(err.message, HttpStatus.INTERNAL_SERVER_ERROR);
    }
  }
@@ -0,0 +1,234 @@
+import { Test, TestingModule } from '@nestjs/testing';
+import { ConflictException } from '@nestjs/common';
+import { getRepositoryToken } from '@nestjs/typeorm';
+import { ConfigService } from '@nestjs/config';
+import { PDFDocument } from 'pdf-lib';
+import { EmailImportService } from './email-import.service';
+import { Attachment } from '../database/entities/attachment.entity';
+import { Email } from '../database/entities/email.entity';
+import { Content } from '../database/entities/content.entity';
+import { CorrespondentEmailMapping } from '../database/entities/correspondent-email-mapping.entity';
+import { Task } from '../database/entities/task.entity';
+import { PaperlessService } from '../paperless/paperless.service';
+import { PdfService } from '../preprocessing/pdf.service';
+import { EmailPageCacheService } from './email-page-cache.service';
+import { ImapFolderService } from './imap-folder.service';
+
+async function pdfMitSeiten(anzahl: number): Promise<Buffer> {
+  const doc = await PDFDocument.create();
+  for (let i = 0; i < anzahl; i++) doc.addPage();
+  return Buffer.from(await doc.save());
+}
+
+describe('EmailImportService – Duplikatprüfung', () => {
+  let service: EmailImportService;
+  let taskQuery: {
+    innerJoin: jest.Mock;
+    where: jest.Mock;
+    andWhere: jest.Mock;
+    getCount: jest.Mock;
+  };
+  let attachmentRepo: { findOne: jest.Mock };
+  let contentRepo: { findOne: jest.Mock };
+  let paperlessService: {
+    checksumExists: jest.Mock;
+    uploadDocument: jest.Mock;
+  };
+
+  beforeEach(async () => {
+    taskQuery = {
+      innerJoin: jest.fn().mockReturnThis(),
+      where: jest.fn().mockReturnThis(),
+      andWhere: jest.fn().mockReturnThis(),
+      getCount: jest.fn().mockResolvedValue(0),
+    };
+    attachmentRepo = {
+      findOne: jest.fn().mockResolvedValue({
+        Id: 7,
+        Checksum: 'abc123',
+        FileName: 'rechnung.pdf',
+      }),
+    };
+    contentRepo = { findOne: jest.fn() };
+    paperlessService = {
+      checksumExists: jest.fn().mockResolvedValue(false),
+      uploadDocument: jest.fn(),
+    };
+
+    const module: TestingModule = await Test.createTestingModule({
+      providers: [
+        EmailImportService,
+        { provide: ConfigService, useValue: { get: jest.fn() } },
+        { provide: getRepositoryToken(Email), useValue: {} },
+        { provide: getRepositoryToken(Attachment), useValue: attachmentRepo },
+        { provide: getRepositoryToken(Content), useValue: contentRepo },
+        {
+          provide: getRepositoryToken(CorrespondentEmailMapping),
+          useValue: {},
+        },
+        {
+          provide: getRepositoryToken(Task),
+          useValue: {
+            createQueryBuilder: jest.fn().mockReturnValue(taskQuery),
+          },
+        },
+        { provide: PaperlessService, useValue: paperlessService },
+        { provide: PdfService, useValue: {} },
+        { provide: EmailPageCacheService, useValue: {} },
+        { provide: ImapFolderService, useValue: {} },
+      ],
+    }).compile();
+
+    service = module.get(EmailImportService);
+  });
+
+  describe('ganzer Anhang', () => {
+    it('erkennt ein Duplikat, dessen Import noch in Paperless verarbeitet wird', async () => {
+      taskQuery.getCount.mockResolvedValue(1);
+
+      await expect(service.checkDuplicate(7)).resolves.toBe(true);
+
+      expect(taskQuery.where).toHaveBeenCalledWith('a.Checksum = :checksum', {
+        checksum: 'abc123',
+      });
+      expect(taskQuery.andWhere).toHaveBeenCalledWith(
+        't.SourceAttachmentRange = :range',
+        { range: 'full' },
+      );
+      expect(paperlessService.checksumExists).not.toHaveBeenCalled();
+    });
+
+    it('erkennt ein Duplikat, das bereits in Paperless liegt', async () => {
+      paperlessService.checksumExists.mockResolvedValue(true);
+
+      await expect(service.checkDuplicate(7)).resolves.toBe(true);
+      expect(paperlessService.checksumExists).toHaveBeenCalledWith('abc123');
+    });
+
+    it('meldet kein Duplikat, wenn weder Import läuft noch Paperless es kennt', async () => {
+      await expect(service.checkDuplicate(7)).resolves.toBe(false);
+    });
+
+    it('meldet kein Duplikat für Anhänge ohne Checksumme', async () => {
+      attachmentRepo.findOne.mockResolvedValue({ Id: 7, Checksum: '' });
+
+      await expect(service.checkDuplicate(7)).resolves.toBe(false);
+      expect(taskQuery.getCount).not.toHaveBeenCalled();
+      expect(paperlessService.checksumExists).not.toHaveBeenCalled();
+    });
+  });
+
+  describe('Teilbereich (Split)', () => {
+    beforeEach(async () => {
+      contentRepo.findOne.mockResolvedValue({
+        Content1: await pdfMitSeiten(3),
+      });
+    });
+
+    it('sucht laufende Importe mit demselben Seitenbereich wie beim Upload', async () => {
+      taskQuery.getCount.mockResolvedValue(1);
+
+      // 999 steht im Assistenten für "bis zur letzten Seite"
+      await expect(
+        service.checkDuplicate(7, { start: 2, end: 999 }),
+      ).resolves.toBe(true);
+
+      expect(taskQuery.where).toHaveBeenCalledWith('a.Checksum = :checksum', {
+        checksum: 'abc123',
+      });
+      expect(taskQuery.andWhere).toHaveBeenCalledWith(
+        't.SourceAttachmentRange = :range',
+        { range: '2-3' },
+      );
+    });
+
+    it('prüft in Paperless die Checksumme des Teil-PDFs', async () => {
+      await expect(
+        service.checkDuplicate(7, { start: 1, end: 1 }),
+      ).resolves.toBe(false);
+
+      const [checksum] = paperlessService.checksumExists.mock
+        .calls[0] as string[];
+      expect(checksum).toMatch(/^[0-9a-f]{32}$/);
+      expect(checksum).not.toBe('abc123');
+    });
+  });
+
+  describe('executeImport', () => {
+    const importDaten = (splitRanges?: { start: number; end: number }[]) => ({
+      emailDate: '2026-09-25',
+      attachments: [
+        {
+          attachmentId: 7,
+          type: 'MAIN' as const,
+          splitRanges,
+          belegnummer: '2026-000123',
+        },
+      ],
+    });
+
+    it('bricht vor Upload und Belegnummer ab, wenn das Dokument schon vorhanden ist', async () => {
+      paperlessService.checksumExists.mockResolvedValue(true);
+
+      await expect(service.executeImport(importDaten())).rejects.toThrow(
+        ConflictException,
+      );
+      expect(paperlessService.uploadDocument).not.toHaveBeenCalled();
+    });
+
+    it('prüft bei geteilten Anhängen jeden Seitenbereich', async () => {
+      const check = jest
+        .spyOn(service, 'checkDuplicate')
+        .mockResolvedValue(true);
+
+      await expect(
+        service.executeImport(importDaten([{ start: 1, end: 2 }])),
+      ).rejects.toThrow(ConflictException);
+      expect(check).toHaveBeenCalledWith(7, { start: 1, end: 2 });
+    });
+
+    it('ignorierte Anhänge werden nicht geprüft', async () => {
+      const check = jest.spyOn(service, 'checkDuplicate');
+      jest
+        .spyOn(
+          service as never as { runImport: () => Promise<unknown> },
+          'runImport',
+        )
+        .mockResolvedValue({});
+      const daten = importDaten();
+      daten.attachments[0] = {
+        ...daten.attachments[0],
+        type: 'IGNORE' as never,
+      };
+
+      await service.executeImport(daten);
+      expect(check).not.toHaveBeenCalled();
+    });
+
+    it('lässt denselben Anhang nicht zweimal gleichzeitig importieren', async () => {
+      let ersterFertig!: () => void;
+      const run = jest
+        .spyOn(
+          service as never as { runImport: () => Promise<unknown> },
+          'runImport',
+        )
+        .mockImplementationOnce(
+          () => new Promise((resolve) => (ersterFertig = () => resolve({}))),
+        )
+        .mockResolvedValue({});
+
+      const erster = service.executeImport(importDaten());
+      await new Promise((r) => setImmediate(r));
+
+      await expect(service.executeImport(importDaten())).rejects.toThrow(
+        ConflictException,
+      );
+
+      ersterFertig();
+      await erster;
+      // Nach Abschluss ist die Sperre wieder aufgehoben
+      await service.executeImport(importDaten());
+      expect(run).toHaveBeenCalledTimes(2);
+    });
+  });
+});
@@ -1,4 +1,10 @@
-import { Injectable, Logger, HttpException, HttpStatus } from '@nestjs/common';
+import {
+  Injectable,
+  Logger,
+  HttpException,
+  HttpStatus,
+  ConflictException,
+} from '@nestjs/common';
 import { ConfigService } from '@nestjs/config';
 import { InjectRepository } from '@nestjs/typeorm';
 import { Repository } from 'typeorm';
@@ -31,6 +37,8 @@ function sanitizeFilename(name: string): string {
@Injectable()
 export class EmailImportService {
  private readonly logger = new Logger(EmailImportService.name);
+  // Anhänge (Checksumme + Seitenbereich), die gerade importiert werden
+  private readonly laufendeImporte = new Set<string>();
  private readonly importJobs = new Map<
    string,
    { message: string; done: boolean }
@@ -237,11 +245,29 @@ export class EmailImportService {
    }
  }

-  // --- Checksum Check for Split Documents ---
-  async checkSplitChecksum(
+  // --- Duplikatprüfung ---
+  /**
+   * Prüft, ob ein Anhang (oder mit `pages` ein Teilbereich davon) schon
+   * importiert wurde. Neben Paperless selbst zählen dabei auch Importe, die
+   * gerade noch in Paperless verarbeitet werden: Deren Checksumme kennt
+   * Paperless erst nach dem Consume, z.B. wenn dieselbe Mail zweimal
+   * weitergeleitet wurde und die erste gerade importiert ist.
+   */
+  async checkDuplicate(
    attachmentId: number,
-    pages: { start: number; end: number },
+    pages?: { start: number; end: number },
  ): Promise<boolean> {
+    const attachment = await this.attachmentRepo.findOne({
+      where: { Id: attachmentId },
+    });
+    if (!attachment) return false;
+
+    if (!pages) {
+      if (!attachment.Checksum) return false;
+      if (await this.hasPendingImport(attachment.Checksum, 'full')) return true;
+      return this.paperlessService.checksumExists(attachment.Checksum);
+    }
+
    const content = await this.contentRepo.findOne({
      where: { AttachmentEntityId: attachmentId },
    });
@@ -251,13 +277,21 @@ export class EmailImportService {
      ignoreEncryption: true,
    });
    const total = pdfDoc.getPageCount();
-    const startIdx = Math.max(1, pages.start) - 1;
-    const endIdx = Math.min(pages.end === 999 ? total : pages.end, total) - 1;
+    // Bereich genauso begrenzen wie executeImport, damit der Range-Key passt
+    const start = Math.max(1, pages.start);
+    const end = Math.min(pages.end === 999 ? total : pages.end, total);
+
+    if (
+      attachment.Checksum &&
+      (await this.hasPendingImport(attachment.Checksum, `${start}-${end}`))
+    ) {
+      return true;
+    }

    const sliced = await PDFDocument.create();
    const indices = Array.from(
-      { length: endIdx - startIdx + 1 },
-      (_, i) => startIdx + i,
+      { length: end - start + 1 },
+      (_, i) => start - 1 + i,
    );
    const copied = await sliced.copyPages(pdfDoc, indices);
    copied.forEach((p) => sliced.addPage(p));
@@ -269,6 +303,25 @@ export class EmailImportService {
    return this.paperlessService.checksumExists(checksum);
  }

+  /**
+   * Läuft für einen Anhang mit dieser Checksumme (egal aus welcher E-Mail)
+   * bereits ein Import desselben Seitenbereichs? Fehlgeschlagene Tasks löscht
+   * der Task-Processor, abgeschlossene findet Paperless selbst.
+   */
+  private async hasPendingImport(
+    checksum: string,
+    range: string,
+  ): Promise<boolean> {
+    const count = await this.taskRepo
+      .createQueryBuilder('t')
+      .innerJoin(Attachment, 'a', 'a.Id = t.SourceAttachmentID')
+      .where('a.Checksum = :checksum', { checksum })
+      .andWhere('t.SourceAttachmentRange = :range', { range })
+      .andWhere('(t.Fertig IS NULL OR t.Fertig = 0)')
+      .getCount();
+    return count > 0;
+  }
+
  // --- Print Preview ---
  async generatePrintPdf(
    attachmentId: number,
@@ -460,6 +513,11 @@ export class EmailImportService {
  }

  // --- Import Logic ---
+  /**
+   * Prüft unmittelbar vor dem Upload – und damit bevor eine Belegnummer
+   * festgelegt wird – nochmals auf Duplikate. Seit dem Öffnen des Assistenten
+   * kann dasselbe Dokument aus einer anderen E-Mail importiert worden sein.
+   */
  async executeImport(data: {
    jobId?: string;
    attachments: {
@@ -478,6 +536,86 @@ export class EmailImportService {
      belegnummer?: string;
    }[];
    emailDate: string;
+  }): Promise<{ success: boolean; results: any[] }> {
+    const sperren = await this.sperreImporte(data.attachments);
+    try {
+      return await this.runImport(data);
+    } finally {
+      sperren.forEach((key) => this.laufendeImporte.delete(key));
+    }
+  }
+
+  /**
+   * Reserviert jeden zu importierenden Anhang (bzw. Seitenbereich) für diesen
+   * Import und wirft einen ConflictException, wenn er schon vorhanden ist oder
+   * parallel importiert wird. Die Reservierung schließt die Lücke zwischen
+   * Prüfung und Upload, in der Paperless noch nichts davon weiß.
+   */
+  private async sperreImporte(
+    attachments: {
+      attachmentId: number;
+      type: string;
+      splitRanges?: { start: number; end: number }[];
+    }[],
+  ): Promise<string[]> {
+    const sperren: string[] = [];
+    try {
+      for (const att of attachments) {
+        if (att.type === 'IGNORE') continue;
+        const attachment = await this.attachmentRepo.findOne({
+          where: { Id: att.attachmentId },
+        });
+        if (!attachment) continue;
+
+        const ranges = att.splitRanges?.length ? att.splitRanges : [undefined];
+        for (const range of ranges) {
+          const bereich = range ? `${range.start}-${range.end}` : 'full';
+          const key = `${attachment.Checksum || `id:${attachment.Id}`}:${bereich}`;
+          const bezeichnung = range
+            ? `„${attachment.FileName}" (Seiten ${bereich})`
+            : `„${attachment.FileName}"`;
+
+          // has + add ohne await dazwischen: atomar gegenüber parallelen Importen
+          if (this.laufendeImporte.has(key)) {
+            throw new ConflictException(
+              `${bezeichnung} wird gerade bereits importiert.`,
+            );
+          }
+          this.laufendeImporte.add(key);
+          sperren.push(key);
+
+          if (await this.checkDuplicate(att.attachmentId, range)) {
+            throw new ConflictException(
+              `${bezeichnung} ist bereits in Paperless vorhanden oder wird dort gerade verarbeitet.`,
+            );
+          }
+        }
+      }
+      return sperren;
+    } catch (err) {
+      sperren.forEach((key) => this.laufendeImporte.delete(key));
+      throw err;
+    }
+  }
+
+  private async runImport(data: {
+    jobId?: string;
+    attachments: {
+      attachmentId: number;
+      type: 'MAIN' | 'ATTACHMENT' | 'IGNORE';
+      paperlessCorrespondentId?: number | null;
+      parentDocumentId?: number | null;
+      splitRanges?: { start: number; end: number }[];
+      barcode?: {
+        x: number;
+        y: number;
+        nummer: string;
+        datum: string;
+        jahr: string;
+      };
+      belegnummer?: string;
+    }[];
+    emailDate: string;
  }): Promise<{ success: boolean; results: any[] }> {
    const tempDir = await fs.mkdtemp(
      path.join(os.tmpdir(), 'paperless-mail-import-'),
@@ -500,7 +638,9 @@ export class EmailImportService {
        if (!content) continue;

        const originalPdfBytes = content.Content1;
-        const baseFilename = sanitizeFilename(attachmentEntity.FileName.replace(/\.pdf$/i, ''));
+        const baseFilename = sanitizeFilename(
+          attachmentEntity.FileName.replace(/\.pdf$/i, ''),
+        );
        const paperlessIds: any = {};

        const uploadPromises = [];