Sync all script optimizations

backend/src/database/migrations/U1709228609__indexed-entities.sql

@@ -0,0 +1 @@
+drop table indexed_entities;

backend/src/database/migrations/V1709228609__indexed-entities.sql

@@ -0,0 +1,11 @@
+create table if not exists indexed_entities (
+    type       varchar(255) not null,
+    entity_id  uuid         not null,
+    tenant_id  uuid         not null,
+    indexed_at timestamptz  not null default now(),
+
+    primary key (type, entity_id)
+);
+
+create index if not exists ix_indexed_entities_tenant on indexed_entities (tenant_id);
+create index if not exists ix_indexed_entities_type on indexed_entities (type);

services/libs/opensearch/src/repo/activity.repo.ts

@@ -1,6 +1,7 @@
 import { DbStore, RepositoryBase } from '@crowd/database'
 import { Logger } from '@crowd/logging'
 import { IDbActivitySyncData } from './activity.data'
+import { IndexedEntityType } from './indexing.data'
 
 export class ActivityRepository extends RepositoryBase<ActivityRepository> {
   constructor(dbStore: DbStore, parentLog: Logger) {
@@ -57,40 +58,20 @@ export class ActivityRepository extends RepositoryBase<ActivityRepository> {
     return results.map((r) => r.id)
   }
 
-  public async getTenantActivitiesForSync(
-    tenantId: string,
-    perPage: number,
-    lastId?: string,
-  ): Promise<string[]> {
-    // eslint-disable-next-line @typescript-eslint/no-explicit-any
-    let results: any[]
-
-    if (lastId) {
-      results = await this.db().any(
-        `
-      select id from activities 
-      where "tenantId" = $(tenantId) and "deletedAt" is null and id > $(lastId)
-      order by id
-      limit ${perPage};
-      `,
-        {
-          tenantId,
-          lastId,
-        },
-      )
-    } else {
-      results = await this.db().any(
-        `
-      select id from activities 
-      where "tenantId" = $(tenantId) and "deletedAt" is null
-      order by id
-      limit ${perPage};
+  public async getTenantActivitiesForSync(tenantId: string, perPage: number): Promise<string[]> {
+    const results = await this.db().any(
+      `
+      select id from activities a
+      left join indexed_entities ie on 
+        a.id = ie.entity_id and ie.type = $(type)
+      where a."tenantId" = $(tenantId) and a."deletedAt" is null and ie.entity_id is null
+      limit ${perPage}
       `,
-        {
-          tenantId,
-        },
-      )
-    }
+      {
+        tenantId,
+        type: IndexedEntityType.ACTIVITY,
+      },
+    )
 
     return results.map((r) => r.id)
   }

services/libs/opensearch/src/repo/indexing.data.ts

@@ -0,0 +1,10 @@
+export interface IEntityData {
+  id: string
+  tenantId: string
+}
+
+export enum IndexedEntityType {
+  ACTIVITY = 'activity',
+  MEMBER = 'member',
+  ORGANIZATION = 'organization',
+}

services/libs/opensearch/src/repo/indexing.repo.ts

@@ -0,0 +1,22 @@
+import { DbStore, RepositoryBase } from '@crowd/database'
+import { Logger } from '@crowd/logging'
+import { IEntityData, IndexedEntityType } from './indexing.data'
+
+export class IndexingRepository extends RepositoryBase<IndexingRepository> {
+  constructor(dbStore: DbStore, parentLog: Logger) {
+    super(dbStore, parentLog)
+  }
+
+  public async markEntitiesIndexed(type: IndexedEntityType, data: IEntityData[]): Promise<void> {
+    if (data.length > 0) {
+      const values = data.map((d) => `('${type}', '${d.id}', '${d.tenantId}')`)
+      const query = `
+        insert into indexed_entities(type, entity_id, tenant_id)
+        values ${values.join(',\n')}
+        on conflict (type, entity_id)
+        do update set indexed_at = now()
+      `
+      await this.db().none(query)
+    }
+  }
+}