Создание облачных приложений Rust: руководство по архитектуре микросервисов

#[derive(Debug, Clone)]
pub struct OrderId(Uuid);
 
#[derive(Debug, Clone)]
pub enum OrderStatus {
    Pending,
    Confirmed,
    Shipped,
    Delivered,
    Cancelled,
}
 
pub struct Order {
    id: OrderId,
    customer_id: CustomerId,
    items: Vec<OrderItem>,
    status: OrderStatus,
    created_at: DateTime<Utc>,
}
 
impl Order {
    pub fn new(customer_id: CustomerId, items: Vec<OrderItem>) -> Result<Self, OrderError> {
        if items.is_empty() {
            return Err(OrderError::EmptyOrder);
        }
        
        Ok(Order {
            id: OrderId(Uuid::new_v4()),
            customer_id,
            items,
            status: OrderStatus::Pending,
            created_at: Utc::now(),
        })
    }
    
    pub fn confirm(&mut self) -> Result<(), OrderError> {
        match self.status {
            OrderStatus::Pending => {
                self.status = OrderStatus::Confirmed;
                Ok(())
            }
            _ => Err(OrderError::InvalidStatusTransition),
        }
    }
}

#[async_trait]
pub trait OrderRepository {
    async fn save(&self, order: Order) -> Result<(), RepositoryError>;
    async fn find_by_id(&self, id: OrderId) -> Result<Option<Order>, RepositoryError>;
    async fn find_by_customer(&self, customer_id: CustomerId) -> Result<Vec<Order>, RepositoryError>;
}
 
#[async_trait]
pub trait PaymentGateway {
    async fn charge(&self, amount: Money, card: CreditCard) -> Result<PaymentResult, PaymentError>;
    async fn refund(&self, payment_id: PaymentId) -> Result<RefundResult, PaymentError>;
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum OrderEvent {
    OrderCreated {
        order_id: OrderId,
        customer_id: CustomerId,
        items: Vec<OrderItem>,
        timestamp: DateTime<Utc>,
    },
    OrderConfirmed {
        order_id: OrderId,
        timestamp: DateTime<Utc>,
    },
    OrderShipped {
        order_id: OrderId,
        tracking_number: String,
        timestamp: DateTime<Utc>,
    },
    OrderCancelled {
        order_id: OrderId,
        reason: String,
        timestamp: DateTime<Utc>,
    },
}
 
impl OrderEvent {
    pub fn apply(events: &[OrderEvent]) -> Result<Order, EventApplyError> {
        if events.is_empty() {
            return Err(EventApplyError::NoEvents);
        }
        
        let mut order = match &events[0] {
            OrderEvent::OrderCreated { order_id, customer_id, items, timestamp } => {
                Order {
                    id: order_id.clone(),
                    customer_id: customer_id.clone(),
                    items: items.clone(),
                    status: OrderStatus::Pending,
                    created_at: *timestamp,
                }
            },
            _ => return Err(EventApplyError::FirstEventMustBeCreation),
        };
        
        for event in events.iter().skip(1) {
            order.apply_event(event)?;
        }
        
        Ok(order)
    }
}

use std::sync::Arc;
use tokio::sync::RwLock;
 
pub struct CircuitBreaker {
    state: Arc<RwLock<CircuitBreakerState>>,
    failure_threshold: u32,
    recovery_timeout: Duration,
}
 
#[derive(Debug)]
enum CircuitBreakerState {
    Closed { failure_count: u32 },
    Open { last_failure: Instant },
    HalfOpen,
}
 
impl CircuitBreaker {
    pub async fn call<F, T, E>(&self, operation: F) -> Result<T, CircuitBreakerError<E>>
    where
        F: Future<Output = Result<T, E>>,
    {
        let state = self.state.read().await;
        
        match *state {
            CircuitBreakerState::Open { last_failure } => {
                if last_failure.elapsed() > self.recovery_timeout {
                    drop(state);
                    let mut write_state = self.state.write().await;
                    *write_state = CircuitBreakerState::HalfOpen;
                } else {
                    return Err(CircuitBreakerError::Open);
                }
            }
            CircuitBreakerState::HalfOpen => {
                // Пропускаем один запрос для проверки
            }
            CircuitBreakerState::Closed { .. } => {
                // Нормальная работа
            }
        }
        
        drop(state);
        
        match operation.await {
            Ok(result) => {
                self.on_success().await;
                Ok(result)
            }
            Err(error) => {
                self.on_failure().await;
                Err(CircuitBreakerError::Operation(error))
            }
        }
    }
}

pub struct OrderSaga {
    steps: Vec<SagaStep>,
    compensations: Vec<CompensationStep>,
}
 
impl OrderSaga {
    pub async fn execute(&mut self) -> Result<SagaResult, SagaError> {
        for (index, step) in self.steps.iter().enumerate() {
            match step.execute().await {
                Ok(result) => {
                    // Сохраняем компенсирующее действие
                    if let Some(compensation) = step.compensation() {
                        self.compensations.push(CompensationStep {
                            step_index: index,
                            action: compensation,
                        });
                    }
                }
                Err(error) => {
                    // Откатываем все выполненные шаги
                    self.compensate().await?;
                    return Err(SagaError::StepFailed(index, error));
                }
            }
        }
        
        Ok(SagaResult::Success)
    }
    
