Architektur im Detail

• 1: Systemkontext und Prinzipien
• 2: Infrastruktur (Bicep)
• 3: Anwendungsarchitektur
• 4: API- und Nachrichten-Referenz
• 5: Datenflüsse
• 6: Abhängigkeiten
• 7: Fähigkeiten-Katalog

1 - Systemkontext und Prinzipien

Zweck

Die dsyr.DataBridge ist eine serverlose Datenintegrations-Bridge auf Azure Functions (.NET 10, Isolated Worker). Sie synchronisiert Datensätze nahezu in Echtzeit zwischen heterogenen Geschäftssystemen, ohne diese direkt zu koppeln. Quelle und Ziel sind über Azure Service Bus entkoppelt, Änderungen werden über Watermarks change-getrieben übertragen.

Die Bridge hält keine Fachdaten dauerhaft. Sie speichert nur Konfiguration und Synchronisationsstände in Azure Table Storage; Nachrichten liegen transient im Service Bus.

Hauptfunktionen

Zielgruppen

Architekturprinzipien

Die zwei Phasen

flowchart LR
    SRC["Quellsystem"] -->|"1. Erkennung (Pull)"| FUNC1["GetChanges<br/>oder Orchestrator + Activity"]
    FUNC1 -->|"QueueMessage"| SB[("Service Bus<br/>Low / Standard / High")]
    SB -->|"2. Verarbeitung (Push)"| FUNC2["Process*QueueItems"]
    FUNC2 -->|"Upsert / Delete / Action"| TGT["Zielsystem"]
    FUNC1 -.->|"Watermark"| WM[("watermark")]
    FUNC2 -.->|"Response-Job / Transitive / Ignored"| SB

• Erkennung: GetChanges (Timer) liest Änderungen aus den Quellsystemen und enqueued sie. Die Durable-Orchestrierung ist eine Alternative, die nur per HTTP gestartet wird und im Regelbetrieb nicht läuft.
• Verarbeitung: Drei Service-Bus-Trigger (je Priorität) dequeuen und schreiben ins Zielsystem.

Diese Entkopplung erlaubt unabhängige Skalierung beider Seiten und überbrückt temporäre Ausfälle des Zielsystems. Details in Datenflüsse.

Serverless-Modell

• Azure Functions v4, .NET 10, Isolated Worker (FUNCTIONS_WORKER_RUNTIME=dotnet-isolated).
• App Service Plan Y1 (Consumption, Windows) als Standard, optional EP1–EP3 über den Bicep-Parameter appServiceSku.
• functionAppScaleLimit begrenzt die Instanzen (Default 12).
• Durable Functions mit Azure-Storage-Provider (DurableFunctionsHub).
• Auslieferung als Run-from-Package-Zip; alternativ als Container-Image (siehe Container-Betrieb).

Betriebsmodell

• Trigger-gesteuert: Timer, Service Bus, HTTP, Durable.
• Single-Message-Processing: autoCompleteMessages = false, manuelles Complete/Abandon; maxConcurrentCalls = 12 je Instanz.
• Function-Timeout: 10 Minuten (host.json).
• Einzelne Functions abschaltbar über AzureWebJobs.<Name>.Disabled; Bicep setzt das über den Parameter disabledFunctions.
• Konfigurations-Cache pro Prozess: Änderungen an den Tabellen greifen erst nach einem Neustart der Function App.

Skalierungsmodell

Abgrenzung

• Keine eigene Benutzeroberfläche. Konfiguration erfolgt in Table Storage, Diagnose über HTTP-Endpunkte und MCP.
• Kein ETL- oder Data-Warehouse-Werkzeug für Massenläufe, sondern inkrementelle Synchronisation.
• Transaktionen und fachliche Konsistenz bleiben in den angebundenen Systemen.

2 - Infrastruktur (Bicep)

Einstiegspunkt ist bicep/main.bicep. Das Deployment läuft im Scope einer bestehenden Resource Group (Bicep-Default, kein explizites targetScope), im Modus Incremental. Die Parameter aus Anwendersicht sind unter Installation beschrieben; hier geht es um den inneren Aufbau.

Module und Parameterfluss

