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.