app-MAIL-temp/app/api/serializer.py

from dataclasses import dataclass
from typing import Optional

from arrow import Arrow
from sqlalchemy import or_, func, case, and_
from sqlalchemy.orm import joinedload

from app.config import PAGE_LIMIT
from app.extensions import db
from app.models import Alias, Contact, EmailLog, Mailbox, AliasMailbox, CustomDomain


@dataclass
class AliasInfo:
    alias: Alias
    mailbox: Mailbox
    mailboxes: [Mailbox]

    nb_forward: int
    nb_blocked: int
    nb_reply: int

    latest_email_log: EmailLog = None
    latest_contact: Contact = None
    custom_domain: Optional[CustomDomain] = None

    def contain_mailbox(self, mailbox_id: int) -> bool:
        return mailbox_id in [m.id for m in self.mailboxes]


def serialize_alias_info(alias_info: AliasInfo) -> dict:
    return {
        # Alias field
        "id": alias_info.alias.id,
        "email": alias_info.alias.email,
        "creation_date": alias_info.alias.created_at.format(),
        "creation_timestamp": alias_info.alias.created_at.timestamp,
        "enabled": alias_info.alias.enabled,
        "note": alias_info.alias.note,
        # activity
        "nb_forward": alias_info.nb_forward,
        "nb_block": alias_info.nb_blocked,
        "nb_reply": alias_info.nb_reply,
    }


def serialize_alias_info_v2(alias_info: AliasInfo) -> dict:
    res = {
        # Alias field
        "id": alias_info.alias.id,
        "email": alias_info.alias.email,
        "creation_date": alias_info.alias.created_at.format(),
        "creation_timestamp": alias_info.alias.created_at.timestamp,
        "enabled": alias_info.alias.enabled,
        "note": alias_info.alias.note,
        "name": alias_info.alias.name,
        # activity
        "nb_forward": alias_info.nb_forward,
        "nb_block": alias_info.nb_blocked,
        "nb_reply": alias_info.nb_reply,
        # mailbox
        "mailbox": {"id": alias_info.mailbox.id, "email": alias_info.mailbox.email},
        "mailboxes": [
            {"id": mailbox.id, "email": mailbox.email}
            for mailbox in alias_info.mailboxes
        ],
        "support_pgp": alias_info.alias.mailbox_support_pgp(),
        "disable_pgp": alias_info.alias.disable_pgp,
        "latest_activity": None,
        "pinned": alias_info.alias.pinned,
    }
    if alias_info.latest_email_log:
        email_log = alias_info.latest_email_log
        contact = alias_info.latest_contact
        # latest activity
        res["latest_activity"] = {
            "timestamp": email_log.created_at.timestamp,
            "action": email_log.get_action(),
            "contact": {
                "email": contact.website_email,
                "name": contact.name,
                "reverse_alias": contact.website_send_to(),
            },
        }
    return res


def serialize_contact(contact: Contact, existed=False) -> dict:
    res = {
        "id": contact.id,
        "creation_date": contact.created_at.format(),
        "creation_timestamp": contact.created_at.timestamp,
        "last_email_sent_date": None,
        "last_email_sent_timestamp": None,
        "contact": contact.website_email,
        "reverse_alias": contact.website_send_to(),
        "reverse_alias_address": contact.reply_email,
        "existed": existed,
    }

    email_log: EmailLog = contact.last_reply()
    if email_log:
        res["last_email_sent_date"] = email_log.created_at.format()
        res["last_email_sent_timestamp"] = email_log.created_at.timestamp

    return res


def get_alias_infos_with_pagination(user, page_id=0, query=None) -> [AliasInfo]:
    ret = []
    q = (
        db.session.query(Alias)
        .options(joinedload(Alias.mailbox))
        .filter(Alias.user_id == user.id)
        .order_by(Alias.created_at.desc())
    )

    if query:
        q = q.filter(
            or_(Alias.email.ilike(f"%{query}%"), Alias.note.ilike(f"%{query}%"))
        )

    q = q.limit(PAGE_LIMIT).offset(page_id * PAGE_LIMIT)

    for alias in q:
        ret.append(get_alias_info(alias))

    return ret


