using System.Net; using System.Net.Http.Json; using System.Text.Json; using ItWorx.KnowledgeHub.Application.Jobs; using ItWorx.KnowledgeHub.Contracts.Jobs; using ItWorx.KnowledgeHub.Contracts.Knowledge; using ItWorx.KnowledgeHub.Domain.Jobs; using ItWorx.KnowledgeHub.Infrastructure.Persistence; using ItWorx.KnowledgeHub.IntegrationTests.Infrastructure; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.DependencyInjection; namespace ItWorx.KnowledgeHub.IntegrationTests.Jobs; [Collection(PostgresTestGroup.Name)] public sealed class BackgroundJobWorkerTests(PostgresContainerFixture postgres) : OwnerSetupTestBase(postgres) { [Fact] public async Task Worker_tags_a_database_snapshot_in_bounded_steps_and_preserves_content_metadata_and_audit() { using var client = Factory.CreateClient(); using var login = await SetupHelper.LoginAsync(client, OwnerEmail, OwnerPassword); Assert.Equal(HttpStatusCode.OK, login.StatusCode); var rootId = await CreatePageAsync(client, null, "Course root", "root markdown", []); var childId = await CreatePageAsync(client, rootId, "Already tagged child", "child markdown", ["course:ccna", "KeepMe"]); var grandchildId = await CreatePageAsync(client, rootId, "Course grandchild", "grandchild markdown", ["existing"]); var csrf = await SetupHelper.GetCsrfTokenAsync(client); using var request = new HttpRequestMessage(HttpMethod.Post, "/api/v1/jobs/course-tagging") { Content = JsonContent.Create(new QueueCourseTaggingJobRequest(rootId, " COURSE:CCNA ")) }; request.Headers.Add(csrf.HeaderName, csrf.Value); request.Headers.Add("Idempotency-Key", "worker-course-tagging"); using var queued = await client.SendAsync(request); Assert.Equal(HttpStatusCode.Accepted, queued.StatusCode); using var queuedJson = JsonDocument.Parse(await queued.Content.ReadAsStringAsync()); var jobId = queuedJson.RootElement.GetProperty("id").GetGuid(); var terminal = await WaitForTerminalAsync(client, jobId, TimeSpan.FromSeconds(30)); Assert.Equal("succeeded", terminal.GetProperty("status").GetString()); Assert.Equal(3, terminal.GetProperty("total").GetInt32()); Assert.Equal(3, terminal.GetProperty("processed").GetInt32()); Assert.Equal(2, terminal.GetProperty("succeeded").GetInt32()); Assert.Equal(1, terminal.GetProperty("skipped").GetInt32()); Assert.Equal(0, terminal.GetProperty("failed").GetInt32()); Assert.Equal(100, terminal.GetProperty("progress").GetInt32()); using var scope = Factory.Services.CreateScope(); var db = scope.ServiceProvider.GetRequiredService(); var repository = scope.ServiceProvider.GetRequiredService(); var root = await repository.FindAsync(rootId, CancellationToken.None); var child = await repository.FindAsync(childId, CancellationToken.None); var grandchild = await repository.FindAsync(grandchildId, CancellationToken.None); Assert.NotNull(root); Assert.NotNull(child); Assert.NotNull(grandchild); Assert.Contains(root.Tags, tag => tag == "course:ccna"); Assert.Contains(child.Tags, tag => tag == "course:ccna"); Assert.Contains(child.Tags, tag => tag == "KeepMe"); Assert.Contains(grandchild.Tags, tag => tag == "course:ccna"); Assert.Equal("root markdown", root.Markdown); Assert.Equal("child markdown", child.Markdown); Assert.Equal("grandchild markdown", grandchild.Markdown); Assert.Equal(2, root.CurrentRevisionNumber); Assert.Equal(1, child.CurrentRevisionNumber); Assert.Equal(2, grandchild.CurrentRevisionNumber); var changedIds = new[] { rootId.ToString("D"), grandchildId.ToString("D") }; Assert.Equal(2, await db.AuditEvents.AsNoTracking() .CountAsync(audit => audit.Action == "page.published" && changedIds.Contains(audit.TargetId))); Assert.Equal(2, await db.OutboxMessages.AsNoTracking() .CountAsync(message => message.EventType == "page.updated.v1" && changedIds.Contains(message.AggregateId))); } [Fact] public async Task Backup_requests_run_through_the_shared_backup_use_case_and_return_no_private_path() { using var client = Factory.CreateClient(); using var login = await SetupHelper.LoginAsync(client, OwnerEmail, OwnerPassword); Assert.Equal(HttpStatusCode.OK, login.StatusCode); var csrf = await SetupHelper.GetCsrfTokenAsync(client); using var request = new HttpRequestMessage(HttpMethod.Post, "/api/v1/jobs/backups") { Content = JsonContent.Create(new QueueBackupJobRequest()) }; request.Headers.Add(csrf.HeaderName, csrf.Value); request.Headers.Add("Idempotency-Key", "worker-backup-job"); using var queued = await client.SendAsync(request); Assert.Equal(HttpStatusCode.Accepted, queued.StatusCode); using var queuedJson = JsonDocument.Parse(await queued.Content.ReadAsStringAsync()); var jobId = queuedJson.RootElement.GetProperty("id").GetGuid(); var terminal = await WaitForTerminalAsync(client, jobId, TimeSpan.FromSeconds(45)); Assert.Equal("succeeded", terminal.GetProperty("status").GetString()); Assert.Equal(1, terminal.GetProperty("total").GetInt32()); Assert.Equal(1, terminal.GetProperty("succeeded").GetInt32()); var result = terminal.GetProperty("result"); Assert.False(result.TryGetProperty("path", out _)); Assert.False(result.TryGetProperty("directory", out _)); Assert.False(string.IsNullOrWhiteSpace(result.GetProperty("backupId").GetString())); using var scope = Factory.Services.CreateScope(); var audit = scope.ServiceProvider.GetRequiredService().AuditEvents; Assert.True(await audit.AsNoTracking().AnyAsync(item => item.Action == "backup.created")); } [Fact] public async Task Worker_persists_only_a_safe_bounded_error_when_backup_storage_fails_with_a_pathful_exception() { var backupPath = Factory.BackupsDirectory; Assert.False(Directory.Exists(backupPath)); Assert.False(File.Exists(backupPath)); const string privatePathMarker = "private-backup-storage-marker"; File.WriteAllText(backupPath, privatePathMarker); try { using var client = Factory.CreateClient(); using var login = await SetupHelper.LoginAsync(client, OwnerEmail, OwnerPassword); Assert.Equal(HttpStatusCode.OK, login.StatusCode); var csrf = await SetupHelper.GetCsrfTokenAsync(client); using var request = new HttpRequestMessage(HttpMethod.Post, "/api/v1/jobs/backups") { Content = JsonContent.Create(new QueueBackupJobRequest()) }; request.Headers.Add(csrf.HeaderName, csrf.Value); request.Headers.Add("Idempotency-Key", "backup-safe-error"); using var queued = await client.SendAsync(request); Assert.Equal(HttpStatusCode.Accepted, queued.StatusCode); using var queuedJson = JsonDocument.Parse(await queued.Content.ReadAsStringAsync()); var jobId = queuedJson.RootElement.GetProperty("id").GetGuid(); var failedAttempt = await WaitForErrorsAsync(client, jobId, TimeSpan.FromSeconds(20)); var errors = failedAttempt.GetProperty("errors"); Assert.NotEqual(0, errors.GetArrayLength()); Assert.Equal("processing_retry", errors[0].GetProperty("code").GetString()); var safeMessage = errors[0].GetProperty("message").GetString(); Assert.Equal("De tijdelijke verwerking is mislukt; de taak wordt opnieuw geprobeerd.", safeMessage); Assert.DoesNotContain(backupPath, safeMessage, StringComparison.OrdinalIgnoreCase); Assert.DoesNotContain(privatePathMarker, safeMessage, StringComparison.Ordinal); Assert.True(safeMessage!.Length <= BackgroundJob.MaxErrorMessageLength); using var cancel = await SetupHelper.PostWithCsrfAsync( client, $"/api/v1/jobs/{jobId:D}/cancel", new { }); Assert.Equal(HttpStatusCode.OK, cancel.StatusCode); var terminal = await WaitForTerminalAsync(client, jobId, TimeSpan.FromSeconds(30)); var terminalStatus = terminal.GetProperty("status").GetString(); Assert.True(terminalStatus is "cancelled" or "failed"); } finally { if (File.Exists(backupPath)) File.Delete(backupPath); } } private static async Task CreatePageAsync( HttpClient client, Guid? parentId, string title, string markdown, IReadOnlyList tags) { using var response = await SetupHelper.PostWithCsrfAsync( client, "/api/v1/pages", new CreatePageRequest(parentId, title, "WikiPage", markdown, null, tags)); Assert.True(response.IsSuccessStatusCode, await response.Content.ReadAsStringAsync()); using var json = JsonDocument.Parse(await response.Content.ReadAsStringAsync()); return json.RootElement.GetProperty("id").GetGuid(); } private static async Task WaitForTerminalAsync(HttpClient client, Guid jobId, TimeSpan timeout) { var stopAt = DateTimeOffset.UtcNow + timeout; while (DateTimeOffset.UtcNow < stopAt) { using var response = await client.GetAsync($"/api/v1/jobs/{jobId:D}"); if (response.StatusCode == HttpStatusCode.OK) { using var json = JsonDocument.Parse(await response.Content.ReadAsStringAsync()); var job = json.RootElement.Clone(); if (job.GetProperty("status").GetString() is "succeeded" or "partially_succeeded" or "failed" or "cancelled") return job; } await Task.Delay(100); } throw new TimeoutException($"Job {jobId:D} did not reach a terminal state within {timeout}."); } private static async Task WaitForErrorsAsync(HttpClient client, Guid jobId, TimeSpan timeout) { var stopAt = DateTimeOffset.UtcNow + timeout; while (DateTimeOffset.UtcNow < stopAt) { using var response = await client.GetAsync($"/api/v1/jobs/{jobId:D}"); if (response.StatusCode == HttpStatusCode.OK) { using var json = JsonDocument.Parse(await response.Content.ReadAsStringAsync()); var job = json.RootElement.Clone(); if (job.GetProperty("errors").GetArrayLength() > 0) return job; } await Task.Delay(100); } throw new TimeoutException($"Job {jobId:D} did not persist an error within {timeout}."); } }