diff --git a/.vscode/extensions.json b/.vscode/extensions.json new file mode 100644 index 0000000..13fe9eb --- /dev/null +++ b/.vscode/extensions.json @@ -0,0 +1,5 @@ +{ + "recommendations": [ + "anysphere.csharp" + ] +} diff --git a/.vscode/settings.json b/.vscode/settings.json new file mode 100644 index 0000000..5e2a033 --- /dev/null +++ b/.vscode/settings.json @@ -0,0 +1,13 @@ +{ + "editor.formatOnSave": true, + "files.insertFinalNewline": true, + "files.trimTrailingWhitespace": true, + "[csharp]": { + "editor.defaultFormatter": "anysphere.csharp", + "editor.formatOnSave": true, + "editor.codeActionsOnSave": { + "source.fixAll": "explicit", + "source.organizeImports": "explicit" + } + } +} diff --git a/CHANGELOG.md b/CHANGELOG.md index 46cf604..c8e1a32 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,14 +5,52 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [Unreleased] + +## [2.2.0] - 2026-07-31 + +### Added +- **Actors facade:** `IDaprActorService` / `DaprActorService` — create typed/weak actor clients and invoke methods with `Result` outcomes; DI `RegisterActors(configure?)` and pipeline `RegisterActorsHandlers()`. +- **Workflows facade:** `IDaprWorkflowService` / `DaprWorkflowService` — schedule, get/wait state, raise event, terminate/suspend/resume/purge, list instance IDs, get history, rerun from event; DI `RegisterWorkflows(configure?)`. +- **Pub/Sub facade rename/expand:** `IDaprPubSubService` — publish with metadata, byte publish, bulk publish; DI `RegisterPubSub()`. +- **State expand:** bulk get/save/delete, `TryDeleteStateAsync`, `QueryStateAsync`, `ExecuteStateTransactionAsync`, optional metadata/consistency/options on existing methods. +- **Client facades:** invocation, binding, secrets, configuration, cryptography, sidecar (`IDapr*Service`) + `Register*` and `RegisterDaprClientFacades()`. +- **Lock facade:** `IDaprLockService` / `DaprLockService` — `LockAsync` / `UnlockAsync` with `Result` outcomes; DI `RegisterLock()` (included in `RegisterDaprClientFacades()`). +- **HA helpers:** `TryHoldAsync` / `DaprWorkLeaseHold` (auto-renew + `Generation` fencing), `LeasedBackgroundService`, `DaprWorkLeaseBootstrap.RunBootstrapUnderLeaseAsync`, `RegisterWorkLeases(storeName)` / `IDaprWorkLeaseOptions`. +- **Work leases rename:** `IDaprWorkLeaseStore` / `DaprWorkLeaseStore` → `IDaprWorkLeaseService` / `DaprWorkLeaseService`. + +### Fixed +- **State get miss on `state.jetstream`:** `GetStateAndETagAsync` / `GetStateAsync` treat NATS/Dapr `key not found` (often gRPC `Internal`) as empty without logging Error — so work-lease first acquire works after release or on a fresh bucket. +- **Results hygiene:** rethrow `OperationCanceledException`; `BadRequest` for empty store/key or pubsub/topic; idempotent `DeleteStateAsync` when key is missing. +- **DI:** `AddDaprClient` is skipped per `IServiceCollection` when `DaprClient` is already registered (no process-wide static flag). +- **Docs:** README and CONTRIBUTING aligned with `net10.0`, RepoUtils Non-Helm bats, and current package version. + +### Changed +- **Work leases layout:** all HA lease types live under `Services/WorkLease/` (`MaksIT.Dapr.Services.WorkLease`) — `DaprWorkLease`, `DaprWorkLeaseHold` / bootstrap / `LeasedBackgroundService`, `DaprWorkLeaseService`, `IDaprWorkLeaseOptions`, `IDaprRuntimeInstanceId`. +- **`GetStateAsync` miss:** returns `Ok(null)` instead of `NotFound` (aligned with ETag get and typical MaksIT optional-read pattern). Check `Value is null` for absence; `!IsSuccess` means infra failure. +- **CancellationToken:** optional `cancellationToken` on facade APIs; lease `ReleaseAsync` forwards it to delete. +- **Invocation:** `DaprInvocationService` uses `DaprClient.CreateInvokableHttpClient` (HTTP POST + JSON) instead of obsolete `InvokeMethodAsync(appId, method…)` helpers. +- **Workflows DI:** `RegisterWorkflows()` calls parameterless `AddDaprWorkflow()` when no configure delegate is passed (SDK 1.18 auto-discovers workflows/activities). +- **Dependencies:** Dapr packages bumped to `1.18.5`. +- **Docs:** README documents each facade (when to use / when not, DI, API), including lock vs work-lease guidance, and a short coordination-pattern chooser. +- **XML docs** on public APIs; `GenerateDocumentationFile` enabled. +- Removed obsolete `assets/badges` pack items (CoverageBadges uses shields.io). +- Package version **2.2.0**. + +### Removed +- Custom pub/sub accept types and `IDaprPubSubWorkHandler` (`DaprPubSubAcceptOutcome`, `DaprPubSubAcceptResult`, `DaprPubSubAck`). Topic controllers should return `MaksIT.Results.Result` via `ToActionResult()` directly (`Ok` ACK, `ServiceUnavailable` retry, `BadRequest` drop). +- `IDaprPublisherService` / `RegisterPublisher` (use `IDaprPubSubService` / `RegisterPubSub`). +- Leftover `DaprWorkLeaseStore` and `PubSub/DaprPubSubWork` types (no obsolete shims). +- `DaprFacadeGuard` helper (inline try/catch + validation in each service). + ## [2.1.0] - 2026-07-30 ### Added -- **HA work leases via Dapr state:** `IDaprWorkLeaseStore` / `DaprWorkLeaseStore` (acquire, renew, release, get) using ETag concurrency — broker-agnostic (NATS KV / Postgres / other state Component). +- **HA work leases via Dapr state:** `IDaprWorkLeaseService` / `DaprWorkLeaseService` (acquire, renew, release, get) using ETag concurrency — broker-agnostic (NATS KV / Postgres / other state Component). - **State ETag API:** `GetStateAndETagAsync` and `TrySaveStateAsync` on `IDaprStateStoreService`. - **Runtime instance id:** `IDaprRuntimeInstanceId` / `DaprRuntimeInstanceIdProvider` (`POD_NAME` in Kubernetes). - **Pub/sub worker helpers:** `IDaprPubSubWorkHandler`, `DaprPubSubAcceptOutcome` / `DaprPubSubAcceptResult`, `DaprPubSubAck` (HTTP ACK/NAK mapping). -- DI: `RegisterWorkLeases()` registers state store + lease store + instance id. +- DI: `RegisterWorkLeases()` registers state store + work-lease service + instance id. ### Changed - Package version **2.1.0**. diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 5fed35d..1f69ead 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -14,7 +14,7 @@ Thank you for your interest in contributing to `MaksIT.Dapr`. ### Prerequisites -- .NET 8 SDK or later +- .NET 10 SDK - Git - PowerShell 7+ (recommended for utility scripts) @@ -27,9 +27,19 @@ dotnet build MaksIT.Dapr.slnx ### Test +Prefer the RepoUtils test engine (coverage + README shields.io badges): + +```powershell +pwsh -File .\utils\engines\test\Invoke-TestEngine.ps1 +``` + +Or double-click `utils\Invoke-TestEngine.bat`. + +Quick local run: + ```bash cd src -dotnet test MaksIT.Dapr.Tests +dotnet test MaksIT.Dapr.slnx ``` ## Commit Message Format @@ -47,7 +57,14 @@ Use: | `(feature):` | New feature or enhancement | | `(bugfix):` | Bug fix | | `(refactor):` | Refactoring without behavior change | -| `(chore):` | Maintenance tasks (dependencies, tooling, docs) | +| `(perf):` | Performance improvement | +| `(test):` | Tests only | +| `(docs):` | Documentation | +| `(build):` | Build / packaging | +| `(ci):` | CI configuration | +| `(style):` | Formatting / style | +| `(chore):` | Maintenance tasks (dependencies, tooling) | +| `(revert):` | Revert a previous change | ### Guidelines @@ -57,10 +74,11 @@ Use: ## Pull Request Checklist -1. Ensure build and tests pass. +1. Ensure build and tests pass (RepoUtils test engine when possible). 2. Update `README.md` if behavior or usage changed. 3. Update `CHANGELOG.md` under the target version. -4. Keep changes scoped and explain rationale in the PR description. +4. If coverage changed, refresh README shields.io badges via the test engine. +5. Keep changes scoped and explain rationale in the PR description. ## Versioning @@ -92,7 +110,7 @@ Builds, tests, packs, and publishes to NuGet and GitHub release flows. pwsh -File .\utils\engines\release\Invoke-ReleasePackage.ps1 ``` -Or run `utils\Invoke-ReleasePackage-Single.bat` (or `-HA.bat` when Helm deploy is wired). +Or double-click `utils\Invoke-ReleasePackage.bat`. Prerequisites: @@ -102,6 +120,8 @@ Prerequisites: Configuration: `utils/engines/release/scriptSettings.json` +Public GitHub releases target `https://github.com/MAKS-IT-COM/maksit-core-dapr` (see release `scriptSettings.json`). + ### Update repo utilities Refreshes `utils/` from maksit-repoutils while preserving local `scriptSettings.json` files. diff --git a/README.md b/README.md index 90ed771..7478f66 100644 --- a/README.md +++ b/README.md @@ -1,206 +1,381 @@ # MaksIT.Dapr -![Line Coverage](https://img.shields.io/badge/Line%20Coverage-41.3%25-yellowgreen) -![Branch Coverage](https://img.shields.io/badge/Branch%20Coverage-25%25-yellow) -![Method Coverage](https://img.shields.io/badge/Method%20Coverage-53.1%25-yellowgreen) +![Line Coverage](https://img.shields.io/badge/Line%20Coverage-41.9%25-yellowgreen) +![Branch Coverage](https://img.shields.io/badge/Branch%20Coverage-40.7%25-yellowgreen) +![Method Coverage](https://img.shields.io/badge/Method%20Coverage-59.5%25-yellowgreen) -This repository hosts the `maksit-dapr` project, which utilizes [Dapr](https://dapr.io/) (Distributed Application Runtime) to facilitate building and managing microservices with ease. The project focuses on implementing a robust, scalable solution leveraging Dapr's building blocks and abstractions. +NuGet facade over [Dapr](https://dapr.io/) for ASP.NET Core: pub/sub, state, invocation, bindings, secrets, configuration, cryptography, sidecar, lock, HA work leases, actors, and workflows — all with `MaksIT.Results` outcomes. ## Table of Contents -- [MaksIT.Dapr](#maksitdapr) - - [Table of Contents](#table-of-contents) - - [Overview](#overview) - - [Features](#features) - - [Getting Started](#getting-started) - - [Installation](#installation) - - [Usage](#usage) - - [Registering Dapr Services](#registering-dapr-services) - - [Injecting and Using Dapr Services](#injecting-and-using-dapr-services) - - [Contributing](#contributing) - - [Contact](#contact) - - [License](#license) +- [Overview](#overview) +- [Result conventions](#result-conventions) +- [Getting Started](#getting-started) +- [Installation](#installation) +- [Registering services](#registering-services) +- [Services](#services) + - [Pub/Sub — `IDaprPubSubService`](#pubsub--idaprpubsubservice) + - [State store — `IDaprStateStoreService`](#state-store--idaprstatestoreservice) + - [Service invocation — `IDaprInvocationService`](#service-invocation--idaprinvocationservice) + - [Bindings — `IDaprBindingService`](#bindings--idaprbindingservice) + - [Secrets — `IDaprSecretService`](#secrets--idaprsecretservice) + - [Configuration — `IDaprConfigurationService`](#configuration--idaprconfigurationservice) + - [Cryptography — `IDaprCryptographyService`](#cryptography--idaprcryptographyservice) + - [Sidecar — `IDaprSidecarService`](#sidecar--idaprsidecarservice) + - [Distributed lock — `IDaprLockService`](#distributed-lock--idaprlockservice) + - [HA work leases — `IDaprWorkLeaseService`](#ha-work-leases--idaprworkleaseservice) + - [Actors — `IDaprActorService`](#actors--idapractorservice) + - [Workflows — `IDaprWorkflowService`](#workflows--idaprworkflowservice) +- [Choosing a coordination pattern](#choosing-a-coordination-pattern) +- [Contributing](#contributing) +- [Contact](#contact) +- [License](#license) ## Overview -`maksit-dapr` serves as a foundational project to explore and implement Dapr-based microservices, demonstrating the integration of Dapr�s pub-sub, bindings, state management, and other building blocks in a distributed system environment. +`MaksIT.Dapr` wraps Dapr building blocks so application code depends on small `IDapr*Service` facades instead of raw `DaprClient` calls. Failures surface as `MaksIT.Results` outcomes that map cleanly to HTTP via `ToActionResult()`. -## Features +| Need | Register | Inject | +|------|----------|--------| +| Publish events | `RegisterPubSub()` | `IDaprPubSubService` | +| Key/value state | `RegisterStateStore()` | `IDaprStateStoreService` | +| Call another Dapr app | `RegisterInvocation()` | `IDaprInvocationService` | +| Trigger external systems (queues, cron, …) | `RegisterBinding()` | `IDaprBindingService` | +| Read secrets from a Component | `RegisterSecrets()` | `IDaprSecretService` | +| Dynamic config + subscribe | `RegisterConfiguration()` | `IDaprConfigurationService` | +| Encrypt/decrypt via Dapr crypto Component | `RegisterCryptography()` | `IDaprCryptographyService` | +| Wait for sidecar / health / metadata | `RegisterSidecar()` | `IDaprSidecarService` | +| Short-lived distributed mutex | `RegisterLock()` | `IDaprLockService` | +| Multi-replica exclusive work (leader/bootstrap) | `RegisterWorkLeases(storeName)` | `IDaprWorkLeaseService` | +| Virtual actors | `RegisterActors(...)` | `IDaprActorService` | +| Durable workflows | `RegisterWorkflows(...)` | `IDaprWorkflowService` | -- **Pub-Sub Integration**: Uses Dapr's pub-sub component for seamless event-driven communication. -- **State Management**: Efficient, distributed state handling across microservices. +`RegisterDaprClientFacades()` registers all `DaprClient`-backed rows above (not actors, workflows, or work leases). + +## Result conventions + +- **Success / failure:** check `result.IsSuccess`. Map to HTTP with `result.ToActionResult()`. +- **Cancellation:** facades rethrow `OperationCanceledException` (do not wrap as a failed `Result`). +- **State misses:** `GetStateAsync` / `GetStateAndETagAsync` return `Ok(null)` / `Ok((null, null))` when the key is absent (including JetStream `key not found` surfaced as gRPC `Internal`). `!IsSuccess` means infrastructure failure. +- **Topic handlers:** return `Result` via `ToActionResult()` — `Ok` ACK, `ServiceUnavailable` retry, `BadRequest` drop. ## Getting Started Ensure that you have the following installed: -- [.NET 8.0 SDK](https://dotnet.microsoft.com/download) -- [Docker](https://www.docker.com/get-started) +- [.NET 10 SDK](https://dotnet.microsoft.com/download) +- [Dapr CLI](https://docs.dapr.io/getting-started/install-dapr-cli/) (for local sidecar runs) +- [Docker](https://www.docker.com/get-started) (optional; used by RepoUtils Linux test validation) ## Installation -To install MaksIT.Core, add the package to your project via NuGet: - ```powershell dotnet add package MaksIT.Dapr ``` -Or manually add it to your .csproj file: +Or in your `.csproj`: -```powershell - +```xml + ``` -## Usage - -### Registering Dapr Services - -To use `maksit-dapr` in your application, you must register the provided services for dependency injection. Follow these steps to integrate Dapr's pub-sub and state management capabilities in your ASP.NET Core application: - -1. **Register the Publisher and State Store Services**: Add these services in `Program.cs` or `Startup.cs`. +## Registering services ```csharp using MaksIT.Dapr.Extensions; var builder = WebApplication.CreateBuilder(args); -// Register Dapr services -builder.Services.RegisterPublisher(); -builder.Services.RegisterStateStore(); +// All DaprClient-backed facades (pub/sub, state, invocation, binding, secrets, …) +builder.Services.RegisterDaprClientFacades(); +// Or register individually: RegisterPubSub(), RegisterStateStore(), … + // HA coordination (leases via Dapr state Component — e.g. persistent NATS KV) -builder.Services.RegisterWorkLeases(); +builder.Services.RegisterWorkLeases("maksit-cicd-state"); // storeName for LeasedBackgroundService + +builder.Services.RegisterActors(options => { + options.Actors.RegisterActor(); +}); +builder.Services.RegisterWorkflows(); var app = builder.Build(); -// Set up Dapr subscriber middleware (optional) -// Only after Webapi Authorization services +// After authentication/authorization middleware app.RegisterSubscriber(); +app.RegisterActorsHandlers(); ``` -2. **Use Controller as a Dapr Subscriber**: +## Services -To designate a controller as a Dapr subscriber, annotate it with the `[Topic("pubsubName", "name")]` attribute: +### Pub/Sub — `IDaprPubSubService` + +**When to use:** fire-and-forget or at-least-once messaging between services (commands, domain events, fan-out). Prefer this over direct broker SDKs so the app stays Component-agnostic (Redis, NATS, Kafka, …). + +**When not to:** request/response RPC → [service invocation](#service-invocation--idaprinvocationservice); long-running orchestrations → [workflows](#workflows--idaprworkflowservice); exclusive multi-replica jobs → [work leases](#ha-work-leases--idaprworkleaseservice). + +| | | +|--|--| +| **DI** | `RegisterPubSub()` (also in `RegisterDaprClientFacades()`) | +| **Pipeline** | `app.RegisterSubscriber()` + `[Topic("pubsubName", "topic")]` on controllers | +| **API** | `PublishEventAsync`, `PublishByteEventAsync`, `BulkPublishEventAsync` | ```csharp -using Dapr; +var result = await pubSub.PublishEventAsync("my-pubsub", "my-topic", command, cancellationToken: ct); +if (!result.IsSuccess) { /* handle */ } [Topic("my-pubsub", "my-topic")] -public class MyController : ControllerBase -{ - [HttpPost("/my-endpoint")] - public IActionResult ReceiveMessage([FromBody] MyCommand payload) - { - - - // Handle message - return Ok(); - } -} +[HttpPost("/my-endpoint")] +public IActionResult Receive([FromBody] MyCommand payload) => + Result.Ok().ToActionResult(); // ACK; ServiceUnavailable → retry; BadRequest → drop ``` -### Injecting and Using Dapr Services +### State store — `IDaprStateStoreService` -With the services registered, you can inject `IDaprPublisherService` and `IDaprStateStoreService` into controllers or other services as needed: +**When to use:** shared key/value state (caches, documents, coordination metadata) through a Dapr state Component. Also the foundation for [work leases](#ha-work-leases--idaprworkleaseservice). + +**When not to:** primary transactional business data that belongs in your product database; use the app’s own data layer for that. + +| | | +|--|--| +| **DI** | `RegisterStateStore()` | +| **API** | `Set` / `Get` / `Delete`, ETag OCC (`GetStateAndETagAsync`, `TrySaveStateAsync`, `TryDeleteStateAsync`), bulk, query, transactions | ```csharp -using MaksIT.Dapr; - -public class MyService -{ - private readonly IDaprPublisherService _publisher; - private readonly IDaprStateStoreService _stateStore; - - public MyService(IDaprPublisherService publisher, IDaprStateStoreService stateStore) - { - _publisher = publisher; - _stateStore = stateStore; - } - - public async Task PublishEventAsync() - { - var command = new MyCommand - { - CommandId = Guid.NewGuid(), - CommandName = "SampleCommand", - Timestamp = DateTime.UtcNow - }; - - var options = new JsonSerializerOptions - { - PropertyNamingPolicy = JsonNamingPolicy.CamelCase, - DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull, - }; - - var payload = JsonSerializer.Serialize(command, options); - - var result = await _publisher.PublishEventAsync("my-pubsub", "my-topic", payload); - if (!result.IsSuccess) - { - // Handle error - } - } - - public async Task SetStateAsync() - { - var saveResult = await _stateStore.SetStateAsync("my-store", "my-key", "my-value"); - if (!saveResult.IsSuccess) - { - // Handle error - } - } - - public async Task GetStateAsync() - { - var stateResult = await _stateStore.GetStateAsync("my-store", "my-key"); - return stateResult.IsSuccess ? stateResult.Value : null; - } - - public async Task DeleteStateAsync() - { - var deleteResult = await _stateStore.DeleteStateAsync("my-store", "my-key"); - if (!deleteResult.IsSuccess) - { - // Handle error - } - } -} +var save = await stateStore.SetStateAsync("my-store", "my-key", "my-value", cancellationToken: ct); +var get = await stateStore.GetStateAsync("my-store", "my-key", cancellationToken: ct); +if (!get.IsSuccess) { /* infra failure */ } +var value = get.Value; // null when key is missing ``` -This setup enables your ASP.NET Core application to utilize Dapr's pub-sub, state management, and other building blocks with minimal boilerplate. +### Service invocation — `IDaprInvocationService` -### HA work leases (multi-replica) +**When to use:** synchronous HTTP-style calls to another Dapr app id (mTLS, retries, and discovery handled by the sidecars). Good for request/response between microservices without hard-coding URLs. -Use Dapr **state** (not product DB lock tables) for bootstrap/sweep/holder coordination. The state Component is infrastructure (e.g. persistent NATS JetStream KV); application code stays broker-agnostic. +**When not to:** broadcast or decoupled events → [pub/sub](#pubsub--idaprpubsubservice); calling non-Dapr external APIs → [bindings](#bindings--idaprbindingservice) or a normal `HttpClient`. + +| | | +|--|--| +| **DI** | `RegisterInvocation()` | +| **API** | `InvokeAsync` overloads (void / request / response / both) — HTTP **POST** via `CreateInvokableHttpClient` | ```csharp -builder.Services.RegisterWorkLeases(); +var result = await invocation.InvokeAsync( + appId: "orders", + methodName: "create", + data: request, + cancellationToken: ct); +``` -// inject IDaprWorkLeaseStore + IDaprRuntimeInstanceId -var acquired = await leases.TryAcquireAsync( +### Bindings — `IDaprBindingService` + +**When to use:** invoke an **output** binding Component (send to a queue, call Twilio, write to blob storage, cron-triggered input handlers on the app side, etc.) without embedding vendor SDKs. + +**When not to:** app-to-app messaging → [pub/sub](#pubsub--idaprpubsubservice) or [invocation](#service-invocation--idaprinvocationservice). + +| | | +|--|--| +| **DI** | `RegisterBinding()` | +| **API** | `InvokeAsync` (fire-and-forget or typed response) | + +```csharp +var result = await bindings.InvokeAsync("my-smtp", "create", emailPayload, cancellationToken: ct); +``` + +### Secrets — `IDaprSecretService` + +**When to use:** load secrets at runtime from a Dapr secret store Component (Kubernetes secrets, Azure Key Vault, local files, …) when the host should not bind every secret into `IConfiguration` up front. + +**When not to:** replace standard ASP.NET `appsettings` / `appsecrets.json` for ordinary host config — MaksIT hosts still prefer configuration binding for app settings; use this facade for Component-backed secret reads from application code. + +| | | +|--|--| +| **DI** | `RegisterSecrets()` | +| **API** | `GetAsync`, `GetBulkAsync` | + +```csharp +var secret = await secrets.GetAsync("my-secret-store", "connection-string", cancellationToken: ct); +``` + +### Configuration — `IDaprConfigurationService` + +**When to use:** read or subscribe to keys from a Dapr configuration Component (feature flags, dynamic settings that change without redeploy). + +**When not to:** static startup configuration already covered by `IConfiguration` / Options — keep using the host configuration stack for that. + +| | | +|--|--| +| **DI** | `RegisterConfiguration()` | +| **API** | `GetAsync`, `SubscribeAsync`, `UnsubscribeAsync` | + +### Cryptography — `IDaprCryptographyService` + +**When to use:** encrypt/decrypt payloads with keys managed by a Dapr cryptography Component (keys stay in the sidecar/Component, not in app memory as raw key material). + +**When not to:** simple password hashing or app-local crypto libraries when you do not need Dapr-managed keys. + +| | | +|--|--| +| **DI** | `RegisterCryptography()` | +| **API** | `EncryptAsync`, `DecryptAsync` | + +### Sidecar — `IDaprSidecarService` + +**When to use:** gate startup or background work until the sidecar is ready; health/outbound checks; inspect metadata; request graceful sidecar shutdown. + +**Typical scenario:** call `WaitForSidecarAsync` before the first [work-lease](#ha-work-leases--idaprworkleaseservice) race or state call on cold start so replicas do not fail while the sidecar is still connecting. + +| | | +|--|--| +| **DI** | `RegisterSidecar()` | +| **API** | `CheckHealthAsync`, `CheckOutboundHealthAsync`, `WaitForSidecarAsync`, `GetMetadataAsync`, `ShutdownAsync` | + +```csharp +var ready = await sidecar.WaitForSidecarAsync(ct); +if (!ready.IsSuccess) { /* abort startup path */ } +``` + +### Distributed lock — `IDaprLockService` + +**When to use:** a short-lived distributed mutex on a Dapr **lock** Component — e.g. protect a critical section across processes for seconds, then unlock. Check `TryLockResponse.Success`; dispose the response (or call `UnlockAsync`) when finished. Expiry is in seconds via the Dapr API. + +**When not to:** multi-replica leader election, bootstrap, or long-running exclusive sweeps → prefer [HA work leases](#ha-work-leases--idaprworkleaseservice) (state-backed TTL, auto-renew, generation fencing, `LeasedBackgroundService` / bootstrap helpers). Per-entity serialized logic → [actors](#actors--idapractorservice). + +| | | +|--|--| +| **DI** | `RegisterLock()` (also in `RegisterDaprClientFacades()`) | +| **API** | `LockAsync` → `TryLockResponse`, `UnlockAsync` → `UnlockResponse` | +| **Component** | Dapr lock store (separate from the state Component used by work leases) | + +```csharp +var locked = await locks.LockAsync("my-lock-store", "invoice-42", runtimeInstance.InstanceId, expiryInSeconds: 15, ct); +if (!locked.IsSuccess) + return; // infra failure + +await using var handle = locked.Value; // DisposeAsync unlocks +if (!handle.Success) + return; // another owner holds the lock + +/* critical section */ +``` + +Or unlock explicitly with `UnlockAsync` when you are not disposing the `TryLockResponse`. + +### HA work leases — `IDaprWorkLeaseService` + +**When to use:** only one replica among many should run a piece of work (DB migrate/bootstrap, periodic sweep, leader-style job). Leases live in a Dapr **state** Component (e.g. persistent NATS JetStream KV) — broker-agnostic, ETag concurrency, optional auto-renew and generation fencing. + +**When not to:** short critical sections with an explicit unlock → [distributed lock](#distributed-lock--idaprlockservice); per-entity serialized state machines → [actors](#actors--idapractorservice); multi-step durable processes → [workflows](#workflows--idaprworkflowservice). + +| | | +|--|--| +| **Namespace** | `MaksIT.Dapr.Services.WorkLease` | +| **DI** | `RegisterWorkLeases()` or `RegisterWorkLeases("state-component-name")` | +| **Also registered** | `IDaprStateStoreService`, `IDaprRuntimeInstanceId` (`POD_NAME` in Kubernetes) | +| **Helpers** | `TryHoldAsync` → `DaprWorkLeaseHold`, `DaprWorkLeaseBootstrap.RunBootstrapUnderLeaseAsync`, `LeasedBackgroundService` | + +```csharp +using MaksIT.Dapr.Services.WorkLease; + +builder.Services.RegisterWorkLeases("maksit-cicd-state"); + +await using var hold = (await leases.TryHoldAsync( storeName: "maksit-cicd-state", workKey: "bootstrap", holderId: runtimeInstance.InstanceId, ttl: TimeSpan.FromMinutes(5), - ct); + autoRenew: true, + cancellationToken: ct)).Value; -if (acquired is { IsSuccess: true, Value: true }) { - try { /* exclusive work */ } - finally { await leases.ReleaseAsync("maksit-cicd-state", "bootstrap", runtimeInstance.InstanceId, CancellationToken.None); } +if (hold is null) + return; // another replica holds the lease + +var stillHeld = await hold.EnsureStillHeldAsync(ct); +if (stillHeld is { IsSuccess: true, Value: true }) { + /* exclusive work; hold.Generation is a fencing token */ } + +// Bootstrap: leader runs under lease; followers poll until ready +await DaprWorkLeaseBootstrap.RunBootstrapUnderLeaseAsync( + leases, + "maksit-cicd-state", + "bootstrap", + runtimeInstance.InstanceId, + TimeSpan.FromMinutes(5), + bootstrap: async ct => { /* migrate / seed */ return Result.Ok(); }, + isReady: async ct => Result.Ok(await db.IsMigratedAsync(ct)), + cancellationToken: ct); ``` -Pub/sub workers: implement `IDaprPubSubWorkHandler` and map with `DaprPubSubAck.ToActionResult` (2xx ACK, 503 Busy → redelivery). +For recurring exclusive sweeps, subclass `LeasedBackgroundService` and implement `ExecuteLeasedAsync` (uses `IDaprWorkLeaseOptions.StoreName` from `RegisterWorkLeases(storeName)`). Before lease races on cold start, call `IDaprSidecarService.WaitForSidecarAsync`. + +### Actors — `IDaprActorService` + +**When to use:** virtual-actor patterns — single-threaded access to an entity id (cart, device, session), turn-based concurrency, reminders/timers. Define actor interfaces and implementations in the product; this facade creates proxies and invokes methods with `Result` outcomes. + +**When not to:** cluster-wide singleton jobs → [work leases](#ha-work-leases--idaprworkleaseservice); multi-step saga across many services → [workflows](#workflows--idaprworkflowservice). + +| | | +|--|--| +| **DI** | `RegisterActors(options => options.Actors.RegisterActor())` | +| **Pipeline** | `app.RegisterActorsHandlers()` | +| **API** | `Create` / `Create`, `InvokeAsync` overloads | + +```csharp +builder.Services.RegisterActors(o => o.Actors.RegisterActor()); +app.RegisterActorsHandlers(); + +var created = actors.Create("cart-42", nameof(CounterActor)); +if (!created.IsSuccess) + return created.ToResult().ToActionResult(); + +var count = await created.Value.IncrementAsync(); +``` + +### Workflows — `IDaprWorkflowService` + +**When to use:** durable, multi-step business processes (order pipeline, approval flow) with history, wait-for-external-event, suspend/resume, and replay. Author `Workflow` / `WorkflowActivity` types in the product (auto-discovered on Dapr SDK 1.18+). + +**When not to:** simple fire-and-forget messages → [pub/sub](#pubsub--idaprpubsubservice); single exclusive background job → [work leases](#ha-work-leases--idaprworkleaseservice). + +| | | +|--|--| +| **DI** | `RegisterWorkflows(configure?)` | +| **API** | `ScheduleAsync`, get/wait state, `RaiseEventAsync`, terminate/suspend/resume/purge, list IDs, history, `RerunFromEventAsync` | + +```csharp +builder.Services.RegisterWorkflows(); + +var scheduled = await workflows.ScheduleAsync( + workflowName: nameof(OrderProcessingWorkflow), + input: order, + instanceId: order.Id, + cancellationToken: ct); + +if (!scheduled.IsSuccess) + return scheduled.ToResult().ToActionResult(); + +var completed = await workflows.WaitForCompletionAsync(scheduled.Value, cancellationToken: ct); +``` + +## Choosing a coordination pattern + +| Scenario | Prefer | +|----------|--------| +| One replica runs migrate/bootstrap/sweep | **Work leases** | +| Short critical section across processes | **Distributed lock** | +| Per-entity single-threaded logic | **Actors** | +| Long-running multi-step process with history | **Workflows** | +| Decoupled async events | **Pub/sub** | +| Sync call to another Dapr app | **Invocation** | ## Contributing -Contributions to this project are welcome! Please fork the repository and submit a pull request with your changes. If you encounter any issues or have feature requests, feel free to open an issue on GitHub. +See [CONTRIBUTING.md](CONTRIBUTING.md) for build/test (RepoUtils), commit format, and release scripts. ## Contact -If you have any questions or need further assistance, feel free to reach out: - - **Email**: [maksym.sadovnychyy@gmail.com](mailto:maksym.sadovnychyy@gmail.com) ## License -See `LICENSE.md`. \ No newline at end of file +See `LICENSE.md`. diff --git a/src/MaksIT.Dapr.Tests/DaprActorServiceTests.cs b/src/MaksIT.Dapr.Tests/DaprActorServiceTests.cs new file mode 100644 index 0000000..98859bc --- /dev/null +++ b/src/MaksIT.Dapr.Tests/DaprActorServiceTests.cs @@ -0,0 +1,66 @@ +using Microsoft.Extensions.Logging; +using Dapr.Actors; +using Dapr.Actors.Client; +using Moq; +using MaksIT.Dapr.Services; + + +namespace MaksIT.Dapr.Tests; + +public class DaprActorServiceTests { + [Fact] + public void Create_ReturnsBadRequest_WhenActorIdEmpty() { + var service = new DaprActorService( + Mock.Of>(), + Mock.Of()); + + var result = service.Create(" ", "MyActor"); + + Assert.False(result.IsSuccess); + } + + [Fact] + public void Create_ReturnsOk_WhenFactorySucceeds() { + var actor = ActorProxy.Create(new ActorId("1"), "MyActor"); + var factory = new Mock(); + factory + .Setup(f => f.Create(It.IsAny(), "MyActor", null)) + .Returns(actor); + + var service = new DaprActorService( + Mock.Of>(), + factory.Object); + + var result = service.Create("1", "MyActor"); + + Assert.True(result.IsSuccess); + Assert.NotNull(result.Value); + } + + [Fact] + public async Task InvokeAsync_ReturnsBadRequest_WhenMethodEmpty() { + var service = new DaprActorService( + Mock.Of>(), + Mock.Of()); + + var result = await service.InvokeAsync("1", "MyActor", " "); + + Assert.False(result.IsSuccess); + } + + [Fact] + public async Task InvokeAsync_ReturnsInternalServerError_WhenFactoryThrows() { + var factory = new Mock(); + factory + .Setup(f => f.Create(It.IsAny(), "MyActor", null)) + .Throws(new InvalidOperationException("sidecar unavailable")); + + var service = new DaprActorService( + Mock.Of>(), + factory.Object); + + var result = await service.InvokeAsync("1", "MyActor", "DoWork"); + + Assert.False(result.IsSuccess); + } +} diff --git a/src/MaksIT.Dapr.Tests/DaprLockServiceTests.cs b/src/MaksIT.Dapr.Tests/DaprLockServiceTests.cs new file mode 100644 index 0000000..60fe693 --- /dev/null +++ b/src/MaksIT.Dapr.Tests/DaprLockServiceTests.cs @@ -0,0 +1,92 @@ +using Microsoft.Extensions.Logging; +using Dapr.Client; +using Moq; +using MaksIT.Dapr.Services; + + +namespace MaksIT.Dapr.Tests; + +#pragma warning disable DAPR_DISTRIBUTEDLOCK + +public class DaprLockServiceTests { + [Fact] + public async Task LockAsync_ReturnsOk_WhenClientSucceeds() { + var response = new TryLockResponse(); + var clientMock = new Mock(); + clientMock + .Setup(x => x.Lock("lock-store", "resource", "owner", 30, It.IsAny())) + .ReturnsAsync(response); + + var service = new DaprLockService(Mock.Of>(), clientMock.Object); + + var result = await service.LockAsync("lock-store", "resource", "owner", 30); + + Assert.True(result.IsSuccess); + Assert.Same(response, result.Value); + } + + [Fact] + public async Task LockAsync_ReturnsBadRequest_WhenArgsInvalid() { + var service = new DaprLockService(Mock.Of>(), Mock.Of()); + + var empty = await service.LockAsync(" ", "resource", "owner", 30); + var badExpiry = await service.LockAsync("lock-store", "resource", "owner", 0); + + Assert.False(empty.IsSuccess); + Assert.False(badExpiry.IsSuccess); + } + + [Fact] + public async Task LockAsync_ReturnsInternalServerError_WhenClientFails() { + var clientMock = new Mock(); + clientMock + .Setup(x => x.Lock(It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny())) + .ThrowsAsync(new InvalidOperationException("lock failed")); + + var service = new DaprLockService(Mock.Of>(), clientMock.Object); + + var result = await service.LockAsync("lock-store", "resource", "owner", 30); + + Assert.False(result.IsSuccess); + } + + [Fact] + public async Task LockAsync_Rethrows_WhenCanceled() { + var clientMock = new Mock(); + clientMock + .Setup(x => x.Lock(It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny())) + .ThrowsAsync(new OperationCanceledException()); + + var service = new DaprLockService(Mock.Of>(), clientMock.Object); + + await Assert.ThrowsAsync(() => + service.LockAsync("lock-store", "resource", "owner", 30)); + } + + [Fact] + public async Task UnlockAsync_ReturnsOk_WhenClientSucceeds() { + var response = new UnlockResponse(LockStatus.Success); + var clientMock = new Mock(); + clientMock + .Setup(x => x.Unlock("lock-store", "resource", "owner", It.IsAny())) + .ReturnsAsync(response); + + var service = new DaprLockService(Mock.Of>(), clientMock.Object); + + var result = await service.UnlockAsync("lock-store", "resource", "owner"); + + Assert.True(result.IsSuccess); + Assert.Same(response, result.Value); + } + + [Fact] + public async Task UnlockAsync_ReturnsBadRequest_WhenArgsEmpty() { + var service = new DaprLockService(Mock.Of>(), Mock.Of()); + + var result = await service.UnlockAsync("lock-store", " ", "owner"); + + Assert.False(result.IsSuccess); + } +} + +#pragma warning restore DAPR_DISTRIBUTEDLOCK diff --git a/src/MaksIT.Dapr.Tests/DaprPublisherServiceTests.cs b/src/MaksIT.Dapr.Tests/DaprPubSubServiceTests.cs similarity index 52% rename from src/MaksIT.Dapr.Tests/DaprPublisherServiceTests.cs rename to src/MaksIT.Dapr.Tests/DaprPubSubServiceTests.cs index cfbd59b..89e94da 100644 --- a/src/MaksIT.Dapr.Tests/DaprPublisherServiceTests.cs +++ b/src/MaksIT.Dapr.Tests/DaprPubSubServiceTests.cs @@ -1,11 +1,12 @@ -using Dapr.Client; -using MaksIT.Dapr.Services; using Microsoft.Extensions.Logging; +using Dapr.Client; using Moq; +using MaksIT.Dapr.Services; + namespace MaksIT.Dapr.Tests; -public class DaprPublisherServiceTests { +public class DaprPubSubServiceTests { [Fact] public async Task PublishEventAsync_ReturnsOk_WhenPublishSucceeds() { var clientMock = new Mock(); @@ -17,8 +18,8 @@ public class DaprPublisherServiceTests { It.IsAny())) .Returns(Task.CompletedTask); - var service = new DaprPublisherService( - Mock.Of>(), + var service = new DaprPubSubService( + Mock.Of>(), clientMock.Object); object payload = new { Name = "payload" }; @@ -38,8 +39,8 @@ public class DaprPublisherServiceTests { It.IsAny())) .ThrowsAsync(new InvalidOperationException("publish failed")); - var service = new DaprPublisherService( - Mock.Of>(), + var service = new DaprPubSubService( + Mock.Of>(), clientMock.Object); object payload = new { Name = "payload" }; @@ -47,4 +48,34 @@ public class DaprPublisherServiceTests { Assert.False(result.IsSuccess); } + + [Fact] + public async Task PublishEventAsync_ReturnsBadRequest_WhenPubsubOrTopicEmpty() { + var service = new DaprPubSubService( + Mock.Of>(), + Mock.Of()); + + var result = await service.PublishEventAsync(" ", "topic", new { }); + + Assert.False(result.IsSuccess); + } + + [Fact] + public async Task PublishEventAsync_Rethrows_WhenCanceled() { + var clientMock = new Mock(); + clientMock + .Setup(x => x.PublishEventAsync( + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny())) + .ThrowsAsync(new OperationCanceledException()); + + var service = new DaprPubSubService( + Mock.Of>(), + clientMock.Object); + + await Assert.ThrowsAsync(() => + service.PublishEventAsync("pubsub", "topic", new { })); + } } diff --git a/src/MaksIT.Dapr.Tests/DaprRuntimeInstanceIdProviderTests.cs b/src/MaksIT.Dapr.Tests/DaprRuntimeInstanceIdProviderTests.cs new file mode 100644 index 0000000..b360039 --- /dev/null +++ b/src/MaksIT.Dapr.Tests/DaprRuntimeInstanceIdProviderTests.cs @@ -0,0 +1,13 @@ +using MaksIT.Dapr.Services.WorkLease; + + +namespace MaksIT.Dapr.Tests; + +public class DaprRuntimeInstanceIdProviderTests { + [Fact] + public void InstanceId_IsNonEmpty() { + var provider = new DaprRuntimeInstanceIdProvider(); + + Assert.False(string.IsNullOrWhiteSpace(provider.InstanceId)); + } +} diff --git a/src/MaksIT.Dapr.Tests/DaprStateStoreServiceTests.cs b/src/MaksIT.Dapr.Tests/DaprStateStoreServiceTests.cs index 6ad6e6f..672c760 100644 --- a/src/MaksIT.Dapr.Tests/DaprStateStoreServiceTests.cs +++ b/src/MaksIT.Dapr.Tests/DaprStateStoreServiceTests.cs @@ -1,7 +1,9 @@ -using Dapr.Client; -using MaksIT.Dapr.Services; using Microsoft.Extensions.Logging; +using Grpc.Core; +using Dapr.Client; using Moq; +using MaksIT.Dapr.Services; + namespace MaksIT.Dapr.Tests; @@ -28,6 +30,17 @@ public class DaprStateStoreServiceTests { Assert.True(result.IsSuccess); } + [Fact] + public async Task SetStateAsync_ReturnsBadRequest_WhenStoreOrKeyEmpty() { + var service = new DaprStateStoreService( + Mock.Of>(), + Mock.Of()); + + var result = await service.SetStateAsync(" ", "key", "value"); + + Assert.False(result.IsSuccess); + } + [Fact] public async Task GetStateAsync_ReturnsOk_WhenStateExists() { var clientMock = new Mock(); @@ -51,7 +64,7 @@ public class DaprStateStoreServiceTests { } [Fact] - public async Task GetStateAsync_ReturnsNotFound_WhenStateIsNull() { + public async Task GetStateAsync_ReturnsOkNull_WhenStateIsNull() { var clientMock = new Mock(); clientMock .Setup(x => x.GetStateAsync( @@ -68,7 +81,7 @@ public class DaprStateStoreServiceTests { var result = await service.GetStateAsync("store", "key"); - Assert.False(result.IsSuccess); + Assert.True(result.IsSuccess); Assert.Null(result.Value); } @@ -92,4 +105,138 @@ public class DaprStateStoreServiceTests { Assert.False(result.IsSuccess); } + + [Fact] + public async Task DeleteStateAsync_ReturnsOk_WhenJetStreamKeyNotFound() { + var clientMock = new Mock(); + clientMock + .Setup(x => x.DeleteStateAsync( + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny>(), + It.IsAny())) + .ThrowsAsync(CreateJetStreamKeyNotFoundException("missing", "store")); + + var logger = new Mock>(); + var service = new DaprStateStoreService(logger.Object, clientMock.Object); + + var result = await service.DeleteStateAsync("store", "key"); + + Assert.True(result.IsSuccess); + logger.Verify( + x => x.Log( + LogLevel.Error, + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny>()), + Times.Never); + } + + [Fact] + public async Task GetStateAndETagAsync_ReturnsEmpty_WhenJetStreamKeyNotFound() { + var clientMock = new Mock(); + clientMock + .Setup(x => x.GetStateAndETagAsync( + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny>(), + It.IsAny())) + .ThrowsAsync(CreateJetStreamKeyNotFoundException("identity-hub-otc-cleanup", "maksit-identity-hub-state")); + + var logger = new Mock>(); + var service = new DaprStateStoreService(logger.Object, clientMock.Object); + + var result = await service.GetStateAndETagAsync("store", "key"); + + Assert.True(result.IsSuccess); + Assert.Null(result.Value.Value); + Assert.Null(result.Value.ETag); + logger.Verify( + x => x.Log( + LogLevel.Error, + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny>()), + Times.Never); + } + + [Fact] + public async Task GetStateAsync_ReturnsOkNull_WhenJetStreamKeyNotFound() { + var clientMock = new Mock(); + clientMock + .Setup(x => x.GetStateAsync( + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny>(), + It.IsAny())) + .ThrowsAsync(CreateJetStreamKeyNotFoundException("missing-key", "store")); + + var logger = new Mock>(); + var service = new DaprStateStoreService(logger.Object, clientMock.Object); + + var result = await service.GetStateAsync("store", "missing-key"); + + Assert.True(result.IsSuccess); + Assert.Null(result.Value); + logger.Verify( + x => x.Log( + LogLevel.Error, + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny>()), + Times.Never); + } + + [Fact] + public async Task GetStateAndETagAsync_ReturnsInternalServerError_WhenOtherFailure() { + var clientMock = new Mock(); + clientMock + .Setup(x => x.GetStateAndETagAsync( + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny>(), + It.IsAny())) + .ThrowsAsync(new InvalidOperationException("connection refused")); + + var service = new DaprStateStoreService( + Mock.Of>(), + clientMock.Object); + + var result = await service.GetStateAndETagAsync("store", "key"); + + Assert.False(result.IsSuccess); + } + + [Fact] + public async Task SetStateAsync_Rethrows_WhenCanceled() { + var clientMock = new Mock(); + clientMock + .Setup(x => x.SaveStateAsync( + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny>(), + It.IsAny())) + .ThrowsAsync(new OperationCanceledException()); + + var service = new DaprStateStoreService( + Mock.Of>(), + clientMock.Object); + + await Assert.ThrowsAsync(() => + service.SetStateAsync("store", "key", "value")); + } + + private static RpcException CreateJetStreamKeyNotFoundException(string key, string storeName) { + var detail = $"fail to get {key} from state store {storeName}: nats: key not found"; + return new RpcException(new Status(StatusCode.Internal, detail)); + } } diff --git a/src/MaksIT.Dapr.Tests/DaprWorkLeaseServiceTests.cs b/src/MaksIT.Dapr.Tests/DaprWorkLeaseServiceTests.cs new file mode 100644 index 0000000..7c56369 --- /dev/null +++ b/src/MaksIT.Dapr.Tests/DaprWorkLeaseServiceTests.cs @@ -0,0 +1,486 @@ +using System.Net; +using System.Text; +using System.Text.Json; +using Microsoft.Extensions.Logging; +using Dapr.Client; +using Moq; +using MaksIT.Results; +using MaksIT.Dapr.Services; +using MaksIT.Dapr.Services.WorkLease; + + +namespace MaksIT.Dapr.Tests; + +public class DaprWorkLeaseServiceTests { + private static Mock CreateStateMock() => new(); + + [Fact] + public async Task TryAcquireAsync_Succeeds_WhenKeyMissing() { + var state = CreateStateMock(); + state + .Setup(s => s.GetStateAndETagAsync( + "store", + "work", + It.IsAny(), + It.IsAny?>(), + It.IsAny())) + .ReturnsAsync(Result<(DaprWorkLease? Value, string? ETag)>.Ok((null, null))); + state + .Setup(s => s.TrySaveStateAsync( + "store", + "work", + It.IsAny(), + null, + It.IsAny(), + It.IsAny?>(), + It.IsAny())) + .ReturnsAsync(Result.Ok(true)); + + var service = new DaprWorkLeaseService(state.Object); + var result = await service.TryAcquireAsync("store", "work", "pod-a", TimeSpan.FromMinutes(1)); + + Assert.True(result.IsSuccess); + Assert.True(result.Value); + } + + [Fact] + public async Task TryAcquireAsync_Fails_WhenHeldByOtherAndNotExpired() { + var lease = new DaprWorkLease("pod-b", DateTimeOffset.UtcNow, DateTimeOffset.UtcNow.AddMinutes(5), 3); + var state = CreateStateMock(); + state + .Setup(s => s.GetStateAndETagAsync( + "store", + "work", + It.IsAny(), + It.IsAny?>(), + It.IsAny())) + .ReturnsAsync(Result<(DaprWorkLease? Value, string? ETag)>.Ok((lease, "1"))); + + var service = new DaprWorkLeaseService(state.Object); + var result = await service.TryAcquireAsync("store", "work", "pod-a", TimeSpan.FromMinutes(1)); + + Assert.True(result.IsSuccess); + Assert.False(result.Value); + } + + [Fact] + public async Task TryAcquireAsync_TakesOver_WhenLeaseExpired_AndBumpsGeneration() { + var lease = new DaprWorkLease("pod-b", DateTimeOffset.UtcNow.AddMinutes(-10), DateTimeOffset.UtcNow.AddMinutes(-5), 7); + DaprWorkLease? saved = null; + var state = CreateStateMock(); + state + .Setup(s => s.GetStateAndETagAsync( + "store", + "work", + It.IsAny(), + It.IsAny?>(), + It.IsAny())) + .ReturnsAsync(Result<(DaprWorkLease? Value, string? ETag)>.Ok((lease, "1"))); + state + .Setup(s => s.TrySaveStateAsync( + "store", + "work", + It.IsAny(), + "1", + It.IsAny(), + It.IsAny?>(), + It.IsAny())) + .Callback?, CancellationToken>( + (_, _, value, _, _, _, _) => saved = value) + .ReturnsAsync(Result.Ok(true)); + + var service = new DaprWorkLeaseService(state.Object); + var result = await service.TryAcquireAsync("store", "work", "pod-a", TimeSpan.FromMinutes(1)); + + Assert.True(result.IsSuccess); + Assert.True(result.Value); + Assert.Equal(8, saved!.Generation); + } + + [Fact] + public async Task TryAcquireAsync_ReturnsBadRequest_WhenTtlNonPositive() { + var service = new DaprWorkLeaseService(Mock.Of()); + var result = await service.TryAcquireAsync("store", "work", "pod-a", TimeSpan.Zero); + + Assert.False(result.IsSuccess); + } + + [Fact] + public async Task TryRenewAsync_Succeeds_WhenSameHolder() { + var lease = new DaprWorkLease("pod-a", DateTimeOffset.UtcNow, DateTimeOffset.UtcNow.AddMinutes(1), 2); + var state = CreateStateMock(); + state + .Setup(s => s.GetStateAndETagAsync( + "store", + "work", + It.IsAny(), + It.IsAny?>(), + It.IsAny())) + .ReturnsAsync(Result<(DaprWorkLease? Value, string? ETag)>.Ok((lease, "2"))); + state + .Setup(s => s.TrySaveStateAsync( + "store", + "work", + It.IsAny(), + "2", + It.IsAny(), + It.IsAny?>(), + It.IsAny())) + .ReturnsAsync(Result.Ok(true)); + + var service = new DaprWorkLeaseService(state.Object); + var result = await service.TryRenewAsync("store", "work", "pod-a", TimeSpan.FromMinutes(5)); + + Assert.True(result.IsSuccess); + Assert.True(result.Value); + } + + [Fact] + public async Task TryRenewAsync_Fails_WhenDifferentHolder() { + var lease = new DaprWorkLease("pod-b", DateTimeOffset.UtcNow, DateTimeOffset.UtcNow.AddMinutes(1)); + var state = CreateStateMock(); + state + .Setup(s => s.GetStateAndETagAsync( + "store", + "work", + It.IsAny(), + It.IsAny?>(), + It.IsAny())) + .ReturnsAsync(Result<(DaprWorkLease? Value, string? ETag)>.Ok((lease, "2"))); + + var service = new DaprWorkLeaseService(state.Object); + var result = await service.TryRenewAsync("store", "work", "pod-a", TimeSpan.FromMinutes(5)); + + Assert.True(result.IsSuccess); + Assert.False(result.Value); + } + + [Fact] + public async Task ReleaseAsync_Deletes_WhenSameHolder() { + var lease = new DaprWorkLease("pod-a", DateTimeOffset.UtcNow, DateTimeOffset.UtcNow.AddMinutes(1)); + var state = CreateStateMock(); + state + .Setup(s => s.GetStateAndETagAsync( + "store", + "work", + It.IsAny(), + It.IsAny?>(), + It.IsAny())) + .ReturnsAsync(Result<(DaprWorkLease? Value, string? ETag)>.Ok((lease, "3"))); + state + .Setup(s => s.DeleteStateAsync( + "store", + "work", + It.IsAny(), + It.IsAny?>(), + It.IsAny())) + .ReturnsAsync(Result.Ok()); + + var service = new DaprWorkLeaseService(state.Object); + var result = await service.ReleaseAsync("store", "work", "pod-a"); + + Assert.True(result.IsSuccess); + state.Verify(s => s.DeleteStateAsync( + "store", + "work", + It.IsAny(), + It.IsAny?>(), + It.IsAny()), Times.Once); + } + + [Fact] + public async Task ReleaseAsync_ReturnsConflict_WhenDifferentHolder() { + var lease = new DaprWorkLease("pod-b", DateTimeOffset.UtcNow, DateTimeOffset.UtcNow.AddMinutes(1)); + var state = CreateStateMock(); + state + .Setup(s => s.GetStateAndETagAsync( + "store", + "work", + It.IsAny(), + It.IsAny?>(), + It.IsAny())) + .ReturnsAsync(Result<(DaprWorkLease? Value, string? ETag)>.Ok((lease, "3"))); + + var service = new DaprWorkLeaseService(state.Object); + var result = await service.ReleaseAsync("store", "work", "pod-a"); + + Assert.False(result.IsSuccess); + } + + [Fact] + public async Task ReleaseAsync_ReturnsOk_WhenKeyMissing() { + var state = CreateStateMock(); + state + .Setup(s => s.GetStateAndETagAsync( + "store", + "work", + It.IsAny(), + It.IsAny?>(), + It.IsAny())) + .ReturnsAsync(Result<(DaprWorkLease? Value, string? ETag)>.Ok((null, null))); + + var service = new DaprWorkLeaseService(state.Object); + var result = await service.ReleaseAsync("store", "work", "pod-a"); + + Assert.True(result.IsSuccess); + } + + [Fact] + public async Task GetAsync_ReturnsLease_WhenPresent() { + var lease = new DaprWorkLease("pod-a", DateTimeOffset.UtcNow, DateTimeOffset.UtcNow.AddMinutes(1), 4); + var state = CreateStateMock(); + state + .Setup(s => s.GetStateAndETagAsync( + "store", + "work", + It.IsAny(), + It.IsAny?>(), + It.IsAny())) + .ReturnsAsync(Result<(DaprWorkLease? Value, string? ETag)>.Ok((lease, "1"))); + + var service = new DaprWorkLeaseService(state.Object); + var result = await service.GetAsync("store", "work"); + + Assert.True(result.IsSuccess); + Assert.Equal("pod-a", result.Value!.HolderId); + Assert.Equal(4, result.Value.Generation); + } + + [Fact] + public async Task TryHoldAsync_ReturnsHold_WhenAcquired() { + var state = CreateStateMock(); + DaprWorkLease? saved = null; + state + .Setup(s => s.GetStateAndETagAsync( + "store", + "work", + It.IsAny(), + It.IsAny?>(), + It.IsAny())) + .ReturnsAsync(() => Result<(DaprWorkLease? Value, string? ETag)>.Ok((saved, saved is null ? null : "1"))); + state + .Setup(s => s.TrySaveStateAsync( + "store", + "work", + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny?>(), + It.IsAny())) + .Callback?, CancellationToken>( + (_, _, value, _, _, _, _) => saved = value) + .ReturnsAsync(Result.Ok(true)); + state + .Setup(s => s.DeleteStateAsync( + "store", + "work", + It.IsAny(), + It.IsAny?>(), + It.IsAny())) + .ReturnsAsync(Result.Ok()); + + var service = new DaprWorkLeaseService(state.Object); + var holdResult = await service.TryHoldAsync("store", "work", "pod-a", TimeSpan.FromMinutes(1), autoRenew: false); + + Assert.True(holdResult.IsSuccess); + Assert.NotNull(holdResult.Value); + Assert.Equal(1, holdResult.Value!.Generation); + + await holdResult.Value.DisposeAsync(); + state.Verify(s => s.DeleteStateAsync( + "store", + "work", + It.IsAny(), + It.IsAny?>(), + It.IsAny()), Times.Once); + } + + [Fact] + public async Task TryHoldAsync_ReturnsNull_WhenBusy() { + var lease = new DaprWorkLease("pod-b", DateTimeOffset.UtcNow, DateTimeOffset.UtcNow.AddMinutes(5), 1); + var state = CreateStateMock(); + state + .Setup(s => s.GetStateAndETagAsync( + "store", + "work", + It.IsAny(), + It.IsAny?>(), + It.IsAny())) + .ReturnsAsync(Result<(DaprWorkLease? Value, string? ETag)>.Ok((lease, "1"))); + + var service = new DaprWorkLeaseService(state.Object); + var holdResult = await service.TryHoldAsync("store", "work", "pod-a", TimeSpan.FromMinutes(1), autoRenew: false); + + Assert.True(holdResult.IsSuccess); + Assert.Null(holdResult.Value); + } +} + +public class DaprWorkLeaseBootstrapTests { + [Fact] + public async Task RunBootstrapUnderLeaseAsync_RunsBootstrap_WhenLeaseAcquired() { + var leases = new Mock(); + var hold = new DaprWorkLeaseHold(leases.Object, "store", "boot", "pod-a", TimeSpan.FromMinutes(1), 1, autoRenew: false); + leases + .Setup(l => l.TryHoldAsync("store", "boot", "pod-a", It.IsAny(), true, It.IsAny())) + .ReturnsAsync(Result.Ok(hold)); + leases + .Setup(l => l.ReleaseAsync("store", "boot", "pod-a", It.IsAny())) + .ReturnsAsync(Result.Ok()); + + var ran = false; + var result = await DaprWorkLeaseBootstrap.RunBootstrapUnderLeaseAsync( + leases.Object, + "store", + "boot", + "pod-a", + TimeSpan.FromMinutes(1), + _ => { + ran = true; + return Task.FromResult(Result.Ok()); + }, + _ => Task.FromResult(Result.Ok(true))); + + Assert.True(result.IsSuccess); + Assert.True(ran); + } + + [Fact] + public async Task RunBootstrapUnderLeaseAsync_WaitsForReady_WhenFollower() { + var leases = new Mock(); + leases + .Setup(l => l.TryHoldAsync("store", "boot", "pod-b", It.IsAny(), true, It.IsAny())) + .ReturnsAsync(Result.Ok(null)); + + var polls = 0; + var result = await DaprWorkLeaseBootstrap.RunBootstrapUnderLeaseAsync( + leases.Object, + "store", + "boot", + "pod-b", + TimeSpan.FromMinutes(1), + _ => Task.FromResult(Result.Ok()), + _ => { + polls++; + return Task.FromResult(Result.Ok(polls >= 2)); + }, + followerPollInterval: TimeSpan.FromMilliseconds(1)); + + Assert.True(result.IsSuccess); + Assert.True(polls >= 2); + } +} + +public class DaprStateStoreETagTests { + [Fact] + public async Task TrySaveStateAsync_ReturnsOkFalse_WhenClientReturnsFalse() { + var clientMock = new Mock(); + clientMock + .Setup(x => x.TrySaveStateAsync( + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny>(), + It.IsAny())) + .ReturnsAsync(false); + + var service = new DaprStateStoreService(Mock.Of>(), clientMock.Object); + var result = await service.TrySaveStateAsync("store", "key", "value", "etag"); + + Assert.True(result.IsSuccess); + Assert.False(result.Value); + } +} + +public class DaprInvocationServiceTests { + [Fact] + public async Task InvokeAsync_ReturnsBadRequest_WhenAppIdEmpty() { + var service = new DaprInvocationService(Mock.Of>(), Mock.Of()); + var result = await service.InvokeAsync(" ", "method"); + Assert.False(result.IsSuccess); + } + + [Fact] + public async Task InvokeAsync_PostsViaInvokableHttpClient_AndReturnsOk() { + var handler = new RecordingHandler(new HttpResponseMessage(HttpStatusCode.OK)); + using var http = new HttpClient(handler) { BaseAddress = new Uri("http://orders/") }; + + var client = new Mock(); + client.Setup(c => c.CreateInvokableHttpClient("orders")).Returns(http); + client.SetupGet(c => c.JsonSerializerOptions).Returns(new JsonSerializerOptions(JsonSerializerDefaults.Web)); + + using var service = new DaprInvocationService(Mock.Of>(), client.Object); + var result = await service.InvokeAsync("orders", "create", new { Id = 1 }); + + Assert.True(result.IsSuccess); + Assert.Equal(HttpMethod.Post, handler.LastMethod); + Assert.Equal(new Uri("http://orders/create"), handler.LastUri); + } + + [Fact] + public async Task InvokeAsync_DeserializesJsonResponse() { + var handler = new RecordingHandler(new HttpResponseMessage(HttpStatusCode.OK) { + Content = new StringContent("""{"total":42}""", Encoding.UTF8, "application/json") + }); + using var http = new HttpClient(handler) { BaseAddress = new Uri("http://orders/") }; + + var client = new Mock(); + client.Setup(c => c.CreateInvokableHttpClient("orders")).Returns(http); + client.SetupGet(c => c.JsonSerializerOptions).Returns(new JsonSerializerOptions(JsonSerializerDefaults.Web)); + + using var service = new DaprInvocationService(Mock.Of>(), client.Object); + var result = await service.InvokeAsync("orders", "total"); + + Assert.True(result.IsSuccess); + Assert.Equal(42, result.Value?.Total); + } + + private sealed class OrderTotal { + public int Total { get; set; } + } + + private sealed class RecordingHandler(HttpResponseMessage response) : HttpMessageHandler { + public HttpMethod? LastMethod { get; private set; } + public Uri? LastUri { get; private set; } + + protected override Task SendAsync(HttpRequestMessage request, CancellationToken cancellationToken) { + LastMethod = request.Method; + LastUri = request.RequestUri; + return Task.FromResult(response); + } + } +} + +public class DaprSidecarServiceTests { + [Fact] + public async Task CheckHealthAsync_ReturnsOk_WhenHealthy() { + var client = new Mock(); + client.Setup(c => c.CheckHealthAsync(It.IsAny())).ReturnsAsync(true); + + var service = new DaprSidecarService(Mock.Of>(), client.Object); + var result = await service.CheckHealthAsync(); + + Assert.True(result.IsSuccess); + Assert.True(result.Value); + } +} + +public class DaprSecretServiceTests { + [Fact] + public async Task GetAsync_ReturnsBadRequest_WhenStoreEmpty() { + var service = new DaprSecretService(Mock.Of>(), Mock.Of()); + var result = await service.GetAsync(" ", "key"); + Assert.False(result.IsSuccess); + } +} + +public class DaprBindingServiceTests { + [Fact] + public async Task InvokeAsync_ReturnsBadRequest_WhenBindingEmpty() { + var service = new DaprBindingService(Mock.Of>(), Mock.Of()); + var result = await service.InvokeAsync(" ", "create", new { }); + Assert.False(result.IsSuccess); + } +} diff --git a/src/MaksIT.Dapr.Tests/DaprWorkLeaseStoreTests.cs b/src/MaksIT.Dapr.Tests/DaprWorkLeaseStoreTests.cs deleted file mode 100644 index 5dac00a..0000000 --- a/src/MaksIT.Dapr.Tests/DaprWorkLeaseStoreTests.cs +++ /dev/null @@ -1,82 +0,0 @@ -using Dapr.Client; -using MaksIT.Dapr.PubSub; -using MaksIT.Dapr.Services; -using Microsoft.AspNetCore.Http; -using Microsoft.AspNetCore.Mvc; -using Microsoft.Extensions.Logging; -using Moq; - - -namespace MaksIT.Dapr.Tests; - -public class DaprWorkLeaseStoreTests { - [Fact] - public async Task TryAcquireAsync_Succeeds_WhenKeyMissing() { - var state = new Mock(); - state - .Setup(s => s.GetStateAndETagAsync("store", "work", It.IsAny())) - .ReturnsAsync(MaksIT.Results.Result<(DaprWorkLease? Value, string? ETag)>.Ok((null, null))); - state - .Setup(s => s.TrySaveStateAsync("store", "work", It.IsAny(), null, It.IsAny())) - .ReturnsAsync(MaksIT.Results.Result.Ok(true)); - - var store = new DaprWorkLeaseStore(state.Object); - var result = await store.TryAcquireAsync("store", "work", "pod-a", TimeSpan.FromMinutes(1)); - - Assert.True(result.IsSuccess); - Assert.True(result.Value); - } - - [Fact] - public async Task TryAcquireAsync_Fails_WhenHeldByOtherAndNotExpired() { - var lease = new DaprWorkLease("pod-b", DateTimeOffset.UtcNow, DateTimeOffset.UtcNow.AddMinutes(5)); - var state = new Mock(); - state - .Setup(s => s.GetStateAndETagAsync("store", "work", It.IsAny())) - .ReturnsAsync(MaksIT.Results.Result<(DaprWorkLease? Value, string? ETag)>.Ok((lease, "1"))); - - var store = new DaprWorkLeaseStore(state.Object); - var result = await store.TryAcquireAsync("store", "work", "pod-a", TimeSpan.FromMinutes(1)); - - Assert.True(result.IsSuccess); - Assert.False(result.Value); - } -} - -public class DaprPubSubAckTests { - [Fact] - public void ToActionResult_Busy_Returns503() { - var result = DaprPubSubAck.ToActionResult(DaprPubSubAcceptResult.Busy("full")); - var objectResult = Assert.IsType(result); - Assert.Equal(StatusCodes.Status503ServiceUnavailable, objectResult.StatusCode); - } - - [Fact] - public void ToActionResult_Accepted_Returns200() { - var result = DaprPubSubAck.ToActionResult(DaprPubSubAcceptResult.Accepted()); - Assert.IsType(result); - } -} - -public class DaprStateStoreETagTests { - [Fact] - public async Task TrySaveStateAsync_ReturnsOkFalse_WhenClientReturnsFalse() { - var clientMock = new Mock(); - clientMock - .Setup(x => x.TrySaveStateAsync( - It.IsAny(), - It.IsAny(), - It.IsAny(), - It.IsAny(), - It.IsAny(), - It.IsAny>(), - It.IsAny())) - .ReturnsAsync(false); - - var service = new DaprStateStoreService(Mock.Of>(), clientMock.Object); - var result = await service.TrySaveStateAsync("store", "key", "value", "etag"); - - Assert.True(result.IsSuccess); - Assert.False(result.Value); - } -} diff --git a/src/MaksIT.Dapr.Tests/DaprWorkflowServiceTests.cs b/src/MaksIT.Dapr.Tests/DaprWorkflowServiceTests.cs new file mode 100644 index 0000000..1f15880 --- /dev/null +++ b/src/MaksIT.Dapr.Tests/DaprWorkflowServiceTests.cs @@ -0,0 +1,70 @@ +using Microsoft.Extensions.Logging; +using Dapr.Workflow; +using Moq; +using MaksIT.Dapr.Services; + + +namespace MaksIT.Dapr.Tests; + +public class DaprWorkflowServiceTests { + [Fact] + public async Task ScheduleAsync_ReturnsBadRequest_WhenWorkflowNameEmpty() { + var service = new DaprWorkflowService( + Mock.Of>(), + Mock.Of()); + + var result = await service.ScheduleAsync(" "); + + Assert.False(result.IsSuccess); + } + + [Fact] + public async Task GetStateAsync_ReturnsBadRequest_WhenInstanceIdEmpty() { + var service = new DaprWorkflowService( + Mock.Of>(), + Mock.Of()); + + var result = await service.GetStateAsync(" "); + + Assert.False(result.IsSuccess); + } + + [Fact] + public async Task RaiseEventAsync_ReturnsBadRequest_WhenEventNameEmpty() { + var service = new DaprWorkflowService( + Mock.Of>(), + Mock.Of()); + + var result = await service.RaiseEventAsync("instance-1", " "); + + Assert.False(result.IsSuccess); + } + + [Fact] + public async Task TerminateAsync_ReturnsBadRequest_WhenInstanceIdEmpty() { + var service = new DaprWorkflowService( + Mock.Of>(), + Mock.Of()); + + var result = await service.TerminateAsync(" "); + + Assert.False(result.IsSuccess); + } + + [Fact] + public async Task ScheduleAsync_ReturnsOk_WhenClientSucceeds() { + var client = new Mock(); + client + .Setup(c => c.ScheduleNewWorkflowAsync("OrderFlow", null, null, null, It.IsAny())) + .ReturnsAsync("instance-1"); + + var service = new DaprWorkflowService( + Mock.Of>(), + client.Object); + + var result = await service.ScheduleAsync("OrderFlow"); + + Assert.True(result.IsSuccess); + Assert.Equal("instance-1", result.Value); + } +} diff --git a/src/MaksIT.Dapr.Tests/MaksIT.Dapr.Tests.csproj b/src/MaksIT.Dapr.Tests/MaksIT.Dapr.Tests.csproj index 3921c40..4a2663d 100644 --- a/src/MaksIT.Dapr.Tests/MaksIT.Dapr.Tests.csproj +++ b/src/MaksIT.Dapr.Tests/MaksIT.Dapr.Tests.csproj @@ -5,6 +5,7 @@ enable enable false + $(NoWarn);xUnit1051 diff --git a/src/MaksIT.Dapr.Tests/ServiceCollectionExtensionsTests.cs b/src/MaksIT.Dapr.Tests/ServiceCollectionExtensionsTests.cs new file mode 100644 index 0000000..b61f07c --- /dev/null +++ b/src/MaksIT.Dapr.Tests/ServiceCollectionExtensionsTests.cs @@ -0,0 +1,113 @@ +using Microsoft.Extensions.DependencyInjection; +using Dapr.Actors.Client; +using Dapr.Client; +using MaksIT.Dapr.Extensions; +using MaksIT.Dapr.Services; +using MaksIT.Dapr.Services.WorkLease; + + +namespace MaksIT.Dapr.Tests; + +public class ServiceCollectionExtensionsTests { + [Fact] + public void RegisterPubSub_AndStateStore_RegisterSingleDaprClient() { + var services = new ServiceCollection(); + services.AddLogging(); + + services.RegisterPubSub(); + services.RegisterStateStore(); + + Assert.Equal(1, services.Count(d => d.ServiceType == typeof(DaprClient))); + Assert.Contains(services, d => d.ServiceType == typeof(IDaprPubSubService)); + Assert.Contains(services, d => d.ServiceType == typeof(IDaprStateStoreService)); + } + + [Fact] + public void RegisterDaprClientFacades_RegistersClientBackedServices() { + var services = new ServiceCollection(); + services.AddLogging(); + + services.RegisterDaprClientFacades(); + + Assert.Equal(1, services.Count(d => d.ServiceType == typeof(DaprClient))); + Assert.Contains(services, d => d.ServiceType == typeof(IDaprPubSubService)); + Assert.Contains(services, d => d.ServiceType == typeof(IDaprStateStoreService)); + Assert.Contains(services, d => d.ServiceType == typeof(IDaprInvocationService)); + Assert.Contains(services, d => d.ServiceType == typeof(IDaprBindingService)); + Assert.Contains(services, d => d.ServiceType == typeof(IDaprSecretService)); + Assert.Contains(services, d => d.ServiceType == typeof(IDaprConfigurationService)); + Assert.Contains(services, d => d.ServiceType == typeof(IDaprCryptographyService)); + Assert.Contains(services, d => d.ServiceType == typeof(IDaprSidecarService)); + Assert.Contains(services, d => d.ServiceType == typeof(IDaprLockService)); + } + + [Fact] + public void RegisterLock_RegistersLockService() { + var services = new ServiceCollection(); + services.AddLogging(); + + services.RegisterLock(); + + Assert.Contains(services, d => d.ServiceType == typeof(IDaprLockService)); + Assert.Equal(1, services.Count(d => d.ServiceType == typeof(DaprClient))); + } + + [Fact] + public void RegisterWorkLeases_RegistersLeaseAndInstanceId() { + var services = new ServiceCollection(); + services.AddLogging(); + + services.RegisterWorkLeases(); + + Assert.Contains(services, d => d.ServiceType == typeof(IDaprWorkLeaseService)); + Assert.Contains(services, d => d.ServiceType == typeof(IDaprRuntimeInstanceId)); + Assert.Contains(services, d => d.ServiceType == typeof(IDaprStateStoreService)); + } + + [Fact] + public void RegisterWorkLeases_WithStoreName_RegistersOptions() { + var services = new ServiceCollection(); + services.AddLogging(); + + services.RegisterWorkLeases("my-state"); + + var provider = services.BuildServiceProvider(); + var options = provider.GetRequiredService(); + Assert.Equal("my-state", options.StoreName); + } + + [Fact] + public void AddDaprClientOnce_IsPerCollection_NotProcessWide() { + var first = new ServiceCollection(); + first.AddLogging(); + first.RegisterStateStore(); + + var second = new ServiceCollection(); + second.AddLogging(); + second.RegisterStateStore(); + + Assert.Equal(1, first.Count(d => d.ServiceType == typeof(DaprClient))); + Assert.Equal(1, second.Count(d => d.ServiceType == typeof(DaprClient))); + } + + [Fact] + public void RegisterActors_RegistersActorService() { + var services = new ServiceCollection(); + services.AddLogging(); + + services.RegisterActors(); + + Assert.Contains(services, d => d.ServiceType == typeof(IDaprActorService)); + Assert.Contains(services, d => d.ServiceType == typeof(IActorProxyFactory)); + } + + [Fact] + public void RegisterWorkflows_RegistersWorkflowService() { + var services = new ServiceCollection(); + services.AddLogging(); + + services.RegisterWorkflows(); + + Assert.Contains(services, d => d.ServiceType == typeof(IDaprWorkflowService)); + } +} diff --git a/src/MaksIT.Dapr/Extensions/ServiceCollectionExtensions.cs b/src/MaksIT.Dapr/Extensions/ServiceCollectionExtensions.cs index eb6f05c..56ace30 100644 --- a/src/MaksIT.Dapr/Extensions/ServiceCollectionExtensions.cs +++ b/src/MaksIT.Dapr/Extensions/ServiceCollectionExtensions.cs @@ -1,36 +1,160 @@ -using MaksIT.Dapr.Services; -using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.DependencyInjection; +using Dapr.Actors.Client; +using Dapr.Actors.Runtime; +using Dapr.Client; +using Dapr.Workflow; +using MaksIT.Dapr.Services; +using MaksIT.Dapr.Services.WorkLease; namespace MaksIT.Dapr.Extensions; +/// +/// DI registration helpers for MaksIT.Dapr services. +/// public static class ServiceCollectionExtensions { - private static bool _isDaprClientRegistered; - private static void AddDaprClientOnce(this IServiceCollection services) { - if (_isDaprClientRegistered) + if (services.Any(d => d.ServiceType == typeof(DaprClient))) return; services.AddDaprClient(); - _isDaprClientRegistered = true; } - public static void RegisterPublisher(this IServiceCollection services) { + /// + /// Registers and a when missing. + /// + public static void RegisterPubSub(this IServiceCollection services) { services.AddDaprClientOnce(); - services.AddSingleton(); + services.AddSingleton(); } + /// + /// Registers and a when missing. + /// public static void RegisterStateStore(this IServiceCollection services) { services.AddDaprClientOnce(); services.AddSingleton(); } /// - /// Registers Dapr state store plus HA work-lease coordination and runtime instance id. + /// Registers . + /// + public static void RegisterInvocation(this IServiceCollection services) { + services.AddDaprClientOnce(); + services.AddSingleton(); + } + + /// + /// Registers . + /// + public static void RegisterBinding(this IServiceCollection services) { + services.AddDaprClientOnce(); + services.AddSingleton(); + } + + /// + /// Registers . + /// + public static void RegisterSecrets(this IServiceCollection services) { + services.AddDaprClientOnce(); + services.AddSingleton(); + } + + /// + /// Registers . + /// + public static void RegisterConfiguration(this IServiceCollection services) { + services.AddDaprClientOnce(); + services.AddSingleton(); + } + + /// + /// Registers . + /// + public static void RegisterCryptography(this IServiceCollection services) { + services.AddDaprClientOnce(); + services.AddSingleton(); + } + + /// + /// Registers . + /// + public static void RegisterSidecar(this IServiceCollection services) { + services.AddDaprClientOnce(); + services.AddSingleton(); + } + + /// + /// Registers . + /// + public static void RegisterLock(this IServiceCollection services) { + services.AddDaprClientOnce(); + services.AddSingleton(); + } + + /// + /// Registers all -backed facades (pub/sub, state, invocation, binding, secrets, + /// configuration, cryptography, sidecar, lock). Actors, workflows, and work leases stay separate. + /// + public static void RegisterDaprClientFacades(this IServiceCollection services) { + services.RegisterPubSub(); + services.RegisterStateStore(); + services.RegisterInvocation(); + services.RegisterBinding(); + services.RegisterSecrets(); + services.RegisterConfiguration(); + services.RegisterCryptography(); + services.RegisterSidecar(); + services.RegisterLock(); + } + + /// + /// Registers Dapr state store plus HA work-lease service and runtime instance id. /// public static void RegisterWorkLeases(this IServiceCollection services) { services.RegisterStateStore(); services.AddSingleton(); - services.AddSingleton(); + services.AddSingleton(); + } + + /// + /// Registers work leases and binds the default Dapr state Component name for . + /// + public static void RegisterWorkLeases(this IServiceCollection services, string storeName) { + if (string.IsNullOrWhiteSpace(storeName)) + throw new ArgumentException("storeName is required.", nameof(storeName)); + + services.RegisterWorkLeases(); + services.AddSingleton(new DaprWorkLeaseOptions { StoreName = storeName }); + } + + /// + /// Registers the Dapr actor runtime and . + /// Use to options.Actors.RegisterActor<T>() and tune idle/drain settings. + /// Pair with . + /// + public static void RegisterActors(this IServiceCollection services, Action? configure = null) { + if (!services.Any(d => d.ServiceType == typeof(IActorProxyFactory))) + services.AddActors(options => configure?.Invoke(options)); + + services.AddSingleton(); + } + + /// + /// Registers Dapr workflow runtime/client and . + /// From Dapr SDK 1.18+, workflows/activities in the entry assembly are auto-discovered by the source generator; + /// pass only for explicit RegisterWorkflow / RegisterActivity or other + /// tuning. Types in referenced assemblies need + /// DaprWorkflowVersioningScanReferences in the host .csproj. + /// + public static void RegisterWorkflows(this IServiceCollection services, Action? configure = null) { + if (!services.Any(d => d.ServiceType == typeof(DaprWorkflowClient) || d.ServiceType == typeof(IDaprWorkflowClient))) { + if (configure is null) + services.AddDaprWorkflow(); + else + services.AddDaprWorkflow(configure); + } + + services.AddSingleton(); } } diff --git a/src/MaksIT.Dapr/Extensions/WebApplicationExtensions.cs b/src/MaksIT.Dapr/Extensions/WebApplicationExtensions.cs index a971866..4ac1457 100644 --- a/src/MaksIT.Dapr/Extensions/WebApplicationExtensions.cs +++ b/src/MaksIT.Dapr/Extensions/WebApplicationExtensions.cs @@ -1,7 +1,11 @@ -using Microsoft.AspNetCore.Builder; +using Microsoft.AspNetCore.Builder; namespace MaksIT.Dapr.Extensions; + +/// +/// ASP.NET Core pipeline helpers for Dapr pub/sub and actors. +/// public static class WebApplicationExtensions { /// @@ -9,7 +13,7 @@ public static class WebApplicationExtensions { /// and authorization but before any other middleware that needs to process incoming HTTP requests. /// This ensures that the UseCloudEvents and MapSubscribeHandler middleware has access to the raw HTTP request data /// before it is processed by other middleware. - /// + /// /// If you need to set controller as subscriber, use [Topic("pubsubName", "name")] attribute /// where: /// @@ -24,9 +28,17 @@ public static class WebApplicationExtensions { /// /// /// - /// + /// The application builder. public static void RegisterSubscriber(this WebApplication app) { app.UseCloudEvents(); app.MapSubscribeHandler(); } + + /// + /// Maps Dapr actor HTTP handlers. Call after + /// (typically near endpoint mapping; avoid HTTPS redirection before actors in development sidecars). + /// + /// The application builder. + public static void RegisterActorsHandlers(this WebApplication app) => + app.MapActorsHandlers(); } diff --git a/src/MaksIT.Dapr/MaksIT.Dapr.csproj b/src/MaksIT.Dapr/MaksIT.Dapr.csproj index 2c55f6c..7b0866d 100644 --- a/src/MaksIT.Dapr/MaksIT.Dapr.csproj +++ b/src/MaksIT.Dapr/MaksIT.Dapr.csproj @@ -1,17 +1,18 @@ - + net10.0 enable enable + true MaksIT.Dapr - 2.1.0 + 2.2.0 Maksym Sadovnychyy MAKS-IT MaksIT.Dapr - MaksIT.Dapr is a facade library for Dapr. + MaksIT.Dapr Result facades for Dapr pub/sub, state, invocation, bindings, secrets, configuration, cryptography, sidecar, lock, actors, workflows, and HA work-lease helpers. dotnet;dapr; https://github.com/MAKS-IT-COM/maksit-core-dapr MIT @@ -21,9 +22,9 @@ - - - + + + @@ -32,10 +33,6 @@ - - - PreserveNewest - diff --git a/src/MaksIT.Dapr/PubSub/DaprPubSubWork.cs b/src/MaksIT.Dapr/PubSub/DaprPubSubWork.cs deleted file mode 100644 index f08cab1..0000000 --- a/src/MaksIT.Dapr/PubSub/DaprPubSubWork.cs +++ /dev/null @@ -1,45 +0,0 @@ -using Microsoft.AspNetCore.Http; -using Microsoft.AspNetCore.Mvc; - - -namespace MaksIT.Dapr.PubSub; - -public enum DaprPubSubAcceptOutcome { - Accepted = 0, - AlreadyHandled = 1, - Busy = 2, - Rejected = 3, -} - -public sealed record DaprPubSubAcceptResult(DaprPubSubAcceptOutcome Outcome, string? Detail = null) { - public static DaprPubSubAcceptResult Accepted(string? detail = null) => new(DaprPubSubAcceptOutcome.Accepted, detail); - public static DaprPubSubAcceptResult AlreadyHandled(string? detail = null) => new(DaprPubSubAcceptOutcome.AlreadyHandled, detail); - public static DaprPubSubAcceptResult Busy(string? detail = null) => new(DaprPubSubAcceptOutcome.Busy, detail); - public static DaprPubSubAcceptResult Rejected(string? detail = null) => new(DaprPubSubAcceptOutcome.Rejected, detail); -} - -/// Product implements accept/claim; HTTP layer maps to Dapr ACK/NAK via . -public interface IDaprPubSubWorkHandler { - Task TryAcceptAsync(TMessage message, CancellationToken cancellationToken = default); -} - -/// Maps accept outcomes to HTTP status codes for Dapr pub/sub delivery. -public static class DaprPubSubAck { - public static IActionResult ToActionResult(DaprPubSubAcceptResult result) => - result.Outcome switch { - DaprPubSubAcceptOutcome.Accepted => new OkObjectResult(new { status = "accepted", detail = result.Detail }), - DaprPubSubAcceptOutcome.AlreadyHandled => new OkObjectResult(new { status = "alreadyHandled", detail = result.Detail }), - DaprPubSubAcceptOutcome.Busy => new ObjectResult(new { status = "busy", detail = result.Detail }) { StatusCode = StatusCodes.Status503ServiceUnavailable }, - DaprPubSubAcceptOutcome.Rejected => new BadRequestObjectResult(new { status = "rejected", detail = result.Detail }), - _ => new StatusCodeResult(StatusCodes.Status500InternalServerError), - }; - - public static Microsoft.AspNetCore.Http.IResult ToHttpResult(DaprPubSubAcceptResult result) => - result.Outcome switch { - DaprPubSubAcceptOutcome.Accepted => Microsoft.AspNetCore.Http.Results.Ok(new { status = "accepted", detail = result.Detail }), - DaprPubSubAcceptOutcome.AlreadyHandled => Microsoft.AspNetCore.Http.Results.Ok(new { status = "alreadyHandled", detail = result.Detail }), - DaprPubSubAcceptOutcome.Busy => Microsoft.AspNetCore.Http.Results.Json(new { status = "busy", detail = result.Detail }, statusCode: StatusCodes.Status503ServiceUnavailable), - DaprPubSubAcceptOutcome.Rejected => Microsoft.AspNetCore.Http.Results.BadRequest(new { status = "rejected", detail = result.Detail }), - _ => Microsoft.AspNetCore.Http.Results.StatusCode(StatusCodes.Status500InternalServerError), - }; -} diff --git a/src/MaksIT.Dapr/Services/DaprActorService.cs b/src/MaksIT.Dapr/Services/DaprActorService.cs new file mode 100644 index 0000000..a57a532 --- /dev/null +++ b/src/MaksIT.Dapr/Services/DaprActorService.cs @@ -0,0 +1,203 @@ +using Microsoft.Extensions.Logging; +using Dapr.Actors; +using Dapr.Actors.Client; +using MaksIT.Results; +using MaksIT.Core.Extensions; + + +namespace MaksIT.Dapr.Services; + +/// +/// Facade over Dapr actors: create typed clients and invoke methods with outcomes. +/// +public interface IDaprActorService { + /// + /// Creates a strongly typed actor client for . + /// + Result Create(string actorId, string actorType) where TActor : IActor; + + /// + /// Creates a weakly typed for dynamic method invocation. + /// + Result Create(string actorId, string actorType); + + /// + /// Invokes an actor method with no request or response payload. + /// + Task InvokeAsync(string actorId, string actorType, string methodName, CancellationToken cancellationToken = default); + + /// + /// Invokes an actor method with a request payload and no response. + /// + Task InvokeAsync(string actorId, string actorType, string methodName, TRequest data, CancellationToken cancellationToken = default); + + /// + /// Invokes an actor method with no request payload and a typed response. + /// + Task> InvokeAsync(string actorId, string actorType, string methodName, CancellationToken cancellationToken = default); + + /// + /// Invokes an actor method with a request payload and a typed response. + /// + Task> InvokeAsync(string actorId, string actorType, string methodName, TRequest data, CancellationToken cancellationToken = default); +} + +/// +/// Default using Dapr's . +/// +public class DaprActorService : IDaprActorService { + private const string ErrorMessage = "MaksIT.Dapr - Actor service error"; + + private readonly IActorProxyFactory _actorProxyFactory; + private readonly ILogger _logger; + + /// + /// Creates an actor service backed by . + /// + public DaprActorService(ILogger logger, IActorProxyFactory actorProxyFactory) { + _logger = logger; + _actorProxyFactory = actorProxyFactory; + } + + /// + public Result Create(string actorId, string actorType) where TActor : IActor { + var validation = ValidateActorKey(actorId, actorType); + if (!validation.IsSuccess) + return Result.BadRequest(default!, validation.Messages.ToArray()); + + try { + var actor = _actorProxyFactory.CreateActorProxy(new ActorId(actorId), actorType); + return Result.Ok(actor); + } + catch (Exception ex) { + _logger.LogError(ex, ErrorMessage); + return Result.InternalServerError(default!, [ErrorMessage, .. ex.ExtractMessages()]); + } + } + + /// + public Result Create(string actorId, string actorType) { + var validation = ValidateActorKey(actorId, actorType); + if (!validation.IsSuccess) + return Result.BadRequest(default!, validation.Messages.ToArray()); + + try { + var actor = _actorProxyFactory.Create(new ActorId(actorId), actorType); + return Result.Ok(actor); + } + catch (Exception ex) { + _logger.LogError(ex, ErrorMessage); + return Result.InternalServerError(default!, [ErrorMessage, .. ex.ExtractMessages()]); + } + } + + /// + public async Task InvokeAsync( + string actorId, + string actorType, + string methodName, + CancellationToken cancellationToken = default) { + var validation = ValidateInvoke(actorId, actorType, methodName); + if (!validation.IsSuccess) + return validation; + + try { + var actor = _actorProxyFactory.Create(new ActorId(actorId), actorType); + await actor.InvokeMethodAsync(methodName, cancellationToken); + return Result.Ok(); + } + catch (OperationCanceledException) { + throw; + } + catch (Exception ex) { + _logger.LogError(ex, ErrorMessage); + return Result.InternalServerError([ErrorMessage, .. ex.ExtractMessages()]); + } + } + + /// + public async Task InvokeAsync( + string actorId, + string actorType, + string methodName, + TRequest data, + CancellationToken cancellationToken = default) { + var validation = ValidateInvoke(actorId, actorType, methodName); + if (!validation.IsSuccess) + return validation; + + try { + var actor = _actorProxyFactory.Create(new ActorId(actorId), actorType); + await actor.InvokeMethodAsync(methodName, data, cancellationToken); + return Result.Ok(); + } + catch (OperationCanceledException) { + throw; + } + catch (Exception ex) { + _logger.LogError(ex, ErrorMessage); + return Result.InternalServerError([ErrorMessage, .. ex.ExtractMessages()]); + } + } + + /// + public async Task> InvokeAsync( + string actorId, + string actorType, + string methodName, + CancellationToken cancellationToken = default) { + var validation = ValidateInvoke(actorId, actorType, methodName); + if (!validation.IsSuccess) + return validation.ToResultOfType(default); + + try { + var actor = _actorProxyFactory.Create(new ActorId(actorId), actorType); + var response = await actor.InvokeMethodAsync(methodName, cancellationToken); + return Result.Ok(response); + } + catch (OperationCanceledException) { + throw; + } + catch (Exception ex) { + _logger.LogError(ex, ErrorMessage); + return Result.InternalServerError(default, [ErrorMessage, .. ex.ExtractMessages()]); + } + } + + /// + public async Task> InvokeAsync( + string actorId, + string actorType, + string methodName, + TRequest data, + CancellationToken cancellationToken = default) { + var validation = ValidateInvoke(actorId, actorType, methodName); + if (!validation.IsSuccess) + return validation.ToResultOfType(default); + + try { + var actor = _actorProxyFactory.Create(new ActorId(actorId), actorType); + var response = await actor.InvokeMethodAsync(methodName, data, cancellationToken); + return Result.Ok(response); + } + catch (OperationCanceledException) { + throw; + } + catch (Exception ex) { + _logger.LogError(ex, ErrorMessage); + return Result.InternalServerError(default, [ErrorMessage, .. ex.ExtractMessages()]); + } + } + + private static Result ValidateActorKey(string actorId, string actorType) { + if (string.IsNullOrWhiteSpace(actorId) || string.IsNullOrWhiteSpace(actorType)) + return Result.BadRequest("actorId and actorType are required."); + return Result.Ok(); + } + + private static Result ValidateInvoke(string actorId, string actorType, string methodName) { + if (string.IsNullOrWhiteSpace(actorId) || string.IsNullOrWhiteSpace(actorType) || string.IsNullOrWhiteSpace(methodName)) + return Result.BadRequest("actorId, actorType, and methodName are required."); + return Result.Ok(); + } +} diff --git a/src/MaksIT.Dapr/Services/DaprBindingService.cs b/src/MaksIT.Dapr/Services/DaprBindingService.cs new file mode 100644 index 0000000..17805ec --- /dev/null +++ b/src/MaksIT.Dapr/Services/DaprBindingService.cs @@ -0,0 +1,88 @@ +using Microsoft.Extensions.Logging; +using Dapr.Client; +using MaksIT.Results; +using MaksIT.Core.Extensions; + + +namespace MaksIT.Dapr.Services; + +/// +/// Dapr output bindings with outcomes. +/// +public interface IDaprBindingService { + /// + /// Invokes a binding operation. + /// + Task InvokeAsync( + string bindingName, + string operation, + TRequest data, + IReadOnlyDictionary? metadata = null, + CancellationToken cancellationToken = default); + + /// + /// Invokes a binding operation and deserializes the response. + /// + Task> InvokeAsync( + string bindingName, + string operation, + TRequest data, + IReadOnlyDictionary? metadata = null, + CancellationToken cancellationToken = default); +} + +/// +/// Default . +/// +public class DaprBindingService( + ILogger logger, + DaprClient client +) : IDaprBindingService { + private const string ErrorMessage = "MaksIT.Dapr - Binding error"; + + /// + public async Task InvokeAsync( + string bindingName, + string operation, + TRequest data, + IReadOnlyDictionary? metadata = null, + CancellationToken cancellationToken = default) { + if (string.IsNullOrWhiteSpace(bindingName) || string.IsNullOrWhiteSpace(operation)) + return Result.BadRequest("bindingName and operation are required."); + + try { + await client.InvokeBindingAsync(bindingName, operation, data, metadata, cancellationToken); + return Result.Ok(); + } + catch (OperationCanceledException) { + throw; + } + catch (Exception ex) { + logger.LogError(ex, ErrorMessage); + return Result.InternalServerError([ErrorMessage, .. ex.ExtractMessages()]); + } + } + + /// + public async Task> InvokeAsync( + string bindingName, + string operation, + TRequest data, + IReadOnlyDictionary? metadata = null, + CancellationToken cancellationToken = default) { + if (string.IsNullOrWhiteSpace(bindingName) || string.IsNullOrWhiteSpace(operation)) + return Result.BadRequest(default, "bindingName and operation are required."); + + try { + var response = await client.InvokeBindingAsync(bindingName, operation, data, metadata, cancellationToken); + return Result.Ok(response); + } + catch (OperationCanceledException) { + throw; + } + catch (Exception ex) { + logger.LogError(ex, ErrorMessage); + return Result.InternalServerError(default, [ErrorMessage, .. ex.ExtractMessages()]); + } + } +} diff --git a/src/MaksIT.Dapr/Services/DaprConfigurationService.cs b/src/MaksIT.Dapr/Services/DaprConfigurationService.cs new file mode 100644 index 0000000..0af042d --- /dev/null +++ b/src/MaksIT.Dapr/Services/DaprConfigurationService.cs @@ -0,0 +1,107 @@ +using Microsoft.Extensions.Logging; +using Dapr.Client; +using MaksIT.Results; +using MaksIT.Core.Extensions; + + +namespace MaksIT.Dapr.Services; + +/// +/// Dapr configuration API with outcomes. +/// +public interface IDaprConfigurationService { + /// + /// Gets configuration items. + /// + Task> GetAsync( + string storeName, + IReadOnlyList keys, + IReadOnlyDictionary? metadata = null, + CancellationToken cancellationToken = default); + + /// + /// Subscribes to configuration changes. + /// + Task> SubscribeAsync( + string storeName, + IReadOnlyList keys, + IReadOnlyDictionary? metadata = null, + CancellationToken cancellationToken = default); + + /// + /// Unsubscribes from configuration changes. + /// + Task UnsubscribeAsync(string storeName, string id, CancellationToken cancellationToken = default); +} + +/// +/// Default . +/// +public class DaprConfigurationService( + ILogger logger, + DaprClient client +) : IDaprConfigurationService { + private const string ErrorMessage = "MaksIT.Dapr - Configuration error"; + + /// + public async Task> GetAsync( + string storeName, + IReadOnlyList keys, + IReadOnlyDictionary? metadata = null, + CancellationToken cancellationToken = default) { + if (string.IsNullOrWhiteSpace(storeName) || keys is null || keys.Count == 0) + return Result.BadRequest(default!, "storeName and keys are required."); + + try { + var response = await client.GetConfiguration(storeName, keys, metadata, cancellationToken); + return Result.Ok(response); + } + catch (OperationCanceledException) { + throw; + } + catch (Exception ex) { + logger.LogError(ex, ErrorMessage); + return Result.InternalServerError(default!, [ErrorMessage, .. ex.ExtractMessages()]); + } + } + + /// + public async Task> SubscribeAsync( + string storeName, + IReadOnlyList keys, + IReadOnlyDictionary? metadata = null, + CancellationToken cancellationToken = default) { + if (string.IsNullOrWhiteSpace(storeName) || keys is null || keys.Count == 0) + return Result.BadRequest(default!, "storeName and keys are required."); + + try { + var response = await client.SubscribeConfiguration(storeName, keys, metadata, cancellationToken); + return Result.Ok(response); + } + catch (OperationCanceledException) { + throw; + } + catch (Exception ex) { + logger.LogError(ex, ErrorMessage); + return Result.InternalServerError(default!, [ErrorMessage, .. ex.ExtractMessages()]); + } + } + + /// + public async Task UnsubscribeAsync(string storeName, string id, CancellationToken cancellationToken = default) { + if (string.IsNullOrWhiteSpace(storeName) || string.IsNullOrWhiteSpace(id)) + return Result.BadRequest("storeName and id are required."); + + try { + _ = await client.UnsubscribeConfiguration(storeName, id, cancellationToken); + return Result.Ok(); + } + catch (OperationCanceledException) { + throw; + } + catch (Exception ex) { + logger.LogError(ex, ErrorMessage); + return Result.InternalServerError([ErrorMessage, .. ex.ExtractMessages()]); + } + } +} diff --git a/src/MaksIT.Dapr/Services/DaprCryptographyService.cs b/src/MaksIT.Dapr/Services/DaprCryptographyService.cs new file mode 100644 index 0000000..113216d --- /dev/null +++ b/src/MaksIT.Dapr/Services/DaprCryptographyService.cs @@ -0,0 +1,96 @@ +#pragma warning disable DAPR_CRYPTOGRAPHY + +using Microsoft.Extensions.Logging; +using Dapr.Client; +using MaksIT.Results; +using MaksIT.Core.Extensions; + + +namespace MaksIT.Dapr.Services; + +/// +/// Dapr cryptography building block with outcomes. +/// +public interface IDaprCryptographyService { + /// + /// Encrypts plaintext bytes. + /// + Task>> EncryptAsync( + string componentName, + ReadOnlyMemory plaintext, + string keyName, + EncryptionOptions options, + CancellationToken cancellationToken = default); + + /// + /// Decrypts ciphertext bytes. + /// + Task>> DecryptAsync( + string componentName, + ReadOnlyMemory ciphertext, + string keyName, + DecryptionOptions? options = null, + CancellationToken cancellationToken = default); +} + +/// +/// Default . +/// +public class DaprCryptographyService( + ILogger logger, + DaprClient client +) : IDaprCryptographyService { + private const string ErrorMessage = "MaksIT.Dapr - Cryptography error"; + + /// + public async Task>> EncryptAsync( + string componentName, + ReadOnlyMemory plaintext, + string keyName, + EncryptionOptions options, + CancellationToken cancellationToken = default) { + if (string.IsNullOrWhiteSpace(componentName) || string.IsNullOrWhiteSpace(keyName)) + return Result>.BadRequest(default, "componentName and keyName are required."); + if (options is null) + return Result>.BadRequest(default, "options are required."); + + try { + var ciphertext = await client.EncryptAsync(componentName, plaintext, keyName, options, cancellationToken); + return Result>.Ok(ciphertext); + } + catch (OperationCanceledException) { + throw; + } + catch (Exception ex) { + logger.LogError(ex, ErrorMessage); + return Result>.InternalServerError(default, [ErrorMessage, .. ex.ExtractMessages()]); + } + } + + /// + public async Task>> DecryptAsync( + string componentName, + ReadOnlyMemory ciphertext, + string keyName, + DecryptionOptions? options = null, + CancellationToken cancellationToken = default) { + if (string.IsNullOrWhiteSpace(componentName) || string.IsNullOrWhiteSpace(keyName)) + return Result>.BadRequest(default, "componentName and keyName are required."); + + try { + var plaintext = options is null + ? await client.DecryptAsync(componentName, ciphertext, keyName, cancellationToken) + : await client.DecryptAsync(componentName, ciphertext, keyName, options, cancellationToken); + return Result>.Ok(plaintext); + } + catch (OperationCanceledException) { + throw; + } + catch (Exception ex) { + logger.LogError(ex, ErrorMessage); + return Result>.InternalServerError(default, [ErrorMessage, .. ex.ExtractMessages()]); + } + } +} + +#pragma warning restore DAPR_CRYPTOGRAPHY diff --git a/src/MaksIT.Dapr/Services/DaprInvocationService.cs b/src/MaksIT.Dapr/Services/DaprInvocationService.cs new file mode 100644 index 0000000..8270798 --- /dev/null +++ b/src/MaksIT.Dapr/Services/DaprInvocationService.cs @@ -0,0 +1,155 @@ +using System.Collections.Concurrent; +using System.Net.Http.Json; +using Microsoft.Extensions.Logging; +using Dapr.Client; +using MaksIT.Results; +using MaksIT.Core.Extensions; + + +namespace MaksIT.Dapr.Services; + +/// +/// Dapr service invocation with outcomes. +/// +public interface IDaprInvocationService { + /// + /// Invokes a method with no request body (HTTP POST). + /// + Task InvokeAsync(string appId, string methodName, CancellationToken cancellationToken = default); + + /// + /// Invokes a method with a JSON request body (HTTP POST). + /// + Task InvokeAsync(string appId, string methodName, TRequest data, CancellationToken cancellationToken = default); + + /// + /// Invokes a method with no request body and deserializes the JSON response (HTTP POST). + /// + Task> InvokeAsync(string appId, string methodName, CancellationToken cancellationToken = default); + + /// + /// Invokes a method with a JSON request body and deserializes the JSON response (HTTP POST). + /// + Task> InvokeAsync(string appId, string methodName, TRequest data, CancellationToken cancellationToken = default); +} + +/// +/// Default using . +/// +public sealed class DaprInvocationService( + ILogger logger, + DaprClient client +) : IDaprInvocationService, IDisposable { + private const string ErrorMessage = "MaksIT.Dapr - Invocation error"; + + private readonly ConcurrentDictionary _clients = new(StringComparer.Ordinal); + private bool _disposed; + + /// + public async Task InvokeAsync(string appId, string methodName, CancellationToken cancellationToken = default) { + if (string.IsNullOrWhiteSpace(appId) || string.IsNullOrWhiteSpace(methodName)) + return Result.BadRequest("appId and methodName are required."); + + try { + using var response = await GetClient(appId).PostAsync(methodName, content: null, cancellationToken); + response.EnsureSuccessStatusCode(); + return Result.Ok(); + } + catch (OperationCanceledException) { + throw; + } + catch (Exception ex) { + logger.LogError(ex, ErrorMessage); + return Result.InternalServerError([ErrorMessage, .. ex.ExtractMessages()]); + } + } + + /// + public async Task InvokeAsync(string appId, string methodName, TRequest data, CancellationToken cancellationToken = default) { + if (string.IsNullOrWhiteSpace(appId) || string.IsNullOrWhiteSpace(methodName)) + return Result.BadRequest("appId and methodName are required."); + + try { + using var response = await GetClient(appId).PostAsJsonAsync(methodName, data, client.JsonSerializerOptions, cancellationToken); + response.EnsureSuccessStatusCode(); + return Result.Ok(); + } + catch (OperationCanceledException) { + throw; + } + catch (Exception ex) { + logger.LogError(ex, ErrorMessage); + return Result.InternalServerError([ErrorMessage, .. ex.ExtractMessages()]); + } + } + + /// + public async Task> InvokeAsync(string appId, string methodName, CancellationToken cancellationToken = default) { + if (string.IsNullOrWhiteSpace(appId) || string.IsNullOrWhiteSpace(methodName)) + return Result.BadRequest(default, "appId and methodName are required."); + + try { + using var response = await GetClient(appId).PostAsync(methodName, content: null, cancellationToken); + response.EnsureSuccessStatusCode(); + var body = await response.Content.ReadFromJsonAsync(client.JsonSerializerOptions, cancellationToken); + return Result.Ok(body); + } + catch (OperationCanceledException) { + throw; + } + catch (Exception ex) { + logger.LogError(ex, ErrorMessage); + return Result.InternalServerError(default, [ErrorMessage, .. ex.ExtractMessages()]); + } + } + + /// + public async Task> InvokeAsync( + string appId, + string methodName, + TRequest data, + CancellationToken cancellationToken = default) { + if (string.IsNullOrWhiteSpace(appId) || string.IsNullOrWhiteSpace(methodName)) + return Result.BadRequest(default, "appId and methodName are required."); + + try { + using var response = await GetClient(appId).PostAsJsonAsync(methodName, data, client.JsonSerializerOptions, cancellationToken); + response.EnsureSuccessStatusCode(); + var body = await response.Content.ReadFromJsonAsync(client.JsonSerializerOptions, cancellationToken); + return Result.Ok(body); + } + catch (OperationCanceledException) { + throw; + } + catch (Exception ex) { + logger.LogError(ex, ErrorMessage); + return Result.InternalServerError(default, [ErrorMessage, .. ex.ExtractMessages()]); + } + } + + /// + public void Dispose() { + if (_disposed) + return; + + _disposed = true; + foreach (var http in _clients.Values) + http.Dispose(); + + _clients.Clear(); + } + + private HttpClient GetClient(string appId) { + ObjectDisposedException.ThrowIf(_disposed, this); + + if (_clients.TryGetValue(appId, out var existing)) + return existing; + + var created = client.CreateInvokableHttpClient(appId); + if (_clients.TryAdd(appId, created)) + return created; + + created.Dispose(); + return _clients[appId]; + } +} diff --git a/src/MaksIT.Dapr/Services/DaprLockService.cs b/src/MaksIT.Dapr/Services/DaprLockService.cs new file mode 100644 index 0000000..bf9c96b --- /dev/null +++ b/src/MaksIT.Dapr/Services/DaprLockService.cs @@ -0,0 +1,97 @@ +#pragma warning disable DAPR_DISTRIBUTEDLOCK + +using Microsoft.Extensions.Logging; +using Dapr.Client; +using MaksIT.Results; +using MaksIT.Core.Extensions; +using MaksIT.Dapr.Services.WorkLease; + + +namespace MaksIT.Dapr.Services; + +/// +/// Dapr distributed lock API with outcomes. +/// +/// +/// Short-lived mutex over a Dapr lock Component. For MaksIT multi-replica leader/bootstrap/sweep +/// coordination prefer (state-backed named leases with renew/hold helpers). +/// +public interface IDaprLockService { + /// + /// Attempts to acquire a lock. Check ; dispose the response when done (or call ). + /// + Task> LockAsync( + string storeName, + string resourceId, + string lockOwner, + int expiryInSeconds, + CancellationToken cancellationToken = default); + + /// + /// Releases a lock held by . + /// + Task> UnlockAsync( + string storeName, + string resourceId, + string lockOwner, + CancellationToken cancellationToken = default); +} + +/// +/// Default . +/// +public class DaprLockService( + ILogger logger, + DaprClient client +) : IDaprLockService { + private const string ErrorMessage = "MaksIT.Dapr - Lock error"; + + /// + public async Task> LockAsync( + string storeName, + string resourceId, + string lockOwner, + int expiryInSeconds, + CancellationToken cancellationToken = default) { + if (string.IsNullOrWhiteSpace(storeName) || string.IsNullOrWhiteSpace(resourceId) || string.IsNullOrWhiteSpace(lockOwner)) + return Result.BadRequest(default!, "storeName, resourceId, and lockOwner are required."); + if (expiryInSeconds <= 0) + return Result.BadRequest(default!, "expiryInSeconds must be positive."); + + try { + var response = await client.Lock(storeName, resourceId, lockOwner, expiryInSeconds, cancellationToken); + return Result.Ok(response); + } + catch (OperationCanceledException) { + throw; + } + catch (Exception ex) { + logger.LogError(ex, ErrorMessage); + return Result.InternalServerError(default!, [ErrorMessage, .. ex.ExtractMessages()]); + } + } + + /// + public async Task> UnlockAsync( + string storeName, + string resourceId, + string lockOwner, + CancellationToken cancellationToken = default) { + if (string.IsNullOrWhiteSpace(storeName) || string.IsNullOrWhiteSpace(resourceId) || string.IsNullOrWhiteSpace(lockOwner)) + return Result.BadRequest(default!, "storeName, resourceId, and lockOwner are required."); + + try { + var response = await client.Unlock(storeName, resourceId, lockOwner, cancellationToken); + return Result.Ok(response); + } + catch (OperationCanceledException) { + throw; + } + catch (Exception ex) { + logger.LogError(ex, ErrorMessage); + return Result.InternalServerError(default!, [ErrorMessage, .. ex.ExtractMessages()]); + } + } +} + +#pragma warning restore DAPR_DISTRIBUTEDLOCK diff --git a/src/MaksIT.Dapr/Services/DaprPubSubService.cs b/src/MaksIT.Dapr/Services/DaprPubSubService.cs new file mode 100644 index 0000000..7862ca0 --- /dev/null +++ b/src/MaksIT.Dapr/Services/DaprPubSubService.cs @@ -0,0 +1,145 @@ +using Microsoft.Extensions.Logging; +using Dapr.Client; +using MaksIT.Results; +using MaksIT.Core.Extensions; + + +namespace MaksIT.Dapr.Services; + +/// +/// Publishes events to a Dapr pub/sub component. +/// +public interface IDaprPubSubService { + /// + /// Publishes to on . + /// + Task PublishEventAsync( + string pubsubName, + string topicName, + object payload, + IReadOnlyDictionary? metadata = null, + CancellationToken cancellationToken = default); + + /// + /// Publishes a raw byte payload. + /// + Task PublishByteEventAsync( + string pubsubName, + string topicName, + ReadOnlyMemory data, + string contentType = "application/json", + IReadOnlyDictionary? metadata = null, + CancellationToken cancellationToken = default); + + /// + /// Publishes multiple events; returns the Dapr bulk response (including failed entries). + /// + Task>> BulkPublishEventAsync( + string pubsubName, + string topicName, + IReadOnlyList events, + Dictionary? metadata = null, + CancellationToken cancellationToken = default); +} + +/// +/// Default using . +/// +public class DaprPubSubService : IDaprPubSubService { + private const string ErrorMessage = "MaksIT.Dapr - Pub/sub error"; + + private readonly DaprClient _client; + private readonly ILogger _logger; + + /// + /// Creates a pub/sub facade backed by . + /// + public DaprPubSubService(ILogger logger, DaprClient client) { + _logger = logger; + _client = client; + } + + /// + public async Task PublishEventAsync( + string pubsubName, + string topicName, + object payload, + IReadOnlyDictionary? metadata = null, + CancellationToken cancellationToken = default) { + if (string.IsNullOrWhiteSpace(pubsubName) || string.IsNullOrWhiteSpace(topicName)) + return Result.BadRequest("pubsubName and topicName are required."); + + try { + if (metadata is null) + await _client.PublishEventAsync(pubsubName, topicName, payload, cancellationToken); + else + await _client.PublishEventAsync( + pubsubName, + topicName, + payload, + metadata as Dictionary ?? metadata.ToDictionary(static kv => kv.Key, static kv => kv.Value), + cancellationToken); + return Result.Ok(); + } + catch (OperationCanceledException) { + throw; + } + catch (Exception ex) { + _logger.LogError(ex, ErrorMessage); + return Result.InternalServerError([ErrorMessage, .. ex.ExtractMessages()]); + } + } + + /// + public async Task PublishByteEventAsync( + string pubsubName, + string topicName, + ReadOnlyMemory data, + string contentType = "application/json", + IReadOnlyDictionary? metadata = null, + CancellationToken cancellationToken = default) { + if (string.IsNullOrWhiteSpace(pubsubName) || string.IsNullOrWhiteSpace(topicName)) + return Result.BadRequest("pubsubName and topicName are required."); + + try { + await _client.PublishByteEventAsync( + pubsubName, + topicName, + data, + contentType, + metadata as Dictionary ?? metadata?.ToDictionary(static kv => kv.Key, static kv => kv.Value), + cancellationToken); + return Result.Ok(); + } + catch (OperationCanceledException) { + throw; + } + catch (Exception ex) { + _logger.LogError(ex, ErrorMessage); + return Result.InternalServerError([ErrorMessage, .. ex.ExtractMessages()]); + } + } + + /// + public async Task>> BulkPublishEventAsync( + string pubsubName, + string topicName, + IReadOnlyList events, + Dictionary? metadata = null, + CancellationToken cancellationToken = default) { + if (string.IsNullOrWhiteSpace(pubsubName) || string.IsNullOrWhiteSpace(topicName)) + return Result>.BadRequest(default!, "pubsubName and topicName are required."); + + try { + var response = await _client.BulkPublishEventAsync(pubsubName, topicName, events, metadata, cancellationToken); + return Result>.Ok(response); + } + catch (OperationCanceledException) { + throw; + } + catch (Exception ex) { + _logger.LogError(ex, ErrorMessage); + return Result>.InternalServerError(default!, [ErrorMessage, .. ex.ExtractMessages()]); + } + } +} diff --git a/src/MaksIT.Dapr/Services/DaprPublisherService.cs b/src/MaksIT.Dapr/Services/DaprPublisherService.cs deleted file mode 100644 index eebb677..0000000 --- a/src/MaksIT.Dapr/Services/DaprPublisherService.cs +++ /dev/null @@ -1,37 +0,0 @@ -using Microsoft.Extensions.Logging; - -using Dapr.Client; - -using MaksIT.Results; -using MaksIT.Core.Extensions; - -namespace MaksIT.Dapr.Services; -public interface IDaprPublisherService { - Task PublishEventAsync(string pubsubName, string topicName, object payload); -} - -public class DaprPublisherService : IDaprPublisherService { - private const string _errorMessage = "MaksIT.Dapr - Event publishing error"; - - private readonly DaprClient _client; - private readonly ILogger _logger; - - public DaprPublisherService( - ILogger logger, - DaprClient client - ) { - _logger = logger; - _client = client; - } - - public async Task PublishEventAsync(string pubsubName, string topicName, object payload) { - try { - await _client.PublishEventAsync(pubsubName, topicName, payload); - return Result.Ok(); - } - catch (Exception ex) { - _logger.LogError(ex, _errorMessage); - return Result.InternalServerError([_errorMessage, .. ex.ExtractMessages()]); - } - } -} \ No newline at end of file diff --git a/src/MaksIT.Dapr/Services/DaprSecretService.cs b/src/MaksIT.Dapr/Services/DaprSecretService.cs new file mode 100644 index 0000000..67e7e20 --- /dev/null +++ b/src/MaksIT.Dapr/Services/DaprSecretService.cs @@ -0,0 +1,82 @@ +using Microsoft.Extensions.Logging; +using Dapr.Client; +using MaksIT.Results; +using MaksIT.Core.Extensions; + + +namespace MaksIT.Dapr.Services; + +/// +/// Dapr secret store operations with outcomes. +/// +public interface IDaprSecretService { + /// + /// Gets a secret by name. + /// + Task>> GetAsync( + string storeName, + string key, + IReadOnlyDictionary? metadata = null, + CancellationToken cancellationToken = default); + + /// + /// Gets all secrets from a store. + /// + Task>>> GetBulkAsync( + string storeName, + IReadOnlyDictionary? metadata = null, + CancellationToken cancellationToken = default); +} + +/// +/// Default . +/// +public class DaprSecretService( + ILogger logger, + DaprClient client +) : IDaprSecretService { + private const string ErrorMessage = "MaksIT.Dapr - Secret error"; + + /// + public async Task>> GetAsync( + string storeName, + string key, + IReadOnlyDictionary? metadata = null, + CancellationToken cancellationToken = default) { + if (string.IsNullOrWhiteSpace(storeName) || string.IsNullOrWhiteSpace(key)) + return Result>.BadRequest(default!, "storeName and key are required."); + + try { + var secrets = await client.GetSecretAsync(storeName, key, metadata, cancellationToken); + return Result>.Ok(secrets); + } + catch (OperationCanceledException) { + throw; + } + catch (Exception ex) { + logger.LogError(ex, ErrorMessage); + return Result>.InternalServerError(default!, [ErrorMessage, .. ex.ExtractMessages()]); + } + } + + /// + public async Task>>> GetBulkAsync( + string storeName, + IReadOnlyDictionary? metadata = null, + CancellationToken cancellationToken = default) { + if (string.IsNullOrWhiteSpace(storeName)) + return Result>>.BadRequest(default!, "storeName is required."); + + try { + var secrets = await client.GetBulkSecretAsync(storeName, metadata, cancellationToken); + return Result>>.Ok(secrets); + } + catch (OperationCanceledException) { + throw; + } + catch (Exception ex) { + logger.LogError(ex, ErrorMessage); + return Result>>.InternalServerError(default!, [ErrorMessage, .. ex.ExtractMessages()]); + } + } +} diff --git a/src/MaksIT.Dapr/Services/DaprSidecarService.cs b/src/MaksIT.Dapr/Services/DaprSidecarService.cs new file mode 100644 index 0000000..19e1524 --- /dev/null +++ b/src/MaksIT.Dapr/Services/DaprSidecarService.cs @@ -0,0 +1,122 @@ +using Microsoft.Extensions.Logging; +using Dapr.Client; +using MaksIT.Results; +using MaksIT.Core.Extensions; + + +namespace MaksIT.Dapr.Services; + +/// +/// Dapr sidecar health and metadata with outcomes. +/// +public interface IDaprSidecarService { + /// + /// Checks sidecar health. + /// + Task> CheckHealthAsync(CancellationToken cancellationToken = default); + + /// + /// Checks outbound health. + /// + Task> CheckOutboundHealthAsync(CancellationToken cancellationToken = default); + + /// + /// Blocks until the sidecar is ready. + /// + Task WaitForSidecarAsync(CancellationToken cancellationToken = default); + + /// + /// Gets sidecar metadata. + /// + Task> GetMetadataAsync(CancellationToken cancellationToken = default); + + /// + /// Requests sidecar shutdown. + /// + Task ShutdownAsync(CancellationToken cancellationToken = default); +} + +/// +/// Default . +/// +public class DaprSidecarService( + ILogger logger, + DaprClient client +) : IDaprSidecarService { + private const string ErrorMessage = "MaksIT.Dapr - Sidecar error"; + + /// + public async Task> CheckHealthAsync(CancellationToken cancellationToken = default) { + try { + var healthy = await client.CheckHealthAsync(cancellationToken); + return Result.Ok(healthy); + } + catch (OperationCanceledException) { + throw; + } + catch (Exception ex) { + logger.LogError(ex, ErrorMessage); + return Result.InternalServerError(false, [ErrorMessage, .. ex.ExtractMessages()]); + } + } + + /// + public async Task> CheckOutboundHealthAsync(CancellationToken cancellationToken = default) { + try { + var healthy = await client.CheckOutboundHealthAsync(cancellationToken); + return Result.Ok(healthy); + } + catch (OperationCanceledException) { + throw; + } + catch (Exception ex) { + logger.LogError(ex, ErrorMessage); + return Result.InternalServerError(false, [ErrorMessage, .. ex.ExtractMessages()]); + } + } + + /// + public async Task WaitForSidecarAsync(CancellationToken cancellationToken = default) { + try { + await client.WaitForSidecarAsync(cancellationToken); + return Result.Ok(); + } + catch (OperationCanceledException) { + throw; + } + catch (Exception ex) { + logger.LogError(ex, ErrorMessage); + return Result.InternalServerError([ErrorMessage, .. ex.ExtractMessages()]); + } + } + + /// + public async Task> GetMetadataAsync(CancellationToken cancellationToken = default) { + try { + var metadata = await client.GetMetadataAsync(cancellationToken); + return Result.Ok(metadata); + } + catch (OperationCanceledException) { + throw; + } + catch (Exception ex) { + logger.LogError(ex, ErrorMessage); + return Result.InternalServerError(default!, [ErrorMessage, .. ex.ExtractMessages()]); + } + } + + /// + public async Task ShutdownAsync(CancellationToken cancellationToken = default) { + try { + await client.ShutdownSidecarAsync(cancellationToken); + return Result.Ok(); + } + catch (OperationCanceledException) { + throw; + } + catch (Exception ex) { + logger.LogError(ex, ErrorMessage); + return Result.InternalServerError([ErrorMessage, .. ex.ExtractMessages()]); + } + } +} diff --git a/src/MaksIT.Dapr/Services/DaprStateStoreService.cs b/src/MaksIT.Dapr/Services/DaprStateStoreService.cs index 12a881b..6fff4e0 100644 --- a/src/MaksIT.Dapr/Services/DaprStateStoreService.cs +++ b/src/MaksIT.Dapr/Services/DaprStateStoreService.cs @@ -1,93 +1,416 @@ -using Dapr.Client; -using MaksIT.Core.Extensions; -using MaksIT.Results; using Microsoft.Extensions.Logging; +using Grpc.Core; +using Dapr.Client; +using MaksIT.Results; +using MaksIT.Core.Extensions; namespace MaksIT.Dapr.Services; +/// +/// Dapr state store operations with outcomes. +/// public interface IDaprStateStoreService { - Task SetStateAsync(string storeName, string key, T value); - Task> GetStateAsync(string storeName, string key); - Task> GetStateAndETagAsync(string storeName, string key, CancellationToken cancellationToken = default); - Task> TrySaveStateAsync(string storeName, string key, T value, string? etag, CancellationToken cancellationToken = default); - Task DeleteStateAsync(string storeName, string key); + /// + /// Saves under in . + /// + Task SetStateAsync( + string storeName, + string key, + T value, + StateOptions? options = null, + IReadOnlyDictionary? metadata = null, + CancellationToken cancellationToken = default); + + /// + /// Gets state for . Missing keys return Ok(null); infra failures are unsuccessful. + /// + Task> GetStateAsync( + string storeName, + string key, + ConsistencyMode? consistencyMode = null, + IReadOnlyDictionary? metadata = null, + CancellationToken cancellationToken = default); + + /// + /// Gets state and ETag for optimistic concurrency. Missing keys return Ok((null, null)). + /// + Task> GetStateAndETagAsync( + string storeName, + string key, + ConsistencyMode? consistencyMode = null, + IReadOnlyDictionary? metadata = null, + CancellationToken cancellationToken = default); + + /// + /// Attempts an ETag-conditional save. is false on conflict. + /// + Task> TrySaveStateAsync( + string storeName, + string key, + T value, + string? etag, + StateOptions? options = null, + IReadOnlyDictionary? metadata = null, + CancellationToken cancellationToken = default); + + /// + /// Deletes . Missing keys are treated as success (idempotent). + /// + Task DeleteStateAsync( + string storeName, + string key, + StateOptions? options = null, + IReadOnlyDictionary? metadata = null, + CancellationToken cancellationToken = default); + + /// + /// Attempts an ETag-conditional delete. + /// + Task> TryDeleteStateAsync( + string storeName, + string key, + string? etag, + StateOptions? options = null, + IReadOnlyDictionary? metadata = null, + CancellationToken cancellationToken = default); + + /// + /// Gets multiple keys from the store. + /// + Task>> GetBulkStateAsync( + string storeName, + IReadOnlyList keys, + int? parallelism = null, + IReadOnlyDictionary? metadata = null, + CancellationToken cancellationToken = default); + + /// + /// Saves multiple items. + /// + Task SaveBulkStateAsync( + string storeName, + IReadOnlyList> items, + CancellationToken cancellationToken = default); + + /// + /// Deletes multiple items. + /// + Task DeleteBulkStateAsync( + string storeName, + IReadOnlyList items, + CancellationToken cancellationToken = default); + + /// + /// Queries state with a JSON query document. + /// + Task>> QueryStateAsync( + string storeName, + string jsonQuery, + IReadOnlyDictionary? metadata = null, + CancellationToken cancellationToken = default); + + /// + /// Executes a state transaction. + /// + Task ExecuteStateTransactionAsync( + string storeName, + IReadOnlyList operations, + IReadOnlyDictionary? metadata = null, + CancellationToken cancellationToken = default); } +/// +/// Default using . +/// public class DaprStateStoreService : IDaprStateStoreService { - private const string ErrorMessage = "MaksIT.Dapr - Data provider error"; + private const string ErrorMessage = "MaksIT.Dapr - State store error"; private readonly DaprClient _client; private readonly ILogger _logger; + /// + /// Creates a state store service backed by . + /// public DaprStateStoreService(ILogger logger, DaprClient client) { _logger = logger; _client = client; } - public async Task SetStateAsync(string storeName, string key, T value) { + /// + public async Task SetStateAsync( + string storeName, + string key, + T value, + StateOptions? options = null, + IReadOnlyDictionary? metadata = null, + CancellationToken cancellationToken = default) { + if (string.IsNullOrWhiteSpace(storeName) || string.IsNullOrWhiteSpace(key)) + return Result.BadRequest("storeName and key are required."); + try { - await _client.SaveStateAsync(storeName, key, value); + await _client.SaveStateAsync(storeName, key, value, options, metadata, cancellationToken); return Result.Ok(); } + catch (OperationCanceledException) { + throw; + } catch (Exception ex) { _logger.LogError(ex, ErrorMessage); return Result.InternalServerError([ErrorMessage, .. ex.ExtractMessages()]); } } - public async Task> GetStateAsync(string storeName, string key) { - try { - var state = await _client.GetStateAsync(storeName, key); - if (state is null) - return Result.NotFound(default, $"State from the store {storeName} with the {key} not found."); + /// + public async Task> GetStateAsync( + string storeName, + string key, + ConsistencyMode? consistencyMode = null, + IReadOnlyDictionary? metadata = null, + CancellationToken cancellationToken = default) { + if (string.IsNullOrWhiteSpace(storeName) || string.IsNullOrWhiteSpace(key)) + return Result.BadRequest(default, "storeName and key are required."); + try { + var state = await _client.GetStateAsync(storeName, key, consistencyMode, metadata, cancellationToken); return Result.Ok(state); } + catch (Exception ex) when (IsStateKeyNotFound(ex)) { + return Result.Ok(default); + } + catch (OperationCanceledException) { + throw; + } catch (Exception ex) { _logger.LogError(ex, ErrorMessage); return Result.InternalServerError(default, [ErrorMessage, .. ex.ExtractMessages()]); } } + /// public async Task> GetStateAndETagAsync( string storeName, string key, + ConsistencyMode? consistencyMode = null, + IReadOnlyDictionary? metadata = null, CancellationToken cancellationToken = default) { + if (string.IsNullOrWhiteSpace(storeName) || string.IsNullOrWhiteSpace(key)) + return Result<(T? Value, string? ETag)>.BadRequest(default, "storeName and key are required."); + try { - var (value, etag) = await _client.GetStateAndETagAsync(storeName, key, cancellationToken: cancellationToken); + var (value, etag) = await _client.GetStateAndETagAsync(storeName, key, consistencyMode, metadata, cancellationToken); return Result<(T? Value, string? ETag)>.Ok((value, etag)); } + catch (Exception ex) when (IsStateKeyNotFound(ex)) { + return Result<(T? Value, string? ETag)>.Ok((default, null)); + } + catch (OperationCanceledException) { + throw; + } catch (Exception ex) { _logger.LogError(ex, ErrorMessage); return Result<(T? Value, string? ETag)>.InternalServerError(default, [ErrorMessage, .. ex.ExtractMessages()]); } } + /// public async Task> TrySaveStateAsync( string storeName, string key, T value, string? etag, + StateOptions? options = null, + IReadOnlyDictionary? metadata = null, CancellationToken cancellationToken = default) { + if (string.IsNullOrWhiteSpace(storeName) || string.IsNullOrWhiteSpace(key)) + return Result.BadRequest(false, "storeName and key are required."); + try { - var saved = await _client.TrySaveStateAsync(storeName, key, value, etag ?? string.Empty, cancellationToken: cancellationToken); + var saved = await _client.TrySaveStateAsync(storeName, key, value, etag ?? string.Empty, options, metadata, cancellationToken); return Result.Ok(saved); } + catch (OperationCanceledException) { + throw; + } catch (Exception ex) { _logger.LogError(ex, ErrorMessage); return Result.InternalServerError(false, [ErrorMessage, .. ex.ExtractMessages()]); } } - public async Task DeleteStateAsync(string storeName, string key) { + /// + public async Task DeleteStateAsync( + string storeName, + string key, + StateOptions? options = null, + IReadOnlyDictionary? metadata = null, + CancellationToken cancellationToken = default) { + if (string.IsNullOrWhiteSpace(storeName) || string.IsNullOrWhiteSpace(key)) + return Result.BadRequest("storeName and key are required."); + try { - await _client.DeleteStateAsync(storeName, key); + await _client.DeleteStateAsync(storeName, key, options, metadata, cancellationToken); return Result.Ok(); } + catch (Exception ex) when (IsStateKeyNotFound(ex)) { + return Result.Ok(); + } + catch (OperationCanceledException) { + throw; + } catch (Exception ex) { _logger.LogError(ex, ErrorMessage); return Result.InternalServerError([ErrorMessage, .. ex.ExtractMessages()]); } } + + /// + public async Task> TryDeleteStateAsync( + string storeName, + string key, + string? etag, + StateOptions? options = null, + IReadOnlyDictionary? metadata = null, + CancellationToken cancellationToken = default) { + if (string.IsNullOrWhiteSpace(storeName) || string.IsNullOrWhiteSpace(key)) + return Result.BadRequest(false, "storeName and key are required."); + + try { + var deleted = await _client.TryDeleteStateAsync(storeName, key, etag ?? string.Empty, options, metadata, cancellationToken); + return Result.Ok(deleted); + } + catch (OperationCanceledException) { + throw; + } + catch (Exception ex) { + _logger.LogError(ex, ErrorMessage); + return Result.InternalServerError(false, [ErrorMessage, .. ex.ExtractMessages()]); + } + } + + /// + public async Task>> GetBulkStateAsync( + string storeName, + IReadOnlyList keys, + int? parallelism = null, + IReadOnlyDictionary? metadata = null, + CancellationToken cancellationToken = default) { + if (string.IsNullOrWhiteSpace(storeName) || keys is null || keys.Count == 0) + return Result>.BadRequest(default!, "storeName and keys are required."); + + try { + var items = await _client.GetBulkStateAsync(storeName, keys, parallelism, metadata, cancellationToken); + return Result>.Ok(items); + } + catch (OperationCanceledException) { + throw; + } + catch (Exception ex) { + _logger.LogError(ex, ErrorMessage); + return Result>.InternalServerError(default!, [ErrorMessage, .. ex.ExtractMessages()]); + } + } + + /// + public async Task SaveBulkStateAsync( + string storeName, + IReadOnlyList> items, + CancellationToken cancellationToken = default) { + if (string.IsNullOrWhiteSpace(storeName) || items is null || items.Count == 0) + return Result.BadRequest("storeName and items are required."); + + try { + await _client.SaveBulkStateAsync(storeName, items, cancellationToken); + return Result.Ok(); + } + catch (OperationCanceledException) { + throw; + } + catch (Exception ex) { + _logger.LogError(ex, ErrorMessage); + return Result.InternalServerError([ErrorMessage, .. ex.ExtractMessages()]); + } + } + + /// + public async Task DeleteBulkStateAsync( + string storeName, + IReadOnlyList items, + CancellationToken cancellationToken = default) { + if (string.IsNullOrWhiteSpace(storeName) || items is null || items.Count == 0) + return Result.BadRequest("storeName and items are required."); + + try { + await _client.DeleteBulkStateAsync(storeName, items, cancellationToken); + return Result.Ok(); + } + catch (OperationCanceledException) { + throw; + } + catch (Exception ex) { + _logger.LogError(ex, ErrorMessage); + return Result.InternalServerError([ErrorMessage, .. ex.ExtractMessages()]); + } + } + + /// + public async Task>> QueryStateAsync( + string storeName, + string jsonQuery, + IReadOnlyDictionary? metadata = null, + CancellationToken cancellationToken = default) { + if (string.IsNullOrWhiteSpace(storeName) || string.IsNullOrWhiteSpace(jsonQuery)) + return Result>.BadRequest(default!, "storeName and jsonQuery are required."); + + try { + var response = await _client.QueryStateAsync(storeName, jsonQuery, metadata, cancellationToken); + return Result>.Ok(response); + } + catch (OperationCanceledException) { + throw; + } + catch (Exception ex) { + _logger.LogError(ex, ErrorMessage); + return Result>.InternalServerError(default!, [ErrorMessage, .. ex.ExtractMessages()]); + } + } + + /// + public async Task ExecuteStateTransactionAsync( + string storeName, + IReadOnlyList operations, + IReadOnlyDictionary? metadata = null, + CancellationToken cancellationToken = default) { + if (string.IsNullOrWhiteSpace(storeName) || operations is null || operations.Count == 0) + return Result.BadRequest("storeName and operations are required."); + + try { + await _client.ExecuteStateTransactionAsync(storeName, operations, metadata, cancellationToken); + return Result.Ok(); + } + catch (OperationCanceledException) { + throw; + } + catch (Exception ex) { + _logger.LogError(ex, ErrorMessage); + return Result.InternalServerError([ErrorMessage, .. ex.ExtractMessages()]); + } + } + + private static bool IsStateKeyNotFound(Exception ex) { + for (var current = ex; current is not null; current = current.InnerException) { + if (current is RpcException rpc && + (rpc.StatusCode == StatusCode.NotFound || ContainsKeyNotFound(rpc.Status.Detail))) + return true; + + if (ContainsKeyNotFound(current.Message)) + return true; + } + + return false; + } + + private static bool ContainsKeyNotFound(string? text) => + !string.IsNullOrEmpty(text) && + text.Contains("key not found", StringComparison.OrdinalIgnoreCase); } diff --git a/src/MaksIT.Dapr/Services/DaprWorkflowService.cs b/src/MaksIT.Dapr/Services/DaprWorkflowService.cs new file mode 100644 index 0000000..0faf36c --- /dev/null +++ b/src/MaksIT.Dapr/Services/DaprWorkflowService.cs @@ -0,0 +1,369 @@ +using Microsoft.Extensions.Logging; +using Dapr.Workflow; +using Dapr.Workflow.Client; +using MaksIT.Results; +using MaksIT.Core.Extensions; + + +namespace MaksIT.Dapr.Services; + +/// +/// Schedules and manages Dapr workflow instances with outcomes. +/// +public interface IDaprWorkflowService { + /// + /// Schedules a new workflow instance. Returns the instance id. + /// + Task> ScheduleAsync( + string workflowName, + object? input = null, + string? instanceId = null, + DateTimeOffset? startTime = null, + CancellationToken cancellationToken = default); + + /// + /// Gets the current state of a workflow instance. + /// + Task> GetStateAsync(string instanceId, bool getInputsAndOutputs = true, CancellationToken cancellationToken = default); + + /// + /// Waits until the workflow has started. + /// + Task> WaitForStartAsync(string instanceId, bool getInputsAndOutputs = true, CancellationToken cancellationToken = default); + + /// + /// Waits until the workflow has completed. + /// + Task> WaitForCompletionAsync(string instanceId, bool getInputsAndOutputs = true, CancellationToken cancellationToken = default); + + /// + /// Raises an external event to a waiting workflow. + /// + Task RaiseEventAsync(string instanceId, string eventName, object? eventPayload = null, CancellationToken cancellationToken = default); + + /// + /// Terminates a running workflow instance. + /// + Task TerminateAsync(string instanceId, object? output = null, CancellationToken cancellationToken = default); + + /// + /// Suspends a running workflow instance. + /// + Task SuspendAsync(string instanceId, string? reason = null, CancellationToken cancellationToken = default); + + /// + /// Resumes a suspended workflow instance. + /// + Task ResumeAsync(string instanceId, string? reason = null, CancellationToken cancellationToken = default); + + /// + /// Purges history for a completed workflow instance. + /// + Task PurgeAsync(string instanceId, CancellationToken cancellationToken = default); + + /// + /// Lists workflow instance IDs with optional pagination. + /// + Task> ListInstanceIdsAsync( + string? continuationToken = null, + int? pageSize = null, + CancellationToken cancellationToken = default); + + /// + /// Gets the full execution history of a workflow instance. + /// + Task>> GetInstanceHistoryAsync( + string instanceId, + CancellationToken cancellationToken = default); + + /// + /// Reruns a workflow from a history event, returning the new instance id. + /// + Task> RerunFromEventAsync( + string sourceInstanceId, + uint eventId, + RerunWorkflowFromEventOptions? options = null, + CancellationToken cancellationToken = default); +} + +/// +/// Default using . +/// +public class DaprWorkflowService : IDaprWorkflowService { + private const string ErrorMessage = "MaksIT.Dapr - Workflow error"; + + private readonly IDaprWorkflowClient _client; + private readonly ILogger _logger; + + /// + /// Creates a workflow facade backed by . + /// + public DaprWorkflowService(ILogger logger, IDaprWorkflowClient client) { + _logger = logger; + _client = client; + } + + /// + public async Task> ScheduleAsync( + string workflowName, + object? input = null, + string? instanceId = null, + DateTimeOffset? startTime = null, + CancellationToken cancellationToken = default) { + if (string.IsNullOrWhiteSpace(workflowName)) + return Result.BadRequest(default!, "workflowName is required."); + + try { + var id = await _client.ScheduleNewWorkflowAsync(workflowName, instanceId, input, startTime, cancellationToken); + return Result.Ok(id); + } + catch (OperationCanceledException) { + throw; + } + catch (Exception ex) { + _logger.LogError(ex, ErrorMessage); + return Result.InternalServerError(default!, [ErrorMessage, .. ex.ExtractMessages()]); + } + } + + /// + public async Task> GetStateAsync( + string instanceId, + bool getInputsAndOutputs = true, + CancellationToken cancellationToken = default) { + var validation = ValidateInstanceId(instanceId); + if (!validation.IsSuccess) + return Result.BadRequest(default!, validation.Messages.ToArray()); + + try { + var state = await _client.GetWorkflowStateAsync(instanceId, getInputsAndOutputs, cancellationToken); + return Result.Ok(state); + } + catch (OperationCanceledException) { + throw; + } + catch (Exception ex) { + _logger.LogError(ex, ErrorMessage); + return Result.InternalServerError(default!, [ErrorMessage, .. ex.ExtractMessages()]); + } + } + + /// + public async Task> WaitForStartAsync( + string instanceId, + bool getInputsAndOutputs = true, + CancellationToken cancellationToken = default) { + var validation = ValidateInstanceId(instanceId); + if (!validation.IsSuccess) + return Result.BadRequest(default!, validation.Messages.ToArray()); + + try { + var state = await _client.WaitForWorkflowStartAsync(instanceId, getInputsAndOutputs, cancellationToken); + return Result.Ok(state); + } + catch (OperationCanceledException) { + throw; + } + catch (Exception ex) { + _logger.LogError(ex, ErrorMessage); + return Result.InternalServerError(default!, [ErrorMessage, .. ex.ExtractMessages()]); + } + } + + /// + public async Task> WaitForCompletionAsync( + string instanceId, + bool getInputsAndOutputs = true, + CancellationToken cancellationToken = default) { + var validation = ValidateInstanceId(instanceId); + if (!validation.IsSuccess) + return Result.BadRequest(default!, validation.Messages.ToArray()); + + try { + var state = await _client.WaitForWorkflowCompletionAsync(instanceId, getInputsAndOutputs, cancellationToken); + return Result.Ok(state); + } + catch (OperationCanceledException) { + throw; + } + catch (Exception ex) { + _logger.LogError(ex, ErrorMessage); + return Result.InternalServerError(default!, [ErrorMessage, .. ex.ExtractMessages()]); + } + } + + /// + public async Task RaiseEventAsync( + string instanceId, + string eventName, + object? eventPayload = null, + CancellationToken cancellationToken = default) { + if (string.IsNullOrWhiteSpace(instanceId) || string.IsNullOrWhiteSpace(eventName)) + return Result.BadRequest("instanceId and eventName are required."); + + try { + await _client.RaiseEventAsync(instanceId, eventName, eventPayload, cancellationToken); + return Result.Ok(); + } + catch (OperationCanceledException) { + throw; + } + catch (Exception ex) { + _logger.LogError(ex, ErrorMessage); + return Result.InternalServerError([ErrorMessage, .. ex.ExtractMessages()]); + } + } + + /// + public async Task TerminateAsync( + string instanceId, + object? output = null, + CancellationToken cancellationToken = default) { + var validation = ValidateInstanceId(instanceId); + if (!validation.IsSuccess) + return validation; + + try { + await _client.TerminateWorkflowAsync(instanceId, output, cancellationToken); + return Result.Ok(); + } + catch (OperationCanceledException) { + throw; + } + catch (Exception ex) { + _logger.LogError(ex, ErrorMessage); + return Result.InternalServerError([ErrorMessage, .. ex.ExtractMessages()]); + } + } + + /// + public async Task SuspendAsync( + string instanceId, + string? reason = null, + CancellationToken cancellationToken = default) { + var validation = ValidateInstanceId(instanceId); + if (!validation.IsSuccess) + return validation; + + try { + await _client.SuspendWorkflowAsync(instanceId, reason, cancellationToken); + return Result.Ok(); + } + catch (OperationCanceledException) { + throw; + } + catch (Exception ex) { + _logger.LogError(ex, ErrorMessage); + return Result.InternalServerError([ErrorMessage, .. ex.ExtractMessages()]); + } + } + + /// + public async Task ResumeAsync( + string instanceId, + string? reason = null, + CancellationToken cancellationToken = default) { + var validation = ValidateInstanceId(instanceId); + if (!validation.IsSuccess) + return validation; + + try { + await _client.ResumeWorkflowAsync(instanceId, reason, cancellationToken); + return Result.Ok(); + } + catch (OperationCanceledException) { + throw; + } + catch (Exception ex) { + _logger.LogError(ex, ErrorMessage); + return Result.InternalServerError([ErrorMessage, .. ex.ExtractMessages()]); + } + } + + /// + public async Task PurgeAsync(string instanceId, CancellationToken cancellationToken = default) { + var validation = ValidateInstanceId(instanceId); + if (!validation.IsSuccess) + return validation; + + try { + await _client.PurgeInstanceAsync(instanceId, cancellationToken); + return Result.Ok(); + } + catch (OperationCanceledException) { + throw; + } + catch (Exception ex) { + _logger.LogError(ex, ErrorMessage); + return Result.InternalServerError([ErrorMessage, .. ex.ExtractMessages()]); + } + } + + /// + public async Task> ListInstanceIdsAsync( + string? continuationToken = null, + int? pageSize = null, + CancellationToken cancellationToken = default) { + try { + var page = await _client.ListInstanceIdsAsync(continuationToken, pageSize, cancellationToken); + return Result.Ok(page); + } + catch (OperationCanceledException) { + throw; + } + catch (Exception ex) { + _logger.LogError(ex, ErrorMessage); + return Result.InternalServerError(default!, [ErrorMessage, .. ex.ExtractMessages()]); + } + } + + /// + public async Task>> GetInstanceHistoryAsync( + string instanceId, + CancellationToken cancellationToken = default) { + var validation = ValidateInstanceId(instanceId); + if (!validation.IsSuccess) + return Result>.BadRequest(default!, validation.Messages.ToArray()); + + try { + var history = await _client.GetInstanceHistoryAsync(instanceId, cancellationToken); + return Result>.Ok(history); + } + catch (OperationCanceledException) { + throw; + } + catch (Exception ex) { + _logger.LogError(ex, ErrorMessage); + return Result>.InternalServerError(default!, [ErrorMessage, .. ex.ExtractMessages()]); + } + } + + /// + public async Task> RerunFromEventAsync( + string sourceInstanceId, + uint eventId, + RerunWorkflowFromEventOptions? options = null, + CancellationToken cancellationToken = default) { + var validation = ValidateInstanceId(sourceInstanceId); + if (!validation.IsSuccess) + return Result.BadRequest(default!, validation.Messages.ToArray()); + + try { + var id = await _client.RerunWorkflowFromEventAsync(sourceInstanceId, eventId, options, cancellationToken); + return Result.Ok(id); + } + catch (OperationCanceledException) { + throw; + } + catch (Exception ex) { + _logger.LogError(ex, ErrorMessage); + return Result.InternalServerError(default!, [ErrorMessage, .. ex.ExtractMessages()]); + } + } + + private static Result ValidateInstanceId(string instanceId) { + if (string.IsNullOrWhiteSpace(instanceId)) + return Result.BadRequest("instanceId is required."); + return Result.Ok(); + } +} diff --git a/src/MaksIT.Dapr/Services/DaprRuntimeInstanceIdProvider.cs b/src/MaksIT.Dapr/Services/WorkLease/DaprRuntimeInstanceIdProvider.cs similarity index 76% rename from src/MaksIT.Dapr/Services/DaprRuntimeInstanceIdProvider.cs rename to src/MaksIT.Dapr/Services/WorkLease/DaprRuntimeInstanceIdProvider.cs index 9948b18..a3e958d 100644 --- a/src/MaksIT.Dapr/Services/DaprRuntimeInstanceIdProvider.cs +++ b/src/MaksIT.Dapr/Services/WorkLease/DaprRuntimeInstanceIdProvider.cs @@ -1,7 +1,13 @@ -namespace MaksIT.Dapr.Services; +namespace MaksIT.Dapr.Services.WorkLease; -/// Stable id for this process/pod (lease holder). + +/// +/// Stable id for this process/pod (lease holder). +/// public interface IDaprRuntimeInstanceId { + /// + /// Identifier used as lease HolderId. + /// string InstanceId { get; } } @@ -9,6 +15,7 @@ public interface IDaprRuntimeInstanceId { /// Prefers POD_NAME in Kubernetes; otherwise host name + process id. /// public sealed class DaprRuntimeInstanceIdProvider : IDaprRuntimeInstanceId { + /// public string InstanceId { get; } = Build(); private static string Build() { diff --git a/src/MaksIT.Dapr/Services/WorkLease/DaprWorkLease.cs b/src/MaksIT.Dapr/Services/WorkLease/DaprWorkLease.cs new file mode 100644 index 0000000..ed4b317 --- /dev/null +++ b/src/MaksIT.Dapr/Services/WorkLease/DaprWorkLease.cs @@ -0,0 +1,16 @@ +namespace MaksIT.Dapr.Services.WorkLease; + + +/// +/// Lease document stored in Dapr state for HA work coordination. +/// +/// Runtime instance currently holding the lease. +/// When this generation was acquired. +/// When the lease expires if not renewed. +/// Monotonic fencing token; bumped on steal / re-acquire by another holder path. +public sealed record DaprWorkLease( + string HolderId, + DateTimeOffset AcquiredAtUtc, + DateTimeOffset ExpiresAtUtc, + long Generation = 0 +); diff --git a/src/MaksIT.Dapr/Services/WorkLease/DaprWorkLeaseHold.cs b/src/MaksIT.Dapr/Services/WorkLease/DaprWorkLeaseHold.cs new file mode 100644 index 0000000..bdbf26e --- /dev/null +++ b/src/MaksIT.Dapr/Services/WorkLease/DaprWorkLeaseHold.cs @@ -0,0 +1,234 @@ +using Microsoft.Extensions.Logging; +using MaksIT.Results; + + +namespace MaksIT.Dapr.Services.WorkLease; + +/// +/// Scoped lease hold: releases on dispose; optional background renew at ~½ TTL. +/// Exposes for fencing long exclusive work. +/// +public sealed class DaprWorkLeaseHold : IAsyncDisposable { + private readonly IDaprWorkLeaseService _leases; + private readonly string _storeName; + private readonly string _workKey; + private readonly string _holderId; + private readonly TimeSpan _ttl; + private readonly CancellationTokenSource _renewCts = new(); + private readonly Task? _renewLoop; + private int _disposed; + + /// + /// Creates a hold after a successful acquire. + /// + public DaprWorkLeaseHold( + IDaprWorkLeaseService leases, + string storeName, + string workKey, + string holderId, + TimeSpan ttl, + long generation, + bool autoRenew) { + _leases = leases; + _storeName = storeName; + _workKey = workKey; + _holderId = holderId; + _ttl = ttl; + Generation = generation; + + if (autoRenew && ttl > TimeSpan.Zero) + _renewLoop = RunRenewLoopAsync(_renewCts.Token); + } + + /// + /// Fencing generation observed at acquire time. + /// + public long Generation { get; } + + /// + /// Renews once. Returns Ok(false) when the lease is no longer held by this holder. + /// + public Task> RenewAsync(CancellationToken cancellationToken = default) => + _leases.TryRenewAsync(_storeName, _workKey, _holderId, _ttl, cancellationToken); + + /// + /// Returns Ok(true) when still held by this holder at the same generation; otherwise Ok(false) or Conflict. + /// + public async Task> EnsureStillHeldAsync(CancellationToken cancellationToken = default) { + var current = await _leases.GetAsync(_storeName, _workKey, cancellationToken).ConfigureAwait(false); + if (!current.IsSuccess) + return current.ToResultOfType(false); + + var lease = current.Value; + if (lease is null) + return Result.Ok(false); + + if (!string.Equals(lease.HolderId, _holderId, StringComparison.Ordinal) || lease.Generation != Generation) + return Result.Conflict(false, "Lease generation or holder changed."); + + if (lease.ExpiresAtUtc <= DateTimeOffset.UtcNow) + return Result.Ok(false); + + return Result.Ok(true); + } + + /// + public async ValueTask DisposeAsync() { + if (Interlocked.Exchange(ref _disposed, 1) != 0) + return; + + try { + _renewCts.Cancel(); + } + catch (ObjectDisposedException) { + // ignore + } + + if (_renewLoop is not null) { + try { + await _renewLoop.ConfigureAwait(false); + } + catch (OperationCanceledException) { + // expected + } + } + + _renewCts.Dispose(); + await _leases.ReleaseAsync(_storeName, _workKey, _holderId, CancellationToken.None).ConfigureAwait(false); + } + + private async Task RunRenewLoopAsync(CancellationToken cancellationToken) { + var delay = TimeSpan.FromTicks(Math.Max(_ttl.Ticks / 2, TimeSpan.FromSeconds(1).Ticks)); + while (!cancellationToken.IsCancellationRequested) { + try { + await Task.Delay(delay, cancellationToken).ConfigureAwait(false); + var renewed = await RenewAsync(cancellationToken).ConfigureAwait(false); + if (!renewed.IsSuccess || !renewed.Value) + break; + } + catch (OperationCanceledException) { + break; + } + } + } +} + +/// +/// Bootstrap helper: one replica runs work under a lease; others wait until ready. +/// +public static class DaprWorkLeaseBootstrap { + /// + /// Leader acquires the lease and runs ; followers poll until true or cancel. + /// + public static async Task RunBootstrapUnderLeaseAsync( + IDaprWorkLeaseService leases, + string storeName, + string workKey, + string holderId, + TimeSpan ttl, + Func> bootstrap, + Func>> isReady, + TimeSpan? followerPollInterval = null, + CancellationToken cancellationToken = default) { + if (bootstrap is null) + return Result.BadRequest("bootstrap is required."); + if (isReady is null) + return Result.BadRequest("isReady is required."); + + var hold = await leases.TryHoldAsync(storeName, workKey, holderId, ttl, autoRenew: true, cancellationToken).ConfigureAwait(false); + if (!hold.IsSuccess) + return hold.ToResult(); + + if (hold.Value is not null) { + await using (hold.Value.ConfigureAwait(false)) { + return await bootstrap(cancellationToken).ConfigureAwait(false); + } + } + + var poll = followerPollInterval ?? TimeSpan.FromSeconds(2); + while (!cancellationToken.IsCancellationRequested) { + var ready = await isReady(cancellationToken).ConfigureAwait(false); + if (!ready.IsSuccess) + return ready.ToResult(); + if (ready.Value) + return Result.Ok(); + + await Task.Delay(poll, cancellationToken).ConfigureAwait(false); + } + + throw new OperationCanceledException(cancellationToken); + } +} + +/// +/// Background loop that runs exclusive work only while a named work lease is held. +/// Failed work values are logged; the host is not crashed. +/// +public abstract class LeasedBackgroundService( + IDaprWorkLeaseService leases, + IDaprRuntimeInstanceId runtimeInstance, + IDaprWorkLeaseOptions options, + ILogger logger +) : Microsoft.Extensions.Hosting.BackgroundService { + /// + /// Product lease key (not the Dapr store name). + /// + protected abstract string WorkKey { get; } + + /// + /// Lease TTL while work runs. + /// + protected virtual TimeSpan LeaseTtl => TimeSpan.FromMinutes(1); + + /// + /// Delay after a successful work cycle. + /// + protected virtual TimeSpan IdleDelay => TimeSpan.FromSeconds(30); + + /// + /// Delay when the lease is busy / not acquired. + /// + protected virtual TimeSpan BusyBackoff => TimeSpan.FromSeconds(5); + + /// + /// Exclusive work while the lease is held. Return unsuccessful to log and continue. + /// + protected abstract Task ExecuteLeasedAsync(DaprWorkLeaseHold hold, CancellationToken stoppingToken); + + /// + protected override async Task ExecuteAsync(CancellationToken stoppingToken) { + while (!stoppingToken.IsCancellationRequested) { + try { + var holdResult = await leases.TryHoldAsync( + options.StoreName, + WorkKey, + runtimeInstance.InstanceId, + LeaseTtl, + autoRenew: true, + stoppingToken).ConfigureAwait(false); + + if (!holdResult.IsSuccess) { + logger.LogError("LeasedBackgroundService acquire failed for {WorkKey}: {Messages}", WorkKey, string.Join("; ", holdResult.Messages)); + await Task.Delay(BusyBackoff, stoppingToken).ConfigureAwait(false); + continue; + } + + if (holdResult.Value is null) { + await Task.Delay(BusyBackoff, stoppingToken).ConfigureAwait(false); + continue; + } + + await using (holdResult.Value.ConfigureAwait(false)) { + var work = await ExecuteLeasedAsync(holdResult.Value, stoppingToken).ConfigureAwait(false); + if (!work.IsSuccess) + logger.LogWarning("LeasedBackgroundService work failed for {WorkKey}: {Messages}", WorkKey, string.Join("; ", work.Messages)); + } + + await Task.Delay(IdleDelay, stoppingToken).ConfigureAwait(false); + } + catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) { + break; + } + } + } +} diff --git a/src/MaksIT.Dapr/Services/WorkLease/DaprWorkLeaseOptions.cs b/src/MaksIT.Dapr/Services/WorkLease/DaprWorkLeaseOptions.cs new file mode 100644 index 0000000..78f87a1 --- /dev/null +++ b/src/MaksIT.Dapr/Services/WorkLease/DaprWorkLeaseOptions.cs @@ -0,0 +1,20 @@ +namespace MaksIT.Dapr.Services.WorkLease; + + +/// +/// Default Dapr state component name for work leases. +/// +public interface IDaprWorkLeaseOptions { + /// + /// Dapr state store Component name used when callers omit storeName. + /// + string StoreName { get; } +} + +/// +/// Default . +/// +public sealed class DaprWorkLeaseOptions : IDaprWorkLeaseOptions { + /// + public required string StoreName { get; init; } +} diff --git a/src/MaksIT.Dapr/Services/DaprWorkLeaseStore.cs b/src/MaksIT.Dapr/Services/WorkLease/DaprWorkLeaseService.cs similarity index 51% rename from src/MaksIT.Dapr/Services/DaprWorkLeaseStore.cs rename to src/MaksIT.Dapr/Services/WorkLease/DaprWorkLeaseService.cs index 0b9087d..022d972 100644 --- a/src/MaksIT.Dapr/Services/DaprWorkLeaseStore.cs +++ b/src/MaksIT.Dapr/Services/WorkLease/DaprWorkLeaseService.cs @@ -1,29 +1,53 @@ using MaksIT.Results; -namespace MaksIT.Dapr.Services; - -public sealed record DaprWorkLease( - string HolderId, - DateTimeOffset AcquiredAtUtc, - DateTimeOffset ExpiresAtUtc -); +namespace MaksIT.Dapr.Services.WorkLease; /// -/// HA work coordination via Dapr state store (broker-agnostic). -/// Keys are product-defined; store name comes from the Dapr Component. +/// HA work coordination via Dapr state (broker-agnostic). +/// Keys are product-defined; store name comes from the Dapr Component or . /// -public interface IDaprWorkLeaseStore { +public interface IDaprWorkLeaseService { + /// + /// Tries to acquire or take over an expired lease for . + /// Task> TryAcquireAsync(string storeName, string workKey, string holderId, TimeSpan ttl, CancellationToken cancellationToken = default); + + /// + /// Extends an existing lease when still holds it. + /// Task> TryRenewAsync(string storeName, string workKey, string holderId, TimeSpan ttl, CancellationToken cancellationToken = default); + + /// + /// Releases the lease when held by . + /// Task ReleaseAsync(string storeName, string workKey, string holderId, CancellationToken cancellationToken = default); + + /// + /// Returns the current lease document, or null when missing. + /// Task> GetAsync(string storeName, string workKey, CancellationToken cancellationToken = default); + + /// + /// Acquires a scoped hold with optional auto-renew. Ok(null) when not acquired; unsuccessful on infra errors. + /// + Task> TryHoldAsync( + string storeName, + string workKey, + string holderId, + TimeSpan ttl, + bool autoRenew = true, + CancellationToken cancellationToken = default); } -public sealed class DaprWorkLeaseStore( +/// +/// Default using ETag concurrency on . +/// +public sealed class DaprWorkLeaseService( IDaprStateStoreService stateStore -) : IDaprWorkLeaseStore { +) : IDaprWorkLeaseService { + /// public async Task> TryAcquireAsync( string storeName, string workKey, @@ -34,9 +58,12 @@ public sealed class DaprWorkLeaseStore( if (!validation.IsSuccess) return validation.ToResultOfType(false); - var existing = await stateStore.GetStateAndETagAsync(storeName, workKey, cancellationToken).ConfigureAwait(false); + var existing = await stateStore.GetStateAndETagAsync( + storeName, + workKey, + cancellationToken: cancellationToken).ConfigureAwait(false); if (!existing.IsSuccess) - return existing.ToResult().ToResultOfType(false); + return existing.ToResultOfType(false); var (lease, etag) = existing.Value; var now = DateTimeOffset.UtcNow; @@ -44,15 +71,26 @@ public sealed class DaprWorkLeaseStore( if (lease is not null && lease.ExpiresAtUtc > now && !string.Equals(lease.HolderId, holderId, StringComparison.Ordinal)) return Result.Ok(false); - var next = new DaprWorkLease(holderId, now, now.Add(ttl)); - // First write: etag may be null/empty when key missing. - var saved = await stateStore.TrySaveStateAsync(storeName, workKey, next, etag, cancellationToken).ConfigureAwait(false); + var generation = lease is null + ? 1L + : string.Equals(lease.HolderId, holderId, StringComparison.Ordinal) + ? lease.Generation + : lease.Generation + 1; + + var next = new DaprWorkLease(holderId, now, now.Add(ttl), generation); + var saved = await stateStore.TrySaveStateAsync( + storeName, + workKey, + next, + etag, + cancellationToken: cancellationToken).ConfigureAwait(false); if (!saved.IsSuccess) return saved; return Result.Ok(saved.Value); } + /// public async Task> TryRenewAsync( string storeName, string workKey, @@ -63,9 +101,12 @@ public sealed class DaprWorkLeaseStore( if (!validation.IsSuccess) return validation.ToResultOfType(false); - var existing = await stateStore.GetStateAndETagAsync(storeName, workKey, cancellationToken).ConfigureAwait(false); + var existing = await stateStore.GetStateAndETagAsync( + storeName, + workKey, + cancellationToken: cancellationToken).ConfigureAwait(false); if (!existing.IsSuccess) - return existing.ToResult().ToResultOfType(false); + return existing.ToResultOfType(false); var (lease, etag) = existing.Value; if (lease is null || !string.Equals(lease.HolderId, holderId, StringComparison.Ordinal)) @@ -73,13 +114,19 @@ public sealed class DaprWorkLeaseStore( var now = DateTimeOffset.UtcNow; var next = lease with { ExpiresAtUtc = now.Add(ttl) }; - var saved = await stateStore.TrySaveStateAsync(storeName, workKey, next, etag, cancellationToken).ConfigureAwait(false); + var saved = await stateStore.TrySaveStateAsync( + storeName, + workKey, + next, + etag, + cancellationToken: cancellationToken).ConfigureAwait(false); if (!saved.IsSuccess) return saved; return Result.Ok(saved.Value); } + /// public async Task ReleaseAsync( string storeName, string workKey, @@ -88,7 +135,10 @@ public sealed class DaprWorkLeaseStore( if (string.IsNullOrWhiteSpace(storeName) || string.IsNullOrWhiteSpace(workKey) || string.IsNullOrWhiteSpace(holderId)) return Result.BadRequest("storeName, workKey, and holderId are required."); - var existing = await stateStore.GetStateAndETagAsync(storeName, workKey, cancellationToken).ConfigureAwait(false); + var existing = await stateStore.GetStateAndETagAsync( + storeName, + workKey, + cancellationToken: cancellationToken).ConfigureAwait(false); if (!existing.IsSuccess) return existing.ToResult(); @@ -99,23 +149,49 @@ public sealed class DaprWorkLeaseStore( if (!string.Equals(lease.HolderId, holderId, StringComparison.Ordinal)) return Result.Conflict("Lease is held by another instance."); - return await stateStore.DeleteStateAsync(storeName, workKey).ConfigureAwait(false); + return await stateStore.DeleteStateAsync(storeName, workKey, cancellationToken: cancellationToken).ConfigureAwait(false); } + /// public async Task> GetAsync( string storeName, string workKey, CancellationToken cancellationToken = default) { if (string.IsNullOrWhiteSpace(storeName) || string.IsNullOrWhiteSpace(workKey)) - return Result.BadRequest(null, "storeName and workKey are required."); + return Result.BadRequest(default, "storeName and workKey are required."); - var existing = await stateStore.GetStateAndETagAsync(storeName, workKey, cancellationToken).ConfigureAwait(false); + var existing = await stateStore.GetStateAndETagAsync( + storeName, + workKey, + cancellationToken: cancellationToken).ConfigureAwait(false); if (!existing.IsSuccess) - return existing.ToResult().ToResultOfType(null); + return existing.ToResultOfType((DaprWorkLease?)null); return Result.Ok(existing.Value.Value); } + /// + public async Task> TryHoldAsync( + string storeName, + string workKey, + string holderId, + TimeSpan ttl, + bool autoRenew = true, + CancellationToken cancellationToken = default) { + var acquired = await TryAcquireAsync(storeName, workKey, holderId, ttl, cancellationToken).ConfigureAwait(false); + if (!acquired.IsSuccess) + return acquired.ToResultOfType((DaprWorkLeaseHold?)null); + if (!acquired.Value) + return Result.Ok(null); + + var current = await GetAsync(storeName, workKey, cancellationToken).ConfigureAwait(false); + if (!current.IsSuccess) + return current.ToResultOfType((DaprWorkLeaseHold?)null); + + var generation = current.Value?.Generation ?? 0; + return Result.Ok(new DaprWorkLeaseHold(this, storeName, workKey, holderId, ttl, generation, autoRenew)); + } + private static Result Validate(string storeName, string workKey, string holderId, TimeSpan ttl) { if (string.IsNullOrWhiteSpace(storeName) || string.IsNullOrWhiteSpace(workKey) || string.IsNullOrWhiteSpace(holderId)) return Result.BadRequest("storeName, workKey, and holderId are required."); diff --git a/utils/engines/release/scriptSettings.json b/utils/engines/release/scriptSettings.json index 4b17cf3..0f54898 100644 --- a/utils/engines/release/scriptSettings.json +++ b/utils/engines/release/scriptSettings.json @@ -16,7 +16,7 @@ "stageLabel": "test", "enabled": true, "project": "..\\..\\..\\src\\MaksIT.Dapr.Tests", - "resultsDir": "..\\..\\..\\testResults" + "resultsDir": "..\\..\\..\\test-results" }, { "name": "QualityGate",