rovenue/funnel/

session.rs

use std::sync::{Arc, Mutex};

use funnel_core::actions::FunnelEffect as CoreEffect;
use funnel_core::answer::AnswerValue;
use funnel_core::machine::{FunnelError, FunnelMachine, Step};
use funnel_core::schema::{
    Funnel, OfferKind, Offering, OfferingPackage, PackageOffer, PackagePrice, PeriodUnit,
    PriceFacts,
};
use funnel_core::session::SessionState;
use serde::{Deserialize, Serialize};

use crate::error::{RovenueError, RovenueResult};
use crate::events::{EventEnvelope, EventQueue, FunnelContext, EVENT_WIRE_VERSION};
use crate::offerings::CoreOfferingProduct;
use crate::placements::types::FunnelWire;
use crate::placements::CoreFunnel;
use crate::time::{rfc3339_millis, Clock};

/// What the host must do, as the façades see it.
///
/// `SetAttribute` is deliberately absent. librovenue intercepts it below
/// and routes it into the existing attributes module, so it never crosses
/// this boundary and is never implemented three times.
#[derive(Debug, Clone, PartialEq)]
pub enum FunnelEffect {
    /// Close the funnel's view. Produced by `close_funnel`.
    Dismiss,
    /// Open a URL. Unlike the web funnel this does not end the funnel: an
    /// app deep-link scheme is legitimate here, and handing off to another
    /// app leaves the session where it was.
    OpenUrl {
        /// The URL to open, exactly as authored.
        url: String,
    },
    // No RequestPermission variant: a permission screen announces itself
    // through the resolved screen's `permission` field, which is
    // declarative and survives a redraw. See funnel_core::actions::FunnelEffect.
    /// Show an alert. Already localized and interpolated.
    ShowAlert {
        /// Optional title.
        title: Option<String>,
        /// The body, never empty.
        message: String,
        /// Optional dismiss-button label.
        button_label: Option<String>,
    },
    /// An app-defined action the funnel author named.
    Custom {
        /// The author's identifier for this action.
        action_id: String,
        /// Parameters as a JSON object string (string-not-record, as above).
        params_json: String,
    },
    /// Buy the selected package through StoreKit / Play Billing.
    Purchase {
        /// The offering package the visitor selected.
        package_id: String,
    },
    /// Restore previous purchases.
    RestorePurchases,
}

/// One step of a funnel, as it crosses the FFI.
///
/// `screen_json` is a raw JSON string rather than a record: the uniffi
/// record→HashMap mapping cannot carry arbitrary nested JSON, and it also
/// means a new block type never forces a UDL edit.
#[derive(Debug, Clone)]
pub struct FunnelStep {
    /// The resolved screen to draw, or `None` once the funnel is over.
    pub screen_json: Option<String>,
    /// What the host must do before or alongside drawing.
    pub effects: Vec<FunnelEffect>,
    /// True once the funnel has ended; `screen_json` is then `None`.
    pub finished: bool,
}

/// A store-fetched price for one package (D13). The façade hands these in
/// through `FunnelSession::set_package_prices` once its async StoreKit /
/// Play Billing lookup resolves — `open` cannot wait on it, so a package
/// row draws unpriced (but titled — see `map_offering`) until this lands.
///
/// The FFI twin of funnel-core's `PackagePrice`: forwarded to the machine
/// (`FunnelMachine::set_package_prices`), so the resolver sees the price
/// when it builds copy (`{{product.*}}`) and computes savings. Matched by
/// `identifier` (the package SLOT identifier, the key
/// `OfferingPackage::identifier` carries); an identifier the offering lacks
/// is ignored. A `None` `localized_period` / `localized_title` keeps what
/// the served offering carried; the numeric facts replace the package's
/// wholesale.
#[derive(Debug, Clone, PartialEq)]
pub struct FunnelPackagePrice {
    pub identifier: String,
    pub localized_price: String,
    pub localized_period: Option<String>,
    pub localized_title: Option<String>,
    /// The price in micros of `currency`.
    pub amount_micros: Option<i64>,
    /// ISO-4217, upper-case.
    pub currency: Option<String>,
    /// `None` for a one-time product.
    pub period_unit: Option<FunnelPeriodUnit>,
    pub period_count: Option<u32>,
    pub localized_price_per_month: Option<String>,
    pub localized_price_per_week: Option<String>,
    /// Present ONLY when the visitor is eligible for it.
    pub offer: Option<FunnelPackageOffer>,
}

/// funnel-core's `PeriodUnit`, as the FFI spells it.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum FunnelPeriodUnit {
    Day,
    Week,
    Month,
    Year,
}

/// funnel-core's `OfferKind`, as the FFI spells it.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum FunnelOfferKind {
    FreeTrial,
    PayAsYouGo,
    PayUpFront,
}

/// funnel-core's `PackageOffer`: an introductory offer the visitor is
/// eligible for. `period_unit` × `period_count` is its whole duration.
#[derive(Debug, Clone, PartialEq)]
pub struct FunnelPackageOffer {
    pub kind: FunnelOfferKind,
    pub period_unit: FunnelPeriodUnit,
    pub period_count: u32,
    pub localized_period: String,
    pub localized_price: Option<String>,
}

