Add initial transfer steps (#1)

.pre-commit-config.yaml

@@ -1,6 +1,6 @@
 repos:
 - repo: https://github.com/Yelp/detect-secrets
-  rev: v1.2.0
+  rev: v1.4.0
   hooks:
   - id: detect-secrets
     args: ['--baseline', '.secrets.baseline']

.secrets.baseline

@@ -1,5 +1,5 @@
 {
-  "version": "1.2.0",
+  "version": "1.4.0",
   "plugins_used": [
     {
       "name": "ArtifactoryDetector"
@@ -20,6 +20,9 @@
     {
       "name": "CloudantDetector"
     },
+    {
+      "name": "DiscordBotTokenDetector"
+    },
     {
       "name": "GitHubTokenDetector"
     },
@@ -104,6 +107,16 @@
       "path": "detect_secrets.filters.heuristic.is_templated_secret"
     }
   ],
-  "results": {},
-  "generated_at": "2022-02-22T08:52:45Z"
+  "results": {
+    "docker-compose.test.yaml": [
+      {
+        "type": "Secret Keyword",
+        "filename": "docker-compose.test.yaml",
+        "hashed_secret": "db553b60422c2f840f659d55e548e4664400ebe5",
+        "is_verified": false,
+        "line_number": 18
+      }
+    ]
+  },
+  "generated_at": "2023-10-22T19:28:32Z"
 }

app/config/index.js

@@ -0,0 +1,9 @@
+const storageConfig = require('./storage')
+const processingConfig = require('./processing')
+const messagingConfig = require('./messaging')
+
+module.exports = {
+  storageConfig,
+  processingConfig,
+  messagingConfig
+}

app/config/messaging.js

@@ -0,0 +1,50 @@
+const Joi = require('joi')
+const { PRODUCTION } = require('../constants/environments')
+
+const schema = Joi.object({
+  messageQueue: {
+    host: Joi.string(),
+    username: Joi.string(),
+    password: Joi.string(),
+    useCredentialChain: Joi.bool().default(false),
+    appInsights: Joi.object()
+  },
+  demographicsTopic: {
+    address: Joi.string()
+  },
+  customerTopic: {
+    address: Joi.string()
+  }
+})
+
+const config = {
+  messageQueue: {
+    host: process.env.MESSAGE_QUEUE_HOST,
+    username: process.env.MESSAGE_QUEUE_USER,
+    password: process.env.MESSAGE_QUEUE_PASSWORD,
+    useCredentialChain: process.env.NODE_ENV === PRODUCTION,
+    appInsights: process.env.NODE_ENV === PRODUCTION ? require('applicationinsights') : undefined
+  },
+  demographicsTopic: {
+    address: process.env.PROCESSING_TOPIC_ADDRESS
+  },
+  customerTopic: {
+    address: process.env.RESPONSE_TOPIC_ADDRESS
+  }
+}
+
+const result = schema.validate(config, {
+  abortEarly: false
+})
+
+if (result.error) {
+  throw new Error(`The messaging config is invalid. ${result.error.message}`)
+}
+
+const demographicsTopic = { ...result.value.messageQueue, ...result.value.demographicsTopic }
+const customerTopic = { ...result.value.messageQueue, ...result.value.customerTopic }
+
+module.exports = {
+  demographicsTopic,
+  customerTopic
+}

app/config/processing.js

@@ -0,0 +1,21 @@
+const Joi = require('joi')
+
+const schema = Joi.object({
+  enabled: Joi.boolean().default(true),
+  pollingInterval: Joi.number().integer().default(60000)
+})
+
+const config = {
+  enabled: process.env.ENABLED,
+  pollingInterval: process.env.POLLING_INTERVAL
+}
+
+const result = schema.validate(config, {
+  abortEarly: false
+})
+
+if (result.error) {
+  throw new Error(`The transfer config is invalid. ${result.error.message}`)
+}
+
+module.exports = result.value

app/config/storage.js

@@ -0,0 +1,31 @@
+const Joi = require('joi')
+
+const schema = Joi.object({
+  enabled: Joi.boolean().default(true),
+  demographicsShareConnectionString: Joi.string().required(),
+  demographicsShareName: Joi.string().required(),
+  demographicsFolder: Joi.string().required(),
+  daxShareConnectionString: Joi.string().required(),
+  daxShareName: Joi.string().required(),
+  daxFolder: Joi.string().required()
+})
+
+const config = {
+  enabled: process.env.ENABLED,
+  demographicsShareConnectionString: process.env.DEMOGRAPHICS_STORAGE_CONNECTION_STRING,
+  demographicsShareName: process.env.DEMOGRAPHICS_STORAGE_SHARE_NAME,
+  demographicsFolder: process.env.DEMOGRAPHICS_STORAGE_FOLDER_NAME,
+  daxShareConnectionString: process.env.DAX_STORAGE_CONNECTION_STRING,
+  daxShareName: process.env.DAX_STORAGE_SHARE_NAME,
+  daxFolder: process.env.DAX_STORAGE_FOLDER_NAME
+}
+
+const result = schema.validate(config, {
+  abortEarly: false
+})
+
+if (result.error) {
+  throw new Error(`The storage config is invalid. ${result.error.message}`)
+}
+
+module.exports = result.value

app/constants/environments.js

@@ -0,0 +1,5 @@
+module.exports = {
+  DEVELOPMENT: 'development',
+  TEST: 'test',
+  PRODUCTION: 'production'
+}

app/constants/source.js

@@ -0,0 +1,3 @@
+module.exports = {
+  SOURCE: 'ffc-pay-demographics'
+}

app/constants/types.js

@@ -0,0 +1,4 @@
+module.exports = {
+  DEMOGRAPHICS: 'uk.gov.uk.pay.demographics.update',
+  CUSTOMER: 'uk.gov.uk.pay.demographics.customer'
+}

app/index.js

@@ -1,13 +1,5 @@
-const server = require('./server')
+require('./insights').setup()
+require('log-timestamp')
+const { start } = require('./processing')
 
-const init = async () => {
-  await server.start()
-  console.log('Server running on %s', server.info.uri)
-}
-
-process.on('unhandledRejection', (err) => {
-  console.log(err)
-  process.exit(1)
-})
-
-init()
+module.exports = (async () => start())()

app/insights.js

@@ -12,4 +12,6 @@ const setup = () => {
   }
 }
 
