Перейти к основному содержимому

Suppliers Layer — Слой поставщиков данных

Версия: 1.0
Дата: 19.04.2026
Статус: Черновик

Обзор​

Suppliers Layer обеспечивает получение сырых данных от различных источников согласно разделу "### 1. Suppliers (Поставщики данных)" в Архитектура платформы. Реализует принцип: "Для каждого поставщика — отдельный адаптер. Ядро системы не подстраивается под форматы поставщиков." Интегрируется с Ingestion Layer через SupplierMessage структуру из Ingestion Layer и поток "Supplier API → Message Queue" из Архитектура платформы.

Архитектура адаптеров: См. диаграмму vitrip_suppliers_architecture.jpg — схема адаптеров для Stuba, HomeToGo и других поставщиков с интеграцией в NATS.

Архитектура Suppliers Layer vitrip.store

Интеграции поставщиков: См. диаграмму vitrip_suppliers_integration.jpg — детальные схемы API интеграций, аутентификации и форматов данных.

Интеграции поставщиков vitrip.store

Поток синхронизации: См. диаграмму vitrip_suppliers_sync_flow.jpg — процесс получения данных от поставщиков и отправки в NATS согласно потоку обновления данных.

Поток синхронизации поставщиков vitrip.store

Архитектура адаптеров​

Принципы проектирования​

Adapter Pattern: Каждый поставщик имеет отдельный адаптер согласно принципу из Архитектура платформы: "Для каждого поставщика — отдельный адаптер. Ядро системы не подстраивается под форматы поставщиков."

// Базовый trait для всех supplier адаптеров
pub trait SupplierAdapter: Send + Sync + 'static {
    type Config: Clone + Send + Sync;
    type RawData: Send + Sync;
    
    fn name(&self) -> &'static str;
    fn fetch_data(&self, config: &Self::Config) -> Result<Vec<Self::RawData>, SupplierError>;
    fn convert_to_message(&self, data: Self::RawData) -> Result<SupplierMessage, ConversionError>;
    fn health_check(&self) -> Result<HealthStatus, SupplierError>;
}

// Интеграция с NATS из reference/ingestion.md  
pub struct SupplierGateway {
    nats_client: nats::Connection,
    adapters: HashMap<String, Box<dyn SupplierAdapter>>,
    scheduler: CronScheduler,
    metrics: Arc<SupplierMetrics>,
}

NATS Integration​

Интеграция с Ingestion Layer: Использует SupplierMessage структуру из Ingestion Layer для отправки данных в NATS согласно потоку "Supplier API → Message Queue" из Архитектура платформы.

// Использует SupplierMessage из reference/ingestion.md
impl SupplierGateway {
    pub async fn publish_to_nats(&self, message: SupplierMessage) -> Result<(), NatsError> {
        let subject = format!("suppliers.{}.{}", 
            message.supplier_id, 
            message.message_type.to_string().to_lowercase()
        );
        
        let payload = serde_json::to_vec(&message)?;
        
        self.nats_client.publish(&subject, payload).await?;
        
        // Метрики для мониторинга
        self.metrics.messages_published
            .with_label_values(&[&message.supplier_id, &message.message_type.to_string()])
            .inc();
            
        Ok(())
    }
}

// MessageType из reference/ingestion.md
use crate::ingestion::{SupplierMessage, MessageType};

Stuba Adapter​

API Integration​

Источник данных: XML/FTP массовые выгрузки согласно Архитектура платформы: "Stuba — XML/FTP массовые выгрузки отелей".

pub struct StubaAdapter {
    ftp_client: FtpClient,
    xml_parser: StubaXmlParser,
    config: StubaConfig,
    rate_limiter: RateLimiter,
}

#[derive(Clone)]
pub struct StubaConfig {
    pub ftp_host: String,
    pub ftp_username: String,
    pub ftp_password: String,
    pub download_path: PathBuf,
    pub file_patterns: Vec<String>,    // ["hotels_*.xml", "prices_*.xml"]
    pub sync_interval: Duration,       // Каждые 4 часа
    pub batch_size: usize,            // 1000 отелей за раз
}

impl SupplierAdapter for StubaAdapter {
    type Config = StubaConfig;
    type RawData = StubaHotelData;
    
