feat: Add team creation invite code system with admin-only access (#249)

* feat: add team invite code functionality

* feat: implement team invite code generation and verification functionality

* refactor: clean up code formatting and remove unused imports in team invite code files

* chore: add audit logs in team creation invite code

* refactor: improve error handling

* feat: add endpoint to list team creation invite codes with pagination and user details

* chore: remove unused user ID from team creation invite code

* refactor: improve code formatting

* refactor: enhance DTO descriptions

* refactor: update team creation invite code handling and validation logic

* fix: patch unit test

* refactor: rename method for retrieving team creation invite codes and improve error handling

* feat: enhance JWT authentication middleware to verify user existence and include email in request

Author: AnujChhikara

.env.example

@@ -27,4 +27,6 @@ TODO_BACKEND_BASE_URL='http://localhost:8000'
 
 CORS_ALLOWED_ORIGINS='http://localhost:3000,http://localhost:8000'
 
-SWAGGER_UI_PATH='/api/schema'
+SWAGGER_UI_PATH='/api/schema'
+
+ADMIN_EMAILS =  "[email protected],[email protected]"

todo/dto/responses/generate_team_creation_invite_code_response.py

@@ -0,0 +1,15 @@
+from pydantic import BaseModel, Field
+
+
+class GenerateTeamCreationInviteCodeResponse(BaseModel):
+    """Response model for team creation invite code generation endpoint.
+
+    Attributes:
+        code: The generated team creation invite code
+        description: Optional description for the code
+        message: Success or status message from the operation
+    """
+
+    code: str = Field(description="The generated team creation invite code")
+    description: str | None = Field(None, description="Optional description for the code")
+    message: str = Field(description="Success message confirming code generation")

todo/dto/responses/get_team_creation_invite_codes_response.py

@@ -0,0 +1,27 @@
+from pydantic import BaseModel, Field
+from typing import List, Optional
+from datetime import datetime
+
+
+class TeamCreationInviteCodeListItemDTO(BaseModel):
+    """DTO for a single team creation invite code in the list."""
+
+    id: str = Field(description="Unique identifier for the team creation invite code")
+    code: str = Field(description="The actual invite code")
+    description: Optional[str] = Field(None, description="Optional description provided when generating the code")
+    created_by: dict = Field(description="User details of who created this code")
+    created_at: datetime = Field(description="Timestamp when the code was created")
+    used_at: Optional[datetime] = Field(None, description="Timestamp when the code was used (null if unused)")
+    used_by: Optional[dict] = Field(None, description="User details of who used this code (null if unused)")
+    is_used: bool = Field(description="Whether this code has been used for team creation")
+
+
+class GetTeamCreationInviteCodesResponse(BaseModel):
+    """Response model for listing all team creation invite codes with pagination links."""
+
+    codes: List[TeamCreationInviteCodeListItemDTO] = Field(
+        description="List of team creation invite codes for current page"
+    )
+    previous_url: Optional[str] = Field(None, description="URL for previous page (null if no previous page)")
+    next_url: Optional[str] = Field(None, description="URL for next page (null if no next page)")
+    message: str = Field(description="Success message")

todo/dto/team_creation_invite_code_dto.py

@@ -0,0 +1,30 @@
+from pydantic import BaseModel, Field
+from typing import Optional
+from datetime import datetime
+
+
+class GenerateTeamCreationInviteCodeDTO(BaseModel):
+    """DTO for generating team creation invite codes.
+
+    Allows admins to create invite codes with an optional description for tracking purposes."""
+
+    description: Optional[str] = None
+
+
+class VerifyTeamCreationInviteCodeDTO(BaseModel):
+    """DTO for verifying team creation invite codes."""
+
+    code: str
+
+
+class TeamCreationInviteCodeDTO(BaseModel):
+    """DTO for team creation invite code data."""
+
+    id: str = Field(description="Unique identifier for the team invite code")
+    code: str = Field(description="The actual invite code")
+    description: Optional[str] = Field(None, description="Optional description provided when generating the code")
+    created_by: str = Field(description="User ID of the admin who generated this code")
+    created_at: datetime = Field(description="Timestamp when the code was created")
+    used_at: Optional[datetime] = Field(None, description="Timestamp when the code was used (null if unused)")
+    used_by: Optional[str] = Field(None, description="User ID who used this code (null if unused)")
+    is_used: bool = Field(description="Whether this code has been used for team creation")

todo/dto/team_dto.py

@@ -9,6 +9,7 @@ class CreateTeamDTO(BaseModel):
     description: Optional[str] = None
     member_ids: Optional[List[str]] = None
     poc_id: Optional[str] = None
+    team_invite_code: str
 
     @validator("member_ids")
     def validate_member_ids(cls, value):

todo/middlewares/jwt_auth.py

@@ -14,6 +14,7 @@
 )
 from todo.constants.messages import AuthErrorMessages, ApiErrors
 from todo.dto.responses.error_response import ApiErrorResponse, ApiErrorDetail
+from todo.repositories.user_repository import UserRepository
 
 
 class JWTAuthenticationMiddleware:
@@ -110,8 +111,14 @@ def _try_refresh(self, request) -> bool:
             return False
 
     def _set_user_data(self, request, payload):
-        """Set user data on request"""
-        request.user_id = payload["user_id"]
+        """Set user data on request with database verification"""
+        user_id = payload["user_id"]
+        user = UserRepository.get_by_id(user_id)
+        if not user:
+            raise TokenInvalidError(AuthErrorMessages.INVALID_TOKEN)
+
+        request.user_id = user_id
+        request.user_email = user.email_id
 
     def _process_response(self, request, response):
         """Process response and set new cookies if token was refreshed"""