    async fn compensate(&mut self) -> Result<(), CompensationError> {
        // Выполняем компенсацию в обратном порядке
        for compensation in self.compensations.iter().rev() {
            compensation.action.execute().await?;
        }
        Ok(())
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum ServiceError {
    NotFound { resource: String, id: String },
    ValidationFailed { field: String, message: String },
    Unauthorized { required_permission: String },
    ExternalServiceUnavailable { service: String, retry_after: Option<Duration> },
    InternalError { correlation_id: Uuid },
}
 
impl ServiceError {
    pub fn to_http_status(&self) -> StatusCode {
        match self {
            ServiceError::NotFound { .. } => StatusCode::NOT_FOUND,
            ServiceError::ValidationFailed { .. } => StatusCode::BAD_REQUEST,
            ServiceError::Unauthorized { .. } => StatusCode::UNAUTHORIZED,
            ServiceError::ExternalServiceUnavailable { .. } => StatusCode::SERVICE_UNAVAILABLE,
            ServiceError::InternalError { .. } => StatusCode::INTERNAL_SERVER_ERROR,
        }
    }
    
    pub fn should_retry(&self) -> bool {
        matches!(self, ServiceError::ExternalServiceUnavailable { .. })
    }
    
    pub fn correlation_id(&self) -> Option<Uuid> {
        match self {
            ServiceError::InternalError { correlation_id } => Some(*correlation_id),
            _ => None,
        }
    }
}

use tokio::sync::broadcast;
use futures_util::StreamExt;
 
pub struct EventBus {
    sender: broadcast::Sender<DomainEvent>,
}
 
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum DomainEvent {
    OrderCreated { order_id: OrderId, customer_id: CustomerId, total: Money },
    PaymentProcessed { payment_id: PaymentId, order_id: OrderId, amount: Money },
    InventoryReserved { product_id: ProductId, quantity: i32, reservation_id: Uuid },
    ShippingScheduled { order_id: OrderId, estimated_delivery: DateTime<Utc> },
}
 
impl EventBus {
    pub fn publish(&self, event: DomainEvent) -> Result<(), EventBusError> {
        match self.sender.send(event) {
            Ok(_) => Ok(()),
            Err(_) => Err(EventBusError::NoSubscribers),
        }
    }
    
    pub fn subscribe(&self) -> broadcast::Receiver<DomainEvent> {
        self.sender.subscribe()
    }
}
 
// Пример подписчика в сервисе уведомлений
pub async fn start_notification_handler(event_bus: EventBus) {
    let mut receiver = event_bus.subscribe();
    
    while let Ok(event) = receiver.recv().await {
        match event {
            DomainEvent::OrderCreated { order_id, customer_id, .. } => {
                if let Err(e) = send_order_confirmation(customer_id, order_id).await {
                    tracing::error!("Failed to send order confirmation: {:?}", e);
                }
            },
            DomainEvent::ShippingScheduled { order_id, estimated_delivery } => {
                if let Err(e) = send_shipping_notification(order_id, estimated_delivery).await {
                    tracing::error!("Failed to send shipping notification: {:?}", e);
                }
            },
            _ => {} // Игнорируем неинтересные события
        }
    }
}

pub struct RetryClient {
    inner: reqwest::Client,
    max_retries: u32,
    base_delay: Duration,
}
 
impl RetryClient {
    pub async fn request_with_retry<T>(&self, mut request_builder: RequestBuilder) -> Result<T, RetryError>
    where
        T: serde::de::DeserializeOwned,
    {
        let mut last_error = None;
        
        for attempt in 0..=self.max_retries {
            let request = request_builder.try_clone().ok_or(RetryError::RequestClone)?;
            
            match request.send().await {
                Ok(response) if response.status().is_success() => {
                    return response.json::<T>().await.map_err(RetryError::Deserialize);
                }
                Ok(response) if response.status().is_server_error() => {
                    last_error = Some(RetryError::ServerError(response.status()));
                }
                Ok(response) => {
                    // Клиентские ошибки не ретраим
                    return Err(RetryError::ClientError(response.status()));
                }
                Err(e) if e.is_timeout() || e.is_connect() => {
                    last_error = Some(RetryError::Network(e));
                }
                Err(e) => {
                    return Err(RetryError::Network(e));
                }
            }
            
            if attempt < self.max_retries {
                let delay = self.base_delay * 2_u32.pow(attempt);
                let jitter = rand::random::<f64>() * 0.1; // 10% jitter
                let total_delay = delay + Duration::from_millis((delay.as_millis() as f64 * jitter) as u64);
                tokio::time::sleep(total_delay).await;
            }
        }
        
        Err(last_error.unwrap_or(RetryError::MaxRetriesExceeded))
    }
}

#[async_trait]
pub trait CacheManager {
    async fn get<T>(&self, key: &str) -> Result<Option<T>, CacheError>
    where
        T: serde::de::DeserializeOwned;
    
    async fn set<T>(&self, key: &str, value: &T, ttl: Duration) -> Result<(), CacheError>
    where
        T: serde::Serialize;
    
    async fn invalidate(&self, pattern: &str) -> Result<(), CacheError>;
}
 
pub async fn handle_product_updated_event(
    event: ProductUpdatedEvent,
    cache: &dyn CacheManager,
) -> Result<(), EventHandlerError> {
    // Инвалидируем все связанные кеши
    cache.invalidate(&format!("product:{}:*", event.product_id)).await?;
    cache.invalidate(&format!("category:{}:*", event.category_id)).await?;
    cache.invalidate("products:featured").await?;
    
    Ok(())
}

FROM rust:1.75-slim as builder
 
WORKDIR /usr/src/app
COPY Cargo.toml Cargo.lock ./
COPY src ./src
 
# Статическая линковка для scratch-образа
ENV RUSTFLAGS="-C target-feature=+crt-static"
RUN cargo build --release --target x86_64-unknown-linux-musl
 
FROM scratch
COPY --from=builder /usr/src/app/target/x86_64-unknown-linux-musl/release/my-service /
COPY --from=builder /etc/ssl/certs/ca-certificates.crt /etc/ssl/certs/
 
EXPOSE 8080
CMD ["/my-service"]

use jsonwebtoken::{decode, encode, DecodingKey, EncodingKey, Header, Validation};
 
#[derive(Debug, Serialize, Deserialize)]
pub struct Claims {
    pub sub: String, // user_id
    pub roles: Vec<String>,
    pub permissions: Vec<String>,
    pub exp: usize,
    pub iat: usize,
    pub iss: String, // issuer
}
 
pub struct JwtService {
    encoding_key: EncodingKey,
    decoding_key: DecodingKey,
    validation: Validation,
}
 
impl JwtService {
    pub fn new(secret: &str) -> Self {
        let mut validation = Validation::default();
        validation.set_issuer(&["order-system"]);
        validation.validate_exp = true;
        
        Self {
            encoding_key: EncodingKey::from_secret(secret.as_ref()),
            decoding_key: DecodingKey::from_secret(secret.as_ref()),
            validation,
        }
    }
 
    pub fn generate_token(&self, user_id: String, roles: Vec<String>, permissions: Vec<String>) -> Result<String, JwtError> {
        let now = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map_err(|_| JwtError::TimeError)?
            .as_secs() as usize;
 
        let claims = Claims {
            sub: user_id,
            roles,
            permissions,
            exp: now + 3600, // 1 час
            iat: now,
            iss: "order-system".to_string(),
        };
 
        encode(&Header::default(), &claims, &self.encoding_key)
            .map_err(JwtError::TokenCreation)
    }
 
    pub fn validate_token(&self, token: &str) -> Result<Claims, JwtError> {
        decode::<Claims>(token, &self.decoding_key, &self.validation)
            .map(|token_data| token_data.claims)
            .map_err(JwtError::TokenValidation)
    }
}
 
// Middleware для проверки авторизации
pub async fn auth_middleware(
    req: ServiceRequest,
    credentials: BearerAuth,
    jwt_service: web::Data<JwtService>,
) -> Result<ServiceRequest, (actix_web::Error, ServiceRequest)> {
    let token = credentials.token();
    
    match jwt_service.validate_token(token) {
        Ok(claims) => {
            req.extensions_mut().insert(claims);
            Ok(req)
        }
        Err(_) => Err((ErrorUnauthorized("Invalid token"), req))
    }
}

#[derive(Debug, sqlx::FromRow)]
pub struct OutboxEvent {
    pub id: Uuid,
    pub aggregate_type: String,
    pub aggregate_id: String,
    pub event_type: String,
    pub event_data: serde_json::Value,
    pub created_at: DateTime<Utc>,
    pub processed_at: Option<DateTime<Utc>>,
}
 
pub struct OutboxProcessor {
    db_pool: PgPool,
    event_bus: EventBus,
}
 
impl OutboxProcessor {
    pub async fn process_pending_events(&self) -> Result<(), OutboxError> {
        let pending_events = sqlx::query_as::<_, OutboxEvent>(
            "SELECT * FROM outbox_events WHERE processed_at IS NULL ORDER BY created_at LIMIT 100"
        )
        .fetch_all(&self.db_pool)
        .await?;
 
        for event in pending_events {
            match self.publish_event(&event).await {
                Ok(_) => {
                    self.mark_as_processed(event.id).await?;
                }
                Err(e) => {
                    tracing::error!("Failed to publish event {}: {:?}", event.id, e);
                    // Продолжаем обработку следующих событий
                }
            }
        }
 
        Ok(())
    }
 
    async fn publish_event(&self, event: &OutboxEvent) -> Result<(), EventPublishError> {
        let domain_event = self.deserialize_event(event)?;
        self.event_bus.publish(domain_event).await
    }
 
    async fn mark_as_processed(&self, event_id: Uuid) -> Result<(), sqlx::Error> {
        sqlx::query("UPDATE outbox_events SET processed_at = $1 WHERE id = $2")
            .bind(Utc::now())
            .bind(event_id)
            .execute(&self.db_pool)
            .await?;
        
        Ok(())
    }
}

use redis::cluster::ClusterClient;
use redis::{AsyncCommands, RedisResult};
 
pub struct CacheService {
    client: ClusterClient,
}
 
impl CacheService {
    pub async fn get_or_set<T, F, Fut>(&self, key: &str, ttl: u64, factory: F) -> RedisResult<T>
    where
        T: serde::Serialize + serde::de::DeserializeOwned,
        F: FnOnce() -> Fut,
        Fut: std::future::Future<Output = T>,
    {
        let mut conn = self.client.get_async_connection().await?;
        
        // Пытаемся получить из кеша
        if let Ok(cached) = conn.get::<_, String>(key).await {
            if let Ok(value) = serde_json::from_str::<T>(&cached) {
                return Ok(value);
            }
        }
        
        // Если в кеше нет - вычисляем
        let value = factory().await;
        
        // Сохраняем в кеш
        let serialized = serde_json::to_string(&value).unwrap();
        let _: () = conn.set_ex(key, serialized, ttl).await?;
        
        Ok(value)
    }
 
    pub async fn invalidate_pattern(&self, pattern: &str) -> RedisResult<()> {
        let mut conn = self.client.get_async_connection().await?;
        let keys: Vec<String> = conn.keys(pattern).await?;
        
        if !keys.is_empty() {
            conn.del(keys).await?;
        }
        
        Ok(())
    }
}

use sqlx::{PgPool, PgPoolOptions};
 
pub async fn create_db_pool(database_url: &str) -> Result<PgPool, sqlx::Error> {
    PgPoolOptions::new()
        .max_connections(20)
        .min_connections(5)
        .acquire_timeout(Duration::from_secs(10))
        .idle_timeout(Duration::from_secs(600))
        .max_lifetime(Duration::from_secs(1800))
        .connect(database_url)
        .await
}

use prometheus::{Counter, Histogram, Gauge, Registry, Encoder, TextEncoder};
 
lazy_static::lazy_static! {
    static ref REGISTRY: Registry = Registry::new();
    
    static ref ORDER_PROCESSING_DURATION: Histogram = Histogram::with_opts(
        prometheus::HistogramOpts::new(
            "order_processing_duration_seconds",
            "Time spent processing orders"
        ).buckets(vec![0.1, 0.5, 1.0, 2.5, 5.0, 10.0])
    ).unwrap();
    
    static ref SAGA_COMPENSATION_COUNT: Counter = Counter::with_opts(
        prometheus::CounterOpts::new(
            "saga_compensations_total",
            "Number of saga compensations executed"
        )
    ).unwrap();
    
    static ref CIRCUIT_BREAKER_STATE: Gauge = Gauge::with_opts(
        prometheus::GaugeOpts::new(
            "circuit_breaker_state",
            "Circuit breaker state (0=closed, 1=half-open, 2=open)"
        )
    ).unwrap();
}
 
pub fn init_metrics() {
    REGISTRY.register(Box::new(ORDER_PROCESSING_DURATION.clone())).unwrap();
    REGISTRY.register(Box::new(SAGA_COMPENSATION_COUNT.clone())).unwrap();
    REGISTRY.register(Box::new(CIRCUIT_BREAKER_STATE.clone())).unwrap();
}
 
pub async fn metrics_handler() -> Result<String, std::fmt::Error> {
    let encoder = TextEncoder::new();
    let metric_families = REGISTRY.gather();
    encoder.encode_to_string(&metric_families)
}

use opentelemetry::{
    global, 
    trace::{TraceError, Tracer, Span, SpanKind, Status},
    Context, KeyValue,
};
use opentelemetry_jaeger::JaegerPipeline;
use tracing_opentelemetry::OpenTelemetrySpanExt;
 
pub fn init_tracing() -> Result<(), TraceError> {
    let tracer = JaegerPipeline::new()
        .with_service_name("order-service")
        .with_endpoint("http://jaeger-collector:14268/api/traces")
        .with_tags(vec![
            KeyValue::new("service.version", env!("CARGO_PKG_VERSION")),
            KeyValue::new("deployment.environment", std::env::var("ENVIRONMENT").unwrap_or_else(|_| "development".to_string())),
        ])
        .install_batch(opentelemetry::runtime::Tokio)?;
        
    global::set_tracer_provider(tracer);
    Ok(())
}
 
#[tracing::instrument(
    name = "process_order",
    fields(
        order.id = %order_id,
        customer.id = %customer_id,
        order.total = %total_amount
    ),
    skip(self)
)]
pub async fn process_order(&self, order_id: OrderId, customer_id: CustomerId, total_amount: Money) -> Result<(), OrderError> {
    let span = tracing::Span::current();
    span.set_attribute(KeyValue::new("order.items_count", items.len() as i64));
    
    // Создаём child span для каждого шага обработки
    let payment_span = span.tracer().start_with_context("process_payment", &Context::current());
    payment_span.set_attribute(KeyValue::new("payment.gateway", "stripe"));
    
    match self.payment_service.charge(total_amount, customer_id).await {
        Ok(payment_result) => {
            payment_span.set_status(Status::Ok);
            payment_span.set_attribute(KeyValue::new("payment.id", payment_result.id.to_string()));
        }
        Err(e) => {
            payment_span.set_status(Status::error(format!("Payment failed: {:?}", e)));
            payment_span.end();
            return Err(OrderError::PaymentFailed(e));
        }
    }
    payment_span.end();
    
    // Аналогично для других шагов...
    
    span.set_status(Status::Ok);
    Ok(())
}

use tracing::{info, warn, error, debug};
use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt, EnvFilter, Registry};
use tracing_subscriber::fmt::layer;
 
