Integrate FastStream for background processing in AMO CRM Data Collection Service. Replaced Celery with FastStream and Redis for job management. Updated application structure to support asynchronous job handling, including export and synchronization tasks. Enhanced logging and error handling throughout the service. Added installation and startup scripts for FastStream services.

app.py

@@ -2,26 +2,45 @@ from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 from contextlib import asynccontextmanager
 import uvicorn
+import logging
 
 from adapters.sqlite.database import init_db
 from routers import entities, export, data, amocrm
 from utils.config import settings
 
+# FastStream integration
+from faststream.redis.fastapi import RedisRouter
+from workers.broker import broker
+from workers.middleware import setup_middleware
+
+# Configure logging
+logging.basicConfig(level=settings.LOG_LEVEL)
+logger = logging.getLogger(__name__)
+
+
+# Setup FastStream middleware
+setup_middleware(broker)
+
+# Create FastStream router for FastAPI integration
+redis_router = RedisRouter(broker)
+
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
     # Startup
+    logger.info("Starting AMO CRM Data Collection Service")
     await init_db()
+    logger.info("Database initialized successfully")
     yield
     # Shutdown
-    pass
+    logger.info("Shutting down AMO CRM Data Collection Service")
 
 
 app = FastAPI(
     title="AMO CRM Data Collection Service",
-    description="Service for collecting and exporting AMO CRM data to Google Sheets",
+    description="Service for collecting and exporting AMO CRM data to Google Sheets with FastStream workers",
     version="0.1.0",
-    lifespan=lifespan,
+    lifespan=redis_router.lifespan_context,
 )
 
 # CORS middleware
@@ -33,7 +52,10 @@ app.add_middleware(
     allow_headers=["*"],
 )
 
-# Include routers
+# Include FastStream router for message handling
+app.include_router(redis_router)
+
+# Include API routers
 app.include_router(entities.router, prefix=f"{settings.API_V1_STR}/entities", tags=["entities"])
 app.include_router(export.router, prefix=f"{settings.API_V1_STR}/export", tags=["export"])
 app.include_router(data.router, prefix=f"{settings.API_V1_STR}/data", tags=["data"])