impl From<FunnelPeriodUnit> for PeriodUnit {
    fn from(unit: FunnelPeriodUnit) -> Self {
        match unit {
            FunnelPeriodUnit::Day => PeriodUnit::Day,
            FunnelPeriodUnit::Week => PeriodUnit::Week,
            FunnelPeriodUnit::Month => PeriodUnit::Month,
            FunnelPeriodUnit::Year => PeriodUnit::Year,
        }
    }
}

impl From<FunnelOfferKind> for OfferKind {
    fn from(kind: FunnelOfferKind) -> Self {
        match kind {
            FunnelOfferKind::FreeTrial => OfferKind::FreeTrial,
            FunnelOfferKind::PayAsYouGo => OfferKind::PayAsYouGo,
            FunnelOfferKind::PayUpFront => OfferKind::PayUpFront,
        }
    }
}

impl From<FunnelPackageOffer> for PackageOffer {
    fn from(offer: FunnelPackageOffer) -> Self {
        PackageOffer {
            kind: offer.kind.into(),
            period_unit: offer.period_unit.into(),
            period_count: offer.period_count,
            localized_period: offer.localized_period,
            localized_price: offer.localized_price,
        }
    }
}

impl From<FunnelPackagePrice> for PackagePrice {
    fn from(price: FunnelPackagePrice) -> Self {
        PackagePrice {
            identifier: price.identifier,
            localized_price: price.localized_price,
            localized_period: price.localized_period,
            localized_title: price.localized_title,
            facts: PriceFacts {
                amount_micros: price.amount_micros,
                currency: price.currency,
                period_unit: price.period_unit.map(Into::into),
                period_count: price.period_count,
                localized_price_per_month: price.localized_price_per_month,
                localized_price_per_week: price.localized_price_per_week,
                offer: price.offer.map(Into::into),
            },
        }
    }
}

/// Attributes a funnel asked to write, collected as they are produced.
///
/// Kept on the session rather than applied inline because `funnel-core` is
/// pure and this wrapper has no network of its own — the owning
/// `RovenueCore` drains them through its existing attributes module.
#[derive(Debug, Clone, Default)]
pub struct PendingAttributes(
    /// Name/value pairs in the order the funnel produced them.
    pub Vec<(String, serde_json::Value)>,
);

/// What the session has already emitted, so each device event fires once:
/// a screen view per screen ARRIVAL (not per interaction on it), one
/// completion, one dismissal.
#[derive(Debug, Default)]
struct Emission {
    last_screen_id: Option<String>,
    finished_emitted: bool,
    dismissed_emitted: bool,
}

/// Prefix of a preview session's presentation id. A preview never emits,
/// so the id is never stamped on an event; the distinct prefix only makes
/// a preview's `state_json` recognisable to whoever reads it.
const PREVIEW_PRESENTATION_PREFIX: &str = "preview_";

/// The persisted session (spec D7): the machine state plus the
/// presentation id, so a restored session keeps its id and counts once.
/// Façades treat it opaquely — store the string, hand it back to
/// `open_funnel`. A pre-wave state (a bare machine state, no wrapper)
/// still restores; it just gets a fresh id.
#[derive(Debug, Deserialize)]
struct StoredSession {
    presentation_id: String,
    state: SessionState,
}

/// Everything a session needs to emit device events, and the only code
/// that can.
///
/// It exists only inside `FunnelSession::open`, which is the one
/// constructor handed an `EventQueue`. A preview session
/// (`FunnelSession::open_preview`) is built with `recorder: None` and is
/// never given a queue, so no code path can reach `emit` from it: zero
/// analytics pollution comes from the type, not from a flag someone could
/// forget to check.
///
/// No `Debug`: the queue and clock behind it don't implement it.
struct EventRecorder {
    queue: Arc<EventQueue>,
    rovenue_id: String,
    context: FunnelContext,
    clock: Arc<dyn Clock>,
    emission: Mutex<Emission>,
}

impl EventRecorder {
    /// Enqueue one device event. Best-effort by design: a telemetry write
    /// must never fail the interaction that produced it. The queue
    /// persists and drains (the server dedups a replayed drain on the
    /// content-hash event id), so a transient failure is retried, not
    /// lost; only a storage failure drops it, and failing the funnel UI
    /// for that would not save it either — but the drop is logged at warn,
    /// never swallowed.
    fn emit(&self, event_type: &str, screen_id: Option<String>) {
        let mut context = self.context.clone();
        context.screen_id = screen_id;
        let envelope = EventEnvelope {
            version: Some(EVENT_WIRE_VERSION),
            event_id: Some(format!("evt_{}", cuid2::create_id())),
            event_type: event_type.to_string(),
            occurred_at: rfc3339_millis(self.clock.now_unix_ms()),
            subscriber_id: if self.rovenue_id.is_empty() {
                None
            } else {
                Some(self.rovenue_id.clone())
            },
            product_id: None,
            amount: None,
            currency: None,
            event_source_url: None,
            identity_context: None,
            funnel_context: Some(context),
        };
        let queued = serde_json::to_string(&envelope)
            .map_err(|_| RovenueError::InvalidArgument())
            .and_then(|json| self.queue.enqueue(&json));
        if let Err(err) = queued {
            self.queue.log_dropped(event_type, &err);
        }
    }

