Fix notifier

packages/database/src/pg/impl/knex.ts

@@ -15,7 +15,7 @@ export const getKnex = () => {
       },
       pool: {
         min: Number(process.env.DB_POOL_MIN || 0),
-        max: Number(process.env.DB_POOL_MAX || 20),
+        max: Number(process.env.DB_POOL_MAX || 10),
       },
     });
   }

packages/functions/src/index.ts

@@ -1,12 +1,12 @@
 /* eslint-disable import/namespace */
 /* eslint-disable @typescript-eslint/no-var-requires */
 require('dotenv').config({ path: __dirname + '/.env' });
-import { BaseRecord, PgChanges, database } from '@buildcore/database';
+import { BaseRecord, database } from '@buildcore/database';
 import { WEN_FUNC } from '@buildcore/interfaces';
 import cors from 'cors';
 import dayjs from 'dayjs';
 import express from 'express';
-import { get, head } from 'lodash';
+import { get } from 'lodash';
 import { flattenObject } from './common';
 import * as onScheduled from './runtime/cron/index';
 import { ScheduledFunction } from './runtime/cron/scheduled';
@@ -60,25 +60,15 @@ Object.entries(flattenObject(onTriggers)).forEach(([name, config]) => {
   app.post(`/${name}`, jsonParser, loggingMiddleware(name), async (req, res) => {
     const pubSubMessage = req.body.message.data;
     const raw = Buffer.from(pubSubMessage, 'base64').toString().trim();
-    const processId = JSON.parse(raw).processId;
-    let snap: PgChanges | undefined = undefined;
-
-    const docRef = database().getCon()('changes').where({ uid: processId });
+    const change = JSON.parse(raw);
     try {
-      snap = head(await docRef);
-      if (!snap) {
-        return;
-      }
-
-      await docRef.delete();
-
       await (config as TriggeredFunction).handler({
-        ...snap.change,
-        prev: snap.change!.prev || undefined,
-        curr: snap.change!.curr || undefined,
+        ...change,
+        prev: change!.prev || undefined,
+        curr: change!.curr || undefined,
       } as PgDocEvent<BaseRecord>);
     } catch (error) {
-      logger.error('onTriggers-error', name, snap, error);
+      logger.error('onTriggers-error', name, change, error);
     } finally {
       res.sendStatus(200);
     }

packages/functions/test/notifier.ts

@@ -65,7 +65,8 @@ const notifier = async () => {
 };
 
 const notifyTriggers = async (channel: string, changeId: string) => {
-  const body = { message: { data: btoa(JSON.stringify({ processId: Number(changeId) })) } };
+  const change = await knex('changes').select('*').where({ uid: changeId });
+  const body = { message: { data: btoa(JSON.stringify(change[0].change)) } };
 
   let error: any = undefined;
   for (let i = 0; i < 5; ++i) {
@@ -80,6 +81,7 @@ const notifyTriggers = async (channel: string, changeId: string) => {
       await new Promise((resolve) => setTimeout(resolve, 1000));
     }
   }
+  await knex('change').delete().where({ uid: changeId });
   throw error;
 };
 

packages/notifier/src/index.ts

@@ -56,10 +56,17 @@ const getTriggerTopic = (topic: string) => {
   return triggerTopics[topic];
 };
 
-const notifyTriggers = async (channel: string, processId: string) => {
-  await getTriggerTopic(channel).publishMessage({
-    data: Buffer.from(JSON.stringify({ processId: Number(processId) })),
-  });
+const notifyTriggers = async (channel: string, changeId: string) => {
+  try {
+    const change = await knex('changes').select('*').where({ uid: changeId });
+    await getTriggerTopic(channel).publishMessage({
+      data: Buffer.from(JSON.stringify(change[0].change)),
+    });
+  } catch (err) {
+    logger.error(err)
+  } finally {
+    await knex('changes').delete().where({ uid: changeId });
+  }
 };
 
 const upserTopic = pubSub.topic('onupsert');