@@ -10,71 +10,121 @@ module Events =
1010 let [<Literal>] CategoryId = " LocationEpoch"
1111 let (| For |) ( locationId , epochId ) = FsCodec.StreamName.compose CategoryId [ LocationId.toString locationId; LocationEpochId.toString epochId]
1212
13- type CarriedForward = { initial : int }
14- type Delta = { delta : int ; transaction : InventoryTransactionId }
15- type Value = { value : int ; transaction : InventoryTransactionId }
13+ type CarriedForward = { initial : int ; recentTransactions : InventoryTransactionId [] }
1614 type Event =
1715 | CarriedForward of CarriedForward
16+ | Added of {| delta : int ; id : InventoryTransactionId |}
17+ | Removed of {| delta : int ; id : InventoryTransactionId |}
18+ | Reset of {| value : int ; id : InventoryTransactionId |}
1819 | Closed
19- | Added of Delta
20- | Removed of Delta
21- | Reset of Value
2220 interface TypeShape.UnionContract.IUnionContract
2321 let codec = FsCodec.NewtonsoftJson.Codec.Create< Event>()
2422
2523module Fold =
2624
27- type Balance = int
28- type OpenState = { count : int ; value : Balance }
29- type State = Initial | Open of OpenState | Closed of Balance
25+ type State =
26+ | Initial
27+ | Open of Record list // reverse order, i.e. most revent first
28+ | Closed of Record list // trimmed
29+ and Record =
30+ | Init of Events.CarriedForward
31+ | Step of Step
32+ and Step = { balance : Balance ; id : InventoryTransactionId }
33+ and Balance = int
3034 let initial = Initial
35+ let (| Current |) = function
36+ | ( Init { initial = bal } | Step { balance = bal }) :: _ -> bal
37+ | [] -> failwith " Cannot transact when no CarriedForward"
3138 let evolve state event =
3239 match event, state with
33- | Events.CarriedForward e, Initial -> Open { count = 0 ; value = e.initial }
34- | Events.Added e, Open bal -> Open { count = bal.count + 1 ; value = bal.value + e.delta }
35- | Events.Removed e, Open bal -> Open { count = bal.count + 1 ; value = bal.value - e.delta }
36- | Events.Reset e, Open bal -> Open { count = bal.count + 1 ; value = e.value }
37- | Events.Closed, Open { value = bal } -> Closed bal
38- | Events.CarriedForward _, ( Open _| Closed _ as x) -> failwithf " CarriedForward : Unexpected %A "
39- | ( Events.Added _| Events.Removed _| Events.Reset _| Events.Closed) as e, ( Initial| Closed _ as s) -> failwithf " Unexpected %A %A "
40+ | Events.CarriedForward e, Initial -> Open [ Init e]
41+ | Events.Added e, Open ( Current cur as log) -> Open ( Step { id = e.id ; balance = cur + e.delta } :: log)
42+ | Events.Removed e, Open ( Current cur as log) -> Open ( Step { id = e.id ; balance = cur - e.delta } :: log)
43+ | Events.Reset e, Open log -> Open ( Step { id = e.id ; balance = e.value } :: log)
44+ | Events.Closed, Open log -> Closed log
45+ | Events.CarriedForward _, ( Open _ | Closed _ as x) -> failwithf " CarriedForward : Unexpected %A "
46+ | ( Events.Added _ | Events.Removed _ | Events.Reset _ | Events.Closed) as e, ( Initial | Closed _ as s) ->
47+ failwithf " Unexpected %A %A "
4048 let fold = Seq.fold evolve
4149
4250/// Holds events accumulated from a series of decisions while also evolving the presented ` state ` to reflect the pended events
4351type private Accumulator () =
4452 let acc = ResizeArray()
4553 member __.Ingest state : 'res * Events.Event list -> 'res * Fold.State = function
46- | res, [] -> res, state
47- | res, [ e] -> acc.Add e; res, Fold.evolve state e
48- | res, xs -> acc.AddRange xs; res, Fold.fold state ( Seq.ofList xs)
54+ | res, [] -> res, state
55+ | res, [ e] -> acc.Add e; res, Fold.evolve state e
56+ | res, xs -> acc.AddRange xs; res, Fold.fold state ( Seq.ofList xs)
4957 member __.Accumulated = List.ofSeq acc
5058
51- type Result < 't > = { balance : Fold .Balance ; result : 't option ; isOpen : bool }
59+ type Result < 't > = { history : Fold .Record list ; result : 't option ; isOpen : bool }
5260
53- let sync ( balanceCarriedForward : Fold.Balance option ) ( decide : Fold.Balance -> 't * Events.Event list ) shouldClose state : Result < 't >* Events.Event list =
61+ let sync ( carriedForward : Events.CarriedForward option )
62+ ( decide : Fold.State -> 't * Events.Event list )
63+ shouldClose
64+ state
65+ : Result < 't > * Events.Event list =
5466
5567 let acc = Accumulator()
56- // We require a CarriedForward event at the start of any Epoch's event stream
68+ // 1. Guarantee a CarriedForward event at the start of any Epoch's event stream
5769 let (), state =
5870 acc.Ingest state <|
5971 match state with
60- | Fold.Initial -> (), [ Events.CarriedForward { initial = Option.get balanceCarriedForward } ]
72+ | Fold.Initial -> (), [ Events.CarriedForward ( Option.get carriedForward ) ]
6173 | Fold.Open _ | Fold.Closed _ -> (), []
62- // Run, unless we determine we're in Closed state
74+ // 2. Transact ( unless we determine we're in Closed state)
6375 let result , state =
6476 acc.Ingest state <|
6577 match state with
66- | Fold.Initial -> failwith " We've just guaranteed not Initial"
67- | Fold.Open { value = bal } -> let r , es = decide bal in Some r, es
68- | Fold.Closed _ -> None, []
69- // Finally (iff we're `Open`, have run a `decide` and `shouldClose`), we generate a Closed event
70- let ( balance , isOpen ), _ =
78+ | Fold.Initial -> failwith " We've just guaranteed not Initial"
79+ | Fold.Open history -> let r , es = decide state in Some r, es
80+ | Fold.Closed _ -> None, []
81+ // 3. Finally (iff we're `Open`, have run a `decide` and `shouldClose`), we generate a Closed event
82+ let ( history , isOpen ), _ =
7183 acc.Ingest state <|
7284 match state with
73- | Fold.Initial -> failwith " Can't be Initial"
74- | Fold.Open ({ value = bal } as openState) when shouldClose openState -> ( bal, false ), [ Events.Closed]
75- | Fold.Open { value = bal } -> ( bal, true ), []
76- | Fold.Closed bal -> ( bal, false ), []
77- { balance = balance; result = result; isOpen = isOpen }, acc.Accumulated
85+ | Fold.Initial -> failwith " Can't be Initial"
86+ | Fold.Open history ->
87+ if shouldClose history then ( history, false ), [ Events.Closed]
88+ else ( history, true ), []
89+ | Fold.Closed history -> ( history, false ), []
90+ { history = history; result = result; isOpen = isOpen }, acc.Accumulated
91+
92+ type DupCheckResult = NotDuplicate | IdempotentInsert of Fold.Balance | DupCarriedForward
93+ let private tryFindDup transactionId ( history : Fold.Record list ) =
94+ let tryMatch : Fold.Record -> Fold.Balance option option = function
95+ | Fold.Step { balance = bal; id = id } when id = transactionId -> Some ( Some bal)
96+ | Fold.Init { recentTransactions = prevs } when prevs |> Array.contains transactionId -> Some None
97+ | _ -> None
98+ match history |> Seq.tryPick tryMatch with
99+ | None -> NotDuplicate
100+ | Some None -> DupCarriedForward
101+ | Some ( Some bal) -> IdempotentInsert bal
102+
103+ type Command =
104+ | Reset of value : int
105+ | Add of delta : int
106+ | Remove of delta : int
107+
108+ type Result = Accepted of Fold.Balance | DupFromPreviousEpoch
109+
110+ let decide transactionId command ( state : Fold.State ) =
111+ match state with
112+ | Fold.Closed _ | Fold.Initial -> failwithf " Cannot apply in state %A "
113+ | Fold.Open history ->
114+
115+ match tryFindDup transactionId history with
116+ | IdempotentInsert bal -> Accepted bal, []
117+ | DupCarriedForward -> DupFromPreviousEpoch, []
118+ | NotDuplicate ->
119+
120+ let e =
121+ match command with
122+ | Reset value -> Events.Reset {| value = value; id = transactionId |}
123+ | Add delta -> Events.Added {| delta = delta; id = transactionId |}
124+ | Remove delta -> Events.Removed {| delta = delta; id = transactionId |}
125+ match Fold.evolve state e with
126+ | Fold.Open ( Fold.Current cur) -> Accepted cur, []
127+ | s -> failwithf " Unexpected state %A "
78128
79129type Service internal ( log , resolve , maxAttempts ) =
80130
0 commit comments