Improve cubejs (#1391)

backend/src/bin/job-generator.ts

@@ -10,7 +10,7 @@ for (const job of jobs) {
     async () => {
       log.info({ job: job.name }, 'Triggering job.')
       try {
-        await job.onTrigger()
+        await job.onTrigger(log)
       } catch (err) {
         log.error(err, { job: job.name }, 'Error while executing a job!')
       }

backend/src/bin/jobs/index.ts

@@ -4,6 +4,7 @@ import weeklyAnalyticsEmailsCoordinator from './weeklyAnalyticsEmailsCoordinator
 import memberScoreCoordinator from './memberScoreCoordinator'
 import checkSqsQueues from './checkSqsQueues'
 import refreshMaterializedViews from './refreshMaterializedViews'
+import refreshMaterializedViewsForCube from './refreshMaterializedViewsForCube'
 import downgradeExpiredPlans from './downgradeExpiredPlans'
 import eagleEyeEmailDigestTicks from './eagleEyeEmailDigestTicks'
 import integrationDataChecker from './integrationDataChecker'
@@ -21,6 +22,7 @@ const jobs: CrowdJob[] = [
   memberScoreCoordinator,
   checkSqsQueues,
   refreshMaterializedViews,
+  refreshMaterializedViewsForCube,
   downgradeExpiredPlans,
   eagleEyeEmailDigestTicks,
   integrationDataChecker,

backend/src/bin/jobs/refreshMaterializedViewsForCube.ts

@@ -0,0 +1,38 @@
+import { Logger, logExecutionTimeV2 } from '@crowd/logging'
+import SequelizeRepository from '../../database/repositories/sequelizeRepository'
+import { CrowdJob } from '../../types/jobTypes'
+
+let processing = false
+
+const job: CrowdJob = {
+  name: 'Refresh Materialized View For Cube',
+  cronTime: '1,31 * * * *',
+  onTrigger: async (log: Logger) => {
+    if (!processing) {
+      processing = true
+    } else {
+      return
+    }
+    const dbOptions = await SequelizeRepository.getDefaultIRepositoryOptions()
+
+    const materializedViews = [
+      'mv_members_cube',
+      'mv_activities_cube',
+      'mv_organizations_cube',
+      'mv_segments_cube',
+    ]
+
+    for (const view of materializedViews) {
+      await logExecutionTimeV2(
+        () =>
+          dbOptions.database.sequelize.query(`REFRESH MATERIALIZED VIEW CONCURRENTLY "${view}"`),
+        log,
+        `Refresh Materialized View ${view}`,
+      )
+    }
+
+    processing = false
+  },
+}
+
+export default job

backend/src/cubejs/.gitignore

@@ -0,0 +1 @@
+/.cubecloud

backend/src/cubejs/schema/Activities.js

@@ -1,218 +1,61 @@
-cube(`Activities`, {
-  sql: `SELECT * FROM public.activities`,
-
-  preAggregations: {
-    Activities: {
-      measures: [Activities.count],
-      dimensions: [
-        Activities.platform,
-        Activities.type,
-        Members.score,
-        Members.location,
-        Members.tenantId,
-        Members.isTeamMember,
-        Members.isBot,
-        Members.isOrganization,
-        Activities.tenantId,
-        Segments.id,
-      ],
-      timeDimension: Activities.date,
-      granularity: `day`,
-      refreshKey: {
-        every: `10 minute`,
-      },
-    },
-    ActivitiesCumulative: {
-      measures: [Activities.cumulativeCount],
-      dimensions: [
-        Activities.platform,
-        Activities.type,
-        Members.score,
-        Members.location,
-        Members.tenantId,
-        Members.isTeamMember,
-        Members.isBot,
-        Members.isOrganization,
-        Activities.tenantId,
-        Segments.id,
-      ],
-      timeDimension: Activities.date,
-      granularity: `day`,
-      refreshKey: {
-        every: `10 minute`,
-      },
-    },
-    ActivitiesPTD: {
-      measures: [Activities.count],
-      dimensions: [
-        Activities.platform,
-        Activities.type,
-        Members.score,
-        Members.location,
-        Members.tenantId,
-        Members.isTeamMember,
-        Members.isBot,
-        Members.isOrganization,
-        Activities.tenantId,
-        Segments.id,
-      ],
-      timeDimension: Activities.date,
-      granularity: `day`,
-      partition_granularity: `month`,
-      refreshKey: {
-        sql: `SELECT MAX("updatedAt") FROM public.activities WHERE ${FILTER_PARAMS.Activities.date.filter(
-          'timestamp',
-        )}`,
-        every: `30 minute`,
-      },
-    },
-    ActivitiesCumulativePTD: {
-      measures: [Activities.cumulativeCount],
-      dimensions: [
-        Activities.platform,
-        Activities.type,
-        Members.score,
-        Members.location,
-        Members.tenantId,
-        Members.isTeamMember,
-        Members.isBot,
-        Members.isOrganization,
-        Activities.tenantId,
-        Segments.id,
-      ],
-      timeDimension: Activities.date,
-      granularity: `day`,
-      partition_granularity: `month`,
-      refreshKey: {
-        sql: `SELECT MAX("updatedAt") FROM public.activities WHERE ${FILTER_PARAMS.Activities.date.filter(
-          'timestamp',
-        )}`,
-        every: `30 minute`,
-      },
-    },
-  },
+cube('Activities', {
+  sql_table: 'mv_activities_cube',
 
   measures: {
     count: {
-      type: `count`,
-      drillMembers: [
-        memberId,
-        sourceid,
-        tenantId,
-        id,
-        updatedbyid,
-        parentid,
-        createdbyid,
-        createdat,
-        updatedat,
-        date,
-      ],
+      sql: `${CUBE}.id`,
+      type: 'count_distinct',
+      drillMembers: [tenantId, date],
     },
     cumulativeCount: {
-      type: `count`,
+      type: 'count',
       rollingWindow: {
-        trailing: `unbounded`,
+        trailing: 'unbounded',
       },
     },
   },
 
   dimensions: {
-    memberId: {
-      sql: `${CUBE}."memberId"`,
-      type: `string`,
-      shown: false,
+    id: {
+      sql: `${CUBE}.id`,
+      type: 'string',
+      primaryKey: true,
     },
 
-    sentimentMood: {
-      case: {
-        when: [
-          { sql: `${CUBE}.sentiment->>'sentiment' is null`, label: `no data` },
-          { sql: `(${CUBE}.sentiment->>'sentiment')::integer < 34`, label: `negative` },
-          { sql: `(${CUBE}.sentiment->>'sentiment')::integer > 66`, label: `positive` },
-        ],
-        else: { label: `neutral` },
-      },
-      type: `string`,
+    iscontribution: {
+      sql: `${CUBE}."isContribution"`,
+      type: 'string',
     },
 
-    sourceid: {
-      sql: `${CUBE}."sourceId"`,
-      type: `string`,
-      shown: false,
+    sentimentMood: {
+      sql: `${CUBE}."sentimentMood"`,
+      type: 'string',
     },
 
     platform: {
-      sql: `platform`,
-      type: `string`,
+      sql: `${CUBE}.platform`,
+      type: 'string',
     },
 
     channel: {
-      sql: `channel`,
-      type: `string`,
+      sql: `${CUBE}.channel`,
+      type: 'string',
     },
 
     tenantId: {
       sql: `${CUBE}."tenantId"`,
-      type: `string`,
+      type: 'string',
       shown: false,
     },
 
-    id: {
-      sql: `id`,
-      type: `string`,
-      primaryKey: true,
-    },
-
     type: {
-      sql: `type`,
-      type: `string`,
-    },
-
-    updatedbyid: {
-      sql: `${CUBE}."updatedById"`,
-      type: `string`,
-      shown: false,
-    },
-
-    iscontribution: {
-      sql: `${CUBE}."isContribution"`,
-      type: `string`,
-      shown: true,
-    },
-
-    parentid: {
-      sql: `${CUBE}."parentId"`,
-      type: `string`,
-      shown: false,
-    },
-
-    createdbyid: {
-      sql: `${CUBE}."createdById"`,
-      type: `string`,
-      shown: false,
-    },
-
-    createdat: {
-      sql: `${CUBE}."createdAt"`,
-      type: `time`,
-      shown: false,
-    },
-
-    updatedat: {
-      sql: `${CUBE}."updatedAt"`,
-      type: `time`,
-      shown: false,
+      sql: `${CUBE}.type`,
+      type: 'string',
     },
 
     date: {
-      sql: `timestamp`,
-      type: `time`,
-    },
-
-    deletedat: {
-      sql: `${CUBE}."deletedAt"`,
-      type: `time`,
-      shown: false,
+      sql: `${CUBE}.timestamp`,
+      type: 'time',
     },
   },
 })