93 lines
3.3 KiB
Python
93 lines
3.3 KiB
Python
 
|
"""Snooze: a conversation leaves the Inbox until a chosen time, then comes back.
|
||
|
|
|
||
|
|
Kept in this database only (the mail server knows nothing of it): a snoozed
|
||
|
|
thread is hidden from the Inbox view and listed under Snoozed. The wake job,
|
||
|
|
run every minute, brings back threads whose time has come — or that got a new
|
||
|
|
message meanwhile, as Gmail does — at the top of the Inbox, the newest message
|
||
|
|
marked unread.
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import datetime as dt
|
||
|
|
import logging
|
||
|
|
import uuid
|
||
|
|
from typing import Any
|
||
|
|
|
||
|
|
from sqlalchemy import or_, select
|
||
|
|
|
||
|
|
from app.db import session_scope
|
||
|
|
from app.events import get_event_bus
|
||
|
|
from app.models import Mailbox, Message, Thread
|
||
|
|
from app.queue import enqueue_job
|
||
|
|
from app.sync.actions import apply_flags_to_messages
|
||
|
|
|
||
|
|
logger = logging.getLogger(__name__)
|
||
|
|
|
||
|
|
MAX_SNOOZE = dt.timedelta(days=366)
|
||
|
|
_SEEN = frozenset({"\\Seen"})
|
||
|
|
|
||
|
|
|
||
|
|
def snooze(thread: Thread, until: dt.datetime, now: dt.datetime) -> None:
|
||
|
|
thread.snoozed_until = until
|
||
|
|
thread.snoozed_at = now
|
||
|
|
|
||
|
|
|
||
|
|
def wake(thread: Thread, now: dt.datetime) -> None:
|
||
|
|
thread.snoozed_until = None
|
||
|
|
thread.snoozed_at = None
|
||
|
|
thread.unsnoozed_at = now
|
||
|
|
|
||
|
|
|
||
|
|
async def wake_due_job(ctx: dict[str, Any]) -> int:
|
||
|
|
"""Cron (every minute): wake snoozed threads that are due. Returns how many."""
|
||
|
|
now = dt.datetime.now(dt.UTC)
|
||
|
|
woken: list[tuple[uuid.UUID, uuid.UUID]] = []
|
||
|
|
unread: list[tuple[uuid.UUID, Message]] = []
|
||
|
|
async with session_scope() as session:
|
||
|
|
due = list(
|
||
|
|
await session.scalars(
|
||
|
|
select(Thread)
|
||
|
|
.where(
|
||
|
|
Thread.snoozed_until.isnot(None),
|
||
|
|
or_(Thread.snoozed_until <= now, Thread.last_message_at > Thread.snoozed_at),
|
||
|
|
)
|
||
|
|
.with_for_update(skip_locked=True)
|
||
|
|
)
|
||
|
|
)
|
||
|
|
for thread in due:
|
||
|
|
on_time = thread.snoozed_until is not None and thread.snoozed_until <= now
|
||
|
|
wake(thread, now)
|
||
|
|
woken.append((thread.user_id, thread.id))
|
||
|
|
if not on_time:
|
||
|
|
continue # a new message woke it: that message is unread already
|
||
|
|
# Back as new: the newest Inbox message becomes unread again.
|
||
|
|
newest = await session.scalar(
|
||
|
|
select(Message)
|
||
|
|
.join(Mailbox, Mailbox.id == Message.mailbox_id)
|
||
|
|
.where(
|
||
|
|
Message.thread_id == thread.id,
|
||
|
|
Message.user_id == thread.user_id,
|
||
|
|
Mailbox.special_use == "inbox",
|
||
|
|
)
|
||
|
|
.order_by(Message.internal_date.desc())
|
||
|
|
.limit(1)
|
||
|
|
)
|
||
|
|
if newest is not None and "\\Seen" in newest.flags:
|
||
|
|
await apply_flags_to_messages(session, [newest], frozenset(), _SEEN)
|
||
|
|
unread.append((thread.user_id, newest))
|
||
|
|
await session.commit()
|
||
|
|
|
||
|
|
bus = get_event_bus()
|
||
|
|
for user_id, message in unread:
|
||
|
|
await enqueue_job("push_flags_job", str(message.id), [], ["\\Seen"])
|
||
|
|
await bus.publish(
|
||
|
|
"flags.changed",
|
||
|
|
{"user_id": str(user_id), "message_id": str(message.id), "flags": list(message.flags)},
|
||
|
|
)
|
||
|
|
for user_id, thread_id in woken:
|
||
|
|
await bus.publish("thread.updated", {"user_id": str(user_id), "thread_id": str(thread_id)})
|
||
|
|
if woken:
|
||
|
|
logger.info("woke %d snoozed thread(s)", len(woken))
|
||
|
|
return len(woken)
|