    /// Record arriving on `page_id` (a `screen_view` per arrival, not per
    /// interaction) and finishing (one `complete`).
    fn after_step(&self, page_id: Option<String>, finished: bool) {
        let mut emission = self.emission.lock().unwrap_or_else(|e| e.into_inner());
        if emission.last_screen_id != page_id {
            emission.last_screen_id = page_id.clone();
            self.emit("funnel_screen_view", page_id);
        }
        if finished && !emission.finished_emitted {
            emission.finished_emitted = true;
            self.emit("funnel_complete", None);
        }
    }

    /// A restore emits nothing — but the tracking must start where the
    /// restored machine stands, or the first post-restore step would
    /// re-emit the arrival it resumes on, and an already finished machine
    /// would emit a completion for a funnel that ended before this wave.
    fn resume_at(&self, page_id: Option<String>, finished: bool) {
        let mut emission = self.emission.lock().unwrap_or_else(|e| e.into_inner());
        emission.last_screen_id = page_id;
        emission.finished_emitted = finished;
    }

    /// `funnel_dismiss`, once only: a close handler firing twice must not
    /// double-count the dismissal — and never after the funnel completed.
    fn dismiss(&self) {
        let mut emission = self.emission.lock().unwrap_or_else(|e| e.into_inner());
        // `dismissed` and `completed` are distinct terminal states (analytics
        // counts a dismissal as a drop-off): closing the sheet of a funnel
        // that already completed is not a second outcome.
        if emission.dismissed_emitted || emission.finished_emitted {
            return;
        }
        emission.dismissed_emitted = true;
        self.emit("funnel_dismiss", None);
    }
}

/// A running funnel, owned by the façade for the life of the presentation.
///
/// No `Debug`: the queue and clock behind it don't implement it, and a
/// debug dump of a live session would be noise anyway.
///
/// Opened through `RovenueCore::open_funnel` (emitting) or the
/// `open_preview_funnel` namespace function (never emitting), never
/// constructed directly: a live session holds the durable event queue and
/// stamps every event with the presentation id, so emission lives here
/// once (W10) rather than in each façade.
pub struct FunnelSession {
    inner: Mutex<FunnelMachine>,
    pending_attributes: Mutex<PendingAttributes>,
    presentation_id: String,
    /// `Some` only for a session `open` built; `None` for a preview, which
    /// therefore has no way to emit. See `EventRecorder`.
    recorder: Option<EventRecorder>,
}

fn map_error(err: FunnelError) -> RovenueError {
    match err {
        FunnelError::UnsupportedConfigVersion { .. } => RovenueError::FunnelConfigUnsupported(),
        FunnelError::EmptyFunnel => RovenueError::FunnelInvalid(),
    }
}

/// The platform's store product id for a package. The machine never reads
/// it (only `localized_price` renders, and prices are the platform's job —
/// core-rs performs no StoreKit / Play Billing I/O), but the field must
/// name the store the build actually buys from.
#[cfg(any(target_os = "ios", target_os = "macos"))]
fn store_product_id(package: &CoreOfferingProduct) -> String {
    package
        .apple_product_id
        .clone()
        .or(package.google_product_id.clone())
        .unwrap_or_default()
}

/// The platform's store product id for a package — see the iOS/macOS
/// twin for why this exists at all.
#[cfg(not(any(target_os = "ios", target_os = "macos")))]
fn store_product_id(package: &CoreOfferingProduct) -> String {
    package
        .google_product_id
        .clone()
        .or(package.apple_product_id.clone())
        .unwrap_or_default()
}

/// The served offering as machine data. No prices cross this boundary
/// (formatting them is the platform's job and core-rs does no store I/O),
/// so package lists draw unpriced until a façade enriches them (D13) —
/// purchasing is unaffected, the host buys by `package_id`. The title DOES
/// cross: it is server content, not a store price, so it draws from the
/// first render — never empty (D7).
fn map_offering(offering: &crate::offerings::CoreOffering) -> Offering {
    Offering {
        identifier: offering.identifier.clone(),
        packages: offering
            .packages
            .iter()
            .map(|p| OfferingPackage {
                identifier: p.package_identifier.clone(),
                product_identifier: store_product_id(p),
                localized_price: String::new(),
                localized_title: Some(if p.display_name.is_empty() {
                    p.package_identifier.clone()
                } else {
                    p.display_name.clone()
                }),
                localized_period: None,
                // The store facts arrive through `set_package_prices`.
                facts: PriceFacts::default(),
            })
            .collect(),
    }
}