pub fn init_logging() {
    let env_filter = EnvFilter::try_from_default_env()
        .unwrap_or_else(|_| EnvFilter::new("info"));
        
    Registry::default()
        .with(env_filter)
        .with(
            layer()
                .json()
                .with_current_span(false)
                .with_span_list(true)
                .with_target(true)
                .with_thread_ids(true)
                .with_thread_names(true)
        )
        .with(tracing_opentelemetry::layer())
        .init();
}
 
// Structured logging в действии
#[tracing::instrument(skip(self))]
pub async fn handle_order_created(&self, event: OrderCreatedEvent) -> Result<(), EventHandlerError> {
    info!(
        order.id = %event.order_id,
        customer.id = %event.customer_id,
        order.amount = %event.total_amount,
        event.timestamp = %event.timestamp,
        "Processing order created event"
    );
    
    match self.send_confirmation_email(event.customer_id, event.order_id).await {
        Ok(_) => {
            info!(
                order.id = %event.order_id,
                action = "email_sent",
                "Order confirmation email sent successfully"
            );
        }
        Err(e) => {
            error!(
                order.id = %event.order_id,
                error = %e,
                action = "email_failed",
                "Failed to send order confirmation email"
            );
            
            // Планируем retry через некоторое время
            self.schedule_retry(event, Duration::from_secs(300)).await?;
        }
    }
    
    Ok(())
}

