Compare commits

...

9 Commits

Author SHA1 Message Date
8f2365d048 remove Features-directory and move the files to the root directory 2026-07-15 15:03:12 +02:00
bfe24eba06 feat(api): Add REST API controllers with RabbitMQ for POST/PUT/DELETE
Controllers:
- EmailProfilesController: CRUD operations (GET sync, POST/PUT/DELETE async via RabbitMQ)
  * GET /api/emailprofiles - List all profiles
  * GET /api/emailprofiles/{id} - Get profile by ID
  * GET /api/emailprofiles/active - List active profiles
  * POST /api/emailprofiles - Create (202 Accepted, queued to RabbitMQ)
  * PUT /api/emailprofiles/{id} - Update (202 Accepted, queued to RabbitMQ)
  * DELETE /api/emailprofiles/{id} - Delete (202 Accepted, queued to RabbitMQ)

- EmailAccountsController: CRUD operations
  * GET /api/emailaccounts - List all accounts
  * GET /api/emailaccounts/{id} - Get account by ID
  * POST /api/emailaccounts - Create (202 Accepted, queued to RabbitMQ)

- EmailHistoryController: Read-only operations
  * GET /api/emailhistory/profile/{profileId} - Get history with pagination
  * GET /api/emailhistory/{id} - Get history by ID

Changes:
- Fix ICommandPublisher constraint: IRequest → IBaseRequest (supports IRequest<T>)
- All POST/PUT/DELETE return HTTP 202 Accepted (async processing)
- All GET operations return HTTP 200 OK (synchronous via MediatR)
- Proper error handling: 404 Not Found for missing resources
2026-07-14 16:39:35 +02:00
0d22fe0b5c docs: Update AGENTS.md and STATUS.md with RabbitMQ and Phase 2 completion
AGENTS.md:
- Add Section 7: RabbitMQ Command Bus Integration (IMPLEMENTED)
- Document ICommandPublisher, RabbitMqCommandPublisher, RabbitMqCommandConsumer
- Add configuration, DI setup, and usage examples
- Document benefits: async processing, horizontal scaling, retries, persistence

STATUS.md:
- Mark Phase 2 (Application Layer) as 100% complete
- Update Phase 3 (Infrastructure Layer) to 15% (RabbitMQ done)
- Document all completed components: DTOs, Commands, Queries, Validators, Mappings
- Update last modified date to 2026-07-14
2026-07-14 16:37:22 +02:00
1ed489532d feat(application): Add AutoMapper profiles for all entities
- EmailProfileMappingProfile: Command→Entity, DTO→Entity, Entity→DTO
- EmailAccountMappingProfile: Command→Entity, Entity→DTO
- EmailHistoryMappingProfile: CreateDto→Entity, UpdateDto partial mapping, Entity→DTO
- EmailAttachmentMappingProfile: CreateDto→Entity, UpdateDto partial mapping, Entity→DTO

All mappings follow Repository<T> pattern with AutoMapper-based CRUD
2026-07-14 16:37:11 +02:00
eda6257145 feat(application): Add MediatR Commands, Queries, and FluentValidation
Commands (5):
- CreateEmailProfileCommand, UpdateEmailProfileCommand, DeleteEmailProfileCommand
- CreateEmailAccountCommand (OAuth2/password conditional validation)
- ProcessEmailCommand (with CreateEmailHistoryDto, UpdateEmailHistoryStatusDto)

Queries (7):
- GetEmailProfilesQuery, GetEmailProfileByIdQuery, GetActiveEmailProfilesQuery
- GetEmailAccountsQuery, GetEmailAccountByIdQuery
- GetEmailHistoryByProfileQuery (with pagination), GetEmailHistoryByIdQuery

Validators (4):
- CreateEmailProfileCommandValidator, UpdateEmailProfileCommandValidator
- CreateEmailAccountCommandValidator, ProcessEmailCommandValidator

