feat(backend): Implement PendingChange Management (Story 5.10)
Implemented complete Human-in-the-Loop approval workflow for AI-proposed changes: Changes: - Created PendingChange DTOs (PendingChangeDto, CreatePendingChangeRequest, ApproveChangeRequest, RejectChangeRequest, PendingChangeFilterDto) - Implemented IPendingChangeService interface with CRUD, approval/rejection, expiration, and deletion operations - Implemented PendingChangeService with full workflow support and tenant isolation - Created McpPendingChangesController REST API with endpoints for listing, approving, rejecting, and deleting pending changes - Implemented PendingChangeApprovedEventHandler to execute approved changes via MediatR commands (Project, Epic, Story, Task CRUD operations) - Created PendingChangeExpirationBackgroundService for auto-expiration of changes after 24 hours - Registered all services and background service in DI container Technical Details: - Status flow: PendingApproval → Approved → Applied (or Rejected/Expired) - Tenant isolation enforced in all operations - Domain events published for audit trail - Event-driven execution using MediatR - Background service runs every 5 minutes to expire old changes - JWT authentication required for all endpoints 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Yaojia Wang
@@ -0,0 +1,410 @@
|
||||
using ColaFlow.Modules.Mcp.Application.DTOs;
|
||||
using ColaFlow.Modules.Mcp.Domain.Entities;
|
||||
using ColaFlow.Modules.Mcp.Domain.Exceptions;
|
||||
using ColaFlow.Modules.Mcp.Domain.Repositories;
|
||||
using ColaFlow.Modules.Mcp.Domain.ValueObjects;
|
||||
using ColaFlow.Modules.ProjectManagement.Application.Common.Interfaces;
|
||||
using MediatR;
|
||||
using Microsoft.AspNetCore.Http;
|
||||
using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace ColaFlow.Modules.Mcp.Application.Services;
|
||||
|
||||
/// <summary>
|
||||
/// Service implementation for PendingChange management
|
||||
/// </summary>
|
||||
public class PendingChangeService : IPendingChangeService
|
||||
{
|
||||
private readonly IPendingChangeRepository _repository;
|
||||
private readonly ITenantContext _tenantContext;
|
||||
private readonly IHttpContextAccessor _httpContextAccessor;
|
||||
private readonly IPublisher _publisher;
|
||||
private readonly ILogger<PendingChangeService> _logger;
|
||||
|
||||
public PendingChangeService(
|
||||
IPendingChangeRepository repository,
|
||||
ITenantContext tenantContext,
|
||||
IHttpContextAccessor httpContextAccessor,
|
||||
IPublisher publisher,
|
||||
ILogger<PendingChangeService> logger)
|
||||
{
|
||||
_repository = repository ?? throw new ArgumentNullException(nameof(repository));
|
||||
_tenantContext = tenantContext ?? throw new ArgumentNullException(nameof(tenantContext));
|
||||
_httpContextAccessor = httpContextAccessor ?? throw new ArgumentNullException(nameof(httpContextAccessor));
|
||||
_publisher = publisher ?? throw new ArgumentNullException(nameof(publisher));
|
||||
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
|
||||
}
|
||||
|
||||
public async Task<PendingChangeDto> CreateAsync(
|
||||
CreatePendingChangeRequest request,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
var tenantId = _tenantContext.GetCurrentTenantId();
|
||||
|
||||
// Get API Key ID from HttpContext (set by MCP authentication middleware)
|
||||
var apiKeyIdNullable = _httpContextAccessor.HttpContext?.Items["ApiKeyId"] as Guid?;
|
||||
if (!apiKeyIdNullable.HasValue)
|
||||
{
|
||||
throw new McpUnauthorizedException("API Key not found in request context");
|
||||
}
|
||||
var apiKeyId = apiKeyIdNullable.Value;
|
||||
|
||||
_logger.LogInformation(
|
||||
"Creating PendingChange: Tool={ToolName}, Operation={Operation}, EntityType={EntityType}, Tenant={TenantId}",
|
||||
request.ToolName, request.Diff.Operation, request.Diff.EntityType, tenantId);
|
||||
|
||||
// Create PendingChange entity
|
||||
var pendingChange = PendingChange.Create(
|
||||
request.ToolName,
|
||||
request.Diff,
|
||||
tenantId,
|
||||
apiKeyId,
|
||||
request.ExpirationHours);
|
||||
|
||||
// Save to database
|
||||
await _repository.AddAsync(pendingChange, cancellationToken);
|
||||
await _repository.SaveChangesAsync(cancellationToken);
|
||||
|
||||
// Publish domain events
|
||||
foreach (var domainEvent in pendingChange.DomainEvents)
|
||||
{
|
||||
await _publisher.Publish(domainEvent, cancellationToken);
|
||||
}
|
||||
pendingChange.ClearDomainEvents();
|
||||
|
||||
_logger.LogInformation(
|
||||
"PendingChange created: {Id}, ExpiresAt={ExpiresAt}",
|
||||
pendingChange.Id, pendingChange.ExpiresAt);
|
||||
|
||||
return MapToDto(pendingChange);
|
||||
}
|
||||
|
||||
public async Task<PendingChangeDto?> GetByIdAsync(
|
||||
Guid id,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
var tenantId = _tenantContext.GetCurrentTenantId();
|
||||
|
||||
var pendingChange = await _repository.GetByIdAsync(id, cancellationToken);
|
||||
if (pendingChange == null)
|
||||
{
|
||||
return null;
|
||||
}
|
||||
|
||||
// Verify tenant isolation
|
||||
if (pendingChange.TenantId != tenantId)
|
||||
{
|
||||
_logger.LogWarning(
|
||||
"Attempted cross-tenant access: PendingChange {Id} belongs to Tenant {OwnerId}, but requested by Tenant {RequesterId}",
|
||||
id, pendingChange.TenantId, tenantId);
|
||||
return null;
|
||||
}
|
||||
|
||||
return MapToDto(pendingChange);
|
||||
}
|
||||
|
||||
public async Task<(List<PendingChangeDto> Items, int TotalCount)> GetPendingChangesAsync(
|
||||
PendingChangeFilterDto filter,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
var tenantId = _tenantContext.GetCurrentTenantId();
|
||||
|
||||
// Build query
|
||||
var query = (await _repository.GetByTenantAsync(tenantId, cancellationToken))
|
||||
.AsEnumerable();
|
||||
|
||||
// Apply filters
|
||||
if (filter.Status.HasValue)
|
||||
{
|
||||
query = query.Where(x => x.Status == filter.Status.Value);
|
||||
}
|
||||
|
||||
if (!string.IsNullOrWhiteSpace(filter.EntityType))
|
||||
{
|
||||
query = query.Where(x => x.Diff.EntityType == filter.EntityType);
|
||||
}
|
||||
|
||||
if (filter.EntityId.HasValue)
|
||||
{
|
||||
query = query.Where(x => x.Diff.EntityId == filter.EntityId.Value);
|
||||
}
|
||||
|
||||
if (filter.ApiKeyId.HasValue)
|
||||
{
|
||||
query = query.Where(x => x.ApiKeyId == filter.ApiKeyId.Value);
|
||||
}
|
||||
|
||||
if (!string.IsNullOrWhiteSpace(filter.ToolName))
|
||||
{
|
||||
query = query.Where(x => x.ToolName == filter.ToolName);
|
||||
}
|
||||
|
||||
if (filter.IncludeExpired == false)
|
||||
{
|
||||
query = query.Where(x => x.Status != PendingChangeStatus.Expired);
|
||||
}
|
||||
|
||||
// Get total count before pagination
|
||||
var totalCount = query.Count();
|
||||
|
||||
// Apply pagination
|
||||
var items = query
|
||||
.OrderByDescending(x => x.CreatedAt)
|
||||
.Skip((filter.Page - 1) * filter.PageSize)
|
||||
.Take(filter.PageSize)
|
||||
.Select(MapToDto)
|
||||
.ToList();
|
||||
|
||||
_logger.LogInformation(
|
||||
"Retrieved {Count}/{Total} PendingChanges for Tenant {TenantId} (Page {Page}/{PageSize})",
|
||||
items.Count, totalCount, tenantId, filter.Page, filter.PageSize);
|
||||
|
||||
return (items, totalCount);
|
||||
}
|
||||
|
||||
public async Task ApproveAsync(
|
||||
Guid pendingChangeId,
|
||||
Guid approvedBy,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
var tenantId = _tenantContext.GetCurrentTenantId();
|
||||
|
||||
var pendingChange = await _repository.GetByIdAsync(pendingChangeId, cancellationToken);
|
||||
if (pendingChange == null)
|
||||
{
|
||||
throw new McpNotFoundException("PendingChange", pendingChangeId.ToString());
|
||||
}
|
||||
|
||||
// Verify tenant isolation
|
||||
if (pendingChange.TenantId != tenantId)
|
||||
{
|
||||
throw new McpForbiddenException(
|
||||
$"Cannot approve PendingChange from different tenant");
|
||||
}
|
||||
|
||||
_logger.LogInformation(
|
||||
"Approving PendingChange {Id} by User {UserId}",
|
||||
pendingChangeId, approvedBy);
|
||||
|
||||
// Domain method validates business rules and raises PendingChangeApprovedEvent
|
||||
pendingChange.Approve(approvedBy);
|
||||
|
||||
await _repository.UpdateAsync(pendingChange, cancellationToken);
|
||||
await _repository.SaveChangesAsync(cancellationToken);
|
||||
|
||||
// Publish domain events (will trigger operation execution)
|
||||
foreach (var domainEvent in pendingChange.DomainEvents)
|
||||
{
|
||||
await _publisher.Publish(domainEvent, cancellationToken);
|
||||
}
|
||||
pendingChange.ClearDomainEvents();
|
||||
|
||||
_logger.LogInformation(
|
||||
"PendingChange {Id} approved successfully",
|
||||
pendingChangeId);
|
||||
}
|
||||
|
||||
public async Task RejectAsync(
|
||||
Guid pendingChangeId,
|
||||
Guid rejectedBy,
|
||||
string reason,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
var tenantId = _tenantContext.GetCurrentTenantId();
|
||||
|
||||
var pendingChange = await _repository.GetByIdAsync(pendingChangeId, cancellationToken);
|
||||
if (pendingChange == null)
|
||||
{
|
||||
throw new McpNotFoundException("PendingChange", pendingChangeId.ToString());
|
||||
}
|
||||
|
||||
// Verify tenant isolation
|
||||
if (pendingChange.TenantId != tenantId)
|
||||
{
|
||||
throw new McpForbiddenException(
|
||||
$"Cannot reject PendingChange from different tenant");
|
||||
}
|
||||
|
||||
_logger.LogInformation(
|
||||
"Rejecting PendingChange {Id} by User {UserId} - Reason: {Reason}",
|
||||
pendingChangeId, rejectedBy, reason);
|
||||
|
||||
// Domain method validates business rules and raises PendingChangeRejectedEvent
|
||||
pendingChange.Reject(rejectedBy, reason);
|
||||
|
||||
await _repository.UpdateAsync(pendingChange, cancellationToken);
|
||||
await _repository.SaveChangesAsync(cancellationToken);
|
||||
|
||||
// Publish domain events
|
||||
foreach (var domainEvent in pendingChange.DomainEvents)
|
||||
{
|
||||
await _publisher.Publish(domainEvent, cancellationToken);
|
||||
}
|
||||
pendingChange.ClearDomainEvents();
|
||||
|
||||
_logger.LogInformation(
|
||||
"PendingChange {Id} rejected successfully",
|
||||
pendingChangeId);
|
||||
}
|
||||
|
||||
public async Task MarkAsAppliedAsync(
|
||||
Guid pendingChangeId,
|
||||
string result,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
var pendingChange = await _repository.GetByIdAsync(pendingChangeId, cancellationToken);
|
||||
if (pendingChange == null)
|
||||
{
|
||||
throw new McpNotFoundException("PendingChange", pendingChangeId.ToString());
|
||||
}
|
||||
|
||||
_logger.LogInformation(
|
||||
"Marking PendingChange {Id} as Applied - Result: {Result}",
|
||||
pendingChangeId, result);
|
||||
|
||||
// Domain method validates business rules and raises PendingChangeAppliedEvent
|
||||
pendingChange.MarkAsApplied(result);
|
||||
|
||||
await _repository.UpdateAsync(pendingChange, cancellationToken);
|
||||
await _repository.SaveChangesAsync(cancellationToken);
|
||||
|
||||
// Publish domain events
|
||||
foreach (var domainEvent in pendingChange.DomainEvents)
|
||||
{
|
||||
await _publisher.Publish(domainEvent, cancellationToken);
|
||||
}
|
||||
pendingChange.ClearDomainEvents();
|
||||
|
||||
_logger.LogInformation(
|
||||
"PendingChange {Id} marked as Applied",
|
||||
pendingChangeId);
|
||||
}
|
||||
|
||||
public async Task<int> ExpireOldChangesAsync(CancellationToken cancellationToken = default)
|
||||
{
|
||||
_logger.LogInformation("Starting expiration check for old PendingChanges");
|
||||
|
||||
var expiredChanges = await _repository.GetExpiredAsync(cancellationToken);
|
||||
|
||||
var count = 0;
|
||||
foreach (var change in expiredChanges)
|
||||
{
|
||||
try
|
||||
{
|
||||
change.Expire();
|
||||
await _repository.UpdateAsync(change, cancellationToken);
|
||||
|
||||
// Publish domain events
|
||||
foreach (var domainEvent in change.DomainEvents)
|
||||
{
|
||||
await _publisher.Publish(domainEvent, cancellationToken);
|
||||
}
|
||||
change.ClearDomainEvents();
|
||||
|
||||
count++;
|
||||
|
||||
_logger.LogWarning(
|
||||
"PendingChange expired: {Id} - {ToolName} {Operation} {EntityType}",
|
||||
change.Id, change.ToolName, change.Diff.Operation, change.Diff.EntityType);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex,
|
||||
"Failed to expire PendingChange {Id}",
|
||||
change.Id);
|
||||
}
|
||||
}
|
||||
|
||||
if (count > 0)
|
||||
{
|
||||
await _repository.SaveChangesAsync(cancellationToken);
|
||||
}
|
||||
|
||||
_logger.LogInformation(
|
||||
"Expired {Count} PendingChanges",
|
||||
count);
|
||||
|
||||
return count;
|
||||
}
|
||||
|
||||
public async Task DeleteAsync(
|
||||
Guid pendingChangeId,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
var tenantId = _tenantContext.GetCurrentTenantId();
|
||||
|
||||
var pendingChange = await _repository.GetByIdAsync(pendingChangeId, cancellationToken);
|
||||
if (pendingChange == null)
|
||||
{
|
||||
throw new McpNotFoundException("PendingChange", pendingChangeId.ToString());
|
||||
}
|
||||
|
||||
// Verify tenant isolation
|
||||
if (pendingChange.TenantId != tenantId)
|
||||
{
|
||||
throw new McpForbiddenException(
|
||||
$"Cannot delete PendingChange from different tenant");
|
||||
}
|
||||
|
||||
// Only allow deletion of Expired or Rejected changes
|
||||
if (pendingChange.Status != PendingChangeStatus.Expired &&
|
||||
pendingChange.Status != PendingChangeStatus.Rejected)
|
||||
{
|
||||
throw new McpValidationException(
|
||||
$"Can only delete PendingChanges with Expired or Rejected status. Current status: {pendingChange.Status}");
|
||||
}
|
||||
|
||||
_logger.LogInformation(
|
||||
"Deleting PendingChange {Id} (Status: {Status})",
|
||||
pendingChangeId, pendingChange.Status);
|
||||
|
||||
await _repository.DeleteAsync(pendingChange, cancellationToken);
|
||||
await _repository.SaveChangesAsync(cancellationToken);
|
||||
|
||||
_logger.LogInformation(
|
||||
"PendingChange {Id} deleted successfully",
|
||||
pendingChangeId);
|
||||
}
|
||||
|
||||
private static PendingChangeDto MapToDto(PendingChange pendingChange)
|
||||
{
|
||||
return new PendingChangeDto
|
||||
{
|
||||
Id = pendingChange.Id,
|
||||
TenantId = pendingChange.TenantId,
|
||||
ApiKeyId = pendingChange.ApiKeyId,
|
||||
ToolName = pendingChange.ToolName,
|
||||
Diff = new DiffPreviewDto
|
||||
{
|
||||
Operation = pendingChange.Diff.Operation,
|
||||
EntityType = pendingChange.Diff.EntityType,
|
||||
EntityId = pendingChange.Diff.EntityId,
|
||||
EntityKey = pendingChange.Diff.EntityKey,
|
||||
BeforeData = pendingChange.Diff.BeforeData,
|
||||
AfterData = pendingChange.Diff.AfterData,
|
||||
ChangedFields = pendingChange.Diff.ChangedFields.Select(f => new DiffFieldDto
|
||||
{
|
||||
FieldName = f.FieldName,
|
||||
DisplayName = f.DisplayName,
|
||||
OldValue = f.OldValue,
|
||||
NewValue = f.NewValue,
|
||||
DiffHtml = f.DiffHtml
|
||||
}).ToList()
|
||||
},
|
||||
Status = pendingChange.Status.ToString(),
|
||||
CreatedAt = pendingChange.CreatedAt,
|
||||
ExpiresAt = pendingChange.ExpiresAt,
|
||||
ApprovedBy = pendingChange.ApprovedBy,
|
||||
ApprovedAt = pendingChange.ApprovedAt,
|
||||
RejectedBy = pendingChange.RejectedBy,
|
||||
RejectedAt = pendingChange.RejectedAt,
|
||||
RejectionReason = pendingChange.RejectionReason,
|
||||
AppliedAt = pendingChange.AppliedAt,
|
||||
ApplicationResult = pendingChange.ApplicationResult,
|
||||
IsExpired = pendingChange.IsExpired(),
|
||||
CanBeApproved = pendingChange.CanBeApproved(),
|
||||
CanBeRejected = pendingChange.CanBeRejected(),
|
||||
Summary = pendingChange.GetSummary()
|
||||
};
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user