use actix_web::{web, HttpResponse};
use sqlx::PgPool;
 
pub async fn health_check(pool: web::Data<PgPool>) -> HttpResponse {
    HttpResponse::Ok().json(serde_json::json!({
        "status": "healthy",
        "timestamp": chrono::Utc::now().to_rfc3339(),
        "service": "order-service",
        "version": env!("CARGO_PKG_VERSION")
    }))
}
 
pub async fn readiness_check(
    pool: web::Data<PgPool>,
    redis_client: web::Data<redis::Client>
) -> HttpResponse {
    let mut checks = std::collections::HashMap::new();
    
    // Проверяем PostgreSQL
    match sqlx::query("SELECT 1").fetch_one(pool.as_ref()).await {
        Ok(_) => { checks.insert("postgres", true); }
        Err(_) => { checks.insert("postgres", false); }
    }
    
    // Проверяем Redis
    match redis_client.get_async_connection().await {
        Ok(mut conn) => {
            match redis::cmd("PING").query_async::<_, String>(&mut conn).await {
                Ok(_) => { checks.insert("redis", true); }
                Err(_) => { checks.insert("redis", false); }
            }
        }
        Err(_) => { checks.insert("redis", false); }
    }
    
    let all_healthy = checks.values().all(|&v| v);
    let status_code = if all_healthy { 200 } else { 503 };
    
    HttpResponse::build(actix_web::http::StatusCode::from_u16(status_code).unwrap())
        .json(serde_json::json!({
            "status": if all_healthy { "ready" } else { "not_ready" },
            "checks": checks,
            "timestamp": chrono::Utc::now().to_rfc3339()
        }))
}