All handlers in same file as commands/queries (AGENTS.md rule #5)
2026-07-14 16:37:00 +02:00
a708799587 refactor(application): Reorganize DTOs - flatten single-DTO folders
- Move EmailAccountDto.cs to Dtos/ (was in EmailAccounts/ subfolder)
- Move EmailProfileDto.cs to Dtos/ (was in EmailProfiles/ subfolder)
- Keep EmailAttachments/ (3 DTOs) and EmailHistories/ (3 DTOs) subfolders
- Update all namespace imports from Dtos.EmailAccounts/EmailProfiles to Dtos
- Simpler structure: single DTOs at root, multiple DTOs in subfolders
2026-07-14 16:36:50 +02:00
b7d65d7d5c refactor(application): Remove IUnitOfWork, add generic IRepository<T> with AutoMapper
- Remove IUnitOfWork pattern (not needed with auto-save repositories)
- Add generic IRepository<T> with CreateAsync<TDto>, UpdateSingleAsync<TDto>, DeleteSingleAsync
- Add UpdateAsync/DeleteAsync for bulk operations
- Add ICommandPublisher interface for RabbitMQ integration
- Add service interfaces: IEmailService, IPdfProcessingService, IDmsService, IEncryptionService, IEmailQueue
- Configure Application DI with MediatR, AutoMapper, FluentValidation
2026-07-14 16:36:41 +02:00
5e8e6a06fe feat(infrastructure): Add RabbitMQ Command Bus integration
- Add RabbitMQ.Client 7.2.1, Microsoft.Extensions.Hosting 10.0.9
- Implement ICommandPublisher interface for async command publishing
- Create RabbitMqCommandPublisher with persistent message delivery
- Create RabbitMqCommandConsumer BackgroundService for command processing
- Add RabbitMqConfiguration with appsettings.json binding
- Configure Infrastructure DI with RabbitMQ services
- Update Program.cs to support appsettings.Secrets.json
- Server: 172.24.12.56:5672, Exchange: emailprofiler.commands
2026-07-14 16:36:23 +02:00
50c21ee628 Refactor MediatR commands and update solution structure
- Consolidated commands and handlers into single files for better organization.
- Updated file naming conventions for commands and queries.
- Added explicit Git operation rules to prevent automatic commits/pushes.
- Introduced new projects and restructured solution file (`legacy` folder).
- Refactored `CreateEmailAccountCommand`, `ProcessEmailCommand`, and others to use `IUnitOfWork`.
- Enhanced `ProcessEmailCommandHandler` with attachment validation and error handling.
- Removed redundant handler files after consolidation.
- Improved code consistency and added `TODO` comments for future enhancements.
2026-07-09 14:37:12 +02:00
56 changed files with 2038 additions and 459 deletions

1
.gitignore vendored
View File

@@ -370,3 +370,4 @@ FodyWeavers.xsd
/EnvelopeGenerator.Server/EnvelopeGenerator.Server/publish-output
/EnvelopeGenerator.Server/EnvelopeGenerator.Server/tekh_softHSM_test.md
/legacy/App
/src/DigitalData.EmailProfiler.API/appsettings.Secrets.json

View File

@@ -55,21 +55,227 @@ var lastPoll = DateTime.UtcNow.AddMinutes(-profile.PollIntervalMinutes); // NEV
- All timestamps in logs and error messages
- All date parameters in queries
### 4. No Commits Without Permission
**NEVER** commit changes to git automatically. Always wait for explicit user instruction to commit.
### 4. Git Operations - NEVER Without Explicit Permission
**CRITICAL**: NEVER execute `git commit` or `git push` commands automatically. ALWAYS wait for explicit user instruction.
**Rules**:
- Only commit when user explicitly says "commit" or "commit this"
- Only push when user explicitly says "push" or "push to remote"
- Stage files with `git add` ONLY when about to commit per user request
### 5. MediatR Command/Query File Organization
**IMPORTANT**: Commands/Queries and their Handlers must be in the SAME file.
**Example**:
```csharp
// ✅ CORRECT - CreateEmailProfileCommand.cs contains BOTH
public record CreateEmailProfileCommand : IRequest<int> { ... }
public class CreateEmailProfileCommandHandler : IRequestHandler<CreateEmailProfileCommand, int> { ... }
// ❌ WRONG - Separate files
// CreateEmailProfileCommand.cs (command only)
// CreateEmailProfileCommandHandler.cs (handler only)
```
**File Naming**:
- Commands: `{Verb}{Entity}Command.cs` (e.g., `CreateEmailProfileCommand.cs`)
- Queries: `{Verb}{Entity}Query.cs` (e.g., `GetEmailProfilesQuery.cs`)
### 6. Repository Pattern - NO UnitOfWork, Generic CRUD with AutoMapper
**CRITICAL**: DO NOT use IUnitOfWork pattern. Use generic repository pattern with AutoMapper-based CRUD operations.
**Key Principles**:
- ✅ Each operation auto-saves changes - NO explicit SaveChangesAsync needed
- ✅ Use `UpdateSingleAsync` / `DeleteSingleAsync` for single-record safety
- ✅ Use `UpdateAsync` / `DeleteAsync` only when intentionally modifying multiple records
- ✅ AutoMapper handles all DTO → Entity mappings
**Pattern**:
```csharp
// IRepository<T> generic interface
public interface IRepository<TEntity> where TEntity : class
{
// Query operations
Task<TEntity?> GetByIdAsync(int id, CancellationToken cancellationToken = default);
Task<IEnumerable<TEntity>> FindAsync(Expression<Func<TEntity, bool>> predicate, ...);
// Create - auto-saves
Task<TEntity> CreateAsync<TDto>(TDto dto, CancellationToken cancellationToken = default);
// Update - auto-saves
Task<int> UpdateAsync<TDto>(Expression<...> predicate, TDto dto, ...); // Multiple records
Task UpdateSingleAsync<TDto>(Expression<...> predicate, TDto dto, ...); // SAFE: Single record only
// Delete - auto-saves
Task<int> DeleteAsync(Expression<...> predicate, ...); // Multiple records
Task DeleteSingleAsync(Expression<...> predicate, ...); // SAFE: Single record only
}
```
**Command Handler Examples**:
```csharp
// ✅ CORRECT - CreateAsync auto-saves
public class CreateEmailProfileCommandHandler(IRepository<EmailProfile> repository)
: IRequestHandler<CreateEmailProfileCommand, int>
{
public async Task<int> Handle(CreateEmailProfileCommand request, CancellationToken cancellationToken)
{
var profile = await repository.CreateAsync(request, cancellationToken);
return profile.Id; // NO SaveChangesAsync needed!
}
}
// ✅ CORRECT - UpdateSingleAsync for safety (throws if 0 or 2+ records match)
public class UpdateEmailProfileCommandHandler(IRepository<EmailProfile> repository)
: IRequestHandler<UpdateEmailProfileCommand, int>
{
public async Task<int> Handle(UpdateEmailProfileCommand request, CancellationToken cancellationToken)
{
await repository.UpdateSingleAsync(p => p.Id == request.Id, request, cancellationToken);
return request.Id; // NO SaveChangesAsync needed!
}
}
// ✅ CORRECT - DeleteSingleAsync for safety (throws if 0 or 2+ records match)
public class DeleteEmailProfileCommandHandler(IRepository<EmailProfile> repository)
: IRequestHandler<DeleteEmailProfileCommand, int>
{
public async Task<int> Handle(DeleteEmailProfileCommand request, CancellationToken cancellationToken)
{
await repository.DeleteSingleAsync(p => p.Id == request.Id, cancellationToken);
return request.Id; // NO SaveChangesAsync needed!
}
}
// ❌ WRONG - Manual entity creation (use AutoMapper instead)
var profile = new EmailProfile
{
ProfileName = request.ProfileName,
EmailAccountId = request.EmailAccountId,
// ... 15 more properties
};
// ❌ WRONG - Using IUnitOfWork (removed)
public CreateEmailProfileCommandHandler(IUnitOfWork unitOfWork) { ... }
// ❌ WRONG - Calling SaveChangesAsync (removed)
await repository.SaveChangesAsync(cancellationToken);
```
**Safety Rules**:
1. **UpdateSingleAsync** - Use for ID-based updates. Throws `InvalidOperationException` if:
- Zero records match (entity not found)
- Multiple records match (predicate too broad)
2. **DeleteSingleAsync** - Use for ID-based deletes. Throws `InvalidOperationException` if:
- Zero records match (entity not found)
- Multiple records match (predicate too broad)
3. **UpdateAsync / DeleteAsync** - Use ONLY when intentionally modifying multiple records:
```csharp
// ✅ CORRECT - Intentional bulk operation
await repository.UpdateAsync(
p => p.EmailAccountId == accountId,
new { IsActive = false },
cancellationToken);
// ✅ Returns count of updated/deleted records
var count = await repository.DeleteAsync(p => p.IsActive == false, cancellationToken);
```
**DTO Mapping Responsibility**:
- Each DTO creator must define their own AutoMapper profile
- Example: `CreateEmailProfileCommand``EmailProfile` mapping must be defined in `EmailProfileMappingProfile.cs`
- Repository implementation uses `IMapper.Map<TEntity>(dto)` internally
---
## Future Enhancements
### HIGH PRIORITY: RabbitMQ Queue Implementation
### 7. RabbitMQ Command Bus Integration (IMPLEMENTED)
**Current State**:
- Email queue is implemented using in-memory `Channel<T>` in `InMemoryEmailQueue.cs`
- Location: `src/DigitalData.EmailProfiler.Infrastructure/Queue/InMemoryEmailQueue.cs`
**Purpose**: Asynchronous command processing via RabbitMQ message broker for POST/PUT/DELETE operations.
**Future Enhancement**:
Replace the in-memory queue with **RabbitMQ** for production resilience and scalability.
**Architecture**:
- **GET Queries**: Synchronous (immediate response via MediatR)
- **POST/PUT/DELETE Commands**: Can be asynchronous (published to RabbitMQ, processed by background worker)
**RabbitMQ Server**:
- Management UI: `http://172.24.12.56:15672`
- AMQP Port: `5672` (default)
- Exchange: `emailprofiler.commands` (Direct)
- Queue: `emailprofiler.command.queue`
- Routing Key: `command`
**Implementation Components**:
1. **ICommandPublisher** (`Application/Common/Interfaces/ICommandPublisher.cs`):
- Interface for publishing commands to message broker
- Generic method: `PublishAsync<TCommand>(TCommand command, CancellationToken)`
2. **RabbitMqCommandPublisher** (`Infrastructure/Messaging/RabbitMqCommandPublisher.cs`):
- Implements `ICommandPublisher`
- Serializes command to JSON with metadata envelope (CommandType, Payload, CorrelationId, PublishedAt)
- Publishes to RabbitMQ exchange with persistent delivery mode
3. **RabbitMqCommandConsumer** (`Infrastructure/Messaging/RabbitMqCommandConsumer.cs`):
- BackgroundService that consumes commands from RabbitMQ
- Deserializes command envelope
- Resolves command type from assembly
- Executes command via MediatR in scoped service
- Acknowledges message on success, requeues on error
4. **RabbitMqConfiguration** (`Infrastructure/Messaging/RabbitMqConfiguration.cs`):
- Configuration model for RabbitMQ connection
- Binds to `appsettings.json` section: `RabbitMq`
**Configuration** (`appsettings.json`):
```json
{
"RabbitMq": {
"HostName": "172.24.12.56",
"Port": 5672,
"UserName": "guest",
"Password": "guest",
"VirtualHost": "/",
"ExchangeName": "emailprofiler.commands",
"QueueName": "emailprofiler.command.queue",
"RoutingKey": "command",
"AutomaticRecoveryEnabled": true,
"NetworkRecoveryIntervalSeconds": 10
}
}
```
**Dependency Injection** (`Infrastructure/DependencyInjection.cs`):
```csharp
services.Configure<RabbitMqConfiguration>(configuration.GetSection(RabbitMqConfiguration.SectionName));
services.AddSingleton<ICommandPublisher, RabbitMqCommandPublisher>();
services.AddHostedService<RabbitMqCommandConsumer>();
```
**Usage in API Controllers** (Future):
```csharp
// Option 1: Synchronous (immediate execution via MediatR)
var result = await _mediator.Send(new CreateEmailProfileCommand(...), cancellationToken);
return Ok(result);
// Option 2: Asynchronous (publish to RabbitMQ for background processing)
await _commandPublisher.PublishAsync(new CreateEmailProfileCommand(...), cancellationToken);
return Accepted(); // HTTP 202 - command queued for processing
```
**Benefits**:
- Decouples API from long-running command processing
- Improves API responsiveness (fire-and-forget)
- Enables horizontal scaling (multiple consumers)
- Automatic retries on failure (requeue mechanism)
- Message persistence (survives application restarts)
---
### HIGH PRIORITY: Email Queue for Outgoing Messages (Future)
**Implementation Steps**:

View File

@@ -25,6 +25,12 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Solution Items", "Solution
STATUS.md = STATUS.md
EndProjectSection
EndProject
Project("{F184B08F-C81C-45F6-A57F-5ABD9991F28F}") = "EmailProfiler.Common", "legacy\App\EmailProfiler.Common\EmailProfiler.Common.vbproj", "{9F748DCD-952E-40A0-9DAD-65BF8A39B231}"
EndProject
Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "legacy", "legacy", "{EAFC1552-2C62-4C00-AE27-47D76FEAE9F5}"
EndProject
Project("{F184B08F-C81C-45F6-A57F-5ABD9991F28F}") = "EmailProfiler.Service", "legacy\App\EmailProfiler.Service\EmailProfiler.Service.vbproj", "{1F3C569B-91DA-427F-8D81-BBCC556B11A4}"
EndProject
Global
GlobalSection(SolutionConfigurationPlatforms) = preSolution
Debug|Any CPU = Debug|Any CPU
@@ -51,6 +57,14 @@ Global
{211FB65F-2406-474E-A426-DA246B250AB8}.Debug|Any CPU.Build.0 = Debug|Any CPU
{211FB65F-2406-474E-A426-DA246B250AB8}.Release|Any CPU.ActiveCfg = Release|Any CPU
{211FB65F-2406-474E-A426-DA246B250AB8}.Release|Any CPU.Build.0 = Release|Any CPU
{9F748DCD-952E-40A0-9DAD-65BF8A39B231}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{9F748DCD-952E-40A0-9DAD-65BF8A39B231}.Debug|Any CPU.Build.0 = Debug|Any CPU
{9F748DCD-952E-40A0-9DAD-65BF8A39B231}.Release|Any CPU.ActiveCfg = Release|Any CPU
{9F748DCD-952E-40A0-9DAD-65BF8A39B231}.Release|Any CPU.Build.0 = Release|Any CPU
{1F3C569B-91DA-427F-8D81-BBCC556B11A4}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{1F3C569B-91DA-427F-8D81-BBCC556B11A4}.Debug|Any CPU.Build.0 = Debug|Any CPU
{1F3C569B-91DA-427F-8D81-BBCC556B11A4}.Release|Any CPU.ActiveCfg = Release|Any CPU
{1F3C569B-91DA-427F-8D81-BBCC556B11A4}.Release|Any CPU.Build.0 = Release|Any CPU
EndGlobalSection
GlobalSection(SolutionProperties) = preSolution
HideSolutionNode = FALSE
@@ -61,6 +75,8 @@ Global
{76ADC1D0-4DFA-0B1E-57C9-2636434A0043} = {02EA681E-C7D8-13C7-8484-4AC65E1B71E8}
{1874A827-C6A5-EB5E-0FE9-30A7200382B7} = {02EA681E-C7D8-13C7-8484-4AC65E1B71E8}
{211FB65F-2406-474E-A426-DA246B250AB8} = {4F20FEFD-9289-42C6-ABA6-8DB236D74559}
{9F748DCD-952E-40A0-9DAD-65BF8A39B231} = {EAFC1552-2C62-4C00-AE27-47D76FEAE9F5}
{1F3C569B-91DA-427F-8D81-BBCC556B11A4} = {EAFC1552-2C62-4C00-AE27-47D76FEAE9F5}
EndGlobalSection
GlobalSection(ExtensibilityGlobals) = postSolution
SolutionGuid = {90E29FDC-F6C6-414F-94BF-25DF61D18060}

105
STATUS.md
View File

@@ -1,6 +1,6 @@
# EmailProfiler - Current Implementation Status
**Last Updated**: 2026-07-07
**Last Updated**: 2026-07-14
---
@@ -48,10 +48,55 @@
---
## 🚧 IN PROGRESS (Phase 2: Application Layer - 5%)
## ✅ COMPLETED (Phase 2: Application Layer - 100%)
### DTOs
-`CommonDtos.cs` - EmailProfileDto, EmailAccountDto, EmailHistoryDto, EmailAttachmentDto
### DTOs (Common/Dtos/{Entity}/)
-`EmailProfiles/EmailProfileDto.cs`
-`EmailAccounts/EmailAccountDto.cs`
-`EmailHistories/EmailHistoryDto.cs`, `CreateEmailHistoryDto.cs`, `UpdateEmailHistoryStatusDto.cs`
-`EmailAttachments/EmailAttachmentDto.cs`, `CreateEmailAttachmentDto.cs`, `UpdateEmailAttachmentStatusDto.cs`
### Repository Interfaces (Generic Pattern - NO UnitOfWork)
-`IRepository<T>` - Generic repository with CreateAsync<TDto>, UpdateSingleAsync<TDto>, DeleteSingleAsync, UpdateAsync, DeleteAsync
### Service Interfaces
-`IEmailService.cs` - IMAP/SMTP operations
-`IPdfProcessingService.cs` - PDF validation and extraction
-`IDmsService.cs` - windream DMS integration
-`IEncryptionService.cs` - Password encryption
-`IEmailQueue.cs` - Email queue operations
-`ICommandPublisher.cs` - RabbitMQ command publishing
### MediatR Commands (Features/*/Commands/)
-`CreateEmailProfileCommand.cs` + Handler
-`UpdateEmailProfileCommand.cs` + Handler
-`DeleteEmailProfileCommand.cs` + Handler
-`CreateEmailAccountCommand.cs` + Handler
-`ProcessEmailCommand.cs` + Handler
### MediatR Queries (Features/*/Queries/)
-`GetEmailProfilesQuery.cs` + Handler
-`GetEmailProfileByIdQuery.cs` + Handler
-`GetActiveEmailProfilesQuery.cs` + Handler
-`GetEmailAccountsQuery.cs` + Handler
-`GetEmailAccountByIdQuery.cs` + Handler
-`GetEmailHistoryByProfileQuery.cs` + Handler (with pagination)
-`GetEmailHistoryByIdQuery.cs` + Handler
### Validators (Features/*/Validators/)
-`CreateEmailProfileCommandValidator.cs`
-`UpdateEmailProfileCommandValidator.cs`
-`CreateEmailAccountCommandValidator.cs` (conditional OAuth2/password validation)
-`ProcessEmailCommandValidator.cs` (with attachment validation)
### AutoMapper Profiles (Common/Mappings/)
-`EmailProfileMappingProfile.cs` (Command→Entity, DTO→Entity, Entity→DTO)
-`EmailAccountMappingProfile.cs`
-`EmailHistoryMappingProfile.cs`
-`EmailAttachmentMappingProfile.cs`
### DI Configuration
-`DependencyInjection.cs` - Registers MediatR, AutoMapper, FluentValidation
### NuGet Packages
- ✅ Application project has:
@@ -61,51 +106,23 @@
---
## ❌ TODO (Phase 2: Application Layer - 95%)
## 🚧 IN PROGRESS (Phase 3: Infrastructure Layer - 15%)
### Repository Interfaces
- `IEmailProfileRepository.cs`
- `IEmailAccountRepository.cs`
- `IEmailHistoryRepository.cs`
- `IEmailProcessRepository.cs`
- `IEmailOutboxRepository.cs`
### RabbitMQ Command Bus (COMPLETED)
- `RabbitMqConfiguration.cs` - Configuration model
- `RabbitMqCommandPublisher.cs` - ICommandPublisher implementation
- `RabbitMqCommandConsumer.cs` - BackgroundService for command consumption
- `DependencyInjection.cs` - Infrastructure DI with RabbitMQ registration
- ✅ Configuration in `appsettings.json` (Server: 172.24.12.56:5672)
### Service Interfaces
- `IEmailService.cs`
- `IPdfProcessingService.cs`
- `IDmsService.cs`
-`IEncryptionService.cs`
-`IEmailQueue.cs`
### MediatR Commands
-`CreateEmailProfileCommand.cs`
-`UpdateEmailProfileCommand.cs`
-`DeleteEmailProfileCommand.cs`
-`ActivateProfileCommand.cs`
-`DeactivateProfileCommand.cs`
- ❌ Similar commands for EmailAccount, EmailProcess, etc.
### MediatR Queries
-`GetEmailProfilesQuery.cs`
-`GetEmailProfileByIdQuery.cs`
-`GetActiveProfilesQuery.cs`
-`GetProfilesDueForPollingQuery.cs`
- ❌ Similar queries for other entities
### Validators
-`CreateEmailProfileCommandValidator.cs`
-`UpdateEmailProfileCommandValidator.cs`
- ❌ Similar validators for all commands
### Mappings
-`MappingProfile.cs` - AutoMapper configuration
### DI Configuration
-`DependencyInjection.cs` - Application layer DI setup
### NuGet Packages (Partial)
- ✅ RabbitMQ.Client 7.2.1
- ✅ Microsoft.Extensions.Hosting 10.0.9
- ✅ Microsoft.Extensions.Options.ConfigurationExtensions 10.0.9
---
## ❌ TODO (Phase 3: Infrastructure Layer - 0%)
## ❌ TODO (Phase 3: Infrastructure Layer - 85%)
### NuGet Packages
- ❌ Microsoft.EntityFrameworkCore.SqlServer

View File

@@ -0,0 +1,70 @@
using DigitalData.EmailProfiler.Application.Common.Dtos;
using DigitalData.EmailProfiler.Application.Common.Interfaces;
using DigitalData.EmailProfiler.Application.EmailAccounts.Commands;
using DigitalData.EmailProfiler.Application.EmailAccounts.Queries;
using MediatR;
using Microsoft.AspNetCore.Mvc;
namespace DigitalData.EmailProfiler.API.Controllers;
/// <summary>
/// Email accounts management API controller
/// </summary>
[ApiController]
[Route("api/[controller]")]
public class EmailAccountsController(
IMediator mediator,
ICommandPublisher commandPublisher) : ControllerBase
{
/// <summary>
/// Get all email accounts
/// </summary>
[HttpGet]
[ProducesResponseType(typeof(IEnumerable<EmailAccountDto>), StatusCodes.Status200OK)]
public async Task<IActionResult> GetAll(CancellationToken cancellationToken)
{
var query = new GetEmailAccountsQuery();
var result = await mediator.Send(query, cancellationToken);
return Ok(result);
}
/// <summary>
/// Get email account by ID
/// </summary>
[HttpGet("{id:int}")]
[ProducesResponseType(typeof(EmailAccountDto), StatusCodes.Status200OK)]
[ProducesResponseType(StatusCodes.Status404NotFound)]
public async Task<IActionResult> GetById(int id, CancellationToken cancellationToken)
{
var query = new GetEmailAccountByIdQuery(id);
var result = await mediator.Send(query, cancellationToken);
if (result == null)
return NotFound(new { Message = $"Email account with ID {id} not found" });
return Ok(result);
}
/// <summary>
/// Create new email account (async via RabbitMQ)
/// </summary>
[HttpPost]
[ProducesResponseType(StatusCodes.Status202Accepted)]
[ProducesResponseType(StatusCodes.Status400BadRequest)]
public async Task<IActionResult> Create(
[FromBody] CreateEmailAccountCommand command,
CancellationToken cancellationToken)
{
// Publish command to RabbitMQ for async processing
await commandPublisher.PublishAsync(command, cancellationToken);
return Accepted(new
{
Message = "Email account creation request queued for processing",
AccountName = command.AccountName
});
}
// Note: Update and Delete operations can be added similarly
// For now, we focus on Create as the main use case
}

View File

@@ -0,0 +1,57 @@
using DigitalData.EmailProfiler.Application.Common.Dtos.EmailHistories;
using DigitalData.EmailProfiler.Application.EmailHistories.Queries;
using MediatR;
using Microsoft.AspNetCore.Mvc;
namespace DigitalData.EmailProfiler.API.Controllers;
/// <summary>
/// Email history API controller (read-only)
/// </summary>
[ApiController]
[Route("api/[controller]")]
public class EmailHistoryController(IMediator mediator) : ControllerBase
{
/// <summary>
/// Get email history by profile ID with pagination
/// </summary>
[HttpGet("profile/{profileId:int}")]
[ProducesResponseType(typeof(IEnumerable<EmailHistoryDto>), StatusCodes.Status200OK)]
public async Task<IActionResult> GetByProfile(
int profileId,
[FromQuery] int pageNumber = 1,
[FromQuery] int pageSize = 50,
CancellationToken cancellationToken = default)
{
var query = new GetEmailHistoryByProfileQuery(profileId, pageNumber, pageSize);
var result = await mediator.Send(query, cancellationToken);
return Ok(new
{
ProfileId = profileId,
PageNumber = pageNumber,
PageSize = pageSize,
Data = result
});
}
/// <summary>
/// Get email history by ID
/// </summary>
[HttpGet("{id:int}")]
[ProducesResponseType(typeof(EmailHistoryDto), StatusCodes.Status200OK)]
[ProducesResponseType(StatusCodes.Status404NotFound)]
public async Task<IActionResult> GetById(int id, CancellationToken cancellationToken)
{
var query = new GetEmailHistoryByIdQuery(id);
var result = await mediator.Send(query, cancellationToken);
if (result == null)
return NotFound(new { Message = $"Email history with ID {id} not found" });
return Ok(result);
}
// Note: Email history is typically managed by ProcessEmailCommand
// No direct CREATE/UPDATE/DELETE endpoints needed
}

View File

@@ -0,0 +1,123 @@
using DigitalData.EmailProfiler.Application.Common.Dtos;
using DigitalData.EmailProfiler.Application.Common.Interfaces;
using DigitalData.EmailProfiler.Application.EmailProfiles.Commands;
using DigitalData.EmailProfiler.Application.EmailProfiles.Queries;
using MediatR;
using Microsoft.AspNetCore.Mvc;
namespace DigitalData.EmailProfiler.API.Controllers;
/// <summary>
/// Email profiles management API controller
/// </summary>
[ApiController]
[Route("api/[controller]")]
public class EmailProfilesController(
IMediator mediator,
ICommandPublisher commandPublisher) : ControllerBase
{
/// <summary>
/// Get all email profiles
/// </summary>
[HttpGet]
[ProducesResponseType(typeof(IEnumerable<EmailProfileDto>), StatusCodes.Status200OK)]
public async Task<IActionResult> GetAll(CancellationToken cancellationToken)
{
var query = new GetEmailProfilesQuery();
var result = await mediator.Send(query, cancellationToken);
return Ok(result);
}
/// <summary>
/// Get email profile by ID
/// </summary>
[HttpGet("{id:int}")]
[ProducesResponseType(typeof(EmailProfileDto), StatusCodes.Status200OK)]
[ProducesResponseType(StatusCodes.Status404NotFound)]
public async Task<IActionResult> GetById(int id, CancellationToken cancellationToken)
{
var query = new GetEmailProfileByIdQuery(id);
var result = await mediator.Send(query, cancellationToken);
if (result == null)
return NotFound(new { Message = $"Email profile with ID {id} not found" });
return Ok(result);
}
/// <summary>
/// Get all active email profiles
/// </summary>
[HttpGet("active")]
[ProducesResponseType(typeof(IEnumerable<EmailProfileDto>), StatusCodes.Status200OK)]
public async Task<IActionResult> GetActive(CancellationToken cancellationToken)
{
var query = new GetActiveEmailProfilesQuery();
var result = await mediator.Send(query, cancellationToken);
return Ok(result);
}
/// <summary>
/// Create new email profile (async via RabbitMQ)
/// </summary>
[HttpPost]
[ProducesResponseType(StatusCodes.Status202Accepted)]
[ProducesResponseType(StatusCodes.Status400BadRequest)]
public async Task<IActionResult> Create(
[FromBody] CreateEmailProfileCommand command,
CancellationToken cancellationToken)
{
// Publish command to RabbitMQ for async processing
await commandPublisher.PublishAsync(command, cancellationToken);
return Accepted(new
{
Message = "Email profile creation request queued for processing",
ProfileName = command.ProfileName
});
}
/// <summary>
/// Update email profile (async via RabbitMQ)
/// </summary>
[HttpPut("{id:int}")]
[ProducesResponseType(StatusCodes.Status202Accepted)]
[ProducesResponseType(StatusCodes.Status400BadRequest)]
public async Task<IActionResult> Update(
int id,
[FromBody] UpdateEmailProfileCommand command,
CancellationToken cancellationToken)
{
// Ensure ID matches route
if (id != command.Id)
return BadRequest(new { Message = "Route ID does not match command ID" });
// Publish command to RabbitMQ for async processing
await commandPublisher.PublishAsync(command, cancellationToken);
return Accepted(new
{
Message = "Email profile update request queued for processing",
Id = command.Id
});
}
/// <summary>
/// Delete email profile (async via RabbitMQ)
/// </summary>
[HttpDelete("{id:int}")]
[ProducesResponseType(StatusCodes.Status202Accepted)]
public async Task<IActionResult> Delete(int id, CancellationToken cancellationToken)
{
var command = new DeleteEmailProfileCommand(id);
// Publish command to RabbitMQ for async processing
await commandPublisher.PublishAsync(command, cancellationToken);
return Accepted(new
{
Message = "Email profile deletion request queued for processing",
Id = id
});
}
}

View File

@@ -9,7 +9,7 @@
<ItemGroup>
<PackageReference Include="Swashbuckle.AspNetCore" Version="6.6.2" />
<PackageReference Include="Microsoft.Extensions.Hosting" Version="8.0.1" />
<PackageReference Include="Microsoft.Extensions.Hosting" Version="10.0.9" />
</ItemGroup>
<ItemGroup>

View File

@@ -1,6 +1,18 @@
using DigitalData.EmailProfiler.API;
using DigitalData.EmailProfiler.Application;
using DigitalData.EmailProfiler.Infrastructure;
var builder = WebApplication.CreateBuilder(args);
// Add appsettings.Secrets.json for sensitive configuration (not committed to git)
builder.Configuration.AddJsonFile("appsettings.Secrets.json", optional: true, reloadOnChange: true);
// Register Application layer (MediatR, AutoMapper, FluentValidation)
builder.Services.AddApplicationServices();
// Register Infrastructure layer (RabbitMQ, Repositories, etc.)
builder.Services.AddInfrastructure(builder.Configuration);
builder.Services.AddHostedService<Worker>();
builder.Services.AddControllers();

View File

@@ -1,54 +0,0 @@
namespace DigitalData.EmailProfiler.Application.Common.Dtos;
public class EmailProfileDto
{
public int Id { get; set; }
public string ProfileName { get; set; } = string.Empty;
public int EmailAccountId { get; set; }
public string? EmailAccountName { get; set; }
public int? ProcessId { get; set; }
public string? ProcessName { get; set; }
public int PollIntervalMinutes { get; set; }
public DateTime? LastPollTime { get; set; }
public bool IsActive { get; set; }
public string? Comment { get; set; }
}
public class EmailAccountDto
{
public int Id { get; set; }
public string AccountName { get; set; } = string.Empty;
public string ImapServer { get; set; } = string.Empty;
public int ImapPort { get; set; }
public string SmtpServer { get; set; } = string.Empty;
public int SmtpPort { get; set; }
public string Username { get; set; } = string.Empty;
public bool UseOAuth2 { get; set; }
public bool IsActive { get; set; }
}
public class EmailHistoryDto
{
public int Id { get; set; }
public int? ProfileId { get; set; }
public string? ProfileName { get; set; }
public string? SenderAddress { get; set; }
public string? Subject { get; set; }
public DateTime? EmailDate { get; set; }
public DateTime? ProcessedDate { get; set; }
public string? Status { get; set; }
public int? ErrorCodeValue { get; set; }
public string? ErrorMessage { get; set; }
public List<EmailAttachmentDto> Attachments { get; set; } = new();
}
public class EmailAttachmentDto
{
public int Id { get; set; }
public string OriginalFileName { get; set; } = string.Empty;
public string FilePath { get; set; } = string.Empty;
public long FileSize { get; set; }
public bool IsEmbeddedFile { get; set; }
public string? Status { get; set; }
public string? ValidationErrorMessage { get; set; }
}

View File

@@ -0,0 +1,17 @@
namespace DigitalData.EmailProfiler.Application.Common.Dtos;
/// <summary>
/// DTO for EmailAccount query results.
/// </summary>
public class EmailAccountDto
{
public int Id { get; set; }
public string AccountName { get; set; } = string.Empty;
public string ImapServer { get; set; } = string.Empty;
public int ImapPort { get; set; }
public string SmtpServer { get; set; } = string.Empty;
public int SmtpPort { get; set; }
public string Username { get; set; } = string.Empty;
public bool UseOAuth2 { get; set; }
public bool IsActive { get; set; }
}

View File

@@ -0,0 +1,15 @@
namespace DigitalData.EmailProfiler.Application.Common.Dtos.EmailAttachments;
/// <summary>
/// DTO for creating EmailAttachment record.
/// AutoMapper profile required: CreateEmailAttachmentDto -> EmailAttachment
/// </summary>
public record CreateEmailAttachmentDto(
int EmailHistoryId,
string OriginalFileName,
string SavedFileName,
string FilePath,
long FileSize,
string Extension,
string ContentType,
byte[] Content);

View File

@@ -0,0 +1,15 @@
namespace DigitalData.EmailProfiler.Application.Common.Dtos.EmailAttachments;
/// <summary>
/// DTO for EmailAttachment query results.
/// </summary>
public class EmailAttachmentDto
{
public int Id { get; set; }
public string OriginalFileName { get; set; } = string.Empty;
public string FilePath { get; set; } = string.Empty;
public long FileSize { get; set; }
public bool IsEmbeddedFile { get; set; }
public string? Status { get; set; }
public string? ValidationErrorMessage { get; set; }
}

View File

@@ -0,0 +1,10 @@
namespace DigitalData.EmailProfiler.Application.Common.Dtos.EmailAttachments;
/// <summary>
/// DTO for updating EmailAttachment status.
/// AutoMapper profile required: UpdateEmailAttachmentStatusDto -> EmailAttachment
/// </summary>
public record UpdateEmailAttachmentStatusDto(
string Status,
int? ValidationErrorCode = null,
string? ValidationErrorMessage = null);

View File

@@ -0,0 +1,15 @@
namespace DigitalData.EmailProfiler.Application.Common.Dtos.EmailHistories;
/// <summary>
/// DTO for creating EmailHistory record.
/// AutoMapper profile required: CreateEmailHistoryDto -> EmailHistory
/// </summary>
public record CreateEmailHistoryDto(
int ProfileId,
string MessageIdHash,
string OriginalMessageId,
string SenderAddress,
DateTime EmailDate,
string Subject,
string? EmailBodyText,
string? EmailBodyHtml);

View File

@@ -0,0 +1,21 @@
using DigitalData.EmailProfiler.Application.Common.Dtos.EmailAttachments;
namespace DigitalData.EmailProfiler.Application.Common.Dtos.EmailHistories;
/// <summary>
/// DTO for EmailHistory query results.
/// </summary>
public class EmailHistoryDto
{
public int Id { get; set; }
public int? ProfileId { get; set; }
public string? ProfileName { get; set; }
public string? SenderAddress { get; set; }
public string? Subject { get; set; }
public DateTime? EmailDate { get; set; }
public DateTime? ProcessedDate { get; set; }
public string? Status { get; set; }
public int? ErrorCodeValue { get; set; }
public string? ErrorMessage { get; set; }
public List<EmailAttachmentDto> Attachments { get; set; } = new();
}

View File

@@ -0,0 +1,11 @@
namespace DigitalData.EmailProfiler.Application.Common.Dtos.EmailHistories;
/// <summary>
/// DTO for updating EmailHistory status.
/// AutoMapper profile required: UpdateEmailHistoryStatusDto -> EmailHistory
/// </summary>
public record UpdateEmailHistoryStatusDto(
string Status,
DateTime? ProcessedDate = null,
int? ErrorCodeValue = null,
string? ErrorMessage = null);

View File

@@ -0,0 +1,18 @@
namespace DigitalData.EmailProfiler.Application.Common.Dtos;
/// <summary>
/// DTO for EmailProfile query results.
/// </summary>
public class EmailProfileDto
{
public int Id { get; set; }
public string ProfileName { get; set; } = string.Empty;
public int EmailAccountId { get; set; }
public string? EmailAccountName { get; set; }
public int? ProcessId { get; set; }
public string? ProcessName { get; set; }
public int PollIntervalMinutes { get; set; }
public DateTime? LastPollTime { get; set; }
public bool IsActive { get; set; }
public string? Comment { get; set; }
}

View File

@@ -0,0 +1,19 @@
using MediatR;
namespace DigitalData.EmailProfiler.Application.Common.Interfaces;
/// <summary>
/// Interface for publishing commands to a message broker (e.g., RabbitMQ)
/// </summary>
public interface ICommandPublisher
{
/// <summary>
/// Publishes a command to the message broker for asynchronous processing
/// </summary>
/// <typeparam name="TCommand">The command type (must implement IBaseRequest - covers both IRequest and IRequest<T>)</typeparam>
/// <param name="command">The command to publish</param>
/// <param name="cancellationToken">Cancellation token</param>
/// <returns>Task representing the publish operation</returns>
Task PublishAsync<TCommand>(TCommand command, CancellationToken cancellationToken = default)
where TCommand : IBaseRequest;
}

View File

@@ -0,0 +1,27 @@
using AutoMapper;
using DigitalData.EmailProfiler.Application.Common.Dtos;
using DigitalData.EmailProfiler.Application.EmailAccounts.Commands;
using DigitalData.EmailProfiler.Domain.Entities;
namespace DigitalData.EmailProfiler.Application.Common.Mappings;
/// <summary>
/// AutoMapper profile for EmailAccount entity mappings.
/// </summary>
public class EmailAccountMappingProfile : Profile
{
public EmailAccountMappingProfile()
{
// Command -> Entity (for Create)
CreateMap<CreateEmailAccountCommand, EmailAccount>()
.ForMember(dest => dest.Id, opt => opt.Ignore()) // Auto-generated
.ForMember(dest => dest.AddedWhen, opt => opt.MapFrom(_ => DateTime.Now))
.ForMember(dest => dest.AddedWho, opt => opt.MapFrom(_ => "System")) // TODO: Get from user context
.ForMember(dest => dest.ChangedWhen, opt => opt.Ignore())
.ForMember(dest => dest.ChangedWho, opt => opt.Ignore())
.ForMember(dest => dest.Profiles, opt => opt.Ignore());
// Entity -> DTO (for Queries)
CreateMap<EmailAccount, EmailAccountDto>();
}
}

View File

@@ -0,0 +1,51 @@
using AutoMapper;
using DigitalData.EmailProfiler.Application.Common.Dtos.EmailAttachments;
using DigitalData.EmailProfiler.Domain.Entities;
using DigitalData.EmailProfiler.Domain.Enums;
namespace DigitalData.EmailProfiler.Application.Common.Mappings;
/// <summary>
/// AutoMapper profile for EmailAttachment entity mappings.
/// </summary>
public class EmailAttachmentMappingProfile : Profile
{
public EmailAttachmentMappingProfile()
{
// DTO -> Entity (for Create)
CreateMap<CreateEmailAttachmentDto, EmailAttachment>()
.ForMember(dest => dest.Id, opt => opt.Ignore()) // Auto-generated
.ForMember(dest => dest.Status, opt => opt.MapFrom(_ => AttachmentStatus.Pending.ToString()))
.ForMember(dest => dest.AddedWhen, opt => opt.MapFrom(_ => DateTime.Now))
.ForMember(dest => dest.IsEmbeddedFile, opt => opt.MapFrom(_ => false))
.ForMember(dest => dest.ParentAttachmentId, opt => opt.Ignore())
.ForMember(dest => dest.AttachmentPosition, opt => opt.MapFrom(_ => 0))
.ForMember(dest => dest.ValidationErrorCode, opt => opt.Ignore())
.ForMember(dest => dest.ValidationErrorMessage, opt => opt.Ignore())
.ForMember(dest => dest.Comment, opt => opt.Ignore())
.ForMember(dest => dest.EmailHistory, opt => opt.Ignore())
.ForMember(dest => dest.ParentAttachment, opt => opt.Ignore())
.ForMember(dest => dest.EmbeddedFiles, opt => opt.Ignore());
// DTO -> Entity (for Update Status)
CreateMap<UpdateEmailAttachmentStatusDto, EmailAttachment>()
.ForMember(dest => dest.Id, opt => opt.Ignore())
.ForMember(dest => dest.EmailHistoryId, opt => opt.Ignore())
.ForMember(dest => dest.OriginalFileName, opt => opt.Ignore())
.ForMember(dest => dest.SavedFileName, opt => opt.Ignore())
.ForMember(dest => dest.FilePath, opt => opt.Ignore())
.ForMember(dest => dest.FileSize, opt => opt.Ignore())
.ForMember(dest => dest.Extension, opt => opt.Ignore())
.ForMember(dest => dest.IsEmbeddedFile, opt => opt.Ignore())
.ForMember(dest => dest.ParentAttachmentId, opt => opt.Ignore())
.ForMember(dest => dest.AttachmentPosition, opt => opt.Ignore())
.ForMember(dest => dest.Comment, opt => opt.Ignore())
.ForMember(dest => dest.AddedWhen, opt => opt.Ignore())
.ForMember(dest => dest.EmailHistory, opt => opt.Ignore())
.ForMember(dest => dest.ParentAttachment, opt => opt.Ignore())
.ForMember(dest => dest.EmbeddedFiles, opt => opt.Ignore());
// Entity -> DTO (for Queries)
CreateMap<EmailAttachment, EmailAttachmentDto>();
}
}

View File

@@ -0,0 +1,59 @@
using AutoMapper;
using DigitalData.EmailProfiler.Application.Common.Dtos.EmailHistories;
using DigitalData.EmailProfiler.Domain.Entities;
using DigitalData.EmailProfiler.Domain.Enums;
namespace DigitalData.EmailProfiler.Application.Common.Mappings;
/// <summary>
/// AutoMapper profile for EmailHistory entity mappings.
/// </summary>
public class EmailHistoryMappingProfile : Profile
{
public EmailHistoryMappingProfile()
{
// DTO -> Entity (for Create)
CreateMap<CreateEmailHistoryDto, EmailHistory>()
.ForMember(dest => dest.Id, opt => opt.Ignore()) // Auto-generated
.ForMember(dest => dest.EmailMessageId, opt => opt.Ignore()) // Auto-generated
.ForMember(dest => dest.Status, opt => opt.MapFrom(_ => EmailStatus.Processing.ToString()))
.ForMember(dest => dest.AddedWhen, opt => opt.MapFrom(_ => DateTime.Now))
.ForMember(dest => dest.ProcessedDate, opt => opt.Ignore())
.ForMember(dest => dest.ErrorCodeValue, opt => opt.Ignore())
.ForMember(dest => dest.ErrorMessage, opt => opt.Ignore())
.ForMember(dest => dest.WindreamDocumentId, opt => opt.Ignore())
.ForMember(dest => dest.Comment, opt => opt.Ignore())
.ForMember(dest => dest.WorkProcess, opt => opt.Ignore())
.ForMember(dest => dest.ImapUid, opt => opt.Ignore())
.ForMember(dest => dest.EmailSubstring1, opt => opt.Ignore())
.ForMember(dest => dest.EmailSubstring2, opt => opt.Ignore())
.ForMember(dest => dest.Profile, opt => opt.Ignore())
.ForMember(dest => dest.Attachments, opt => opt.Ignore());
// DTO -> Entity (for Update Status)
CreateMap<UpdateEmailHistoryStatusDto, EmailHistory>()
.ForMember(dest => dest.Id, opt => opt.Ignore())
.ForMember(dest => dest.ProfileId, opt => opt.Ignore())
.ForMember(dest => dest.MessageIdHash, opt => opt.Ignore())
.ForMember(dest => dest.OriginalMessageId, opt => opt.Ignore())
.ForMember(dest => dest.SenderAddress, opt => opt.Ignore())
.ForMember(dest => dest.EmailDate, opt => opt.Ignore())
.ForMember(dest => dest.Subject, opt => opt.Ignore())
.ForMember(dest => dest.EmailBodyText, opt => opt.Ignore())
.ForMember(dest => dest.EmailBodyHtml, opt => opt.Ignore())
.ForMember(dest => dest.EmailMessageId, opt => opt.Ignore())
.ForMember(dest => dest.AddedWhen, opt => opt.Ignore())
.ForMember(dest => dest.WindreamDocumentId, opt => opt.Ignore())
.ForMember(dest => dest.Comment, opt => opt.Ignore())
.ForMember(dest => dest.WorkProcess, opt => opt.Ignore())
.ForMember(dest => dest.ImapUid, opt => opt.Ignore())
.ForMember(dest => dest.EmailSubstring1, opt => opt.Ignore())
.ForMember(dest => dest.EmailSubstring2, opt => opt.Ignore())
.ForMember(dest => dest.Profile, opt => opt.Ignore())
.ForMember(dest => dest.Attachments, opt => opt.Ignore());
// Entity -> DTO (for Queries)
CreateMap<EmailHistory, EmailHistoryDto>()
.ForMember(dest => dest.ProfileName, opt => opt.MapFrom(src => src.Profile != null ? src.Profile.ProfileName : null));
}
}

View File

@@ -0,0 +1,45 @@
using AutoMapper;
using DigitalData.EmailProfiler.Application.Common.Dtos;
using DigitalData.EmailProfiler.Application.EmailProfiles.Commands;
using DigitalData.EmailProfiler.Domain.Entities;
namespace DigitalData.EmailProfiler.Application.Common.Mappings;
/// <summary>
/// AutoMapper profile for EmailProfile entity mappings.
/// </summary>
public class EmailProfileMappingProfile : Profile
{
public EmailProfileMappingProfile()
{
// Command -> Entity (for Create)
CreateMap<CreateEmailProfileCommand, EmailProfile>()
.ForMember(dest => dest.Id, opt => opt.Ignore()) // Auto-generated
.ForMember(dest => dest.AddedWhen, opt => opt.MapFrom(_ => DateTime.Now))
.ForMember(dest => dest.AddedWho, opt => opt.MapFrom(_ => "System")) // TODO: Get from user context
.ForMember(dest => dest.ChangedWhen, opt => opt.Ignore())
.ForMember(dest => dest.ChangedWho, opt => opt.Ignore())
.ForMember(dest => dest.LastPollTime, opt => opt.Ignore())
.ForMember(dest => dest.EmailAccount, opt => opt.Ignore())
.ForMember(dest => dest.EmailProcess, opt => opt.Ignore());
// Command -> Entity (for Update)
CreateMap<UpdateEmailProfileCommand, EmailProfile>()
.ForMember(dest => dest.Id, opt => opt.Ignore()) // Don't update ID
.ForMember(dest => dest.EmailAccountId, opt => opt.Ignore()) // Don't change account
.ForMember(dest => dest.ProcessId, opt => opt.Ignore()) // Don't change process
.ForMember(dest => dest.AddedWhen, opt => opt.Ignore())
.ForMember(dest => dest.AddedWho, opt => opt.Ignore())
.ForMember(dest => dest.ChangedWhen, opt => opt.MapFrom(_ => DateTime.Now))
.ForMember(dest => dest.ChangedWho, opt => opt.MapFrom(_ => "System")) // TODO: Get from user context
.ForMember(dest => dest.LastPollTime, opt => opt.Ignore())
.ForMember(dest => dest.EmailAccount, opt => opt.Ignore())
.ForMember(dest => dest.EmailProcess, opt => opt.Ignore());
// Entity -> DTO (for Queries)
CreateMap<EmailProfile, EmailProfileDto>()
.ForMember(dest => dest.EmailAccountName, opt => opt.MapFrom(src => src.EmailAccount != null ? src.EmailAccount.AccountName : null))
.ForMember(dest => dest.ProcessName, opt => opt.MapFrom(src => src.EmailProcess != null ? src.EmailProcess.ProcessName : null))
.ForMember(dest => dest.LastPollTime, opt => opt.MapFrom(src => src.LastPollTime));
}
}

View File

@@ -0,0 +1,30 @@
using System.Reflection;
using FluentValidation;
using Microsoft.Extensions.DependencyInjection;
namespace DigitalData.EmailProfiler.Application;
/// <summary>
/// Dependency injection configuration for Application layer.
/// </summary>
public static class DependencyInjection
{
public static IServiceCollection AddApplicationServices(this IServiceCollection services)
{
var assembly = Assembly.GetExecutingAssembly();
// MediatR - Register all handlers
services.AddMediatR(config =>
{
config.RegisterServicesFromAssembly(assembly);
});
// AutoMapper - Register all profiles
services.AddAutoMapper(assembly);
// FluentValidation - Register all validators
services.AddValidatorsFromAssembly(assembly);
return services;
}
}

View File

@@ -1,6 +1,8 @@
using DigitalData.EmailProfiler.Application.Interfaces.Repositories;
using DigitalData.EmailProfiler.Domain.Entities;
using MediatR;
namespace DigitalData.EmailProfiler.Application.Features.EmailAccounts.Commands;
namespace DigitalData.EmailProfiler.Application.EmailAccounts.Commands;
/// <summary>
/// Command to create a new email account.
@@ -22,3 +24,13 @@ public record CreateEmailAccountCommand : IRequest<int>
public string? EncryptedClientSecret { get; init; } // Already encrypted by client
public bool IsActive { get; init; } = true;
}
public class CreateEmailAccountCommandHandler(IRepository<EmailAccount> repository)
: IRequestHandler<CreateEmailAccountCommand, int>
{
public async Task<int> Handle(CreateEmailAccountCommand request, CancellationToken cancellationToken)
{
var account = await repository.CreateAsync(request, cancellationToken);
return account.Id;
}
}

