tele 0.1.23

use super::*;

/// Long-polling source configuration.
#[derive(Clone, Debug)]
pub struct PollingConfig {
    /// Polling timeout passed to `getUpdates` in seconds.
    ///
    /// When greater than zero, runtime requires at least one second of timeout
    /// budget headroom from `min(client.request_timeout, client.total_timeout)`.
    /// If budget is smaller, polling returns `Error::Configuration`.
    /// Set this value to `0` for explicit short polling.
    pub poll_timeout_seconds: u16,
    pub limit: Option<u8>,
    pub allowed_updates: Option<Vec<AllowedUpdate>>,
    pub disable_webhook_on_start: bool,
    pub drop_pending_updates_on_start: bool,
    pub dedupe_window_size: usize,
    pub persist_offset_path: Option<PathBuf>,
}

impl Default for PollingConfig {
    fn default() -> Self {
        Self {
            poll_timeout_seconds: 30,
            limit: None,
            allowed_updates: None,
            disable_webhook_on_start: true,
            drop_pending_updates_on_start: false,
            dedupe_window_size: 2048,
            persist_offset_path: None,
        }
    }
}

impl PollingConfig {
    pub fn allowed_updates(
        mut self,
        allowed_updates: impl IntoIterator<Item = AllowedUpdate>,
    ) -> Self {
        self.set_allowed_updates(allowed_updates);
        self
    }

    pub fn allowed_update_kinds(
        mut self,
        kinds: impl IntoIterator<Item = UpdateKind>,
    ) -> Result<Self> {
        self.set_allowed_update_kinds(kinds)?;
        Ok(self)
    }

    pub fn set_allowed_updates(
        &mut self,
        allowed_updates: impl IntoIterator<Item = AllowedUpdate>,
    ) -> &mut Self {
        self.allowed_updates = Some(allowed_updates.into_iter().collect());
        self
    }

    pub fn set_allowed_update_kinds(
        &mut self,
        kinds: impl IntoIterator<Item = UpdateKind>,
    ) -> Result<&mut Self> {
        self.allowed_updates = Some(AllowedUpdate::from_kinds(kinds)?);
        Ok(self)
    }

    pub fn clear_allowed_updates(&mut self) -> &mut Self {
        self.allowed_updates = None;
        self
    }

    pub fn validate(&self) -> Result<()> {
        let request = GetUpdatesRequest {
            limit: self.limit,
            allowed_updates: self.allowed_updates.clone(),
            ..GetUpdatesRequest::default()
        };

        request.validate().map_err(|error| match error {
            Error::InvalidRequest { reason } => Error::Configuration {
                reason: format!("invalid polling config: {reason}"),
            },
            error => error,
        })?;

        Ok(())
    }

    fn resolve_poll_timeout_seconds(
        &self,
        request_timeout: Duration,
        total_timeout: Option<Duration>,
    ) -> Result<u16> {
        self.validate()?;

        let request_budget =
            total_timeout.map_or(request_timeout, |total| total.min(request_timeout));

        // Keep one second of headroom so transport timeout does not preempt long polling.
        let max_poll_timeout = request_budget
            .checked_sub(Duration::from_secs(1))
            .map_or(0, |timeout| {
                timeout.as_secs().min(u64::from(u16::MAX)) as u16
            });

        if self.poll_timeout_seconds == 0 {
            return Ok(0);
        }

        if self.poll_timeout_seconds > max_poll_timeout {
            let total_timeout_display = total_timeout.map_or_else(
                || "none".to_owned(),
                |timeout| format!("{}ms", timeout.as_millis()),
            );
            return Err(Error::Configuration {
                reason: format!(
                    "poll_timeout_seconds={} exceeds timeout budget headroom of {}s, got request_timeout={}ms and total_timeout={}; reduce poll_timeout_seconds, increase timeouts, or set poll_timeout_seconds=0 for short polling",
                    self.poll_timeout_seconds,
                    max_poll_timeout,
                    request_timeout.as_millis(),
                    total_timeout_display
                ),
            });
        }

        Ok(self.poll_timeout_seconds)
    }
}

/// Result of dispatching one update through router + middleware chain.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum DispatchOutcome {
    Handled { update_id: i64 },
    Ignored { update_id: i64 },
    Failed { update_id: i64 },
}

