Create should_disable

app/email_utils.py

@@ -41,7 +41,16 @@ from app.config import (
 from app.dns_utils import get_mx_domains
 from app.extensions import db
 from app.log import LOG
-from app.models import Mailbox, User, SentAlert, CustomDomain, SLDomain, Contact
+from app.models import (
+    Mailbox,
+    User,
+    SentAlert,
+    CustomDomain,
+    SLDomain,
+    Contact,
+    Alias,
+    EmailLog,
+)
 from app.utils import random_string, convert_to_id, convert_to_alphanumeric
 
 
@@ -893,3 +902,22 @@ def normalize_reply_email(reply_email: str) -> str:
             ret.append(c)
 
     return "".join(ret)
+
+
+def should_disable(alias: Alias) -> bool:
+    """Disable an alias if it has more than 5 bounces in the last 24h"""
+    yesterday = arrow.now().shift(days=-1)
+    nb_bounced_last_24h = (
+        db.session.query(EmailLog)
+        .join(Contact, EmailLog.contact_id == Contact.id)
+        .filter(
+            EmailLog.bounced.is_(True),
+            EmailLog.is_reply.is_(False),
+            EmailLog.created_at > yesterday,
+        )
+        .filter(Contact.alias_id == alias.id)
+        .count()
+    )
+    if nb_bounced_last_24h > 5:
+        return True
+    return False

app/models.py

@@ -1384,6 +1384,7 @@ class EmailLog(db.Model, ModelMixin):
 
     contact = db.relationship(Contact, backref="email_logs")
     mailbox = db.relationship("Mailbox", lazy="joined", foreign_keys=[mailbox_id])
+    user = db.relationship(User)
 
     def bounced_mailbox(self) -> str:
         if self.bounced_mailbox_id:

tests/test_email_utils.py

@@ -22,9 +22,10 @@ from app.email_utils import (
     encode_text,
     EmailEncoding,
     replace,
+    should_disable,
 )
 from app.extensions import db
-from app.models import User, CustomDomain
+from app.models import User, CustomDomain, Alias, Contact, EmailLog
 from tests.utils import login
 
 # flake8: noqa: E101, W191
@@ -551,3 +552,37 @@ def test_encode_text():
     assert encode_text("mèo méo") == "mèo méo"
     assert encode_text("mèo méo", EmailEncoding.BASE64) == "bcOobyBtw6lv"
     assert encode_text("mèo méo", EmailEncoding.QUOTED) == "m=C3=A8o m=C3=A9o"
+
+
+def test_should_disable(flask_client):
+    user = User.create(
+        email="a@b.c",
+        password="password",
+        name="Test User",
+        activated=True,
+        include_sender_in_reverse_alias=True,
+    )
+    alias = Alias.create_new_random(user)
+    db.session.commit()
+
+    assert not should_disable(alias)
+
+    # create a lot of bounce on this alias
+    contact = Contact.create(
+        user_id=user.id,
+        alias_id=alias.id,
+        website_email="contact@example.com",
+        reply_email="rep@sl.local",
+        commit=True,
+    )
+    for _ in range(20):
+        EmailLog.create(
+            user_id=user.id, contact_id=contact.id, commit=True, bounced=True
+        )
+
+    assert should_disable(alias)
+
+    # should not affect another alias
+    alias2 = Alias.create_new_random(user)
+    db.session.commit()
+    assert not should_disable(alias2)