    fn name(&self) -> &'static str {
        "stuba"
    }
    
    fn fetch_data(&self, config: &Self::Config) -> Result<Vec<Self::RawData>, SupplierError> {
        // 1. Подключение к FTP серверу
        self.ftp_client.connect(&config.ftp_host)?;
        self.ftp_client.login(&config.ftp_username, &config.ftp_password)?;
        
        // 2. Поиск новых XML файлов
        let files = self.find_new_xml_files(&config.file_patterns)?;
        
        let mut all_hotels = Vec::new();
        
        for file_path in files {
            // 3. Загрузка XML файла
            let xml_content = self.download_xml_file(&file_path)?;
            
            // 4. Парсинг XML в структуры
            let hotels = self.xml_parser.parse_hotels(&xml_content)?;
            all_hotels.extend(hotels);
            
            // Rate limiting между файлами
            self.rate_limiter.wait_if_needed().await?;
        }
        
        Ok(all_hotels)
    }
    
    fn convert_to_message(&self, data: Self::RawData) -> Result<SupplierMessage, ConversionError> {
        Ok(SupplierMessage {
            supplier_id: "stuba".to_string(),
            message_type: MessageType::HotelData,
            payload: serde_json::to_value(&data)?,
            timestamp: chrono::Utc::now(),
            correlation_id: format!("stuba_{}", uuid::Uuid::new_v4()),
        })
    }
}

XML Data Structures​

// Stuba-специфичные структуры данных
#[derive(Debug, Deserialize, Clone)]
pub struct StubaHotelData {
    #[serde(rename = "@id")]
    pub hotel_id: String,
    #[serde(rename = "@name")]
    pub name: String,
    pub location: StubaLocation,
    pub amenities: Option<StubaAmenities>,
    pub rooms: Vec<StubaRoom>,
    pub images: Option<Vec<StubaImage>>,
    pub description: Option<StubaDescription>,
}

#[derive(Debug, Deserialize, Clone)]
pub struct StubaLocation {
    pub coordinates: Option<StubaCoordinates>,
    pub address: StubaAddress,
    pub region: String,
    pub country: String,
}

#[derive(Debug, Deserialize, Clone)]
pub struct StubaCoordinates {
    #[serde(rename = "@lat")]
    pub latitude: f64,
    #[serde(rename = "@lng")]  
    pub longitude: f64,
}

// XML парсер специально для Stuba формата
impl StubaXmlParser {
    pub fn parse_hotels(&self, xml_content: &str) -> Result<Vec<StubaHotelData>, ParseError> {
        let mut reader = quick_xml::Reader::from_str(xml_content);
        reader.trim_text(true);
        
        let mut hotels = Vec::new();
        let mut buf = Vec::new();
        
        loop {
            match reader.read_event(&mut buf)? {
                Event::Start(ref e) if e.name() == b"hotel" => {
                    let hotel = self.parse_single_hotel(&mut reader)?;
                    hotels.push(hotel);
                }
                Event::Eof => break,
                _ => (),
            }
            buf.clear();
        }
        
        Ok(hotels)
    }
}

FTP Integration​

// FTP клиент для загрузки XML файлов
impl StubaAdapter {
    async fn find_new_xml_files(&self, patterns: &[String]) -> Result<Vec<String>, SupplierError> {
        let mut new_files = Vec::new();
        
        // Получение списка файлов с FTP
        let file_list = self.ftp_client.list(None)?;
        
        for line in file_list.lines() {
            let file_info = self.parse_ftp_line(line)?;
            
            // Проверка паттернов файлов
            for pattern in patterns {
                if glob::Pattern::new(pattern)?.matches(&file_info.name) {
                    // Проверка что файл новее последней синхронизации
                    if file_info.modified_time > self.last_sync_time() {
                        new_files.push(file_info.name);
                    }
                }
            }
        }
        
        Ok(new_files)
    }
    
