New 'import' job to trigger return log and supplementary flag checks

app/controllers/jobs.controller.js

@@ -6,13 +6,14 @@
  */
 
 const ExportService = require('../services/jobs/export/export.service.js')
+const ImportLicence = require('../services/jobs/import/import-licences.service.js')
 const ProcessLicenceUpdates = require('../services/jobs/licence-updates/process-licence-updates.js')
+const ProcessReturnLogsService = require('../services/jobs/return-logs/process-return-logs.service.js')
 const ProcessSessionStorageCleanupService = require('../services/jobs/session-cleanup/process-session-storage-cleanup.service.js')
 const ProcessTimeLimitedLicencesService = require('../services/jobs/time-limited/process-time-limited-licences.service.js')
-const ProcessReturnLogsService = require('../services/jobs/return-logs/process-return-logs.service.js')
 
-const redirectStatusCode = 204
-const notFoundStatusCode = 404
+const NO_CONTENT_STATUS_CODE = 204
+const NOT_FOUND_STATUS_CODE = 404
 
 /**
  * Triggers export of all relevant tables to CSV and then uploads them to S3
@@ -26,32 +27,38 @@ const notFoundStatusCode = 404
 async function exportDb (_request, h) {
   ExportService.go()
 
-  return h.response().code(redirectStatusCode)
+  return h.response().code(NO_CONTENT_STATUS_CODE)
+}
+
+async function ImportLicences (_request, h) {
+  ImportLicence.go()
+
+  return h.response().code(NO_CONTENT_STATUS_CODE)
 }
 
 async function licenceUpdates (_request, h) {
   ProcessLicenceUpdates.go()
 
-  return h.response().code(redirectStatusCode)
+  return h.response().code(NO_CONTENT_STATUS_CODE)
 }
 
 async function sessionCleanup (_request, h) {
   ProcessSessionStorageCleanupService.go()
 
-  return h.response().code(redirectStatusCode)
+  return h.response().code(NO_CONTENT_STATUS_CODE)
 }
 
 async function timeLimited (_request, h) {
   ProcessTimeLimitedLicencesService.go()
 
-  return h.response().code(redirectStatusCode)
+  return h.response().code(NO_CONTENT_STATUS_CODE)
 }
 
 async function returnLogs (request, h) {
   const { cycle } = request.params
 
   if (!['summer', 'all-year'].includes(cycle)) {
-    return h.response().code(notFoundStatusCode)
+    return h.response().code(NOT_FOUND_STATUS_CODE)
   }
 
   let licenceReference
@@ -62,11 +69,12 @@ async function returnLogs (request, h) {
 
   ProcessReturnLogsService.go(cycle, licenceReference)
 
-  return h.response().code(redirectStatusCode)
+  return h.response().code(NO_CONTENT_STATUS_CODE)
 }
 
 module.exports = {
   exportDb,
+  ImportLicences,
   licenceUpdates,
   returnLogs,
   sessionCleanup,

app/routes/jobs.routes.js

@@ -17,6 +17,20 @@ const routes = [
       }
     }
   },
+  {
+    method: 'POST',
+    path: '/jobs/import-licences',
+    options: {
+      handler: JobsController.ImportLicences,
+      app: {
+        plainOutput: true
+      },
+      auth: false,
+      plugins: {
+        crumb: false
+      }
+    }
+  },
   {
     method: 'POST',
     path: '/jobs/licence-updates',

app/services/jobs/import/fetch-licences.service.js

@@ -0,0 +1,40 @@
+'use strict'
+
+/**
+ * Fetches all licences that are in NALD and WRLS
+ * @module FetchLicences
+ */
+
+const { db } = require('../../../../db/db.js')
+
+/**
+ * Fetches all licences that are in NALD and WRLS
+ *
+ * A licence is valid if at least one licence version is not in draft
+ *
+ * @returns {Promise<object[]>} - An array of licences
+ */
+async function go () {
+  const query = `
+    SELECT DISTINCT ON (nal."LIC_NO")
+      l.id as id,
+      TO_DATE(NULLIF(nal."EXPIRY_DATE", 'null'), 'DD/MM/YYYY') AS expired_date,
+      TO_DATE(NULLIF(nal."LAPSED_DATE", 'null'), 'DD/MM/YYYY') AS lapsed_date,
+      TO_DATE(NULLIF(nal."EXPIRY_DATE", 'null'), 'DD/MM/YYYY') AS revoked_date
+    FROM import."NALD_ABS_LIC_VERSIONS" nalv
+      INNER JOIN import."NALD_ABS_LICENCES" nal
+        ON nal."ID" = nalv."AABL_ID"
+        AND nal."FGAC_REGION_CODE" = nalv."FGAC_REGION_CODE"
+      INNER JOIN
+        public.licences l ON l.licence_ref = nal."LIC_NO"
+    WHERE nalv."STATUS" <> 'DRAFT'
+  `
+
+  const { rows } = await db.raw(query)
+
+  return rows
+}
+
+module.exports = {
+  go
+}

