Handle api SIGTERM gracefully (#368)

Author: Uros Marolt

backend/package.json

@@ -27,7 +27,8 @@
     "lint": "npx eslint .",
     "format": "npx prettier --write .",
     "script:process-integration": "SERVICE=script ts-node ./src/bin/scripts/process-integration.ts",
-    "script:change-tenant-plan": "SERVICE=script ts-node ./src/bin/scripts/change-tenant-plan.ts"
+    "script:change-tenant-plan": "SERVICE=script ts-node ./src/bin/scripts/change-tenant-plan.ts",
+    "script:process-webhook": "SERVICE=script ts-node ./src/bin/scripts/process-webhook.ts"
   },
   "dependencies": {
     "@aws-sdk/client-comprehend": "^3.159.0",

backend/src/bin/api.ts

@@ -1,11 +1,31 @@
 import { getServiceLogger } from '../utils/logging'
 import api from '../api'
 import { API_CONFIG } from '../config'
+import { timeout } from '../utils/timing'
 
 const PORT = API_CONFIG.port || 8080
 
 const log = getServiceLogger()
 
-api.listen(PORT, () => {
+const server = api.listen(PORT, () => {
   log.info(`Listening on port ${PORT}`)
 })
+
+process.on('SIGTERM', async () => {
+  log.warn('Detected SIGTERM signal, started exiting!')
+  await new Promise<void>((resolve) => {
+    server.close((err) => {
+      if (err) {
+        log.error(err, 'Error while closing server!')
+        resolve()
+      } else {
+        log.info('Server closed successfully!')
+        resolve()
+      }
+    })
+  })
+
+  log.info('Exiting in 5 seconds...')
+  await timeout(5000)
+  process.exit(0)
+})

backend/src/bin/nodejs-worker.ts

@@ -10,6 +10,7 @@ import { createChildLogger, getServiceLogger, Logger } from '../utils/logging'
 import { deleteMessage, receiveMessage, sendMessage } from '../utils/sqs'
 import { timeout } from '../utils/timing'
 import { processIntegration, processIntegrationCheck } from './worker/integrations'
+import { processWebhook } from '../serverless/integrations/workers/githubWebhookWorker'
 
 /* eslint-disable no-constant-condition */
 
@@ -137,6 +138,9 @@ async function handleMessages() {
         case NodeWorkerMessageType.DB_OPERATIONS:
           processFunction = processDbOperationsMessage
           break
+        case NodeWorkerMessageType.PROCESS_WEBHOOK:
+          processFunction = processWebhook
+          break
 
         default:
           keep = true

backend/src/bin/scripts/process-integration.ts

@@ -41,7 +41,7 @@ const sections = [
   },
   {
     header: 'Process Integration',
-    content: 'Trigger processing of a single integration.',
+    content: 'Trigger processing of integrations.',
   },
   {
     header: 'Options',
@@ -58,9 +58,9 @@ if (parameters.help || !parameters.integration) {
   setImmediate(async () => {
     const integrationIds = parameters.integration.split(',')
     const onboarding = parameters.onboarding
+    const options = await SequelizeRepository.getDefaultIRepositoryOptions()
 
     for (const integrationId of integrationIds) {
-      const options = await SequelizeRepository.getDefaultIRepositoryOptions()
       const integration = await options.database.integration.findOne({
         where: { id: integrationId },
       })

backend/src/bin/scripts/process-webhook.ts

@@ -0,0 +1,77 @@
+import commandLineArgs from 'command-line-args'
+import commandLineUsage from 'command-line-usage'
+import * as fs from 'fs'
+import path from 'path'
+import { createServiceLogger } from '../../utils/logging'
+import SequelizeRepository from '../../database/repositories/sequelizeRepository'
+import { sendNodeWorkerMessage } from '../../serverless/utils/nodeWorkerSQS'
+import IncomingWebhookRepository from '../../database/repositories/incomingWebhookRepository'
+import { WebhookState } from '../../types/webhooks'
+import { NodeWorkerProcessWebhookMessage } from '../../types/mq/nodeWorkerProcessWebhookMessage'
+
+const banner = fs.readFileSync(path.join(__dirname, 'banner.txt'), 'utf8')
+
+const log = createServiceLogger()
+
+const options = [
+  {
+    name: 'webhook',
+    alias: 'w',
+    typeLabel: '{underline webhookId}',
+    type: String,
+    description:
+      'The unique ID of webhook that you would like to process. Use comma delimiter when sending multiple webhooks.',
+  },
+  {
+    name: 'help',
+    alias: 'h',
+    type: Boolean,
+    description: 'Print this usage guide.',
+  },
+]
+const sections = [
+  {
+    content: banner,
+    raw: true,
+  },
+  {
+    header: 'Process Webhook',
+    content: 'Trigger processing of webhooks.',
+  },
+  {
+    header: 'Options',
+    optionList: options,
+  },
+]
+
+const usage = commandLineUsage(sections)
+const parameters = commandLineArgs(options)
+
+if (parameters.help || !parameters.webhook) {
+  console.log(usage)
+} else {
+  setImmediate(async () => {
+    const webhookIds = parameters.webhook.split(',')
+    const options = await SequelizeRepository.getDefaultIRepositoryOptions()
+    const repo = new IncomingWebhookRepository(options)
+
+    for (const webhookId of webhookIds) {
+      const webhook = await repo.findById(webhookId)
+
+      if (!webhook) {
+        log.error({ webhookId }, 'Webhook not found!')
+        process.exit(1)
+      } else if (webhook.state !== WebhookState.PENDING) {
+        log.error({ webhookId }, 'Webhook is not in pending state!')
+        process.exit(1)
+      } else {
+        log.info({ webhookId }, 'Webhook found - triggering SQS message!')
+        await sendNodeWorkerMessage(
+          webhook.tenantId,
+          new NodeWorkerProcessWebhookMessage(webhook.tenantId, webhook.id),
+        )
+      }
+    }
+    process.exit(0)
+  })
+}

backend/src/database/migrations/U1671451319__incoming-webhooks.sql

@@ -0,0 +1 @@
+drop table "incomingWebhooks";

backend/src/database/migrations/V1671451319__incoming-webhooks.sql

@@ -0,0 +1,16 @@
+create table "incomingWebhooks" (
+    id              uuid         not null,
+    "tenantId"      uuid         not null,
+    "integrationId" uuid         not null,
+    state           varchar(255) not null,
+
+    type            varchar(255) not null,
+    payload         json         not null,
+
+    "processedAt"   timestamptz  null,
+    error           json         null,
+
+    "createdAt"     timestamptz  not null default now(),
+
+    primary key (id)
+);

backend/src/database/repositories/incomingWebhookRepository.ts

@@ -0,0 +1,156 @@
+import { v4 as uuid } from 'uuid'
+import { QueryTypes } from 'sequelize'
+import {
+  DbIncomingWebhookInsertData,
+  IncomingWebhookData,
+  WebhookState,
+} from '../../types/webhooks'
+import { IRepositoryOptions } from './IRepositoryOptions'
+import { RepositoryBase } from './repositoryBase'
+
+/* eslint-disable class-methods-use-this */
+
+export default class IncomingWebhookRepository extends RepositoryBase<
+  IncomingWebhookData,
+  string,
+  DbIncomingWebhookInsertData,
+  unknown,
+  unknown
+> {
+  public constructor(options: IRepositoryOptions) {
+    super(options, true)
+  }
+
+  async create(data: DbIncomingWebhookInsertData): Promise<IncomingWebhookData> {
+    const transaction = this.transaction
+
+    const id = uuid()
+
+    const results = await this.seq.query(
+      `
+    insert into "incomingWebhooks"(id, "tenantId", "integrationId", state, type, payload)
+    values(:id, :tenantId, :integrationId, :state, :type, :payload)
+    returning "createdAt"
+    `,
+      {
+        replacements: {
+          id,
+          tenantId: data.tenantId,
+          integrationId: data.integrationId,
+          type: data.type,
+          state: WebhookState.PENDING,
+          payload: JSON.stringify(data.payload),
+        },
+        type: QueryTypes.INSERT,
+        transaction,
+      },
+    )
+
+    return {
+      id,
+      state: WebhookState.PENDING,
+      ...data,
+      processedAt: null,
+      error: null,
+      createdAt: results[0][0].createdAt.toISOString(),
+    }
+  }
+
+  override async findById(id: string): Promise<IncomingWebhookData> {
+    const transaction = this.transaction
+
+    const seq = this.seq
+
+    const results = await seq.query(
+      `
+      select id,
+             "tenantId",
+             "integrationId",
+             state,
+             type,
+             payload,
+             "processedAt",
+             error,
+             "createdAt"
+      from "incomingWebhooks"
+      where id = :id
+    `,
+      {
+        replacements: {
+          id,
+        },
+        type: QueryTypes.SELECT,
+        transaction,
+      },
+    )
+
+    if (results.length === 0) {
+      return null
+    }
+
+    const data = results[0] as any
+
+    return {
+      id: data.id,
+      tenantId: data.tenantId,
+      integrationId: data.integrationId,
+      state: data.state,
+      type: data.type,
+      payload: data.payload,
+      processedAt: data.processedAt ? data.processedAt.toISOString() : null,
+      error: data.error,
+      createdAt: data.createdAt.toISOString(),
+    }
+  }
+
+  async markCompleted(id: string): Promise<void> {
+    const transaction = this.transaction
+
+    const [, rowCount] = await this.seq.query(
+      `
+      update "incomingWebhooks"
+      set state = :state,
+          "processedAt" = now()
+      where id = :id
+    `,
+      {
+        replacements: {
+          id,
+          state: WebhookState.PROCESSED,
+        },
+        type: QueryTypes.UPDATE,
+        transaction,
+      },
+    )
+
+    if (rowCount !== 1) {
+      throw new Error(`Failed to mark webhook '${id}' as completed!`)
+    }
+  }
+
+  async markError(id: string, error: any): Promise<void> {
+    const transaction = this.transaction
+
+    const [, rowCount] = await this.seq.query(
+      `
+      update "incomingWebhooks"
+      set state = :state,
+          error = :error
+      where id = :id
+    `,
+      {
+        replacements: {
+          id,
+          state: WebhookState.ERROR,
+          error: JSON.stringify(error),
+        },
+        type: QueryTypes.UPDATE,
+        transaction,
+      },
+    )
+
+    if (rowCount !== 1) {
+      throw new Error(`Failed to mark webhook '${id}' as error!`)
+    }
+  }
+}

backend/src/database/repositories/repositoryBase.ts

@@ -37,19 +37,29 @@ export abstract class RepositoryBase<
     return this.options.database
   }
 
-  abstract create(data: TCreate): Promise<TData>
+  async create(data: TCreate): Promise<TData> {
+    throw new Error('Method not implemented.')
+  }
 
-  abstract update(id: TId, data: TUpdate): Promise<TData>
+  async update(id: TId, data: TUpdate): Promise<TData> {
+    throw new Error('Method not implemented.')
+  }
 
   async destroy(id: TId): Promise<void> {
     return this.destroyAll([id])
   }
 
-  abstract destroyAll(ids: TId[]): Promise<void>
+  async destroyAll(ids: TId[]): Promise<void> {
+    throw new Error('Method not implemented.')
+  }
 
-  abstract findById(id: TId): Promise<TData>
+  async findById(id: TId): Promise<TData> {
+    throw new Error('Method not implemented.')
+  }
 
-  abstract findAndCountAll(criteria: TCriteria): Promise<PageData<TData>>
+  async findAndCountAll(criteria: TCriteria): Promise<PageData<TData>> {
+    throw new Error('Method not implemented.')
+  }
 
   async findAll(criteria: TCriteria): Promise<TData[]> {
     const copy = { ...criteria }