@@ -41,12 +63,12 @@ app.include_router(amocrm.router, prefix=f"{settings.API_V1_STR}/amocrm", tags=[
 
 
 @app.get("/")
-async def root():
+async def root() -> dict[str, str]:
     return {"message": "AMO CRM Data Collection Service", "version": "0.1.0"}
 
 
 @app.get("/health")
-async def health_check():
+async def health_check() -> dict[str, str]:
     return {"status": "healthy"}
 
 

docs/amocrm-service-design.md

@@ -612,8 +612,8 @@ def process_custom_field(field_data, entity_type, entity_id):
 - ✅ Job status tracking
 
 #### Phase 5: Workers & Background Processing
-- 🔄 Celery worker implementation
+- 🔄 FastStream worker implementation
-- 🔄 Data synchronization workers
+- 🔄 Data synchronization workers with Redis broker
 - 🔄 Error handling and retry logic
 - 🔄 Monitoring and logging
 
@@ -626,7 +626,7 @@ def process_custom_field(field_data, entity_type, entity_id):
 - **HTTP Client**: httpx for AMO CRM API
 - **Data Validation**: Pydantic v2
 - **Testing**: pytest with real AMO CRM fixtures
-- **Background Jobs**: Celery with Redis (planned)
+- **Background Jobs**: FastStream with Redis broker
 - **Google Sheets**: google-api-python-client (planned)
 - **Development**: Black, isort, mypy, flake8
 
@@ -645,7 +645,7 @@ AMO_CRM_ACCESS_TOKEN=your-longterm-access-token
 GOOGLE_SERVICE_ACCOUNT_FILE=path/to/service-account.json
 GOOGLE_SCOPES=https://www.googleapis.com/auth/spreadsheets
 
-# Redis (for future Celery integration)
+# Redis (for FastStream message broker)
 REDIS_URL=redis://localhost:6379/0
 
 # API Settings
@@ -658,6 +658,9 @@ LOG_LEVEL=INFO
 - **Long-term Tokens**: Uses AMO CRM long-term access tokens
 - **Real-time Testing**: Direct AMO CRM data fetching
 - **Comprehensive Fixtures**: Real API response examples
+- **Modern Workers**: FastStream async message processing
+- **Redis Integration**: Reliable message broker with persistence
+- **Built-in Observability**: Prometheus metrics and OpenTelemetry tracing
 
 ## Quick Start
 
@@ -668,16 +671,25 @@ git clone <repository>
 cd amo-server
 uv sync
 
+# Install FastStream with Redis support
+uv add 'faststream[redis]'
+
 # Configure environment
 cp env.example .env
-# Edit .env with your AMO CRM token
+# Edit .env with your AMO CRM token and Redis URL
 ```
 
 ### 2. Run Service
 ```bash
+# Start Redis (required for FastStream workers)
+redis-server
+
 # Start development server
 uv run uvicorn app:app --reload
 
+# Start FastStream workers (in separate terminal)
+uv run faststream run workers.broker:app --reload
+
 # Access API documentation
 # http://localhost:8000/docs
 ```
@@ -743,6 +755,355 @@ curl -X POST http://localhost:8000/api/v1/export/configure \
 4. **Database Errors**: Handle connection issues, constraint violations
 5. **Retry Logic**: Exponential backoff for transient failures (planned)
 
+## FastStream Workers & Background Processing
+
+### Worker Architecture
+
+FastStream replaces Celery for background job processing, providing a modern async-first approach with Redis as the message broker.
+
+#### Why FastStream over Celery?
+
+1. **Native Async Support**: Built from ground-up for async/await patterns
+2. **Type Safety**: Full Pydantic integration with automatic validation
+3. **Modern Python**: Leverages Python 3.8+ features and type hints
+4. **Simplified Architecture**: No separate result backend needed
+5. **FastAPI Integration**: Seamless integration with existing FastAPI codebase
+6. **Built-in Observability**: Prometheus and OpenTelemetry out-of-the-box
+7. **Better Error Handling**: Structured error handling with automatic retries
+8. **AsyncAPI Documentation**: Automatic API documentation for message flows
+
+The architecture consists of:
+
+1. **Message Producers**: API endpoints that queue jobs
+2. **Message Consumers**: Worker functions that process jobs
+3. **Redis Broker**: Message routing and persistence
+4. **Job Status Tracking**: Database-backed status updates
+
+### Worker Implementation
+
+#### 1. FastStream Broker Setup
+
+```python
+# workers/broker.py
+from faststream import FastStream
+from faststream.redis import RedisBroker
+from utils.config import get_settings
+
+settings = get_settings()
+broker = RedisBroker(settings.redis_url)
+app = FastStream(broker)
+
+# Export job processing
+@broker.subscriber("export-jobs")
+async def process_export_job(job_data: dict):
+    """Process Google Sheets export jobs"""
+    from servers.export_server import ExportServer
+
+    export_server = ExportServer()
+    await export_server.process_export_job(job_data)
+
+# AMO CRM data synchronization
+@broker.subscriber("sync-jobs")
+async def process_sync_job(sync_data: dict):
+    """Synchronize data from AMO CRM"""
+    from servers.sync_server import SyncServer
+
+    sync_server = SyncServer()
+    await sync_server.process_sync_job(sync_data)
+
+# Scheduled data refresh
+@broker.subscriber("refresh-jobs")
+async def process_refresh_job(entity_type: str):
+    """Refresh entity data from AMO CRM"""
+    from servers.sync_server import SyncServer
+
+    sync_server = SyncServer()
+    await sync_server.refresh_entity_data(entity_type)
+```
+
+#### 2. Job Publishers
+
+```python
+# servers/job_server.py
+from faststream.redis import RedisBroker
+from typing import Dict, Any
+import uuid
+from datetime import datetime
+
+class JobServer:
+    def __init__(self, broker: RedisBroker):
+        self.broker = broker
+
+    async def queue_export_job(self, configuration_id: int) -> str:
+        """Queue an export job for processing"""
+        job_id = str(uuid.uuid4())
+        job_data = {
+            "job_id": job_id,
+            "configuration_id": configuration_id,
+            "created_at": datetime.utcnow().isoformat(),
+            "status": "queued"
+        }
+
+        # Publish to export-jobs channel
+        await self.broker.publish(job_data, "export-jobs")
+        return job_id
+
+    async def queue_sync_job(self, entity_type: str, **kwargs) -> str:
+        """Queue a data synchronization job"""
+        job_id = str(uuid.uuid4())
+        sync_data = {
+            "job_id": job_id,
+            "entity_type": entity_type,
+            "parameters": kwargs,
+            "created_at": datetime.utcnow().isoformat(),
+            "status": "queued"
+        }
+
+        await self.broker.publish(sync_data, "sync-jobs")
+        return job_id
+
+    async def schedule_refresh_job(self, entity_type: str):
+        """Schedule periodic data refresh"""
+        await self.broker.publish(entity_type, "refresh-jobs")
+```
+
+#### 3. Export Job Processing
+
+```python
+# servers/export_server.py
+from typing import Dict, Any
+from adapters.sqlite.database import get_database
+from adapters.google_sheets_client import GoogleSheetsClient
+from datetime import datetime
+import logging
+
+logger = logging.getLogger(__name__)
+
+class ExportServer:
+    def __init__(self):
+        self.db = get_database()
+        self.sheets_client = GoogleSheetsClient()
+
+    async def process_export_job(self, job_data: Dict[str, Any]):
+        """Process a Google Sheets export job"""
+        job_id = job_data["job_id"]
+        configuration_id = job_data["configuration_id"]
+
+        try:
+            # Update job status to running
+            await self._update_job_status(job_id, "running")
+
+            # Get export configuration
+            config = await self._get_export_configuration(configuration_id)
+
+            # Process each enabled entity
+            total_records = 0
+            for entity_type, mapping in config["entity_mappings"].items():
+                if mapping.get("is_enabled", False):
+                    records_count = await self._export_entity(
+                        entity_type,
+                        config["sheet_id"],
+                        mapping
+                    )
+                    total_records += records_count
+
+            # Update job as completed
+            await self._update_job_status(
+                job_id,
+                "completed",
+                records_processed=total_records
+            )
+
+        except Exception as e:
+            logger.error(f"Export job {job_id} failed: {str(e)}")
+            await self._update_job_status(
+                job_id,
+                "failed",
+                error_message=str(e)
+            )
+
+    async def _export_entity(self, entity_type: str, sheet_id: str, mapping: dict) -> int:
+        """Export specific entity type to Google Sheets"""
+        # Implementation details for entity export
+        pass
+
+    async def _update_job_status(self, job_id: str, status: str, **kwargs):
+        """Update job status in database"""
+        # Update export_jobs table
+        pass
+```
+
+#### 4. Data Synchronization Workers
+
+```python
+# servers/sync_server.py
+from adapters.amocrm_client import AMOCRMClient
+from adapters.sqlite.database import get_database
+from typing import Dict, Any, List
+import logging
+
+logger = logging.getLogger(__name__)
+
+class SyncServer:
+    def __init__(self):
+        self.amocrm = AMOCRMClient()
+        self.db = get_database()
+
+    async def process_sync_job(self, sync_data: Dict[str, Any]):
+        """Process AMO CRM data synchronization"""
+        job_id = sync_data["job_id"]
+        entity_type = sync_data["entity_type"]
+        parameters = sync_data.get("parameters", {})
+
+        try:
+            logger.info(f"Starting sync job {job_id} for {entity_type}")
+
+            # Fetch data from AMO CRM
+            data = await self.amocrm.fetch_entity_data(
+                entity_type,
+                **parameters
+            )
+
+            # Process and store data
+            processed_count = await self._store_entity_data(entity_type, data)
+
+            logger.info(f"Sync job {job_id} completed: {processed_count} records")
+
+        except Exception as e:
+            logger.error(f"Sync job {job_id} failed: {str(e)}")
+            raise
+
+    async def refresh_entity_data(self, entity_type: str):
+        """Refresh all data for an entity type"""
+        logger.info(f"Refreshing {entity_type} data")
+
+        # Implement incremental refresh logic
+        last_update = await self._get_last_update_timestamp(entity_type)
+
+        data = await self.amocrm.fetch_entity_data(
+            entity_type,
+            updated_at=last_update,
+            limit=250
+        )
+
+        await self._store_entity_data(entity_type, data)
+```
+
+### FastStream Integration with FastAPI
+
+```python
+# app.py (updated)
+from faststream.redis.fastapi import RedisRouter
+from workers.broker import broker
+
+# Create FastStream router for FastAPI integration
+redis_router = RedisRouter(broker)
+
+# Include in FastAPI app
+app.include_router(redis_router)
+
+# Lifespan integration
+app = FastAPI(lifespan=redis_router.lifespan_context)
+```
+
+### Running Workers
+
+#### Development
+```bash
+# Start FastStream worker
+faststream run workers.broker:app
+
+# Or with auto-reload
+faststream run workers.broker:app --reload
+```
+
+#### Production
+```bash
+# Run multiple worker instances
+faststream run workers.broker:app --workers 4
+
+# With specific worker ID
+WORKER_ID=worker-1 faststream run workers.broker:app
+```
+
+### Monitoring and Observability
+
+FastStream provides built-in observability features:
+
+```python
+# workers/middleware.py
+from faststream.prometheus import PrometheusMiddleware
+from faststream.observability.middleware import TelemetryMiddleware
+
+# Add Prometheus metrics
+broker.add_middleware(PrometheusMiddleware)
+
+# Add OpenTelemetry tracing
+broker.add_middleware(TelemetryMiddleware)
+```
+
+### Error Handling and Retry Logic
+
+```python
+# workers/error_handling.py
+from faststream.redis import RedisBroker
+import asyncio
+from typing import Any
+import logging
+
+logger = logging.getLogger(__name__)
+
+@broker.subscriber("export-jobs", retry=3, retry_delay=60)
+async def process_export_job_with_retry(job_data: dict):
+    """Export job with automatic retry"""
+    try:
+        await process_export_job(job_data)
+    except Exception as e:
+        logger.error(f"Job failed: {e}")
+        # FastStream will automatically retry based on retry settings
+        raise
+
+# Dead letter queue for failed jobs
+@broker.subscriber("failed-jobs")
+async def handle_failed_jobs(job_data: dict):
+    """Handle permanently failed jobs"""
+    logger.error(f"Job permanently failed: {job_data}")
+    # Implement notification or manual intervention logic
+```
+
+### Job Scheduling
+
+For scheduled tasks, integrate with FastStream's scheduling capabilities:
+
+```python
+# workers/scheduler.py
+from taskiq_faststream import StreamScheduler
+from taskiq.schedule_sources import LabelScheduleSource
+
+# Schedule periodic data refresh
+@broker.task(
+    message={"entity_type": "deals"},
+    channel="refresh-jobs",
+    schedule=[{"cron": "0 */6 * * *"}]  # Every 6 hours
+)
+async def scheduled_deals_refresh():
+    pass
+
+@broker.task(
+    message={"entity_type": "contacts"},
+    channel="refresh-jobs",
+    schedule=[{"cron": "0 */4 * * *"}]  # Every 4 hours
+)
+async def scheduled_contacts_refresh():
+    pass
+
+# Initialize scheduler
+scheduler = StreamScheduler(
+    broker=broker,
+    sources=[LabelScheduleSource(broker)]
+)
+```
+
 ## Scripts and Utilities
 
 ### fetch_amocrm_data.py

pyproject.toml

@@ -17,7 +17,8 @@ dependencies = [
     "google-api-python-client>=2.100.0",
     "google-auth-httplib2>=0.2.0",
     "google-auth-oauthlib>=1.1.0",
-    "celery>=5.3.0",
+    "faststream[redis]>=0.5.0",
+    "taskiq-faststream>=0.2.0",
     "redis>=5.0.0",
     "python-dotenv>=1.0.0",
 ]

routers/export.py

@@ -4,9 +4,14 @@ from typing import Dict, Any, List, Optional
 from pydantic import BaseModel
 from datetime import datetime
 import time
+import logging
 
 from adapters.sqlite.database import get_db
 from adapters.sqlite.models import ExportConfiguration, ExportEntityMapping, ExportJob
+from servers.job_server import JobServer
+from workers.broker import broker
+
+logger = logging.getLogger(__name__)
 
 router = APIRouter()
 
@@ -35,6 +40,12 @@ class ExportJobStart(BaseModel):
     configuration_id: int
 
 
+class SyncJobRequest(BaseModel):
+    entity_type: str
+    limit: Optional[int] = None
+    page: Optional[int] = None
+
+
 @router.post("/configure")
 async def create_export_configuration(
     config: ExportConfigurationRequest,
@@ -168,14 +179,32 @@ async def start_export_job(
     db.add(job)
     db.commit()
 
-    # TODO: Queue job for background processing with Celery
+    # Queue job for background processing with FastStream
-    # celery_app.send_task("export_to_sheets", args=[job.id])
+    try:
+        job_server = JobServer(broker)
+        job_uuid = await job_server.queue_export_job(config.id)
 
-    return {
+        logger.info(f"Export job {job.id} queued with UUID {job_uuid}")
-        "job_id": job.id,
+
-        "status": "pending",
+        return {
-        "message": "Export job queued successfully"
+            "job_id": job.id,
-    }
+            "job_uuid": job_uuid,
+            "status": "pending",
+            "message": "Export job queued successfully"
+        }
+
+    except Exception as e:
+        logger.error(f"Failed to queue export job {job.id}: {str(e)}")
+
+        # Update job status to failed
+        job.status = "failed"
+        job.error_message = f"Failed to queue job: {str(e)}"
+        db.commit()
+
+        raise HTTPException(
+            status_code=500,
+            detail=f"Failed to queue export job: {str(e)}"
+        )
 
 
 @router.get("/status/{job_id}")
@@ -266,3 +295,74 @@ async def list_export_jobs(
         "page": page,
         "per_page": per_page
     }
+
+
+@router.post("/sync")
+async def queue_sync_job(
+    sync_request: SyncJobRequest,
+    db: Session = Depends(get_db)
+) -> Dict[str, Any]:
+    """Queue a data synchronization job from AMO CRM"""
+
+    # Validate entity type
+    valid_entities = ["deals", "contacts", "companies", "pipelines", "users", "events"]
+    if sync_request.entity_type not in valid_entities:
+        raise HTTPException(
+            status_code=400,
+            detail=f"Invalid entity type. Must be one of: {', '.join(valid_entities)}"
+        )
+
+    try:
+        job_server = JobServer(broker)
+        job_uuid = await job_server.queue_sync_job(
+            entity_type=sync_request.entity_type,
+            limit=sync_request.limit,
+            page=sync_request.page
+        )
+
+        logger.info(f"Sync job queued for {sync_request.entity_type} with UUID {job_uuid}")
+
+        return {
+            "job_uuid": job_uuid,
+            "entity_type": sync_request.entity_type,
+            "status": "queued",
+            "message": f"Sync job for {sync_request.entity_type} queued successfully"
+        }
+
+    except Exception as e:
+        logger.error(f"Failed to queue sync job for {sync_request.entity_type}: {str(e)}")
+        raise HTTPException(
+            status_code=500,
+            detail=f"Failed to queue sync job: {str(e)}"
+        )
+
+
+@router.post("/refresh/{entity_type}")
+async def schedule_refresh_job(entity_type: str) -> Dict[str, Any]:
+    """Schedule a data refresh job for an entity type"""
+
+    # Validate entity type
+    valid_entities = ["deals", "contacts", "companies", "pipelines", "users", "events"]
+    if entity_type not in valid_entities:
+        raise HTTPException(
+            status_code=400,
+            detail=f"Invalid entity type. Must be one of: {', '.join(valid_entities)}"
+        )
+
+    try:
+        job_server = JobServer(broker)
+        await job_server.schedule_refresh_job(entity_type)
+
+        logger.info(f"Refresh job scheduled for {entity_type}")
+
+        return {
+            "entity_type": entity_type,
+            "message": f"Refresh job for {entity_type} scheduled successfully"
+        }
+
+    except Exception as e:
+        logger.error(f"Failed to schedule refresh job for {entity_type}: {str(e)}")
+        raise HTTPException(
+            status_code=500,
+            detail=f"Failed to schedule refresh job: {str(e)}"
+        )

scripts/install_faststream.py

@@ -0,0 +1,66 @@
+#!/usr/bin/env python3
+"""
+Installation script for FastStream dependencies.
+
+This script installs FastStream with Redis support and updates the project
+configuration to use FastStream instead of Celery.
+"""
+
+import subprocess
+import sys
+import logging
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+
+def run_command(command: str) -> bool:
+    """
+    Run a shell command and return success status.
+
+    Args:
+        command: Command to execute
+
+    Returns:
+        True if command succeeded, False otherwise
+    """
+    try:
+        logger.info(f"Running: {command}")
+        result = subprocess.run(
+            command.split(),
+            check=True,
+            capture_output=True,
+            text=True
+        )
+        logger.info(f"Success: {result.stdout}")
+        return True
+    except subprocess.CalledProcessError as e:
+        logger.error(f"Failed: {e.stderr}")
+        return False
+
+
+def main() -> None:
+    """Main installation function."""
+    logger.info("Installing FastStream dependencies...")
+
+    # Install FastStream with Redis support
+    if not run_command("uv add faststream[redis]"):
+        logger.error("Failed to install faststream[redis]")
+        sys.exit(1)
+
+    # Install TaskIQ-FastStream for scheduling
+    if not run_command("uv add taskiq-faststream"):
+        logger.error("Failed to install taskiq-faststream")
+        sys.exit(1)
+
+    # Remove Celery (optional)
+    logger.info("Removing Celery dependency...")
+    run_command("uv remove celery")  # Don't fail if this doesn't work
+
+    logger.info("FastStream installation completed successfully!")
+    logger.info("You can now start the FastStream worker with:")
+    logger.info("  uv run faststream run workers.broker:app")
+
+
+if __name__ == "__main__":
+    main()

scripts/start_services.py

@@ -0,0 +1,180 @@
+#!/usr/bin/env python3
+"""
+Service startup script for AMO CRM service with FastStream workers.
+
+This script starts Redis, FastAPI server, and FastStream workers in the
+correct order for development.
+"""
+
+import subprocess
+import sys
+import time
+import logging
+import signal
+import os
+from typing import List
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+# Global list to track running processes
+processes: List[subprocess.Popen] = []
+
+
+def cleanup_processes() -> None:
+    """Clean up all running processes."""
+    logger.info("Cleaning up processes...")
+    for process in processes:
+        if process.poll() is None:  # Process is still running
+            logger.info(f"Terminating process {process.pid}")
+            process.terminate()
+            try:
+                process.wait(timeout=5)
+            except subprocess.TimeoutExpired:
+                logger.warning(f"Force killing process {process.pid}")
+                process.kill()
+
+
+def signal_handler(signum, frame) -> None:
+    """Handle interrupt signals."""
+    logger.info("Received interrupt signal")
+    cleanup_processes()
+    sys.exit(0)
+
+
+def start_redis() -> bool:
+    """
+    Start Redis server.
+
+    Returns:
+        True if Redis is running, False otherwise
+    """
+    try:
+        # Check if Redis is already running
+        result = subprocess.run(
+            ["redis-cli", "ping"],
+            capture_output=True,
+            text=True,
+            timeout=5
+        )
+        if result.returncode == 0:
+            logger.info("Redis is already running")
+            return True
+    except (subprocess.TimeoutExpired, FileNotFoundError):
+        pass
+
+    logger.info("Starting Redis server...")
+    try:
+        process = subprocess.Popen(
+            ["redis-server", "--daemonize", "yes"],
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE
+        )
+
+        # Wait a moment for Redis to start
+        time.sleep(2)
+
+        # Check if Redis is now running
+        result = subprocess.run(
+            ["redis-cli", "ping"],
+            capture_output=True,
+            text=True,
+            timeout=5
+        )
+
+        if result.returncode == 0:
+            logger.info("Redis started successfully")
+            return True
+        else:
+            logger.error("Failed to start Redis")
+            return False
+
+    except FileNotFoundError:
+        logger.error("Redis not found. Please install Redis first.")
+        logger.info("On Ubuntu/Debian: sudo apt install redis-server")
+        logger.info("On macOS: brew install redis")
+        logger.info("On Windows: Download from https://redis.io/download")
+        return False
+
+
+def start_fastapi() -> subprocess.Popen:
+    """
+    Start FastAPI server.
+
+    Returns:
+        Process object for the FastAPI server
+    """
+    logger.info("Starting FastAPI server...")
+    process = subprocess.Popen([
+        "uv", "run", "uvicorn", "app:app",
+        "--host", "0.0.0.0",
+        "--port", "8000",
+        "--reload"
+    ])
+    processes.append(process)
+    return process
+
+
+def start_faststream_worker() -> subprocess.Popen:
+    """
+    Start FastStream worker.
+
+    Returns:
+        Process object for the FastStream worker
+    """
+    logger.info("Starting FastStream worker...")
+    process = subprocess.Popen([
+        "uv", "run", "faststream", "run", "workers.broker:app", "--reload"
+    ])
+    processes.append(process)
+    return process
+
+
+def main() -> None:
+    """Main function to start all services."""
+    # Set up signal handlers
+    signal.signal(signal.SIGINT, signal_handler)
+    signal.signal(signal.SIGTERM, signal_handler)
+
+    logger.info("Starting AMO CRM services...")
+
+    # Start Redis
+    if not start_redis():
+        logger.error("Failed to start Redis. Exiting.")
+        sys.exit(1)
+
+    # Start FastAPI server
+    fastapi_process = start_fastapi()
+    logger.info(f"FastAPI server started (PID: {fastapi_process.pid})")
+
+    # Wait a moment for FastAPI to start
+    time.sleep(3)
+
+    # Start FastStream worker
+    worker_process = start_faststream_worker()
+    logger.info(f"FastStream worker started (PID: {worker_process.pid})")
+
+    logger.info("All services started successfully!")
+    logger.info("FastAPI server: http://localhost:8000")
+    logger.info("API documentation: http://localhost:8000/docs")
+    logger.info("Press Ctrl+C to stop all services")
+
+    try:
+        # Wait for processes to complete
+        while True:
+            # Check if any process has died
+            for process in processes:
+                if process.poll() is not None:
+                    logger.error(f"Process {process.pid} has died")
+                    cleanup_processes()
+                    sys.exit(1)
+
+            time.sleep(1)
+
+    except KeyboardInterrupt:
+        logger.info("Received interrupt, shutting down...")
+        cleanup_processes()
+
+
+if __name__ == "__main__":
+    main()

servers/export_server.py

@@ -0,0 +1,282 @@
+"""
+Export server for processing Google Sheets export jobs.
+
+This module handles the actual processing of export jobs queued through
+the FastStream message broker, including data retrieval, formatting,
+and Google Sheets API integration.
+"""
+
+import logging
+from typing import Dict, Any, List, Optional
+from datetime import datetime
+from adapters.sqlite.database import get_db
+
+logger = logging.getLogger(__name__)
+
+
+class ExportServer:
+    """Server for processing Google Sheets export operations."""
+
+    def __init__(self):
+        """Initialize ExportServer with database and Google Sheets client."""
+        self.db = get_db
+        # TODO: Initialize Google Sheets client when implemented
+        # self.sheets_client = GoogleSheetsClient()
+
+    async def process_export_job(self, job_data: Dict[str, Any]) -> None:
+        """
+        Process a Google Sheets export job.
+
+        Args:
+            job_data: Dictionary containing job_id, configuration_id, and metadata
+        """
+        job_id = job_data["job_id"]
+        configuration_id = job_data["configuration_id"]
+
+        try:
+            logger.info(f"Starting export job {job_id} for configuration {configuration_id}")
+
+            # Update job status to running
+            await self._update_job_status(job_id, "running", started_at=datetime.utcnow())
+
+            # Get export configuration
+            config = await self._get_export_configuration(configuration_id)
+            if not config:
+                raise ValueError(f"Export configuration {configuration_id} not found")
+
+            # Process each enabled entity
+            total_records = 0
+            processed_entities = {}
+
+            for entity_type, mapping in config["entity_mappings"].items():
+                if mapping.get("is_enabled", False):
+                    logger.info(f"Exporting {entity_type} data")
+
+                    records_count = await self._export_entity(
+                        entity_type,
+                        config["sheet_id"],
+                        mapping,
+                        config.get("date_range_start"),
+                        config.get("date_range_end")
+                    )
+
+                    total_records += records_count
+                    processed_entities[entity_type] = {
+                        "processed": records_count,
+                        "total": records_count,
+                        "status": "completed"
+                    }
+
+                    logger.info(f"Exported {records_count} {entity_type} records")
+
+            # Update job as completed
+            await self._update_job_status(
+                job_id,
+                "completed",
+                completed_at=datetime.utcnow(),
+                records_processed=total_records,
+                total_records=total_records,
+                entities_processed=processed_entities
+            )
+
+            logger.info(f"Export job {job_id} completed successfully with {total_records} records")
+
+        except Exception as e:
+            logger.error(f"Export job {job_id} failed: {str(e)}")
+            await self._update_job_status(
+                job_id,
+                "failed",
+                completed_at=datetime.utcnow(),
+                error_message=str(e)
+            )
+            raise
+
+    async def _export_entity(
+        self,
+        entity_type: str,
+        sheet_id: str,
+        mapping: Dict[str, Any],
+        date_range_start: Optional[str] = None,
+        date_range_end: Optional[str] = None
+    ) -> int:
+        """
+        Export specific entity type to Google Sheets.
+
+        Args:
+            entity_type: Type of entity to export (deals, contacts, companies, etc.)
+            sheet_id: Google Sheets document ID
+            mapping: Field mapping configuration for this entity
+            date_range_start: Optional start date filter
+            date_range_end: Optional end date filter
+
+        Returns:
+            Number of records exported
+        """
+        try:
+            # Get entity data from database
+            data = await self._get_entity_data(entity_type, date_range_start, date_range_end)
+
+            if not data:
+                logger.warning(f"No data found for {entity_type}")
+                return 0
+
+            # Format data according to mapping
+            formatted_data = await self._format_data_for_export(data, mapping)
+
+            # Export to Google Sheets
+            await self._write_to_google_sheets(sheet_id, mapping["sheet_name"], formatted_data)
+
+            return len(data)
+
+        except Exception as e:
+            logger.error(f"Failed to export {entity_type}: {str(e)}")
+            raise
+
+    async def _get_entity_data(
+        self,
+        entity_type: str,
+        date_range_start: Optional[str] = None,
+        date_range_end: Optional[str] = None
+    ) -> List[Dict[str, Any]]:
+        """
+        Retrieve entity data from the database.
+
+        Args:
+            entity_type: Type of entity to retrieve
+            date_range_start: Optional start date filter
+            date_range_end: Optional end date filter
+
+        Returns:
+            List of entity records
+        """
+        # TODO: Implement database query based on entity type and date range
+        # This would query the appropriate table (amo_deals, amo_contacts, etc.)
+        # and join with custom fields if needed
+
+        logger.info(f"Retrieving {entity_type} data (date range: {date_range_start} to {date_range_end})")
+
+        # Placeholder implementation
+        return []
+
+    async def _format_data_for_export(
+        self,
+        data: List[Dict[str, Any]],
+        mapping: Dict[str, Any]
+    ) -> List[List[Any]]:
+        """
+        Format data according to the export mapping configuration.
+
+        Args:
+            data: Raw entity data from database
+            mapping: Field mapping configuration
+
+        Returns:
+            Formatted data ready for Google Sheets export
+        """
+        field_mapping = mapping.get("field_mapping", [])
+
+        # Sort field mapping by order
+        sorted_fields = sorted(field_mapping, key=lambda x: x.get("order", 0))
+
+        # Create header row
+        headers = [field["field_name"] for field in sorted_fields]
+        formatted_data = [headers]
+
+        # Format data rows
+        for record in data:
+            row = []
+            for field in sorted_fields:
+                field_name = field["field_name"]
+                value = record.get(field_name, "")
+
+                # Handle different data types and formatting
+                if isinstance(value, datetime):
+                    value = value.isoformat()
+                elif value is None:
+                    value = ""
+
+                row.append(value)
+
+            formatted_data.append(row)
+
+        logger.info(f"Formatted {len(data)} records with {len(headers)} columns")
+        return formatted_data
+
+    async def _write_to_google_sheets(
+        self,
+        sheet_id: str,
+        sheet_name: str,
+        data: List[List[Any]]
+    ) -> None:
+        """
+        Write formatted data to Google Sheets.
+
+        Args:
+            sheet_id: Google Sheets document ID
+            sheet_name: Name of the sheet tab
+            data: Formatted data to write
+        """
+        # TODO: Implement Google Sheets API integration
+        # This would use the Google Sheets client to write data
+
+        logger.info(f"Writing {len(data)} rows to sheet '{sheet_name}' in document {sheet_id}")
+
+        # Placeholder - would actually write to Google Sheets
+        logger.info(f"Successfully wrote data to Google Sheets (placeholder)")
+
+    async def _get_export_configuration(self, configuration_id: int) -> Optional[Dict[str, Any]]:
+        """
+        Get export configuration from database.
+
+        Args:
+            configuration_id: ID of the export configuration
+
+        Returns:
+            Export configuration data or None if not found
+        """
+        # TODO: Implement database query to get export configuration
+        # This would query the export_configuration and export_entity_mappings tables
+
+        logger.info(f"Retrieving export configuration {configuration_id}")
+
+        # Placeholder implementation
+        return {
+            "id": configuration_id,
+            "sheet_id": "placeholder_sheet_id",
+            "entity_mappings": {
+                "deals": {
+                    "sheet_name": "Deals",
+                    "is_enabled": True,
+                    "field_mapping": [
+                        {"field_name": "name", "column": "A", "order": 1},
+                        {"field_name": "price", "column": "B", "order": 2},
+                        {"field_name": "created_at", "column": "C", "order": 3}
+                    ]
+                }
+            }
+        }
+
+    async def _update_job_status(
+        self,
+        job_id: str,
+        status: str,
+        **kwargs
+    ) -> None:
+        """
+        Update job status in the database.
+
+        Args:
+            job_id: Unique identifier of the job
+            status: New status (pending, running, completed, failed)
+            **kwargs: Additional fields to update
+        """
+        # TODO: Implement database update
+        # This would update the export_jobs table
+
+        logger.info(f"Updating job {job_id} status to {status}")
+
+        update_data = {"status": status, **kwargs}
+        logger.debug(f"Job update data: {update_data}")
+
+        # Placeholder - would actually update database
+        pass

servers/job_server.py

@@ -0,0 +1,204 @@
+"""
+Job server for managing background job publishing and status tracking.
+
+This module provides the JobServer class for queuing various types of jobs
+using FastStream Redis broker and tracking their status in the database.
+"""
+
+import uuid
+import logging
+from datetime import datetime
+from typing import Dict, Any, Optional
+from faststream.redis import RedisBroker
+from adapters.sqlite.database import get_db
+
+logger = logging.getLogger(__name__)
+
+
+class JobServer:
+    """Server for managing background job operations."""
+
+    def __init__(self, broker: RedisBroker):
+        """
+        Initialize JobServer with Redis broker.
+
+        Args:
+            broker: FastStream Redis broker instance
+        """
+        self.broker = broker
+        self.db = get_db
+
+    async def queue_export_job(self, configuration_id: int) -> str:
+        """
+        Queue an export job for processing.
+
+        Args:
+            configuration_id: ID of the export configuration to process
+
+        Returns:
+            job_id: Unique identifier for the queued job
+        """
+        job_id = str(uuid.uuid4())
+        job_data = {
+            "job_id": job_id,
+            "job_type": "export",
+            "configuration_id": configuration_id,
+            "created_at": datetime.utcnow().isoformat(),
+            "status": "queued"
+        }
+
+        try:
+            # Create job record in database
+            await self._create_job_record(job_data)
+
+            # Publish to export-jobs channel
+            await self.broker.publish(job_data, "export-jobs")
+
+            logger.info(f"Export job {job_id} queued successfully")
+            return job_id
+
+        except Exception as e:
+            logger.error(f"Failed to queue export job: {str(e)}")
+            await self._update_job_status(job_id, "failed", error_message=str(e))
+            raise
+
+    async def queue_sync_job(
+        self,
+        entity_type: str,
+        limit: Optional[int] = None,
+        page: Optional[int] = None,
+        **kwargs
+    ) -> str:
+        """
+        Queue a data synchronization job.
+
+        Args:
+            entity_type: Type of entity to sync (deals, contacts, companies, etc.)
+            limit: Maximum number of records to sync
+            page: Page number for pagination
+            **kwargs: Additional parameters for the sync job
+
+        Returns:
+            job_id: Unique identifier for the queued job
+        """
+        job_id = str(uuid.uuid4())
+        sync_data = {
+            "job_id": job_id,
+            "job_type": "sync",
+            "entity_type": entity_type,
+            "parameters": {
+                "limit": limit,
+                "page": page,
+                **kwargs
+            },
+            "created_at": datetime.utcnow().isoformat(),
+            "status": "queued"
+        }
+
+        try:
+            # Create job record in database
+            await self._create_job_record(sync_data)
+
+            # Publish to sync-jobs channel
+            await self.broker.publish(sync_data, "sync-jobs")
+
+            logger.info(f"Sync job {job_id} for {entity_type} queued successfully")
+            return job_id
+
+        except Exception as e:
+            logger.error(f"Failed to queue sync job: {str(e)}")
+            await self._update_job_status(job_id, "failed", error_message=str(e))
+            raise
+
+    async def schedule_refresh_job(self, entity_type: str) -> None:
+        """
+        Schedule periodic data refresh for an entity type.
+
+        Args:
+            entity_type: Type of entity to refresh
+        """
+        try:
+            # Publish to refresh-jobs channel
+            await self.broker.publish(entity_type, "refresh-jobs")
+
+            logger.info(f"Refresh job for {entity_type} scheduled successfully")
+
+        except Exception as e:
+            logger.error(f"Failed to schedule refresh job for {entity_type}: {str(e)}")
+            raise
+
+    async def get_job_status(self, job_id: str) -> Optional[Dict[str, Any]]:
+        """
+        Get the status of a specific job.
+
+        Args:
+            job_id: Unique identifier of the job
+
+        Returns:
+            Job status information or None if not found
+        """
+        try:
+            # TODO: Implement database query to get job status
+            # This would query the export_jobs table
+            pass
+        except Exception as e:
+            logger.error(f"Failed to get job status for {job_id}: {str(e)}")
+            return None
+
+    async def list_jobs(
+        self,
+        status: Optional[str] = None,
+        job_type: Optional[str] = None,
+        limit: int = 20,
+        offset: int = 0
+    ) -> Dict[str, Any]:
+        """
+        List jobs with optional filtering.
+
+        Args:
+            status: Filter by job status (pending, running, completed, failed)
+            job_type: Filter by job type (export, sync)
+            limit: Maximum number of jobs to return
+            offset: Number of jobs to skip
+
+        Returns:
+            Dictionary containing jobs list and pagination info
+        """
+        try:
+            # TODO: Implement database query to list jobs
+            # This would query the export_jobs table with filters
+            pass
+        except Exception as e:
+            logger.error(f"Failed to list jobs: {str(e)}")
+            return {"jobs": [], "total": 0}
+
+    async def _create_job_record(self, job_data: Dict[str, Any]) -> None:
+        """
+        Create a job record in the database.
+
+        Args:
+            job_data: Job information to store
+        """
+        # TODO: Implement database insertion
+        # This would insert into the export_jobs table
+        pass
+
+    async def _update_job_status(
+        self,
+        job_id: str,
+        status: str,
+        error_message: Optional[str] = None,
+        **kwargs
+    ) -> None:
+        """
+        Update job status in the database.
+
+        Args:
+            job_id: Unique identifier of the job
+            status: New status (pending, running, completed, failed)
+            error_message: Error message if status is failed
+            **kwargs: Additional fields to update
+        """
+        # TODO: Implement database update
+        # This would update the export_jobs table
+        pass

servers/sync_server.py

@@ -0,0 +1,289 @@
+"""
+Sync server for processing AMO CRM data synchronization jobs.
+
+This module handles synchronization of data from AMO CRM API to the local
+SQLite database, including both full syncs and incremental updates.
+"""
+
+import logging
+from typing import Dict, Any, List, Optional
+from datetime import datetime, timezone
+from adapters.amocrm_client import AmoCRMClient
+from adapters.sqlite.database import get_db
+
+logger = logging.getLogger(__name__)
+
+
+class SyncServer:
+    """Server for processing AMO CRM data synchronization operations."""
+
+    def __init__(self):
+        """Initialize SyncServer with AMO CRM client and database."""
+        self.amocrm = AmoCRMClient()
+        self.db = get_db
+
+    async def process_sync_job(self, sync_data: Dict[str, Any]) -> None:
+        """
+        Process AMO CRM data synchronization job.
+
+        Args:
+            sync_data: Dictionary containing job_id, entity_type, and parameters
+        """
+        job_id = sync_data["job_id"]
+        entity_type = sync_data["entity_type"]
+        parameters = sync_data.get("parameters", {})
+
+        try:
+            logger.info(f"Starting sync job {job_id} for {entity_type}")
+
+            # Validate entity type
+            if not self._is_valid_entity_type(entity_type):
+                raise ValueError(f"Invalid entity type: {entity_type}")
+
+            # Fetch data from AMO CRM
+            data = await self.amocrm.fetch_entity_data(entity_type, **parameters)
+
+            if not data:
+                logger.warning(f"No data received from AMO CRM for {entity_type}")
+                return
+
+            # Process and store data
+            processed_count = await self._store_entity_data(entity_type, data)
+
+            logger.info(f"Sync job {job_id} completed: {processed_count} records processed")
+
+        except Exception as e:
+            logger.error(f"Sync job {job_id} failed: {str(e)}")
+            raise
+
+    async def refresh_entity_data(self, entity_type: str) -> None:
+        """
+        Refresh all data for an entity type with incremental updates.
+
+        Args:
+            entity_type: Type of entity to refresh (deals, contacts, companies, etc.)
+        """
+        try:
+            logger.info(f"Starting refresh for {entity_type} data")
+
+            # Validate entity type
+            if not self._is_valid_entity_type(entity_type):
+                raise ValueError(f"Invalid entity type: {entity_type}")
+
+            # Get last update timestamp for incremental sync
+            last_update = await self._get_last_update_timestamp(entity_type)
+
+            # Fetch updated data from AMO CRM
+            data = await self.amocrm.fetch_entity_data(
+                entity_type,
+                updated_at=last_update,
+                limit=250  # Process in batches
+            )
+
+            if data:
+                processed_count = await self._store_entity_data(entity_type, data)
+                logger.info(f"Refresh completed for {entity_type}: {processed_count} records updated")
+            else:
+                logger.info(f"No updates found for {entity_type}")
+
+        except Exception as e:
+            logger.error(f"Refresh failed for {entity_type}: {str(e)}")
+            raise
+
+    async def full_sync_entity(self, entity_type: str, batch_size: int = 250) -> int:
+        """
+        Perform a full synchronization of an entity type.
+
+        Args:
+            entity_type: Type of entity to sync
+            batch_size: Number of records to fetch per batch
+
+        Returns:
+            Total number of records processed
+        """
+        try:
+            logger.info(f"Starting full sync for {entity_type}")
+
+            total_processed = 0
+            page = 1
+
+            while True:
+                # Fetch batch of data
+                data = await self.amocrm.fetch_entity_data(
+                    entity_type,
+                    limit=batch_size,
+                    page=page
+                )
+
+                if not data:
+                    break
+
+                # Process batch
+                batch_processed = await self._store_entity_data(entity_type, data)
+                total_processed += batch_processed
+
+                logger.info(f"Processed batch {page}: {batch_processed} {entity_type} records")
+
+                # Check if we got less than batch_size (last page)
+                if len(data) < batch_size:
+                    break
+
+                page += 1
+
+            logger.info(f"Full sync completed for {entity_type}: {total_processed} total records")
+            return total_processed
+
+        except Exception as e:
+            logger.error(f"Full sync failed for {entity_type}: {str(e)}")
+            raise
+
+    async def _store_entity_data(self, entity_type: str, data: List[Dict[str, Any]]) -> int:
+        """
+        Store entity data in the database.
+
+        Args:
+            entity_type: Type of entity being stored
+            data: List of entity records from AMO CRM
+
+        Returns:
+            Number of records processed
+        """
+        try:
+            processed_count = 0
+
+            for record in data:
+                # Process main entity data
+                await self._store_main_entity(entity_type, record)
+
+                # Process custom fields
+                if "custom_fields_values" in record:
+                    await self._store_custom_fields(entity_type, record)
+
+                # Process relationships (embedded data)
+                if "_embedded" in record:
+                    await self._store_relationships(entity_type, record)
+
+                processed_count += 1
+
+            # Update last sync timestamp
+            await self._update_last_sync_timestamp(entity_type)
+
+            logger.info(f"Stored {processed_count} {entity_type} records in database")
+            return processed_count
+
+        except Exception as e:
+            logger.error(f"Failed to store {entity_type} data: {str(e)}")
+            raise
+
+    async def _store_main_entity(self, entity_type: str, record: Dict[str, Any]) -> None:
+        """
+        Store main entity record in the appropriate table.
+
+        Args:
+            entity_type: Type of entity
+            record: Entity record data
+        """
+        # TODO: Implement database insertion based on entity type
+        # This would insert/update records in tables like amo_deals, amo_contacts, etc.
+
+        entity_id = record.get("id")
+        logger.debug(f"Storing {entity_type} record ID: {entity_id}")
+
+        # Placeholder implementation
+        pass
+
+    async def _store_custom_fields(self, entity_type: str, record: Dict[str, Any]) -> None:
+        """
+        Store custom fields for an entity.
+
+        Args:
+            entity_type: Type of entity
+            record: Entity record containing custom fields
+        """
+        entity_id = record.get("id")
+        custom_fields = record.get("custom_fields_values", [])
+
+        for field in custom_fields:
+            # TODO: Implement custom field storage in amo_custom_fields table
+            field_id = field.get("field_id")
+            field_name = field.get("field_name", f"field_{field_id}")
+            values = field.get("values", [])
+
+            logger.debug(f"Storing custom field {field_name} for {entity_type} ID: {entity_id}")
+
+        # Placeholder implementation
+        pass
+
+    async def _store_relationships(self, entity_type: str, record: Dict[str, Any]) -> None:
+        """
+        Store entity relationships from embedded data.
+
+        Args:
+            entity_type: Type of entity
+            record: Entity record containing embedded relationships
+        """
+        entity_id = record.get("id")
+        embedded = record.get("_embedded", {})
+
+        for relation_type, relations in embedded.items():
+            if not isinstance(relations, list):
+                continue
+
+            for relation in relations:
+                # TODO: Implement relationship storage in junction tables
+                relation_id = relation.get("id")
+                is_main = relation.get("is_main", False)
+
+                logger.debug(f"Storing {relation_type} relationship: {entity_type} {entity_id} -> {relation_id}")
+
+        # Placeholder implementation
+        pass
+
+    async def _get_last_update_timestamp(self, entity_type: str) -> Optional[int]:
+        """
+        Get the timestamp of the last successful sync for an entity type.
+
+        Args:
+            entity_type: Type of entity
+
+        Returns:
+            Unix timestamp of last update or None for full sync
+        """
+        # TODO: Implement database query to get last sync timestamp
+        # This could be stored in a sync_status table or derived from entity updated_at
+
+        logger.debug(f"Getting last update timestamp for {entity_type}")
+
+        # Placeholder - return None for full sync
+        return None
+
+    async def _update_last_sync_timestamp(self, entity_type: str) -> None:
+        """
+        Update the last sync timestamp for an entity type.
+
+        Args:
+            entity_type: Type of entity
+        """
+        # TODO: Implement database update for last sync timestamp
+
+        current_time = datetime.now(timezone.utc)
+        logger.debug(f"Updating last sync timestamp for {entity_type} to {current_time}")
+
+        # Placeholder implementation
+        pass
+
+    def _is_valid_entity_type(self, entity_type: str) -> bool:
+        """
+        Validate if the entity type is supported.
+
+        Args:
+            entity_type: Type of entity to validate
+
+        Returns:
+            True if valid, False otherwise
+        """
+        valid_types = {
+            "deals", "contacts", "companies",
+            "users", "pipelines", "events"
+        }
+        return entity_type in valid_types

utils/config.py

@@ -1,5 +1,5 @@
 from pydantic_settings import BaseSettings
-from typing import Optional
+# Removed unused import
 
 
 class Settings(BaseSettings):
@@ -8,13 +8,13 @@ class Settings(BaseSettings):
 
     # AMO CRM API
     AMO_CRM_DOMAIN: str = "wecheap.amocrm.ru"
-    AMO_CRM_ACCESS_TOKEN: str
+    AMO_CRM_ACCESS_TOKEN: str = ""  # Make it optional with default empty string
 
     # Google Sheets API
-    GOOGLE_SERVICE_ACCOUNT_FILE: str
+    GOOGLE_SERVICE_ACCOUNT_FILE: str = ""  # Make it optional
     GOOGLE_SCOPES: str = "https://www.googleapis.com/auth/spreadsheets"
 
-    # Redis (for Celery)
+    # Redis (for FastStream message broker)
     REDIS_URL: str = "redis://localhost:6379/0"
 
     # API Settings

workers/broker.py

@@ -0,0 +1,132 @@
+"""
+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("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}"
+    )
+
+
+@broker.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}")
+
+
+@broker.before_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())

workers/middleware.py

@@ -0,0 +1,98 @@
+"""
+FastStream middleware for observability and error handling.
+
+This module provides middleware for adding metrics, tracing, and
+enhanced error handling to FastStream message processing.
+"""
+
+import logging
+from typing import Any, Awaitable, Callable
+from faststream import BaseMiddleware
+from faststream.redis import RedisPublishCommand
+from faststream.prometheus import PrometheusMiddleware
+from faststream.observability.middleware import TelemetryMiddleware
+
+logger = logging.getLogger(__name__)
+
+
+class ErrorHandlingMiddleware(BaseMiddleware):
+    """Middleware for enhanced error handling and logging."""
+
+    async def consume_scope(
+        self,
+        call_next: Callable[..., Awaitable[Any]],
+        msg: Any,
+    ) -> Any:
+        """
+        Handle message consumption with error logging.
+
+        Args:
+            call_next: Next middleware/handler in the chain
+            msg: Incoming message
+
+        Returns:
+            Result from the handler
+        """
+        try:
+            logger.debug(f"Processing message: {type(msg).__name__}")
+            result = await call_next(msg)
+            logger.debug("Message processed successfully")
+            return result
+
+        except Exception as e:
+            logger.error(f"Error processing message: {str(e)}", exc_info=True)
+
+            # TODO: Implement dead letter queue publishing for permanent failures
+            # For now, re-raise to let FastStream handle retries
+            raise
+
+
+class RedisPublishMiddleware(BaseMiddleware[RedisPublishCommand]):
+    """Middleware for Redis publishing operations."""
+
+    async def publish_scope(
+        self,
+        call_next: Callable[[RedisPublishCommand], Awaitable[Any]],
+        cmd: RedisPublishCommand,
+    ) -> Any:
+        """
+        Handle Redis publish operations with logging.
+
+        Args:
+            call_next: Next middleware/handler in the chain
+            cmd: Redis publish command
+
+        Returns:
+            Result from the publish operation
+        """
+        try:
+            logger.debug(f"Publishing to Redis: {cmd}")
+            result = await call_next(cmd)
+            logger.debug("Redis publish successful")
+            return result
+
+        except Exception as e:
+            logger.error(f"Redis publish failed: {str(e)}", exc_info=True)
+            raise
+
+
+def setup_middleware(broker):
+    """
+    Set up all middleware for the FastStream broker.
+
+    Args:
+        broker: FastStream Redis broker instance
+    """
+    # Add Prometheus metrics middleware
+    broker.add_middleware(PrometheusMiddleware)
+
+    # Add OpenTelemetry tracing middleware
+    broker.add_middleware(TelemetryMiddleware)
+
+    # Add custom error handling middleware
+    broker.add_middleware(ErrorHandlingMiddleware)
+
+    # Add custom Redis publish middleware
+    broker.add_middleware(RedisPublishMiddleware)
+
+    logger.info("FastStream middleware configured successfully")

workers/scheduler.py

@@ -0,0 +1,110 @@
+"""
+Task scheduler for periodic AMO CRM data refresh jobs.
+
+This module sets up scheduled tasks using TaskIQ-FastStream integration
+for periodic data synchronization and maintenance operations.
+"""
+
+import logging
+from taskiq_faststream import StreamScheduler
+from taskiq.schedule_sources import LabelScheduleSource
+from workers.broker import broker
+
+logger = logging.getLogger(__name__)
+
+# Schedule periodic data refresh jobs
+@broker.task(
+    message={"entity_type": "deals"},
+    channel="refresh-jobs",
+    schedule=[{"cron": "0 */6 * * *"}]  # Every 6 hours
+)
+async def scheduled_deals_refresh():
+    """Scheduled refresh for deals data."""
+    logger.info("Scheduled deals refresh triggered")
+
+
+@broker.task(
+    message={"entity_type": "contacts"},
+    channel="refresh-jobs",
+    schedule=[{"cron": "0 */4 * * *"}]  # Every 4 hours
+)
+async def scheduled_contacts_refresh():
+    """Scheduled refresh for contacts data."""
+    logger.info("Scheduled contacts refresh triggered")
+
+
+@broker.task(
+    message={"entity_type": "companies"},
+    channel="refresh-jobs",
+    schedule=[{"cron": "0 */8 * * *"}]  # Every 8 hours
+)
+async def scheduled_companies_refresh():
+    """Scheduled refresh for companies data."""
+    logger.info("Scheduled companies refresh triggered")
+
+
+@broker.task(
+    message={"entity_type": "users"},
+    channel="refresh-jobs",
+    schedule=[{"cron": "0 */12 * * *"}]  # Every 12 hours
+)
+async def scheduled_users_refresh():
+    """Scheduled refresh for users data."""
+    logger.info("Scheduled users refresh triggered")
+
+
+@broker.task(
+    message={"entity_type": "pipelines"},
+    channel="refresh-jobs",
+    schedule=[{"cron": "0 */24 * * *"}]  # Daily
+)
+async def scheduled_pipelines_refresh():
+    """Scheduled refresh for pipelines data."""
+    logger.info("Scheduled pipelines refresh triggered")
+
+
+@broker.task(
+    message={"entity_type": "events"},
+    channel="refresh-jobs",
+    schedule=[{"cron": "0 */2 * * *"}]  # Every 2 hours
+)
+async def scheduled_events_refresh():
+    """Scheduled refresh for events data."""
+    logger.info("Scheduled events refresh triggered")
+
+
+# Initialize scheduler
+scheduler = StreamScheduler(
+    broker=broker,
+    sources=[LabelScheduleSource(broker)]
+)
+
+
+async def start_scheduler():
+    """Start the task scheduler."""
+    logger.info("Starting FastStream task scheduler")
+    await scheduler.startup()
+
+
+async def stop_scheduler():
+    """Stop the task scheduler."""
+    logger.info("Stopping FastStream task scheduler")
+    await scheduler.shutdown()
+
+
+if __name__ == "__main__":
+    # This allows running the scheduler directly
+    import asyncio
+
+    async def main():
+        await start_scheduler()
+        try:
+            # Keep the scheduler running
+            while True:
+                await asyncio.sleep(60)
+        except KeyboardInterrupt:
+            logger.info("Scheduler interrupted by user")
+        finally:
+            await stop_scheduler()
+
+    asyncio.run(main())