impl DispatchOutcome {
    pub fn update_id(self) -> i64 {
        match self {
            Self::Handled { update_id }
            | Self::Ignored { update_id }
            | Self::Failed { update_id } => update_id,
        }
    }

    pub fn is_handled(self) -> bool {
        matches!(self, Self::Handled { .. })
    }

    pub fn is_failed(self) -> bool {
        matches!(self, Self::Failed { .. })
    }
}

/// Pluggable update input source used by `BotEngine`.
pub trait UpdateSource: Send + 'static {
    fn poll<'a>(&'a mut self) -> SourceFuture<'a>;

    fn commit<'a>(&'a mut self, _outcomes: &'a [DispatchOutcome]) -> SourceCommitFuture<'a> {
        Box::pin(async { Ok(()) })
    }
}

/// Exponential backoff policy for source-side polling errors.
#[derive(Clone, Debug, PartialEq)]
pub struct SourceErrorBackoffConfig {
    pub base_delay: Duration,
    pub max_delay: Duration,
    pub jitter_ratio: f64,
}

impl Default for SourceErrorBackoffConfig {
    fn default() -> Self {
        Self {
            base_delay: Duration::from_millis(500),
            max_delay: Duration::from_secs(30),
            jitter_ratio: 0.2,
        }
    }
}

impl SourceErrorBackoffConfig {
    pub fn validate(&self) -> Result<()> {
        if self.base_delay.is_zero() {
            return Err(Error::Configuration {
                reason: "source_error_backoff base_delay must be greater than zero".to_owned(),
            });
        }
        if self.max_delay.is_zero() {
            return Err(Error::Configuration {
                reason: "source_error_backoff max_delay must be greater than zero".to_owned(),
            });
        }
        if self.base_delay > self.max_delay {
            return Err(Error::Configuration {
                reason: "source_error_backoff base_delay must not exceed max_delay".to_owned(),
            });
        }
        if !self.jitter_ratio.is_finite() || !(0.0..=1.0).contains(&self.jitter_ratio) {
            return Err(Error::Configuration {
                reason: "source_error_backoff jitter_ratio must be finite and between 0.0 and 1.0"
                    .to_owned(),
            });
        }

        Ok(())
    }
}

/// Shared engine configuration independent from input source implementation.
#[derive(Clone, Debug)]
pub struct EngineConfig {
    pub idle_delay: Duration,
    pub error_delay: Duration,
    /// Optional exponential backoff for repeated source errors.
    ///
    /// When enabled, this takes precedence over `error_delay`.
    pub source_error_backoff: Option<SourceErrorBackoffConfig>,
    /// Keep the engine running after retryable source-side polling errors.
    ///
    /// Non-retryable source errors remain fatal even when this is enabled.
    pub continue_on_source_error: bool,
    /// Continue dispatching later updates after a handler returns an error.
    ///
    /// In long-polling mode, continued handler failures are committed as processed together with
    /// the rest of the completed batch. Fail-fast dispatch still commits only the successful
    /// prefix before the failed update.
    pub continue_on_handler_error: bool,
    /// Maximum number of update handlers to run concurrently.
    ///
    /// This is applied only when `continue_on_handler_error` is true.
    pub max_handler_concurrency: usize,
}

impl Default for EngineConfig {
    fn default() -> Self {
        Self {
            idle_delay: Duration::from_millis(100),
            error_delay: Duration::from_millis(500),
            source_error_backoff: None,
            continue_on_source_error: true,
            continue_on_handler_error: true,
            max_handler_concurrency: 1,
        }
    }
}

impl EngineConfig {
    pub fn validate(&self) -> Result<()> {
        if self.max_handler_concurrency == 0 {
            return Err(Error::Configuration {
                reason: "max_handler_concurrency must be at least 1".to_owned(),
            });
        }
        if self.idle_delay.is_zero() {
            return Err(Error::Configuration {
                reason: "idle_delay must be greater than zero".to_owned(),
            });
        }
        if self.continue_on_source_error
            && self.source_error_backoff.is_none()
            && self.error_delay.is_zero()
        {
            return Err(Error::Configuration {
                reason: "error_delay must be greater than zero when source errors are retried without backoff".to_owned(),
            });
        }
        if let Some(backoff) = self.source_error_backoff.as_ref() {
            backoff.validate()?;
        }

        Ok(())
    }
}

