feat(mojaloop/#3488): enable batch processing

docs/technical/central-ledger/assets/diagrams/sequence/seq-position-1.3.1-prepare-batch.plantuml

@@ -0,0 +1,271 @@
+/'*****
+ 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>
+
+ * Georgi Georgiev <georgi.georgiev@modusbox.com>
+ * Rajiv Mothilal <rajiv.mothilal@modusbox.com>
+ * Miguel de Barros <miguel.debarros@modusbox.com>
+ * Vijay Kumar Guthi <vijaya.guthi@infitx.com>
+ --------------
+ ******'/
+
+@startuml
+' declate title
+title 1.3.1. Position Prepare Facade
+
+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 Prepare Facade" as FACADE_POSITION_PREPARE
+collections "Notification-Topic" as TOPIC_NOTIFICATIONS
+database "Central Store" as DB
+
+box "Central Service" #LightYellow
+    participant FACADE_POSITION_PREPARE
+    participant DB
+    participant TOPIC_NOTIFICATIONS
+end box
+
+' start flow
+activate FACADE_POSITION_PREPARE
+group Prepare Position Batch Processing
+    ' DB Trans: This is where 1st DB Transaction would start in 2 DB transacation future model for horizontal scaling
+    FACADE_POSITION_PREPARE -> FACADE_POSITION_PREPARE: Loop through batch and build list of transferIds and calculate sumTransfersInBatch,\nchecking all in Batch are for the correct Paricipant and Currency\n<color #FF0000><b>Error code:</b> 2001, 3100</color>
+    FACADE_POSITION_PREPARE -> DB: Retrieve current state of all transfers in array from DB with select whereIn\n(FYI: The two DB transaction model needs to add a mini-state step here (RECEIVED_PREPARE => RECEIVDED_PREPARE_PROCESSING) so that the transfers are left alone if processing has started)
+    activate DB
+    hnote over DB #lightyellow
+        transferStateChange
+        transferParticipant
+    end note
+    DB --> FACADE_POSITION_PREPARE: Return current state of all selected transfers from DB
+    deactivate DB
+    FACADE_POSITION_PREPARE <-> FACADE_POSITION_PREPARE: Validate current state (transferStateChange.transferStateId == 'RECEIVED_PREPARE')\n<color #FF0000><b>Error code:</b> 2001</color> against failing transfers\nBatch is not rejected as a whole.
+
+    note right of FACADE_POSITION_PREPARE #lightgray
+        List of transfers used during processing
+        **reservedTransfers** is list of transfers to be processed in the batch
+        **abortedTransfers** is the list of transfers in the incorrect state going into the process. Currently the transferStateChange is set to ABORTED - this should only be done if not already in a final state (idempotency)
+        **processedTransfers** is the list of transfers that have gone through the position management algorithm. Both successful and failed trasnfers appear here as the order and "running position" against each is necessary for reconciliation
+
+        Scalar intermidate values used in the algorithm
+        **transferAmount** = payload.amount.amount
+        **sumTransfersInBatch** = SUM amount against each Transfer in batch
+        **currentPosition** = participantPosition.value
+        **reservedPosition** = participantPosition.{original}reservedValue
+        **effectivePosition** = currentPosition + reservedPosition
+        **heldPosition** = effectivePosition + sumTransfersInBatch
+        **availablePosition** = //if settlement model delay is IMMEDIATE then:// settlementBalance + participantLimit(NetDebitCap) - effectivePosition, //otherwise:// participantLimit(NetDebitCap) - effectivePosition
+        **sumReserved** = SUM of transfers that have met rule criteria and processed
+    end note
+    note over FACADE_POSITION_PREPARE,DB
+        Going to reserve the sum of the valid transfers in the batch against the Participants Positon in the Currency of this batch
+        and calculate the available position for the Participant to use
+    end note
+    FACADE_POSITION_PREPARE -> DB: Select effectivePosition FOR UPDATE from DB for Payer
+    activate DB
+    hnote over DB #lightyellow
+        participantPosition
+    end note
+    DB --> FACADE_POSITION_PREPARE: Return effectivePosition (currentPosition and reservedPosition) from DB for Payer
+    deactivate DB
+    FACADE_POSITION_PREPARE -> FACADE_POSITION_PREPARE: Increment reservedValue to heldPosition\n(reservedValue = reservedPosition + sumTransfersInBatch)
+    FACADE_POSITION_PREPARE -> DB: Persist reservedValue
+    activate DB
+    hnote over DB #lightyellow
+        UPDATE **participantPosition**
+        SET reservedValue += sumTransfersInBatch
+    end note
+    deactivate DB
+    ' DB Trans: This is where 1st DB Transaction would end in 2 DB transacation future model for horizontal scaling
+
+
+    FACADE_POSITION_PREPARE -> DB: Request position limits for Payer Participant
+    activate DB
+    hnote over DB #lightyellow
+        FROM **participantLimit**
+        WHERE participantLimit.limitTypeId = 'NET-DEBIT-CAP'
+        AND participantLimit.participantId = payload.payerFsp
+        AND participantLimit.currencyId = payload.amount.currency
+    end note
+    DB --> FACADE_POSITION_PREPARE: Return position limits
+    deactivate DB
+    FACADE_POSITION_PREPARE <-> FACADE_POSITION_PREPARE: **availablePosition** = //if settlement model delay is IMMEDIATE then://\n settlementBalance + participantLimit(NetDebitCap) - effectivePosition\n //otherwise://\n participantLimit(NetDebitCap) - effectivePosition\n(same as = (settlementBalance?) + netDebitCap - currentPosition - reservedPosition)
+    note over FACADE_POSITION_PREPARE,DB
+        For each transfer in the batch, validate the availablility of position to meet the transfer amount
+        this will be as per the position algorithm documented below
+    end note
+    FACADE_POSITION_PREPARE <-> FACADE_POSITION_PREPARE: Validate availablePosition for each tranfser (see algorithm below)\n<color #FF0000><b>Error code:</b> 4001</color>
+    note right of FACADE_POSITION_PREPARE #lightgray
+        01: sumReserved = 0 // Record the sum of the transfers we allow to progress to RESERVED
+        02: sumProcessed =0 // Record the sum of the transfers already processed in this batch
+        03: processedTransfers = {} // The list of processed transfers - so that we can store the additional information around the decision. Most importantly the "running" position
+        04: foreach transfer in reservedTransfers
+            05: sumProcessed += transfer.amount // the total processed so far **(NEED TO UPDATE IN CODE)**
+            06: if availablePosition >= transfer.amount
+                07: transfer.state = "RESERVED"
+                08: availablePosition -= preparedTransfer.amount
+                09: sumRESERVED += preparedTransfer.amount
+            10: else
+                11: preparedTransfer.state = "ABORTED"
+                12: preparedTransfer.reason = "Net Debit Cap exceeded by this request at this time, please try again later"
+            13: end if
+            14: runningPosition = currentPosition + sumReserved // the initial value of the Participants position plus the total value that has been accepted in the batch so far
+            15: runningReservedValue = sumTransfersInBatch - sumProcessed + reservedPosition **(NEED TO UPDATE IN CODE)** // the running down of the total reserved value at the begining of the batch.
+            16: Add transfer to the processedTransfer list recording the transfer state and running position and reserved values { transferState, transfer, rawMessage, transferAmount,  runningPosition, runningReservedValue }
+        16: end foreach
+    end note
+    note over FACADE_POSITION_PREPARE,DB
+        Once the outcome for all transfers is known,update the Participant's position and remove the reserved amount associated with the batch
+        (If there are any alarm limits, process those returning limits in which the threshold has been breached)
+        Do a bulk insert of the trasnferStateChanges associated with processing, using the result to complete the participantPositionChange and bulk insert of these to persist the running position
+    end note
+    FACADE_POSITION_PREPARE->FACADE_POSITION_PREPARE: Assess any limit thresholds on the final position\nadding to alarm list if triggered
+
+    ' DB Trans: This is where 2nd DB Transaction would start in 2 DB transacation future model for horizontal scaling
+    FACADE_POSITION_PREPARE->DB: Persist latest position **value** and **reservedValue** to DB for Payer
+        hnote over DB #lightyellow
+            UPDATE **participantPosition**
+            SET value += sumRESERVED,
+            reservedValue -= sumTransfersInBatch
+        end note
+        activate DB
+        deactivate DB
+
+    FACADE_POSITION_PREPARE -> DB: Bulk persist transferStateChange for all processedTransfers
+    hnote over DB #lightyellow
+            batch INSERT **transferStateChange**
+            select for update from transfer table where transferId in ([transferBatch.transferId,...])
+            build list of transferStateChanges from transferBatch
+
+    end note
+    activate DB
+    deactivate DB
+
+    FACADE_POSITION_PREPARE->FACADE_POSITION_PREPARE: Populate batchParticipantPositionChange from the resultant transferStateChange and the earlier processedTransfer list
+
+    note right of FACADE_POSITION_PREPARE #lightgray
+        Effectively:
+            SET transferStateChangeId = processedTransfer.transferStateChangeId,
+            participantPositionId = preparedTransfer.participantPositionId,
+            value = preparedTransfer.positionValue,
+            reservedValue = preparedTransfer.positionReservedValue
+    end note
+    FACADE_POSITION_PREPARE -> DB: Bulk persist the participant position change for all processedTransfers
+    hnote over DB #lightyellow
+            batch INSERT **participantPositionChange**
+    end note
+    activate DB
+    deactivate DB
+    ' DB Trans: This is where 2nd DB Transaction would end in 2 DB transacation future model for horizontal scaling
+
+    deactivate FACADE_POSITION_PREPARE
+    alt Calculate & Validate Latest Position Prepare (success)
+        note right of FACADE_POSITION_PREPARE #yellow
+            Message:
+            {
+                id: <transferMessage.transferId>
+                from: <transferMessage.payerFsp>,
+                to: <transferMessage.payeeFsp>,
+                type: application/json
+                content: {
+                    headers: <transferHeaders>,
+                    payload: <transferMessage>
+                },
+                metadata: {
+                    event: {
+                        id: <uuid>,
+                        responseTo: <previous.uuid>,
+                        type: transfer,
+                        action: prepare,
+                        createdAt: <timestamp>,
+                        state: {
+                            status: "success",
+                            code: 0
+                        }
+                    }
+                }
+            }
+        end note
+        FACADE_POSITION_PREPARE -> TOPIC_NOTIFICATIONS: Publish Notification event\n<color #FF0000><b>Error code:</b> 2003</color>
+        activate TOPIC_NOTIFICATIONS
+        deactivate TOPIC_NOTIFICATIONS
+    else Calculate & Validate Latest Position Prepare (failure)
+        note right of FACADE_POSITION_PREPARE #red: Validation failure!
+
+        group Persist Transfer State (with transferState='ABORTED' on position check fail)
+            FACADE_POSITION_PREPARE -> DB: Request to persist transfer\n<color #FF0000><b>Error code:</b> 2003</color>
+            activate DB
+            note right of FACADE_POSITION_PREPARE #lightgray
+                transferStateChange.state = "ABORTED",
+                transferStateChange.reason = "Net Debit Cap exceeded by this request at this time, please try again later"
+            end note
+            hnote over DB #lightyellow
+                transferStateChange
+            end note
+            DB --> FACADE_POSITION_PREPARE: Return success
+            deactivate DB
+        end
+
+        note right of FACADE_POSITION_PREPARE #yellow
+            Message:
+            {
+                id: <transferMessage.transferId>
+                from: <ledgerName>,
+                to: <transferMessage.payerFsp>,
+                type: application/json
+                content: {
+                    headers: <transferHeaders>,
+                    payload: {
+                        "errorInformation": {
+                            "errorCode": 4001,
+                            "errorDescription": "Payer FSP insufficient liquidity",
+                            "extensionList": <transferMessage.extensionList>
+                    }
+                },
+                metadata: {
+                    event: {
+                        id: <uuid>,
+                        responseTo: <previous.uuid>,
+                        type: notification,
+                        action: prepare,
+                        createdAt: <timestamp>,
+                        state: {
+                            status: 'error',
+                            code: <errorInformation.errorCode>
+                            description: <errorInformation.errorDescription>
+                        }
+                    }
+                }
+            }
+        end note
+        FACADE_POSITION_PREPARE -> TOPIC_NOTIFICATIONS: Publish Notification (failure) event for Payer\n<color #FF0000><b>Error code:</b> 2003</color>
+        activate TOPIC_NOTIFICATIONS
+        deactivate TOPIC_NOTIFICATIONS
+        deactivate FACADE_POSITION_PREPARE
+    end
+end
+
+@enduml