amo-server/docs/amocrm-service-design.md

# AMO CRM Data Collection Service - Design Document

## Overview

This service collects data from AMO CRM API (wecheap.amocrm.ru) and stores it in SQLite database for further export to Google Sheets. The service handles AMO CRM's HAL+JSON format and maintains proper relationships between entities.

**Key Features:**
- Simple API without authentication
- Direct AMO CRM integration with long-term tokens
- Real-time data fetching and testing
- Unified export configuration for all entities
- Background job processing for exports

## Architecture

```
┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐
│   AMO CRM API   │    │   Our Service   │    │  Google Sheets  │
│                 │    │                 │    │                 │
│ HAL+JSON Format │◄──►│ SQLite Database │───►│   Export API    │
│                 │    │                 │    │                 │
└─────────────────┘    └─────────────────┘    └─────────────────┘
```

### Components
- **Adapters**: Database connections, AMO CRM API client, Google Sheets client
- **Routers**: API endpoints (no authentication required)
- **Servers**: Business logic for data processing and export
- **Workers**: Background data collection and processing
- **Scripts**: Utilities for fetching real AMO CRM data

## Data Entities

### 1. Deals (Сделки)
- Primary entity representing sales opportunities
- Links to: Contacts, Companies, Users, Pipelines
- Contains custom fields and standard fields

### 2. Contacts (Контакты)
- Individual people in the CRM
- Links to: Companies, Deals, Users
- Contains communication history

### 3. Companies (Компании)
- Organizations in the CRM
- Links to: Contacts, Deals, Users
- Contains company-specific custom fields

### 4. Pipelines (Воронки)
- Sales process definitions
- Contains stages and automation rules
- Referenced by Deals

### 5. Users (Пользователи)
- CRM users (employees)
- Referenced by all other entities as owners/responsible persons

### 6. Events (События)
- Activity logs (filtered types only):
  - `incoming_call`
  - `outgoing_call`
  - `lead_status_changed`
- Links to related entities

## Database Schema

### Core Tables

