system-prompts-and-models-of-ai-tools/salesflow-saas/backend/app/workers/message_tasks.py

from app.workers.celery_app import celery_app
from app.config import get_settings
from app.database import SessionLocal
from datetime import datetime, timezone
import logging

logger = logging.getLogger(__name__)
settings = get_settings()


@celery_app.task(name="app.workers.message_tasks.send_scheduled_messages")
def send_scheduled_messages():
    """Send messages that are scheduled for delivery."""
    from app.models.message import Message
    from sqlalchemy import select, and_

    logger.info("Processing scheduled messages...")

    with SessionLocal() as db:
        now = datetime.now(timezone.utc)

        pending_messages = db.execute(
            select(Message).where(
                and_(
                    Message.status == "pending",
                    Message.scheduled_at <= now if hasattr(Message, 'scheduled_at') else True,
                )
            ).limit(50)
        ).scalars().all()

        sent_count = 0
        failed_count = 0

        for msg in pending_messages:
            try:
                if msg.channel == "whatsapp":
                    result = _send_whatsapp_message(msg.lead_id, msg.content, str(msg.tenant_id))
                elif msg.channel == "email":
                    result = _send_email_message(msg)
                elif msg.channel == "sms":
                    result = _send_sms_message(msg)
                else:
                    logger.warning(f"Unknown channel: {msg.channel}")
                    continue

                msg.status = "sent"
                msg.sent_at = now
                sent_count += 1

            except Exception as e:
                logger.error(f"Failed to send message {msg.id}: {e}")
                msg.status = "failed"
                failed_count += 1

        db.commit()
        logger.info(f"Sent {sent_count}, failed {failed_count} of {len(pending_messages)} messages")

    return {"sent": sent_count, "failed": failed_count}


@celery_app.task(name="app.workers.message_tasks.send_whatsapp", bind=True, max_retries=3)
def send_whatsapp(self, phone: str, message: str, tenant_id: str):
    """Send a WhatsApp message via Business API."""
    from app.integrations.whatsapp import send_whatsapp_message
    from app.models.message import Message

    logger.info(f"Sending WhatsApp to {phone}")

    try:
        result = send_whatsapp_message(phone, message)

        # Store message record
        with SessionLocal() as db:
            msg = Message(
                tenant_id=tenant_id,
                channel="whatsapp",
                direction="outbound",
                content=message,
                status="sent",
                sent_at=datetime.now(timezone.utc),
                metadata={"phone": phone, "wa_message_id": result.get("messages", [{}])[0].get("id", "")},
            )
            db.add(msg)
            db.commit()

        logger.info(f"WhatsApp sent successfully to {phone}")
        return {"status": "sent", "phone": phone}

    except Exception as e:
        logger.error(f"WhatsApp send failed to {phone}: {e}")
        raise self.retry(exc=e, countdown=60 * (self.request.retries + 1))


@celery_app.task(name="app.workers.message_tasks.send_email", bind=True, max_retries=3)
def send_email(self, to_email: str, subject: str, body: str, tenant_id: str):
    """Send an email via configured provider."""
    from app.integrations.email_sender import send_email as smtp_send
    from app.models.message import Message

    logger.info(f"Sending email to {to_email}")

    try:
        smtp_send(to_email, subject, body)

        with SessionLocal() as db:
            msg = Message(
                tenant_id=tenant_id,
                channel="email",
                direction="outbound",
                content=body,
                status="sent",
                sent_at=datetime.now(timezone.utc),
                metadata={"to": to_email, "subject": subject},
            )
            db.add(msg)
            db.commit()

        logger.info(f"Email sent successfully to {to_email}")
        return {"status": "sent", "email": to_email}

    except Exception as e:
        logger.error(f"Email send failed to {to_email}: {e}")
        raise self.retry(exc=e, countdown=60 * (self.request.retries + 1))


@celery_app.task(name="app.workers.message_tasks.send_sms", bind=True, max_retries=3)
def send_sms(self, phone: str, message: str, tenant_id: str):
    """Send SMS via Unifonic."""
    from app.integrations.sms import send_sms as unifonic_send
    from app.models.message import Message

    logger.info(f"Sending SMS to {phone}")

    try:
        unifonic_send(phone, message)

        with SessionLocal() as db:
            msg = Message(
                tenant_id=tenant_id,
                channel="sms",
                direction="outbound",
                content=message,
                status="sent",
                sent_at=datetime.now(timezone.utc),
                metadata={"phone": phone},
            )
            db.add(msg)
            db.commit()

        logger.info(f"SMS sent successfully to {phone}")
        return {"status": "sent", "phone": phone}

    except Exception as e:
        logger.error(f"SMS send failed to {phone}: {e}")
        raise self.retry(exc=e, countdown=60 * (self.request.retries + 1))


def _send_whatsapp_message(lead_id, content, tenant_id):
    """Helper to send WhatsApp from message record."""
    from app.integrations.whatsapp import send_whatsapp_message
    from app.models.lead import Lead

    with SessionLocal() as db:
        lead = db.get(Lead, lead_id)
        if lead and lead.phone:
            return send_whatsapp_message(lead.phone, content)
    return None


def _send_email_message(msg):
    """Helper to send email from message record."""
    from app.integrations.email_sender import send_email as smtp_send
    from app.models.lead import Lead

    with SessionLocal() as db:
        lead = db.get(Lead, msg.lead_id) if msg.lead_id else None
        if lead and lead.email:
            smtp_send(lead.email, "Dealix - متابعة", msg.content)
    return None


def _send_sms_message(msg):
    """Helper to send SMS from message record."""
    from app.integrations.sms import send_sms as unifonic_send
    from app.models.lead import Lead

    with SessionLocal() as db:
        lead = db.get(Lead, msg.lead_id) if msg.lead_id else None
        if lead and lead.phone:
            unifonic_send(lead.phone, msg.content)
    return None