impl FunnelSession {
    /// Open a funnel presentation on `funnel`'s published document.
    ///
    /// The offering comes from `funnel.offering` (see `map_offering`); an
    /// empty `funnel_id` — a cached or bundled response predating the
    /// field — is refused loudly rather than emitted unattributable.
    ///
    /// A fresh open emits `funnel_view` plus `funnel_screen_view` for the
    /// first screen. A restored open emits nothing and keeps the stored
    /// presentation id.
    pub fn open(
        funnel: CoreFunnel,
        locale: String,
        today: Option<String>,
        restored_state_json: Option<String>,
        queue: Arc<EventQueue>,
        rovenue_id: String,
        clock: Arc<dyn Clock>,
    ) -> RovenueResult<Self> {
        if funnel.funnel_id.trim().is_empty() {
            return Err(RovenueError::InvalidArgument());
        }
        let document: Funnel =
            serde_json::from_str(&funnel.config_json).map_err(|_| RovenueError::FunnelInvalid())?;

        // A malformed offering or a stale stored session must not stop the
        // funnel from opening: both degrade to "absent" rather than
        // throwing.
        let offering: Option<Offering> = funnel.offering.as_ref().map(map_offering);
        let (restored, presentation_id, was_restored) = split_restored(&restored_state_json);

        let machine = FunnelMachine::new(
            document,
            locale,
            today,
            offering,
            funnel.links.into(),
            restored,
        )
        .map_err(map_error)?;

        let (variant_id, experiment_key) = funnel
            .presented_context
            .map(|c| (c.variant_id, c.experiment_key))
            .unwrap_or((None, None));
        let recorder = EventRecorder {
            queue,
            rovenue_id,
            context: FunnelContext {
                funnel_id: funnel.funnel_id,
                session_id: presentation_id.clone(),
                placement_id: funnel.placement_identifier,
                placement_revision: funnel.placement_revision,
                revision: funnel.revision,
                screen_id: None,
                variant_id,
                experiment_key,
            },
            clock,
            emission: Mutex::new(Emission::default()),
        };
        let (page, finished) = (
            machine.state().current_page_id.clone(),
            machine.current().finished,
        );
        if was_restored {
            recorder.resume_at(page, finished);
        } else {
            recorder.emit("funnel_view", None);
            recorder.after_step(page, finished);
        }
        Ok(Self::from_machine(machine, presentation_id, Some(recorder)))
    }

    /// Open a NON-EMITTING preview of a funnel, for preview tools (the
    /// operator apps' "Test on device").
    ///
    /// `payload_json` is exactly the object `GET /v1/placements` serves
    /// under its `funnel` key (`identifier`, `id`, `revision`,
    /// `configFormatVersion`, `config`, `offering`), decoded by the same
    /// wire type the placements client uses, so a preview consumes the
    /// production input. The dashboard's preview-document route serves the
    /// same shape for a draft.
    ///
    /// The session resolves, navigates, evaluates conditions and variables
    /// and produces effects exactly like one from `open` — it is the same
    /// machine and the same conversion — but it is built without an
    /// `EventQueue`, so it holds no `EventRecorder` and cannot emit: no
    /// `funnel_*` event, no experiment exposure (no variant is drawn; the
    /// payload is one funnel). It needs no configured core and no API key.
    /// A preview always starts fresh: there is nothing to restore.
    ///
    /// An empty `id` is accepted here (nothing is attributed). A payload
    /// that does not decode, or whose `config` is not a funnel document,
    /// is `FunnelInvalid`; a config format newer than this build is
    /// `FunnelConfigUnsupported`. A payload with no offering opens fine —
    /// its package lists simply draw empty, as in production.
    pub fn open_preview(
        payload_json: String,
        locale: String,
        today: Option<String>,
    ) -> RovenueResult<Self> {
        let wire: FunnelWire =
            serde_json::from_str(&payload_json).map_err(|_| RovenueError::FunnelInvalid())?;
        let document: Funnel =
            serde_json::from_value(wire.config).map_err(|_| RovenueError::FunnelInvalid())?;
        let offering = wire
            .offering
            .map(crate::offerings::client::map_offering)
            .as_ref()
            .map(map_offering);
        let machine = FunnelMachine::new(document, locale, today, offering, wire.links, None)
            .map_err(map_error)?;
        Ok(Self::from_machine(
            machine,
            format!("{PREVIEW_PRESENTATION_PREFIX}{}", cuid2::create_id()),
            None,
        ))
    }

    fn from_machine(
        machine: FunnelMachine,
        presentation_id: String,
        recorder: Option<EventRecorder>,
    ) -> Self {
        Self {
            inner: Mutex::new(machine),
            pending_attributes: Mutex::new(PendingAttributes::default()),
            presentation_id,
            recorder,
        }
    }

    /// The presentation id this session stamps. Test-only: lets emission
    /// tests prove one id across events without parsing `state_json`.
    #[doc(hidden)]
    pub fn test_presentation_id(&self) -> String {
        self.presentation_id.clone()
    }

    /// Record arrival/finish on the recorder, if this session has one.
    fn after_step(&self, page_id: Option<String>, finished: bool) {
        if let Some(recorder) = &self.recorder {
            recorder.after_step(page_id, finished);
        }
    }

    fn convert(&self, step: Step) -> FunnelStep {
        let mut effects = Vec::new();
        for effect in step.effects {
            match effect {
                // Absorbed here: recorded for the owning core to flush
                // through its existing attributes module.
                CoreEffect::SetAttribute { name, value } => {
                    self.pending_attributes
                        .lock()
                        .unwrap_or_else(|e| e.into_inner())
                        .0
                        .push((name, value));
                }
                CoreEffect::Dismiss => effects.push(FunnelEffect::Dismiss),
                CoreEffect::OpenUrl { url } => effects.push(FunnelEffect::OpenUrl { url }),
                CoreEffect::ShowAlert {
                    title,
                    message,
                    button_label,
                } => effects.push(FunnelEffect::ShowAlert {
                    title,
                    message,
                    button_label,
                }),
                CoreEffect::Custom { action_id, params } => effects.push(FunnelEffect::Custom {
                    action_id,
                    params_json: serde_json::to_string(&params).unwrap_or_else(|_| "{}".into()),
                }),
                CoreEffect::Purchase { package_id } => {
                    effects.push(FunnelEffect::Purchase { package_id })
                }
                CoreEffect::RestorePurchases => effects.push(FunnelEffect::RestorePurchases),
            }
        }
        FunnelStep {
            screen_json: step
                .screen
                .as_ref()
                .and_then(|s| serde_json::to_string(s).ok()),
            effects,
            finished: step.finished,
        }
    }

