# Datenflüsse

> Sequenzdiagramme für Erkennung (Timer und Durable), Verarbeitung, Response-Jobs und Fehlerpfad; Persistenz der Daten; Watermark-Logik; DeadLetterMover.

---

Site [llms.txt](/llms.txt)

---

Die konzeptuelle Beschreibung steht unter [Wie es funktioniert](/docs/ueberblick/wie-es-funktioniert/).
Hier folgen die Abläufe auf Klassenebene.

## Erkennung (Timer)

```mermaid
sequenceDiagram
    autonumber
    participant Timer as GetChanges (Timer)
    participant Init as InitializationService
    participant PJ as ProcessJobsForRetrievalInOrderService
    participant Job as JobService
    participant WM as WatermarkService
    participant Conn as Source-Connector
    participant SBS as ServiceBusService
    participant SB as Service Bus

    Timer->>Init: EnsureInitializedAsync()
    Note over Init: einmalig: Storage → Connections → Mappings → Jobs
    Timer->>PJ: Execute()
    PJ->>Job: GetJobsToRun()
    Note over Job,WM: GetAllWatermarks() (Bulk) für LastCronRun
    loop je Job nach Order
        PJ->>WM: GetWatermarkByJobName(job)
        PJ->>Conn: RetrieveChangesAsync(mapping, job, watermark, filter)
        loop seitenweise
            Conn-->>PJ: QueueMessage (serialisiert)
            PJ->>SBS: AddMessageOrSendAsync(msg, priority)
            alt Batch voll (200) oder Größenlimit
                SBS->>SB: SendMessagesAsync(batch)
            end
            PJ->>PJ: lastChangeTracker = msg.ChangeTracker
        end
        PJ->>SBS: FlushBatch()
        PJ->>WM: UpdateWatermarkByJobName(newWatermark, lastCronRun)
    end
```

- Die Serialisierung (Key-Aufbau → Feld-Mappings → Source-Operationen / Resolver) passiert
  im Connector (`ConnectorBase.Serialize`).
- Die Watermark wird **erst am Ende** eines Jobs geschrieben. Bricht der Lauf vorher ab,
  wird beim nächsten Lauf erneut ab dem alten Stand gelesen; `SkipOutdatedMessages` fängt
  Doppel-Nachrichten im Ziel ab.
- Bei reinem Zeitstempel-Vorrücken ohne fachliche Änderung wird `JobActionType.TimeStamp`
  verwendet.
- `MetricsService` zählt `ChangesDetected` je Job.

## Erkennung (Durable)

```mermaid
sequenceDiagram
    autonumber
    participant HTTP as StartGetChangesOrchestration
    participant Orch as GetChangesOrchestrator
    participant State as JobState (Entity)
    participant Act as GetChangesActivity
    participant Conn as Source-Connector
    participant SBS as ServiceBusService

    HTTP-->>HTTP: 202 + InstanceId
    loop je Job aus GetJobsToRun
        Orch->>State: InProgress(jobName)?
        alt in Bearbeitung
            Orch->>Orch: überspringen
        else
            Orch->>State: Add(jobName)
            loop do … while HasMoreRecord
                Orch->>Act: CallActivity(JobActivityParam)
                Act->>Conn: RetrieveChangesAsync(…, MaxBatchSize=3)
                Act->>SBS: SendMessage(record, priority)
                Act-->>Orch: Page, Cookie, Watermark, HasMoreRecord
            end
            Orch->>State: Remove(jobName)
        end
    end
```

Der Durable-Pfad persistiert die Watermark nicht und ist im Timer-Betrieb nicht aktiv.

## Verarbeitung (Service Bus → Zielsystem)

```mermaid
sequenceDiagram
    autonumber
    participant SB as Service Bus Queue
    participant PQI as Process*QueueItems
    participant Init as InitializationService
    participant PQM as ProcessQueueMessageService
    participant Conn as Target-Connector
    participant SBS as ServiceBusService

    SB->>PQI: Nachricht
    PQI->>PQI: Lock-Renewal-Task starten (30 s)
    PQI->>Init: EnsureInitializedAsync()
    PQI->>PQM: Execute(busMessage)
    PQM->>PQM: QueueMessage deserialisieren, Job + Mapping laden
    alt DeliveryCount > 1
        PQM->>PQM: Backoff DeliveryCount² · 1000 ms
    end
    alt SkipOutdatedMessages und Quelle älter als Ziel
        PQM->>SBS: SendMessage → sbqignored
    else NewUpdate
        PQM->>Conn: Upsert(message, mapping[, responseMapping, job])
        opt Response vorhanden
            PQM->>SBS: SendMessage(response, priority)
        end
    else Delete
        PQM->>Conn: Delete(message, mapping)
    else InvokeAction
        PQM->>Conn: InvokeAction(message, job)
    end
    alt Erfolg
        PQI->>SB: CompleteMessageAsync()
    else TransitiveException
        PQM->>SBS: SendMessage → sbqtransitive*
        PQI->>SB: CompleteMessageAsync()
    else andere Exception
        PQI->>SB: AbandonMessageAsync(+ Error-Properties)
        Note over SB: nach maxDeliveryCount ⇒ Dead-Letter
    end
    PQI->>PQI: Lock-Renewal stoppen
```