app/services/jobs/import/import-licences.service.js

@@ -0,0 +1,111 @@
+'use strict'
+
+/**
+ * Extracts and imports licence from NALD
+ * @module ImportLicenceService
+ */
+// const pMap = require('p-map')
+
+let pMap;
+
+(async () => {
+  pMap = (await import('p-map')).default
+})()
+
+const FetchLicences = require('./fetch-licences.service.js')
+const ProcessImportLicence = require('./process-import-licence.service.js')
+const config = require('../../../../config/jobs.config.js')
+const { calculateAndLogTimeTaken, currentTimeInNanoseconds } = require('../../../lib/general.lib.js')
+
+const { batchSize } = config.importLicence
+
+/**
+ * Processes NALD licences due for import on a nightly basis
+ *
+ * Overnight the {@link https://github.com/DEFRA/water-abstraction-import | water-abstraction-import} app imports each
+ * abstraction licence found in NALD. New licences are added, existing ones are updated.
+ *
+ * When an existing licence is imported, there are some additional processes we need to trigger. The first is
+ * `DetermineSupplementaryBillingFlagsService`, which checks whether a licence needs to be flagged for supplementary
+ * billing if a change has been made.
+ *
+ * For the same reason, `ProcessLicenceReturnLogsService` needs to check if any changes need to be made to a licence's
+ * return logs.
+ *
+ * Because we are migrating from the legacy apps we couldn't trigger these processes in **water-abstraction-import**.
+ * Instead, we have this job. The one caveat is that it needs to be scheduled to run _after_ the
+ * {@link https://github.com/DEFRA/water-abstraction-team/blob/main/jobs/import.md#nald-import | NALD import job},
+ * (specifically once the NALD data has been extracted and imported into the `import` schema), but _before_ the
+ * {@link https://github.com/DEFRA/water-abstraction-team/blob/main/jobs/import.md#nald-import | Licence import job}
+ * runs. This is so these processes can see the differences between the NALD licence record and ours, to determine
+ * if and what they need to do.
+ *
+ * > If a licence in NALD does not have a status of DRAFT, and at least one non-draft licence version then it will be
+ * excluded
+ *
+ */
+async function go () {
+  try {
+    const startTime = currentTimeInNanoseconds()
+
+    const licences = await FetchLicences.go()
+
+    await _processLicences(licences)
+
+    calculateAndLogTimeTaken(startTime, `Importing ${licences.length} licences from NALD`)
+  } catch (error) {
+    global.GlobalNotifier.omfg('Importing Licence job failed', null, error)
+  }
+}
+
+/**
+ * Process Licences
+ *
+ * When we process the licences we want to batch them into smaller chunks in an attempt to make the
+ * import process as efficient as possible
+ *
+ *
+ * @param {object[]} licences - An array of licences
+ *
+ * @private
+ */
+async function _processLicences (licences) {
+  // for (let i = 0; i < licences.length; i += batchSize) {
+  //   const batch = licences.slice(i, i + batchSize)
+  //
+  //   await _processBatch(batch)
+  //
+  //   console.log('Batch end: ' + i)
+  // }
+
+  if (!pMap) {
+    throw new Error('pMap is not yet loaded.')
+  }
+
+  const mapper = async (licence) => {
+    return ProcessImportLicence.go(licence)
+  }
+
+  await pMap(licences, mapper, { concurrency: 10 })
+}
+
+// /**
+//  * Process Batch
+//  *
+//  * We need to process the licences in the current batch as efficiently as possible.
+//  *
+//  * @param {object[]} batch - a licence
+//  *
+//  * @private
+//  */
+// async function _processBatch (batch) {
+//   return Promise.allSettled(
+//     batch.map(async (licence) => {
+//       await ProcessImportLicence.go(licence)
+//     })
+//   )
+// }
+
+module.exports = {
+  go
+}