```sql
-- Users table
CREATE TABLE amo_users (
    id INTEGER PRIMARY KEY,
    name TEXT NOT NULL,
    email TEXT,
    is_active BOOLEAN DEFAULT TRUE,
    created_at INTEGER,
    updated_at INTEGER,
    raw_data JSON -- Original AMO CRM data
);

-- Pipelines table
CREATE TABLE amo_pipelines (
    id INTEGER PRIMARY KEY,
    name TEXT NOT NULL,
    sort INTEGER,
    is_main BOOLEAN DEFAULT FALSE,
    is_unsorted BOOLEAN DEFAULT FALSE,
    is_archive BOOLEAN DEFAULT FALSE,
    account_id INTEGER,
    created_at INTEGER,
    updated_at INTEGER,
    raw_data JSON
);

-- Pipeline stages
CREATE TABLE amo_pipeline_stages (
    id INTEGER PRIMARY KEY,
    pipeline_id INTEGER REFERENCES amo_pipelines(id),
    name TEXT NOT NULL,
    sort INTEGER,
    is_editable BOOLEAN DEFAULT TRUE,
    color TEXT,
    created_at INTEGER,
    updated_at INTEGER,
    raw_data JSON
);

-- Companies table
CREATE TABLE amo_companies (
    id INTEGER PRIMARY KEY,
    name TEXT NOT NULL,
    responsible_user_id INTEGER REFERENCES amo_users(id),
    group_id INTEGER,
    created_by INTEGER REFERENCES amo_users(id),
    updated_by INTEGER REFERENCES amo_users(id),
    created_at INTEGER,
    updated_at INTEGER,
    closest_task_at INTEGER,
    is_deleted BOOLEAN DEFAULT FALSE,
    raw_data JSON
);

-- Contacts table
CREATE TABLE amo_contacts (
    id INTEGER PRIMARY KEY,
    name TEXT NOT NULL,
    first_name TEXT,
    last_name TEXT,
    responsible_user_id INTEGER REFERENCES amo_users(id),
    group_id INTEGER,
    created_by INTEGER REFERENCES amo_users(id),
    updated_by INTEGER REFERENCES amo_users(id),
    created_at INTEGER,
    updated_at INTEGER,
    closest_task_at INTEGER,
    is_deleted BOOLEAN DEFAULT FALSE,
    raw_data JSON
);

-- Deals table
CREATE TABLE amo_deals (
    id INTEGER PRIMARY KEY,
    name TEXT NOT NULL,
    price INTEGER DEFAULT 0,
    responsible_user_id INTEGER REFERENCES amo_users(id),
    group_id INTEGER,
    status_id INTEGER REFERENCES amo_pipeline_stages(id),
    pipeline_id INTEGER REFERENCES amo_pipelines(id),
    loss_reason_id INTEGER,
    created_by INTEGER REFERENCES amo_users(id),
    updated_by INTEGER REFERENCES amo_users(id),
    closed_at INTEGER,
    created_at INTEGER,
    updated_at INTEGER,
    closest_task_at INTEGER,
    is_deleted BOOLEAN DEFAULT FALSE,
    raw_data JSON
);

-- Events table
CREATE TABLE amo_events (
    id INTEGER PRIMARY KEY,
    type TEXT NOT NULL CHECK (type IN ('incoming_call', 'outgoing_call', 'lead_status_changed')),
    entity_id INTEGER,
    entity_type TEXT,
    created_by INTEGER REFERENCES amo_users(id),
    created_at INTEGER,
    value_after JSON,
    value_before JSON,
    account_id INTEGER,
    raw_data JSON
);

-- Relations tables
CREATE TABLE amo_deal_contacts (
    deal_id INTEGER REFERENCES amo_deals(id),
    contact_id INTEGER REFERENCES amo_contacts(id),
    is_main BOOLEAN DEFAULT FALSE,
    PRIMARY KEY (deal_id, contact_id)
);

CREATE TABLE amo_deal_companies (
    deal_id INTEGER REFERENCES amo_deals(id),
    company_id INTEGER REFERENCES amo_companies(id),
    is_main BOOLEAN DEFAULT FALSE,
    PRIMARY KEY (deal_id, company_id)
);

CREATE TABLE amo_contact_companies (
    contact_id INTEGER REFERENCES amo_contacts(id),
    company_id INTEGER REFERENCES amo_companies(id),
    is_main BOOLEAN DEFAULT FALSE,
    PRIMARY KEY (contact_id, company_id)
);
```

### Custom Fields Handling

```sql
-- Universal custom fields table
CREATE TABLE amo_custom_fields (
    id INTEGER PRIMARY KEY,
    entity_type TEXT NOT NULL, -- 'deals', 'contacts', 'companies'
    entity_id INTEGER NOT NULL,
    field_id INTEGER NOT NULL,
    field_name TEXT NOT NULL,
    field_type TEXT NOT NULL, -- 'text', 'numeric', 'checkbox', 'select', 'multiselect', 'date', 'url', 'textarea'
    field_value TEXT,
    field_value_numeric REAL,
    field_value_date INTEGER,
    is_custom BOOLEAN DEFAULT TRUE,
    created_at INTEGER DEFAULT (strftime('%s', 'now')),
    updated_at INTEGER DEFAULT (strftime('%s', 'now'))
);

-- Indexes for performance
CREATE INDEX idx_custom_fields_entity ON amo_custom_fields(entity_type, entity_id);
CREATE INDEX idx_custom_fields_field ON amo_custom_fields(field_id);
CREATE INDEX idx_custom_fields_name ON amo_custom_fields(field_name);
```

### Google Sheets Export Configuration

