feat: discover and parse project-maintained affiliations (CM-361) (#4280)

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

Author: skwowet

services/apps/git_integration/src/crowdgit/database/crud.py

@@ -5,6 +5,7 @@
 
 from crowdgit.enums import RepositoryPriority, RepositoryState
 from crowdgit.errors import RepoLockingError
+from crowdgit.models.affiliation_info import RepoAffiliationRegistry
 from crowdgit.models.repository import Repository
 from crowdgit.models.service_execution import ServiceExecution
 from crowdgit.settings import (
@@ -552,3 +553,303 @@ async def save_service_execution(service_execution: ServiceExecution) -> None:
             f"error: {e}"
         )
         # Do not re-raise - we don't want metrics saving to disrupt main operations
+
+
+async def get_repo_affiliation_registry(repo_id: str) -> RepoAffiliationRegistry | None:
+    sql_query = """
+        SELECT "repoId", "filePath", "fileHash", "status", "snapshot", "lastRunAt"
+        FROM git."repoAffiliationRegistry"
+        WHERE "repoId" = $1
+    """
+    result = await fetchrow(sql_query, (repo_id,))
+    if not result:
+        return None
+
+    return RepoAffiliationRegistry.from_db(dict(result))
+
+
+async def upsert_repo_affiliation_registry(registry: RepoAffiliationRegistry) -> None:
+    snapshot_json = registry.snapshot_for_db()
+    sql_query = """
+        INSERT INTO git."repoAffiliationRegistry" (
+            "repoId", "filePath", "fileHash", "status", "snapshot", "lastRunAt", "updatedAt"
+        )
+        VALUES ($1, $2, $3, $4, $5::jsonb, NOW(), NOW())
+        ON CONFLICT ("repoId") DO UPDATE SET
+            "filePath" = EXCLUDED."filePath",
+            "fileHash" = EXCLUDED."fileHash",
+            "status" = EXCLUDED."status",
+            "snapshot" = EXCLUDED."snapshot",
+            "lastRunAt" = NOW(),
+            "updatedAt" = NOW()
+    """
+    await execute(
+        sql_query,
+        (
+            registry.repo_id,
+            registry.file_path,
+            registry.file_hash,
+            registry.status,
+            snapshot_json,
+        ),
+    )
+
+
+async def find_many_member_ids_by_identities(identities: list[dict]) -> list[dict]:
+    if not identities:
+        return []
+
+    values_parts: list[str] = []
+    params: list[str | bool | int] = []
+    param_index = 1
+    for idx, identity in enumerate(identities):
+        values_parts.append(
+            f"(${param_index}::int, ${param_index + 1}::text, ${param_index + 2}::boolean,"
+            f" ${param_index + 3}::text, ${param_index + 4}::text)"
+        )
+        params.extend(
+            [
+                idx,
+                identity["type"],
+                identity.get("verified", True),
+                identity.get("platform"),
+                identity["value"],
+            ]
+        )
+        param_index += 5
+
+    matches_by_idx: dict[int, set[str]] = {}
+    rows = await query(
+        f"""
+        WITH input_identities (idx, identity_type, verified, platform, value) AS (
+            VALUES {", ".join(values_parts)}
+        )
+        SELECT i.idx, mi."memberId"
+        FROM input_identities i
+        LEFT JOIN "memberIdentities" mi
+            ON mi.type = i.identity_type
+            AND mi.verified = i.verified
+            AND lower(mi.value) = lower(i.value)
+            AND mi.platform = i.platform
+            AND mi."deletedAt" IS NULL
+        ORDER BY i.idx
+        """,
+        tuple(params),
+    )
+    for row in rows:
+        if row["memberId"] is None:
+            continue
+        matches_by_idx.setdefault(row["idx"], set()).add(str(row["memberId"]))
+
+    results: list[dict] = []
+    for idx, identity in enumerate(identities):
+        member_ids = matches_by_idx.get(idx, set())
+        member_id = next(iter(member_ids)) if len(member_ids) == 1 else None
+        results.append(
+            {
+                "type": identity["type"],
+                "platform": identity.get("platform"),
+                "value": identity["value"],
+                "verified": identity.get("verified", True),
+                "member_id": member_id,
+            }
+        )
+
+    return results
+
+
+async def find_many_organization_ids_by_identities(identities: list[dict]) -> list[dict]:
+    if not identities:
+        return []
+
+    values_parts: list[str] = []
+    params: list[str | bool | int] = []
+    param_index = 1
+    for idx, identity in enumerate(identities):
+        values_parts.append(
+            f"(${param_index}::int, ${param_index + 1}::text, ${param_index + 2}::boolean,"
+            f" ${param_index + 3}::text, ${param_index + 4}::text)"
+        )
+        params.extend(
+            [
+                idx,
+                identity["type"],
+                identity.get("verified", True),
+                identity["platform"],
+                identity["value"],
+            ]
+        )
+        param_index += 5
+
+    matches_by_idx: dict[int, set[str]] = {}
+    rows = await query(
+        f"""
+        WITH input_identities (idx, identity_type, verified, platform, value) AS (
+            VALUES {", ".join(values_parts)}
+        )
+        SELECT i.idx, oi."organizationId"
+        FROM input_identities i
+        LEFT JOIN "organizationIdentities" oi
+            ON oi.type = i.identity_type
+            AND oi.verified = i.verified
+            AND oi.platform = i.platform
+            AND lower(oi.value) = lower(i.value)
+        ORDER BY i.idx
+        """,
+        tuple(params),
+    )
+    for row in rows:
+        if row["organizationId"] is None:
+            continue
+        matches_by_idx.setdefault(row["idx"], set()).add(str(row["organizationId"]))
+
+    results: list[dict] = []
+    for idx, identity in enumerate(identities):
+        organization_ids = matches_by_idx.get(idx, set())
+        organization_id = next(iter(organization_ids)) if len(organization_ids) == 1 else None
+        results.append(
+            {
+                "type": identity["type"],
+                "platform": identity["platform"],
+                "value": identity["value"],
+                "verified": identity.get("verified", True),
+                "organization_id": organization_id,
+            }
+        )
+
+    return results
+
+
+async def fetch_member_organizations(member_ids: list[str]) -> list[dict]:
+    if not member_ids:
+        return []
+
+    return await query(
+        """
+        SELECT "memberId", "organizationId", "dateStart", "dateEnd", source
+        FROM "memberOrganizations"
+        WHERE "memberId" = ANY($1::uuid[])
+            AND "deletedAt" IS NULL
+        """,
+        (member_ids,),
+    )
+
+
+async def fetch_segment_affiliations(member_ids: list[str], segment_id: str) -> list[dict]:
+    """MSA rows are per segment — filter by segment_id so guards match this repo's project."""
+    if not member_ids:
+        return []
+
+    return await query(
+        """
+        SELECT "memberId", "segmentId", "organizationId", "dateStart", "dateEnd", verified
+        FROM "memberSegmentAffiliations"
+        WHERE "memberId" = ANY($1::uuid[])
+            AND "segmentId" = $2::uuid
+            AND "deletedAt" IS NULL
+            AND "organizationId" IS NOT NULL
+        """,
+        (member_ids, segment_id),
+    )
+
+
+async def insert_member_organizations(rows: list[dict]) -> None:
+    if not rows:
+        return
+
+    undated_rows: list[tuple] = []
+    open_ended_rows: list[tuple] = []
+    dated_rows: list[tuple] = []
+
+    for row in rows:
+        params = (
+            row["member_id"],
+            row["organization_id"],
+            row.get("date_start"),
+            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:
+            open_ended_rows.append(params)
+        else:
+            dated_rows.append(params)
+
+    insert_sql = """
+        INSERT INTO "memberOrganizations"(
+            "memberId",
+            "organizationId",
+            "dateStart",
+            "dateEnd",
+            title,
+            source,
+            "createdAt",
+            "updatedAt"
+        )
+        VALUES ($1, $2, $3, $4, NULL, $5, NOW(), NOW())
+    """
+
+    if undated_rows:
+        sql = (
+            insert_sql
+            + """
+            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
+            + """
+            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
+            + """
+            ON CONFLICT ("memberId", "organizationId", "dateStart", "dateEnd")
+                WHERE ("deletedAt" IS NULL)
+            DO NOTHING
+        """
+        )
+        await executemany(sql, dated_rows)
+
+
+async def insert_member_segment_affiliations(rows: list[dict]) -> None:
+    if not rows:
+        return
+
+    await executemany(
+        """
+        INSERT INTO "memberSegmentAffiliations"(
+            id,
+            "memberId",
+            "segmentId",
+            "organizationId",
+            "dateStart",
+            "dateEnd"
+        )
+        VALUES (gen_random_uuid(), $1, $2, $3, $4, $5)
+        """,
+        [
+            (
+                row["member_id"],
+                row["segment_id"],
+                row["organization_id"],
+                row.get("date_start"),
+                row.get("date_end"),
+            )
+            for row in rows
+        ],
+    )

services/apps/git_integration/src/crowdgit/enums.py

@@ -18,6 +18,9 @@ class ErrorCode(str, Enum):
     NO_MAINTAINER_FOUND = "no-maintainer-found"
     MAINTAINER_ANALYSIS_FAILED = "maintainer-analysis-failed"
     MAINTAINER_INTERVAL_NOT_ELAPSED = "maintainer-interval-not-elapsed"
+    NO_AFFILIATION_FILE = "no-affiliation-file"
+    AFFILIATION_ANALYSIS_FAILED = "affiliation-analysis-failed"
+    AFFILIATION_INTERVAL_NOT_ELAPSED = "affiliation-interval-not-elapsed"
     CLEANUP_FAILED = "cleanup-failed"
     PARENT_REPO_INVALID = "parent-repo-invalid"
     REONBOARDING_REQUIRED = "reonboarding-required"
@@ -67,11 +70,19 @@ class ExecutionStatus(str, Enum):
     FAILURE = "failure"
 
 
+class AffiliationRegistryStatus(str, Enum):
+    SUCCESS = "success"
+    NOT_FOUND = "not_found"
+    UNUSABLE = "unusable"
+    ERROR = "error"
+
+
 class OperationType(str, Enum):
     """Service operation types for metrics tracking"""
 
     CLONE = "Clone"
     COMMIT = "Commit"
     MAINTAINER = "Maintainer"
+    AFFILIATION = "Affiliation"
     SOFTWARE_VALUE = "SoftwareValue"
     VULNERABILITY_SCAN = "VulnerabilityScanner"

services/apps/git_integration/src/crowdgit/errors.py

@@ -104,6 +104,27 @@ class MaintainerIntervalNotElapsedError(CrowdGitError):
     ai_cost: int = 0
 
 
+@dataclass
+class AffiliationFileNotFoundError(CrowdGitError):
+    error_message: str = "No affiliation file found in this repository"
+    error_code: ErrorCode = ErrorCode.NO_AFFILIATION_FILE
+    ai_cost: float = 0.0
+
+
+@dataclass
+class AffiliationAnalysisError(CrowdGitError):
+    error_message: str = "Could not parse the affiliation file"
+    error_code: ErrorCode = ErrorCode.AFFILIATION_ANALYSIS_FAILED
+    retain_file_hash: bool = False
+
+
+@dataclass
+class AffiliationIntervalNotElapsedError(CrowdGitError):
+    error_message: str = "Too soon since the last affiliation run"
+    error_code: ErrorCode = ErrorCode.AFFILIATION_INTERVAL_NOT_ELAPSED
+    ai_cost: float = 0.0
+
+
 @dataclass
 class ParentRepoInvalidError(CrowdGitError):
     error_message: str = "Parent repository is not valid or not found"