class Conversation(ChatRecord):
    record_type = 'conversation'

    def mark_non_distinct(self):
        database = self._get_database()
        database.save([{'_id': Database._encode_id(self.id),
                        'distinct_by_participants': True}])

    @classmethod
    def equal_record(cls, record1, record2):
        keys = ['distinct_by_participants', 'title', 'meta']
        for key in keys:
            if ((key in record1) != (key in record2)) or \
               (key in record1 and record1[key] != record2[key]):
                return False
        return True

    @classmethod
    def new(cls, conversation_id, user_id):
        return Conversation(RecordID(cls.record_type, conversation_id),
                            user_id,
                            [RoleAccessControlEntry(
                             cls.get_admin_role(conversation_id),
                             ACCESS_CONTROL_ENTRY_LEVEL_WRITE),
                             RoleAccessControlEntry(
                             cls.get_participant_role(conversation_id),
                             ACCESS_CONTROL_ENTRY_LEVEL_READ)])

    @classmethod
    def get_participant_role(cls, conversation_id):
        return "%s-participant-%s" % (cls.record_type, conversation_id)

    @classmethod
    def get_admin_role(cls, conversation_id):
        return "%s-admin-%s" % (cls.record_type, conversation_id)

    def get_user_conversation_acl(self):
        return [RoleAccessControlEntry(
                self.get_admin_role(self.id.key),
                ACCESS_CONTROL_ENTRY_LEVEL_WRITE),
                RoleAccessControlEntry(
                self.get_participant_role(self.id.key),
                ACCESS_CONTROL_ENTRY_LEVEL_WRITE)]

    @classmethod
    def __uc_to_conversation(cls, uc):
        c = uc['_transient']['conversation']
        c['unread_count'] = uc['unread_count']
        c['last_message_ref'] = c.get('last_message', None)
        c['last_read_message_ref'] = uc.get('last_read_message', None)
        c['last_message'] = None
        c['last_read_message'] = None
        return Conversation.from_record(c)

    @classmethod
    def fetch_all_with_paging(cls, page, page_size, order='desc'):
        ucs = UserConversation.fetch_all_with_paging(page, page_size, order)
        result = [cls.__uc_to_conversation(uc)
                  for uc in ucs]
        result = [c for c in result if c is not None]
        participants, admins = cls.__get_participants_and_admins(result)
        for row in result:
            key = row.id.key
            row['admin_ids'] = admins[key]
            row['participant_ids'] = participants[key]
        return result

    @classmethod
    def fetch_one(cls, conversation_id, with_uc=True):
        result = None
        if with_uc:
            uc = UserConversation.fetch_one(conversation_id)
            if uc:
                result = cls.__uc_to_conversation(uc)
        else:
            if result is None:
                result = super(Conversation, cls).fetch_one(conversation_id)

        if result is None:
            msg = "Conversation not found,conversation_id=%s" %\
                  (conversation_id)
            raise SkygearChatException(msg)

        participants, admins = cls.__get_participants_and_admins([result])
        key = result.id.key
        result['admin_ids'] = admins[key]
        result['participant_ids'] = participants[key]
        return result

    @classmethod
    def __get_participants_and_admins(cls, conversations):
        database = cls._get_database()
        conversation_ids = [c.id.key for c in conversations]

        admins = {}
        participants = {}
        for key in conversation_ids:
            admins[key] = []
            participants[key] = []
        predicate = Predicate(conversation__in=list(set(conversation_ids)))
        query_result = database.query(
                       Query(UserConversation.record_type, predicate=predicate,
                             limit=None)
                       )
        for row in query_result:
            conversation_id = row['conversation'].recordID.key
            if row['is_admin']:
                admins[conversation_id].append(row['user'].recordID.key)
            participants[conversation_id].append(row['user'].recordID.key)

        return participants, admins

    @classmethod
    def get_message_acl(cls, conversation_id):
        return [RoleAccessControlEntry(
                cls.get_participant_role(conversation_id),
                ACCESS_CONTROL_ENTRY_LEVEL_WRITE)]

    @classmethod
    def exists(cls, conversation_id):
        conversation = Conversation.fetch_one(conversation_id)
        return not (conversation is None) class ConversationAlreadyExistsException(SkygearException):
    def __init__(self, conversation_id):
        super().__init__(
            "Conversation with the participants already exists",
            InvalidArgument,
            {'conversation_id': conversation_id}
        ) class InvalidArgumentException(SkygearChatException):
    def __init__(self, message=None, arguments=None):
        message = message or "This operation is not supported."
        arguments = arguments if isinstance(arguments, list) else []
        super().__init__(
            message,
            InvalidArgument,
            {
                'arguments': arguments,
            }
        ) class NotAdminConversationException(SkygearChatException):
    def __init__(self):
        super().__init__(
            "user is not an admin, permission denied",
            PermissionDenied
        ) class NotInConversationException(SkygearChatException):
    def __init__(self):
        super().__init__(
            "user is not in the conversation, permission denied",
            PermissionDenied
        ) class NotSupportedException(SkygearChatException):
    def __init__(self, message=None):
        message = message or "This operation is not supported."
        super().__init__(
            message,
            NotSupported
        ) class SkygearChatException(SkygearException):
    pass def send_after_conversation_created_hook(conversation, participant_ids):
    participants = __get_users_by_user_ids(participant_ids)
    data = {'conversation': conversation, 'participants': participants}
    __send_hook(AFTER_CONVERSATION_CREATED_HOOK, data) def send_after_conversation_deleted_hook(conversation, participant_ids):
    participants = __get_users_by_user_ids(participant_ids)
    data = {'conversation': conversation, 'participants': participants}
    __send_hook(AFTER_CONVERSATION_DELETED_HOOK, data) def send_after_conversation_updated_hook(conversation, participant_ids):
    participants = __get_users_by_user_ids(participant_ids)
    data = {'conversation': conversation, 'participants': participants}
    __send_hook(AFTER_CONVERSATION_UPDATED_HOOK, data) def send_after_users_added_to_conversation_hook(conversation,
                                                participant_ids,
                                                new_user_ids):

    participants = __get_users_by_user_ids(participant_ids)
    new_users = __get_users_by_user_ids(new_user_ids)
    data = {'conversation': conversation,
            'participants': participants,
            'new_users': new_users}
    __send_hook(AFTER_USERS_ADDED_TO_CONVERSATION_HOOK, data) def send_after_users_removed_from_conversation_hook(conversation,
                                                    participant_ids,
                                                    old_user_ids):
    participants = __get_users_by_user_ids(participant_ids)
    old_users = __get_users_by_user_ids(old_user_ids)
    data = {'conversation': conversation,
            'participants': participants,
            'old_users': old_users}
    __send_hook(AFTER_USERS_REMOVED_FROM_CONVERSATION_HOOK, data) class Message(ChatRecord):
    record_type = 'message'

    def delete(self) -> None:
        """
        Soft-delete a message
        - Mark message as deleted
        - Update last_message and last_read_message
        """

        if self['deleted']:
            raise AlreadyDeletedException()

        self['deleted'] = True
        self.save()

    def getReceiptList(self):
        """
        Returns a list of message receipt statuses.
        """
        receipts = list()
        with db.conn() as conn:
            cur = conn.execute('''
                SELECT receipt.user, read_at, delivered_at
                FROM %(schema_name)s.receipt
                WHERE
                    "message" = %(message_id)s AND
                    (read_at IS NOT NULL or delivered_at is NOT NULL)
                ''', {
                    'schema_name': AsIs(_get_schema_name()),
                    'message_id': self.id.key
                }
            )

            for row in cur:
                receipts.append({
                    'user': row['user'],
                    'read_at': to_rfc3339_or_none(row['read_at']),
                    'delivered_at': to_rfc3339_or_none(row['delivered_at'])
                })
        return receipts

    def updateMessageStatus(self, conn) -> bool:
        """
        Update the message status field by querying the database for
        all receipt statuses.
        """
        cur = conn.execute('''
            WITH
              read_count AS (
                SELECT DISTINCT COUNT(receipt.user) as count
                FROM %(schema_name)s.receipt
                WHERE message = %(message_id)s
                    AND read_at IS NOT NULL
              ),
              participant_count AS (
                SELECT count(1) as count
                FROM %(schema_name)s.user_conversation
                WHERE conversation = %(conversation_id)s
              )
            UPDATE %(schema_name)s.message
            SET _updated_at = NOW(),
                message_status =
                  CASE
                    WHEN read_count.count = 0 THEN 'delivered'
                    WHEN read_count.count < participant_count.count
                        THEN 'some_read'
                    ELSE 'all_read'
                  END
            FROM read_count, participant_count
            WHERE _id = %(message_id)s
            RETURNING _updated_at, message_status
            ''', {
                'schema_name': AsIs(_get_schema_name()),
                'message_id': self.id.key,
                'conversation_id': self.conversation_id
            }
        )

        row = cur.fetchone()
        if row is not None:
            self['_updated_at'] = row[0]
            self['message_status'] = row[1]

    def notifyParticipants(self, event_type='update') -> None:
        result = UserConversation.\
            fetch_all_by_conversation_id(self.conversation_id)
        participants = [row['user'].recordID.key for row in result]
        _publish_record_event(participants,
                              "message",
                              event_type,
                              self)

    @property
    def conversation_id(self):
        return self['conversation'].recordID.key

    @classmethod
    def fetch_all_by_conversation_id(cls, conversation_id, limit,
                                     before_time=None, before_message_id=None,
                                     after_time=None, after_message_id=None,
                                     order=None, deleted=False):
        database = cls._get_database()
        predicate = Predicate(conversation__eq=conversation_id,
                              deleted__eq=deleted)
        if before_time is not None:
            predicate = predicate & Predicate(_created_at__lt=before_time)
        if before_message_id is not None:
            before_message = Message.fetch_one(before_message_id)
            predicate = predicate & Predicate(seq__lt=before_message['seq'])
        if after_time is not None:
            predicate = predicate & Predicate(_created_at__gt=after_time)
        if after_message_id is not None:
            after_message = Message.fetch_one(after_message_id)
            predicate = predicate & Predicate(seq__gt=after_message['seq'])
        query = Query('message', predicate=predicate, limit=limit)
        if order != 'edited_at':
            order = '_created_at'
        query.add_order(order, 'desc')
        result = database.query(query)

        # remove deleted message content
        if deleted:
            for message in result:
                if message['deleted']:
                    Message.clear_message_content(message)
        return result

    @classmethod
    def fetch_all_by_conversation_id_and_seq(cls,
                                             conversation_id,
                                             from_seq,
                                             to_seq):
        database = cls._get_database()
        predicate = Predicate(seq__lte=to_seq,
                              conversation__eq=conversation_id,
                              deleted__eq=False)
        if from_seq >= 0:
            predicate = predicate & Predicate(seq__gte=from_seq)

        query = Query('message', predicate=predicate, limit=None)
        return database.query(query)

    @classmethod
    def clear_message_content(cls, message):
        keys_to_delete = [
            'message_status',
            'body',
            'meta_data',
            'attachment',
            'revision'
        ]

        for key in keys_to_delete:
            message[key] = None
            del message[key] def _publish_record_event(user_ids: [],
                          record_type: str,
                          event: str,
                          record: Record) -> None:
    if len(user_ids) == 0:
        return
    _publish_event(user_ids, event, {
        'event_type': event,
        'type': 'record',
        'record_type': record_type,
        'record': serialize_record(record)
    }) class RolesHelper(object):
    def __init__(self, container):
        self.container = container

    def assign(self, users, roles):
        return self.container.send_action('role:assign', {
            'roles': roles,
            'users': users
        })

    def revoke(self, users, roles):
        return self.container.send_action('role:revoke', {
            'roles': roles,
            'users': users
        })

    def set_roles(self, users, roles, flag):
        result = None
        if flag:
            result = self.assign(users, roles)
        else:
            result = self.revoke(users, roles)
        return result class UserConversation(ChatRecord):
    record_type = 'user_conversation'

    @classmethod
    def new(cls, conversation, user_id):
        record = cls(None,
                     user_id,
                     conversation.get_user_conversation_acl())
        record['user'] = Reference(RecordID('user', user_id))
        record['conversation'] = Reference(conversation.id)
        record['unread_count'] = 0
        record['is_admin'] = False
        return record

    def get_hash(self):
        return UserConversation.\
               get_consistent_hash(self['conversation'].recordID.key,
                                   self['user'].recordID.key)

    def mark_admin(self, flag):
        database = self._get_database()
        record = Record(self.id, self.owner_id, self.acl)
        record['is_admin'] = flag
        record['user'] = self['user']
        record['conversation'] = self['conversation']
        database.save([record])

    @classmethod
    def get_consistent_hash(cls, conversation_id, user_id):
        seed = conversation_id + user_id
        sha = hashlib.sha256(bytes(seed, 'utf8'))
        return str(uuid.UUID(bytes=sha.digest()[0:16]))

    @property
    def id(self):
        return RecordID(self.record_type, self.get_hash())

    @classmethod
    def exists(cls, record, check_is_admin=False):
        record = cls.fetch_one(record.id.key)
        return (record is not None) and\
               (not check_is_admin or record['is_admin'])

    @classmethod
    def fetch_all_with_paging(cls, page, page_size, order='desc'):
        database = cls._get_database()
        offset = (page - 1) * page_size
        query_result = database.query(
                       Query(cls.record_type,
                             predicate=Predicate(user__eq=current_user_id()),
                             offset=offset,
                             limit=page_size,
                             include=["conversation", "user"])
                       .add_order('_updated_at', order))
        return [uc for uc in query_result]

    @classmethod
    def fetch_all_by_conversation_id(cls, conversation_id):
        database = cls._get_database()
        predicate = Predicate(conversation__eq=conversation_id)
        records = database.query(Query(cls.record_type,
                                       predicate=predicate,
                                       include=["conversation", "user"],
                                       limit=None))
        return [UserConversation.from_record(record) for record in records]

    @classmethod
    def fetch_one(cls,
                  conversation_id,
                  user_id=None):
        database = cls._get_database()
        if user_id is None:
            user_id = current_user_id()
        predicate = Predicate(user__eq=user_id,
                              conversation__eq=conversation_id)
        query_result = database.query(
                       Query(cls.record_type,
                             predicate=predicate,
                             limit=1,
                             include=["conversation", "user"]))
        return UserConversation.from_record(query_result[0])\
            if len(query_result) == 1 else None def _get_container():
    return SkygearContainer(api_key=skyoptions.masterkey,
                            user_id=current_user_id()) def _get_schema_name():
    return "app_%s" % skyoptions.appname def current_context_has_master_key():
    # FIXME: skygear-server does not pass access_key_type information
    # yet. This is a temporary workaround before skygear-server has support.
    return 'access_key_type' not in current_context() or \
        current_context().get('access_key_type', '') == 'master'
from uuid import uuid4
from psycopg2.extensions import AsIs
from skygear.encoding import serialize_record
from skygear.models import Record
from skygear.utils import db
from skygear.utils.context import current_user_id
from .conversation import Conversation
from .exc import (ConversationAlreadyExistsException, InvalidArgumentException,
                  NotAdminConversationException, NotInConversationException,
                  NotSupportedException, SkygearChatException)
from .hooks import (send_after_conversation_created_hook,
                    send_after_conversation_deleted_hook,
                    send_after_conversation_updated_hook,
                    send_after_users_added_to_conversation_hook,
                    send_after_users_removed_from_conversation_hook)
from .message import Message
from .pubsub import _publish_record_event
from .roles import RolesHelper
from .user_conversation import UserConversation
from .utils import (_get_container, _get_schema_name,
                    current_context_has_master_key)
import skygear





def __validate_user_is_admin(conversation_id):
    uc = UserConversation.fetch_one(conversation_id)
    if uc is None:
        raise NotInConversationException()
    if not uc['is_admin']:
        raise NotAdminConversationException()


def __validate_conversation(participants):
    first_row = None
    with db.conn() as conn:
        result = conn.execute("""
                              SELECT c._id
                                   FROM %(schema_name)s.conversation AS c
                              WHERE c._id IN (
                                  SELECT uc.conversation
                                  FROM %(schema_name)s.user_conversation AS uc
                                  WHERE uc.conversation IN (
                                      SELECT uc.conversation
                                      FROM %(schema_name)s.user_conversation
                                      AS uc
                                      WHERE uc.user IN %(user_ids)s
                                      GROUP BY uc.conversation
                                      HAVING
                                          COUNT(DISTINCT uc.user) = %(count)s)
                                  GROUP BY uc.conversation
                                  HAVING COUNT(DISTINCT uc.user) = %(count)s)
                              ORDER BY c._created_at DESC
                              """, {'schema_name': AsIs(_get_schema_name()),
                                    'user_ids': tuple(participants),
                                    'count': len(participants)})
        first_row = result.first()
    valid = first_row is None
    if not valid:
        raise ConversationAlreadyExistsException(first_row[0])