app/services/jobs/import/process-import-licence.service.js

@@ -0,0 +1,39 @@
+'use strict'
+
+/**
+ * Process licence for the licences imported from NALD
+ * @module ProcessImportLicence
+ */
+
+const DetermineSupplementaryBillingFlagsService = require('../../import/determine-supplementary-billing-flags.service.js')
+const ProcessLicenceReturnLogsService = require('../return-logs/process-licence-return-logs.service.js')
+
+/**
+ * Process licence for the licences imported from NALD
+ *
+ * When the licence exists in the WRLS databse and we are improting from NALD.
+ *
+ * We need to check if the licence needs to be flagged for supplementary billing and
+ * determine if any new return logs need to be created depending on the current cycle.
+ *
+ * @param {object} licence - a licence
+ */
+async function go (licence) {
+  try {
+    const endDates = {
+      expiredDate: licence.expired_date,
+      lapsedDate: licence.lapsed_date,
+      revokedDate: licence.revoked_date
+    }
+
+    await DetermineSupplementaryBillingFlagsService.go(endDates, licence.id)
+
+    await ProcessLicenceReturnLogsService.go(licence.id)
+  } catch (error) {
+    global.GlobalNotifier.omfg(`Importing licence ${licence.id}`, null, error)
+  }
+}
+
+module.exports = {
+  go
+}

config/jobs.config.js

@@ -0,0 +1,19 @@
+'use strict'
+
+/**
+ * Config values used for jobs
+ * @module JobsConfig
+ */
+
+// We require dotenv directly in each config file to support unit tests that depend on this this subset of config.
+// Requiring dotenv in multiple places has no effect on the app when running for real.
+require('dotenv').config()
+
+const config = {
+  importLicence: {
+    batchSize: parseInt(process.env.IMORT_LICENCE_BATCH_SIZE) || 10
+  }
+
+}
+
+module.exports = config

package.json

@@ -55,6 +55,7 @@
     "knex": "^2.5.1",
     "nunjucks": "^3.2.4",
     "objection": "^3.1.1",
+    "p-map": "^7.0.2",
     "pg": "^8.11.2",
     "pg-query-stream": "^4.5.2",
     "request": "^2.88.2",

test/controllers/jobs.controller.test.js

@@ -10,10 +10,11 @@ const { expect } = Code
 
 // Things we need to stub
 const ExportService = require('../../app/services/jobs/export/export.service.js')
+const ImportLicence = require('../../app/services/jobs/import/import-licences.service.js')
 const ProcessLicenceUpdatesService = require('../../app/services/jobs/licence-updates/process-licence-updates.js')
+const ProcessReturnLogsService = require('../../app/services/jobs/return-logs/process-return-logs.service.js')
 const ProcessSessionStorageCleanupService = require('../../app/services/jobs/session-cleanup/process-session-storage-cleanup.service.js')
 const ProcessTimeLimitedLicencesService = require('../../app/services/jobs/time-limited/process-time-limited-licences.service.js')
-const ProcessReturnLogsService = require('../../app/services/jobs/return-logs/process-return-logs.service.js')
 
 // For running our service
 const { init } = require('../../app/server.js')
@@ -58,6 +59,26 @@ describe('Jobs controller', () => {
     })
   })
 
+  describe('/jobs/import-licence', () => {
+    describe('POST', () => {
+      beforeEach(() => {
+        options = { method: 'POST', url: '/jobs/import-licences' }
+      })
+
+      describe('when the request succeeds', () => {
+        beforeEach(async () => {
+          Sinon.stub(ImportLicence, 'go').resolves()
+        })
+
+        it('returns a 204 response', async () => {
+          const response = await server.inject(options)
+
+          expect(response.statusCode).to.equal(204)
+        })
+      })
+    })
+  })
+
   describe('/jobs/licence-updates', () => {
     describe('POST', () => {
       beforeEach(() => {