Merge branch 'created-at-updated-at-bugfix'

Author: themarolt

backend/src/services/activityService.ts

@@ -169,7 +169,9 @@ export default class ActivityService extends LoggerBase {
         )
 
         record = await ActivityRepository.create(data, repositoryOptions)
-        await insertActivities(QUEUE_CLIENT(), [{ ...data, id: record.id }], true)
+        await insertActivities(QUEUE_CLIENT(), [
+          { ...data, id: record.id, createdAt: record.id, createdById: record.createdById },
+        ])
 
         // Only track activity's platform and timestamp and memberId. It is completely annonymous.
         telemetryTrack(

services/apps/activities_worker/src/activities/createConversations.ts

@@ -181,7 +181,7 @@ export async function createConversations(): Promise<ICreateConversationsResult>
   if (toUpdate.length > 0) {
     for (const batch of partition(toUpdate, 100)) {
       try {
-        const results = await insertActivities(svc.queue, batch, true)
+        const results = await insertActivities(svc.queue, batch)
         activitiesAddedToConversations += results.length
       } catch (err) {
         svc.log.error(err, 'Error linking activities to conversations')

services/apps/cron_service/src/jobs/nangoConnectionCleanup.job.ts

@@ -36,10 +36,10 @@ const job: IJobDefinition = {
       const integration = singleOrDefault(allIntegrations, (i) => i.id === connection.connectionId)
 
       if (!integration) {
-        // check if connection.created is older than 1 day
+        // check if connection.created is older than 7 days
         const created = new Date(connection.createdAt)
 
-        if (created < new Date(Date.now() - 24 * 60 * 60 * 1000)) {
+        if (created < new Date(Date.now() - 7 * 24 * 60 * 60 * 1000)) {
           ctx.log.info(`Deleting stale connection ${connection.connectionId}`)
           await deleteNangoConnection(connection.integration, connection.connectionId)
         }

services/apps/data_sink_worker/src/service/activity.data.ts

@@ -39,6 +39,7 @@ export interface IActivityUpdateData {
   url?: string
   organizationId?: string
   platform?: PlatformType
+  createdAt?: string
 }
 
 export interface ISentimentActivityInput {

services/apps/data_sink_worker/src/service/activity.service.ts

@@ -182,7 +182,10 @@ export default class ActivityService extends LoggerBase {
         }
 
         if (!isObjectEmpty(toUpdate)) {
-          this.log.debug({ activityId: id }, 'Updating activity.')
+          this.log.debug(
+            { activityId: id, createdAt: original.createdAt },
+            'Updating activity in database.',
+          )
 
           // use insert instead of update to avoid using pg protocol with questdb
           try {
@@ -211,6 +214,7 @@ export default class ActivityService extends LoggerBase {
                 isBotActivity: memberInfo.isBot,
                 isTeamMemberActivity: memberInfo.isTeamMember,
                 importHash: original.importHash,
+                createdAt: original.createdAt,
               },
             ])
             await createOrUpdateRelations(queryExecutor, {

services/libs/data-access-layer/src/activities/ilp.ts

@@ -15,7 +15,6 @@ const logger = getServiceChildLogger('insert-activities')
 export async function insertActivities(
   queueClient: IQueue,
   activities: IDbActivityCreateData[],
-  update = false,
 ): Promise<string[]> {
   const now = moment().toISOString()
 
@@ -37,7 +36,7 @@ export async function insertActivities(
         ...activity,
 
         id,
-        updatedAt: update || !activity.updatedAt ? now : moment(activity.updatedAt).toISOString(),
+        updatedAt: now,
         createdAt: activity.createdAt ? moment(activity.createdAt).toISOString() : now,
         timestamp: activity.timestamp ? moment(activity.timestamp).toISOString() : now,
         attributes: objectToBytes(tryToUnwrapAttributes(activity.attributes)),

services/libs/data-access-layer/src/activities/update.ts

@@ -90,7 +90,7 @@ export async function updateActivities(
     qdb,
     async (activity) => {
       const newActivity = await mapNewActivity(activity, mapActivity)
-      await insertActivities(queueClient, [newActivity], true)
+      await insertActivities(queueClient, [newActivity])
       const changedRelations = getChangedRelationshipFields(activity, newActivity)
       if (Object.keys(changedRelations).length > 0) {
         await updateActivityRelationsById(pgQx, {

services/libs/data-access-layer/src/old/apps/data_sink_worker/repo/activity.data.ts

@@ -24,6 +24,7 @@ export interface IDbActivity {
   sentiment: ISentimentAnalysisResult
   organizationId?: string
   deletedAt?: string
+  createdAt?: string
 }
 
 export interface IDbActivityCreateData {