def __update_roles(container, users, role, flag):
    roles_helper = RolesHelper(container)
    roles_helper.set_roles(users, [role], flag)


def __update_participant_roles(container, conversation_id, users, flag):
    __update_roles(container,
                   users,
                   Conversation.get_participant_role(conversation_id),
                   flag)


def __update_admin_roles(container, conversation_id, users, flag):
    __update_roles(container,
                   users,
                   Conversation.get_admin_role(conversation_id),
                   flag)


def __update_admin_flags(container, conversation_id, user_ids, flag):
    for user_id in user_ids:
        uc = UserConversation.fetch_one(conversation_id, user_id=user_id)
        if uc is None:
            c = Conversation.new(conversation_id, user_id)
            uc = UserConversation.new(c, user_id)
        uc.mark_admin(flag)


def notify_users(record,
                 unchanged_participants,
                 old_participants,
                 new_participants):

    _publish_record_event(unchanged_participants,
                          "conversation",
                          "update",
                          record)

    _publish_record_event(old_participants,
                          "conversation",
                          "delete",
                          record)

    _publish_record_event(new_participants,
                          "conversation",
                          "create",
                          record)


def handle_conversation_before_delete(record, conn):
    raise NotSupportedException("Call chat:delete_conversation instead.")


def handle_leave_conversation(conversation_id):
    uc = UserConversation.fetch_one(conversation_id)
    if uc is None:
        raise NotInConversationException()
    uc.delete()
    return {'status': 'OK'}


def handle_add_participants(conversation_id,
                            participant_ids,
                            is_first_time=False):

    existing_participants = []
    for participant in participant_ids:

        uc = UserConversation.fetch_one(conversation_id,