ruststream-pulsar 0.7.1

//! The contexts a Pulsar subscription hands its handlers, and the keys that read them.
//!
//! Pulsar topics are a retained log, so the two things a body needs to reposition itself are
//! per-delivery state: where this message sits ([`Position`]) and the handle that moves the
//! subscription ([`SeekHandle`]). Both ride [`PulsarContext`], built once per delivery; a batch
//! spans many deliveries, so [`PulsarBatchContext`] carries only the subscription-scoped half.

use ruststream::{BuildBatchContext, BuildContext, ContextField, Field, Positioned};

use crate::message::{PulsarMessage, PulsarPosition};
use crate::subscriber::PulsarSeeker;

/// The per-delivery context of a Pulsar subscription: this message's position in the topic's
/// retained log, and the subscription's own seeker.
///
/// The runtime builds one per dispatched delivery (see [`BuildContext`]), and handlers read its
/// fields by key - [`Position`] and [`SeekHandle`] - through the `Ctx` extractor or
/// `ctx.context(..)`. A body that repositions its own subscription names this type as its `C`
/// axis and needs nothing else.
///
/// Building it costs no allocation: the position is the delivery's message id, and the seeker
/// is a clone of the channel to the subscription's driver task.
///
/// # Examples
///
/// ```
/// # mod demo {
/// use ruststream_pulsar::prelude::*;
/// use serde::Deserialize;
///
/// #[derive(Deserialize)]
/// struct Job {
///     id: u64,
///     rewind: bool,
/// }
///
/// /// Logs where each job sits, and replays the retained backlog when a job asks for it.
/// #[subscriber(PulsarSubscription::new("jobs", "workers"))]
/// async fn work(job: &Job, ctx: &mut Context<'_, PulsarContext>) -> HandlerOutcome {
///     tracing::info!(job = job.id, position = ?ctx.context(Position), "job received");
///     if job.rewind
///         && ctx
///             .context(SeekHandle)
///             .seek(PulsarPosition::earliest())
///             .await
///             .is_err()
///     {
///         return HandlerOutcome::retry();
///     }
///     HandlerOutcome::ack()
/// }
/// # }
/// # fn main() {}
/// ```
#[derive(Debug)]
pub struct PulsarContext {
    position: PulsarPosition,
    seeker: PulsarSeeker,
}

impl BuildContext<PulsarMessage> for PulsarContext {
    fn build(msg: &PulsarMessage) -> Self {
        Self {
            position: Positioned::position(msg),
            seeker: msg.seeker(),
        }
    }
}

/// The subscription-scoped batch context of a Pulsar subscription: its seeker, shared by every
/// delivery of the batch.
///
/// The runtime builds one per dispatched batch from the batch's first delivery (see
/// [`BuildBatchContext`]), and a batch body reads it by key - [`SeekHandle`] - through
/// `ctx.context(..)`. A [`Position`] has no place here: a batch spans many deliveries, so where
/// to seek rides the elements themselves (a `&[Message<H, T>]` batch reads it off each element's
/// header contract). Keeping this a separate type from [`PulsarContext`] is what rejects a batch
/// body asking for per-delivery fields at compile time.
///
/// Pulsar's client hands over one delivery at a time, so a batch here is assembled on the client
/// from the size the mount site's `batch(n)` names. The body sees the same context, seeker and
/// settlement as on a broker that batches on the wire.
///
/// # Examples
///
/// ```
/// # mod demo {
/// use ruststream_pulsar::prelude::*;
/// use serde::Deserialize;
///
/// #[derive(Deserialize)]
/// struct Job {
///     id: u64,
///     rewind: bool,
/// }
///
/// /// A batch that carries the rewind marker replays the retained backlog once it is settled:
/// /// the next batch opens at the beginning of the log.
/// #[subscriber(PulsarSubscription::new("jobs.bulk", "bulk-workers"))]
/// async fn work(jobs: &[Job], ctx: &mut Context<'_, PulsarBatchContext>) -> HandlerOutcome {
///     let first = jobs.first().map(|job| job.id);
///     tracing::info!(count = jobs.len(), ?first, "jobs received");
///     if jobs.iter().any(|job| job.rewind)
///         && ctx
///             .context(SeekHandle)
///             .seek(PulsarPosition::earliest())
///             .await
///             .is_err()
///     {
///         return HandlerOutcome::retry();
///     }
///     HandlerOutcome::ack()
/// }
///
/// #[ruststream::app]
/// fn app() -> impl App {
///     RustStream::new(AppInfo::new("jobs", "0.1.0")).with_broker(
///         PulsarBroker::new("pulsar://localhost:6650"),
///         |b| {
///             b.include(work.batch(nonzero!(100)));
///         },
///     )
/// }
/// # }
/// # fn main() {}
/// ```
#[derive(Debug)]
pub struct PulsarBatchContext {
    seeker: PulsarSeeker,
}

