diff --git a/services/apps/git_integration/src/crowdgit/database/crud.py b/services/apps/git_integration/src/crowdgit/database/crud.py index e98a25a06a..7b58db668f 100644 --- a/services/apps/git_integration/src/crowdgit/database/crud.py +++ b/services/apps/git_integration/src/crowdgit/database/crud.py @@ -720,6 +720,20 @@ async def find_many_organization_ids_by_identities(identities: list[dict]) -> li return results +async def fetch_organizations(org_ids: list[str]) -> list[dict]: + if not org_ids: + return [] + + return await query( + """ + SELECT id, "isAffiliationBlocked" + FROM organizations + WHERE id = ANY($1::uuid[]) + """, + (org_ids,), + ) + + async def fetch_member_organizations(member_ids: list[str]) -> list[dict]: if not member_ids: return [] @@ -751,9 +765,9 @@ async def fetch_segment_affiliations(member_ids: list[str], segment_id: str) -> ) -async def insert_member_organizations(rows: list[dict]) -> None: +async def insert_member_organizations(rows: list[dict]) -> list[dict]: if not rows: - return + return [] undated_rows: list[tuple] = [] open_ended_rows: list[tuple] = [] @@ -767,8 +781,10 @@ async def insert_member_organizations(rows: list[dict]) -> None: row.get("date_end"), row["source"], ) + date_start = row.get("date_start") date_end = row.get("date_end") + if date_start is None and date_end is None: undated_rows.append(params) elif date_end is None: @@ -787,41 +803,91 @@ async def insert_member_organizations(rows: list[dict]) -> None: "createdAt", "updatedAt" ) - VALUES ($1, $2, $3, $4, NULL, $5, NOW(), NOW()) """ - if undated_rows: - sql = ( - insert_sql - + """ + returning_sql = """ + RETURNING id, "memberId", "organizationId" + """ + + buckets = [ + ( + undated_rows, + """ ON CONFLICT ("memberId", "organizationId") WHERE ("dateStart" IS NULL AND "dateEnd" IS NULL AND "deletedAt" IS NULL) DO NOTHING - """ - ) - await executemany(sql, undated_rows) - - if open_ended_rows: - sql = ( - insert_sql - + """ + """, + ), + ( + open_ended_rows, + """ ON CONFLICT ("memberId", "organizationId", "dateStart") WHERE ("dateEnd" IS NULL AND "deletedAt" IS NULL) DO NOTHING - """ - ) - await executemany(sql, open_ended_rows) - - if dated_rows: - sql = ( - insert_sql - + """ + """, + ), + ( + dated_rows, + """ ON CONFLICT ("memberId", "organizationId", "dateStart", "dateEnd") WHERE ("deletedAt" IS NULL) DO NOTHING + """, + ), + ] + + created_rows: list[dict] = [] + + for bucket_rows, conflict_sql in buckets: + if not bucket_rows: + continue + + values_parts: list[str] = [] + params: list = [] + param_index = 1 + + for member_id, organization_id, date_start, date_end, source in bucket_rows: + values_parts.append( + f"(${param_index}, ${param_index + 1}, ${param_index + 2}, " + f"${param_index + 3}, NULL, ${param_index + 4}, NOW(), NOW())" + ) + params.extend([member_id, organization_id, date_start, date_end, source]) + param_index += 5 + + created_rows.extend( + await query( + insert_sql + f" VALUES {', '.join(values_parts)}" + conflict_sql + returning_sql, + tuple(params), + ) + ) + + return created_rows + + +async def insert_member_organization_affiliation_overrides(rows: list[dict]) -> None: + if not rows: + return + + await executemany( """ + INSERT INTO "memberOrganizationAffiliationOverrides"( + id, + "memberId", + "memberOrganizationId", + "allowAffiliation" ) - await executemany(sql, dated_rows) + VALUES (gen_random_uuid(), $1, $2, $3) + ON CONFLICT ("memberId", "memberOrganizationId") DO NOTHING + """, + [ + ( + row["member_id"], + row["member_organization_id"], + row["allow_affiliation"], + ) + for row in rows + ], + ) async def insert_member_segment_affiliations(rows: list[dict]) -> None: diff --git a/services/apps/git_integration/src/crowdgit/services/affiliation/affiliation_service.py b/services/apps/git_integration/src/crowdgit/services/affiliation/affiliation_service.py index f3b009a76a..0e164f232e 100644 --- a/services/apps/git_integration/src/crowdgit/services/affiliation/affiliation_service.py +++ b/services/apps/git_integration/src/crowdgit/services/affiliation/affiliation_service.py @@ -11,10 +11,12 @@ from crowdgit.database.crud import ( fetch_member_organizations, + fetch_organizations, fetch_segment_affiliations, find_many_member_ids_by_identities, find_many_organization_ids_by_identities, get_repo_affiliation_registry, + insert_member_organization_affiliation_overrides, insert_member_organizations, insert_member_segment_affiliations, save_service_execution, @@ -870,6 +872,16 @@ async def apply_affiliations( else: segment_affiliations_by_member.setdefault(member_id, []).append(row) + blocked_org_ids: set[str] = set() + if resolved_stints: + org_ids = list({organization_id for _, organization_id, _ in resolved_stints}) + organizations = await fetch_organizations(org_ids) + blocked_org_ids = { + str(organization["id"]) + for organization in organizations + if organization["isAffiliationBlocked"] + } + mo_inserts: list[dict] = [] msa_inserts: list[dict] = [] @@ -896,10 +908,14 @@ async def apply_affiliations( } ) - if not self.has_existing_stint( - existing_msas, organization_id, date_start, date_end - ) and not self.is_blocked_by_deleted_row( - deleted_msas, organization_id, date_start, date_end + if ( + str(organization_id) not in blocked_org_ids + and not self.has_existing_stint( + existing_msas, organization_id, date_start, date_end + ) + and not self.is_blocked_by_deleted_row( + deleted_msas, organization_id, date_start, date_end + ) ): msa_inserts.append( { @@ -911,7 +927,21 @@ async def apply_affiliations( } ) - await insert_member_organizations(mo_inserts) + created_member_organizations = await insert_member_organizations(mo_inserts) + + if blocked_org_ids: + override_rows = [ + { + "member_id": mo["memberId"], + "member_organization_id": mo["id"], + "allow_affiliation": False, + } + for mo in created_member_organizations + if str(mo["organizationId"]) in blocked_org_ids + ] + if override_rows: + await insert_member_organization_affiliation_overrides(override_rows) + await insert_member_segment_affiliations(msa_inserts) async def process_affiliations(