/// Long-polling update source that only fetches updates and tracks offsets.
#[derive(Clone)]
pub struct LongPollingSource {
    client: Client,
    config: PollingConfig,
    next_offset: Option<i64>,
    seen_update_ids: HashSet<i64>,
    seen_update_order: VecDeque<i64>,
    offset_loaded: bool,
    offset_overridden: bool,
    pending_persisted_offset: Option<i64>,
    validated_offset_storage_path: Option<PathBuf>,
    prepared: bool,
}

impl LongPollingSource {
    pub fn new(client: Client) -> Self {
        Self {
            client,
            config: PollingConfig::default(),
            next_offset: None,
            seen_update_ids: HashSet::new(),
            seen_update_order: VecDeque::new(),
            offset_loaded: false,
            offset_overridden: false,
            pending_persisted_offset: None,
            validated_offset_storage_path: None,
            prepared: false,
        }
    }

    /// Applies polling config and validates timeout budget immediately.
    pub fn with_config(mut self, config: PollingConfig) -> Result<Self> {
        self.set_config(config)?;
        Ok(self)
    }

    /// Returns the current long-polling configuration.
    ///
    /// Use [`Self::set_config`] or dedicated setters to change values so runtime caches stay
    /// consistent.
    pub fn config(&self) -> &PollingConfig {
        &self.config
    }

    /// Replaces long-polling configuration after validating request options and timeout budget.
    pub fn set_config(&mut self, config: PollingConfig) -> Result<&mut Self> {
        let _ = self.resolved_poll_timeout_seconds(&config)?;
        Ok(self.apply_config(config))
    }

    fn apply_config(&mut self, config: PollingConfig) -> &mut Self {
        let dedupe_window_size_changed =
            self.config.dedupe_window_size != config.dedupe_window_size;
        if self.config.persist_offset_path != config.persist_offset_path {
            self.invalidate_offset_storage_cache();
        }
        if self.config.disable_webhook_on_start != config.disable_webhook_on_start
            || self.config.drop_pending_updates_on_start != config.drop_pending_updates_on_start
        {
            self.prepared = false;
        }
        self.config = config;
        if dedupe_window_size_changed {
            self.trim_seen_update_ids();
        }
        self
    }

    /// Sets the `getUpdates` long-poll timeout in seconds.
    pub fn set_poll_timeout_seconds(&mut self, poll_timeout_seconds: u16) -> Result<&mut Self> {
        let mut config = self.config.clone();
        config.poll_timeout_seconds = poll_timeout_seconds;
        self.set_config(config)
    }

    /// Sets the in-memory duplicate-update window and trims cached IDs immediately.
    pub fn set_dedupe_window_size(&mut self, dedupe_window_size: usize) -> &mut Self {
        if self.config.dedupe_window_size != dedupe_window_size {
            self.config.dedupe_window_size = dedupe_window_size;
            self.trim_seen_update_ids();
        }
        self
    }

    /// Validates timeout budget and returns resolved poll timeout seconds.
    pub fn validate_timeout_budget(&self) -> Result<u16> {
        self.effective_poll_timeout_seconds()
    }

    pub fn next_offset(&self) -> Option<i64> {
        self.next_offset
    }

    /// Overrides the next polling offset and makes the override authoritative.
    ///
    /// This also clears the in-memory dedupe window so callers can intentionally rewind or clear
    /// offsets without stale local state suppressing redelivered updates.
    pub fn set_next_offset(&mut self, offset: Option<i64>) -> &mut Self {
        self.next_offset = offset;
        self.offset_loaded = true;
        self.offset_overridden = true;
        self.pending_persisted_offset = None;
        self.seen_update_ids.clear();
        self.seen_update_order.clear();
        self
    }

    /// Enables offset persistence with a builder-style API.
    pub fn with_offset_persistence_path(mut self, path: impl Into<PathBuf>) -> Self {
        self.set_offset_persistence_path(path);
        self
    }

    /// Disables offset persistence with a builder-style API.
    pub fn clear_offset_persistence_path(mut self) -> Self {
        self.clear_offset_persistence();
        self
    }