    async fn download_xml_file(&self, file_path: &str) -> Result<String, SupplierError> {
        // Загрузка файла с прогресс-баром
        let mut cursor = std::io::Cursor::new(Vec::new());
        
        self.ftp_client.retr(file_path, &mut cursor)?;
        
        let content = String::from_utf8(cursor.into_inner())?;
        
        // Валидация XML перед возвратом
        quick_xml::Reader::from_str(&content).check_end()?;
        
        Ok(content)
    }
}

HomeToGo Adapter​

GraphQL Integration​

Источник данных: GraphQL API согласно Архитектура платформы: "HomeToGo — GraphQL API для поиска и цен".

pub struct HomeToGoAdapter {
    graphql_client: GraphQLClient,
    api_key: String,
    config: HomeToGoConfig,
    rate_limiter: RateLimiter,
}

#[derive(Clone)]
pub struct HomeToGoConfig {
    pub api_endpoint: String,           // "https://api.hometogo.com/graphql"
    pub api_key: String,               // Sandbox или Production ключ
    pub timeout: Duration,             // 30 секунд таймаут
    pub rate_limit: u32,              // 100 requests/minute
    pub batch_size: usize,            // 50 accommodation за запрос
    pub regions: Vec<String>,         // ["prague", "berlin", "paris"]
}

impl SupplierAdapter for HomeToGoAdapter {
    type Config = HomeToGoConfig;
    type RawData = HomeToGoAccommodation;
    
    fn fetch_data(&self, config: &Self::Config) -> Result<Vec<Self::RawData>, SupplierError> {
        let mut all_accommodations = Vec::new();
        
        for region in &config.regions {
            // 1. GraphQL запрос для региона
            let query = self.build_region_query(region, config.batch_size)?;
            
            // 2. Пагинация через все результаты
            let mut offset = 0;
            loop {
                let response = self.execute_graphql_query(&query, offset)?;
                
                if response.data.accommodations.is_empty() {
                    break;
                }
                
                all_accommodations.extend(response.data.accommodations);
                offset += config.batch_size;
                
                // Rate limiting между запросами
                self.rate_limiter.wait().await?;
            }
        }
        
        Ok(all_accommodations)
    }
    
    fn convert_to_message(&self, data: Self::RawData) -> Result<SupplierMessage, ConversionError> {
        Ok(SupplierMessage {
            supplier_id: "hometogo".to_string(),
            message_type: MessageType::HotelData,
            payload: serde_json::to_value(&data)?,
            timestamp: chrono::Utc::now(),
            correlation_id: format!("hometogo_{}", uuid::Uuid::new_v4()),
        })
    }
}

GraphQL Queries​

