Merge branch 'main' into chore/update-pvtr-version

Author: gaspergrom

backend/.env.dist.local

@@ -151,6 +151,7 @@ CROWD_TEMPORAL_ENCRYPTION_KEY_ID=local
 CROWD_TEMPORAL_ENCRYPTION_KEY=FweBMRnGCLshER8FlSvNusQA6G3MRUKt
 
 # Temporal — packages namespace
+CROWD_PACKAGES_TEMPORAL_SERVER_URL=localhost:7233
 CROWD_PACKAGES_TEMPORAL_NAMESPACE=default
 
 # Seach sync api

backend/config/custom-environment-variables.json

@@ -180,6 +180,12 @@
     "certificate": "CROWD_TEMPORAL_CERTIFICATE",
     "privateKey": "CROWD_TEMPORAL_PRIVATE_KEY"
   },
+  "packagesTemporal": {
+    "serverUrl": "CROWD_PACKAGES_TEMPORAL_SERVER_URL",
+    "namespace": "CROWD_PACKAGES_TEMPORAL_NAMESPACE",
+    "certificate": "CROWD_TEMPORAL_CERTIFICATE",
+    "privateKey": "CROWD_TEMPORAL_PRIVATE_KEY"
+  },
   "searchSyncApi": {
     "baseUrl": "CROWD_SEARCH_SYNC_API_URL"
   },

backend/config/default.json

@@ -43,6 +43,7 @@
     "isEnabled": "false"
   },
   "temporal": {},
+  "packagesTemporal": {},
   "searchSyncApi": {},
   "encryption": {},
   "openStatusApi": {},

backend/src/conf/index.ts

@@ -86,6 +86,14 @@ export const PACKAGES_DB_CONFIG: IDatabaseConfig | undefined = config.has('packa
   ? config.get<IDatabaseConfig>('packagesDb')
   : undefined
 
+// packages_worker (npm/maven/pypi/osv/security-contacts/...) runs in its own Temporal
+// namespace, separate from the API's default namespace — see CROWD_PACKAGES_TEMPORAL_NAMESPACE.
+export const PACKAGES_TEMPORAL_CONFIG: ITemporalConfig | undefined = config.has(
+  'packagesTemporal.namespace',
+)
+  ? config.get<ITemporalConfig>('packagesTemporal')
+  : undefined
+
 export const SEGMENT_CONFIG: SegmentConfiguration = config.get<SegmentConfiguration>('segment')
 
 export const COMPREHEND_CONFIG: ComprehendConfiguration =

backend/src/db/packagesTemporal.ts

@@ -0,0 +1,45 @@
+import { IS_DEV_ENV, SERVICE } from '@crowd/common'
+import { Client, Connection, getDataConverter } from '@crowd/temporal'
+
+import { PACKAGES_TEMPORAL_CONFIG } from '@/conf'
+
+let _init: Promise<Client> | undefined
+
+// Separate connection from the API's default req.temporal client — packages_worker
+// (npm/maven/pypi/osv/security-contacts/...) polls task queues in its own Temporal
+// namespace (CROWD_PACKAGES_TEMPORAL_NAMESPACE), not the API's default namespace.
+export function getPackagesTemporalClient(): Promise<Client> {
+  if (!_init) {
+    if (!PACKAGES_TEMPORAL_CONFIG?.serverUrl) {
+      throw new Error('Packages Temporal is not configured — set CROWD_PACKAGES_TEMPORAL_NAMESPACE')
+    }
+
+    const cfg = PACKAGES_TEMPORAL_CONFIG
+    _init = Connection.connect({
+      address: cfg.serverUrl,
+      tls:
+        cfg.certificate && cfg.privateKey
+          ? {
+              clientCertPair: {
+                crt: Buffer.from(cfg.certificate, 'base64'),
+                key: Buffer.from(cfg.privateKey, 'base64'),
+              },
+            }
+          : undefined,
+    })
+      .then(
+        async (connection) =>
+          new Client({
+            connection,
+            namespace: cfg.namespace,
+            identity: SERVICE,
+            dataConverter: IS_DEV_ENV ? undefined : await getDataConverter(),
+          }),
+      )
+      .catch((err) => {
+        _init = undefined
+        throw err
+      })
+  }
+  return _init
+}

scripts/services/security-contacts-worker.yaml

@@ -7,6 +7,7 @@ x-env-args: &env-args
   SHELL: /bin/sh
   SUPPRESS_NO_CONFIG_WARNING: 'true'
   CROWD_TEMPORAL_TASKQUEUE: packages-worker
+  CROWD_TEMPORAL_NAMESPACE: ${CROWD_PACKAGES_TEMPORAL_NAMESPACE}
 
 services:
   security-contacts-worker:

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: