feat(webhook-retries): produce and consume retries (3) (#1940)

package.json

@@ -142,6 +142,8 @@
     "slick-carousel": "1.8.1",
     "sortablejs": "~1.13.0",
     "spark-md5": "^3.0.1",
+    "sqs-consumer": "^5.5.0",
+    "sqs-producer": "^2.1.0",
     "text-encoding": "^0.7.0",
     "toastr": "^2.1.4",
     "triple-beam": "^1.3.0",
@@ -154,7 +156,8 @@
     "web-streams-polyfill": "^3.0.3",
     "whatwg-fetch": "^3.6.2",
     "winston": "^3.3.3",
-    "winston-cloudwatch": "^2.5.2"
+    "winston-cloudwatch": "^2.5.2",
+    "zod": "^3.0.0"
   },
   "devDependencies": {
     "@babel/core": "^7.14.3",

src/app/config/feature-manager/types.ts

@@ -79,6 +79,7 @@ export interface IVerifiedFields {
 
 export interface IWebhookVerifiedContent {
   signingSecretKey: string
+  webhookQueueUrl: string
 }
 
 export interface IIntranet {

src/app/config/feature-manager/webhook-verified-content.config.ts

@@ -10,6 +10,12 @@ const webhookVerifiedContentFeature: RegisterableFeature<FeatureNames.WebhookVer
       default: null,
       env: 'SIGNING_SECRET_KEY',
     },
+    webhookQueueUrl: {
+      doc: 'URL of AWS SQS queue for webhook retries',
+      format: String,
+      default: '',
+      env: 'WEBHOOK_SQS_URL',
+    },
   },
 }
 

src/app/constants/timezone.ts

@@ -0,0 +1 @@
+export const TIMEZONE = 'Asia/Singapore'

src/app/models/submission.server.model.ts

