worker/

queue.rs

use std::{
    convert::{TryFrom, TryInto},
    marker::PhantomData,
};

use crate::{env::EnvBinding, Date, Error, Result};
use js_sys::Array;
use serde::{de::DeserializeOwned, Serialize};
use wasm_bindgen::{prelude::*, JsCast};
use wasm_bindgen_futures::JsFuture;
use worker_sys::{Message as MessageSys, MessageBatch as MessageBatchSys, Queue as EdgeQueue};

/// A batch of messages that are sent to a consumer Worker.
#[derive(Debug)]
pub struct MessageBatch<T> {
    inner: MessageBatchSys,
    phantom: PhantomData<T>,
}

impl<T> MessageBatch<T> {
    /// The name of the Queue that belongs to this batch.
    pub fn queue(&self) -> String {
        self.inner.queue().unwrap().into()
    }

    /// Marks every message to be retried in the next batch.
    pub fn retry_all(&self) {
        self.inner.retry_all(JsValue::null()).unwrap();
    }

    /// Marks every message to be retried in the next batch with options.
    pub fn retry_all_with_options(&self, queue_retry_options: &QueueRetryOptions) {
        self.inner
            // SAFETY: QueueRetryOptions is controlled by this module and all data in it is serializable to a js value.
            .retry_all(serde_wasm_bindgen::to_value(&queue_retry_options).unwrap())
            .unwrap();
    }

    /// Marks every message acknowledged in the batch.
    pub fn ack_all(&self) {
        self.inner.ack_all().unwrap();
    }

    /// Iterator for raw messages in the message batch. Ordering of messages is not guaranteed.
    pub fn raw_iter(&self) -> RawMessageIter {
        let messages = self.inner.messages().unwrap();
        RawMessageIter {
            range: 0..messages.length(),
            array: messages,
        }
    }
}

impl<T: DeserializeOwned> MessageBatch<T> {
    /// An array of messages in the batch. Ordering of messages is not guaranteed.
    pub fn messages(&self) -> Result<Vec<Message<T>>> {
        self.iter().collect()
    }

    /// Iterator for messages in the message batch. Ordering of messages is not guaranteed.
    pub fn iter(&self) -> MessageIter<T> {
        let messages = self.inner.messages().unwrap();
        MessageIter {
            range: 0..messages.length(),
            array: messages,
            marker: PhantomData,
        }
    }
}

impl<T> From<MessageBatchSys> for MessageBatch<T> {
    fn from(value: MessageBatchSys) -> Self {
        Self {
            inner: value,
            phantom: PhantomData,
        }
    }
}

/// A message that is sent to a consumer Worker.
#[derive(Debug)]
pub struct Message<T> {
    inner: MessageSys,
    body: T,
}

impl<T> Message<T> {
    /// The body of the message.
    pub fn body(&self) -> &T {
        &self.body
    }

    /// The body of the message.
    pub fn into_body(self) -> T {
        self.body
    }

    /// The raw body of the message.
    pub fn raw_body(&self) -> JsValue {
        self.inner().body().unwrap()
    }
}

impl<T> TryFrom<RawMessage> for Message<T>
where
    T: DeserializeOwned,
{
    type Error = Error;

    fn try_from(value: RawMessage) -> std::result::Result<Self, Self::Error> {
        let body = serde_wasm_bindgen::from_value(value.body())?;
        Ok(Self {
            inner: value.inner,
            body,
        })
    }
}

/// A message that is sent to a consumer Worker.
#[derive(Debug)]
pub struct RawMessage {
    inner: MessageSys,
}

impl RawMessage {
    /// The body of the message.
    pub fn body(&self) -> JsValue {
        self.inner.body().unwrap()
    }
}

impl From<MessageSys> for RawMessage {
    fn from(value: MessageSys) -> Self {
        Self { inner: value }
    }
}

trait MessageSysInner {
    fn inner(&self) -> &MessageSys;
}

impl MessageSysInner for RawMessage {
    fn inner(&self) -> &MessageSys {
        &self.inner
    }
}

impl<T> MessageSysInner for Message<T> {
    fn inner(&self) -> &MessageSys {
        &self.inner
    }
}

#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
#[derive(Debug)]
/// Optional configuration when marking a message or a batch of messages for retry.
pub struct QueueRetryOptions {
    delay_seconds: Option<u32>,
}

#[derive(Debug)]
pub struct QueueRetryOptionsBuilder {
    delay_seconds: Option<u32>,
}

impl QueueRetryOptionsBuilder {
    /// Creates a new retry options builder.
    pub fn new() -> Self {
        Self {
            delay_seconds: None,
        }
    }

    #[must_use]
    /// The number of seconds to delay a message for within the queue, before it can be delivered to a consumer
    pub fn with_delay_seconds(mut self, delay_seconds: u32) -> Self {
        self.delay_seconds = Some(delay_seconds);
        self
    }

    /// Build the retry options.
    pub fn build(self) -> QueueRetryOptions {
        QueueRetryOptions {
            delay_seconds: self.delay_seconds,
        }
    }
}

pub trait MessageExt {
    /// A unique, system-generated ID for the message.
    fn id(&self) -> String;

    /// A timestamp when the message was sent.
    fn timestamp(&self) -> Date;

    /// Marks message to be retried.
    fn retry(&self);

    /// Marks message to be retried with options.
    fn retry_with_options(&self, queue_retry_options: &QueueRetryOptions);

    /// Marks message acknowledged.
    fn ack(&self);
}

impl<T: MessageSysInner> MessageExt for T {
    /// A unique, system-generated ID for the message.
    fn id(&self) -> String {
        self.inner().id().unwrap().into()
    }

    /// A timestamp when the message was sent.
    fn timestamp(&self) -> Date {
        Date::from(self.inner().timestamp().unwrap())
    }

    /// Marks message to be retried.
    fn retry(&self) {
        self.inner().retry(JsValue::null()).unwrap();
    }

    /// Marks message to be retried with options.
    fn retry_with_options(&self, queue_retry_options: &QueueRetryOptions) {
        self.inner()
            // SAFETY: QueueRetryOptions is controlled by this module and all data in it is serializable to a js value.
            .retry(serde_wasm_bindgen::to_value(&queue_retry_options).unwrap())
            .unwrap();
    }

    /// Marks message acknowledged.
    fn ack(&self) {
        self.inner().ack().unwrap();
    }
}

#[derive(Debug)]
pub struct MessageIter<T> {
    range: std::ops::Range<u32>,
    array: Array,
    marker: PhantomData<T>,