From 1132865acaa283f195d0d51cef841ddb25c9ff90 Mon Sep 17 00:00:00 2001 From: StuartFerguson Date: Tue, 8 Sep 2026 20:39:44 +0100 Subject: [PATCH 1/2] ignore launch settings --- .gitignore | 1 + 1 file changed, 1 insertion(+) diff --git a/.gitignore b/.gitignore index 5bf2a2c..8937902 100644 --- a/.gitignore +++ b/.gitignore @@ -356,3 +356,4 @@ MigrationBackup/ /TransactionProcessing.MerchantPos/merchant.db /TransactionProcessing.MerchantPos/merchant.db-shm /TransactionProcessing.MerchantPos/merchant.db-wal +/TransactionProcessor.SystemSetupTool/Properties/launchSettings.json From afcd36b5e9908f084b767010e62a958c9dbe2fef Mon Sep 17 00:00:00 2001 From: StuartFerguson Date: Tue, 8 Sep 2026 20:39:48 +0100 Subject: [PATCH 2/2] tweaks to setup tool to support catchup service --- .../CatchupServiceClient.cs | 174 ++++++++++++++ .../EstateSetupFunctions.cs | 35 ++- .../EventStoreFunctions.cs | 224 ++++++++++++++---- .../Program.cs | 19 +- .../appsettings.json | 17 +- .../identityserverconfig.json | 6 +- 6 files changed, 400 insertions(+), 75 deletions(-) create mode 100644 TransactionProcessor.SystemSetupTool/CatchupServiceClient.cs diff --git a/TransactionProcessor.SystemSetupTool/CatchupServiceClient.cs b/TransactionProcessor.SystemSetupTool/CatchupServiceClient.cs new file mode 100644 index 0000000..2f6175b --- /dev/null +++ b/TransactionProcessor.SystemSetupTool/CatchupServiceClient.cs @@ -0,0 +1,174 @@ +using System; +using System.Collections.Generic; +using System.Net; +using System.Net.Http; +using System.Net.Http.Json; +using System.Text; +using System.Text.Json; +using System.Threading; +using System.Threading.Tasks; + +namespace TransactionProcessor.SystemSetupTool +{ + public sealed record SubscriptionConfigurationRequest + { + public string? SubscriptionId { get; init; } + public string? SecondaryIndexName { get; init; } + public string? EndpointUrl { get; init; } + public string? Tag { get; init; } + public int TimeoutSeconds { get; init; } = 30; + public int RetryMaxAttempts { get; init; } = 3; + public int RetryDelaySeconds { get; init; } = 1; + public int CheckpointBatchSize { get; init; } = 100; + public bool ContinueOnParked { get; init; } + public AuthenticationConfiguration? Authentication { get; init; } + public bool SoftDeleteParked { get; init; } = true; + } + + public interface ICatchupServiceClient + { + Task>> GetEndpointsAsync(CancellationToken cancellationToken = default); + Task> CreateEndpointAsync(EndpointDefinition endpoint, CancellationToken cancellationToken = default); + Task>> GetSubscriptionConfigurationsAsync(CancellationToken cancellationToken = default); + Task> CreateSubscriptionConfigurationAsync(SubscriptionConfigurationRequest request, CancellationToken cancellationToken = default); + Task StartSubscriptionAsync(string subscriptionId, CancellationToken cancellationToken = default); + Task> CreateIndexAsync(string indexName, string payload, CancellationToken cancellationToken = default); + Task> GetIndexAsync(String indexName,CancellationToken cancellationToken = default); + } + + public record CatchupApiResponse(HttpStatusCode StatusCode, string? Content, string? ContentType) + { + public bool IsSuccessStatusCode => (int)StatusCode is >= 200 and < 300; + } + + public sealed record CatchupApiResponse(HttpStatusCode StatusCode, T? Value, string? Content, string? ContentType) + : CatchupApiResponse(StatusCode, Content, ContentType); + + public sealed class CatchupServiceClient(Func urlResolver, HttpClient httpClient) : ICatchupServiceClient + { + private readonly Func UrlResolver = urlResolver; + private static readonly JsonSerializerOptions JsonOptions = new(JsonSerializerDefaults.Web); + + public Task> GetIndexAsync(String indexName, CancellationToken cancellationToken = default) => + SendRawAsync(HttpMethod.Get, $"indexes/{Escape(indexName)}", null, cancellationToken); + + public Task>> GetEndpointsAsync(CancellationToken cancellationToken = default) => + SendJsonAsync>(HttpMethod.Get, "endpoints", null, cancellationToken); + + public Task> CreateEndpointAsync(EndpointDefinition endpoint, CancellationToken cancellationToken = default) => + SendJsonAsync(HttpMethod.Post, "endpoints", endpoint, cancellationToken); + + public Task>> GetSubscriptionConfigurationsAsync(CancellationToken cancellationToken = default) => + SendJsonAsync>(HttpMethod.Get, "subscriptions/config", null, cancellationToken); + + public Task> CreateSubscriptionConfigurationAsync(SubscriptionConfigurationRequest request, CancellationToken cancellationToken = default) => + SendJsonAsync(HttpMethod.Post, "subscriptions/config", request, cancellationToken); + + public Task StartSubscriptionAsync(string subscriptionId, CancellationToken cancellationToken = default) => + SendAsync(HttpMethod.Post, $"subscriptions/config/{Escape(subscriptionId)}/start", null, cancellationToken); + + public Task> CreateIndexAsync(string indexName, string payload, CancellationToken cancellationToken = default) => + SendRawAsync(HttpMethod.Post, $"indexes/{Escape(indexName)}", payload, cancellationToken); + + private async Task> SendJsonAsync(HttpMethod method, string path, object? body, CancellationToken cancellationToken) + { + var response = await SendAsync(method, path, body, cancellationToken); + T? value = default; + if (response.IsSuccessStatusCode && !string.IsNullOrWhiteSpace(response.Content)) + value = JsonSerializer.Deserialize(response.Content, JsonOptions); + + return new(response.StatusCode, value, response.Content, response.ContentType); + } + + private async Task> SendRawAsync(HttpMethod method, string path, string? body, CancellationToken cancellationToken) + { + var response = await SendAsync(method, path, body is null ? null : new StringContent(body, Encoding.UTF8, "application/json"), cancellationToken); + return new(response.StatusCode, response.Content, response.Content, response.ContentType); + } + + private async Task SendAsync(HttpMethod method, string path, object? body, CancellationToken cancellationToken) + { + using var content = body is null ? null : JsonContent.Create(body, options: JsonOptions); + return await SendAsync(method, path, content, cancellationToken); + } + + private async Task SendAsync(HttpMethod method, string path, HttpContent? content, CancellationToken cancellationToken) { + var baseUrl = this.UrlResolver("CatchupService"); + + string fullUrl = $"{baseUrl}/{path}"; + + + using var request = new HttpRequestMessage(method, fullUrl) { Content = content }; + using var response = await httpClient.SendAsync(request, HttpCompletionOption.ResponseHeadersRead, cancellationToken); + var responseContent = await response.Content.ReadAsStringAsync(cancellationToken); + return new(response.StatusCode, string.IsNullOrWhiteSpace(responseContent) ? null : responseContent, response.Content.Headers.ContentType?.ToString()); + } + + private static string Escape(string value) => Uri.EscapeDataString(value); + + private static void AddQuery(ICollection query, string name, string? value) + { + if (!string.IsNullOrWhiteSpace(value)) + query.Add($"{Escape(name)}={Escape(value)}"); + } + + private static string AddQueryString(string path, IReadOnlyCollection query) => + query.Count == 0 ? path : $"{path}?{string.Join('&', query)}"; + } + + public sealed record SubscriptionDefinition( + string SubscriptionId, + string SecondaryIndexName, + string EndpointUrl, + string Tag, + TimeoutSettings Timeout, + RetrySettings Retry, + CheckpointSettings Checkpoint, + bool ContinueOnParked = false, + AuthenticationConfiguration? Authentication = null) + { + public int? EndpointId { get; init; } + public bool Enabled { get; init; } = true; + public SubscriptionOperationalState OperationalState { get; init; } = SubscriptionOperationalState.Healthy; + public string? OperationalReason { get; init; } + public bool SoftDeleteParked { get; init; } = true; + + public Uri Endpoint => new(EndpointUrl, UriKind.Absolute); + } + + public sealed record RetrySettings(int MaxAttempts, TimeSpan Delay) + { + public static RetrySettings Default { get; } = new(3, TimeSpan.FromSeconds(1)); + } + + public sealed record CheckpointSettings(int BatchSize) + { + public static CheckpointSettings Default { get; } = new(100); + } + + public sealed record TimeoutSettings(TimeSpan RequestTimeout) + { + public static TimeoutSettings Default { get; } = new(TimeSpan.FromSeconds(30)); + } + + public sealed record AuthenticationConfiguration(string? Scheme, IReadOnlyDictionary Parameters) + { + public static AuthenticationConfiguration Empty { get; } = new(null, new Dictionary()); + } + + + public enum SubscriptionOperationalState + { + Healthy = 0, + Stopped = 1, + Faulted = 2 + } + + public sealed record EndpointDefinition( + string Name, + string Url, + AuthenticationConfiguration? Authentication = null) + { + public Uri Uri => new(Url, UriKind.Absolute); + } +} diff --git a/TransactionProcessor.SystemSetupTool/EstateSetupFunctions.cs b/TransactionProcessor.SystemSetupTool/EstateSetupFunctions.cs index a9cb600..25dd0e3 100644 --- a/TransactionProcessor.SystemSetupTool/EstateSetupFunctions.cs +++ b/TransactionProcessor.SystemSetupTool/EstateSetupFunctions.cs @@ -1,13 +1,14 @@ -using System; -using System.Collections.Generic; -using System.Linq; -using System.Threading; -using System.Threading.Tasks; +using Microsoft.AspNetCore.DataProtection; using SecurityService.Client; using SecurityService.DataTransferObjects; using Shared.Exceptions; using Shared.Results; using SimpleResults; +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; using TransactionProcessor.Client; using TransactionProcessor.DataTransferObjects; using TransactionProcessor.DataTransferObjects.Requests.Contract; @@ -53,6 +54,11 @@ public async Task SetupEstate(CancellationToken cancellationToken) { return ResultHelpers.CreateFailure(createEstateResult); this.EstateId = createEstateResult.Data; + + Result getEstateResult = await this.GetEstate(this.EstateId, cancellationToken); + if (getEstateResult.IsFailed) + return ResultHelpers.CreateFailure(createEstateResult); + var createUserResult = await this.CreateEstateUser(cancellationToken); if (createUserResult.IsFailed) return ResultHelpers.CreateFailure(createUserResult); @@ -77,6 +83,15 @@ public async Task SetupEstate(CancellationToken cancellationToken) { return Result.Success(); } + private static string ResolveUserPassword(String password) { + if (!string.IsNullOrWhiteSpace(password)) + { + return password; + } + return Environment.GetEnvironmentVariable("DEFAULT_USER_PASSWORD"); + } + + private static string ResolveServiceClientSecret() { string secret = Environment.GetEnvironmentVariable("TRANSACTIONPROCESSOR_SERVICECLIENT_SECRET"); @@ -90,8 +105,10 @@ private static string ResolveServiceClientSecret() private async Task> GetEstate(Guid estateId, CancellationToken cancellationToken) { - Result estateResponse = await this.TransactionProcessorClient.GetEstate(this.TokenResponse.AccessToken, estateId, cancellationToken); - return estateResponse; + var estateResponse = await this.TransactionProcessorClient.GetEstates(this.TokenResponse.AccessToken, estateId, cancellationToken); + if (estateResponse.IsFailed) + return ResultHelpers.CreateFailure(estateResponse); + return Result.Success(estateResponse.Data.Single()); } private async Task> CreateEstate(CancellationToken cancellationToken) { @@ -131,7 +148,7 @@ private async Task CreateEstateUser(CancellationToken cancellationToken) FamilyName = this.EstateConfig.User.FamilyName, GivenName = this.EstateConfig.User.GivenName, MiddleName = this.EstateConfig.User.MiddleName, - Password = this.EstateConfig.User.Password + Password = ResolveUserPassword(this.EstateConfig.User.Password) }; return await this.TransactionProcessorClient.CreateEstateUser(this.TokenResponse.AccessToken, this.EstateId, createEstateUserRequest, cancellationToken); } @@ -371,7 +388,7 @@ public async Task CreateMerchant(Merchant merchant, FamilyName = merchant.User.FamilyName, GivenName = merchant.User.GivenName, MiddleName = merchant.User.MiddleName, - Password = merchant.User.Password + Password = ResolveUserPassword(merchant.User.Password) }; var createMerchantUserResult = await this.TransactionProcessorClient.CreateMerchantUser(this.TokenResponse.AccessToken, this.EstateId, createMerchantRequest.MerchantId.Value, createMerchantUserRequest, cancellationToken); if (createMerchantUserResult.IsFailed) diff --git a/TransactionProcessor.SystemSetupTool/EventStoreFunctions.cs b/TransactionProcessor.SystemSetupTool/EventStoreFunctions.cs index 7520edd..aecc424 100644 --- a/TransactionProcessor.SystemSetupTool/EventStoreFunctions.cs +++ b/TransactionProcessor.SystemSetupTool/EventStoreFunctions.cs @@ -1,11 +1,14 @@ using System; using System.Collections.Generic; +using System.ComponentModel.DataAnnotations; using System.IO; using System.Linq; +using System.Net; using System.Threading; using System.Threading.Tasks; using EventStore.Client; using KurrentDB.Client; +using Shared.Results; using SimpleResults; namespace TransactionProcessor.SystemSetupTool; @@ -18,77 +21,195 @@ public class EventStoreFunctions{ private const string MessagingServiceGroup = "Messaging Service"; private readonly KurrentDBProjectionManagementClient ProjectionClient; + private readonly ICatchupServiceClient CatchupServiceClient; - private readonly KurrentDBPersistentSubscriptionsClient PersistentSubscriptionsClient; - - public EventStoreFunctions(KurrentDBProjectionManagementClient projectionClient, KurrentDBPersistentSubscriptionsClient persistentSubscriptionsClient){ + public EventStoreFunctions(KurrentDBProjectionManagementClient projectionClient,ICatchupServiceClient catchupServiceClient){ this.ProjectionClient = projectionClient; - this.PersistentSubscriptionsClient = persistentSubscriptionsClient; + this.CatchupServiceClient = catchupServiceClient; } private static PersistentSubscriptionSettings CreatePersistentSettings(Int32 retryCount = 0) => new PersistentSubscriptionSettings(resolveLinkTos: true, maxRetryCount: retryCount, startFrom:new StreamPosition(0)); public async Task SetupEventStore(CancellationToken cancellationToken) { - await this.DeployProjections(cancellationToken); - await this.SetupSubscriptions(cancellationToken); + Result projectionResult = await this.DeployProjections(cancellationToken); + if (projectionResult.IsFailed) + return ResultHelpers.CreateFailure(projectionResult); + + Result indexResult = await this.CreateIndexes(cancellationToken); + if (indexResult.IsFailed) + return ResultHelpers.CreateFailure(indexResult); + Result subscriptionsResult = await this.SetupSubscriptions(cancellationToken); + if (subscriptionsResult.IsFailed) + return ResultHelpers.CreateFailure(subscriptionsResult); + return Result.Success(); } - private async Task SetupSubscriptions(CancellationToken cancellationToken){ - List<(String streamName, String groupName, Int32 retryCount)> subscriptions = [ - ("$ce-TransactionAggregate", TransactionProcessorGroup, 0), - ("$ce-SettlementAggregate", TransactionProcessorGroup, 0), - ("$ce-VoucherAggregate", TransactionProcessorGroup, 0), - ("$ce-FloatAggregate", TransactionProcessorGroup, 0), - ("$ce-MerchantStatementAggregate", TransactionProcessorGroup, 0), - ("$ce-ContractAggregate", TransactionProcessorGroup, 0), - ("$ce-EstateAggregate", TransactionProcessorGroup, 0), - ("$ce-MerchantAggregate", TransactionProcessorGroup, 0), - ("$ce-CallbackMessageAggregate", TransactionProcessorGroup, 0), - ("$ce-ReconciliationAggregate", TransactionProcessorGroup, 0), - ("$ce-FileAggregate", TransactionProcessorGroup, 0), - ("$ce-FileImportLogAggregate", TransactionProcessorGroup, 0), - ("$ce-OperatorAggregate", TransactionProcessorGroup, 0), - ("$ce-MerchantBalanceArchive", TransactionProcessorGroup, 0), - - ("$ce-TransactionAggregate", TransactionProcessorDomainGroup, 0), - ("$ce-SettlementAggregate", TransactionProcessorDomainGroup, 0), - ("$ce-FloatAggregate", TransactionProcessorDomainGroup, 0), - ("$ce-MerchantStatementForDateAggregate", TransactionProcessorDomainGroup, 0), - - ("$ce-EstateAggregate", TransactionProcessorOrderedGroup, 1), - ("$ce-SettlementAggregate", TransactionProcessorOrderedGroup, 1), - ("$ce-VoucherAggregate", TransactionProcessorOrderedGroup, 1), - ("$ce-TransactionAggregate", TransactionProcessorOrderedGroup, 0), - ("$ce-MerchantStatementAggregate", TransactionProcessorOrderedGroup, 0), - ("$ce-EstateAggregate", TransactionProcessorOrderedGroup, 0), - - ("$ce-FileAggregate", FileProcessorGroup, 0), - ("$ce-FileImportLogAggregate", FileProcessorGroup, 0), - - ("$ce-EmailAggregate", MessagingServiceGroup, 0), - ("$ce-SMSAggregate", MessagingServiceGroup, 0) - ]; + private async Task CreateIndexes(CancellationToken cancellationToken) { - foreach ((String streamName, String groupName, Int32 retryCount) subscription in subscriptions){ - Boolean exists = false; - try{ - await this.PersistentSubscriptionsClient.GetInfoToStreamAsync(subscription.streamName, subscription.groupName, cancellationToken: cancellationToken, deadline: TimeSpan.FromSeconds(30)); - exists = true; - } - catch(PersistentSubscriptionNotFoundException){ - exists = false; + // Now create any secondary indexes + String merchantBalanceArchiveIndexName = "merchant-balance-archive-events"; + String merchantBalanceArchiveIndexFilter = "{\r\n \"filter\": \"rec => rec.value != null && rec.value.merchantId != null && ['MerchantCreatedEvent', 'ManualDepositMadeEvent', 'AutomaticDepositMadeEvent', 'TransactionHasStartedEvent', 'TransactionHasBeenCompletedEvent', 'SettledMerchantFeeAddedToTransactionEvent', 'WithdrawalMadeEvent'].includes(rec.schema.name)\",\r\n \"fields\": [\r\n {\r\n \"name\": \"merchantid\",\r\n \"selector\": \"rec => rec.value.merchantId\",\r\n \"type\": \"INDEX_FIELD_TYPE_STRING\"\r\n }\r\n ],\r\n \"start\": true\r\n}"; + + CatchupApiResponse indexDetails = await this.CatchupServiceClient.GetIndexAsync(merchantBalanceArchiveIndexName, cancellationToken); + + if (indexDetails.StatusCode == HttpStatusCode.NotFound){ + await this.CatchupServiceClient.CreateIndexAsync(merchantBalanceArchiveIndexName, merchantBalanceArchiveIndexFilter, cancellationToken); + } + + return Result.Success(); + } + + //private async Task SetupSubscriptions(CancellationToken cancellationToken){ + // List<(String streamName, String groupName, Int32 retryCount)> subscriptions = [ + // ("$ce-TransactionAggregate", TransactionProcessorGroup, 0), + // ("$ce-SettlementAggregate", TransactionProcessorGroup, 0), + // ("$ce-VoucherAggregate", TransactionProcessorGroup, 0), + // ("$ce-FloatAggregate", TransactionProcessorGroup, 0), + // ("$ce-MerchantStatementAggregate", TransactionProcessorGroup, 0), + // ("$ce-ContractAggregate", TransactionProcessorGroup, 0), + // ("$ce-EstateAggregate", TransactionProcessorGroup, 0), + // ("$ce-MerchantAggregate", TransactionProcessorGroup, 0), + // ("$ce-CallbackMessageAggregate", TransactionProcessorGroup, 0), + // ("$ce-ReconciliationAggregate", TransactionProcessorGroup, 0), + // ("$ce-FileAggregate", TransactionProcessorGroup, 0), + // ("$ce-FileImportLogAggregate", TransactionProcessorGroup, 0), + // ("$ce-OperatorAggregate", TransactionProcessorGroup, 0), + // ("$ce-MerchantBalanceArchive", TransactionProcessorGroup, 0), + + // ("$ce-TransactionAggregate", TransactionProcessorDomainGroup, 0), + // ("$ce-SettlementAggregate", TransactionProcessorDomainGroup, 0), + // ("$ce-FloatAggregate", TransactionProcessorDomainGroup, 0), + // ("$ce-MerchantStatementForDateAggregate", TransactionProcessorDomainGroup, 0), + + // ("$ce-EstateAggregate", TransactionProcessorOrderedGroup, 1), + // ("$ce-SettlementAggregate", TransactionProcessorOrderedGroup, 1), + // ("$ce-VoucherAggregate", TransactionProcessorOrderedGroup, 1), + // ("$ce-TransactionAggregate", TransactionProcessorOrderedGroup, 0), + // ("$ce-MerchantStatementAggregate", TransactionProcessorOrderedGroup, 0), + // ("$ce-EstateAggregate", TransactionProcessorOrderedGroup, 0), + + // ("$ce-FileAggregate", FileProcessorGroup, 0), + // ("$ce-FileImportLogAggregate", FileProcessorGroup, 0), + + // ("$ce-EmailAggregate", MessagingServiceGroup, 0), + // ("$ce-SMSAggregate", MessagingServiceGroup, 0) + // ]; + + // foreach ((String streamName, String groupName, Int32 retryCount) subscription in subscriptions){ + // Boolean exists = false; + // try{ + // await this.PersistentSubscriptionsClient.GetInfoToStreamAsync(subscription.streamName, subscription.groupName, cancellationToken: cancellationToken, deadline: TimeSpan.FromSeconds(30)); + // exists = true; + // } + // catch(PersistentSubscriptionNotFoundException){ + // exists = false; + // } + + // if (exists == false){ + // await this.PersistentSubscriptionsClient.CreateToStreamAsync(subscription.streamName, subscription.groupName, CreatePersistentSettings(subscription.retryCount), cancellationToken: cancellationToken, deadline: TimeSpan.FromSeconds(30)); + // } + // } + + // return Result.Success(); + //} + + private async Task SetupSubscriptions(CancellationToken cancellationToken) + { + // First Step is to setup the endpoints + EndpointDefinition transactionProcessorEndpoint = new EndpointDefinition( "Transaction Processor", "http://localhost:5002/api/domainevents"); + EndpointDefinition fileProcessorEndpoint = new EndpointDefinition( "File Processor", "http://localhost:5009/api/domainevents"); + EndpointDefinition messagingServiceEndpoint = new EndpointDefinition( "Messaging Service", "http://localhost:5006/api/domainevents"); + + List endpointDefinitions = [transactionProcessorEndpoint, fileProcessorEndpoint, messagingServiceEndpoint]; + + CatchupApiResponse> existingEndpoints = await this.CatchupServiceClient.GetEndpointsAsync(cancellationToken); + + foreach (EndpointDefinition endpointDefinition in endpointDefinitions) { + if (existingEndpoints.Value.SingleOrDefault(x => x.Name == endpointDefinition.Name) == null) + { + var result = await this.CatchupServiceClient.CreateEndpointAsync(endpointDefinition, cancellationToken); + if (result.IsSuccessStatusCode == false) + return Result.Failure($"Error creating endpoint name {endpointDefinition.Name} Status Code = {result.StatusCode}"); } + } + + // Refresh the list of endpoints now + existingEndpoints = await this.CatchupServiceClient.GetEndpointsAsync(cancellationToken); + + // Now create any secondary indexes + String merchantBalanceArchiveIndexName = "merchant-balance-archive-events"; + + List<(String IndexName, String endpointName, String tag)> subscriptions = [ + ("$idx-ce-EstateAggregate", "Transaction Processor", "Main"), + ("$idx-ce-TransactionAggregate", "Transaction Processor", "Main"), + ("$idx-ce-SettlementAggregate", "Transaction Processor", "Main"), + ("$idx-ce-VoucherAggregate", "Transaction Processor", "Main"), + ("$idx-ce-FloatAggregate", "Transaction Processor", "Main"), + ("$idx-ce-MerchantStatementAggregate", "Transaction Processor", "Main"), + ("$idx-ce-SettlementAggregate", "Transaction Processor", "Main"), + ("$idx-ce-ContractAggregate", "Transaction Processor", "Main"), + ("$idx-ce-MerchantAggregate", "Transaction Processor", "Main"), + ("$idx-ce-CallbackMessageAggregate", "Transaction Processor", "Main"), + ("$idx-ce-ReconciliationAggregate", "Transaction Processor", "Main"), + ("$idx-ce-FileAggregate", "Transaction Processor", "Main"), + ("$idx-ce-FileImportLogAggregate", "Transaction Processor", "Main"), + ("$idx-ce-OperatorAggregate", "Transaction Processor", "Main"), + + ("$idx-ce-TransactionAggregate", "Transaction Processor", "Domain"), + ("$idx-ce-SettlementAggregate", "Transaction Processor", "Domain"), + ("$idx-ce-FloatAggregate", "Transaction Processor", "Domain"), + + ("$idx-ce-EstateAggregate", "Transaction Processor", "Ordered"), + ("$idx-ce-SettlementAggregate", "Transaction Processor", "Ordered"), + ("$idx-ce-VoucherAggregate", "Transaction Processor", "Ordered"), + ("$idx-ce-TransactionAggregate", "Transaction Processor", "Ordered"), + ("$idx-ce-MerchantStatementAggregate", "Transaction Processor", "Ordered"), + ($"$idx-user-{merchantBalanceArchiveIndexName}", "Transaction Processor", "Ordered"), + + ("$idx-ce-FileAggregate", "File Processor", "Main"), + ("$idx-ce-FileImportLogAggregate", "File Processor", "Main"), + + ("$idx-ce-EmailAggregate", "Messaging Service", "Main"), + ("$idx-ce-SMSAggregate", "Messaging Service", "Main"), + ]; - if (exists == false){ - await this.PersistentSubscriptionsClient.CreateToStreamAsync(subscription.streamName, subscription.groupName, CreatePersistentSettings(subscription.retryCount), cancellationToken: cancellationToken, deadline: TimeSpan.FromSeconds(30)); + foreach ((String IndexName, String endpointName, String tag) subscription in subscriptions) + { + var endpoint = existingEndpoints.Value.SingleOrDefault(x => x.Name == subscription.endpointName); + if (endpoint == null) + return Result.Failure($"Endpoint name {subscription.endpointName} not found"); + // build the subscription id + String subscriptionId = $"{subscription.IndexName}_{subscription.endpointName}_{subscription.tag}"; + + CatchupApiResponse> subscriptionConfigList = await this.CatchupServiceClient.GetSubscriptionConfigurationsAsync(cancellationToken); + + if (subscriptionConfigList.Value.SingleOrDefault(x => x.SubscriptionId == subscriptionId) == null) { + SubscriptionConfigurationRequest subscriptionConfigurationRequest = new(); + subscriptionConfigurationRequest = subscriptionConfigurationRequest with { + ContinueOnParked = true, + CheckpointBatchSize = 100, + EndpointUrl = endpoint.Url, + RetryDelaySeconds = 5, + RetryMaxAttempts = 0, + SecondaryIndexName = subscription.IndexName, + SoftDeleteParked = true, + SubscriptionId = subscriptionId, + Tag = subscription.tag, + TimeoutSeconds = 30, + }; + + var result = await this.CatchupServiceClient.CreateSubscriptionConfigurationAsync(subscriptionConfigurationRequest, cancellationToken); + if (result.IsSuccessStatusCode == false) + return Result.Failure($"Error subscription configuration id {subscriptionConfigurationRequest.SubscriptionId} Status Code = {result.StatusCode}"); } } return Result.Success(); } + private async Task DeployProjections(CancellationToken cancellationToken) { IAsyncEnumerable currentProjectionsList = this.ProjectionClient.ListAllAsync(cancellationToken: cancellationToken); var currentProjections = new List(); @@ -107,6 +228,7 @@ private async Task DeployProjections(CancellationToken cancellationToken projection.Contains("TransactionProcessorSubscriptionStreamBuilder", StringComparison.Ordinal) || projection.Contains("EstateAggregator", StringComparison.Ordinal) || projection.Contains("MerchantAggregator", StringComparison.Ordinal) || + projection.Contains("MerchantBalanceAggregator", StringComparison.Ordinal) || projection.Contains("CallbackHandlerEnricher", StringComparison.Ordinal)) { continue; diff --git a/TransactionProcessor.SystemSetupTool/Program.cs b/TransactionProcessor.SystemSetupTool/Program.cs index 4b23259..80e8204 100644 --- a/TransactionProcessor.SystemSetupTool/Program.cs +++ b/TransactionProcessor.SystemSetupTool/Program.cs @@ -32,6 +32,7 @@ class Program private static KurrentDBProjectionManagementClient ProjectionClient; private static KurrentDBPersistentSubscriptionsClient PersistentSubscriptionsClient; + private static ICatchupServiceClient CatchupServiceClient; static async Task Main(string[] args) { @@ -47,7 +48,17 @@ static async Task Main(string[] args) { Func securityResolver = s => { return ConfigurationReader.GetValue("SecurityServiceUri"); }; Func transactionProcessorResolver = s => { return ConfigurationReader.GetValue("TransactionProcessorApi"); }; Func fileProcessorResolver = s => { return ConfigurationReader.GetValue("FileProcessorApi"); }; - HttpClientHandler handler = new(); + Func catchupServiceResolver = s => { return ConfigurationReader.GetValue("CatchupService"); }; + HttpClientHandler handler = new() + { + ServerCertificateCustomValidationCallback = (message, + cert, + chain, + errors) => + { + return true; + } + }; HttpClient client = new(handler); Program.SecurityServiceClient = new SecurityServiceClient(securityResolver, client, Serialise, Deserialise); @@ -58,14 +69,16 @@ static async Task Main(string[] args) { Program.ProjectionClient = new (settings); Program.PersistentSubscriptionsClient = new (settings); + CatchupServiceClient = new CatchupServiceClient(catchupServiceResolver, client); + Mode setupMode = Mode.EstateSetup; - String configFileName = "setupconfig.staging.json"; + String configFileName = "setupconfig.json"; FileProcessingOptions fileProcessingOptions = await Program.GetFileProfileConfig(cancellationToken); IdentityServerConfiguration identityServerConfiguration = await Program.GetIdentityServerConfig(cancellationToken); IdentityServerFunctions identityServerFunctions = new(Program.SecurityServiceClient, identityServerConfiguration); - EventStoreFunctions eventStoreFunctions = new(Program.ProjectionClient, Program.PersistentSubscriptionsClient); + EventStoreFunctions eventStoreFunctions = new(Program.ProjectionClient, CatchupServiceClient); Result result = setupMode switch { Mode.SecuritySetup => await identityServerFunctions.CreateConfig(cancellationToken), diff --git a/TransactionProcessor.SystemSetupTool/appsettings.json b/TransactionProcessor.SystemSetupTool/appsettings.json index fde9f02..aaa0e61 100644 --- a/TransactionProcessor.SystemSetupTool/appsettings.json +++ b/TransactionProcessor.SystemSetupTool/appsettings.json @@ -1,16 +1,17 @@ { "AppSettings": { // Local (Docker) - //"SecurityServiceUri": "https://127.0.0.1:5001", - //"TransactionProcessorApi": "http://127.0.0.1:5002", - //"FileProcessorApi": "http://127.0.0.1:5009", - //"EventStoreAddress": "esdb://admin:changeit@127.0.0.1:4113?tls=false&tlsVerifyCert=false" + "SecurityServiceUri": "https://127.0.0.1:5001", + "TransactionProcessorApi": "http://127.0.0.1:5002", + "FileProcessorApi": "http://127.0.0.1:5009", + "CatchupService": "http://127.0.0.1:8080", + "EventStoreAddress": "esdb://admin:changeit@127.0.0.1:2113?tls=false&tlsVerifyCert=false" // Staging - "SecurityServiceUri": "https://192.168.1.163:5001", - "TransactionProcessorApi": "http://192.168.1.163:5002", - "FileProcessorApi": "http://192.168.1.163:5009", - "EventStoreAddress": "esdb://admin:changeit@192.168.1.163:2113?tls=false&tlsVerifyCert=false" + //"SecurityServiceUri": "https://192.168.1.163:5001", + //"TransactionProcessorApi": "http://192.168.1.163:5002", + //"FileProcessorApi": "http://192.168.1.163:5009", + //"EventStoreAddress": "esdb://admin:changeit@192.168.1.163:2113?tls=false&tlsVerifyCert=false" // Production //"SecurityServiceUri": "https://192.168.1.155:5001", diff --git a/TransactionProcessor.SystemSetupTool/identityserverconfig.json b/TransactionProcessor.SystemSetupTool/identityserverconfig.json index c9ddba8..eb5e438 100644 --- a/TransactionProcessor.SystemSetupTool/identityserverconfig.json +++ b/TransactionProcessor.SystemSetupTool/identityserverconfig.json @@ -110,13 +110,11 @@ ], "client_redirect_uris": [ "https://localhost:5004/signin-oidc", - "https://192.168.1.86:5004/signin-oidc", - "https://192.168.1.167:5004/signin-oidc" + "https://192.168.1.86:5004/signin-oidc" ], "client_post_logout_redirect_uris": [ "https://localhost:5004/signout-oidc", - "https://192.168.1.86:5004/signout-oidc", - "https://192.168.1.167:5004/signout-oidc" + "https://192.168.1.86:5004/signout-oidc" ], "require_consent": true, "allow_offline_access": true