refactor: enhance PostgreSQL model definitions and integrate dual-write service for task assignments and user roles

Author: AnujChhikara

todo/migrations/0001_initial_setup.py

@@ -6,23 +6,45 @@
 from todo.models.task_assignment import TaskAssignmentModel
 from todo.repositories.common.mongo_repository import MongoRepository
 from todo.models.common.pyobjectid import PyObjectId
+from todo.services.enhanced_dual_write_service import EnhancedDualWriteService
 
 
 class TaskAssignmentRepository(MongoRepository):
     collection_name = TaskAssignmentModel.collection_name
 
     @classmethod
     def create(cls, task_assignment: TaskAssignmentModel) -> TaskAssignmentModel:
-        """
-        Creates a new task assignment.
-        """
         collection = cls.get_collection()
         task_assignment.created_at = datetime.now(timezone.utc)
         task_assignment.updated_at = None
 
         task_assignment_dict = task_assignment.model_dump(mode="json", by_alias=True, exclude_none=True)
         insert_result = collection.insert_one(task_assignment_dict)
         task_assignment.id = insert_result.inserted_id
+
+        dual_write_service = EnhancedDualWriteService()
+        task_assignment_data = {
+            "task_id": str(task_assignment.task_id),
+            "assignee_id": str(task_assignment.assignee_id),
+            "user_type": task_assignment.user_type,
+            "team_id": str(task_assignment.team_id) if task_assignment.team_id else None,
+            "is_active": task_assignment.is_active,
+            "created_by": str(task_assignment.created_by),
+            "updated_by": str(task_assignment.updated_by) if task_assignment.updated_by else None,
+            "created_at": task_assignment.created_at,
+            "updated_at": task_assignment.updated_at,
+        }
+
+        dual_write_success = dual_write_service.create_document(
+            collection_name="task_assignments", data=task_assignment_data, mongo_id=str(task_assignment.id)
+        )
+
+        if not dual_write_success:
+            import logging
+
+            logger = logging.getLogger(__name__)
+            logger.warning(f"Failed to sync task assignment {task_assignment.id} to Postgres")
+
         return task_assignment
 
     @classmethod
@@ -124,11 +146,13 @@ def update_assignment(
 
     @classmethod
     def delete_assignment(cls, task_id: str, user_id: str) -> bool:
-        """
-        Soft delete a task assignment by setting is_active to False.
-        """
         collection = cls.get_collection()
         try:
+            # Get current assignment first
+            current_assignment = cls.get_by_task_id(task_id)
+            if not current_assignment:
+                return False
+
             # Try with ObjectId first
             result = collection.update_one(
                 {"task_id": ObjectId(task_id), "is_active": True},
@@ -152,17 +176,45 @@ def delete_assignment(cls, task_id: str, user_id: str) -> bool:
                         }
                     },
                 )
+
+            if result.modified_count > 0:
+                # Sync to PostgreSQL
+                dual_write_service = EnhancedDualWriteService()
+                assignment_data = {
+                    "task_id": str(current_assignment.task_id),
+                    "assignee_id": str(current_assignment.assignee_id),
+                    "user_type": current_assignment.user_type,
+                    "team_id": str(current_assignment.team_id) if current_assignment.team_id else None,
+                    "is_active": False,
+                    "created_by": str(current_assignment.created_by),
+                    "updated_by": str(user_id),
+                    "created_at": current_assignment.created_at,
+                    "updated_at": datetime.now(timezone.utc),
+                }
+
+                dual_write_success = dual_write_service.update_document(
+                    collection_name="task_assignments", data=assignment_data, mongo_id=str(current_assignment.id)
+                )
+
+                if not dual_write_success:
+                    import logging
+
+                    logger = logging.getLogger(__name__)
+                    logger.warning(f"Failed to sync task assignment deletion {current_assignment.id} to Postgres")
+
             return result.modified_count > 0
         except Exception:
             return False
 
     @classmethod
     def update_executor(cls, task_id: str, executor_id: str, user_id: str) -> bool:
-        """
-        Update the executor_id for the active assignment of the given task_id.
-        """
         collection = cls.get_collection()
         try:
+            # Get current assignment first
+            current_assignment = cls.get_by_task_id(task_id)
+            if not current_assignment:
+                return False
+
             result = collection.update_one(
                 {"task_id": ObjectId(task_id), "is_active": True},
                 {
@@ -187,17 +239,45 @@ def update_executor(cls, task_id: str, executor_id: str, user_id: str) -> bool:
                         }
                     },
                 )
