mirror of
https://github.com/suitenumerique/messages.git
synced 2026-08-17 21:25:41 +02:00
Jean-Baptiste PENRATH
a8d8e5b435
A reply carrying In-Reply-To to a message we already hold was still rejected when its subject differed, because the delivery path required both a reference match and an identical canonical subject. A subject edited mid-conversation therefore started a brand new thread. The import path had its own laxer rule (message-ids only), so the very same conversation was grouped differently depending on whether it was imported or received over SMTP. Both paths now share find_thread_for_message: In-Reply-To is trusted on its own (RFC 8621 §3 only allows splitting on subject, never requires it), while References — which some clients recycle to start unrelated topics — still needs a matching canonical subject. The canonicalization also accepts "Re:subject" with no space, common on mobile clients.
707 lines
29 KiB
Python
707 lines
29 KiB
Python
"""Handles inbound email delivery logic: receiving messages and delivering to mailboxes."""
|
|
|
|
# pylint: disable=broad-exception-caught
|
|
|
|
import hashlib
|
|
import logging
|
|
import re
|
|
import uuid
|
|
from contextlib import contextmanager, nullcontext
|
|
|
|
from django.core.exceptions import ValidationError
|
|
from django.db import connection, transaction
|
|
from django.db.utils import Error as DjangoDbError
|
|
from django.utils import timezone
|
|
|
|
from jmap_email import (
|
|
first_address_email,
|
|
first_address_name,
|
|
first_msgid,
|
|
sent_at_to_datetime,
|
|
)
|
|
from jmap_email.types import JmapEmail
|
|
|
|
from core import enums, models
|
|
from core.ai.call_label import assign_label_to_thread
|
|
from core.ai.thread_summarizer import summarize_thread
|
|
from core.ai.utils import (
|
|
get_messages_from_thread,
|
|
is_ai_summary_enabled,
|
|
is_auto_labels_enabled,
|
|
)
|
|
from core.mda.utils import thread_snippet
|
|
from core.services.importer.labels import (
|
|
compute_labels_and_flags,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
TOKEN_THRESHOLD_FOR_SUMMARY = 200 # Minimum token count to trigger summarization
|
|
MINIMUM_MESSAGES_FOR_SUMMARY = 3 # Minimum number of messages to trigger summarization
|
|
|
|
# Advisory-lock namespace for inbound delivery. Distinct ``classid`` from the
|
|
# blob-cohort locks (see core.services.tiered_storage) so the two never
|
|
# collide in Postgres' single global advisory-lock keyspace.
|
|
_ADVISORY_LOCK_CLASSID_INBOUND = 0x696E626E # 'inbn' in ASCII
|
|
|
|
|
|
@contextmanager
|
|
def inbound_mailbox_lock(mailbox_id: uuid.UUID):
|
|
"""Serialize inbound message creation for one mailbox.
|
|
|
|
Dedup (does a message with this ``mime_id`` already exist?) and
|
|
thread-bucketing (does a thread already exist for this ``In-Reply-To`` /
|
|
``References``?) are both read-then-decide: two concurrent inbound
|
|
deliveries to the same mailbox could each read "nothing there yet" and
|
|
both create a row, yielding duplicate Messages or two parallel Threads
|
|
with no reconcile path. Holding a per-mailbox Postgres advisory lock for
|
|
the duration of the find-or-create makes that critical section
|
|
cluster-wide serial.
|
|
|
|
Must be called inside ``transaction.atomic()`` — ``pg_advisory_xact_lock``
|
|
binds the lock to the current transaction and releases it on
|
|
commit/rollback. The lock is held only across the (DB-only) find-or-create
|
|
work; slow steps (spam scoring, AI summary/labels) run outside it.
|
|
"""
|
|
# First 4 bytes of the mailbox UUID as a signed int32 — the two-arg
|
|
# pg_advisory_xact_lock(classid, objid) form takes two int4s. A 2^-32
|
|
# false-share collision is operationally invisible given the short,
|
|
# DB-only critical section it guards.
|
|
objid = int.from_bytes(mailbox_id.bytes[:4], byteorder="big", signed=True)
|
|
with connection.cursor() as cursor:
|
|
cursor.execute(
|
|
"SELECT pg_advisory_xact_lock(%s, %s)",
|
|
[_ADVISORY_LOCK_CLASSID_INBOUND, objid],
|
|
)
|
|
yield
|
|
|
|
|
|
def _canonicalize_subject(subject: str | None) -> str:
|
|
"""Strip leading ``Re:`` / ``Fwd:`` (and i18n variants) for thread match."""
|
|
# ``\s*`` after the colon, not ``\s+``: mobile clients regularly send
|
|
# "Re:subject" with no space, and a prefix left in place makes two
|
|
# messages of the same conversation compare as different subjects.
|
|
return re.sub(
|
|
r"^((re|fwd|fw|rep|tr|rép)\s*:\s*)+",
|
|
"",
|
|
(subject or "").lower(),
|
|
flags=re.IGNORECASE,
|
|
).strip()
|
|
|
|
|
|
def _parent_messages(mime_ids: list[str], mailbox: models.Mailbox):
|
|
"""Messages of ``mailbox`` whose Message-ID is one of ``mime_ids``, newest first."""
|
|
return (
|
|
models.Message.objects.filter(
|
|
mime_id__in=mime_ids,
|
|
thread__accesses__mailbox=mailbox,
|
|
)
|
|
.select_related("thread")
|
|
.order_by("-created_at") # Prefer newer matches if multiple found
|
|
)
|
|
|
|
|
|
def find_thread_for_message(
|
|
parsed_email: JmapEmail, mailbox: models.Mailbox
|
|
) -> models.Thread | None:
|
|
"""Attempt to find the existing thread an incoming message belongs to.
|
|
|
|
Two levels of evidence, strongest first:
|
|
|
|
1. ``In-Reply-To`` — an explicit, unambiguous pointer to a single
|
|
parent. When that parent is already in the mailbox the message
|
|
belongs to its thread whatever the subject says: participants
|
|
rewrite subjects mid-conversation, and RFC 8621 §3 only *allows*
|
|
splitting a thread on subject, it never requires it.
|
|
2. ``References`` — a chain some MUAs recycle across unrelated
|
|
conversations (replying to an old mail to start a new topic), so a
|
|
matching canonical subject is required before trusting it.
|
|
|
|
Shared by the SMTP delivery path and the importers: a mailbox that is
|
|
imported and then kept in sync over SMTP must be threaded by one rule,
|
|
otherwise the same conversation splits differently on each path.
|
|
"""
|
|
in_reply_to = first_msgid(parsed_email.get("inReplyTo"))
|
|
references = [
|
|
ref for ref in (parsed_email.get("references") or []) if ref != in_reply_to
|
|
]
|
|
|
|
if in_reply_to:
|
|
parent = _parent_messages([in_reply_to], mailbox).first()
|
|
if parent:
|
|
return parent.thread
|
|
|
|
if references:
|
|
incoming_subject_canonical = _canonicalize_subject(parsed_email.get("subject"))
|
|
for parent in _parent_messages(references, mailbox):
|
|
if _canonicalize_subject(parent.subject) == incoming_subject_canonical:
|
|
return parent.thread
|
|
|
|
return None
|
|
|
|
|
|
def _create_thread(parsed_email: JmapEmail, mailbox: models.Mailbox) -> models.Thread:
|
|
"""Create a new thread."""
|
|
|
|
snippet = thread_snippet(
|
|
parsed_email,
|
|
fallback=parsed_email.get("subject") or "(No snippet available)",
|
|
)
|
|
|
|
# Truncate subject to 255 characters if it exceeds max_length
|
|
thread_subject = parsed_email.get("subject")
|
|
if thread_subject and len(thread_subject) > 255:
|
|
thread_subject = thread_subject[:255]
|
|
|
|
thread = models.Thread.objects.create(
|
|
subject=thread_subject,
|
|
snippet=snippet,
|
|
)
|
|
# Create a thread access for the sender mailbox
|
|
models.ThreadAccess.objects.create(
|
|
thread=thread,
|
|
mailbox=mailbox,
|
|
role=enums.ThreadAccessRoleChoices.EDITOR,
|
|
)
|
|
|
|
return thread
|
|
|
|
|
|
def _record_divergent_rcpt(
|
|
postmark: dict, recipient_email: str, parsed_email: JmapEmail
|
|
) -> None:
|
|
"""Record the envelope RCPT TO in ``postmark`` when it diverges from the
|
|
MIME addressees.
|
|
|
|
In the happy path the RCPT is one of the visible To/Cc addresses and we
|
|
store nothing (keeps ``postmark`` NULL). When it isn't — a BCC'd copy, an
|
|
alias/catch-all, or plus-addressed delivery — the RCPT is the only record
|
|
of how this mailbox actually received the mail, so we keep it. Matching is
|
|
case-insensitive on the address.
|
|
"""
|
|
rcpt = (recipient_email or "").strip().lower()
|
|
if not rcpt:
|
|
return
|
|
visible = {
|
|
(entry.get("email") or "").strip().lower()
|
|
for field_name in ("to", "cc")
|
|
for entry in (parsed_email.get(field_name) or [])
|
|
}
|
|
if rcpt not in visible:
|
|
postmark["rcpt_to"] = recipient_email
|
|
|
|
|
|
def _create_message_from_inbound( # pylint: disable=too-many-arguments
|
|
recipient_email: str,
|
|
parsed_email: JmapEmail,
|
|
raw_data: bytes,
|
|
mailbox: models.Mailbox,
|
|
is_import: bool = False,
|
|
is_import_sender: bool = False,
|
|
imap_labels: list[str] | None = None,
|
|
imap_flags: list[str] | None = None,
|
|
channel: models.Channel | None = None,
|
|
is_spam: bool = False,
|
|
is_trashed: bool = False,
|
|
is_archived: bool = False,
|
|
is_outbound: bool = False,
|
|
blob: "models.Blob | None" = None,
|
|
postmark: dict | None = None,
|
|
) -> models.Message | None:
|
|
"""Create a message and thread from parsed email data.
|
|
|
|
Used for inbound delivery, imports, and outbound submission.
|
|
Returns the created Message on success, or None on failure.
|
|
Callers that only need a boolean can check truthiness of the return value.
|
|
|
|
When ``is_outbound`` is True:
|
|
- ``is_sender`` is forced to True
|
|
- No blob is created (the caller handles DKIM signing + blob via prepare_outbound_message)
|
|
- AI features (summary, auto-labels) are skipped
|
|
- The message is created as a draft (finalized later by prepare_outbound_message)
|
|
|
|
Warning: messages imported here could be is_sender=True.
|
|
"""
|
|
# pylint: disable=too-many-locals,too-many-branches,too-many-statements
|
|
message_flags = {}
|
|
|
|
mime_id = first_msgid(parsed_email.get("messageId")) or None
|
|
|
|
# Dedup, thread-bucketing and the message INSERT form one read-then-write
|
|
# critical section. Serialize it per mailbox under a Postgres advisory lock
|
|
# (held only across this DB-only work) so concurrent inbound deliveries to
|
|
# the same mailbox cannot create duplicate Messages or split a conversation
|
|
# into two parallel Threads. Imports are a single-writer backfill path and
|
|
# skip the lock to avoid serializing bulk loads.
|
|
lock_ctx = nullcontext() if is_import else inbound_mailbox_lock(mailbox.id)
|
|
with transaction.atomic(), lock_ctx:
|
|
# Recheck for an already-stored copy now that we hold the lock.
|
|
# deliver_inbound_message dedups before queueing, but the async
|
|
# processing path and concurrent deliveries can still reach here twice
|
|
# for the same Message-ID; this makes creation idempotent per
|
|
# (mailbox, mime_id).
|
|
# Dedup key: the message's own Message-ID when it has one, else the
|
|
# raw-bytes sha256 (== the blob's sha256). The fallback matters on the
|
|
# import path — Drafts and locally-generated mail often carry no
|
|
# Message-ID, and without it a resumed/re-run import would re-deliver
|
|
# them. sha256 only matches byte-identical copies (the same message
|
|
# imported twice), so it can't collapse genuinely distinct mail.
|
|
existing_message = None
|
|
if not is_outbound:
|
|
if mime_id:
|
|
existing_message = models.Message.objects.filter(
|
|
mime_id=mime_id, thread__accesses__mailbox=mailbox
|
|
).first()
|
|
elif is_import:
|
|
raw_sha256 = hashlib.sha256(raw_data).digest()
|
|
existing_message = models.Message.objects.filter(
|
|
blob__sha256=raw_sha256, thread__accesses__mailbox=mailbox
|
|
).first()
|
|
if existing_message:
|
|
logger.info(
|
|
"Duplicate inbound message %s (MIME ID: %s) in mailbox %s; "
|
|
"skipping create",
|
|
existing_message.id,
|
|
mime_id,
|
|
mailbox.id,
|
|
)
|
|
# Dedup hit on ``(mailbox, mime_id)``: this call did NOT create
|
|
# the row. The common cause is a DUPLICATE INBOUND EMAIL — an
|
|
# upstream MTA redelivered the same Message-ID (SMTP retry /
|
|
# greylisting / relay double-send), so the original ``Message``
|
|
# already exists; a concurrent reprocess could land here too,
|
|
# but the prefork ``time_limit`` / lock-TTL coupling makes that
|
|
# structurally rare. Signal the caller so it skips the
|
|
# non-idempotent finalize side effects (events, draft replies,
|
|
# autoreply, the ``message.delivered`` webhook) that already
|
|
# ran for the original create — re-running would duplicate them.
|
|
# pylint: disable-next=protected-access
|
|
existing_message._created_now = False # noqa: SLF001
|
|
return existing_message
|
|
|
|
# --- 3. Find or Create Thread --- #
|
|
try:
|
|
thread = find_thread_for_message(parsed_email, mailbox)
|
|
if not thread:
|
|
thread = _create_thread(parsed_email, mailbox)
|
|
|
|
except (DjangoDbError, ValidationError) as e:
|
|
logger.error(
|
|
"Failed to find or create thread for %s: %s", recipient_email, e
|
|
)
|
|
# Returning from inside the atomic block would commit any partial
|
|
# writes (e.g. a thread without its message); roll back instead.
|
|
transaction.set_rollback(True)
|
|
return None # Indicate failure
|
|
except Exception as e:
|
|
logger.exception(
|
|
"Unexpected error finding/creating thread for %s: %s",
|
|
recipient_email,
|
|
e,
|
|
)
|
|
transaction.set_rollback(True)
|
|
return None
|
|
|
|
if is_import:
|
|
# get labels from parsed_email
|
|
labels, message_flags = compute_labels_and_flags(
|
|
parsed_email, imap_labels, imap_flags
|
|
)
|
|
for label in labels:
|
|
try:
|
|
label_obj, _ = models.Label.objects.get_or_create(
|
|
name=label, mailbox=mailbox
|
|
)
|
|
thread.labels.add(label_obj)
|
|
except Exception as e:
|
|
logger.exception("Error creating label %s: %s", label, e)
|
|
continue
|
|
|
|
# Apply labels from channel settings (e.g., widget channel tags)
|
|
if channel and channel.settings:
|
|
channel_tags = channel.settings.get("tags", [])
|
|
for tag_id in channel_tags:
|
|
try:
|
|
label_obj = models.Label.objects.get(id=tag_id, mailbox=mailbox)
|
|
thread.labels.add(label_obj)
|
|
except models.Label.DoesNotExist:
|
|
logger.warning(
|
|
"Label %s not found for channel %s, skipping",
|
|
tag_id,
|
|
channel.id,
|
|
)
|
|
except Exception as e:
|
|
logger.exception(
|
|
"Error adding label %s from channel: %s", tag_id, e
|
|
)
|
|
|
|
# --- 4. Get or Create Sender Contact --- #
|
|
sender_email = first_address_email(parsed_email.get("from"))
|
|
sender_name = first_address_name(parsed_email.get("from"))
|
|
|
|
if not sender_email:
|
|
logger.warning(
|
|
"Inbound message for %s missing 'From' email, using fallback.",
|
|
recipient_email,
|
|
)
|
|
sender_email = (
|
|
f"unknown-sender@{mailbox.domain.name}" # Use recipient's domain
|
|
)
|
|
sender_name = sender_name or "Unknown Sender"
|
|
|
|
try:
|
|
# Validate sender_email format before saving
|
|
models.Contact(email=sender_email).full_clean(
|
|
exclude=["mailbox", "name"]
|
|
) # Validate email format
|
|
|
|
sender_contact, created = models.Contact.objects.get_or_create(
|
|
email=sender_email,
|
|
mailbox=mailbox, # Associate contact with the recipient mailbox
|
|
defaults={
|
|
"name": sender_name or sender_email.split("@")[0],
|
|
"email": sender_email, # Ensure correct casing is saved
|
|
},
|
|
)
|
|
if created:
|
|
logger.info(
|
|
"Created contact for sender %s in mailbox %s",
|
|
sender_email,
|
|
mailbox.id,
|
|
)
|
|
|
|
except ValidationError as e:
|
|
logger.error(
|
|
"Validation error for sender contact %s in mailbox %s: %s. Using fallback.",
|
|
sender_email,
|
|
mailbox.id,
|
|
e,
|
|
)
|
|
# Fallback: Use a generic placeholder contact if validation fails
|
|
sender_email = f"invalid-sender@{mailbox.domain.name}"
|
|
sender_name = "Invalid Sender Address"
|
|
sender_contact, _ = models.Contact.objects.get_or_create(
|
|
email=sender_email,
|
|
mailbox=mailbox,
|
|
defaults={"name": sender_name, "email": sender_email},
|
|
)
|
|
except DjangoDbError as e:
|
|
logger.error(
|
|
"DB error getting/creating sender contact %s in mailbox %s: %s",
|
|
sender_email,
|
|
mailbox.id,
|
|
e,
|
|
)
|
|
transaction.set_rollback(True)
|
|
return None # Indicate failure
|
|
except Exception as e:
|
|
logger.exception(
|
|
"Unexpected error with sender contact %s in mailbox %s: %s",
|
|
sender_email,
|
|
mailbox.id,
|
|
e,
|
|
)
|
|
transaction.set_rollback(True)
|
|
return None
|
|
|
|
# --- 5. Create Message --- #
|
|
try:
|
|
# Can we get a parent message for reference?
|
|
# TODO: validate this doesn't create security issues
|
|
parent_message = None
|
|
parent_msg_id = first_msgid(parsed_email.get("inReplyTo"))
|
|
if parent_msg_id:
|
|
parent_message = models.Message.objects.filter(
|
|
mime_id=parent_msg_id, thread=thread
|
|
).first()
|
|
|
|
# Truncate subject to 255 characters if it exceeds max_length
|
|
subject = parsed_email.get("subject")
|
|
if subject and len(subject) > 255:
|
|
subject = subject[:255]
|
|
|
|
is_sender = is_outbound or (is_import and is_import_sender)
|
|
sent_at = sent_at_to_datetime(parsed_email.get("sentAt"))
|
|
|
|
# The Blob INSERT and the Message INSERT must commit together
|
|
# so the GC sweep never sees the Blob row without its
|
|
# referencing FK on ``Message.blob``. Outbound messages have
|
|
# no blob yet — ``prepare_outbound_message`` adds it later.
|
|
# ``postmark`` is assembled by the caller (the inbound task builds
|
|
# the pipeline verdicts + envelope-RCPT divergence); here we only
|
|
# persist it.
|
|
with transaction.atomic():
|
|
# Reuse the ingest blob (inbound queue path) so a message has
|
|
# ONE blob from ingest through to here — no second plaintext
|
|
# copy, no re-encrypt. Imports have no ingest blob and pass
|
|
# raw bytes; outbound gets its blob later from the send path.
|
|
# Import bodies go straight to the object-storage tier (best
|
|
# effort): bulk archives must not park gigabytes in Postgres
|
|
# waiting for the periodic offload.
|
|
if blob is None and not is_outbound:
|
|
blob = models.Blob.objects.create_blob(
|
|
content=raw_data,
|
|
content_type="message/rfc822",
|
|
prefer_offloaded=is_import,
|
|
)
|
|
|
|
message = models.Message.objects.create(
|
|
postmark=postmark or None,
|
|
thread=thread,
|
|
sender=sender_contact,
|
|
subject=subject,
|
|
blob=blob,
|
|
mime_id=first_msgid(parsed_email.get("messageId")) or None,
|
|
parent=parent_message,
|
|
sent_at=(None if is_outbound else (sent_at or timezone.now())),
|
|
is_draft=is_outbound, # Outbound: draft until prepare_outbound_message finalizes
|
|
is_sender=is_sender,
|
|
is_trashed=is_trashed,
|
|
# Keep timestamps in lockstep with the booleans, as the
|
|
# flag endpoint does — a NULL trashed_at/archived_at on a
|
|
# trashed/archived row breaks restore, ordering and any
|
|
# auto-purge that keys off the timestamp.
|
|
trashed_at=(timezone.now() if is_trashed else None),
|
|
is_archived=is_archived,
|
|
archived_at=(timezone.now() if is_archived else None),
|
|
is_spam=is_spam,
|
|
has_attachments=len(parsed_email.get("attachments", [])) > 0,
|
|
channel=channel,
|
|
)
|
|
if is_import:
|
|
# We need to set the created_at field to the date of the message
|
|
# because the inbound message is not created at the same time as the message is received
|
|
message.created_at = sent_at or timezone.now()
|
|
# Extract flags handled via ThreadAccess (not Message fields)
|
|
import_is_unread = message_flags.pop("is_unread", True)
|
|
import_is_starred = message_flags.pop("_starred", False)
|
|
|
|
for flag, value in message_flags.items():
|
|
if hasattr(message, flag):
|
|
setattr(message, flag, value)
|
|
message.save(
|
|
update_fields=[
|
|
"created_at",
|
|
*message_flags.keys(),
|
|
]
|
|
)
|
|
# Update ThreadAccess for read/starred state
|
|
access = models.ThreadAccess.objects.filter(
|
|
thread=thread, mailbox=mailbox
|
|
).first()
|
|
if access:
|
|
update_fields = []
|
|
# Sent messages are always considered read by the sender
|
|
if (is_sender or not import_is_unread) and (
|
|
access.read_at is None or message.created_at > access.read_at
|
|
):
|
|
access.read_at = message.created_at
|
|
update_fields.append("read_at")
|
|
if import_is_starred and access.starred_at is None:
|
|
access.starred_at = message.created_at
|
|
update_fields.append("starred_at")
|
|
if update_fields:
|
|
access.save(update_fields=update_fields)
|
|
elif is_sender:
|
|
access = models.ThreadAccess.objects.filter(
|
|
thread=thread, mailbox=mailbox
|
|
).first()
|
|
if access:
|
|
access.read_at = message.created_at
|
|
access.save(update_fields=["read_at"])
|
|
except (DjangoDbError, ValidationError) as e:
|
|
logger.error("Failed to create message in thread %s: %s", thread.id, e)
|
|
transaction.set_rollback(True)
|
|
return None # Indicate failure
|
|
except Exception as e:
|
|
logger.exception(
|
|
"Unexpected error creating message in thread %s: %s",
|
|
thread.id,
|
|
e,
|
|
)
|
|
transaction.set_rollback(True)
|
|
return None
|
|
|
|
# --- 6. Create Recipient Contacts and Links --- #
|
|
# deduplicate recipients
|
|
recipient_types_to_process = []
|
|
for type_choice, type_name in [
|
|
(models.MessageRecipientTypeChoices.TO, "to"),
|
|
(models.MessageRecipientTypeChoices.CC, "cc"),
|
|
(models.MessageRecipientTypeChoices.BCC, "bcc"),
|
|
]:
|
|
recipients = list(
|
|
{
|
|
frozenset(recipient.items())
|
|
for recipient in (parsed_email.get(type_name) or [])
|
|
}
|
|
)
|
|
recipient_types_to_process.append(
|
|
(type_choice, [dict(recipient) for recipient in recipients])
|
|
)
|
|
|
|
for recipient_type, recipients_list in recipient_types_to_process:
|
|
for recipient_data in recipients_list:
|
|
email = recipient_data.get("email")
|
|
name = recipient_data.get("name")
|
|
if not email:
|
|
logger.warning(
|
|
"Skipping recipient with no email address for message %s.",
|
|
message.id,
|
|
)
|
|
continue
|
|
|
|
try:
|
|
models.Contact(email=email).full_clean(
|
|
exclude=["mailbox", "name"]
|
|
) # Validate
|
|
recipient_contact, created = models.Contact.objects.get_or_create(
|
|
email=email,
|
|
mailbox=mailbox, # Associate contact with the recipient mailbox
|
|
defaults={"name": name or email.split("@")[0], "email": email},
|
|
)
|
|
if created:
|
|
logger.info(
|
|
"Created contact for recipient %s in mailbox %s",
|
|
email,
|
|
mailbox.id,
|
|
)
|
|
|
|
# Create the link between message and contact (use get_or_create to handle duplicates)
|
|
defaults = {}
|
|
if is_import and not message.is_draft:
|
|
defaults["delivery_status"] = (
|
|
enums.MessageDeliveryStatusChoices.SENT_EXTERNAL
|
|
)
|
|
models.MessageRecipient.objects.get_or_create(
|
|
message=message,
|
|
contact=recipient_contact,
|
|
type=recipient_type,
|
|
defaults=defaults,
|
|
)
|
|
except ValidationError as e:
|
|
logger.warning(
|
|
"Validation error creating recipient contact/link (%s) for message %s: %s",
|
|
email,
|
|
message.id,
|
|
e,
|
|
)
|
|
# Continue processing other recipients even if one fails validation
|
|
except DjangoDbError as e:
|
|
logger.error(
|
|
"DB error creating recipient contact/link (%s) for message %s: %s",
|
|
email,
|
|
message.id,
|
|
e,
|
|
)
|
|
# Potentially return False here if one recipient failure should stop all?
|
|
# For now, log and continue.
|
|
except Exception as e:
|
|
logger.exception(
|
|
"Unexpected error with recipient contact/link %s for msg %s: %s",
|
|
email,
|
|
message.id,
|
|
e,
|
|
)
|
|
# Log and continue
|
|
|
|
# --- 7. Process Attachments if present --- #
|
|
# if parsed_email.get("attachments"):
|
|
# _process_attachments(message, parsed_email["attachments"], mailbox)
|
|
|
|
# --- 8. Final Updates --- #
|
|
try:
|
|
# Update snippet using the new message's body if possible
|
|
# (This assumes the subject was used for the initial snippet if body was empty)
|
|
new_snippet = thread_snippet(
|
|
parsed_email,
|
|
fallback=parsed_email.get("subject", ""),
|
|
)
|
|
|
|
if new_snippet:
|
|
thread.snippet = new_snippet
|
|
thread.save(update_fields=["snippet"])
|
|
|
|
# Do not trigger AI features on import, spam, or outbound
|
|
if not is_import and not is_spam and not is_outbound:
|
|
# Update summary if needed is ai is enabled
|
|
if is_ai_summary_enabled():
|
|
messages = get_messages_from_thread(thread)
|
|
token_count = sum(message.get_tokens_count() for message in messages)
|
|
|
|
# Only summarize if the thread has enough content (more than 200 tokens or at least 3 messages)
|
|
if (
|
|
token_count >= TOKEN_THRESHOLD_FOR_SUMMARY
|
|
or len(messages) >= MINIMUM_MESSAGES_FOR_SUMMARY
|
|
):
|
|
new_summary = summarize_thread(thread)
|
|
if new_summary:
|
|
thread.summary = new_summary
|
|
thread.save(update_fields=["summary"])
|
|
|
|
# Assign labels to the thread (skip if channel already applied tags)
|
|
has_channel_tags = (
|
|
channel and channel.settings and channel.settings.get("tags")
|
|
)
|
|
if is_auto_labels_enabled() and not has_channel_tags:
|
|
assign_label_to_thread(thread, mailbox.id)
|
|
|
|
except Exception as e:
|
|
logger.exception(
|
|
"Error updating thread %s after message delivery: %s",
|
|
thread.id,
|
|
e,
|
|
)
|
|
# Don't return False here, delivery was successful
|
|
|
|
thread.update_stats()
|
|
|
|
logger.info(
|
|
"Successfully delivered message %s to mailbox %s (Thread: %s)",
|
|
message.id,
|
|
mailbox.id,
|
|
thread.id,
|
|
)
|
|
# Freshly created this call — the caller may run the one-shot finalize
|
|
# side effects (see the dedup branch above for why this matters).
|
|
# pylint: disable-next=protected-access
|
|
message._created_now = True # noqa: SLF001
|
|
return message # Return created Message on success (truthy), None on failure
|
|
|
|
|
|
# def _process_attachments(
|
|
# message: models.Message, attachment_data: list[Dict], mailbox: models.Mailbox
|
|
# ) -> None:
|
|
# """
|
|
# Process attachments found during email parsing.
|
|
|
|
# Creates Blob records for each attachment and links them to the message.
|
|
|
|
# Args:
|
|
# message: The message object to link attachments to
|
|
# attachment_data: List of attachment data dictionaries from parsing
|
|
# mailbox: The mailbox that owns these attachments
|
|
# """
|
|
# for attachment_info in attachment_data:
|
|
# try:
|
|
# # Check if we have content to store
|
|
# if "content" in attachment_info and attachment_info["content"]:
|
|
# # Create a blob for this attachment using the mailbox method
|
|
# content = attachment_info["content"]
|
|
# blob = mailbox.create_blob(
|
|
# content=content,
|
|
# content_type=attachment_info["type"],
|
|
# )
|
|
|
|
# # Create an attachment record linking to this blob
|
|
# attachment = models.Attachment.objects.create(
|
|
# name=attachment_info.get("name", "unnamed"),
|
|
# blob=blob,
|
|
# mailbox=mailbox,
|
|
# )
|
|
|
|
# # Link the attachment to the message
|
|
# message.attachments.add(attachment)
|
|
# except Exception as e:
|
|
# logger.exception("Error processing attachment: %s", e)
|