144 lines
5.7 KiB
Python
144 lines
5.7 KiB
Python
"""Mailbox sync Taskiq tasks — Outlook pull + triage + ingest on own stream."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import os
|
|
from datetime import datetime,timezone
|
|
|
|
import redis.asyncio as redis
|
|
from dotenv import load_dotenv
|
|
from taskiq import TaskiqEvents
|
|
|
|
from db_setup import session_scope
|
|
import department.models # noqa: F401
|
|
from inbox.models import MailboxSyncRun
|
|
from inbox.views import Email
|
|
from taskiq_management.broker_setup import MAX_RETRIES,RETRY_DELAY
|
|
from taskiq_management.mailbox_sync_broker_setup import MAILBOX_SYNC_QUEUE_NAME,mailbox_sync_broker
|
|
from taskiq_management.middleware import PermanentTaskError
|
|
|
|
load_dotenv()
|
|
|
|
logger=logging.getLogger("inbox.mailbox_sync")
|
|
REDIS_URL=os.getenv("REDIS_URL","redis://localhost:6379/0")
|
|
_LOCK_KEY="inbox:mailbox_sync:lock"
|
|
_LOCK_TTL=900
|
|
_CONSUMER_GROUP=os.getenv("TASKIQ_CONSUMER_GROUP","taskiq")
|
|
# taskiq-redis listen() skips XAUTOCLAIM while this key exists. redis-py Lock has
|
|
# no TTL by default, so SIGKILL (compose rebuild) leaves it forever and the Sync
|
|
# button stays on a running row nobody will finish.
|
|
_AUTOCLAIM_KEY=f"autoclaim:{_CONSUMER_GROUP}:{MAILBOX_SYNC_QUEUE_NAME}"
|
|
|
|
|
|
@mailbox_sync_broker.on_event(TaskiqEvents.WORKER_STARTUP)
|
|
async def _drop_stale_autoclaim(_state) -> None:
|
|
client=redis.from_url(REDIS_URL,decode_responses=True)
|
|
try:
|
|
if await client.delete(_AUTOCLAIM_KEY):
|
|
logger.warning("dropped stale autoclaim lock %s",_AUTOCLAIM_KEY)
|
|
finally:
|
|
await client.aclose()
|
|
|
|
|
|
async def _fail(run_id:str,error:str) -> dict:
|
|
async with session_scope() as session:
|
|
await MailboxSyncRun.update_run(session,run_id,{
|
|
"status":"failed",
|
|
"error":error,
|
|
"finished_at":datetime.now(timezone.utc),
|
|
})
|
|
return {"status":"failed","error":error}
|
|
|
|
|
|
@mailbox_sync_broker.task(
|
|
task_name="inbox.sync_mailbox",
|
|
retry_on_error=True,
|
|
max_retries=MAX_RETRIES,
|
|
delay=RETRY_DELAY,
|
|
)
|
|
async def sync_mailbox(run_id:str) -> dict:
|
|
if not run_id or not str(run_id).strip():
|
|
raise PermanentTaskError("run_id is required")
|
|
run_id=str(run_id).strip()
|
|
|
|
client=redis.from_url(REDIS_URL,decode_responses=True)
|
|
try:
|
|
acquired=await client.set(_LOCK_KEY,run_id,nx=True,ex=_LOCK_TTL)
|
|
if not acquired:
|
|
holder=await client.get(_LOCK_KEY)
|
|
# Crash/restart redelivers the same run_id while the TTL lock is
|
|
# still set. Failing that as "another sync" strands the lock until
|
|
# expiry and every later click also bounces.
|
|
if holder==run_id:
|
|
await client.expire(_LOCK_KEY,_LOCK_TTL)
|
|
logger.warning("mailbox sync %s reclaimed its own stale lock",run_id)
|
|
else:
|
|
logger.warning("mailbox sync %s skipped: lock held by %s",run_id,holder)
|
|
return await _fail(run_id,"another mailbox sync is already running")
|
|
|
|
try:
|
|
async with session_scope() as session:
|
|
row=await MailboxSyncRun.get_by_id(session,run_id)
|
|
if not row:
|
|
raise PermanentTaskError(f"sync run {run_id} not found")
|
|
if row.status in ("completed","failed"):
|
|
logger.info("mailbox sync %s already %s, skipping",run_id,row.status)
|
|
return {"status":row.status,"error":row.error}
|
|
await MailboxSyncRun.update_run(session,run_id,{
|
|
"status":"running",
|
|
"started_at":datetime.now(timezone.utc),
|
|
"error":None,
|
|
})
|
|
top=row.top or 100
|
|
skip=row.skip or 0
|
|
test_on=True if row.test_on is None else bool(row.test_on)
|
|
|
|
async with session_scope() as session:
|
|
service=Email(session=session)
|
|
if not service.token:
|
|
return await _fail(run_id,"EMAIL_API_TOKEN is not configured")
|
|
|
|
async def on_progress(processed,expected,entries):
|
|
ingested=sum(1 for e in entries if e.get("status") in ("ingested","known"))
|
|
skipped=sum(1 for e in entries if e.get("status")=="skipped")
|
|
errors=sum(1 for e in entries if e.get("status")=="error")
|
|
await MailboxSyncRun.update_run(session,run_id,{
|
|
"entries":list(entries),
|
|
"triage":{
|
|
"expected":expected,
|
|
"processed":processed,
|
|
"ingested":ingested,
|
|
"skipped":skipped,
|
|
"errors":errors,
|
|
"total":expected,
|
|
},
|
|
})
|
|
|
|
try:
|
|
summary=await service.run_mailbox_sync_page(
|
|
top=top,skip=skip,test_on=test_on,on_progress=on_progress,
|
|
)
|
|
except Exception as e:
|
|
logger.exception("mailbox sync failed for run %s",run_id)
|
|
return await _fail(run_id,str(e))
|
|
|
|
await MailboxSyncRun.update_run(session,run_id,{
|
|
"status":"completed",
|
|
"entries":summary["entries"],
|
|
"triage":summary["triage"],
|
|
"error":None,
|
|
"finished_at":datetime.now(timezone.utc),
|
|
})
|
|
return {
|
|
"status":"completed",
|
|
"triage":summary["triage"],
|
|
"entries":len(summary["entries"]),
|
|
}
|
|
finally:
|
|
current=await client.get(_LOCK_KEY)
|
|
if current==run_id:
|
|
await client.delete(_LOCK_KEY)
|
|
finally:
|
|
await client.aclose()
|