Add a reactor to reserve and dole out invoice numbers

.gitignore

@@ -1,2 +1,4 @@
 obj/
 bin/
+
+*.DotSettings.user

App/App.fsproj

@@ -0,0 +1,33 @@
+<Project Sdk="Microsoft.NET.Sdk.Web">
+
+  <PropertyGroup>
+    <TargetFramework>net7.0</TargetFramework>
+    <RootNamespace>eqx_blog</RootNamespace>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <Compile Include="CheckpointStore.fs" />
+    <Compile Include="InvoiceNumberReactor.fs" />
+    <Compile Include="Program.fs" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <PackageReference Include="Equinox" Version="4.0.0-rc.7" />
+    <PackageReference Include="Equinox.MessageDb" Version="4.0.0-rc.7" />
+    <PackageReference Include="FsCodec" Version="3.0.0-rc.9" />
+    <PackageReference Include="FsCodec.SystemTextJson" Version="3.0.0-rc.9" />
+    <PackageReference Include="FSharp.UMX" Version="1.1.0" />
+    <PackageReference Include="Microsoft.Extensions.Hosting" Version="7.0.0" />
+    <PackageReference Include="Propulsion" Version="3.0.0-beta.6" />
+    <PackageReference Include="Propulsion.MessageDb" Version="3.0.0-beta.5.1" />
+    <PackageReference Include="Serilog" Version="2.12.0" />
+    <PackageReference Include="Serilog.Sinks.Console" Version="4.1.0" />
+    <PackageReference Include="TypeShape" Version="10.0.0" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\InvoiceNumbering\InvoiceNumbering.fsproj" />
+    <ProjectReference Include="..\Invoice\Invoice.fsproj" />
+  </ItemGroup>
+
+</Project>

App/CheckpointStore.fs

@@ -0,0 +1,58 @@
+module CheckpointStore
+
+open Equinox
+open Propulsion.Feed
+
+module Events =
+  type Event =
+    | Checkpoint of {| pos: int64 |}
+    interface TypeShape.UnionContract.IUnionContract
+
+  let codec = FsCodec.SystemTextJson.Codec.Create<Event>()
+
+module Fold =
+  type State = int64 option
+  let initial = None
+
+  let evolve _ =
+    function
+    | Events.Checkpoint x -> Some x.pos
+
+  let fold: State -> Events.Event seq -> State = Seq.fold evolve
+
+let streamId = StreamId.gen2 SourceId.toString id
+type CheckpointService internal (resolve: string -> StreamId -> Equinox.Decider<Events.Event, Fold.State>) =
+  member _.SetCheckpoint(source, tranche, group: string, pos) =
+    let category = TrancheId.toString tranche + ":position"
+    let streamId = streamId (source, group)
+    let decider = resolve category streamId
+    decider.Transact (function
+          | None -> [ Events.Checkpoint {| pos = pos |} ]
+          | Some curr when curr < pos -> [ Events.Checkpoint {| pos = pos |} ]
+          | Some _ -> [])
+
+  member _.ReadCheckpoint(source, tranche, group) =
+    let category = TrancheId.toString tranche + ":position"
+    let streamId = StreamId.ofRaw $"{SourceId.toString source}_{group}"
+    let decider = resolve category streamId
+    decider.Query(id)
+
+let create resolve = CheckpointService(resolve)
+
+type CheckpointStore(service: CheckpointService, consumerGroup, defaultCheckpointFrequency) =
+  interface IFeedCheckpointStore with
+    member this.Commit(source, tranche, pos) =
+        service.SetCheckpoint(source, tranche, consumerGroup, Position.toInt64 pos)
+
+    member this.Start(source, tranche, establishOrigin) =
+      async {
+        let! maybePos = service.ReadCheckpoint(source, tranche, consumerGroup)
+
+        let! pos =
+          match maybePos, establishOrigin with
+          | Some pos, _ -> async { return  Position.parse pos }
+          | None, Some f -> f
+          | None, None -> async { return Position.initial }
+
+        return struct (defaultCheckpointFrequency, pos)
+      }