+
+            if result.modified_count > 0:
+                # Sync to PostgreSQL
+                dual_write_service = EnhancedDualWriteService()
+                assignment_data = {
+                    "task_id": str(current_assignment.task_id),
+                    "assignee_id": str(executor_id),
+                    "user_type": "user",
+                    "team_id": str(current_assignment.team_id) if current_assignment.team_id else None,
+                    "is_active": current_assignment.is_active,
+                    "created_by": str(current_assignment.created_by),
+                    "updated_by": str(user_id),
+                    "created_at": current_assignment.created_at,
+                    "updated_at": datetime.now(timezone.utc),
+                }
+
+                dual_write_success = dual_write_service.update_document(
+                    collection_name="task_assignments", data=assignment_data, mongo_id=str(current_assignment.id)
+                )
+
+                if not dual_write_success:
+                    import logging
+
+                    logger = logging.getLogger(__name__)
+                    logger.warning(f"Failed to sync task assignment update {current_assignment.id} to Postgres")
+
             return result.modified_count > 0
         except Exception:
             return False
 
     @classmethod
     def deactivate_by_task_id(cls, task_id: str, user_id: str) -> bool:
-        """
-        Deactivate all assignments for a specific task by setting is_active to False.
-        """
         collection = cls.get_collection()
         try:
+            # Get all active assignments for this task
+            active_assignments = cls.get_by_task_id(task_id)
+            if not active_assignments:
+                return False
+
             # Try with ObjectId first
             result = collection.update_many(
                 {"task_id": ObjectId(task_id), "is_active": True},
@@ -221,6 +301,32 @@ def deactivate_by_task_id(cls, task_id: str, user_id: str) -> bool:
                         }
                     },
                 )
+
+            if result.modified_count > 0:
+                # Sync to PostgreSQL for each assignment
+                dual_write_service = EnhancedDualWriteService()
+                assignment_data = {
+                    "task_id": str(active_assignments.task_id),
+                    "assignee_id": str(active_assignments.assignee_id),
+                    "user_type": active_assignments.user_type,
+                    "team_id": str(active_assignments.team_id) if active_assignments.team_id else None,
+                    "is_active": False,
+                    "created_by": str(active_assignments.created_by),
+                    "updated_by": str(user_id),
+                    "created_at": active_assignments.created_at,
+                    "updated_at": datetime.now(timezone.utc),
+                }
+
+                dual_write_success = dual_write_service.update_document(
+                    collection_name="task_assignments", data=assignment_data, mongo_id=str(active_assignments.id)
+                )
+
+                if not dual_write_success:
+                    import logging
+
+                    logger = logging.getLogger(__name__)
+                    logger.warning(f"Failed to sync task assignment deactivation {active_assignments.id} to Postgres")
+
             return result.modified_count > 0
         except Exception:
             return False

todo/repositories/task_assignment_repository.py

@@ -16,6 +16,8 @@
     TaskStatus,
 )
 from todo.repositories.team_repository import UserTeamDetailsRepository
+from todo.services.enhanced_dual_write_service import EnhancedDualWriteService
+from todo.models.postgres import PostgresTask, PostgresDeferredDetails
 
 
 class TaskRepository(MongoRepository):
@@ -210,6 +212,31 @@ def create(cls, task: TaskModel) -> TaskModel:
                     insert_result = tasks_collection.insert_one(task_dict, session=session)
 
                     task.id = insert_result.inserted_id
+
+                    dual_write_service = EnhancedDualWriteService()
+                    task_data = {
+                        "title": task.title,
+                        "description": task.description,
+                        "priority": task.priority,
+                        "status": task.status,
+                        "display_id": task.displayId,
+                        "created_by": str(task.createdBy),
+                        "updated_by": str(task.updatedBy) if task.updatedBy else None,
+                        "is_deleted": task.isDeleted,
+                        "created_at": task.createdAt,
+                        "updated_at": task.updatedAt,
+                    }
+
+                    dual_write_success = dual_write_service.create_document(
+                        collection_name="tasks", data=task_data, mongo_id=str(task.id)
+                    )
+
+                    if not dual_write_success:
+                        import logging
+
+                        logger = logging.getLogger(__name__)
+                        logger.warning(f"Failed to sync task {task.id} to Postgres")
+
                     return task
 
             except Exception as e:
@@ -259,9 +286,6 @@ def delete_by_id(cls, task_id: ObjectId, user_id: str) -> TaskModel | None:
 
     @classmethod
     def update(cls, task_id: str, update_data: dict) -> TaskModel | None:
-        """
-        Updates a specific task by its ID with the given data.
-        """
         if not isinstance(update_data, dict):
             raise ValueError("update_data must be a dictionary.")
 
