Add payment processing (#4)

app/config/index.js

@@ -5,12 +5,16 @@ const { development, production, test } = require('./constants').environments
 
 // Define config schema
 const schema = Joi.object({
-  env: Joi.string().valid(development, test, production).default(development)
+  env: Joi.string().valid(development, test, production).default(development),
+  paymentProcessingInterval: Joi.number().default(1000),
+  processingBatchSize: Joi.number().default(1000)
 })
 
 // Build config
 const config = {
-  env: process.env.NODE_ENV
+  env: process.env.NODE_ENV,
+  paymentProcessingInterval: process.env.PROCESSING_INTERVAL,
+  processingBatchSize: process.env.PROCESSING_BATCH_SIZE
 }
 
 // Validate config

app/data/models/completed-invoice-line.js

@@ -3,6 +3,7 @@ module.exports = (sequelize, DataTypes) => {
     completedInvoiceLineId: { type: DataTypes.INTEGER, primaryKey: true, autoIncrement: true },
     completedPaymentRequestId: DataTypes.INTEGER,
     standardCode: DataTypes.STRING,
+    schemeCode: DataTypes.STRING,
     accountCode: DataTypes.STRING,
     fundCode: DataTypes.STRING,
     description: DataTypes.STRING,

app/data/models/completed-payment-request.js

@@ -33,7 +33,7 @@ module.exports = (sequelize, DataTypes) => {
     })
     completedPaymentRequest.hasMany(models.completedInvoiceLine, {
       foreignKey: 'completedPaymentRequestId',
-      as: 'completedInvoiceLines'
+      as: 'invoiceLines'
     })
     completedPaymentRequest.belongsTo(models.batch, {
       foreignKey: 'batchId',

app/data/models/hold.js

@@ -2,8 +2,9 @@ module.exports = (sequelize, DataTypes) => {
   const hold = sequelize.define('hold', {
     holdId: { type: DataTypes.INTEGER, primaryKey: true, autoIncrement: true },
     holdCategoryId: DataTypes.INTEGER,
-    applied: DataTypes.DATE,
-    removed: DataTypes.DATE
+    frn: DataTypes.BIGINT,
+    added: DataTypes.DATE,
+    closed: DataTypes.DATE
   },
   {
     tableName: 'holds',

app/data/models/invoice-line.js

@@ -3,6 +3,7 @@ module.exports = (sequelize, DataTypes) => {
     invoiceLineId: { type: DataTypes.INTEGER, primaryKey: true, autoIncrement: true },
     paymentRequestId: DataTypes.INTEGER,
     standardCode: DataTypes.STRING,
+    schemeCode: DataTypes.STRING,
     accountCode: DataTypes.STRING,
     fundCode: DataTypes.STRING,
     description: DataTypes.STRING,

app/data/models/scheme.js

@@ -1,7 +1,8 @@
 module.exports = (sequelize, DataTypes) => {
   const scheme = sequelize.define('scheme', {
     schemeId: { type: DataTypes.INTEGER, primaryKey: true, autoIncrement: true },
-    name: DataTypes.STRING
+    name: DataTypes.STRING,
+    active: DataTypes.BOOLEAN
   },
   {
     tableName: 'schemes',

app/index.js

@@ -1,5 +1,6 @@
 require('./insights').setup()
 const messageService = require('./messaging')
+const paymentProcessing = require('./processing')
 
 process.on('SIGTERM', async () => {
   await messageService.stop()
@@ -13,4 +14,5 @@ process.on('SIGINT', async () => {
 
 module.exports = (async function startService () {
   await messageService.start()
+  await paymentProcessing.start()
 }())

app/processing/calculate-delta.js

@@ -0,0 +1,6 @@
+
+const calculateDelta = async (paymentRequest, previousPaymentRequests) => {
+  return [paymentRequest]
+}
+
+module.exports = calculateDelta

app/processing/complete-payment-requests.js

@@ -0,0 +1,30 @@
+const db = require('../data')
+
+const completePaymentRequest = async (scheduleId, paymentRequests) => {
+  const transaction = await db.sequelize.transaction()
+  try {
+    const schedule = await db.schedule.findByPk(scheduleId, { transaction })
+
+    // Check if completed already in case of duplicate processing
+    if (schedule.completed === null) {
+      await db.schedule.update({ completed: new Date() }, { where: { scheduleId }, transaction })
+      for (const paymentRequest of paymentRequests) {
+        // Extract data values from Sequelize object if exists
+        const completedPaymentRequest = paymentRequest.dataValues ?? paymentRequest
+        const savedCompletedPaymentRequest = await db.completedPaymentRequest.create(completedPaymentRequest, { transaction })
+        for (const invoiceLine of paymentRequest.invoiceLines) {
+          // Extract data values from Sequelize object if exists
+          const completedInvoiceLine = invoiceLine.dataValues ?? invoiceLine
+          completedInvoiceLine.completedPaymentRequestId = savedCompletedPaymentRequest.completedPaymentRequestId
+          await db.completedInvoiceLine.create(completedInvoiceLine, { transaction })
+        }
+      }
+    }
+    await transaction.commit()
+  } catch (error) {
+    await transaction.rollback()
+    throw (error)
+  }
+}
+
+module.exports = completePaymentRequest

app/processing/get-completed-payment-requests.js

@@ -0,0 +1,17 @@
+const db = require('../data')
+
+const getCompletedPaymentRequests = async (schemeId, frn, marketingYear) => {
+  return db.completedPaymentRequest.findAll({
+    where: {
+      schemeId,
+      frn,
+      marketingYear
+    },
+    include: [{
+      model: db.completedInvoiceLine,
+      as: 'invoiceLines'
+    }]
+  })
+}
+
+module.exports = getCompletedPaymentRequests

app/processing/get-payment-requests.js

@@ -0,0 +1,123 @@
+const db = require('../data')
+const moment = require('moment')
+const config = require('../config')
+
+const getPaymentRequests = async (started = new Date()) => {
+  const transaction = await db.sequelize.transaction()
+  try {
+    const paymentRequests = await getScheduledPaymentRequests(started, transaction)
+    const paymentRequestsWithoutPending = await removePending(paymentRequests, started, transaction)
+    const paymentRequestsWithoutHolds = await removeHolds(paymentRequestsWithoutPending, transaction)
+    const uniquePaymentRequests = removeDuplicates(paymentRequestsWithoutHolds)
+    const cappedPaymentRequests = restrictToBatchSize(uniquePaymentRequests)
+    await updateScheduled(cappedPaymentRequests, started, transaction)
+    await transaction.commit()
+    return cappedPaymentRequests
+  } catch (error) {
+    await transaction.rollback()
+    throw (error)
+  }
+}
+
+const getScheduledPaymentRequests = async (started, transaction) => {
+  return db.schedule.findAll({
+    transaction,
+    order: ['planned'],
+    include: [{
+      model: db.paymentRequest,
+      as: 'paymentRequest',
+      required: true,
+      include: [{
+        model: db.invoiceLine,
+        as: 'invoiceLines',
+        required: true
+      }, {
+        model: db.scheme,
+        as: 'scheme',
+        required: true
+      }]
+    }],
+    where: {
+      '$paymentRequest.scheme.active$': true,
+      planned: { [db.Sequelize.Op.lte]: started },
+      completed: null,
+      [db.Sequelize.Op.or]: [{
+        started: null
+      }, {
+        started: { [db.Sequelize.Op.lte]: moment(started).subtract(5, 'minutes').toDate() }
+      }]
+    }
+  })
+}
+
+const removePending = async (scheduledPaymentRequests, started, transaction) => {
+  const pending = await getPending(started, transaction)
+  return scheduledPaymentRequests.filter(x =>
+    !pending.some(y => y.paymentRequest.schemeId === x.paymentRequest.schemeId && y.paymentRequest.frn === x.paymentRequest.frn && y.paymentRequest.marketingYear === x.paymentRequest.marketingYear))
+}
+
+const getPending = async (started, transaction) => {
+  return db.schedule.findAll({
+    where: {
+      completed: null,
+      started: { [db.Sequelize.Op.gt]: moment(started).subtract(5, 'minutes').toDate() }
+    },
+    include: [{
+      model: db.paymentRequest,
+      as: 'paymentRequest'
+    }],
+    transaction
+  })
+}
+
+const removeHolds = async (scheduledPaymentRequests, transaction) => {
+  const holds = await getHolds(transaction)
+  return scheduledPaymentRequests.filter(x =>
+    !holds.some(y => y.holdCategory.schemeId === x.paymentRequest.schemeId && y.frn === x.paymentRequest.frn))
+}
+
+const getHolds = async (transaction) => {
+  return db.hold.findAll({
+    where: { closed: null },
+    include: [{
+      model: db.holdCategory,
+      as: 'holdCategory'
+    }],
+    transaction
+  })
+}
+
+const removeDuplicates = (scheduledPaymentRequests) => {
+  return scheduledPaymentRequests.reduce((x, y) => {
+    const isDuplicate = (currentSchedule) => {
+      return x.some((schedule) => {
+        return (schedule.paymentRequest.schemeId === currentSchedule.paymentRequest.schemeId &&
+          schedule.paymentRequest.frn === currentSchedule.paymentRequest.frn &&
+          schedule.paymentRequest.marketingYear === currentSchedule.paymentRequest.marketingYear)
+      })
+    }
+
+    if (isDuplicate(y)) {
+      return x
+    } else {
+      return [...x, y]
+    }
+  }, [])
+}
+
+const restrictToBatchSize = (scheduledPaymentRequests) => {
+  return scheduledPaymentRequests.slice(0, config.processingBatchSize)
+}
+
+const updateScheduled = async (scheduledPaymentRequests, started, transaction) => {
+  for (const scheduledPaymentRequest of scheduledPaymentRequests) {
+    await db.schedule.update({ started }, {
+      where: {
+        scheduleId: scheduledPaymentRequest.scheduleId
+      },
+      transaction
+    })
+  }
+}
+
+module.exports = getPaymentRequests

app/processing/index.js

@@ -0,0 +1,16 @@
+const processPaymentRequests = require('./process-payment-requests')
+const config = require('../config')
+
+const start = async () => {
+  try {
+    await processPaymentRequests()
+  } catch (err) {
+    console.error(err)
+  } finally {
+    setTimeout(start, config.paymentProcessingInterval)
+  }
+}
+
+module.exports = {
+  start
+}

app/processing/map-account-codes.js

@@ -0,0 +1,20 @@
+const db = require('../data')
+
+const mapAccountCodes = async (paymentRequest) => {
+  for (const invoiceLine of paymentRequest.invoiceLines) {
+    const accountCode = await db.accountCode.findOne({
+      include: [{
+        model: db.schemeCode,
+        as: 'schemeCode'
+      }],
+      where: {
+        '$schemeCode.schemeCode$': invoiceLine.schemeCode,
+        lineDescription: invoiceLine.description
+      }
+    })
+
+    invoiceLine.accountCode = paymentRequest.ledger === 'AP' ? accountCode.accountCodeAP : accountCode.accountCodeAR
+  }
+}
+
+module.exports = mapAccountCodes

app/processing/process-payment-requests.js

@@ -0,0 +1,21 @@
+const getPaymentRequests = require('./get-payment-requests')
+const mapAccountCodes = require('./map-account-codes')
+const completePaymentRequests = require('./complete-payment-requests')
+const transformPaymentRequest = require('./transform-payment-request')
+
+const processPaymentRequests = async () => {
+  const scheduledPaymentRequests = await getPaymentRequests()
+  for (const scheduledPaymentRequest of scheduledPaymentRequests) {
+    await processPaymentRequest(scheduledPaymentRequest)
+  }
+}
+
+const processPaymentRequest = async (scheduledPaymentRequest) => {
+  const paymentRequests = await transformPaymentRequest(scheduledPaymentRequest.paymentRequest)
+  for (const paymentRequest of paymentRequests) {
+    await mapAccountCodes(paymentRequest)
+  }
+  await completePaymentRequests(scheduledPaymentRequest.scheduleId, paymentRequests)
+}
+
+module.exports = processPaymentRequests

app/processing/transform-payment-request.js

@@ -0,0 +1,15 @@
+const calculateDelta = require('./calculate-delta')
+const getCompletedPaymentRequests = require('./get-completed-payment-requests')
+
+const transformPaymentRequest = async (paymentRequest) => {
+  // Check to see if payment request has had a previous payment request.
+  // if yes, need to treat as post payment adjustment and calculate Delta which can result in payment request splitting
+  const previousPaymentRequests = await getCompletedPaymentRequests(paymentRequest.schemeId, paymentRequest.frn, paymentRequest.marketingYear)
+  if (previousPaymentRequests.length) {
+    return calculateDelta(paymentRequest, previousPaymentRequests)
+  }
+  // otherwise original payment request does not require further processing so can be returned without modification
+  return [paymentRequest]
+}
+
+module.exports = transformPaymentRequest

changelog/db.changelog-1.3.xml

@@ -0,0 +1,14 @@
+<?xml version="1.1" encoding="UTF-8" standalone="no"?>
+<databaseChangeLog xmlns="http://www.liquibase.org/xml/ns/dbchangelog" xmlns:ext="http://www.liquibase.org/xml/ns/dbchangelog-ext" xmlns:pro="http://www.liquibase.org/xml/ns/pro" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://www.liquibase.org/xml/ns/dbchangelog-ext http://www.liquibase.org/xml/ns/dbchangelog/dbchangelog-ext.xsd http://www.liquibase.org/xml/ns/pro http://www.liquibase.org/xml/ns/pro/liquibase-pro-3.9.xsd http://www.liquibase.org/xml/ns/dbchangelog http://www.liquibase.org/xml/ns/dbchangelog/dbchangelog-3.9.xsd">
+  <changeSet author="John Watson" id="36">
+    <tagDatabase tag="v1.3.0" />
+  </changeSet>
+  <changeSet author="John Watson" id="37">
+    <addColumn tableName="invoiceLines">
+      <column name="schemeCode" afterColumn="sbi" type="VARCHAR(10)" />
+    </addColumn>
+    <addColumn tableName="completedInvoiceLines">
+      <column name="schemeCode" afterColumn="sbi" type="VARCHAR(10)" />
+    </addColumn>
+  </changeSet>
+</databaseChangeLog>

changelog/db.changelog.xml

@@ -7,4 +7,5 @@
   <include  file="changelog/db.changelog-1.0.xml"/>
   <include  file="changelog/db.changelog-1.1.xml"/>
   <include  file="changelog/db.changelog-1.2.xml"/>
+  <include  file="changelog/db.changelog-1.3.xml"/>
 </databaseChangeLog>

package.json

@@ -1,6 +1,6 @@
 {
   "name": "ffc-sfi-payments",
-  "version": "1.1.1",
+  "version": "1.2.0",
   "description": "FFC SFI payment services",
   "homepage": "https://github.com/DEFRA/ffc-sfi-payments",
   "main": "app/index.js",

scripts/start

@@ -9,8 +9,8 @@ projectRoot="$(a="/$0"; a=${a%/*}; a=${a:-.}; a=${a#/}/; cd "$a/.." || return; p
 
 cd "${projectRoot}"
 
-docker-compose down -v
-docker-compose -f docker-compose.migrate.yaml down -v
+docker-compose down
+docker-compose -f docker-compose.migrate.yaml down
 # Ensure container images are up to date
 docker-compose -f docker-compose.migrate.yaml run database-up
 

scripts/test

@@ -43,11 +43,13 @@ compose() {
 (
   cd "${projectRoot}"
 
-  # Guarantee clean environment
+ # Guarantee clean environment
   compose down -v
+  docker-compose -f docker-compose.migrate.yaml -p "${service}-test" down
 
   # Ensure container images are up to date
   compose build
+  docker-compose -f docker-compose.migrate.yaml -p "${service}-test" run --rm database-up
 
   # Run tests
   compose run ${service} ${command}