Neu veröffentlicht: E-Commerce mit Power Pages, Stripe & Analytics

· David Göschel · Architektur  · 7 minuten Lesezeit

Capture Intake als eigenes Modul, weniger Orchestrierung im Express Endpoint

Mein /ingest Endpoint wusste zu viel über Validierung, Content Hash, PostgreSQL, S3, BullMQ und Rollback. Ich habe diese Regeln in ein eigenes Capture Intake Modul gezogen, ohne das Verhalten zu ändern.

Mein /ingest Endpoint wusste zu viel über Validierung, Content Hash, PostgreSQL, S3, BullMQ und Rollback. Ich habe diese Regeln in ein eigenes Capture Intake Modul gezogen, ohne das Verhalten zu ändern.

Inhalt

Warum ich den Ingest Endpoint nochmal angefasst habe

Der Capture Flow lief. Der Endpoint nahm den Request an, schrieb den Datensatz nach PostgreSQL, lud das Bild nach S3, stellte den Job in BullMQ und gab 202 zurück. Funktional war das in Ordnung.

Beim Lesen des Codes fiel mir trotzdem auf, dass routes/ingest.ts zu viel über die eigentliche Capture-Annahme wusste. Die Route validierte den Payload, berechnete den Content Hash, normalisierte Bildgrößen, behandelte Duplikate, orchestrierte den Upload und führte im Fehlerfall den Rollback aus. Für einen Express Endpoint war das zu viel Verantwortung in einer einzigen Datei.

So ein Endpoint ist anfangs bequem. Später wird er teuer. Jede weitere Regel landet an derselben Stelle. Jeder neue Fehlerfall erweitert dieselbe Orchestrierung. Jede Änderung am Persistenzpfad zwingt dich zurück in die Route.

Woran die Route zu viel wusste

Vor dem Refactoring sah der relevante Pfad grob so aus:

router.post('/', async (req: Request, res: Response) => {
  const payload = req.body as PlatformAnalysisPayload;
  const userId = req.auth!.userId;
  payload.capturedBy.userId = userId;

  const captureId = crypto.randomUUID();
  const embeddingText = buildEmbeddingText(payload);
  const contentHash = computeContentHash(embeddingText);

  await ensureUser(userId);
  await insertCapture(buildInsertCaptureInput(payload, { id: captureId, userId, contentHash }));

  const blobName = await uploadImage(payload.image.base64, payload.image.mimeType, userId);
  await updateCaptureBlobName(captureId, blobName);
  await ingestionQueue.add('ingest', { captureId, userId }, { jobId: captureId });

  res.status(202).json({ captureId });
});

Der Code war nicht falsch. Er lag nur am falschen Ort. Die Route war nicht mehr nur HTTP Adapter, sondern bereits ein halbes Fachmodul.

Das Problem daran ist nicht nur Dateigröße. Das eigentliche Problem ist Wissen. Sobald eine Route wissen muss, in welcher Reihenfolge PostgreSQL, S3 und BullMQ angesprochen werden, ist die Fachregel nicht mehr konzentriert. Damit rutscht die Regel an den Rand des Systems.

Welche Regeln gleich bleiben mussten

Beim Umbau wollte ich keine versteckte Verhaltensänderung einführen. Vier Regeln mussten exakt gleich bleiben.

  1. Ein doppelter Capture bleibt ein 409 mit DUPLICATE_CAPTURE.
  2. Wenn S3 Upload oder Queue Handoff scheitern, wird weiter vollständig aufgeräumt.
  3. Der Quota Zähler bleibt Best Effort und blockiert keinen erfolgreichen Capture.
  4. Die JWT Prüfung bleibt im Auth Middleware Pfad und wandert nicht in das neue Modul.

Gerade Punkt zwei war wichtig. Ein Refactoring wirkt schnell sauber, wenn man Aufräumlogik auslagert. Es ist aber nur dann wirklich sauber, wenn die Aufräumlogik danach noch dieselben Konsistenzregeln durchsetzt wie vorher.

Das neue Capture Intake Modul

Ich habe die komplette Capture-Annahme in backend/src/services/capture-intake.ts gezogen. Die Route bleibt jetzt dünn und übersetzt nur noch HTTP in einen Modulaufruf.

router.post('/', async (req: Request, res: Response) => {
  const payload = req.body as PlatformAnalysisPayload;

  try {
    const { captureId } = await acceptCapture({
      payload,
      userId: req.auth!.userId,
    });

    res.status(202).json({ captureId });
  } catch (error: unknown) {
    if (error instanceof InvalidCapturePayloadError) {
      res.status(400).json({ error: error.message });
      return;
    }

    if (error instanceof DuplicateCaptureError) {
      res.status(409).json({
        error: error.message,
        code: 'DUPLICATE_CAPTURE',
        captureId: error.captureId,
      });
      return;
    }

    const message = error instanceof Error ? error.message : 'Unknown error';
    res.status(500).json({ error: message });
  }
});

