Merge branch 'main' into deprecate-questdb-CM-420

Author: themarolt

backend/src/services/integrationService.ts

@@ -252,9 +252,12 @@ export default class IntegrationService {
       }
 
       if (IntegrationService.isCodePlatform(platform) && platform !== PlatformType.GIT) {
-        this.gitConnectOrUpdate({
-          remotes: repositories,
-        })
+        await this.gitConnectOrUpdate(
+          {
+            remotes: repositories,
+          },
+          txOptions,
+        )
       }
 
       return integration
@@ -833,7 +836,11 @@ export default class IntegrationService {
   }
 
   async githubNangoConnect(settings, mapping, integrationId?: string) {
-    const transaction = await SequelizeRepository.createTransaction(this.options)
+    const existingTransaction = SequelizeRepository.getTransaction(this.options)
+
+    const transaction =
+      existingTransaction || (await SequelizeRepository.createTransaction(this.options))
+
     const txOptions = {
       ...this.options,
       transaction,
@@ -881,7 +888,9 @@ export default class IntegrationService {
         )
       }
 
-      await SequelizeRepository.commitTransaction(transaction)
+      if (!existingTransaction) {
+        await SequelizeRepository.commitTransaction(transaction)
+      }
 
       await this.options.temporal.workflow.start('syncGithubIntegration', {
         taskQueue: 'nango',
@@ -896,7 +905,9 @@ export default class IntegrationService {
       return integration
     } catch (err) {
       this.options.log.error(err, 'Error while creating or updating GitHub integration!')
-      await SequelizeRepository.rollbackTransaction(transaction)
+      if (!existingTransaction) {
+        await SequelizeRepository.rollbackTransaction(transaction)
+      }
       throw err
     }
   }
@@ -1280,7 +1291,9 @@ export default class IntegrationService {
       return null
     }
 
-    const existingTransaction = SequelizeRepository.getTransaction(options || this.options)
+    const currentOptions = options || this.options
+    const existingTransaction =
+      currentOptions.transaction || SequelizeRepository.getTransaction(currentOptions)
     const transaction =
       existingTransaction || (await SequelizeRepository.createTransaction(options || this.options))
     let integration

services/apps/git_integration/src/crowdgit/services/clone/clone_service.py

@@ -62,7 +62,7 @@ async def _perform_minimal_clone(self, path: str, remote: str) -> None:
         )
         self.logger.info("Initializing minimal clone")
         await run_shell_command(
-            ["git", "clone", "--depth=1", "--no-tags", "--single-branch", remote, path], cwd=path
+            ["git", "clone", "--depth=1", "--no-tags", "--single-branch", remote, "."], cwd=path
         )
         self.logger.info("Minimal clone initialized successfully")
 
@@ -280,7 +280,9 @@ async def _calculate_batch_depth(self, repo_path: str, remote: str) -> int:
     async def _perform_full_clone(self, repo_path: str, remote: str):
         """Perform full repository clone"""
         self.logger.info(f"Performing full clone for repo {remote}...")
-        await run_shell_command(["git", "clone", remote, repo_path], cwd=repo_path)
+        await run_shell_command(
+            ["git", "clone", "--no-tags", "--single-branch", remote, "."], cwd=repo_path
+        )
         self.logger.info(f"Successfully completed full clone of repository: {remote}")
 
     async def has_default_branch_changed(self, remote: str, saved_branch: str | None) -> bool:

services/apps/script_executor_worker/src/activities.ts

@@ -1,3 +1,7 @@
+import {
+  blockMemberOrganizationAffiliation,
+  getOrganizationMembers,
+} from './activities/block-organization-affiliation'
 import {
   findDuplicateMembersAfterDate,
   moveMemberActivityRelations,
@@ -22,6 +26,7 @@ import {
   findMemberIdentitiesGroupedByPlatform,
   findMemberMergeActions,
 } from './activities/dissect-member'
+import { calculateMemberAffiliations } from './activities/fix-activity-foreign-keys'
 import {
   getBotMembersWithOrgAffiliation,
   removeBotMemberOrganization,
@@ -78,4 +83,7 @@ export {
   getBotMembersWithOrgAffiliation,
   removeBotMemberOrganization,
   unlinkOrganizationFromBotActivities,
+  blockMemberOrganizationAffiliation,
+  getOrganizationMembers,
+  calculateMemberAffiliations,
 }

services/apps/script_executor_worker/src/activities/block-organization-affiliation.ts

@@ -0,0 +1,38 @@
+import { pgpQx } from '@crowd/data-access-layer'
+import { changeOverride } from '@crowd/data-access-layer/src/member_organization_affiliation_overrides'
+import OrganizationRepository from '@crowd/data-access-layer/src/old/apps/script_executor_worker/organization.repo'
+import { IMemberOrganization } from '@crowd/types'
+
+import { svc } from '../main'
+
+export async function getOrganizationMembers(
+  organizationId: string,
+  limit = 100,
+  offset = 0,
+): Promise<IMemberOrganization[]> {
+  try {
+    const orgRepo = new OrganizationRepository(svc.postgres.reader.connection(), svc.log)
+    return orgRepo.findOrganizationMembers(organizationId, limit, offset)
+  } catch (error) {
+    svc.log.error(error, 'Error getting organization members!')
+    throw error
+  }
+}
+
+export async function blockMemberOrganizationAffiliation(
+  memberId: string,
+  memberOrganizationId: string,
+): Promise<void> {
+  try {
+    const qx = pgpQx(svc.postgres.writer.connection())
+    return changeOverride(qx, {
+      memberId,
+      memberOrganizationId,
+      allowAffiliation: false,
+      isPrimaryWorkExperience: false,
+    })
+  } catch (error) {
+    svc.log.error(error, 'Error blocking organization affiliation!')
+    throw error
+  }
+}

services/apps/script_executor_worker/src/activities/fix-activity-foreign-keys/index.ts

@@ -0,0 +1,30 @@
+import { refreshMemberOrganizationAffiliations } from '@crowd/data-access-layer/src/member-organization-affiliation'
+import MergeActionRepository from '@crowd/data-access-layer/src/old/apps/script_executor_worker/mergeAction.repo'
+import { EntityType } from '@crowd/data-access-layer/src/old/apps/script_executor_worker/types'
+import { pgpQx } from '@crowd/data-access-layer/src/queryExecutor'
+import { IMergeAction } from '@crowd/types'
+
+import { svc } from '../../main'
+
+export async function findMergeActionsWithDeletedSecondaryEntities(
+  limit: number,
+  offset: number,
+  entityType: EntityType,
+): Promise<IMergeAction[]> {
+  try {
+    const mergeActionRepo = new MergeActionRepository(svc.postgres.reader.connection(), svc.log)
+    return mergeActionRepo.findMergeActionsWithDeletedSecondaryEntities(limit, offset, entityType)
+  } catch (error) {
+    svc.log.error(error, `Error getting merge actions with deleted secondary ${entityType}s !`)
+    throw error
+  }
+}
+
+export async function calculateMemberAffiliations(memberId: string): Promise<void> {
+  try {
+    const qx = pgpQx(svc.postgres.writer.connection())
+    await refreshMemberOrganizationAffiliations(qx, memberId)
+  } catch (err) {
+    throw new Error(err)
+  }
+}

services/apps/script_executor_worker/src/types.ts

@@ -56,3 +56,8 @@ export interface IDedupActivityRelationsArgs extends IScriptBatchTestArgs {
   groupsPerRun?: number
   cursor?: Omit<IActivityRelationDuplicateGroup, 'activityIds'>
 }
+
+export interface IBlockOrganizationAffiliationArgs {
+  organizationId: string
+  offset?: number
+}

services/apps/script_executor_worker/src/workflows.ts

@@ -1,3 +1,4 @@
+import { blockOrganizationAffiliation } from './workflows/block-organization-affiliation'
 import { cleanupDuplicateMembers } from './workflows/cleanup/duplicate-members'
 import { cleanupMembers } from './workflows/cleanup/members'
 import { cleanupOrganizations } from './workflows/cleanup/organizations'
@@ -22,4 +23,5 @@ export {
   processLLMVerifiedMerges,
   cleanupDuplicateMembers,
   fixBotMembersAffiliation,
+  blockOrganizationAffiliation,
 }

services/apps/script_executor_worker/src/workflows/block-organization-affiliation.ts

@@ -0,0 +1,46 @@
+import { continueAsNew, proxyActivities } from '@temporalio/workflow'
+
+import * as activities from '../activities'
+import { IBlockOrganizationAffiliationArgs } from '../types'
+import { chunkArray } from '../utils/common'
+
+const { getOrganizationMembers, blockMemberOrganizationAffiliation, calculateMemberAffiliations } =
+  proxyActivities<typeof activities>({
+    startToCloseTimeout: '30 minutes',
+  })
+
+export async function blockOrganizationAffiliation(
+  args: IBlockOrganizationAffiliationArgs,
+): Promise<void> {
+  const MEMBERS_PER_RUN = 500
+  const BATCH_SIZE = 50
+  const OFFSET = args.offset ?? 0
+
+  const memberOrganizations = await getOrganizationMembers(
+    args.organizationId,
+    MEMBERS_PER_RUN,
+    OFFSET,
+  )
+
+  if (memberOrganizations.length === 0) {
+    console.log('No more organization members to block!')
+    return
+  }
+
+  // Step 1: Block all affiliations in batches
+  for (const chunk of chunkArray(memberOrganizations, BATCH_SIZE)) {
+    await Promise.all(chunk.map((mo) => blockMemberOrganizationAffiliation(mo.memberId, mo.id)))
+  }
+
+  // Step 2: Deduplicate memberIds and calculate affiliations
+  const uniqueMemberIds = Array.from(new Set(memberOrganizations.map((mo) => mo.memberId)))
+  for (const chunk of chunkArray(uniqueMemberIds, BATCH_SIZE)) {
+    await Promise.all(chunk.map((memberId) => calculateMemberAffiliations(memberId)))
+  }
+
+  // Step 3: Continue pagination
+  await continueAsNew<typeof blockOrganizationAffiliation>({
+    ...args,
+    offset: OFFSET + MEMBERS_PER_RUN,
+  })
+}

services/libs/data-access-layer/src/old/apps/integration_sync_worker/member.repo.ts

@@ -79,19 +79,6 @@ export class MemberRepository extends RepositoryBase<MemberRepository> {
                               where mnm."memberId" = $(id)
                                 and m2."deletedAt" is null
                               group by mnm."memberId"),
-            member_tags as (select mt."memberId",
-                                    json_agg(
-                                            json_build_object(
-                                                    'id', t.id,
-                                                    'name', t.name
-                                                )
-                                        )           as all_tags,
-                                    jsonb_agg(t.id) as all_ids
-                            from "memberTags" mt
-                                      inner join tags t on mt."tagId" = t.id
-                            where mt."memberId" = $(id)
-                              and t."deletedAt" is null
-                            group by mt."memberId"),
             member_organizations as (select mo."memberId",
                                             os."segmentId",
                                             json_agg(
@@ -157,7 +144,6 @@ export class MemberRepository extends RepositoryBase<MemberRepository> {
 
               i.identities,
               coalesce(mo.all_organizations, json_build_array()) as organizations,
-              coalesce(mt.all_tags, json_build_array())          as tags,
               coalesce(tmd.to_merge_ids, array []::text[])       as "toMergeIds",
               coalesce(nmd.no_merge_ids, array []::text[])       as "noMergeIds"
         from "memberSegments" ms
@@ -166,7 +152,6 @@ export class MemberRepository extends RepositoryBase<MemberRepository> {
                 inner join activity_data ad on ms."memberId" = ad."memberId" and ms."segmentId" = ad."segmentId"
                 left join to_merge_data tmd on m.id = tmd."memberId"
                 left join no_merge_data nmd on m.id = nmd."memberId"
-                left join member_tags mt on ms."memberId" = mt."memberId"
                 left join member_organizations mo on ms."memberId" = mo."memberId" and ms."segmentId" = mo."segmentId"
         where ms."memberId" = $(id)
           and m."deletedAt" is null;`,

services/libs/data-access-layer/src/old/apps/script_executor_worker/member.repo.ts

@@ -208,7 +208,6 @@ class MemberRepository {
       { name: 'memberSegmentAffiliations', conditions: ['memberId'] },
       { name: 'memberSegmentsAgg', conditions: ['memberId'] },
       { name: 'memberSegments', conditions: ['memberId'] },
-      { name: 'memberTags', conditions: ['memberId'] },
       { name: 'memberToMerge', conditions: ['memberId', 'toMergeId'] },
       { name: 'memberToMergeRaw', conditions: ['memberId', 'toMergeId'] },
       { name: 'members', conditions: ['id'] },