diff --git a/ScheduledTasks/ReplayParkedQueue/Replay Parked Queues.xml b/ScheduledTasks/ReplayParkedQueue/Replay Parked Queues.xml index 8f1230e..c72a83b 100644 Binary files a/ScheduledTasks/ReplayParkedQueue/Replay Parked Queues.xml and b/ScheduledTasks/ReplayParkedQueue/Replay Parked Queues.xml differ diff --git a/ScheduledTasks/ReplayParkedQueue/ReplayParkedQueue.ps1 b/ScheduledTasks/ReplayParkedQueue/ReplayParkedQueue.ps1 index e46066d..441c091 100644 --- a/ScheduledTasks/ReplayParkedQueue/ReplayParkedQueue.ps1 +++ b/ScheduledTasks/ReplayParkedQueue/ReplayParkedQueue.ps1 @@ -79,7 +79,7 @@ try { Write-Trace "INFO" "Querying persistent subscriptions" $subscriptions = Invoke-RestMethod ` -Method GET ` - -Uri "$BaseUrl/subscriptions" ` + -Uri "$BaseUrl/subscriptions/status" ` -Headers $AuthHeader } catch { @@ -92,19 +92,8 @@ catch { # ========================= foreach ($sub in $subscriptions) { - $stream = $sub.eventStreamId - $group = $sub.groupName - - #Write-Trace "INFO" "Processing subscription [$stream][$group]" - - $streamEncoded = if ($stream -eq '$all') { - '%24all' - } - else { - [System.Web.HttpUtility]::UrlEncode($stream) - } - - $infoUrl = "$BaseUrl/subscriptions/$streamEncoded/$group/info" + $subscriptionId = $sub.subscriptionId + $infoUrl = "$BaseUrl/subscriptions/$subscriptionId/status" try { $info = Invoke-RestMethod ` @@ -113,17 +102,17 @@ foreach ($sub in $subscriptions) { -Headers $AuthHeader } catch { - Write-Trace "WARN" "Failed to get info for [$stream][$group]: $($_.Exception.Message)" + Write-Trace "WARN" "Failed to get info for [$subscriptionId]: $($_.Exception.Message)" continue } + #Write-Trace "INFO" "[$subscriptionId] $info" + $parkedEventCount = $info.parkedEventCount + #Write-Trace "INFO" "[$subscriptionId] parkedEventCount=$parkedEventCount" - $parkedCount = $info.parkedMessageCount - #Write-Trace "INFO" "[$stream][$group] parkedMessageCount=$parkedCount" - - if ($parkedCount -gt 0) { - $replayUrl = "$BaseUrl/subscriptions/$streamEncoded/$group/replayParked?from=0" + if ($parkedEventCount -gt 0) { + $replayUrl = "$BaseUrl/subscriptions/$subscriptionId/replay" - Write-Trace "INFO" "Replaying [$parkedCount] parked messages for [$stream][$group]" + Write-Trace "INFO" "Replaying [$parkedEventCount] parked messages for [$subscriptionId]" try { Invoke-RestMethod ` @@ -131,14 +120,14 @@ foreach ($sub in $subscriptions) { -Uri $replayUrl ` -Headers $AuthHeader - Write-Trace "INFO" "Replay triggered successfully for [$stream][$group]" + Write-Trace "INFO" "Replay triggered successfully for [$subscriptionId]" } catch { - Write-Trace "ERROR" "Replay failed for [$stream][$group]: $($_.Exception.Message)" + Write-Trace "ERROR" "Replay failed for [$subscriptionId]: $($_.Exception.Message)" } } #else { - #Write-Trace "INFO" "No parked messages for [$stream][$group]" + # Write-Trace "INFO" "No parked messages for [$subscriptionId]" #} } diff --git a/ScheduledTasks/ReplayParkedQueue/ReplayParkedQueue_EventStore.ps1 b/ScheduledTasks/ReplayParkedQueue/ReplayParkedQueue_EventStore.ps1 new file mode 100644 index 0000000..e46066d --- /dev/null +++ b/ScheduledTasks/ReplayParkedQueue/ReplayParkedQueue_EventStore.ps1 @@ -0,0 +1,145 @@ +param ( + [Parameter(Mandatory = $true)] + [string]$BaseUrl, + + [Parameter()] + [System.Management.Automation.PSCredential]$Credential, + + [string]$LogDirectory = "C:\home\txnproc\trace", + [int]$LogRetentionDays = 7 +) + +# ========================= +# Logging setup +# ========================= +if (-not (Test-Path $LogDirectory)) { + New-Item -ItemType Directory -Path $LogDirectory -Force | Out-Null +} + +function Get-LogFilePath { + $date = (Get-Date).ToString("yyyy-MM-dd") + Join-Path $LogDirectory "ReplayParkedSubscriptions-$date.log" +} + +function Write-Trace { + param ( + [string]$Level, + [string]$Message + ) + + $timestamp = (Get-Date).ToString("yyyy-MM-dd HH:mm:ss.fff") + $entry = "$timestamp [$Level] $Message" + Add-Content -Path (Get-LogFilePath) -Value $entry +} + +# ========================= +# Log retention cleanup +# ========================= +$cutoffDate = (Get-Date).Date.AddDays(-$LogRetentionDays) + +Get-ChildItem -Path $LogDirectory -Filter "ReplayParkedSubscriptions-*.log" -File | + Where-Object { + if ($_.Name -match 'ReplayParkedSubscriptions-(\d{4}-\d{2}-\d{2})\.log') { + [DateTime]::ParseExact($matches[1], 'yyyy-MM-dd', $null) -lt $cutoffDate + } + else { + $false + } + } | + ForEach-Object { + try { + Remove-Item $_.FullName -Force + } + catch { + Write-Trace "WARN" "Failed to remove stale log file $($_.FullName): $($_.Exception.Message)" + } + } + +Write-Trace "INFO" "Script started. BaseUrl=$BaseUrl" + +# ========================= +# Auth header +# ========================= +$AuthHeader = @{} +if ($null -ne $Credential) { + $networkCredential = $Credential.GetNetworkCredential() + $username = $networkCredential.UserName + $password = $networkCredential.Password + + $AuthHeader.Authorization = "Basic " + + [Convert]::ToBase64String( + [Text.Encoding]::ASCII.GetBytes("$username`:$password") + ) +} + +# ========================= +# Query subscriptions +# ========================= +try { + Write-Trace "INFO" "Querying persistent subscriptions" + $subscriptions = Invoke-RestMethod ` + -Method GET ` + -Uri "$BaseUrl/subscriptions" ` + -Headers $AuthHeader +} +catch { + Write-Trace "ERROR" "Failed to query subscriptions: $($_.Exception.Message)" + exit 1 +} + +# ========================= +# Process subscriptions +# ========================= +foreach ($sub in $subscriptions) { + + $stream = $sub.eventStreamId + $group = $sub.groupName + + #Write-Trace "INFO" "Processing subscription [$stream][$group]" + + $streamEncoded = if ($stream -eq '$all') { + '%24all' + } + else { + [System.Web.HttpUtility]::UrlEncode($stream) + } + + $infoUrl = "$BaseUrl/subscriptions/$streamEncoded/$group/info" + + try { + $info = Invoke-RestMethod ` + -Method GET ` + -Uri $infoUrl ` + -Headers $AuthHeader + } + catch { + Write-Trace "WARN" "Failed to get info for [$stream][$group]: $($_.Exception.Message)" + continue + } + + $parkedCount = $info.parkedMessageCount + #Write-Trace "INFO" "[$stream][$group] parkedMessageCount=$parkedCount" + + if ($parkedCount -gt 0) { + $replayUrl = "$BaseUrl/subscriptions/$streamEncoded/$group/replayParked?from=0" + + Write-Trace "INFO" "Replaying [$parkedCount] parked messages for [$stream][$group]" + + try { + Invoke-RestMethod ` + -Method POST ` + -Uri $replayUrl ` + -Headers $AuthHeader + + Write-Trace "INFO" "Replay triggered successfully for [$stream][$group]" + } + catch { + Write-Trace "ERROR" "Replay failed for [$stream][$group]: $($_.Exception.Message)" + } + } + #else { + #Write-Trace "INFO" "No parked messages for [$stream][$group]" + #} +} + +Write-Trace "INFO" "Script completed" diff --git a/ScheduledTasks/ReplayParkedQueue/Tests/ReplayParkedQueue.Tests.ps1 b/ScheduledTasks/ReplayParkedQueue/Tests/ReplayParkedQueue.Tests.ps1 index f17cf70..5b9a7dc 100644 --- a/ScheduledTasks/ReplayParkedQueue/Tests/ReplayParkedQueue.Tests.ps1 +++ b/ScheduledTasks/ReplayParkedQueue/Tests/ReplayParkedQueue.Tests.ps1 @@ -1,5 +1,5 @@ Describe 'ReplayParkedQueue.ps1' { - It 'replays parked messages for subscriptions with parked counts and encodes $all streams' { + It 'replays parked messages for subscriptions with parked counts' { $scriptUnderTest = (Resolve-Path (Join-Path (Split-Path $PSScriptRoot -Parent) 'ReplayParkedQueue.ps1')).Path $logDirectory = Join-Path $TestDrive 'replay-logs-1' $securePassword = [System.Security.SecureString]::new() @@ -9,12 +9,10 @@ Describe 'ReplayParkedQueue.ps1' { $credential = [pscredential]::new('alice', $securePassword) $subscriptions = @( [pscustomobject]@{ - eventStreamId = 'stream-1' - groupName = 'group-1' + subscriptionId = 'subscription-1' }, [pscustomobject]@{ - eventStreamId = '$all' - groupName = 'group-2' + subscriptionId = 'subscription-2' } ) $captured = [pscustomobject]@{ @@ -28,30 +26,30 @@ Describe 'ReplayParkedQueue.ps1' { $subscriptions } -ParameterFilter { $Method -eq 'GET' -and - $Uri -eq 'https://queue.example/subscriptions' + $Uri -eq 'https://queue.example/subscriptions/status' } Mock Invoke-RestMethod { $captured.InfoUris.Add($Uri) | Out-Null - @{ parkedMessageCount = 2 } + @{ parkedEventCount = 2 } } -ParameterFilter { $Method -eq 'GET' -and - $Uri -eq 'https://queue.example/subscriptions/stream-1/group-1/info' + $Uri -eq 'https://queue.example/subscriptions/subscription-1/status' } Mock Invoke-RestMethod { $captured.InfoUris.Add($Uri) | Out-Null - @{ parkedMessageCount = 0 } + @{ parkedEventCount = 0 } } -ParameterFilter { $Method -eq 'GET' -and - $Uri -eq 'https://queue.example/subscriptions/%24all/group-2/info' + $Uri -eq 'https://queue.example/subscriptions/subscription-2/status' } Mock Invoke-RestMethod { $captured.ReplayUris.Add($Uri) | Out-Null } -ParameterFilter { $Method -eq 'POST' -and - $Uri -eq 'https://queue.example/subscriptions/stream-1/group-1/replayParked?from=0' + $Uri -eq 'https://queue.example/subscriptions/subscription-1/replay' } & $scriptUnderTest 'https://queue.example' -Credential $credential -LogDirectory $logDirectory @@ -59,11 +57,11 @@ Describe 'ReplayParkedQueue.ps1' { $captured.SubscriptionCalls | Should -Be 1 $captured.InfoUris.Count | Should -Be 2 $captured.InfoUris | Should -Be @( - 'https://queue.example/subscriptions/stream-1/group-1/info', - 'https://queue.example/subscriptions/%24all/group-2/info' + 'https://queue.example/subscriptions/subscription-1/status', + 'https://queue.example/subscriptions/subscription-2/status' ) $captured.ReplayUris | Should -Be @( - 'https://queue.example/subscriptions/stream-1/group-1/replayParked?from=0' + 'https://queue.example/subscriptions/subscription-1/replay' ) } @@ -77,8 +75,7 @@ Describe 'ReplayParkedQueue.ps1' { $credential = [pscredential]::new('alice', $securePassword) $subscriptions = @( [pscustomobject]@{ - eventStreamId = 'stream-1' - groupName = 'group-1' + subscriptionId = 'subscription-1' } ) $captured = [pscustomobject]@{ @@ -92,22 +89,22 @@ Describe 'ReplayParkedQueue.ps1' { $subscriptions } -ParameterFilter { $Method -eq 'GET' -and - $Uri -eq 'https://queue.example/subscriptions' + $Uri -eq 'https://queue.example/subscriptions/status' } Mock Invoke-RestMethod { $captured.InfoCalls++ - @{ parkedMessageCount = 0 } + @{ parkedEventCount = 0 } } -ParameterFilter { $Method -eq 'GET' -and - $Uri -eq 'https://queue.example/subscriptions/stream-1/group-1/info' + $Uri -eq 'https://queue.example/subscriptions/subscription-1/status' } Mock Invoke-RestMethod { $captured.ReplayCalls++ } -ParameterFilter { $Method -eq 'POST' -and - $Uri -like 'https://queue.example/subscriptions/*/*/replayParked?from=0' + $Uri -like 'https://queue.example/subscriptions/*/replay' } & $scriptUnderTest 'https://queue.example' -Credential $credential -LogDirectory $logDirectory @@ -129,7 +126,7 @@ Describe 'ReplayParkedQueue.ps1' { @() } -ParameterFilter { $Method -eq 'GET' -and - $Uri -eq 'https://queue.example/subscriptions' + $Uri -eq 'https://queue.example/subscriptions/status' } & $scriptUnderTest 'https://queue.example' -LogDirectory $logDirectory diff --git a/TransactionProcessing.MerchantPos/Program.cs b/TransactionProcessing.MerchantPos/Program.cs index dbf2323..db6a3c4 100644 --- a/TransactionProcessing.MerchantPos/Program.cs +++ b/TransactionProcessing.MerchantPos/Program.cs @@ -35,6 +35,7 @@ .SetBasePath(builder.Environment.ContentRootPath) .AddJsonFile("appsettings.json", optional: true, reloadOnChange: true) .AddJsonFile($"appsettings.{builder.Environment.EnvironmentName}.json", optional: true, reloadOnChange: true) + .AddJsonFile($"appsettings.local.json", optional: true, reloadOnChange: true) .AddEnvironmentVariables() .AddCommandLine(args); diff --git a/TransactionProcessor.SystemSetupTool/EventStoreFunctions.cs b/TransactionProcessor.SystemSetupTool/EventStoreFunctions.cs index aecc424..8ceb073 100644 --- a/TransactionProcessor.SystemSetupTool/EventStoreFunctions.cs +++ b/TransactionProcessor.SystemSetupTool/EventStoreFunctions.cs @@ -161,7 +161,8 @@ private async Task SetupSubscriptions(CancellationToken cancellationToke ("$idx-ce-TransactionAggregate", "Transaction Processor", "Domain"), ("$idx-ce-SettlementAggregate", "Transaction Processor", "Domain"), ("$idx-ce-FloatAggregate", "Transaction Processor", "Domain"), - + ("$idx-ce-MerchantStatementForDateAggregate", "Transaction Processor", "Domain"), + ("$idx-ce-EstateAggregate", "Transaction Processor", "Ordered"), ("$idx-ce-SettlementAggregate", "Transaction Processor", "Ordered"), ("$idx-ce-VoucherAggregate", "Transaction Processor", "Ordered"),