View File

@@ -0,0 +1,26 @@
using AutoMapper;
using DigitalData.EmailProfiler.Application.Common.Dtos;
using DigitalData.EmailProfiler.Application.Interfaces.Repositories;
using DigitalData.EmailProfiler.Domain.Entities;
using MediatR;
namespace DigitalData.EmailProfiler.Application.EmailAccounts.Queries;
/// <summary>
/// Query to get email account by ID.
/// </summary>
public record GetEmailAccountByIdQuery(int Id) : IRequest<EmailAccountDto?>;
public class GetEmailAccountByIdQueryHandler(
IRepository<EmailAccount> repository,
IMapper mapper)
: IRequestHandler<GetEmailAccountByIdQuery, EmailAccountDto?>
{
public async Task<EmailAccountDto?> Handle(
GetEmailAccountByIdQuery request,
CancellationToken cancellationToken)
{
var account = await repository.GetByIdAsync(request.Id, cancellationToken);
return account != null ? mapper.Map<EmailAccountDto>(account) : null;
}
}

View File

@@ -0,0 +1,26 @@
using AutoMapper;
using DigitalData.EmailProfiler.Application.Common.Dtos;
using DigitalData.EmailProfiler.Application.Interfaces.Repositories;
using DigitalData.EmailProfiler.Domain.Entities;
using MediatR;
namespace DigitalData.EmailProfiler.Application.EmailAccounts.Queries;
/// <summary>
/// Query to get all email accounts.
/// </summary>
public record GetEmailAccountsQuery : IRequest<IEnumerable<EmailAccountDto>>;
public class GetEmailAccountsQueryHandler(
IRepository<EmailAccount> repository,
IMapper mapper)
: IRequestHandler<GetEmailAccountsQuery, IEnumerable<EmailAccountDto>>
{
public async Task<IEnumerable<EmailAccountDto>> Handle(
GetEmailAccountsQuery request,
CancellationToken cancellationToken)
{
var accounts = await repository.GetAllAsync(cancellationToken);
return mapper.Map<IEnumerable<EmailAccountDto>>(accounts);
}
}