flowchart TD
    MAIN["main.bicep"]

    SA["storageaccount.bicep"]
    AI["applicationinsights.bicep"]
    ASP["appserviceplan.bicep"]
    SB["servicebus.bicep"]
    ALERT["metricalerts.bicep<br/>if enableAlerts"]
    DASH["dashboard.bicep<br/>if enableGrafanaDashboard"]
    FUNC["azurefunction.bicep"]
    ROLES["roleassignments.bicep<br/>if enableManagedIdentity"]

    MAIN --> SA & AI & ASP & SB & ALERT & DASH & FUNC & ROLES

    SA -->|"id, name"| FUNC
    AI -->|"connectionString, id"| FUNC
    AI -->|"id"| DASH
    ASP -->|"id"| FUNC
    SB -->|"authorizationRuleId, name"| FUNC
    SB -->|"id"| ALERT
    FUNC -->|"principalId"| ROLES
    SA -->|"id"| ROLES
    SB -->|"id"| ROLES

Die Abhängigkeiten ergeben sich über Modul-Outputs, es gibt keine Zyklen. Zusätzlich legt main.bicep ein leeres Tracking-Deployment für die Microsoft-Partner-Attribution an.

Ressourcen

Queues

Alle Queues haben deadLetteringOnMessageExpiration = true, TTL defaultMessageTimeToLive (Default P14D) und lockDuration (Default PT30S).

Die Schreibweise standart ist in Code und Bicep konsistent und darf nicht geändert werden, weil bestehende Deployments die Queue unter diesem Namen besitzen.

App-Settings der Function App

alwaysOn wird nicht gesetzt. Für Premium-Pläne kann das manuell nachgezogen werden.

Managed Identity und Rollen

Die System-Identity der Function App wird immer angelegt. Mit enableManagedIdentity = true entfallen die Schlüssel in den App-Settings, stattdessen weist roleassignments.bicep folgende Rollen zu:

SettingsService liefert Connection-Strings tolerant (null statt Exception); StorageService und ServiceBusService bauen ihre Clients aus dem Connection-String, wenn vorhanden, sonst über DefaultAzureCredential.

Alerts

Mit enableAlerts entstehen eine Action Group (E-Mail an notificationEmail), vier DLQ-Alerts auf DeadletteredMessages (Info ≥ 1, Warning > 50, Error > 200, Critical > 500, Schwellen parametrisierbar) und ein Queue-Längen-Alert auf ActiveMessages > 300000. Alle Alerts sind namespace-weit (EntityName = *). Anwendersicht: Monitoring & Alerts.

Netzwerk

Es gibt keine VNet-Integration, keine Private Endpoints und keine IP-Restriktionen. Storage, Service Bus und Function App sind öffentlich erreichbar; Transport ist TLS 1.2 bzw. HTTPS. Netzwerk-Isolation wäre eine Erweiterung des Bicep-Deployments.

Parameterdateien

bicep/params/databridge.sample.bicepparam ist die Vorlage mit allen Parametern. Kundenspezifische Parameterdateien liegen daneben im Repository und werden von cicd/manage/deploy_bicep.yml verwendet. storageKind erlaubt nur StorageV2; ältere Parameterdateien mit Storage schlagen an der Validierung fehl.

Marketplace-Artefakte

• bicep/marketplace/mainTemplate.json wird mit GenerateMarketplace.sh aus main.bicep erzeugt und muss nach jeder Bicep-Änderung neu generiert und committet werden.
• createUiDefinition.json ist Handarbeit und bildet die Parameter auf den Marketplace-Assistenten ab.

CI/CD

• cicd/manage/build_and_deploy_functionapp.yml baut das Zip, setzt die Version (VersionMajor.VersionMinor.yyyy.MMdd) und veröffentlicht ein globales Artefakt.
• cicd/manage/deploy_bicep.yml rollt Infrastruktur und Zip je Kunde aus.
• cicd/manage/build_docfx_and_publish.yml baut und veröffentlicht diese Dokumentation (der Dateiname stammt noch aus der DocFX-Zeit).

3 - Anwendungsarchitektur

Projekte (src/DataBridge.slnx)

Dependency Injection (Program.cs)