@@ -281,7 +305,37 @@ def update(cls, task_id: str, update_data: dict) -> TaskModel | None:
         )
 
         if updated_task_doc:
-            return TaskModel(**updated_task_doc)
+            task_model = TaskModel(**updated_task_doc)
+
+            dual_write_service = EnhancedDualWriteService()
+            task_data = {
+                "title": task_model.title,
+                "description": task_model.description,
+                "priority": task_model.priority,
+                "status": task_model.status,
+                "display_id": task_model.displayId,
+                "created_by": str(task_model.createdBy),
+                "updated_by": str(task_model.updatedBy) if task_model.updatedBy else None,
+                "is_deleted": task_model.isDeleted,
+                "created_at": task_model.createdAt,
+                "updated_at": task_model.updatedAt,
+            }
+
+            dual_write_success = dual_write_service.update_document(
+                collection_name="tasks", data=task_data, mongo_id=str(task_model.id)
+            )
+
+            if not dual_write_success:
+                import logging
+
+                logger = logging.getLogger(__name__)
+                logger.warning(f"Failed to sync task update {task_model.id} to Postgres")
+
+            # Handle deferred details if present in update_data
+            if "deferredDetails" in update_data:
+                cls._handle_deferred_details_sync(task_id, update_data["deferredDetails"])
+
+            return task_model
         return None
 
     @classmethod
@@ -307,3 +361,30 @@ def get_by_ids(cls, task_ids: List[str]) -> List[TaskModel]:
         object_ids = [ObjectId(task_id) for task_id in task_ids]
         cursor = tasks_collection.find({"_id": {"$in": object_ids}})
         return [TaskModel(**doc) for doc in cursor]
+
+    @classmethod
+    def _handle_deferred_details_sync(cls, task_id: str, deferred_details: dict) -> None:
+        """Handle deferred details synchronization to PostgreSQL"""
+        try:
+            postgres_task = PostgresTask.objects.get(mongo_id=task_id)
+
+            if deferred_details:
+                deferred_details_data = {
+                    "task": postgres_task,
+                    "deferred_at": deferred_details.get("deferredAt"),
+                    "deferred_till": deferred_details.get("deferredTill"),
+                    "deferred_by": str(deferred_details.get("deferredBy")),
+                }
+
+                PostgresDeferredDetails.objects.update_or_create(task=postgres_task, defaults=deferred_details_data)
+            else:
+                # Remove deferred details if None
+                PostgresDeferredDetails.objects.filter(task=postgres_task).delete()
+
+        except PostgresTask.DoesNotExist:
+            pass
+        except Exception as e:
+            import logging
+
+            logger = logging.getLogger(__name__)
+            logger.warning(f"Failed to sync deferred details to PostgreSQL for task {task_id}: {str(e)}")

todo/repositories/task_repository.py

@@ -4,6 +4,7 @@
 from todo.repositories.common.mongo_repository import MongoRepository
 from todo.models.team_creation_invite_code import TeamCreationInviteCodeModel
 from todo.repositories.user_repository import UserRepository
+from todo.services.enhanced_dual_write_service import EnhancedDualWriteService
 
 
 class TeamCreationInviteCodeRepository(MongoRepository):
@@ -38,13 +39,34 @@ def validate_and_consume_code(cls, code: str, used_by: str) -> Optional[dict]:
 
     @classmethod
     def create(cls, team_invite_code: TeamCreationInviteCodeModel) -> TeamCreationInviteCodeModel:
-        """Create a new team invite code."""
         collection = cls.get_collection()
         team_invite_code.created_at = datetime.now(timezone.utc)
 
         code_dict = team_invite_code.model_dump(mode="json", by_alias=True, exclude_none=True)
         insert_result = collection.insert_one(code_dict)
         team_invite_code.id = insert_result.inserted_id
+
+        dual_write_service = EnhancedDualWriteService()
+        invite_code_data = {
+            "code": team_invite_code.code,
+            "description": team_invite_code.description,
+            "is_used": team_invite_code.is_used,
+            "created_by": str(team_invite_code.created_by),
+            "used_by": str(team_invite_code.used_by) if team_invite_code.used_by else None,
+            "created_at": team_invite_code.created_at,
+            "used_at": team_invite_code.used_at,
+        }
+
+        dual_write_success = dual_write_service.create_document(
+            collection_name="team_creation_invite_codes", data=invite_code_data, mongo_id=str(team_invite_code.id)
+        )
+
+        if not dual_write_success:
+            import logging
+
+            logger = logging.getLogger(__name__)
+            logger.warning(f"Failed to sync team creation invite code {team_invite_code.id} to Postgres")
+
         return team_invite_code
 
     @classmethod