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

app/controllers/jobs.controller.js

@@ -6,10 +6,11 @@
  */
 
 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 NO_CONTENT_STATUS_CODE = 204
 const NOT_FOUND_STATUS_CODE = 404
@@ -29,6 +30,12 @@ async function exportDb(_request, h) {
   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()
 
@@ -54,13 +61,22 @@ async function returnLogs(request, h) {
     return h.response().code(NOT_FOUND_STATUS_CODE)
   }
 
+  let licenceReference
+
   ProcessReturnLogsService.go(cycle)
 
+  if (h.request.payload !== null && h.request.payload.licenceReference) {
+    licenceReference = h.request.payload.licenceReference
+  }
+
+  ProcessReturnLogsService.go(cycle, licenceReference)
+
   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,83 @@
+'use strict'
+
+/**
+ * Extracts and imports licence from NALD
+ * @module ImportLicenceService
+ */
+
+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) {
+  if (!pMap) {
+    throw new Error('pMap is not yet loaded.')
+  }
+
+  const mapper = async (licence) => {
+    return ProcessImportLicence.go(licence)
+  }
+
+  await pMap(licences, mapper, { concurrency: batchSize })
+}
+
+module.exports = {
+  go
+}

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

@@ -0,0 +1,44 @@
+'use strict'
+
+/**
+ * Process licence for the licences imported from NALD
+ * @module ProcessImportLicence
+ */
+
+const DetermineLicenceEndDateChangedService = require('../../import/determine-licence-end-date-changed.service')
+const ProcessBillingFlagService = require('../../licences/supplementary/process-billing-flag.service')
+const GenerateReturnLogsService = require('../../import/generate-return-logs.service')
+
+/**
+ * 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 payload = {
+      expiredDate: licence.expired_date,
+      lapsedDate: licence.lapsed_date,
+      revokedDate: licence.revoked_date,
+      licenceId: licence.id
+    }
+
+    const licenceEndDateChanged = await DetermineLicenceEndDateChangedService.go(payload, licence.id)
+
+    if (licenceEndDateChanged) {
+      ProcessBillingFlagService.go(payload)
+      GenerateReturnLogsService.go(licence.id, payload)
+    }
+  } catch (error) {
+    global.GlobalNotifier.omfg(`Importing licence ${licence.id}`, null, error)
+  }
+}
+
+module.exports = {
+  go
+}

config/jobs.config.js

@@ -0,0 +1,18 @@
+'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')
@@ -60,6 +61,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(() => {