View File

@@ -0,0 +1,56 @@
using DigitalData.EmailProfiler.Application.EmailAccounts.Commands;
using FluentValidation;
namespace DigitalData.EmailProfiler.Application.EmailAccounts.Validators;
/// <summary>
/// Validator for CreateEmailAccountCommand.
/// </summary>
public class CreateEmailAccountCommandValidator : AbstractValidator<CreateEmailAccountCommand>
{
public CreateEmailAccountCommandValidator()
{
RuleFor(x => x.AccountName)
.NotEmpty().WithMessage("Account name is required")
.MaximumLength(100).WithMessage("Account name must not exceed 100 characters");
RuleFor(x => x.Username)
.NotEmpty().WithMessage("Username is required")
.MaximumLength(200).WithMessage("Username must not exceed 200 characters")
.EmailAddress().WithMessage("Username must be a valid email address");
RuleFor(x => x.ImapServer)
.NotEmpty().WithMessage("IMAP server is required")
.MaximumLength(200).WithMessage("IMAP server must not exceed 200 characters");
RuleFor(x => x.ImapPort)
.GreaterThan(0).WithMessage("IMAP port must be greater than 0")
.LessThanOrEqualTo(65535).WithMessage("IMAP port must not exceed 65535");
RuleFor(x => x.SmtpServer)
.NotEmpty().WithMessage("SMTP server is required")
.MaximumLength(200).WithMessage("SMTP server must not exceed 200 characters");
RuleFor(x => x.SmtpPort)
.GreaterThan(0).WithMessage("SMTP port must be greater than 0")
.LessThanOrEqualTo(65535).WithMessage("SMTP port must not exceed 65535");
// OAuth2 validation
RuleFor(x => x.TenantId)
.NotEmpty().WithMessage("Tenant ID is required for OAuth2")
.When(x => x.UseOAuth2);
RuleFor(x => x.ClientId)
.NotEmpty().WithMessage("Client ID is required for OAuth2")
.When(x => x.UseOAuth2);
RuleFor(x => x.EncryptedClientSecret)
.NotEmpty().WithMessage("Client secret is required for OAuth2")
.When(x => x.UseOAuth2);
// Password validation (non-OAuth2)
RuleFor(x => x.EncryptedPassword)
.NotEmpty().WithMessage("Password is required")
.When(x => !x.UseOAuth2);
}
}

