fix: prevent affiliations for project-registry member organizations (#4321)

Signed-off-by: Yeganathan S <63534555+skwowet@users.noreply.github.com>

Author: skwowet

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:

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(