```sql
-- Single unified export configuration
CREATE TABLE export_configuration (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT NOT NULL,
    sheet_id TEXT NOT NULL,
    date_range_start INTEGER,
    date_range_end INTEGER,
    is_active BOOLEAN DEFAULT TRUE,
    created_at INTEGER DEFAULT (strftime('%s', 'now')),
    updated_at INTEGER DEFAULT (strftime('%s', 'now'))
);

-- Entity-specific sheet mappings within the configuration
CREATE TABLE export_entity_mappings (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    configuration_id INTEGER REFERENCES export_configuration(id),
    entity_type TEXT NOT NULL, -- 'deals', 'contacts', 'companies', 'pipelines', 'users', 'events'
    sheet_name TEXT NOT NULL, -- Sheet tab name for this entity
    field_mapping JSON NOT NULL, -- [{"field_name": "name", "column": "A", "order": 1}]
    is_enabled BOOLEAN DEFAULT TRUE,
    created_at INTEGER DEFAULT (strftime('%s', 'now')),
    updated_at INTEGER DEFAULT (strftime('%s', 'now'))
);

-- Export jobs tracking
CREATE TABLE export_jobs (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    configuration_id INTEGER REFERENCES export_configuration(id),
    status TEXT NOT NULL DEFAULT 'pending', -- 'pending', 'running', 'completed', 'failed'
    records_processed INTEGER DEFAULT 0,
    total_records INTEGER DEFAULT 0,
    error_message TEXT,
    started_at INTEGER,
    completed_at INTEGER,
    created_at INTEGER DEFAULT (strftime('%s', 'now'))
);
```

## API Endpoints

**Base URL**: `http://localhost:8000/api/v1`
**Authentication**: None required

### 1. Entity Management

#### GET /entities
List all entities with statistics
```json
{
  "entities": [
    {
      "type": "deals",
      "count": 1500,
      "last_updated": "2024-01-15T10:30:00Z",
      "date_range": {
        "earliest": "2017-01-01T00:00:00Z",
        "latest": "2024-01-15T10:30:00Z"
      }
    },
    {
      "type": "contacts",
      "count": 3200,
      "last_updated": "2024-01-15T09:45:00Z",
      "date_range": {
        "earliest": "2017-03-15T00:00:00Z",
        "latest": "2024-01-15T09:45:00Z"
      }
    }
  ]
}
```

#### GET /entities/{entity_type}/fields
List fields with statistics and examples
```json
{
  "entity_type": "deals",
  "fields": [
    {
      "name": "name",
      "type": "text",
      "is_custom": false,
      "usage_count": 1500,
      "examples": ["Deal #1", "Important client deal", "Q4 opportunity"],
      "null_count": 0
    },
    {
      "name": "custom_priority",
      "type": "select",
      "is_custom": true,
      "field_id": 123456,
      "usage_count": 1200,
      "examples": ["High", "Medium", "Low"],
      "null_count": 300,
      "possible_values": ["High", "Medium", "Low"]
    }
  ]
}
```

### 2. AMO CRM Integration

#### GET /amocrm/info
Get AMO CRM connection information
```json
{
  "domain": "wecheap.amocrm.ru",
  "has_token": true,
  "base_url": "https://wecheap.amocrm.ru/api/v4",
  "supported_entities": [
    "users", "pipelines", "companies",
    "contacts", "deals", "events"
  ]
}
```

#### GET /amocrm/fetch/{entity_type}
Fetch data directly from AMO CRM API
- **Parameters**: `limit` (default: 10), `page` (default: 1)
- **Supported entities**: users, pipelines, companies, contacts, deals, events
- **Returns**: Raw AMO CRM HAL+JSON response

#### GET /amocrm/fetch/custom_fields/{entity_type}
Fetch custom fields metadata from AMO CRM
- **Supported entities**: leads, contacts, companies
- **Returns**: Custom fields definitions with enums

#### GET /amocrm/fetch/all
Fetch all data from AMO CRM (limited amounts for testing)
- **Returns**: Complete data set from all entities

#### POST /amocrm/sync/{entity_type}
Fetch data from AMO CRM and store it in database
- **Parameters**: `limit` (default: 250), `page` (default: 1)
- **Returns**: Sync status and fetched data

### 3. Google Sheets Export

