diff --git a/EventSourcing.NetCore.sln b/EventSourcing.NetCore.sln
index 1bd6322c0..302298d2b 100644
--- a/EventSourcing.NetCore.sln
+++ b/EventSourcing.NetCore.sln
@@ -477,6 +477,20 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "SwaggerUi.Aspire.Hosting",
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Core.Scheduling", "Core.Scheduling\Core.Scheduling.csproj", "{3B754142-CDA4-430F-8DF7-5DC0D204EE35}"
EndProject
+Project("{F2A71F9B-5D33-465A-A702-920D77279786}") = "ECommerce.Domain", "Sample\ECommerce.Equinox\ECommerce.Domain\ECommerce.Domain.fsproj", "{7C22079D-C359-40DE-8E5F-1DDD9CDD06DD}"
+EndProject
+Project("{F2A71F9B-5D33-465A-A702-920D77279786}") = "ECommerce.Reactor", "Sample\ECommerce.Equinox\ECommerce.Reactor\ECommerce.Reactor.fsproj", "{C3D0553F-5786-4417-97FD-B65440620274}"
+EndProject
+Project("{F2A71F9B-5D33-465A-A702-920D77279786}") = "ECommerce.Api", "Sample\ECommerce.Equinox\ECommerce.Api\ECommerce.Api.fsproj", "{1F43CF34-CD69-4EEB-BF64-5AD649CE3AF5}"
+EndProject
+Project("{F2A71F9B-5D33-465A-A702-920D77279786}") = "ECommerce.Infrastructure", "Sample\ECommerce.Equinox\ECommerce.Infrastructure\ECommerce.Infrastructure.fsproj", "{1C310D3A-BFC5-4984-BB11-4579067CFFB5}"
+EndProject
+Project("{F2A71F9B-5D33-465A-A702-920D77279786}") = "ECommerce.Tests", "Sample\ECommerce.Equinox\ECommerce.Tests\ECommerce.Tests.fsproj", "{16B7A55B-E2E6-4CE2-896A-EF0F02259482}"
+EndProject
+Project("{F2A71F9B-5D33-465A-A702-920D77279786}") = "ECommerce.FeedConsumer", "Sample\ECommerce.Equinox\ECommerce.FeedConsumer\ECommerce.FeedConsumer.fsproj", "{F2D522B1-BDF0-4345-919D-42DD79D4BB24}"
+EndProject
+Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "ECommerce.Equinox", "ECommerce.Equinox", "{006643C6-E0B6-48E6-ABC6-9BE3DCB293D8}"
+EndProject
Global
GlobalSection(SolutionConfigurationPlatforms) = preSolution
Debug|Any CPU = Debug|Any CPU
@@ -1079,6 +1093,18 @@ Global
{3B754142-CDA4-430F-8DF7-5DC0D204EE35}.Debug|Any CPU.Build.0 = Debug|Any CPU
{3B754142-CDA4-430F-8DF7-5DC0D204EE35}.Release|Any CPU.ActiveCfg = Release|Any CPU
{3B754142-CDA4-430F-8DF7-5DC0D204EE35}.Release|Any CPU.Build.0 = Release|Any CPU
+ {1F43CF34-CD69-4EEB-BF64-5AD649CE3AF5}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+ {1F43CF34-CD69-4EEB-BF64-5AD649CE3AF5}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {7C22079D-C359-40DE-8E5F-1DDD9CDD06DD}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+ {7C22079D-C359-40DE-8E5F-1DDD9CDD06DD}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {F2D522B1-BDF0-4345-919D-42DD79D4BB24}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+ {F2D522B1-BDF0-4345-919D-42DD79D4BB24}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {1C310D3A-BFC5-4984-BB11-4579067CFFB5}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+ {1C310D3A-BFC5-4984-BB11-4579067CFFB5}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {C3D0553F-5786-4417-97FD-B65440620274}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+ {C3D0553F-5786-4417-97FD-B65440620274}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {16B7A55B-E2E6-4CE2-896A-EF0F02259482}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+ {16B7A55B-E2E6-4CE2-896A-EF0F02259482}.Debug|Any CPU.Build.0 = Debug|Any CPU
EndGlobalSection
GlobalSection(SolutionProperties) = preSolution
HideSolutionNode = FALSE
@@ -1276,6 +1302,13 @@ Global
{D4CD0C59-65CC-4051-A77E-2C944F3C27C0} = {A8E25331-55E9-4D1A-87A4-136EC4D2A4B5}
{CCFE487A-A5C5-4DAC-8815-AC486B9625BC} = {0570E45A-2EB6-4C4C-84E4-2C80E1FECEB5}
{3B754142-CDA4-430F-8DF7-5DC0D204EE35} = {0570E45A-2EB6-4C4C-84E4-2C80E1FECEB5}
+ {006643C6-E0B6-48E6-ABC6-9BE3DCB293D8} = {A7186B6B-D56D-4AEF-B6B7-FAA827764C34}
+ {1F43CF34-CD69-4EEB-BF64-5AD649CE3AF5} = {006643C6-E0B6-48E6-ABC6-9BE3DCB293D8}
+ {7C22079D-C359-40DE-8E5F-1DDD9CDD06DD} = {006643C6-E0B6-48E6-ABC6-9BE3DCB293D8}
+ {F2D522B1-BDF0-4345-919D-42DD79D4BB24} = {006643C6-E0B6-48E6-ABC6-9BE3DCB293D8}
+ {1C310D3A-BFC5-4984-BB11-4579067CFFB5} = {006643C6-E0B6-48E6-ABC6-9BE3DCB293D8}
+ {C3D0553F-5786-4417-97FD-B65440620274} = {006643C6-E0B6-48E6-ABC6-9BE3DCB293D8}
+ {16B7A55B-E2E6-4CE2-896A-EF0F02259482} = {006643C6-E0B6-48E6-ABC6-9BE3DCB293D8}
EndGlobalSection
GlobalSection(ExtensibilityGlobals) = postSolution
SolutionGuid = {A5F55604-2FF3-43B7-B657-4F18E6E95D3B}
diff --git a/Sample/ECommerce.Equinox/Directory.Build.props b/Sample/ECommerce.Equinox/Directory.Build.props
new file mode 100644
index 000000000..398982871
--- /dev/null
+++ b/Sample/ECommerce.Equinox/Directory.Build.props
@@ -0,0 +1,6 @@
+
+
+
+ true
+
+
diff --git a/Sample/ECommerce.Equinox/ECommerce.Api/Controllers/ConfirmedFeedController.fs b/Sample/ECommerce.Equinox/ECommerce.Api/Controllers/ConfirmedFeedController.fs
new file mode 100644
index 000000000..6fc9012c4
--- /dev/null
+++ b/Sample/ECommerce.Equinox/ECommerce.Api/Controllers/ConfirmedFeedController.fs
@@ -0,0 +1,59 @@
+namespace ECommerce.Api.Controllers
+
+open Microsoft.AspNetCore.Mvc
+
+open ECommerce.Domain
+
+type TranchesDto = { activeEpochs : TrancheReferenceDto[] }
+ and TrancheReferenceDto = { seriesId : ConfirmedSeriesId; epochId : ConfirmedEpochId }
+
+module TranchesDto =
+
+ let ofEpochId epochId =
+ { activeEpochs = [| { seriesId = ConfirmedSeriesId.wellKnownId; epochId = epochId } |]}
+
+type SliceDto = { closed : bool; carts : CartDto[]; position : ConfirmedCheckpoint; checkpoint : ConfirmedCheckpoint }
+ and CartDto = { id : CartId; items : ItemDto[] }
+ and ItemDto = { productId : ProductId; unitPrice : decimal; quantity : int }
+
+module CartDto =
+
+ let ofDto (x : ConfirmedEpoch.Events.Cart) : CartDto =
+ { id = x.cartId
+ items = [| for x in x.items -> { productId = x.productId; unitPrice = x.unitPrice; quantity = x.quantity } |] }
+
+module Checkpoint =
+
+ let ofEpochAndOffset (epoch : ConfirmedEpochId) (offset : int) =
+ ConfirmedCheckpoint.ofEpochAndOffset epoch offset
+
+ let ofState (epochId : ConfirmedEpochId) (s : ConfirmedEpoch.Reader.StateDto) =
+ ConfirmedCheckpoint.ofEpochContent epochId s.closed s.carts.Length
+
+[]
+type ConfirmedFeedController(series : ConfirmedSeries.Service, epochs : ConfirmedEpoch.Reader.Service) =
+ inherit ControllerBase()
+
+ []
+ member _.ListTranches() : Async = async {
+ let! active = series.ReadIngestionEpochId()
+ return TranchesDto.ofEpochId active
+ }
+
+ []
+ member _.ReadTranche(epoch : ConfirmedEpochId) : Async = async {
+ let! state = epochs.Read(epoch)
+ // TOCONSIDER closed should control cache header
+ let pos, checkpoint = Checkpoint.ofEpochAndOffset epoch 0, Checkpoint.ofState epoch state
+ return { closed = state.closed; carts = Array.map CartDto.ofDto state.carts; position = pos; checkpoint = checkpoint }
+ }
+
+ []
+ member _.Poll(token : System.Nullable) : Async = async {
+ let pos = if token.HasValue then token.Value else ConfirmedCheckpoint.initial
+ let epochId, offset = ConfirmedCheckpoint.toEpochAndOffset pos
+ let! state = epochs.Read(epochId)
+ // TOCONSIDER closed should control cache header
+ let pos, checkpoint = Checkpoint.ofEpochAndOffset epochId offset, Checkpoint.ofState epochId state
+ return { closed = state.closed; carts = Array.skip offset state.carts |> Array.map CartDto.ofDto; position = pos; checkpoint = checkpoint }
+ }
diff --git a/Sample/ECommerce.Equinox/ECommerce.Api/Controllers/ShoppingCartsController.fs b/Sample/ECommerce.Equinox/ECommerce.Api/Controllers/ShoppingCartsController.fs
new file mode 100644
index 000000000..2fecb7de0
--- /dev/null
+++ b/Sample/ECommerce.Equinox/ECommerce.Api/Controllers/ShoppingCartsController.fs
@@ -0,0 +1,85 @@
+namespace ECommerce.Api.Controllers
+
+open ECommerce.Domain
+open Microsoft.AspNetCore.Mvc
+open System
+
+type InitializeShoppingCartRequest = { clientId : Guid Nullable }
+type AddProductRequest = { productId : Guid; quantity : int }
+type RemoveProductRequest = { productId : Guid; price : decimal }
+
+[]
+[]
+type ShoppingCartsController(carts : ShoppingCart.Service, cartsDenormalized : ShoppingCartSummary.Service) =
+ inherit ControllerBase()
+
+ []
+ member _.InitializeCart([] request : InitializeShoppingCartRequest) : Async = async {
+ if obj.ReferenceEquals(null, request) then nameof request |> nullArg
+
+ // TODO in these samples in general, the semantics should be extended to be more representative of the real world
+ // - you don't want to create orphan carts esp if they need to retry this API call
+ // - you may want to allow users to shop before logging in, keeping the clientId in a cookie.
+ // After some time, you let them log in, but then you need to merge the content into their real cart
+ // For now, the code remains in line with the C# version
+ let cartId = CartId.generate();
+ do! carts.Initialize(cartId, ClientId.parse request.clientId)
+ return CreatedResult("api/ShoppingCarts", cartId) :> _
+ }
+
+ []
+ member _.AddProduct([] id : Guid Nullable, [] request : AddProductRequest) : Async = async {
+ if obj.ReferenceEquals(null, request) then nameof request |> nullArg
+
+ let CartId.ParseGuid cartId, ProductId.Parse productId = id, request.productId
+ do! carts.Add(cartId, productId, request.quantity)
+ return OkResult() :> _
+ }
+
+ []
+ member _.RemoveProduct([] id : Guid Nullable, [] request : RemoveProductRequest) : Async = async {
+ if obj.ReferenceEquals(null, request) then nameof request |> nullArg
+
+ let CartId.ParseGuid cartId, ProductId.Parse productId = id, request.productId
+ do! carts.Remove(cartId, productId, request.price)
+ return OkResult() :> _
+ }
+
+ []
+ member _.ConfirmCart([] id : Guid Nullable(*, [] request : ConfirmShoppingCartRequest*)) : Async = async {
+// if obj.ReferenceEquals(null, request) then nameof request |> nullArg // TODO only relevant if we follow version-contingent style
+
+ let (CartId.ParseGuid cartId) = id
+ do! carts.Confirm(cartId, DateTimeOffset.UtcNow)
+ return OkResult() :> _
+ }
+
+ /// Reads from write side
+ []
+ member _.Get([] id : Guid Nullable) : Async = async {
+ let (CartId.ParseGuid cartId) = id
+ match! carts.Read cartId with
+ | Some (res : ShoppingCart.Details.View) -> return OkObjectResult res :> _
+ | None -> return NotFoundResult() :> _
+ }
+
+ /// Reads from denormalized view
+ []
+ member _.GetSummary([] id : Guid Nullable) : Async = async {
+ let (CartId.ParseGuid cartId) = id
+ match! cartsDenormalized.Read cartId with
+ | Some (res : ShoppingCartSummary.Details.View) -> return OkObjectResult res :> _
+ | None -> return NotFoundResult() :> _
+ }
+
+ (* TODO we dont produce a list like this atm - not porting for the moment as having an arbitrarily growing list like this does not really make
+ sense; instead, we'll produce a cart summaries API
+ [HttpGet]
+ public Task> Get(
+ [FromServices] Func>> query,
+ CancellationToken ct,
+ [FromQuery] int pageNumber = 1,
+ [FromQuery] int pageSize = 20
+ ) =>
+ query(GetCarts.From(pageNumber, pageSize), ct);
+*)
diff --git a/Sample/ECommerce.Equinox/ECommerce.Api/ECommerce.Api.fsproj b/Sample/ECommerce.Equinox/ECommerce.Api/ECommerce.Api.fsproj
new file mode 100644
index 000000000..37076391f
--- /dev/null
+++ b/Sample/ECommerce.Equinox/ECommerce.Api/ECommerce.Api.fsproj
@@ -0,0 +1,28 @@
+
+
+
+ net6.0
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ <_ContentIncludedByDefault Remove="Properties\launchSettings.json" />
+
+
+
diff --git a/Sample/ECommerce.Equinox/ECommerce.Api/Program.fs b/Sample/ECommerce.Equinox/ECommerce.Api/Program.fs
new file mode 100644
index 000000000..4f22e1837
--- /dev/null
+++ b/Sample/ECommerce.Equinox/ECommerce.Api/Program.fs
@@ -0,0 +1,82 @@
+module ECommerce.Api.Program
+
+open ECommerce
+open Microsoft.AspNetCore.Hosting
+open Microsoft.Extensions.DependencyInjection
+open Serilog
+open System
+
+type Configuration(tryGet) =
+ inherit Args.Configuration(tryGet)
+
+let [] AppName = "ECommerce.Web"
+
+module Args =
+
+ open Argu
+
+ type [] Parameters =
+ | [] Verbose
+ | [] PrometheusPort of int
+ | [] Cosmos of ParseResults
+ | [] Dynamo of ParseResults
+ | [] Esdb of ParseResults
+ | [] Sss of ParseResults
+ interface IArgParserTemplate with
+ member a.Usage = a |> function
+ | Verbose -> "request verbose logging."
+ | PrometheusPort _ -> "port from which to expose a Prometheus /metrics endpoint. Default: off (optional if environment variable PROMETHEUS_PORT specified)"
+ | Cosmos _ -> "specify CosmosDB input parameters"
+ | Dynamo _ -> "specify DynamoDB input parameters"
+ | Esdb _ -> "specify EventStore input parameters"
+ | Sss _ -> "specify SqlStreamStore input parameters"
+ and []
+ Arguments(c : Configuration, p : ParseResults) =
+ member val Verbose = p.Contains Verbose
+ member val PrometheusPort = p.TryGetResult PrometheusPort |> Option.orElseWith (fun () -> c.PrometheusPort)
+ member val CacheSizeMb = 10
+ member val StoreArgs : Args.StoreArgs =
+ match p.TryGetSubCommand() with
+ | Some (Parameters.Cosmos cosmos) -> Args.StoreArgs.Cosmos (Args.Cosmos.Arguments(c, cosmos))
+ | Some (Parameters.Dynamo dynamo) -> Args.StoreArgs.Dynamo (Args.Dynamo.Arguments(c, dynamo))
+ | Some (Parameters.Esdb es) -> Args.StoreArgs.Esdb (Args.Esdb.Arguments(c, es))
+ | Some (Parameters.Sss sss) -> Args.StoreArgs.Sss (Args.Sss.Arguments(c, sss))
+ | _ -> p.Raise "Must specify one of cosmos, dynamo, esdb or sss for store"
+ member x.VerboseStore = Args.StoreArgs.verboseRequested x.StoreArgs
+ member x.Connect(): Store.Config =
+ let cache = Equinox.Cache (AppName, sizeMb = x.CacheSizeMb)
+ Args.StoreArgs.connectTarget x.StoreArgs cache
+
+ /// Parse the commandline; can throw exceptions in response to missing arguments and/or `-h`/`--help` args
+ let parse tryGetConfigValue argv =
+ let programName = Reflection.Assembly.GetEntryAssembly().GetName().Name
+ let parser = ArgumentParser.Create(programName = programName)
+ Arguments(Configuration tryGetConfigValue, parser.ParseCommandLine argv)
+
+let run (args : Args.Arguments) =
+ let store = args.Connect()
+ let carts = Domain.ShoppingCart.Config.create store
+ let registerServices (services: IServiceCollection) =
+ services.AddSingleton(carts) |> ignore
+ WebHostBuilder()
+ .UseKestrel()
+ .UseSerilog()
+ .ConfigureServices(registerServices)
+ .UseStartup()
+ .Build()
+ .Run()
+
+[]
+let main argv =
+ try let args = Args.parse EnvVar.tryGet argv
+ let metrics = Sinks.tags AppName |> Sinks.equinoxMetricsOnly
+ try Log.Logger <- LoggerConfiguration()
+ .Configure(args.Verbose)
+ .MinimumLevel.Override("Microsoft.AspNetCore", Serilog.Events.LogEventLevel.Warning)
+ .Sinks(metrics, args.VerboseStore)
+ .CreateLogger()
+ try run args; 0
+ with e -> Log.Fatal(e, "Exiting"); 2
+ finally Log.CloseAndFlush()
+ with:? Argu.ArguParseException as e -> eprintfn $"%s{e.Message}"; 1
+ | e -> eprintfn $"Exception %s{e.Message}"; 1
diff --git a/Sample/ECommerce.Equinox/ECommerce.Api/Startup.fs b/Sample/ECommerce.Equinox/ECommerce.Api/Startup.fs
new file mode 100644
index 000000000..718752888
--- /dev/null
+++ b/Sample/ECommerce.Equinox/ECommerce.Api/Startup.fs
@@ -0,0 +1,31 @@
+namespace ECommerce.Api
+
+open Microsoft.AspNetCore.Builder
+open Microsoft.Extensions.DependencyInjection
+open Microsoft.Extensions.Hosting
+open Prometheus
+open Serilog
+
+type Startup() =
+
+ member _.ConfigureServices(services : IServiceCollection) : unit =
+ services.AddMvc() |> ignore
+ services.AddControllers()
+ .AddNewtonsoftJson() |> ignore
+ // TODO AddSwaggerGen
+
+ // This method gets called by the runtime. Use this method to configure the HTTP request pipeline.
+ member _.Configure(app : IApplicationBuilder, env : IHostEnvironment) : unit =
+ if env.IsDevelopment() then
+ app.UseDeveloperExceptionPage() |> ignore
+
+ app
+ .UseRouting()
+ //.UseAuthorization()
+ .UseSerilogRequestLogging() // see https://nblumhardt.com/2019/10/serilog-in-aspnetcore-3/
+ .UseEndpoints(fun endpoints ->
+ endpoints.MapControllers() |> ignore
+ endpoints.MapMetrics() |> ignore)
+ |> ignore
+// app.UseSwagger();
+// app.UseSwaggerUI
diff --git a/Sample/ECommerce.Equinox/ECommerce.Domain/Config.fs b/Sample/ECommerce.Equinox/ECommerce.Domain/Config.fs
new file mode 100644
index 000000000..28aa7b952
--- /dev/null
+++ b/Sample/ECommerce.Equinox/ECommerce.Domain/Config.fs
@@ -0,0 +1,85 @@
+module ECommerce.Domain.Config
+
+let log = Serilog.Log.ForContext("isMetric", true)
+let createDecider category = Equinox.Decider.resolve log category
+
+module Memory =
+
+ let create codec initial fold store : Equinox.Category<_, _, _> =
+ Equinox.MemoryStore.MemoryStoreCategory(store, codec, fold, initial)
+
+module EventCodec =
+
+ open FsCodec.SystemTextJson
+
+ let private defaultOptions = Options.Create(autoTypeSafeEnumToJsonString = true)
+ let genJsonElement<'t when 't :> TypeShape.UnionContract.IUnionContract> =
+ CodecJsonElement.Create<'t>(options = defaultOptions)
+ let gen<'t when 't :> TypeShape.UnionContract.IUnionContract> =
+ Codec.Create<'t>(options = defaultOptions)
+
+let private defaultCacheDuration = System.TimeSpan.FromMinutes 20.
+
+module Cosmos =
+
+ let private createCached codec initial fold accessStrategy (context, cache) =
+ let cacheStrategy = Equinox.CosmosStore.CachingStrategy.SlidingWindow (cache, defaultCacheDuration)
+ Equinox.CosmosStore.CosmosStoreCategory(context, codec, fold, initial, cacheStrategy, accessStrategy)
+
+ let createUnoptimized codec initial fold (context, cache) =
+ let accessStrategy = Equinox.CosmosStore.AccessStrategy.Unoptimized
+ createCached codec initial fold accessStrategy (context, cache)
+
+ let createSnapshotted codec initial fold (isOrigin, toSnapshot) (context, cache) =
+ let accessStrategy = Equinox.CosmosStore.AccessStrategy.Snapshot (isOrigin, toSnapshot)
+ createCached codec initial fold accessStrategy (context, cache)
+
+ let createRollingState codec initial fold toSnapshot (context, cache) =
+ let accessStrategy = Equinox.CosmosStore.AccessStrategy.RollingState toSnapshot
+ createCached codec initial fold accessStrategy (context, cache)
+
+module Dynamo =
+
+ let private createCached codec initial fold accessStrategy (context, cache) =
+ let cacheStrategy = Equinox.DynamoStore.CachingStrategy.SlidingWindow (cache, defaultCacheDuration)
+ Equinox.DynamoStore.DynamoStoreCategory(context, codec |> FsCodec.Deflate.EncodeUncompressed, fold, initial, cacheStrategy, accessStrategy)
+
+ let createUnoptimized codec initial fold (context, cache) =
+ let accessStrategy = Equinox.DynamoStore.AccessStrategy.Unoptimized
+ createCached codec initial fold accessStrategy (context, cache)
+
+ let createSnapshotted codec initial fold (isOrigin, toSnapshot) (context, cache) =
+ let accessStrategy = Equinox.DynamoStore.AccessStrategy.Snapshot (isOrigin, toSnapshot)
+ createCached codec initial fold accessStrategy (context, cache)
+
+ let createRollingState codec initial fold toSnapshot (context, cache) =
+ let accessStrategy = Equinox.DynamoStore.AccessStrategy.RollingState toSnapshot
+ createCached codec initial fold accessStrategy (context, cache)
+
+module Esdb =
+
+ let private createCached codec initial fold accessStrategy (context, cache) =
+ let cacheStrategy = Equinox.EventStoreDb.CachingStrategy.SlidingWindow (cache, defaultCacheDuration)
+ Equinox.EventStoreDb.EventStoreCategory(context, codec, fold, initial, cacheStrategy, ?access = accessStrategy)
+ let createUnoptimized codec initial fold (context, cache) =
+ createCached codec initial fold None (context, cache)
+ let createLatestKnownEvent codec initial fold (context, cache) =
+ createCached codec initial fold (Some Equinox.EventStoreDb.AccessStrategy.LatestKnownEvent) (context, cache)
+
+module Sss =
+
+ let private createCached codec initial fold accessStrategy (context, cache) =
+ let cacheStrategy = Equinox.SqlStreamStore.CachingStrategy.SlidingWindow (cache, defaultCacheDuration)
+ Equinox.SqlStreamStore.SqlStreamStoreCategory(context, codec, fold, initial, cacheStrategy, ?access = accessStrategy)
+ let createUnoptimized codec initial fold (context, cache) =
+ createCached codec initial fold None (context, cache)
+ let createLatestKnownEvent codec initial fold (context, cache) =
+ createCached codec initial fold (Some Equinox.SqlStreamStore.AccessStrategy.LatestKnownEvent) (context, cache)
+
+[]
+type Store<'t> =
+ | Memory of Equinox.MemoryStore.VolatileStore<'t>
+ | Cosmos of Equinox.CosmosStore.CosmosStoreContext * Equinox.Core.ICache
+ | Dynamo of Equinox.DynamoStore.DynamoStoreContext * Equinox.Core.ICache
+ | Esdb of Equinox.EventStoreDb.EventStoreContext * Equinox.Core.ICache
+ | Sss of Equinox.SqlStreamStore.SqlStreamStoreContext * Equinox.Core.ICache
diff --git a/Sample/ECommerce.Equinox/ECommerce.Domain/ConfirmedEpoch.fs b/Sample/ECommerce.Equinox/ECommerce.Domain/ConfirmedEpoch.fs
new file mode 100644
index 000000000..5de89457c
--- /dev/null
+++ b/Sample/ECommerce.Equinox/ECommerce.Domain/ConfirmedEpoch.fs
@@ -0,0 +1,124 @@
+/// Tracks all Confirmed Carts in the system
+/// - Used to walk back through the history of all Carts in the system in approximate order of their processing
+/// - Limited to a certain reasonable count of items; snapshot of Items in an epoch needs to stay a sensible size
+/// The ConfirmedSeries holds a pointer to the current active epoch for each series
+/// Each successive epoch is identified by an index, i.e. ConfirmedEpoch-0_0, then ConfirmedEpoch-0_1
+module ECommerce.Domain.ConfirmedEpoch
+
+let [] CategoryName = "ConfirmedEpoch"
+let streamId epochId = FsCodec.StreamId.gen2 ConfirmedSeriesId.toString ConfirmedEpochId.toString (ConfirmedSeriesId.wellKnownId, epochId)
+
+// NB - these types and the union case names reflect the actual storage formats and hence need to be versioned with care
+[]
+module Events =
+
+ type Ingested = { carts : Cart[] }
+ and Cart = { cartId : CartId; items : Item[] }
+ and Item = { productId : ProductId; unitPrice : decimal; quantity : int }
+ type Event =
+ | Ingested of Ingested
+ | Closed
+ interface TypeShape.UnionContract.IUnionContract
+ let codec = Store.Codec.gen
+ let codecJsonElement = Store.Codec.genJsonElement
+
+let ofShoppingCartView cartId (view : ShoppingCart.Details.View) : Events.Cart =
+ { cartId = cartId; items = [| for i in view.items -> { productId = i.productId; unitPrice = i.unitPrice; quantity = i.quantity } |] }
+
+let itemId (x : Events.Cart) : CartId = x.cartId
+let (|ItemIds|) : Events.Cart[] -> CartId[] = Array.map itemId
+
+module Fold =
+
+ type State = CartId[] * bool
+ let initial = [||], false
+ let evolve (ids, closed) = function
+ | Events.Ingested { carts = ItemIds ingestedIds } -> (Array.append ids ingestedIds, closed)
+ | Events.Closed -> (ids, true)
+
+ let fold : State -> Events.Event seq -> State = Seq.fold evolve
+
+let notAlreadyIn (ids : CartId seq) =
+ let ids = System.Collections.Generic.HashSet ids
+ fun (x : Events.Cart) -> (not << ids.Contains) x.cartId
+
+/// Manages ingestion of only items not already in the list
+/// Yields residual net of items already present in this epoch
+// NOTE See feedSource template for more advanced version handling splitting large input requests where epoch limit is strict
+let decide shouldClose candidates (currentIds, closed as state) : ExactlyOnceIngester.IngestResult<_,_> * Events.Event[] =
+ match closed, candidates |> Array.filter (notAlreadyIn currentIds) with
+ | false, fresh ->
+ let added, events =
+ match fresh with
+ | [||] -> [||], [||]
+ | ItemIds freshIds ->
+ let closing = shouldClose currentIds freshIds
+ let ingestEvent = Events.Ingested { carts = fresh }
+ freshIds, if closing then [| ingestEvent ; Events.Closed |] else [| ingestEvent |]
+ let _, closed = Fold.fold state events
+ { accepted = added; closed = closed; residual = [||] }, events
+ | true, fresh ->
+ { accepted = [||]; closed = true; residual = fresh }, [||]
+
+// NOTE see feedSource for example of separating Service logic into Ingestion and Read Services in order to vary the folding and/or state held
+type Service internal
+ ( shouldClose : CartId[] -> CartId[] -> bool, // let outer layers decide whether ingestion should trigger closing of the batch
+ resolve : ConfirmedEpochId -> Equinox.Decider) =
+
+ /// Ingest the supplied items. Yields relevant elements of the post-state to enable generation of stats
+ /// and facilitate deduplication of incoming items in order to avoid null store round-trips where possible
+ member _.Ingest(epochId, carts) =
+ let decider = resolve epochId
+ // NOTE decider which will initially transact against potentially stale cached state, which will trigger a
+ // resync if another writer has gotten in before us. This is a conscious decision in this instance; the bulk
+ // of writes are presumed to be coming from within this same process
+ decider.Transact(decide shouldClose carts, load = Equinox.LoadOption.AnyCachedValue)
+
+ /// Returns all the items currently held in the stream (Not using AllowStale on the assumption this needs to see updates from other apps)
+ member _.Read epochId : Async =
+ let decider = resolve epochId
+ decider.Query id
+
+module Config =
+
+ let private create_ shouldClose cat = Service(shouldClose, streamId >> Store.createDecider cat)
+ let private (|Category|) = function
+ | Store.Config.Memory store -> Store.Memory.create CategoryName Events.codec Fold.initial Fold.fold store
+ | Store.Config.Cosmos (context, cache) -> Store.Cosmos.createUnoptimized CategoryName Events.codecJsonElement Fold.initial Fold.fold (context, cache)
+ | Store.Config.Dynamo (context, cache) -> Store.Dynamo.createUnoptimized CategoryName Events.codec Fold.initial Fold.fold (context, cache)
+ | Store.Config.Esdb (context, cache) -> Store.Esdb.createUnoptimized CategoryName Events.codec Fold.initial Fold.fold (context, cache)
+ | Store.Config.Sss (context, cache) -> Store.Sss.createUnoptimized CategoryName Events.codec Fold.initial Fold.fold (context, cache)
+ let shouldClose maxItemsPerEpoch candidateItems currentItems = Array.length currentItems + Array.length candidateItems >= maxItemsPerEpoch
+ let create maxItemsPerEpoch (Category cat) = create_ (shouldClose maxItemsPerEpoch) cat
+
+/// Custom Fold and caching logic compared to the IngesterService
+/// - When reading, we want the full Items
+/// - Caching only for one minute
+/// - There's no value in using the snapshot as it does not have the full state
+module Reader =
+
+ type State = Events.Cart[] * bool
+ let initial = [||], false
+ let evolve (es, closed) = function
+ | Events.Ingested e -> Array.append es e.carts, closed
+ | Events.Closed -> (es, true)
+ let fold : State -> Events.Event seq -> State = Seq.fold evolve
+
+ type StateDto = { closed : bool; carts : Events.Cart[] }
+
+ type Service internal (resolve : ConfirmedEpochId -> Equinox.Decider) =
+
+ /// Returns all the items currently held in the stream
+ member _.Read(epochId) : Async =
+ let decider = resolve epochId
+ decider.Query(fun (carts, closed) -> { closed = closed; carts = carts })
+
+ module Config =
+
+ let private (|Category|) = function
+ | Store.Config.Memory store -> Store.Memory.create CategoryName Events.codec initial fold store
+ | Store.Config.Cosmos (context, cache) -> Store.Cosmos.createUnoptimized CategoryName Events.codecJsonElement initial fold (context, cache)
+ | Store.Config.Dynamo (context, cache) -> Store.Dynamo.createUnoptimized CategoryName Events.codec initial fold (context, cache)
+ | Store.Config.Esdb (context, cache) -> Store.Esdb.createUnoptimized CategoryName Events.codec initial fold (context, cache)
+ | Store.Config.Sss (context, cache) -> Store.Sss.createUnoptimized CategoryName Events.codec initial fold (context, cache)
+ let create (Category cat) = Service(streamId >> Store.createDecider cat)
diff --git a/Sample/ECommerce.Equinox/ECommerce.Domain/ConfirmedIngester.fs b/Sample/ECommerce.Equinox/ECommerce.Domain/ConfirmedIngester.fs
new file mode 100644
index 000000000..6d719b2a8
--- /dev/null
+++ b/Sample/ECommerce.Equinox/ECommerce.Domain/ConfirmedIngester.fs
@@ -0,0 +1,28 @@
+module ECommerce.Domain.ConfirmedIngester
+
+type Service internal (tip : ExactlyOnceIngester.Service<_, _, _, _>) =
+
+ member _.IngestMany(originEpochId, cartSummaries) : Async =
+ tip.IngestMany(originEpochId, cartSummaries)
+
+ /// Slot the item into the series of epochs.
+ /// Returns true if it got added this time, i.e. idempotent retries don't count
+ member x.TryIngestCartSummary(originEpochId, cartSummary : ConfirmedEpoch.Events.Cart) : Async = async {
+ let! ingested = x.IngestMany(originEpochId, [| cartSummary |])
+ return ingested |> Seq.contains cartSummary.cartId }
+
+ /// Efficiently determine a valid ingestion origin epoch
+ member _.ActiveIngestionEpochId() =
+ tip.ActiveIngestionEpochId()
+
+module Config =
+
+ let create_ maxItemsPerEpoch linger store =
+ let series = ConfirmedSeries.Config.create store
+ let epochs = ConfirmedEpoch.Config.create maxItemsPerEpoch store
+ let log = Serilog.Log.ForContext()
+ let tip = ExactlyOnceIngester.create log linger (series.ReadIngestionEpochId, series.MarkIngestionEpochId) (epochs.Ingest, Array.toSeq)
+ Service(tip)
+ let create linger store =
+ let maxItemsPerEpoch = 10_000
+ create_ maxItemsPerEpoch linger store
diff --git a/Sample/ECommerce.Equinox/ECommerce.Domain/ConfirmedSeries.fs b/Sample/ECommerce.Equinox/ECommerce.Domain/ConfirmedSeries.fs
new file mode 100644
index 000000000..68d3d24e8
--- /dev/null
+++ b/Sample/ECommerce.Equinox/ECommerce.Domain/ConfirmedSeries.fs
@@ -0,0 +1,60 @@
+/// Maintains a pointer into the Epoch chain for a given Series
+/// Allows the Ingester to determine the current Epoch into which it commence writing via ReadIngestionEpochId
+/// As an Epoch is marked `Closed`, the Ingester will mark a new Epoch `Started` on this aggregate via MarkIngestionEpochId
+module ECommerce.Domain.ConfirmedSeries
+
+let [] CategoryName = "ConfirmedSeries"
+// TOCONSIDER: if you need multiple lists series/epochs in a single system, the Series and Epoch streams should have a SeriesId in the stream name
+// See also the implementation in the feedSource template, where the Series aggregate also functions as an index of series held in the system
+let streamId () = FsCodec.StreamId.gen ConfirmedSeriesId.toString ConfirmedSeriesId.wellKnownId
+
+// NB - these types and the union case names reflect the actual storage formats and hence need to be versioned with care
+[]
+module Events =
+
+ type Event =
+ | Started of {| epochId : ConfirmedEpochId |}
+ | Snapshotted of {| active : ConfirmedEpochId |}
+ interface TypeShape.UnionContract.IUnionContract
+ let codec = Store.Codec.gen
+ let codecJsonElement = Store.Codec.genJsonElement
+
+module Fold =
+
+ type State = ConfirmedEpochId option
+ let initial = None
+ let private evolve _state = function
+ | Events.Started e -> Some e.epochId
+ | Events.Snapshotted e -> Some e.active
+ let fold : State -> Events.Event seq -> State = Seq.fold evolve
+
+ let isOrigin = function Events.Snapshotted _ -> true | _ -> false
+ let toSnapshot s = Events.Snapshotted {| active = Option.get s |}
+
+let interpret epochId (state : Fold.State) = [|
+ if state |> Option.forall (fun cur -> cur < epochId) && epochId >= ConfirmedEpochId.initial then
+ yield Events.Started {| epochId = epochId |}|]
+
+type Service internal (resolve : unit -> Equinox.Decider) =
+
+ /// Determines the current active epoch
+ /// Uses cached values as epoch transitions are rare, and caller needs to deal with the inherent race condition in any case
+ member _.ReadIngestionEpochId() : Async =
+ let decider = resolve ()
+ decider.Query(Option.defaultValue ConfirmedEpochId.initial)
+
+ /// Mark specified `epochId` as live for the purposes of ingesting
+ /// Writers are expected to react to having writes to an epoch denied (due to it being Closed) by anointing a successor via this
+ member _.MarkIngestionEpochId epochId : Async =
+ let decider = resolve ()
+ decider.Transact(interpret epochId, load = Equinox.LoadOption.AnyCachedValue)
+
+module Config =
+
+ let private (|Category|) = function
+ | Store.Config.Memory store -> Store.Memory.create CategoryName Events.codec Fold.initial Fold.fold store
+ | Store.Config.Cosmos (context, cache) -> Store.Cosmos.createSnapshotted CategoryName Events.codecJsonElement Fold.initial Fold.fold (Fold.isOrigin, Fold.toSnapshot) (context, cache)
+ | Store.Config.Dynamo (context, cache) -> Store.Dynamo.createSnapshotted CategoryName Events.codec Fold.initial Fold.fold (Fold.isOrigin, Fold.toSnapshot) (context, cache)
+ | Store.Config.Esdb (context, cache) -> Store.Esdb.createUnoptimized CategoryName Events.codec Fold.initial Fold.fold (context, cache)
+ | Store.Config.Sss (context, cache) -> Store.Sss.createUnoptimized CategoryName Events.codec Fold.initial Fold.fold (context, cache)
+ let create (Category cat) = Service(streamId >> Store.createDecider cat)
diff --git a/Sample/ECommerce.Equinox/ECommerce.Domain/ECommerce.Domain.fsproj b/Sample/ECommerce.Equinox/ECommerce.Domain/ECommerce.Domain.fsproj
new file mode 100644
index 000000000..537e5fd1e
--- /dev/null
+++ b/Sample/ECommerce.Equinox/ECommerce.Domain/ECommerce.Domain.fsproj
@@ -0,0 +1,31 @@
+
+
+
+ net6.0
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/Sample/ECommerce.Equinox/ECommerce.Domain/ExactlyOnceIngester.fs b/Sample/ECommerce.Equinox/ECommerce.Domain/ExactlyOnceIngester.fs
new file mode 100644
index 000000000..12d7dba82
--- /dev/null
+++ b/Sample/ECommerce.Equinox/ECommerce.Domain/ExactlyOnceIngester.fs
@@ -0,0 +1,89 @@
+/// Generic Service that controls the deterministic application of requests that insert items into a chain of `Epoch` streams
+/// - the `Series` aggregate maintains a pointer to the current Epoch
+/// - as `Epoch`s complete (have `Closed` events logged), we update the `active` Epoch in the Series to reference the new one
+/// The fact that each request walks forward from a given start point until it either gets to append (or encounters a prior insertion)
+/// means we can guarantee the insertion/deduplication to be idempotent and insert exactly once per completed execution
+module ECommerce.Domain.ExactlyOnceIngester
+
+open FSharp.UMX // %
+
+type IngestResult<'req, 'res> = { accepted : 'res[]; closed : bool; residual : 'req[] }
+
+module Internal =
+
+ let unknown<[]'m> = UMX.tag -1
+ let next<[]'m> (value : int<'m>) = UMX.tag<'m>(UMX.untag value + 1)
+
+/// Ensures any given item is only added to the series exactly once by virtue of the following protocol:
+/// 1. Caller obtains an origin epoch via ActiveIngestionEpochId, storing that alongside the source item
+/// 2. Caller deterministically obtains that origin epoch to supply to Ingest/TryIngest such that retries can be idempotent
+type Service<[]'id, 'req, 'res, 'outcome> internal
+ ( log : Serilog.ILogger,
+ readActiveEpoch : unit -> Async>,
+ markActiveEpoch : int<'id> -> Async,
+ ingest : int<'id> * 'req [] -> Async>,
+ mapResults : 'res [] -> 'outcome seq,
+ linger) =
+
+ let uninitializedSentinel : int = %Internal.unknown
+ let mutable currentEpochId_ = uninitializedSentinel
+ let currentEpochId () = if currentEpochId_ <> uninitializedSentinel then Some %currentEpochId_ else None
+
+ let tryIngest (reqs : (int<'id> * 'req)[][]) =
+ let rec aux ingestedItems items = async {
+ let epochId = items |> Array.minBy fst |> fst
+ let epochItems, futureEpochItems = items |> Array.partition (fun (e, _ : 'req) -> e = epochId)
+ let! res = ingest (epochId, Array.map snd epochItems)
+ let ingestedItemIds = Array.append ingestedItems res.accepted
+ let logLevel =
+ if res.residual.Length <> 0 || futureEpochItems.Length <> 0 || Array.isEmpty res.accepted then Serilog.Events.LogEventLevel.Information
+ else Serilog.Events.LogEventLevel.Debug
+ log.Write(logLevel, "Added {count}/{total} items to {epochId} Residual {residual} Future {future}",
+ res.accepted.Length, epochItems.Length, epochId, res.residual.Length, futureEpochItems.Length)
+ let nextEpochId = Internal.next epochId
+ let pushedToNextEpoch = res.residual |> Array.map (fun x -> nextEpochId, x)
+ match Array.append pushedToNextEpoch futureEpochItems with
+ | [||] ->
+ // Any writer noticing we've moved to a new Epoch shares the burden of marking it active in the Series
+ let newActiveEpochId = if res.closed then nextEpochId else epochId
+ if currentEpochId_ < %newActiveEpochId then
+ log.Information("Marking {epochId} active", newActiveEpochId)
+ do! markActiveEpoch newActiveEpochId
+ System.Threading.Interlocked.CompareExchange(¤tEpochId_, %newActiveEpochId, currentEpochId_) |> ignore
+ return ingestedItemIds
+ | remaining -> return! aux ingestedItemIds remaining }
+ aux [||] (Array.concat reqs)
+
+ /// In the overall processing using an Ingester, we frequently have a Scheduler running N streams concurrently
+ /// If each thread works in isolation, they'll conflict with each other as they feed the Items into the batch in epochs.Ingest
+ /// Instead, we enable concurrent requests to coalesce by having requests converge in this AsyncBatchingGate
+ /// This has the following critical effects:
+ /// - Traffic to CosmosDB is naturally constrained to a single flight in progress
+ /// (BatchingGate does not release next batch for execution until current has succeeded or throws)
+ /// - RU consumption for writing to the batch is optimized (1 write inserting 1 event document vs N writers writing N)
+ /// - Peak throughput is more consistent as latency is not impacted by the combination of having to:
+ /// a) back-off, re-read and retry if there's a concurrent write Optimistic Concurrency Check failure when writing the stream
+ /// b) enter a prolonged period of retries if multiple concurrent writes trigger rate limiting and 429s from CosmosDB
+ /// c) readers will less frequently encounter sustained 429s on the batch
+ let batchedIngest = Equinox.Core.Batching.Batcher(tryIngest, linger)
+
+ /// Run the requests over a chain of epochs.
+ /// Returns the subset that actually got handled this time around (exclusive of items that did not trigger events per idempotency rules).
+ member _.IngestMany(originEpoch, reqs) : Async<'outcome seq> = async {
+ if Array.isEmpty reqs then return Seq.empty else
+
+ let! results = batchedIngest.Execute [| for x in reqs -> originEpoch, x |]
+ return results |> mapResults
+ }
+
+ /// Exposes the current high water mark epoch - i.e. the tip epoch to which appends are presently being applied.
+ /// The fact that any Ingest call for a given item (or set of items) always commences from the same origin is key to exactly once insertion guarantee.
+ /// Caller should first store this alongside the item in order to deterministically be able to start from the same origin in idempotent retry cases.
+ /// Uses cached values as epoch transitions are rare, and caller needs to deal with the inherent race condition in any case
+ member _.ActiveIngestionEpochId() : Async> =
+ match currentEpochId () with
+ | Some currentEpochId -> async { return currentEpochId }
+ | None -> readActiveEpoch()
+
+let create log linger (readIngestionEpoch, markIngestionEpoch) (apply, mapResult) =
+ Service<'id, 'req, 'res, 'outcome>(log, readIngestionEpoch, markIngestionEpoch, apply, mapResult, linger = linger)
diff --git a/Sample/ECommerce.Equinox/ECommerce.Domain/RandomProductPriceCalculator.fs b/Sample/ECommerce.Equinox/ECommerce.Domain/RandomProductPriceCalculator.fs
new file mode 100644
index 000000000..662842a88
--- /dev/null
+++ b/Sample/ECommerce.Equinox/ECommerce.Domain/RandomProductPriceCalculator.fs
@@ -0,0 +1,12 @@
+namespace ECommerce.Domain
+
+type RandomProductPriceCalculator() =
+
+ let productPrices = System.Collections.Concurrent.ConcurrentDictionary()
+
+ member _.Calculate(productId : ProductId) : Async = async {
+ let r = System.Random()
+ let calc _ = (r.NextDouble() |> decimal) * 100m
+ let price : decimal = productPrices.GetOrAdd(productId, calc)
+ return price
+ }
diff --git a/Sample/ECommerce.Equinox/ECommerce.Domain/ShoppingCart.fs b/Sample/ECommerce.Equinox/ECommerce.Domain/ShoppingCart.fs
new file mode 100644
index 000000000..4d815e3ee
--- /dev/null
+++ b/Sample/ECommerce.Equinox/ECommerce.Domain/ShoppingCart.fs
@@ -0,0 +1,169 @@
+module ECommerce.Domain.ShoppingCart
+
+open Propulsion.Internal
+
+let [] CategoryName = "ShoppingCart"
+let streamId = FsCodec.StreamId.gen CartId.toString
+let private catId = CategoryId(CategoryName, streamId, FsCodec.StreamId.dec CartId.(|Parse|))
+
+module Events =
+
+ type Event =
+ | Initialized of {| clientId : ClientId |}
+ | ItemAdded of {| productId : ProductId; quantity : int; unitPrice : decimal |}
+ | ItemRemoved of {| productId : ProductId; (*; quantity : int;*) unitPrice : decimal |}
+ | Confirmed of {| confirmedAt : System.DateTimeOffset |}
+ | Registering of {| originEpoch : ConfirmedEpochId |}
+ interface TypeShape.UnionContract.IUnionContract
+ let codec = Store.Codec.gen
+ let codecJsonElement = Store.Codec.genJsonElement
+
+module Reactions =
+
+ let [] (|For|_|) = catId.TryDecode
+ let dec = Streams.Codec.dec
+ let config = catId.StreamName, dec
+ let [] (|Decode|_|) = function
+ | struct (For id, _) & Streams.Decode dec events -> ValueSome struct (id, events)
+ | _ -> ValueNone
+ let chooseConfirmed = function
+ | Events.Confirmed _ -> ValueSome ()
+ | _ -> ValueNone
+ let [] (|Confirmed|_|) events = Seq.rev events |> Seq.tryPickV chooseConfirmed
+ let chooseNotRegistering = function
+ | Events.Registering _ -> ValueNone
+ | _ -> ValueSome ()
+ let [] (|StateChanged|_|) events = Seq.rev events |> Seq.tryPickV chooseNotRegistering
+
+module Fold =
+
+ type Item = { productId : ProductId; quantity : int; unitPrice : decimal }
+
+ type Status = Pending | Confirmed
+ type State =
+ { clientId : ClientId option
+ status : Status; items : Item array
+ confirmedAt : System.DateTimeOffset option
+ confirmedOriginEpoch : ConfirmedEpochId option }
+ let initial = { clientId = None; status = Status.Pending; items = Array.empty; confirmedAt = None; confirmedOriginEpoch = None }
+ let isClosed (s : State) = match s.status with Confirmed -> true | Pending -> false
+ module ItemList =
+ let keys (x : Item) = x.productId, x.unitPrice
+ let add (productId, price, quantity) (current : Item seq) =
+ let newItemKeys = productId, price
+ let mkItem (productId, price, quantity) = { productId = productId; quantity = quantity; unitPrice = price }
+ let mutable merged = false
+ [| for x in current do
+ if newItemKeys = keys x then
+ mkItem (productId, price, x.quantity + quantity)
+ merged <- true
+ else
+ x
+ if not merged then
+ mkItem (productId, price, quantity) |]
+ let remove (productId, price) (current : Item[]) =
+ current |> Array.where (fun x -> keys x <> (productId, price))
+ let private evolve s = function
+ | Events.Initialized e -> { s with clientId = Some e.clientId }
+ | Events.ItemAdded e -> { s with items = s.items |> ItemList.add (e.productId, e.unitPrice, e.quantity) }
+ | Events.ItemRemoved e -> { s with items = s.items |> ItemList.remove (e.productId, e.unitPrice) }
+ | Events.Confirmed e -> { s with status = Confirmed; confirmedAt = Some e.confirmedAt }
+ | Events.Registering e -> { s with confirmedOriginEpoch = Some e.originEpoch }
+ let fold = Seq.fold evolve
+
+let decideInitialize clientId (s : Fold.State) = [|
+ if Option.isNone s.clientId then
+ Events.Initialized {| clientId = clientId |} |]
+
+let decideAdd calculatePrice productId quantity state = async {
+ match state with
+ | s when Fold.isClosed s -> return invalidOp $"Adding product item for cart in '%A{s.status}' status is not allowed."
+ | _ ->
+ let! price = calculatePrice (productId, quantity)
+ return (), [| Events.ItemAdded {| productId = productId; unitPrice = price; quantity = quantity |} |] }
+
+let decideRemove (productId, price) = function
+ | s when Fold.isClosed s -> invalidOp $"Removing product item for cart in '%A{s.status}' status is not allowed."
+ | _ ->
+ [| Events.ItemRemoved {| productId = productId; unitPrice = price |} |]
+
+let decideConfirm at = function
+ | s when Fold.isClosed s -> [||]
+ | _ -> [| Events.Confirmed {| confirmedAt = at |} |]
+
+module Details =
+
+ type View = { (* id *) clientId : ClientId; status : Fold.Status; items : Item[] }
+ and Item = { productId : ProductId; unitPrice : decimal; quantity : int }
+
+ let render = function
+ | ({ clientId = None } : Fold.State) -> None
+ | { clientId = Some clientId } as s ->
+ let items = [| for { productId = productId; quantity = q; unitPrice = p } in s.items ->
+ { productId = productId; unitPrice = p; quantity = q } |]
+ Some { clientId = clientId; status = s.status; items = items }
+
+let summarizeWithOriginEpoch getActiveEpochId state = async {
+ match state with
+ | s when not (Fold.isClosed s) -> return failwith "Unexpected"
+ | { confirmedOriginEpoch = Some originEpoch } as s ->
+ return (Details.render s |> Option.get, originEpoch), [||]
+ | { confirmedOriginEpoch = None } as s ->
+ let! originEpoch = getActiveEpochId ()
+ return (Details.render s |> Option.get, originEpoch), [| Events.Registering {| originEpoch = originEpoch |} |] }
+
+type Service internal (resolve : CartId -> Equinox.Decider, calculatePrice : ProductId * int -> Async) =
+
+ member _.Initialize(cartId, clientId) =
+ let decider = resolve cartId
+ decider.Transact(decideInitialize clientId)
+
+ member _.Add(cartId, productId, quantity) =
+ let decider = resolve cartId
+ decider.Transact(decideAdd calculatePrice productId quantity)
+
+ member _.Remove(cartId, productId, price) =
+ let decider = resolve cartId
+ decider.Transact(decideRemove (productId, price))
+
+ member _.Confirm(cartId, at) =
+ let decider = resolve cartId
+ decider.Transact(decideConfirm at)
+
+ // NOTE doing this does not fulfil the CQRS principle to the letter
+ // However, its not unrealistic for this demo in that
+ // a) it means you can read your writes immediately
+ // b) it's not unreasonable in efficiency terms
+ // - on Cosmos, you pay only 1RU to read through the cache with the etag
+ // - on EventStoreDB you are reading forward from a cached stream and hence are typically doing a roundtrip that does not send any events
+ member _.Read(cartId) : Async =
+ let decider = resolve cartId
+ decider.Query(Details.render)
+
+ /// Summarizes the contents of the cart
+ /// Decides the tranche from where the insertion into the PoolTranches is to commence
+ member _.SummarizeWithOriginEpoch(cartId, getActiveEpochId) : Async =
+ let decider = resolve cartId
+ decider.Transact(summarizeWithOriginEpoch getActiveEpochId)
+
+ /// Render view (and emit version on which it was based) for Denormalizer to store
+ member _.SummarizeWithVersion(cartId) : Async =
+ let decider = resolve cartId
+ decider.QueryEx(fun c -> Details.render c.State, c.Version)
+
+module Config =
+
+ // Adapts the external Pricing algorithm interface shape (see IProductPriceCalculator) to what's required by `type Service`
+ let calculatePrice (pricer : ProductId -> Async) (productId, _quantity) : Async =
+ pricer productId
+
+ let private (|Category|) = function
+ | Store.Config.Memory store -> Store.Memory.create CategoryName Events.codec Fold.initial Fold.fold store
+ | Store.Config.Cosmos (context, cache) -> Store.Cosmos.createUnoptimized CategoryName Events.codecJsonElement Fold.initial Fold.fold (context, cache)
+ | Store.Config.Dynamo (context, cache) -> Store.Dynamo.createUnoptimized CategoryName Events.codec Fold.initial Fold.fold (context, cache)
+ | Store.Config.Esdb (context, cache) -> Store.Esdb.createUnoptimized CategoryName Events.codec Fold.initial Fold.fold (context, cache)
+ | Store.Config.Sss (context, cache) -> Store.Sss.createUnoptimized CategoryName Events.codec Fold.initial Fold.fold (context, cache)
+ let create_ pricer (Category cat) = Service(streamId >> Store.createDecider cat, calculatePrice pricer)
+ let create =
+ let defaultCalculator = RandomProductPriceCalculator()
+ create_ defaultCalculator.Calculate
diff --git a/Sample/ECommerce.Equinox/ECommerce.Domain/ShoppingCartSummary.fs b/Sample/ECommerce.Equinox/ECommerce.Domain/ShoppingCartSummary.fs
new file mode 100644
index 000000000..4ee02d6db
--- /dev/null
+++ b/Sample/ECommerce.Equinox/ECommerce.Domain/ShoppingCartSummary.fs
@@ -0,0 +1,71 @@
+module ECommerce.Domain.ShoppingCartSummary
+
+let [] CategoryName = "ShoppingCartSummary"
+
+let streamId = FsCodec.StreamId.gen CartId.toString
+
+module Events =
+
+ type Ingested = { version : int64; value : Summary }
+ and Summary = { items : Item[]; status : Status; clientId : ClientId }
+ and Item = { productId : ProductId; unitPrice : decimal; quantity : int }
+ and Status = Pending | Confirmed
+ type Event =
+ | Ingested of Ingested
+ interface TypeShape.UnionContract.IUnionContract
+ let codec = Store.Codec.gen
+ let codecJsonElement = Store.Codec.genJsonElement
+
+module Fold =
+
+ type State = Events.Ingested option
+ let initial = None
+ let private evolve _s = function
+ | Events.Ingested e -> Some e
+ let fold = Seq.fold evolve
+ let toSnapshot (s : State) = s.Value |> Events.Ingested
+
+module Details =
+
+ type View = { (* id *) clientId : ClientId; status : Events.Status; items : Item[] }
+ and Item = { productId : ProductId; unitPrice : decimal; quantity : int }
+
+ let render : Fold.State -> View option = function
+ | None -> None
+ | Some { value = v } ->
+ let items = [| for { productId = productId; quantity = q; unitPrice = p } in v.items ->
+ { productId = productId; unitPrice = p; quantity = q } |]
+ Some { clientId = v.clientId; status = v.status; items = items }
+
+module Ingest =
+
+ let summarizeShoppingCartView (view : ShoppingCart.Details.View) : Events.Summary =
+ let mapStatus = function
+ | ShoppingCart.Fold.Pending -> Events.Pending
+ | ShoppingCart.Fold.Confirmed -> Events.Confirmed
+ { clientId = view.clientId; status = mapStatus view.status
+ items = [| for i in view.items -> { productId = i.productId; unitPrice = i.unitPrice; quantity = i.quantity } |] }
+
+ let decide (version : int64, value : Events.Summary): Fold.State -> bool * Events.Event[] = function
+ | Some { version = v } when v >= version -> false, [||]
+ | None -> false, [||]
+ | _ -> true, [| Events.Ingested { version = version; value = value } |]
+
+type Service internal (resolve : CartId -> Equinox.Decider) =
+
+ member _.Read(cartId) : Async =
+ let decider = resolve cartId
+ decider.Query(Details.render)
+
+ member _.TryIngest(cartId, version, value) : Async =
+ let decider = resolve cartId
+ decider.Transact(Ingest.decide (version, Ingest.summarizeShoppingCartView value))
+
+module Config =
+
+ let private (|Category|) = function
+ | Store.Config.Memory store -> Store.Memory.create CategoryName Events.codec Fold.initial Fold.fold store
+ | Store.Config.Cosmos (context, cache) -> Store.Cosmos.createRollingState CategoryName Events.codecJsonElement Fold.initial Fold.fold Fold.toSnapshot (context, cache)
+ | Store.Config.Dynamo (context, cache) -> Store.Dynamo.createRollingState CategoryName Events.codec Fold.initial Fold.fold Fold.toSnapshot (context, cache)
+ | Store.Config.Esdb _ | Store.Config.Sss _ -> failwith "Not implemented: For EventStore/Sss its suggested to do a cached read from the write side"
+ let create (Category cat) = Service(streamId >> Store.createDecider cat)
diff --git a/Sample/ECommerce.Equinox/ECommerce.Domain/Store.fs b/Sample/ECommerce.Equinox/ECommerce.Domain/Store.fs
new file mode 100644
index 000000000..e1b0592cf
--- /dev/null
+++ b/Sample/ECommerce.Equinox/ECommerce.Domain/Store.fs
@@ -0,0 +1,87 @@
+module Store
+
+module Metrics =
+
+ let log = Serilog.Log.ForContext("isMetric", true)
+
+let createDecider category = Equinox.Decider.forStream Metrics.log category
+
+module Memory =
+
+ let create name codec initial fold store : Equinox.Category<_, _, _> =
+ Equinox.MemoryStore.MemoryStoreCategory(store, name, FsCodec.Compression.EncodeUncompressed codec, fold, initial)
+
+module Codec =
+
+ open FsCodec.SystemTextJson
+ let private defaultOptions = Options.Create(autoTypeSafeEnumToJsonString = true)
+ let gen<'t when 't :> TypeShape.UnionContract.IUnionContract> =
+ Codec.Create<'t>(options = defaultOptions)
+ let genJsonElement<'t when 't :> TypeShape.UnionContract.IUnionContract> =
+ CodecJsonElement.Create<'t>(options = defaultOptions)
+
+let private defaultCacheDuration = System.TimeSpan.FromMinutes 20.
+
+module Cosmos =
+
+ let private createCached name codec initial fold accessStrategy (context, cache) =
+ let cacheStrategy = Equinox.CachingStrategy.SlidingWindow (cache, defaultCacheDuration)
+ Equinox.CosmosStore.CosmosStoreCategory(context, name, codec, fold, initial, accessStrategy, cacheStrategy)
+
+ let createUnoptimized name codec initial fold (context, cache) =
+ let accessStrategy = Equinox.CosmosStore.AccessStrategy.Unoptimized
+ createCached name codec initial fold accessStrategy (context, cache)
+
+ let createSnapshotted name codec initial fold (isOrigin, toSnapshot) (context, cache) =
+ let accessStrategy = Equinox.CosmosStore.AccessStrategy.Snapshot (isOrigin, toSnapshot)
+ createCached name codec initial fold accessStrategy (context, cache)
+
+ let createRollingState name codec initial fold toSnapshot (context, cache) =
+ let accessStrategy = Equinox.CosmosStore.AccessStrategy.RollingState toSnapshot
+ createCached name codec initial fold accessStrategy (context, cache)
+
+module Dynamo =
+
+ let private createCached name codec initial fold accessStrategy (context, cache) =
+ let cacheStrategy = Equinox.CachingStrategy.SlidingWindow (cache, defaultCacheDuration)
+ Equinox.DynamoStore.DynamoStoreCategory(context, name, FsCodec.Compression.EncodeTryCompress codec, fold, initial, accessStrategy, cacheStrategy)
+
+ let createUnoptimized name codec initial fold (context, cache) =
+ let accessStrategy = Equinox.DynamoStore.AccessStrategy.Unoptimized
+ createCached name codec initial fold accessStrategy (context, cache)
+
+ let createSnapshotted name codec initial fold (isOrigin, toSnapshot) (context, cache) =
+ let accessStrategy = Equinox.DynamoStore.AccessStrategy.Snapshot (isOrigin, toSnapshot)
+ createCached name codec initial fold accessStrategy (context, cache)
+
+ let createRollingState name codec initial fold toSnapshot (context, cache) =
+ let accessStrategy = Equinox.DynamoStore.AccessStrategy.RollingState toSnapshot
+ createCached name codec initial fold accessStrategy (context, cache)
+
+module Esdb =
+
+ let private createCached name codec initial fold accessStrategy (context, cache) =
+ let cacheStrategy = Equinox.CachingStrategy.SlidingWindow (cache, defaultCacheDuration)
+ Equinox.EventStoreDb.EventStoreCategory(context, name, codec, fold, initial, accessStrategy, cacheStrategy)
+ let createUnoptimized name codec initial fold (context, cache) =
+ createCached name codec initial fold Equinox.EventStoreDb.AccessStrategy.Unoptimized (context, cache)
+ let createLatestKnownEvent name codec initial fold (context, cache) =
+ createCached name codec initial fold Equinox.EventStoreDb.AccessStrategy.LatestKnownEvent (context, cache)
+
+module Sss =
+
+ let private createCached name codec initial fold accessStrategy (context, cache) =
+ let cacheStrategy = Equinox.CachingStrategy.SlidingWindow (cache, defaultCacheDuration)
+ Equinox.SqlStreamStore.SqlStreamStoreCategory(context, name, codec, fold, initial, accessStrategy, cacheStrategy)
+ let createUnoptimized name codec initial fold (context, cache) =
+ createCached name codec initial fold Equinox.SqlStreamStore.AccessStrategy.Unoptimized (context, cache)
+ let createLatestKnownEvent name codec initial fold (context, cache) =
+ createCached name codec initial fold Equinox.SqlStreamStore.AccessStrategy.LatestKnownEvent (context, cache)
+
+[]
+type Config =
+ | Memory of Equinox.MemoryStore.VolatileStore)>
+ | Cosmos of Equinox.CosmosStore.CosmosStoreContext * Equinox.Cache
+ | Dynamo of Equinox.DynamoStore.DynamoStoreContext * Equinox.Cache
+ | Esdb of Equinox.EventStoreDb.EventStoreContext * Equinox.Cache
+ | Sss of Equinox.SqlStreamStore.SqlStreamStoreContext * Equinox.Cache
diff --git a/Sample/ECommerce.Equinox/ECommerce.Domain/Streams.fs b/Sample/ECommerce.Equinox/ECommerce.Domain/Streams.fs
new file mode 100644
index 000000000..95dbfe02c
--- /dev/null
+++ b/Sample/ECommerce.Equinox/ECommerce.Domain/Streams.fs
@@ -0,0 +1,22 @@
+module Streams
+
+open Serilog
+
+module Codec =
+
+ let dec<'E when 'E :> TypeShape.UnionContract.IUnionContract> : Propulsion.Sinks.Codec<'E> =
+ FsCodec.SystemTextJson.Codec.Create<'E>() // options = Options.Default
+
+ let private renderBody (x: Propulsion.Sinks.EventBody) = System.Text.Encoding.UTF8.GetString(x.Span)
+
+ // Uses the supplied codec to decode the supplied event record (iff at LogEventLevel.Debug, failures are logged, citing `stream` and `.Data`)
+ let internal tryDecode<'E> (codec: Propulsion.Sinks.Codec<'E>) (streamName: FsCodec.StreamName) event =
+ match codec.Decode event with
+ | ValueNone when Log.IsEnabled Serilog.Events.LogEventLevel.Debug ->
+ Log.ForContext("eventData", renderBody event.Data)
+ .Debug("Codec {type} Could not decode {eventType} in {stream}", codec.GetType().FullName, event.EventType, streamName)
+ ValueNone
+ | x -> x
+
+let (|Decode|) codec struct (stream, events: Propulsion.Sinks.Event[]): 'E[] =
+ events |> Propulsion.Internal.Array.chooseV (Codec.tryDecode codec stream)
diff --git a/Sample/ECommerce.Equinox/ECommerce.Domain/Types.fs b/Sample/ECommerce.Equinox/ECommerce.Domain/Types.fs
new file mode 100644
index 000000000..394cbcb0f
--- /dev/null
+++ b/Sample/ECommerce.Equinox/ECommerce.Domain/Types.fs
@@ -0,0 +1,78 @@
+namespace ECommerce.Domain
+
+open FSharp.UMX
+open System
+
+module Guid =
+
+ let inline toStringN (x : Guid) = x.ToString "N"
+
+type ProductId = Guid
+and [] productId
+module ProductId =
+ let toString (x : ProductId) : string = (UMX.untag x).ToString("N")
+ let parse (value : Guid) : ProductId = %value
+ let (|Parse|) = parse
+
+type ClientId = Guid
+and [] clientId
+module ClientId =
+ let toString (x : ClientId) : string = (UMX.untag x).ToString("N")
+ let parse (value : Guid Nullable) : ClientId =
+ if not value.HasValue || value.Value = Guid.Empty then raise <| ArgumentOutOfRangeException(nameof value)
+ %value.Value
+ let (|Parse|) = parse
+
+type CartId = Guid
+and [] cartId
+module CartId =
+ let toString (x : CartId) : string = (UMX.untag x).ToString("N")
+ let (|Parse|) : string -> CartId = Guid.Parse >> UMX.tag
+ let parse (value : Guid Nullable) : CartId =
+ if not value.HasValue || value.Value = Guid.Empty then raise <| ArgumentOutOfRangeException(nameof value)
+ %value.Value
+ let (|ParseGuid|) = parse
+ let generate () : CartId = Guid.NewGuid() |> Nullable |> parse
+
+(* At present, there's only a single series of Confirmed carts; this extension point could be used
+ e.g. if one was to implement multi-tenancy, the tenantId would become the seriesId *)
+
+type [] confirmedSeriesId
+type ConfirmedSeriesId = int
+module ConfirmedSeriesId =
+ let wellKnownId = 0
+ let toString (value : ConfirmedSeriesId) : string = string %value
+
+type [] confirmedEpochId
+type ConfirmedEpochId = int
+module ConfirmedEpochId =
+ let initial = 0
+ let value (value : ConfirmedEpochId) : int = %value
+ let parse (value : int) : ConfirmedEpochId = %value
+ let next (value : ConfirmedEpochId) : ConfirmedEpochId = % (%value + 1)
+ let toString (value : ConfirmedEpochId) : string = string %value
+
+type [] confirmedCheckpoint
+type ConfirmedCheckpoint = int64
+module ConfirmedCheckpoint =
+
+ let initial : ConfirmedCheckpoint = %0L
+ let factor = 1_000_000L
+
+ let ofEpochAndOffset (epoch : ConfirmedEpochId) offset : ConfirmedCheckpoint =
+ int64 (ConfirmedEpochId.value epoch) * factor + int64 offset |> UMX.tag
+
+ let ofEpochContent (epoch : ConfirmedEpochId) isClosed count : ConfirmedCheckpoint =
+ let epoch, offset =
+ if isClosed then ConfirmedEpochId.next epoch, 0
+ else epoch, count
+ ofEpochAndOffset epoch offset
+
+ let toEpochAndOffset (value : ConfirmedCheckpoint) : ConfirmedEpochId * int =
+ let d, r = Math.DivRem(%value, factor)
+ (ConfirmedEpochId.parse (int d)), int r
+
+/// Handles symmetric generation and decoding of StreamNames composed of a series of elements via the FsCodec.StreamId helpers
+type internal CategoryId<'elements>(name, gen: 'elements -> FsCodec.StreamId, dec: FsCodec.StreamId -> 'elements) =
+ member _.StreamName = gen >> FsCodec.StreamName.create name
+ member _.TryDecode = FsCodec.StreamName.tryFind name >> ValueOption.map dec
diff --git a/Sample/ECommerce.Equinox/ECommerce.FeedConsumer/ApiClient.fs b/Sample/ECommerce.Equinox/ECommerce.FeedConsumer/ApiClient.fs
new file mode 100644
index 000000000..ae4a956ce
--- /dev/null
+++ b/Sample/ECommerce.Equinox/ECommerce.FeedConsumer/ApiClient.fs
@@ -0,0 +1,88 @@
+module ECommerce.FeedConsumer.ApiClient
+
+open FSharp.UMX
+open System.Net.Http
+
+open ECommerce.FeedConsumer.Domain
+
+(* The feed presents a Tranche (series of epochs) per FC *)
+
+module TrancheId =
+
+ let toFcId (x : Propulsion.Feed.TrancheId) : FcId = %x
+ let ofFcId (x : FcId) : Propulsion.Feed.TrancheId = %x
+
+type TicketsEpochId = int
+ and [] ticketsEpochId
+
+[]
+type TicketsTranchesDto = { activeEpochs : TrancheReferenceDto[] }
+ and TrancheReferenceDto = { fc : FcId; epochId : TicketsEpochId }
+
+(* Each Tranche response includes a checkpoint, which can be presented to Poll in order to resume consumption *)
+
+type TicketsCheckpoint = int64
+ and [] ticketsCheckpoint
+module TicketsCheckpoint =
+ let ofPosition (x : Propulsion.Feed.Position) : TicketsCheckpoint = %x
+ let toPosition (x : TicketsCheckpoint) : Propulsion.Feed.Position = %x
+ let toStreamIndex (x : TicketsCheckpoint) : int64 = %x
+
+type ItemDto = { id : TicketId; payload : string }
+type SliceDto = { closed : bool; tickets : ItemDto[]; position : TicketsCheckpoint; checkpoint : TicketsCheckpoint }
+
+type Session(client: HttpClient) =
+
+ member _.Send(req : HttpRequestMessage) : Async =
+ client.Send2(req)
+
+type TicketsClient(session: Session) =
+
+ let basePath = "api/tickets"
+
+ member _.ActiveFcs() : Async = async {
+ let request = HttpReq.get () |> HttpReq.withPath basePath
+ let! response = session.Send request
+ let! body = response |> HttpRes.deserializeOkStj
+ return [| for f in body.activeEpochs -> f.fc |]
+ }
+
+ member _.ReadPage(fc : FcId, index : int) : Async = async {
+ let request = HttpReq.post () |> HttpReq.withPathf "%s/%O/%d" basePath fc index
+ let! response = session.Send request
+ return! response |> HttpRes.deserializeOkStj
+ }
+
+ member _.Poll(fc : FcId, checkpoint: TicketsCheckpoint) : Async = async {
+ let request = HttpReq.create () |> HttpReq.withPathf "%s/%O/slice/%O" basePath fc checkpoint
+ let! response = session.Send request
+ return! response |> HttpRes.deserializeOkStj
+ }
+
+type Session with
+
+ member session.Tickets = TicketsClient session
+
+type TicketsFeed(baseUri) =
+
+ let client = new HttpClient(BaseAddress = baseUri)
+ let tickets = Session(client).Tickets
+
+ let batch pg items : Propulsion.Feed.Page =
+ { checkpoint = TicketsCheckpoint.toPosition pg.checkpoint; items = items; isTail = not pg.closed }
+
+ // TODO add retries - consumer loop will abort if this throws
+ member _.Poll(trancheId, pos) = async {
+ let checkpoint = TicketsCheckpoint.ofPosition pos
+ let! pg = tickets.Poll(TrancheId.toFcId trancheId, checkpoint)
+ let baseIndex = TicketsCheckpoint.toStreamIndex pg.position
+ let map (x : ItemDto) : Ingester.PipelineEvent.Item = { id = x.id; payload = x.payload }
+ let items = pg.tickets |> Array.mapi (fun i x -> Ingester.PipelineEvent.ofIndexAndItem (baseIndex + int64 i) (map x))
+ return batch pg items
+ }
+
+ // TODO add retries - consumer loop will not commence if this emits an exception
+ member _.ReadTranches() : Async = async {
+ let! activeFcs = tickets.ActiveFcs()
+ return [| for f in activeFcs -> TrancheId.ofFcId f |]
+ }
diff --git a/Sample/ECommerce.Equinox/ECommerce.FeedConsumer/ECommerce.FeedConsumer.fsproj b/Sample/ECommerce.Equinox/ECommerce.FeedConsumer/ECommerce.FeedConsumer.fsproj
new file mode 100644
index 000000000..d3587691f
--- /dev/null
+++ b/Sample/ECommerce.Equinox/ECommerce.FeedConsumer/ECommerce.FeedConsumer.fsproj
@@ -0,0 +1,22 @@
+
+
+
+ Exe
+ net6.0
+ 5
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/Sample/ECommerce.Equinox/ECommerce.FeedConsumer/Infrastructure.fs b/Sample/ECommerce.Equinox/ECommerce.FeedConsumer/Infrastructure.fs
new file mode 100644
index 000000000..384c5f0de
--- /dev/null
+++ b/Sample/ECommerce.Equinox/ECommerce.FeedConsumer/Infrastructure.fs
@@ -0,0 +1,183 @@
+[]
+module ECommerce.FeedConsumer.Infrastructure
+
+type Async with
+ /// Re-raise an exception so that the current stacktrace is preserved
+ static member Raise(e : #exn) : Async<'T> = Async.FromContinuations (fun (_,ec,_) -> ec e)
+
+open System.Text
+
+type StringBuilder with
+ member sb.Appendf fmt = Printf.ksprintf (ignore << sb.Append) fmt
+ member sb.Appendfn fmt = Printf.ksprintf (ignore << sb.AppendLine) fmt
+
+ static member inline Build(builder : StringBuilder -> unit) =
+ let instance = StringBuilder() // TOCONSIDER PooledStringBuilder.GetInstance()
+ builder instance
+ instance.ToString()
+
+open System
+open System.Net
+open System.Net.Http
+open System.Runtime.Serialization
+
+/// Operations on System.Net.HttpRequestMessage
+module HttpReq =
+
+ /// Creates an HTTP GET request.
+ let inline create () = new HttpRequestMessage()
+
+ /// Assigns a method to an HTTP request.
+ let inline withMethod (m : HttpMethod) (req : HttpRequestMessage) =
+ req.Method <- m
+ req
+
+ /// Creates an HTTP GET request.
+ let inline get () = create ()
+
+ /// Creates an HTTP POST request.
+ let inline post () = create () |> withMethod HttpMethod.Post
+
+ /// Assigns a path to an HTTP request.
+ let inline withUri (u : Uri) (req : HttpRequestMessage) =
+ req.RequestUri <- u
+ req
+
+ /// Assigns a path to an HTTP request.
+ let inline withPath (p : string) (req : HttpRequestMessage) =
+ req |> withUri (Uri(p, UriKind.Relative))
+
+ /// Assigns a path to a Http request using printf-like formatting.
+ let inline withPathf fmt =
+ Printf.ksprintf withPath fmt
+
+type HttpContent with
+ member c.ReadAsString() = async {
+ match c with
+ | null -> return null
+ | c -> return! c.ReadAsStringAsync() |> Async.AwaitTask
+ }
+
+ // only intended for logging under control of InvalidHttpResponseException, hence the esoteric name
+ member internal c.ReadAsStringDiapered() = async {
+ try return! c.ReadAsString()
+ with :? ObjectDisposedException -> return ""
+ }
+
+type HttpClient with
+ ///
+ /// Drop-in replacement for HttpClient.SendAsync which addresses known timeout issues
+ ///
+ /// HttpRequestMessage to be submitted.
+ member client.Send2(msg : HttpRequestMessage) = async {
+ let! ct = Async.CancellationToken
+ try return! client.SendAsync(msg, ct) |> Async.AwaitTask
+ // address https://github.com/dotnet/corefx/issues/20296
+ with :? System.Threading.Tasks.TaskCanceledException ->
+ let message =
+ match client.BaseAddress with
+ | null -> "HTTP request timeout"
+ | baseAddr -> sprintf "HTTP request timeout [%O]" baseAddr
+
+ return! Async.Raise(TimeoutException message)
+ }
+
+/// Exception indicating an unexpected response received by an Http Client
+type InvalidHttpResponseException =
+ inherit Exception
+
+ // TODO: include headers
+ val private userMessage : string
+ val private requestMethod : string
+ val RequestUri : Uri
+ val RequestBody : string
+ val StatusCode : HttpStatusCode
+ val ReasonPhrase : string
+ val ResponseBody : string
+
+ member e.RequestMethod = HttpMethod(e.requestMethod)
+
+ private new (userMessage : string, requestMethod : HttpMethod, requestUri : Uri, requestBody : string,
+ statusCode : HttpStatusCode, reasonPhrase : string, responseBody : string,
+ ?innerException : exn) =
+ {
+ inherit Exception(message = null, innerException = defaultArg innerException null) ; userMessage = userMessage ;
+ requestMethod = string requestMethod ; RequestUri = requestUri ; RequestBody = requestBody ;
+ StatusCode = statusCode ; ReasonPhrase = reasonPhrase ; ResponseBody = responseBody
+ }
+
+ override e.Message =
+ StringBuilder.Build(fun sb ->
+ sb.Appendfn "%s %O RequestUri=%O HttpStatusCode=%O" e.userMessage e.RequestMethod e.RequestUri e.StatusCode
+ let getBodyString str = if String.IsNullOrWhiteSpace str then "" else str
+ sb.Appendfn "RequestBody=%s" (getBodyString e.RequestBody)
+ sb.Appendfn "ResponseBody=%s" (getBodyString e.ResponseBody))
+
+ interface ISerializable with
+ member e.GetObjectData(si : SerializationInfo, sc : StreamingContext) =
+ let add name (value:obj) = si.AddValue(name, value)
+ base.GetObjectData(si, sc) ; add "userMessage" e.userMessage ;
+ add "requestUri" e.RequestUri ; add "requestMethod" e.requestMethod ; add "requestBody" e.RequestBody
+ add "statusCode" e.StatusCode ; add "reasonPhrase" e.ReasonPhrase ; add "responseBody" e.ResponseBody
+
+ new (si : SerializationInfo, sc : StreamingContext) =
+ let get name = si.GetValue(name, typeof<'a>) :?> 'a
+ {
+ inherit Exception(si, sc) ; userMessage = get "userMessage" ;
+ RequestUri = get "requestUri" ; requestMethod = get "requestMethod" ; RequestBody = get "requestBody" ;
+ StatusCode = get "statusCode" ; ReasonPhrase = get "reasonPhrase" ; ResponseBody = get "responseBody"
+ }
+
+ static member Create(userMessage : string, response : HttpResponseMessage, ?innerException : exn) = async {
+ let request = response.RequestMessage
+ let! responseBodyC = response.Content.ReadAsStringDiapered() |> Async.StartChild
+ let! requestBody = request.Content.ReadAsStringDiapered()
+ let! responseBody = responseBodyC
+ return InvalidHttpResponseException(
+ userMessage, request.Method, request.RequestUri, requestBody,
+ response.StatusCode, response.ReasonPhrase, responseBody,
+ ?innerException = innerException)
+ }
+
+ static member Create(response : HttpResponseMessage, ?innerException : exn) =
+ InvalidHttpResponseException.Create("HTTP request yielded unexpected response.", response, ?innerException = innerException)
+
+type HttpResponseMessage with
+
+ /// Raises an InvalidHttpResponseException if the response status code does not match expected value.
+ member response.EnsureStatusCode(expectedStatusCode : HttpStatusCode) = async {
+ if response.StatusCode <> expectedStatusCode then
+ let! exn = InvalidHttpResponseException.Create("Http request yielded unanticipated HTTP Result.", response)
+ do raise exn
+ }
+
+ /// Asynchronously deserializes the json response content using the supplied `deserializer`, without validating the `StatusCode`
+ /// The decoder routine to apply to the body content. Exceptions are wrapped in exceptions containing the offending content.
+ member response.InterpretContent<'Decoded>(deserializer : string -> 'Decoded) : Async<'Decoded> = async {
+ let! content = response.Content.ReadAsString()
+ try return deserializer content
+ with e ->
+ let! exn = InvalidHttpResponseException.Create("HTTP response could not be decoded.", response, e)
+ return raise exn
+ }
+
+ /// Asynchronously deserializes the json response content using the supplied `deserializer`, validating the `StatusCode` is `expectedStatusCode`
+ /// check that status code matches supplied code or raise a InvalidHttpResponseException if it doesn't.
+ /// The decoder routine to apply to the body content. Exceptions are wrapped in exceptions containing the offending content.
+ member response.Interpret<'Decoded>(expectedStatusCode : HttpStatusCode, deserializer : string -> 'Decoded) : Async<'Decoded> = async {
+ do! response.EnsureStatusCode expectedStatusCode
+ return! response.InterpretContent deserializer
+ }
+
+module HttpRes =
+
+// let codec = ECommerce.Domain.Config.EventCodec.forUnion
+ let serdes = FsCodec.SystemTextJson.Serdes(FsCodec.SystemTextJson.Options.Create())
+
+ /// Deserialize body using default Json.Net profile - throw with content details if StatusCode is unexpected or decoding fails
+ let deserializeExpectedStj<'t> expectedStatusCode (res : HttpResponseMessage) =
+ res.Interpret(expectedStatusCode, serdes.Deserialize<'t>)
+
+ /// Deserialize body using default Json.Net profile - throw with content details if StatusCode is not OK or decoding fails
+ let deserializeOkStj<'t> =
+ deserializeExpectedStj<'t> HttpStatusCode.OK
diff --git a/Sample/ECommerce.Equinox/ECommerce.FeedConsumer/Ingester.fs b/Sample/ECommerce.Equinox/ECommerce.FeedConsumer/Ingester.fs
new file mode 100644
index 000000000..f1ac96400
--- /dev/null
+++ b/Sample/ECommerce.Equinox/ECommerce.FeedConsumer/Ingester.fs
@@ -0,0 +1,63 @@
+module ECommerce.FeedConsumer.Ingester
+
+open ECommerce.FeedConsumer.Domain
+open System
+
+type Outcome = { added : int; notReady : int; dups : int }
+
+/// Gathers stats based on the outcome of each Span processed for periodic emission
+type Stats(log, statsInterval, stateInterval, logExternalStats) =
+ inherit Propulsion.Streams.Stats(log, statsInterval, stateInterval)
+
+ let mutable added, notReady, dups = 0, 0, 0
+
+ override _.HandleOk outcome =
+ added <- added + outcome.added
+ notReady <- notReady + outcome.notReady
+ dups <- dups + outcome.dups
+
+ override _.HandleExn(log, exn) =
+ log.Information(exn, "Unhandled")
+
+ override _.DumpStats() =
+ if added <> 0 || notReady <> 0 || dups <> 0 then
+ log.Information("👉Added {added} Not Yet Shipped {notReady} Duplicates {dups}", added, notReady, dups)
+ added <- 0; notReady <- 0; dups <- 0
+ base.DumpStats()
+ logExternalStats log
+
+module PipelineEvent =
+
+ type Item = { id : TicketId; payload : string }
+ let ofIndexAndItem index (item : Item) =
+ FsCodec.Core.TimelineEvent.Create(
+ index,
+ "eventType",
+ Unchecked.defaultof<_>,
+ context = item)
+ let (|ItemsForFc|_|) = function
+ | FsCodec.StreamName.Split (_, FsCodec.StreamId.Parse 2 [|_ ; FcId.Parse fc|]), (s : Propulsion.Sinks.Event[]) ->
+ Some (fc, s |> Seq.map (fun e -> Unchecked.unbox- e.Context))
+ | _ -> None
+
+let handle maxDop stream span: Async = async {
+ match stream, span with
+ | PipelineEvent.ItemsForFc (_fc, items) ->
+ // Take chunks of max 1000 in order to make handler latency be less 'lumpy'
+ // What makes sense in terms of a good chunking size will vary depending on the workload in question
+ let ticketIds = seq { for x in items -> x.id } |> Seq.truncate 1000 |> Seq.toArray
+ let maybeAccept = Seq.distinct ticketIds |> Seq.mapi (fun i _x -> async {
+ do! Async.Sleep(TimeSpan.FromSeconds 1.)
+ return if i % 3 = 1 then Some 42 else None
+ })
+ let! results = Async.Parallel(maybeAccept, maxDegreeOfParallelism=maxDop)
+ let ready = results |> Array.choose id
+ let maybeAdd = ready |> Seq.mapi (fun i _x -> async {
+ do! Async.Sleep(TimeSpan.FromSeconds 1.)
+ return if i % 2 = 1 then Some 42 else None
+ })
+ let! added = Async.Parallel(maybeAdd, maxDegreeOfParallelism=maxDop)
+ let outcome = { added = Seq.length added; notReady = results.Length - ready.Length; dups = results.Length - ticketIds.Length }
+ return Propulsion.Sinks.StreamResult.PartiallyProcessed ticketIds.Length, outcome
+ | x -> return failwithf "Unexpected stream %O" x
+}
diff --git a/Sample/ECommerce.Equinox/ECommerce.FeedConsumer/Program.fs b/Sample/ECommerce.Equinox/ECommerce.FeedConsumer/Program.fs
new file mode 100644
index 000000000..3e5d073ed
--- /dev/null
+++ b/Sample/ECommerce.Equinox/ECommerce.FeedConsumer/Program.fs
@@ -0,0 +1,136 @@
+module ECommerce.FeedConsumer.Program
+
+open Serilog
+open System
+
+type Configuration(tryGet) =
+ inherit Args.Configuration(tryGet)
+
+ let get key = match tryGet key with Some value -> value | None -> failwith $"Missing Argument/Environment Variable %s{key}"
+ member _.BaseUri = get "API_BASE_URI"
+ member _.Group = get "API_CONSUMER_GROUP"
+
+let [] AppName = "FeedConsumer"
+
+module Args =
+
+ open Argu
+
+ []
+ type Parameters =
+ | [] Verbose
+ | [] PrometheusPort of int
+
+ | [] Group of string
+ | [] BaseUri of string
+
+ | [] MaxReadAhead of int
+ | [] FcsDop of int
+ | [] TicketsDop of int
+
+ | [] Cosmos of ParseResults
+ | [] Dynamo of ParseResults
+ | [] Esdb of ParseResults
+ | [] Sss of ParseResults
+ interface IArgParserTemplate with
+ member a.Usage = a |> function
+ | Verbose -> "request verbose logging."
+ | PrometheusPort _ -> "port from which to expose a Prometheus /metrics endpoint. Default: off (optional if environment variable PROMETHEUS_PORT specified)"
+ | Group _ -> "specify Api Consumer Group Id. (optional if environment variable API_CONSUMER_GROUP specified)"
+ | BaseUri _ -> "specify Api endpoint. (optional if environment variable API_BASE_URI specified)"
+ | MaxReadAhead _ -> "maximum number of batches to let processing get ahead of completion. Default: 8."
+ | FcsDop _ -> "maximum number of FCs to process in parallel. Default: 4"
+ | TicketsDop _ -> "maximum number of Tickets to process in parallel (per FC). Default: 4"
+ | Cosmos _ -> "specify CosmosDB input parameters"
+ | Dynamo _ -> "specify DynamoDB input parameters"
+ | Esdb _ -> "specify EventStore input parameters"
+ | Sss _ -> "specify SqlStreamStore input parameters"
+
+ type Arguments(c : Configuration, a : ParseResults) =
+ member val Verbose = a.Contains Verbose
+ member val PrometheusPort = a.TryGetResult PrometheusPort |> Option.orElseWith (fun () -> c.PrometheusPort)
+ member val CacheSizeMb = 10
+ member val SourceId = a.TryGetResult Group |> Option.defaultWith (fun () -> c.Group) |> Propulsion.Feed.SourceId.parse
+ member val BaseUri = a.TryGetResult BaseUri |> Option.defaultWith (fun () -> c.BaseUri) |> Uri
+ member val MaxReadAhead = a.GetResult(MaxReadAhead,8)
+ member val FcsDop = a.TryGetResult FcsDop |> Option.defaultValue 4
+ member val TicketsDop = a.TryGetResult TicketsDop |> Option.defaultValue 4
+ member val StatsInterval = TimeSpan.FromMinutes 1.
+ member val StateInterval = TimeSpan.FromMinutes 5.
+ member val CheckpointInterval = TimeSpan.FromHours 1.
+ member val TailSleepInterval = TimeSpan.FromSeconds 1.
+ member val ConsumerGroupName = "default"
+ member val StoreArgs : Args.StoreArgs =
+ match a.TryGetSubCommand() with
+ | Some (Parameters.Cosmos cosmos) -> Args.StoreArgs.Cosmos (Args.Cosmos.Arguments(c, cosmos))
+ | Some (Parameters.Dynamo dynamo) -> Args.StoreArgs.Dynamo (Args.Dynamo.Arguments(c, dynamo))
+ | Some (Parameters.Esdb es) -> Args.StoreArgs.Esdb (Args.Esdb.Arguments(c, es))
+ | _ -> a.Raise "Must specify one of cosmos, dynamo or esdb for store"
+ member x.VerboseStore = Args.StoreArgs.verboseRequested x.StoreArgs
+ member x.DumpStoreMetrics = Args.StoreArgs.dumpMetrics x.StoreArgs
+ member x.Connect() : Store.Config * Propulsion.Feed.IFeedCheckpointStore =
+ let cache = Equinox.Cache(AppName, sizeMb = x.CacheSizeMb)
+ let createCheckpoints = Args.Checkpoints.create (x.ConsumerGroupName, x.CheckpointInterval) Store.Metrics.log
+ match x.StoreArgs with
+ | Args.StoreArgs.Cosmos a ->
+ let context = a.Connect() |> Async.RunSynchronously
+ let store = Store.Config.Cosmos (context, cache)
+ store, createCheckpoints (Args.Checkpoints.Config.Cosmos (context, cache))
+ | Args.StoreArgs.Dynamo a ->
+ let context = a.Connect()
+ let store = Store.Config.Dynamo (context, cache)
+ store, createCheckpoints (Args.Checkpoints.Config.Dynamo (context, cache))
+ | Args.StoreArgs.Esdb a ->
+ let context = a.Connect(Log.Logger, AppName, EventStore.Client.NodePreference.Leader) |> EventStoreContext.create
+ let store = Store.Config.Esdb (context, cache)
+ let checkpointStore = a.ConnectCheckpointStore(cache)
+ store, createCheckpoints checkpointStore
+ | Args.StoreArgs.Sss a ->
+ let context = a.Connect() |> SqlStreamStoreContext.create
+ let store = Store.Config.Sss (context, cache)
+ let checkpointStore = a.CreateCheckpointStoreSql(x.ConsumerGroupName)
+ store, checkpointStore
+
+ /// Parse the commandline; can throw exceptions in response to missing arguments and/or `-h`/`--help` args
+ let parse tryGetConfigValue argv =
+ let programName = Reflection.Assembly.GetEntryAssembly().GetName().Name
+ let parser = ArgumentParser.Create(programName = programName)
+ Arguments(Configuration tryGetConfigValue, parser.ParseCommandLine argv)
+
+let build (args : Args.Arguments) =
+ let _store, checkpoints = args.Connect() // TODO wireup to use store in handler
+
+ let log = Log.forGroup args.SourceId // needs to have a `group` tag for Propulsion.Streams Prometheus metrics
+ let sink =
+ let handle = Ingester.handle args.TicketsDop
+ let stats = Ingester.Stats(log, args.StatsInterval, args.StateInterval, logExternalStats = args.DumpStoreMetrics)
+ Propulsion.Sinks.Factory.StartConcurrent(log, args.MaxReadAhead, args.FcsDop, handle, stats)
+ let pumpSource =
+ let feed = ApiClient.TicketsFeed args.BaseUri
+ let source =
+ Propulsion.Feed.FeedSource(
+ log, args.StatsInterval, args.SourceId, args.TailSleepInterval,
+ checkpoints, sink)
+ source.Start(feed.ReadTranches, fun t p -> feed.Poll(t, p))
+ sink, pumpSource
+
+open Propulsion.Internal // AwaitKeyboardInterruptAsTaskCanceledException
+
+let run args = async {
+ let sink, source = build args
+ use _ = args.PrometheusPort |> Option.map startMetricsServer |> Option.toObj
+ return! [| Async.AwaitKeyboardInterruptAsTaskCanceledException()
+ source.AwaitWithStopOnCancellation()
+ sink.AwaitWithStopOnCancellation()
+ |] |> Async.Parallel |> Async.Ignore }
+
+[]
+let main argv =
+ try let args = Args.parse EnvVar.tryGet argv
+ try let metrics = Sinks.equinoxAndPropulsionFeedConsumerMetrics (Sinks.tags AppName)
+ Log.Logger <- LoggerConfiguration().Configure(args.Verbose).Sinks(metrics, args.VerboseStore).CreateLogger()
+ try run args |> Async.RunSynchronously; 0
+ with e when not (e :? System.Threading.Tasks.TaskCanceledException) -> Log.Fatal(e, "Exiting"); 2
+ finally Log.CloseAndFlush()
+ with:? Argu.ArguParseException as e -> eprintfn $"%s{e.Message}"; 1
+ | e -> eprintfn $"Exception %s{e.Message}"; 1
diff --git a/Sample/ECommerce.Equinox/ECommerce.FeedConsumer/Types.fs b/Sample/ECommerce.Equinox/ECommerce.FeedConsumer/Types.fs
new file mode 100644
index 000000000..1ad1ab9d4
--- /dev/null
+++ b/Sample/ECommerce.Equinox/ECommerce.FeedConsumer/Types.fs
@@ -0,0 +1,17 @@
+namespace ECommerce.FeedConsumer.Domain
+
+open FSharp.UMX // see https://github.com/fsprojects/FSharp.UMX - % operator and ability to apply units of measure to Guid+strings
+
+type [] fcId
+type FcId = string
+module FcId =
+ let toString (value : FcId) : string = %value
+ let parse (value : string) : FcId = let raw = value in % raw
+ let (|Parse|) = parse
+
+type [] ticketId
+type TicketId = string
+module TicketId =
+ let toString (value : TicketId) : string = %value
+ let parse (value : string) : TicketId = let raw = value in % raw
+ let (|Parse|) = parse
diff --git a/Sample/ECommerce.Equinox/ECommerce.Infrastructure/Args.fs b/Sample/ECommerce.Equinox/ECommerce.Infrastructure/Args.fs
new file mode 100644
index 000000000..e139546d7
--- /dev/null
+++ b/Sample/ECommerce.Equinox/ECommerce.Infrastructure/Args.fs
@@ -0,0 +1,288 @@
+/// Commandline arguments and/or secrets loading specifications
+module Args
+
+open System
+
+let [] REGION = "EQUINOX_DYNAMO_REGION"
+let [] SERVICE_URL = "EQUINOX_DYNAMO_SERVICE_URL"
+let [] ACCESS_KEY = "EQUINOX_DYNAMO_ACCESS_KEY_ID"
+let [] SECRET_KEY = "EQUINOX_DYNAMO_SECRET_ACCESS_KEY"
+let [] TABLE = "EQUINOX_DYNAMO_TABLE"
+let [] INDEX_TABLE = "EQUINOX_DYNAMO_TABLE_INDEX"
+
+type Configuration(tryGet : string -> string option) =
+
+ let get key = match tryGet key with Some value -> value | None -> failwith $"Missing Argument/Environment Variable %s{key}"
+ member val tryGet = tryGet
+
+ member _.CosmosConnection = get "EQUINOX_COSMOS_CONNECTION"
+ member _.CosmosDatabase = get "EQUINOX_COSMOS_DATABASE"
+ member _.CosmosContainer = get "EQUINOX_COSMOS_CONTAINER"
+
+ member _.DynamoServiceUrl = get SERVICE_URL
+ member _.DynamoAccessKey = get ACCESS_KEY
+ member _.DynamoSecretKey = get SECRET_KEY
+ member _.DynamoTable = get TABLE
+ member _.DynamoRegion = tryGet REGION
+
+ member _.EventStoreConnection = get "EQUINOX_ES_CONNECTION"
+ // member _.EventStoreCredentials = get "EQUINOX_ES_CREDENTIALS"
+ member _.MaybeEventStoreConnection = tryGet "EQUINOX_ES_CONNECTION"
+ member _.MaybeEventStoreCredentials = tryGet "EQUINOX_ES_CREDENTIALS"
+
+ member _.SqlStreamStoreConnection = get "SQLSTREAMSTORE_CONNECTION"
+ member _.SqlStreamStoreCredentials = tryGet "SQLSTREAMSTORE_CREDENTIALS"
+ member _.SqlStreamStoreCredentialsCheckpoints = tryGet "SQLSTREAMSTORE_CREDENTIALS_CHECKPOINTS"
+ member _.SqlStreamStoreDatabase = get "SQLSTREAMSTORE_DATABASE"
+ member _.SqlStreamStoreContainer = get "SQLSTREAMSTORE_CONTAINER"
+
+ member x.PrometheusPort = tryGet "PROMETHEUS_PORT" |> Option.map int
+
+// Type used to represent where checkpoints (for either the FeedConsumer position, or for a Reactor's Event Store subscription position) will be stored
+// In a typical app you don't have anything like this as you'll simply use your primary Event Store (see)
+module Checkpoints =
+
+ []
+ type Config =
+ | Cosmos of Equinox.CosmosStore.CosmosStoreContext * Equinox.Cache
+ | Dynamo of Equinox.DynamoStore.DynamoStoreContext * Equinox.Cache
+ (* Propulsion.EventStoreDb does not implement a native checkpoint storage mechanism,
+ perhaps port https://github.com/absolutejam/Propulsion.EventStoreDB ?
+ or fork/finish https://github.com/jet/dotnet-templates/pull/81
+ alternately one could use a SQL Server DB via Propulsion.SqlStreamStore
+
+ For now, we store the Checkpoints in one of the above stores as this sample uses one for the read models anyway *)
+
+ let create (consumerGroup, checkpointInterval) storeLog: Config -> Propulsion.Feed.IFeedCheckpointStore = function
+ | Config.Cosmos (context, cache) ->
+ Propulsion.Feed.ReaderCheckpoint.CosmosStore.create storeLog (consumerGroup, checkpointInterval) (context, cache)
+ | Config.Dynamo (context, cache) ->
+ Propulsion.Feed.ReaderCheckpoint.DynamoStore.create storeLog (consumerGroup, checkpointInterval) (context, cache)
+ let createCheckpointStore (group, checkpointInterval, store) : Propulsion.Feed.IFeedCheckpointStore =
+ let checkpointStore =
+ match store with
+ | Store.Config.Cosmos (context, cache) -> Config.Cosmos (context, cache)
+ | Store.Config.Dynamo (context, cache) -> Config.Dynamo (context, cache)
+ | Store.Config.Esdb _
+ | Store.Config.Memory _
+ | Store.Config.Sss _ -> failwith "unexpected"
+ create (group, checkpointInterval) Store.Metrics.log checkpointStore
+
+open Argu
+
+module Cosmos =
+
+ type [] Parameters =
+ | [] Verbose
+ | [] ConnectionMode of Microsoft.Azure.Cosmos.ConnectionMode
+ | [] Connection of string
+ | [] Database of string
+ | [] Container of string
+ | [] Timeout of float
+ | [] Retries of int
+ | [] RetriesWaitTime of float
+ interface IArgParserTemplate with
+ member p.Usage = p |> function
+ | Verbose _ -> "request verbose logging."
+ | ConnectionMode _ -> "override the connection mode. Default: Direct."
+ | Connection _ -> "specify a connection string for a Cosmos account. (optional if environment variable EQUINOX_COSMOS_CONNECTION specified)"
+ | Database _ -> "specify a database name for Cosmos store. (optional if environment variable EQUINOX_COSMOS_DATABASE specified)"
+ | Container _ -> "specify a container name for Cosmos store. (optional if environment variable EQUINOX_COSMOS_CONTAINER specified)"
+ | Timeout _ -> "specify operation timeout in seconds (default: 5)."
+ | Retries _ -> "specify operation retries (default: 1)."
+ | RetriesWaitTime _ -> "specify max wait-time for retry when being throttled by Cosmos in seconds (default: 5)"
+ type Arguments(c : Configuration, p : ParseResults) =
+ let connection = p.GetResult(Connection, fun () -> c.CosmosConnection)
+ let discovery = Equinox.CosmosStore.Discovery.ConnectionString connection
+ let mode = p.TryGetResult ConnectionMode
+ let timeout = p.GetResult(Timeout, 5.) |> TimeSpan.FromSeconds
+ let retries = p.GetResult(Retries, 1)
+ let maxRetryWaitTime = p.GetResult(RetriesWaitTime, 5.) |> TimeSpan.FromSeconds
+ let connector = Equinox.CosmosStore.CosmosStoreConnector(discovery, timeout, retries, maxRetryWaitTime, ?mode = mode)
+ let database = p.GetResult(Database, fun () -> c.CosmosDatabase)
+ let container = p.GetResult(Container, fun () -> c.CosmosContainer)
+ member val Verbose = p.Contains Verbose
+ member _.Connect() = connector.Connect("Target", database, container)
+
+module Dynamo =
+
+ type [] Parameters =
+ | [] Verbose
+ | [] RegionProfile of string
+ | [] ServiceUrl of string
+ | [] AccessKey of string
+ | [] SecretKey of string
+ | [] Table of string
+ | [] Retries of int
+ | [] RetriesTimeoutS of float
+ interface IArgParserTemplate with
+ member p.Usage = p |> function
+ | Verbose -> "Include low level Store logging."
+ | RegionProfile _ -> "specify an AWS Region (aka System Name, e.g. \"us-east-1\") to connect to using the implicit AWS SDK/tooling config and/or environment variables etc. Optional if:\n" +
+ "1) $" + REGION + " specified OR\n" +
+ "2) Explicit `ServiceUrl`/$" + SERVICE_URL + "+`AccessKey`/$" + ACCESS_KEY + "+`Secret Key`/$" + SECRET_KEY + " specified.\n" +
+ "See https://docs.aws.amazon.com/cli/latest/userguide/cli-configure-envvars.html for details"
+ | ServiceUrl _ -> "specify a server endpoint for a Dynamo account. (Not applicable if `ServiceRegion`/$" + REGION + " specified; Optional if $" + SERVICE_URL + " specified)"
+ | AccessKey _ -> "specify an access key id for a Dynamo account. (Not applicable if `ServiceRegion`/$" + REGION + " specified; Optional if $" + ACCESS_KEY + " specified)"
+ | SecretKey _ -> "specify a secret access key for a Dynamo account. (Not applicable if `ServiceRegion`/$" + REGION + " specified; Optional if $" + SECRET_KEY + " specified)"
+ | Table _ -> "specify a table name for the primary store. (optional if $" + TABLE + " specified)"
+ | Retries _ -> "specify operation retries (default: 1)."
+ | RetriesTimeoutS _ -> "specify max wait-time including retries in seconds (default: 5)"
+ type Arguments(c : Configuration, p : ParseResults) =
+ let conn = match p.TryGetResult RegionProfile |> Option.orElseWith (fun () -> c.DynamoRegion) with
+ | Some systemName ->
+ Choice1Of2 systemName
+ | None ->
+ let serviceUrl = p.TryGetResult ServiceUrl |> Option.defaultWith (fun () -> c.DynamoServiceUrl)
+ let accessKey = p.TryGetResult AccessKey |> Option.defaultWith (fun () -> c.DynamoAccessKey)
+ let secretKey = p.TryGetResult SecretKey |> Option.defaultWith (fun () -> c.DynamoSecretKey)
+ Choice2Of2 (serviceUrl, accessKey, secretKey)
+ let retries = p.GetResult(Retries, 1)
+ let timeout = p.GetResult(RetriesTimeoutS, 5.) |> TimeSpan.FromSeconds
+ let connector = match conn with
+ | Choice1Of2 systemName ->
+ Equinox.DynamoStore.DynamoStoreConnector(systemName, timeout, retries)
+ | Choice2Of2 (serviceUrl, accessKey, secretKey) ->
+ Equinox.DynamoStore.DynamoStoreConnector(serviceUrl, accessKey, secretKey, timeout, retries)
+ let table = p.TryGetResult Table |> Option.defaultWith (fun () -> c.DynamoTable)
+ member val Verbose = p.Contains Verbose
+ member _.Connect() = connector.CreateClient().CreateContext("Main", table)
+
+module Esdb =
+
+ []
+ type Parameters =
+ | [] Verbose
+ | [] Connection of string
+ | [] Credentials of string
+ | [] Timeout of float
+ | [] Retries of int
+
+ | [] Cosmos of ParseResults
+ | [] Dynamo of ParseResults
+ interface IArgParserTemplate with
+ member a.Usage = a |> function
+ | Verbose -> "Include low level Store logging."
+ | Connection _ -> "EventStore Connection String. (optional if environment variable EQUINOX_ES_CONNECTION specified)"
+ | Credentials _ -> "Credentials string for EventStore (used as part of connection string, but NOT logged). Default: use EQUINOX_ES_CREDENTIALS environment variable (or assume no credentials)"
+ | Timeout _ -> "specify operation timeout in seconds. Default: 20."
+ | Retries _ -> "specify operation retries. Default: 3."
+
+ // Feed Consumer app needs somewhere to store checkpoints
+ // Here we align with the structure of the commandline parameters for the Reactor app and also require a Dynamo or Cosmos instance to be specified
+ | Cosmos _ -> "CosmosDB (Checkpoint/Target) Store parameters (Not applicable for Web app)."
+ | Dynamo _ -> "DynamoDB (Checkpoint/Target) Store parameters (Not applicable to Web app)."
+
+ type Arguments(c : Configuration, a : ParseResults) =
+ let connectionStringLoggable = a.TryGetResult Connection |> Option.defaultWith (fun () -> c.EventStoreConnection)
+ let credentials = a.TryGetResult Credentials |> Option.orElseWith (fun () -> c.MaybeEventStoreCredentials)
+ let retries = a.GetResult(Retries, 3)
+ let timeout = a.GetResult(Timeout, 20.) |> TimeSpan.FromSeconds
+ member _.Verbose = a.Contains Verbose
+
+ member x.Connect(log : Serilog.ILogger, appName, nodePreference) : Equinox.EventStoreDb.EventStoreConnection =
+ log.Information("EventStore {discovery}", connectionStringLoggable)
+ let discovery = match credentials with Some x -> String.Join(";", connectionStringLoggable, x) | None -> connectionStringLoggable
+ |> Equinox.EventStoreDb.Discovery.ConnectionString
+ let tags=["M", Environment.MachineName; "I", Guid.NewGuid() |> string]
+ Equinox.EventStoreDb.EventStoreConnector(timeout, retries, tags = tags)
+ .Establish(appName, discovery, Equinox.EventStoreDb.ConnectionStrategy.ClusterSingle nodePreference)
+
+ member _.SecondaryStoreArgs : SecondaryStoreArgs =
+ match a.GetSubCommand() with
+ | Cosmos cosmos -> SecondaryStoreArgs.Cosmos (Cosmos.Arguments(c, cosmos))
+ | Dynamo dynamo -> SecondaryStoreArgs.Dynamo (Dynamo.Arguments(c, dynamo))
+ | _ -> a.Raise "Must specify `cosmos` or `dynamo` target store when source is `esdb`"
+
+ member x.ConnectCheckpointStore(cache) =
+ match x.SecondaryStoreArgs with
+ | SecondaryStoreArgs.Cosmos a ->
+ let context = a.Connect() |> Async.RunSynchronously
+ Checkpoints.Config.Cosmos (context, cache)
+ | SecondaryStoreArgs.Dynamo a ->
+ let context = a.Connect()
+ Checkpoints.Config.Dynamo (context, cache)
+
+ and []
+ SecondaryStoreArgs =
+ | Cosmos of Cosmos.Arguments
+ | Dynamo of Dynamo.Arguments
+
+module Sss =
+
+ // TOCONSIDER: add DB connectors other than MsSql
+ type [] Parameters =
+ | [] Connection of string
+ | [] Credentials of string
+ | [] Schema of string
+ | [] CheckpointsConnection of string
+ | [] CheckpointsCredentials of string
+ // | [] BatchSize of int
+ interface IArgParserTemplate with
+ member p.Usage = p |> function
+ | Connection _ -> "Connection string for SqlStreamStore db. Optional if SQLSTREAMSTORE_CONNECTION specified"
+ | Credentials _ -> "Credentials string for SqlStreamStore db (used as part of connection string, but NOT logged). Default: use SQLSTREAMSTORE_CREDENTIALS environment variable (or assume no credentials)"
+ | Schema _ -> "Database schema name"
+ | CheckpointsConnection _ ->"Connection string for Checkpoints sql db. Optional if SQLSTREAMSTORE_CONNECTION_CHECKPOINTS specified. Default: same as `Connection`"
+ | CheckpointsCredentials _ ->"Credentials string for Checkpoints sql db. (used as part of checkpoints connection string, but NOT logged). Default (when no `CheckpointsConnection`: use `Credentials. Default (when `CheckpointsConnection` specified): use SQLSTREAMSTORE_CREDENTIALS_CHECKPOINTS environment variable (or assume no credentials)"
+ // | BatchSize _ -> "Maximum events to request from feed. Default: 512"
+
+ type Arguments(c : Configuration, p : ParseResults) =
+ // let batchSize = p.GetResult(BatchSize, 512)
+ let connection = p.TryGetResult Connection |> Option.defaultWith (fun () -> c.SqlStreamStoreConnection)
+ let credentials = p.TryGetResult Credentials |> Option.orElseWith (fun () -> c.SqlStreamStoreCredentials) |> Option.toObj
+ let schema = p.GetResult(Schema, null)
+
+ member x.Connect() =
+ let conn, creds, schema, autoCreate = connection, credentials, schema, false
+ let sssConnectionString = String.Join(";", conn, creds)
+ Serilog.Log.Information("SqlStreamStore MsSql Connection {connectionString} Schema {schema} AutoCreate {autoCreate}", conn, schema, autoCreate)
+ let rawStore = Equinox.SqlStreamStore.MsSql.Connector(sssConnectionString, schema, autoCreate=autoCreate).Connect() |> Async.RunSynchronously
+ Equinox.SqlStreamStore.SqlStreamStoreConnection(rawStore)
+ member x.BuildCheckpointsConnectionString() =
+ let c, cs =
+ match p.TryGetResult CheckpointsConnection, p.TryGetResult CheckpointsCredentials with
+ | Some c, Some p -> c, String.Join(";", c, p)
+ | None, Some p -> let c = connection in c, String.Join(";", c, p)
+ | None, None -> let c = connection in c, String.Join(";", c, credentials)
+ | Some cc, None -> let p = c.SqlStreamStoreCredentialsCheckpoints |> Option.toObj
+ cc, String.Join(";", cc, p)
+ Serilog.Log.Information("Checkpoints MsSql Connection {connectionString}", c)
+ cs
+ member x.CreateCheckpointStoreSql(groupName) : Propulsion.Feed.IFeedCheckpointStore =
+ let connectionString = x.BuildCheckpointsConnectionString()
+ Propulsion.SqlStreamStore.ReaderCheckpoint.Service(connectionString, groupName)
+
+type []
+ StoreArgs =
+ | Cosmos of Cosmos.Arguments
+ | Dynamo of Dynamo.Arguments
+ | Esdb of Esdb.Arguments
+ | Sss of Sss.Arguments
+
+module StoreArgs =
+
+ let connectTarget targetStore cache =
+ match targetStore with
+ | StoreArgs.Cosmos a ->
+ let context = a.Connect() |> Async.RunSynchronously
+ Store.Config.Cosmos (context, cache)
+ | StoreArgs.Dynamo a ->
+ let context = a.Connect()
+ Store.Config.Dynamo (context, cache)
+ | StoreArgs.Esdb a ->
+ let context = a.Connect(Serilog.Log.Logger, "Main", EventStore.Client.NodePreference.Leader) |> EventStoreContext.create
+ Store.Config.Esdb (context, cache)
+ | StoreArgs.Sss a ->
+ let context = a.Connect() |> SqlStreamStoreContext.create
+ Store.Config.Sss (context, cache)
+ let verboseRequested = function
+ | StoreArgs.Cosmos a -> a.Verbose
+ | StoreArgs.Dynamo a -> a.Verbose
+ | StoreArgs.Esdb a -> a.Verbose
+ | StoreArgs.Sss a -> false
+ let dumpMetrics = function
+ | StoreArgs.Cosmos _ -> Equinox.CosmosStore.Core.Log.InternalMetrics.dump
+ | StoreArgs.Dynamo _ -> Equinox.DynamoStore.Core.Log.InternalMetrics.dump
+ | StoreArgs.Esdb _ -> Equinox.EventStoreDb.Log.InternalMetrics.dump
+ | StoreArgs.Sss _ -> Equinox.SqlStreamStore.Log.InternalMetrics.dump
diff --git a/Sample/ECommerce.Equinox/ECommerce.Infrastructure/ECommerce.Infrastructure.fsproj b/Sample/ECommerce.Equinox/ECommerce.Infrastructure/ECommerce.Infrastructure.fsproj
new file mode 100644
index 000000000..271393a4f
--- /dev/null
+++ b/Sample/ECommerce.Equinox/ECommerce.Infrastructure/ECommerce.Infrastructure.fsproj
@@ -0,0 +1,35 @@
+
+
+
+ net6.0
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/Sample/ECommerce.Equinox/ECommerce.Infrastructure/Infrastructure.fs b/Sample/ECommerce.Equinox/ECommerce.Infrastructure/Infrastructure.fs
new file mode 100644
index 000000000..da00aece9
--- /dev/null
+++ b/Sample/ECommerce.Equinox/ECommerce.Infrastructure/Infrastructure.fs
@@ -0,0 +1,173 @@
+[]
+module Helpers
+
+open Serilog
+open System
+
+module Log =
+
+ let isStoreMetrics x = Serilog.Filters.Matching.WithProperty("isMetric").Invoke x
+ let forGroup group = Log.ForContext("group", group)
+
+module EnvVar =
+
+ let tryGet varName : string option = Environment.GetEnvironmentVariable varName |> Option.ofObj
+
+type Equinox.CosmosStore.CosmosStoreContext with
+
+ member x.LogConfiguration(role, databaseId: string, containerId: string) =
+ Log.Information("CosmosStore {role:l} {db}/{container} Tip maxEvents {maxEvents} maxSize {maxJsonLen} Query maxItems {queryMaxItems}",
+ role, databaseId, containerId, x.TipOptions.MaxEvents, x.TipOptions.MaxJsonLength, x.QueryOptions.MaxItems)
+
+type Equinox.CosmosStore.CosmosStoreClient with
+
+ member x.CreateContext(role: string, databaseId, containerId, tipMaxEvents, ?queryMaxItems, ?tipMaxJsonLength, ?skipLog) =
+ let c = Equinox.CosmosStore.CosmosStoreContext(x, databaseId, containerId, tipMaxEvents, ?queryMaxItems = queryMaxItems, ?tipMaxJsonLength = tipMaxJsonLength)
+ if skipLog = Some true then () else c.LogConfiguration(role, databaseId, containerId)
+ c
+
+module CosmosStoreConnector =
+
+ let private get (role: string) (client: Microsoft.Azure.Cosmos.CosmosClient) databaseId containerId =
+ Log.Information("CosmosDB {role} Database {database} Container {container}", role, databaseId, containerId)
+ client.GetDatabase(databaseId).GetContainer(containerId)
+ let getSource c = get "Source" c
+ let getLeases c = get "Leases" c
+ let getSourceAndLeases client databaseId containerId auxContainerId =
+ getSource client databaseId containerId, getLeases client databaseId auxContainerId
+
+type Equinox.CosmosStore.CosmosStoreConnector with
+
+ member private x.LogConfiguration(role, databaseId: string, containers: string[]) =
+ let o = x.Options
+ let timeout, retries429, timeout429 = o.RequestTimeout, o.MaxRetryAttemptsOnRateLimitedRequests, o.MaxRetryWaitTimeOnRateLimitedRequests
+ Log.Information("CosmosDB {role} {mode} {endpointUri} {db} {containers} timeout {timeout}s Throttling retries {retries}, max wait {maxRetryWaitTime}s",
+ role, o.ConnectionMode, x.Endpoint, databaseId, containers, timeout.TotalSeconds, retries429, let t = timeout429.Value in t.TotalSeconds)
+ member private x.CreateAndInitialize(role, databaseId, containers) =
+ x.LogConfiguration(role, databaseId, containers)
+ x.CreateAndInitialize(databaseId, containers)
+ // member private x.Connect(role, databaseId, containers) =
+ // x.LogConfiguration(role, databaseId, containers)
+ // x.Connect(databaseId, containers)
+ member private x.ConnectContexts(role, databaseId, containerId, ?auxContainerId): Async<_ * Equinox.CosmosStore.CosmosStoreContext> = async {
+ let! cosmosClient = x.CreateAndInitialize(role, databaseId, [| yield containerId; yield! Option.toList auxContainerId |])
+ let client = Equinox.CosmosStore.CosmosStoreClient(cosmosClient)
+ let contexts = client.CreateContext(role, databaseId, containerId, tipMaxEvents = 256, queryMaxItems = 100)
+ return cosmosClient, contexts }
+ /// Connect to the database (including verifying and warming up relevant containers), establish relevant CosmosStoreContexts required by Domain
+ member x.Connect(role, databaseId, containerId: string) = async {
+ let! _client, contexts = x.ConnectContexts(role, databaseId, containerId)
+ return contexts }
+ member x.ConnectWithFeed(databaseId, containerId, auxContainerId) = async {
+ let! client, context = x.ConnectContexts("Main", databaseId, containerId, auxContainerId)
+ let source, leases = CosmosStoreConnector.getSourceAndLeases client databaseId containerId auxContainerId
+ return context, source, leases }
+
+type Equinox.DynamoStore.DynamoStoreConnector with
+
+ member x.LogConfiguration() =
+ Log.Information("DynamoStore {endpoint} Timeout {timeoutS}s Retries {retries}",
+ x.Endpoint, (let t = x.Timeout in t.TotalSeconds), x.Retries)
+
+ member x.CreateClient() =
+ x.LogConfiguration()
+ x.CreateDynamoStoreClient()
+
+type Equinox.DynamoStore.DynamoStoreClient with
+
+ member x.CreateContext(role, table, ?queryMaxItems, ?maxBytes, ?archiveTableName: string) =
+ let queryMaxItems = defaultArg queryMaxItems 100
+ let c = Equinox.DynamoStore.DynamoStoreContext(x, table, queryMaxItems = queryMaxItems, ?maxBytes = maxBytes, ?archiveTableName = archiveTableName)
+ Log.Information("DynamoStore {role:l} Table {table} Archive {archive} Tip thresholds: {maxTipBytes}b {maxTipEvents}e Query paging {queryMaxItems} items",
+ role, table, Option.toObj archiveTableName, c.TipOptions.MaxBytes, Option.toNullable c.TipOptions.MaxEvents, c.QueryOptions.MaxItems)
+ c
+
+type Equinox.DynamoStore.DynamoStoreContext with
+
+ member context.CreateCheckpointService(consumerGroupName, cache, log, ?checkpointInterval) =
+ let checkpointInterval = defaultArg checkpointInterval (TimeSpan.FromHours 1.)
+ Propulsion.Feed.ReaderCheckpoint.DynamoStore.create log (consumerGroupName, checkpointInterval) (context, cache)
+
+module EventStoreContext =
+
+ let create (storeConnection : Equinox.EventStoreDb.EventStoreConnection) =
+ Equinox.EventStoreDb.EventStoreContext(storeConnection, batchSize = 200)
+
+module SqlStreamStoreContext =
+
+ let create (storeConnection : Equinox.SqlStreamStore.SqlStreamStoreConnection) =
+ Equinox.SqlStreamStore.SqlStreamStoreContext(storeConnection, batchSize = 200)
+
+/// Equinox and Propulsion provide metrics as properties in log emissions
+/// These helpers wire those to pass through virtual Log Sinks that expose them as Prometheus metrics.
+module Sinks =
+
+ let tags appName = ["app", appName]
+
+ let equinoxMetricsOnly tags (l : LoggerConfiguration) =
+ l.WriteTo.Sink(Equinox.CosmosStore.Core.Log.InternalMetrics.Stats.LogSink())
+ .WriteTo.Sink(Equinox.CosmosStore.Prometheus.LogSink(tags))
+ .WriteTo.Sink(Equinox.DynamoStore.Core.Log.InternalMetrics.Stats.LogSink())
+ .WriteTo.Sink(Equinox.DynamoStore.Prometheus.LogSink(tags))
+ .WriteTo.Sink(Equinox.EventStoreDb.Log.InternalMetrics.Stats.LogSink())
+
+ let equinoxAndPropulsionMetrics tags (l : LoggerConfiguration) =
+ l |> equinoxMetricsOnly tags
+ |> _.WriteTo.Sink(Propulsion.Prometheus.LogSink(tags))
+
+ let equinoxAndPropulsionReactorMetrics tags (l : LoggerConfiguration) =
+ l |> equinoxAndPropulsionMetrics tags
+ |> _.WriteTo.Sink(Propulsion.CosmosStore.Prometheus.LogSink(tags))
+ .WriteTo.Sink(Propulsion.Feed.Prometheus.LogSink(tags)) // Esdb and Dynamo indirectly provide metrics via Feed
+
+ let equinoxAndPropulsionFeedConsumerMetrics tags (l : LoggerConfiguration) =
+ l |> equinoxAndPropulsionMetrics tags
+ |> _.WriteTo.Sink(Propulsion.Feed.Prometheus.LogSink(tags))
+
+ let console (configuration : LoggerConfiguration) =
+ let t = "[{Timestamp:HH:mm:ss} {Level:u3}] {Message:lj} {NewLine}{Exception}"
+ configuration.WriteTo.Console(theme=Sinks.SystemConsole.Themes.AnsiConsoleTheme.Code, outputTemplate=t)
+
+[]
+type Logging() =
+
+ []
+ static member Configure(configuration : LoggerConfiguration, ?verbose) =
+ configuration
+ .Destructure.FSharpTypes()
+ .Enrich.FromLogContext()
+ |> fun c -> if verbose = Some true then c.MinimumLevel.Debug() else c
+
+ []
+ static member private Sinks(configuration : LoggerConfiguration, configureMetricsSinks, configureConsoleSink, ?isMetric) =
+ let configure (a : Configuration.LoggerSinkConfiguration) : unit =
+ a.Logger(configureMetricsSinks >> ignore) |> ignore // unconditionally feed all log events to the metrics sinks
+ a.Logger(fun l -> // but filter what gets emitted to the console sink
+ let l = match isMetric with None -> l | Some predicate -> l.Filter.ByExcluding(Func predicate)
+ configureConsoleSink l |> ignore)
+ |> ignore
+ configuration.WriteTo.Async(bufferSize = 65536, blockWhenFull = true, configure = System.Action<_> configure)
+
+ []
+ static member Sinks(configuration : LoggerConfiguration, configureMetricsSinks, verboseStore) =
+ configuration.Sinks(configureMetricsSinks, Sinks.console, ?isMetric = if verboseStore then None else Some Log.isStoreMetrics)
+
+/// A typical app will likely have health checks etc, implying the wireup would be via `UseMetrics()` and thus not use this ugly code directly
+let startMetricsServer port : IDisposable =
+ let metricsServer = new Prometheus.KestrelMetricServer(port = port)
+ let ms = metricsServer.Start()
+ Log.Information("Prometheus /metrics endpoint on port {port}", port)
+ { new IDisposable with member x.Dispose() = ms.Stop(); (metricsServer :> IDisposable).Dispose() }
+
+module Exception =
+
+ let dump verboseStore (log : ILogger) (exn : exn) =
+ match exn with
+ | :? Microsoft.Azure.Cosmos.CosmosException as e
+ when (e.StatusCode = System.Net.HttpStatusCode.TooManyRequests
+ || e.StatusCode = System.Net.HttpStatusCode.ServiceUnavailable)
+ && not verboseStore -> ()
+ | Equinox.DynamoStore.Exceptions.ProvisionedThroughputExceeded
+ | :? TimeoutException when not verboseStore -> ()
+ | _ ->
+ log.Information(exn, "Unhandled")
diff --git a/Sample/ECommerce.Equinox/ECommerce.Infrastructure/SourceConfig.fs b/Sample/ECommerce.Equinox/ECommerce.Infrastructure/SourceConfig.fs
new file mode 100644
index 000000000..a8553efd2
--- /dev/null
+++ b/Sample/ECommerce.Equinox/ECommerce.Infrastructure/SourceConfig.fs
@@ -0,0 +1,114 @@
+namespace global
+
+open System
+open System.Threading.Tasks
+
+[]
+type SourceConfig =
+ | Memory of store : Equinox.MemoryStore.VolatileStore)>
+ | Cosmos of monitoredContainer: Microsoft.Azure.Cosmos.Container
+ * leasesContainer: Microsoft.Azure.Cosmos.Container
+ * checkpoints: CosmosFeedConfig
+ * tailSleepInterval: TimeSpan
+ * statsInterval: TimeSpan
+ | Dynamo of indexContext: Equinox.DynamoStore.DynamoStoreContext
+ * checkpoints: Propulsion.Feed.IFeedCheckpointStore
+ * loading: Propulsion.DynamoStore.EventLoadMode
+ * startFromTail: bool
+ * batchSizeCutoff: int
+ * tailSleepInterval: TimeSpan
+ * statsInterval: TimeSpan
+ | Esdb of client : EventStore.Client.EventStoreClient
+ * checkpoints : Propulsion.Feed.IFeedCheckpointStore
+ * hydrateBodies : bool
+ * startFromTail : bool
+ * batchSize : int
+ * tailSleepInterval : TimeSpan
+ * statsInterval : TimeSpan
+ | Sss of client : SqlStreamStore.IStreamStore
+ * checkpoints : Propulsion.Feed.IFeedCheckpointStore
+ * hydrateBodies : bool
+ * startFromTail : bool
+ * batchSize : int
+ * tailSleepInterval : TimeSpan
+ * statsInterval : TimeSpan
+and [] CosmosFeedConfig =
+ | Ephemeral of processorName : string
+ | Persistent of processorName : string * startFromTail : bool * maxItems : int option * lagFrequency : TimeSpan
+and [] DynamoLoadModeConfig =
+ | Hydrate of monitoredContext : Equinox.DynamoStore.DynamoStoreContext * hydrationConcurrency : int
+
+module SourceConfig =
+ module Memory =
+ open Propulsion.MemoryStore
+ let start log (sink: Propulsion.Sinks.SinkPipeline) (categories: string[])
+ (store: Equinox.MemoryStore.VolatileStore<_>): Propulsion.Pipeline * (TimeSpan -> Task) =
+ let source = MemoryStoreSource(log, store, categories, sink)
+ source.Start(), fun _propagationDelay -> source.Monitor.AwaitCompletion(ignoreSubsequent = false)
+ module Cosmos =
+ open Propulsion.CosmosStore
+ let start log (sink: Propulsion.Sinks.SinkPipeline) categories
+ (monitoredContainer, leasesContainer, checkpointConfig, tailSleepInterval, statsInterval): Propulsion.Pipeline * (TimeSpan -> Task) =
+ let parseFeedDoc = EquinoxSystemTextJsonParser.ofCategories categories
+ let source =
+ match checkpointConfig with
+ | Ephemeral processorName ->
+ let withStartTime1sAgo (x: Microsoft.Azure.Cosmos.ChangeFeedProcessorBuilder) =
+ x.WithStartTime(let t = DateTime.UtcNow in t.AddSeconds -1.)
+ let lagFrequency = TimeSpan.FromMinutes 1.
+ CosmosStoreSource(log, statsInterval, monitoredContainer, leasesContainer, processorName, parseFeedDoc, sink,
+ startFromTail = true, customize = withStartTime1sAgo, tailSleepInterval = tailSleepInterval,
+ lagEstimationInterval = lagFrequency).Start()
+ | Persistent (processorName, startFromTail, maxItems, lagFrequency) ->
+ CosmosStoreSource(log, statsInterval, monitoredContainer, leasesContainer, processorName, parseFeedDoc, sink,
+ startFromTail = startFromTail, ?maxItems = maxItems, tailSleepInterval = tailSleepInterval,
+ lagEstimationInterval = lagFrequency).Start()
+ source, fun propagationDelay -> source.Monitor.AwaitCompletion(propagationDelay, ignoreSubsequent = false)
+ module Dynamo =
+ open Propulsion.DynamoStore
+ let create (log, storeLog) (sink: Propulsion.Sinks.SinkPipeline) categories
+ (indexContext, checkpoints, loadMode, startFromTail, batchSizeCutoff, tailSleepInterval, statsInterval) trancheIds =
+ DynamoStoreSource(
+ log, statsInterval,
+ indexContext, batchSizeCutoff, tailSleepInterval,
+ checkpoints, sink, loadMode, categories = categories,
+ startFromTail = startFromTail, storeLog = storeLog, ?trancheIds = trancheIds)
+ let start (log, storeLog) sink categories (indexContext, checkpoints, loadMode, startFromTail, batchSizeCutoff, tailSleepInterval, statsInterval)
+ : Propulsion.Pipeline * (TimeSpan -> Task) =
+ let source = create (log, storeLog) sink categories (indexContext, checkpoints, loadMode, startFromTail, batchSizeCutoff, tailSleepInterval, statsInterval) None
+ let source = source.Start()
+ source, fun propagationDelay -> source.Monitor.AwaitCompletion(propagationDelay, ignoreSubsequent = false)
+ module Esdb =
+ open Propulsion.EventStoreDb
+ let start log (sink: Propulsion.Sinks.SinkPipeline) categories
+ (client, checkpoints, withData, startFromTail, batchSize, tailSleepInterval, statsInterval): Propulsion.Pipeline * (TimeSpan -> Task) =
+ let source =
+ EventStoreSource(
+ log, statsInterval,
+ client, batchSize, tailSleepInterval,
+ checkpoints, sink, categories, withData = withData, startFromTail = startFromTail)
+ let source = source.Start()
+ source, fun propagationDelay -> source.Monitor.AwaitCompletion(propagationDelay, ignoreSubsequent = false)
+ module Sss =
+ open Propulsion.SqlStreamStore
+ let start log (sink: Propulsion.Sinks.SinkPipeline) categoryFilter
+ (client, checkpoints, withData, startFromTail, batchSize, tailSleepInterval, statsInterval) : Propulsion.Pipeline * (TimeSpan -> Task) =
+ let source =
+ SqlStreamStoreSource(
+ log, statsInterval,
+ client, batchSize, tailSleepInterval,
+ checkpoints, sink, categoryFilter, withData = withData, startFromTail = startFromTail)
+ let source = source.Start()
+ source, fun propagationDelay -> source.Monitor.AwaitCompletion(propagationDelay, ignoreSubsequent = false)
+
+ let start (log, storeLog) sink categories: SourceConfig -> Propulsion.Pipeline * (TimeSpan -> Task) = function
+ | SourceConfig.Memory volatileStore ->
+ Memory.start log sink categories volatileStore
+ | SourceConfig.Cosmos (monitored, leases, checkpointConfig, tailSleepInterval, statsInterval) ->
+ Cosmos.start log sink categories (monitored, leases, checkpointConfig, tailSleepInterval, statsInterval)
+ | SourceConfig.Dynamo (indexContext, checkpoints, loadMode, startFromTail, batchSizeCutoff, tailSleepInterval, statsInterval) ->
+ Dynamo.start (log, storeLog) sink categories (indexContext, checkpoints, loadMode, startFromTail, batchSizeCutoff, tailSleepInterval, statsInterval)
+ | SourceConfig.Esdb (client, checkpoints, hydrateBodies, startFromTail, batchSize, tailSleepInterval, statsInterval) ->
+ Esdb.start log sink categories (client, checkpoints, hydrateBodies, startFromTail, batchSize, tailSleepInterval, statsInterval)
+ | SourceConfig.Sss (client, checkpoints, hydrateBodies, startFromTail, batchSize, tailSleepInterval, statsInterval) ->
+ Sss.start log sink categories (client, checkpoints, hydrateBodies, startFromTail, batchSize, tailSleepInterval, statsInterval)
diff --git a/Sample/ECommerce.Equinox/ECommerce.Reactor/ConfirmedHandler.fs b/Sample/ECommerce.Equinox/ECommerce.Reactor/ConfirmedHandler.fs
new file mode 100644
index 000000000..979889ec8
--- /dev/null
+++ b/Sample/ECommerce.Equinox/ECommerce.Reactor/ConfirmedHandler.fs
@@ -0,0 +1,19 @@
+module ECommerce.Reactor.ConfirmedHandler
+
+open ECommerce.Domain
+
+type Service internal (carts : ShoppingCart.Service, ingester : ConfirmedIngester.Service) =
+
+ member _.TrySummarizeConfirmed(cartId) : Async = async {
+ let! cartSummary, originEpoch = carts.SummarizeWithOriginEpoch(cartId, ingester.ActiveIngestionEpochId)
+ return! ingester.TryIngestCartSummary(originEpoch, ConfirmedEpoch.ofShoppingCartView cartId cartSummary) }
+
+module Config =
+
+ let create_ linger (sourceStore, cosmosStore) =
+ let carts = ShoppingCart.Config.create sourceStore
+ let ingester = ConfirmedIngester.Config.create linger cosmosStore
+ Service(carts, ingester)
+ let create store =
+ let defaultLinger = System.TimeSpan.FromMilliseconds 200.
+ create_ defaultLinger store
diff --git a/Sample/ECommerce.Equinox/ECommerce.Reactor/ECommerce.Reactor.fsproj b/Sample/ECommerce.Equinox/ECommerce.Reactor/ECommerce.Reactor.fsproj
new file mode 100644
index 000000000..41a5fce6f
--- /dev/null
+++ b/Sample/ECommerce.Equinox/ECommerce.Reactor/ECommerce.Reactor.fsproj
@@ -0,0 +1,24 @@
+
+
+
+ Exe
+ net6.0
+ 5
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/Sample/ECommerce.Equinox/ECommerce.Reactor/Program.fs b/Sample/ECommerce.Equinox/ECommerce.Reactor/Program.fs
new file mode 100644
index 000000000..cd65852e1
--- /dev/null
+++ b/Sample/ECommerce.Equinox/ECommerce.Reactor/Program.fs
@@ -0,0 +1,140 @@
+module ECommerce.Reactor.Program
+
+open Serilog
+open System
+open System.Threading.Tasks
+
+let [] AppName = "ECommerce.Reactor"
+
+module Args =
+
+ open Argu
+
+ []
+ type Parameters =
+ | [] Verbose
+ | [] PrometheusPort of int
+ | [] ConsumerGroupName of string
+ | [] MaxReadAhead of int
+ | [] MaxWriters of int
+ | [] StateIntervalM of float
+ | [] IdleDelayMs of int
+ | [] WakeForResults
+ | [] Cosmos of ParseResults
+ | [] Dynamo of ParseResults
+ | [] Esdb of ParseResults
+ interface IArgParserTemplate with
+ member a.Usage = a |> function
+ | Verbose -> "request Verbose Logging. Default: off."
+ | PrometheusPort _ -> "port from which to expose a Prometheus /metrics endpoint. Default: off (optional if environment variable PROMETHEUS_PORT specified)"
+ | ConsumerGroupName _ -> "Projector consumer group name."
+ | MaxReadAhead _ -> "maximum number of batches to let processing get ahead of completion. Default: 16."
+ | MaxWriters _ -> "maximum number of concurrent streams on which to process at any time. Default: 8."
+ | StateIntervalM _ -> "Interval at which to report Propulsion Statistics. Default: 10"
+ | IdleDelayMs _ -> "Idle delay for scheduler. Default 1000ms"
+ | WakeForResults -> "Wake for all results to provide optimal throughput"
+ | Cosmos _ -> "specify CosmosDB input parameters."
+ | Dynamo _ -> "specify DynamoDB input parameters."
+ | Esdb _ -> "specify EventStoreDB input parameters."
+
+ type Arguments(c : SourceArgs.Configuration, p : ParseResults) =
+ let maxReadAhead = p.GetResult(MaxReadAhead, 2)
+ let maxConcurrentProcessors = p.GetResult(MaxWriters, 8)
+ let consumerGroupName = p.GetResult ConsumerGroupName
+ member _.ProcessorParams() = Log.Information("Reacting... {consumerGroupName}, reading {maxReadAhead} ahead, {dop} writers",
+ consumerGroupName, maxReadAhead, maxConcurrentProcessors)
+ (consumerGroupName, maxReadAhead, maxConcurrentProcessors)
+ member val Verbose = p.Contains Verbose
+ member val PrometheusPort = p.TryGetResult PrometheusPort |> Option.orElseWith (fun () -> c.PrometheusPort)
+ member val CacheSizeMb = 10
+ member val StatsInterval = TimeSpan.FromMinutes 1.
+ member val StateInterval = p.GetResult(StateIntervalM, 10.) |> TimeSpan.FromMinutes
+ member val PurgeInterval = TimeSpan.FromHours 1.
+ member val IdleDelay = p.GetResult(IdleDelayMs, 1000) |> TimeSpan.FromMilliseconds
+ member val WakeForResults = p.Contains WakeForResults
+ member val Store : Choice =
+ match p.GetSubCommand() with
+ | Cosmos a -> Choice1Of3 <| SourceArgs.Cosmos.Arguments(c, a)
+ | Dynamo a -> Choice2Of3 <| SourceArgs.Dynamo.Arguments(c, a)
+ | Esdb a -> Choice3Of3 <| SourceArgs.Esdb.Arguments(c, a)
+ | a -> p.Raise $"Unexpected Store subcommand %A{a}"
+ member x.VerboseStore = match x.Store with
+ | Choice1Of3 s -> s.Verbose
+ | Choice2Of3 s -> s.Verbose
+ | Choice3Of3 s -> s.Verbose
+ member x.DumpStoreMetrics = match x.Store with
+ | Choice1Of3 _ -> Equinox.CosmosStore.Core.Log.InternalMetrics.dump
+ | Choice2Of3 _ -> Equinox.DynamoStore.Core.Log.InternalMetrics.dump
+ | Choice3Of3 _ -> Equinox.EventStoreDb.Log.InternalMetrics.dump
+
+ member x.ConnectStoreSourceAndTarget() : Store.Config * (ILogger -> string -> SourceConfig) * Store.Config=
+ let cache = Equinox.Cache (AppName, sizeMb = x.CacheSizeMb)
+ match x.Store with
+ | Choice1Of3 a ->
+ let context, monitored, leases = a.ConnectWithFeed() |> Async.RunSynchronously
+ let buildSourceConfig _log groupName =
+ let startFromTail, maxItems, tailSleepInterval, lagFrequency = a.MonitoringParams
+ let checkpointConfig = CosmosFeedConfig.Persistent (groupName, startFromTail, maxItems, lagFrequency)
+ SourceConfig.Cosmos (monitored, leases, checkpointConfig, tailSleepInterval, x.StatsInterval)
+ let store = Store.Config.Cosmos (context, cache)
+ store, buildSourceConfig, store
+ | Choice2Of3 a ->
+ let context = a.Connect()
+ let buildSourceConfig log groupName =
+ let indexContext, startFromTail, batchSizeCutoff, tailSleepInterval, streamsDop = a.MonitoringParams(log)
+ let checkpoints = a.CreateCheckpointStore(groupName, cache)
+ let load = Propulsion.DynamoStore.WithData (streamsDop, context)
+ SourceConfig.Dynamo (indexContext, checkpoints, load, startFromTail, batchSizeCutoff, tailSleepInterval, x.StatsInterval)
+ let store = Store.Config.Dynamo (context, cache)
+ store, buildSourceConfig, store
+ | Choice3Of3 a ->
+ let connection = a.Connect(Log.Logger, AppName, EventStore.Client.NodePreference.Leader)
+ let context = connection |> EventStoreContext.create
+ let store = Store.Config.Esdb (context, cache)
+ let targetStore = a.ConnectTarget(cache)
+ let buildSourceConfig log groupName =
+ let startFromTail, maxItems, tailSleepInterval = a.MonitoringParams(log)
+ let checkpoints = a.CreateCheckpointStore(groupName, targetStore)
+ let hydrateBodies = true
+ SourceConfig.Esdb (connection.ReadConnection, checkpoints, hydrateBodies, startFromTail, maxItems, tailSleepInterval, x.StatsInterval)
+ store, buildSourceConfig, targetStore
+
+ /// Parse the commandline; can throw exceptions in response to missing arguments and/or `-h`/`--help` args
+ let parse tryGetConfigValue argv : Arguments =
+ let programName = System.Reflection.Assembly.GetEntryAssembly().GetName().Name
+ let parser = ArgumentParser.Create(programName = programName)
+ Arguments(SourceArgs.Configuration tryGetConfigValue, parser.ParseCommandLine argv)
+
+open Propulsion.Internal // AwaitKeyboardInterruptAsTaskCanceledException
+
+let build (args : Args.Arguments) =
+ let consumerGroupName, maxReadAhead, maxConcurrentStreams = args.ProcessorParams()
+ let store, buildSourceConfig, targetStore = args.ConnectStoreSourceAndTarget()
+ let log = Log.forGroup consumerGroupName // needs to have a `group` tag for Propulsion.Streams Prometheus metrics
+ let handle = Reactor.Config.create (store, targetStore)
+ let stats = Reactor.Stats(log, args.StatsInterval, args.StateInterval, args.VerboseStore, logExternalStats = args.DumpStoreMetrics)
+ let sink = Reactor.Config.StartSink(log, stats, handle, maxReadAhead, maxConcurrentStreams,
+ wakeForResults = args.WakeForResults, idleDelay = args.IdleDelay, purgeInterval = args.PurgeInterval)
+ let source, _awaitSource =
+ let sourceConfig = buildSourceConfig log consumerGroupName
+ Reactor.Config.StartSource(log, sink, sourceConfig)
+ sink, source
+
+let run (args : Args.Arguments) = async {
+ use _ = args.PrometheusPort |> Option.map startMetricsServer |> Option.toObj
+ let sink, source = build args
+ return! [| Async.AwaitKeyboardInterruptAsTaskCanceledException()
+ source.AwaitWithStopOnCancellation()
+ sink.AwaitWithStopOnCancellation()
+ |] |> Async.Parallel |> Async.Ignore }
+
+[]
+let main argv =
+ try let args = Args.parse EnvVar.tryGet argv
+ let metrics = Sinks.tags AppName |> Sinks.equinoxAndPropulsionReactorMetrics
+ try Log.Logger <- LoggerConfiguration().Configure(verbose=args.Verbose).Sinks(metrics, args.VerboseStore).CreateLogger()
+ try run args |> Async.RunSynchronously; 0
+ with e when not (e :? TaskCanceledException) -> Log.Fatal(e, "Exiting"); 2
+ finally Log.CloseAndFlush()
+ with:? Argu.ArguParseException as e -> eprintfn $"%s{e.Message}"; 1
+ | e -> eprintfn $"Exception %s{e.Message}"; 1
diff --git a/Sample/ECommerce.Equinox/ECommerce.Reactor/Reactor.fs b/Sample/ECommerce.Equinox/ECommerce.Reactor/Reactor.fs
new file mode 100644
index 000000000..f6081741e
--- /dev/null
+++ b/Sample/ECommerce.Equinox/ECommerce.Reactor/Reactor.fs
@@ -0,0 +1,65 @@
+module ECommerce.Reactor.Reactor
+
+open ECommerce.Domain
+open Metrics
+
+/// Gathers stats based on the outcome of each Span processed for emission, at intervals controlled by `StreamsConsumer`
+type Stats(log, statsInterval, stateInterval, verboseStore, ?logExternalStats) =
+ inherit Propulsion.Streams.Stats(log, statsInterval, stateInterval)
+
+ let mutable ok, skipped, na = 0, 0, 0
+
+ override _.HandleOk res =
+ observeReactorOutcome res
+ match res with
+ | Outcome.Ok (used, unused) -> ok <- ok + used; skipped <- skipped + unused
+ | Outcome.Skipped count -> skipped <- skipped + count
+ | Outcome.NotApplicable count -> na <- na + count
+ override _.HandleExn(log, exn) =
+ Exception.dump verboseStore log exn
+
+ override _.DumpStats() =
+ if ok <> 0 || skipped <> 0 || na <> 0 then
+ log.Information(" used {ok} skipped {skipped} n/a {na}", ok, skipped, na)
+ ok <- 0; skipped <- 0; na <- 0
+ match logExternalStats with None -> () | Some f -> f Serilog.Log.Logger
+ base.DumpStats()
+
+let reactionCategories = [| ShoppingCart.CategoryName |]
+
+let handle
+ (cartSummary : ShoppingCartSummaryHandler.Service)
+ (confirmedCarts : ConfirmedHandler.Service)
+ stream span : Async = async {
+ match struct (stream, span) with
+ | ShoppingCart.Reactions.Decode (cartId, events) ->
+ match events with
+ | ShoppingCart.Reactions.Confirmed ->
+ let! _done = confirmedCarts.TrySummarizeConfirmed(cartId) in ()
+ | _ -> ()
+ match events with
+ | ShoppingCart.Reactions.StateChanged ->
+ let! worked, version' = cartSummary.TryIngestSummary(cartId)
+ let outcome = if worked then Outcome.Ok (1, Array.length span - 1) else Outcome.Skipped span.Length
+ return Propulsion.Sinks.StreamResult.OverrideNextIndex version', outcome
+ | _ -> return Propulsion.Sinks.StreamResult.AllProcessed, Outcome.NotApplicable span.Length
+ | x -> return failwith $"Invalid event %A{x}" } // should be filtered by isReactionStream
+
+module Config =
+
+ let create (sourceStore, targetStore) =
+ let cartSummary = ShoppingCartSummaryHandler.Config.create (sourceStore, targetStore)
+ let confirmedCarts = ConfirmedHandler.Config.create (sourceStore, targetStore)
+ handle cartSummary confirmedCarts
+
+type Config private () =
+
+ static member StartSink(log : Serilog.ILogger, stats : Stats,
+ handle,// : (FsCodec.StreamName * Propulsion.Sinks.Event[]) ->
+ // Async<(Propulsion.Sinks.StreamResult * Outcome)>,
+ maxReadAhead : int, maxConcurrentStreams : int, ?wakeForResults, ?idleDelay, ?purgeInterval) =
+ Propulsion.Sinks.Factory.StartConcurrent(log, maxReadAhead, maxConcurrentStreams, handle, stats,
+ ?wakeForResults = wakeForResults, ?idleDelay = idleDelay, ?purgeInterval = purgeInterval)
+
+ static member StartSource(log, sink, sourceConfig) =
+ SourceConfig.start (log, Store.Metrics.log) sink reactionCategories sourceConfig
diff --git a/Sample/ECommerce.Equinox/ECommerce.Reactor/ReactorMetrics.fs b/Sample/ECommerce.Equinox/ECommerce.Reactor/ReactorMetrics.fs
new file mode 100644
index 000000000..5a167d558
--- /dev/null
+++ b/Sample/ECommerce.Equinox/ECommerce.Reactor/ReactorMetrics.fs
@@ -0,0 +1,31 @@
+module ECommerce.Reactor.Metrics
+
+let baseName stat = "Ecommerce_reactor_" + stat
+let baseDesc desc = "Ecommerce: Reactor " + desc
+
+module private Counter =
+
+ let private make (config : Prometheus.CounterConfiguration) name desc =
+ let ctr = Prometheus.Metrics.CreateCounter(name, desc, config)
+ fun tagValues (c : float) -> ctr.WithLabels(tagValues).Inc(c)
+
+ let create (tagNames, tagValues) stat desc =
+ let config = Prometheus.CounterConfiguration(LabelNames = tagNames)
+ make config (baseName stat) (baseDesc desc) tagValues
+
+let observeOutcomeStatus s = Counter.create ([| "status" |],[| s |]) "outcome_total" "Outcome"
+
+[]
+type Outcome =
+ /// Handler processed the span, with counts of used vs unused known event types
+ | Ok of used : int * unused : int
+ /// Handler processed the span, but idempotency checks resulted in no writes being applied; includes count of decoded events
+ | Skipped of count : int
+ /// Handler determined the events were not relevant to its duties and performed no actions
+ /// e.g. wrong category, events that dont imply a state change
+ | NotApplicable of count : int
+
+let observeReactorOutcome = function
+ | Outcome.Ok (used, _)-> observeOutcomeStatus "ok" (float used)
+ | Outcome.Skipped c -> observeOutcomeStatus "ignored" (float c)
+ | Outcome.NotApplicable c -> observeOutcomeStatus "handled" (float c)
diff --git a/Sample/ECommerce.Equinox/ECommerce.Reactor/ShoppingCartSummaryHandler.fs b/Sample/ECommerce.Equinox/ECommerce.Reactor/ShoppingCartSummaryHandler.fs
new file mode 100644
index 000000000..94c6ef970
--- /dev/null
+++ b/Sample/ECommerce.Equinox/ECommerce.Reactor/ShoppingCartSummaryHandler.fs
@@ -0,0 +1,19 @@
+module ECommerce.Reactor.ShoppingCartSummaryHandler
+
+open ECommerce.Domain
+
+type Service internal (source : ShoppingCart.Service, destination : ShoppingCartSummary.Service) =
+
+ member _.TryIngestSummary(cartId) : Async = async {
+ match! source.SummarizeWithVersion(cartId) with
+ | Some view, version' ->
+ let! worked = destination.TryIngest(cartId, version', view)
+ return worked, version'
+ | None, version' -> return false, version' }
+
+module Config =
+
+ let create (sourceStore, targetStore) =
+ let source = ShoppingCart.Config.create sourceStore
+ let destination = ShoppingCartSummary.Config.create targetStore
+ Service(source, destination)
diff --git a/Sample/ECommerce.Equinox/ECommerce.Reactor/SourceArgs.fs b/Sample/ECommerce.Equinox/ECommerce.Reactor/SourceArgs.fs
new file mode 100644
index 000000000..d19fd1945
--- /dev/null
+++ b/Sample/ECommerce.Equinox/ECommerce.Reactor/SourceArgs.fs
@@ -0,0 +1,256 @@
+module SourceArgs
+
+open Argu
+open Serilog
+open System
+
+type Configuration(tryGet) =
+ inherit Args.Configuration(tryGet)
+ member _.DynamoIndexTable = tryGet Args.INDEX_TABLE
+
+module Cosmos =
+
+ type [] Parameters =
+ | [] Verbose
+ | [