// GraphQL запросы для HomeToGo API
impl HomeToGoAdapter {
    fn build_region_query(&self, region: &str, limit: usize) -> Result<String, QueryError> {
        let query = format!(r#"
            query GetAccommodations($region: String!, $limit: Int!, $offset: Int!) {{
                accommodations(
                    region: $region, 
                    limit: $limit, 
                    offset: $offset,
                    filters: {{ 
                        accommodationType: HOTEL,
                        status: ACTIVE 
                    }}
                ) {{
                    id
                    name
                    description
                    location {{
                        latitude
                        longitude
                        address {{
                            street
                            city
                            country
                            postalCode
                        }}
                    }}
                    amenities {{
                        name
                        category
                    }}
                    images {{
                        url
                        caption
                        isPrimary
                    }}
                    rating {{
                        overall
                        cleanliness
                        service
                        location
                        reviewCount
                    }}
                    priceRange {{
                        min
                        max
                        currency
                    }}
                }}
            }}
        "#);
        
        Ok(query)
    }
    
    async fn execute_graphql_query(&self, query: &str, offset: usize) -> Result<HomeToGoResponse, GraphQLError> {
        let variables = json!({
            "region": region,
            "limit": self.config.batch_size,
            "offset": offset
        });
        
        let request = GraphQLRequest {
            query: query.to_string(),
            variables: Some(variables),
        };
        
        let response = self.graphql_client
            .post(&self.config.api_endpoint)
            .header("Authorization", format!("Bearer {}", self.api_key))
            .header("Content-Type", "application/json")
            .json(&request)
            .timeout(self.config.timeout)
            .send()
            .await?;
            
        if !response.status().is_success() {
            return Err(GraphQLError::HttpError(response.status()));
        }
        
        let graphql_response: HomeToGoResponse = response.json().await?;
        
        if let Some(errors) = &graphql_response.errors {
            return Err(GraphQLError::GraphQLErrors(errors.clone()));
        }
        
        Ok(graphql_response)
    }
}

Data Structures​

// HomeToGo API response structures
#[derive(Debug, Deserialize, Clone)]
pub struct HomeToGoResponse {
    pub data: HomeToGoData,
    pub errors: Option<Vec<GraphQLError>>,
}

#[derive(Debug, Deserialize, Clone)]
pub struct HomeToGoData {
    pub accommodations: Vec<HomeToGoAccommodation>,
}

#[derive(Debug, Deserialize, Clone)]
pub struct HomeToGoAccommodation {
    pub id: String,
    pub name: String,
    pub description: Option<String>,
    pub location: HomeToGoLocation,
    pub amenities: Vec<HomeToGoAmenity>,
    pub images: Vec<HomeToGoImage>,
    pub rating: Option<HomeToGoRating>,
    pub price_range: Option<HomeToGoPriceRange>,
}

#[derive(Debug, Deserialize, Clone)]
pub struct HomeToGoLocation {
    pub latitude: f64,
    pub longitude: f64,
    pub address: HomeToGoAddress,
}

#[derive(Debug, Deserialize, Clone)]
pub struct HomeToGoAddress {
    pub street: Option<String>,
    pub city: String,
    pub country: String,
    #[serde(rename = "postalCode")]
    pub postal_code: Option<String>,
}

Other Suppliers Adapter​

Extensible Architecture​

Будущие поставщики: REST/XML согласно Архитектура платформы: "Others — будущие поставщики через REST/XML".

// Универсальный адаптер для REST API поставщиков
pub struct RestSupplierAdapter {
    http_client: reqwest::Client,
    config: RestSupplierConfig,
    transformer: Box<dyn DataTransformer>,
}

#[derive(Clone)]
pub struct RestSupplierConfig {
    pub supplier_name: String,         // "ratehawk", "booking", "expedia"
    pub base_url: String,             // "https://api.ratehawk.com/v1"
    pub auth_method: AuthMethod,       // API key, OAuth2, Basic Auth
    pub endpoints: HashMap<String, RestEndpoint>,
    pub rate_limit: u32,              // requests per minute
    pub data_format: DataFormat,       // JSON, XML
}

#[derive(Clone)]
pub enum AuthMethod {
    ApiKey { header: String, key: String },
    OAuth2 { token_url: String, client_id: String, client_secret: String },
    BasicAuth { username: String, password: String },
    BearerToken { token: String },
}

#[derive(Clone)]
pub struct RestEndpoint {
    pub path: String,                 // "/hotels"
    pub method: HttpMethod,           // GET, POST
    pub query_params: Vec<String>,    // ["region", "limit", "offset"]
    pub request_body: Option<String>, // JSON template for POST requests
}

// Trait для трансформации данных от разных поставщиков
pub trait DataTransformer: Send + Sync {
    fn transform(&self, raw_data: serde_json::Value) -> Result<Vec<GenericHotelData>, TransformError>;
}

// Универсальная структура для любого поставщика
#[derive(Debug, Serialize, Deserialize, Clone)]
pub struct GenericHotelData {
    pub supplier_id: String,
    pub external_id: String,
    pub name: String,
    pub location: GenericLocation,
    pub amenities: Vec<String>,
    pub metadata: serde_json::Value,  // Supplier-specific data
}

Dynamic Configuration​

// Конфигурация поставщиков из базы данных или файлов
impl RestSupplierAdapter {
    pub async fn load_supplier_config(supplier_name: &str) -> Result<RestSupplierConfig, ConfigError> {
        // Загрузка конфигурации из PostgreSQL или YAML файлов
        let config_query = r#"
            SELECT config_data 
            FROM supplier_configs 
            WHERE supplier_name = $1 AND active = true
        "#;
        
        let config_json: serde_json::Value = sqlx::query_scalar(config_query)
            .bind(supplier_name)
            .fetch_one(&db_pool)
            .await?;
            
        let config: RestSupplierConfig = serde_json::from_value(config_json)?;
        
        Ok(config)
    }
    
