amo-server/workers/broker.py

"""
FastStream broker setup for AMO CRM service background processing.

This module sets up the Redis-based message broker and defines worker
functions for processing export jobs, data synchronization, and scheduled tasks.
"""

import logging
from typing import Dict, Any
from faststream import FastStream
from faststream.redis import RedisBroker
from utils.config import settings

# Configure logging
logging.basicConfig(level=settings.LOG_LEVEL)
logger = logging.getLogger(__name__)

# Initialize Redis broker
broker = RedisBroker(settings.REDIS_URL)
app = FastStream(broker)


@broker.subscriber("export-jobs")
async def process_export_job(job_data: Dict[str, Any]) -> None:
    """
    Process Google Sheets export jobs.

    Args:
        job_data: Dictionary containing job_id, configuration_id, and metadata
    """
    logger.info(f"Processing export job: {job_data.get('job_id')}")

    try:
        from servers.export_server import ExportServer

        export_server = ExportServer()
        await export_server.process_export_job(job_data)

        logger.info(f"Export job {job_data.get('job_id')} completed successfully")

    except Exception as e:
        logger.error(f"Export job {job_data.get('job_id')} failed: {str(e)}")
        raise


@broker.subscriber("sync-jobs")
async def process_sync_job(sync_data: Dict[str, Any]) -> None:
    """
    Process AMO CRM data synchronization jobs.

    Args:
        sync_data: Dictionary containing job_id, entity_type, and parameters
    """
    logger.info(f"Processing sync job: {sync_data.get('job_id')} for {sync_data.get('entity_type')}")

    try:
        from servers.sync_server import SyncServer

        sync_server = SyncServer()
        await sync_server.process_sync_job(sync_data)

        logger.info(f"Sync job {sync_data.get('job_id')} completed successfully")

    except Exception as e:
        logger.error(f"Sync job {sync_data.get('job_id')} failed: {str(e)}")
        raise


@broker.subscriber("refresh-jobs")
async def process_refresh_job(entity_type: str) -> None:
    """
    Process scheduled entity data refresh jobs.

    Args:
        entity_type: Type of entity to refresh (deals, contacts, companies, etc.)
    """
    logger.info(f"Processing refresh job for entity type: {entity_type}")

    try:
        from servers.sync_server import SyncServer

        sync_server = SyncServer()
        await sync_server.refresh_entity_data(entity_type)

        logger.info(f"Refresh job for {entity_type} completed successfully")

    except Exception as e:
        logger.error(f"Refresh job for {entity_type} failed: {str(e)}")
        raise


@broker.subscriber("full-sync-jobs")
async def process_full_sync_job(job_data: Dict[str, Any]) -> None:
    """
    Process full synchronization jobs for AMO CRM entities.

    Supports syncing individual entity types or all entities when entity_type='all'.
    This is a long-running operation that can take several hours.

    Args:
        job_data: Dictionary containing job_id, entity_type, batch_size, etc.
    """
    job_id = job_data.get("job_id", "unknown")
    entity_type = job_data.get("entity_type")
    batch_size = job_data.get("batch_size", 250)

    if not entity_type:
        logger.error(f"Full sync job {job_id} missing entity_type")
        raise ValueError("entity_type is required for full sync job")

    logger.info(f"Processing full sync job {job_id} for entity type: {entity_type}")

    try:
        from servers.sync_server import SyncServer

        sync_server = SyncServer()

        # Handle "all" entity type
        if entity_type == "all":
            # IMPORTANT: Sync order matters due to foreign key constraints
            # 1. Users must be synced first (referenced by all other entities)
            # 2. Pipelines must be before deals (deals reference pipeline stages)
            # 3. Companies, contacts, deals, events can follow
            all_entities = ["users", "pipelines", "companies", "contacts", "deals", "events"]
            total_all_records = 0

            logger.info(f"Starting full sync for ALL entities: {all_entities}")

            for entity in all_entities:
                logger.info(f"Starting full sync for {entity} (part of 'all' job)")
                try:
                    records = await sync_server.full_sync_entity(entity, batch_size=batch_size)
                    total_all_records += records
                    logger.info(f"Completed full sync for {entity}: {records} records")
                except Exception as entity_error:
                    logger.error(f"Error syncing {entity} in 'all' job: {str(entity_error)}")
                    # Continue with next entity even if one fails
                    continue

            logger.info(
                f"Full sync job {job_id} for ALL entities completed: "
                f"{total_all_records} total records across all entities"
            )
        else:
            # Single entity sync
            total_records = await sync_server.full_sync_entity(entity_type, batch_size=batch_size)
            logger.info(f"Full sync job {job_id} for {entity_type} completed: {total_records} records")

    except Exception as e:
        logger.error(f"Full sync job {job_id} failed: {str(e)}")
        raise


@broker.subscriber("failed-jobs")
async def handle_failed_jobs(job_data: Dict[str, Any]) -> None:
    """
    Handle permanently failed jobs (dead letter queue).

    Args:
        job_data: Failed job data for logging and potential manual intervention
    """
    logger.error(f"Job permanently failed: {job_data}")

    # TODO: Implement notification system (email, Slack, etc.)
    # TODO: Store failed jobs for manual review

    # For now, just log the failure
    job_id = job_data.get("job_id", "unknown")
    job_type = job_data.get("job_type", "unknown")
    error_message = job_data.get("error_message", "No error message provided")

    logger.error(
        f"DEAD LETTER QUEUE - Job ID: {job_id}, "
        f"Type: {job_type}, Error: {error_message}"
    )


@app.after_startup
async def startup_handler() -> None:
    """Handle broker startup tasks."""
    logger.info("FastStream broker started successfully")
    logger.info(f"Connected to Redis: {settings.REDIS_URL}")


@app.on_shutdown
async def shutdown_handler() -> None:
    """Handle broker shutdown tasks."""
    logger.info("FastStream broker shutting down")


if __name__ == "__main__":
    # This allows running the worker directly with: python workers/broker.py
    import asyncio
    asyncio.run(app.run())