feat(mojaloop/#3488): enable batch processing

docs/.vuepress/config.js

@@ -684,6 +684,10 @@ module.exports = {
                     }
                   ]
                 },
+                {
+                  title: "Position Handler Bin Support",
+                  path: "central-ledger/transfers/1.3.4-position-handler-consume-binning"
+                },
               ]
             },
             {

docs/technical/central-ledger/assets/diagrams/sequence/seq-position-1.3.4.1-binning.plantuml

@@ -0,0 +1,143 @@
+/'*****
+ License
+ --------------
+ Copyright © 2017 Bill & Melinda Gates Foundation
+ The Mojaloop files are made available by the Bill & Melinda Gates Foundation under the Apache License, Version 2.0 (the "License") and you may not use these files except in compliance with the License. You may obtain a copy of the License at
+ http://www.apache.org/licenses/LICENSE-2.0
+ Unless required by applicable law or agreed to in writing, the Mojaloop files are distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.
+ Contributors
+ --------------
+ This is the official list of the Mojaloop project contributors for this file.
+ Names of the original copyright holders (individuals or organizations)
+ should be listed with a '*' in the first column. People who have
+ contributed from an organization can be listed under the organization
+ that actually holds the copyright for their contributions (see the
+ Gates Foundation organization for an example). Those individuals should have
+ their names indented and be marked with a '-'. Email address can be added
+ optionally within square brackets <email>.
+ * Gates Foundation
+ - Name Surname <name.surname@gatesfoundation.com>
+
+ * Vijay Kumar Guthi <vijaya.guthi@infitx.com>
+
+ --------------
+ ******'/
+
+@startuml
+' declate title
+title 1.3.4.1. Position Messages Binning
+
+autonumber
+
+' Actor Keys:
+'   boundary - APIs/Interfaces, etc
+'   collections - Kafka Topics
+'   control - Kafka Consumers
+'   entity - Database Access Objects
+'   database - Database Persistence Store
+
+' declare actors
+control "Position Handler" as POS_HANDLER
+
+entity "Domain Bin Processor" as BIN_PROCESSOR
+entity "Prepare Domain" as DOMAIN_PREPARE
+entity "Fulfil Domain" as DOMAIN_FULFIL
+entity "Abort Domain" as DOMAIN_ABORT
+entity "Timeout Domain" as DOMAIN_TIMEOUT
+entity "Transfer Facade" as FACADE_TRANSFER
+entity "Account Facade" as FACADE_ACCOUNT
+collections "Notification-Topic" as TOPIC_NOTIFICATIONS
+' database "Central Store" as DB
+
+box "Central Service" #LightYellow
+    participant POS_HANDLER
+    participant BIN_PROCESSOR
+    participant DOMAIN_FULFIL
+    participant DOMAIN_ABORT
+    participant DOMAIN_TIMEOUT
+    participant DOMAIN_PREPARE
+    participant FACADE_TRANSFER
+    participant FACADE_ACCOUNT
+    ' participant DB
+    participant TOPIC_NOTIFICATIONS
+end box
+
+' start flow
+activate POS_HANDLER
+group Position Handler Consume
+    POS_HANDLER -> POS_HANDLER: Consumes messages from the Position topic \nand stores them in memory
+    loop for each message in the list
+        POS_HANDLER -> POS_HANDLER: Assign message to account-bin by accountID and child action-bin by event-action\n (References to the messagses to be stored in bins, no duplication of messages)
+        POS_HANDLER -> POS_HANDLER: Audit message
+    end
+    group <color #blue>DB TRANSACTION</color>
+        POS_HANDLER -> BIN_PROCESSOR: Call Bin Processor with the list of account-bins
+
+        BIN_PROCESSOR -> FACADE_TRANSFER: Pre Bulk fetch all transferStateChanges for all the transferIds in the account-bin
+        BIN_PROCESSOR -> FACADE_ACCOUNT: Pre Bulk fetch all position and settlement account balances for the account-bin\nAcquire lock on position
+
+        loop for each account-bin in the list
+            note right of BIN_PROCESSOR #lightblue
+                **Note**: Order of processing for various actions
+                1. Fulfil - Work in progress is first priority
+                2. Abort & Timeout - Freeing funds for future transfers is priority before tackling new requests
+                3. Prepare - Processing new requests is last priority
+            end note
+
+            BIN_PROCESSOR -> DOMAIN_FULFIL: Process Fulfil Bin
+            note right of BIN_PROCESSOR
+                Input: Bin, Pre-fetched account balances and transferStateChanges
+                Output: Accumulated changes in position and transferStateChanges
+            end note
+            activate DOMAIN_FULFIL
+            DOMAIN_FULFIL --> BIN_PROCESSOR: Result
+            deactivate DOMAIN_FULFIL
+
+            BIN_PROCESSOR -> DOMAIN_ABORT: Process Abort Bin
+            note right of BIN_PROCESSOR
+                Input: Bin, previously accumulated account balances and transferStateChanges
+                Output: Accumulated changes in position and transferStateChanges
+            end note
+            activate DOMAIN_ABORT
+            DOMAIN_ABORT --> BIN_PROCESSOR: Result
+            deactivate DOMAIN_ABORT
+
+            BIN_PROCESSOR -> DOMAIN_TIMEOUT: Process Timeout Bin
+            note right of BIN_PROCESSOR
+                Input: Bin, previously accumulated account balances and transferStateChanges
+                Output: Accumulated changes in position and transferStateChanges
+            end note
+            activate DOMAIN_TIMEOUT
+            DOMAIN_TIMEOUT --> BIN_PROCESSOR: Result
+            deactivate DOMAIN_TIMEOUT
+
+            BIN_PROCESSOR -> DOMAIN_PREPARE: Process Prepare Batch
+            note right of BIN_PROCESSOR
+                Input: Bin, previously accumulated account balances and transferStateChanges
+                Output: Accumulated changes in position and transferStateChanges
+            end note
+            activate DOMAIN_PREPARE
+            DOMAIN_PREPARE --> BIN_PROCESSOR: Result
+            deactivate DOMAIN_PREPARE
+            BIN_PROCESSOR -> FACADE_TRANSFER: Bulk insert accumulated transferStateChanges
+            BIN_PROCESSOR -> FACADE_TRANSFER: Bulk get the transferStateChangeIds for transferids using select whereIn
+            BIN_PROCESSOR -> BIN_PROCESSOR: Mutate accumulated positionChanges with transferStateChangeIds
+            BIN_PROCESSOR -> FACADE_ACCOUNT: Bulk insert accumulated positionChanges
+        end
+
+        BIN_PROCESSOR --> POS_HANDLER: Return bins with results
+        alt return is true
+            POS_HANDLER -> POS_HANDLER: Commit Kafka Offset
+            POS_HANDLER -> POS_HANDLER: Commit DB Transaction
+            loop for each result in the account-bin
+                POS_HANDLER -> POS_HANDLER: Produce notification message
+                POS_HANDLER -> POS_HANDLER: Audit notification
+            end
+        else return is false or unhandled exception
+            POS_HANDLER -> POS_HANDLER: Rollback DB Transaction
+            POS_HANDLER -> POS_HANDLER: Audit error
+        end
+    end
+end
+deactivate POS_HANDLER
+@enduml