App/InvoiceNumberReactor.fs

@@ -0,0 +1,45 @@
+module InvoiceNumberReactor
+
+open System
+open Microsoft.Extensions.Hosting
+open Propulsion.Internal
+open Propulsion.MessageDb
+
+type HostedService internal (connectionString, log, service: InvoiceNumberingReactor.Service, checkpointService: CheckpointStore.CheckpointService) =
+
+  inherit BackgroundService()
+
+  [<Literal>]
+  let GroupName = "InvoiceNumberingReactor"
+
+  let categories = [| Invoice.Category |]
+
+  let checkpoints = CheckpointStore.CheckpointStore(checkpointService, GroupName, TimeSpan.FromSeconds 10)
+
+  override _.ExecuteAsync(ct) =
+    let computation =
+      async {
+        use sink = service.Sink(log)
+
+        use src =
+          MessageDbSource(
+            log,
+            statsInterval = TimeSpan.FromMinutes 1,
+            connectionString = connectionString,
+            batchSize = 1000,
+            // Controls the time to wait once fully caught up
+            // before requesting a new batch of events
+            tailSleepInterval = TimeSpan.FromMilliseconds 100,
+            checkpoints = checkpoints,
+            sink = sink,
+            // An array of message-db categories to subscribe to
+            // Propulsion guarantees that events within streams are
+            // handled in order, it makes no guarantees across streams (Even within categories)
+            categories = categories
+          )
+            .Start()
+
+        do! src.AwaitWithStopOnCancellation()
+      }
+
+    Async.StartAsTask(computation, cancellationToken = ct)

App/Program.fs

@@ -3,6 +3,7 @@ module Program
 open System
 open Microsoft.AspNetCore.Builder
 open Microsoft.Extensions.Hosting
+open Microsoft.Extensions.DependencyInjection
 open Equinox.MessageDb
 open Serilog
 
@@ -22,46 +23,72 @@ let writeUrl =
 let readUrl =
   Environment.tryGetEnv "MESSAGE_DB_REPLICA_URL" |> Option.defaultValue writeUrl
 
-let connection = MessageDbConnector(writeUrl, readUrl).Establish()
+let connection = MessageDbClient(writeUrl, readUrl)
 let context = MessageDbContext(connection)
 let caching = CachingStrategy.SlidingWindow(cache, TimeSpan.FromMinutes(20))
 
-let service =
-  MessageDbCategory(context, Invoice.Events.codec, Invoice.Fold.fold, Invoice.Fold.initial, caching)
-  |> Equinox.Decider.resolve log
-  |> Invoice.create
+type Services() =
+  static member Create(context, codec, fold, initial, ?access) =
+    MessageDbCategory(context, codec, fold, initial, caching, ?access = access)
+
+  static member InvoiceService(context) =
+    let codec = Invoice.Events.codec
+    let fold, initial = Invoice.Fold.fold, Invoice.Fold.initial
+    Services.Create(context, codec, fold, initial)
+    |> Equinox.Decider.resolve log
+    |> Invoice.create
+
+  static member InvoiceNumberService(context) =
+    let codec = InvoiceNumbering.Events.codec
+    let fold, initial = InvoiceNumbering.Fold.fold, InvoiceNumbering.Fold.initial
+    let snapshot = InvoiceNumbering.Fold.snapshotEventType, InvoiceNumbering.Fold.toSnapshot
+    let access = AccessStrategy.AdjacentSnapshots snapshot
+    Services.Create(context, codec, fold, initial, access)
+    |> Equinox.Decider.resolve log
+    |> InvoiceNumbering.create
+
+  static member InvoiceNumberReactorService(context) =
+    let invoiceService = Services.InvoiceService(context)
+    let numberService = Services.InvoiceNumberService(context)
+    InvoiceNumberingReactor.Service(numberService, invoiceService)
+
+  static member CheckpointService(context) =
+    let codec = CheckpointStore.Events.codec
+    let fold, initial = CheckpointStore.Fold.fold, CheckpointStore.Fold.initial
+    let access = AccessStrategy.LatestKnownEvent
+    Services.Create(context, codec, fold, initial, access)
+    |> Equinox.Decider.resolve log
+    |> CheckpointStore.create
+
+  static member HostedReactor(context) =
+    let reactorService = Services.InvoiceNumberReactorService(context)
+    let checkpointService = Services.CheckpointService(context)
+    new InvoiceNumberReactor.HostedService(readUrl, log, reactorService, checkpointService)
 
 let builder = WebApplication.CreateBuilder()