    /// Enables offset persistence and revalidates the new storage target before polling.
    pub fn set_offset_persistence_path(&mut self, path: impl Into<PathBuf>) -> &mut Self {
        let path = path.into();
        if self.config.persist_offset_path.as_ref() != Some(&path) {
            self.config.persist_offset_path = Some(path);
            self.invalidate_offset_storage_cache();
        }
        self
    }

    /// Disables offset persistence and clears cached storage validation state.
    pub fn clear_offset_persistence(&mut self) -> &mut Self {
        if self.config.persist_offset_path.is_some() {
            self.config.persist_offset_path = None;
            self.pending_persisted_offset = None;
            self.invalidate_offset_storage_cache();
        }
        self
    }

    fn invalidate_offset_storage_cache(&mut self) {
        self.validated_offset_storage_path = None;
        if self.next_offset.is_none() && !self.offset_overridden {
            self.offset_loaded = false;
        }
    }

    async fn ensure_prepared(&mut self) -> Result<()> {
        self.ensure_offset_loaded().await?;

        if self.prepared {
            return Ok(());
        }

        if self.config.disable_webhook_on_start {
            let request = DeleteWebhookRequest {
                drop_pending_updates: self.config.drop_pending_updates_on_start.then_some(true),
            };
            self.client.updates().delete_webhook(&request).await?;
        }

        self.prepared = true;
        Ok(())
    }

    fn apply_committed_update(&mut self, update_id: i64) -> bool {
        let candidate = update_id.saturating_add(1);
        let next = Some(
            self.next_offset
                .map_or(candidate, |current| current.max(candidate)),
        );
        let changed = next != self.next_offset;
        self.next_offset = next;
        changed
    }

    async fn ensure_offset_loaded(&mut self) -> Result<()> {
        self.ensure_offset_storage_target_validated().await?;

        if self.offset_loaded {
            return Ok(());
        }

        if self.next_offset.is_none()
            && let Some(path) = self.config.persist_offset_path.as_deref()
        {
            self.next_offset = load_persisted_polling_offset_async(path.to_path_buf()).await?;
        }

        self.offset_loaded = true;
        Ok(())
    }

    async fn ensure_offset_storage_target_validated(&mut self) -> Result<()> {
        let Some(path) = self.config.persist_offset_path.as_deref() else {
            self.validated_offset_storage_path = None;
            return Ok(());
        };
        if self.validated_offset_storage_path.as_deref() == Some(path) {
            return Ok(());
        }

        let path = path.to_path_buf();
        let path = normalize_file_storage_target_async(path, "polling offset snapshot").await?;
        self.config.persist_offset_path = Some(path.clone());
        self.validated_offset_storage_path = Some(path);
        Ok(())
    }

    async fn flush_pending_persisted_offset(&mut self) -> Result<()> {
        let Some(next_offset) = self.pending_persisted_offset else {
            return Ok(());
        };

        self.ensure_offset_storage_target_validated().await?;
        let Some(path) = self.config.persist_offset_path.clone() else {
            self.pending_persisted_offset = None;
            return Ok(());
        };

        persist_polling_offset_async(path, Some(next_offset)).await?;
        self.pending_persisted_offset = None;
        Ok(())
    }

    fn is_duplicate_update(&self, update_id: i64) -> bool {
        if self.config.dedupe_window_size == 0 {
            return false;
        }
        self.seen_update_ids.contains(&update_id)
    }

    fn remember_update(&mut self, update_id: i64) {
        if self.config.dedupe_window_size == 0 {
            return;
        }

        if !self.seen_update_ids.insert(update_id) {
            return;
        }

        self.seen_update_order.push_back(update_id);
        while self.seen_update_order.len() > self.config.dedupe_window_size {
            if let Some(oldest) = self.seen_update_order.pop_front() {
                self.seen_update_ids.remove(&oldest);
            }
        }
    }

    fn trim_seen_update_ids(&mut self) {
        if self.config.dedupe_window_size == 0 {
            self.seen_update_ids.clear();
            self.seen_update_order.clear();
            return;
        }

        while self.seen_update_order.len() > self.config.dedupe_window_size {
            if let Some(oldest) = self.seen_update_order.pop_front() {
                self.seen_update_ids.remove(&oldest);
            }
        }
    }