@@ -9,6 +9,7 @@ import {
   IEmailSubmissionSchema,
   IEncryptedSubmissionSchema,
   IEncryptSubmissionModel,
+  IPopulatedWebhookSubmission,
   ISubmissionModel,
   ISubmissionSchema,
   IWebhookResponse,
@@ -17,6 +18,7 @@ import {
   SubmissionCursorData,
   SubmissionMetadata,
   SubmissionType,
+  SubmissionWebhookInfo,
   WebhookData,
   WebhookView,
 } from '../../types'
@@ -180,10 +182,13 @@ const EncryptSubmissionSchema = new Schema<
  * which will be posted to the webhook URL.
  */
 EncryptSubmissionSchema.methods.getWebhookView = function (
-  this: IEncryptedSubmissionSchema,
+  this: IEncryptedSubmissionSchema | IPopulatedWebhookSubmission,
 ): WebhookView {
+  const formId = this.populated('form')
+    ? String(this.form._id)
+    : String(this.form)
   const webhookData: WebhookData = {
-    formId: String(this.form),
+    formId,
     submissionId: String(this._id),
     encryptedContent: this.encryptedContent,
     verifiedContent: this.verifiedContent,
@@ -208,6 +213,22 @@ EncryptSubmissionSchema.statics.addWebhookResponse = function (
   ).exec()
 }
 
+EncryptSubmissionSchema.statics.retrieveWebhookInfoById = function (
+  this: IEncryptSubmissionModel,
+  submissionId: string,
+): Promise<SubmissionWebhookInfo | null> {
+  return this.findById(submissionId)
+    .populate('form', 'webhook')
+    .then((populatedSubmission: IPopulatedWebhookSubmission | null) => {
+      if (!populatedSubmission) return null
+      return {
+        webhookUrl: populatedSubmission.form.webhook?.url ?? '',
+        isRetryEnabled: !!populatedSubmission.form.webhook?.isRetryEnabled,
+        webhookView: populatedSubmission.getWebhookView(),
+      }
+    })
+}
+
 EncryptSubmissionSchema.statics.findSingleMetadata = function (
   this: IEncryptSubmissionModel,
   formId: string,

src/app/modules/submission/encrypt-submission/encrypt-submission.controller.ts

@@ -343,15 +343,14 @@ const submitEncryptModeForm: RequestHandler = async (req, res) => {
   })
 
   // Fire webhooks if available
-  // Note that we push data to webhook endpoints on a best effort basis
-  // As such, we should not await on these post requests
+  // To avoid being coupled to latency of receiving system,
+  // do not await on webhook
   const webhookUrl = form.webhook?.url
   if (webhookUrl) {
-    void WebhookFactory.sendWebhook(
+    void WebhookFactory.sendInitialWebhook(
       submission,
       webhookUrl,
-    ).andThen((response) =>
-      WebhookFactory.saveWebhookRecord(submission._id, response),
+      !!form.webhook?.isRetryEnabled,
     )
   }
 

src/app/modules/submission/submission.errors.ts

@@ -12,7 +12,7 @@ export class ConflictError extends ApplicationError {
 }
 
 export class SubmissionNotFoundError extends ApplicationError {
-  constructor(message: string) {
+  constructor(message = 'Submission not found for given ID') {
     super(message)
   }
 }

src/app/modules/webhook/__tests__/webhook.service.spec.ts

@@ -138,6 +138,7 @@ describe('webhook.service', () => {
         'X-FormSG-Signature': `t=${MOCK_EPOCH},s=${testEncryptedSubmission._id},f=${testEncryptedForm._id},v1=${testSignature}`,
       },
       maxRedirects: 0,
+      timeout: 10000,
     }
   })
 
@@ -237,7 +238,7 @@ describe('webhook.service', () => {
 
       // Act
       const actual = await sendWebhook(
-        testEncryptedSubmission,
+        testEncryptedSubmission.getWebhookView(),
         MOCK_WEBHOOK_URL,
       )
 
@@ -258,7 +259,7 @@ describe('webhook.service', () => {
 
       // Act
       const actual = await sendWebhook(
-        testEncryptedSubmission,
+        testEncryptedSubmission.getWebhookView(),
         MOCK_WEBHOOK_URL,
       )
 
@@ -295,7 +296,7 @@ describe('webhook.service', () => {
 
       // Act
       const actual = await sendWebhook(
-        testEncryptedSubmission,
+        testEncryptedSubmission.getWebhookView(),
         MOCK_WEBHOOK_URL,
       )
 
@@ -323,7 +324,7 @@ describe('webhook.service', () => {
 
       // Act
       const actual = await sendWebhook(
-        testEncryptedSubmission,
+        testEncryptedSubmission.getWebhookView(),
         MOCK_WEBHOOK_URL,
       )
 
@@ -356,7 +357,7 @@ describe('webhook.service', () => {
 
       // Act
       const actual = await sendWebhook(
-        testEncryptedSubmission,
+        testEncryptedSubmission.getWebhookView(),
         MOCK_WEBHOOK_URL,
       )
 
@@ -386,7 +387,7 @@ describe('webhook.service', () => {
 
       // Act
       const actual = await sendWebhook(
-        testEncryptedSubmission,
+        testEncryptedSubmission.getWebhookView(),
         MOCK_WEBHOOK_URL,
       )
 

src/app/modules/webhook/webhook.constants.ts

@@ -0,0 +1,46 @@
+import config from '../../config/config'
+
+import { RetryInterval } from './webhook.types'
+
+/**
+ * Current version of queue message format.
+ */
+export const QUEUE_MESSAGE_VERSION = 0
+
+// Conversion to seconds
+const hours = (h: number) => h * 60 * 60
+const minutes = (m: number) => m * 60
+
+/**
+ * Encodes retry policy.
+ * Element 0 is time to wait + jitter before
+ * retrying the first time, element 1 is time to wait
+ * to wait + jitter before 2nd time, etc.
+ * All units are in seconds.
+ */
+export const RETRY_INTERVALS: RetryInterval[] = config.isDev
+  ? [
+      { base: 10, jitter: 5 },
+      { base: 20, jitter: 5 },
+      { base: 30, jitter: 5 },
+    ]
+  : [
+      { base: minutes(5), jitter: minutes(1) },
+      { base: hours(1), jitter: minutes(30) },
+      { base: hours(2), jitter: minutes(30) },
+      { base: hours(4), jitter: minutes(30) },
+      { base: hours(8), jitter: minutes(30) },
+      { base: hours(24), jitter: minutes(30) },
+    ]
+
+/**
+ * Max possible delay for a message, as specified by AWS.
+ */
+export const MAX_DELAY_SECONDS = minutes(15)
+
+/**
+ * Tolerance allowed for determining if a message is due to be sent.
+ * If a message's next attempt is scheduled either in the past or this
+ * number of seconds in the future, it will be sent.
+ */
+export const DUE_TIME_TOLERANCE_SECONDS = minutes(1)