fix: throttle how many times we update updatedAt

Signed-off-by: Uroš Marolt <uros@marolt.me>

Author: themarolt

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

@@ -28,7 +28,7 @@ export class OrganizationService extends LoggerBase {
   ): Promise<string> {
     const id = await this.store.transactionally(async (txStore) => {
       const qe = dbStoreQx(txStore)
-      const id = await findOrCreateOrganization(qe, source, data, integrationId)
+      const id = await findOrCreateOrganization(qe, source, data, integrationId, true)
       return id
     })
 

services/libs/data-access-layer/src/organizations/base.ts

@@ -367,6 +367,7 @@ export async function updateOrganization(
   qe: QueryExecutor,
   organizationId: string,
   data: Partial<IDbOrganizationInput>,
+  throttleUpdatedAt = false,
 ): Promise<string | null> {
   const columns = Object.keys(data)
   if (columns.length === 0) {
@@ -376,9 +377,13 @@ export async function updateOrganization(
   const updatedAt = new Date()
   columns.push('updatedAt')
 
+  const updatedAtExpr = throttleUpdatedAt
+    ? `CASE WHEN "updatedAt" < now() - interval '30 minutes' THEN $(updatedAt) ELSE "updatedAt" END`
+    : `$(updatedAt)`
+
   const query = `
     update organizations set
-      ${columns.map((c) => `"${c}" = $(${c})`).join(',\n')}
+      ${columns.map((c) => `"${c}" = ${c === 'updatedAt' ? updatedAtExpr : `$(${c})`}`).join(',\n')}
     where id = $(organizationId)
     returning id;
   `
@@ -443,6 +448,7 @@ export async function findOrCreateOrganization(
   source: string,
   data: IOrganization,
   integrationId?: string,
+  throttleUpdatedAt = false,
 ): Promise<string | undefined> {
   const verifiedIdentities = data.identities ? data.identities.filter((i) => i.verified) : []
 
@@ -524,7 +530,8 @@ export async function findOrCreateOrganization(
       if (Object.keys(processed.organization).length > 0) {
         log.info({ orgId: existing.id }, `Updating organization!`)
         await logExecutionTimeV2(
-          async () => updateOrganization(qe, existing.id, processed.organization),
+          async () =>
+            updateOrganization(qe, existing.id, processed.organization, throttleUpdatedAt),
           log,
           'organizationService -> findOrCreateOrganization -> updateOrganization',
         )