#### POST /export/configure
Create or update unified export configuration for all entities
```json
{
  "name": "Q1 2024 AMO CRM Export",
  "sheet_id": "1BxiMVs0XRA5nFMdKvBdBZjgmUUqptlbs74OgvE2upms",
  "date_range_start": "2024-01-01T00:00:00Z",
  "date_range_end": "2024-03-31T23:59:59Z",
  "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},
        {"field_name": "custom_priority", "column": "D", "order": 4}
      ]
    },
    "contacts": {
      "sheet_name": "Contacts",
      "is_enabled": true,
      "field_mapping": [
        {"field_name": "name", "column": "A", "order": 1},
        {"field_name": "first_name", "column": "B", "order": 2},
        {"field_name": "last_name", "column": "C", "order": 3},
        {"field_name": "created_at", "column": "D", "order": 4}
      ]
    },
    "companies": {
      "sheet_name": "Companies",
      "is_enabled": true,
      "field_mapping": [
        {"field_name": "name", "column": "A", "order": 1},
        {"field_name": "created_at", "column": "B", "order": 2}
      ]
    },
    "pipelines": {
      "sheet_name": "Pipelines",
      "is_enabled": false,
      "field_mapping": [
        {"field_name": "name", "column": "A", "order": 1},
        {"field_name": "is_main", "column": "B", "order": 2}
      ]
    },
    "users": {
      "sheet_name": "Users",
      "is_enabled": false,
      "field_mapping": [
        {"field_name": "name", "column": "A", "order": 1},
        {"field_name": "email", "column": "B", "order": 2}
      ]
    },
    "events": {
      "sheet_name": "Events",
      "is_enabled": true,
      "field_mapping": [
        {"field_name": "type", "column": "A", "order": 1},
        {"field_name": "entity_type", "column": "B", "order": 2},
        {"field_name": "created_at", "column": "C", "order": 3}
      ]
    }
  }
}
```

#### POST /export/start
Start async export job
```json
{
  "configuration_id": 1
}
```
Response:
```json
{
  "job_id": 42,
  "status": "pending",
  "message": "Export job queued successfully"
}
```

#### GET /export/status/{job_id}
Get export job status
```json
{
  "job_id": 42,
  "status": "running",
  "progress": {
    "records_processed": 2100,
    "total_records": 4200,
    "percentage": 50,
    "entities": {
      "deals": {"processed": 1500, "total": 1500, "status": "completed"},
      "contacts": {"processed": 600, "total": 2200, "status": "running"},
      "companies": {"processed": 0, "total": 300, "status": "pending"},
      "events": {"processed": 0, "total": 200, "status": "pending"}
    }
  },
  "started_at": "2024-01-15T10:30:00Z",
  "estimated_completion": "2024-01-15T10:35:00Z"
}
```

#### GET /export/jobs
List all export jobs
```json
{
  "jobs": [
    {
      "job_id": 42,
      "configuration_name": "Q1 2024 AMO CRM Export",
      "status": "completed",
      "records_processed": 4200,
      "total_records": 4200,
      "entities_processed": {
        "deals": 1500,
        "contacts": 2200,
        "companies": 300,
        "events": 200
      },
      "started_at": "2024-01-15T10:30:00Z",
      "completed_at": "2024-01-15T10:35:00Z"
    }
  ],
  "total": 1,
  "page": 1,
  "per_page": 20
}
```

### 4. Data Ingestion (Worker API)

#### POST /data/{entity_type}
Put data into database (used by workers)
```json
{
  "data": [
    {
      "id": 123456,
      "name": "Important Deal",
      "price": 50000,
      "responsible_user_id": 789,
      "pipeline_id": 456,
      "status_id": 142,
      "created_at": 1640995200,
      "updated_at": 1704067200,
      "custom_fields": [
        {
          "field_id": 123456,
          "field_name": "Priority",
          "field_type": "select",
          "values": [{"value": "High", "enum_id": 1}]
        }
      ],
      "_embedded": {
        "contacts": [{"id": 111, "is_main": true}],
        "companies": [{"id": 222, "is_main": true}]
      }
    }
  ],
  "sync_mode": "upsert" // "insert", "upsert", "replace"
}
```

