lakehouse/crates/vectord/src/activation.rs

//! Profile activation tracking (Phase 41 PRD).
//!
//! Phase 41 PRD called out `crates/vectord/src/activation.rs` with
//! `ActivationTracker` + background-job pattern. The activation
//! handler itself lives in `service.rs::activate_profile` (200+ lines
//! of warm-up + bucket binding that's wired to VectorState); this
//! module provides the type the PRD named and a single-flight guard
//! that satisfies the PRD gate "refuse new activation if one is
//! pending/running."
//!
//! Handler extraction (moving the body of `activate_profile` here)
//! is deliberately NOT in this commit — it's a module-structure
//! refactor, not a semantic change. When that lands, the inline
//! `tokio::spawn` in `service.rs` moves into `ActivationTracker::start`
//! and the HTTP handler shrinks to ~20 lines of validate + start +
//! respond-202.

use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::RwLock;

/// Tracks in-flight profile activations. The PRD's "single-flight guard"
/// lives here: callers check `is_pending` before starting a new activation
/// and register via `mark_pending` if they proceed. On completion, they
/// call `mark_complete` so the next caller can start.
///
/// Per-profile granularity — activating profile A doesn't block B.
#[derive(Clone, Default)]
pub struct ActivationTracker {
    pending: Arc<RwLock<HashMap<String, String>>>, // profile_id → job_id
}

impl ActivationTracker {
    pub fn new() -> Self {
        Self::default()
    }

    /// Check if a profile has an activation already running. Returns the
    /// in-flight job_id if so. Safe to call without holding a lock.
    pub async fn is_pending(&self, profile_id: &str) -> Option<String> {
        self.pending.read().await.get(profile_id).cloned()
    }

    /// Register a new activation as pending. Returns false if an
    /// activation is already running for the same profile (caller should
    /// return 409 Conflict or surface the existing job_id). Returns true
    /// on successful registration.
    pub async fn mark_pending(&self, profile_id: &str, job_id: &str) -> bool {
        let mut guard = self.pending.write().await;
        if guard.contains_key(profile_id) {
            return false;
        }
        guard.insert(profile_id.to_string(), job_id.to_string());
        true
    }

    /// Remove the pending marker when activation finishes (success OR
    /// failure — both free the slot for the next caller).
    pub async fn mark_complete(&self, profile_id: &str) {
        self.pending.write().await.remove(profile_id);
    }

    /// How many activations are currently in-flight across all profiles.
    pub async fn in_flight_count(&self) -> usize {
        self.pending.read().await.len()
    }
}

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

    #[tokio::test]
    async fn empty_tracker_has_no_pending() {
        let t = ActivationTracker::new();
        assert_eq!(t.in_flight_count().await, 0);
        assert!(t.is_pending("any-profile").await.is_none());
    }

    #[tokio::test]
    async fn mark_pending_registers_the_job() {
        let t = ActivationTracker::new();
        assert!(t.mark_pending("profile-A", "job-1").await);
        assert_eq!(t.in_flight_count().await, 1);
        assert_eq!(t.is_pending("profile-A").await, Some("job-1".into()));
    }

    #[tokio::test]
    async fn single_flight_guard_refuses_second_activation_same_profile() {
        // PRD Phase 41 gate: "refuse new activation if one is
        // pending/running." Same profile twice → second call returns
        // false, caller must surface the in-flight job_id.
        let t = ActivationTracker::new();
        assert!(t.mark_pending("profile-A", "job-1").await);
        assert!(!t.mark_pending("profile-A", "job-2").await);
        // Still the first job — second registration didn't overwrite.
        assert_eq!(t.is_pending("profile-A").await, Some("job-1".into()));
    }

    #[tokio::test]
    async fn different_profiles_dont_block_each_other() {
        // Per-profile granularity — activating A doesn't block B.
        let t = ActivationTracker::new();
        assert!(t.mark_pending("profile-A", "job-1").await);
        assert!(t.mark_pending("profile-B", "job-2").await);
        assert_eq!(t.in_flight_count().await, 2);
    }

    #[tokio::test]
    async fn mark_complete_frees_the_slot() {
        let t = ActivationTracker::new();
        t.mark_pending("profile-A", "job-1").await;
        t.mark_complete("profile-A").await;
        assert_eq!(t.in_flight_count().await, 0);
        // Next activation can now proceed.
        assert!(t.mark_pending("profile-A", "job-2").await);
    }
}