diff --git a/AGENTS.md b/AGENTS.md index b1c5fad81..fd380a65f 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -8,6 +8,7 @@ Ignore the "archived" directory. - `build.sh` is passing without errors or warnings - Test coverage is greater than 80% and `coverage.sh` is not failing - Problems are not hidden, problems are addressed. +- Report breaking changes and ask how to handle them, do not assume migrations are required. # C# Code Style diff --git a/clean.sh b/clean.sh index 6661aba3b..ad7170f24 100755 --- a/clean.sh +++ b/clean.sh @@ -12,3 +12,5 @@ rm -rf src/Core/obj rm -rf src/Main/bin rm -rf src/Main/obj + +rm -rf tests/Core.Tests/TestResults diff --git a/src/Core/Core.csproj b/src/Core/Core.csproj index 3a5c20e56..2b8e87b72 100644 --- a/src/Core/Core.csproj +++ b/src/Core/Core.csproj @@ -6,4 +6,9 @@ net10.0 + + + + + diff --git a/src/Core/Storage/ContentStorageDbContext.cs b/src/Core/Storage/ContentStorageDbContext.cs new file mode 100644 index 000000000..f3e031a9e --- /dev/null +++ b/src/Core/Storage/ContentStorageDbContext.cs @@ -0,0 +1,178 @@ +using System.Globalization; +using KernelMemory.Core.Storage.Entities; +using Microsoft.EntityFrameworkCore; + +namespace KernelMemory.Core.Storage; + +/// +/// Database context for content storage. +/// Manages Content and Operations tables with proper SQLite configuration. +/// +public class ContentStorageDbContext : DbContext +{ + public DbSet Content { get; set; } = null!; + public DbSet Operations { get; set; } = null!; + + public ContentStorageDbContext(DbContextOptions options) : base(options) + { + } + + protected override void OnModelCreating(ModelBuilder modelBuilder) + { + base.OnModelCreating(modelBuilder); + + // Configure Content table + modelBuilder.Entity(entity => + { + // Hardcoded table name as per specification + entity.ToTable("km_content"); + + // Primary key + entity.HasKey(e => e.Id); + + // Required fields + entity.Property(e => e.Id) + .IsRequired() + .HasMaxLength(32); // Cuid2 is typically 25-32 characters + + entity.Property(e => e.Content) + .IsRequired(); + + entity.Property(e => e.MimeType) + .IsRequired() + .HasMaxLength(255); + + entity.Property(e => e.ByteSize) + .IsRequired(); + + entity.Property(e => e.Ready) + .IsRequired(); + + // DateTimeOffset stored as ISO 8601 string in SQLite + entity.Property(e => e.ContentCreatedAt) + .IsRequired() + .HasConversion( + v => v.ToString("O"), + v => DateTimeOffset.Parse(v, CultureInfo.InvariantCulture)); + + entity.Property(e => e.RecordCreatedAt) + .IsRequired() + .HasConversion( + v => v.ToString("O"), + v => DateTimeOffset.Parse(v, CultureInfo.InvariantCulture)); + + entity.Property(e => e.RecordUpdatedAt) + .IsRequired() + .HasConversion( + v => v.ToString("O"), + v => DateTimeOffset.Parse(v, CultureInfo.InvariantCulture)); + + // Optional fields + entity.Property(e => e.Title) + .IsRequired() + .HasDefaultValue(string.Empty); + + entity.Property(e => e.Description) + .IsRequired() + .HasDefaultValue(string.Empty); + + // JSON fields - store as TEXT in SQLite + entity.Property(e => e.TagsJson) + .IsRequired() + .HasColumnType("TEXT") + .HasDefaultValue("[]"); + + entity.Property(e => e.MetadataJson) + .IsRequired() + .HasColumnType("TEXT") + .HasDefaultValue("{}"); + + // Ignore computed properties (not stored in database) + entity.Ignore(e => e.Tags); + entity.Ignore(e => e.Metadata); + + // Indexes + entity.HasIndex(e => e.Ready) + .HasDatabaseName("IX_km_content_Ready"); + }); + + // Configure Operations table + modelBuilder.Entity(entity => + { + // Hardcoded table name as per specification + entity.ToTable("km_operations"); + + // Primary key + entity.HasKey(e => e.Id); + + // Required fields + entity.Property(e => e.Id) + .IsRequired() + .HasMaxLength(32); + + entity.Property(e => e.Complete) + .IsRequired(); + + entity.Property(e => e.Cancelled) + .IsRequired(); + + entity.Property(e => e.ContentId) + .IsRequired() + .HasMaxLength(32); + + // DateTimeOffset stored as ISO 8601 string in SQLite + entity.Property(e => e.Timestamp) + .IsRequired() + .HasConversion( + v => v.ToString("O"), + v => DateTimeOffset.Parse(v, CultureInfo.InvariantCulture)); + + entity.Property(e => e.LastFailureReason) + .IsRequired() + .HasDefaultValue(string.Empty); + + // Nullable DateTimeOffset for locking + entity.Property(e => e.LastAttemptTimestamp) + .HasConversion( + v => v.HasValue ? v.Value.ToString("O") : null, + v => v == null ? (DateTimeOffset?)null : DateTimeOffset.Parse(v, CultureInfo.InvariantCulture)); + + // JSON fields - store as TEXT in SQLite + entity.Property(e => e.PlannedStepsJson) + .IsRequired() + .HasColumnType("TEXT") + .HasDefaultValue("[]"); + + entity.Property(e => e.CompletedStepsJson) + .IsRequired() + .HasColumnType("TEXT") + .HasDefaultValue("[]"); + + entity.Property(e => e.RemainingStepsJson) + .IsRequired() + .HasColumnType("TEXT") + .HasDefaultValue("[]"); + + entity.Property(e => e.PayloadJson) + .IsRequired() + .HasColumnType("TEXT") + .HasDefaultValue("{}"); + + // Ignore computed properties (not stored in database) + entity.Ignore(e => e.PlannedSteps); + entity.Ignore(e => e.CompletedSteps); + entity.Ignore(e => e.RemainingSteps); + entity.Ignore(e => e.Payload); + + // Indexes as per specification + entity.HasIndex(e => new { e.ContentId, e.Timestamp }) + .HasDatabaseName("IX_km_operations_ContentId_Timestamp"); + + entity.HasIndex(e => new { e.Complete, e.Timestamp }) + .HasDatabaseName("IX_km_operations_Complete_Timestamp"); + + entity.HasIndex(e => e.Timestamp) + .HasDatabaseName("IX_km_operations_Timestamp"); + }); + } +} diff --git a/src/Core/Storage/ContentStorageService.cs b/src/Core/Storage/ContentStorageService.cs new file mode 100644 index 000000000..8141d7c7d --- /dev/null +++ b/src/Core/Storage/ContentStorageService.cs @@ -0,0 +1,567 @@ +using System.Text; +using System.Text.Json; +using KernelMemory.Core.Storage.Entities; +using KernelMemory.Core.Storage.Models; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.Logging; + +namespace KernelMemory.Core.Storage; + +/// +/// Implementation of IContentStorage using SQLite with queue-based execution. +/// Follows two-phase write pattern with distributed locking. +/// +public class ContentStorageService : IContentStorage +{ + private readonly ContentStorageDbContext _context; + private readonly ICuidGenerator _cuidGenerator; + private readonly ILogger _logger; + + public ContentStorageService( + ContentStorageDbContext context, + ICuidGenerator cuidGenerator, + ILogger logger) + { + this._context = context; + this._cuidGenerator = cuidGenerator; + this._logger = logger; + } + + /// + /// Upserts content following the two-phase write pattern. + /// + /// + /// + [System.Diagnostics.CodeAnalysis.SuppressMessage("Design", "CA1031:Do not catch general exception types", Justification = "Best-effort error handling for phase 2 and processing - operation is queued successfully in phase 1")] + public async Task UpsertAsync(UpsertRequest request, CancellationToken cancellationToken = default) + { + // Generate ID if not provided + var contentId = string.IsNullOrWhiteSpace(request.Id) + ? this._cuidGenerator.Generate() + : request.Id; + + this._logger.LogInformation("Starting upsert operation for content ID: {ContentId}", contentId); + + // Phase 1: Queue the operation (MUST succeed) + var operationId = await this.QueueUpsertOperationAsync(contentId, request, cancellationToken).ConfigureAwait(false); + this._logger.LogDebug("Phase 1 complete: Operation {OperationId} queued for content {ContentId}", operationId, contentId); + + // Phase 2: Try to cancel superseded operations (best effort) + try + { + await this.TryCancelSupersededUpsertOperationsAsync(contentId, operationId, cancellationToken).ConfigureAwait(false); + this._logger.LogDebug("Phase 2 complete: Cancelled superseded operations for content {ContentId}", contentId); + } + catch (Exception ex) + { + // Best effort - log but don't fail + this._logger.LogWarning(ex, "Phase 2 failed to cancel superseded operations for content {ContentId} - continuing anyway", contentId); + } + + // Processing: Try to process the new operation synchronously + try + { + await this.TryProcessNextOperationAsync(contentId, cancellationToken).ConfigureAwait(false); + this._logger.LogDebug("Processing complete for content {ContentId}", contentId); + } + catch (Exception ex) + { + // Log but don't fail - operation is queued and will be processed eventually + this._logger.LogWarning(ex, "Failed to process operation synchronously for content {ContentId} - will be processed by background worker", contentId); + } + + return contentId; + } + + /// + /// Deletes content following the two-phase write pattern. + /// + /// + /// + [System.Diagnostics.CodeAnalysis.SuppressMessage("Design", "CA1031:Do not catch general exception types", Justification = "Best-effort error handling for phase 2 and processing - operation is queued successfully in phase 1")] + public async Task DeleteAsync(string id, CancellationToken cancellationToken = default) + { + this._logger.LogInformation("Starting delete operation for content ID: {ContentId}", id); + + // Phase 1: Queue the operation (MUST succeed) + var operationId = await this.QueueDeleteOperationAsync(id, cancellationToken).ConfigureAwait(false); + this._logger.LogDebug("Phase 1 complete: Operation {OperationId} queued for content {ContentId}", operationId, id); + + // Phase 2: Try to cancel ALL previous operations (best effort) + try + { + await this.TryCancelAllOperationsAsync(id, operationId, cancellationToken).ConfigureAwait(false); + this._logger.LogDebug("Phase 2 complete: Cancelled all previous operations for content {ContentId}", id); + } + catch (Exception ex) + { + // Best effort - log but don't fail + this._logger.LogWarning(ex, "Phase 2 failed to cancel previous operations for content {ContentId} - continuing anyway", id); + } + + // Processing: Try to process the new operation synchronously + try + { + await this.TryProcessNextOperationAsync(id, cancellationToken).ConfigureAwait(false); + this._logger.LogDebug("Processing complete for content {ContentId}", id); + } + catch (Exception ex) + { + // Log but don't fail - operation is queued and will be processed eventually + this._logger.LogWarning(ex, "Failed to process operation synchronously for content {ContentId} - will be processed by background worker", id); + } + } + + /// + /// Retrieves content by ID. + /// + /// + /// + public async Task GetByIdAsync(string id, CancellationToken cancellationToken = default) + { + var record = await this._context.Content + .AsNoTracking() + .FirstOrDefaultAsync(c => c.Id == id, cancellationToken).ConfigureAwait(false); + + if (record == null) + { + return null; + } + + return new ContentDto + { + Id = record.Id, + Content = record.Content, + MimeType = record.MimeType, + ByteSize = record.ByteSize, + ContentCreatedAt = record.ContentCreatedAt, + RecordCreatedAt = record.RecordCreatedAt, + RecordUpdatedAt = record.RecordUpdatedAt, + Title = record.Title, + Description = record.Description, + Tags = record.Tags, + Metadata = record.Metadata + }; + } + + /// + /// Counts total number of content records. + /// + /// + public async Task CountAsync(CancellationToken cancellationToken = default) + { + return await this._context.Content.LongCountAsync(cancellationToken).ConfigureAwait(false); + } + + // ========== Phase 1: Queue Operations (REQUIRED) ========== + + /// + /// Phase 1: Queue an upsert operation. Must succeed. + /// + /// + /// + /// + private async Task QueueUpsertOperationAsync(string contentId, UpsertRequest request, CancellationToken cancellationToken) + { + var operation = new OperationRecord + { + Id = this._cuidGenerator.Generate(), + Complete = false, + Cancelled = false, + ContentId = contentId, + Timestamp = DateTimeOffset.UtcNow, + PlannedSteps = ["upsert"], + CompletedSteps = [], + RemainingSteps = ["upsert"], + PayloadJson = JsonSerializer.Serialize(request), + LastFailureReason = string.Empty, + LastAttemptTimestamp = null + }; + + this._context.Operations.Add(operation); + await this._context.SaveChangesAsync(cancellationToken).ConfigureAwait(false); + + this._logger.LogDebug("Queued upsert operation {OperationId} for content {ContentId}", operation.Id, contentId); + return operation.Id; + } + + /// + /// Phase 1: Queue a delete operation. Must succeed. + /// + /// + /// + private async Task QueueDeleteOperationAsync(string contentId, CancellationToken cancellationToken) + { + var operation = new OperationRecord + { + Id = this._cuidGenerator.Generate(), + Complete = false, + Cancelled = false, + ContentId = contentId, + Timestamp = DateTimeOffset.UtcNow, + PlannedSteps = ["delete"], + CompletedSteps = [], + RemainingSteps = ["delete"], + PayloadJson = JsonSerializer.Serialize(new { Id = contentId }), + LastFailureReason = string.Empty, + LastAttemptTimestamp = null + }; + + this._context.Operations.Add(operation); + await this._context.SaveChangesAsync(cancellationToken).ConfigureAwait(false); + + this._logger.LogDebug("Queued delete operation {OperationId} for content {ContentId}", operation.Id, contentId); + return operation.Id; + } + + // ========== Phase 2: Optimize Queue (OPTIONAL - Best Effort) ========== + + /// + /// Phase 2: Try to cancel superseded upsert operations (best effort). + /// Only cancels incomplete Upsert operations older than the new one. + /// Does NOT cancel Delete operations. + /// + /// + /// + /// + private async Task TryCancelSupersededUpsertOperationsAsync(string contentId, string newOperationId, CancellationToken cancellationToken) + { + // Find incomplete operations with same ContentId and older Timestamp + // Exclude Delete operations (they must complete) + var timestamp = await this._context.Operations + .Where(o => o.Id == newOperationId) + .Select(o => o.Timestamp) + .FirstOrDefaultAsync(cancellationToken).ConfigureAwait(false); + + var superseded = await this._context.Operations + .Where(o => o.ContentId == contentId + && o.Id != newOperationId + && !o.Complete + && o.Timestamp < timestamp + && o.PlannedStepsJson.Contains("upsert")) // Only cancel upserts + .ToListAsync(cancellationToken).ConfigureAwait(false); + + foreach (var op in superseded) + { + op.Cancelled = true; + this._logger.LogDebug("Cancelled superseded operation {OperationId} for content {ContentId}", op.Id, contentId); + } + + if (superseded.Count > 0) + { + await this._context.SaveChangesAsync(cancellationToken).ConfigureAwait(false); + } + } + + /// + /// Phase 2: Try to cancel ALL previous operations for delete (best effort). + /// Cancels all incomplete operations older than the delete operation. + /// + /// + /// + /// + private async Task TryCancelAllOperationsAsync(string contentId, string newOperationId, CancellationToken cancellationToken) + { + // Find incomplete operations with same ContentId and older Timestamp + var timestamp = await this._context.Operations + .Where(o => o.Id == newOperationId) + .Select(o => o.Timestamp) + .FirstOrDefaultAsync(cancellationToken).ConfigureAwait(false); + + var superseded = await this._context.Operations + .Where(o => o.ContentId == contentId + && o.Id != newOperationId + && !o.Complete + && o.Timestamp < timestamp) + .ToListAsync(cancellationToken).ConfigureAwait(false); + + foreach (var op in superseded) + { + op.Cancelled = true; + this._logger.LogDebug("Cancelled operation {OperationId} for content {ContentId} due to delete", op.Id, contentId); + } + + if (superseded.Count > 0) + { + await this._context.SaveChangesAsync(cancellationToken).ConfigureAwait(false); + } + } + + // ========== Processing: Execute Operations ========== + + /// + /// Try to process the next operation for a content ID. + /// Skips locked operations (no recovery attempts). + /// + /// + /// + [System.Diagnostics.CodeAnalysis.SuppressMessage("Design", "CA1031:Do not catch general exception types", Justification = "Catch all to ensure operation failure is logged and content remains locked for retry")] + private async Task TryProcessNextOperationAsync(string contentId, CancellationToken cancellationToken) + { + // Step 1: Get next operation to process + var operation = await this._context.Operations + .Where(o => o.ContentId == contentId && !o.Complete) + .OrderBy(o => o.Timestamp) + .FirstOrDefaultAsync(cancellationToken).ConfigureAwait(false); + + if (operation == null) + { + this._logger.LogDebug("No operations to process for content {ContentId}", contentId); + return; + } + + // Check if operation is locked + if (operation.LastAttemptTimestamp.HasValue) + { + this._logger.LogDebug("Operation {OperationId} is locked - skipping (no recovery)", operation.Id); + return; // Skip locked operations + } + + // If cancelled, mark complete and skip execution + if (operation.Cancelled) + { + this._logger.LogDebug("Operation {OperationId} was cancelled - marking complete", operation.Id); + operation.Complete = true; + operation.LastFailureReason = "Cancelled"; + await this._context.SaveChangesAsync(cancellationToken).ConfigureAwait(false); + + // Try to process next operation recursively + await this.TryProcessNextOperationAsync(contentId, cancellationToken).ConfigureAwait(false); + return; + } + + // Step 2: Acquire lock (Transaction 1) + var lockAcquired = await this.TryAcquireLockAsync(operation.Id, contentId, cancellationToken).ConfigureAwait(false); + if (!lockAcquired) + { + this._logger.LogDebug("Failed to acquire lock for operation {OperationId} - another VM got there first", operation.Id); + return; // Another VM got the lock + } + + this._logger.LogDebug("Lock acquired for operation {OperationId}", operation.Id); + + try + { + // Step 3: Execute planned steps + await this.ExecuteStepsAsync(operation, cancellationToken).ConfigureAwait(false); + + // Step 4: Complete and unlock (Transaction 2) + await this.CompleteAndUnlockAsync(operation.Id, contentId, cancellationToken).ConfigureAwait(false); + + this._logger.LogInformation("Operation {OperationId} completed successfully for content {ContentId}", operation.Id, contentId); + + // Step 5: Process next operation (if any) + await this.TryProcessNextOperationAsync(contentId, cancellationToken).ConfigureAwait(false); + } + catch (Exception ex) + { + // Update failure reason + operation.LastFailureReason = ex.Message; + await this._context.SaveChangesAsync(cancellationToken).ConfigureAwait(false); + + this._logger.LogError(ex, "Operation {OperationId} failed - content {ContentId} remains locked", operation.Id, contentId); + throw; // Propagate error (operation and content remain locked) + } + } + + /// + /// Step 2: Try to acquire lock on operation and content atomically. + /// Uses raw SQL for atomic UPDATE with WHERE clause check. + /// + /// + /// + /// + private async Task TryAcquireLockAsync(string operationId, string contentId, CancellationToken cancellationToken) + { + // Start a transaction for atomic lock acquisition + using var transaction = await this._context.Database.BeginTransactionAsync(cancellationToken).ConfigureAwait(false); + + try + { + var now = DateTimeOffset.UtcNow.ToString("O"); // ISO 8601 format + + // Lock operation - only if LastAttemptTimestamp IS NULL + const string OperationSql = @" + UPDATE km_operations + SET LastAttemptTimestamp = @p0 + WHERE Id = @p1 + AND LastAttemptTimestamp IS NULL"; + + var operationRows = await this._context.Database.ExecuteSqlRawAsync( + OperationSql, + [now, operationId], + cancellationToken).ConfigureAwait(false); + + if (operationRows == 0) + { + // Failed to lock operation - already locked + await transaction.RollbackAsync(cancellationToken).ConfigureAwait(false); + return false; + } + + // Lock content - set Ready = false (only if Ready = true) + const string ContentSql = @" + UPDATE km_content + SET Ready = 0, + RecordUpdatedAt = @p0 + WHERE Id = @p1"; + + // Note: We don't check Ready = true because content might not exist yet (insert case) + // We execute this to ensure content is locked if it exists + await this._context.Database.ExecuteSqlRawAsync( + ContentSql, + [now, contentId], + cancellationToken).ConfigureAwait(false); + + // Commit transaction + await transaction.CommitAsync(cancellationToken).ConfigureAwait(false); + return true; + } + catch + { + await transaction.RollbackAsync(cancellationToken).ConfigureAwait(false); + throw; + } + } + + /// + /// Step 3: Execute all remaining steps for an operation. + /// + /// + /// + private async Task ExecuteStepsAsync(OperationRecord operation, CancellationToken cancellationToken) + { + foreach (var step in operation.RemainingSteps) + { + this._logger.LogDebug("Executing step '{Step}' for operation {OperationId}", step, operation.Id); + + switch (step) + { + case "upsert": + await this.ExecuteUpsertStepAsync(operation, cancellationToken).ConfigureAwait(false); + break; + case "delete": + await this.ExecuteDeleteStepAsync(operation, cancellationToken).ConfigureAwait(false); + break; + default: + throw new InvalidOperationException($"Unknown step type: {step}"); + } + + // Move step from Remaining to Completed + var completed = operation.CompletedSteps.Concat([step]).ToArray(); + var remaining = operation.RemainingSteps.Where(s => s != step).ToArray(); + + operation.CompletedSteps = completed; + operation.RemainingSteps = remaining; + + await this._context.SaveChangesAsync(cancellationToken).ConfigureAwait(false); + this._logger.LogDebug("Step '{Step}' completed for operation {OperationId}", step, operation.Id); + } + } + + /// + /// Execute upsert step: delete existing + create new (if exists). + /// + /// + /// + private async Task ExecuteUpsertStepAsync(OperationRecord operation, CancellationToken cancellationToken) + { + var request = JsonSerializer.Deserialize(operation.PayloadJson) + ?? throw new InvalidOperationException($"Failed to deserialize upsert payload for operation {operation.Id}"); + + var now = DateTimeOffset.UtcNow; + var contentCreatedAt = request.ContentCreatedAt ?? now; + + // Delete existing record if it exists + var existing = await this._context.Content.FirstOrDefaultAsync(c => c.Id == operation.ContentId, cancellationToken).ConfigureAwait(false); + if (existing != null) + { + this._context.Content.Remove(existing); + this._logger.LogDebug("Deleted existing content {ContentId} for upsert", operation.ContentId); + } + + // Create new record + var content = new ContentRecord + { + Id = operation.ContentId, + Content = request.Content, + MimeType = request.MimeType, + ByteSize = Encoding.UTF8.GetByteCount(request.Content), + Ready = false, // Will be set to true when operation completes + ContentCreatedAt = contentCreatedAt, + RecordCreatedAt = existing?.RecordCreatedAt ?? now, // Preserve original creation time if exists + RecordUpdatedAt = now, + Title = request.Title, + Description = request.Description, + Tags = request.Tags, + Metadata = request.Metadata + }; + + this._context.Content.Add(content); + await this._context.SaveChangesAsync(cancellationToken).ConfigureAwait(false); + + this._logger.LogDebug("Created new content record {ContentId}", operation.ContentId); + } + + /// + /// Execute delete step: delete content if exists (idempotent). + /// + /// + /// + private async Task ExecuteDeleteStepAsync(OperationRecord operation, CancellationToken cancellationToken) + { + var existing = await this._context.Content.FirstOrDefaultAsync(c => c.Id == operation.ContentId, cancellationToken).ConfigureAwait(false); + + if (existing != null) + { + this._context.Content.Remove(existing); + await this._context.SaveChangesAsync(cancellationToken).ConfigureAwait(false); + this._logger.LogDebug("Deleted content {ContentId}", operation.ContentId); + } + else + { + this._logger.LogDebug("Content {ContentId} not found - delete is idempotent, no error", operation.ContentId); + } + } + + /// + /// Step 4: Complete operation and unlock content. + /// + /// + /// + /// + private async Task CompleteAndUnlockAsync(string operationId, string contentId, CancellationToken cancellationToken) + { + using var transaction = await this._context.Database.BeginTransactionAsync(cancellationToken).ConfigureAwait(false); + + try + { + var now = DateTimeOffset.UtcNow.ToString("O"); + + // Mark operation complete + const string OperationSql = @" + UPDATE km_operations + SET Complete = 1 + WHERE Id = @p0"; + + await this._context.Database.ExecuteSqlRawAsync(OperationSql, [operationId], cancellationToken).ConfigureAwait(false); + + // Unlock content (set Ready = true) + const string ContentSql = @" + UPDATE km_content + SET Ready = 1, + RecordUpdatedAt = @p0 + WHERE Id = @p1"; + + await this._context.Database.ExecuteSqlRawAsync(ContentSql, [now, contentId], cancellationToken).ConfigureAwait(false); + + await transaction.CommitAsync(cancellationToken).ConfigureAwait(false); + this._logger.LogDebug("Operation {OperationId} completed and content {ContentId} unlocked", operationId, contentId); + } + catch + { + await transaction.RollbackAsync(cancellationToken).ConfigureAwait(false); + throw; + } + } +} diff --git a/src/Core/Storage/CuidGenerator.cs b/src/Core/Storage/CuidGenerator.cs new file mode 100644 index 000000000..4ab894399 --- /dev/null +++ b/src/Core/Storage/CuidGenerator.cs @@ -0,0 +1,21 @@ +using Visus.Cuid; + +namespace KernelMemory.Core.Storage; + +/// +/// Default implementation of ICuidGenerator using Cuid.Net library. +/// +public class CuidGenerator : ICuidGenerator +{ + /// + /// Generates a new lowercase Cuid2 identifier. + /// + /// A unique lowercase Cuid2 string. + public string Generate() + { + // Create new Cuid2 with default length (24 characters) + // Cuid2 generates lowercase IDs by default + var cuid = new Cuid2(); + return cuid.ToString(); + } +} diff --git a/src/Core/Storage/Entities/ContentRecord.cs b/src/Core/Storage/Entities/ContentRecord.cs new file mode 100644 index 000000000..3fcb0de0b --- /dev/null +++ b/src/Core/Storage/Entities/ContentRecord.cs @@ -0,0 +1,54 @@ +using System.Text.Json; + +namespace KernelMemory.Core.Storage.Entities; + +/// +/// Entity representing a content record in the Content table. +/// Source of truth for all content in the system. +/// +public class ContentRecord +{ + // Mandatory fields + public string Id { get; set; } = string.Empty; + public string Content { get; set; } = string.Empty; + public string MimeType { get; set; } = string.Empty; + public long ByteSize { get; set; } + public bool Ready { get; set; } + public DateTimeOffset ContentCreatedAt { get; set; } + public DateTimeOffset RecordCreatedAt { get; set; } + public DateTimeOffset RecordUpdatedAt { get; set; } + + // Optional fields + public string Title { get; set; } = string.Empty; + public string Description { get; set; } = string.Empty; + + // JSON-backed fields (stored as JSON strings in SQLite) + // Tags array + public string TagsJson { get; set; } = "[]"; + + /// + /// Gets or sets the tags array. Not mapped to database - uses TagsJson for persistence. + /// + [System.Diagnostics.CodeAnalysis.SuppressMessage("Performance", "CA1819:Properties should not return arrays")] + public string[] Tags + { + get => string.IsNullOrWhiteSpace(this.TagsJson) + ? [] + : JsonSerializer.Deserialize(this.TagsJson) ?? []; + set => this.TagsJson = JsonSerializer.Serialize(value); + } + + // Metadata key-value pairs + public string MetadataJson { get; set; } = "{}"; + + /// + /// Gets or sets the metadata dictionary. Not mapped to database - uses MetadataJson for persistence. + /// + public Dictionary Metadata + { + get => string.IsNullOrWhiteSpace(this.MetadataJson) + ? new Dictionary() + : JsonSerializer.Deserialize>(this.MetadataJson) ?? new Dictionary(); + set => this.MetadataJson = JsonSerializer.Serialize(value); + } +} diff --git a/src/Core/Storage/Entities/OperationRecord.cs b/src/Core/Storage/Entities/OperationRecord.cs new file mode 100644 index 000000000..7de369916 --- /dev/null +++ b/src/Core/Storage/Entities/OperationRecord.cs @@ -0,0 +1,80 @@ +using System.Text.Json; + +namespace KernelMemory.Core.Storage.Entities; + +/// +/// Entity representing an operation in the Operations table. +/// Used for queue-based processing with distributed locking. +/// +public class OperationRecord +{ + public string Id { get; set; } = string.Empty; + public bool Complete { get; set; } + public bool Cancelled { get; set; } + public string ContentId { get; set; } = string.Empty; + public DateTimeOffset Timestamp { get; set; } + public string LastFailureReason { get; set; } = string.Empty; + + /// + /// When last attempt was made (nullable). Used for distributed locking. + /// If NOT NULL and Complete=false: operation is locked (executing or crashed). + /// + public DateTimeOffset? LastAttemptTimestamp { get; set; } + + // JSON-backed array fields + public string PlannedStepsJson { get; set; } = "[]"; + + /// + /// Gets or sets the planned steps array. Not mapped to database - uses PlannedStepsJson for persistence. + /// + [System.Diagnostics.CodeAnalysis.SuppressMessage("Performance", "CA1819:Properties should not return arrays")] + public string[] PlannedSteps + { + get => string.IsNullOrWhiteSpace(this.PlannedStepsJson) + ? [] + : JsonSerializer.Deserialize(this.PlannedStepsJson) ?? []; + set => this.PlannedStepsJson = JsonSerializer.Serialize(value); + } + + public string CompletedStepsJson { get; set; } = "[]"; + + /// + /// Gets or sets the completed steps array. Not mapped to database - uses CompletedStepsJson for persistence. + /// + [System.Diagnostics.CodeAnalysis.SuppressMessage("Performance", "CA1819:Properties should not return arrays")] + public string[] CompletedSteps + { + get => string.IsNullOrWhiteSpace(this.CompletedStepsJson) + ? [] + : JsonSerializer.Deserialize(this.CompletedStepsJson) ?? []; + set => this.CompletedStepsJson = JsonSerializer.Serialize(value); + } + + public string RemainingStepsJson { get; set; } = "[]"; + + /// + /// Gets or sets the remaining steps array. Not mapped to database - uses RemainingStepsJson for persistence. + /// + [System.Diagnostics.CodeAnalysis.SuppressMessage("Performance", "CA1819:Properties should not return arrays")] + public string[] RemainingSteps + { + get => string.IsNullOrWhiteSpace(this.RemainingStepsJson) + ? [] + : JsonSerializer.Deserialize(this.RemainingStepsJson) ?? []; + set => this.RemainingStepsJson = JsonSerializer.Serialize(value); + } + + // Payload stored as JSON + public string PayloadJson { get; set; } = "{}"; + + /// + /// Gets or sets the payload object. Not mapped to database - uses PayloadJson for persistence. + /// + public object? Payload + { + get => string.IsNullOrWhiteSpace(this.PayloadJson) + ? null + : JsonSerializer.Deserialize(this.PayloadJson); + set => this.PayloadJson = value == null ? "{}" : JsonSerializer.Serialize(value); + } +} diff --git a/src/Core/Storage/Exceptions/ContentStorageException.cs b/src/Core/Storage/Exceptions/ContentStorageException.cs new file mode 100644 index 000000000..4b691da10 --- /dev/null +++ b/src/Core/Storage/Exceptions/ContentStorageException.cs @@ -0,0 +1,84 @@ +namespace KernelMemory.Core.Storage.Exceptions; + +/// +/// Base exception for content storage errors. +/// +public class ContentStorageException : Exception +{ + public ContentStorageException() + { + } + + public ContentStorageException(string message) : base(message) + { + } + + public ContentStorageException(string message, Exception innerException) : base(message, innerException) + { + } +} + +/// +/// Exception thrown when a content record is not found. +/// +public class ContentNotFoundException : ContentStorageException +{ + public string ContentId { get; } + + public ContentNotFoundException(string contentId) + : base($"Content with ID '{contentId}' was not found.") + { + this.ContentId = contentId; + } + + public ContentNotFoundException(string contentId, string message) + : base(message) + { + this.ContentId = contentId; + } + + public ContentNotFoundException() : base() + { + this.ContentId = string.Empty; + } + + public ContentNotFoundException(string message, Exception innerException) : base(message, innerException) + { + this.ContentId = string.Empty; + } +} + +/// +/// Exception thrown when an operation fails during processing. +/// +public class OperationFailedException : ContentStorageException +{ + public string OperationId { get; } + + public OperationFailedException(string operationId, string message) + : base(message) + { + this.OperationId = operationId; + } + + public OperationFailedException(string operationId, string message, Exception innerException) + : base(message, innerException) + { + this.OperationId = operationId; + } + + public OperationFailedException() : base() + { + this.OperationId = string.Empty; + } + + public OperationFailedException(string message) : base(message) + { + this.OperationId = string.Empty; + } + + public OperationFailedException(string message, Exception innerException) : base(message, innerException) + { + this.OperationId = string.Empty; + } +} diff --git a/src/Core/Storage/IContentStorage.cs b/src/Core/Storage/IContentStorage.cs new file mode 100644 index 000000000..919f5a1f5 --- /dev/null +++ b/src/Core/Storage/IContentStorage.cs @@ -0,0 +1,46 @@ +using KernelMemory.Core.Storage.Exceptions; +using KernelMemory.Core.Storage.Models; + +namespace KernelMemory.Core.Storage; + +/// +/// Interface for content storage operations. +/// Provides queue-based write operations with eventual consistency. +/// +public interface IContentStorage +{ + /// + /// Upserts content. Creates new record if ID is empty, replaces existing if ID is provided. + /// Operation is queued and processed asynchronously. + /// + /// The upsert request containing content and metadata. + /// Cancellation token. + /// The ID of the content record (newly generated or existing). + /// Thrown if queueing the operation fails. + Task UpsertAsync(UpsertRequest request, CancellationToken cancellationToken = default); + + /// + /// Deletes content by ID. Idempotent - no error if record doesn't exist. + /// Operation is queued and processed asynchronously. + /// + /// The content ID to delete. + /// Cancellation token. + /// Task representing the async operation. + /// Thrown if queueing the operation fails. + Task DeleteAsync(string id, CancellationToken cancellationToken = default); + + /// + /// Retrieves content by ID. + /// + /// The content ID to retrieve. + /// Cancellation token. + /// The content DTO, or null if not found. + Task GetByIdAsync(string id, CancellationToken cancellationToken = default); + + /// + /// Counts total number of content records. + /// + /// Cancellation token. + /// Total count of content records. + Task CountAsync(CancellationToken cancellationToken = default); +} diff --git a/src/Core/Storage/ICuidGenerator.cs b/src/Core/Storage/ICuidGenerator.cs new file mode 100644 index 000000000..6f8606688 --- /dev/null +++ b/src/Core/Storage/ICuidGenerator.cs @@ -0,0 +1,13 @@ +namespace KernelMemory.Core.Storage; + +/// +/// Interface for generating Cuid2 identifiers. +/// +public interface ICuidGenerator +{ + /// + /// Generates a new lowercase Cuid2 identifier. + /// + /// A unique lowercase Cuid2 string. + string Generate(); +} diff --git a/src/Core/Storage/Models/ContentDto.cs b/src/Core/Storage/Models/ContentDto.cs new file mode 100644 index 000000000..3c61377b7 --- /dev/null +++ b/src/Core/Storage/Models/ContentDto.cs @@ -0,0 +1,21 @@ +namespace KernelMemory.Core.Storage.Models; + +/// +/// Data transfer object for content records. +/// Clean representation without operational fields like Ready flag. +/// +public class ContentDto +{ + public string Id { get; set; } = string.Empty; + public string Content { get; set; } = string.Empty; + public string MimeType { get; set; } = string.Empty; + public long ByteSize { get; set; } + public DateTimeOffset ContentCreatedAt { get; set; } + public DateTimeOffset RecordCreatedAt { get; set; } + public DateTimeOffset RecordUpdatedAt { get; set; } + public string Title { get; set; } = string.Empty; + public string Description { get; set; } = string.Empty; + [System.Diagnostics.CodeAnalysis.SuppressMessage("Performance", "CA1819:Properties should not return arrays")] + public string[] Tags { get; set; } = []; + public Dictionary Metadata { get; set; } = new(); +} diff --git a/src/Core/Storage/Models/UpsertRequest.cs b/src/Core/Storage/Models/UpsertRequest.cs new file mode 100644 index 000000000..087ad3b5f --- /dev/null +++ b/src/Core/Storage/Models/UpsertRequest.cs @@ -0,0 +1,50 @@ +namespace KernelMemory.Core.Storage.Models; + +/// +/// Request model for upserting content. +/// If ID is empty, a new one will be generated. +/// If ID is provided, the record will be replaced if it exists, or created if it doesn't. +/// +public class UpsertRequest +{ + /// + /// Content ID. If empty, a new one will be generated. + /// + public string Id { get; set; } = string.Empty; + + /// + /// The actual string content to store. + /// + public string Content { get; set; } = string.Empty; + + /// + /// MIME type of the content. + /// + public string MimeType { get; set; } = string.Empty; + + /// + /// Optional content creation date. If not provided, current time is used. + /// + public DateTimeOffset? ContentCreatedAt { get; set; } + + /// + /// Optional title. + /// + public string Title { get; set; } = string.Empty; + + /// + /// Optional description. + /// + public string Description { get; set; } = string.Empty; + + /// + /// Optional tags. + /// + [System.Diagnostics.CodeAnalysis.SuppressMessage("Performance", "CA1819:Properties should not return arrays")] + public string[] Tags { get; set; } = []; + + /// + /// Optional metadata key-value pairs. + /// + public Dictionary Metadata { get; set; } = new(); +} diff --git a/src/Directory.Packages.props b/src/Directory.Packages.props index fc4481b00..afb6c8f29 100644 --- a/src/Directory.Packages.props +++ b/src/Directory.Packages.props @@ -4,7 +4,9 @@ + + diff --git a/tests/Core.Tests/Config/AppConfigTests.cs b/tests/Core.Tests/Config/AppConfigTests.cs index d455109de..6f3dfab8f 100644 --- a/tests/Core.Tests/Config/AppConfigTests.cs +++ b/tests/Core.Tests/Config/AppConfigTests.cs @@ -10,7 +10,7 @@ namespace Core.Tests.Config; /// /// Tests for AppConfig validation and default configuration /// -public class AppConfigTests +public sealed class AppConfigTests { [Fact] public void CreateDefault_ShouldCreateValidConfiguration() diff --git a/tests/Core.Tests/Config/ConfigParserTests.cs b/tests/Core.Tests/Config/ConfigParserTests.cs index d6e6e53bd..cdc349ea6 100644 --- a/tests/Core.Tests/Config/ConfigParserTests.cs +++ b/tests/Core.Tests/Config/ConfigParserTests.cs @@ -7,7 +7,7 @@ namespace Core.Tests.Config; /// /// Tests for ConfigParser - loading and parsing configuration files /// -public class ConfigParserTests +public sealed class ConfigParserTests { [Fact] public void LoadFromFile_WhenFileMissing_ShouldReturnDefaultConfig() diff --git a/tests/Core.Tests/Config/ContentIndexConfigTests.cs b/tests/Core.Tests/Config/ContentIndexConfigTests.cs index 108e351b1..1c405052a 100644 --- a/tests/Core.Tests/Config/ContentIndexConfigTests.cs +++ b/tests/Core.Tests/Config/ContentIndexConfigTests.cs @@ -6,7 +6,7 @@ namespace Core.Tests.Config; /// /// Tests for Content Index configuration validation /// -public class ContentIndexConfigTests +public sealed class ContentIndexConfigTests { [Fact] public void LoadFromFile_WithPostgresContentIndex_ShouldValidate() diff --git a/tests/Core.Tests/Config/EmbeddingsConfigTests.cs b/tests/Core.Tests/Config/EmbeddingsConfigTests.cs index f97564bfc..cc7cf71b4 100644 --- a/tests/Core.Tests/Config/EmbeddingsConfigTests.cs +++ b/tests/Core.Tests/Config/EmbeddingsConfigTests.cs @@ -6,7 +6,7 @@ namespace Core.Tests.Config; /// /// Tests for Embeddings configuration validation /// -public class EmbeddingsConfigTests +public sealed class EmbeddingsConfigTests { [Fact] public void LoadFromFile_WithOllamaEmbeddings_ShouldValidate() diff --git a/tests/Core.Tests/Config/SearchIndexConfigTests.cs b/tests/Core.Tests/Config/SearchIndexConfigTests.cs index 5d849524e..82cd1fb5b 100644 --- a/tests/Core.Tests/Config/SearchIndexConfigTests.cs +++ b/tests/Core.Tests/Config/SearchIndexConfigTests.cs @@ -6,7 +6,7 @@ namespace Core.Tests.Config; /// /// Tests for Search Index configuration validation /// -public class SearchIndexConfigTests +public sealed class SearchIndexConfigTests { [Fact] public void LoadFromFile_WithGraphSearchIndex_ShouldExpandTildePath() diff --git a/tests/Core.Tests/Config/StorageConfigTests.cs b/tests/Core.Tests/Config/StorageConfigTests.cs index 616b9ac38..c1136447a 100644 --- a/tests/Core.Tests/Config/StorageConfigTests.cs +++ b/tests/Core.Tests/Config/StorageConfigTests.cs @@ -6,7 +6,7 @@ namespace Core.Tests.Config; /// /// Tests for Storage configuration validation and parsing /// -public class StorageConfigTests +public sealed class StorageConfigTests { [Fact] public void LoadFromFile_WithDiskStorage_ShouldExpandTildePath() diff --git a/tests/Core.Tests/Core.Tests.csproj b/tests/Core.Tests/Core.Tests.csproj index cd434d381..63252b381 100644 --- a/tests/Core.Tests/Core.Tests.csproj +++ b/tests/Core.Tests/Core.Tests.csproj @@ -6,12 +6,16 @@ false - $(NoWarn);CA1707;CA1307 + + + $(NoWarn);CA1707;CA1307;CA1812;xUnit1030 + + all diff --git a/tests/Core.Tests/Storage/ContentStorageIntegrationTests.cs b/tests/Core.Tests/Storage/ContentStorageIntegrationTests.cs new file mode 100644 index 000000000..e85bbf8bb --- /dev/null +++ b/tests/Core.Tests/Storage/ContentStorageIntegrationTests.cs @@ -0,0 +1,396 @@ +using KernelMemory.Core.Storage; +using KernelMemory.Core.Storage.Models; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.Logging; +using Moq; + +namespace Core.Tests.Storage; + +/// +/// Integration tests for ContentStorageService using real SQLite database files. +/// Tests the full stack including database schema, migrations, and persistence. +/// +public sealed class ContentStorageIntegrationTests : IDisposable +{ + private readonly string _tempDbPath; + private readonly ContentStorageDbContext _context; + private readonly ContentStorageService _service; + private readonly Mock> _mockLogger; + + public ContentStorageIntegrationTests() + { + // Use temporary SQLite file for integration tests + _tempDbPath = Path.Combine(Path.GetTempPath(), $"test_km_{Guid.NewGuid()}.db"); + + var options = new DbContextOptionsBuilder() + .UseSqlite($"Data Source={_tempDbPath}") + .Options; + + _context = new ContentStorageDbContext(options); + _context.Database.EnsureCreated(); + + _mockLogger = new Mock>(); + + // Use real CuidGenerator for integration tests + var cuidGenerator = new CuidGenerator(); + _service = new ContentStorageService(_context, cuidGenerator, _mockLogger.Object); + } + + public void Dispose() + { + _context.Dispose(); + + // Clean up temporary database file + if (File.Exists(_tempDbPath)) + { + File.Delete(_tempDbPath); + } + + GC.SuppressFinalize(this); + } + + [Fact] + public async Task DatabaseSchema_IsCreatedCorrectlyAsync() + { + // Assert - Verify tables exist + var tables = await _context.Database.SqlQueryRaw( + "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name") + .ToListAsync().ConfigureAwait(false); + + Assert.Contains("km_content", tables); + Assert.Contains("km_operations", tables); + } + + [Fact] + public async Task ContentTable_HasCorrectIndexesAsync() + { + // Assert - Verify indexes on Content table + var indexes = await _context.Database.SqlQueryRaw( + "SELECT name FROM sqlite_master WHERE type='index' AND tbl_name='km_content'") + .ToListAsync().ConfigureAwait(false); + + Assert.Contains(indexes, idx => idx.Contains("Ready")); + } + + [Fact] + public async Task OperationsTable_HasCorrectIndexesAsync() + { + // Assert - Verify indexes on Operations table + var indexes = await _context.Database.SqlQueryRaw( + "SELECT name FROM sqlite_master WHERE type='index' AND tbl_name='km_operations'") + .ToListAsync().ConfigureAwait(false); + + Assert.Contains(indexes, idx => idx.Contains("ContentId")); + Assert.Contains(indexes, idx => idx.Contains("Complete")); + Assert.Contains(indexes, idx => idx.Contains("Timestamp")); + } + + [Fact] + public async Task FullWorkflow_UpsertRetrieveDeleteAsync() + { + // Arrange + var request = new UpsertRequest + { + Content = "Integration test content", + MimeType = "text/plain", + Title = "Integration Test", + Description = "Testing full workflow", + Tags = ["integration", "test"], + Metadata = new Dictionary + { + ["source"] = "integration_test", + ["version"] = "1.0" + } + }; + + // Act 1: Upsert + var contentId = await _service.UpsertAsync(request).ConfigureAwait(false); + await Task.Delay(200).ConfigureAwait(false); // Wait for processing + + // Assert 1: Content exists + var content = await _service.GetByIdAsync(contentId).ConfigureAwait(false); + Assert.NotNull(content); + Assert.Equal("Integration test content", content.Content); + Assert.Equal("text/plain", content.MimeType); + Assert.Equal("Integration Test", content.Title); + Assert.Equal("Testing full workflow", content.Description); + Assert.Equal(2, content.Tags.Length); + Assert.Equal(2, content.Metadata.Count); + + // Act 2: Update + var updateRequest = new UpsertRequest + { + Id = contentId, + Content = "Updated content", + MimeType = "text/html", + Title = "Updated Title" + }; + await _service.UpsertAsync(updateRequest).ConfigureAwait(false); + await Task.Delay(200).ConfigureAwait(false); // Wait for processing + + // Assert 2: Content is updated + var updatedContent = await _service.GetByIdAsync(contentId).ConfigureAwait(false); + Assert.NotNull(updatedContent); + Assert.Equal("Updated content", updatedContent.Content); + Assert.Equal("text/html", updatedContent.MimeType); + Assert.Equal("Updated Title", updatedContent.Title); + + // Act 3: Delete + await _service.DeleteAsync(contentId).ConfigureAwait(false); + await Task.Delay(200).ConfigureAwait(false); // Wait for processing + + // Assert 3: Content is deleted + var deletedContent = await _service.GetByIdAsync(contentId).ConfigureAwait(false); + Assert.Null(deletedContent); + } + + [Fact] + public async Task RealCuidGenerator_GeneratesValidIdsAsync() + { + // Act - Create multiple content items with real CUID generation + var ids = new List(); + for (int i = 0; i < 10; i++) + { + var id = await _service.UpsertAsync(new UpsertRequest + { + Content = $"Content {i}", + MimeType = "text/plain" + }).ConfigureAwait(false); + ids.Add(id); + } + + // Assert - All IDs should be unique and non-empty + Assert.Equal(10, ids.Distinct().Count()); + Assert.All(ids, id => + { + Assert.NotEmpty(id); + Assert.True(id.Length >= 20); // CUIDs are typically 25-32 chars + }); + } + + [Fact] + public async Task Persistence_SurvivesDatabaseReopenAsync() + { + // Arrange - Create content + var request = new UpsertRequest + { + Id = "persistent_test", + Content = "Persistent content", + MimeType = "text/plain", + Title = "Persistence Test" + }; + + await _service.UpsertAsync(request).ConfigureAwait(false); + await Task.Delay(200).ConfigureAwait(false); // Wait for processing + + // Act - Dispose and recreate context (simulates app restart) + await _context.DisposeAsync().ConfigureAwait(false); + + var options = new DbContextOptionsBuilder() + .UseSqlite($"Data Source={_tempDbPath}") + .Options; + + using var newContext = new ContentStorageDbContext(options); + var newService = new ContentStorageService( + newContext, + new CuidGenerator(), + _mockLogger.Object); + + // Assert - Content should still exist + var content = await newService.GetByIdAsync("persistent_test").ConfigureAwait(false); + Assert.NotNull(content); + Assert.Equal("Persistent content", content.Content); + Assert.Equal("Persistence Test", content.Title); + } + + [Fact] + public async Task MultipleOperations_ProcessInOrderAsync() + { + // Arrange + var contentId = "ordered_test"; + var operations = new List(); + + // Act - Create multiple operations quickly + for (int i = 1; i <= 5; i++) + { + await _service.UpsertAsync(new UpsertRequest + { + Id = contentId, + Content = $"Version {i}", + MimeType = "text/plain" + }).ConfigureAwait(false); + await Task.Delay(10).ConfigureAwait(false); // Small delay to ensure timestamp order + } + + // Wait for all operations to process + await Task.Delay(500).ConfigureAwait(false); + + // Assert - Final content should be the last version + var content = await _service.GetByIdAsync(contentId).ConfigureAwait(false); + Assert.NotNull(content); + Assert.Equal("Version 5", content.Content); + } + + [Fact] + public async Task OperationQueue_HandlesFailureGracefullyAsync() + { + // This test verifies that operations are queued even if processing might fail + // Arrange + var contentId = "failure_test"; + + // Act - Queue operation + await _service.UpsertAsync(new UpsertRequest + { + Id = contentId, + Content = "Test content", + MimeType = "text/plain" + }).ConfigureAwait(false); + + // Assert - Operation should be queued (Phase 1 always succeeds) + var operation = await _context.Operations + .FirstOrDefaultAsync(o => o.ContentId == contentId).ConfigureAwait(false); + + Assert.NotNull(operation); + Assert.False(operation.Complete || operation.Cancelled); + } + + [Fact] + public async Task DateTimeOffset_IsStoredAndRetrievedCorrectlyAsync() + { + // Arrange - Use specific timestamp + var specificDate = new DateTimeOffset(2024, 6, 15, 14, 30, 0, TimeSpan.FromHours(-7)); + var request = new UpsertRequest + { + Content = "Date test content", + MimeType = "text/plain", + ContentCreatedAt = specificDate + }; + + // Act + var contentId = await _service.UpsertAsync(request).ConfigureAwait(false); + await Task.Delay(200).ConfigureAwait(false); // Wait for processing + + // Assert + var content = await _service.GetByIdAsync(contentId).ConfigureAwait(false); + Assert.NotNull(content); + Assert.Equal(specificDate, content.ContentCreatedAt); + } + + [Fact] + public async Task JsonSerialization_HandlesComplexMetadataAsync() + { + // Arrange - Complex metadata with special characters + var request = new UpsertRequest + { + Content = "Test content", + MimeType = "text/plain", + Tags = ["tag with spaces", "tag-with-dashes", "tag_with_underscores"], + Metadata = new Dictionary + { + ["key with spaces"] = "value with spaces", + ["special-chars"] = "value!@#$%^&*()", + ["unicode"] = "你好世界🌍", + ["json-like"] = "{\"nested\": \"value\"}" + } + }; + + // Act + var contentId = await _service.UpsertAsync(request).ConfigureAwait(false); + await Task.Delay(200).ConfigureAwait(false); // Wait for processing + + // Assert + var content = await _service.GetByIdAsync(contentId).ConfigureAwait(false); + Assert.NotNull(content); + Assert.Equal(3, content.Tags.Length); + Assert.Equal("tag with spaces", content.Tags[0]); + Assert.Equal(4, content.Metadata.Count); + Assert.Equal("你好世界🌍", content.Metadata["unicode"]); + Assert.Equal("{\"nested\": \"value\"}", content.Metadata["json-like"]); + } + + [Fact] + public async Task CountAsync_ReflectsDatabaseStateAsync() + { + // Arrange - Initial count + var initialCount = await _service.CountAsync().ConfigureAwait(false); + + // Act - Add 3 items + for (int i = 0; i < 3; i++) + { + await _service.UpsertAsync(new UpsertRequest + { + Content = $"Content {i}", + MimeType = "text/plain" + }).ConfigureAwait(false); + } + await Task.Delay(300).ConfigureAwait(false); // Wait for processing + + // Assert - Count increased by 3 + var afterAddCount = await _service.CountAsync().ConfigureAwait(false); + Assert.Equal(initialCount + 3, afterAddCount); + + // Act - Delete 1 item + var content = await _context.Content.FirstAsync().ConfigureAwait(false); + await _service.DeleteAsync(content.Id).ConfigureAwait(false); + await Task.Delay(200).ConfigureAwait(false); // Wait for processing + + // Assert - Count decreased by 1 + var afterDeleteCount = await _service.CountAsync().ConfigureAwait(false); + Assert.Equal(afterAddCount - 1, afterDeleteCount); + } + + [Fact] + public async Task EmptyStringFields_AreHandledCorrectlyAsync() + { + // Arrange - Use empty strings for optional fields + var request = new UpsertRequest + { + Content = "Content only", + MimeType = "text/plain", + Title = string.Empty, + Description = string.Empty, + Tags = [], + Metadata = new Dictionary() + }; + + // Act + var contentId = await _service.UpsertAsync(request).ConfigureAwait(false); + await Task.Delay(200).ConfigureAwait(false); // Wait for processing + + // Assert + var content = await _service.GetByIdAsync(contentId).ConfigureAwait(false); + Assert.NotNull(content); + Assert.Equal(string.Empty, content.Title); + Assert.Equal(string.Empty, content.Description); + Assert.Empty(content.Tags); + Assert.Empty(content.Metadata); + } + + [Fact] + public async Task ConcurrentWrites_ToSameContent_AreSerializedCorrectlyAsync() + { + // Arrange + var contentId = "concurrent_integration_test"; + + // Act - Fire multiple concurrent upserts + var tasks = Enumerable.Range(1, 10).Select(i => + _service.UpsertAsync(new UpsertRequest + { + Id = contentId, + Content = $"Concurrent Version {i}", + MimeType = "text/plain" + })).ToList(); + + await Task.WhenAll(tasks).ConfigureAwait(false); + await Task.Delay(1000).ConfigureAwait(false); // Wait for all operations to process + + // Assert - Should have exactly one content record (last one wins) + var content = await _service.GetByIdAsync(contentId).ConfigureAwait(false); + Assert.NotNull(content); + Assert.StartsWith("Concurrent Version", content.Content); + + // Verify only one content record exists with this ID + var count = await _context.Content.CountAsync(c => c.Id == contentId).ConfigureAwait(false); + Assert.Equal(1, count); + } +} diff --git a/tests/Core.Tests/Storage/ContentStorageServiceTests.cs b/tests/Core.Tests/Storage/ContentStorageServiceTests.cs new file mode 100644 index 000000000..21342641c --- /dev/null +++ b/tests/Core.Tests/Storage/ContentStorageServiceTests.cs @@ -0,0 +1,522 @@ +using KernelMemory.Core.Storage; +using KernelMemory.Core.Storage.Entities; +using KernelMemory.Core.Storage.Models; +using Microsoft.Data.Sqlite; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.Logging; +using Moq; + +namespace Core.Tests.Storage; + +/// +/// Unit tests for ContentStorageService using in-memory SQLite database. +/// Tests cover the queue-based execution model and two-phase write pattern. +/// +public sealed class ContentStorageServiceTests : IDisposable +{ + private readonly SqliteConnection _connection; + private readonly ContentStorageDbContext _context; + private readonly Mock _mockCuidGenerator; + private readonly Mock> _mockLogger; + private readonly ContentStorageService _service; + private int _cuidCounter; + + public ContentStorageServiceTests() + { + // Use in-memory SQLite for fast isolated tests + _connection = new SqliteConnection("DataSource=:memory:"); + _connection.Open(); + + var options = new DbContextOptionsBuilder() + .UseSqlite(_connection) + .Options; + + _context = new ContentStorageDbContext(options); + _context.Database.EnsureCreated(); + + // Mock CUID generator with predictable IDs + _mockCuidGenerator = new Mock(); + _cuidCounter = 0; + _mockCuidGenerator + .Setup(x => x.Generate()) + .Returns(() => $"test_id_{++_cuidCounter:D5}"); + + _mockLogger = new Mock>(); + + _service = new ContentStorageService(_context, _mockCuidGenerator.Object, _mockLogger.Object); + } + + public void Dispose() + { + _context.Dispose(); + _connection.Dispose(); + GC.SuppressFinalize(this); + } + + [Fact] + public async Task UpsertAsync_WithEmptyId_GeneratesNewIdAsync() + { + // Arrange + var request = new UpsertRequest + { + Content = "Test content", + MimeType = "text/plain", + Title = "Test" + }; + + // Act + var resultId = await _service.UpsertAsync(request).ConfigureAwait(false); + + // Assert + Assert.Equal("test_id_00001", resultId); // First generated ID + + // Verify content was created + var content = await _service.GetByIdAsync(resultId).ConfigureAwait(false); + Assert.NotNull(content); + Assert.Equal("Test content", content.Content); + Assert.Equal("text/plain", content.MimeType); + Assert.Equal("Test", content.Title); + } + + [Fact] + public async Task UpsertAsync_WithProvidedId_UsesProvidedIdAsync() + { + // Arrange + var request = new UpsertRequest + { + Id = "custom_id_123", + Content = "Test content", + MimeType = "text/plain" + }; + + // Act + var resultId = await _service.UpsertAsync(request).ConfigureAwait(false); + + // Assert + Assert.Equal("custom_id_123", resultId); + + // Verify content was created + var content = await _service.GetByIdAsync(resultId).ConfigureAwait(false); + Assert.NotNull(content); + Assert.Equal("Test content", content.Content); + } + + [Fact] + public async Task UpsertAsync_ReplacesExistingContentAsync() + { + // Arrange - Create initial content + var initialRequest = new UpsertRequest + { + Id = "test_id_replace", + Content = "Initial content", + MimeType = "text/plain", + Title = "Initial Title" + }; + await _service.UpsertAsync(initialRequest).ConfigureAwait(false); + + // Wait for processing to complete + await Task.Delay(100).ConfigureAwait(false); + + // Act - Replace with new content + var replaceRequest = new UpsertRequest + { + Id = "test_id_replace", + Content = "Replaced content", + MimeType = "text/html", + Title = "New Title" + }; + await _service.UpsertAsync(replaceRequest).ConfigureAwait(false); + + // Wait for processing to complete + await Task.Delay(100).ConfigureAwait(false); + + // Assert + var content = await _service.GetByIdAsync("test_id_replace").ConfigureAwait(false); + Assert.NotNull(content); + Assert.Equal("Replaced content", content.Content); + Assert.Equal("text/html", content.MimeType); + Assert.Equal("New Title", content.Title); + } + + [Fact] + public async Task UpsertAsync_StoresTagsAsync() + { + // Arrange + var request = new UpsertRequest + { + Content = "Test content", + MimeType = "text/plain", + Tags = ["tag1", "tag2", "tag3"] + }; + + // Act + var resultId = await _service.UpsertAsync(request).ConfigureAwait(false); + await Task.Delay(100).ConfigureAwait(false); // Wait for processing + + // Assert + var content = await _service.GetByIdAsync(resultId).ConfigureAwait(false); + Assert.NotNull(content); + Assert.Equal(3, content.Tags.Length); + Assert.Contains("tag1", content.Tags); + Assert.Contains("tag2", content.Tags); + Assert.Contains("tag3", content.Tags); + } + + [Fact] + public async Task UpsertAsync_StoresMetadataAsync() + { + // Arrange + var request = new UpsertRequest + { + Content = "Test content", + MimeType = "text/plain", + Metadata = new Dictionary + { + ["key1"] = "value1", + ["key2"] = "value2" + } + }; + + // Act + var resultId = await _service.UpsertAsync(request).ConfigureAwait(false); + await Task.Delay(100).ConfigureAwait(false); // Wait for processing + + // Assert + var content = await _service.GetByIdAsync(resultId).ConfigureAwait(false); + Assert.NotNull(content); + Assert.Equal(2, content.Metadata.Count); + Assert.Equal("value1", content.Metadata["key1"]); + Assert.Equal("value2", content.Metadata["key2"]); + } + + [Fact] + public async Task UpsertAsync_CalculatesByteSizeAsync() + { + // Arrange + var testContent = "Test content with some length"; + var request = new UpsertRequest + { + Content = testContent, + MimeType = "text/plain" + }; + + // Act + var resultId = await _service.UpsertAsync(request).ConfigureAwait(false); + await Task.Delay(100).ConfigureAwait(false); // Wait for processing + + // Assert + var content = await _service.GetByIdAsync(resultId).ConfigureAwait(false); + Assert.NotNull(content); + Assert.Equal(System.Text.Encoding.UTF8.GetByteCount(testContent), content.ByteSize); + } + + [Fact] + public async Task UpsertAsync_UsesCustomContentCreatedAtAsync() + { + // Arrange + var customDate = new DateTimeOffset(2024, 1, 1, 12, 0, 0, TimeSpan.Zero); + var request = new UpsertRequest + { + Content = "Test content", + MimeType = "text/plain", + ContentCreatedAt = customDate + }; + + // Act + var resultId = await _service.UpsertAsync(request).ConfigureAwait(false); + await Task.Delay(100).ConfigureAwait(false); // Wait for processing + + // Assert + var content = await _service.GetByIdAsync(resultId).ConfigureAwait(false); + Assert.NotNull(content); + Assert.Equal(customDate, content.ContentCreatedAt); + } + + [Fact] + public async Task DeleteAsync_RemovesExistingContentAsync() + { + // Arrange - Create content first + var request = new UpsertRequest + { + Id = "test_id_delete", + Content = "Content to delete", + MimeType = "text/plain" + }; + await _service.UpsertAsync(request).ConfigureAwait(false); + await Task.Delay(100).ConfigureAwait(false); // Wait for processing + + // Verify content exists + var contentBefore = await _service.GetByIdAsync("test_id_delete").ConfigureAwait(false); + Assert.NotNull(contentBefore); + + // Act - Delete the content + await _service.DeleteAsync("test_id_delete").ConfigureAwait(false); + await Task.Delay(100).ConfigureAwait(false); // Wait for processing + + // Assert - Content should be gone + var contentAfter = await _service.GetByIdAsync("test_id_delete").ConfigureAwait(false); + Assert.Null(contentAfter); + } + + [Fact] + public async Task DeleteAsync_IsIdempotentAsync() + { + // Act - Delete non-existent content (should not throw) + await _service.DeleteAsync("non_existent_id").ConfigureAwait(false); + await Task.Delay(100).ConfigureAwait(false); // Wait for processing + + // Assert - No exception thrown, verify content doesn't exist + var content = await _service.GetByIdAsync("non_existent_id").ConfigureAwait(false); + Assert.Null(content); + } + + [Fact] + public async Task GetByIdAsync_ReturnsNullForNonExistentAsync() + { + // Act + var content = await _service.GetByIdAsync("non_existent_id").ConfigureAwait(false); + + // Assert + Assert.Null(content); + } + + [Fact] + public async Task CountAsync_ReturnsCorrectCountAsync() + { + // Arrange - Create multiple content records + for (int i = 0; i < 5; i++) + { + await _service.UpsertAsync(new UpsertRequest + { + Content = $"Content {i}", + MimeType = "text/plain" + }).ConfigureAwait(false); + } + await Task.Delay(500).ConfigureAwait(false); // Wait for all to process + + // Act + var count = await _service.CountAsync().ConfigureAwait(false); + + // Assert + Assert.Equal(5, count); + } + + [Fact] + public async Task UpsertAsync_QueuesOperationSuccessfullyAsync() + { + // Arrange + var request = new UpsertRequest + { + Content = "Test content", + MimeType = "text/plain" + }; + + // Act + var resultId = await _service.UpsertAsync(request).ConfigureAwait(false); + + // Assert - Operation should be queued + var operation = await _context.Operations + .FirstOrDefaultAsync(o => o.ContentId == resultId).ConfigureAwait(false); + + Assert.NotNull(operation); + Assert.False(operation.Complete); + Assert.False(operation.Cancelled); + Assert.Contains("upsert", operation.PlannedSteps); + } + + [Fact] + public async Task DeleteAsync_QueuesOperationSuccessfullyAsync() + { + // Arrange + var contentId = "test_delete_queue"; + + // Act + await _service.DeleteAsync(contentId).ConfigureAwait(false); + + // Assert - Operation should be queued + var operation = await _context.Operations + .FirstOrDefaultAsync(o => o.ContentId == contentId).ConfigureAwait(false); + + Assert.NotNull(operation); + Assert.False(operation.Complete); + Assert.False(operation.Cancelled); + Assert.Contains("delete", operation.PlannedSteps); + } + + [Fact] + public async Task ConcurrentUpserts_LastOneWinsAsync() + { + // Arrange + var contentId = "concurrent_test"; + + // Act - Simulate concurrent upserts + var task1 = _service.UpsertAsync(new UpsertRequest + { + Id = contentId, + Content = "Version 1", + MimeType = "text/plain" + }); + + var task2 = _service.UpsertAsync(new UpsertRequest + { + Id = contentId, + Content = "Version 2", + MimeType = "text/plain" + }); + + var task3 = _service.UpsertAsync(new UpsertRequest + { + Id = contentId, + Content = "Version 3", + MimeType = "text/plain" + }); + + await Task.WhenAll(task1, task2, task3).ConfigureAwait(false); + await Task.Delay(300).ConfigureAwait(false); // Wait for all operations to process + + // Assert - Last version should win + var content = await _service.GetByIdAsync(contentId).ConfigureAwait(false); + Assert.NotNull(content); + Assert.Equal("Version 3", content.Content); // Latest should win + } + + [Fact] + public async Task OperationCancellation_SupersededUpsertsAsync() + { + // Arrange + var contentId = "cancellation_test"; + + // Act - Create multiple upsert operations + await _service.UpsertAsync(new UpsertRequest + { + Id = contentId, + Content = "Version 1", + MimeType = "text/plain" + }).ConfigureAwait(false); + + await _service.UpsertAsync(new UpsertRequest + { + Id = contentId, + Content = "Version 2", + MimeType = "text/plain" + }).ConfigureAwait(false); + + await Task.Delay(500).ConfigureAwait(false); // Wait for processing + + // Assert - Verify operations were queued (Phase 1 always succeeds) + var operations = await _context.Operations + .Where(o => o.ContentId == contentId) + .OrderBy(o => o.Timestamp) + .ToListAsync().ConfigureAwait(false); + + Assert.Equal(2, operations.Count); + + // Eventually, the final content should be Version 2 (last write wins) + var content = await _service.GetByIdAsync(contentId).ConfigureAwait(false); + Assert.NotNull(content); + Assert.Equal("Version 2", content.Content); + } + + [Fact] + public async Task Delete_CancelsAllPreviousOperationsAsync() + { + // Arrange + var contentId = "delete_cancellation_test"; + + // Create multiple upsert operations + await _service.UpsertAsync(new UpsertRequest + { + Id = contentId, + Content = "Version 1", + MimeType = "text/plain" + }).ConfigureAwait(false); + + await _service.UpsertAsync(new UpsertRequest + { + Id = contentId, + Content = "Version 2", + MimeType = "text/plain" + }).ConfigureAwait(false); + + // Act - Delete should queue a delete operation and try to cancel previous ops + await _service.DeleteAsync(contentId).ConfigureAwait(false); + await Task.Delay(500).ConfigureAwait(false); // Wait for processing + + // Assert - Delete operation was queued (Phase 1 always succeeds) + var deleteOps = await _context.Operations + .Where(o => o.ContentId == contentId && o.PlannedStepsJson.Contains("delete")) + .ToListAsync().ConfigureAwait(false); + + Assert.NotEmpty(deleteOps); + + // Eventually, content should be deleted (delete is the last operation) + var content = await _service.GetByIdAsync(contentId).ConfigureAwait(false); + Assert.Null(content); + } + + [Fact] + public async Task RecordTimestamps_AreSetCorrectlyAsync() + { + // Arrange + var beforeCreate = DateTimeOffset.UtcNow.AddSeconds(-1); + var request = new UpsertRequest + { + Content = "Test content", + MimeType = "text/plain" + }; + + // Act + var resultId = await _service.UpsertAsync(request).ConfigureAwait(false); + await Task.Delay(100).ConfigureAwait(false); // Wait for processing + var afterCreate = DateTimeOffset.UtcNow.AddSeconds(1); + + // Assert + var content = await _service.GetByIdAsync(resultId).ConfigureAwait(false); + Assert.NotNull(content); + Assert.InRange(content.RecordCreatedAt, beforeCreate, afterCreate); + Assert.InRange(content.RecordUpdatedAt, beforeCreate, afterCreate); + } + + [Fact] + public async Task EmptyContent_IsAllowedAsync() + { + // Arrange - Empty content should be allowed + var request = new UpsertRequest + { + Content = string.Empty, + MimeType = "text/plain" + }; + + // Act + var resultId = await _service.UpsertAsync(request).ConfigureAwait(false); + await Task.Delay(100).ConfigureAwait(false); // Wait for processing + + // Assert + var content = await _service.GetByIdAsync(resultId).ConfigureAwait(false); + Assert.NotNull(content); + Assert.Equal(string.Empty, content.Content); + Assert.Equal(0, content.ByteSize); + } + + [Fact] + public async Task UpsertAsync_HandlesLargeContentAsync() + { + // Arrange - Create large content (1MB) + var largeContent = new string('x', 1024 * 1024); + var request = new UpsertRequest + { + Content = largeContent, + MimeType = "text/plain" + }; + + // Act + var resultId = await _service.UpsertAsync(request).ConfigureAwait(false); + await Task.Delay(1000).ConfigureAwait(false); // Wait longer for large content processing + + // Assert + var content = await _service.GetByIdAsync(resultId).ConfigureAwait(false); + Assert.NotNull(content); + Assert.Equal(largeContent.Length, content.Content.Length); + Assert.True(content.ByteSize >= 1024 * 1024); // Should be at least 1MB (UTF-8 encoding) + } +} diff --git a/tests/Directory.Packages.props b/tests/Directory.Packages.props index d47ecb08e..cb833b6cd 100644 --- a/tests/Directory.Packages.props +++ b/tests/Directory.Packages.props @@ -4,7 +4,9 @@ + +