Die eigentliche Fachregel sitzt jetzt in acceptCapture():

export async function acceptCapture({
  payload,
  userId,
}: CaptureIntakeInput): Promise<CaptureIntakeResult> {
  validateCapturePayload(payload);

  const captureId = crypto.randomUUID();
  const embeddingText = buildEmbeddingText(capturePayload);
  const contentHash = computeContentHash(embeddingText);

  await ensureUser(userId);
  await insertCapture(...);

  const blobName = await uploadImage(...);
  await updateCaptureBlobName(captureId, blobName);
  await ingestionQueue.add('ingest', { captureId, userId }, { jobId: captureId });

  incrementCaptures(userId).catch(...);

  return { captureId };
}

Der große Unterschied ist nicht, dass jetzt weniger Code existiert. Der Unterschied ist, dass die Capture-Annahme jetzt einen benannten Ort hat. Wenn ich später am Verhalten etwas ändere, suche ich nicht mehr in einem Route Handler nach verstreuter Persistenzlogik. Ich gehe direkt in das Modul, das genau diese Verantwortung trägt.

Warum das Refactoring in der Praxis etwas bringt

Für mich hat der Umbau drei konkrete Folgen.

Erstens bleibt die Express Route lesbar. Wenn ich nur verstehen will, wie HTTP Fehler in Statuscodes übersetzt werden, sehe ich genau das und nicht gleichzeitig den halben Persistenzpfad.

Zweitens hängt der eigentliche Capture Pfad nicht mehr an Express. Wenn ich denselben Ablauf später aus einem anderen Einstiegspunkt aufrufen will, etwa aus einem Import oder Replay Pfad, muss ich nicht mit Request und Response Objekten arbeiten.

Drittens ist der Rollback jetzt dort, wo auch die Gefahr entsteht. PostgreSQL, S3 und BullMQ werden in einem zusammenhängenden Ablauf aufgerufen. Der gleiche Ort weiß auch, wie aufgeräumt wird, wenn einer dieser Schritte scheitert. Das ist die Stelle, an der ich im Fehlerfall wirklich nachsehen will.

Was sich bewusst nicht geändert hat

Ich habe das Modul nicht künstlich allgemeiner gemacht, als es heute sein muss. Es ist noch kein universeller Import Pfad. Es ist keine abstrakte Pipeline Engine. Es ist auch keine neue Schicht für alles, was mit Ingest zu tun hat.

Das Modul bedient genau einen aktuellen Aufrufer, nämlich POST /ingest. Das reicht für diesen Schritt. Die Tiefe kommt hier nicht aus maximaler Flexibilität. Der Gewinn entsteht daraus, dass eine konkrete Regel nun vollständig an einem Ort sitzt.

Ich habe auch die bestehende Reihenfolge bewusst beibehalten. Erst wird der Capture Datensatz angelegt. Dann folgt der Bild Upload. Danach wird der Blob Name persistiert. Danach geht der Job in die Queue. So bleibt das Verhalten für den laufenden Stack stabil.

Nach dem Test

Nach dem Umbau habe ich den Capture Ablauf wieder durchgeklickt. Der Request kam weiter mit 202 zurück, der Datensatz wurde gespeichert und der Worker lief wie vorher an. Genau das wollte ich sehen.

Das war für mich der wichtigste Punkt an diesem Refactoring. Nicht nur der Code ist klarer geworden. Der produktive Pfad hat sich dabei nicht heimlich verbogen.