    async fn commit_update_ids(&mut self, update_ids: &[i64]) -> Result<()> {
        if update_ids.is_empty() {
            return self.flush_pending_persisted_offset().await;
        }

        let previous_offset = self.next_offset;
        for update_id in update_ids {
            let _ = self.apply_committed_update(*update_id);
            self.remember_update(*update_id);
        }
        if self.next_offset != previous_offset {
            self.pending_persisted_offset = self.next_offset;
        }

        self.flush_pending_persisted_offset().await
    }

    fn effective_poll_timeout_seconds(&self) -> Result<u16> {
        self.resolved_poll_timeout_seconds(&self.config)
    }

    fn resolved_poll_timeout_seconds(&self, config: &PollingConfig) -> Result<u16> {
        config.resolve_poll_timeout_seconds(
            self.client.request_timeout(),
            self.client.total_timeout(),
        )
    }
}

impl UpdateSource for LongPollingSource {
    fn poll<'a>(&'a mut self) -> SourceFuture<'a> {
        Box::pin(async move {
            self.config.validate()?;
            self.ensure_prepared().await?;
            self.flush_pending_persisted_offset().await?;

            let mut request =
                GetUpdatesRequest::with_timeout(self.effective_poll_timeout_seconds()?);
            request.offset = self.next_offset;
            request.limit = self.config.limit;
            request.allowed_updates = self.config.allowed_updates.clone();

            let updates = self.client.updates().get_updates_once(&request).await?;

            let mut deduped = Vec::with_capacity(updates.len());
            let mut batch_seen = HashSet::new();
            for update in updates {
                if self.is_duplicate_update(update.update_id)
                    || !batch_seen.insert(update.update_id)
                {
                    continue;
                }
                deduped.push(update);
            }

            Ok(deduped)
        })
    }

    fn commit<'a>(&'a mut self, outcomes: &'a [DispatchOutcome]) -> SourceCommitFuture<'a> {
        Box::pin(async move {
            let update_ids = outcomes
                .iter()
                .map(|outcome| outcome.update_id())
                .collect::<Vec<_>>();
            self.commit_update_ids(&update_ids).await
        })
    }
}

#[derive(Clone, Debug, Default, serde::Serialize, serde::Deserialize)]
struct PollingOffsetSnapshot {
    #[serde(default = "default_polling_offset_snapshot_version")]
    version: u8,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    next_offset: Option<i64>,
}

fn default_polling_offset_snapshot_version() -> u8 {
    1
}

fn load_persisted_polling_offset(path: &Path) -> Result<Option<i64>> {
    let Some(raw) =
        read_optional_storage_file(path, "polling offset snapshot", "polling offset read")?
    else {
        return Ok(None);
    };

    if raw.is_empty() {
        return Ok(None);
    }

    let snapshot: PollingOffsetSnapshot = serde_json::from_slice(&raw).map_err(|source| {
        storage_decode_error(
            "polling offset decode",
            "polling offset snapshot",
            path,
            source,
        )
    })?;
    validate_polling_offset_snapshot(&snapshot).map_err(|source| {
        storage_snapshot_error(
            "polling offset validate",
            "polling offset snapshot",
            path,
            source,
        )
    })?;
    Ok(snapshot.next_offset)
}

fn validate_polling_offset_snapshot(snapshot: &PollingOffsetSnapshot) -> Result<()> {
    if snapshot.version != default_polling_offset_snapshot_version() {
        return Err(invalid_request(format!(
            "unsupported polling offset snapshot version `{}`",
            snapshot.version
        )));
    }
    if snapshot.next_offset.is_some_and(|offset| offset < 0) {
        return Err(invalid_request(
            "polling offset snapshot next_offset must not be negative",
        ));
    }

    Ok(())
}

fn persist_polling_offset(path: &Path, next_offset: Option<i64>) -> Result<()> {
    let snapshot = PollingOffsetSnapshot {
        version: default_polling_offset_snapshot_version(),
        next_offset,
    };
    validate_polling_offset_snapshot(&snapshot)?;
    let encoded = serde_json::to_vec(&snapshot).map_err(|source| {
        storage_encode_error("polling offset encode", "polling offset snapshot", source)
    })?;
    write_file_atomic(path, encoded.as_slice(), "polling offset snapshot")?;
    Ok(())
}