    pub async fn register_new_supplier(&self, config: RestSupplierConfig) -> Result<(), RegistrationError> {
        // Валидация конфигурации
        self.validate_config(&config)?;
        
        // Тест соединения с API поставщика  
        self.test_connection(&config).await?;
        
        // Регистрация в системе
        let insert_query = r#"
            INSERT INTO supplier_configs (supplier_name, config_data, active, created_at)
            VALUES ($1, $2, true, now())
        "#;
        
        sqlx::query(insert_query)
            .bind(&config.supplier_name)
            .bind(serde_json::to_value(&config)?)
            .execute(&self.db_pool)
            .await?;
            
        // Уведомление Ingestion Layer о новом поставщике
        self.notify_ingestion_layer(&config.supplier_name).await?;
        
        Ok(())
    }
}

Scheduling and Synchronization​

Cron-Based Scheduling​

// Планировщик синхронизации с поставщиками
pub struct SyncScheduler {
    scheduler: tokio_cron_scheduler::JobScheduler,
    suppliers: HashMap<String, Arc<dyn SupplierAdapter>>,
    nats_gateway: Arc<SupplierGateway>,
}

impl SyncScheduler {
    pub async fn start(&self) -> Result<(), SchedulerError> {
        // Stuba: каждые 4 часа (большие XML файлы)
        self.scheduler.add(Job::new_async("0 */4 * * *", |_uuid, _l| {
            Box::pin(async {
                self.sync_supplier("stuba").await;
            })
        })?).await?;
        
        // HomeToGo: каждый час (цены меняются часто)
        self.scheduler.add(Job::new_async("0 * * * *", |_uuid, _l| {
            Box::pin(async {
                self.sync_supplier("hometogo").await;
            })
        })?).await?;
        
        // Другие поставщики: настраиваемое расписание
        for (supplier_name, adapter) in &self.suppliers {
            if let Some(schedule) = adapter.get_sync_schedule() {
                self.add_custom_schedule(supplier_name, &schedule).await?;
            }
        }
        
        self.scheduler.start().await?;
        
        Ok(())
    }
    
