Guards and authorization

use std::sync::Arc;
use lilyrs::consumer::{Consumer, ConsumerError, Extensions, Injectable, ServiceTrait};
use lilyrs::queue::{
    DeliveryContext, Json, Local, QueueDeliveryExchange, QueueGuard, QueueHandlerError,
    QueueMiddleware, QueuePipelineComponentInitError, Service, async_trait, guard,
    queue, queue_service,
};
use serde::Deserialize;

#[derive(Clone)]
struct DeliveryAudit { event_id: String }

struct Audit;
#[async_trait]
impl QueueMiddleware for Audit {
    async fn new(_: Arc<Extensions>) -> Result<Self, QueuePipelineComponentInitError> {
        Ok(Self)
    }
    async fn before_delivery(&self, exchange: &mut QueueDeliveryExchange<'_>)
        -> Result<(), QueueHandlerError>
    {
        let event_id = exchange.context().event_id().into_inner().to_string();
        exchange.insert_local(DeliveryAudit { event_id })?;
        Ok(())
    }
}

#[derive(Injectable)]
#[service(lifetime = "Scoped")]
struct OrderPolicy;
impl ServiceTrait for OrderPolicy {}
impl OrderPolicy {
    fn accepts(&self, context: &DeliveryContext) -> bool {
        context.schema_version().into_inner() == 1
    }
}

struct OrderGuard;
#[async_trait]
impl QueueGuard for OrderGuard {
    async fn new(_: Arc<Extensions>) -> Result<Self, QueuePipelineComponentInitError> {
        Ok(Self)
    }
    async fn can_activate(&self, exchange: &mut QueueDeliveryExchange<'_>)
        -> Result<(), QueueHandlerError>
    {
        let policy = exchange.service::<OrderPolicy>().await
            .map_err(|error| QueueHandlerError::retryable_with_source("POLICY_UNAVAILABLE", error))?;
        if policy.accepts(exchange.context()) { Ok(()) }
        else { Err(QueueHandlerError::permanent("ORDER_NOT_ADMITTED")) }
    }
}

#[derive(Deserialize)]
struct OrderCreated { order_id: String }
#[derive(Injectable)]
#[service(lifetime = "Singleton")]
struct OrderWorker;
impl ServiceTrait for OrderWorker {}

#[queue_service]
#[guard(OrderGuard)]
impl OrderWorker {
    #[queue("orders.created", version = 1, content = "json")]
    async fn created(&self, Service(_policy): Service<OrderPolicy>,
        Local(audit): Local<DeliveryAudit>, Json(event): Json<OrderCreated>)
        -> Result<(), QueueHandlerError>
    {
        if event.order_id.is_empty() {
            return Err(QueueHandlerError::permanent("ORDER_ID_REQUIRED"));
        }
        println!("event={}, order={}", audit.event_id, event.order_id);
        Ok(())
    }
}

#[tokio::main]
async fn main() -> Result<(), ConsumerError> {
    Consumer::builder().middleware::<Audit>().run().await
}