Alle Artikel der Serie

  1. Vision und Systemübersicht: Chrome Extension, RAG-Architektur, Projekthintergrund: Artikel lesen
  2. RAG-System Aufbau: Qdrant, Embeddings, Cosine-Ähnlichkeit in TypeScript: Artikel lesen
  3. AI Provider Abstraktion: Ollama vs. OpenAI, Interface-Design, kein Vendor-Lock-in: Artikel lesen
  4. Chrome Extension MV3: Drei isolierte Laufzeitkontexte, Message Passing, Strategy Pattern: Artikel lesen
  5. Docker Compose Strategie: Override-Pattern, von lokal zu Azure: Artikel lesen
  6. Ollama lokal vs. Docker: Die Entscheidung und ihre Konsequenzen: Artikel lesen
  7. Ollama Auto-Pull Entrypoint: Automatisiertes Modell-Setup beim Container-Start: Artikel lesen
  8. tsconfig und Vite: Node16 vs. bundler, warum Vite eigene Regeln hat: Artikel lesen
  9. Instagram Caption mit MutationObserver vollständig laden: Artikel lesen
  10. Chrome Extension Foundation mit Health-Dot und Retry-Queue: Artikel lesen
  11. Phase 2 Features: Shadow DOM Overlay, Tailwind v4, Duplicate Detection: Artikel lesen
  12. Race Condition bei der Plattformerkennung: Wie ein UI-Event die Instagram-Erkennung bricht: Artikel lesen
  13. PostId-Extraktion in zwei Instagram-Layouts: querySelector vs. Ancestor-Traversal: Artikel lesen
  14. Instagram Karussell vollständig erfassen mit MutationObserver: Lazy-Loading, Observer-before-click, Timeout-Fallback: Artikel lesen
  15. Notiz und Tags beim Screenshot-Speichern: Artikel lesen
  16. Instagram Tastatur-Shortcuts blockieren Chrome Extension Eingaben: Artikel lesen
  17. Lowercase-Normalisierung und Duplikat-Erkennung im Tag-Input: Artikel lesen
  18. Zitadel Login V2 in Docker Compose: drei versteckte Fehler: Artikel lesen
  19. PKCE OAuth in einer Chrome MV3 Extension: Artikel lesen
  20. React Frontend mit react-oidc-context und Zitadel: Artikel lesen
  21. Vite Build-Time-Umgebungsvariablen in Docker: Artikel lesen
  22. Event-Driven Ingestion mit BullMQ und Redis: Artikel lesen
  23. MinIO statt Azurite: S3-kompatible Objektspeicherung lokal und auf Hetzner: Artikel lesen
  24. access_token, id_token und der Userinfo-Endpoint: was wohin gehört:: Artikel lesen
  25. Qdrant Multi-Tenancy: Pro Nutzer eine eigene Collection: Artikel lesen
  26. Wenn Backend und Frontend unterschiedliche Typen kennen: Artikel lesen
  27. Zitadel Bootstrap entfernt: Host-Header-Bug und manuelles Setup: Artikel lesen
  28. Backend Code Review: sechs Probleme vor dem Launch behoben: Artikel lesen
  29. Traefik statt NGINX: Reverse Proxy für einen wachsenden Docker-Compose-Stack: Artikel lesen
  30. Zweischichtiges Rate Limiting: Traefik und express-rate-limit mit Redis: Artikel lesen
  31. DSGVO Art. 17 korrekt implementieren: Promise.allSettled und Export-Batching: Artikel lesen
  32. Embedding-Modell-Lock-in: Warum mxbai-embed-large eine Produktionsentscheidung für immer ist: Artikel lesen
  33. Docker Volumes in Produktion: Named Volumes, Bind Mounts und der Hetzner-Volume-Trick: Artikel lesen
  34. Zwei Sicherheitslücken vor dem Launch: Redis ohne Auth und ein offener Qdrant-Admin-Port: Artikel lesen
  35. Traefik als einziger Einstiegspunkt im Docker Compose Stack: Artikel lesen
  36. Zitadel hinter Traefik richtig verdrahten mit Issuer, JWKS und Login V2: Artikel lesen
  37. Frontend reparieren wenn der nginx Healthcheck an localhost scheitert: Artikel lesen
  38. Observability für meinen Docker Compose Stack mit Bull Board und Dozzle: Artikel lesen
  39. Qdrant Dashboard sicher öffnen mit lokalem Traefik und SSH Tunnel: Artikel lesen
  40. Diagnose: Warum mein Chunking trotz Tokenisierung noch scheiterte: Artikel lesen
  41. Entscheidung: Warum ich den Chunk auf 1500 Tokens gesetzt habe: Artikel lesen
  42. Implementierung: Wie ich den Embedding Workflow in mehrere saubere Schritte zerlegt habe: Artikel lesen
  43. Validierung: Wie ich Chunking, Speicherung und Suche wieder zusammenbringe: Artikel lesen
  44. ‘Capture Intake als eigenes Modul, weniger Orchestrierung im Express Endpoint’ (dieser Artikel)

Du arbeitest gerade an einem Express Backend, in dem ein Endpoint langsam zu viel über Persistenz und Rollback weiß? Lass uns das gemeinsam einschätzen.

Teilen:
Zurück zum Blog

Ähnliche Beiträge

Alle Beiträge ansehen
Event-Driven Ingestion mit BullMQ und Redis

Event-Driven Ingestion mit BullMQ und Redis

POST /ingest blockierte die Extension, bis Embedding und Qdrant-Upsert fertig waren. Mit BullMQ und Redis wird der Ingest asynchron: 202 sofort, Verarbeitung im Hintergrund, Statusabfrage über GET /captures/:id/status.