add script

Author: garrrikkotua

services/apps/integration_run_worker/package.json

@@ -16,7 +16,8 @@
     "tsc-check": "./node_modules/.bin/tsc --noEmit",
     "script:onboard-integration": "SERVICE=script TS_NODE_TRANSPILE_ONLY=true node -r tsconfig-paths/register -r ts-node/register src/bin/onboard-integration.ts",
     "script:process-repo": "SERVICE=script TS_NODE_TRANSPILE_ONLY=true node -r tsconfig-paths/register -r ts-node/register src/bin/process-repo.ts",
-    "script:trigger-stream-processed": "SERVICE=script TS_NODE_TRANSPILE_ONLY=true node -r tsconfig-paths/register -r ts-node/register src/bin/trigger-stream-processed.ts"
+    "script:trigger-stream-processed": "SERVICE=script TS_NODE_TRANSPILE_ONLY=true node -r tsconfig-paths/register -r ts-node/register src/bin/trigger-stream-processed.ts",
+    "script:continue-run": "SERVICE=script TS_NODE_TRANSPILE_ONLY=true node -r tsconfig-paths/register -r ts-node/register src/bin/continue-run.ts"
   },
   "dependencies": {
     "@crowd/common": "file:../../libs/common",

services/apps/integration_run_worker/src/bin/continue-run.ts

@@ -0,0 +1,41 @@
+import { DB_CONFIG, SQS_CONFIG } from '@/conf'
+import { DbStore, getDbConnection } from '@crowd/database'
+import { getServiceLogger } from '@crowd/logging'
+import { IntegrationRunWorkerEmitter, getSqsClient } from '@crowd/sqs'
+import IntegrationRunRepository from '@/repo/integrationRun.repo'
+import { IntegrationRunState } from '@crowd/types'
+
+const log = getServiceLogger()
+
+const processArguments = process.argv.slice(2)
+
+const runId = processArguments[0]
+
+setImmediate(async () => {
+  const sqsClient = getSqsClient(SQS_CONFIG())
+  const emitter = new IntegrationRunWorkerEmitter(sqsClient, log)
+  await emitter.init()
+
+  const dbConnection = await getDbConnection(DB_CONFIG(), 1)
+  const store = new DbStore(log, dbConnection)
+
+  const repo = new IntegrationRunRepository(store, log)
+
+  const run = await repo.findIntegrationRunById(runId)
+
+  if (run) {
+    if (run.state != IntegrationRunState.PENDING) {
+      log.warn(`Integration run is not pending, setting to pending!`)
+
+      await repo.restart(run.id)
+    }
+
+    log.info(`Triggering integration run for ${runId}!`)
+
+    await emitter.triggerIntegrationRun(run.tenantId, run.platform, run.integrationId, true)
+    process.exit(0)
+  } else {
+    log.error({ run }, 'Run not found!')
+    process.exit(1)
+  }
+})

services/apps/integration_run_worker/src/repo/integrationRun.repo.ts

@@ -371,4 +371,46 @@ export default class IntegrationRunRepository extends RepositoryBase<Integration
 
     return result.settings
   }
+
+  public async findIntegrationRunById(runId: string): Promise<{
+    id: string
+    state: IntegrationRunState
+    tenantId: string
+    integrationId: string
+    platform: string
+  } | null> {
+    const result = await this.db().oneOrNone(
+      `
+      select r.id, r.state, r."tenantId", r."integrationId", i.platform
+      from integration.runs r
+      inner join integrations i on r."integrationId" = i.id
+      where r.id = $(runId)
+    `,
+      {
+        runId,
+      },
+    )
+
+    return result
+  }
+
+  public async restart(runId: string): Promise<void> {
+    const result = await this.db().result(
+      `
+      update integration.runs
+         set state = $(state),
+             "processedAt" = null,
+              error = null,
+              "delayedUntil" = null,
+             "updatedAt" = now()
+       where id = $(runId)
+    `,
+      {
+        runId,
+        state: IntegrationRunState.PENDING,
+      },
+    )
+
+    this.checkUpdateRowCount(result.rowCount, 1)
+  }
 }