use prometheus::{register_histogram_vec, register_counter_vec, register_gauge_vec};
 
lazy_static::lazy_static! {
    static ref HTTP_REQUEST_DURATION: prometheus::HistogramVec = register_histogram_vec!(
        "http_request_duration_seconds",
        "HTTP request duration in seconds",
        &["method", "endpoint", "status_code"],
        vec![0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0]
    ).unwrap();
    
    static ref SAGA_STEPS: prometheus::CounterVec = register_counter_vec!(
        "saga_steps_total",
        "Total number of saga steps executed",
        &["saga_type", "step_name", "status"]
    ).unwrap();
    
    static ref ACTIVE_CONNECTIONS: prometheus::GaugeVec = register_gauge_vec!(
        "active_connections",
        "Number of active connections",
        &["service", "connection_type"]
    ).unwrap();
}
 
pub struct MetricsMiddleware;
 
impl<S> Transform<S, ServiceRequest> for MetricsMiddleware 
where
    S: Service<ServiceRequest, Response = ServiceResponse, Error = Error> + 'static,
{
    type Response = ServiceResponse;
    type Error = Error;
    type Transform = MetricsMiddlewareService<S>;
    type InitError = ();
    type Future = Ready<Result<Self::Transform, Self::InitError>>;
 
    fn new_transform(&self, service: S) -> Self::Future {
        ready(Ok(MetricsMiddlewareService { service }))
    }
}
 
