lukasandClaude Opus 4.8
e0a45c5a46
- config: WORKER_ROLE (sync|jobs|all), INTERACTIVE_CONCURRENCY, BULK_CONCURRENCY. - queue: named queues (interactive/bulk) + enqueue_job() router; sync_bootstrap goes to bulk, all other jobs to interactive. - workers.run: single entrypoint; role=jobs runs both arq consumers (per-queue timeout/concurrency), role=sync runs the supervisor, all=both (dev default). - docker-compose: worker-sync + worker-jobs services (shared image, nofile ulimit); backend/Dockerfile; Makefile worker/worker-sync/worker-jobs targets. - tests repointed to app.queue.get_arq_pool; supervisor behavior unchanged. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
117 lines
4.1 KiB
Python
117 lines
4.1 KiB
Python
"""Mirror local drafts to the account's IMAP Drafts folder (SPEC §6.5 extension).
|
|
|
|
Drafts are stored server-side as ``\\Draft`` messages so they can be continued on
|
|
any standard mail client. IMAP messages are immutable, so each edit APPENDs a new
|
|
copy and expunges the previous one (tracked by ``drafts.imap_uid``). Runs in the
|
|
worker (credential decryption is worker-only, §10.2).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import logging
|
|
import uuid
|
|
from typing import Any
|
|
|
|
from sqlalchemy import select
|
|
|
|
from app.db import session_scope
|
|
from app.models import Account, Draft, Mailbox, Message
|
|
from app.outbox.attachments import load_attachments
|
|
from app.outbox.mime_build import build_mime
|
|
from app.sync.provider_factory import build_provider
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_DRAFT_FLAGS = frozenset({"\\Draft", "\\Seen"})
|
|
|
|
|
|
async def _ensure_drafts_folder(provider: Any, account: Account, user_id: uuid.UUID) -> str:
|
|
"""Return the account's Drafts folder path, creating one if it is missing."""
|
|
path = "Drafts"
|
|
await provider.create_mailbox(path)
|
|
remote = await provider.list_mailboxes()
|
|
uidvalidity = next((r.uidvalidity for r in remote if r.path == path), 0)
|
|
async with session_scope() as session:
|
|
existing = await session.scalar(
|
|
select(Mailbox).where(Mailbox.account_id == account.id, Mailbox.special_use == "drafts")
|
|
)
|
|
if existing is not None:
|
|
return existing.path
|
|
session.add(
|
|
Mailbox(
|
|
account_id=account.id,
|
|
user_id=user_id,
|
|
path=path,
|
|
special_use="drafts",
|
|
uidvalidity=uidvalidity,
|
|
)
|
|
)
|
|
return path
|
|
|
|
|
|
async def push_draft_to_imap_job(ctx: dict[str, Any], draft_id: str) -> None:
|
|
"""APPEND the draft to the Drafts folder and expunge the previous copy."""
|
|
did = uuid.UUID(draft_id)
|
|
async with session_scope() as session:
|
|
draft = await session.get(Draft, did)
|
|
if draft is None:
|
|
return
|
|
account = await session.get(Account, draft.account_id)
|
|
if account is None:
|
|
return
|
|
user_id = draft.user_id
|
|
old_uid = draft.imap_uid
|
|
reply_to = (
|
|
await session.get(Message, draft.reply_to_message_id)
|
|
if draft.reply_to_message_id
|
|
else None
|
|
)
|
|
attachments = await load_attachments(session, user_id, draft)
|
|
_, sent_raw, _, _ = build_mime(account, draft, reply_to, attachments)
|
|
drafts_mb = await session.scalar(
|
|
select(Mailbox).where(Mailbox.account_id == account.id, Mailbox.special_use == "drafts")
|
|
)
|
|
drafts_path = drafts_mb.path if drafts_mb else None
|
|
account_obj = account
|
|
|
|
provider = build_provider(account_obj)
|
|
new_uid = 0
|
|
try:
|
|
if drafts_path is None:
|
|
drafts_path = await _ensure_drafts_folder(provider, account_obj, user_id)
|
|
new_uid = await provider.append(drafts_path, sent_raw, _DRAFT_FLAGS)
|
|
if old_uid:
|
|
with contextlib.suppress(Exception):
|
|
await provider.delete_messages(drafts_path, [old_uid])
|
|
finally:
|
|
await provider.close()
|
|
|
|
if new_uid:
|
|
async with session_scope() as session:
|
|
draft = await session.get(Draft, did)
|
|
if draft is not None:
|
|
draft.imap_uid = new_uid
|
|
|
|
|
|
async def remove_imap_draft_job(ctx: dict[str, Any], account_id: str, uid: int) -> None:
|
|
"""Expunge a draft copy from the account's Drafts folder (on discard)."""
|
|
aid = uuid.UUID(account_id)
|
|
async with session_scope() as session:
|
|
account = await session.get(Account, aid)
|
|
if account is None:
|
|
return
|
|
drafts_mb = await session.scalar(
|
|
select(Mailbox).where(Mailbox.account_id == aid, Mailbox.special_use == "drafts")
|
|
)
|
|
drafts_path = drafts_mb.path if drafts_mb else None
|
|
account_obj = account
|
|
if drafts_path is None:
|
|
return
|
|
provider = build_provider(account_obj)
|
|
try:
|
|
with contextlib.suppress(Exception):
|
|
await provider.delete_messages(drafts_path, [int(uid)])
|
|
finally:
|
|
await provider.close()
|