feat(backend): Implement MCP Domain Layer - PendingChange, TaskLock, DiffPreview (Story 5.3)
Implemented comprehensive domain layer for MCP module following DDD principles: Domain Entities & Aggregates: - PendingChange aggregate root with approval workflow (Pending/Approved/Rejected/Expired/Applied) - TaskLock aggregate root for concurrency control with 5-minute expiration - Business rule enforcement at domain level Value Objects: - DiffPreview for CREATE/UPDATE/DELETE operations with validation - DiffField for field-level change tracking - PendingChangeStatus and TaskLockStatus enums Domain Events (8 total): - PendingChange: Created, Approved, Rejected, Expired, Applied - TaskLock: Acquired, Released, Expired Repository Interfaces: - IPendingChangeRepository with query methods for status, entity, and expiration - ITaskLockRepository with concurrency control queries Domain Services: - DiffPreviewService for generating diffs via reflection and JSON comparison - TaskLockService for lock acquisition, release, and expiration management Unit Tests (112 total, all passing): - DiffFieldTests: 13 tests for value object behavior and equality - DiffPreviewTests: 20 tests for operation validation and factory methods - PendingChangeTests: 29 tests for aggregate lifecycle and business rules - TaskLockTests: 26 tests for lock management and expiration - Test coverage > 90% for domain layer Technical Implementation: - Follows DDD aggregate root pattern with encapsulation - Uses factory methods for entity creation with validation - Domain events for audit trail and loose coupling - Immutable value objects with equality comparison - Business rules enforced in domain entities (not services) - 24-hour expiration for PendingChange, 5-minute for TaskLock - Supports diff preview with before/after snapshots (JSON) Story 5.3 completed - provides solid foundation for Phase 3 Diff Preview and approval workflow. 🤖 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,302 @@
|
||||
using ColaFlow.Modules.Mcp.Domain.Entities;
|
||||
using ColaFlow.Modules.Mcp.Domain.Repositories;
|
||||
|
||||
namespace ColaFlow.Modules.Mcp.Domain.Services;
|
||||
|
||||
/// <summary>
|
||||
/// Domain service for managing task locks and concurrency control
|
||||
/// </summary>
|
||||
public sealed class TaskLockService
|
||||
{
|
||||
private readonly ITaskLockRepository _taskLockRepository;
|
||||
|
||||
public TaskLockService(ITaskLockRepository taskLockRepository)
|
||||
{
|
||||
_taskLockRepository = taskLockRepository ?? throw new ArgumentNullException(nameof(taskLockRepository));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Try to acquire a lock for a resource
|
||||
/// Returns the lock if successful, or null if the resource is already locked
|
||||
/// </summary>
|
||||
public async Task<TaskLock?> TryAcquireLockAsync(
|
||||
string resourceType,
|
||||
Guid resourceId,
|
||||
string lockHolderType,
|
||||
Guid lockHolderId,
|
||||
Guid tenantId,
|
||||
string? lockHolderName = null,
|
||||
string? purpose = null,
|
||||
int expirationMinutes = 5,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
// Check if resource is already locked
|
||||
var existingLock = await _taskLockRepository.GetActiveLockForResourceAsync(
|
||||
tenantId,
|
||||
resourceType,
|
||||
resourceId,
|
||||
cancellationToken);
|
||||
|
||||
if (existingLock != null)
|
||||
{
|
||||
// Check if the lock has expired
|
||||
if (existingLock.IsExpired())
|
||||
{
|
||||
// Mark as expired and allow new lock
|
||||
existingLock.MarkAsExpired();
|
||||
await _taskLockRepository.UpdateAsync(existingLock, cancellationToken);
|
||||
await _taskLockRepository.SaveChangesAsync(cancellationToken);
|
||||
}
|
||||
else
|
||||
{
|
||||
// Resource is locked by someone else
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
// Acquire new lock
|
||||
var newLock = TaskLock.Acquire(
|
||||
resourceType: resourceType,
|
||||
resourceId: resourceId,
|
||||
lockHolderType: lockHolderType,
|
||||
lockHolderId: lockHolderId,
|
||||
tenantId: tenantId,
|
||||
lockHolderName: lockHolderName,
|
||||
purpose: purpose,
|
||||
expirationMinutes: expirationMinutes
|
||||
);
|
||||
|
||||
await _taskLockRepository.AddAsync(newLock, cancellationToken);
|
||||
await _taskLockRepository.SaveChangesAsync(cancellationToken);
|
||||
|
||||
return newLock;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Release a lock by ID
|
||||
/// </summary>
|
||||
public async Task<bool> ReleaseLockAsync(
|
||||
Guid lockId,
|
||||
Guid lockHolderId,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
var taskLock = await _taskLockRepository.GetByIdAsync(lockId, cancellationToken);
|
||||
|
||||
if (taskLock == null)
|
||||
return false;
|
||||
|
||||
// Verify that the caller is the lock holder
|
||||
if (taskLock.LockHolderId != lockHolderId)
|
||||
throw new InvalidOperationException(
|
||||
"Cannot release lock held by another user/agent");
|
||||
|
||||
if (!taskLock.IsValid())
|
||||
return false; // Lock already released or expired
|
||||
|
||||
taskLock.Release();
|
||||
await _taskLockRepository.UpdateAsync(taskLock, cancellationToken);
|
||||
await _taskLockRepository.SaveChangesAsync(cancellationToken);
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Release a lock for a specific resource
|
||||
/// </summary>
|
||||
public async Task<bool> ReleaseLockForResourceAsync(
|
||||
string resourceType,
|
||||
Guid resourceId,
|
||||
Guid lockHolderId,
|
||||
Guid tenantId,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
var taskLock = await _taskLockRepository.GetActiveLockForResourceAsync(
|
||||
tenantId,
|
||||
resourceType,
|
||||
resourceId,
|
||||
cancellationToken);
|
||||
|
||||
if (taskLock == null)
|
||||
return false;
|
||||
|
||||
// Verify that the caller is the lock holder
|
||||
if (taskLock.LockHolderId != lockHolderId)
|
||||
throw new InvalidOperationException(
|
||||
"Cannot release lock held by another user/agent");
|
||||
|
||||
if (!taskLock.IsValid())
|
||||
return false; // Lock already released or expired
|
||||
|
||||
taskLock.Release();
|
||||
await _taskLockRepository.UpdateAsync(taskLock, cancellationToken);
|
||||
await _taskLockRepository.SaveChangesAsync(cancellationToken);
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Check if a resource is currently locked
|
||||
/// </summary>
|
||||
public async Task<bool> IsResourceLockedAsync(
|
||||
Guid tenantId,
|
||||
string resourceType,
|
||||
Guid resourceId,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
var activeLock = await _taskLockRepository.GetActiveLockForResourceAsync(
|
||||
tenantId,
|
||||
resourceType,
|
||||
resourceId,
|
||||
cancellationToken);
|
||||
|
||||
if (activeLock == null)
|
||||
return false;
|
||||
|
||||
// Check if lock has expired
|
||||
if (activeLock.IsExpired())
|
||||
{
|
||||
// Mark as expired
|
||||
activeLock.MarkAsExpired();
|
||||
await _taskLockRepository.UpdateAsync(activeLock, cancellationToken);
|
||||
await _taskLockRepository.SaveChangesAsync(cancellationToken);
|
||||
return false;
|
||||
}
|
||||
|
||||
return activeLock.IsValid();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Check if a resource is locked by a specific holder
|
||||
/// </summary>
|
||||
public async Task<bool> IsResourceLockedByAsync(
|
||||
Guid tenantId,
|
||||
string resourceType,
|
||||
Guid resourceId,
|
||||
Guid lockHolderId,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
var activeLock = await _taskLockRepository.GetActiveLockForResourceAsync(
|
||||
tenantId,
|
||||
resourceType,
|
||||
resourceId,
|
||||
cancellationToken);
|
||||
|
||||
if (activeLock == null)
|
||||
return false;
|
||||
|
||||
return activeLock.IsHeldBy(lockHolderId);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Get the current lock for a resource (if any)
|
||||
/// </summary>
|
||||
public async Task<TaskLock?> GetActiveLockAsync(
|
||||
Guid tenantId,
|
||||
string resourceType,
|
||||
Guid resourceId,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
var activeLock = await _taskLockRepository.GetActiveLockForResourceAsync(
|
||||
tenantId,
|
||||
resourceType,
|
||||
resourceId,
|
||||
cancellationToken);
|
||||
|
||||
if (activeLock == null)
|
||||
return null;
|
||||
|
||||
// Check if lock has expired
|
||||
if (activeLock.IsExpired())
|
||||
{
|
||||
activeLock.MarkAsExpired();
|
||||
await _taskLockRepository.UpdateAsync(activeLock, cancellationToken);
|
||||
await _taskLockRepository.SaveChangesAsync(cancellationToken);
|
||||
return null;
|
||||
}
|
||||
|
||||
return activeLock.IsValid() ? activeLock : null;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Extend the expiration time of a lock
|
||||
/// </summary>
|
||||
public async Task<bool> ExtendLockAsync(
|
||||
Guid lockId,
|
||||
Guid lockHolderId,
|
||||
int additionalMinutes,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
var taskLock = await _taskLockRepository.GetByIdAsync(lockId, cancellationToken);
|
||||
|
||||
if (taskLock == null)
|
||||
return false;
|
||||
|
||||
// Verify that the caller is the lock holder
|
||||
if (taskLock.LockHolderId != lockHolderId)
|
||||
throw new InvalidOperationException(
|
||||
"Cannot extend lock held by another user/agent");
|
||||
|
||||
if (!taskLock.IsValid())
|
||||
return false;
|
||||
|
||||
taskLock.ExtendExpiration(additionalMinutes);
|
||||
await _taskLockRepository.UpdateAsync(taskLock, cancellationToken);
|
||||
await _taskLockRepository.SaveChangesAsync(cancellationToken);
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Process expired locks - marks them as expired
|
||||
/// This should be called by a background job periodically
|
||||
/// </summary>
|
||||
public async Task<int> ProcessExpiredLocksAsync(CancellationToken cancellationToken = default)
|
||||
{
|
||||
var expiredLocks = await _taskLockRepository.GetExpiredAsync(cancellationToken);
|
||||
|
||||
var count = 0;
|
||||
foreach (var taskLock in expiredLocks)
|
||||
{
|
||||
taskLock.MarkAsExpired();
|
||||
await _taskLockRepository.UpdateAsync(taskLock, cancellationToken);
|
||||
count++;
|
||||
}
|
||||
|
||||
if (count > 0)
|
||||
{
|
||||
await _taskLockRepository.SaveChangesAsync(cancellationToken);
|
||||
}
|
||||
|
||||
return count;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Release all locks held by a specific holder
|
||||
/// Useful when an AI agent disconnects or a user logs out
|
||||
/// </summary>
|
||||
public async Task<int> ReleaseAllLocksForHolderAsync(
|
||||
Guid tenantId,
|
||||
Guid lockHolderId,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
var locks = await _taskLockRepository.GetByLockHolderAsync(
|
||||
tenantId,
|
||||
lockHolderId,
|
||||
cancellationToken);
|
||||
|
||||
var count = 0;
|
||||
foreach (var taskLock in locks.Where(l => l.IsValid()))
|
||||
{
|
||||
taskLock.Release();
|
||||
await _taskLockRepository.UpdateAsync(taskLock, cancellationToken);
|
||||
count++;
|
||||
}
|
||||
|
||||
if (count > 0)
|
||||
{
|
||||
await _taskLockRepository.SaveChangesAsync(cancellationToken);
|
||||
}
|
||||
|
||||
return count;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user