pub struct MetricsMiddlewareService<S> {
    service: S,
}
 
impl<S> Service<ServiceRequest> for MetricsMiddlewareService<S>
where
    S: Service<ServiceRequest, Response = ServiceResponse, Error = Error>,
{
    type Response = ServiceResponse;
    type Error = Error;
    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>>>>;
 
    fn poll_ready(&self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        self.service.poll_ready(cx)
    }
 
    fn call(&self, req: ServiceRequest) -> Self::Future {
        let start_time = std::time::Instant::now();
        let method = req.method().to_string();
        let path = req.path().to_string();
        
        let fut = self.service.call(req);
        
        Box::pin(async move {
            let result = fut.await;
            let duration = start_time.elapsed().as_secs_f64();
            
            match &result {
                Ok(response) => {
                    HTTP_REQUEST_DURATION
                        .with_label_values(&[&method, &path, &response.status().as_str()])
                        .observe(duration);
                }
                Err(_) => {
                    HTTP_REQUEST_DURATION
                        .with_label_values(&[&method, &path, "500"])
                        .observe(duration);
                }
            }
            
            result
        })
    }
}

use serde::{Deserialize, Serialize};
use serde_with::{serde_as, DisplayFromStr};
 
// Оптимизированная структура для передачи между сервисами
#[serde_as]
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct OrderSummary {
    #[serde_as(as = "DisplayFromStr")]
    pub id: Uuid,
    #[serde_as(as = "DisplayFromStr")]
    pub customer_id: Uuid,
    pub status: OrderStatus,
    #[serde(with = "rust_decimal::serde::float")]
    pub total_amount: Decimal,
    pub items_count: u32,
    #[serde(with = "chrono::serde::ts_seconds")]
    pub created_at: DateTime<Utc>,
}
 
// Используем бинарную сериализацию для внутренних вызовов
use bincode::{serialize, deserialize};
 
pub async fn send_internal_message<T: serde::Serialize>(
    client: &reqwest::Client,
    url: &str,
    payload: &T
) -> Result<(), reqwest::Error> {
    let binary_data = serialize(payload).unwrap();
    
    client
        .post(url)
        .header("Content-Type", "application/octet-stream")
        .body(binary_data)
        .send()
        .await?;
    
    Ok(())
}

use sqlx::{PgPool, Row};
use tokio_cron_scheduler::{JobScheduler, Job};
 
pub struct BackupManager {
    primary_pool: PgPool,
    replica_pool: PgPool,
}
 
