Background Job: Celery

uv add "celery[redis]"

# workers/celery_app.py
from celery import Celery
from app.core.config import get_settings

settings = get_settings()

celery = Celery(
    "notes",
    broker=settings.redis_url,             # job কোথায় রাখা হবে
    backend=settings.redis_url,            # ফলাফল কোথায় (দরকার হলে)
    include=["app.workers.tasks"],
)

celery.conf.update(
    task_acks_late=True,                   # কাজ শেষ হলে তবেই "নিলাম" বলবে
    task_reject_on_worker_lost=True,       # worker মরে গেলে job আবার queue তে
    worker_prefetch_multiplier=1,          # একবারে একটার বেশি job ধরে রাখবে না
    task_time_limit=300,                   # ৫ মিনিটের বেশি চললে মেরে ফেলো
    task_soft_time_limit=270,
    timezone="Asia/Dhaka",
)

# workers/tasks.py
import logging
import httpx
from app.workers.celery_app import celery

logger = logging.getLogger(__name__)


@celery.task(
    bind=True,
    autoretry_for=(httpx.HTTPError,),       # এই error হলে নিজে থেকে retry
    retry_backoff=True,                      # 1s, 2s, 4s, 8s ...
    retry_backoff_max=600,
    retry_jitter=True,
    max_retries=6,
)
def send_sms(self, phone: str, message: str, dedup_key: str) -> None:
    if already_sent(dedup_key):              # idempotent: দুইবার চললেও একবারই SMS
        return
    with httpx.Client(timeout=10) as client:
        r = client.post(SMS_URL, json={"to": phone, "text": message}, headers=SMS_HEADERS)
        r.raise_for_status()
    mark_sent(dedup_key)
    logger.info("SMS পাঠানো হলো %s", mask(phone))


@celery.task
def build_invoice(order_id: int) -> str:
    pdf = render_invoice_pdf(order_id)
    key = upload_to_storage(pdf, f"invoices/{order_id}.pdf")
    return key

@router.post("/orders", status_code=201)
async def create_order(...):
    order = await svc.create(...)                     # database এ save + commit
    send_sms.delay(order.phone, f"Order #{order.id} নেওয়া হয়েছে", dedup_key=f"order-sms-{order.id}")
    build_invoice.delay(order.id)
    return order

uv run celery -A app.workers.celery_app worker --loglevel=info --concurrency=4

build_invoice.delay(order.id)          # ✓ ছোট, worker নিজে সর্বশেষ data আনবে
build_invoice.delay(order.dict())      # ✗ বড়, আর queue তে থাকার সময় data পুরনো হয়ে যেতে পারে

from celery.schedules import crontab

celery.conf.beat_schedule = {
    "daily-sales-report": {
        "task": "app.workers.tasks.daily_report",
        "schedule": crontab(hour=2, minute=0),          # প্রতিদিন রাত ২টা (ঢাকা)
    },
    "cleanup-expired-tokens": {
        "task": "app.workers.tasks.cleanup_tokens",
        "schedule": crontab(minute=0),                  # প্রতি ঘণ্টায়
    },
    "reconcile-payments": {
        "task": "app.workers.tasks.reconcile_payments",
        "schedule": 300.0,                               # প্রতি ৫ মিনিটে
    },
}

uv run celery -A app.workers.celery_app beat --loglevel=info

services:
  api:
    build: .
    command: uvicorn app.main:app --host 0.0.0.0 --port 8000
  worker:
    build: .                               # একই image
    command: celery -A app.workers.celery_app worker --concurrency=4 --loglevel=info
  beat:
    build: .
    command: celery -A app.workers.celery_app beat --loglevel=info
  redis:
    image: redis:7-alpine