    fn with<R>(&self, f: impl FnOnce(&mut FunnelMachine) -> R) -> R {
        let mut guard = self.inner.lock().unwrap_or_else(|e| e.into_inner());
        f(&mut guard)
    }

    /// One interaction: run the machine, convert, then emit for whatever
    /// screen that landed on (and for finishing).
    fn interact(&self, f: impl FnOnce(&mut FunnelMachine) -> Step) -> FunnelStep {
        let (step, page, finished) = self.with(|m| {
            let step = f(m);
            let finished = step.finished;
            let page = m.state().current_page_id.clone();
            (step, page, finished)
        });
        let out = self.convert(step);
        self.after_step(page, finished);
        out
    }

    /// The screen to draw right now, with no effects.
    pub fn current(&self) -> FunnelStep {
        let step = self.with(|m| m.current());
        self.convert(step)
    }

    /// Record an answer and re-resolve, so a `visibleWhen` keyed on it
    /// takes effect on the same screen.
    pub fn submit(&self, block_id: String, answer_json: String) -> FunnelStep {
        // An answer that will not parse is stored as null rather than
        // dropped: "asked and not answered" is a state the evaluator knows
        // how to route, and a silent no-op is not.
        let answer: AnswerValue = serde_json::from_str(&answer_json).unwrap_or(AnswerValue::Null);
        self.interact(|m| m.submit(&block_id, answer))
    }

    /// Record which offering package a package list tap selected.
    pub fn select_package(&self, identifier: String) -> FunnelStep {
        self.interact(|m| m.select_package(&identifier))
    }

    /// Hand in store-fetched prices for the current offering's packages
    /// (D13). Façades call this once their async StoreKit / Play Billing
    /// lookup resolves — `open` cannot wait on it, so it draws first
    /// without a price. Forwarded to the machine, which keeps them on its
    /// offering (so every later draw carries them) and re-resolves: copy
    /// reading `{{product.*}}` and the computed savings see the prices.
    /// `identifier` matches the package SLOT identifier (what
    /// `select_package` also takes); one the offering lacks is ignored.
    /// Does not advance the session or emit anything.
    pub fn set_package_prices(&self, prices: Vec<FunnelPackagePrice>) -> FunnelStep {
        let step =
            self.with(|m| m.set_package_prices(prices.into_iter().map(Into::into).collect()));
        self.convert(step)
    }

    /// Run the action chain at `action_path` (`cta`, `el.<blockId>` or
    /// `opt.<urlEncodedValue>`). An address nothing answers to is a no-op.
    pub fn tap(&self, action_path: String) -> FunnelStep {
        self.interact(|m| m.tap(&action_path))
    }

    /// Return to the previous screen. A no-op when there is no history.
    pub fn back(&self) -> FunnelStep {
        self.interact(|m| m.back())
    }

    /// The persisted session: machine state plus the presentation id. The
    /// façade stores this string and hands it back to `open_funnel` on the
    /// next launch (spec D7).
    pub fn state_json(&self) -> String {
        self.with(|m| {
            let stored = StoredSessionRef {
                presentation_id: &self.presentation_id,
                state: m.state(),
            };
            serde_json::to_string(&stored).unwrap_or_else(|_| "{}".into())
        })
    }

    /// Emit `funnel_dismiss`. The façade calls it when the host closes the
    /// view or on a `Dismiss` effect. Once only: a close handler firing
    /// twice must not double-count the dismissal — and never after the
    /// funnel completed.
    ///
    /// A preview session has no recorder, so this is a no-op there.
    pub fn dismiss(&self) {
        if let Some(recorder) = &self.recorder {
            recorder.dismiss();
        }
    }

    /// Attributes the funnel asked to write, drained for the owning core to
    /// flush. Not part of the FFI surface.
    pub fn take_pending_attributes(&self) -> Vec<(String, serde_json::Value)> {
        std::mem::take(
            &mut self
                .pending_attributes
                .lock()
                .unwrap_or_else(|e| e.into_inner())
                .0,
        )
    }
}

/// `state_json` shapes, split out so `open` stays readable.
#[derive(Debug, Serialize)]
struct StoredSessionRef<'a> {
    presentation_id: &'a str,
    state: &'a SessionState,
}