impl BackupManager {
    pub async fn schedule_health_checks(&self) -> Result<(), Box<dyn std::error::Error>> {
        let sched = JobScheduler::new().await?;
        
        let primary_pool = self.primary_pool.clone();
        sched.add(
            Job::new_async("0 */5 * * * *", move |_uuid, _l| {
                let pool = primary_pool.clone();
                Box::pin(async move {
                    if let Err(e) = check_database_health(&pool).await {
                        tracing::error!("Primary database health check failed: {:?}", e);
                        // Отправляем алерт в Slack/PagerDuty
                        send_alert("Primary DB unhealthy", &format!("{:?}", e)).await;
                    }
                })
            })?
        ).await?;
        
        let replica_pool = self.replica_pool.clone();
        sched.add(
            Job::new_async("30 */5 * * * *", move |_uuid, _l| {
                let pool = replica_pool.clone();
                Box::pin(async move {
                    match check_replication_lag(&pool).await {
                        Ok(lag) if lag > Duration::from_secs(60) => {
                            tracing::warn!("Replication lag is {} seconds", lag.as_secs());
                            send_alert("High replication lag", &format!("Lag: {}s", lag.as_secs())).await;
                        }
                        Err(e) => {
                            tracing::error!("Failed to check replication lag: {:?}", e);
                            send_alert("Replica DB check failed", &format!("{:?}", e)).await;
                        }
                        _ => {}
                    }
                })
            })?
        ).await?;
        
        sched.start().await?;
        Ok(())
    }
}
 
async fn check_database_health(pool: &PgPool) -> Result<(), sqlx::Error> {
    // Проверяем простой SELECT
    sqlx::query("SELECT 1").fetch_one(pool).await?;
    
    // Проверяем операции записи
    sqlx::query("INSERT INTO health_check (timestamp) VALUES ($1)")
        .bind(chrono::Utc::now())
        .execute(pool)
        .await?;
    
    // Очищаем тестовые записи старше часа
    sqlx::query("DELETE FROM health_check WHERE timestamp < $1")
        .bind(chrono::Utc::now() - chrono::Duration::hours(1))
        .execute(pool)
        .await?;
    
    Ok(())
}
 
async fn check_replication_lag(replica_pool: &PgPool) -> Result<Duration, sqlx::Error> {
    let row = sqlx::query(
        "SELECT EXTRACT(EPOCH FROM (now() - pg_last_xact_replay_timestamp())) as lag"
    )
    .fetch_one(replica_pool)
    .await?;
    
    let lag_seconds: f64 = row.get("lag");
    Ok(Duration::from_secs_f64(lag_seconds))
}

use std::env;
use reqwest::Client;
use serde_json::Value;
 
pub struct SecretManager {
    vault_client: Client,
    vault_addr: String,
    vault_token: String,
}
 
impl SecretManager {
    pub fn new() -> Result<Self, Box<dyn std::error::Error>> {
        let vault_addr = env::var("VAULT_ADDR")
            .unwrap_or_else(|_| "http://vault:8200".to_string());
        let vault_token = env::var("VAULT_TOKEN")
            .expect("VAULT_TOKEN environment variable required");
        
        Ok(Self {
            vault_client: Client::new(),
            vault_addr,
            vault_token,
        })
    }
    
    pub async fn get_secret(&self, path: &str) -> Result<Value, reqwest::Error> {
        let url = format!("{}/v1/secret/data/{}", self.vault_addr, path);
        
        let response = self.vault_client
            .get(&url)
            .header("X-Vault-Token", &self.vault_token)
            .send()
            .await?;
        
        if !response.status().is_success() {
            return Err(reqwest::Error::from(response.error_for_status().unwrap_err()));
        }
        
        let json: Value = response.json().await?;
        Ok(json["data"]["data"].clone())
    }
    
    pub async fn get_database_url(&self) -> Result<String, Box<dyn std::error::Error>> {
        let secrets = self.get_secret("database/postgres").await?;
        
        let username = secrets["username"].as_str().unwrap();
        let password = secrets["password"].as_str().unwrap(); 
        let host = secrets["host"].as_str().unwrap();
        let database = secrets["database"].as_str().unwrap();
        
        Ok(format!("postgresql://{}:{}@{}/{}", username, password, host, database))
    }
}

// .cargo/config.toml
[build]
rustc-wrapper = "sccache"
 
[target.x86_64-unknown-linux-musl]
linker = "rust-lld"

FROM rust:1.75-slim as deps-builder
WORKDIR /usr/src/app
# Копируем только файлы зависимостей
COPY Cargo.toml Cargo.lock ./
# Создаём фиктивный main.rs для сборки зависимостей
RUN mkdir src && echo "fn main() {}" > src/main.rs
RUN cargo build --release --target x86_64-unknown-linux-musl
RUN rm -rf src
 