async fn load_persisted_polling_offset_async(path: PathBuf) -> Result<Option<i64>> {
    run_blocking_io(move || load_persisted_polling_offset(path.as_path())).await
}

async fn persist_polling_offset_async(path: PathBuf, next_offset: Option<i64>) -> Result<()> {
    run_blocking_io(move || persist_polling_offset(path.as_path(), next_offset)).await
}

/// Sink side of a channel-backed update source.
#[derive(Clone)]
pub struct UpdateSink {
    sender: mpsc::Sender<Update>,
}

impl UpdateSink {
    pub fn new(sender: mpsc::Sender<Update>) -> Self {
        Self { sender }
    }

    pub async fn send(&self, update: Update) -> Result<()> {
        self.sender
            .send(update)
            .await
            .map_err(|_| runtime_error("update sink channel is closed"))?;
        Ok(())
    }
}

/// Source side of a channel-backed update source.
pub struct ChannelUpdateSource {
    receiver: mpsc::Receiver<Update>,
    max_batch: usize,
    in_flight: VecDeque<Update>,
}

impl ChannelUpdateSource {
    pub fn new(receiver: mpsc::Receiver<Update>) -> Self {
        Self {
            receiver,
            max_batch: 32,
            in_flight: VecDeque::new(),
        }
    }

    pub fn with_max_batch(mut self, max_batch: usize) -> Result<Self> {
        if max_batch == 0 {
            return Err(Error::Configuration {
                reason: "channel update source max_batch must be at least 1".to_owned(),
            });
        }
        self.max_batch = max_batch;
        Ok(self)
    }
}

impl UpdateSource for ChannelUpdateSource {
    fn poll<'a>(&'a mut self) -> SourceFuture<'a> {
        Box::pin(async move {
            if self.in_flight.is_empty() {
                let Some(first) = self.receiver.recv().await else {
                    return Err(runtime_error("update source channel is closed"));
                };

                self.in_flight.push_back(first);

                while self.in_flight.len() < self.max_batch {
                    match self.receiver.try_recv() {
                        Ok(update) => self.in_flight.push_back(update),
                        Err(mpsc::error::TryRecvError::Empty) => break,
                        Err(mpsc::error::TryRecvError::Disconnected) => break,
                    }
                }
            }

            Ok(self.in_flight.iter().cloned().collect())
        })
    }

    fn commit<'a>(&'a mut self, outcomes: &'a [DispatchOutcome]) -> SourceCommitFuture<'a> {
        Box::pin(async move {
            for outcome in outcomes {
                let Some(front) = self.in_flight.front() else {
                    return Err(runtime_error(
                        "channel update source commit received more outcomes than in-flight updates",
                    ));
                };
                if front.update_id != outcome.update_id() {
                    return Err(runtime_error(
                        "channel update source commit must acknowledge an ordered update prefix",
                    ));
                }
                let _ = self.in_flight.pop_front();
            }
            Ok(())
        })
    }
}

