fix: affiliation registry writes and expected-run reporting

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

Author: skwowet

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

@@ -546,7 +546,7 @@ async def upsert_repo_affiliation_registry(registry: RepoAffiliationRegistry) ->
         INSERT INTO git."repoAffiliationRegistry" (
             "repoId", "filePath", "fileHash", "status", "snapshot", "lastRunAt", "updatedAt"
         )
-        VALUES ($1, $2, $3, $4, $5, NOW(), NOW())
+        VALUES ($1, $2, $3, $4, $5::jsonb, NOW(), NOW())
         ON CONFLICT ("repoId") DO UPDATE SET
             "filePath" = EXCLUDED."filePath",
             "fileHash" = EXCLUDED."fileHash",
@@ -576,7 +576,8 @@ async def find_many_member_ids_by_identities(identities: list[dict]) -> list[dic
     param_index = 1
     for idx, identity in enumerate(identities):
         values_parts.append(
-            f"(${param_index}, ${param_index + 1}, ${param_index + 2}, ${param_index + 3}, ${param_index + 4})"
+            f"(${param_index}::int, ${param_index + 1}::text, ${param_index + 2}::boolean,"
+            f" ${param_index + 3}::text, ${param_index + 4}::text)"
         )
         params.extend(
             [
@@ -638,7 +639,8 @@ async def find_many_organization_ids_by_identities(identities: list[dict]) -> li
     param_index = 1
     for idx, identity in enumerate(identities):
         values_parts.append(
-            f"(${param_index}, ${param_index + 1}, ${param_index + 2}, ${param_index + 3})"
+            f"(${param_index}::int, ${param_index + 1}::text,"
+            f" ${param_index + 2}::boolean, ${param_index + 3}::text)"
         )
         params.extend(
             [

services/apps/git_integration/src/crowdgit/models/affiliation_info.py

@@ -4,6 +4,7 @@
 from datetime import datetime
 from typing import Any
 
+import orjson
 from loguru import logger
 from pydantic import BaseModel, TypeAdapter, ValidationError
 
@@ -71,6 +72,14 @@ def from_db(cls, db_data: dict[str, Any]) -> RepoAffiliationRegistry:
 
     @staticmethod
     def _parse_snapshot(snapshot) -> list[AffiliationInfoItem] | None:
+        if isinstance(snapshot, str | bytes):
+            try:
+                snapshot = orjson.loads(snapshot)
+            except orjson.JSONDecodeError as error:
+                logger.warning(
+                    f"Invalid affiliation snapshot JSON in registry, will re-parse: {error}"
+                )
+                return None
         if isinstance(snapshot, dict) and "affiliations" in snapshot:
             snapshot = snapshot["affiliations"]
         try:
@@ -79,7 +88,7 @@ def _parse_snapshot(snapshot) -> list[AffiliationInfoItem] | None:
             logger.warning(f"Invalid affiliation snapshot in registry, will re-parse: {error}")
             return None
 
-    def snapshot_for_db(self) -> list[dict] | None:
+    def snapshot_for_db(self) -> str | None:
         if self.snapshot is None:
             return None
-        return [item.model_dump() for item in self.snapshot]
+        return orjson.dumps([item.model_dump() for item in self.snapshot]).decode()

services/apps/git_integration/src/crowdgit/services/affiliation/affiliation_service.py

@@ -858,21 +858,13 @@ async def process_affiliations(
             self.logger.info(f"Finished with {len(affiliations)} rows from {latest_file_path}")
 
         except AffiliationIntervalNotElapsedError as e:
-            execution_status = ExecutionStatus.FAILURE
-            error_message = e.error_message
-            error_code = e.error_code.value
+            self.logger.info(e.error_message)
 
         except AffiliationFileNotFoundError as e:
-            execution_status = ExecutionStatus.FAILURE
-            error_message = e.error_message
-            error_code = e.error_code.value
             ai_cost = e.ai_cost
-            self.logger.info(error_message)
+            self.logger.info(e.error_message)
 
         except AffiliationAnalysisError as e:
-            execution_status = ExecutionStatus.FAILURE
-            error_message = e.error_message
-            error_code = e.error_code.value
             await upsert_repo_affiliation_registry(
                 RepoAffiliationRegistry(
                     repo_id=repository.id,
@@ -888,7 +880,13 @@ async def process_affiliations(
                     else (registry.snapshot if registry else None),
                 )
             )
-            self.logger.warning(error_message)
+            if e.retain_file_hash:
+                self.logger.info(e.error_message)
+            else:
+                execution_status = ExecutionStatus.FAILURE
+                error_message = e.error_message
+                error_code = e.error_code.value
+                self.logger.warning(error_message)
 
         except Exception as e:
             execution_status = ExecutionStatus.FAILURE