Running async journeys while others wait

/// A payment client: it charges an order's amount to a card, and gives back a receipt, or nothing
/// when the card has expired, or the error of a network that failed; and it says which card
/// replaced one, if any did. Each answer comes over the network.
#[async_trait]
trait Payments: Send + Sync {
    async fn charge(
        &self,
        order: &str,
        card: &str,
        amount: i64,
    ) -> Result<Option<String>, NetworkDown>;
    async fn replacement(&self, card: &str) -> Option<String>;
}

/// A carrier: it books a parcel of this weight in kg and gives back its tracking number, or the
/// error of a booking system that failed, then says whether it collects the parcel. Each answer
/// comes over the network.
#[async_trait]
trait Carrier: Send + Sync {
    async fn book(&self, order: &str, weight: i64) -> Result<String, BookingFailed>;
    async fn collects(&self, weight: i64) -> bool;
}

/// The network the fakes answer over: it hands the thread back once before the answer comes, as a
/// real call does while it waits.
async fn network() {
    tokio::task::yield_now().await;
}

/// The shop's bank, faked: a card is its holder's name and the year it expires, and the bank
/// declines a card that expired before this year. Its network fails on any amount over 100. It
/// prints what it does once it has answered.
struct Bank {
    this_year: u32,
}

#[async_trait]
impl Payments for Bank {
    async fn charge(
        &self,
        order: &str,
        card: &str,
        amount: i64,
    ) -> Result<Option<String>, NetworkDown> {
        network().await;
        if amount > 100 {
            println!("the bank's network fails while charging order {order}");
            return Err(NetworkDown);
        }
        if expiry(card).is_some_and(|year| year < self.this_year) {
            println!("the bank declines card {card}: it has expired");
            return Ok(None);
        }
        println!("the bank charges {amount} for order {order}");
        Ok(Some(format!("R-{order}")))
    }

    /// A card that expired in the last three years was replaced by its holder's card that expires
    /// in three years. An older one was not.
    async fn replacement(&self, card: &str) -> Option<String> {
        network().await;
        if expiry(card)? + 3 < self.this_year {
            println!("the bank says nothing replaced card {card}");
            return None;
        }
        let (holder, _) = card.rsplit_once('-')?;
        let replacement = format!("{holder}-{}", self.this_year + 3);
        println!("the bank says card {card} was replaced by {replacement}");
        Some(replacement)
    }
}

#[step]
impl Price {
    async fn run(self, #[contributor] mut contributor: Contributor<'_>) -> Result<Outcome, Error> {
        contributor.contribute("amount", self.price - self.discount.unwrap_or(0));
        Self::succeed()
    }
}

#[step]
impl Ship {
    async fn run(
        self,
        #[service] carrier: &dyn Carrier,
        #[contributor] mut contributor: Contributor<'_>,
        #[reporter] mut reporter: AsyncStepReporter<'_>,
    ) -> Result<Outcome, Error> {
        if self.gift_card {
            return Self::skip_because("nothing-to-ship");
        }
        if self.weight > 80 {
            return Self::fail("over-parcel-limit");
        }
        let tracking = carrier.book(&self.order, self.weight).await?;
        contributor.contribute("tracking", tracking);
        if !carrier.collects(self.weight).await {
            return Self::fail("too-heavy");
        }
        reporter
            .info(format!("the parcel leaves with receipt {}", self.receipt))
            .await?;
        Self::succeed()
    }
}

/// What the shop does for its policies: tell a customer something.
#[async_trait]
trait Messages: Sync {
    async fn tell(&self, customer: &str, message: &str);
}

/// What the shop does for its policies: give an order's money back.
#[async_trait]
trait Refunds: Sync {
    async fn refund(&self, receipt: &str, amount: i64);
}

#[role]
#[async_trait]
impl Refunds for Orders {
    async fn refund(&self, receipt: &str, amount: i64) {
        network().await;
        println!("the bank gives {amount} back on receipt {receipt}");
    }
}

#[role]
#[async_trait]
impl Messages for Orders {
    async fn tell(&self, customer: &str, message: &str) {
        network().await;
        println!("{} to {customer}: {message}", self.channel);
    }
}

/// Deals with a shipping that fails for good. It gives up a parcel over the shop's limit rather
/// than try it again, since the limit is the same on every attempt, whichever carrier comes. And
/// whatever gave the shipping up, it refunds the order's charge.
#[derive(Default)]
struct FailedShipping;

#[step_policy(name = "failed_shipping")]
impl FailedShipping {
    #[on_step_retry]
    async fn before_retrying(
        &self,
        #[reason] reason: Option<Reason>,
        #[from_workflow] receipt: String,
        #[from_workflow] amount: i64,
        #[role] refunds: &dyn Refunds,
        #[contributor] contributor: Contributor<'_>,
    ) -> Result<Option<FailWorkflow>, Error> {
        if reason.as_ref().map(Reason::code) == Some("over-parcel-limit") {
            self.refund(refunds, contributor, &receipt, amount, "over-parcel-limit")
                .await;
            return Self::fail_workflow("no-carrier-can-take-it");
        }
        Self::carry_on()
    }

    #[on_step_failure]
    async fn given_up(
        &self,
        #[cause] cause: StepFailureCause,
        #[from_workflow] receipt: String,
        #[from_workflow] amount: i64,
        #[role] refunds: &dyn Refunds,
        #[contributor] contributor: Contributor<'_>,
    ) -> Result<Option<FailWorkflow>, Error> {
        self.refund(refunds, contributor, &receipt, amount, &cause.to_string())
            .await;
        Self::carry_on()
    }

    /// Gives the order's money back, and records why.
    async fn refund(
        &self,
        refunds: &dyn Refunds,
        mut contributor: Contributor<'_>,
        receipt: &str,
        amount: i64,
        why: &str,
    ) {
        refunds.refund(receipt, amount).await;
        contributor.contribute("refund", format!("{amount} back on {receipt}: {why}"));
    }
}

/// An instance of the shop's workflow for an order paid with this card.
fn order(
    order: &str,
    customer: &str,
    price: i64,
    discount: Option<i64>,
    parcel: Parcel,
    card: &str,
) -> Result<Instance<Orders, Asynchronous>, Box<dyn std::error::Error>> {
    Ok(Orders::builder()
        .channel("email")
        .initial_data(NewOrder {
            order: order.to_owned(),
            customer: customer.to_owned(),
            price,
            discount,
            parcel,
            card: Some(card.to_owned()),
            gift_card: false,
        })
        .create()?)
}

/// Runs an order's journey with an executor of its own.
async fn journey(instance: Instance<Orders, Asynchronous>) -> Result<JourneyResult, Refusal> {
    local_executor_for!(Orders).run(instance).await
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let (ann, gil, jo) = tokio::join!(
        journey(order(
            "A-1",
            "Ann",
            40,
            Some(10),
            Parcel {
                weight: 2,
                size: [30, 20, 10],
            },
            "ann-2028",
        )?),
        journey(order(
            "A-7",
            "Gil",
            45,
            None,
            Parcel {
                weight: 60,
                size: [120, 80, 60],
            },
            "gil-2027",
        )?),
        journey(order(
            "A-8",
            "Jo",
            30,
            None,
            Parcel {
                weight: 1,
                size: [20, 15, 5],
            },
            "jo-2024",
        )?),
    );
    for result in [ann?, gil?, jo?] {
        println!("order {} {}", result.journey_id, result.status.kind());
    }
    Ok(())
}