View File

@@ -0,0 +1,25 @@
using AutoMapper;
using DigitalData.EmailProfiler.Application.Common.Dtos.EmailHistories;
using DigitalData.EmailProfiler.Application.Interfaces.Repositories;
using MediatR;
namespace DigitalData.EmailProfiler.Application.EmailHistories.Queries;
/// <summary>
/// Query to get email history by ID with attachments.
/// </summary>
public record GetEmailHistoryByIdQuery(int Id) : IRequest<EmailHistoryDto?>;
public class GetEmailHistoryByIdQueryHandler(
IEmailHistoryRepository repository,
IMapper mapper)
: IRequestHandler<GetEmailHistoryByIdQuery, EmailHistoryDto?>
{
public async Task<EmailHistoryDto?> Handle(
GetEmailHistoryByIdQuery request,
CancellationToken cancellationToken)
{
var history = await repository.GetWithAttachmentsAsync(request.Id, cancellationToken);
return history != null ? mapper.Map<EmailHistoryDto>(history) : null;
}
}

View File

@@ -0,0 +1,48 @@
using AutoMapper;
using DigitalData.EmailProfiler.Application.Common.Dtos.EmailHistories;
using DigitalData.EmailProfiler.Application.Interfaces.Repositories;
using MediatR;
namespace DigitalData.EmailProfiler.Application.EmailHistories.Queries;
/// <summary>
/// Query to get email history by profile with pagination.
/// </summary>
public record GetEmailHistoryByProfileQuery(
int ProfileId,
int PageNumber = 1,
int PageSize = 50) : IRequest<EmailHistoryPagedResult>;
/// <summary>
/// Paged result for email history.
/// </summary>
public record EmailHistoryPagedResult(
IEnumerable<EmailHistoryDto> Items,
int TotalCount,
int PageNumber,
int PageSize);
public class GetEmailHistoryByProfileQueryHandler(
IEmailHistoryRepository repository,
IMapper mapper)
: IRequestHandler<GetEmailHistoryByProfileQuery, EmailHistoryPagedResult>
{
public async Task<EmailHistoryPagedResult> Handle(
GetEmailHistoryByProfileQuery request,
CancellationToken cancellationToken)
{
var (items, totalCount) = await repository.GetByProfileIdAsync(
request.ProfileId,
request.PageNumber,
request.PageSize,
cancellationToken);
var dtos = mapper.Map<IEnumerable<EmailHistoryDto>>(items);
return new EmailHistoryPagedResult(
dtos,
totalCount,
request.PageNumber,
request.PageSize);
}
}

View File

@@ -0,0 +1,180 @@
using DigitalData.EmailProfiler.Application.Common.Dtos.EmailAttachments;
using DigitalData.EmailProfiler.Application.Common.Dtos.EmailHistories;
using DigitalData.EmailProfiler.Application.Interfaces.Repositories;
using DigitalData.EmailProfiler.Application.Interfaces.Services;
using DigitalData.EmailProfiler.Domain.Entities;
using DigitalData.EmailProfiler.Domain.Enums;
using DigitalData.EmailProfiler.Domain.Events;
using DigitalData.EmailProfiler.Domain.Exceptions;
using DigitalData.EmailProfiler.Domain.Services;
using MediatR;
namespace DigitalData.EmailProfiler.Application.EmailProcessing.Commands;
/// <summary>
/// Command to process a single email from a profile.
/// This is the core email processing logic.
/// </summary>
public record ProcessEmailCommand : IRequest<int>
{
public int ProfileId { get; init; }
public string MessageId { get; init; } = string.Empty;
public string Sender { get; init; } = string.Empty;
public DateTime ReceivedDate { get; init; }
public string Subject { get; init; } = string.Empty;
public string? BodyText { get; init; }
public string? BodyHtml { get; init; }
public List<AttachmentData> Attachments { get; init; } = new();
}
/// <summary>
/// Attachment data for email processing.
/// </summary>
public record AttachmentData(
string FileName,
byte[] Content,
string ContentType,
long SizeBytes);
public class ProcessEmailCommandHandler(
IEmailProfileRepository profileRepository,
IEmailHistoryRepository historyRepository,
IRepository<EmailAttachment> attachmentRepository,
IPublisher publisher,
MessageIdGenerator messageIdGenerator,
IPdfProcessingService pdfService,
IDmsService dmsService)
: IRequestHandler<ProcessEmailCommand, int>
{
public async Task<int> Handle(ProcessEmailCommand request, CancellationToken cancellationToken)
{
// 1. Get profile with related entities
var profile = await profileRepository.GetWithRelatedEntitiesAsync(request.ProfileId, cancellationToken)
?? throw new DomainException($"Profile with ID {request.ProfileId} not found");
// 2. Generate message ID hash for duplicate detection
var messageId = messageIdGenerator.Generate(
request.MessageId,
request.Sender,
request.ReceivedDate,
request.Subject);
// 3. Check for duplicates
var isDuplicate = await historyRepository.IsDuplicateAsync(messageId.Hash, cancellationToken);
if (isDuplicate)
{
throw new ValidationException("Email already processed (duplicate detected)", ErrorCode.DuplicateMessageId);
}
// 4. Create email history record using DTO approach
var historyDto = new CreateEmailHistoryDto(
ProfileId: profile.Id,
MessageIdHash: messageId.Hash,
OriginalMessageId: request.MessageId,
SenderAddress: request.Sender,
EmailDate: request.ReceivedDate,
Subject: request.Subject,
EmailBodyText: request.BodyText,
EmailBodyHtml: request.BodyHtml);
var createdHistory = await historyRepository.CreateAsync(historyDto, cancellationToken);
try
{
// 5. Process attachments
foreach (var attachmentData in request.Attachments)
{
var attachmentDto = new CreateEmailAttachmentDto(
EmailHistoryId: createdHistory.Id,
OriginalFileName: attachmentData.FileName,
SavedFileName: attachmentData.FileName, // TODO: Generate unique name
FilePath: string.Empty, // TODO: Save to disk and get path
FileSize: attachmentData.SizeBytes,
Extension: Path.GetExtension(attachmentData.FileName),
ContentType: attachmentData.ContentType,
Content: attachmentData.Content);
var attachment = await attachmentRepository.CreateAsync(attachmentDto, cancellationToken);
// Validate PDF attachments - need to retrieve and update the entity
if (attachmentData.ContentType.Contains("pdf", StringComparison.OrdinalIgnoreCase))
{
using var stream = new MemoryStream(attachmentData.Content);
var isValidPdf = await pdfService.IsValidPdfAsync(stream, cancellationToken);
// For domain methods like MarkAsValid/MarkAsCorrupt, we need to get the entity
var attachmentEntity = await attachmentRepository.GetByIdAsync(attachment.Id, cancellationToken);
if (attachmentEntity != null)
{
if (isValidPdf)
{
attachmentEntity.MarkAsValid();
}
else
{
attachmentEntity.MarkAsCorrupt(ErrorCode.PdfStructureInvalid, "Invalid PDF structure");
}
// Update directly using UpdateSingleAsync for safety
await attachmentRepository.UpdateSingleAsync(
a => a.Id == attachmentEntity.Id,
new UpdateEmailAttachmentStatusDto(
Status: attachmentEntity.Status ?? string.Empty,
ValidationErrorCode: attachmentEntity.ValidationErrorCode,
ValidationErrorMessage: attachmentEntity.ValidationErrorMessage),
cancellationToken);
}
}
}
// 6. Archive to DMS if configured
if (profile.EmailProcess != null && profile.EmailProcess.EnableWindreamImport)
{
// TODO: Implement DMS archiving with indexing steps
// This will be implemented based on ProcessSteps and IndexingSteps
}
// 7. Mark as processed - retrieve entity for domain method
var historyEntity = await historyRepository.GetByIdAsync(createdHistory.Id, cancellationToken);
if (historyEntity != null)
{
historyEntity.MarkAsProcessed();
await historyRepository.UpdateSingleAsync(
h => h.Id == historyEntity.Id,
new UpdateEmailHistoryStatusDto(
Status: historyEntity.Status ?? string.Empty,
ProcessedDate: historyEntity.ProcessedDate),
cancellationToken);
}
// 8. Publish domain event
await publisher.Publish(
new EmailProcessedEvent(
createdHistory.Id,
profile.Id,
messageId.Hash,
EmailStatus.Processed),
cancellationToken);
return createdHistory.Id;
}
catch (Exception ex)
{
// Mark as failed - retrieve entity for domain method
var historyEntity = await historyRepository.GetByIdAsync(createdHistory.Id, cancellationToken);
if (historyEntity != null)
{
historyEntity.MarkAsFailed(ErrorCode.AttachmentExtractionFailed, ex.Message);
await historyRepository.UpdateSingleAsync(
h => h.Id == historyEntity.Id,
new UpdateEmailHistoryStatusDto(
Status: historyEntity.Status ?? string.Empty,
ErrorCodeValue: historyEntity.ErrorCodeValue,
ErrorMessage: historyEntity.ErrorMessage),
cancellationToken);
}
throw;
}
}
}

