refactor: update task assignment models and repository to streamline data handling and enhance dual-write service integration

Author: AnujChhikara

todo/migrations/0001_initial_setup.py

@@ -180,27 +180,19 @@ class Migration(migrations.Migration):
                 ("id", models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID")),
                 ("mongo_id", models.CharField(blank=True, max_length=24, null=True, unique=True)),
                 ("task_mongo_id", models.CharField(max_length=24)),
-                ("user_mongo_id", models.CharField(max_length=24)),
-                ("team_mongo_id", models.CharField(blank=True, max_length=24, null=True)),
+                ("assignee_id", models.CharField(max_length=24)),
                 (
-                    "status",
+                    "user_type",
                     models.CharField(
-                        choices=[
-                            ("ASSIGNED", "Assigned"),
-                            ("IN_PROGRESS", "In Progress"),
-                            ("COMPLETED", "Completed"),
-                            ("REJECTED", "Rejected"),
-                        ],
-                        default="ASSIGNED",
-                        max_length=20,
+                        choices=[("user", "User"), ("team", "Team")],
+                        max_length=10,
                     ),
                 ),
-                ("assigned_at", models.DateTimeField(default=django.utils.timezone.now)),
-                ("started_at", models.DateTimeField(blank=True, null=True)),
-                ("completed_at", models.DateTimeField(blank=True, null=True)),
+                ("team_id", models.CharField(blank=True, max_length=24, null=True)),
+                ("is_active", models.BooleanField(default=True)),
                 ("created_at", models.DateTimeField(default=django.utils.timezone.now)),
                 ("updated_at", models.DateTimeField(blank=True, null=True)),
-                ("assigned_by", models.CharField(max_length=24)),
+                ("created_by", models.CharField(max_length=24)),
                 ("updated_by", models.CharField(blank=True, max_length=24, null=True)),
                 ("last_sync_at", models.DateTimeField(auto_now=True)),
                 (
@@ -218,12 +210,12 @@ class Migration(migrations.Migration):
                 "indexes": [
                     models.Index(fields=["mongo_id"], name="postgres_ta_mongo_i_326fa9_idx"),
                     models.Index(fields=["task_mongo_id"], name="postgres_ta_task_mo_95ca3b_idx"),
-                    models.Index(fields=["user_mongo_id"], name="postgres_ta_user_mo_d13fa3_idx"),
-                    models.Index(fields=["team_mongo_id"], name="postgres_ta_team_mo_a0605f_idx"),
-                    models.Index(fields=["status"], name="postgres_ta_status_8b9698_idx"),
+                    models.Index(fields=["assignee_id"], name="postgres_ta_assignee_95ca3b_idx"),
+                    models.Index(fields=["user_type"], name="postgres_ta_user_typ_d13fa3_idx"),
+                    models.Index(fields=["team_id"], name="postgres_ta_team_id_a0605f_idx"),
+                    models.Index(fields=["is_active"], name="postgres_ta_is_acti_8b9698_idx"),
                     models.Index(fields=["sync_status"], name="postgres_ta_sync_st_385c3f_idx"),
                 ],
-                "unique_together": {("task_mongo_id", "user_mongo_id")},
             },
         ),
         migrations.CreateModel(

todo/models/postgres/task_assignment.py

@@ -12,28 +12,19 @@ class PostgresTaskAssignment(models.Model):
 
     # Assignment fields
     task_mongo_id = models.CharField(max_length=24)  # MongoDB ObjectId as string
-    user_mongo_id = models.CharField(max_length=24)  # MongoDB ObjectId as string
-    team_mongo_id = models.CharField(max_length=24, null=True, blank=True)  # MongoDB ObjectId as string
-
-    # Status
-    STATUS_CHOICES = [
-        ("ASSIGNED", "Assigned"),
-        ("IN_PROGRESS", "In Progress"),
-        ("COMPLETED", "Completed"),
-        ("REJECTED", "Rejected"),
-    ]
-
-    status = models.CharField(max_length=20, choices=STATUS_CHOICES, default="ASSIGNED")
+    assignee_id = models.CharField(max_length=24)  # MongoDB ObjectId as string (user or team ID)
+    user_type = models.CharField(max_length=10, choices=[("user", "User"), ("team", "Team")])  # user or team
+    team_id = models.CharField(
+        max_length=24, null=True, blank=True
+    )  # MongoDB ObjectId as string (only for team assignments)
+    is_active = models.BooleanField(default=True)  # Match MongoDB approach
 
     # Timestamps
-    assigned_at = models.DateTimeField(default=timezone.now)
-    started_at = models.DateTimeField(null=True, blank=True)
-    completed_at = models.DateTimeField(null=True, blank=True)
     created_at = models.DateTimeField(default=timezone.now)
     updated_at = models.DateTimeField(null=True, blank=True)
 
     # References
-    assigned_by = models.CharField(max_length=24)  # MongoDB ObjectId as string
+    created_by = models.CharField(max_length=24)  # MongoDB ObjectId as string
     updated_by = models.CharField(max_length=24, null=True, blank=True)  # MongoDB ObjectId as string
 
     # Sync metadata
@@ -51,18 +42,18 @@ class PostgresTaskAssignment(models.Model):
 
     class Meta:
         db_table = "postgres_task_assignments"
-        unique_together = ["task_mongo_id", "user_mongo_id"]
         indexes = [
             models.Index(fields=["mongo_id"]),
             models.Index(fields=["task_mongo_id"]),
-            models.Index(fields=["user_mongo_id"]),
-            models.Index(fields=["team_mongo_id"]),
-            models.Index(fields=["status"]),
+            models.Index(fields=["assignee_id"]),
+            models.Index(fields=["user_type"]),
+            models.Index(fields=["team_id"]),
+            models.Index(fields=["is_active"]),
             models.Index(fields=["sync_status"]),
         ]
 
     def __str__(self):
-        return f"Task {self.task_mongo_id} assigned to User {self.user_mongo_id}"
+        return f"Task {self.task_mongo_id} assigned to {self.user_type} {self.assignee_id}"
 
     def save(self, *args, **kwargs):
         if not self.pk:  # New instance

todo/repositories/task_assignment_repository.py

@@ -24,15 +24,15 @@ def create(cls, task_assignment: TaskAssignmentModel) -> TaskAssignmentModel:
 
         dual_write_service = EnhancedDualWriteService()
         task_assignment_data = {
-            "task_id": str(task_assignment.task_id),
+            "task_mongo_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,
+            "created_by": str(task_assignment.created_by),
+            "updated_by": str(task_assignment.updated_by) if task_assignment.updated_by else None,
         }
 
         dual_write_success = dual_write_service.create_document(
@@ -130,6 +130,31 @@ def update_assignment(
                 },
             )
 
+            # Sync deactivation to PostgreSQL
+            if current_assignment:
+                dual_write_service = EnhancedDualWriteService()
+                deactivation_data = {
+                    "task_mongo_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_at": current_assignment.created_at,
+                    "updated_at": datetime.now(timezone.utc),
+                    "created_by": str(current_assignment.created_by),
+                    "updated_by": str(user_id),
+                }
+
+                dual_write_success = dual_write_service.update_document(
+                    collection_name="task_assignments", data=deactivation_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 deactivation {current_assignment.id} to Postgres")
+
             new_assignment = TaskAssignmentModel(
                 _id=PyObjectId(),
                 task_id=PyObjectId(task_id),
@@ -181,15 +206,15 @@ def delete_assignment(cls, task_id: str, user_id: str) -> bool:
                 # Sync to PostgreSQL
                 dual_write_service = EnhancedDualWriteService()
                 assignment_data = {
-                    "task_id": str(current_assignment.task_id),
+                    "task_mongo_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),
+                    "created_by": str(current_assignment.created_by),
+                    "updated_by": str(user_id),
                 }
 
                 dual_write_success = dual_write_service.update_document(
@@ -244,15 +269,15 @@ def update_executor(cls, task_id: str, executor_id: str, user_id: str) -> bool:
                 # Sync to PostgreSQL
                 dual_write_service = EnhancedDualWriteService()
                 assignment_data = {
-                    "task_id": str(current_assignment.task_id),
+                    "task_mongo_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),
+                    "created_by": str(current_assignment.created_by),
+                    "updated_by": str(user_id),
                 }
 
                 dual_write_success = dual_write_service.update_document(
@@ -306,15 +331,15 @@ def deactivate_by_task_id(cls, task_id: str, user_id: str) -> bool:
                 # Sync to PostgreSQL for each assignment
                 dual_write_service = EnhancedDualWriteService()
                 assignment_data = {
-                    "task_id": str(active_assignments.task_id),
+                    "task_mongo_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),
+                    "created_by": str(active_assignments.created_by),
+                    "updated_by": str(user_id),
                 }
 
                 dual_write_success = dual_write_service.update_document(

todo/repositories/task_repository.py

@@ -208,23 +208,31 @@ def create(cls, task: TaskModel) -> TaskModel:
                     task.createdAt = datetime.now(timezone.utc)
                     task.updatedAt = None
 
+                    # Ensure createdAt is properly set
+                    if not task.createdAt:
+                        task.createdAt = datetime.now(timezone.utc)
+
                     task_dict = task.model_dump(mode="json", by_alias=True, exclude_none=True)
                     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,
+                        "displayId": task.displayId,
+                        "isAcknowledged": task.isAcknowledged,
+                        "isDeleted": task.isDeleted,
+                        "startedAt": task.startedAt,
+                        "dueAt": task.dueAt,
+                        "createdAt": task.createdAt or datetime.now(timezone.utc),
+                        "updatedAt": task.updatedAt,
+                        "createdBy": str(task.createdBy),
+                        "updatedBy": str(task.updatedBy) if task.updatedBy else None,
                     }
 
                     dual_write_success = dual_write_service.create_document(
@@ -313,12 +321,15 @@ def update(cls, task_id: str, update_data: dict) -> TaskModel | None:
                 "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,
+                "displayId": task_model.displayId,
+                "isAcknowledged": task_model.isAcknowledged,
+                "isDeleted": task_model.isDeleted,
+                "startedAt": task_model.startedAt,
+                "dueAt": task_model.dueAt,
+                "createdAt": task_model.createdAt,
+                "updatedAt": task_model.updatedAt,
+                "createdBy": str(task_model.createdBy),
+                "updatedBy": str(task_model.updatedBy) if task_model.updatedBy else None,
             }
 
             dual_write_success = dual_write_service.update_document(

todo/services/dual_write_service.py

@@ -115,9 +115,12 @@ def update_document(self, collection_name: str, mongo_id: str, data: Dict[str, A
                 labels = postgres_data.pop("labels", []) if collection_name == "tasks" else []
 
                 postgres_instance = postgres_model.objects.get(mongo_id=mongo_id)
+                preserve_fields = {"created_at", "mongo_id"}
+
                 for field, value in postgres_data.items():
-                    if hasattr(postgres_instance, field):
+                    if hasattr(postgres_instance, field) and field not in preserve_fields:
                         setattr(postgres_instance, field, value)
+
                 postgres_instance.sync_status = "SYNCED"
                 postgres_instance.sync_error = None
                 postgres_instance.save()
@@ -309,15 +312,13 @@ def _transform_task_assignment_data(self, data: Dict[str, Any]) -> Dict[str, Any
         """Transform task assignment data for Postgres."""
         return {
             "task_mongo_id": str(data.get("task_mongo_id", "")),
-            "user_mongo_id": str(data.get("user_mongo_id", "")),
-            "team_mongo_id": str(data.get("team_mongo_id", "")) if data.get("team_mongo_id") else None,
-            "status": data.get("status", "ASSIGNED"),
-            "assigned_at": data.get("assigned_at"),
-            "started_at": data.get("started_at"),
-            "completed_at": data.get("completed_at"),
+            "assignee_id": str(data.get("assignee_id", "")),
+            "user_type": data.get("user_type", "user"),
+            "team_id": str(data.get("team_id", "")) if data.get("team_id") else None,
+            "is_active": data.get("is_active", True),
             "created_at": data.get("created_at"),
             "updated_at": data.get("updated_at"),
-            "assigned_by": str(data.get("assigned_by", "")),
+            "created_by": str(data.get("created_by", "")),
             "updated_by": str(data.get("updated_by", "")) if data.get("updated_by") else None,
         }