/// See `FunnelSession::open`: wrapper preferred, legacy bare state
/// accepted, garbage degrades to fresh.
fn split_restored(restored_state_json: &Option<String>) -> (Option<SessionState>, String, bool) {
    let fresh_id = || format!("sess_{}", cuid2::create_id());
    match restored_state_json {
        Some(raw) => {
            if let Ok(stored) = serde_json::from_str::<StoredSession>(raw) {
                (Some(stored.state), stored.presentation_id, true)
            } else if let Ok(legacy) = serde_json::from_str::<SessionState>(raw) {
                (Some(legacy), fresh_id(), true)
            } else {
                (None, fresh_id(), false)
            }
        }
        None => (None, fresh_id(), false),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::cache::CacheStore;
    use crate::events::EventsClient;
    use crate::placements::CorePresentedContext;
    use crate::time::SystemClock;
    use crate::transport::http_client::HttpClient;

    const MINIMAL: &str = r#"{"configFormatVersion":1,"id":"f","identifier":"t","revision":1,
        "defaultLocale":"en","target":"native","pagesOrder":["a"],
        "pages":[{"id":"a","type":"custom","question_id":"q",
                  "blocks":[{"id":"b1","type":"heading","text":{"en":"Hi"}}]}]}"#;

    /// Two screens: `s1` advances on a bare `cta` tap, `s2` closes.
    const TWO_PAGE: &str = r#"{"configFormatVersion":1,"id":"f","identifier":"t","revision":1,
        "defaultLocale":"en","target":"native","pagesOrder":["s1","s2"],
        "pages":[{"id":"s1","type":"custom","question_id":"q",
                  "blocks":[{"id":"b1","type":"heading","text":{"en":"Hi"}}]},
                 {"id":"s2","type":"custom",
                  "actions":{"next":[{"type":"close_funnel"}]}}]}"#;

    fn test_env() -> (Arc<CacheStore>, Arc<EventQueue>) {
        let store = Arc::new(CacheStore::open_in_memory().unwrap());
        // Nothing listens here: emission persists + retains, never posts.
        // That is exactly what these tests assert — the queued envelopes.
        let http = Arc::new(
            HttpClient::new("http://127.0.0.1:1".to_string(), "pk_test".to_string())
                .with_max_attempts(1),
        );
        let events = Arc::new(EventsClient::new(http));
        let queue = Arc::new(EventQueue::new(Arc::clone(&store), events));
        (store, queue)
    }

    fn core_funnel(config_json: &str) -> CoreFunnel {
        CoreFunnel {
            placement_identifier: "plc_test".to_string(),
            placement_revision: 3,
            funnel_identifier: "t".to_string(),
            funnel_id: "fnl_test".to_string(),
            revision: 1,
            config_format_version: 1,
            config_json: config_json.to_string(),
            offering: None,
            presented_context: Some(CorePresentedContext {
                placement_id: "plc_test".to_string(),
                funnel_id: "t".to_string(),
                variant_id: Some("var_a".to_string()),
                experiment_key: Some("exp_1".to_string()),
                revision: 3,
            }),
            served_from_fallback: false,
            links: crate::placements::FunnelProjectLinks::default(),
        }
    }

    fn open_session(
        config_json: &str,
        restored: Option<String>,
    ) -> (FunnelSession, Arc<CacheStore>) {
        let (store, queue) = test_env();
        let clock: Arc<dyn Clock> = Arc::new(SystemClock);
        let session = FunnelSession::open(
            core_funnel(config_json),
            "en".to_string(),
            None,
            restored,
            queue,
            "rov_test".to_string(),
            clock,
        )
        .expect("session opens");
        (session, store)
    }

    /// The queued device envelopes, oldest first, parsed (never hand-built).
    fn queued(store: &CacheStore) -> Vec<EventEnvelope> {
        store
            .list_queued_events(usize::MAX)
            .unwrap()
            .into_iter()
            .map(|row| serde_json::from_str::<EventEnvelope>(&row.envelope_json).unwrap())
            .collect()
    }

    struct Collector(Arc<Mutex<Vec<crate::logging::LogRecord>>>);
    impl crate::logging::LogSink for Collector {
        fn on_log(&self, record: crate::logging::LogRecord) {
            self.0.lock().unwrap().push(record);
        }
    }

    /// Emission is best-effort, but a storage failure must be VISIBLE: a
    /// silent `let _ =` hid a missing `event_queue` table (MIGRATION_V11
    /// edited in place) that dropped every funnel event. How it could fail:
    /// nothing logged; logged below warn (the default threshold hides it);
    /// the record does not say which event was lost; the failure breaks
    /// opening the funnel instead of being contained.
    #[test]
    fn a_failed_enqueue_is_logged_at_warn_and_does_not_fail_the_funnel() {
        let store = Arc::new(CacheStore::open_in_memory().unwrap());
        store
            .with_conn(|c| c.execute_batch("DROP TABLE event_queue"))
            .unwrap();
        let logger = Arc::new(crate::logging::Logger::new(crate::logging::LogLevel::Warn));
        let records = Arc::new(Mutex::new(Vec::new()));
        logger.set_sink(Arc::new(Collector(Arc::clone(&records))));
        let http = Arc::new(
            HttpClient::new("http://127.0.0.1:1".to_string(), "pk_test".to_string())
                .with_max_attempts(1),
        );
        let queue = Arc::new(
            EventQueue::new(Arc::clone(&store), Arc::new(EventsClient::new(http)))
                .with_logger(logger),
        );

        // Opening arrives on the first page, which emits funnel_screen_view.
        FunnelSession::open(
            core_funnel(TWO_PAGE),
            "en".to_string(),
            None,
            None,
            queue,
            "rov_test".to_string(),
            Arc::new(SystemClock),
        )
        .expect("a telemetry failure never fails the funnel");

        let got = records.lock().unwrap();
        let dropped: Vec<_> = got
            .iter()
            .filter(|r| r.message.contains("funnel_screen_view"))
            .collect();
        assert_eq!(dropped.len(), 1, "one warn per lost event");
        assert_eq!(dropped[0].level, crate::logging::LogLevel::Warn);
        assert_eq!(
            dropped[0].fields.get("event_type").map(String::as_str),
            Some("funnel_screen_view")
        );
    }

    fn types_of(store: &CacheStore) -> Vec<String> {
        queued(store).into_iter().map(|e| e.event_type).collect()
    }

    #[test]
    fn a_fresh_open_emits_view_then_the_first_screen() {
        let (session, store) = open_session(TWO_PAGE, None);

        assert_eq!(types_of(&store), vec!["funnel_view", "funnel_screen_view"]);
        let rows = queued(&store);
        let first = &rows[0];
        assert_eq!(first.funnel_context.as_ref().unwrap().screen_id, None);
        assert_eq!(
            rows[1].funnel_context.as_ref().unwrap().screen_id,
            Some("s1".to_string())
        );

        // One presentation id across both, non-empty, namespaced.
        let id = session.test_presentation_id();
        assert!(id.starts_with("sess_"), "unexpected id {id}");
        for e in &rows {
            let ctx = e.funnel_context.as_ref().unwrap();
            assert_eq!(ctx.session_id, id);
            assert_eq!(ctx.funnel_id, "fnl_test");
            assert_eq!(ctx.placement_id, "plc_test");
            assert_eq!(ctx.placement_revision, 3);
            assert_eq!(ctx.revision, 1);
            assert_eq!(ctx.variant_id.as_deref(), Some("var_a"));
            assert_eq!(ctx.experiment_key.as_deref(), Some("exp_1"));
            assert_eq!(e.subscriber_id.as_deref(), Some("rov_test"));
            assert!(e.occurred_at.contains('T') && e.occurred_at.ends_with('Z'));
        }
        // Distinct envelopes: the server dedups replays, not siblings.
        assert_ne!(rows[0].event_id, rows[1].event_id);
    }

    #[test]
    fn tap_to_screen_two_then_finish_drains_view_screen_screen_complete() {
        let (session, store) = open_session(TWO_PAGE, None);

        let step = session.tap("cta".to_string());
        let json: serde_json::Value =
            serde_json::from_str(&step.screen_json.expect("screen two")).unwrap();
        assert_eq!(json["screenId"], "s2");

        let end = session.tap("cta".to_string());
        assert!(end.finished, "s2's next chain closes the funnel");

        assert_eq!(
            types_of(&store),
            vec![
                "funnel_view",
                "funnel_screen_view",
                "funnel_screen_view",
                "funnel_complete"
            ]
        );
        let rows = queued(&store);
        let screens: Vec<Option<String>> = rows
            .iter()
            .map(|e| e.funnel_context.as_ref().unwrap().screen_id.clone())
            .collect();
        assert_eq!(
            screens,
            vec![None, Some("s1".to_string()), Some("s2".to_string()), None]
        );
        let id = session.test_presentation_id();
        assert!(!id.is_empty());
        for e in &rows {
            assert_eq!(e.funnel_context.as_ref().unwrap().session_id, id);
        }
    }

    #[test]
    fn an_address_nothing_answers_to_emits_nothing() {
        let (session, store) = open_session(TWO_PAGE, None);
        session.tap("nope".to_string());
        assert_eq!(types_of(&store), vec!["funnel_view", "funnel_screen_view"]);
    }

    #[test]
    fn a_restore_emits_nothing_and_keeps_the_session_id() {
        let (session, store) = open_session(TWO_PAGE, None);
        session.tap("cta".to_string());
        assert_eq!(types_of(&store).len(), 3);
        let id = session.test_presentation_id();
        let saved = session.state_json();
        drop(session);

        let clock: Arc<dyn Clock> = Arc::new(SystemClock);
        let (store_r, queue_r) = test_env();
        let resumed = FunnelSession::open(
            core_funnel(TWO_PAGE),
            "en".to_string(),
            None,
            Some(saved),
            queue_r,
            "rov_test".to_string(),
            clock,
        )
        .expect("restore opens");
        assert_eq!(resumed.test_presentation_id(), id);
        assert_eq!(types_of(&store_r).len(), 0, "a restore emits nothing");
        // …and resumes where it left off: one more tap finishes.
        let end = resumed.tap("cta".to_string());
        assert!(end.finished);
        assert_eq!(types_of(&store_r), vec!["funnel_complete"]);
        assert_eq!(
            queued(&store_r)[0]
                .funnel_context
                .as_ref()
                .unwrap()
                .session_id,
            id
        );
    }

    #[test]
    fn a_legacy_bare_state_restores_with_a_fresh_id_and_emits_nothing() {
        let (session, _) = open_session(TWO_PAGE, None);
        session.submit("b1".to_string(), "\"answered\"".to_string());
        let saved: serde_json::Value = serde_json::from_str(&session.state_json()).unwrap();
        let legacy = serde_json::to_string(&saved["state"]).unwrap();
        let old_id = session.test_presentation_id();
        drop(session);

        let (store, queue) = test_env();
        let clock: Arc<dyn Clock> = Arc::new(SystemClock);
        let resumed = FunnelSession::open(
            core_funnel(TWO_PAGE),
            "en".to_string(),
            None,
            Some(legacy),
            queue,
            "rov_test".to_string(),
            clock,
        )
        .expect("legacy state restores");
        assert_ne!(resumed.test_presentation_id(), old_id);
        assert_eq!(types_of(&store).len(), 0);
        // The answers survived the legacy restore.
        let json: serde_json::Value =
            serde_json::from_str(&resumed.current().screen_json.expect("a screen")).unwrap();
        assert_eq!(json["screenId"], "s1");
    }

    #[test]
    fn dismiss_emits_funnel_dismiss_once() {
        let (session, store) = open_session(TWO_PAGE, None);
        session.dismiss();
        session.dismiss();
        assert_eq!(
            types_of(&store),
            vec!["funnel_view", "funnel_screen_view", "funnel_dismiss"]
        );
        let rows = queued(&store);
        assert_eq!(rows[2].funnel_context.as_ref().unwrap().screen_id, None);
        assert_eq!(
            rows[2].funnel_context.as_ref().unwrap().session_id,
            session.test_presentation_id()
        );
    }

    #[test]
    fn an_empty_funnel_id_is_refused_before_anything_emits() {
        let (store, queue) = test_env();
        let clock: Arc<dyn Clock> = Arc::new(SystemClock);
        let mut funnel = core_funnel(TWO_PAGE);
        funnel.funnel_id = String::new();
        let err = FunnelSession::open(
            funnel,
            "en".to_string(),
            None,
            None,
            queue,
            "rov_test".to_string(),
            clock,
        )
        .err()
        .expect("empty funnel id is refused");
        assert_eq!(err.kind, crate::error::ErrorKind::InvalidArgument);
        assert_eq!(types_of(&store).len(), 0);
    }

    #[test]
    fn a_step_carries_the_screen_as_json() {
        let (session, _) = open_session(MINIMAL, None);
        let step = session.current();
        let json = step.screen_json.expect("a screen");
        assert!(json.contains("\"screenId\":\"a\""));
        assert!(json.contains("\"text\":\"Hi\""));
        assert!(!step.finished);
    }

    #[test]
    fn a_config_above_the_supported_version_is_refused() {
        let newer = MINIMAL.replace("\"configFormatVersion\":1", "\"configFormatVersion\":99");
        let (store, queue) = test_env();
        let clock: Arc<dyn Clock> = Arc::new(SystemClock);
        let err = FunnelSession::open(
            core_funnel(&newer),
            "en".to_string(),
            None,
            None,
            queue,
            "rov_test".to_string(),
            clock,
        )
        .err()
        .expect("newer config is refused");
        assert_eq!(err.kind, crate::error::ErrorKind::FunnelConfigUnsupported);
        assert_eq!(types_of(&store).len(), 0);
    }

    #[test]
    fn a_config_that_does_not_parse_is_refused_as_invalid() {
        let (store, queue) = test_env();
        let clock: Arc<dyn Clock> = Arc::new(SystemClock);
        let err = FunnelSession::open(
            core_funnel("not json"),
            "en".to_string(),
            None,
            None,
            queue,
            "rov_test".to_string(),
            clock,
        )
        .err()
        .expect("unparseable config is refused");
        assert_eq!(err.kind, crate::error::ErrorKind::FunnelInvalid);
        assert_eq!(types_of(&store).len(), 0);
    }

    #[test]
    fn a_malformed_offering_does_not_stop_the_funnel_opening() {
        // The offering now travels inside CoreFunnel (mapped, never parsed
        // as JSON), so "malformed" degrades to no offering — the funnel
        // still opens and still emits.
        let (session, store) = open_session(MINIMAL, None);
        assert!(session.current().screen_json.is_some());
        assert_eq!(types_of(&store).len(), 2);
    }

    #[test]
    fn set_attribute_is_absorbed_and_never_reaches_a_facade() {
        let config = r#"{"configFormatVersion":1,"id":"f","identifier":"t","revision":1,
            "defaultLocale":"en","target":"native","pagesOrder":["a"],
            "pages":[{"id":"a","type":"custom","question_id":"q",
                      "actions":{"cta":[{"type":"set_attribute","name":"goal",
                                         "value":{"kind":"literal","value":"focus"}}]}}]}"#;
        let (session, _) = open_session(config, None);
        let step = session.tap("cta".into());
        assert!(
            step.effects.is_empty(),
            "set_attribute must not cross the FFI"
        );
        assert_eq!(
            session.take_pending_attributes(),
            vec![("goal".to_string(), serde_json::json!("focus"))]
        );
        assert!(
            session.take_pending_attributes().is_empty(),
            "draining twice must not repeat"
        );
    }

    #[test]
    fn state_json_round_trips_into_a_resumed_session() {
        let (session, _) = open_session(MINIMAL, None);
        session.submit("b1".into(), "\"answered\"".into());
        let saved = session.state_json();

        let (resumed, _) = open_session(MINIMAL, Some(saved));
        let json = resumed.current().screen_json.expect("a screen");
        assert!(json.contains("\"screenId\":\"a\""));
    }

    #[test]
    fn an_unparseable_answer_becomes_null_rather_than_vanishing() {
        let (session, _) = open_session(MINIMAL, None);
        session.submit("b1".into(), "not json".into());
        let states_answers = session.with(|m| m.state().answers.get("q").cloned());
        assert_eq!(states_answers, Some(AnswerValue::Null));
    }
}