impl BuildBatchContext<PulsarMessage> for PulsarBatchContext {
    fn build(first: &PulsarMessage) -> Self {
        Self {
            seeker: first.seeker(),
        }
    }
}

/// The key reading this delivery's [`PulsarPosition`] out of [`PulsarContext`].
///
/// The value is the delivery's own message id, so it carries the framework's pinned contract:
/// seeking to it redelivers exactly this message.
///
/// A message id is not `Copy`, so `ctx.context(Position)` borrows it and only the `Ctx<Position>`
/// extractor - which must yield an owned value - copies one per delivery.
///
/// # Examples
///
/// ```
/// use ruststream_pulsar::Position;
/// use ruststream_pulsar::prelude::*;
/// # #[derive(serde::Deserialize)]
/// # struct Job { id: u64 }
///
/// #[subscriber(PulsarSubscription::new("jobs", "auditors"))]
/// async fn audit(job: &Job, Ctx(at): Ctx<Position>) -> HandlerOutcome {
///     tracing::info!(job = job.id, position = ?at, "job audited");
///     HandlerOutcome::ack()
/// }
/// ```
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash)]
pub struct Position;

impl ContextField for Position {
    type Context = PulsarContext;
    type Value = PulsarPosition;

    fn read(self, src: &PulsarContext) -> PulsarPosition {
        src.position.clone()
    }
}

impl Field<PulsarContext> for Position {
    type Value<'a> = &'a PulsarPosition;

    fn get(self, src: &PulsarContext) -> &PulsarPosition {
        &src.position
    }
}

/// The key reading the subscription's [`PulsarSeeker`] out of [`PulsarContext`] (and out of
/// [`PulsarBatchContext`] on the batch path): the reposition handle every delivery of the
/// subscription shares.
///
/// One seek covers every topic and every partition of the subscription's consumer.
///
/// # Examples
///
/// ```
/// use ruststream_pulsar::SeekHandle;
/// use ruststream_pulsar::prelude::*;
/// # #[derive(serde::Deserialize)]
/// # struct Job { id: u64, poisoned_until: Option<u64> }
///
/// /// Skips forward past a region the producer marked poisoned.
/// #[subscriber(PulsarSubscription::new("jobs", "workers"))]
/// async fn work(job: &Job, Ctx(seeker): Ctx<SeekHandle>) -> HandlerOutcome {
///     if let Some(resume_at) = job.poisoned_until
///         && seeker.seek(PulsarPosition::timestamp(resume_at)).await.is_err()
///     {
///         return HandlerOutcome::retry();
///     }
///     HandlerOutcome::ack()
/// }
/// ```
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash)]
pub struct SeekHandle;

impl ContextField for SeekHandle {
    type Context = PulsarContext;
    type Value = PulsarSeeker;

    fn read(self, src: &PulsarContext) -> PulsarSeeker {
        src.seeker.clone()
    }
}

impl Field<PulsarContext> for SeekHandle {
    type Value<'a> = &'a PulsarSeeker;

    fn get(self, src: &PulsarContext) -> &PulsarSeeker {
        &src.seeker
    }
}

impl Field<PulsarBatchContext> for SeekHandle {
    type Value<'a> = &'a PulsarSeeker;

    fn get(self, src: &PulsarBatchContext) -> &PulsarSeeker {
        &src.seeker
    }
}