@@ -156,6 +163,7 @@ def get_current_user_info(request) -> dict:
 
     user_info = {
         "user_id": request.user_id,
+        "email": request.user_email,
     }
 
     return user_info

todo/models/team_creation_invite_code.py

@@ -0,0 +1,31 @@
+from bson import ObjectId
+from pydantic import Field, validator
+from typing import ClassVar
+from datetime import datetime, timezone
+
+from todo.models.common.document import Document
+from todo.models.common.pyobjectid import PyObjectId
+
+
+class TeamCreationInviteCodeModel(Document):
+    """
+    Model for team creation invite codes.
+    """
+
+    collection_name: ClassVar[str] = "team_creation_invite_codes"
+
+    code: str = Field(description="The actual invite code")
+    description: str | None = None
+    created_by: PyObjectId
+    created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
+    used_at: datetime | None = None
+    used_by: PyObjectId | None = None
+    is_used: bool = False
+
+    @validator("created_by", "used_by")
+    def validate_object_id(cls, v):
+        if v is None:
+            return v
+        if not ObjectId.is_valid(v):
+            raise ValueError(f"Invalid ObjectId: {v}")
+        return ObjectId(v)

todo/repositories/team_creation_invite_code_repository.py

@@ -0,0 +1,90 @@
+from typing import Optional, List
+from datetime import datetime, timezone
+
+from todo.repositories.common.mongo_repository import MongoRepository
+from todo.models.team_creation_invite_code import TeamCreationInviteCodeModel
+from todo.repositories.user_repository import UserRepository
+
+
+class TeamCreationInviteCodeRepository(MongoRepository):
+    """Repository for team creation invite code operations."""
+
+    collection_name = TeamCreationInviteCodeModel.collection_name
+
+    @classmethod
+    def is_code_valid(cls, code: str) -> Optional[dict]:
+        """Check if a code is available for use (unused)."""
+        collection = cls.get_collection()
+        try:
+            code_data = collection.find_one({"code": code, "is_used": False})
+            return code_data
+        except Exception as e:
+            raise Exception(f"Error checking if code is valid: {e}")
+
+    @classmethod
+    def validate_and_consume_code(cls, code: str, used_by: str) -> Optional[dict]:
+        """Validate and consume a code in one atomic operation using findOneAndUpdate."""
+        collection = cls.get_collection()
+        try:
+            current_time = datetime.now(timezone.utc)
+            result = collection.find_one_and_update(
+                {"code": code, "is_used": False},
+                {"$set": {"is_used": True, "used_by": used_by, "used_at": current_time.isoformat()}},
+                return_document=True,
+            )
+            return result
+        except Exception as e:
+            raise Exception(f"Error validating and consuming code: {e}")
+
+    @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
+        return team_invite_code
+
+    @classmethod
+    def get_all_codes(cls, page: int = 1, limit: int = 10) -> tuple[List[dict], int]:
+        """Get paginated team creation invite codes with user details for created_by and used_by."""
+        collection = cls.get_collection()
+        try:
+            skip = (page - 1) * limit
+
+            total_count = collection.count_documents({})
+
+            codes = list(collection.find().sort("created_at", -1).skip(skip).limit(limit))
+
+            enhanced_codes = []
+            for code in codes:
+                created_by_user = None
+                used_by_user = None
+
+                if code.get("created_by"):
+                    user = UserRepository.get_by_id(str(code["created_by"]))
+                    if user:
+                        created_by_user = {"id": str(user.id), "name": user.name}
+
+                if code.get("used_by"):
+                    user = UserRepository.get_by_id(str(code["used_by"]))
+                    if user:
+                        used_by_user = {"id": str(user.id), "name": user.name}
+
+                enhanced_code = {
+                    "id": str(code["_id"]),
+                    "code": code["code"],
+                    "description": code.get("description"),
+                    "created_at": code.get("created_at"),
+                    "used_at": code.get("used_at"),
+                    "is_used": code.get("is_used", False),
+                    "created_by": created_by_user or {},
+                    "used_by": used_by_user,
+                }
+                enhanced_codes.append(enhanced_code)
+
+            return enhanced_codes, total_count
+        except Exception as e:
+            raise Exception(f"Error getting all codes with user details: {e}")

todo/serializers/create_team_serializer.py

@@ -13,6 +13,7 @@ class CreateTeamSerializer(serializers.Serializer):
     description = serializers.CharField(max_length=500, required=False, allow_blank=True)
     member_ids = serializers.ListField(child=serializers.CharField(), required=False, default=list)
     poc_id = serializers.CharField(required=False, allow_null=True, allow_blank=True)
+    team_invite_code = serializers.CharField(max_length=20, min_length=6)
 
     def validate_poc_id(self, value):
         if not value or not value.strip():

todo/serializers/team_creation_invite_code_serializer.py

@@ -0,0 +1,18 @@
+from rest_framework import serializers
+
+
+class GenerateTeamCreationInviteCodeSerializer(serializers.Serializer):
+    """Serializer for generating team creation invite codes."""
+
+    description = serializers.CharField(
+        max_length=500,
+        required=False,
+        allow_blank=True,
+        help_text="Optional description for the team creation invite code (e.g., 'Code for marketing team')",
+    )
+
+
+class VerifyTeamCreationInviteCodeSerializer(serializers.Serializer):
+    """Serializer for verifying team creation invite codes."""
+
+    code = serializers.CharField(max_length=100)