Reliable billing: outbox, concurrency safety, Compose and integration tests

.config/dotnet-tools.json

@@ -0,0 +1,7 @@
+{
+  "version": 1,
+  "isRoot": true,
+  "tools": {
+    "dotnet-ef": { "version": "8.0.11", "commands": ["dotnet-ef"] }
+  }
+}

.dockerignore

@@ -0,0 +1,7 @@
+**/bin
+**/obj
+.git
+.vs
+.tools
+.env
+artifacts

.gitattributes

@@ -0,0 +1,10 @@
+* text=auto
+*.cs text eol=lf
+*.csproj text eol=lf
+*.json text eol=lf
+*.md text eol=lf
+*.yaml text eol=lf
+*.yml text eol=lf
+*.ps1 text eol=lf
+*.py text eol=lf
+Dockerfile text eol=lf

.github/workflows/ci.yml

@@ -0,0 +1,23 @@
+name: billing-ci
+on: [push, pull_request]
+permissions:
+  contents: read
+jobs:
+  integration:
+    runs-on: ubuntu-latest
+    timeout-minutes: 15
+    steps:
+      - uses: actions/checkout@v4
+      - name: Start full stack and apply migrations
+        run: docker compose up --build -d
+      - name: Test against PostgreSQL and RabbitMQ
+        run: docker compose --profile test run --build --rm tests
+      - name: Smoke and outage recovery
+        shell: pwsh
+        run: ./scripts/verify.ps1
+      - name: Logs
+        if: failure()
+        run: docker compose logs --no-color
+      - name: Cleanup
+        if: always()
+        run: docker compose down -v

.gitignore

@@ -482,3 +482,5 @@ $RECYCLE.BIN/
 
 # Vim temporary swap files
 *.swp
+
+.tools/

Directory.Build.props

@@ -0,0 +1,6 @@
+<Project>
+  <PropertyGroup>
+    <TreatWarningsAsErrors>true</TreatWarningsAsErrors>
+    <AnalysisLevel>latest</AnalysisLevel>
+  </PropertyGroup>
+</Project>

Dockerfile

@@ -0,0 +1,24 @@
+FROM mcr.microsoft.com/dotnet/sdk:8.0 AS restore
+WORKDIR /src
+COPY Directory.Build.props global.json ./
+COPY ["Subscription & Billing/Subscription & Billing.csproj", "Subscription & Billing/"]
+RUN dotnet restore "Subscription & Billing/Subscription & Billing.csproj"
+
+FROM restore AS test-restore
+COPY tests/Billing.Tests/Billing.Tests.csproj tests/Billing.Tests/
+RUN dotnet restore tests/Billing.Tests/Billing.Tests.csproj
+
+FROM test-restore AS test
+COPY . .
+ENTRYPOINT ["dotnet", "test", "tests/Billing.Tests/Billing.Tests.csproj", "-c", "Release", "--no-restore", "--logger", "console;verbosity=normal"]
+
+FROM restore AS build
+COPY . .
+RUN dotnet publish "Subscription & Billing/Subscription & Billing.csproj" -c Release --no-restore -o /app /p:UseAppHost=false
+
+FROM mcr.microsoft.com/dotnet/aspnet:8.0 AS runtime
+WORKDIR /app
+COPY --from=build /app .
+USER app
+EXPOSE 8080
+ENTRYPOINT ["dotnet", "Billing.dll"]

README.md

