Add return requirements clean up step to import (#1017)

https://eaflood.atlassian.net/browse/WATER-4655
https://eaflood.atlassian.net/browse/WATER-4651

> Part of the work to migrate management of return requirements from NALD to WRLS

Having made return versions and requirements visible to our internal users they have spotted that there are return requirements displayed in WRLS that don't exist.

We know why this is: NALD allows users to delete records. This means a user can create a record in NALD on day 1; we'll see and import the record that night. They can then delete the record (in most cases, it is because a mistake was spotted) on day 2. The import _never deletes_ records, so the errant return requirement remains.

With us taking over management of return requirements, though, it has become important to try to get the 2 systems in sync as much as possible before the import is switched off.

This change will add a new step to the return versions import job to tackle clean-up. First, it will focus on return requirements that no longer exist in NALD and that do not have a return log in WRLS.

We'll look at return requirements that _are_ linked to return logs in future changes.

src/modules/return-versions/controller.js

@@ -1,10 +1,10 @@
 'use strict'
 
-const ImportJob = require('./jobs/import.js')
+const CleanJob = require('./jobs/clean.js')
 
 async function importReturnVersions (request, h) {
- await request.messageQueue.deleteQueue(ImportJob.JOB_NAME)
- await request.messageQueue.publish(ImportJob.createMessage())
+ await request.messageQueue.deleteQueue(CleanJob.JOB_NAME)
+ await request.messageQueue.publish(CleanJob.createMessage())
 
  return h.response().code(204)
 }

src/modules/return-versions/jobs/clean.js

@@ -0,0 +1,57 @@
+'use strict'
+
+const { pool } = require('../../../lib/connectors/db.js')
+const Queries = require('../lib/clean-queries.js')
+const ImportJob = require('./import.js')
+
+const JOB_NAME = 'return-versions.clean'
+
+function createMessage () {
+ return {
+ name: JOB_NAME,
+ options: {
+ singletonKey: JOB_NAME
+ }
+ }
+}
+
+async function handler () {
+ try {
+ global.GlobalNotifier.omg(`${JOB_NAME}: started`)
+
+ // Delete any return requirement points linked to deleted NALD return requirements
+ await pool.query(Queries.cleanPoints)
+
+ // Delete any return requirement purposes linked to deleted NALD return requirements
+ await pool.query(Queries.cleanPurposes)
+
+ // Delete any return requirements linked to deleted NALD return requirements
+ await pool.query(Queries.cleanRequirements)
+
+ // Delete any return versions that have no return requirements and that are linked to deleted return versions
+ await pool.query(Queries.cleanVersions)
+
+ // Update the mod logs to remove the return version ID for where the return version has now been deleted
+ await pool.query(Queries.cleanModLogs)
+ } catch (error) {
+ global.GlobalNotifier.omfg(`${JOB_NAME}: errored`, error)
+ throw error
+ }
+}
+
+async function onComplete (messageQueue, job) {
+ if (!job.failed) {
+ await messageQueue.publish(ImportJob.createMessage())
+
+ global.GlobalNotifier.omg(`${JOB_NAME}: finished`)
+ } else {
+ global.GlobalNotifier.omg(`${JOB_NAME}: failed`)
+ }
+}
+
+module.exports = {
+ JOB_NAME,
+ createMessage,
+ handler,
+ onComplete
+}

src/modules/return-versions/jobs/import.js

@@ -1,7 +1,7 @@
 'use strict'
 
 const { pool } = require('../../../lib/connectors/db.js')
-const Queries = require('../lib/queries.js')
+const Queries = require('../lib/import-queries.js')
 
 const JOB_NAME = 'return-versions.import'
 

src/modules/return-versions/lib/clean-queries.js

@@ -0,0 +1,128 @@
+'use strict'
+
+const cleanPoints = `
+ WITH nald_return_requirements AS (
+ SELECT concat_ws(':', nrf."FGAC_REGION_CODE", nrf."ID") AS nald_id
+ FROM "import"."NALD_RET_FORMATS" nrf
+ )
+ DELETE FROM water.return_requirement_points rrp WHERE rrp.return_requirement_id IN (
+ SELECT
+ rr.return_requirement_id
+ FROM
+ water.return_requirements rr
+ WHERE
+ NOT EXISTS (
+ SELECT 1
+ FROM nald_return_requirements nrr
+ WHERE rr.external_id = nrr.nald_id
+ )
+ AND NOT EXISTS (
+ SELECT 1
+ FROM "returns"."returns" rl
+ WHERE
+ rl.return_requirement = rr.legacy_id::varchar
+ LIMIT 1
+ )
+ );
+`
+
+const cleanPurposes = `
+ WITH nald_return_requirements AS (
+ SELECT concat_ws(':', nrf."FGAC_REGION_CODE", nrf."ID") AS nald_id
+ FROM "import"."NALD_RET_FORMATS" nrf
+ )
+ DELETE FROM water.return_requirement_purposes rrp WHERE rrp.return_requirement_id IN (
+ SELECT
+ rr.return_requirement_id
+ FROM
+ water.return_requirements rr
+ WHERE
+ NOT EXISTS (
+ SELECT 1
+ FROM nald_return_requirements nrr
+ WHERE rr.external_id = nrr.nald_id
+ )
+ AND NOT EXISTS (
+ SELECT 1
+ FROM "returns"."returns" rl
+ WHERE
+ rl.return_requirement = rr.legacy_id::varchar
+ LIMIT 1
+ )
+ );
+`
+
+const cleanRequirements = `
+ WITH nald_return_requirements AS (
+ SELECT concat_ws(':', nrf."FGAC_REGION_CODE", nrf."ID") AS nald_id
+ FROM "import"."NALD_RET_FORMATS" nrf
+ )
+ DELETE FROM water.return_requirements WHERE return_requirement_id IN (
+ SELECT
+ rr.return_requirement_id
+ FROM
+ water.return_requirements rr
+ WHERE
+ NOT EXISTS (
+ SELECT 1
+ FROM nald_return_requirements nrr
+ WHERE rr.external_id = nrr.nald_id
+ )
+ AND NOT EXISTS (
+ SELECT 1
+ FROM "returns"."returns" rl
+ WHERE
+ rl.return_requirement = rr.legacy_id::varchar
+ LIMIT 1
+ )
+ );
+`
+
+const cleanVersions = `
+ WITH nald_return_versions AS (
+ SELECT concat_ws(':', nv."FGAC_REGION_CODE", nv."AABL_ID", nv."VERS_NO") AS nald_id
+ FROM "import"."NALD_RET_VERSIONS" nv
+ )
+ DELETE FROM water.return_versions WHERE return_version_id IN (
+ SELECT
+ rv.return_version_id
+ FROM
+ water.return_versions rv
+ WHERE
+ NOT EXISTS (
+ SELECT 1
+ FROM nald_return_versions nrv
+ WHERE rv.external_id = nrv.nald_id
+ )
+ AND NOT EXISTS (
+ SELECT 1
+ FROM water.return_requirements rr
+ WHERE
+ rr.return_version_id = rv.return_version_id
+ LIMIT 1
+ )
+ );
+`
+
+const cleanModLogs = `
+ UPDATE
+ water.mod_logs ml
+ SET
+ return_version_id = NULL
+ WHERE
+ ml.return_version_id IS NOT NULL
+ AND ml.return_version_id NOT IN (
+ SELECT
+ rv.return_version_id
+ FROM
+ water.return_versions rv
+ );
+`
+
+module.exports = {
+ cleanPoints,
+ cleanPurposes,
+ cleanRequirements,
+ cleanVersions,
+ cleanModLogs
+}

src/modules/return-versions/plugin.js

@@ -2,20 +2,27 @@
 
 const cron = require('node-cron')
 
+const CleanJob = require('./jobs/clean.js')
 const ImportJob = require('./jobs/import.js')
 
 const config = require('../../../config')
 
 async function register (server, _options) {
+ // Register clean return versions job
+ await server.messageQueue.subscribe(CleanJob.JOB_NAME, CleanJob.handler)
+ await server.messageQueue.onComplete(CleanJob.JOB_NAME, (executedJob) => {
+ return CleanJob.onComplete(server.messageQueue, executedJob)
+ })
+
  // Register import return versions job
  await server.messageQueue.subscribe(ImportJob.JOB_NAME, ImportJob.handler)
  await server.messageQueue.onComplete(ImportJob.JOB_NAME, (executedJob) => {
  return ImportJob.onComplete(executedJob)
  })
 
- // Set up import on cron job
+ // Schedule clean job using cron. The clean job will then queue the import job in its onComplete
  cron.schedule(config.import.returnVersions.schedule, async () => {
- await server.messageQueue.publish(ImportJob.createMessage())
+ await server.messageQueue.publish(CleanJob.createMessage())
  })
 }