webmail-client/backend/workers/worker.py

"""Worker job callables and queue function-sets (refactor §1, §6).

The process is launched via ``workers.run`` (``WORKER_ROLE`` decides which arq
workers run). This module only defines the job callables and the two queue
function-sets; the runner wires them onto the ``interactive`` / ``bulk`` queues.
The sync role runs the coordinator directly (see ``workers.run``), not via arq.
"""

from __future__ import annotations

import logging
import uuid
from typing import Any

from arq import func

from app.indexer.attachments import dedupe_existing_job
from app.outbox.draft_sync import push_draft_to_imap_job, remove_imap_draft_job
from app.outbox.jobs import file_sent_copy_job, send_job
from app.push.service import push_new_mail_job
from app.sync.actions import (
    ACTION_MAX_TRIES,
    delete_thread_job,
    empty_folder_job,
    move_job,
    move_thread_to_folder_job,
    push_flags_job,
    restore_thread_job,
)
from app.sync.engine import bootstrap_and_backfill, refresh_account_job

logger = logging.getLogger(__name__)


async def sync_bootstrap_account(ctx: dict[str, Any], account_id: str) -> int:
    """Bootstrap + backfill an account (enqueued on account creation)."""
    logger.info("sync.bootstrap_account start account=%s", account_id)
    count = await bootstrap_and_backfill(uuid.UUID(account_id))
    logger.info("sync.bootstrap_account done account=%s messages=%d", account_id, count)
    return count


# Fast, interactive user actions (§6): short timeout, high concurrency. The ones
# pushing a change to the mail server retry (arq ``Retry``) for ~24 hours while
# the server is unreachable — more tries than arq's default of 5 allows.
_RETRYING_ACTIONS: list[Any] = [
    push_flags_job,
    move_job,
    move_thread_to_folder_job,
    restore_thread_job,
    delete_thread_job,
    empty_folder_job,
    file_sent_copy_job,
]

INTERACTIVE_FUNCTIONS: list[Any] = [
    *(func(job, max_tries=ACTION_MAX_TRIES) for job in _RETRYING_ACTIONS),
    send_job,
    push_draft_to_imap_job,
    remove_imap_draft_job,
    refresh_account_job,
    push_new_mail_job,
]

# Bulk work (§6): long timeout, low concurrency, isolated from interactive jobs.
BULK_FUNCTIONS = [
    sync_bootstrap_account,
    # Resumable (each batch is saved), so a long first run may span timeouts.
    func(dedupe_existing_job, timeout=6 * 3600),
]