WIP

PlanTempus.Components/Outbox/OutboxListener.cs

@@ -2,6 +2,7 @@ using Microsoft.ApplicationInsights;
 using Microsoft.Extensions.Hosting;
 using Npgsql;
 using PlanTempus.Core.Database.ConnectionFactory;
+using PlanTempus.Core.Telemetry;
 
 namespace PlanTempus.Components.Outbox;
 
@@ -10,19 +11,23 @@ public class OutboxListener : BackgroundService
 	private readonly IDbConnectionFactory _connectionFactory;
 	private readonly ICommandHandler _commandHandler;
 	private readonly TelemetryClient _telemetryClient;
+	private readonly IMessageChannel<string> _notificationChannel;
 
 	public OutboxListener(
 		IDbConnectionFactory connectionFactory,
 		ICommandHandler commandHandler,
-		TelemetryClient telemetryClient)
+		TelemetryClient telemetryClient,
+		IMessageChannel<string> notificationChannel)
 	{
 		_connectionFactory = connectionFactory;
 		_commandHandler = commandHandler;
 		_telemetryClient = telemetryClient;
+		_notificationChannel = notificationChannel;
 	}
 
 	protected override async Task ExecuteAsync(CancellationToken stoppingToken)
 	{
+		Console.WriteLine("OutboxListener starting - listening for outbox_messages");
 		_telemetryClient.TrackTrace("OutboxListener starting - listening for outbox_messages");
 
 		while (!stoppingToken.IsCancellationRequested)
@@ -33,6 +38,7 @@ public class OutboxListener : BackgroundService
 			}
 			catch (Exception ex) when (ex is not OperationCanceledException)
 			{
+				Console.WriteLine($"OutboxListener error: {ex.Message}");
 				_telemetryClient.TrackException(ex);
 				await Task.Delay(5000, stoppingToken);
 			}
@@ -44,18 +50,10 @@ public class OutboxListener : BackgroundService
 		await using var conn = (NpgsqlConnection)_connectionFactory.Create();
 		await conn.OpenAsync(stoppingToken);
 
-		conn.Notification += async (_, e) =>
+		conn.Notification += (_, e) =>
 		{
-			_telemetryClient.TrackTrace($"Outbox notification received: {e.Payload}");
-
-			try
-			{
-				await _commandHandler.Handle(new ProcessOutboxCommand());
-			}
-			catch (Exception ex)
-			{
-				_telemetryClient.TrackException(ex);
-			}
+			Console.WriteLine($"Notification event received: {e.Payload}");
+			_notificationChannel.Writer.TryWrite(e.Payload);
 		};
 
 		await using (var cmd = new NpgsqlCommand("LISTEN outbox_messages;", conn))
@@ -63,14 +61,32 @@ public class OutboxListener : BackgroundService
 			await cmd.ExecuteNonQueryAsync(stoppingToken);
 		}
 
+		Console.WriteLine("OutboxListener now listening on outbox_messages channel");
 		_telemetryClient.TrackTrace("OutboxListener now listening on outbox_messages channel");
 
 		// Process any pending messages on startup
+		Console.WriteLine("Processing pending messages on startup...");
 		await _commandHandler.Handle(new ProcessOutboxCommand());
 
 		while (!stoppingToken.IsCancellationRequested)
 		{
 			await conn.WaitAsync(stoppingToken);
+
+			while (_notificationChannel.Reader.TryRead(out var payload))
+			{
+				Console.WriteLine($"Outbox notification received from channel: {payload}");
+				_telemetryClient.TrackTrace($"Outbox notification received: {payload}");
+
+				try
+				{
+					await _commandHandler.Handle(new ProcessOutboxCommand());
+				}
+				catch (Exception ex)
+				{
+					Console.WriteLine($"Error processing outbox: {ex.Message}");
+					_telemetryClient.TrackException(ex);
+				}
+			}
 		}
 	}
 

PlanTempus.Components/Outbox/ProcessOutboxHandler.cs

@@ -13,23 +13,28 @@ public class ProcessOutboxHandler(
 {
 	public async Task<CommandResponse> Handle(ProcessOutboxCommand command)
 	{
+		Console.WriteLine($"ProcessOutboxHandler started");
 		telemetryClient.TrackTrace($"ProcessOutboxHandler started");
 
 		var messages = await outboxService.GetPendingAsync(command.BatchSize);
 
+		Console.WriteLine($"ProcessOutboxHandler found {messages.Count} pending messages");
 		telemetryClient.TrackTrace($"ProcessOutboxHandler found {messages.Count} pending messages");
 
 		foreach (var message in messages)
 		{
 			try
 			{
+				Console.WriteLine($"Processing message {message.Id} of type {message.Type}");
 				telemetryClient.TrackTrace($"Processing message {message.Id} of type {message.Type}");
 				await ProcessMessageAsync(message);
 				await outboxService.MarkAsSentAsync(message.Id);
+				Console.WriteLine($"Message {message.Id} marked as sent");
 				telemetryClient.TrackTrace($"Message {message.Id} marked as sent");
 			}
 			catch (Exception ex)
 			{
+				Console.WriteLine($"Message {message.Id} failed: {ex.Message}");
 				telemetryClient.TrackTrace($"Message {message.Id} failed: {ex.Message}");
 				await outboxService.MarkAsFailedAsync(message.Id, ex.Message);
 			}

PlanTempus.Components/PlanTempus.Components.csproj

@@ -5,10 +5,6 @@
 		<ImplicitUsings>enable</ImplicitUsings>
 	</PropertyGroup>
 
-	<ItemGroup>
-		<ProjectReference Include="..\Core\PlanTempus.Core.csproj" />
-	</ItemGroup>
-
 	<ItemGroup>
 	  <Folder Include="Users\Delete\" />
 	  <Folder Include="Users\Update\" />
@@ -21,4 +17,8 @@
 	  <PackageReference Include="Microsoft.AspNetCore.Mvc.Core" Version="2.3.0" />
 	</ItemGroup>
 
+	<ItemGroup>
+	  <ProjectReference Include="..\PlanTempus.Core\PlanTempus.Core.csproj" />
+	</ItemGroup>
+
 </Project>