`MetricsService` zählt `MessagesProcessed` mit dem Ausgang (Success, Transitive, Ignored,
Failed).

## Response-Job

```mermaid
sequenceDiagram
    autonumber
    participant ProcA as Process*QueueItems (Job A: CRM → BC)
    participant BC as Target-Connector BC
    participant SB as Service Bus
    participant ProcB as Process*QueueItems (ResponseJob: BC → CRM)
    participant CRM as Target-Connector CRM

    ProcA->>BC: Upsert(message, mapping, responseMapping, job)
    BC-->>ProcA: QueueMessage mit erzeugtem Ziel-Schlüssel
    ProcA->>SB: SendMessage(responseMessage, priority)
    SB->>ProcB: Response-Nachricht
    ProcB->>CRM: Upsert (z. B. BC-Belegnummer zurückschreiben)
```

## Fehlerpfad

```mermaid
flowchart TD
    MSG["Queue-Nachricht"] --> PROC["ProcessQueueMessageService.Execute"]
    PROC -->|Erfolg| DONE["Complete"]
    PROC -->|TransitiveException| TRANS["sbqtransitive*<br/>kein Trigger, liegt bis TTL"]
    PROC -->|veraltet| IGN["sbqignored<br/>kein Trigger"]
    PROC -->|andere Exception| AB["Abandon + Error-Properties"]
    AB --> RETRY{"DeliveryCount<br/>< maxDeliveryCount?"}
    RETRY -->|ja| MSG
    RETRY -->|nein| DLQ["Dead-Letter-Subqueue"]
    DLQ -.->|manuell| DLM["DeadLetterMover"]
    TRANS -.->|manuell| DLM
    DLM -.->|Resubmit ohne Error-Properties| MSG
```

Transitive- und Ignored-Queues werden **nicht** automatisch weiterverarbeitet. Sie dienen
der Isolation und der Analyse. Ein Requeue erfolgt über den `DeadLetterMover`.

### DeadLetterMover

Konsolenanwendung unter `src/DeadLetterMover`. Sie liest eine Queue oder deren
`$DeadLetterQueue`, entscheidet anhand der `ErrorMessage` per Muster (Resubmit, None, Skip)
und sendet Nachrichten ohne Error-Properties zurück in die Original-Queue, dedupliziert über
`SourceKey`. Connection-String, Queue-Name und der Schalter `UseDeadletter` stehen als
Konstanten im Code und werden vor dem Lauf angepasst. Der Connection-String darf nicht
eingecheckt werden.

## Persistenz

| Datenart | Ort | Lebensdauer |
|---|---|---|
| Konfiguration | Table Storage `connections`, `jobs`, `mappings` | persistent, manuell gepflegt |
| Watermarks (`Watermark`, `LastCronRun`) | Table Storage `watermark` | persistent je Job |
| Nachrichten | Service-Bus-Queues | transient, TTL `P14D` |
| Durable-State | Azure Storage `DurableFunctionsHub` | laufzeitgebunden |
| Telemetrie | Application Insights | 90 Tage |
| Fachdaten | nur Quell- und Zielsysteme | – |

## Watermark-Logik

Die Wertlogik liegt in `WatermarkValue` (`Models/Common`), der Service persistiert nur.

- `GetWatermarkByJobName` liefert bei fehlendem Eintrag eine nicht-persistierte Entity;
  `LastCronRun == null` bedeutet „Job ist fällig“.
- `GetAllWatermarks` ist eine Bulk-Abfrage auf PartitionKey `watermark` und wird vom
  Scheduler genutzt, damit nicht je Job eine Einzelabfrage anfällt.
- `WatermarkValue.Initialize(type, stored)` setzt Typ-Defaults (`date` → `0001-01-01`,
  `datetime` → `0001-01-01T00:00:00Z`). Der Typ wird getrimmt und case-insensitiv geparst;
  ein unbekannter Typ belässt den gespeicherten Wert und erzeugt eine Warnung.
- `Advance(changeTracker)` vergleicht typabhängig mit `InvariantCulture` und
  `RoundtripKind`.
- `UpdateTableEntry` schreibt **ohne ETag** (Last-Writer-Wins). Der HTTP-Endpunkt
  `ExecuteJob` kann parallel zum Timer laufen; ein veralteter Stand ist immer älter und
  führt zu Wiederholungen, nicht zu Lücken.
- Reset: Feld `Watermark` in der Tabelle leeren → vollständiger Neulauf. Anwendersicht:
  [Watermarks](/docs/konfiguration/watermarks/).