active-call 0.3.56

use super::processor::Processor;
use crate::event::{EventSender, SessionEvent};
use crate::media::{AudioFrame, get_timestamp};
use anyhow::Result;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Duration;
use tokio_util::sync::CancellationToken;
use tracing::info;

pub struct InactivityProcessor {
    last_received: Arc<AtomicU64>,
}

impl InactivityProcessor {
    pub fn new(
        track_id: String,
        timeout: Duration,
        event_sender: EventSender,
        cancel_token: CancellationToken,
    ) -> Self {
        let last_received = Arc::new(AtomicU64::new(get_timestamp()));
        let last_received_clone = last_received.clone();
        let track_id_clone = track_id.clone();

        crate::spawn(async move {
            let mut interval = tokio::time::interval(Duration::from_secs(1));
            loop {
                tokio::select! {
                    _ = cancel_token.cancelled() => break,
                    _ = interval.tick() => {
                        let last = last_received_clone.load(Ordering::SeqCst);
                        let now = get_timestamp();
                        if now > last && now - last > timeout.as_millis() as u64 {
                            info!(track_id = track_id_clone, "Inactivity timeout reached, sending inactivity event");
                            let _ = event_sender.send(SessionEvent::Inactivity {
                                track_id: track_id_clone.clone(),
                                timestamp: now,
                            });
                            break;
                        }
                    }
                }
            }
        });

        Self { last_received }
    }
}

impl Processor for InactivityProcessor {
    fn process_frame(&mut self, _frame: &mut AudioFrame) -> Result<()> {
        self.last_received.store(get_timestamp(), Ordering::SeqCst);
        Ok(())
    }
}