@@ -0,0 +1,85 @@
+# Subscription Billing
+
+ASP.NET Core backend: запись оплаты, создание подписки и асинхронная активация.
+PostgreSQL 16 · RabbitMQ 4 · EF Core 8 · MassTransit 8 · .NET 8.
+
+## Запуск
+
+Нужен Docker с Compose. Из корня репозитория:
+
+```sh
+docker compose up --build -d
+```
+
+Открой [Swagger](http://localhost:8080/swagger). Compose запускает БД, брокер,
+миграции, API и worker. Первая сборка скачивает образы. Данные сохраняются в volumes.
+Остановка: `docker compose down`. Локальные пароли заданы в Compose;
+их можно заменить через `.env` (`POSTGRES_PASSWORD`, `RABBITMQ_PASSWORD`).
+
+## Архитектура
+
+```mermaid
+flowchart LR
+    C[Client + Idempotency-Key] --> A[API: validation + concurrency limit]
+    A --> S[SubscriptionService]
+    S -->|one transaction| DB[(PostgreSQL: subscription + payment + response + outbox)]
+    DB --> D[MassTransit outbox delivery]
+    D -->|publisher confirm| R[RabbitMQ durable queue]
+    R --> W[Worker: retry + inbox]
+    W -->|conditional activation| DB
+    W -->|failed after retries| E[Error queue]
+```
+
+- **Деньги и событие атомарны:** outbox записывается в той же транзакции.
+  При недоступном брокере запрос сохраняется, доставка продолжится после восстановления.
+- **Параллельные повторы безопасны:** PostgreSQL advisory lock по ключу,
+  сохранённый ответ, `409` при смене тела. Нет блокировок в памяти одного сервера.
+- **Доставка at-least-once:** inbox и условный UPDATE защищают активацию от дублей.
+  Необработанные события после трёх повторов попадают в `subscription-activation_error`.
+- **Контроль нагрузки:** максимум 64 одновременных API-запроса, затем `429`;
+  prefetch 32, 8 обработчиков на worker; индекс и cursor pagination с лимитом 100.
+- **Эксплуатация:** отдельные API/worker, миграции перед запуском, контейнер без root,
+  JSON-логи, `traceId` в ProblemDetails, `/health/live` и `/health/ready`.
+  API остаётся готовым при сбое брокера, пока доступна БД; worker требует брокер.
+
+## Попробовать
+
+1. `POST /api/users`: `{"name":"Stepan","email":"stepan@example.com"}`.
+2. `POST /api/subscriptions/create`: `{"userId":1,"plan":"basic"}`,
+   заголовок `Idempotency-Key: 75be42df-1323-48c2-a317-55bb64445a1b`.
+3. Ответ `202` и `Location`; `GET /api/subscriptions/1` покажет `isActive: true`
+   после обработки события. Повтор шага 2 вернёт исходный ответ без второй оплаты.
+
+Цены задаёт сервер: basic = 990 RUB, pro = 1990 RUB. Список:
+`GET /api/subscriptions?userId=1&afterId=0&limit=20`. Полный контракт в Swagger.
+
+## Проверка
+
+```sh
+docker compose --profile test run --build --rm tests
+```
+
+xUnit + WebApplicationFactory проверяют настоящий PostgreSQL и RabbitMQ:
+20 параллельных повторов, откат после вставки платежа, конфликт ключа,
+валидацию, серверную цену, пагинацию, дубли событий и активацию.
+Тесты используют отдельную БД `billing_tests` и отдельный exchange/queue.
+
+`pwsh ./scripts/verify.ps1` проверяет полный стек, отключает брокер,
+перезапускает API и проверяет доставку сохранённого события после восстановления.
+GitHub Actions запускает обе проверки. [Архитектурные решения и масштабирование](docs/architecture.md).
+
+Мини-бенчмарк записи (Python 3, без пакетов):
+`python scripts/load.py --requests 200 --concurrency 20`.
+Он измеряет p50/p95 и число принятых запросов в секунду, затем проверяет активацию.
+[Результат проверок и условия замера](docs/verification.md).
+
+## Границы проекта
+
+Это демонстрация надёжности backend, оплата симулируется: платёжного провайдера,
+авторизации, возвратов и периодических списаний пока нет. Compose открыт только
+на localhost. Для публичного сервиса нужны auth, проверенные webhooks провайдера,
+TLS и управление секретами. Высокая нагрузка требует замеров, а не обещаний RPS.
+
+Контракт изменён: клиентский `amount` удалён, POST теперь возвращает `202`.
+Начальная миграция предназначена для новой БД; старую БД, созданную через
+`EnsureCreated`, нужно переносить отдельно с резервной копией.

Subscription & Billing.sln

@@ -5,6 +5,10 @@ VisualStudioVersion = 17.11.35222.181
 MinimumVisualStudioVersion = 10.0.40219.1
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Subscription & Billing", "Subscription & Billing\Subscription & Billing.csproj", "{355B9D45-5658-43C0-B86D-5135298F19E9}"
 EndProject
+Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "tests", "tests", "{4127394C-64CD-400E-9F55-570D4DE21232}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Billing.Tests", "tests\Billing.Tests\Billing.Tests.csproj", "{321D0816-D5EA-4EAC-B948-169278FD8EC1}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -15,11 +19,18 @@ Global
 		{355B9D45-5658-43C0-B86D-5135298F19E9}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{355B9D45-5658-43C0-B86D-5135298F19E9}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{355B9D45-5658-43C0-B86D-5135298F19E9}.Release|Any CPU.Build.0 = Release|Any CPU
+		{321D0816-D5EA-4EAC-B948-169278FD8EC1}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{321D0816-D5EA-4EAC-B948-169278FD8EC1}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{321D0816-D5EA-4EAC-B948-169278FD8EC1}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{321D0816-D5EA-4EAC-B948-169278FD8EC1}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
 		SolutionGuid = {A288BAF6-7662-485E-AB6E-A4FCF8F7AA78}
 	EndGlobalSection
+	GlobalSection(NestedProjects) = preSolution
+		{321D0816-D5EA-4EAC-B948-169278FD8EC1} = {4127394C-64CD-400E-9F55-570D4DE21232}
+	EndGlobalSection
 EndGlobal

Subscription & Billing/AppDbContext.cs

@@ -1,17 +1,31 @@
+using MassTransit;
 using Microsoft.EntityFrameworkCore;
 using Subscription_Backend.Models;
-using System.Collections.Generic;
 
-namespace Subscription_Backend
-{
-    public class AppDbContext : DbContext
-    {
-        public AppDbContext(DbContextOptions<AppDbContext> options) : base(options) { }
+namespace Subscription_Backend;
 
-        public DbSet<User> Users => Set<User>();
-        public DbSet<Subscription> Subscriptions => Set<Subscription>();
-        public DbSet<Payment> Payments => Set<Payment>();
-        public DbSet<IdempotencyRecord> IdempotencyRecords => Set<IdempotencyRecord>();
+public sealed class AppDbContext(DbContextOptions<AppDbContext> options) : DbContext(options)
+{
+    public DbSet<User> Users => Set<User>();
+    public DbSet<Subscription> Subscriptions => Set<Subscription>();
+    public DbSet<Payment> Payments => Set<Payment>();
+    public DbSet<IdempotencyRecord> IdempotencyRecords => Set<IdempotencyRecord>();
 
+    protected override void OnModelCreating(ModelBuilder model)
+    {
+        model.Entity<User>().Property(x => x.Name).HasMaxLength(120);
+        model.Entity<User>().Property(x => x.Email).HasMaxLength(254);
+        model.Entity<User>().HasIndex(x => x.Email).IsUnique();
+        model.Entity<Subscription>().Property(x => x.Plan).HasMaxLength(32);
+        model.Entity<Subscription>().HasOne<User>().WithMany().HasForeignKey(x => x.UserId);
+        model.Entity<Subscription>().HasIndex(x => new { x.UserId, x.Id });
+        model.Entity<Payment>().HasIndex(x => x.SubscriptionId).IsUnique();
+        model.Entity<Payment>().Property(x => x.Amount).HasPrecision(18, 2);
+        model.Entity<Payment>().ToTable(t => t.HasCheckConstraint("CK_Payment_Amount", "\"Amount\" > 0"));
+        model.Entity<IdempotencyRecord>().HasKey(x => x.Key);
+        model.Entity<IdempotencyRecord>().Property(x => x.RequestHash).HasMaxLength(64);
+        model.AddInboxStateEntity();
+        model.AddOutboxMessageEntity();
+        model.AddOutboxStateEntity();
     }
 }

Subscription & Billing/Consumers/SubscriptionActivationConsumer.cs

@@ -1,40 +1,21 @@
 using MassTransit;
 using Microsoft.EntityFrameworkCore;
 using Subscription_Backend.Messages;
-using Subscription_Backend.Models;
 
 namespace Subscription_Backend.Consumers;
 
-public class SubscriptionActivationConsumer : IConsumer<SubscriptionPaidEvent>
+public sealed class SubscriptionActivationConsumer(AppDbContext db,
+    ILogger<SubscriptionActivationConsumer> logger) : IConsumer<SubscriptionPaidEvent>
 {
-    private readonly AppDbContext _db;
-    private readonly ILogger<SubscriptionActivationConsumer> _logger;
-
-    public SubscriptionActivationConsumer(AppDbContext db, ILogger<SubscriptionActivationConsumer> logger)
-    {
-        _db = db;
-        _logger = logger;
-    }
-
     public async Task Consume(ConsumeContext<SubscriptionPaidEvent> context)
     {
-        var message = context.Message;
-        _logger.LogInformation($"[Worker] �������� ������� ������ ��� �������� {message.SubscriptionId}");
-
-        // ������ Random Exception. 
-        // ���� ���� ������ ���� �� ���� � Retry ��������� �������������.
-
-        var subscription = await _db.Subscriptions.FindAsync(message.SubscriptionId);
-
-        if (subscription == null)
-        {
-            _logger.LogError($"�������� {message.SubscriptionId} �� �������!");
-            return;
-        }
-
-        subscription.IsActive = true;
-        await _db.SaveChangesAsync();
-
-        _logger.LogInformation($"[Worker] �������� {message.SubscriptionId} ������� ������������!");
+        var id = context.Message.SubscriptionId;
+        if (!await db.Payments.AnyAsync(x => x.SubscriptionId == id && x.Amount == context.Message.Amount,
+                context.CancellationToken))
+            throw new InvalidOperationException($"Payment for subscription {id} is missing or inconsistent.");
+        // Also safe for different message IDs carrying the same business event.
+        var changed = await db.Subscriptions.Where(x => x.Id == id && !x.IsActive)
+            .ExecuteUpdateAsync(s => s.SetProperty(x => x.IsActive, true), context.CancellationToken);
+        logger.LogInformation("Activation processed for {SubscriptionId}; changed {Changed}", id, changed);
     }
-}
+}

Subscription & Billing/Controllers/SubscriptionWorkflowController.cs

@@ -1,34 +1,46 @@
 using Microsoft.AspNetCore.Mvc;
+using Microsoft.EntityFrameworkCore;
 using Subscription_Backend.Dtos;
-using Subscription_Backend.Models;
 using Subscription_Backend.Services;
 
 namespace Subscription_Backend.Controllers;
 
 [ApiController]
 [Route("api/subscriptions")]
-public class SubscriptionWorkflowController : ControllerBase
+public sealed class SubscriptionWorkflowController(SubscriptionService service, AppDbContext db) : ControllerBase
 {
-    private readonly SubscriptionService _service;
-
-    public SubscriptionWorkflowController(SubscriptionService service)
+    [HttpPost("create")]
+    [ProducesResponseType<SubscriptionResponse>(StatusCodes.Status202Accepted)]
+    [ProducesResponseType<ProblemDetails>(StatusCodes.Status400BadRequest)]
+    [ProducesResponseType<ProblemDetails>(StatusCodes.Status404NotFound)]
+    [ProducesResponseType<ProblemDetails>(StatusCodes.Status409Conflict)]
+    public async Task<ActionResult<SubscriptionResponse>> Create(CreateSubscriptionRequest request,
+        [FromHeader(Name = "Idempotency-Key")] Guid key, CancellationToken ct)
     {
-        _service = service;
+        if (key == Guid.Empty)
+            return Problem(statusCode: 400, detail: "A non-empty UUID Idempotency-Key header is required.");
+        var result = await service.CreateSubscriptionAsync(request, key, ct);
+        return AcceptedAtAction(nameof(Get), new { id = result.Id }, result);
     }
 
-    [HttpPost("create")]
-    public async Task<ActionResult<Subscription>> Create(
-    CreateSubscriptionRequest request,
-    [FromHeader(Name = "Idempotency-Key")] Guid idempotencyKey)
+    [HttpGet("{id:int}")]
+    [ProducesResponseType<SubscriptionResponse>(StatusCodes.Status200OK)]
+    [ProducesResponseType<ProblemDetails>(StatusCodes.Status404NotFound)]
+    public async Task<ActionResult<SubscriptionResponse>> Get(int id, CancellationToken ct)
     {
-        if (idempotencyKey == Guid.Empty)
-            return BadRequest("Idempotency-Key header is required");
-
-        var subscription = await _service.CreateSubscriptionAsync(
-            request,
-            idempotencyKey);
-
-        return Ok(subscription);
+        var result = await db.Subscriptions.AsNoTracking().Where(x => x.Id == id)
+            .Select(x => new SubscriptionResponse(x.Id, x.UserId, x.Plan, x.IsActive)).SingleOrDefaultAsync(ct);
+        return result is null ? NotFound() : Ok(result);
     }
 
+    [HttpGet]
+    public async Task<ActionResult<IReadOnlyList<SubscriptionResponse>>> List(
+        [FromQuery] int userId, [FromQuery] int afterId = 0, [FromQuery] int limit = 20, CancellationToken ct = default)
+    {
+        if (userId < 1 || afterId < 0 || limit is < 1 or > 100)
+            return Problem(statusCode: 400, detail: "userId > 0, afterId >= 0, limit between 1 and 100 required.");
+        return Ok(await db.Subscriptions.AsNoTracking().Where(x => x.UserId == userId && x.Id > afterId)
+            .OrderBy(x => x.Id).Take(limit)
+            .Select(x => new SubscriptionResponse(x.Id, x.UserId, x.Plan, x.IsActive)).ToListAsync(ct));
+    }
 }