View File

@@ -0,0 +1,50 @@
using DigitalData.EmailProfiler.Application.EmailProcessing.Commands;
using FluentValidation;
namespace DigitalData.EmailProfiler.Application.EmailProcessing.Validators;
/// <summary>
/// Validator for ProcessEmailCommand.
/// </summary>
public class ProcessEmailCommandValidator : AbstractValidator<ProcessEmailCommand>
{
public ProcessEmailCommandValidator()
{
RuleFor(x => x.ProfileId)
.GreaterThan(0).WithMessage("Profile ID must be greater than 0");
RuleFor(x => x.MessageId)
.NotEmpty().WithMessage("Message ID is required")
.MaximumLength(500).WithMessage("Message ID must not exceed 500 characters");
RuleFor(x => x.Sender)
.NotEmpty().WithMessage("Sender is required")
.MaximumLength(200).WithMessage("Sender must not exceed 200 characters")
.EmailAddress().WithMessage("Sender must be a valid email address");
RuleFor(x => x.Subject)
.NotEmpty().WithMessage("Subject is required")
.MaximumLength(500).WithMessage("Subject must not exceed 500 characters");
RuleFor(x => x.ReceivedDate)
.NotEmpty().WithMessage("Received date is required")
.LessThanOrEqualTo(DateTime.Now.AddDays(1)).WithMessage("Received date cannot be in the future");
RuleFor(x => x.Attachments)
.NotNull().WithMessage("Attachments collection cannot be null");
RuleForEach(x => x.Attachments).ChildRules(attachment =>
{
attachment.RuleFor(a => a.FileName)
.NotEmpty().WithMessage("Attachment file name is required")
.MaximumLength(500).WithMessage("Attachment file name must not exceed 500 characters");
attachment.RuleFor(a => a.Content)
.NotEmpty().WithMessage("Attachment content is required");
attachment.RuleFor(a => a.SizeBytes)
.GreaterThan(0).WithMessage("Attachment size must be greater than 0")
.LessThanOrEqualTo(100 * 1024 * 1024).WithMessage("Attachment size must not exceed 100 MB");
});
}
}

View File

@@ -0,0 +1,28 @@
using DigitalData.EmailProfiler.Application.Interfaces.Repositories;
using DigitalData.EmailProfiler.Domain.Entities;
using MediatR;
namespace DigitalData.EmailProfiler.Application.EmailProfiles.Commands;
/// <summary>
/// Command to create a new email profile.
/// </summary>
public record CreateEmailProfileCommand : IRequest<int>
{
public string ProfileName { get; init; } = string.Empty;
public int EmailAccountId { get; init; }
public int? ProcessId { get; init; }
public string? ValidationSql { get; init; }
public int PollIntervalMinutes { get; init; } = 15;
public bool IsActive { get; init; } = true;
}
public class CreateEmailProfileCommandHandler(IRepository<EmailProfile> repository)
: IRequestHandler<CreateEmailProfileCommand, int>
{
public async Task<int> Handle(CreateEmailProfileCommand request, CancellationToken cancellationToken)
{
var profile = await repository.CreateAsync(request, cancellationToken);
return profile.Id;
}
}

View File

@@ -0,0 +1,20 @@
using DigitalData.EmailProfiler.Application.Interfaces.Repositories;
using DigitalData.EmailProfiler.Domain.Entities;
using MediatR;
namespace DigitalData.EmailProfiler.Application.EmailProfiles.Commands;
/// <summary>
/// Command to delete an email profile.
/// </summary>
public record DeleteEmailProfileCommand(int Id) : IRequest<int>;
public class DeleteEmailProfileCommandHandler(IRepository<EmailProfile> repository)
: IRequestHandler<DeleteEmailProfileCommand, int>
{
public async Task<int> Handle(DeleteEmailProfileCommand request, CancellationToken cancellationToken)
{
await repository.DeleteSingleAsync(p => p.Id == request.Id, cancellationToken);
return request.Id;
}
}

View File

@@ -0,0 +1,27 @@
using DigitalData.EmailProfiler.Application.Interfaces.Repositories;
using DigitalData.EmailProfiler.Domain.Entities;
using MediatR;
namespace DigitalData.EmailProfiler.Application.EmailProfiles.Commands;
/// <summary>
/// Command to update an existing email profile.
/// </summary>
public record UpdateEmailProfileCommand : IRequest<int>
{
public int Id { get; init; }
public string ProfileName { get; init; } = string.Empty;
public string? ValidationSql { get; init; }
public int PollIntervalMinutes { get; init; }
public bool IsActive { get; init; }
}
public class UpdateEmailProfileCommandHandler(IRepository<EmailProfile> repository)
: IRequestHandler<UpdateEmailProfileCommand, int>
{
public async Task<int> Handle(UpdateEmailProfileCommand request, CancellationToken cancellationToken)
{
await repository.UpdateSingleAsync(p => p.Id == request.Id, request, cancellationToken);
return request.Id;
}
}

View File

@@ -0,0 +1,25 @@
using AutoMapper;
using DigitalData.EmailProfiler.Application.Common.Dtos;
using DigitalData.EmailProfiler.Application.Interfaces.Repositories;
using MediatR;
namespace DigitalData.EmailProfiler.Application.EmailProfiles.Queries;
/// <summary>
/// Query to get all active email profiles.
/// </summary>
public record GetActiveEmailProfilesQuery : IRequest<IEnumerable<EmailProfileDto>>;
public class GetActiveEmailProfilesQueryHandler(
IEmailProfileRepository repository,
IMapper mapper)
: IRequestHandler<GetActiveEmailProfilesQuery, IEnumerable<EmailProfileDto>>
{
public async Task<IEnumerable<EmailProfileDto>> Handle(
GetActiveEmailProfilesQuery request,
CancellationToken cancellationToken)
{
var profiles = await repository.GetActiveProfilesAsync(cancellationToken);
return mapper.Map<IEnumerable<EmailProfileDto>>(profiles);
}
}

View File

@@ -0,0 +1,25 @@
using AutoMapper;
using DigitalData.EmailProfiler.Application.Common.Dtos;
using DigitalData.EmailProfiler.Application.Interfaces.Repositories;
using MediatR;
namespace DigitalData.EmailProfiler.Application.EmailProfiles.Queries;
/// <summary>
/// Query to get email profile by ID with related entities.
/// </summary>
public record GetEmailProfileByIdQuery(int Id) : IRequest<EmailProfileDto?>;
public class GetEmailProfileByIdQueryHandler(
IEmailProfileRepository repository,
IMapper mapper)
: IRequestHandler<GetEmailProfileByIdQuery, EmailProfileDto?>
{
public async Task<EmailProfileDto?> Handle(
GetEmailProfileByIdQuery request,
CancellationToken cancellationToken)
{
var profile = await repository.GetWithRelatedEntitiesAsync(request.Id, cancellationToken);
return profile != null ? mapper.Map<EmailProfileDto>(profile) : null;
}
}

View File

@@ -0,0 +1,26 @@
using AutoMapper;
using DigitalData.EmailProfiler.Application.Common.Dtos;
using DigitalData.EmailProfiler.Application.Interfaces.Repositories;
using DigitalData.EmailProfiler.Domain.Entities;
using MediatR;
namespace DigitalData.EmailProfiler.Application.EmailProfiles.Queries;
/// <summary>
/// Query to get all email profiles.
/// </summary>
public record GetEmailProfilesQuery : IRequest<IEnumerable<EmailProfileDto>>;
public class GetEmailProfilesQueryHandler(
IRepository<EmailProfile> repository,
IMapper mapper)
: IRequestHandler<GetEmailProfilesQuery, IEnumerable<EmailProfileDto>>
{
public async Task<IEnumerable<EmailProfileDto>> Handle(
GetEmailProfilesQuery request,
CancellationToken cancellationToken)
{
var profiles = await repository.GetAllAsync(cancellationToken);
return mapper.Map<IEnumerable<EmailProfileDto>>(profiles);
}
}

View File

@@ -0,0 +1,28 @@
using DigitalData.EmailProfiler.Application.EmailProfiles.Commands;
using FluentValidation;
namespace DigitalData.EmailProfiler.Application.EmailProfiles.Validators;
/// <summary>
/// Validator for CreateEmailProfileCommand.
/// </summary>
public class CreateEmailProfileCommandValidator : AbstractValidator<CreateEmailProfileCommand>
{
public CreateEmailProfileCommandValidator()
{
RuleFor(x => x.ProfileName)
.NotEmpty().WithMessage("Profile name is required")
.MaximumLength(100).WithMessage("Profile name must not exceed 100 characters");
RuleFor(x => x.EmailAccountId)
.GreaterThan(0).WithMessage("Email account ID must be greater than 0");
RuleFor(x => x.PollIntervalMinutes)
.GreaterThanOrEqualTo(1).WithMessage("Poll interval must be at least 1 minute")
.LessThanOrEqualTo(1440).WithMessage("Poll interval must not exceed 1440 minutes (24 hours)");
RuleFor(x => x.ValidationSql)
.MaximumLength(1000).WithMessage("Validation SQL must not exceed 1000 characters")
.When(x => !string.IsNullOrEmpty(x.ValidationSql));
}
}

View File

@@ -0,0 +1,28 @@
using DigitalData.EmailProfiler.Application.EmailProfiles.Commands;
using FluentValidation;
namespace DigitalData.EmailProfiler.Application.EmailProfiles.Validators;
/// <summary>
/// Validator for UpdateEmailProfileCommand.
/// </summary>
public class UpdateEmailProfileCommandValidator : AbstractValidator<UpdateEmailProfileCommand>
{
public UpdateEmailProfileCommandValidator()
{
RuleFor(x => x.Id)
.GreaterThan(0).WithMessage("Profile ID must be greater than 0");
RuleFor(x => x.ProfileName)
.NotEmpty().WithMessage("Profile name is required")
.MaximumLength(100).WithMessage("Profile name must not exceed 100 characters");
RuleFor(x => x.PollIntervalMinutes)
.GreaterThanOrEqualTo(1).WithMessage("Poll interval must be at least 1 minute")
.LessThanOrEqualTo(1440).WithMessage("Poll interval must not exceed 1440 minutes (24 hours)");
RuleFor(x => x.ValidationSql)
.MaximumLength(1000).WithMessage("Validation SQL must not exceed 1000 characters")
.When(x => !string.IsNullOrEmpty(x.ValidationSql));
}
}

View File

@@ -1,37 +0,0 @@
using DigitalData.EmailProfiler.Application.Interfaces.Repositories;
using DigitalData.EmailProfiler.Domain.Entities;
using MediatR;
namespace DigitalData.EmailProfiler.Application.Features.EmailAccounts.Commands;
public class CreateEmailAccountCommandHandler(IUnitOfWork unitOfWork)
: IRequestHandler<CreateEmailAccountCommand, int>
{
public async Task<int> Handle(CreateEmailAccountCommand request, CancellationToken cancellationToken)
{
var account = new EmailAccount
{
AccountName = request.AccountName,
Username = request.Username,
ImapServer = request.ImapServer,
ImapPort = request.ImapPort,
ImapUseSsl = request.ImapUseSsl,
SmtpServer = request.SmtpServer,
SmtpPort = request.SmtpPort,
SmtpUseSsl = request.SmtpUseSsl,
UseOAuth2 = request.UseOAuth2,
EncryptedPassword = request.EncryptedPassword,
TenantId = request.TenantId,
ClientId = request.ClientId,
EncryptedClientSecret = request.EncryptedClientSecret,
IsActive = request.IsActive,
AddedWhen = DateTime.Now,
AddedWho = "System" // TODO: Get from current user context
};
var createdAccount = await unitOfWork.EmailAccounts.AddAsync(account, cancellationToken);
await unitOfWork.SaveChangesAsync(cancellationToken);
return createdAccount.Id;
}
}

View File

@@ -1,28 +0,0 @@
using MediatR;
namespace DigitalData.EmailProfiler.Application.Features.EmailProcessing.Commands;
/// <summary>
/// Command to process a single email from a profile.
/// This is the core email processing logic.
/// </summary>
public record ProcessEmailCommand : IRequest<int>
{
public int ProfileId { get; init; }
public string MessageId { get; init; } = string.Empty;
public string Sender { get; init; } = string.Empty;
public DateTime ReceivedDate { get; init; }
public string Subject { get; init; } = string.Empty;
public string? BodyText { get; init; }
public string? BodyHtml { get; init; }
public List<AttachmentData> Attachments { get; init; } = new();
}
/// <summary>
/// Attachment data for email processing.
/// </summary>
public record AttachmentData(
string FileName,
byte[] Content,
string ContentType,
long SizeBytes);

View File

