Merge branch 'main' into bugfix/editTags

backend/src/services/memberService.ts

@@ -336,6 +336,9 @@ export default class MemberService extends LoggerBase {
 
           // Collect unique domains
           for (const email of data.emails) {
+            if (!email) {
+              continue
+            }
             const domain = email.split('@')[1]
             emailDomains.add(domain)
           }

frontend/src/modules/automation/components/automation-form.vue

@@ -139,7 +139,7 @@
 
 <script setup>
 import {
-  computed, defineProps, defineEmits, reactive, ref, watch,
+  computed, defineProps, defineEmits, reactive, ref, watch, onMounted,
 } from 'vue';
 import AppDrawer from '@/shared/drawer/drawer.vue';
 import { required } from '@vuelidate/validators';
@@ -154,6 +154,7 @@ import { useAutomationStore } from '@/modules/automation/store';
 import Message from '@/shared/message/message';
 import { i18n } from '@/i18n';
 import formChangeDetector from '@/shared/form/form-change';
+import { useStore } from 'vuex';
 
 const props = defineProps({
   modelValue: {
@@ -177,6 +178,9 @@ const emit = defineEmits(['update:modelValue', 'update:automation']);
 
 const { createAutomation, updateAutomation, getAutomations } = useAutomationStore();
 
+const store = useStore();
+const fetchIntegrations = () => store.dispatch('integration/doFetch');
+
 const isDrawerOpen = computed({
   get() {
     return props.modelValue;
@@ -287,6 +291,9 @@ const doSubmit = () => {
   }
 };
 
+onMounted(() => {
+  fetchIntegrations();
+});
 </script>
 
 <script>

frontend/src/modules/automation/pages/automation-list-page.vue

@@ -109,6 +109,7 @@
 
     <!-- Add/Edit Webhook form drawer -->
     <app-automation-form
+      v-if="openAutomationForm"
       v-model="openAutomationForm"
       v-model:automation="editAutomation"
       :type="automationFormType"

services/apps/integration_data_worker/package.json

@@ -13,7 +13,9 @@
     "lint": "./node_modules/.bin/eslint --ext .ts src --max-warnings=0",
     "format": "./node_modules/.bin/prettier --write \"src/**/*.ts\"",
     "format-check": "./node_modules/.bin/prettier --check .",
-    "tsc-check": "./node_modules/.bin/tsc --noEmit"
+    "tsc-check": "./node_modules/.bin/tsc --noEmit",
+    "script:process-data": "SERVICE=script TS_NODE_TRANSPILE_ONLY=true node -r tsconfig-paths/register -r ts-node/register src/bin/process-data.ts",
+    "script:process-data-for-tenant": "SERVICE=script TS_NODE_TRANSPILE_ONLY=true node -r tsconfig-paths/register -r ts-node/register src/bin/process-data-for-tenant.ts"
   },
   "dependencies": {
     "@crowd/common": "file:../../libs/common",

services/apps/integration_data_worker/src/bin/process-data-for-tenant.ts

@@ -0,0 +1,44 @@
+import { DB_CONFIG, SQS_CONFIG } from '@/conf'
+import IntegrationDataRepository from '@/repo/integrationData.repo'
+import { DbStore, getDbConnection } from '@crowd/database'
+import { getServiceLogger } from '@crowd/logging'
+import { IntegrationDataWorkerEmitter, getSqsClient } from '@crowd/sqs'
+import { IntegrationStreamDataState } from '@crowd/types'
+
+const log = getServiceLogger()
+
+const processArguments = process.argv.slice(2)
+
+if (processArguments.length !== 1) {
+  log.error('Expected 1 argument: tenantId')
+  process.exit(1)
+}
+
+const tenantId = processArguments[0]
+
+setImmediate(async () => {
+  const sqsClient = getSqsClient(SQS_CONFIG())
+  const emitter = new IntegrationDataWorkerEmitter(sqsClient, log)
+  await emitter.init()
+
+  const dbConnection = getDbConnection(DB_CONFIG(), 1)
+  const store = new DbStore(log, dbConnection)
+  const repo = new IntegrationDataRepository(store, log)
+
+  const dataIds = await repo.getDataForTenant(tenantId)
+
+  for (const dataId of dataIds) {
+    const info = await repo.getDataInfo(dataId)
+
+    if (info) {
+      if (info.state !== IntegrationStreamDataState.PENDING) {
+        await repo.resetStream(dataId)
+      }
+
+      await emitter.triggerDataProcessing(info.tenantId, info.integrationType, dataId)
+    } else {
+      log.error({ dataId }, 'Data stream not found!')
+      process.exit(1)
+    }
+  }
+})

services/apps/integration_data_worker/src/bin/process-data.ts

@@ -0,0 +1,42 @@
+import { DB_CONFIG, SQS_CONFIG } from '@/conf'
+import IntegrationDataRepository from '@/repo/integrationData.repo'
+import { DbStore, getDbConnection } from '@crowd/database'
+import { getServiceLogger } from '@crowd/logging'
+import { IntegrationDataWorkerEmitter, getSqsClient } from '@crowd/sqs'
+import { IntegrationStreamDataState } from '@crowd/types'
+
+const log = getServiceLogger()
+
+const processArguments = process.argv.slice(2)
+
+if (processArguments.length !== 1) {
+  log.error('Expected 1 argument: dataId')
+  process.exit(1)
+}
+
+const dataIds = processArguments[0].split(',')
+
+setImmediate(async () => {
+  const sqsClient = getSqsClient(SQS_CONFIG())
+  const emitter = new IntegrationDataWorkerEmitter(sqsClient, log)
+  await emitter.init()
+
+  const dbConnection = getDbConnection(DB_CONFIG(), 1)
+  const store = new DbStore(log, dbConnection)
+  const repo = new IntegrationDataRepository(store, log)
+
+  for (const dataId of dataIds) {
+    const info = await repo.getDataInfo(dataId)
+
+    if (info) {
+      if (info.state !== IntegrationStreamDataState.PENDING) {
+        await repo.resetStream(dataId)
+      }
+
+      await emitter.triggerDataProcessing(info.tenantId, info.integrationType, dataId)
+    } else {
+      log.error({ dataId }, 'Data stream not found!')
+      process.exit(1)
+    }
+  }
+})

services/apps/integration_data_worker/src/repo/integrationData.repo.ts

@@ -42,6 +42,15 @@ export default class IntegrationDataRepository extends RepositoryBase<Integratio
       where d.id = $(dataId);
   `
 
+  private readonly getDataForTenantQuery = `
+    select
+      d.id
+    from 
+     integration."apiData" d
+    where 
+     d."tenantId" = $(tenantId)
+  `
+
   public async getDataInfo(dataId: string): Promise<IApiDataInfo | null> {
     const results = await this.db().oneOrNone(this.getDataInfoQuery, {
       dataId,
@@ -236,4 +245,30 @@ export default class IntegrationDataRepository extends RepositoryBase<Integratio
 
     this.checkUpdateRowCount(result.rowCount, 1)
   }
+
+  public async resetStream(dataId: string): Promise<void> {
+    const result = await this.db().result(
+      `update integration."apiData"
+       set  state = $(state),
+            error = null,
+            "delayedUntil" = null,
+            "processedAt" = null,
+            "updatedAt" = now()
+       where id = $(dataId)`,
+      {
+        dataId,
+        state: IntegrationStreamDataState.PENDING,
+      },
+    )
+
+    this.checkUpdateRowCount(result.rowCount, 1)
+  }
+
+  public async getDataForTenant(tenantId: string): Promise<string[]> {
+    const results = await this.db().manyOrNone(this.getDataForTenantQuery, {
+      tenantId,
+    })
+
+    return results.map((r) => r.id)
+  }
 }

services/libs/conversations/src/repo/conversation.repo.ts

@@ -62,7 +62,7 @@ export class ConversationRepository extends RepositoryBase<ConversationRepositor
     const now = new Date()
     const result = await this.db().result(
       `insert into "conversationSettings"(id, "tenantId", "createdAt", "updatedAt", "createdById")
-       values ($(id), $(tenantId), $(now), $(now), (select "createdById" from tenants where id = $(tenantId) limit 1));`,
+       values ($(id), $(tenantId), $(now), $(now), (select coalesce("createdById", "updatedById") as "createdById" from tenants where id = $(tenantId) limit 1));`,
       {
         id,
         tenantId,