-module.exports = { setup }
+module.exports = {
+  setup
+}

app/messaging/create-message.js

@@ -0,0 +1,13 @@
+const { SOURCE } = require('../constants/source')
+
+const createMessage = (body, type) => {
+  return {
+    body,
+    type,
+    source: SOURCE
+  }
+}
+
+module.exports = {
+  createMessage
+}

app/messaging/index.js

@@ -0,0 +1,7 @@
+const { createMessage } = require('./create-message')
+const { sendMessage } = require('./send-message')
+
+module.exports = {
+  createMessage,
+  sendMessage
+}

app/messaging/send-message.js

@@ -0,0 +1,20 @@
+const { MessageSender } = require('ffc-messaging')
+const { messageConfig } = require('../config')
+const { createMessage } = require('./create-message')
+const { CUSTOMER } = require('../constants/types')
+
+const sendMessage = async (body, type) => {
+  const message = createMessage(body, type)
+  const topic = getTopic(type)
+  const sender = new MessageSender(topic)
+  await sender.sendMessage(message)
+  await sender.closeConnection()
+}
+
+const getTopic = (type) => {
+  return type === CUSTOMER ? messageConfig.customerTopic : messageConfig.demographicsTopic
+}
+
+module.exports = {
+  sendMessage
+}

app/processing/create-customer-update.js

@@ -0,0 +1,13 @@
+const createCustomerUpdate = (content) => {
+  // once file format known, need to convert to update
+  return {
+    trader: '',
+    vendor: '',
+    sbi: '',
+    frn: ''
+  }
+}
+
+module.exports = {
+  createCustomerUpdate
+}

app/processing/create-dax-update.js

@@ -0,0 +1,7 @@
+const createDaxUpdate = (data) => {
+  return data
+}
+
+module.exports = {
+  createDaxUpdate
+}

app/processing/create-demographics-update.js

@@ -0,0 +1,8 @@
+const createDemographicsUpdate = (content) => {
+  // once file format known, need to convert to update
+  return content
+}
+
+module.exports = {
+  createDemographicsUpdate
+}

app/processing/index.js

@@ -0,0 +1,18 @@
+const { processingConfig } = require('../config')
+const { processFiles } = require('./process-files')
+
+const start = async () => {
+  try {
+    if (processingConfig.enabled) {
+      await processFiles()
+    }
+  } catch (err) {
+    console.error(err)
+  } finally {
+    setTimeout(start, processingConfig.pollingInterval)
+  }
+}
+
+module.exports = {
+  start
+}

app/processing/process-file.js

@@ -0,0 +1,25 @@
+const { sendMessage } = require('../messaging')
+const { downloadFile, uploadFile, deleteFile } = require('../storage')
+const { createDemographicsUpdate } = require('./create-demographics-update')
+const { DEMOGRAPHICS, CUSTOMER } = require('../constants/types')
+const { createCustomerUpdate } = require('./create-customer-update')
+const { createDaxUpdate } = require('./create-dax-update')
+
+const processFile = async (file) => {
+  console.log(`processing demographics file: ${file}`)
+  const data = await downloadFile(file)
+  const demographicsData = createDemographicsUpdate(data)
+  await sendMessage(demographicsData, DEMOGRAPHICS)
+  // Not every update will have customer mapping, once we know file spec we can update this flow
+  const customerData = createCustomerUpdate(data)
+  await sendMessage(customerData, CUSTOMER)
+  // Need to further understand how and where updates are published from the TL
+  const daxData = createDaxUpdate(data)
+  await uploadFile(file, daxData)
+  await deleteFile(file)
+  console.log(`processed demographics file: ${file}`)
+}
+
+module.exports = {
+  processFile
+}

app/processing/process-files.js

@@ -0,0 +1,17 @@
+const { getDemographicsFiles } = require('../storage')
+const { processFile } = require('./process-file')
+
+const processFiles = async () => {
+  const files = await getDemographicsFiles()
+  for (const file of files) {
+    try {
+      await processFile(file)
+    } catch (err) {
+      console.error(err)
+    }
+  }
+}
+
+module.exports = {
+  processFiles
+}