## Data Processing Logic

### 1. HAL+JSON Parsing
- Extract main entity data from root level
- Process `_embedded` section for related entities
- Handle `_links` section for additional relationships
- Parse custom fields from `custom_fields_values` array

### 2. Date Inference
- Convert Unix timestamps to readable dates
- Validate dates are between 2017-2026
- Handle null/missing dates appropriately

### 3. Custom Fields Processing
```python
def process_custom_field(field_data, entity_type, entity_id):
    """
    Process custom field from AMO CRM format
    """
    field_record = {
        'entity_type': entity_type,
        'entity_id': entity_id,
        'field_id': field_data['field_id'],
        'field_name': field_data['field_name'],
        'field_type': field_data['field_type'],
        'is_custom': True
    }

    # Handle different field types
    if field_data['field_type'] in ['text', 'textarea', 'url']:
        field_record['field_value'] = field_data['values'][0]['value']
    elif field_data['field_type'] == 'numeric':
        field_record['field_value_numeric'] = float(field_data['values'][0]['value'])
        field_record['field_value'] = str(field_data['values'][0]['value'])
    elif field_data['field_type'] == 'date':
        field_record['field_value_date'] = int(field_data['values'][0]['value'])
        field_record['field_value'] = str(field_data['values'][0]['value'])
    elif field_data['field_type'] in ['select', 'multiselect']:
        # Handle enum values
        values = [v['value'] for v in field_data['values']]
        field_record['field_value'] = ', '.join(values) if len(values) > 1 else values[0]

    return field_record
```

### 4. Relationship Mapping
- Extract relationships from `_embedded` section
- Create junction table records for many-to-many relationships
- Handle `is_main` flags for primary relationships

## Implementation Status

### ✅ Completed Features

#### Phase 1: Core Infrastructure
- ✅ FastAPI application structure with uv
- ✅ SQLite adapter with migrations (Alembic)
- ✅ Complete entity models with relationships
- ✅ AMO CRM API client for wecheap.amocrm.ru

#### Phase 2: Data Ingestion
- ✅ HAL+JSON parser implementation
- ✅ Data ingestion endpoints for all entities
- ✅ Custom fields processing with type detection
- ✅ Relationship mapping (many-to-many)

#### Phase 3: API Development
- ✅ Entity listing endpoints with statistics
- ✅ Field statistics endpoints with examples
- ✅ Export configuration management
- ✅ Direct AMO CRM integration endpoints

#### Testing & Utilities
- ✅ Real AMO CRM response examples
- ✅ Comprehensive test suite
- ✅ Data fetching script for real responses
- ✅ API documentation (OpenAPI/Swagger)

### 🔄 In Progress / TODO

#### Phase 4: Google Sheets Integration
- 🔄 Google Sheets API client implementation
- 🔄 Actual export processing
- ✅ Export job management structure
- ✅ Job status tracking

#### Phase 5: Workers & Background Processing
- 🔄 FastStream worker implementation
- 🔄 Data synchronization workers with Redis broker
- 🔄 Error handling and retry logic
- 🔄 Monitoring and logging

## Technology Stack

- **Framework**: FastAPI (with uvicorn)
- **Package Manager**: uv (modern Python package management)
- **Database**: SQLite with SQLAlchemy 2.0 ORM
- **Migrations**: Alembic
- **HTTP Client**: httpx for AMO CRM API
- **Data Validation**: Pydantic v2
- **Testing**: pytest with real AMO CRM fixtures
- **Background Jobs**: FastStream with Redis broker
- **Google Sheets**: google-api-python-client (planned)
- **Development**: Black, isort, mypy, flake8

## Configuration