FROM deps-builder as app-builder
COPY src ./src
# Пересобираем только изменённый код
RUN touch src/main.rs && cargo build --release --target x86_64-unknown-linux-musl
 
FROM scratch
COPY --from=app-builder /usr/src/app/target/x86_64-unknown-linux-musl/release/service /
CMD ["/service"]

[profile.release]
opt-level = "z"  # Оптимизация по размеру
lto = true
codegen-units = 1
panic = "abort"
strip = "symbols"

use reqwest::Client;
use serde_xml_rs::{from_str, to_string};
 
pub struct SoapClient {
    client: Client,
    endpoint: String,
}
 
impl SoapClient {
    pub async fn call_legacy_service<T, R>(&self, request: T) -> Result<R, SoapError>
    where
        T: serde::Serialize,
        R: serde::de::DeserializeOwned,
    {
        let soap_body = format!(
            r#"<?xml version="1.0" encoding="UTF-8"?>
            <soap:Envelope xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/">
                <soap:Body>
                    {}
                </soap:Body>
            </soap:Envelope>"#,
            to_string(&request)?
        );
 
        let response = self
            .client
            .post(&self.endpoint)
            .header("Content-Type", "text/xml; charset=utf-8")
            .header("SOAPAction", "processOrder")
            .body(soap_body)
            .send()
            .await?;
 
        if !response.status().is_success() {
            return Err(SoapError::HttpError(response.status()));
        }
 
        let body = response.text().await?;
        let parsed: R = from_str(&body)?;
        Ok(parsed)
    }
}

use std::collections::HashMap;
use tokio::sync::RwLock;
use std::sync::Arc;
 
pub struct ConnectionManager {
    connections: Arc<RwLock<HashMap<Uuid, Connection>>>,
}
 
impl ConnectionManager {
    pub async fn add_connection(&self, id: Uuid, conn: Connection) {
        let mut connections = self.connections.write().await;
        connections.insert(id, conn);
        
        // Периодически чистим мёртвые соединения
        tokio::spawn({
            let connections = self.connections.clone();
            async move {
                let mut interval = tokio::time::interval(Duration::from_secs(60));
                loop {
                    interval.tick().await;
                    let mut conns = connections.write().await;
                    conns.retain(|_, conn| !conn.is_closed());
                }
            }
        });
    }
}

use tokio::signal;
use tokio::sync::oneshot;
use tracing::{info, error};
 
pub async fn run_with_graceful_shutdown<F>(app_future: F) 
where
    F: Future<Output = Result<(), Box<dyn std::error::Error + Send + Sync>>>,
{
    let (shutdown_tx, shutdown_rx) = oneshot::channel();
    
    // Обработчик сигналов
    tokio::spawn(async move {
        match signal::ctrl_c().await {
            Ok(_) => {
                info!("Received SIGINT, starting graceful shutdown");
                let _ = shutdown_tx.send(());
            }
            Err(e) => {
                error!("Failed to install SIGINT handler: {}", e);
            }
        }
    });
    
    // Основное приложение
    tokio::select! {
        result = app_future => {
            if let Err(e) = result {
                error!("Application error: {:?}", e);
            }
        }
        _ = shutdown_rx => {
            info!("Shutdown signal received, stopping application");
        }
    }
    
    // Даём время на завершение активных requests
    tokio::time::sleep(Duration::from_secs(10)).await;
    info!("Graceful shutdown complete");
}

pub async fn process_complex_operation(&self) -> Result<ProcessResult, ProcessError> {
    let inventory_result = self.inventory_service
        .reserve_items(&items)
        .await
        .map_err(ProcessError::InventoryError)?;
    
    let payment_result = self.payment_service
        .charge_customer(&customer, &amount)
        .await
        .map_err(|e| {
            // Компенсируем резерв товара
            let _ = self.inventory_service.release_reservation(inventory_result.reservation_id);
            ProcessError::PaymentError(e)
        })?;
    
    let shipping_result = self.shipping_service
        .schedule_delivery(&order)
        .await
        .map_err(|e| {
            // Компенсируем и резерв, и платёж
            let _ = self.inventory_service.release_reservation(inventory_result.reservation_id);
            let _ = self.payment_service.refund(payment_result.transaction_id);
            ProcessError::ShippingError(e)
        })?;
    
    Ok(ProcessResult {
        inventory: inventory_result,
        payment: payment_result,
        shipping: shipping_result,
    })
}

[dependencies]
console-subscriber = { version = "0.2", features = ["parking_lot"] }
 
[profile.dev]
debug = true  # Нужно для tokio-console

fn main() {
    console_subscriber::init();
    tokio::runtime::Builder::new_multi_thread()
        .enable_all()
        .build()
        .unwrap()
        .block_on(async_main())
}