models.py

import re
from typing import Any, Dict, List, Optional, Union

from django.db import models
from django.db.models import Q
from django.template.defaultfilters import slugify
from django.utils.translation import gettext_lazy as _

from aleksis.apps.matrix.matrix import do_matrix_request
from aleksis.core.mixins import ExtensibleModel, ExtensiblePolymorphicModel
from aleksis.core.models import Group, Person
from aleksis.core.util.core_helpers import get_site_preferences


class MatrixProfile(ExtensibleModel):
    """Model for a Matrix profile."""

    matrix_id = models.CharField(max_length=255, verbose_name=_("Matrix ID"), unique=True)
    person = models.OneToOneField(
        Person,
        on_delete=models.CASCADE,
        verbose_name=_("Person"),
        null=True,
        blank=True,
        related_name="matrix_profile",
    )

    @classmethod
    def build_matrix_id(cls, username, homeserver: Optional[str] = None):
        homeserver = homeserver or get_site_preferences()["matrix__homeserver_ids"]
        return f"@{username}:{homeserver}"

    @classmethod
    def from_person(cls, person: Person, commit: bool = False) -> Union["MatrixProfile", None]:
        if hasattr(person, "matrix_profile"):
            return person.matrix_profile
        if not person.user:
            raise ValueError("Person must have a user.")
        if not get_site_preferences()["matrix__homeserver_ids"]:
            return None
        new_profile = MatrixProfile(
            matrix_id=cls.build_matrix_id(person.user.username), person=person
        )
        if commit:
            new_profile.save()
        return new_profile

    class Meta:
        verbose_name = _("Matrix profile")
        verbose_name_plural = _("Matrix profiles")


class MatrixRoom(ExtensiblePolymorphicModel):
    """Model for a Matrix room."""

    room_id = models.CharField(max_length=255, verbose_name=_("Room ID"), unique=True)
    alias = models.CharField(max_length=255, verbose_name=_("Alias"), unique=True, blank=True)
    group = models.OneToOneField(
        Group,
        on_delete=models.CASCADE,
        verbose_name=_("Group"),
        related_name="matrix_room",
    )

    @classmethod
    def from_group(self, group: Group):
        """Create a Matrix room from a group."""
        from .matrix import MatrixException, do_matrix_request

        try:
            room = MatrixRoom.objects.get(group=group)
        except MatrixRoom.DoesNotExist:
            room = MatrixRoom(group=group)

        if room.room_id:
            # Existing room, check if still accessible
            r = do_matrix_request("GET", f"directory/list/room/{room.room_id}")
        else:
            # Room does not exist, create it
            alias = slugify(group.short_name or group.name)
            profiles_to_invite = list(
                self.get_profiles_for_group(group).values_list("matrix_id", flat=True)
            )

            alias_found = False
            while not alias_found:
                try:
                    r = self._create_room(group.name, alias, profiles_to_invite)
                    alias_found = True
                except MatrixException as e:
                    print(e.args, get_site_preferences()["matrix__disambiguate_room_aliases"])
                    if (
                        not get_site_preferences()["matrix__disambiguate_room_aliases"]
                        or e.args[0].get("errcode") != "M_ROOM_IN_USE"
                    ):
                        raise MatrixException(*e.args)

                match = re.match(r"^(.*)-(\d+)$", alias)
                if match:
                    # Counter found, increase
                    prefix = match.group(1)
                    counter = int(match.group(2)) + 1
                    alias = f"{prefix}-{counter}"
                else:
                    # Counter not found, add one
                    alias = f"{alias}-2"

            room.room_id = r["room_id"]
            room.alias = r["room_alias"]
            room.save()
        return room

    @classmethod
    def _create_room(self, name, alias, invite: List[str]):
        from .matrix import do_matrix_request

        body = {"preset": "private_chat", "name": name, "room_alias_name": alias, "invite": invite}
        r = do_matrix_request("POST", "createRoom", body=body)

        return r

    @property
    def power_levels(self) -> Dict[str, int]:
        """Return the power levels for this room."""
        from aleksis.apps.matrix.matrix import do_matrix_request

        r = do_matrix_request("GET", f"rooms/{self.room_id}/state")

        event = list(filter(lambda x: x["type"] == "m.room.power_levels", r))
        user_levels = event[0]["content"]["users"]

        return user_levels

    @property
    def members(self) -> List[str]:
        from aleksis.apps.matrix.matrix import do_matrix_request

        r = do_matrix_request(
            "GET", f"rooms/{self.room_id}/members", body={"membership": ["join", "invite"]}
        )
        return [m["state_key"] for m in r["chunk"]]

    def _invite(self, profile: MatrixProfile) -> Dict[str, Any]:
        """Invite a user to this room."""
        from aleksis.apps.matrix.matrix import do_matrix_request

        r = do_matrix_request(
            "POST",
            f"rooms/{self.room_id}/invite",
            body={"user_id": profile.matrix_id},
        )
        return r

    def _set_power_levels(self, power_levels: Dict[str, int]) -> Dict[str, Any]:
        """Set the power levels for this room."""
        r = do_matrix_request(
            "PUT",
            f"rooms/{self.room_id}/state/m.room.power_levels/",
            body={"users": power_levels},
        )
        return r

    @classmethod
    def get_profiles_for_group(cls, group: Group):
        """Get all profile objects for the members/owners of a group."""
        existing_profiles = MatrixProfile.objects.filter(
            Q(person__member_of=group) | Q(person__owner_of=group)
        )
        profiles_to_create = []
        for person in (
            Person.objects.filter(user__isnull=False)
            .filter(Q(member_of=group) | Q(owner_of=group))
            .exclude(matrix_profile__in=existing_profiles)
            .distinct()
        ):
            new_profile = MatrixProfile.from_person(person)
            if new_profile:
                profiles_to_create.append(new_profile)
        MatrixProfile.objects.bulk_create(profiles_to_create)

        all_profiles = MatrixProfile.objects.filter(
            Q(person__in=group.members.all()) | Q(person__in=group.owners.all())
        )

        return all_profiles

    def get_profiles(self):
        """Get all profile objects for the members/owners of this group."""
        return self.get_profiles_for_group(self.group)

    def sync_profiles(self):
        """Sync profiles for this room."""
        all_profiles = self.get_profiles()
        members = self.members

        # Invite all users who are not in the room yet
        for profile in all_profiles:
            if profile.matrix_id not in members:
                # Now invite
                self._invite(profile)

        # Set power levels for all users
        # Mod = 50 = Owners
        # User = 0 = Members
        user_levels = self.power_levels
        for profile in all_profiles:
            if profile.person in self.group.owners.all():
                user_levels[profile.matrix_id] = 50
            elif profile.person in self.group.members.all():
                user_levels[profile.matrix_id] = 0
        self._set_power_levels(user_levels)

    def sync(self):
        """Sync this room."""
        self.sync_profiles()

    class Meta:
        verbose_name = _("Matrix room")
        verbose_name_plural = _("Matrix rooms")
        permissions = (("use_group_in_matrix", "Can use group in Matrix"),)


class MatrixSpace(MatrixRoom):
    children = models.ManyToManyField(
        to=MatrixRoom, verbose_name=_("Child rooms/spaces"), blank=True, related_name="parents"
    )

    class Meta:
        verbose_name = _("Matrix space")
        verbose_name_plural = _("Matrix spaces")