From 12288fcd10eef3685d0b41e067341004085281c3 Mon Sep 17 00:00:00 2001 From: cherninkiy Date: Fri, 15 May 2026 14:02:47 +0300 Subject: [PATCH 1/9] docs(adr): add ADR002 DbContext composition pattern - define composition approach using IEntityTypeConfiguration - each service owns its DbSet declarations, shared configs in Shared/ - ApiGateway is migration owner for common tables (documents, processed_messages) - Worker relies on existing schema, does not create migrations for common tables --- docs/adr/ADR002_DbContext_Composition.md | 231 ++++++++++++++++++++++ docs/adr/ADR002_DbContext_Composition.mmd | 30 +++ 2 files changed, 261 insertions(+) create mode 100644 docs/adr/ADR002_DbContext_Composition.md create mode 100644 docs/adr/ADR002_DbContext_Composition.mmd diff --git a/docs/adr/ADR002_DbContext_Composition.md b/docs/adr/ADR002_DbContext_Composition.md new file mode 100644 index 0000000..0075231 --- /dev/null +++ b/docs/adr/ADR002_DbContext_Composition.md @@ -0,0 +1,231 @@ +# ADR-002: Использование композиции для DbContext вместо наследования или дублирования + +| **Статус** | **Дата** | **Автор** | +|------------|----------|-----------| +| Принято | 2026-05-15 | Черников Дмитрий | + +## Контекст (условия задачи) + +В системе имеется два сервиса: **ApiGateway** и **Worker**. Оба используют единую базу данных PostgreSQL, но работают с разными наборами таблиц: + +| Таблица | ApiGateway | Worker | +|---------|------------|--------| +| `documents` | ✅ (чтение/запись) | ✅ (чтение/запись) | +| `processed_messages` | ✅ (чтение/запись) | ✅ (чтение/запись) | +| `outbox` | ✅ (только запись) | ❌ | +| `workflow_checkpoints` | ❌ | ✅ | +| `agent_definitions` | ❌ | ✅ | + +Первоначально в MVP была **отдельная реализация DbContext в каждом сервисе** без общего кода. Это привело к дублированию конфигураций общих сущностей (`documents`, `processed_messages`). Позже была предпринята попытка использовать **наследование от базового класса `Shared.AppDbContext`**, чтобы сократить дублирование. Однако наследование создало проблемы с миграциями (конфликт создания одних и тех же таблиц) и нарушило принцип разделения ответственности (базовый класс «знал» о всех таблицах). + +Теперь система выходит за рамки MVP, ожидается развитие и, возможно, появление новых сервисов. Необходимо выбрать устойчивое архитектурное решение для работы с DbContext. + +## Решение (композиция) + +**Принято решение использовать композицию** через: +- Вынос всех конфигураций сущностей в `Shared/Configurations` в виде классов, реализующих `IEntityTypeConfiguration`. +- Создание отдельного `DbContext` для каждого сервиса, который применяет только необходимые конфигурации. +- Назначение одного сервиса (ApiGateway) владельцем миграций для общих таблиц (`documents`, `processed_messages`). Worker использует эти таблицы через существующую схему, не создавая миграции для них. + +### Архитектурная схема композиции + +```mermaid +flowchart LR + subgraph Shared + C1["DocumentConfiguration"] + C2["ProcessedMessageConfiguration"] + C3["OutboxConfiguration"] + C4["WorkflowCheckpointConfiguration"] + C5["AgentDefinitionConfiguration"] + end + + subgraph ApiGateway + A["GatewayDbContext"] + A1["DbSet"] + A2["DbSet"] + A3["DbSet"] + A --> C1 + A --> C2 + A --> C3 + end + + subgraph Worker + B["WorkerDbContext"] + B1["DbSet"] + B2["DbSet"] + B4["DbSet"] + B5["DbSet"] + B --> C1 + B --> C2 + B --> C4 + B --> C5 + end +``` + +### Ключевые элементы реализации + +1. **Конфигурации сущностей в Shared** + Каждая сущность получает отдельный конфигурационный класс, расположенный в папке `Shared/Configurations`. Конфигурации содержат имя таблицы, ключи, индексы, ограничения и типы колонок. + +2. **GatewayDbContext (владелец общих таблиц)** + - Определяет `DbSet` для `Document`, `ProcessedMessage`, `OutboxMessage`. + - В `OnModelCreating` применяет соответствующие конфигурации. + - Миграции создаются только для этого контекста. Он отвечает за создание и обновление схемы для `documents` и `processed_messages`. + +3. **WorkerDbContext (потребитель общих таблиц)** + - Определяет `DbSet` для `Document`, `ProcessedMessage`, `WorkflowCheckpoint`, `AgentDefinition`. + - Применяет конфигурации всех используемых сущностей. + - **Не создаёт миграции** для общих таблиц (полагается на `GatewayDbContext`). При необходимости может иметь собственные миграции только для своих уникальных таблиц, но в production миграции не применяются (или применяются с осторожностью). + +4. **Управление миграциями** + - В `Program.cs` ApiGateway вызывается `dbContext.Database.MigrateAsync()` – создаёт/обновляет общую схему. + - В `Program.cs` Worker **не вызывается** `MigrateAsync()`. Вместо этого используется `dbContext.Database.EnsureCreated()` (только для разработки) или полагается на существование схемы. В production Worker просто подключается к уже подготовленной базе. + +### Код примера + +**Shared/Configurations/DocumentConfiguration.cs** +```csharp +public class DocumentConfiguration : IEntityTypeConfiguration +{ + public void Configure(EntityTypeBuilder builder) + { + builder.ToTable("documents"); + builder.HasKey(e => e.Id); + builder.Property(e => e.Filename).HasMaxLength(512).IsRequired(); + builder.Property(e => e.Status).HasConversion().IsRequired(); + builder.Property(e => e.FilePath).HasMaxLength(1024).IsRequired(); + builder.Property(e => e.CreatedAt).IsRequired(); + } +} +``` + +**ApiGateway/Data/GatewayDbContext.cs** +```csharp +public class GatewayDbContext : DbContext +{ + public GatewayDbContext(DbContextOptions options) : base(options) { } + public DbSet Documents => Set(); + public DbSet ProcessedMessages => Set(); + public DbSet OutboxMessages => Set(); + + protected override void OnModelCreating(ModelBuilder modelBuilder) + { + modelBuilder.ApplyConfiguration(new DocumentConfiguration()); + modelBuilder.ApplyConfiguration(new ProcessedMessageConfiguration()); + modelBuilder.ApplyConfiguration(new OutboxConfiguration()); + } +} +``` + +**Worker/Data/WorkerDbContext.cs** +```csharp +public class WorkerDbContext : DbContext +{ + public WorkerDbContext(DbContextOptions options) : base(options) { } + public DbSet Documents => Set(); + public DbSet ProcessedMessages => Set(); + public DbSet WorkflowCheckpoints => Set(); + public DbSet AgentDefinitions => Set(); + + protected override void OnModelCreating(ModelBuilder modelBuilder) + { + modelBuilder.ApplyConfiguration(new DocumentConfiguration()); + modelBuilder.ApplyConfiguration(new ProcessedMessageConfiguration()); + modelBuilder.ApplyConfiguration(new WorkflowCheckpointConfiguration()); + modelBuilder.ApplyConfiguration(new AgentDefinitionConfiguration()); + } +} +``` + +## Альтернативы, рассмотренные и отклонённые + +### Альтернатива 1: Отдельная реализация в каждом сервисе (без общего кода) + +**Суть:** Каждый сервис содержит свой собственный `DbContext`, полностью независимо определяющий нужные сущности и их конфигурации. Общие таблицы (`documents`, `processed_messages`) описываются дважды (в ApiGateway и в Worker). + +**Плюсы:** +- Полная независимость сервисов. +- Миграции не конфликтуют (каждый сервис свою схему создаёт отдельно – но в реальности при одной БД это приведёт к конфликту создания одних и тех же таблиц). +- Простота понимания (нет shared-зависимостей). + +**Минусы:** +- **Дублирование кода** – конфигурации общих таблиц повторяются. При изменении схемы нужно править в двух местах. +- **Рассогласование** – можно случайно изменить конфигурацию в одном сервисе, а во втором забыть, что вызовет ошибки времени выполнения. +- **Проблема миграций** – при одной БД оба контекста попытаются создать одни и те же таблицы, что приведёт к ошибке `already exists`. Решение – ручное управление миграциями (один контекст «владелец», второй – игнорирует общие таблицы). Но это не очевидно из кода. + +**Почему отклонена:** Дублирование и риск рассинхронизации неприемлемы для проекта, выходящего за рамки MVP. Подход не масштабируется при добавлении третьего сервиса. + +### Альтернатива 2: Наследование от общего базового DbContext + +**Суть:** Создаётся базовый класс `Shared.AppDbContext`, содержащий общие `DbSet` и их конфигурации. ApiGateway и Worker наследуют от него и добавляют свои таблицы. Используется текущая реализация (до принятия ADR-002). + +**Плюсы:** +- Устранено дублирование общих таблиц. +- Единое место изменения схемы общих сущностей. +- Простота добавления новых таблиц в базовый класс. + +**Минусы:** +- **Проблема миграций** – EF Core при создании миграции для наследника пытается включить в неё все таблицы из базового класса. При запуске миграции в ApiGateway и Worker будут попытки повторно создать `documents` и `processed_messages`. Это приводит к конфликтам или дублирующемуся коду миграций. +- **Нарушение SRP** – базовый класс «знает» о таблицах обоих сервисов. Изменение в Worker (например, новое поле в `workflow_checkpoints`) требует изменения общего класса, хотя ApiGateway это не нужно. +- **Сложность тестирования** – невозможно изолированно протестировать WorkerDbContext без поднятия всей схемы ApiGateway. +- **Разрастание базового класса** – при добавлении новых сервисов все таблицы будут скапливаться в одном месте, создавая «God DbContext». + +**Почему отклонена:** Наследование DbContext в EF Core – признанный анти-паттерн для production-проектов с общей БД. Проблемы с миграциями и нарушение SRP перевешивают удобство устранения дублирования. + +## Плюсы решения (композиция) + +1. **Устранение дублирования** – конфигурации вынесены в Shared и переиспользуются. Изменение схемы общих таблиц происходит в одном месте. +2. **Разделение ответственности** – каждый сервис определяет только те таблицы, которые ему нужны. Нет «знания» о чужих сущностях. +3. **Чистые миграции** – только один сервис (ApiGateway) управляет схемой общих таблиц. Worker не создаёт миграций для `documents` и `processed_messages`, что исключает конфликты. +4. **Гибкость** – при добавлении нового сервиса (например, ReportingService) достаточно создать его собственный `DbContext` и применить нужные конфигурации. Не нужно менять общий базовый класс. +5. **Тестируемость** – каждый контекст можно тестировать изолированно (InMemory, Testcontainers) с минимальным набором сущностей. +6. **Поддержка разных СУБД в будущем** – если потребуется разнести базы данных, это будет легко сделать, так как контексты полностью независимы. + +## Минусы решения и их смягчение + +| Минус | Смягчение | +|-------|-----------| +| **Небольшое дублирование кода** – каждый контекст всё равно объявляет `DbSet` для общих сущностей. | Объём кода мал, явное объявление повышает читаемость и инкапсуляцию. Это не дублирование логики, а декларация зависимостей. | +| **Риск несинхронного применения конфигураций** – можно в одном сервисе применить устаревшую конфигурацию. | Конфигурации берутся из общей папки `Shared/Configurations`. Если конфигурация изменена, она автоматически изменится для всех. | +| **Worker не управляет миграциями общих таблиц** – при изменении схемы нужно помнить, что миграцию создаёт только ApiGateway. | Документировано в ADR и в коде (комментарии в `Program.cs` Worker). Автоматизированные тесты накатывают миграцию Gateway перед запуском тестов Worker. | +| **Дополнительная сложность для новичков** – нестандартный паттерн. | Композиция с `IEntityTypeConfiguration` – это стандартная рекомендация Microsoft. Документация и явные комментарии помогут. | + +## Последствия (Consequences) + +### Что меняется в проекте + +- **Удаляются** существующие классы `Shared.AppDbContext` и наследники. +- **Создаётся** папка `Shared/Configurations` с классами конфигураций для всех сущностей. +- **Создаются** `ApiGateway/Data/GatewayDbContext` и `Worker/Data/WorkerDbContext` (новые имена, чтобы не путать со старыми). +- **Обновляется** `Program.cs` ApiGateway: остается вызов `MigrateAsync()`. +- **Обновляется** `Program.cs` Worker: убирается `MigrateAsync()`, при необходимости добавляется `EnsureCreated()` только для разработки или проверка существования схемы. +- **DI-регистрация** обновляется: вместо `Shared.AppDbContext` регистрируются `GatewayDbContext` и `WorkerDbContext`. + +### Риски + +- **При развёртывании новой версии** необходимо сначала запустить ApiGateway (чтобы применить миграции), затем Worker. В противном случае Worker может временно работать со старой схемой, что не критично, но может вызвать ошибки, если изменения ломающие. Это решается оркестрацией (Docker Compose с условиями, K8s initContainer). +- **Worker при старте не проверяет актуальность схемы** – если миграции не были накачены, Worker упадёт с ошибкой о несуществующей колонке. В production это контролируется последовательностью запуска. + +### Что нужно донести до команды + +- При добавлении новой таблицы или изменении существующей **конфигурация создаётся/изменяется в `Shared/Configurations`**. +- **Миграции** создаются только от `GatewayDbContext`. Команда: `dotnet ef migrations add --context GatewayDbContext --startup-project ApiGateway --project ApiGateway`. +- Worker **никогда не накатывает миграции на production**. Он только читает схему, созданную ApiGateway. +- Если в будущем потребуется, чтобы Worker владел своими уникальными таблицами (не общими), можно добавить миграции для `WorkerDbContext`, но перед этим нужно убедиться, что они не создают `documents` и `processed_messages` повторно. Это достигается удалением этих строк из миграции вручную или использованием `modelBuilder.Ignore()` в WorkerDbContext (тогда Worker вообще не будет знать об этих таблицах, и придётся работать через SQL или отдельный репозиторий). + +## Итог + +Композиция с вынесением конфигураций в `IEntityTypeConfiguration` и созданием отдельных `DbContext` для каждого сервиса признана наиболее подходящим архитектурным решением для развивающейся системы. Она обеспечивает: + +- Отсутствие дублирования кода. +- Разделение ответственности. +- Управляемость миграций. +- Гибкость для добавления новых сервисов. + +Альтернативы (отдельные реализации в каждом сервисе и наследование) отклонены из-за дублирования кода, конфликтов миграций и нарушения принципов SOLID. + +--- + +**Дата принятия:** 2026-05-15 +**Автор:** Черников Дмитрий +**Утверждено:** для реализации композиции во всех сервисах системы. \ No newline at end of file diff --git a/docs/adr/ADR002_DbContext_Composition.mmd b/docs/adr/ADR002_DbContext_Composition.mmd new file mode 100644 index 0000000..9d49dc8 --- /dev/null +++ b/docs/adr/ADR002_DbContext_Composition.mmd @@ -0,0 +1,30 @@ +flowchart LR + subgraph Shared + C1["DocumentConfiguration"] + C2["ProcessedMessageConfiguration"] + C3["OutboxConfiguration"] + C4["WorkflowCheckpointConfiguration"] + C5["AgentDefinitionConfiguration"] + end + + subgraph ApiGateway + A["GatewayDbContext"] + A1["DbSet"] + A2["DbSet"] + A3["DbSet"] + A --> C1 + A --> C2 + A --> C3 + end + + subgraph Worker + B["WorkerDbContext"] + B1["DbSet"] + B2["DbSet"] + B4["DbSet"] + B5["DbSet"] + B --> C1 + B --> C2 + B --> C4 + B --> C5 + end \ No newline at end of file From d67587a05983f68655c55afc91b5264c7a290484 Mon Sep 17 00:00:00 2001 From: cherninkiy Date: Fri, 22 May 2026 22:45:37 +0300 Subject: [PATCH 2/9] refactor(shared): extract entity configurations and base repository - add Microsoft.EntityFrameworkCore 8.0.x and Relational to Shared.csproj - create IEntityTypeConfiguration for all 5 entities in Shared/Configurations/ - split IDocumentRepository into base interface + IOutboxRepository - create DocumentRepositoryBase with common SQL operations --- .../AgentDefinitionConfiguration.cs | 18 +++++ .../Configurations/DocumentConfiguration.cs | 18 +++++ .../Configurations/OutboxConfiguration.cs | 18 +++++ .../ProcessedMessageConfiguration.cs | 15 ++++ .../WorkflowCheckpointConfiguration.cs | 20 +++++ src/Shared/Interfaces/IDocumentRepository.cs | 18 ++++- .../Repositories/DocumentRepositoryBase.cs | 77 +++++++++++++++++++ src/Shared/Shared.csproj | 2 + 8 files changed, 183 insertions(+), 3 deletions(-) create mode 100644 src/Shared/Configurations/AgentDefinitionConfiguration.cs create mode 100644 src/Shared/Configurations/DocumentConfiguration.cs create mode 100644 src/Shared/Configurations/OutboxConfiguration.cs create mode 100644 src/Shared/Configurations/ProcessedMessageConfiguration.cs create mode 100644 src/Shared/Configurations/WorkflowCheckpointConfiguration.cs create mode 100644 src/Shared/Repositories/DocumentRepositoryBase.cs diff --git a/src/Shared/Configurations/AgentDefinitionConfiguration.cs b/src/Shared/Configurations/AgentDefinitionConfiguration.cs new file mode 100644 index 0000000..010b436 --- /dev/null +++ b/src/Shared/Configurations/AgentDefinitionConfiguration.cs @@ -0,0 +1,18 @@ +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Metadata.Builders; +using Shared.Models; + +namespace Shared.Configurations; + +public class AgentDefinitionConfiguration : IEntityTypeConfiguration +{ + public void Configure(EntityTypeBuilder builder) + { + builder.ToTable("agent_definitions"); + builder.HasKey(e => e.Id); + builder.Property(e => e.Name).HasMaxLength(128).IsRequired(); + builder.Property(e => e.HandlerType).HasMaxLength(512).IsRequired(); + builder.Property(e => e.Activities).HasColumnType("jsonb"); + builder.HasIndex(e => e.Name).IsUnique(); + } +} \ No newline at end of file diff --git a/src/Shared/Configurations/DocumentConfiguration.cs b/src/Shared/Configurations/DocumentConfiguration.cs new file mode 100644 index 0000000..bcae39b --- /dev/null +++ b/src/Shared/Configurations/DocumentConfiguration.cs @@ -0,0 +1,18 @@ +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Metadata.Builders; +using Shared.Models; + +namespace Shared.Configurations; + +public class DocumentConfiguration : IEntityTypeConfiguration +{ + public void Configure(EntityTypeBuilder builder) + { + builder.ToTable("documents"); + builder.HasKey(e => e.Id); + builder.Property(e => e.Filename).HasMaxLength(512).IsRequired(); + builder.Property(e => e.Status).HasConversion().IsRequired(); + builder.Property(e => e.FilePath).HasMaxLength(1024).IsRequired(); + builder.Property(e => e.CreatedAt).IsRequired(); + } +} \ No newline at end of file diff --git a/src/Shared/Configurations/OutboxConfiguration.cs b/src/Shared/Configurations/OutboxConfiguration.cs new file mode 100644 index 0000000..0262c88 --- /dev/null +++ b/src/Shared/Configurations/OutboxConfiguration.cs @@ -0,0 +1,18 @@ +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Metadata.Builders; +using Shared.Models; + +namespace Shared.Configurations; + +public class OutboxConfiguration : IEntityTypeConfiguration +{ + public void Configure(EntityTypeBuilder builder) + { + builder.ToTable("outbox"); + builder.HasKey(e => e.Id); + builder.Property(e => e.DocumentId).IsRequired(); + builder.Property(e => e.MessagePayload).HasColumnType("jsonb").IsRequired(); + builder.Property(e => e.CreatedAt).IsRequired(); + builder.HasIndex(e => e.ProcessedAt).HasFilter("\"processed_at\" IS NULL"); + } +} \ No newline at end of file diff --git a/src/Shared/Configurations/ProcessedMessageConfiguration.cs b/src/Shared/Configurations/ProcessedMessageConfiguration.cs new file mode 100644 index 0000000..7f4553f --- /dev/null +++ b/src/Shared/Configurations/ProcessedMessageConfiguration.cs @@ -0,0 +1,15 @@ +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Metadata.Builders; +using Shared.Models; + +namespace Shared.Configurations; + +public class ProcessedMessageConfiguration : IEntityTypeConfiguration +{ + public void Configure(EntityTypeBuilder builder) + { + builder.ToTable("processed_messages"); + builder.HasKey(e => e.MessageId); + builder.HasIndex(e => e.MessageId); + } +} \ No newline at end of file diff --git a/src/Shared/Configurations/WorkflowCheckpointConfiguration.cs b/src/Shared/Configurations/WorkflowCheckpointConfiguration.cs new file mode 100644 index 0000000..8ea4a3f --- /dev/null +++ b/src/Shared/Configurations/WorkflowCheckpointConfiguration.cs @@ -0,0 +1,20 @@ +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Metadata.Builders; +using Shared.Models; + +namespace Shared.Configurations; + +public class WorkflowCheckpointConfiguration : IEntityTypeConfiguration +{ + public void Configure(EntityTypeBuilder builder) + { + builder.ToTable("workflow_checkpoints"); + builder.HasKey(e => e.Id); + builder.Property(e => e.AgentName).HasMaxLength(128).IsRequired(); + builder.Property(e => e.CurrentActivity).HasMaxLength(128).IsRequired(); + builder.Property(e => e.StateData).HasColumnType("text"); + builder.Property(e => e.ErrorMessage).HasMaxLength(4096); + builder.HasIndex(e => new { e.AgentName, e.DocumentId }); + builder.HasIndex(e => e.IsCompleted); + } +} \ No newline at end of file diff --git a/src/Shared/Interfaces/IDocumentRepository.cs b/src/Shared/Interfaces/IDocumentRepository.cs index 44dc54f..9e38c36 100644 --- a/src/Shared/Interfaces/IDocumentRepository.cs +++ b/src/Shared/Interfaces/IDocumentRepository.cs @@ -2,15 +2,27 @@ namespace Shared.Interfaces; +/// +/// Base document repository interface with common operations. +/// Implemented by both ApiGateway and Worker repositories. +/// public interface IDocumentRepository { - Task CreateAsync(DocumentDto document, OutboxMessage outboxMessage, CancellationToken cancellationToken = default); Task GetByIdAsync(Guid id, CancellationToken cancellationToken = default); Task> GetAllAsync(CancellationToken cancellationToken = default); Task TryUpdateStatusAsync(Guid id, DocumentStatus fromStatus, DocumentStatus toStatus, string? errorMessage = null, CancellationToken cancellationToken = default); Task UpdateTextAsync(Guid id, string? extractedText, DocumentStatus status, CancellationToken cancellationToken = default); - Task> GetOutboxPendingAsync(CancellationToken cancellationToken = default); - Task MarkOutboxProcessedAsync(Guid id, CancellationToken cancellationToken = default); Task MarkMessageProcessedAsync(Guid messageId, Guid documentId, CancellationToken cancellationToken = default); Task IsMessageProcessedAsync(Guid messageId, CancellationToken cancellationToken = default); +} + +/// +/// Outbox-specific repository interface. +/// Only implemented by ApiGateway repository. +/// +public interface IOutboxRepository +{ + Task CreateAsync(DocumentDto document, OutboxMessage outboxMessage, CancellationToken cancellationToken = default); + Task> GetOutboxPendingAsync(CancellationToken cancellationToken = default); + Task MarkOutboxProcessedAsync(Guid id, CancellationToken cancellationToken = default); } \ No newline at end of file diff --git a/src/Shared/Repositories/DocumentRepositoryBase.cs b/src/Shared/Repositories/DocumentRepositoryBase.cs new file mode 100644 index 0000000..c2e3137 --- /dev/null +++ b/src/Shared/Repositories/DocumentRepositoryBase.cs @@ -0,0 +1,77 @@ +using Microsoft.EntityFrameworkCore; +using Shared.Models; + +namespace Shared.Repositories; + +public abstract class DocumentRepositoryBase where TContext : DbContext +{ + protected readonly TContext Context; + + protected DocumentRepositoryBase(TContext context) + { + Context = context; + } + + public virtual async Task GetByIdAsync(Guid id, CancellationToken ct = default) + { + return await Context.Set().FirstOrDefaultAsync(d => d.Id == id, ct); + } + + public virtual async Task AddAsync(DocumentDto document, CancellationToken ct = default) + { + await Context.Set().AddAsync(document, ct); + await Context.SaveChangesAsync(ct); + } + + public virtual async Task> GetAllAsync(CancellationToken ct = default) + { + return await Context.Set().OrderByDescending(d => d.CreatedAt).ToListAsync(ct); + } + + public virtual async Task TryUpdateStatusAsync( + Guid id, DocumentStatus fromStatus, DocumentStatus toStatus, + string? errorMessage = null, CancellationToken ct = default) + { + var fromStatusInt = (int)fromStatus; + var toStatusInt = (int)toStatus; + + int rows; + if (toStatus == DocumentStatus.Processing) + { + rows = await Context.Database.ExecuteSqlRawAsync( + "UPDATE documents SET status = {0}, started_at = {1} WHERE id = {2} AND status = {3}", + toStatusInt, DateTime.UtcNow, id, fromStatusInt, ct); + } + else + { + rows = await Context.Database.ExecuteSqlRawAsync( + "UPDATE documents SET status = {0}, completed_at = {1}, error_message = {2} WHERE id = {3} AND status = {4}", + toStatusInt, DateTime.UtcNow, errorMessage!, id, fromStatusInt, ct); + } + + return rows > 0; + } + + public virtual async Task UpdateTextAsync(Guid id, string? extractedText, DocumentStatus status, CancellationToken ct = default) + { + await Context.Database.ExecuteSqlRawAsync( + "UPDATE documents SET extracted_text = {0}, status = {1}, completed_at = {2} WHERE id = {3}", + extractedText!, (int)status, DateTime.UtcNow, id, ct); + } + + public virtual async Task MarkMessageProcessedAsync(Guid messageId, Guid documentId, CancellationToken ct = default) + { + Context.Set().Add(new ProcessedMessage + { + MessageId = messageId, + DocumentId = documentId, + ProcessedAt = DateTime.UtcNow + }); + await Context.SaveChangesAsync(ct); + } + + public virtual async Task IsMessageProcessedAsync(Guid messageId, CancellationToken ct = default) + { + return await Context.Set().AnyAsync(p => p.MessageId == messageId, ct); + } +} \ No newline at end of file diff --git a/src/Shared/Shared.csproj b/src/Shared/Shared.csproj index b2fe3e1..7be5969 100644 --- a/src/Shared/Shared.csproj +++ b/src/Shared/Shared.csproj @@ -9,6 +9,8 @@ + + \ No newline at end of file From 12e73723e2027e6ac60a7e43bf09fcde948ef6fa Mon Sep 17 00:00:00 2001 From: cherninkiy Date: Fri, 22 May 2026 22:46:40 +0300 Subject: [PATCH 3/9] refactor(db): create service-specific DbContext classes - create GatewayDbContext in ApiGateway with Documents, ProcessedMessages, OutboxMessages - create WorkerDbContext in Worker with Documents, ProcessedMessages, WorkflowCheckpoints, AgentDefinitions - each context applies only its required IEntityTypeConfiguration - remove old shared AppDbContext --- src/ApiGateway/Data/GatewayDbContext.cs | 25 +++++++++++++++++++++++ src/Worker/Data/WorkerDbContext.cs | 27 +++++++++++++++++++++++++ 2 files changed, 52 insertions(+) create mode 100644 src/ApiGateway/Data/GatewayDbContext.cs create mode 100644 src/Worker/Data/WorkerDbContext.cs diff --git a/src/ApiGateway/Data/GatewayDbContext.cs b/src/ApiGateway/Data/GatewayDbContext.cs new file mode 100644 index 0000000..b671ff3 --- /dev/null +++ b/src/ApiGateway/Data/GatewayDbContext.cs @@ -0,0 +1,25 @@ +using Microsoft.EntityFrameworkCore; +using Shared.Configurations; +using Shared.Models; + +namespace ApiGateway.Data; + +/// +/// ApiGateway-specific DbContext with Outbox support. +/// Applies shared entity configurations and defines DbSets for this service. +/// +public class GatewayDbContext : DbContext +{ + public GatewayDbContext(DbContextOptions options) : base(options) { } + + public DbSet Documents => Set(); + public DbSet ProcessedMessages => Set(); + public DbSet OutboxMessages => Set(); + + protected override void OnModelCreating(ModelBuilder modelBuilder) + { + modelBuilder.ApplyConfiguration(new DocumentConfiguration()); + modelBuilder.ApplyConfiguration(new ProcessedMessageConfiguration()); + modelBuilder.ApplyConfiguration(new OutboxConfiguration()); + } +} \ No newline at end of file diff --git a/src/Worker/Data/WorkerDbContext.cs b/src/Worker/Data/WorkerDbContext.cs new file mode 100644 index 0000000..2fc2b50 --- /dev/null +++ b/src/Worker/Data/WorkerDbContext.cs @@ -0,0 +1,27 @@ +using Microsoft.EntityFrameworkCore; +using Shared.Configurations; +using Shared.Models; + +namespace Worker.Data; + +/// +/// Worker-specific DbContext with WorkflowCheckpoint and AgentDefinition support. +/// Applies shared entity configurations and defines DbSets for this service. +/// +public class WorkerDbContext : DbContext +{ + public WorkerDbContext(DbContextOptions options) : base(options) { } + + public DbSet Documents => Set(); + public DbSet ProcessedMessages => Set(); + public DbSet WorkflowCheckpoints => Set(); + public DbSet AgentDefinitions => Set(); + + protected override void OnModelCreating(ModelBuilder modelBuilder) + { + modelBuilder.ApplyConfiguration(new DocumentConfiguration()); + modelBuilder.ApplyConfiguration(new ProcessedMessageConfiguration()); + modelBuilder.ApplyConfiguration(new WorkflowCheckpointConfiguration()); + modelBuilder.ApplyConfiguration(new AgentDefinitionConfiguration()); + } +} \ No newline at end of file From 528664df1cc6abcc5a7ba9ba901f7e655ae0b098 Mon Sep 17 00:00:00 2001 From: cherninkiy Date: Fri, 22 May 2026 22:48:12 +0300 Subject: [PATCH 4/9] refactor(db): create service-specific document repositories - create ApiGateway DocumentRepository with Outbox support (implements IOutboxRepository) - create Worker DocumentRepository with WorkflowCheckpoint queries - both inherit from DocumentRepositoryBase --- .../Repositories/DocumentRepository.cs | 43 +++++++++++++++++++ src/Worker/Repositories/DocumentRepository.cs | 35 +++++++++++++++ 2 files changed, 78 insertions(+) create mode 100644 src/ApiGateway/Repositories/DocumentRepository.cs create mode 100644 src/Worker/Repositories/DocumentRepository.cs diff --git a/src/ApiGateway/Repositories/DocumentRepository.cs b/src/ApiGateway/Repositories/DocumentRepository.cs new file mode 100644 index 0000000..3750495 --- /dev/null +++ b/src/ApiGateway/Repositories/DocumentRepository.cs @@ -0,0 +1,43 @@ +using ApiGateway.Data; +using Microsoft.EntityFrameworkCore; +using Shared.Interfaces; +using Shared.Models; +using Shared.Repositories; + +namespace ApiGateway.Repositories; + +/// +/// ApiGateway-specific document repository. +/// Inherits shared SQL operations from DocumentRepositoryBase. +/// Implements both IDocumentRepository and IOutboxRepository. +/// +public class DocumentRepository : DocumentRepositoryBase, IDocumentRepository, IOutboxRepository +{ + public DocumentRepository(GatewayDbContext context) : base(context) { } + + public async Task CreateAsync(DocumentDto document, OutboxMessage outboxMessage, CancellationToken ct = default) + { + await Context.Documents.AddAsync(document, ct); + await Context.OutboxMessages.AddAsync(outboxMessage, ct); + await Context.SaveChangesAsync(ct); + } + + public async Task> GetOutboxPendingAsync(CancellationToken ct = default) + { + return await Context.OutboxMessages + .Where(o => o.ProcessedAt == null) + .OrderBy(o => o.CreatedAt) + .Take(50) + .ToListAsync(ct); + } + + public async Task MarkOutboxProcessedAsync(Guid id, CancellationToken ct = default) + { + var message = await Context.OutboxMessages.FirstOrDefaultAsync(o => o.Id == id, ct); + if (message != null) + { + message.ProcessedAt = DateTime.UtcNow; + await Context.SaveChangesAsync(ct); + } + } +} \ No newline at end of file diff --git a/src/Worker/Repositories/DocumentRepository.cs b/src/Worker/Repositories/DocumentRepository.cs new file mode 100644 index 0000000..937d3a6 --- /dev/null +++ b/src/Worker/Repositories/DocumentRepository.cs @@ -0,0 +1,35 @@ +using Microsoft.EntityFrameworkCore; +using Shared.Interfaces; +using Shared.Models; +using Shared.Repositories; +using Worker.Data; + +namespace Worker.Repositories; + +/// +/// Worker-specific document repository. +/// Inherits shared SQL operations from DocumentRepositoryBase. +/// Worker does not use Outbox — it consumes messages from RabbitMQ. +/// +public class DocumentRepository : DocumentRepositoryBase, IDocumentRepository +{ + public DocumentRepository(WorkerDbContext context) : base(context) { } + + public async Task> GetCompletedCheckpointsAsync( + string agentName, Guid documentId, CancellationToken ct = default) + { + return await Context.WorkflowCheckpoints + .Where(c => c.AgentName == agentName && c.DocumentId == documentId && c.IsCompleted && !c.IsFailed) + .OrderBy(c => c.CreatedAt) + .ToListAsync(ct); + } + + public async Task GetLastCheckpointAsync( + string agentName, Guid documentId, CancellationToken ct = default) + { + return await Context.WorkflowCheckpoints + .Where(c => c.AgentName == agentName && c.DocumentId == documentId) + .OrderByDescending(c => c.UpdatedAt) + .FirstOrDefaultAsync(ct); + } +} \ No newline at end of file From 5c27410842eae291b55ef64e9510688b890aebcd Mon Sep 17 00:00:00 2001 From: cherninkiy Date: Fri, 22 May 2026 22:49:03 +0300 Subject: [PATCH 5/9] refactor(api-gateway): update to use new DbContext and repository structure - update OutboxPublisher to use IOutboxRepository instead of IDocumentRepository - update DocumentService to accept IOutboxRepository for CreateDocumentAsync - update Program.cs to use GatewayDbContext with EnsureCreated - register both IDocumentRepository and IOutboxRepository in DI --- .../BackgroundServices/OutboxPublisher.cs | 23 ++------ .../Extensions/ServiceCollectionExtensions.cs | 25 +++----- src/ApiGateway/Program.cs | 58 ++----------------- src/ApiGateway/Services/DocumentService.cs | 35 ++--------- 4 files changed, 21 insertions(+), 120 deletions(-) diff --git a/src/ApiGateway/BackgroundServices/OutboxPublisher.cs b/src/ApiGateway/BackgroundServices/OutboxPublisher.cs index 52e8916..0ce9a70 100644 --- a/src/ApiGateway/BackgroundServices/OutboxPublisher.cs +++ b/src/ApiGateway/BackgroundServices/OutboxPublisher.cs @@ -5,18 +5,6 @@ namespace ApiGateway.BackgroundServices; -/// -/// Background service that implements the Transactional Outbox pattern. -/// Polls the outbox table every 5s, publishes pending messages to RabbitMQ via MassTransit, -/// then marks them as processed. This guarantees at-least-once delivery without -/// relying on a distributed transaction between the database and message broker. -/// -/// Workflow: -/// 1. POST /upload → document + outbox row saved in same DB transaction -/// 2. OutboxPublisher picks up unprocessed rows (processed_at IS NULL) -/// 3. Deserializes JSON payload → publishes to RabbitMQ -/// 4. Marks row as processed → won't be picked up again -/// public class OutboxPublisher : BackgroundService { private readonly IServiceScopeFactory _scopeFactory; @@ -39,14 +27,13 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken) try { using var scope = _scopeFactory.CreateScope(); - var repository = scope.ServiceProvider.GetRequiredService(); - var pendingMessages = await repository.GetOutboxPendingAsync(stoppingToken); + var outboxRepository = scope.ServiceProvider.GetRequiredService(); + var pendingMessages = await outboxRepository.GetOutboxPendingAsync(stoppingToken); foreach (var message in pendingMessages) { try { - // Deserialize payload — handle corrupt messages gracefully to avoid blocking the queue PdfProcessingCommand? command = null; try { @@ -55,7 +42,7 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken) catch (JsonException ex) { _logger.LogError(ex, "Corrupt outbox message payload for {OutboxId}, marking as processed to unblock queue", message.Id); - await repository.MarkOutboxProcessedAsync(message.Id, stoppingToken); + await outboxRepository.MarkOutboxProcessedAsync(message.Id, stoppingToken); continue; } @@ -65,8 +52,7 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken) _logger.LogInformation("Published outbox message {OutboxId} for document {DocumentId}", message.Id, message.DocumentId); } - // Mark as processed so it won't be picked up again - await repository.MarkOutboxProcessedAsync(message.Id, stoppingToken); + await outboxRepository.MarkOutboxProcessedAsync(message.Id, stoppingToken); } catch (Exception ex) { @@ -79,7 +65,6 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken) _logger.LogError(ex, "Error in OutboxPublisher loop"); } - // Poll interval — balance between latency (shorter) and DB load (longer) await Task.Delay(TimeSpan.FromSeconds(5), stoppingToken); } diff --git a/src/ApiGateway/Extensions/ServiceCollectionExtensions.cs b/src/ApiGateway/Extensions/ServiceCollectionExtensions.cs index 26830aa..c0a95b2 100644 --- a/src/ApiGateway/Extensions/ServiceCollectionExtensions.cs +++ b/src/ApiGateway/Extensions/ServiceCollectionExtensions.cs @@ -1,4 +1,5 @@ using ApiGateway.Data; +using ApiGateway.Repositories; using ApiGateway.Storage; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.Hosting; @@ -8,16 +9,6 @@ namespace ApiGateway.Extensions; public static class ServiceCollectionExtensions { - /// - /// Configures the database provider based on environment and connection string availability. - /// - /// - Testing environment → in-memory database (for unit tests) - /// - Production with connection string → PostgreSQL with snake_case naming - /// - Fallback (e.g., local dev without PostgreSQL) → in-memory database - /// - /// Extracted from Program.cs to comply with SRP — the entry point should - /// only compose services, not decide which database to use. - /// public static IServiceCollection AddDatabase( this IServiceCollection services, IConfiguration configuration, @@ -27,18 +18,18 @@ public static IServiceCollection AddDatabase( if (environment.IsEnvironment("Testing")) { - services.AddDbContext(options => + services.AddDbContext(options => options.UseInMemoryDatabase("TestDb")); } else if (!string.IsNullOrWhiteSpace(connectionString)) { - services.AddDbContext(options => + services.AddDbContext(options => options.UseNpgsql(connectionString) .UseSnakeCaseNamingConvention()); } else { - services.AddDbContext(options => + services.AddDbContext(options => options.UseInMemoryDatabase("TestDb")); } @@ -49,10 +40,6 @@ public static IServiceCollection AddApplicationServices(this IServiceCollection { var storageProvider = configuration.GetValue("Storage__Provider") ?? "local"; - // Determine storage path: - // 1. Explicit config via Storage__LocalPath or Storage:LocalPath - // 2. /app/storage if running in Docker (directory exists or explicitly configured) - // 3. Fallback to temp directory for local dev and tests var localPath = configuration.GetValue("Storage__LocalPath") ?? configuration.GetValue("Storage:LocalPath"); @@ -71,7 +58,9 @@ public static IServiceCollection AddApplicationServices(this IServiceCollection }); } - services.AddScoped(); + services.AddScoped(); + services.AddScoped(sp => sp.GetRequiredService()); + services.AddScoped(sp => sp.GetRequiredService()); return services; } diff --git a/src/ApiGateway/Program.cs b/src/ApiGateway/Program.cs index 8fcdee2..d4beddf 100644 --- a/src/ApiGateway/Program.cs +++ b/src/ApiGateway/Program.cs @@ -10,35 +10,14 @@ using Scalar.AspNetCore; using Serilog; -// ------------------------------------------------------------ -// Program.cs – Application entry point -// ------------------------------------------------------------ -// This file wires up the entire API Gateway workflow: -// 1. Configures the database via AddDatabase() extension (PostgreSQL or in-memory). -// 2. Sets up MassTransit with RabbitMQ (skipped in Development to avoid external deps). -// 3. Registers application services, including the DocumentService and the OutboxPublisher background service. -// 4. Adds controllers and Swagger for API documentation. -// 5. Ensures the database schema is created on startup. -// The workflow follows the transactional outbox pattern: uploads are stored in the DB and an outbox row is created; the OutboxPublisher later publishes the message to RabbitMQ. var builder = WebApplication.CreateBuilder(args); -// ── Serilog (structured logging) ── -// Replaces default ILogger with Serilog + CompactJsonFormatter for production-grade -// structured log output. ReadFrom.Configuration picks up Serilog sections from appsettings. builder.Host.UseSerilog((context, loggerConfig) => loggerConfig.ReadFrom.Configuration(context.Configuration) .WriteTo.Console(new Serilog.Formatting.Compact.CompactJsonFormatter())); - // ── Database (PostgreSQL via EF Core) ── - // Delegated to AddDatabase() extension method for SRP compliance. - // See ServiceCollectionExtensions.AddDatabase() for logic. - builder.Services.AddDatabase(builder.Configuration, builder.Environment); +builder.Services.AddDatabase(builder.Configuration, builder.Environment); -// ── MassTransit + RabbitMQ ── -// Publishes PdfProcessingCommand messages. The OutboxPublisher -// background service handles reliable delivery via the outbox table. -// Add MassTransit only when RabbitMQ host is configured. -// Skip in unit tests (Testing environment) to avoid external dependencies. var rabbitHost = builder.Configuration.GetValue("RabbitMq:Host"); if (!string.IsNullOrWhiteSpace(rabbitHost) && !builder.Environment.IsEnvironment("Testing")) { @@ -57,79 +36,54 @@ }); } -// ── Health Checks ── builder.Services.AddHealthChecks() - .AddDbContextCheck("postgres"); -// RabbitMQ health check (custom) — only when configured + .AddDbContextCheck("postgres"); + if (!string.IsNullOrWhiteSpace(rabbitHost) && !builder.Environment.IsEnvironment("Testing")) { builder.Services.AddHealthChecks() .AddCheck("rabbitmq"); } -// ── Application Services ── -// DocumentService orchestrates upload → outbox → file storage. -// OutboxPublisher polls unprocessed outbox rows every 5s and publishes via MassTransit. builder.Services.AddApplicationServices(builder.Configuration); builder.Services.AddScoped(); -// Register the OutboxPublisher only when RabbitMQ is configured (not in unit tests). + if (!string.IsNullOrWhiteSpace(rabbitHost) && !builder.Environment.IsEnvironment("Testing")) { builder.Services.AddHostedService(); } -// ── JWT Authentication ── -// Delegated to AddGatewayAuthentication() extension method for SRP compliance. -// See AuthenticationExtensions.AddGatewayAuthentication() for logic. -// Supports two modes: -// - Production: validates against Jwt:Authority (external identity provider) -// - Development: self-signed tokens via Jwt:SecretKey + /auth/token endpoint -// Testing environment skips auth entirely. builder.Services.AddGatewayAuthentication(builder.Configuration, builder.Environment); -// ── Controllers + OpenAPI ── -// Using Microsoft.AspNetCore.OpenApi (built-in for .NET 10) instead of Swashbuckle. -// Scalar.AspNetCore provides the API explorer UI at /scalar. builder.Services.AddControllers(); builder.Services.AddEndpointsApiExplorer(); builder.Services.AddOpenApi(); var app = builder.Build(); -// ── Auto-create database tables (Dev only) ── -// Uses init.sql in PostgreSQL's docker-entrypoint-initdb.d for production-like setup. using (var scope = app.Services.CreateScope()) { - var db = scope.ServiceProvider.GetRequiredService(); + var db = scope.ServiceProvider.GetRequiredService(); db.Database.EnsureCreated(); } -// ── OpenAPI endpoint + Scalar UI ── -// /openapi/v1.json — OpenAPI 3.1 specification -// /scalar — interactive API documentation (modern Swagger UI alternative) app.MapOpenApi(); app.MapScalarApiReference(); -// JWT Authentication middleware (skipped in Testing) app.UseGatewayAuthentication(app.Environment); -// Prometheus metrics — must be before MapControllers to capture request metrics app.UseHttpMetrics(); app.MapControllers(); -// Health check endpoints -// /health/live: quick probe, runs no dependency checks app.MapHealthChecks("/health/live", new Microsoft.AspNetCore.Diagnostics.HealthChecks.HealthCheckOptions { Predicate = _ => false }); -// /health/ready: checks all dependencies (Postgres, RabbitMQ) app.MapHealthChecks("/health/ready"); app.MapMetrics(); app.Run(); -// Exposed for integration testing with WebApplicationFactory -public partial class Program { } +public partial class Program { } \ No newline at end of file diff --git a/src/ApiGateway/Services/DocumentService.cs b/src/ApiGateway/Services/DocumentService.cs index 3d6dbcd..ad38319 100644 --- a/src/ApiGateway/Services/DocumentService.cs +++ b/src/ApiGateway/Services/DocumentService.cs @@ -4,43 +4,28 @@ namespace ApiGateway.Services; -// ------------------------------------------------------------ -// DocumentService – Core business logic for the API Gateway -// ------------------------------------------------------------ -// This service implements the upload flow using the transactional outbox pattern: -// * Saves the uploaded PDF to the configured IFileStorage implementation. -// * Persists a DocumentDto with status 'Uploaded'. -// * Creates an OutboxMessage containing a PdfProcessingCommand. -// * Both the document and outbox row are saved in a single DB transaction. -// The OutboxPublisher background service later reads pending outbox rows and -// publishes the command to RabbitMQ for asynchronous processing by the Worker. public class DocumentService { private static readonly Counter UploadCount = Metrics .CreateCounter("document_upload_total", "Total number of uploaded documents."); private readonly IDocumentRepository _repository; + private readonly IOutboxRepository _outboxRepository; private readonly IFileStorage _fileStorage; private readonly ILogger _logger; public DocumentService( IDocumentRepository repository, + IOutboxRepository outboxRepository, IFileStorage fileStorage, ILogger logger) { _repository = repository; + _outboxRepository = outboxRepository; _fileStorage = fileStorage; _logger = logger; } - /// - /// Core upload logic with Transactional Outbox. - /// 1. Save file to storage (local volume or MinIO) - /// 2. Create document record (status=uploaded) - /// 3. Create outbox row with serialized PdfProcessingCommand - /// Steps 2+3 happen in the same DB transaction (see DocumentRepository.CreateAsync). - /// The OutboxPublisher background service will later pick up the outbox row and publish to RabbitMQ. - /// public async Task CreateDocumentAsync(Stream fileStream, string filename, CancellationToken cancellationToken = default) { var documentId = Guid.NewGuid(); @@ -71,8 +56,7 @@ public async Task CreateDocumentAsync(Stream fileStream, string CreatedAt = DateTime.UtcNow }; - // Atomic insert: document + outbox in one transaction - await _repository.CreateAsync(document, outboxMessage, cancellationToken); + await _outboxRepository.CreateAsync(document, outboxMessage, cancellationToken); UploadCount.Inc(); _logger.LogInformation("Document {DocumentId} uploaded, filename: {Filename}, outbox message: {OutboxId}", documentId, filename, outboxMessage.Id); @@ -84,13 +68,9 @@ public async Task CreateDocumentAsync(Stream fileStream, string }; } - /// - /// Returns all documents ordered by creation date (newest first). - /// public async Task> GetAllDocumentsAsync(CancellationToken cancellationToken = default) { var documents = await _repository.GetAllAsync(cancellationToken); - // Ensure deterministic ordering: newest first by CreatedAt, then by Id to break ties. var ordered = documents .OrderByDescending(d => d.CreatedAt) .ThenByDescending(d => d.Id) @@ -105,13 +85,6 @@ public async Task> GetAllDocumentsAsync(CancellationToken return ordered; } - /// - /// Returns extracted text for a document based on its status: - /// - Completed → return text with 200 - /// - Processing/Uploaded → return 202 (retry later) - /// - Failed → return 409 (error message included) - /// - Not found → return 404 - /// public async Task<(DocumentDto? Document, string? ErrorMessage, int? StatusCode)> GetDocumentTextAsync(Guid id, CancellationToken cancellationToken = default) { var document = await _repository.GetByIdAsync(id, cancellationToken); From 544e49e8aa964199454019113322c090171b1b08 Mon Sep 17 00:00:00 2001 From: cherninkiy Date: Fri, 22 May 2026 22:50:08 +0300 Subject: [PATCH 6/9] refactor(worker): update to use WorkerDbContext and new repository - update Program.cs to use WorkerDbContext with EnsureCreated - update PostgreSqlCheckpointStore to use WorkerDbContext - register Worker DocumentRepository in DI --- src/Worker/Data/PostgreSqlCheckpointStore.cs | 4 ++-- src/Worker/Program.cs | 6 +++--- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/src/Worker/Data/PostgreSqlCheckpointStore.cs b/src/Worker/Data/PostgreSqlCheckpointStore.cs index 25c4bda..21802e1 100644 --- a/src/Worker/Data/PostgreSqlCheckpointStore.cs +++ b/src/Worker/Data/PostgreSqlCheckpointStore.cs @@ -14,10 +14,10 @@ namespace Worker.Data; /// public class PostgreSqlCheckpointStore : ICheckpointStore { - private readonly AppDbContext _context; + private readonly WorkerDbContext _context; private readonly ILogger _logger; - public PostgreSqlCheckpointStore(AppDbContext context, ILogger logger) + public PostgreSqlCheckpointStore(WorkerDbContext context, ILogger logger) { _context = context; _logger = logger; diff --git a/src/Worker/Program.cs b/src/Worker/Program.cs index 95aa2f4..4115761 100644 --- a/src/Worker/Program.cs +++ b/src/Worker/Program.cs @@ -37,12 +37,12 @@ var configuration = hostContext.Configuration; // ── Database (PostgreSQL via EF Core) ── - services.AddDbContext(options => + services.AddDbContext(options => options.UseNpgsql(configuration.GetConnectionString("DefaultConnection")) .UseSnakeCaseNamingConvention()); // ── Repository ── - services.AddScoped(); + services.AddScoped(); // ── File Storage ── // Uses local Docker volume shared with ApiGateway for MVP. @@ -114,7 +114,7 @@ // Auto-create database tables (Dev only) using (var scope = host.Services.CreateScope()) { - var db = scope.ServiceProvider.GetRequiredService(); + var db = scope.ServiceProvider.GetRequiredService(); db.Database.EnsureCreated(); } From 9fb02f2a197786b773386ba310bb11a6bc08bf70 Mon Sep 17 00:00:00 2001 From: cherninkiy Date: Fri, 22 May 2026 22:58:49 +0300 Subject: [PATCH 7/9] refactor(tests): update unit tests for new repository structure - update DocumentServiceTests to use IOutboxRepository mock - update OutboxPublisherTests to use IOutboxRepository mock - fix CreateDocumentAsync test to verify outbox repository call --- .../DocumentServiceTests.cs | 27 +++------- .../OutboxPublisherTests.cs | 49 +++++-------------- 2 files changed, 20 insertions(+), 56 deletions(-) diff --git a/tests/ApiGateway.UnitTests/DocumentServiceTests.cs b/tests/ApiGateway.UnitTests/DocumentServiceTests.cs index d17411b..5499e59 100644 --- a/tests/ApiGateway.UnitTests/DocumentServiceTests.cs +++ b/tests/ApiGateway.UnitTests/DocumentServiceTests.cs @@ -9,43 +9,41 @@ namespace ApiGateway.UnitTests; public class DocumentServiceTests { private readonly Mock _repositoryMock; + private readonly Mock _outboxRepositoryMock; private readonly Mock _fileStorageMock; private readonly DocumentService _service; public DocumentServiceTests() { _repositoryMock = new Mock(); + _outboxRepositoryMock = new Mock(); _fileStorageMock = new Mock(); var loggerMock = new Mock>(); - _service = new DocumentService(_repositoryMock.Object, _fileStorageMock.Object, loggerMock.Object); + _service = new DocumentService(_repositoryMock.Object, _outboxRepositoryMock.Object, _fileStorageMock.Object, loggerMock.Object); } [Fact] public async Task CreateDocumentAsync_SavesFileAndCreatesOutbox() { - // Arrange var filename = "test.pdf"; - var content = new MemoryStream(new byte[] { 0x25, 0x50, 0x44, 0x46 }); // PDF magic bytes + var content = new MemoryStream(new byte[] { 0x25, 0x50, 0x44, 0x46 }); var expectedPath = $"/storage/{Guid.NewGuid()}.pdf"; _fileStorageMock.Setup(x => x.SaveAsync(It.IsAny(), It.IsAny(), It.IsAny())) .ReturnsAsync(expectedPath); - _repositoryMock.Setup(x => x.CreateAsync(It.IsAny(), It.IsAny(), It.IsAny())) + _outboxRepositoryMock.Setup(x => x.CreateAsync(It.IsAny(), It.IsAny(), It.IsAny())) .Returns(Task.CompletedTask); - // Act var result = await _service.CreateDocumentAsync(content, filename); - // Assert Assert.Equal("accepted", result.Status); Assert.NotEqual(Guid.Empty, result.DocumentId); _fileStorageMock.Verify(x => x.SaveAsync(It.IsAny(), It.IsAny(), It.IsAny()), Times.Once); - _repositoryMock.Verify(x => x.CreateAsync(It.IsAny(), It.IsAny(), It.IsAny()), Times.Once); + _outboxRepositoryMock.Verify(x => x.CreateAsync(It.IsAny(), It.IsAny(), It.IsAny()), Times.Once); } [Fact] public async Task GetAllDocumentsAsync_ReturnsOrderedList() { - // Arrange var docs = new List { new() { Id = Guid.NewGuid(), Filename = "a.pdf", Status = DocumentStatus.Completed, CreatedAt = DateTime.UtcNow.AddMinutes(-1) }, @@ -54,26 +52,21 @@ public async Task GetAllDocumentsAsync_ReturnsOrderedList() _repositoryMock.Setup(x => x.GetAllAsync(It.IsAny())) .ReturnsAsync(docs); - // Act var result = await _service.GetAllDocumentsAsync(); - // Assert Assert.Equal(2, result.Count); - Assert.Equal("b.pdf", result[0].Filename); // newest first + Assert.Equal("b.pdf", result[0].Filename); Assert.Equal("a.pdf", result[1].Filename); } [Fact] public async Task GetDocumentTextAsync_ReturnsNotFound_ForMissingId() { - // Arrange _repositoryMock.Setup(x => x.GetByIdAsync(It.IsAny(), It.IsAny())) .ReturnsAsync((DocumentDto?)null); - // Act var (doc, error, statusCode) = await _service.GetDocumentTextAsync(Guid.NewGuid()); - // Assert Assert.Null(doc); Assert.Equal(404, statusCode); Assert.Equal("Document not found", error); @@ -82,7 +75,6 @@ public async Task GetDocumentTextAsync_ReturnsNotFound_ForMissingId() [Fact] public async Task GetDocumentTextAsync_Returns200_ForCompletedDocument() { - // Arrange var id = Guid.NewGuid(); var doc = new DocumentDto { @@ -94,10 +86,8 @@ public async Task GetDocumentTextAsync_Returns200_ForCompletedDocument() _repositoryMock.Setup(x => x.GetByIdAsync(id, It.IsAny())) .ReturnsAsync(doc); - // Act var (result, error, statusCode) = await _service.GetDocumentTextAsync(id); - // Assert Assert.Equal(200, statusCode); Assert.Equal("Hello World", result?.ExtractedText); Assert.Null(error); @@ -106,7 +96,6 @@ public async Task GetDocumentTextAsync_Returns200_ForCompletedDocument() [Fact] public async Task GetDocumentTextAsync_Returns409_ForFailedDocument() { - // Arrange var id = Guid.NewGuid(); var doc = new DocumentDto { @@ -117,10 +106,8 @@ public async Task GetDocumentTextAsync_Returns409_ForFailedDocument() _repositoryMock.Setup(x => x.GetByIdAsync(id, It.IsAny())) .ReturnsAsync(doc); - // Act var (_, error, statusCode) = await _service.GetDocumentTextAsync(id); - // Assert Assert.Equal(409, statusCode); Assert.Contains("OCR failed", error); } diff --git a/tests/ApiGateway.UnitTests/OutboxPublisherTests.cs b/tests/ApiGateway.UnitTests/OutboxPublisherTests.cs index 936f0fd..d2f49f5 100644 --- a/tests/ApiGateway.UnitTests/OutboxPublisherTests.cs +++ b/tests/ApiGateway.UnitTests/OutboxPublisherTests.cs @@ -9,20 +9,12 @@ namespace ApiGateway.UnitTests; -/// -/// Unit tests for the OutboxPublisher background service. -/// -/// The OutboxPublisher implements the transactional outbox pattern: -/// it polls the outbox table, publishes pending messages via MassTransit, -/// then marks them as processed. These tests verify its reliability -/// under various failure conditions. -/// public class OutboxPublisherTests { private readonly Mock _scopeFactoryMock; private readonly Mock _scopeMock; private readonly Mock _busMock; - private readonly Mock _repositoryMock; + private readonly Mock _outboxRepositoryMock; private readonly Mock> _loggerMock; private readonly OutboxPublisher _publisher; @@ -32,17 +24,16 @@ public OutboxPublisherTests() _scopeMock = new Mock(); _busMock = new Mock(); _loggerMock = new Mock>(); - _repositoryMock = new Mock(); + _outboxRepositoryMock = new Mock(); - // Set up scope factory to return a scope that provides the repository _scopeFactoryMock .Setup(x => x.CreateScope()) .Returns(_scopeMock.Object); var serviceProviderMock = new Mock(); serviceProviderMock - .Setup(x => x.GetService(typeof(IDocumentRepository))) - .Returns(_repositoryMock.Object); + .Setup(x => x.GetService(typeof(IOutboxRepository))) + .Returns(_outboxRepositoryMock.Object); _scopeMock .Setup(x => x.ServiceProvider) @@ -57,7 +48,6 @@ public OutboxPublisherTests() [Fact] public async Task ExecuteAsync_PublishesPendingMessagesAndMarksProcessed() { - // Arrange var documentId = Guid.NewGuid(); var outboxId = Guid.NewGuid(); var command = new PdfProcessingCommand @@ -78,35 +68,31 @@ public async Task ExecuteAsync_PublishesPendingMessagesAndMarksProcessed() } }; - _repositoryMock + _outboxRepositoryMock .Setup(x => x.GetOutboxPendingAsync(It.IsAny())) .ReturnsAsync(pendingMessages); - _repositoryMock + _outboxRepositoryMock .Setup(x => x.MarkOutboxProcessedAsync(outboxId, It.IsAny())) .Returns(Task.CompletedTask); - // Use a cancellation token source to stop the loop after one iteration using var cts = new CancellationTokenSource(); var executeTask = _publisher.StartAsync(cts.Token); - // Allow one loop iteration to complete await Task.Delay(500); await cts.CancelAsync(); - // Assert _busMock.Verify(x => x.Publish( It.IsAny(), It.IsAny()), Times.Once); - _repositoryMock.Verify(x => x.MarkOutboxProcessedAsync( + _outboxRepositoryMock.Verify(x => x.MarkOutboxProcessedAsync( outboxId, It.IsAny()), Times.Once); } [Fact] public async Task ExecuteAsync_SkipsCorruptMessagesAndContinues() { - // Arrange var pendingMessages = new List { new() @@ -118,11 +104,11 @@ public async Task ExecuteAsync_SkipsCorruptMessagesAndContinues() } }; - _repositoryMock + _outboxRepositoryMock .Setup(x => x.GetOutboxPendingAsync(It.IsAny())) .ReturnsAsync(pendingMessages); - _repositoryMock + _outboxRepositoryMock .Setup(x => x.MarkOutboxProcessedAsync(It.IsAny(), It.IsAny())) .Returns(Task.CompletedTask); @@ -132,19 +118,17 @@ public async Task ExecuteAsync_SkipsCorruptMessagesAndContinues() await Task.Delay(500); await cts.CancelAsync(); - // Assert: corrupt message is marked as processed (unblock queue) but not published _busMock.Verify(x => x.Publish( It.IsAny(), It.IsAny()), Times.Never); - _repositoryMock.Verify(x => x.MarkOutboxProcessedAsync( + _outboxRepositoryMock.Verify(x => x.MarkOutboxProcessedAsync( It.IsAny(), It.IsAny()), Times.Once); } [Fact] public async Task ExecuteAsync_ContinuesOnPublishFailure() { - // Arrange var documentId = Guid.NewGuid(); var command = new PdfProcessingCommand { @@ -164,11 +148,10 @@ public async Task ExecuteAsync_ContinuesOnPublishFailure() } }; - _repositoryMock + _outboxRepositoryMock .Setup(x => x.GetOutboxPendingAsync(It.IsAny())) .ReturnsAsync(pendingMessages); - // Simulate publish failure on first attempt _busMock .Setup(x => x.Publish(It.IsAny(), It.IsAny())) .ThrowsAsync(new InvalidOperationException("RabbitMQ unavailable")); @@ -179,30 +162,24 @@ public async Task ExecuteAsync_ContinuesOnPublishFailure() await Task.Delay(500); await cts.CancelAsync(); - // Assert: exception is caught, loop continues, message is NOT marked processed - // (it will be retried on the next poll cycle) - _repositoryMock.Verify(x => x.MarkOutboxProcessedAsync( + _outboxRepositoryMock.Verify(x => x.MarkOutboxProcessedAsync( It.IsAny(), It.IsAny()), Times.Never); } [Fact] public async Task ExecuteAsync_StopsOnCancellation() { - // Arrange - _repositoryMock + _outboxRepositoryMock .Setup(x => x.GetOutboxPendingAsync(It.IsAny())) .ReturnsAsync(new List()); using var cts = new CancellationTokenSource(); - // Act: start and immediately cancel var executeTask = _publisher.StartAsync(cts.Token); await cts.CancelAsync(); - // Wait a brief moment for the loop to observe cancellation await Task.Delay(200); - // Assert: no exception thrown, service stops gracefully Assert.True(executeTask.IsCompletedSuccessfully); } } \ No newline at end of file From 54661f150b9be7c1b461ec4dd1519b54baa2a03c Mon Sep 17 00:00:00 2001 From: cherninkiy Date: Fri, 22 May 2026 23:04:57 +0300 Subject: [PATCH 8/9] refactor(db): remove old AppDbContext and DocumentRepository files - delete ApiGateway/Data/AppDbContext.cs (replaced by GatewayDbContext) - delete ApiGateway/Data/DocumentRepository.cs (replaced by Repositories/DocumentRepository) - delete Worker/Data/AppDbContext.cs (replaced by WorkerDbContext) - delete Worker/Data/DocumentRepository.cs (replaced by Repositories/DocumentRepository) --- src/ApiGateway/Data/AppDbContext.cs | 48 ---------- src/ApiGateway/Data/DocumentRepository.cs | 107 ---------------------- src/Worker/Data/AppDbContext.cs | 64 ------------- src/Worker/Data/DocumentRepository.cs | 92 ------------------- 4 files changed, 311 deletions(-) delete mode 100644 src/ApiGateway/Data/AppDbContext.cs delete mode 100644 src/ApiGateway/Data/DocumentRepository.cs delete mode 100644 src/Worker/Data/AppDbContext.cs delete mode 100644 src/Worker/Data/DocumentRepository.cs diff --git a/src/ApiGateway/Data/AppDbContext.cs b/src/ApiGateway/Data/AppDbContext.cs deleted file mode 100644 index 5f439a5..0000000 --- a/src/ApiGateway/Data/AppDbContext.cs +++ /dev/null @@ -1,48 +0,0 @@ -using Microsoft.EntityFrameworkCore; -using Shared.Models; - -namespace ApiGateway.Data; - -public class AppDbContext : DbContext -{ - public AppDbContext(DbContextOptions options) : base(options) { } - - public DbSet Documents => Set(); - public DbSet OutboxMessages => Set(); - public DbSet ProcessedMessages => Set(); - - protected override void OnModelCreating(ModelBuilder modelBuilder) - { - // Status is stored as INTEGER (matching document_statuses lookup table). - // C# enum DocumentStatus maps directly to int (0 = Uploaded, 1 = Processing, 2 = Completed, 3 = Failed). - // No JOIN needed in services — status int is sufficient for all business logic. - modelBuilder.Entity(entity => - { - entity.ToTable("documents"); - entity.HasKey(e => e.Id); - entity.Property(e => e.Filename).HasMaxLength(512).IsRequired(); - entity.Property(e => e.Status) - .HasConversion() - .IsRequired(); - entity.Property(e => e.FilePath).HasMaxLength(1024).IsRequired(); - entity.Property(e => e.CreatedAt).IsRequired(); - }); - - modelBuilder.Entity(entity => - { - entity.ToTable("outbox"); - entity.HasKey(e => e.Id); - entity.Property(e => e.DocumentId).IsRequired(); - entity.Property(e => e.MessagePayload).HasColumnType("jsonb").IsRequired(); - entity.Property(e => e.CreatedAt).IsRequired(); - entity.HasIndex(e => e.ProcessedAt).HasFilter("\"processed_at\" IS NULL"); - }); - - modelBuilder.Entity(entity => - { - entity.ToTable("processed_messages"); - entity.HasKey(e => e.MessageId); - entity.HasIndex(e => e.MessageId); - }); - } -} diff --git a/src/ApiGateway/Data/DocumentRepository.cs b/src/ApiGateway/Data/DocumentRepository.cs deleted file mode 100644 index 8492448..0000000 --- a/src/ApiGateway/Data/DocumentRepository.cs +++ /dev/null @@ -1,107 +0,0 @@ -using Microsoft.EntityFrameworkCore; -using Shared.Interfaces; -using Shared.Models; - -namespace ApiGateway.Data; - -public class DocumentRepository : IDocumentRepository -{ - private readonly AppDbContext _context; - - public DocumentRepository(AppDbContext context) - { - _context = context; - } - - public async Task CreateAsync(DocumentDto document, OutboxMessage outboxMessage, CancellationToken cancellationToken = default) - { - await _context.Documents.AddAsync(document, cancellationToken); - await _context.OutboxMessages.AddAsync(outboxMessage, cancellationToken); - await _context.SaveChangesAsync(cancellationToken); - } - - public async Task GetByIdAsync(Guid id, CancellationToken cancellationToken = default) - { - return await _context.Documents.FirstOrDefaultAsync(d => d.Id == id, cancellationToken); - } - - public async Task> GetAllAsync(CancellationToken cancellationToken = default) - { - return await _context.Documents - .OrderByDescending(d => d.CreatedAt) - .ToListAsync(cancellationToken); - } - - public async Task TryUpdateStatusAsync(Guid id, DocumentStatus fromStatus, DocumentStatus toStatus, string? errorMessage = null, CancellationToken cancellationToken = default) - { - // Atomic optimistic lock using raw SQL to avoid race conditions. - // Uses int status values (matching DocumentStatus enum) — no JOIN needed. - var fromStatusInt = (int)fromStatus; - var toStatusInt = (int)toStatus; - - int rows; - if (toStatus == DocumentStatus.Processing) - { - rows = await _context.Database.ExecuteSqlRawAsync( - "UPDATE documents SET status = {0}, started_at = NOW() WHERE id = {1} AND status = {2}", - toStatusInt, id, fromStatusInt, - cancellationToken); - } - else - { - rows = await _context.Database.ExecuteSqlRawAsync( - "UPDATE documents SET status = {0}, completed_at = NOW(), error_message = {1} WHERE id = {2} AND status = {3}", - toStatusInt, - errorMessage ?? (object)DBNull.Value, - id, fromStatusInt, - cancellationToken); - } - - return rows > 0; - } - - public async Task UpdateTextAsync(Guid id, string? extractedText, DocumentStatus status, CancellationToken cancellationToken = default) - { - await _context.Database.ExecuteSqlRawAsync( - "UPDATE documents SET extracted_text = {0}, status = {1}, completed_at = NOW() WHERE id = {2}", - extractedText ?? (object)DBNull.Value, - (int)status, - id, - cancellationToken); - } - - public async Task> GetOutboxPendingAsync(CancellationToken cancellationToken = default) - { - return await _context.OutboxMessages - .Where(o => o.ProcessedAt == null) - .OrderBy(o => o.CreatedAt) - .Take(50) - .ToListAsync(cancellationToken); - } - - public async Task MarkOutboxProcessedAsync(Guid id, CancellationToken cancellationToken = default) - { - var message = await _context.OutboxMessages.FirstOrDefaultAsync(o => o.Id == id, cancellationToken); - if (message != null) - { - message.ProcessedAt = DateTime.UtcNow; - await _context.SaveChangesAsync(cancellationToken); - } - } - - public async Task MarkMessageProcessedAsync(Guid messageId, Guid documentId, CancellationToken cancellationToken = default) - { - _context.ProcessedMessages.Add(new ProcessedMessage - { - MessageId = messageId, - DocumentId = documentId, - ProcessedAt = DateTime.UtcNow - }); - await _context.SaveChangesAsync(cancellationToken); - } - - public async Task IsMessageProcessedAsync(Guid messageId, CancellationToken cancellationToken = default) - { - return await _context.ProcessedMessages.AnyAsync(p => p.MessageId == messageId, cancellationToken); - } -} \ No newline at end of file diff --git a/src/Worker/Data/AppDbContext.cs b/src/Worker/Data/AppDbContext.cs deleted file mode 100644 index e82e625..0000000 --- a/src/Worker/Data/AppDbContext.cs +++ /dev/null @@ -1,64 +0,0 @@ -using Microsoft.EntityFrameworkCore; -using Shared.Models; - -namespace Worker.Data; - -public class AppDbContext : DbContext -{ - public AppDbContext(DbContextOptions options) : base(options) { } - - public DbSet Documents => Set(); - public DbSet ProcessedMessages => Set(); - public DbSet WorkflowCheckpoints => Set(); - public DbSet AgentDefinitions => Set(); - - protected override void OnModelCreating(ModelBuilder modelBuilder) - { - // Status is stored as INTEGER (matching document_statuses lookup table). - modelBuilder.Entity(entity => - { - entity.ToTable("documents"); - entity.HasKey(e => e.Id); - entity.Property(e => e.Filename).HasMaxLength(512).IsRequired(); - entity.Property(e => e.Status) - .HasConversion() - .IsRequired(); - entity.Property(e => e.FilePath).HasMaxLength(1024).IsRequired(); - entity.Property(e => e.CreatedAt).IsRequired(); - }); - - modelBuilder.Entity(entity => - { - entity.ToTable("processed_messages"); - entity.HasKey(e => e.MessageId); - entity.HasIndex(e => e.MessageId); - }); - - // ── Workflow Checkpoints ── - // Stores agent execution state for durable workflows. - // If a worker crashes, the agent resumes from the last checkpoint. - modelBuilder.Entity(entity => - { - entity.ToTable("workflow_checkpoints"); - entity.HasKey(e => e.Id); - entity.Property(e => e.AgentName).HasMaxLength(128).IsRequired(); - entity.Property(e => e.CurrentActivity).HasMaxLength(128).IsRequired(); - entity.Property(e => e.StateData).HasColumnType("text"); - entity.Property(e => e.ErrorMessage).HasMaxLength(4096); - entity.HasIndex(e => new { e.AgentName, e.DocumentId }); - entity.HasIndex(e => e.IsCompleted); - }); - - // ── Agent Definitions ── - // Registry of available agents for dynamic discovery. - modelBuilder.Entity(entity => - { - entity.ToTable("agent_definitions"); - entity.HasKey(e => e.Id); - entity.Property(e => e.Name).HasMaxLength(128).IsRequired(); - entity.Property(e => e.HandlerType).HasMaxLength(512).IsRequired(); - entity.Property(e => e.Activities).HasColumnType("jsonb"); - entity.HasIndex(e => e.Name).IsUnique(); - }); - } -} diff --git a/src/Worker/Data/DocumentRepository.cs b/src/Worker/Data/DocumentRepository.cs deleted file mode 100644 index b3f42a7..0000000 --- a/src/Worker/Data/DocumentRepository.cs +++ /dev/null @@ -1,92 +0,0 @@ -using Microsoft.EntityFrameworkCore; -using Shared.Interfaces; -using Shared.Models; - -namespace Worker.Data; - -public class DocumentRepository : IDocumentRepository -{ - private readonly AppDbContext _context; - - public DocumentRepository(AppDbContext context) - { - _context = context; - } - - public async Task CreateAsync(DocumentDto document, OutboxMessage outboxMessage, CancellationToken cancellationToken = default) - { - await _context.Documents.AddAsync(document, cancellationToken); - await _context.SaveChangesAsync(cancellationToken); - } - - public async Task GetByIdAsync(Guid id, CancellationToken cancellationToken = default) - { - return await _context.Documents.FirstOrDefaultAsync(d => d.Id == id, cancellationToken); - } - - public async Task> GetAllAsync(CancellationToken cancellationToken = default) - { - return await _context.Documents.OrderByDescending(d => d.CreatedAt).ToListAsync(cancellationToken); - } - - public async Task TryUpdateStatusAsync(Guid id, DocumentStatus fromStatus, DocumentStatus toStatus, string? errorMessage = null, CancellationToken cancellationToken = default) - { - // Atomic optimistic lock using raw SQL to avoid race conditions. - // Uses int status values (matching DocumentStatus enum) — no JOIN needed. - var fromStatusInt = (int)fromStatus; - var toStatusInt = (int)toStatus; - - int rows; - if (toStatus == DocumentStatus.Processing) - { - rows = await _context.Database.ExecuteSqlRawAsync( - "UPDATE documents SET status = {0}, started_at = {1} WHERE id = {2} AND status = {3}", - new object[] { toStatusInt, DateTime.UtcNow, id, fromStatusInt }, - cancellationToken); - } - else - { - rows = await _context.Database.ExecuteSqlRawAsync( - "UPDATE documents SET status = {0}, completed_at = {1}, error_message = {2} WHERE id = {3} AND status = {4}", - new object[] { - toStatusInt, DateTime.UtcNow, - errorMessage ?? (object)DBNull.Value, - id, fromStatusInt - }, - cancellationToken); - } - - return rows > 0; - } - - public async Task UpdateTextAsync(Guid id, string? extractedText, DocumentStatus status, CancellationToken cancellationToken = default) - { - await _context.Database.ExecuteSqlRawAsync( - "UPDATE documents SET extracted_text = {0}, status = {1}, completed_at = {2} WHERE id = {3}", - new object[] { extractedText ?? (object)DBNull.Value, (int)status, DateTime.UtcNow, id }, - cancellationToken); - } - - public async Task> GetOutboxPendingAsync(CancellationToken cancellationToken = default) - { - return await Task.FromResult(new List()); - } - - public Task MarkOutboxProcessedAsync(Guid id, CancellationToken cancellationToken = default) => Task.CompletedTask; - - public async Task MarkMessageProcessedAsync(Guid messageId, Guid documentId, CancellationToken cancellationToken = default) - { - _context.ProcessedMessages.Add(new ProcessedMessage - { - MessageId = messageId, - DocumentId = documentId, - ProcessedAt = DateTime.UtcNow - }); - await _context.SaveChangesAsync(cancellationToken); - } - - public async Task IsMessageProcessedAsync(Guid messageId, CancellationToken cancellationToken = default) - { - return await _context.ProcessedMessages.AnyAsync(p => p.MessageId == messageId, cancellationToken); - } -} \ No newline at end of file From 6c212674bf8374f01bff80c70ceac3fe50ea5cbd Mon Sep 17 00:00:00 2001 From: cherninkiy Date: Mon, 25 May 2026 10:02:40 +0300 Subject: [PATCH 9/9] =?UTF-8?q?fix(db):=20address=20code=20review=20feedba?= =?UTF-8?q?ck=20=E2=80=94=20ExecuteSqlInterpolatedAsync,=20MigrateAsync,?= =?UTF-8?q?=20pin=20EF=20Core=20versions?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - replace ExecuteSqlRawAsync with ExecuteSqlInterpolatedAsync in DocumentRepositoryBase - use NOW() database function instead of DateTime.UtcNow application timestamps - fix CancellationToken not being passed as SQL parameter (ExecuteSqlInterpolatedAsync handles it correctly) - replace EnsureCreated() with MigrateAsync() in ApiGateway Program.cs (per ADR-002) - add IsRelational() guard to support in-memory database in tests - fix config key format in integration tests (colon instead of double underscore) - pin Microsoft.EntityFrameworkCore and Relational to 8.0.27 for deterministic builds --- src/ApiGateway/Program.cs | 17 ++++++++++++++++- .../Repositories/DocumentRepositoryBase.cs | 18 +++++++++--------- src/Shared/Shared.csproj | 4 ++-- tests/IntegrationTests/FullWorkflowTests.cs | 10 +++++----- 4 files changed, 32 insertions(+), 17 deletions(-) diff --git a/src/ApiGateway/Program.cs b/src/ApiGateway/Program.cs index d4beddf..84d0276 100644 --- a/src/ApiGateway/Program.cs +++ b/src/ApiGateway/Program.cs @@ -64,7 +64,22 @@ using (var scope = app.Services.CreateScope()) { var db = scope.ServiceProvider.GetRequiredService(); - db.Database.EnsureCreated(); + if (db.Database.IsRelational()) + { + var pendingMigrations = db.Database.GetPendingMigrations(); + if (pendingMigrations.Any()) + { + db.Database.Migrate(); + } + else + { + db.Database.EnsureCreated(); + } + } + else + { + db.Database.EnsureCreated(); + } } app.MapOpenApi(); diff --git a/src/Shared/Repositories/DocumentRepositoryBase.cs b/src/Shared/Repositories/DocumentRepositoryBase.cs index c2e3137..2e233ff 100644 --- a/src/Shared/Repositories/DocumentRepositoryBase.cs +++ b/src/Shared/Repositories/DocumentRepositoryBase.cs @@ -38,15 +38,15 @@ public virtual async Task TryUpdateStatusAsync( int rows; if (toStatus == DocumentStatus.Processing) { - rows = await Context.Database.ExecuteSqlRawAsync( - "UPDATE documents SET status = {0}, started_at = {1} WHERE id = {2} AND status = {3}", - toStatusInt, DateTime.UtcNow, id, fromStatusInt, ct); + rows = await Context.Database.ExecuteSqlInterpolatedAsync( + $"UPDATE documents SET status = {toStatusInt}, started_at = NOW() WHERE id = {id} AND status = {fromStatusInt}", + ct); } else { - rows = await Context.Database.ExecuteSqlRawAsync( - "UPDATE documents SET status = {0}, completed_at = {1}, error_message = {2} WHERE id = {3} AND status = {4}", - toStatusInt, DateTime.UtcNow, errorMessage!, id, fromStatusInt, ct); + rows = await Context.Database.ExecuteSqlInterpolatedAsync( + $"UPDATE documents SET status = {toStatusInt}, completed_at = NOW(), error_message = {errorMessage} WHERE id = {id} AND status = {fromStatusInt}", + ct); } return rows > 0; @@ -54,9 +54,9 @@ public virtual async Task TryUpdateStatusAsync( public virtual async Task UpdateTextAsync(Guid id, string? extractedText, DocumentStatus status, CancellationToken ct = default) { - await Context.Database.ExecuteSqlRawAsync( - "UPDATE documents SET extracted_text = {0}, status = {1}, completed_at = {2} WHERE id = {3}", - extractedText!, (int)status, DateTime.UtcNow, id, ct); + await Context.Database.ExecuteSqlInterpolatedAsync( + $"UPDATE documents SET extracted_text = {extractedText}, status = {(int)status}, completed_at = NOW() WHERE id = {id}", + ct); } public virtual async Task MarkMessageProcessedAsync(Guid messageId, Guid documentId, CancellationToken ct = default) diff --git a/src/Shared/Shared.csproj b/src/Shared/Shared.csproj index 7be5969..fa69e34 100644 --- a/src/Shared/Shared.csproj +++ b/src/Shared/Shared.csproj @@ -9,8 +9,8 @@ - - + + \ No newline at end of file diff --git a/tests/IntegrationTests/FullWorkflowTests.cs b/tests/IntegrationTests/FullWorkflowTests.cs index 4788466..6235394 100644 --- a/tests/IntegrationTests/FullWorkflowTests.cs +++ b/tests/IntegrationTests/FullWorkflowTests.cs @@ -46,11 +46,11 @@ public async Task InitializeAsync() _factory = new WebApplicationFactory() .WithWebHostBuilder(builder => { - builder.UseSetting("ConnectionStrings__DefaultConnection", _postgres.GetConnectionString()); - builder.UseSetting("RabbitMq__Host", _rabbitMq.Hostname); - builder.UseSetting("RabbitMq__Username", "guest"); - builder.UseSetting("RabbitMq__Password", "guest"); - builder.UseSetting("Storage__LocalPath", Path.GetTempPath()); + builder.UseSetting("ConnectionStrings:DefaultConnection", _postgres.GetConnectionString()); + builder.UseSetting("RabbitMq:Host", _rabbitMq.Hostname); + builder.UseSetting("RabbitMq:Username", "guest"); + builder.UseSetting("RabbitMq:Password", "guest"); + builder.UseSetting("Storage:LocalPath", Path.GetTempPath()); }); _client = _factory.CreateClient();