# Anwendungsarchitektur

> Projekte, Dependency Injection, host.json, Functions, Service-Schicht, Connector-Basisklassen, Transformations-Pipeline und Domänenobjekte.

---

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

---

## Projekte (`src/DataBridge.slnx`)

| Projekt | Typ | Zweck |
|---|---|---|
| `DataBridge` | Azure Functions, net10.0, isolated | Host: Trigger, DI, Telemetrie-Bootstrap, Durable Functions, MCP-Endpunkt |
| `DataBridge.Core` | Class Library | Services, Connectoren, Resolver, Mapper, Source-Operationen, Modelle, Telemetrie |
| `DataBridge.Test` | xUnit v3 + Moq (Microsoft Testing Platform) | Unit-Tests; Ordnerstruktur spiegelt `DataBridge.Core` |
| `DeadLetterMover` | Konsolenanwendung | Requeue von Dead-Letter-Nachrichten |

## 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`

| Bereich | Einstellung | Wert |
|---|---|---|
| Allgemein | `functionTimeout` | 10 min |
| AI-Sampling | `excludedTypes` | `Request;Exception` |
| Durable | `hubName` | `DurableFunctionsHub`, Azure-Storage-Provider, Distributed Tracing V2 |
| Storage Queues | `batchSize` / `maxDequeueCount` | 1 / 5 (nur für Durable relevant) |
| Service Bus | `autoCompleteMessages` | `false` |
| Service Bus | `maxConcurrentCalls` | 12 |
| Service Bus | `maxAutoLockRenewalDuration` | 5 min |

## Functions

| Function | Trigger | Aufgabe |
|---|---|---|
| `GetChanges` | Timer `%GetChangesTimerExpression%` | Initialisiert thread-sicher über `InitializationService` und ruft `ProcessJobsForRetrievalInOrderService.Execute` auf |
| `ProcessStandardQueueItems`, `ProcessHighPriorityQueueItems`, `ProcessLowPriorityQueueItems` | Service Bus `sbqexport`, `sbqhighexport`, `sbqlowexport` | Verarbeiten je eine Nachricht end-to-end über `ProcessQueueMessageService.Execute` |
| `ExecuteJob` | HTTP POST | Stößt einen Job per Name über `ProcessJobsForRetrievalInOrderService.ExecuteByJobName` an |
| `ExecuteHttpRequest` | HTTP POST | Ad-hoc-HTTP-GET mit eigenem `HttpClient` (NTLM oder anonym), unabhängig von den Connectoren |
| `GetInfo` | HTTP GET | Assembly-Version |
| `CheckList` | HTTP GET | `InitializationService.CheckAsync()`: Mappings gegen Live-Metadaten der Systeme |
| `Validate` | HTTP GET | Erzwingt die Initialisierung, liefert das Fehlerlog |
| `GetJobs`, `GetJobByRowKey` | HTTP GET `job`, `job/{rowKey}` | Geladene Job-Konfiguration |
| `GetMappings`, `GetMappingByRowKey` | HTTP GET `mapping`, `mapping/{name}` | Geladene Mappings, Suche per Name mit Fallback auf RowKey |
| `Mcp`, `McpGet` | HTTP POST/GET `mcp` | MCP-Server; GET antwortet mit 405, weil SSE-Streams nicht unterstützt werden |
| `StartGetChangesOrchestration`, `GetChangesOrchestrator`, `GetChangesActivity`, `JobState` | HTTP, Orchestration, Activity, Entity | Optionale Durable-Variante der Erkennung |

Details zu Request und Response: [API-Referenz](../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

```mermaid
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.

| Service | Verantwortung | Wichtige Methoden |
|---|---|---|
| `InitializationService` | Initialisierung, Ready/Error-Zustand | `EnsureInitializedAsync`, `CheckAsync`, `ResetInitialization`, `GetErrorLogs`, `HasError` |
| `SettingsService` | App-Settings, Connection-Strings | `GetConnectionString` mit Fallback-Kette `ConnectionStrings:` → `SERVICEBUSCONNSTR_` → `SQLCONNSTR_` → `CUSTOMCONNSTR_` → App-Setting; `GetStorageAccountName`, `GetServiceBusFullyQualifiedNamespace` |
| `StorageService` | Table Storage | `ValidateAndInitialize`, `GetTableEntries<T>`, `UpdateTableEntry<T>` (Upsert Merge, ohne ETag) |
| `ConnectionService` | Connector je Connection | `Initialize` mit Verbindungstest, `IsValidConnection`, `GetConnector` |
| `MappingService` | Mappings gruppiert nach Name | `Initialize`, `GetMapping`, `ValidateMappings` |
| `JobService` | Jobs, Validierung, Cron | `Initialize`, `GetJobsToRun` (NCrontab, 6 Felder), `GetJobByName` |
| `WatermarkService` | Persistenz des Stands | `GetWatermarkByJobName`, `GetAllWatermarks` (Bulk), `UpdateWatermarkByJobName`; Wertlogik in `WatermarkValue` |
| `ServiceBusService` | Enqueue | `SendMessageAsync`, `AddMessageOrSendAsync` (Batch bis 200 oder Größenlimit), `FlushBatch`; Priorität → Queue |
| `ProcessJobsForRetrievalInOrderService` | Erkennung | `Execute` (aktive Jobs nach `Order`), `ExecuteByJobName`, `ExecuteJob` |
| `ProcessQueueMessageService` | Verarbeitung | `Execute`: Deserialisierung, Backoff, Upsert/Delete/Action, Response, Transitive, Ignored |
| `CheckListService` | Diagnose | `Check()` gegen Live-Metadaten der Connectoren |
| `MetricsService` | Custom Metrics | pre-aggregiert über `GetMetric`, kein Sampling |
| `McpToolService` | MCP-Tool-Logik | read-only, siehe [MCP-Server](/docs/mcp-server/) |

```mermaid
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](/docs/ueberblick/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.

| `ConnectionTypes` | Connector |
|---|---|
| `D365CE` | `CeConnector` |
| `D365BC` | `BcConnector` |
| `D365NAV` | `NavConnector` |
| `OPENAPI` | `OpenApiConnector` |
| `MSSQL`, `PGSQL`, `MYSQL`, `ORACLE` | `MsSqlConnector`, `PostgreSqlConnector`, `MySqlConnector`, `OracleConnector` |

## Transformations-Pipeline

```mermaid
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

| Modell | Zweck |
|---|---|
| `ConnectionTableEntry` | PartitionKey `connection`, RowKey = Name, `Type`, `Json` = `ConnectionSetting` |
| `JobTableEntry` / `Job` | PartitionKey `job`; Source/Target-Connection, `Order`, `Filter`, `Mapping`, `WatermarkType`/`WatermarkField`, `Schedule`, `Action`, `Priority`, `ResponseJob`, `Source`/`Target` (job-eigene Pfade) |
| `MappingTableEntry` / `TableMapping` + `FieldMapping` | Feldzuordnung, `TargetKey`/`TargetKeys` (mit `KeyRank`), `SourceOperation`, `TargetOperation` |
| `WatermarkTableEntry` | PartitionKey `watermark`, RowKey = Job-Name, `Watermark`, `LastCronRun` |
| `QueueMessage` + `QueueMessageProperty` | Transportobjekt, siehe [Nachrichtenschema](../api-referenz/#nachrichtenschema-service-bus) |
| `TableMetadata` / `ColumnMetadata` | Schema-Informationen für MCP und CheckList |
| Enums | `ConnectionTypes`, `JobActionType`, `JobPriorityType`, `TargetKeyType`, `LookupType`, `WatermarkType` |

## Querschnitt

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