• FunctionsApplication.CreateBuilder(args) (moderner Einstiegspunkt statt new HostBuilder()), mit builder.UseMiddleware<InvocationContextMiddleware>(), die jede Invocation in einen InvocationContext einbettet und am Ende das gepufferte Log flusht.
• Application Insights über AddApplicationInsightsTelemetryWorkerService, Adaptive Sampling deaktiviert; DependencySamplingProcessor per AddApplicationInsightsTelemetryProcessor. Der Standard-AI-Logger-Provider ist auf LogLevel.None gesetzt, einziger Pfad nach AI ist der BufferingLoggerProvider.
• Alle Services sind Singletons: MetricsService, SettingsService, StorageService, ConnectionService, MappingService, JobService, WatermarkService, ServiceBusService, ProcessQueueMessageService, ProcessJobsForRetrievalInOrderService, CheckListService, InitializationService, McpToolService.
• Log-Level: Kategorien DataBridge und Function auf Trace, der Puffer-Provider filtert auf Debug.

host.json

Functions

Details zu Request und Response: API-Referenz.

Besonderheiten der Trigger

• Einmal-Initialisierung: GetChanges, ProcessQueueItems, JobApi, MappingApi und McpToolService halten je eine eigene statische InitializationGuard-Instanz (DataBridge.Core/Services/Initialization/InitializationGuard.cs) — sie kapselt das Lock+Flag-Muster, das vorher in jeder Klasse dupliziert war. Bewusst nicht geteilt: ein Fehler in einer Function zwingt keine der anderen zur Neu-Prüfung. Bei Fehlern wird InitializationService.ResetInitialization() gerufen und beim nächsten Aufruf neu initialisiert (Self-Repair). GetChanges fängt dafür zusätzlich EndpointNotFoundException und CommunicationException.
• Lock-Renewal: ProcessQueueItems erneuert den Service-Bus-Lock in einem Hintergrund-Task alle 30 Sekunden, bis die Nachricht abgeschlossen ist.
• Fehlerstempel: Bei Verarbeitungsfehlern werden ErrorMessage, ErrorTimeStamp, ErrorStackTrace und OperationId als Application Properties gesetzt und die Nachricht abandoned. Nach maxDeliveryCount Versuchen landet sie in der Dead-Letter-Subqueue.

Durable-Variante

flowchart TD
    HTTP["StartGetChangesOrchestration"] --> ORCH["GetChangesOrchestrator"]
    ORCH -->|"je Job aus GetJobsToRun"| CHK{"JobState.InProgress?"}
    CHK -->|ja| SKIP["überspringen"]
    CHK -->|nein| ADD["JobState.Add"]
    ADD --> LOOP["CallActivity(GetChangesActivity)"]
    LOOP -->|HasMoreRecord| LOOP
    LOOP -->|fertig| REM["JobState.Remove"]
    ORCH -. Fehler .-> REM

Die Activity liest seitenweise (MaxBatchSize = 3) und sendet jede Nachricht einzeln. Jobs laufen sequentiell. Der Durable-Pfad schreibt die Watermark nicht zurück und ist im Timer-Betrieb nicht aktiv.

Service-Schicht

InitializationService (bis v2.3 CoreService) ist der Einstieg. Er lädt beim ersten Aufruf Storage → Connections → Mappings → Jobs und stellt nur noch den Ready/Error-Zustand bereit. Er ist kein Service-Locator mehr: Functions, die z. B. ProcessJobsForRetrievalInOrderService oder ProcessQueueMessageService brauchen, injizieren diese direkt statt über InitializationService zu routen.

flowchart TB
    INIT["InitializationService"] --> STORE["StorageService"] & CONN["ConnectionService"] & MAP["MappingService"] & JOB["JobService"] & CHK["CheckListService"]
    GC["GetChanges/ExecuteJob"] --> INIT & PJRS["ProcessJobsForRetrievalInOrderService"]
    PQI["ProcessQueueItems"] --> INIT & PQMS["ProcessQueueMessageService"]
    PJRS --> JOB & MAP & CONN & WM["WatermarkService"] & SBS["ServiceBusService"]
    PQMS --> JOB & MAP & CONN & SBS
    JOB --> WM
    WM --> STORE
    CONN --> CONNECTORS["Connectoren"]
    MCP["McpToolService"] --> INIT & JOB & MAP & CONN & SBS

Connector-Basisklassen

Die Hierarchie ist unter Architektur → Connector-Hierarchie abgebildet. Aufgaben der Basisklassen:

• ConnectorBase<TRecord, TConnectionSetting>: Logger, ConnectionSetting, Lookup-Cache (ConcurrentDictionary<ThreeStringKey,string>), Culture, generische Serialize-Pipeline, Filter-Templating ({{…}}), optionale Key-Vault-Auflösung der Settings über keyVaultUrl + DefaultAzureCredential, GetTableMetadata. Vertragsmethoden: Connect, RetrieveChangesAsync, Upsert, Delete, ProcessMappingSerialization, Key-Aufbau.
• HttpConnectorBase: HTTP-Client mit Kompression, Basic, OAuth2 Client Credentials (konfigurierbare tokenUrl), API-Key oder Managed Identity (nur bei authmethod: ManagedIdentity, Scope aus der Connection), Token-Refresh.
• SqlConnectorBase<TConnection>: typisierte DbCommand/DataAdapter, SELECT 1 als Verbindungstest, seitenweises SELECT mit ORDER BY watermarkField, Metadaten über INFORMATION_SCHEMA (Oracle: USER_*/ALL_*), Schema-Cache. Upsert und Delete werfen NotImplementedException; SQL ist nur Quelle.
• CeConnector: Dataverse ServiceClient, RetrieveEntityChangesRequest mit DataVersion-Token, Batch-Pre-Caching von Lookups, EmbedChildren.
• OpenApiConnector: Pfade, Wrapper-Schlüssel und Typen aus der OpenAPI-Spezifikation, pro Prozess gecacht.

Transformations-Pipeline

flowchart LR
    REC["Quell-Record<br/>JObject / DataRow / Entity"] --> KEY["Key-Aufbau → QueueMessage.Id"]
    KEY --> FM{"SourceField<br/>beginnt mit '_'?"}
    FM -->|ja| SOP["SourceOperation-Handler<br/>LookupValue, ReadFromMessage,<br/>ActivityParty, LinkToRecord, EmbedChildren"]
    FM -->|nein| RES["TargetValueMapper → AttributeResolver<br/>Typ-Coercion nach Metadaten"]
    SOP --> ATTR["QueueMessage.Attributes"]
    RES --> ATTR

• AttributeResolver (Resolvers/Common, D365Bc, D365Ce, D365Nav): Typ-Coercion auf String-Basis (Value + TypeName in der Nachricht). CE hat zusätzlich Picklist, Lookup, Money, State/Status, MultiSelect.
• TargetValueMapper (Connectors/<X>/<X>TargetValueMapper.cs, SqlTargetValueMapper<T>): wählt anhand des Ziel-Metadatentyps den Resolver. Registry einmalig als FrozenDictionary; Coverage-Tests in DataBridge.Test machen fehlende Registrierungen zum Build-Fehler.
• Source-Operationen (ISourceOperationHandler): per Reflection registriert, Konstruktor (TConnector, ILogger). Felder mit _-Präfix sind virtuell.

Domänenobjekte

Querschnitt

• Telemetrie (DataBridge.Core/Telemetry): BufferingLoggerProvider, InvocationContext (AsyncLocal, operation_id), DependencySamplingProcessor, MetricsService. Verhalten und Steuerung: Logging.
• Konstanten (DataBridge.Core/Constants): QueueNames, StorageTables, MessageProperties, EnviromentVariables, Errors, MappingDefaultValues.
• Exceptions: TransitiveException steuert die Ablage in Transitive-Queues.

4 - API- und Nachrichten-Referenz

Authentifizierung

Alle HTTP-Functions verwenden AuthorizationLevel.Function. Der Funktionsschlüssel wird als Query-Parameter code oder Header x-functions-key übergeben. Es gibt keine Easy-Auth- oder Entra-Konfiguration im Deployment. Timer-, Service-Bus- und Durable-Trigger haben keinen HTTP-Endpunkt.

Routen ohne Route-Attribut folgen der Konvention /api/<FunctionName>.

HTTP-Endpunkte

GET /api/GetInfo

Liefert die Assembly-Version, keine Gesundheitsprüfung.

{ "Version": "2.3.2026.0926" }

POST /api/ExecuteJob

Stößt einen Job aus der Tabelle jobs manuell an (Erkennungsphase).

{ "JobName": "<RowKey des Jobs>" }

Ablauf: JobName prüfen → EnsureInitializedAsync → ProcessJobForRetrieval. Antwort 200 OK; Fehler propagieren als 500. Der Aufruf läuft ohne Job-Locking parallel zum Timer, siehe Datenflüsse → Watermark.

POST /api/ExecuteHttpRequest

Diagnose-Endpunkt: setzt eine rohe HTTP-GET-Anfrage mit eigenem HttpClient ab, unabhängig von den konfigurierten Connectoren.