/// Creates a webhook-friendly channel source pair.
pub fn channel_source(buffer: usize) -> Result<(UpdateSink, ChannelUpdateSource)> {
    if buffer == 0 {
        return Err(Error::Configuration {
            reason: "channel source buffer must be at least 1".to_owned(),
        });
    }
    let (sender, receiver) = mpsc::channel(buffer);
    Ok((UpdateSink::new(sender), ChannelUpdateSource::new(receiver)))
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn validates_polling_offset_snapshot_metadata() {
        let mut snapshot = PollingOffsetSnapshot {
            version: default_polling_offset_snapshot_version(),
            next_offset: Some(1),
        };
        assert!(validate_polling_offset_snapshot(&snapshot).is_ok());

        snapshot.version = snapshot.version.saturating_add(1);
        assert!(matches!(
            validate_polling_offset_snapshot(&snapshot),
            Err(Error::InvalidRequest { .. })
        ));

        snapshot.version = default_polling_offset_snapshot_version();
        snapshot.next_offset = Some(-1);
        assert!(matches!(
            validate_polling_offset_snapshot(&snapshot),
            Err(Error::InvalidRequest { .. })
        ));
    }

    #[tokio::test]
    async fn explicit_offset_override_skips_persisted_offset_load() -> Result<()> {
        let timestamp = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map_or(0_u128, |duration| duration.as_nanos());
        let offset_path = std::env::temp_dir().join(format!(
            "tele-offset-explicit-override-{}-{timestamp}.json",
            std::process::id()
        ));
        let changed_offset_path = std::env::temp_dir().join(format!(
            "tele-offset-explicit-override-changed-{}-{timestamp}.json",
            std::process::id()
        ));
        let _ = fs::remove_file(&offset_path);
        let _ = fs::remove_file(&changed_offset_path);
        persist_polling_offset(&offset_path, Some(42))?;
        persist_polling_offset(&changed_offset_path, Some(77))?;

        let client = Client::builder("http://127.0.0.1:9")?
            .bot_token("123:abc")?
            .build()?;
        let mut source = LongPollingSource::new(client).with_offset_persistence_path(&offset_path);

        source.set_next_offset(None);
        source.set_offset_persistence_path(&changed_offset_path);
        source.ensure_offset_loaded().await?;

        assert_eq!(source.next_offset(), None);

        let _ = fs::remove_file(&offset_path);
        let _ = fs::remove_file(&changed_offset_path);
        Ok(())
    }

    #[tokio::test]
    async fn enabling_offset_persistence_reloads_when_no_offset_is_authoritative() -> Result<()> {
        let timestamp = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map_or(0_u128, |duration| duration.as_nanos());
        let offset_path = std::env::temp_dir().join(format!(
            "tele-offset-enable-persistence-{}-{timestamp}.json",
            std::process::id()
        ));
        let _ = fs::remove_file(&offset_path);
        persist_polling_offset(&offset_path, Some(77))?;

        let client = Client::builder("http://127.0.0.1:9")?
            .bot_token("123:abc")?
            .build()?;
        let mut source = LongPollingSource::new(client);
        source.ensure_offset_loaded().await?;
        assert_eq!(source.next_offset(), None);

        source.set_offset_persistence_path(&offset_path);
        source.ensure_offset_loaded().await?;

        assert_eq!(source.next_offset(), Some(77));

        let _ = fs::remove_file(&offset_path);
        Ok(())
    }

    #[tokio::test]
    async fn offset_persistence_path_is_normalized_after_validation() -> Result<()> {
        let timestamp = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map_or(0_u128, |duration| duration.as_nanos());
        let root = std::env::temp_dir().join(format!(
            "tele-offset-normalized-path-{}-{timestamp}",
            std::process::id()
        ));
        let nested = root.join("nested");
        fs::create_dir_all(&nested).map_err(|source| {
            storage_error(
                "test offset mkdir",
                format!("failed to create offset test directory: {source}"),
                true,
            )
        })?;
        let path = nested.join("..").join("offset.json");
        let expected = root.canonicalize().map_err(|source| {
            storage_error(
                "test offset canonicalize",
                format!("failed to canonicalize offset test root: {source}"),
                true,
            )
        })?;

        let client = Client::builder("http://127.0.0.1:9")?
            .bot_token("123:abc")?
            .build()?;
        let mut source = LongPollingSource::new(client).with_offset_persistence_path(path);
        source.ensure_offset_loaded().await?;

        assert_eq!(
            source.config.persist_offset_path.as_deref(),
            Some(expected.join("offset.json").as_path())
        );
        assert_eq!(
            source.validated_offset_storage_path.as_deref(),
            Some(expected.join("offset.json").as_path())
        );
        let _ = fs::remove_dir_all(root);
        Ok(())
    }

    #[tokio::test]
    async fn offset_persistence_path_change_invalidates_validation_cache() -> Result<()> {
        let timestamp = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map_or(0_u128, |duration| duration.as_nanos());
        let root = std::env::temp_dir().join(format!(
            "tele-offset-path-change-{}-{timestamp}",
            std::process::id()
        ));
        fs::create_dir_all(&root).map_err(|source| {
            storage_error(
                "test mkdir",
                format!("failed to create test root: {source}"),
                true,
            )
        })?;
        let valid_path = root.join("offset.json");
        let blocked_parent = root.join("not-a-directory");
        fs::write(&blocked_parent, b"not a directory").map_err(|source| {
            storage_error(
                "test write",
                format!("failed to create blocked path: {source}"),
                true,
            )
        })?;
        let invalid_path = blocked_parent.join("offset.json");

        let client = Client::builder("http://127.0.0.1:9")?
            .bot_token("123:abc")?
            .build()?;
        let mut source = LongPollingSource::new(client).with_offset_persistence_path(&valid_path);
        source.set_next_offset(Some(42));
        source.ensure_offset_loaded().await?;
        assert_eq!(
            source.validated_offset_storage_path.as_deref(),
            Some(valid_path.as_path())
        );

        source.set_offset_persistence_path(invalid_path);
        let result = source.ensure_offset_loaded().await;

        assert!(matches!(result, Err(Error::Storage { .. })));
        let _ = fs::remove_dir_all(root);
        Ok(())
    }

    #[test]
    fn set_config_resets_prepared_when_webhook_startup_policy_changes() -> Result<()> {
        let client = Client::builder("http://127.0.0.1:9")?
            .bot_token("123:abc")?
            .build()?;
        let mut source = LongPollingSource::new(client);
        source.prepared = true;

        let config = PollingConfig {
            disable_webhook_on_start: false,
            ..PollingConfig::default()
        };
        source.set_config(config)?;

        assert!(!source.prepared);
        Ok(())
    }

    #[test]
    fn dedupe_window_reconfiguration_trims_cached_update_ids() -> Result<()> {
        let client = Client::builder("http://127.0.0.1:9")?
            .bot_token("123:abc")?
            .build()?;
        let mut source = LongPollingSource::new(client);

        source.remember_update(1);
        source.remember_update(2);
        source.remember_update(3);
        source.set_dedupe_window_size(2);

        assert!(!source.is_duplicate_update(1));
        assert!(source.is_duplicate_update(2));
        assert!(source.is_duplicate_update(3));

        source.set_config(PollingConfig {
            dedupe_window_size: 0,
            ..PollingConfig::default()
        })?;

        assert!(!source.is_duplicate_update(2));
        assert!(source.seen_update_ids.is_empty());
        assert!(source.seen_update_order.is_empty());
        Ok(())
    }

    #[tokio::test]
    async fn explicit_offset_override_clears_dedupe_window() -> Result<()> {
        let client = Client::builder("http://127.0.0.1:9")?
            .bot_token("123:abc")?
            .build()?;
        let mut source = LongPollingSource::new(client);

        source.commit_update_ids(&[10]).await?;
        assert_eq!(source.next_offset(), Some(11));
        assert!(source.is_duplicate_update(10));

        source.set_next_offset(Some(10));

        assert_eq!(source.next_offset(), Some(10));
        assert!(!source.is_duplicate_update(10));
        assert!(source.offset_loaded);

        Ok(())
    }

    #[tokio::test]
    async fn failed_offset_persistence_keeps_committed_memory_progress() -> Result<()> {
        let timestamp = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map_or(0_u128, |duration| duration.as_nanos());
        let root = std::env::temp_dir().join(format!(
            "tele-offset-persist-failure-{}-{timestamp}",
            std::process::id()
        ));
        let offset_path = root.join("offset.json");
        fs::create_dir_all(&offset_path).map_err(|source| {
            storage_error(
                "test mkdir",
                format!("failed to create blocked offset path: {source}"),
                true,
            )
        })?;

        let client = Client::builder("http://127.0.0.1:9")?
            .bot_token("123:abc")?
            .build()?;
        let mut source = LongPollingSource::new(client).with_offset_persistence_path(&offset_path);

        let result = source.commit_update_ids(&[41]).await;

        assert!(matches!(result, Err(Error::Storage { .. })));
        assert_eq!(source.next_offset(), Some(42));
        assert!(source.is_duplicate_update(41));
        assert_eq!(source.pending_persisted_offset, Some(42));

        fs::remove_dir(&offset_path).map_err(|source| {
            storage_error(
                "test rmdir",
                format!("failed to unblock offset path: {source}"),
                true,
            )
        })?;
        source.flush_pending_persisted_offset().await?;

        assert_eq!(source.pending_persisted_offset, None);
        assert_eq!(load_persisted_polling_offset(&offset_path)?, Some(42));

        let _ = fs::remove_dir_all(root);
        Ok(())
    }
}