def get_alias_infos_with_pagination_v3(
    user, page_id=0, query=None, sort=None, alias_filter=None, mailbox_id=None
) -> [AliasInfo]:
    # subquery on alias annotated with nb_reply, nb_blocked, nb_forward, max_created_at, latest_email_log_created_at
    alias_activity_subquery = (
        db.session.query(
            Alias.id,
            func.sum(case([(EmailLog.is_reply, 1)], else_=0)).label("nb_reply"),
            func.sum(
                case(
                    [(and_(EmailLog.is_reply.is_(False), EmailLog.blocked), 1)],
                    else_=0,
                )
            ).label("nb_blocked"),
            func.sum(
                case(
                    [
                        (
                            and_(
                                EmailLog.is_reply.is_(False),
                                EmailLog.blocked.is_(False),
                            ),
                            1,
                        )
                    ],
                    else_=0,
                )
            ).label("nb_forward"),
            func.max(EmailLog.created_at).label("latest_email_log_created_at"),
        )
        .join(EmailLog, Alias.id == EmailLog.alias_id, isouter=True)
        .filter(Alias.user_id == user.id)
        .group_by(Alias.id)
        .subquery()
    )

    alias_contact_subquery = (
        db.session.query(Alias.id, func.max(Contact.id).label("max_contact_id"))
        .join(Contact, Alias.id == Contact.alias_id, isouter=True)
        .filter(Alias.user_id == user.id)
        .group_by(Alias.id)
        .subquery()
    )

    latest_activity = case(
        [
            (Alias.created_at > EmailLog.created_at, Alias.created_at),
            (Alias.created_at < EmailLog.created_at, EmailLog.created_at),
        ],
        else_=Alias.created_at,
    )

    q = (
        db.session.query(
            Alias,
            Contact,
            EmailLog,
            CustomDomain,
            alias_activity_subquery.c.nb_reply,
            alias_activity_subquery.c.nb_blocked,
            alias_activity_subquery.c.nb_forward,
        )
        .options(joinedload(Alias.hibp_breaches))
        .join(Contact, Alias.id == Contact.alias_id, isouter=True)
        .join(CustomDomain, Alias.custom_domain_id == CustomDomain.id, isouter=True)
        .join(EmailLog, Contact.id == EmailLog.contact_id, isouter=True)
        .filter(Alias.id == alias_activity_subquery.c.id)
        .filter(Alias.id == alias_contact_subquery.c.id)
        .filter(
            or_(
                EmailLog.created_at
                == alias_activity_subquery.c.latest_email_log_created_at,
                and_(
                    # no email log yet for this alias
                    alias_activity_subquery.c.latest_email_log_created_at.is_(None),
                    # to make sure only 1 contact is returned in this case
                    or_(
                        Contact.id == alias_contact_subquery.c.max_contact_id,
                        alias_contact_subquery.c.max_contact_id.is_(None),
                    ),
                ),
            )
        )
    )

    if query:
        q = q.filter(
            or_(
                Alias.email.ilike(f"%{query}%"),
                Alias.note.ilike(f"%{query}%"),
                # can't use match() here as it uses to_tsquery that expected a tsquery input
                # Alias.ts_vector.match(query),
                Alias.ts_vector.op("@@")(func.plainto_tsquery("english", query)),
                Alias.name.ilike(f"%{query}%"),
            )
        )

    if mailbox_id:
        q = q.join(
            AliasMailbox, Alias.id == AliasMailbox.alias_id, isouter=True
        ).filter(
            or_(Alias.mailbox_id == mailbox_id, AliasMailbox.mailbox_id == mailbox_id)
        )

    if alias_filter == "enabled":
        q = q.filter(Alias.enabled)
    elif alias_filter == "disabled":
        q = q.filter(Alias.enabled.is_(False))

    q = q.order_by(Alias.pinned.desc())

    if sort == "old2new":
        q = q.order_by(Alias.created_at)
    elif sort == "new2old":
        q = q.order_by(Alias.created_at.desc())
    elif sort == "a2z":
        q = q.order_by(Alias.email)
    elif sort == "z2a":
        q = q.order_by(Alias.email.desc())
    elif alias_filter == "hibp":
        q = q.filter(Alias.hibp_breaches.any())
    else:
        # default sorting
        q = q.order_by(latest_activity.desc())

    q = list(q.limit(PAGE_LIMIT).offset(page_id * PAGE_LIMIT))

    ret = []
    for alias, contact, email_log, custom_domain, nb_reply, nb_blocked, nb_forward in q:
        ret.append(
            AliasInfo(
                alias=alias,
                mailbox=alias.mailbox,
                mailboxes=alias.mailboxes,
                nb_forward=nb_forward,
                nb_blocked=nb_blocked,
                nb_reply=nb_reply,
                latest_email_log=email_log,
                latest_contact=contact,
                custom_domain=custom_domain,
            )
        )

    return ret


def get_alias_info(alias: Alias) -> AliasInfo:
    q = (
        db.session.query(Contact, EmailLog)
        .filter(Contact.alias_id == alias.id)
        .filter(EmailLog.contact_id == Contact.id)
    )

    alias_info = AliasInfo(
        alias=alias,
        nb_blocked=0,
        nb_forward=0,
        nb_reply=0,
        mailbox=alias.mailbox,
        mailboxes=[alias.mailbox],
    )

    for _, el in q:
        if el.is_reply:
            alias_info.nb_reply += 1
        elif el.blocked:
            alias_info.nb_blocked += 1
        else:
            alias_info.nb_forward += 1

    return alias_info


def get_alias_info_v2(alias: Alias, mailbox=None) -> AliasInfo:
    if not mailbox:
        mailbox = alias.mailbox

    q = (
        db.session.query(Contact, EmailLog)
        .filter(Contact.alias_id == alias.id)
        .filter(EmailLog.contact_id == Contact.id)
    )

    latest_activity: Arrow = alias.created_at
    latest_email_log = None
    latest_contact = None

    alias_info = AliasInfo(
        alias=alias,
        nb_blocked=0,
        nb_forward=0,
        nb_reply=0,
        mailbox=mailbox,
        mailboxes=[mailbox],
    )

    for m in alias._mailboxes:
        alias_info.mailboxes.append(m)

    # remove duplicates
    # can happen that alias.mailbox_id also appears in AliasMailbox table
    alias_info.mailboxes = list(set(alias_info.mailboxes))

    for contact, email_log in q:
        if email_log.is_reply:
            alias_info.nb_reply += 1
        elif email_log.blocked:
            alias_info.nb_blocked += 1
        else:
            alias_info.nb_forward += 1

        if email_log.created_at > latest_activity:
            latest_activity = email_log.created_at
            latest_email_log = email_log
            latest_contact = contact

    alias_info.latest_contact = latest_contact
    alias_info.latest_email_log = latest_email_log

    return alias_info


def get_alias_contacts(alias, page_id: int) -> [dict]:
    q = (
        Contact.query.filter_by(alias_id=alias.id)
        .order_by(Contact.id.desc())
        .limit(PAGE_LIMIT)
        .offset(page_id * PAGE_LIMIT)
    )

    res = []
    for fe in q.all():
        res.append(serialize_contact(fe))

    return res