feat: Webhook-Warteschlange für sequenzielle, deduplizierte Verarbeitung
Build and Push Multi-Platform Images / build-and-push (push) Successful in 37s

Der Webhook reiht eingehende Dokument-IDs nur noch in eine Warteschlange ein
und antwortet sofort (status "queued"). Ein separater Intervall-Prozess
(WebhookQueueService) arbeitet die IDs nacheinander ab – ohne Überschneidung,
falls ein Dokument mehrfach kurz hintereinander gespeichert wird:

- Jede ID kommt nur einmal in der Liste vor (Dedup).
- Beim Verarbeitungsstart wird die ID sofort entfernt; ein erneutes Feuern
  während der Verarbeitung reiht sie wieder ein (ein weiterer Lauf folgt).
- Es läuft immer nur eine Verarbeitung gleichzeitig (isProcessing-Guard).
- Prüfintervall sehr kurz (WEBHOOK_QUEUE_INTERVAL_MS, Default 1000 ms).

Status-Recording (last_webhook_call) + GET /api/webhook/status wandern in den
Queue-Service und liefern zusätzlich die aktuelle Warteschlangen-Größe.
Frontend-Tab zeigt Status "In Warteschlange" und die Queue-Größe an.
Unit-Test deckt Dedup, Remove-on-start, Re-Enqueue und No-Parallel ab.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
bjoernpoettkerandClaude Opus 4.8 committed 2026-06-29 17:09:23 +02:00
1 parent b77a283ab2
commit 66a2cccd20
6 files changed
+299 -106

No files matched your search

@@ -9,11 +9,8 @@ import {
UseGuards,
BadRequestException,
} from '@nestjs/common';
import { InjectRepository } from '@nestjs/typeorm';
import { Repository } from 'typeorm';
import { ApiKeyGuard } from '../auth/api-key.guard';
import { PaperlessProcessorService } from '../paperless/paperless-processor.service';
import { Setting } from '../database/entities/setting.entity';
import { WebhookQueueService } from './webhook-queue.service';
export interface PaperlessWebhookPayload {
doc_url?: string;
@@ -22,18 +19,11 @@ export interface PaperlessWebhookPayload {
[key: string]: any;
}
// Tag (Schlüssel) des Settings-Eintrags, der den letzten Webhook-Aufruf festhält.
const LAST_WEBHOOK_CALL_TAG = 'last_webhook_call';
@Controller('api/webhook')
export class WebhookController {
private readonly logger = new Logger(WebhookController.name);
constructor(
private readonly paperlessProcessor: PaperlessProcessorService,
@InjectRepository(Setting)
private readonly settingRepo: Repository<Setting>,
) {}
constructor(private readonly webhookQueue: WebhookQueueService) {}
@UseGuards(ApiKeyGuard)
@Post('paperless')
@@ -44,7 +34,7 @@ export class WebhookController {
this.logger.warn(
`Webhook ohne ermittelbare Dokument-ID: ${JSON.stringify(payload)}`,
);
await this.recordWebhookCall({
await this.webhookQueue.recordStatus({
documentId: null,
action: payload.action,
status: 'bad-request',
@@ -55,54 +45,22 @@ export class WebhookController {
}
this.logger.log(
`Webhook: action=${payload.action}, document=${documentId}`,
`Webhook: action=${payload.action}, document=${documentId} → eingereiht`,
);
try {
const result =
await this.paperlessProcessor.processDocumentById(documentId);
const status = result.processed ? 'processed' : 'skipped';
await this.recordWebhookCall({
documentId,
action: payload.action,
status,
reason: result.reason,
});
return { status, reason: result.reason };
} catch (err) {
// Fehler tolerieren (wie der bisherige Cron) und 200 zurückgeben,
// damit Paperless keine Retry-Schleife startet.
const message = err instanceof Error ? err.message : String(err);
this.logger.error(
`Fehler bei Webhook-Verarbeitung von Dokument ${documentId}: ${message}`,
);
await this.recordWebhookCall({
documentId,
action: payload.action,
status: 'error',
message,
});
return { status: 'error', message };
}
// Sofort einreihen und antworten; die eigentliche Verarbeitung übernimmt
// der separate Queue-Prozess (sequenziell, dedupliziert).
this.webhookQueue.enqueue(documentId, payload.action);
return { status: 'queued', documentId };
}
/**
* Liefert Informationen zum letzten Webhook-Aufruf (Zeitpunkt, Dokument,
* Ergebnis). Über die globalen Guards per JWT oder API-Key zugänglich.
* Liefert Informationen zum letzten Webhook-Vorgang (Zeitpunkt, Dokument,
* Ergebnis) und die aktuelle Warteschlangen-Größe. Über die globalen Guards
* per JWT oder API-Key zugänglich.
*/
@Get('status')
async getWebhookStatus(): Promise<{ lastCall: unknown }> {
const setting = await this.settingRepo.findOneBy({
Tag: LAST_WEBHOOK_CALL_TAG,
});
if (!setting?.Wert) {
return { lastCall: null };
}
try {
return { lastCall: JSON.parse(setting.Wert) };
} catch {
return { lastCall: { raw: setting.Wert } };
}
async getWebhookStatus(): Promise<{ lastCall: unknown; queueSize: number }> {
return this.webhookQueue.getStatus();
}
/**
@@ -124,47 +82,4 @@ export class WebhookController {
}
return null;
}
/**
* Hält den letzten Webhook-Aufruf in der Settings-Tabelle fest
* (Tag "last_webhook_call"). Fehler beim Speichern werden nur geloggt und
* beeinflussen die Webhook-Antwort nicht.
*/
private async recordWebhookCall(info: {
documentId: number | null;
action?: string;
status: string;
reason?: string;
message?: string;
}): Promise<void> {
try {
const value = JSON.stringify({
at: new Date().toISOString(),
documentId: info.documentId,
action: info.action ?? null,
status: info.status,
...(info.reason ? { reason: info.reason } : {}),
...(info.message ? { message: info.message.slice(0, 100) } : {}),
});
let setting = await this.settingRepo.findOneBy({
Tag: LAST_WEBHOOK_CALL_TAG,
});
if (!setting) {
setting = this.settingRepo.create({
Typ: 0,
Tag: LAST_WEBHOOK_CALL_TAG,
Wert: value,
});
} else {
setting.Wert = value;
}
await this.settingRepo.save(setting);
} catch (err) {
this.logger.error(
'Konnte letzten Webhook-Aufruf nicht in den Settings speichern',
err instanceof Error ? err.stack : String(err),
);
}
}
}