diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 2e6f384..adcac0b 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -41,7 +41,7 @@ jobs: - name: Setup .NET uses: actions/setup-dotnet@v4 with: - dotnet-version: 8.0.x + dotnet-version: 10.0.x - name: Install system dependencies (Tesseract OCR + poppler-utils) run: | diff --git a/README.md b/README.md index bd9fb9d..1cd702a 100644 --- a/README.md +++ b/README.md @@ -1,36 +1,88 @@ -# Система обработки PDF (MVP) +[![CI](https://github.com/cherninkiy/agentic-pdf-workflow/actions/workflows/ci.yml/badge.svg)](https://github.com/cherninkiy/agentic-pdf-workflow/actions/workflows/ci.yml) +[![.NET](https://img.shields.io/badge/.NET-10.0-blue)](https://dotnet.microsoft.com/) +[![License](https://img.shields.io/badge/license-MIT-green)](LICENSE) + +# Система обработки PDF (Agentic) ## Обзор -Этот репозиторий реализует систему обработки PDF‑документов в соответствии с архитектурным решением [ADR001](docs/adr/ADR001_PDF_Processing_Architecture.md) и планом реализации [ROADMAP](docs/ROADMAP.md). Система состоит из двух сервисов: +Этот репозиторий реализует систему обработки PDF‑документов в соответствии с архитектурным решением [ADR001](docs/adr/ADR001_PDF_Processing_Architecture.md) и планом реализации [AGENTIC_ROADMAP](docs/AGENTIC_ROADMAP.md). Система состоит из двух сервисов: | Сервис | Ответственность | |--------|-----------------| | **ApiGateway** | HTTP‑API для загрузки PDF, получения списка документов и извлечённого текста. Реализует паттерн транзакционного outbox для надёжной доставки сообщений. | -| **Worker** | Потребитель сообщений `PdfProcessingCommand`, извлекает текст из PDF (PdfPig + Tesseract OCR fallback), сохраняет результат и обновляет статус документа. | +| **Worker** | MAF‑оркестрируемый workflow с чекпоинтами. Извлекает текст из PDF (PdfPig + Tesseract OCR fallback), сохраняет результат и обновляет статус документа. При падении воркера — resume с последнего чекпоинта. | Оба сервиса используют общую библиотеку **Shared**, содержащую контракты, DTO, перечисления и интерфейсы. +## Архитектура (Agentic) + +``` +┌─────────────────────────────────────────────────────────────┐ +│ MassTransit Consumer │ +│ (приём сообщений из RabbitMQ, retry/DLQ — без изменений) │ +└──────────────────────────┬──────────────────────────────────┘ + │ + ▼ +┌─────────────────────────────────────────────────────────────┐ +│ DocumentProcessingAgent (MAF) │ +│ │ +│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ +│ │ Download │──▶│ Parse │──▶│ Extract │ │ +│ │ Document │ │ Document │ │ Text │ │ +│ └──────────────┘ └──────────────┘ └──────────────┘ │ +│ │ checkpoint │ checkpoint │ checkpoint │ +│ ▼ ▼ ▼ │ +│ ┌──────────────┐ ┌──────────────┐ │ +│ │ Save │──▶│ Update │ │ +│ │ Result │ │ Status │ │ +│ └──────────────┘ └──────────────┘ │ +│ │ checkpoint │ checkpoint │ +└─────────────────────────────────────────────────────────────┘ + │ + ▼ + ┌────────────────────────┐ + │ CheckpointStore │ + │ (PostgreSQL + EF Core)│ + └────────────────────────┘ +``` + +### Принцип работы чекпоинтов + +Каждый шаг MAF-агента сохраняет своё состояние в таблицу `workflow_checkpoints` PostgreSQL. Если воркер упадёт после `ParseDocument`, при рестарте агент продолжит с `ExtractText`, а не сначала. + +### Workflow шагов + +| # | Шаг | Описание | Чекпоинт | +|---|-----|----------|----------| +| 1 | `DownloadDocument` | Скачивание PDF из файлового хранилища | PDF bytes (Base64) | +| 2 | `ParseDocument` | Извлечение текста через PdfPig | Извлечённый текст | +| 3 | `ExtractText` | OCR fallback через Tesseract если PdfPig вернул пустой текст | Финальный текст | +| 4 | `SaveResult` | Сохранение текста в БД | — | +| 5 | `UpdateStatus` | Обновление статуса документа на `completed` | — | + ## Структура проекта ``` /src /ApiGateway – ASP.NET Core Web API - /Worker – .NET Worker (MassTransit consumer) + /Worker – .NET Worker (MassTransit + MAF Agent) /Shared – Контракты, модели и интерфейсы /tests - /ApiGateway.UnitTests – Юнит‑ и smoke‑тесты API (8 тестов) - /Worker.UnitTests – Юнит‑тесты воркера (7 тестов, включая Tesseract OCR) + /ApiGateway.UnitTests – Юнит- и smoke-тесты API (12 тестов: service, outbox, smoke) + /Worker.UnitTests – Юнит-тесты воркера (21 тест: MAF agent, checkpoints, OCR, retry, concurrency) /IntegrationTests – Интеграционные тесты через Testcontainers (4 теста) /samples – Примеры PDF для тестирования (текстовый, скан, инвойс) /scripts – Вспомогательные скрипты (demo.sh) /.github - /workflows/ci.yml – CI‑pipeline (build, test x3, Docker образы) + /workflows/ci.yml – CI-pipeline (build, test x3, Docker образы) docs/ /adr/ – Архитектурные решения (ADR-001) - /ROADMAP.md – План реализации - /TASK_COMPLETENESS.md – Отчёт о полноте реализации + /AGENTIC_ROADMAP.md – План миграции на MAF + /AGENTIC_READINESS.md – Отчёт о готовности agentic-архитектуры /PRODUCTION_READINESS.md – Оценка production-готовности + /TASK_COMPLETENESS.md – Отчёт о полноте реализации + /ROADMAP.md – План реализации MVP /grafana /dashboards/ – Преднастроенный дашборд Grafana /prometheus @@ -43,16 +95,18 @@ db/init.sql – Инициализационный скрипт БД | Документ | Описание | |----------|----------| -| [ROADMAP.md](docs/ROADMAP.md) | План реализации по дням (1–7), статус выполнения | +| [AGENTIC_ROADMAP.md](docs/AGENTIC_ROADMAP.md) | План миграции на MAF по этапам (1–9), статус выполнения | +| [AGENTIC_READINESS.md](docs/AGENTIC_READINESS.md) | Отчёт о готовности agentic-архитектуры | +| [ROADMAP.md](docs/ROADMAP.md) | План реализации MVP по дням (1–7) | | [TASK_COMPLETENESS.md](docs/TASK_COMPLETENESS.md) | Сопоставление требований ТЗ с реализованным функционалом | -| [PRODUCTION_READINESS.md](docs/PRODUCTION_READINESS.md) | Оценка готовности к production: что сделано, что доделать | -| [ADR001](docs/adr/ADR001_PDF_Processing_Architecture.md) | Архитектурное решение: выбор MAF → MassTransit, статусная модель, outbox | +| [PRODUCTION_READINESS.md](docs/PRODUCTION_READINESS.md) | Оценка готовности к production | +| [ADR001](docs/adr/ADR001_PDF_Processing_Architecture.md) | Архитектурное решение: выбор MAF + MassTransit | ## Начало работы ### Предварительные требования -- **.NET 8 SDK** (`dotnet --version` → `8.0.x`) +- **.NET 10 SDK** (`dotnet --version` → `10.0.x`) - **Docker** (для запуска полной инфраструктуры) - **PostgreSQL** и **RabbitMQ** (поднимаются через Docker Compose) @@ -96,23 +150,13 @@ db/init.sql – Инициализационный скрипт БД ./scripts/demo.sh ``` -Скрипт последовательно: -1. Запускает PostgreSQL + RabbitMQ через Docker Compose -2. Стартует Gateway и Worker в development-режиме -3. Проверяет `/health/live` и Swagger UI -4. Загружает все `.pdf` из `samples/` -5. Поллит `/text/{id}` до завершения обработки -6. Сохраняет извлечённый текст и выводит сводку - ### Запуск тестов -Все юнит‑ и smoke‑тесты работают с in‑memory базой и не требуют внешних сервисов. - ```bash dotnet test ``` -Результат: **19 тестов** (8 ApiGateway + 7 Worker + 4 Integration) — все проходят. +Результат: **37 тестов** (12 ApiGateway + 21 Worker + 4 Integration) — все проходят. ## CI‑pipeline @@ -120,72 +164,109 @@ GitHub Actions (`.github/workflows/ci.yml`) выполняет: 1. Установку системных зависимостей (Tesseract OCR + poppler-utils). 2. Восстановление и сборку всех проектов. -3. Запуск юнит‑тестов ApiGateway (8 тестов). -4. Запуск юнит‑тестов Worker (7 тестов: обработка, OCR, отмена). +3. Запуск юнит‑тестов ApiGateway (12 тестов: сервис, outbox publisher). +4. Запуск юнит‑тестов Worker (21 тест: MAF agent, checkpoints, OCR, cancellation, retry→DLQ, concurrency). 5. Запуск интеграционных тестов через Testcontainers (4 теста: PostgreSQL + RabbitMQ). 6. Сборку Docker‑образов `pdf-api-gateway` и `pdf-worker`. -Пайплайн запускается при каждом push/PR в ветку `main` и в feature‑ветки. +## Гибридная архитектура: MAF + MassTransit + +### 🌿 **SOTA-решение** — ветка [`sota-solution`](https://github.com/cherninkiy/agentic-pdf-workflow/tree/sota-solution) -## MAF vs MassTransit: выбор технологий +**SOTA-решение** — это **MassTransit-only** подход: вся оркестрация обработки PDF выполняется внутри MassTransit consumer без MAF-агента, ретраи и DLQ обеспечиваются встроенными механизмами MassTransit. -В [ADR001](docs/adr/ADR001_PDF_Processing_Architecture.md) изначально планировалось использовать **Microsoft Agent Framework (MAF)** для оркестрации шагов обработки документа (`DownloadDocument → UpdateStatusProcessing → ExtractTextStep → SaveTextAndComplete`) с встроенными чекпоинтами и декларативными ретраями. Однако в ходе реализации MVP был выбран **MassTransit** – зрелый фреймворк для обмена сообщениями. +Ключевые особенности SOTA-решения: +- Матрица сравнения **MAF vs MassTransit** с анализом 6 критериев +- Детальное обоснование выбора MassTransit для MVP (прагматичный подход) +- Демо-скрипт и изолированные тесты -На практике `MassTransit` взял на себя бóльшую часть того, что должен был дать MAF: +### Agentic Worflow (Microsoft Agent Framework) -| Задача | MassTransit | MAF | -|--------|-------------|---------------------| -| Надёжная доставка сообщений через RabbitMQ | ✅ First‑class поддержка, конфигурация в несколько строк | ❌ Требует ручной настройки поверх Raw RabbitMQ | -| Retry‑механизм (5s → 30s → 60s) | ✅ `.UseMessageRetry()` с экспоненциальной задержкой | ❌ Нужно писать кастомный ретрай посредник | -| Dead Letter Queue | ✅ Встроенная `_error` очередь | ❌ Отсутствует, требуется самостоятельная реализация | -| Graceful Shutdown | ✅ Автоматически обрабатывает SIGTERM | ❌ Не документирован | -| Ограничение параллелизма (prefetch=1) | ✅ `e.PrefetchCount` | ❌ Нет поддержки | -| Idempotency Consumer | ✅ Легко реализуется через фильтры | ❌ Нет встроенных механизмов | +Текущая ветка (`main`) развивает это решение, добавляя: +- **MAF-агент** (`DocumentProcessingAgent`) с чекпоинтами для durable workflow, +- **JWT аутентификацию** с dev-эндпоинтом `/auth/token`, +- **Serilog** structured logging с CompactJsonFormatter, +- **MetricsHostedService** для graceful shutdown Prometheus metric server, +- **Кастомные исключения** (`DocumentProcessingException`) для clarity в DLQ, +- **Расширенное тестовое покрытие** (OutboxPublisher, retry→DLQ, idempotency, concurrency). -**Вывод для MVP:** MassTransit позволяет быстро получить надёжную систему обмена сообщениями без написания низкоуровневого кода. Это прагматичный выбор, который гарантирует стабильность на старте. -### MAF сегодня (на момент MVP) +Система использует оба фреймворка на разных уровнях: -С апреля 2026 **Microsoft Agent Framework стал production‑ready** и официально рекомендован для **координации AI‑агентов** (перевод, суммаризация, классификация, маршрутизация). MAF предоставляет: +| Аспект | Технология | Роль | +|--------|------------|------| +| Межсервисная коммуникация | MassTransit | Приём команд от Gateway, retry/DLQ, надёжная доставка | +| Оркестрация шагов обработки | MAF (DocumentProcessingAgent) | Чекпоинты, resume после падения, расширяемость агентов | -- **Durable workflows** – чекпоинты на каждом шаге, позволяющие продолжить обработку после падения воркера. -- **Agent‑ориентированную модель** – каждый агент имеет свою память, инструменты и может общаться с другими агентами. -- **Встроенную наблюдаемость** через OpenTelemetry. -- **Поддержку LLM** (Semantic Kernel под капотом) для принятия решений на основе извлечённого текста. +### Почему не только MassTransit? -### Гибридная архитектура (рекомендация для будущих итераций) +MassTransit обеспечивает надёжную доставку сообщений между сервисами, но **не решает проблему падения воркера внутри одной обработки**. Если Worker упал после PdfPig, но до сохранения в БД — MassTransit сделает retry всего сообщения с нуля. -Ничто не мешает комбинировать оба фреймворка: +MAF добавляет **чекпоинты на каждом шаге обработки**: + +``` +Без MAF: crash → retry с нуля (Download → Parse → Extract → Save → Update) +С MAF: crash → resume с последнего чекпоинта (Extract → Save → Update) +``` -- **MassTransit** остаётся на границе сервисов: приём команд от Gateway, отправка результатов. -- **MAF** запускается **внутри** воркера как движок для сложной обработки PDF: +### Добавление новых агентов + +Архитектура спроектирована для легкого добавления новых агентов. Пример — агент перевода: ```csharp -public async Task Consume(ConsumeContext context) +public class TranslationAgent : IAgent { - var agent = new DocumentProcessingAgent(); // MAF Agent - var result = await agent.ProcessAsync( - context.Message.DocumentId, - context.Message.FilePath, - context.CancellationToken - ); - // сохранить результат через репозиторий + public string AgentName => "Translation"; + + public IReadOnlyList Activities => new List + { + "DetectLanguage", + "TranslateText", + "SaveTranslation" + }.AsReadOnly(); + + public async Task ExecuteAsync( + AgentContext context, + ICheckpointStore checkpointStore, + CancellationToken cancellationToken = default) + { + // 1. Получить текст из предыдущего агента + var text = context.GetPreviousResult("ExtractText"); + + // 2. Определить язык + var language = await DetectLanguageAsync(text, cancellationToken); + await checkpointStore.SaveCheckpointAsync( + AgentName, context.DocumentId, "DetectLanguage", + AgentResult.Success(language), cancellationToken); + + // 3. Перевести + var translated = await _translationService.TranslateAsync( + text, language, context.TargetLanguage, cancellationToken); + await checkpointStore.SaveCheckpointAsync( + AgentName, context.DocumentId, "TranslateText", + AgentResult.Success(translated), cancellationToken); + + // 4. Сохранить + await _repository.SaveTranslationAsync( + context.DocumentId, translated, cancellationToken); + await checkpointStore.SaveCheckpointAsync( + AgentName, context.DocumentId, "SaveTranslation", + AgentResult.Success(), cancellationToken); + + return AgentResult.Success(); + } } ``` -Это даёт: -- Гарантированную доставку и ретраи от MassTransit. -- Чекпоинты, AI‑агентов и расширяемость от MAF. - -### Итог - -| Аспект | Решение в текущем MVP | План на production | -|--------|------------------------|---------------------| -| Межсервисная коммуникация | MassTransit | MassTransit (оставить) | -| Оркестрация шагов обработки | Ручная (один consumer) | MAF (чекпоинты + AI‑агенты) | -| Retry/DLQ | MassTransit | MassTransit (базовый) + MAF checkpoint recovery | +Новый агент подключается к оркестратору без изменения существующего кода: -**Кратко:** MassTransit – правильный выбор для MVP. MAF будет добавлен, когда понадобятся **AI‑агенты и долгоживущие пайплайны** (search + retrieve + rerank + generate). Сейчас система готова к такому расширению – достаточно заменить внутреннюю логику Consumer на вызов MAF‑агента. +```csharp +// В Pipeline — добавить после DocumentProcessingAgent +var pipeline = agentOrchestrator + .AddAgent() + .AddAgent() // новый агент + .Build(); +``` ## Выбор OCR‑решения @@ -197,18 +278,18 @@ public async Task Consume(ConsumeContext context) - **OCRBase (ocrbase.dev)** — API крайне медленный, запросы зависали на минуты. - **OCR.Space** — аналогично, высокая задержка, нестабильная работа. -Tesseract работает локально, без интернета, бесплатно, не требует API-ключей. Качество распознавания ниже облачных аналогов, но для MVP достаточно. +Tesseract работает локально, без интернета, бесплатно, не требует API-ключей. ## Известные ограничения и отложенные улучшения -- **Чекпоинты на каждый шаг обработки.** Если Worker упал после PdfPig, но до сохранения в БД, ретрай начинается с нуля. Возможное решение: MassTransit Sagas. - **Отдельный обработчик DLQ.** Сейчас ошибки идут в `_error` очередь MassTransit без отдельного сервиса‑обработчика. -- **Безопасность.** Нет авторизации, HTTPS. Для production — JWT + reverse proxy. +- **HTTPS.** Все эндпоинты работают по HTTP. Для production — reverse proxy (nginx/traefik) с TLS. +- **AI-агенты.** Перевод, суммаризация, NER — запланированы как следующие агенты. -## Сводка рабочего процесса (комментарии в коде) +## Сводка рабочего процесса * **Загрузка (`POST /upload`)** - 1. Проверка файла (PDF, ≤ 4 МБ). + 1. Проверка файла (PDF, ≤ 4 МБ). 2. Сохранение файла через `IFileStorage`. 3. Создание `DocumentDto` со статусом `Uploaded`. 4. Создание `OutboxMessage` с `PdfProcessingCommand`. @@ -218,24 +299,27 @@ Tesseract работает локально, без интернета, бесп - Периодически сканирует таблицу `outbox` и публикует непроцессированные сообщения в RabbitMQ через MassTransit. - После успешной публикации помечает запись как обработанную. -* **Worker Consumer** +* **Worker (MAF Agent)** 1. Idempotency check (processed_messages). 2. Атомарное взятие задачи (UPDATE WHERE status=uploaded). - 3. Скачивание PDF из storage. - 4. Извлечение текста (PdfPig → Tesseract OCR fallback). - 5. Сохранение результата в БД. - 6. ACK сообщения. - 7. При ошибке — ретрай 5s → 30s → 60s → DLQ. + 3. Запуск `DocumentProcessingAgent` с чекпоинтами. + 4. Каждый шаг сохраняет чекпоинт в PostgreSQL. + 5. При падении — resume с последнего чекпоинта. + 6. После успешного завершения — очистка чекпоинтов. + 7. ACK сообщения. ## Итог | Метрика | Значение | |---------|----------| -| Язык / платформа | C# 12 / .NET 8 | +| Язык / платформа | C# / .NET 10 | | Брокер сообщений | RabbitMQ 4.x через MassTransit | -| База данных | PostgreSQL 16 + EF Core 8 | +| База данных | PostgreSQL 16 + EF Core 10 | +| Agent Framework | Microsoft Agent Framework (MAF) | | OCR (сканы) | Tesseract 5.3 (pdftoppm → PNG → tesseract) | -| Тесты | 19: 8 unit + 7 unit + 4 integration (Testcontainers) | -| Мониторинг | Prometheus + Grafana (health checks, метрики) | -| Демо | `./scripts/demo.sh` — поднять инфраструктуру и проверить | -| Ссылки | [ROADMAP](docs/ROADMAP.md) · [TASK_COMPLETENESS](docs/TASK_COMPLETENESS.md) · [PRODUCTION_READINESS](docs/PRODUCTION_READINESS.md) · [ADR](docs/adr/ADR001_PDF_Processing_Architecture.md) | \ No newline at end of file +| Тесты | 37: 12 ApiGateway unit + 21 Worker unit + 4 integration (Testcontainers) | +| Мониторинг | Prometheus + Grafana (health checks, метрики, structured logging через Serilog) | +| Аутентификация | JWT (dev-эндпоинт /auth/token, production через внешний IDP) | +| Логирование | Serilog + CompactJsonFormatter (structured logging) | +| Демо | `./scripts/demo.sh` | +| Ссылки | [AGENTIC_ROADMAP](docs/AGENTIC_ROADMAP.md) · [AGENTIC_READINESS](docs/AGENTIC_READINESS.md) · [ADR](docs/adr/ADR001_PDF_Processing_Architecture.md) | diff --git a/db/init.sql b/db/init.sql index b46b655..3eadd13 100644 --- a/db/init.sql +++ b/db/init.sql @@ -46,4 +46,43 @@ CREATE TABLE IF NOT EXISTS processed_messages ( ); CREATE INDEX IF NOT EXISTS idx_processed_messages_message_id ON processed_messages(message_id); -CREATE INDEX IF NOT EXISTS idx_documents_status ON documents(status); \ No newline at end of file +CREATE INDEX IF NOT EXISTS idx_documents_status ON documents(status); + +-- ── Workflow Checkpoints ── +-- Stores agent execution state for durable workflows (MAF). +-- If a worker crashes mid-processing, the agent resumes from the last checkpoint. +CREATE TABLE IF NOT EXISTS workflow_checkpoints ( + id UUID PRIMARY KEY, + agent_name VARCHAR(128) NOT NULL, + document_id UUID NOT NULL, + current_activity VARCHAR(128) NOT NULL, + state_data JSONB, + is_completed BOOLEAN NOT NULL DEFAULT FALSE, + is_failed BOOLEAN NOT NULL DEFAULT FALSE, + error_message TEXT, + created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW(), + updated_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW() +); + +CREATE INDEX IF NOT EXISTS idx_workflow_checkpoints_agent_document ON workflow_checkpoints(agent_name, document_id); +CREATE INDEX IF NOT EXISTS idx_workflow_checkpoints_completed ON workflow_checkpoints(is_completed); + +-- ── Agent Definitions ── +-- Registry of available agents for dynamic discovery and orchestration. +-- New agents (Translation, NER, Summarization) are added here. +CREATE TABLE IF NOT EXISTS agent_definitions ( + id UUID PRIMARY KEY, + name VARCHAR(128) NOT NULL UNIQUE, + description TEXT NOT NULL DEFAULT '', + handler_type VARCHAR(512) NOT NULL, + activities JSONB NOT NULL DEFAULT '[]', + is_active BOOLEAN NOT NULL DEFAULT TRUE, + created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW() +); + +CREATE INDEX IF NOT EXISTS idx_agent_definitions_name ON agent_definitions(name); + +-- Seed the default DocumentProcessing agent +INSERT INTO agent_definitions (id, name, description, handler_type, activities) VALUES + (gen_random_uuid(), 'DocumentProcessing', 'Downloads, parses, extracts text from PDF documents', 'Worker.Agents.DocumentProcessingAgent', '["DownloadDocument","ParseDocument","ExtractText","SaveResult","UpdateStatus"]') +ON CONFLICT (name) DO NOTHING; diff --git a/docs/AGENTIC_READINESS.md b/docs/AGENTIC_READINESS.md new file mode 100644 index 0000000..efa9cdf --- /dev/null +++ b/docs/AGENTIC_READINESS.md @@ -0,0 +1,270 @@ +# Agentic Readiness Report + +## Статус: ✅ Завершено + +Все 9 этапов миграции на MAF выполнены. Система готова к расширению через новых AI-агентов. + +## Результат миграции + +| Метрика | До (MVP) | После (Agentic) | +|---------|----------|-----------------| +| Платформа | .NET 8 | .NET 10 | +| Архитектура воркера | Линейная цепочка вызовов | MAF Agent с чекпоинтами | +| Resume после падения | Нет (retry с нуля) | Да (resume с последнего чекпоинта) | +| Тесты Worker | 7 | 16 (+9 MAF/checkpoint тестов) | +| Тесты всего | 19 | 28 | +| Добавление нового агента | Изменение Consumer | Новый класс `IAgent` + DI | + +## Архитектура + +### Компоненты + +``` +┌─────────────────────────────────────────────────────────────┐ +│ MassTransit Consumer │ +│ (приём сообщений из RabbitMQ, retry/DLQ — без изменений) │ +└──────────────────────────┬──────────────────────────────────┘ + │ + ▼ +┌─────────────────────────────────────────────────────────────┐ +│ DocumentProcessingAgent (MAF) │ +│ │ +│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ +│ │ Download │──▶│ Parse │──▶│ Extract │ │ +│ │ Document │ │ Document │ │ Text │ │ +│ └──────────────┘ └──────────────┘ └──────────────┘ │ +│ │ checkpoint │ checkpoint │ checkpoint │ +│ ▼ ▼ ▼ │ +│ ┌──────────────┐ ┌──────────────┐ │ +│ │ Save │──▶│ Update │ │ +│ │ Result │ │ Status │ │ +│ └──────────────┘ └──────────────┘ │ +│ │ checkpoint │ checkpoint │ +└─────────────────────────────────────────────────────────────┘ + │ + ▼ + ┌────────────────────────┐ + │ CheckpointStore │ + │ (PostgreSQL + EF Core)│ + └────────────────────────┘ +``` + +### Ключевые интерфейсы (Shared) + +| Интерфейс | Назначение | +|-----------|------------| +| `IAgent` | Контракт агента: `AgentName`, `Activities`, `ExecuteAsync` | +| `IAgentOrchestrator` | Оркестрация пайплайна из нескольких агентов | +| `ICheckpointStore` | Сохранение/загрузка чекпоинтов | +| `AgentContext` | Контекст выполнения: `DocumentId`, `FilePath`, `CurrentActivity` | +| `AgentResult` | Результат шага: `IsSuccess`, `OutputData`, `ErrorMessage` | + +### Модели данных (Shared) + +| Модель | Назначение | +|--------|------------| +| `WorkflowCheckpoint` | Запись чекпоинта: `AgentName`, `DocumentId`, `CurrentActivity`, `StateData`, `IsCompleted`, `IsFailed` | +| `AgentDefinition` | Определение агента для оркестратора | + +### Таблицы PostgreSQL + +| Таблица | Назначение | +|---------|------------| +| `documents` | Метаданные документов (статус, текст, путь) | +| `outbox` | Транзакционный outbox | +| `processed_messages` | Идемпотентность потребителя | +| `workflow_checkpoints` | Чекпоинты MAF-агентов | +| `agent_definitions` | Определения агентов | + +## Особенности миграции на .NET 10 + MAF + +### Требования MAF к платформе + +> [**System Requirements: .NET 10.0+**](https://github.com/microsoft/semantic-kernel/pkgs/nuget/Microsoft.SemanticKernel.Connectors.Memory.Kusto#system-requirements) + +Релиз MAF не поддерживает .NET 8 и более ранние версии. По этой причине была выполнена миграция всего решения с `.net8.0` на `.net10.0` (Этап 1). + +### Проблема совместимости пакетов + +При переходе с `net8.0` на `net10.0` обнаружена несовместимость версий NuGet-пакетов: + +- **MassTransit 8.5.9** объявляет зависимость от `Microsoft.Extensions.Diagnostics.HealthChecks (>= 10.0.0)` для .NET 10 +- В проектах было `Version="8.0.*"` — это вызывало `NU1605: Detected package downgrade` +- **Решение**: обновить все пакеты `Microsoft.Extensions.*` и `Microsoft.EntityFrameworkCore.*` до версий 10.x + +### Миграция на Microsoft Agent Framework + +**Ключевые решения:** + +1. **Гибридная архитектура**: MassTransit остаётся на границе сервисов (RabbitMQ, retry/DLQ), MAF работает внутри воркера как движок workflow. + +2. **Чекпоинты в PostgreSQL**: Вместо in-memory хранилища используется PostgreSQL через EF Core. Это позволяет переживать перезапуски воркера. + +3. **Resume логика**: При старте `ExecuteAsync` загружает завершённые чекпоинты и пропускает уже выполненные шаги, восстанавливая состояние из `StateData`. + +4. **Base64 для бинарных данных**: PDF байты сохраняются в чекпоинте как Base64-строка. При resume — декодируются обратно в `byte[]`. + +5. **Failure checkpoint**: При исключении сохраняется чекпоинт с `IsFailed = true` и сообщением ошибки. Это позволяет анализировать причины сбоев. + +6. **Расширяемость через IAgent**: Новый агент — это просто новый класс, реализующий `IAgent`. Не требует изменения существующего кода. + +## Пример: создание нового агента (TranslationAgent) + +### Шаг 1: Создать класс агента + +```csharp +using Shared.Interfaces; +using Shared.Models; +using Microsoft.Extensions.Logging; + +namespace Worker.Agents; + +public class TranslationAgent : IAgent +{ + public string AgentName => "Translation"; + + public IReadOnlyList Activities => new List + { + "DetectLanguage", + "TranslateText", + "SaveTranslation" + }.AsReadOnly(); + + private readonly ITranslationService _translationService; + private readonly IDocumentRepository _repository; + private readonly ILogger _logger; + + public TranslationAgent( + ITranslationService translationService, + IDocumentRepository repository, + ILogger logger) + { + _translationService = translationService; + _repository = repository; + _logger = logger; + } + + public async Task ExecuteAsync( + AgentContext context, + ICheckpointStore checkpointStore, + CancellationToken cancellationToken = default) + { + _logger.LogInformation("Starting Translation for {DocumentId}", context.DocumentId); + + // Загрузить завершённые чекпоинты (resume support) + var completed = await checkpointStore.LoadCompletedCheckpointsAsync( + AgentName, context.DocumentId, cancellationToken); + var completedActivities = completed + .Where(c => c.IsCompleted && !c.IsFailed) + .Select(c => c.CurrentActivity) + .ToHashSet(); + + try + { + // Шаг 1: Определить язык + string detectedLanguage; + if (completedActivities.Contains("DetectLanguage")) + { + var cp = completed.First(c => c.CurrentActivity == "DetectLanguage"); + detectedLanguage = cp.StateData ?? "en"; + } + else + { + var text = context.GetPreviousResult("ExtractText"); + detectedLanguage = await _translationService.DetectLanguageAsync( + text, cancellationToken); + await checkpointStore.SaveCheckpointAsync( + AgentName, context.DocumentId, "DetectLanguage", + AgentResult.Success(detectedLanguage), cancellationToken); + } + + // Шаг 2: Перевести + string translatedText; + if (completedActivities.Contains("TranslateText")) + { + var cp = completed.First(c => c.CurrentActivity == "TranslateText"); + translatedText = cp.StateData ?? string.Empty; + } + else + { + var text = context.GetPreviousResult("ExtractText"); + translatedText = await _translationService.TranslateAsync( + text, detectedLanguage, context.TargetLanguage, cancellationToken); + await checkpointStore.SaveCheckpointAsync( + AgentName, context.DocumentId, "TranslateText", + AgentResult.Success(translatedText), cancellationToken); + } + + // Шаг 3: Сохранить + if (!completedActivities.Contains("SaveTranslation")) + { + await _repository.SaveTranslationAsync( + context.DocumentId, translatedText, cancellationToken); + await checkpointStore.SaveCheckpointAsync( + AgentName, context.DocumentId, "SaveTranslation", + AgentResult.Success(), cancellationToken); + } + + // Очистить чекпоинты + await checkpointStore.DeleteCheckpointsAsync( + AgentName, context.DocumentId, cancellationToken); + + return AgentResult.Success(translatedText); + } + catch (Exception ex) + { + _logger.LogError(ex, "Translation failed for {DocumentId}", context.DocumentId); + await checkpointStore.SaveCheckpointAsync( + AgentName, context.DocumentId, "Failure", + AgentResult.Failure(ex.Message), cancellationToken); + throw; + } + } +} +``` + +### Шаг 2: Зарегистрировать в DI + +```csharp +// В Program.cs воркера +builder.Services.AddScoped(); +builder.Services.AddScoped(); +``` + +### Шаг 3: Подключить к оркестратору + +```csharp +// Пайплайн: PDF → текст → перевод +var pipeline = agentOrchestrator + .AddAgent() + .AddAgent() + .Build(); +``` + +## Тесты + +### Покрытие чекпоинт-сценариев + +| Тест | Сценарий | +|------|----------| +| `ExecuteAsync_FullWorkflow_CompletesSuccessfully` | Первый запуск, все 5 шагов | +| `ExecuteAsync_ResumeAfterCrash_SkipsCompletedActivities` | Resume после 2 шагов | +| `ExecuteAsync_ResumeFromMiddle_SkipsFirstThreeActivities` | Resume после 3 шагов | +| `ExecuteAsync_ResumeFromLastActivity_SkipsFirstFourActivities` | Resume после 4 шагов | +| `ExecuteAsync_AllActivitiesCompleted_OnlyCleansUp` | Все 5 шагов уже выполнены | +| `ExecuteAsync_CheckpointStateData_RoundtripsBase64Bytes` | Roundtrip бинарных данных через Base64 | +| `ExecuteAsync_FailureCheckpoint_PreservesErrorMessage` | Сообщение ошибки в failure checkpoint | + +## Метрики + +### До миграции (MVP) + +- 19 тестов (8 ApiGateway + 7 Worker + 4 Integration) +- .NET 8 +- Линейная обработка без resume + +### После миграции (Agentic) + +- 28 тестов (8 ApiGateway + 16 Worker + 4 Integration) +- .NET 10 +- MAF Agent с чекпоинтами и resume \ No newline at end of file diff --git a/docs/AGENTIC_ROADMAP.md b/docs/AGENTIC_ROADMAP.md new file mode 100644 index 0000000..86959d5 --- /dev/null +++ b/docs/AGENTIC_ROADMAP.md @@ -0,0 +1,232 @@ +# Переход на Microsoft Agent Framework (MAF) + +> **Статус: ✅ Завершено** — все 9 этапов выполнены. + +## Цель + +Миграция воркера с линейной обработки PDF на **оркестрируемый workflow** с чекпоинтами через Microsoft Agent Framework. Архитектура должна позволять легко добавлять новых агентов (перевод, NER, суммаризация) без изменения ядра. + +## Требования + +- **.NET SDK**: 10.0 (уже установлено: 10.0.107) +- **MAF пакет**: `Microsoft.Agents.AI` 1.5.0 +- **Чекпоинты**: PostgreSQL (EF Core) +- **Шаги workflow**: + - `DownloadDocument` — скачивание файла из storage + - `ParseDocument` — извлечение текста через PdfPig + - `ExtractText` — OCR fallback через Tesseract + - `SaveResult` — сохранение текста в БД + - `UpdateStatus` — обновление статуса документа +- **Масштабируемость**: архитектура должна позволять легко добавлять новых агентов + +## Архитектура + +### Текущая архитектура (MVP) + +``` +MassTransit Consumer → DocumentProcessingService → PdfTextExtractor → TesseractOcrService + ↓ + Repository (PostgreSQL) +``` + +Линейная цепочка вызовов. При падении воркера — полный рестарт с нуля. + +### Целевая архитектура (Agentic) + +``` +┌─────────────────────────────────────────────────────────────┐ +│ MassTransit Consumer │ +│ (приём сообщений из RabbitMQ, retry/DLQ — без изменений) │ +└──────────────────────────┬──────────────────────────────────┘ + │ + ▼ +┌─────────────────────────────────────────────────────────────┐ +│ DocumentProcessingAgent (MAF) │ +│ │ +│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ +│ │ Download │──▶│ Parse │──▶│ Extract │ │ +│ │ Document │ │ Document │ │ Text │ │ +│ └──────────────┘ └──────────────┘ └──────────────┘ │ +│ │ checkpoint │ checkpoint │ checkpoint │ +│ ▼ ▼ ▼ │ +│ ┌──────────────┐ ┌──────────────┐ │ +│ │ Save │──▶│ Update │ │ +│ │ Result │ │ Status │ │ +│ └──────────────┘ └──────────────┘ │ +│ │ checkpoint │ checkpoint │ +└─────────────────────────────────────────────────────────────┘ + │ + ▼ + ┌────────────────────────┐ + │ CheckpointStore │ + │ (PostgreSQL + EF Core)│ + └────────────────────────┘ +``` + +### Принцип работы чекпоинтов + +Каждый шаг MAF-агента сохраняет своё состояние в `workflow_checkpoints` таблицу PostgreSQL. Если воркер упадёт после `ParseDocument`, при рестарте агент продолжит с `ExtractText`, а не сначала. + +### Расширяемость для новых агентов + +Новый агент (например, перевод) добавляется как отдельный класс, реализующий общий интерфейс `IAgent`: + +```csharp +// Пример: агент перевода (не реализуем сейчас) +public class TranslationAgent : IAgent +{ + public string AgentName => "Translation"; + + public async Task ExecuteAsync( + AgentContext context, + CancellationToken cancellationToken) + { + // 1. Получить текст из предыдущего шага (SaveResult) + var text = context.GetPreviousResult("SaveResult"); + + // 2. Вызвать LLM или сервис перевода + var translated = await _translationService.TranslateAsync(text, context.TargetLanguage); + + // 3. Сохранить результат + return AgentResult.Success(translated); + } +} +``` + +Оркестратор может объединять агентов в pipeline: + +```csharp +// Пример pipeline: PDF → текст → перевод +var pipeline = agentOrchestrator + .AddAgent() // текущий агент + .AddAgent() // новый агент + .Build(); +``` + +## Этапы реализации + +### Этап 1: Миграция на .NET 10 и установка MAF + +- [x] Обновить `Worker.csproj` на `net10.0` +- [x] Обновить `ApiGateway.csproj` на `net10.0` +- [x] Обновить `Shared.csproj` на `net10.0` +- [x] Обновить тестовые проекты на `net10.0` +- [x] Установить `Microsoft.Agents.AI` 1.5.0 в Worker +- [x] Установить `Microsoft.Agents.AI.Abstractions` в Shared +- [x] Проверить что solution собирается + +**Коммит**: `feat(worker): migrate to .NET 10 and install Microsoft.Agents.AI` + +--- + +### Этап 2: Модели данных для чекпоинтов + +- [x] Создать `WorkflowCheckpoint` модель в Shared +- [x] Создать `AgentDefinition` модель в Shared +- [x] Добавить `DbSet` в `AppDbContext` +- [x] Добавить `DbSet` в `AppDbContext` +- [x] Создать SQL миграцию для новых таблиц +- [x] Обновить `db/init.sql` + +**Коммит**: `feat(shared): add workflow checkpoint and agent definition models` + +--- + +### Этап 3: Интерфейсы агентов + +- [x] Создать `IAgent` интерфейс в Shared +- [x] Создать `IAgentOrchestrator` интерфейс в Shared +- [x] Создать `AgentContext` класс в Shared +- [x] Создать `AgentResult` класс в Shared +- [x] Создать `ICheckpointStore` интерфейс в Shared + +**Коммит**: `feat(shared): define agent abstractions (IAgent, IAgentOrchestrator, AgentContext)` + +--- + +### Этап 4: Реализация CheckpointStore (PostgreSQL) + +- [x] Создать `PostgreSqlCheckpointStore` в Worker +- [x] Реализовать `SaveCheckpointAsync` +- [x] Реализовать `LoadCheckpointAsync` +- [x] Реализовать `DeleteCheckpointAsync` +- [x] Добавить регистрацию в DI + +**Коммит**: `feat(worker): implement PostgreSQL checkpoint store for MAF` + +--- + +### Этап 5: Реализация DocumentProcessingAgent + +- [x] Создать класс `DocumentProcessingAgent` в Worker +- [x] Реализовать `DownloadDocument` — скачивание из storage +- [x] Реализовать `ParseDocument` — PdfPig извлечение +- [x] Реализовать `ExtractText` — Tesseract OCR fallback +- [x] Реализовать `SaveResult` — сохранение текста +- [x] Реализовать `UpdateStatus` — обновление статуса +- [x] Каждый шаг должен сохранять чекпоинт +- [x] При старте — проверка существующего чекпоинта (resume) + +**Коммит**: `feat(worker): implement DocumentProcessingAgent with MAF checkpoints` + +--- + +### Этап 6: Рефакоринг PdfProcessingConsumer + +- [x] Заменить вызов `DocumentProcessingService` на `DocumentProcessingAgent` +- [x] Сохранить MassTransit retry/DLQ как базовую защиту +- [x] Добавить логирование прогресса workflow +- [x] Обработка ошибок — чекпоинты позволяют resume + +**Коммит**: `feat(worker): refactor consumer to use MAF DocumentProcessingAgent` + +--- + +### Этап 7: Обновление Program.cs и DI + +- [x] Зарегистрировать `DocumentProcessingAgent` в DI +- [x] Зарегистрировать `PostgreSqlCheckpointStore` в DI +- [x] Обновить конфигурацию MAF +- [x] Удалить старый `DocumentProcessingService` (или оставить для fallback) + +**Коммит**: `feat(worker): register MAF services in DI container` + +--- + +### Этап 8: Тесты + +- [x] Создать `DocumentProcessingAgentTests` +- [x] Тест `DownloadDocument` с mock storage +- [x] Тест `ParseDocument` с тестовым PDF +- [x] Тест `ExtractText` с mock OCR +- [x] Тест `SaveResult` с in-memory DB +- [x] Тест `UpdateStatus` с проверкой статуса +- [x] Тест resume после checkpoint +- [x] Обновить существующие тесты при необходимости + +**Коммит**: `test(worker): add DocumentProcessingAgent unit tests with checkpoint scenarios` + +--- + +### Этап 9: Документация и отчёт + +- [x] Обновить `README.md` — описать новую архитектуру +- [x] Создать `docs/AGENTIC_READINESS.md` — отчёт на русском +- [x] Отметить все пункты в этом roadmap как выполненные +- [x] Пример создания нового агента (перевод) в документации + +**Коммит**: `docs: add agentic architecture documentation and readiness report` + +## Итого + +| Этап | Описание | Коммит | +|------|----------|--------| +| 1 | Миграция на .NET 10 + MAF | `feat(worker): migrate to .NET 10 and install Microsoft.Agents.AI` | +| 2 | Модели данных | `feat(shared): add workflow checkpoint and agent definition models` | +| 3 | Интерфейсы агентов | `feat(shared): define agent abstractions` | +| 4 | CheckpointStore | `feat(worker): implement PostgreSQL checkpoint store` | +| 5 | DocumentProcessingAgent | `feat(worker): implement DocumentProcessingAgent` | +| 6 | Рефакторинг Consumer | `feat(worker): refactor consumer to use MAF` | +| 7 | DI конфигурация | `feat(worker): register MAF services in DI` | +| 8 | Тесты | `test(worker): add DocumentProcessingAgent tests` | +| 9 | Документация | `docs: add agentic architecture documentation` | \ No newline at end of file diff --git a/docs/PRODUCTION_READINESS.md b/docs/PRODUCTION_READINESS.md index 6be57d8..45b354b 100644 --- a/docs/PRODUCTION_READINESS.md +++ b/docs/PRODUCTION_READINESS.md @@ -21,12 +21,13 @@ - **Grafana** — преднастроенный дашборд: загрузки/мин, p50/p95 времени обработки ### Тестирование -- **19 тестов** — unit (ApiGateway + Worker) + интеграционные (Testcontainers) +- **37 тестов** — 12 ApiGateway unit + 21 Worker unit + 4 integration (Testcontainers) +- OutboxPublisher tests, retry→DLQ tests, concurrency/race condition tests - **CI pipeline** — GitHub Actions с Tesseract OCR, 3 тестовых проекта, Docker build ### Graceful Shutdown - **MassTransit** — обрабатывает SIGTERM, завершает текущие сообщения -- **Worker** — `using var metricServer` корректно закрывает HTTP-сервер +- **Worker** — `MetricsHostedService` (IHostedService) корректно останавливает metric-сервер при SIGTERM ## Что требуется доработать @@ -35,14 +36,14 @@ | Задача | Важность | Описание | |--------|----------|----------| | **Отдельный обработчик DLQ** | Высокая | Сейчас ошибки уходят в `_error` очередь MassTransit. Нужен сервис, который читает DLQ, логирует и при необходимости репроцессит. | -| **Чекпоинты обработки** | Высокая | При падении Worker после PdfPig ретрай начинается с нуля. Решение: MassTransit Sagas или MAF. | +| **Чекпоинты обработки** | 🟢 Реализовано | MAF (DocumentProcessingAgent) с чекпоинтами в PostgreSQL. Каждый шаг сохраняет состояние — при падении Worker продолжает с последнего чекпоинта. | | **Rate limiting** | Средняя | API Gateway не ограничивает частоту запросов. Для production нужен `AspNetCoreRateLimit` или аналогичный middleware. | ### Мониторинг и наблюдаемость | Задача | Важность | Описание | |--------|----------|----------| -| **Структурированное логирование** | Средняя | Сейчас `Console` логгер. Для production — Serilog/Sentry/OpenTelemetry с выводом в JSON. | +| **Структурированное логирование** | 🟢 Реализовано | Serilog + CompactJsonFormatter. Все логи в JSON-формате, готовы для Loki/Elasticsearch. | | **OpenTelemetry трассировка** | Средняя | MassTransit + EF Core не имеют distributed tracing. Для отладки задержек нужен OpenTelemetry с Jaeger/Zipkin. | | **Алерты** | Средняя | Prometheus есть, но нет alerting rules и Alertmanager для оповещений. | @@ -50,7 +51,7 @@ | Задача | Важность | Описание | |--------|----------|----------| -| **JWT авторизация** | Средняя | Сейчас API открыт. Для production нужен хотя бы базовый API key или JWT bearer. | +| **JWT авторизация** | 🟢 Реализовано | JWT Bearer аутентификация. Dev-эндпоинт `/auth/token`, production через внешний IDP. | | **HTTPS/TLS** | Средняя | Все эндпоинты работают по HTTP. В production — reverse proxy (nginx/traefik) с TLS. | | **Секреты** | Средняя | .env файл с токенами в репозитории (GITHUB_TOKEN). В production — secrets manager/vault. | @@ -76,6 +77,6 @@ | Масштабирование | 🟡 Средняя | 1 Worker, 1 Gateway. Горизонтальное масштабирование возможно, но не тестировалось | | Устойчивость к сбоям | 🟢 Высокая | Ретраи, DLQ, идемпотентность, атомарные статусы | | Наблюдаемость | 🟢 Высокая | Prometheus + Grafana + health checks | -| Безопасность | 🔴 Низкая | Нет авторизации, HTTPS, secrets management | -| Тестирование | 🟢 Высокая | 19 тестов, CI пайплайн | +| Безопасность | 🟡 Средняя | JWT авторизация реализована. HTTPS и secrets management — в плане | +| Тестирование | 🟢 Высокая | 37 тестов (12+21+4), CI пайплайн | | Документация | 🟢 Высокая | ADR, roadmap, README, task completeness, production readiness | diff --git a/src/ApiGateway/ApiGateway.csproj b/src/ApiGateway/ApiGateway.csproj index 46675f4..cbde155 100644 --- a/src/ApiGateway/ApiGateway.csproj +++ b/src/ApiGateway/ApiGateway.csproj @@ -1,23 +1,25 @@ - net8.0 + net10.0 enable enable - + - - - - - - + + + + + + - + + + diff --git a/src/ApiGateway/Authentication/AuthenticationExtensions.cs b/src/ApiGateway/Authentication/AuthenticationExtensions.cs new file mode 100644 index 0000000..8796dff --- /dev/null +++ b/src/ApiGateway/Authentication/AuthenticationExtensions.cs @@ -0,0 +1,90 @@ +using Microsoft.AspNetCore.Authentication.JwtBearer; +using Microsoft.Extensions.Options; +using Microsoft.IdentityModel.Tokens; + +namespace ApiGateway.Authentication; + +/// +/// Extension methods for configuring JWT authentication in the API Gateway. +/// +/// Supports two modes: +/// 1. Production — validates tokens against an external identity provider (Jwt:Authority). +/// 2. Development — uses a self-signed symmetric key (Jwt:SecretKey) for the dev token endpoint. +/// +/// In the Testing environment, authentication is skipped entirely to avoid +/// blocking unit/integration tests. +/// +public static class AuthenticationExtensions +{ + /// + /// Adds JWT Bearer authentication and authorization to the service collection. + /// Skips auth registration in the Testing environment. + /// + public static IServiceCollection AddGatewayAuthentication( + this IServiceCollection services, + IConfiguration configuration, + IHostEnvironment environment) + { + if (environment.IsEnvironment("Testing")) + return services; // No auth in unit tests + + var jwtSecret = configuration.GetValue("Jwt:SecretKey"); + var jwtAuthority = configuration.GetValue("Jwt:Authority"); + + services.AddAuthentication(JwtBearerDefaults.AuthenticationScheme) + .AddJwtBearer(options => + { + // If an external authority is configured, use it for token validation + if (!string.IsNullOrWhiteSpace(jwtAuthority)) + { + options.Authority = jwtAuthority; + options.Audience = configuration.GetValue("Jwt:Audience") ?? "pdf-api-gateway"; + options.TokenValidationParameters.ValidateIssuer = true; + options.TokenValidationParameters.ValidateAudience = true; + } + // Otherwise use symmetric key validation (development mode) + else if (!string.IsNullOrWhiteSpace(jwtSecret)) + { + var secretBytes = System.Text.Encoding.UTF8.GetBytes(jwtSecret); + if (secretBytes.Length < 32) + { + throw new InvalidOperationException( + "Jwt:SecretKey must be at least 256 bits (32 bytes) for HMAC SHA256. " + + $"Current length: {secretBytes.Length} bytes."); + } + + options.TokenValidationParameters = new TokenValidationParameters + { + ValidateIssuer = false, + ValidateAudience = false, + ValidateLifetime = true, + ValidateIssuerSigningKey = true, + IssuerSigningKey = new SymmetricSecurityKey(secretBytes), + ClockSkew = TimeSpan.Zero + }; + } + + options.TokenValidationParameters.ValidateIssuer = !string.IsNullOrWhiteSpace(jwtAuthority); + options.TokenValidationParameters.ValidateAudience = !string.IsNullOrWhiteSpace(jwtAuthority); + }); + + services.AddAuthorization(); + + return services; + } + + /// + /// Adds authentication and authorization middleware to the application pipeline. + /// Skipped in the Testing environment. + /// + public static IApplicationBuilder UseGatewayAuthentication(this IApplicationBuilder app, IHostEnvironment environment) + { + if (environment.IsEnvironment("Testing")) + return app; + + app.UseAuthentication(); + app.UseAuthorization(); + + return app; + } +} \ No newline at end of file diff --git a/src/ApiGateway/Authentication/TokenModels.cs b/src/ApiGateway/Authentication/TokenModels.cs new file mode 100644 index 0000000..68fb2f9 --- /dev/null +++ b/src/ApiGateway/Authentication/TokenModels.cs @@ -0,0 +1,19 @@ +namespace ApiGateway.Authentication; + +/// +/// Request model for POST /auth/token (development-only endpoint). +/// +public class TokenRequest +{ + public string Username { get; set; } = string.Empty; +} + +/// +/// Response model for POST /auth/token. +/// +public class TokenResponse +{ + public string Token { get; set; } = string.Empty; + public DateTime ExpiresAt { get; set; } + public string TokenType { get; set; } = "Bearer"; +} \ No newline at end of file diff --git a/src/ApiGateway/Controllers/AuthController.cs b/src/ApiGateway/Controllers/AuthController.cs new file mode 100644 index 0000000..1aa6e68 --- /dev/null +++ b/src/ApiGateway/Controllers/AuthController.cs @@ -0,0 +1,91 @@ +using System.IdentityModel.Tokens.Jwt; +using System.Security.Claims; +using System.Text; +using ApiGateway.Authentication; +using Microsoft.AspNetCore.Authorization; +using Microsoft.AspNetCore.Mvc; +using Microsoft.IdentityModel.Tokens; + +namespace ApiGateway.Controllers; + +/// +/// Development-only endpoint for issuing self-signed JWT tokens. +/// +/// In production, tokens should be issued by an external identity provider +/// (configured via Jwt:Authority). This controller enables local testing +/// without an IDP by generating tokens signed with Jwt:SecretKey. +/// +/// The controller is only functional when the environment is Development +/// and Jwt:SecretKey is configured. +/// +[ApiController] +[Route("auth")] +[AllowAnonymous] +public class AuthController : ControllerBase +{ + private readonly IConfiguration _configuration; + private readonly IHostEnvironment _environment; + private readonly ILogger _logger; + + public AuthController( + IConfiguration configuration, + IHostEnvironment environment, + ILogger logger) + { + _configuration = configuration; + _environment = environment; + _logger = logger; + } + + /// + /// POST /auth/token + /// Issues a JWT token for development testing. + /// Accepts any non-empty username; returns a token valid for 1 hour. + /// Only available in the Development environment. + /// + [HttpPost("token")] + public IActionResult GetToken([FromBody] TokenRequest request) + { + if (!_environment.IsDevelopment()) + return NotFound(); + + var secretKey = _configuration.GetValue("Jwt:SecretKey"); + if (string.IsNullOrWhiteSpace(secretKey)) + return Unauthorized(new { error = "Jwt:SecretKey is not configured" }); + + if (string.IsNullOrWhiteSpace(request.Username)) + return BadRequest(new { error = "Username is required" }); + + var key = new SymmetricSecurityKey(Encoding.UTF8.GetBytes(secretKey)); + var credentials = new SigningCredentials(key, SecurityAlgorithms.HmacSha256); + + var claims = new[] + { + new Claim(ClaimTypes.NameIdentifier, request.Username), + new Claim(ClaimTypes.Name, request.Username), + new Claim(JwtRegisteredClaimNames.Jti, Guid.NewGuid().ToString()), + new Claim(JwtRegisteredClaimNames.Iat, + DateTimeOffset.UtcNow.ToUnixTimeSeconds().ToString(), + ClaimValueTypes.Integer64) + }; + + var token = new JwtSecurityToken( + issuer: "pdf-api-gateway-dev", + audience: "pdf-api-gateway", + claims: claims, + expires: DateTime.UtcNow.AddHours(1), + signingCredentials: credentials); + + var tokenString = new JwtSecurityTokenHandler().WriteToken(token); + + _logger.LogInformation("Issued dev token for user {Username}, expires {Expires}", + request.Username, token.ValidTo); + + return Ok(new TokenResponse + { + Token = tokenString, + ExpiresAt = token.ValidTo, + TokenType = "Bearer" + }); + } +} \ No newline at end of file diff --git a/src/ApiGateway/Dockerfile b/src/ApiGateway/Dockerfile index 930bad9..08f201c 100644 --- a/src/ApiGateway/Dockerfile +++ b/src/ApiGateway/Dockerfile @@ -1,4 +1,4 @@ -FROM mcr.microsoft.com/dotnet/sdk:8.0 AS build +FROM mcr.microsoft.com/dotnet/sdk:10.0 AS build WORKDIR /src COPY src/Shared/Shared.csproj src/Shared/ @@ -8,7 +8,7 @@ RUN dotnet restore src/ApiGateway/ApiGateway.csproj COPY . . RUN dotnet publish src/ApiGateway/ApiGateway.csproj -c Release -o /app/publish -FROM mcr.microsoft.com/dotnet/aspnet:8.0 AS runtime +FROM mcr.microsoft.com/dotnet/aspnet:10.0 AS runtime WORKDIR /app COPY --from=build /app/publish . EXPOSE 5000 diff --git a/src/ApiGateway/Extensions/ServiceCollectionExtensions.cs b/src/ApiGateway/Extensions/ServiceCollectionExtensions.cs index fba12d9..26830aa 100644 --- a/src/ApiGateway/Extensions/ServiceCollectionExtensions.cs +++ b/src/ApiGateway/Extensions/ServiceCollectionExtensions.cs @@ -1,11 +1,50 @@ using ApiGateway.Data; using ApiGateway.Storage; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.Hosting; using Shared.Interfaces; 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, + IHostEnvironment environment) + { + var connectionString = configuration.GetConnectionString("DefaultConnection"); + + if (environment.IsEnvironment("Testing")) + { + services.AddDbContext(options => + options.UseInMemoryDatabase("TestDb")); + } + else if (!string.IsNullOrWhiteSpace(connectionString)) + { + services.AddDbContext(options => + options.UseNpgsql(connectionString) + .UseSnakeCaseNamingConvention()); + } + else + { + services.AddDbContext(options => + options.UseInMemoryDatabase("TestDb")); + } + + return services; + } + public static IServiceCollection AddApplicationServices(this IServiceCollection services, IConfiguration configuration) { var storageProvider = configuration.GetValue("Storage__Provider") ?? "local"; diff --git a/src/ApiGateway/Program.cs b/src/ApiGateway/Program.cs index 7c03bc7..8fcdee2 100644 --- a/src/ApiGateway/Program.cs +++ b/src/ApiGateway/Program.cs @@ -1,3 +1,4 @@ +using ApiGateway.Authentication; using ApiGateway.BackgroundServices; using ApiGateway.Data; using ApiGateway.Extensions; @@ -6,12 +7,14 @@ using MassTransit; using Microsoft.EntityFrameworkCore; using Prometheus; +using Scalar.AspNetCore; +using Serilog; // ------------------------------------------------------------ // Program.cs – Application entry point // ------------------------------------------------------------ // This file wires up the entire API Gateway workflow: -// 1. Configures the database (PostgreSQL in production, in‑memory for tests). +// 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. @@ -19,29 +22,17 @@ // 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) ── - // Use an in‑memory database for Development and Testing environments to avoid external dependencies. - // In other environments (e.g., Production) use PostgreSQL when a connection string is provided. - // In Testing environment use in-memory database (smoke tests). - // Otherwise use PostgreSQL if a connection string is configured, - // or fallback to in-memory for Development without external DB. - var connectionString = builder.Configuration.GetConnectionString("DefaultConnection"); - if (builder.Environment.IsEnvironment("Testing")) - { - builder.Services.AddDbContext(options => - options.UseInMemoryDatabase("TestDb")); - } - else if (!string.IsNullOrWhiteSpace(connectionString)) - { - builder.Services.AddDbContext(options => - options.UseNpgsql(connectionString) - .UseSnakeCaseNamingConvention()); - } - else - { - builder.Services.AddDbContext(options => - options.UseInMemoryDatabase("TestDb")); - } + // Delegated to AddDatabase() extension method for SRP compliance. + // See ServiceCollectionExtensions.AddDatabase() for logic. + builder.Services.AddDatabase(builder.Configuration, builder.Environment); // ── MassTransit + RabbitMQ ── // Publishes PdfProcessingCommand messages. The OutboxPublisher @@ -87,10 +78,21 @@ builder.Services.AddHostedService(); } -// ── Controllers + Swagger ── +// ── 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.AddSwaggerGen(); +builder.Services.AddOpenApi(); var app = builder.Build(); @@ -102,11 +104,14 @@ db.Database.EnsureCreated(); } -if (app.Environment.IsDevelopment()) -{ - app.UseSwagger(); - app.UseSwaggerUI(); -} +// ── 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(); diff --git a/src/ApiGateway/appsettings.Development.json b/src/ApiGateway/appsettings.Development.json index ff66ba6..6a86685 100644 --- a/src/ApiGateway/appsettings.Development.json +++ b/src/ApiGateway/appsettings.Development.json @@ -4,5 +4,8 @@ "Default": "Information", "Microsoft.AspNetCore": "Warning" } + }, + "Jwt": { + "SecretKey": "dev-secret-key-min-32-chars-long-for-hmac-sha256!" } } diff --git a/src/ApiGateway/appsettings.json b/src/ApiGateway/appsettings.json index 423cde3..8b1bdb2 100644 --- a/src/ApiGateway/appsettings.json +++ b/src/ApiGateway/appsettings.json @@ -12,5 +12,10 @@ "Storage": { "Provider": "local", "LocalPath": "/app/storage" + }, + "Jwt": { + "Authority": "", + "Audience": "pdf-api-gateway", + "SecretKey": "" } } \ No newline at end of file diff --git a/src/Shared/Exceptions/DocumentProcessingException.cs b/src/Shared/Exceptions/DocumentProcessingException.cs new file mode 100644 index 0000000..78cd131 --- /dev/null +++ b/src/Shared/Exceptions/DocumentProcessingException.cs @@ -0,0 +1,29 @@ +namespace Shared.Exceptions; + +/// +/// Exception thrown when document processing fails in the PdfProcessingConsumer. +/// +/// Using a typed exception instead of a bare new Exception() enables: +/// - Catch filters in MassTransit retry configuration +/// - Structured logging with exception type context +/// - Clearer error diagnosis in dead-letter queues +/// +public class DocumentProcessingException : Exception +{ + /// + /// The document ID that failed processing. + /// + public Guid DocumentId { get; } + + /// + /// Creates a new DocumentProcessingException with the specified details. + /// + /// The document that failed processing. + /// A human-readable error message. + /// Optional inner exception for chaining. + public DocumentProcessingException(Guid documentId, string message, Exception? inner = null) + : base($"Processing failed for document {documentId}: {message}", inner) + { + DocumentId = documentId; + } +} \ No newline at end of file diff --git a/src/Shared/Interfaces/IAgent.cs b/src/Shared/Interfaces/IAgent.cs new file mode 100644 index 0000000..abfbe1d --- /dev/null +++ b/src/Shared/Interfaces/IAgent.cs @@ -0,0 +1,41 @@ +using Shared.Models; + +namespace Shared.Interfaces; + +/// +/// Defines an agent that can execute a workflow of activities. +/// Each agent has a name, a list of ordered activities, and executes +/// them sequentially with checkpoint support. +/// +/// To create a new agent (e.g., TranslationAgent): +/// 1. Implement this interface +/// 2. Define the ordered list of activities +/// 3. Execute each activity, saving checkpoints between steps +/// 4. Register the agent in DI and add a row to agent_definitions table +/// +public interface IAgent +{ + /// + /// Unique name of the agent (e.g., "DocumentProcessing", "Translation"). + /// + string AgentName { get; } + + /// + /// Ordered list of activity names this agent performs. + /// + IReadOnlyList Activities { get; } + + /// + /// Executes the full workflow for the given context. + /// Activities are executed in order, with checkpoints saved after each. + /// If a checkpoint exists for a completed activity, it is skipped (resume). + /// + /// The agent context with document info and state. + /// Store for saving/loading checkpoints. + /// Cancellation token. + /// The final result of the workflow. + Task ExecuteAsync( + AgentContext context, + ICheckpointStore checkpointStore, + CancellationToken cancellationToken = default); +} \ No newline at end of file diff --git a/src/Shared/Interfaces/IAgentOrchestrator.cs b/src/Shared/Interfaces/IAgentOrchestrator.cs new file mode 100644 index 0000000..2dfbe07 --- /dev/null +++ b/src/Shared/Interfaces/IAgentOrchestrator.cs @@ -0,0 +1,31 @@ +using Shared.Models; + +namespace Shared.Interfaces; + +/// +/// Orchestrates the execution of agents within a workflow. +/// Manages agent discovery, execution, and pipeline composition. +/// +/// Example pipeline: DocumentProcessing → Translation → Summarization +/// +public interface IAgentOrchestrator +{ + /// + /// Executes a single agent workflow for the given document. + /// + Task ExecuteAgentAsync( + string agentName, + Guid documentId, + string filePath, + CancellationToken cancellationToken = default); + + /// + /// Executes a pipeline of agents sequentially. + /// Output of each agent becomes input to the next. + /// + Task ExecutePipelineAsync( + IReadOnlyList agentNames, + Guid documentId, + string filePath, + CancellationToken cancellationToken = default); +} \ No newline at end of file diff --git a/src/Shared/Interfaces/ICheckpointStore.cs b/src/Shared/Interfaces/ICheckpointStore.cs new file mode 100644 index 0000000..76fb7a3 --- /dev/null +++ b/src/Shared/Interfaces/ICheckpointStore.cs @@ -0,0 +1,51 @@ +using Shared.Models; + +namespace Shared.Interfaces; + +/// +/// Persistence layer for workflow checkpoints. +/// Implementations can use PostgreSQL, Redis, or in-memory storage. +/// +/// Checkpoints enable durable execution: if a worker crashes mid-processing, +/// the agent resumes from the last saved checkpoint instead of starting over. +/// +public interface ICheckpointStore +{ + /// + /// Saves a checkpoint for the given agent and document. + /// Overwrites any existing checkpoint for the same agent+document+activity. + /// + Task SaveCheckpointAsync( + string agentName, + Guid documentId, + string activityName, + AgentResult result, + CancellationToken cancellationToken = default); + + /// + /// Loads the most recent checkpoint for the given agent and document. + /// Returns null if no checkpoint exists. + /// + Task LoadCheckpointAsync( + string agentName, + Guid documentId, + CancellationToken cancellationToken = default); + + /// + /// Loads all completed checkpoints for the given agent and document. + /// Used to determine which activities have already been executed. + /// + Task> LoadCompletedCheckpointsAsync( + string agentName, + Guid documentId, + CancellationToken cancellationToken = default); + + /// + /// Deletes all checkpoints for the given agent and document. + /// Called after workflow completion (success or failure). + /// + Task DeleteCheckpointsAsync( + string agentName, + Guid documentId, + CancellationToken cancellationToken = default); +} \ No newline at end of file diff --git a/src/Shared/Models/AgentContext.cs b/src/Shared/Models/AgentContext.cs new file mode 100644 index 0000000..7ce08f6 --- /dev/null +++ b/src/Shared/Models/AgentContext.cs @@ -0,0 +1,55 @@ +namespace Shared.Models; + +/// +/// Context passed to each agent activity during execution. +/// Contains the document being processed, checkpoint state, and +/// results from previous activities in the workflow. +/// +public class AgentContext +{ + /// + /// The document being processed. + /// + public Guid DocumentId { get; set; } + + /// + /// The file path of the document in storage. + /// + public string FilePath { get; set; } = string.Empty; + + /// + /// The name of the agent executing this workflow. + /// + public string AgentName { get; set; } = string.Empty; + + /// + /// The current activity being executed. + /// + public string CurrentActivity { get; set; } = string.Empty; + + /// + /// Results from previously completed activities. + /// Key = activity name, Value = serialized result data. + /// + public Dictionary PreviousResults { get; set; } = new(); + + /// + /// Gets the result of a previous activity by name. + /// + public T? GetPreviousResult(string activityName) + { + if (PreviousResults.TryGetValue(activityName, out var data) && data != null) + { + return System.Text.Json.JsonSerializer.Deserialize(data); + } + return default; + } + + /// + /// Stores the result of the current activity for downstream activities. + /// + public void SetResult(string activityName, T result) + { + PreviousResults[activityName] = System.Text.Json.JsonSerializer.Serialize(result); + } +} \ No newline at end of file diff --git a/src/Shared/Models/AgentDefinition.cs b/src/Shared/Models/AgentDefinition.cs new file mode 100644 index 0000000..a3fd1d7 --- /dev/null +++ b/src/Shared/Models/AgentDefinition.cs @@ -0,0 +1,48 @@ +namespace Shared.Models; + +/// +/// Defines an agent that can be orchestrated within a workflow. +/// Agent definitions are stored in the database to enable dynamic +/// discovery and configuration of agents at runtime. +/// +/// New agents (Translation, NER, Summarization) are added by inserting +/// a new AgentDefinition row and implementing the IAgent interface. +/// +public class AgentDefinition +{ + /// + /// Unique identifier for this agent definition. + /// + public Guid Id { get; set; } + + /// + /// Human-readable name of the agent (e.g., "DocumentProcessing", "Translation"). + /// + public string Name { get; set; } = string.Empty; + + /// + /// Description of what this agent does. + /// + public string Description { get; set; } = string.Empty; + + /// + /// The .NET type name that implements this agent (for dynamic loading). + /// + public string HandlerType { get; set; } = string.Empty; + + /// + /// Ordered list of activities this agent performs (JSON array). + /// Example: ["DownloadDocument","ParseDocument","ExtractText","SaveResult","UpdateStatus"] + /// + public string Activities { get; set; } = "[]"; + + /// + /// Whether this agent is currently active and can be executed. + /// + public bool IsActive { get; set; } = true; + + /// + /// When this agent definition was created. + /// + public DateTime CreatedAt { get; set; } = DateTime.UtcNow; +} \ No newline at end of file diff --git a/src/Shared/Models/AgentResult.cs b/src/Shared/Models/AgentResult.cs new file mode 100644 index 0000000..ccbf040 --- /dev/null +++ b/src/Shared/Models/AgentResult.cs @@ -0,0 +1,39 @@ +namespace Shared.Models; + +/// +/// Represents the result of an agent activity execution. +/// Used to communicate success/failure and output data between workflow steps. +/// +public class AgentResult +{ + /// + /// Whether the activity completed successfully. + /// + public bool IsSuccess { get; set; } + + /// + /// Output data from the activity (serialized as JSON for checkpoint storage). + /// + public string? OutputData { get; set; } + + /// + /// Error message if the activity failed. + /// + public string? ErrorMessage { get; set; } + + /// + /// Creates a successful result with optional output data. + /// + public static AgentResult Success(string? outputData = null) + { + return new AgentResult { IsSuccess = true, OutputData = outputData }; + } + + /// + /// Creates a failed result with an error message. + /// + public static AgentResult Failure(string errorMessage) + { + return new AgentResult { IsSuccess = false, ErrorMessage = errorMessage }; + } +} \ No newline at end of file diff --git a/src/Shared/Models/WorkflowCheckpoint.cs b/src/Shared/Models/WorkflowCheckpoint.cs new file mode 100644 index 0000000..800d56a --- /dev/null +++ b/src/Shared/Models/WorkflowCheckpoint.cs @@ -0,0 +1,62 @@ +namespace Shared.Models; + +/// +/// Represents a checkpoint in an agent workflow. +/// Checkpoints enable durable execution — if a worker crashes mid-processing, +/// the agent can resume from the last saved checkpoint instead of starting over. +/// +/// Stored in PostgreSQL via EF Core (Worker project owns the DbContext). +/// +public class WorkflowCheckpoint +{ + /// + /// Unique identifier for this checkpoint record. + /// + public Guid Id { get; set; } + + /// + /// The agent that owns this checkpoint (e.g., "DocumentProcessing"). + /// + public string AgentName { get; set; } = string.Empty; + + /// + /// The document being processed — links to the documents table. + /// + public Guid DocumentId { get; set; } + + /// + /// The current step in the workflow (e.g., "DownloadDocument", "ParseDocument"). + /// + public string CurrentActivity { get; set; } = string.Empty; + + /// + /// Serialized state data for the current step (JSON). + /// Contains step-specific data needed to resume execution. + /// + public string? StateData { get; set; } + + /// + /// Whether this checkpoint represents a completed workflow. + /// + public bool IsCompleted { get; set; } + + /// + /// Whether this checkpoint represents a failed workflow. + /// + public bool IsFailed { get; set; } + + /// + /// Error message if the workflow failed at this checkpoint. + /// + public string? ErrorMessage { get; set; } + + /// + /// When this checkpoint was created. + /// + public DateTime CreatedAt { get; set; } = DateTime.UtcNow; + + /// + /// When this checkpoint was last updated. + /// + public DateTime UpdatedAt { get; set; } = DateTime.UtcNow; +} \ No newline at end of file diff --git a/src/Shared/Shared.csproj b/src/Shared/Shared.csproj index e1bdf0a..b2fe3e1 100644 --- a/src/Shared/Shared.csproj +++ b/src/Shared/Shared.csproj @@ -1,13 +1,14 @@  - net8.0 + net10.0 enable enable + \ No newline at end of file diff --git a/src/Worker/Agents/DocumentProcessingAgent.cs b/src/Worker/Agents/DocumentProcessingAgent.cs new file mode 100644 index 0000000..08ddcab --- /dev/null +++ b/src/Worker/Agents/DocumentProcessingAgent.cs @@ -0,0 +1,272 @@ +using Microsoft.Extensions.Logging; +using Shared.Interfaces; +using Shared.Models; +using Worker.Services; + +namespace Worker.Agents; + +/// +/// MAF agent that orchestrates the PDF document processing workflow. +/// +/// Workflow activities (executed in order): +/// 1. DownloadDocument — download PDF from file storage +/// 2. ParseDocument — extract text via PdfPig +/// 3. ExtractText — OCR fallback via Tesseract if PdfPig returns empty +/// 4. SaveResult — save extracted text to database +/// 5. UpdateStatus — mark document as completed +/// +/// Each activity saves a checkpoint after execution. If the worker crashes, +/// the agent resumes from the last completed activity instead of starting over. +/// +/// Reuses existing services: PdfTextExtractor, TesseractOcrService, IDocumentRepository. +/// +public class DocumentProcessingAgent : IAgent +{ + public string AgentName => "DocumentProcessing"; + + public IReadOnlyList Activities => new List + { + "DownloadDocument", + "ParseDocument", + "ExtractText", + "SaveResult", + "UpdateStatus" + }.AsReadOnly(); + + private readonly PdfTextExtractor _textExtractor; + private readonly IDocumentRepository _repository; + private readonly IFileStorage _fileStorage; + private readonly ILogger _logger; + + public DocumentProcessingAgent( + PdfTextExtractor textExtractor, + IDocumentRepository repository, + IFileStorage fileStorage, + ILogger logger) + { + _textExtractor = textExtractor; + _repository = repository; + _fileStorage = fileStorage; + _logger = logger; + } + + /// + /// Executes the full document processing workflow with checkpoint support. + /// Skips already-completed activities (resume after crash). + /// + public async Task ExecuteAsync( + AgentContext context, + ICheckpointStore checkpointStore, + CancellationToken cancellationToken = default) + { + _logger.LogInformation("Starting {Agent} workflow for document {DocumentId}", + AgentName, context.DocumentId); + + // Load completed checkpoints to determine which activities to skip + var completedCheckpoints = await checkpointStore.LoadCompletedCheckpointsAsync( + AgentName, context.DocumentId, cancellationToken); + var completedActivities = completedCheckpoints + .Where(c => c.IsCompleted && !c.IsFailed) + .Select(c => c.CurrentActivity) + .ToHashSet(); + + try + { + // ── Activity 1: DownloadDocument ── + byte[] pdfBytes; + if (completedActivities.Contains("DownloadDocument")) + { + _logger.LogInformation("Skipping DownloadDocument (already completed)"); + // Restore PDF bytes from checkpoint state + var checkpoint = completedCheckpoints.First(c => c.CurrentActivity == "DownloadDocument"); + pdfBytes = Convert.FromBase64String(checkpoint.StateData ?? string.Empty); + } + else + { + pdfBytes = await ExecuteDownloadDocumentAsync(context, checkpointStore, cancellationToken); + } + + // ── Activity 2: ParseDocument ── + string? parsedText; + if (completedActivities.Contains("ParseDocument")) + { + _logger.LogInformation("Skipping ParseDocument (already completed)"); + var checkpoint = completedCheckpoints.First(c => c.CurrentActivity == "ParseDocument"); + parsedText = checkpoint.StateData; + } + else + { + parsedText = await ExecuteParseDocumentAsync(pdfBytes, context, checkpointStore, cancellationToken); + } + + // ── Activity 3: ExtractText (OCR fallback) ── + string? extractedText; + if (completedActivities.Contains("ExtractText")) + { + _logger.LogInformation("Skipping ExtractText (already completed)"); + var checkpoint = completedCheckpoints.First(c => c.CurrentActivity == "ExtractText"); + extractedText = checkpoint.StateData; + } + else + { + extractedText = await ExecuteExtractTextAsync(parsedText, pdfBytes, context, checkpointStore, cancellationToken); + } + + // ── Activity 4: SaveResult ── + if (!completedActivities.Contains("SaveResult")) + { + await ExecuteSaveResultAsync(extractedText, context, checkpointStore, cancellationToken); + } + else + { + _logger.LogInformation("Skipping SaveResult (already completed)"); + } + + // ── Activity 5: UpdateStatus ── + if (!completedActivities.Contains("UpdateStatus")) + { + await ExecuteUpdateStatusAsync(context, checkpointStore, cancellationToken); + } + else + { + _logger.LogInformation("Skipping UpdateStatus (already completed)"); + } + + // Clean up checkpoints after successful completion + await checkpointStore.DeleteCheckpointsAsync(AgentName, context.DocumentId, cancellationToken); + + _logger.LogInformation("Document {DocumentId} processed successfully", context.DocumentId); + return AgentResult.Success(); + } + catch (Exception ex) + { + _logger.LogError(ex, "Document processing failed for {DocumentId}", context.DocumentId); + // Save failure checkpoint + await checkpointStore.SaveCheckpointAsync( + AgentName, context.DocumentId, "Failure", + AgentResult.Failure(ex.Message), cancellationToken); + throw; + } + } + + /// + /// Activity 1: Downloads the PDF file from storage. + /// Stores PDF bytes as Base64 in checkpoint for resume support. + /// + private async Task ExecuteDownloadDocumentAsync( + AgentContext context, + ICheckpointStore checkpointStore, + CancellationToken cancellationToken) + { + cancellationToken.ThrowIfCancellationRequested(); + _logger.LogInformation("Activity: DownloadDocument for {DocumentId}", context.DocumentId); + + await using var stream = await _fileStorage.GetAsync(context.FilePath, cancellationToken); + using var memoryStream = new MemoryStream(); + await stream.CopyToAsync(memoryStream, cancellationToken); + var pdfBytes = memoryStream.ToArray(); + + // Save checkpoint with PDF bytes as Base64 (for resume) + await checkpointStore.SaveCheckpointAsync( + AgentName, context.DocumentId, "DownloadDocument", + AgentResult.Success(Convert.ToBase64String(pdfBytes)), cancellationToken); + + _logger.LogInformation("Downloaded {Bytes} bytes for {DocumentId}", pdfBytes.Length, context.DocumentId); + return pdfBytes; + } + + /// + /// Activity 2: Parses PDF text using PdfPig. + /// Returns extracted text or null if PDF is scanned (needs OCR). + /// + private async Task ExecuteParseDocumentAsync( + byte[] pdfBytes, + AgentContext context, + ICheckpointStore checkpointStore, + CancellationToken cancellationToken) + { + cancellationToken.ThrowIfCancellationRequested(); + _logger.LogInformation("Activity: ParseDocument for {DocumentId}", context.DocumentId); + + // PdfTextExtractor handles PdfPig internally + var text = await _textExtractor.ExtractTextAsync(pdfBytes, cancellationToken); + + await checkpointStore.SaveCheckpointAsync( + AgentName, context.DocumentId, "ParseDocument", + AgentResult.Success(text), cancellationToken); + + _logger.LogInformation("ParseDocument extracted {Length} chars for {DocumentId}", + text?.Length ?? 0, context.DocumentId); + return text; + } + + /// + /// Activity 3: OCR fallback — if PdfPig returned empty, use Tesseract. + /// PdfTextExtractor already handles the fallback internally. + /// + private async Task ExecuteExtractTextAsync( + string? parsedText, + byte[] pdfBytes, + AgentContext context, + ICheckpointStore checkpointStore, + CancellationToken cancellationToken) + { + cancellationToken.ThrowIfCancellationRequested(); + _logger.LogInformation("Activity: ExtractText for {DocumentId}", context.DocumentId); + + // If ParseDocument already got text, no need for OCR + var finalText = !string.IsNullOrWhiteSpace(parsedText) ? parsedText : null; + + await checkpointStore.SaveCheckpointAsync( + AgentName, context.DocumentId, "ExtractText", + AgentResult.Success(finalText), cancellationToken); + + _logger.LogInformation("ExtractText result: {Length} chars for {DocumentId}", + finalText?.Length ?? 0, context.DocumentId); + return finalText; + } + + /// + /// Activity 4: Saves extracted text to the database. + /// + private async Task ExecuteSaveResultAsync( + string? extractedText, + AgentContext context, + ICheckpointStore checkpointStore, + CancellationToken cancellationToken) + { + cancellationToken.ThrowIfCancellationRequested(); + _logger.LogInformation("Activity: SaveResult for {DocumentId}", context.DocumentId); + + await _repository.UpdateTextAsync( + context.DocumentId, extractedText, DocumentStatus.Processing, cancellationToken); + + await checkpointStore.SaveCheckpointAsync( + AgentName, context.DocumentId, "SaveResult", + AgentResult.Success(), cancellationToken); + + _logger.LogInformation("Saved text for {DocumentId}", context.DocumentId); + } + + /// + /// Activity 5: Updates document status to Completed. + /// + private async Task ExecuteUpdateStatusAsync( + AgentContext context, + ICheckpointStore checkpointStore, + CancellationToken cancellationToken) + { + cancellationToken.ThrowIfCancellationRequested(); + _logger.LogInformation("Activity: UpdateStatus for {DocumentId}", context.DocumentId); + + await _repository.TryUpdateStatusAsync( + context.DocumentId, DocumentStatus.Processing, DocumentStatus.Completed, + cancellationToken: cancellationToken); + + await checkpointStore.SaveCheckpointAsync( + AgentName, context.DocumentId, "UpdateStatus", + AgentResult.Success(), cancellationToken); + + _logger.LogInformation("Document {DocumentId} marked as Completed", context.DocumentId); + } +} \ No newline at end of file diff --git a/src/Worker/Consumers/PdfProcessingConsumer.cs b/src/Worker/Consumers/PdfProcessingConsumer.cs index 4055fc2..caa3d4b 100644 --- a/src/Worker/Consumers/PdfProcessingConsumer.cs +++ b/src/Worker/Consumers/PdfProcessingConsumer.cs @@ -1,31 +1,48 @@ using MassTransit; using Microsoft.Extensions.Logging; +using Shared.Exceptions; +using Shared.Interfaces; using Shared.Models; -using Worker.Services; +using Worker.Agents; namespace Worker.Consumers; /// /// MassTransit consumer for PdfProcessingCommand messages. /// -/// Processing pipeline: -/// 1. Idempotency check — skip if message already processed -/// 2. Optimistic lock — claim document via UPDATE WHERE status=uploaded -/// 3. Download PDF from shared storage -/// 4. Extract text via PdfPig (fallback to Tesseract OCR if needed) -/// 5. Save extracted text + mark message processed (single transaction) +/// This consumer acts as the entry point between the message broker (RabbitMQ) +/// and the MAF agent workflow. MassTransit handles: +/// - Message delivery and retry (5s → 30s → 60s delays) +/// - Dead letter queue after 3 failed retries +/// - Idempotent message handling /// -/// On failure: throw exception → MassTransit retries with delays (5s, 30s, 60s) -/// After 3 retries: message moves to error queue (DLQ) +/// The actual document processing is delegated to DocumentProcessingAgent, +/// which orchestrates the workflow with checkpoint-based durability: +/// 1. DownloadDocument — download PDF from storage +/// 2. ParseDocument — extract text via PdfPig +/// 3. ExtractText — OCR fallback via Tesseract +/// 4. SaveResult — save text to database +/// 5. UpdateStatus — mark document as completed +/// +/// If the worker crashes mid-processing, the agent resumes from the last +/// checkpoint instead of starting over. /// public class PdfProcessingConsumer : IConsumer { - private readonly DocumentProcessingService _processingService; + private readonly DocumentProcessingAgent _agent; + private readonly ICheckpointStore _checkpointStore; + private readonly IDocumentRepository _repository; private readonly ILogger _logger; - public PdfProcessingConsumer(DocumentProcessingService processingService, ILogger logger) + public PdfProcessingConsumer( + DocumentProcessingAgent agent, + ICheckpointStore checkpointStore, + IDocumentRepository repository, + ILogger logger) { - _processingService = processingService; + _agent = agent; + _checkpointStore = checkpointStore; + _repository = repository; _logger = logger; } @@ -35,16 +52,53 @@ public async Task Consume(ConsumeContext context) _logger.LogInformation("Received processing command for document {DocumentId}, retry: {RetryCount}", command.DocumentId, command.RetryCount); - var success = await _processingService.ProcessDocumentAsync(command.DocumentId, command.MessageId, context.CancellationToken); + // ── Idempotency check (at message level) ── + // The agent has its own checkpoint-based idempotency, + // but this check prevents unnecessary processing of duplicate messages. + if (await _repository.IsMessageProcessedAsync(command.MessageId, context.CancellationToken)) + { + _logger.LogInformation("Message {MessageId} already processed, skipping", command.MessageId); + return; + } + + // ── Build agent context ── + var agentContext = new AgentContext + { + DocumentId = command.DocumentId, + FilePath = command.FilePath, + AgentName = _agent.AgentName, + CurrentActivity = _agent.Activities.First() + }; - if (!success) + try { - _logger.LogWarning("Processing failed for document {DocumentId}, sending to error queue", command.DocumentId); - // Throwing signals MassTransit to apply retry policy, then move to error queue - throw new Exception($"Processing failed for document {command.DocumentId}"); + // ── Execute the MAF agent workflow ── + var result = await _agent.ExecuteAsync(agentContext, _checkpointStore, context.CancellationToken); + + if (!result.IsSuccess) + { + _logger.LogWarning("Agent workflow failed for document {DocumentId}: {Error}", + command.DocumentId, result.ErrorMessage); + throw new DocumentProcessingException(command.DocumentId, result.ErrorMessage!); + } + + // ── Mark message as processed (idempotency) ── + await _repository.MarkMessageProcessedAsync(command.MessageId, command.DocumentId, context.CancellationToken); + + _logger.LogInformation("Document {DocumentId} processed successfully, message {MessageId} consumed", + command.DocumentId, command.MessageId); } + catch (Exception ex) + { + _logger.LogError(ex, "Processing failed for document {DocumentId}", command.DocumentId); - _logger.LogInformation("Document {DocumentId} processed successfully, message {MessageId} consumed", - command.DocumentId, command.MessageId); + // Update document status to Failed + await _repository.TryUpdateStatusAsync( + command.DocumentId, DocumentStatus.Processing, DocumentStatus.Failed, + errorMessage: ex.Message, cancellationToken: context.CancellationToken); + + // Throw to trigger MassTransit retry → DLQ after 3 attempts + throw; + } } } \ No newline at end of file diff --git a/src/Worker/Data/AppDbContext.cs b/src/Worker/Data/AppDbContext.cs index 64cc7a0..577c02a 100644 --- a/src/Worker/Data/AppDbContext.cs +++ b/src/Worker/Data/AppDbContext.cs @@ -9,6 +9,8 @@ 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) { @@ -31,5 +33,32 @@ protected override void OnModelCreating(ModelBuilder modelBuilder) 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("jsonb"); + 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(); + }); } -} \ No newline at end of file +} diff --git a/src/Worker/Data/PostgreSqlCheckpointStore.cs b/src/Worker/Data/PostgreSqlCheckpointStore.cs new file mode 100644 index 0000000..25c4bda --- /dev/null +++ b/src/Worker/Data/PostgreSqlCheckpointStore.cs @@ -0,0 +1,134 @@ +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.Logging; +using Shared.Interfaces; +using Shared.Models; + +namespace Worker.Data; + +/// +/// PostgreSQL-backed implementation of ICheckpointStore. +/// Stores workflow checkpoints in the workflow_checkpoints table. +/// +/// Checkpoints enable durable execution: if a worker crashes mid-processing, +/// the agent resumes from the last saved checkpoint instead of starting over. +/// +public class PostgreSqlCheckpointStore : ICheckpointStore +{ + private readonly AppDbContext _context; + private readonly ILogger _logger; + + public PostgreSqlCheckpointStore(AppDbContext context, ILogger logger) + { + _context = context; + _logger = logger; + } + + /// + /// Saves a checkpoint for the given agent, document, and activity. + /// Uses upsert semantics: inserts new or updates existing checkpoint. + /// + public async Task SaveCheckpointAsync( + string agentName, + Guid documentId, + string activityName, + AgentResult result, + CancellationToken cancellationToken = default) + { + var existing = await _context.WorkflowCheckpoints + .FirstOrDefaultAsync(c => c.AgentName == agentName + && c.DocumentId == documentId + && c.CurrentActivity == activityName, + cancellationToken); + + var errorMessage = result.ErrorMessage?.Length > 4096 + ? result.ErrorMessage[..4096] + : result.ErrorMessage; + + if (existing != null) + { + // Update existing checkpoint + existing.StateData = result.OutputData; + existing.IsCompleted = result.IsSuccess; + existing.IsFailed = !result.IsSuccess; + existing.ErrorMessage = errorMessage; + existing.UpdatedAt = DateTime.UtcNow; + } + else + { + // Create new checkpoint + _context.WorkflowCheckpoints.Add(new WorkflowCheckpoint + { + Id = Guid.NewGuid(), + AgentName = agentName, + DocumentId = documentId, + CurrentActivity = activityName, + StateData = result.OutputData, + IsCompleted = result.IsSuccess, + IsFailed = !result.IsSuccess, + ErrorMessage = errorMessage, + CreatedAt = DateTime.UtcNow, + UpdatedAt = DateTime.UtcNow + }); + } + + await _context.SaveChangesAsync(cancellationToken); + _logger.LogDebug("Checkpoint saved: {AgentName}/{DocumentId}/{Activity}", + agentName, documentId, activityName); + } + + /// + /// Loads the most recent checkpoint for the given agent and document. + /// Returns null if no checkpoint exists (first run). + /// + public async Task LoadCheckpointAsync( + string agentName, + Guid documentId, + CancellationToken cancellationToken = default) + { + return await _context.WorkflowCheckpoints + .Where(c => c.AgentName == agentName && c.DocumentId == documentId) + .OrderByDescending(c => c.UpdatedAt) + .FirstOrDefaultAsync(cancellationToken); + } + + /// + /// Loads all completed checkpoints for the given agent and document. + /// Used to determine which activities have already been executed (for resume). + /// + public async Task> LoadCompletedCheckpointsAsync( + string agentName, + Guid documentId, + CancellationToken cancellationToken = default) + { + var checkpoints = await _context.WorkflowCheckpoints + .Where(c => c.AgentName == agentName + && c.DocumentId == documentId + && c.IsCompleted && !c.IsFailed) + .OrderBy(c => c.CreatedAt) + .ToListAsync(cancellationToken); + + return checkpoints.AsReadOnly(); + } + + /// + /// Deletes all checkpoints for the given agent and document. + /// Called after workflow completion (success or failure) to clean up. + /// + public async Task DeleteCheckpointsAsync( + string agentName, + Guid documentId, + CancellationToken cancellationToken = default) + { + var checkpoints = await _context.WorkflowCheckpoints + .Where(c => c.AgentName == agentName && c.DocumentId == documentId) + .ToListAsync(cancellationToken); + + if (checkpoints.Count > 0) + { + _context.WorkflowCheckpoints.RemoveRange(checkpoints); + await _context.SaveChangesAsync(cancellationToken); + _logger.LogDebug("Checkpoints deleted: {AgentName}/{DocumentId} ({Count} records)", + agentName, documentId, checkpoints.Count); + } + } +} \ No newline at end of file diff --git a/src/Worker/Dockerfile b/src/Worker/Dockerfile index 942fed6..87f2fe3 100644 --- a/src/Worker/Dockerfile +++ b/src/Worker/Dockerfile @@ -1,4 +1,4 @@ -FROM mcr.microsoft.com/dotnet/sdk:8.0 AS build +FROM mcr.microsoft.com/dotnet/sdk:10.0 AS build WORKDIR /src COPY src/Shared/Shared.csproj src/Shared/ @@ -8,7 +8,7 @@ RUN dotnet restore src/Worker/Worker.csproj COPY . . RUN dotnet publish src/Worker/Worker.csproj -c Release -o /app/publish -FROM mcr.microsoft.com/dotnet/aspnet:8.0 AS runtime +FROM mcr.microsoft.com/dotnet/aspnet:10.0 AS runtime RUN apt-get update && apt-get install -y --no-install-recommends \ tesseract-ocr \ tesseract-ocr-eng \ diff --git a/src/Worker/Program.cs b/src/Worker/Program.cs index d06d13d..95aa2f4 100644 --- a/src/Worker/Program.cs +++ b/src/Worker/Program.cs @@ -5,8 +5,9 @@ using Microsoft.Extensions.Diagnostics.HealthChecks; using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Logging; -using Prometheus; +using Serilog; using Shared.Interfaces; +using Worker.Agents; using Worker.Consumers; using Worker.Data; using Worker.Services; @@ -14,15 +15,23 @@ // ── Worker Host ── // Console application running as a generic host with MassTransit consumer. -// Processes PdfProcessingCommand messages from RabbitMQ: -// 1. Idempotency check (processed_messages table) -// 2. Optimistic lock (UPDATE documents SET status='processing' WHERE status='uploaded') -// 3. Download PDF from storage -// 4. Extract text via PdfPig (fallback to Tesseract OCR if empty) -// 5. Save text + mark message processed in one transaction -// 6. ACK on success, throw on failure → MassTransit retry with delays +// Uses hybrid architecture: +// - MassTransit handles message delivery, retry, DLQ (inter-service boundary) +// - MAF DocumentProcessingAgent handles workflow orchestration with checkpoints +// +// Processing workflow (inside MAF agent): +// 1. DownloadDocument — download PDF from storage +// 2. ParseDocument — extract text via PdfPig +// 3. ExtractText — OCR fallback via Tesseract +// 4. SaveResult — save text to database +// 5. UpdateStatus — mark document as completed +// +// Each activity saves a checkpoint. If worker crashes, agent resumes from last checkpoint. var host = Host.CreateDefaultBuilder(args) + .UseSerilog((context, loggerConfig) => + loggerConfig.ReadFrom.Configuration(context.Configuration) + .WriteTo.Console(new Serilog.Formatting.Compact.CompactJsonFormatter())) .ConfigureServices((hostContext, services) => { var configuration = hostContext.Configuration; @@ -50,6 +59,24 @@ // ── Application Services ── services.AddScoped(); + + // ── Prometheus metrics as a hosted service ── + // Wraps MetricServer in IHostedService so it stops gracefully on SIGTERM + services.AddHostedService(); + + // ── MAF Agent ── + // DocumentProcessingAgent orchestrates the PDF processing workflow + // with checkpoint-based durability. Registered as scoped so each + // message gets a fresh agent instance with its own state. + services.AddScoped(); + + // ── Checkpoint Store ── + // PostgreSQL-backed checkpoint storage for durable agent execution. + // Enables resume after worker crash — agent continues from last checkpoint. + services.AddScoped(); + + // Keep DocumentProcessingService for backward compatibility + // (can be removed in future iterations) services.AddScoped(); // ── MassTransit + RabbitMQ Consumer ── @@ -82,18 +109,8 @@ }); }); }) - .ConfigureLogging(logging => - { - logging.ClearProviders(); - logging.AddConsole(); - }) .Build(); -// ── Prometheus metrics server ── -// Serves /metrics on a separate port so Prometheus can scrape worker metrics -using var metricServer = new MetricServer(port: 5091); -metricServer.Start(); - // Auto-create database tables (Dev only) using (var scope = host.Services.CreateScope()) { diff --git a/src/Worker/Services/MetricsHostedService.cs b/src/Worker/Services/MetricsHostedService.cs new file mode 100644 index 0000000..45fe677 --- /dev/null +++ b/src/Worker/Services/MetricsHostedService.cs @@ -0,0 +1,42 @@ +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using Prometheus; + +namespace Worker.Services; + +/// +/// Hosted service that wraps the Prometheus MetricServer for proper +/// lifecycle integration with the generic host. On SIGTERM, the host +/// calls StopAsync(), which stops the metric server gracefully. +/// +/// Without this wrapper, the MetricServer created via "using var" in +/// Program.cs would remain hanging after host shutdown. +/// +public class MetricsHostedService : IHostedService +{ + private readonly MetricServer _metricServer; + private readonly ILogger _logger; + private readonly int _port; + + public MetricsHostedService(IConfiguration configuration, ILogger logger) + { + _logger = logger; + _port = configuration.GetValue("Metrics:Port", 5091); + _metricServer = new MetricServer(port: _port); + } + + public Task StartAsync(CancellationToken cancellationToken) + { + _logger.LogInformation("Starting Prometheus metric server on port {Port}", _port); + _metricServer.Start(); + return Task.CompletedTask; + } + + public Task StopAsync(CancellationToken cancellationToken) + { + _logger.LogInformation("Stopping Prometheus metric server"); + _metricServer.Stop(); + return Task.CompletedTask; + } +} \ No newline at end of file diff --git a/src/Worker/Services/PdfTextExtractor.cs b/src/Worker/Services/PdfTextExtractor.cs index 38a03dd..12dcf1d 100644 --- a/src/Worker/Services/PdfTextExtractor.cs +++ b/src/Worker/Services/PdfTextExtractor.cs @@ -14,7 +14,7 @@ public PdfTextExtractor(ILogger logger, IOCRService? ocrServic _ocrService = ocrService; } - public async Task ExtractTextAsync(byte[] pdfContent, CancellationToken cancellationToken = default) + public virtual async Task ExtractTextAsync(byte[] pdfContent, CancellationToken cancellationToken = default) { string? extractedText = null; diff --git a/src/Worker/Worker.csproj b/src/Worker/Worker.csproj index 97ec08a..8a64d90 100644 --- a/src/Worker/Worker.csproj +++ b/src/Worker/Worker.csproj @@ -2,21 +2,24 @@ Exe - net8.0 + net10.0 enable enable - + - - - + + + + + + - + diff --git a/src/Worker/appsettings.json b/src/Worker/appsettings.json index 10b859c..0cc6225 100644 --- a/src/Worker/appsettings.json +++ b/src/Worker/appsettings.json @@ -18,5 +18,8 @@ }, "OcrBase": { "ApiKey": "" + }, + "Metrics": { + "Port": 5091 } } \ No newline at end of file diff --git a/tests/ApiGateway.UnitTests/ApiGateway.UnitTests.csproj b/tests/ApiGateway.UnitTests/ApiGateway.UnitTests.csproj index 9ec9e3b..05049df 100644 --- a/tests/ApiGateway.UnitTests/ApiGateway.UnitTests.csproj +++ b/tests/ApiGateway.UnitTests/ApiGateway.UnitTests.csproj @@ -1,7 +1,7 @@ - net8.0 + net10.0 enable enable @@ -11,7 +11,7 @@ - + @@ -22,8 +22,8 @@ - - + + - + \ No newline at end of file diff --git a/tests/ApiGateway.UnitTests/OutboxPublisherTests.cs b/tests/ApiGateway.UnitTests/OutboxPublisherTests.cs new file mode 100644 index 0000000..936f0fd --- /dev/null +++ b/tests/ApiGateway.UnitTests/OutboxPublisherTests.cs @@ -0,0 +1,208 @@ +using System.Text.Json; +using ApiGateway.BackgroundServices; +using MassTransit; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; +using Moq; +using Shared.Interfaces; +using Shared.Models; + +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> _loggerMock; + private readonly OutboxPublisher _publisher; + + public OutboxPublisherTests() + { + _scopeFactoryMock = new Mock(); + _scopeMock = new Mock(); + _busMock = new Mock(); + _loggerMock = new Mock>(); + _repositoryMock = 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); + + _scopeMock + .Setup(x => x.ServiceProvider) + .Returns(serviceProviderMock.Object); + + _publisher = new OutboxPublisher( + _scopeFactoryMock.Object, + _busMock.Object, + _loggerMock.Object); + } + + [Fact] + public async Task ExecuteAsync_PublishesPendingMessagesAndMarksProcessed() + { + // Arrange + var documentId = Guid.NewGuid(); + var outboxId = Guid.NewGuid(); + var command = new PdfProcessingCommand + { + DocumentId = documentId, + MessageId = Guid.NewGuid(), + FilePath = "/test.pdf" + }; + + var pendingMessages = new List + { + new() + { + Id = outboxId, + DocumentId = documentId, + MessagePayload = JsonSerializer.Serialize(command), + CreatedAt = DateTime.UtcNow + } + }; + + _repositoryMock + .Setup(x => x.GetOutboxPendingAsync(It.IsAny())) + .ReturnsAsync(pendingMessages); + + _repositoryMock + .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( + outboxId, It.IsAny()), Times.Once); + } + + [Fact] + public async Task ExecuteAsync_SkipsCorruptMessagesAndContinues() + { + // Arrange + var pendingMessages = new List + { + new() + { + Id = Guid.NewGuid(), + DocumentId = Guid.NewGuid(), + MessagePayload = "not-valid-json", + CreatedAt = DateTime.UtcNow + } + }; + + _repositoryMock + .Setup(x => x.GetOutboxPendingAsync(It.IsAny())) + .ReturnsAsync(pendingMessages); + + _repositoryMock + .Setup(x => x.MarkOutboxProcessedAsync(It.IsAny(), It.IsAny())) + .Returns(Task.CompletedTask); + + using var cts = new CancellationTokenSource(); + var executeTask = _publisher.StartAsync(cts.Token); + + 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( + It.IsAny(), It.IsAny()), Times.Once); + } + + [Fact] + public async Task ExecuteAsync_ContinuesOnPublishFailure() + { + // Arrange + var documentId = Guid.NewGuid(); + var command = new PdfProcessingCommand + { + DocumentId = documentId, + MessageId = Guid.NewGuid(), + FilePath = "/test.pdf" + }; + + var pendingMessages = new List + { + new() + { + Id = Guid.NewGuid(), + DocumentId = documentId, + MessagePayload = JsonSerializer.Serialize(command), + CreatedAt = DateTime.UtcNow + } + }; + + _repositoryMock + .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")); + + using var cts = new CancellationTokenSource(); + var executeTask = _publisher.StartAsync(cts.Token); + + 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( + It.IsAny(), It.IsAny()), Times.Never); + } + + [Fact] + public async Task ExecuteAsync_StopsOnCancellation() + { + // Arrange + _repositoryMock + .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 diff --git a/tests/IntegrationTests/IntegrationTests.csproj b/tests/IntegrationTests/IntegrationTests.csproj index 242b5c1..5983453 100644 --- a/tests/IntegrationTests/IntegrationTests.csproj +++ b/tests/IntegrationTests/IntegrationTests.csproj @@ -1,7 +1,7 @@ - net8.0 + net10.0 enable enable false @@ -15,8 +15,7 @@ - - + diff --git a/tests/Worker.UnitTests/ConcurrencyTests.cs b/tests/Worker.UnitTests/ConcurrencyTests.cs new file mode 100644 index 0000000..396c2f3 --- /dev/null +++ b/tests/Worker.UnitTests/ConcurrencyTests.cs @@ -0,0 +1,171 @@ +using MassTransit; +using Microsoft.Extensions.Logging; +using Moq; +using Shared.Interfaces; +using Shared.Models; +using Worker.Agents; +using Worker.Consumers; +using Worker.Services; + +namespace Worker.UnitTests; + +/// +/// Tests for race conditions in concurrent document processing. +/// +/// Uses a real DocumentProcessingAgent with mocked dependencies to verify +/// that concurrent duplicate messages are handled gracefully. +/// +/// In production, PrefetchCount = 1 prevents concurrent message delivery +/// within a single worker instance, but multiple worker replicas can still +/// receive the same message (at-least-once delivery). +/// +public class ConcurrencyTests +{ + private readonly Mock _checkpointStoreMock; + private readonly Mock _repositoryMock; + private readonly Mock _fileStorageMock; + private readonly Mock> _loggerMock; + private readonly PdfProcessingConsumer _consumer; + + public ConcurrencyTests() + { + _checkpointStoreMock = new Mock(); + _repositoryMock = new Mock(); + _fileStorageMock = new Mock(); + _loggerMock = new Mock>(); + + // Create a real DocumentProcessingAgent with mocked dependencies + var extractorLoggerMock = new Mock>(); + var ocrServiceMock = Mock.Of(); + var textExtractor = new PdfTextExtractor(extractorLoggerMock.Object, ocrServiceMock); + + var agentLoggerMock = new Mock>(); + var agent = new DocumentProcessingAgent( + textExtractor, + _repositoryMock.Object, + _fileStorageMock.Object, + agentLoggerMock.Object); + + _consumer = new PdfProcessingConsumer( + agent, + _checkpointStoreMock.Object, + _repositoryMock.Object, + _loggerMock.Object); + } + + [Fact] + public async Task Consume_ConcurrentDuplicate_TryUpdateStatusReturnsFalseForSecondWorker() + { + // Arrange — simulate two workers with the same document + var documentId = Guid.NewGuid(); + var command = new PdfProcessingCommand + { + DocumentId = documentId, + MessageId = Guid.NewGuid(), + FilePath = "/test.pdf", + RetryCount = 0 + }; + + var consumeContextMock = new Mock>(); + consumeContextMock.Setup(x => x.Message).Returns(command); + consumeContextMock.Setup(x => x.CancellationToken).Returns(CancellationToken.None); + + // No existing checkpoints + _checkpointStoreMock + .Setup(x => x.LoadCompletedCheckpointsAsync( + "DocumentProcessing", documentId, It.IsAny())) + .ReturnsAsync(Array.Empty()); + + // Mock file storage to return a minimal PDF + var fakePdfBytes = new byte[] { 0x25, 0x50, 0x44, 0x46 }; // %PDF header + _fileStorageMock + .Setup(x => x.GetAsync("/test.pdf", It.IsAny())) + .ReturnsAsync(new MemoryStream(fakePdfBytes)); + + // Mock text extraction + var extractorLoggerMock = new Mock>(); + var ocrServiceMock = Mock.Of(); + var textExtractorMock = new Mock(extractorLoggerMock.Object, ocrServiceMock) + { + CallBase = true + }; + textExtractorMock + .Setup(x => x.ExtractTextAsync(It.IsAny(), It.IsAny())) + .ReturnsAsync("Extracted text"); + + _repositoryMock + .Setup(x => x.IsMessageProcessedAsync(command.MessageId, It.IsAny())) + .ReturnsAsync(false); + + _repositoryMock + .Setup(x => x.MarkMessageProcessedAsync(command.MessageId, documentId, It.IsAny())) + .Returns(Task.CompletedTask); + + // Simulate race: first worker already set status to Completed, + // so TryUpdateStatusAsync returns false for this worker + _repositoryMock + .Setup(x => x.TryUpdateStatusAsync( + documentId, + DocumentStatus.Processing, + DocumentStatus.Completed, + It.IsAny(), + It.IsAny())) + .ReturnsAsync(false); + + // Act + await _consumer.Consume(consumeContextMock.Object); + + // Assert: agent was called, but TryUpdateStatusAsync returned false + // meaning another worker already finished this document + _repositoryMock.Verify(x => x.TryUpdateStatusAsync( + documentId, + DocumentStatus.Processing, + DocumentStatus.Completed, + null, + It.IsAny()), Times.Once); + } + + [Fact] + public async Task Consume_TwoWorkersSameDocument_SecondWorkerHandlesGracefully() + { + // Simulate what happens when two consumers receive the same message + // from two worker instances (at-least-once delivery) + var documentId = Guid.NewGuid(); + var command = new PdfProcessingCommand + { + DocumentId = documentId, + MessageId = Guid.NewGuid(), + FilePath = "/test.pdf", + RetryCount = 0 + }; + + var repositoryMock2 = new Mock(); + repositoryMock2 + .Setup(x => x.IsMessageProcessedAsync(command.MessageId, It.IsAny())) + .ReturnsAsync(true); // Already processed by worker 1 + + var consumer2 = new PdfProcessingConsumer( + // Use the same agent instance, but the idempotency check happens first + new DocumentProcessingAgent( + new PdfTextExtractor( + new Mock>().Object, + Mock.Of()), + repositoryMock2.Object, + Mock.Of(), + new Mock>().Object), + _checkpointStoreMock.Object, + repositoryMock2.Object, + _loggerMock.Object); + + var consumeContextMock = new Mock>(); + consumeContextMock.Setup(x => x.Message).Returns(command); + consumeContextMock.Setup(x => x.CancellationToken).Returns(CancellationToken.None); + + // Act — worker 2 receives duplicate + await consumer2.Consume(consumeContextMock.Object); + + // Assert: file storage was never accessed (processing skipped before agent call) + _fileStorageMock.Verify(x => x.GetAsync( + It.IsAny(), It.IsAny()), Times.Never); + } +} \ No newline at end of file diff --git a/tests/Worker.UnitTests/DocumentProcessingAgentTests.cs b/tests/Worker.UnitTests/DocumentProcessingAgentTests.cs new file mode 100644 index 0000000..2c9ac03 --- /dev/null +++ b/tests/Worker.UnitTests/DocumentProcessingAgentTests.cs @@ -0,0 +1,465 @@ +using System.IO; +using Microsoft.Extensions.Logging; +using Moq; +using Shared.Interfaces; +using Shared.Models; +using Worker.Agents; +using Worker.Services; + +namespace Worker.UnitTests; + +/// +/// Unit tests for DocumentProcessingAgent. +/// Tests the MAF agent workflow orchestration with mocked dependencies. +/// +public class DocumentProcessingAgentTests +{ + private readonly Mock _repositoryMock; + private readonly Mock _fileStorageMock; + private readonly Mock _checkpointStoreMock; + private readonly Mock _textExtractorMock; + private readonly Mock> _loggerMock; + private readonly DocumentProcessingAgent _agent; + + public DocumentProcessingAgentTests() + { + _repositoryMock = new Mock(); + _fileStorageMock = new Mock(); + _checkpointStoreMock = new Mock(); + _loggerMock = new Mock>(); + + // Create PdfTextExtractor with mocked OCR (null disables OCR fallback) + var extractorLoggerMock = new Mock>(); + _textExtractorMock = new Mock(extractorLoggerMock.Object, Mock.Of()) + { + CallBase = true + }; + + _agent = new DocumentProcessingAgent( + _textExtractorMock.Object, + _repositoryMock.Object, + _fileStorageMock.Object, + _loggerMock.Object); + } + + [Fact] + public void AgentName_ReturnsDocumentProcessing() + { + Assert.Equal("DocumentProcessing", _agent.AgentName); + } + + [Fact] + public void Activities_ReturnsFiveActivities() + { + var activities = _agent.Activities; + + Assert.Equal(5, activities.Count); + Assert.Equal("DownloadDocument", activities[0]); + Assert.Equal("ParseDocument", activities[1]); + Assert.Equal("ExtractText", activities[2]); + Assert.Equal("SaveResult", activities[3]); + Assert.Equal("UpdateStatus", activities[4]); + } + + [Fact] + public async Task ExecuteAsync_FullWorkflow_CompletesSuccessfully() + { + // Arrange + var documentId = Guid.NewGuid(); + var context = new AgentContext + { + DocumentId = documentId, + FilePath = "/test/sample.pdf", + AgentName = _agent.AgentName, + CurrentActivity = _agent.Activities.First() + }; + + // No existing checkpoints (first run) + _checkpointStoreMock + .Setup(x => x.LoadCompletedCheckpointsAsync(_agent.AgentName, documentId, It.IsAny())) + .ReturnsAsync(Array.Empty()); + + // Mock file storage — return a minimal PDF-like byte array + var fakePdfBytes = new byte[] { 0x25, 0x50, 0x44, 0x46 }; // %PDF header + var memoryStream = new MemoryStream(fakePdfBytes); + _fileStorageMock + .Setup(x => x.GetAsync("/test/sample.pdf", It.IsAny())) + .ReturnsAsync(memoryStream); + + // Mock text extraction — return some text + _textExtractorMock + .Setup(x => x.ExtractTextAsync(It.IsAny(), It.IsAny())) + .ReturnsAsync("Extracted text from PDF"); + + // Mock repository + _repositoryMock + .Setup(x => x.UpdateTextAsync(documentId, "Extracted text from PDF", DocumentStatus.Processing, It.IsAny())) + .Returns(Task.CompletedTask); + + _repositoryMock + .Setup(x => x.TryUpdateStatusAsync(documentId, DocumentStatus.Processing, DocumentStatus.Completed, It.IsAny(), It.IsAny())) + .ReturnsAsync(true); + + // Act + var result = await _agent.ExecuteAsync(context, _checkpointStoreMock.Object); + + // Assert + Assert.True(result.IsSuccess); + + // Verify all 5 checkpoints were saved + _checkpointStoreMock.Verify(x => x.SaveCheckpointAsync( + _agent.AgentName, documentId, "DownloadDocument", + It.IsAny(), It.IsAny()), Times.Once); + + _checkpointStoreMock.Verify(x => x.SaveCheckpointAsync( + _agent.AgentName, documentId, "ParseDocument", + It.IsAny(), It.IsAny()), Times.Once); + + _checkpointStoreMock.Verify(x => x.SaveCheckpointAsync( + _agent.AgentName, documentId, "ExtractText", + It.IsAny(), It.IsAny()), Times.Once); + + _checkpointStoreMock.Verify(x => x.SaveCheckpointAsync( + _agent.AgentName, documentId, "SaveResult", + It.IsAny(), It.IsAny()), Times.Once); + + _checkpointStoreMock.Verify(x => x.SaveCheckpointAsync( + _agent.AgentName, documentId, "UpdateStatus", + It.IsAny(), It.IsAny()), Times.Once); + + // Verify cleanup + _checkpointStoreMock.Verify(x => x.DeleteCheckpointsAsync( + _agent.AgentName, documentId, It.IsAny()), Times.Once); + } + + [Fact] + public async Task ExecuteAsync_ResumeAfterCrash_SkipsCompletedActivities() + { + // Arrange + var documentId = Guid.NewGuid(); + var context = new AgentContext + { + DocumentId = documentId, + FilePath = "/test/sample.pdf", + AgentName = _agent.AgentName, + CurrentActivity = _agent.Activities.First() + }; + + // Simulate crash after DownloadDocument and ParseDocument completed + var completedCheckpoints = new List + { + new() { AgentName = "DocumentProcessing", DocumentId = documentId, CurrentActivity = "DownloadDocument", IsCompleted = true, StateData = Convert.ToBase64String(new byte[] { 0x25, 0x50, 0x44, 0x46 }) }, + new() { AgentName = "DocumentProcessing", DocumentId = documentId, CurrentActivity = "ParseDocument", IsCompleted = true, StateData = "Parsed text" } + }; + + _checkpointStoreMock + .Setup(x => x.LoadCompletedCheckpointsAsync(_agent.AgentName, documentId, It.IsAny())) + .ReturnsAsync(completedCheckpoints); + + // Mock text extraction — should use parsed text from checkpoint + _textExtractorMock + .Setup(x => x.ExtractTextAsync(It.IsAny(), It.IsAny())) + .ReturnsAsync("Parsed text"); + + _repositoryMock + .Setup(x => x.UpdateTextAsync(documentId, It.IsAny(), DocumentStatus.Processing, It.IsAny())) + .Returns(Task.CompletedTask); + + _repositoryMock + .Setup(x => x.TryUpdateStatusAsync(documentId, DocumentStatus.Processing, DocumentStatus.Completed, It.IsAny(), It.IsAny())) + .ReturnsAsync(true); + + // Act + var result = await _agent.ExecuteAsync(context, _checkpointStoreMock.Object); + + // Assert + Assert.True(result.IsSuccess); + + // Verify DownloadDocument and ParseDocument checkpoints were NOT saved again + _checkpointStoreMock.Verify(x => x.SaveCheckpointAsync( + _agent.AgentName, documentId, "DownloadDocument", + It.IsAny(), It.IsAny()), Times.Never); + + _checkpointStoreMock.Verify(x => x.SaveCheckpointAsync( + _agent.AgentName, documentId, "ParseDocument", + It.IsAny(), It.IsAny()), Times.Never); + + // Verify remaining activities were executed + _checkpointStoreMock.Verify(x => x.SaveCheckpointAsync( + _agent.AgentName, documentId, "ExtractText", + It.IsAny(), It.IsAny()), Times.Once); + + _checkpointStoreMock.Verify(x => x.SaveCheckpointAsync( + _agent.AgentName, documentId, "SaveResult", + It.IsAny(), It.IsAny()), Times.Once); + + _checkpointStoreMock.Verify(x => x.SaveCheckpointAsync( + _agent.AgentName, documentId, "UpdateStatus", + It.IsAny(), It.IsAny()), Times.Once); + } + + [Fact] + public async Task ExecuteAsync_ResumeFromMiddle_SkipsFirstThreeActivities() + { + // Arrange + var documentId = Guid.NewGuid(); + var context = new AgentContext + { + DocumentId = documentId, + FilePath = "/test/sample.pdf", + AgentName = _agent.AgentName, + CurrentActivity = _agent.Activities.First() + }; + + // Simulate crash after DownloadDocument, ParseDocument, and ExtractText completed + var completedCheckpoints = new List + { + new() { AgentName = "DocumentProcessing", DocumentId = documentId, CurrentActivity = "DownloadDocument", IsCompleted = true, StateData = Convert.ToBase64String(new byte[] { 0x25, 0x50, 0x44, 0x46 }) }, + new() { AgentName = "DocumentProcessing", DocumentId = documentId, CurrentActivity = "ParseDocument", IsCompleted = true, StateData = "Parsed text from PDF" }, + new() { AgentName = "DocumentProcessing", DocumentId = documentId, CurrentActivity = "ExtractText", IsCompleted = true, StateData = "Final extracted text" } + }; + + _checkpointStoreMock + .Setup(x => x.LoadCompletedCheckpointsAsync(_agent.AgentName, documentId, It.IsAny())) + .ReturnsAsync(completedCheckpoints); + + _repositoryMock + .Setup(x => x.UpdateTextAsync(documentId, "Final extracted text", DocumentStatus.Processing, It.IsAny())) + .Returns(Task.CompletedTask); + + _repositoryMock + .Setup(x => x.TryUpdateStatusAsync(documentId, DocumentStatus.Processing, DocumentStatus.Completed, It.IsAny(), It.IsAny())) + .ReturnsAsync(true); + + // Act + var result = await _agent.ExecuteAsync(context, _checkpointStoreMock.Object); + + // Assert + Assert.True(result.IsSuccess); + + // First 3 activities should be skipped + _checkpointStoreMock.Verify(x => x.SaveCheckpointAsync( + _agent.AgentName, documentId, "DownloadDocument", + It.IsAny(), It.IsAny()), Times.Never); + + _checkpointStoreMock.Verify(x => x.SaveCheckpointAsync( + _agent.AgentName, documentId, "ParseDocument", + It.IsAny(), It.IsAny()), Times.Never); + + _checkpointStoreMock.Verify(x => x.SaveCheckpointAsync( + _agent.AgentName, documentId, "ExtractText", + It.IsAny(), It.IsAny()), Times.Never); + + // Last 2 activities should execute + _checkpointStoreMock.Verify(x => x.SaveCheckpointAsync( + _agent.AgentName, documentId, "SaveResult", + It.IsAny(), It.IsAny()), Times.Once); + + _checkpointStoreMock.Verify(x => x.SaveCheckpointAsync( + _agent.AgentName, documentId, "UpdateStatus", + It.IsAny(), It.IsAny()), Times.Once); + + // Verify cleanup + _checkpointStoreMock.Verify(x => x.DeleteCheckpointsAsync( + _agent.AgentName, documentId, It.IsAny()), Times.Once); + } + + [Fact] + public async Task ExecuteAsync_ResumeFromLastActivity_SkipsFirstFourActivities() + { + // Arrange + var documentId = Guid.NewGuid(); + var context = new AgentContext + { + DocumentId = documentId, + FilePath = "/test/sample.pdf", + AgentName = _agent.AgentName, + CurrentActivity = _agent.Activities.First() + }; + + // Simulate crash after 4 of 5 activities completed + var completedCheckpoints = new List + { + new() { AgentName = "DocumentProcessing", DocumentId = documentId, CurrentActivity = "DownloadDocument", IsCompleted = true, StateData = Convert.ToBase64String(new byte[] { 0x25, 0x50, 0x44, 0x46 }) }, + new() { AgentName = "DocumentProcessing", DocumentId = documentId, CurrentActivity = "ParseDocument", IsCompleted = true, StateData = "Parsed text" }, + new() { AgentName = "DocumentProcessing", DocumentId = documentId, CurrentActivity = "ExtractText", IsCompleted = true, StateData = "Extracted text" }, + new() { AgentName = "DocumentProcessing", DocumentId = documentId, CurrentActivity = "SaveResult", IsCompleted = true } + }; + + _checkpointStoreMock + .Setup(x => x.LoadCompletedCheckpointsAsync(_agent.AgentName, documentId, It.IsAny())) + .ReturnsAsync(completedCheckpoints); + + _repositoryMock + .Setup(x => x.TryUpdateStatusAsync(documentId, DocumentStatus.Processing, DocumentStatus.Completed, It.IsAny(), It.IsAny())) + .ReturnsAsync(true); + + // Act + var result = await _agent.ExecuteAsync(context, _checkpointStoreMock.Object); + + // Assert + Assert.True(result.IsSuccess); + + // First 4 activities should be skipped + _checkpointStoreMock.Verify(x => x.SaveCheckpointAsync( + _agent.AgentName, documentId, "DownloadDocument", + It.IsAny(), It.IsAny()), Times.Never); + + _checkpointStoreMock.Verify(x => x.SaveCheckpointAsync( + _agent.AgentName, documentId, "ParseDocument", + It.IsAny(), It.IsAny()), Times.Never); + + _checkpointStoreMock.Verify(x => x.SaveCheckpointAsync( + _agent.AgentName, documentId, "ExtractText", + It.IsAny(), It.IsAny()), Times.Never); + + _checkpointStoreMock.Verify(x => x.SaveCheckpointAsync( + _agent.AgentName, documentId, "SaveResult", + It.IsAny(), It.IsAny()), Times.Never); + + // Only UpdateStatus should execute + _checkpointStoreMock.Verify(x => x.SaveCheckpointAsync( + _agent.AgentName, documentId, "UpdateStatus", + It.IsAny(), It.IsAny()), Times.Once); + + // Verify cleanup + _checkpointStoreMock.Verify(x => x.DeleteCheckpointsAsync( + _agent.AgentName, documentId, It.IsAny()), Times.Once); + } + + [Fact] + public async Task ExecuteAsync_AllActivitiesCompleted_OnlyCleansUp() + { + // Arrange + var documentId = Guid.NewGuid(); + var context = new AgentContext + { + DocumentId = documentId, + FilePath = "/test/sample.pdf", + AgentName = _agent.AgentName, + CurrentActivity = _agent.Activities.First() + }; + + // All 5 activities already completed (e.g., reprocessing after cleanup failure) + var completedCheckpoints = new List + { + new() { AgentName = "DocumentProcessing", DocumentId = documentId, CurrentActivity = "DownloadDocument", IsCompleted = true, StateData = Convert.ToBase64String(new byte[] { 0x25, 0x50, 0x44, 0x46 }) }, + new() { AgentName = "DocumentProcessing", DocumentId = documentId, CurrentActivity = "ParseDocument", IsCompleted = true, StateData = "Parsed text" }, + new() { AgentName = "DocumentProcessing", DocumentId = documentId, CurrentActivity = "ExtractText", IsCompleted = true, StateData = "Extracted text" }, + new() { AgentName = "DocumentProcessing", DocumentId = documentId, CurrentActivity = "SaveResult", IsCompleted = true }, + new() { AgentName = "DocumentProcessing", DocumentId = documentId, CurrentActivity = "UpdateStatus", IsCompleted = true } + }; + + _checkpointStoreMock + .Setup(x => x.LoadCompletedCheckpointsAsync(_agent.AgentName, documentId, It.IsAny())) + .ReturnsAsync(completedCheckpoints); + + // Act + var result = await _agent.ExecuteAsync(context, _checkpointStoreMock.Object); + + // Assert + Assert.True(result.IsSuccess); + + // No activity checkpoints should be saved — all skipped + _checkpointStoreMock.Verify(x => x.SaveCheckpointAsync( + _agent.AgentName, documentId, It.IsAny(), + It.IsAny(), It.IsAny()), Times.Never); + + // Cleanup should still run + _checkpointStoreMock.Verify(x => x.DeleteCheckpointsAsync( + _agent.AgentName, documentId, It.IsAny()), Times.Once); + } + + [Fact] + public async Task ExecuteAsync_CheckpointStateData_RoundtripsBase64Bytes() + { + // Arrange + var documentId = Guid.NewGuid(); + var context = new AgentContext + { + DocumentId = documentId, + FilePath = "/test/sample.pdf", + AgentName = _agent.AgentName, + CurrentActivity = _agent.Activities.First() + }; + + // Simulate a realistic PDF byte array roundtrip through checkpoint + var originalPdfBytes = new byte[] { 0x25, 0x50, 0x44, 0x46, 0x2D, 0x31, 0x2E, 0x34 }; // %PDF-1.4 + var base64State = Convert.ToBase64String(originalPdfBytes); + + var completedCheckpoints = new List + { + new() { AgentName = "DocumentProcessing", DocumentId = documentId, CurrentActivity = "DownloadDocument", IsCompleted = true, StateData = base64State } + }; + + _checkpointStoreMock + .Setup(x => x.LoadCompletedCheckpointsAsync(_agent.AgentName, documentId, It.IsAny())) + .ReturnsAsync(completedCheckpoints); + + // Capture the bytes passed to ExtractText to verify roundtrip + byte[]? capturedBytes = null; + _textExtractorMock + .Setup(x => x.ExtractTextAsync(It.IsAny(), It.IsAny())) + .Callback((bytes, _) => capturedBytes = bytes) + .ReturnsAsync("Roundtrip text"); + + _repositoryMock + .Setup(x => x.UpdateTextAsync(documentId, "Roundtrip text", DocumentStatus.Processing, It.IsAny())) + .Returns(Task.CompletedTask); + + _repositoryMock + .Setup(x => x.TryUpdateStatusAsync(documentId, DocumentStatus.Processing, DocumentStatus.Completed, It.IsAny(), It.IsAny())) + .ReturnsAsync(true); + + // Act + var result = await _agent.ExecuteAsync(context, _checkpointStoreMock.Object); + + // Assert + Assert.True(result.IsSuccess); + Assert.NotNull(capturedBytes); + Assert.Equal(originalPdfBytes, capturedBytes); + } + + [Fact] + public async Task ExecuteAsync_FailureCheckpoint_PreservesErrorMessage() + { + // Arrange + var documentId = Guid.NewGuid(); + var context = new AgentContext + { + DocumentId = documentId, + FilePath = "/test/sample.pdf", + AgentName = _agent.AgentName, + CurrentActivity = _agent.Activities.First() + }; + + _checkpointStoreMock + .Setup(x => x.LoadCompletedCheckpointsAsync(_agent.AgentName, documentId, It.IsAny())) + .ReturnsAsync(Array.Empty()); + + // Mock file storage — throw exception on download + _fileStorageMock + .Setup(x => x.GetAsync("/test/sample.pdf", It.IsAny())) + .ThrowsAsync(new IOException("Storage unavailable")); + + // Capture the AgentResult passed to SaveCheckpointAsync for "Failure" + AgentResult? capturedResult = null; + _checkpointStoreMock + .Setup(x => x.SaveCheckpointAsync( + _agent.AgentName, documentId, "Failure", + It.IsAny(), It.IsAny())) + .Callback((_, _, _, result, _) => capturedResult = result) + .Returns(Task.CompletedTask); + + // Act & Assert + var ex = await Assert.ThrowsAsync( + () => _agent.ExecuteAsync(context, _checkpointStoreMock.Object)); + + Assert.Contains("Storage unavailable", ex.Message); + + // Verify the failure checkpoint captured the error message + Assert.NotNull(capturedResult); + Assert.False(capturedResult.IsSuccess); + Assert.Contains("Storage unavailable", capturedResult.ErrorMessage); + } +} diff --git a/tests/Worker.UnitTests/DocumentProcessingServiceTests.cs b/tests/Worker.UnitTests/DocumentProcessingServiceTests.cs index 4662560..a7fc596 100644 --- a/tests/Worker.UnitTests/DocumentProcessingServiceTests.cs +++ b/tests/Worker.UnitTests/DocumentProcessingServiceTests.cs @@ -19,9 +19,10 @@ public DocumentProcessingServiceTests() _fileStorageMock = new Mock(); var loggerMock = new Mock>(); - // Create PdfTextExtractor with null OCR (PdfPig only mode) + // Create PdfTextExtractor without OCR service (PdfPig only mode) + // IOCRService parameter is optional — passing null disables OCR fallback var extractorLoggerMock = new Mock>(); - _textExtractorMock = new Mock(extractorLoggerMock.Object, (IOCRService?)null) + _textExtractorMock = new Mock(extractorLoggerMock.Object, Mock.Of()) { CallBase = true }; diff --git a/tests/Worker.UnitTests/PdfProcessingConsumerRetryTests.cs b/tests/Worker.UnitTests/PdfProcessingConsumerRetryTests.cs new file mode 100644 index 0000000..553168f --- /dev/null +++ b/tests/Worker.UnitTests/PdfProcessingConsumerRetryTests.cs @@ -0,0 +1,182 @@ +using MassTransit; +using Microsoft.Extensions.Logging; +using Moq; +using Shared.Exceptions; +using Shared.Interfaces; +using Shared.Models; +using Worker.Agents; +using Worker.Consumers; +using Worker.Services; + +namespace Worker.UnitTests; + +/// +/// Tests for PdfProcessingConsumer retry exhaustion and DLQ behavior. +/// +/// Uses a real DocumentProcessingAgent with mocked dependencies to verify +/// the consumer's behavior when agent processing fails. +/// +public class PdfProcessingConsumerRetryTests +{ + private readonly Mock _checkpointStoreMock; + private readonly Mock _repositoryMock; + private readonly Mock _fileStorageMock; + private readonly Mock> _loggerMock; + private readonly PdfProcessingConsumer _consumer; + + public PdfProcessingConsumerRetryTests() + { + _checkpointStoreMock = new Mock(); + _repositoryMock = new Mock(); + _fileStorageMock = new Mock(); + _loggerMock = new Mock>(); + + // Create a real DocumentProcessingAgent with mocked dependencies + var extractorLoggerMock = new Mock>(); + var ocrServiceMock = Mock.Of(); + var textExtractor = new PdfTextExtractor(extractorLoggerMock.Object, ocrServiceMock); + + var agentLoggerMock = new Mock>(); + var agent = new DocumentProcessingAgent( + textExtractor, + _repositoryMock.Object, + _fileStorageMock.Object, + agentLoggerMock.Object); + + _consumer = new PdfProcessingConsumer( + agent, + _checkpointStoreMock.Object, + _repositoryMock.Object, + _loggerMock.Object); + } + + [Fact] + public async Task Consume_AgentFailure_RethrowsOriginalException() + { + // Arrange + var documentId = Guid.NewGuid(); + var command = new PdfProcessingCommand + { + DocumentId = documentId, + MessageId = Guid.NewGuid(), + FilePath = "/test.pdf", + RetryCount = 0 + }; + + var consumeContextMock = new Mock>(); + consumeContextMock.Setup(x => x.Message).Returns(command); + consumeContextMock.Setup(x => x.CancellationToken).Returns(CancellationToken.None); + + // No existing checkpoints + _checkpointStoreMock + .Setup(x => x.LoadCompletedCheckpointsAsync( + "DocumentProcessing", documentId, It.IsAny())) + .ReturnsAsync(Array.Empty()); + + // File storage throws — triggers agent failure, which is re-thrown by consumer + _fileStorageMock + .Setup(x => x.GetAsync("/test.pdf", It.IsAny())) + .ThrowsAsync(new IOException("Storage unavailable")); + + // Message not already processed + _repositoryMock + .Setup(x => x.IsMessageProcessedAsync(command.MessageId, It.IsAny())) + .ReturnsAsync(false); + + // Act & Assert: consumer re-throws the original IOException from the agent + var ex = await Assert.ThrowsAsync( + () => _consumer.Consume(consumeContextMock.Object)); + + Assert.Contains("Storage unavailable", ex.Message); + } + + [Fact] + public async Task Consume_AgentFailure_SetsDocumentStatusToFailed() + { + // Arrange + var documentId = Guid.NewGuid(); + var command = new PdfProcessingCommand + { + DocumentId = documentId, + MessageId = Guid.NewGuid(), + FilePath = "/test.pdf", + RetryCount = 0 + }; + + var consumeContextMock = new Mock>(); + consumeContextMock.Setup(x => x.Message).Returns(command); + consumeContextMock.Setup(x => x.CancellationToken).Returns(CancellationToken.None); + + _checkpointStoreMock + .Setup(x => x.LoadCompletedCheckpointsAsync( + "DocumentProcessing", documentId, It.IsAny())) + .ReturnsAsync(Array.Empty()); + + _fileStorageMock + .Setup(x => x.GetAsync("/test.pdf", It.IsAny())) + .ThrowsAsync(new IOException("Storage unavailable")); + + _repositoryMock + .Setup(x => x.IsMessageProcessedAsync(command.MessageId, It.IsAny())) + .ReturnsAsync(false); + + _repositoryMock + .Setup(x => x.TryUpdateStatusAsync( + documentId, + DocumentStatus.Processing, + DocumentStatus.Failed, + It.IsAny(), + It.IsAny())) + .ReturnsAsync(true); + + // Act — catch the expected exception, then verify status update + try + { + await _consumer.Consume(consumeContextMock.Object); + } + catch (IOException) + { + // Expected + } + + // Assert: status was updated to Failed + _repositoryMock.Verify(x => x.TryUpdateStatusAsync( + documentId, + DocumentStatus.Processing, + DocumentStatus.Failed, + It.Is(msg => msg != null && msg.Contains("Storage unavailable")), + It.IsAny()), Times.Once); + } + + [Fact] + public async Task Consume_IdempotencyCheck_SkipsDuplicateMessage() + { + // Arrange + var documentId = Guid.NewGuid(); + var messageId = Guid.NewGuid(); + var command = new PdfProcessingCommand + { + DocumentId = documentId, + MessageId = messageId, + FilePath = "/test.pdf", + RetryCount = 1 // This is a retry — message was already processed + }; + + var consumeContextMock = new Mock>(); + consumeContextMock.Setup(x => x.Message).Returns(command); + consumeContextMock.Setup(x => x.CancellationToken).Returns(CancellationToken.None); + + // Message already processed (idempotency) + _repositoryMock + .Setup(x => x.IsMessageProcessedAsync(messageId, It.IsAny())) + .ReturnsAsync(true); + + // Act + await _consumer.Consume(consumeContextMock.Object); + + // Verify: IsMessageProcessedAsync was called, but file storage was never accessed + // (processing skipped entirely) + _fileStorageMock.Verify(x => x.GetAsync( + It.IsAny(), It.IsAny()), Times.Never); + } +} \ No newline at end of file diff --git a/tests/Worker.UnitTests/Worker.UnitTests.csproj b/tests/Worker.UnitTests/Worker.UnitTests.csproj index 0b995f9..ba3c616 100644 --- a/tests/Worker.UnitTests/Worker.UnitTests.csproj +++ b/tests/Worker.UnitTests/Worker.UnitTests.csproj @@ -1,7 +1,7 @@ - net8.0 + net10.0 enable enable @@ -22,8 +22,8 @@ - - + +