    async fn sync_supplier(&self, supplier_name: &str) -> Result<(), SyncError> {
        let start_time = Instant::now();
        
        match self.suppliers.get(supplier_name) {
            Some(adapter) => {
                info!("Starting sync for supplier: {}", supplier_name);
                
                // 1. Health check перед синхронизацией
                adapter.health_check()?;
                
                // 2. Получение данных от поставщика
                let config = self.load_supplier_config(supplier_name).await?;
                let raw_data = adapter.fetch_data(&config)?;
                
                info!("Fetched {} records from {}", raw_data.len(), supplier_name);
                
                // 3. Конвертация и отправка в NATS
                for data_chunk in raw_data.chunks(100) { // Батчами по 100
                    for data_item in data_chunk {
                        let message = adapter.convert_to_message(data_item.clone())?;
                        self.nats_gateway.publish_to_nats(message).await?;
                    }
                    
                    // Небольшая пауза между батчами
                    tokio::time::sleep(Duration::from_millis(100)).await;
                }
                
                let duration = start_time.elapsed();
                info!("Completed sync for {} in {:?}", supplier_name, duration);
                
                // Метрики
                self.record_sync_metrics(supplier_name, raw_data.len(), duration);
                
                Ok(())
            }
            None => {
                warn!("Supplier {} not found", supplier_name);
                Err(SyncError::SupplierNotFound(supplier_name.to_string()))
            }
        }
    }
}

Error Handling and Monitoring​

Comprehensive Error Handling​

// Типы ошибок для адаптеров поставщиков
#[derive(Debug, thiserror::Error)]
pub enum SupplierError {
    #[error("Network error: {0}")]
    NetworkError(#[from] reqwest::Error),
    
    #[error("Authentication failed for supplier {supplier}: {reason}")]
    AuthenticationError { supplier: String, reason: String },
    
    #[error("Rate limit exceeded for supplier {supplier}")]
    RateLimitError { supplier: String },
    
    #[error("Data parsing error: {0}")]
    ParseError(#[from] ParseError),
    
    #[error("FTP error: {0}")]
    FtpError(String),
    
    #[error("GraphQL error: {0:?}")]
    GraphQLError(Vec<GraphQLError>),
    
    #[error("Configuration error: {0}")]
    ConfigError(String),
}

// Error recovery strategies
impl SupplierAdapter for StubaAdapter {
    fn fetch_data(&self, config: &Self::Config) -> Result<Vec<Self::RawData>, SupplierError> {
        let mut retries = 0;
        const MAX_RETRIES: u32 = 3;
        
        loop {
            match self.fetch_data_impl(config) {
                Ok(data) => return Ok(data),
                Err(SupplierError::NetworkError(_)) if retries < MAX_RETRIES => {
                    retries += 1;
                    warn!("Network error, retry {}/{}", retries, MAX_RETRIES);
                    
                    // Exponential backoff
                    let delay = Duration::from_secs(2_u64.pow(retries));
                    std::thread::sleep(delay);
                }
                Err(SupplierError::RateLimitError { .. }) => {
                    warn!("Rate limit hit, waiting 60 seconds");
                    std::thread::sleep(Duration::from_secs(60));
                }
                Err(e) => return Err(e),
            }
        }
    }
}

Monitoring and Alerting​

// Prometheus метрики для suppliers layer
pub struct SupplierMetrics {
    pub fetch_duration: prometheus::HistogramVec,
    pub fetch_success_total: prometheus::CounterVec,
    pub fetch_error_total: prometheus::CounterVec,
    pub records_fetched: prometheus::CounterVec,
    pub api_rate_limit_hits: prometheus::CounterVec,
}

impl SupplierMetrics {
    pub fn new() -> Self {
        Self {
            fetch_duration: prometheus::register_histogram_vec!(
                "supplier_fetch_duration_seconds",
                "Time spent fetching data from suppliers",
                &["supplier"],
                prometheus::exponential_buckets(0.1, 2.0, 15).unwrap()
            ).unwrap(),
            
            fetch_success_total: prometheus::register_counter_vec!(
                "supplier_fetch_success_total",
                "Successful data fetch operations",
                &["supplier"]
            ).unwrap(),
            
            fetch_error_total: prometheus::register_counter_vec!(
                "supplier_fetch_error_total",
                "Failed data fetch operations",
                &["supplier", "error_type"]
            ).unwrap(),
            
            records_fetched: prometheus::register_counter_vec!(
                "supplier_records_fetched_total",
                "Total number of records fetched",
                &["supplier", "data_type"]
            ).unwrap(),
        }
    }
    
    pub fn record_fetch_success(&self, supplier: &str, duration: Duration, record_count: usize) {
        self.fetch_duration.with_label_values(&[supplier]).observe(duration.as_secs_f64());
        self.fetch_success_total.with_label_values(&[supplier]).inc();
        self.records_fetched.with_label_values(&[supplier, "hotel"]).inc_by(record_count as u64);
    }
}

// Health checks для каждого поставщика
impl SupplierAdapter for HomeToGoAdapter {
    fn health_check(&self) -> Result<HealthStatus, SupplierError> {
        // Простой GraphQL query для проверки доступности
        let health_query = r#"
            query HealthCheck {
                __schema {
                    queryType {
                        name
                    }
                }
            }
        "#;
        
        let start_time = Instant::now();
        let response = self.execute_simple_query(&health_query)?;
        let latency = start_time.elapsed();
        
        if response.errors.is_some() {
            return Ok(HealthStatus::Degraded { 
                reason: "GraphQL errors in health check".to_string(),
                latency 
            });
        }
        
        if latency > Duration::from_secs(5) {
            return Ok(HealthStatus::Degraded { 
                reason: "High latency".to_string(),
                latency 
            });
        }
        
        Ok(HealthStatus::Healthy { latency })
    }
}

Development and Testing​

Integration Tests​

// Тесты интеграции с реальными API поставщиков
#[cfg(test)]
mod integration_tests {
    use super::*;
    
    #[tokio::test]
    async fn test_stuba_adapter_integration() {
        let config = StubaConfig::from_env();
        let adapter = StubaAdapter::new(config.clone());
        
        // Тест health check
        let health = adapter.health_check().unwrap();
        assert!(matches!(health, HealthStatus::Healthy { .. }));
        
        // Тест получения данных (небольшая выборка)
        let mut test_config = config;
        test_config.batch_size = 10;
        
        let data = adapter.fetch_data(&test_config).unwrap();
        assert!(!data.is_empty());
        
        // Тест конвертации в SupplierMessage
        let message = adapter.convert_to_message(data[0].clone()).unwrap();
        assert_eq!(message.supplier_id, "stuba");
        assert_eq!(message.message_type, MessageType::HotelData);
    }
    
    #[tokio::test]
    async fn test_hometogo_adapter_graphql() {
        let config = HomeToGoConfig::sandbox();
        let adapter = HomeToGoAdapter::new(config.clone());
        
        // Тест GraphQL запроса
        let query = adapter.build_region_query("prague", 5).unwrap();
        let response = adapter.execute_graphql_query(&query, 0).await.unwrap();
        
        assert!(response.data.accommodations.len() <= 5);
        
        // Валидация структуры данных
        for accommodation in response.data.accommodations {
            assert!(!accommodation.id.is_empty());
            assert!(!accommodation.name.is_empty());
        }
    }
}

// Mock адаптеры для unit тестов
pub struct MockSupplierAdapter {
    pub mock_data: Vec<serde_json::Value>,
    pub should_fail: bool,
}

impl SupplierAdapter for MockSupplierAdapter {
    type Config = ();
    type RawData = serde_json::Value;
    
    fn fetch_data(&self, _config: &Self::Config) -> Result<Vec<Self::RawData>, SupplierError> {
        if self.should_fail {
            Err(SupplierError::NetworkError(
                reqwest::Error::from(std::io::Error::new(std::io::ErrorKind::Other, "Mock error"))
            ))
        } else {
            Ok(self.mock_data.clone())
        }
    }
    
    fn convert_to_message(&self, data: Self::RawData) -> Result<SupplierMessage, ConversionError> {
        Ok(SupplierMessage {
            supplier_id: "mock".to_string(),
            message_type: MessageType::HotelData,
            payload: data,
            timestamp: chrono::Utc::now(),
            correlation_id: uuid::Uuid::new_v4().to_string(),
        })
    }
}

Deployment and Configuration​

Kubernetes Deployment​

# Suppliers Layer Deployment
apiVersion: apps/v1
kind: Deployment
metadata:
  name: suppliers-gateway
spec:
  replicas: 2
  selector:
    matchLabels:
      app: suppliers-gateway
  template:
    metadata:
      labels:
        app: suppliers-gateway
    spec:
      containers:
      - name: gateway
        image: vitrip/suppliers-gateway:v1.0
        env:
        - name: NATS_URL
          value: "nats://nats-cluster:4222"
        - name: POSTGRES_URL
          valueFrom:
            secretKeyRef:
              name: postgres-credentials
              key: url
        - name: STUBA_FTP_PASSWORD
          valueFrom:
            secretKeyRef:
              name: supplier-credentials
              key: stuba-ftp-password
        - name: HOMETOGO_API_KEY
          valueFrom:
            secretKeyRef:
              name: supplier-credentials
              key: hometogo-api-key
        resources:
          requests:
            memory: "512Mi"
            cpu: "300m"
          limits:
            memory: "1Gi"
            cpu: "500m"
        livenessProbe:
          httpGet:
            path: /health
            port: 8080
          initialDelaySeconds: 30
          periodSeconds: 30
        readinessProbe:
          httpGet:
            path: /ready
            port: 8080
          initialDelaySeconds: 5
          periodSeconds: 10

---
# CronJob для регулярной синхронизации
apiVersion: batch/v1
kind: CronJob
metadata:
  name: stuba-sync
spec:
  schedule: "0 */4 * * *"  # Каждые 4 часа
  jobTemplate:
    spec:
      template:
        spec:
          containers:
          - name: stuba-sync
            image: vitrip/suppliers-gateway:v1.0
            command: ["./sync-stuba"]
            envFrom:
            - secretRef:
                name: supplier-credentials
          restartPolicy: OnFailure

Связанная документация​