@@ -1,127 +0,0 @@
using DigitalData.EmailProfiler.Application.Interfaces.Repositories;
using DigitalData.EmailProfiler.Application.Interfaces.Services;
using DigitalData.EmailProfiler.Domain.Entities;
using DigitalData.EmailProfiler.Domain.Enums;
using DigitalData.EmailProfiler.Domain.Events;
using DigitalData.EmailProfiler.Domain.Exceptions;
using DigitalData.EmailProfiler.Domain.Services;
using MediatR;
namespace DigitalData.EmailProfiler.Application.Features.EmailProcessing.Commands;
public class ProcessEmailCommandHandler(
IUnitOfWork unitOfWork,
IPublisher publisher,
MessageIdGenerator messageIdGenerator,
IPdfProcessingService pdfService,
IDmsService dmsService)
: IRequestHandler<ProcessEmailCommand, int>
{
public async Task<int> Handle(ProcessEmailCommand request, CancellationToken cancellationToken)
{
// 1. Get profile with related entities
var profile = await unitOfWork.EmailProfiles.GetWithRelatedEntitiesAsync(request.ProfileId, cancellationToken)
?? throw new DomainException($"Profile with ID {request.ProfileId} not found");
// 2. Generate message ID hash for duplicate detection
var messageId = messageIdGenerator.Generate(
request.MessageId,
request.Sender,
request.ReceivedDate,
request.Subject);
// 3. Check for duplicates
var isDuplicate = await unitOfWork.EmailHistories.IsDuplicateAsync(messageId.Hash, cancellationToken);
if (isDuplicate)
{
throw new ValidationException("Email already processed (duplicate detected)", ErrorCode.DuplicateMessageId);
}
// 4. Create email history record
var emailHistory = new EmailHistory
{
ProfileId = profile.Id,
MessageIdHash = messageId.Hash,
OriginalMessageId = request.MessageId,
SenderAddress = request.Sender,
EmailDate = request.ReceivedDate,
Subject = request.Subject,
EmailBodyText = request.BodyText,
EmailBodyHtml = request.BodyHtml,
Status = EmailStatus.Processing.ToString(),
AddedWhen = DateTime.Now
};
var createdHistory = await unitOfWork.EmailHistories.AddAsync(emailHistory, cancellationToken);
await unitOfWork.SaveChangesAsync(cancellationToken);
try
{
// 5. Process attachments
foreach (var attachmentData in request.Attachments)
{
var attachment = new EmailAttachment
{
EmailHistoryId = createdHistory.Id,
OriginalFileName = attachmentData.FileName,
SavedFileName = attachmentData.FileName, // TODO: Generate unique name
FilePath = string.Empty, // TODO: Save to disk and get path
FileSize = attachmentData.SizeBytes,
Extension = Path.GetExtension(attachmentData.FileName),
Status = AttachmentStatus.Pending.ToString(),
AddedWhen = DateTime.Now
};
// Validate PDF attachments
if (attachmentData.ContentType.Contains("pdf", StringComparison.OrdinalIgnoreCase))
{
using var stream = new MemoryStream(attachmentData.Content);
var isValidPdf = await pdfService.IsValidPdfAsync(stream, cancellationToken);
if (isValidPdf)
{
attachment.MarkAsValid();
}
else
{
attachment.MarkAsCorrupt(ErrorCode.PdfStructureInvalid, "Invalid PDF structure");
}
}
createdHistory.Attachments.Add(attachment);
}
// 6. Archive to DMS if configured
if (profile.EmailProcess != null && profile.EmailProcess.EnableWindreamImport)
{
// TODO: Implement DMS archiving with indexing steps
// This will be implemented based on ProcessSteps and IndexingSteps
}
// 7. Mark as processed
createdHistory.MarkAsProcessed();
await unitOfWork.EmailHistories.UpdateAsync(createdHistory, cancellationToken);
await unitOfWork.SaveChangesAsync(cancellationToken);
// 8. Publish domain event
await publisher.Publish(
new EmailProcessedEvent(
createdHistory.Id,
profile.Id,
messageId.Hash,
EmailStatus.Processed),
cancellationToken);
return createdHistory.Id;
}
catch (Exception ex)
{
// Mark as failed
createdHistory.MarkAsFailed(ErrorCode.AttachmentExtractionFailed, ex.Message);
await unitOfWork.EmailHistories.UpdateAsync(createdHistory, cancellationToken);
await unitOfWork.SaveChangesAsync(cancellationToken);
throw;
}
}
}

View File

@@ -1,16 +0,0 @@
using MediatR;
namespace DigitalData.EmailProfiler.Application.Features.EmailProfiles.Commands;
/// <summary>
/// Command to create a new email profile.
/// </summary>
public record CreateEmailProfileCommand : IRequest<int>
{
public string ProfileName { get; init; } = string.Empty;
public int EmailAccountId { get; init; }
public int? ProcessId { get; init; }
public string? ValidationSql { get; init; }
public int PollIntervalMinutes { get; init; } = 15;
public bool IsActive { get; init; } = true;
}

View File

@@ -1,29 +0,0 @@
using DigitalData.EmailProfiler.Application.Interfaces.Repositories;
using DigitalData.EmailProfiler.Domain.Entities;
using MediatR;
namespace DigitalData.EmailProfiler.Application.Features.EmailProfiles.Commands;
public class CreateEmailProfileCommandHandler(IUnitOfWork unitOfWork)
: IRequestHandler<CreateEmailProfileCommand, int>
{
public async Task<int> Handle(CreateEmailProfileCommand request, CancellationToken cancellationToken)
{
var profile = new EmailProfile
{
ProfileName = request.ProfileName,
EmailAccountId = request.EmailAccountId,
ProcessId = request.ProcessId,
ValidationSql = request.ValidationSql,
PollIntervalMinutes = request.PollIntervalMinutes,
IsActive = request.IsActive,
AddedWhen = DateTime.Now,
AddedWho = "System" // TODO: Get from current user context
};
var createdProfile = await unitOfWork.EmailProfiles.AddAsync(profile, cancellationToken);
await unitOfWork.SaveChangesAsync(cancellationToken);
return createdProfile.Id;
}
}

View File

@@ -1,8 +0,0 @@
using MediatR;
namespace DigitalData.EmailProfiler.Application.Features.EmailProfiles.Commands;
/// <summary>
/// Command to delete an email profile.
/// </summary>
public record DeleteEmailProfileCommand(int Id) : IRequest<Unit>;

View File

@@ -1,20 +0,0 @@
using DigitalData.EmailProfiler.Application.Interfaces.Repositories;
using DigitalData.EmailProfiler.Domain.Exceptions;
using MediatR;
namespace DigitalData.EmailProfiler.Application.Features.EmailProfiles.Commands;
public class DeleteEmailProfileCommandHandler(IUnitOfWork unitOfWork)
: IRequestHandler<DeleteEmailProfileCommand, Unit>
{
public async Task<Unit> Handle(DeleteEmailProfileCommand request, CancellationToken cancellationToken)
{
var profile = await unitOfWork.EmailProfiles.GetByIdAsync(request.Id, cancellationToken)
?? throw new DomainException($"Email profile with ID {request.Id} not found");
await unitOfWork.EmailProfiles.DeleteAsync(profile, cancellationToken);
await unitOfWork.SaveChangesAsync(cancellationToken);
return Unit.Value;
}
}

View File

@@ -1,15 +0,0 @@
using MediatR;
namespace DigitalData.EmailProfiler.Application.Features.EmailProfiles.Commands;
/// <summary>
/// Command to update an existing email profile.
/// </summary>
public record UpdateEmailProfileCommand : IRequest<Unit>
{
public int Id { get; init; }
public string ProfileName { get; init; } = string.Empty;
public string? ValidationSql { get; init; }
public int PollIntervalMinutes { get; init; }
public bool IsActive { get; init; }
}

View File

@@ -1,27 +0,0 @@
using DigitalData.EmailProfiler.Application.Interfaces.Repositories;
using DigitalData.EmailProfiler.Domain.Exceptions;
using MediatR;
namespace DigitalData.EmailProfiler.Application.Features.EmailProfiles.Commands;
public class UpdateEmailProfileCommandHandler(IUnitOfWork unitOfWork)
: IRequestHandler<UpdateEmailProfileCommand, Unit>
{
public async Task<Unit> Handle(UpdateEmailProfileCommand request, CancellationToken cancellationToken)
{
var profile = await unitOfWork.EmailProfiles.GetByIdAsync(request.Id, cancellationToken)
?? throw new DomainException($"Email profile with ID {request.Id} not found");
profile.ProfileName = request.ProfileName;
profile.ValidationSql = request.ValidationSql;
profile.PollIntervalMinutes = request.PollIntervalMinutes;
profile.IsActive = request.IsActive;
profile.ChangedWhen = DateTime.Now;
profile.ChangedWho = "System"; // TODO: Get from current user context
await unitOfWork.EmailProfiles.UpdateAsync(profile, cancellationToken);
await unitOfWork.SaveChangesAsync(cancellationToken);
return Unit.Value;
}
}

View File

@@ -3,17 +3,58 @@ using System.Linq.Expressions;
namespace DigitalData.EmailProfiler.Application.Interfaces.Repositories;
/// <summary>
/// Base repository interface for common CRUD operations.
/// Base repository interface for common CRUD operations with AutoMapper support.
/// Changes are automatically saved after each operation - no explicit SaveChanges needed.
/// </summary>
/// <typeparam name="T">Entity type</typeparam>
public interface IRepository<T> where T : class
/// <typeparam name="TEntity">Entity type</typeparam>
public interface IRepository<TEntity> where TEntity : class
{
Task<T?> GetByIdAsync(int id, CancellationToken cancellationToken = default);
Task<IEnumerable<T>> GetAllAsync(CancellationToken cancellationToken = default);
Task<IEnumerable<T>> FindAsync(Expression<Func<T, bool>> predicate, CancellationToken cancellationToken = default);
Task<T> AddAsync(T entity, CancellationToken cancellationToken = default);
Task UpdateAsync(T entity, CancellationToken cancellationToken = default);
Task DeleteAsync(T entity, CancellationToken cancellationToken = default);
Task<bool> ExistsAsync(Expression<Func<T, bool>> predicate, CancellationToken cancellationToken = default);
Task<int> CountAsync(Expression<Func<T, bool>>? predicate = null, CancellationToken cancellationToken = default);
// ==================== QUERY OPERATIONS ====================
Task<TEntity?> GetByIdAsync(int id, CancellationToken cancellationToken = default);
Task<IEnumerable<TEntity>> GetAllAsync(CancellationToken cancellationToken = default);
Task<IEnumerable<TEntity>> FindAsync(Expression<Func<TEntity, bool>> predicate, CancellationToken cancellationToken = default);
Task<TEntity?> FirstOrDefaultAsync(Expression<Func<TEntity, bool>> predicate, CancellationToken cancellationToken = default);
Task<bool> ExistsAsync(Expression<Func<TEntity, bool>> predicate, CancellationToken cancellationToken = default);
Task<int> CountAsync(Expression<Func<TEntity, bool>>? predicate = null, CancellationToken cancellationToken = default);
// ==================== CREATE OPERATIONS ====================
/// <summary>
/// Create entity from DTO using AutoMapper and save immediately.
/// </summary>
/// <returns>Created entity with ID populated</returns>
Task<TEntity> CreateAsync<TDto>(TDto dto, CancellationToken cancellationToken = default) where TDto : class;
// ==================== UPDATE OPERATIONS ====================
/// <summary>
/// Update ALL entities matching expression using DTO via AutoMapper.
/// WARNING: This can update multiple records. Use UpdateSingleAsync for single-record updates.
/// </summary>
/// <returns>Number of entities updated</returns>
Task<int> UpdateAsync<TDto>(Expression<Func<TEntity, bool>> predicate, TDto dto, CancellationToken cancellationToken = default) where TDto : class;
/// <summary>
/// Update SINGLE entity matching expression using DTO via AutoMapper.
/// SAFETY: Throws exception if zero or multiple entities match the predicate.
/// </summary>
/// <exception cref="InvalidOperationException">Thrown when zero or multiple entities match</exception>
Task UpdateSingleAsync<TDto>(Expression<Func<TEntity, bool>> predicate, TDto dto, CancellationToken cancellationToken = default) where TDto : class;
// ==================== DELETE OPERATIONS ====================
/// <summary>
/// Delete ALL entities matching expression.
/// WARNING: This can delete multiple records. Use DeleteSingleAsync for single-record deletes.
/// </summary>
/// <returns>Number of entities deleted</returns>
Task<int> DeleteAsync(Expression<Func<TEntity, bool>> predicate, CancellationToken cancellationToken = default);
/// <summary>
/// Delete SINGLE entity matching expression.
/// SAFETY: Throws exception if zero or multiple entities match the predicate.
/// </summary>
/// <exception cref="InvalidOperationException">Thrown when zero or multiple entities match</exception>
Task DeleteSingleAsync(Expression<Func<TEntity, bool>> predicate, CancellationToken cancellationToken = default);
}

View File

@@ -1,33 +0,0 @@
namespace DigitalData.EmailProfiler.Application.Interfaces.Repositories;
/// <summary>
/// Unit of Work pattern for transaction management.
/// </summary>
public interface IUnitOfWork : IDisposable
{
IEmailAccountRepository EmailAccounts { get; }
IEmailProfileRepository EmailProfiles { get; }
IEmailProcessRepository EmailProcesses { get; }
IEmailHistoryRepository EmailHistories { get; }
IEmailOutboxRepository EmailOutbox { get; }
/// <summary>
/// Save all changes to the database.
/// </summary>
Task<int> SaveChangesAsync(CancellationToken cancellationToken = default);
/// <summary>
/// Begin a database transaction.
/// </summary>
Task BeginTransactionAsync(CancellationToken cancellationToken = default);
/// <summary>
/// Commit the current transaction.
/// </summary>
Task CommitTransactionAsync(CancellationToken cancellationToken = default);
/// <summary>
/// Rollback the current transaction.
/// </summary>
Task RollbackTransactionAsync(CancellationToken cancellationToken = default);
}