Antwort: 200 OK mit dem rohen Antwort-Body des Zielservers.

GET /api/CheckList

EnsureInitializedAsync → InitializationService.CheckAsync(). Prüft Connections, Job-Connections, eindeutige Order, Mapping-Existenz sowie Tabellen und Felder gegen die Live-Metadaten der Connectoren. Antwort 200 OK mit serialisiertem CheckResponse, auch wenn Befunde vorliegen.

GET /api/Validate

Erzwingt die Initialisierung. 200 OK bei Erfolg, sonst 500; Body text/plain mit InitializationService.GetErrorLogs().

GET /api/job, GET /api/job/{rowKey}

Geladene Jobs (alle bzw. einer). 404, wenn der RowKey unbekannt ist.

GET /api/mapping, GET /api/mapping/{name}

Geladene Mappings. Die Suche läuft über den Mapping-Namen (PartitionKey), Fallback RowKey.

POST /api/mcp

Streamable-HTTP-Endpunkt des MCP-Servers. GET /api/mcp antwortet mit 405, weil kein SSE-Stream angeboten wird. Tools und Nutzung: MCP-Server.

GET /api/StartGetChangesOrchestration

Startet die Durable-Orchestrierung. Antwort 202 Accepted mit der Instance-ID und den Standard-Status-URLs des Durable-Clients.

Trigger-Bindungen

Die Datei sample.local.settings.json setzt lokal * */2 * * * *. Effektiv gilt immer der App-Setting-Wert der jeweiligen Umgebung.

Nachrichtenschema (Service Bus)

Body: JSON-serialisiertes QueueMessage (Newtonsoft.Json).

Application Properties:

SourceKey nutzt der DeadLetterMover zur Deduplizierung. OperationId ist die W3C-Root-ID und erlaubt die Korrelation mit Application Insights.

Fehlerverhalten

5 - Datenflüsse

Die konzeptuelle Beschreibung steht unter Wie es funktioniert. Hier folgen die Abläufe auf Klassenebene.

Erkennung (Timer)

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)

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)

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

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

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

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.

6 - Abhängigkeiten

Plattform

• net10.0 in allen Projekten, ImplicitUsings und Nullable aktiv.
• Azure Functions v4, Isolated Worker, OutputType=Exe.
• Tests auf xUnit v3 mit Microsoft Testing Platform. Aufruf mit dotnet test --project …; das Positionsargument funktioniert nicht mehr.

Projekte

flowchart LR
    HOST["DataBridge (Host)"] --> CORE["DataBridge.Core"]
    DLM["DeadLetterMover"] --> CORE
    TEST["DataBridge.Test"] --> CORE
    TEST --> HOST

NuGet-Pakete

Die Versionen werden zentral in src/Directory.Packages.props gepflegt (Central Package Management); die .csproj-Dateien referenzieren Pakete nur noch ohne Version. Ein Update passiert also an genau einer Stelle.

Sicherheitsprüfung beim Build: Directory.Packages.props aktiviert NuGet Audit für alle Projekte, inklusive transitiver Pakete, und meldet Treffer jeder Schwere. Ab „moderate“ (NU1902–NU1904) gelten sie als Fehler, „low“ (NU1901) bleibt eine Warnung — solche Lücken betreffen oft Funktionen, die gar nicht genutzt werden, und sollen kein Release blockieren. Ein Paket mit einer Sicherheitslücke ab „moderate“ lässt damit Restore und Build lokal, im Dockerfile und in der Pipeline scheitern. Abhilfe ist ein Update der betroffenen Version; liegt die Lücke in einem transitiven Paket, dieses mit eigener Version in Directory.Packages.props aufnehmen. Weil neue Advisories jederzeit erscheinen, kann ein bisher grüner Stand ohne Code-Änderung rot werden. Ist der Advisory-Feed nicht erreichbar, bleibt es bei der Warnung NU1900, offline bauen funktioniert also weiter.

Hier nur die Rolle der wichtigsten Pakete:

Alle DB-Treiber werden immer mitgeladen, auch wenn nur ein Teil genutzt wird.

Azure-Dienste

Externe Systeme

Architekturkritische Kopplungen

7 - Fähigkeiten-Katalog

Integration und Datenfluss

Konnektivität

Transformation und Mapping

Konfiguration und Steuerung

Resilienz

Observability

Infrastruktur und Auslieferung

Qualitätssicherung