from __future__ import annotations from django.db.models import Count, Prefetch, QuerySet from rest_framework import status, viewsets from rest_framework.decorators import action from rest_framework.permissions import IsAuthenticated from rest_framework.request import Request from rest_framework.response import Response from apps.groups.models import Group, GroupInvite, GroupMember, GroupRole, JoinRequest, JoinRequestStatus from apps.groups.permissions import IsGroupAdmin, IsGroupMember from apps.groups.serializers import ( GroupCreateSerializer, GroupDetailSerializer, GroupInviteCreateSerializer, GroupInviteSerializer, GroupListSerializer, GroupMemberSerializer, GroupUpdateSerializer, JoinRequestSerializer, JoinViaInviteSerializer, RoleUpdateSerializer, ) class GroupViewSet(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] def get_serializer_class(self): if self.action == "create": return GroupCreateSerializer if self.action in ("update", "partial_update"): return GroupUpdateSerializer if self.action == "retrieve": return GroupDetailSerializer return GroupListSerializer def get_queryset(self) -> QuerySet[Group]: user = self.request.user qs = Group.objects.filter(memberships__user=user).distinct() qs = qs.annotate(_member_count=Count("memberships")) if self.action in ("list", "retrieve"): qs = qs.prefetch_related( Prefetch( "memberships", queryset=GroupMember.objects.select_related("user").order_by("joined_at"), ) ) return qs def get_object(self) -> Group: obj = super().get_object() # Cache the requesting user's membership for serializers try: obj._user_membership = obj.memberships.get(user=self.request.user) except GroupMember.DoesNotExist: obj._user_membership = None return obj def perform_create(self, serializer: GroupCreateSerializer) -> Group: return serializer.save() def perform_destroy(self, instance: Group) -> None: # Only admin can delete/dissolve the group if not GroupMember.objects.filter( group=instance, user=self.request.user, role=GroupRole.ADMIN ).exists(): from rest_framework.exceptions import PermissionDenied raise PermissionDenied("Only group admins can delete the group.") instance.delete() # ---- Members ---- @action(detail=True, methods=["get"], permission_classes=[IsAuthenticated, IsGroupMember]) def members(self, request: Request, pk: str | None = None) -> Response: """List all members of the group.""" group = self.get_object() memberships = group.memberships.select_related("user").order_by("joined_at") serializer = GroupMemberSerializer(memberships, many=True) return Response(serializer.data) @action( detail=True, methods=["delete"], url_path="members/(?P[^/.]+)", permission_classes=[IsAuthenticated, IsGroupAdmin], ) def remove_member(self, request: Request, pk: str | None = None, user_id: str | None = None) -> Response: """Admin removes a member from the group.""" group = self.get_object() try: membership = GroupMember.objects.get(group=group, user_id=user_id) except GroupMember.DoesNotExist: return Response({"error": "Member not found."}, status=status.HTTP_404_NOT_FOUND) if membership.user == request.user: return Response( {"error": "Admins cannot remove themselves. Use leave instead, or transfer admin first."}, status=status.HTTP_400_BAD_REQUEST, ) membership.delete() return Response(status=status.HTTP_204_NO_CONTENT) @action( detail=True, methods=["patch"], url_path="members/(?P[^/.]+)/role", permission_classes=[IsAuthenticated, IsGroupAdmin], ) def update_member_role(self, request: Request, pk: str | None = None, user_id: str | None = None) -> Response: """Admin transfers admin role or changes member role.""" group = self.get_object() serializer = RoleUpdateSerializer(data=request.data, context={"group": group}) serializer.is_valid(raise_exception=True) try: membership = GroupMember.objects.get(group=group, user_id=user_id) except GroupMember.DoesNotExist: return Response({"error": "Member not found."}, status=status.HTTP_404_NOT_FOUND) membership.role = serializer.validated_data["role"] membership.save(update_fields=["role"]) if serializer.validated_data["role"] == GroupRole.ADMIN and membership.user != request.user: # Downgrade the current admin to member GroupMember.objects.filter(group=group, user=request.user).update(role=GroupRole.MEMBER) return Response(GroupMemberSerializer(membership).data) @action(detail=True, methods=["post"], permission_classes=[IsAuthenticated, IsGroupMember]) def leave(self, request: Request, pk: str | None = None) -> Response: """Member leaves the group. If admin is last admin, dissolve the group.""" group = self.get_object() membership = GroupMember.objects.filter(group=group, user=request.user).first() if not membership: return Response({"error": "You are not a member of this group."}, status=status.HTTP_400_BAD_REQUEST) if membership.role == GroupRole.ADMIN: admin_count = GroupMember.objects.filter(group=group, role=GroupRole.ADMIN).count() if admin_count <= 1: # Last admin leaving — dissolve the group group.delete() return Response({"detail": "You were the last admin. The group has been dissolved."}) membership.delete() return Response({"detail": "You have left the group."}) # ---- Invites ---- @action(detail=True, methods=["get", "post"], permission_classes=[IsAuthenticated, IsGroupAdmin]) def invites(self, request: Request, pk: str | None = None) -> Response: """List or create invites for the group.""" group = self.get_object() if request.method == "GET": invites_qs = group.invites.select_related("created_by").order_by("-created_at") serializer = GroupInviteSerializer(invites_qs, many=True, context={"request": request}) return Response(serializer.data) serializer = GroupInviteCreateSerializer( data=request.data, context={"group": group, "request": request}, ) serializer.is_valid(raise_exception=True) invite = serializer.save() return Response( GroupInviteSerializer(invite, context={"request": request}).data, status=status.HTTP_201_CREATED, ) @action( detail=True, methods=["delete"], url_path="invites/(?P[^/.]+)", permission_classes=[IsAuthenticated, IsGroupAdmin], ) def revoke_invite(self, request: Request, pk: str | None = None, invite_id: str | None = None) -> Response: """Revoke an invite by deactivating it.""" group = self.get_object() try: invite = GroupInvite.objects.get(id=invite_id, group=group) except GroupInvite.DoesNotExist: return Response({"error": "Invite not found."}, status=status.HTTP_404_NOT_FOUND) invite.is_active = False invite.save(update_fields=["is_active"]) return Response(status=status.HTTP_204_NO_CONTENT) # ---- Join Requests ---- @action(detail=True, methods=["get"], permission_classes=[IsAuthenticated, IsGroupAdmin]) def requests(self, request: Request, pk: str | None = None) -> Response: """List pending join requests for the group (admin only).""" group = self.get_object() join_requests = group.join_requests.select_related("user", "invite").order_by("-created_at") serializer = JoinRequestSerializer(join_requests, many=True) return Response(serializer.data) @action( detail=True, methods=["post"], url_path="requests/(?P[^/.]+)/approve", permission_classes=[IsAuthenticated, IsGroupAdmin], ) def approve_request(self, request: Request, pk: str | None = None, request_id: str | None = None) -> Response: """Approve a pending join request.""" group = self.get_object() try: join_request = JoinRequest.objects.get(id=request_id, group=group, status=JoinRequestStatus.PENDING) except JoinRequest.DoesNotExist: return Response({"error": "Pending join request not found."}, status=status.HTTP_404_NOT_FOUND) join_request.status = JoinRequestStatus.APPROVED join_request.save(update_fields=["status"]) GroupMember.objects.get_or_create( group=group, user=join_request.user, defaults={"role": GroupRole.MEMBER}, ) # Increment invite use count if join_request.invite: join_request.invite.use_count += 1 join_request.invite.save(update_fields=["use_count"]) return Response(JoinRequestSerializer(join_request).data) @action( detail=True, methods=["post"], url_path="requests/(?P[^/.]+)/reject", permission_classes=[IsAuthenticated, IsGroupAdmin], ) def reject_request(self, request: Request, pk: str | None = None, request_id: str | None = None) -> Response: """Reject a pending join request.""" group = self.get_object() try: join_request = JoinRequest.objects.get(id=request_id, group=group, status=JoinRequestStatus.PENDING) except JoinRequest.DoesNotExist: return Response({"error": "Pending join request not found."}, status=status.HTTP_404_NOT_FOUND) join_request.status = JoinRequestStatus.REJECTED join_request.save(update_fields=["status"]) return Response(JoinRequestSerializer(join_request).data) class JoinGroupViewSet(viewsets.GenericViewSet): """Public(ish) endpoint for joining a group via an invite code.""" permission_classes = [IsAuthenticated] @action(detail=False, methods=["get"], url_path="(?P[^/.]+)") def validate_invite(self, request: Request, code: str | None = None) -> Response: """Check if an invite code is valid and show group info.""" try: invite = GroupInvite.objects.select_related("group", "group__created_by").get(code=code) except GroupInvite.DoesNotExist: return Response({"error": "Invalid invite code."}, status=status.HTTP_404_NOT_FOUND) if not invite.is_active: return Response({"error": "This invite is no longer active."}, status=status.HTTP_410_GONE) if invite.max_uses > 0 and invite.use_count >= invite.max_uses: return Response({"error": "This invite has reached its maximum uses."}, status=status.HTTP_410_GONE) return Response({ "group": { "id": invite.group.id, "name": invite.group.name, "description": invite.group.description, "created_by_email": invite.group.created_by.email, "member_count": invite.group.memberships.count(), }, "invite": { "code": str(invite.code), "created_by_email": invite.created_by.email, }, }) @action(detail=False, methods=["post"], url_path="(?P[^/.]+)") def join(self, request: Request, code: str | None = None) -> Response: """Join a group via invite code.""" try: invite = GroupInvite.objects.select_related("group").get(code=code) except GroupInvite.DoesNotExist: return Response({"error": "Invalid invite code."}, status=status.HTTP_404_NOT_FOUND) if not invite.is_active: return Response({"error": "This invite is no longer active."}, status=status.HTTP_410_GONE) if invite.max_uses > 0 and invite.use_count >= invite.max_uses: return Response({"error": "This invite has reached its maximum uses."}, status=status.HTTP_410_GONE) group = invite.group # Check if already a member if GroupMember.objects.filter(group=group, user=request.user).exists(): return Response( {"detail": "You are already a member of this group.", "group_id": group.id}, status=status.HTTP_200_OK, ) # Check for existing pending request existing_request = JoinRequest.objects.filter( group=group, user=request.user, status=JoinRequestStatus.PENDING ).first() if existing_request: return Response( JoinRequestSerializer(existing_request).data, status=status.HTTP_200_OK, ) # Create join request or add directly (direct join for now — simple invite flow) member = GroupMember.objects.create(group=group, user=request.user, role=GroupRole.MEMBER) invite.use_count += 1 invite.save(update_fields=["use_count"]) # Also create a join request record for tracking JoinRequest.objects.create( group=group, user=request.user, invite=invite, status=JoinRequestStatus.APPROVED, ) serializer = GroupDetailSerializer(group, context={"request": request}) return Response(serializer.data, status=status.HTTP_201_CREATED)