+builder.Services.AddHostedService(fun _ -> Services.HostedReactor(context)) |> ignore
 let app = builder.Build()
 
-let raiseInvoice body =
-  task {
-    let id = Guid.NewGuid() |> Invoice.InvoiceId.ofGuid
-    do! service.Raise(id, body)
-    return id
-  }
+let service = Services.InvoiceService(context)
 
-app.MapPost("/", Func<_, _>(raiseInvoice)) |> ignore
+app.MapPost("/", Func<_, _>(fun body -> task {
+  let id = Guid.NewGuid() |> Invoice.InvoiceId.ofGuid
+  do! service.Raise(id, body)
+  return id
+})) |> ignore
 
-let finalizeInvoice id =
-  task {
-    do! service.Finalize(id)
-    return "OK"
-  }
+app.MapPost("/{id}/finalize", Func<_, _>(fun id -> task {
+  do! service.Finalize(id)
+  return "OK"
+})) |> ignore
 
-app.MapPost("/{id}/finalize", Func<_, _>(finalizeInvoice)) |> ignore
+app.MapPost("/{id}/record-payment", Func<_, _, _>(fun id payment -> task {
+  do! service.RecordPayment(id, payment)
+  return "OK"
+})) |> ignore
 
-let recordPayment id payment =
-  task {
-    do! service.RecordPayment(id, payment)
-    return "OK"
-  }
-
-app.MapPost("/{id}/record-payment", Func<_, _, _>(recordPayment)) |> ignore
-
-let readInvoice id =
-  task { return! service.ReadInvoice(id) }
-
-app.MapGet("/{id}", Func<_, _>(readInvoice)) |> ignore
+app.MapGet("/{id}", Func<_, _>(fun id -> task {
+  return! service.ReadInvoice(id)
+})) |> ignore
 
 app.Run()

App/Properties/launchSettings.json

@@ -1,37 +1,13 @@
-{
-  "iisSettings": {
-    "windowsAuthentication": false,
-    "anonymousAuthentication": true,
-    "iisExpress": {
-      "applicationUrl": "http://localhost:52842",
-      "sslPort": 44350
-    }
-  },
-  "profiles": {
-    "http": {
-      "commandName": "Project",
-      "dotnetRunMessages": true,
-      "launchBrowser": true,
-      "applicationUrl": "http://localhost:5244",
-      "environmentVariables": {
-        "ASPNETCORE_ENVIRONMENT": "Development"
-      }
-    },
-    "https": {
-      "commandName": "Project",
-      "dotnetRunMessages": true,
-      "launchBrowser": true,
-      "applicationUrl": "https://localhost:7026;http://localhost:5244",
-      "environmentVariables": {
-        "ASPNETCORE_ENVIRONMENT": "Development"
-      }
-    },
-    "IIS Express": {
-      "commandName": "IISExpress",
-      "launchBrowser": true,
-      "environmentVariables": {
-        "ASPNETCORE_ENVIRONMENT": "Development"
-      }
-    }
-  }
-}
+{
+  "profiles": {
+    "http": {
+      "commandName": "Project",
+      "dotnetRunMessages": true,
+      "launchBrowser": false,
+      "applicationUrl": "http://localhost:5244",
+      "environmentVariables": {
+        "ASPNETCORE_ENVIRONMENT": "Development"
+      }
+    }
+  }
+}