### Environment Variables
```env
# Database
DATABASE_URL=sqlite:///./amo_data.db

# AMO CRM API (wecheap.amocrm.ru)
AMO_CRM_DOMAIN=wecheap.amocrm.ru
AMO_CRM_ACCESS_TOKEN=your-longterm-access-token

# Google Sheets API (optional)
GOOGLE_SERVICE_ACCOUNT_FILE=path/to/service-account.json
GOOGLE_SCOPES=https://www.googleapis.com/auth/spreadsheets

# Redis (for FastStream message broker)
REDIS_URL=redis://localhost:6379/0

# API Settings
API_V1_STR=/api/v1
LOG_LEVEL=INFO
```

### Key Features
- **No Authentication**: Simple public API
- **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

### 1. Setup
```bash
# Clone and setup
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 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
```

### 3. Test AMO CRM Connection
```bash
# Check connection
curl http://localhost:8000/api/v1/amocrm/info

# Fetch real data
curl http://localhost:8000/api/v1/amocrm/fetch/users?limit=5

# Get all data for testing
uv run python scripts/fetch_amocrm_data.py
```

### 4. Example API Usage
```bash
# List entities
curl http://localhost:8000/api/v1/entities/

# Get deal fields
curl http://localhost:8000/api/v1/entities/deals/fields

# Fetch deals from AMO CRM
curl http://localhost:8000/api/v1/amocrm/fetch/deals?limit=10

# Create export configuration
curl -X POST http://localhost:8000/api/v1/export/configure \
  -H "Content-Type: application/json" \
  -d '{"name": "Test Export", "sheet_id": "123", "entity_mappings": {...}}'
```

## Security Considerations

1. **API Keys**: Store AMO CRM tokens securely in environment variables
2. **No Authentication**: Current API has no authentication (as requested)
3. **Rate Limiting**: Consider implementing rate limiting for production
4. **Data Validation**: All incoming data is validated with Pydantic
5. **CORS**: Currently allows all origins (configure for production)

## Performance Considerations

1. **Database Indexing**: Proper indexes on foreign keys and search fields
2. **Batch Processing**: Process data in batches to avoid memory issues
3. **Caching**: Use Redis for frequently accessed data
4. **Async Processing**: Use async/await for I/O operations
5. **Pagination**: Implement cursor-based pagination for large datasets

## Monitoring & Logging

1. **Application Metrics**: Track API response times, error rates
2. **Data Metrics**: Monitor data freshness, sync status
3. **Export Metrics**: Track export job success rates, processing times
4. **Error Tracking**: Comprehensive error logging and alerting
5. **Health Checks**: Implement health check endpoints

## Error Handling

1. **AMO CRM API Errors**: Handle rate limits, timeouts, authentication errors
2. **Google Sheets API Errors**: Handle quota limits, permission errors (planned)
3. **Data Validation Errors**: Proper error messages for invalid data
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.postgres.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.postgres.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
- **Purpose**: Fetch real AMO CRM responses for testing
- **Usage**: `uv run python scripts/fetch_amocrm_data.py`
- **Output**:
  - Individual JSON files in `tests/fixtures/real_responses/`
  - Python fixtures in `tests/fixtures/real_amocrm_responses.py`
  - Summary of fetched data

### Features
- Fetches all entity types from wecheap.amocrm.ru
- Includes custom fields metadata
- Proper error handling with troubleshooting tips
- Generates both JSON and Python fixtures

## Testing

### Test Structure
- **Unit Tests**: `tests/test_amocrm_examples.py` - Test AMO CRM response structure
- **API Tests**: `tests/test_api.py` - Test all API endpoints with real data
- **Fixtures**: Real AMO CRM responses in multiple formats

### Running Tests
```bash
# Run all tests
uv run pytest

# Run with coverage
uv run pytest --cov=. --cov-report=html

# Run specific test file
uv run pytest tests/test_api.py -v
```

## Development Workflow

### Code Quality
```bash
# Format code
uv run black .
uv run isort .

# Lint code
uv run flake8

# Type checking
uv run mypy .
```

### Database Migrations
```bash
# Generate migration
alembic revision --autogenerate -m "Description"

# Apply migrations
alembic upgrade head
```