View File

@@ -0,0 +1,32 @@
using DigitalData.EmailProfiler.Application.Common.Interfaces;
using DigitalData.EmailProfiler.Infrastructure.Messaging;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
namespace DigitalData.EmailProfiler.Infrastructure;
/// <summary>
/// Dependency injection configuration for Infrastructure layer
/// </summary>
public static class DependencyInjection
{
/// <summary>
/// Adds Infrastructure layer services to the DI container
/// </summary>
public static IServiceCollection AddInfrastructure(
this IServiceCollection services,
IConfiguration configuration)
{
// Register RabbitMQ configuration
services.Configure<RabbitMqConfiguration>(
configuration.GetSection(RabbitMqConfiguration.SectionName));
// Register RabbitMQ command publisher
services.AddSingleton<ICommandPublisher, RabbitMqCommandPublisher>();
// Register RabbitMQ command consumer as hosted service
services.AddHostedService<RabbitMqCommandConsumer>();
return services;
}
}

View File

@@ -8,6 +8,13 @@
<ItemGroup>
<ProjectReference Include="..\DigitalData.EmailProfiler.Domain\DigitalData.EmailProfiler.Domain.csproj" />
<ProjectReference Include="..\DigitalData.EmailProfiler.Application\DigitalData.EmailProfiler.Application.csproj" />
</ItemGroup>
<ItemGroup>
<PackageReference Include="Microsoft.Extensions.Hosting" Version="10.0.9" />
<PackageReference Include="Microsoft.Extensions.Options.ConfigurationExtensions" Version="10.0.9" />
<PackageReference Include="RabbitMQ.Client" Version="7.2.1" />
</ItemGroup>
</Project>

View File

@@ -0,0 +1,188 @@
using System.Text;
using System.Text.Json;
using DigitalData.EmailProfiler.Infrastructure.Messaging;
using MediatR;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
namespace DigitalData.EmailProfiler.Infrastructure.Messaging;
/// <summary>
/// Background service that consumes commands from RabbitMQ and executes them via MediatR
/// </summary>
public class RabbitMqCommandConsumer(
IOptions<RabbitMqConfiguration> config,
IServiceProvider ServiceProvider,
ILogger<RabbitMqCommandConsumer>? Logger) : BackgroundService
{
private readonly RabbitMqConfiguration Config = config.Value ?? throw new ArgumentNullException(nameof(config));
private IConnection? Connection;
private IChannel? _channel;
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
Logger?.LogInformation("RabbitMQ Command Consumer starting...");
try
{
// Create connection factory
var factory = new ConnectionFactory
{
HostName = Config.HostName,
Port = Config.Port,
UserName = Config.UserName,
Password = Config.Password,
VirtualHost = Config.VirtualHost,
AutomaticRecoveryEnabled = Config.AutomaticRecoveryEnabled,
NetworkRecoveryInterval = TimeSpan.FromSeconds(Config.NetworkRecoveryIntervalSeconds)
};
// Create connection and channel
Connection = await factory.CreateConnectionAsync(stoppingToken);
_channel = await Connection.CreateChannelAsync(cancellationToken: stoppingToken);
// Set prefetch count (process one message at a time)
await _channel.BasicQosAsync(prefetchSize: 0, prefetchCount: 1, global: false, cancellationToken: stoppingToken);
// Create async consumer
var consumer = new AsyncEventingBasicConsumer(_channel);
consumer.ReceivedAsync += async (sender, ea) =>
{
try
{
await ProcessMessageAsync(ea, stoppingToken);
await _channel.BasicAckAsync(ea.DeliveryTag, multiple: false, cancellationToken: stoppingToken);
}
catch (Exception ex)
{
Logger?.LogError(ex, "Error processing message {MessageId}", ea.BasicProperties?.MessageId);
// Reject and requeue on error (be careful with infinite loops!)
await _channel.BasicNackAsync(ea.DeliveryTag, multiple: false, requeue: true, cancellationToken: stoppingToken);
}
};
// Start consuming
await _channel.BasicConsumeAsync(
queue: Config.QueueName,
autoAck: false,
consumer: consumer,
cancellationToken: stoppingToken);
Logger?.LogInformation(
"RabbitMQ Command Consumer started. Listening on queue: {QueueName}",
Config.QueueName);
// Keep running until cancellation requested
// Consumer will process messages in the background via event handlers
var tcs = new TaskCompletionSource();
stoppingToken.Register(() => tcs.SetResult());
await tcs.Task;
}
catch (OperationCanceledException)
{
Logger?.LogInformation("RabbitMQ Command Consumer is stopping due to cancellation");
}
catch (Exception ex)
{
Logger?.LogError(ex, "Fatal error in RabbitMQ Command Consumer");
throw;
}
}
private async Task ProcessMessageAsync(BasicDeliverEventArgs ea, CancellationToken cancellationToken)
{
var messageId = ea.BasicProperties?.MessageId ?? "unknown";
var commandType = ea.BasicProperties?.Type ?? "unknown";
Logger?.LogInformation(
"Processing command {CommandType} with MessageId {MessageId}",
commandType, messageId);
// Deserialize message envelope
var json = Encoding.UTF8.GetString(ea.Body.ToArray());
var envelope = JsonSerializer.Deserialize<CommandEnvelope>(json);
if (envelope == null)
{
Logger?.LogWarning("Failed to deserialize command envelope for MessageId {MessageId}", messageId);
return;
}
// Get command type from assembly
var type = Type.GetType(envelope.CommandType);
if (type == null)
{
Logger?.LogWarning(
"Command type {CommandType} not found in assembly for MessageId {MessageId}",
envelope.CommandType, messageId);
return;
}
// Deserialize command payload
var command = JsonSerializer.Deserialize(envelope.Payload, type);
if (command == null)
{
Logger?.LogWarning(
"Failed to deserialize command payload for MessageId {MessageId}",
messageId);
return;
}
// Create scope and execute command via MediatR
using var scope = ServiceProvider.CreateScope();
var mediator = scope.ServiceProvider.GetRequiredService<IMediator>();
try
{
// Send command to MediatR (fire-and-forget)
await mediator.Send(command, cancellationToken);
Logger?.LogInformation(
"Successfully processed command {CommandType} with MessageId {MessageId}",
commandType, messageId);
}
catch (Exception ex)
{
Logger?.LogError(
ex,
"Error executing command {CommandType} with MessageId {MessageId}",
commandType, messageId);
throw;
}
}
public override async Task StopAsync(CancellationToken cancellationToken)
{
Logger?.LogInformation("RabbitMQ Command Consumer stopping...");
if (_channel != null)
{
await _channel.CloseAsync(cancellationToken);
_channel.Dispose();
}
if (Connection != null)
{
await Connection.CloseAsync(cancellationToken);
Connection.Dispose();
}
await base.StopAsync(cancellationToken);
}
/// <summary>
/// Message envelope for command deserialization
/// </summary>
private class CommandEnvelope
{
public string CommandType { get; set; } = string.Empty;
public string Payload { get; set; } = string.Empty;
public DateTime PublishedAt { get; set; }
public string CorrelationId { get; set; } = string.Empty;
}
}

View File

@@ -0,0 +1,158 @@
using System.Text;
using System.Text.Json;
using DigitalData.EmailProfiler.Application.Common.Interfaces;
using MediatR;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using RabbitMQ.Client;
namespace DigitalData.EmailProfiler.Infrastructure.Messaging;
/// <summary>
/// RabbitMQ implementation of ICommandPublisher for asynchronous command processing
/// </summary>
public class RabbitMqCommandPublisher : ICommandPublisher, IDisposable
{
private readonly ILogger<RabbitMqCommandPublisher> Logger;
private readonly RabbitMqConfiguration Config;
private readonly IConnection Connection;
private readonly IChannel Channel;
private bool Disposed { get; set; }
public RabbitMqCommandPublisher(
IOptions<RabbitMqConfiguration> config,
ILogger<RabbitMqCommandPublisher> logger)
{
Config = config.Value ?? throw new ArgumentNullException(nameof(config));
Logger = logger ?? throw new ArgumentNullException(nameof(logger));
try
{
// Create connection factory
var factory = new ConnectionFactory
{
HostName = Config.HostName,
Port = Config.Port,
UserName = Config.UserName,
Password = Config.Password,
VirtualHost = Config.VirtualHost,
AutomaticRecoveryEnabled = Config.AutomaticRecoveryEnabled,
NetworkRecoveryInterval = TimeSpan.FromSeconds(Config.NetworkRecoveryIntervalSeconds)
};
// Create connection and channel
Connection = factory.CreateConnectionAsync().GetAwaiter().GetResult();
Channel = Connection.CreateChannelAsync().GetAwaiter().GetResult();
// Declare exchange (fanout for broadcasting commands)
Channel.ExchangeDeclareAsync(
exchange: Config.ExchangeName,
type: ExchangeType.Direct,
durable: true,
autoDelete: false).GetAwaiter().GetResult();
// Declare queue
Channel.QueueDeclareAsync(
queue: Config.QueueName,
durable: true,
exclusive: false,
autoDelete: false,
arguments: null).GetAwaiter().GetResult();
// Bind queue to exchange
Channel.QueueBindAsync(
queue: Config.QueueName,
exchange: Config.ExchangeName,
routingKey: Config.RoutingKey).GetAwaiter().GetResult();
Logger.LogInformation(
"RabbitMQ connection established: {HostName}:{Port}, Exchange: {Exchange}, Queue: {Queue}",
Config.HostName, Config.Port, Config.ExchangeName, Config.QueueName);
}
catch (Exception ex)
{
Logger.LogError(ex, "Failed to establish RabbitMQ connection");
throw;
}
}
/// <summary>
/// Publishes a command to RabbitMQ for asynchronous processing
/// </summary>
public async Task PublishAsync<TCommand>(TCommand command, CancellationToken cancellationToken = default)
where TCommand : IBaseRequest
{
ObjectDisposedException.ThrowIf(Disposed, typeof(RabbitMqCommandPublisher));
try
{
// Create message envelope with metadata
var envelope = new CommandEnvelope
{
CommandType = typeof(TCommand).AssemblyQualifiedName!,
Payload = JsonSerializer.Serialize(command),
PublishedAt = DateTime.Now,
CorrelationId = Guid.NewGuid().ToString()
};
// Serialize to JSON
var json = JsonSerializer.Serialize(envelope);
var body = Encoding.UTF8.GetBytes(json);
// Set message properties
var properties = new BasicProperties
{
Persistent = true,
ContentType = "application/json",
MessageId = envelope.CorrelationId,
Timestamp = new AmqpTimestamp(DateTimeOffset.Now.ToUnixTimeSeconds()),
Type = typeof(TCommand).Name
};
// Publish to exchange
await Channel.BasicPublishAsync(
exchange: Config.ExchangeName,
routingKey: Config.RoutingKey,
mandatory: false,
basicProperties: properties,
body: body,
cancellationToken: cancellationToken);
Logger.LogInformation(
"Published command {CommandType} with CorrelationId {CorrelationId} to RabbitMQ",
typeof(TCommand).Name, envelope.CorrelationId);
}
catch (Exception ex)
{
Logger.LogError(ex, "Failed to publish command {CommandType} to RabbitMQ", typeof(TCommand).Name);
throw;
}
}
public void Dispose()
{
if (Disposed)
return;
Channel?.Dispose();
Connection?.Dispose();
Disposed = true;
Logger.LogInformation("RabbitMQ connection disposed");
}
/// <summary>
/// Message envelope for command serialization
/// </summary>
private class CommandEnvelope
{
public string CommandType { get; set; } = string.Empty;
public string Payload { get; set; } = string.Empty;
public DateTime PublishedAt { get; set; }
public string CorrelationId { get; set; } = string.Empty;
}
}

View File

@@ -0,0 +1,62 @@
namespace DigitalData.EmailProfiler.Infrastructure.Messaging;
/// <summary>
/// Configuration for RabbitMQ connection
/// </summary>
public class RabbitMqConfiguration
{
/// <summary>
/// Configuration section name in appsettings.json
/// </summary>
public const string SectionName = "RabbitMQ";
/// <summary>
/// RabbitMQ server hostname
/// </summary>
public string HostName { get; set; } = "localhost";
/// <summary>
/// RabbitMQ AMQP port (default: 5672)
/// </summary>
public int Port { get; set; } = 5672;
/// <summary>
/// RabbitMQ username
/// </summary>
public string UserName { get; set; } = "guest";
/// <summary>
/// RabbitMQ password
/// </summary>
public string Password { get; set; } = "guest";
/// <summary>
/// Virtual host (default: /)
/// </summary>
public string VirtualHost { get; set; } = "/";
/// <summary>
/// Exchange name for commands
/// </summary>
public string ExchangeName { get; set; } = "emailprofiler.commands";
/// <summary>
/// Queue name for commands
/// </summary>
public string QueueName { get; set; } = "emailprofiler.command.queue";
/// <summary>
/// Routing key for commands
/// </summary>
public string RoutingKey { get; set; } = "command";
/// <summary>
/// Enable automatic recovery on connection failure
/// </summary>
public bool AutomaticRecoveryEnabled { get; set; } = true;
/// <summary>
/// Network recovery interval in seconds
/// </summary>
public int NetworkRecoveryIntervalSeconds { get; set; } = 10;
}