Merge pull request #3445 from uselagoon/subscription-improvements

Consolidate api subscription events

services/api/src/clients/pubSub.ts

@@ -8,6 +8,12 @@ import { query } from '../util/db';
 import { Sql as environmentSql } from '../resources/environment/sql';
 import { ResolverFn } from '../resources';
 
+export const EVENTS = {
+  DEPLOYMENT: 'api.subscription.deployment',
+  BACKUP: 'api.subscription.backup',
+  TASK: 'api.subscription.task'
+};
+
 export const config = {
   host: getConfigFromEnv('RABBITMQ_HOST', 'broker'),
   user: getConfigFromEnv('RABBITMQ_USERNAME', 'guest'),

services/api/src/resources/backup/resolvers.ts

@@ -6,7 +6,8 @@ import { getConfigFromEnv } from '../../util/config';
 import { query, isPatchEmpty, knex } from '../../util/db';
 import {
   pubSub,
-  createEnvironmentFilteredSubscriber
+  createEnvironmentFilteredSubscriber,
+  EVENTS
 } from '../../clients/pubSub';
 import S3 from 'aws-sdk/clients/s3';
 import { Sql } from './sql';
@@ -15,7 +16,6 @@ import { Sql as environmentSql } from '../environment/sql';
 import { Helpers as environmentHelpers } from '../environment/helpers';
 import { Helpers as projectHelpers } from '../project/helpers';
 import { getEnvVarsByProjectId } from '../env-variables/resolvers';
-import { EVENTS } from './events';
 import { logger } from '../../loggers/logger';
 
 export const getRestoreLocation: ResolverFn = async (
@@ -191,7 +191,7 @@ export const addBackup: ResolverFn = async (
   const rows = await query(sqlClientPool, Sql.selectBackup(insertId));
   const backup = R.prop(0, rows);
 
-  pubSub.publish(EVENTS.BACKUP.ADDED, backup);
+  pubSub.publish(EVENTS.BACKUP, backup);
 
   userActivityLogger(`User deployed backup '${backupId}' to '${environment.name}' on project '${environment.project}'`, {
     project: '',
@@ -222,9 +222,6 @@ export const deleteBackup: ResolverFn = async (
 
   await query(sqlClientPool, Sql.deleteBackup(backupId));
 
-  const rows = await query(sqlClientPool, Sql.selectBackupByBackupId(backupId));
-  pubSub.publish(EVENTS.BACKUP.DELETED, R.prop(0, rows));
-
   userActivityLogger(`User deleted backup '${backupId}'`, {
     project: '',
     event: 'api:deleteBackup',
@@ -278,7 +275,7 @@ export const addRestore: ResolverFn = async (
   rows = await query(sqlClientPool, Sql.selectBackupByBackupId(backupId));
   const backupData = R.prop(0, rows);
 
-  pubSub.publish(EVENTS.BACKUP.UPDATED, backupData);
+  pubSub.publish(EVENTS.BACKUP, backupData);
 
   // Allow creating restore data w/o executing the restore
   if (execute === false) {
@@ -389,7 +386,7 @@ export const updateRestore: ResolverFn = async (
   rows = await query(sqlClientPool, Sql.selectBackupByBackupId(backupId));
   const backupData = R.prop(0, rows);
 
-  pubSub.publish(EVENTS.BACKUP.UPDATED, backupData);
+  pubSub.publish(EVENTS.BACKUP, backupData);
 
   userActivityLogger(`User updated restore '${backupId}'`, {
     project: '',
@@ -418,7 +415,5 @@ export const getRestoreByBackupId: ResolverFn = async (
 };
 
 export const backupSubscriber = createEnvironmentFilteredSubscriber([
-  EVENTS.BACKUP.ADDED,
-  EVENTS.BACKUP.UPDATED,
-  EVENTS.BACKUP.DELETED
+  EVENTS.BACKUP
 ]);

services/api/src/resources/deployment/resolvers.ts

@@ -13,13 +13,13 @@ import {
 import { ResolverFn } from '../';
 import {
   pubSub,
-  createEnvironmentFilteredSubscriber
+  createEnvironmentFilteredSubscriber,
+  EVENTS
 } from '../../clients/pubSub';
 import { getConfigFromEnv, getLagoonRouteFromEnv } from '../../util/config';
 import { knex, query, isPatchEmpty } from '../../util/db';
 import { Sql } from './sql';
 import { Helpers } from './helpers';
-import { EVENTS } from './events';
 import { Helpers as environmentHelpers } from '../environment/helpers';
 import { Helpers as projectHelpers } from '../project/helpers';
 // @ts-ignore
@@ -342,7 +342,7 @@ export const addDeployment: ResolverFn = async (
   const rows = await query(sqlClientPool, Sql.selectDeployment(insertId));
   const deployment = R.prop(0, rows);
 
-  pubSub.publish(EVENTS.DEPLOYMENT.ADDED, deployment);
+  pubSub.publish(EVENTS.DEPLOYMENT, deployment);
   return deployment;
 };
 
@@ -440,7 +440,7 @@ export const updateDeployment: ResolverFn = async (
   const rows = await query(sqlClientPool, Sql.selectDeployment(id));
   const deployment = R.prop(0, rows);
 
-  pubSub.publish(EVENTS.DEPLOYMENT.UPDATED, deployment);
+  pubSub.publish(EVENTS.DEPLOYMENT, deployment);
 
   userActivityLogger(`User updated deployment '${id}'`, {
     project: '',
@@ -1360,6 +1360,5 @@ export const bulkDeployEnvironmentLatest: ResolverFn = async (
 };
 
 export const deploymentSubscriber = createEnvironmentFilteredSubscriber([
-  EVENTS.DEPLOYMENT.ADDED,
-  EVENTS.DEPLOYMENT.UPDATED
+  EVENTS.DEPLOYMENT
 ]);

services/api/src/resources/task/helpers.ts

@@ -3,9 +3,8 @@ import { Pool } from 'mariadb';
 import { sendToLagoonLogs } from '@lagoon/commons/dist/logs/lagoon-logger';
 import { createTaskTask, createMiscTask } from '@lagoon/commons/dist/tasks';
 import { query } from '../../util/db';
-import { pubSub } from '../../clients/pubSub';
+import { pubSub, EVENTS } from '../../clients/pubSub';
 import { Sql } from './sql';
-import { EVENTS } from './events';
 import { Sql as projectSql } from '../project/sql';
 import { Sql as environmentSql } from '../environment/sql';
 import convertDateToMYSQLDateTimeFormat from '../../util/convertDateToMYSQLDateTimeFormat';
@@ -67,7 +66,7 @@ export const Helpers = (sqlClientPool: Pool) => ({
     let rows = await query(sqlClientPool, Sql.selectTask(insertId));
     const taskData = R.prop(0, rows);
 
-    pubSub.publish(EVENTS.TASK.ADDED, taskData);
+    pubSub.publish(EVENTS.TASK, taskData);
 
     // Allow creating task data w/o executing the task
     if (execute === false) {
@@ -199,8 +198,6 @@ export const Helpers = (sqlClientPool: Pool) => ({
       }
     }
 
-    pubSub.publish(EVENTS.TASK.ADDED, jobSpec);
-
     try {
       await createMiscTask(
         {

services/api/src/resources/task/resolvers.ts

@@ -2,11 +2,11 @@ import * as R from 'ramda';
 import { ResolverFn } from '../';
 import {
   pubSub,
-  createEnvironmentFilteredSubscriber
+  createEnvironmentFilteredSubscriber,
+  EVENTS
 } from '../../clients/pubSub';
 import { knex, query, isPatchEmpty } from '../../util/db';
 import { Sql } from './sql';
-import { EVENTS } from './events';
 import { Helpers } from './helpers';
 import { Filters } from './filters';
 import { Helpers as environmentHelpers } from '../environment/helpers';
@@ -393,7 +393,7 @@ export const updateTask: ResolverFn = async (
   const rows = await query(sqlClientPool, Sql.selectTask(id));
   const taskData = R.prop(0, rows);
 
-  pubSub.publish(EVENTS.TASK.UPDATED, taskData);
+  pubSub.publish(EVENTS.TASK, taskData);
 
   userActivityLogger(`User updated task '${id}'`, {
     project: '',
@@ -790,6 +790,5 @@ export const taskDrushUserLogin: ResolverFn = async (
 };
 
 export const taskSubscriber = createEnvironmentFilteredSubscriber([
-  EVENTS.TASK.ADDED,
-  EVENTS.TASK.UPDATED
+  EVENTS.TASK
 ]);