scout_rs 0.95.95

use std::collections::HashMap;

use anyhow::{anyhow, Result};
use postgrest::Postgrest;
use serde::{Deserialize, Serialize};

use crate::constants::env;
use crate::models::serde_helpers::serialized_object_key_set;
use crate::models::{Artifact, Event, ProductCompatibility, Session};
#[cfg(feature = "media")]
use crate::scan::{page_from_rows, time_id_keyset_or, ScanCursor, ScanPage, ScanRequest, ScanRow};

/// Pulls the `message` field out of a PostgREST error body, so a failure reports what Postgres
/// said rather than the whole JSON envelope.
pub(crate) fn parse_postgrest_error_message(body: &str) -> Option<String> {
    let v: serde_json::Value = serde_json::from_str(body).ok()?;
    let msg = v.as_object()?.get("message")?.as_str()?;
    Some(msg.to_string())
}

/// Response type for get_connectivity_for_artifact RPC (connectivity_with_coordinates).
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub struct ConnectivityWithCoordinates {
    pub id: Option<i64>,
    pub session_id: Option<i64>,
    pub device_id: Option<i64>,
    pub inserted_at: Option<String>,
    pub timestamp_start: Option<String>,
    pub signal: Option<f64>,
    pub noise: Option<f64>,
    pub altitude: Option<f64>,
    pub heading: Option<f64>,
    pub latitude: Option<f64>,
    pub longitude: Option<f64>,
    pub h14_index: Option<String>,
    pub h13_index: Option<String>,
    pub h12_index: Option<String>,
    pub h11_index: Option<String>,
    pub battery_percentage: Option<f32>,
    pub frequency_hz: Option<f32>,
    pub bandwidth_hz: Option<f32>,
    pub associated_station: Option<String>,
    pub mode: Option<String>,
    pub battery_volts: Option<f32>,
    #[serde(default)]
    pub link_type: crate::models::LinkType,
    pub quality_normalized: Option<f32>,
    pub secondary_battery_percentage: Option<f32>,
    pub secondary_battery_volts: Option<f32>,
    pub frame_center_latitude: Option<f64>,
    pub frame_center_longitude: Option<f64>,
    pub frame_center_altitude: Option<f64>,
    pub base_device_online: Option<bool>,
    pub frame_center_h14_index: Option<String>,
    pub frame_center_h13_index: Option<String>,
    pub frame_center_h12_index: Option<String>,
    pub frame_center_h11_index: Option<String>,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct UserSessionTimeFilters {
    pub user_id: String,
}

impl UserSessionTimeFilters {
    pub fn new(user_id: impl Into<String>) -> Self {
        Self {
            user_id: user_id.into(),
        }
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DatabaseConfig {
    pub rest_url: String,
    pub scout_api_key: String,
    pub supabase_api_key: String,
    /// When set, used as apikey + Bearer auth to bypass RLS; device api key is not sent.
    pub service_role_key: Option<String>,
    /// When true and service_role_key is set, connect() uses service key. When false, connect() uses device API key (required for identify/RPC).
    pub use_service_key: bool,
}

/// Ensures a Supabase base URL ends with the PostgREST path.
fn normalize_rest_url(mut url: String) -> String {
    if !url.ends_with("/rest/v1") {
        if url.ends_with('/') {
            url.push_str("rest/v1");
        } else {
            url.push_str("/rest/v1");
        }
    }
    url
}

impl DatabaseConfig {
    /// Creates a new database config from environment variables with an optional Scout device API key
    /// If the scout device API key is not provided, it will fall back to the environment variable.
    ///
    /// The environment variables are:
    /// See [`crate::constants::env`] for variable names.
    pub fn from_env() -> Result<Self> {
        dotenv::dotenv().ok();

        let rest_url =
            normalize_rest_url(std::env::var(env::SCOUT_DATABASE_REST_URL).map_err(|_| {
                anyhow!(
                    "{} environment variable is required",
                    env::SCOUT_DATABASE_REST_URL
                )
            })?);

        let service_role_key = std::env::var(env::SUPABASE_SERVICE_ROLE_KEY).ok();
        let use_service_key = std::env::var(env::SCOUT_USE_SERVICE_KEY)
            .map(|v| v == "1" || v.eq_ignore_ascii_case("true"))
            .unwrap_or(false);

        let (scout_api_key, supabase_api_key) = if use_service_key && service_role_key.is_some() {
            (
                std::env::var(env::SCOUT_DEVICE_API_KEY).unwrap_or_default(),
                std::env::var(env::SUPABASE_PUBLIC_API_KEY).unwrap_or_default(),
            )
        } else {
            let scout_api_key = std::env::var(env::SCOUT_DEVICE_API_KEY).map_err(|_| {
                anyhow!(
                    "{} environment variable not found",
                    env::SCOUT_DEVICE_API_KEY
                )
            })?;
            let supabase_api_key = std::env::var(env::SUPABASE_PUBLIC_API_KEY).map_err(|_| {
                anyhow!(
                    "{} environment variable is required for Supabase access",
                    env::SUPABASE_PUBLIC_API_KEY
                )
            })?;
            (scout_api_key, supabase_api_key)
        };

        Ok(DatabaseConfig {
            rest_url,
            scout_api_key,
            supabase_api_key,
            service_role_key,
            use_service_key,
        })
    }

    /// Service-role only config from env (for edge services: video-tagger, tracker).
    /// Needs [`env::SCOUT_DATABASE_REST_URL`] and [`env::SUPABASE_SERVICE_ROLE_KEY`]. Returns None if either is missing.
    pub fn from_env_service_role() -> Option<Self> {
        let rest_url = normalize_rest_url(std::env::var(env::SCOUT_DATABASE_REST_URL).ok()?);
        let service_role_key = std::env::var(env::SUPABASE_SERVICE_ROLE_KEY).ok()?;
        Some(Self::service_role(rest_url, service_role_key))
    }

    /// Service-role config from a known key (no env required); bypasses RLS.
    pub fn service_role(rest_url: impl Into<String>, service_role_key: impl Into<String>) -> Self {
        DatabaseConfig {
            rest_url: normalize_rest_url(rest_url.into()),
            scout_api_key: String::new(),
            supabase_api_key: String::new(),
            service_role_key: Some(service_role_key.into()),
            use_service_key: true,
        }
    }

    /// Gets the PostgREST endpoint URL
    pub fn get_rest_url(&self) -> &str {
        &self.rest_url
    }

    /// Gets the Scout API key for custom authentication
    pub fn get_scout_api_key(&self) -> &str {
        &self.scout_api_key
    }

    /// Gets the Supabase API key for PostgREST access
    pub fn get_supabase_api_key(&self) -> &str {
        &self.supabase_api_key
    }

    /// Base URL for Supabase Edge Functions, derived from [`Self::get_rest_url`].
    pub fn get_functions_url(&self) -> Result<String> {
        let rest = self.get_rest_url().trim_end_matches('/');
        let base = rest
            .strip_suffix("/rest/v1")
            .ok_or_else(|| anyhow!("rest URL must end with /rest/v1: {rest}"))?;
        Ok(format!("{base}/functions/v1"))
    }
}

pub struct ScoutDbClient {
    config: DatabaseConfig,
    client: Option<Postgrest>,
}

impl std::fmt::Debug for ScoutDbClient {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("ScoutDbClient")
            .field("config", &self.config)
            .field(
                "client",
                if self.client.is_some() {
                    &"Connected"
                } else {
                    &"Disconnected"
                },
            )
            .finish()
    }
}

impl ScoutDbClient {
    pub fn new(config: DatabaseConfig) -> Self {
        Self {
            config,
            client: None,
        }
    }

    /// Establishes a connection to the database via PostgREST
    pub fn connect(&mut self) -> Result<()> {
        let rest_url = self.config.get_rest_url();

        let client = if self.config.use_service_key {
            if let Some(ref key) = self.config.service_role_key {
                Postgrest::new(rest_url)
                    .insert_header("apikey", key)
                    .insert_header("Authorization", format!("Bearer {}", key))
            } else {
                Postgrest::new(rest_url)
                    .insert_header("apikey", self.config.get_supabase_api_key())
                    .insert_header("api_key", &format!("{}", self.config.get_scout_api_key()))
            }
        } else {
            Postgrest::new(rest_url)
                .insert_header("apikey", self.config.get_supabase_api_key())
                .insert_header("api_key", &format!("{}", self.config.get_scout_api_key()))
        };

        self.client = Some(client);

        Ok(())
    }

    /// Gets the PostgREST client, ensuring connection is established
    pub fn get_client(&mut self) -> Result<&Postgrest> {
        if self.client.is_none() {
            self.connect()?;
        }

        self.client
            .as_ref()
            .ok_or_else(|| anyhow!("No PostgREST client available"))
    }

    /// Closes the database connection
    pub fn disconnect(&mut self) {
        if self.client.is_some() {
            self.client = None;
        }
    }

    /// Fetches an `id` or `(time_column, id)` ascending keyset page.
    #[cfg(feature = "media")]
    pub(crate) async fn scan_page<T>(&mut self, request: ScanRequest) -> Result<ScanPage<T>>
    where
        T: for<'de> serde::Deserialize<'de> + ScanRow,
    {
        if !(1..=500).contains(&request.batch_size) {
            return Err(anyhow!("batch_size must be between 1 and 500"));
        }

        let table = request.table.clone();
        let select = request.select.clone();
        let time_column = request.time_column.clone();
        let batch_size = request.batch_size;
        let cursor = request.cursor.clone();

        let rows: Vec<T> = self
            .query(|client| {
                let order = format!("{time_column}.asc,id.asc");
                let query = client
                    .from(table.as_str())
                    .select(select.as_str())
                    .order(order.as_str())
                    .limit(batch_size + 1);
                match &cursor {
                    ScanCursor::Start => query,
                    ScanCursor::AfterTimeId { last_time, last_id } => {
                        query.or(time_id_keyset_or(&time_column, last_time, *last_id))
                    }
                }
            })
            .await?;

        Ok(page_from_rows(rows, batch_size))
    }

    /// Executes a query and returns the results
    pub async fn query<T>(
        &mut self,
        query_builder: impl FnOnce(&Postgrest) -> postgrest::Builder,
    ) -> Result<Vec<T>>
    where
        T: for<'de> serde::Deserialize<'de>,
    {
        let client = self.get_client()?;

        let builder = query_builder(client);
        let response = builder.execute().await?;

        let body = response.text().await?;

        // Try to parse as the expected type first
        if let Ok(results) = serde_json::from_str::<Vec<T>>(&body) {
            Ok(results)
        } else {
            // If that fails, try to parse as an error response
            if let Ok(error_response) = serde_json::from_str::<serde_json::Value>(&body) {
                if let Some(error_msg) = error_response.get("error") {
                    return Err(anyhow!("Database error: {}", error_msg));
                } else if let Some(message) = error_response.get("message") {
                    return Err(anyhow!("Database message: {}", message));
                } else {
                    return Err(anyhow!("Database returned unexpected format: {}", body));
                }
            } else {
                return Err(anyhow!(
                    "Failed to parse database response as JSON: {}",
                    body
                ));
            }
        }
    }

    /// Executes a query that returns a single row
    pub async fn query_one<T>(
        &mut self,
        query_builder: impl FnOnce(&Postgrest) -> postgrest::Builder,
    ) -> Result<T>
    where
        T: for<'de> serde::Deserialize<'de>,
    {
        let client = self.get_client()?;

        let builder = query_builder(client);
        let response = builder.execute().await?;

        let body = response.text().await?;
        let results: Vec<T> = serde_json::from_str(&body)?;

        if results.is_empty() {
            return Err(anyhow!("No results found"));
        }

        Ok(results.into_iter().next().unwrap())
    }

    /// Executes a query that doesn't return results (INSERT, UPDATE, DELETE)
    pub async fn execute(
        &mut self,
        query_builder: impl FnOnce(&Postgrest) -> postgrest::Builder,
    ) -> Result<()> {
        let client = self.get_client()?;

        let builder = query_builder(client);
        let response = builder.execute().await?;

        let status = response.status();
        if !status.is_success() {
            let error_text = response.text().await?;
            return Err(anyhow!(
                "Operation failed: HTTP {} - {}",
                status,
                error_text
            ));
        }

        Ok(())
    }

    /// Calls a PostgREST function and deserializes its result. Accepts any function the
    /// deployment exposes, including ones added outside Scout's schema.
    pub async fn rpc<P, T>(&mut self, function: &str, params: &P) -> Result<T>
    where
        P: serde::Serialize + ?Sized,
        T: for<'de> serde::Deserialize<'de>,
    {
        let body = self.rpc_body(function, params).await?;
        serde_json::from_str(&body).map_err(|error| {
            anyhow!(
                "Failed to parse {} response: {} - {}",
                function,
                error,
                body
            )
        })
    }

    /// Calls a PostgREST function, discarding its result.
    pub async fn rpc_execute<P>(&mut self, function: &str, params: &P) -> Result<()>
    where
        P: serde::Serialize + ?Sized,
    {
        self.rpc_body(function, params).await.map(|_| ())
    }

    async fn rpc_body<P>(&mut self, function: &str, params: &P) -> Result<String>
    where
        P: serde::Serialize + ?Sized,
    {
        let payload = serde_json::to_string(params)?;
        let client = self.get_client()?;
        let response = client.rpc(function, payload).execute().await?;

        let status = response.status();
        let body = response.text().await?;
        if !status.is_success() {
            let detail = parse_postgrest_error_message(&body).unwrap_or(body);
            return Err(anyhow!("{} failed: HTTP {} - {}", function, status, detail));
        }

        Ok(body)
    }

    /// Inserts data into a table
    pub async fn insert<T>(&mut self, table: &str, data: &T) -> Result<Vec<T>>
    where
        T: for<'de> serde::Deserialize<'de> + serde::Serialize,
    {
        let client = self.get_client()?;

        let json_data = serde_json::to_string(data)?;

        let response = client.from(table).insert(&json_data).execute().await?;

        let body = response.text().await?;

        // Try to parse as the expected type first
        if let Ok(results) = serde_json::from_str::<Vec<T>>(&body) {
            Ok(results)
        } else {
            // If that fails, try to parse as an error response
            if let Ok(error_response) = serde_json::from_str::<serde_json::Value>(&body) {
                if let Some(error_msg) = error_response.get("error") {
                    return Err(anyhow!("Database insert error: {}", error_msg));
                } else if let Some(message) = error_response.get("message") {
                    return Err(anyhow!("Database insert message: {}", message));
                } else {
                    return Err(anyhow!(
                        "Database insert returned unexpected format: {}",
                        body
                    ));
                }
            } else {
                return Err(anyhow!(
                    "Failed to parse database insert response as JSON: {}",
                    body
                ));
            }
        }
    }

    /// Inserts multiple items in a single bulk operation
    pub async fn insert_bulk<T>(&mut self, table: &str, data: &[T]) -> Result<Vec<T>>
    where
        T: for<'de> serde::Deserialize<'de> + serde::Serialize,
    {
        let client = self.get_client()?;

        let json_data = serde_json::to_string(data)?;

        let response = client.from(table).insert(&json_data).execute().await?;

        let body = response.text().await?;

        // Try to parse as the expected type first
        if let Ok(results) = serde_json::from_str::<Vec<T>>(&body) {
            Ok(results)
        } else {
            // If that fails, try to parse as an error response
            if let Ok(error_response) = serde_json::from_str::<serde_json::Value>(&body) {
                if let Some(error_msg) = error_response.get("error") {
                    return Err(anyhow!("Database bulk insert error: {}", error_msg));
                } else if let Some(message) = error_response.get("message") {
                    return Err(anyhow!("Database bulk insert message: {}", message));
                } else {
                    return Err(anyhow!(
                        "Database bulk insert returned unexpected format: {}",
                        body
                    ));
                }
            } else {
                return Err(anyhow!(
                    "Failed to parse database bulk insert response as JSON: {}",
                    body
                ));
            }
        }
    }

    /// Bulk insert split by serialized JSON key set (PostgREST needs uniform columns per request).
    pub async fn insert_bulk_batched_by_shape<T>(
        &mut self,
        table: &str,
        rows: &[T],
    ) -> Result<Vec<T>>
    where
        T: Serialize + Clone + for<'de> Deserialize<'de>,
    {
        if rows.is_empty() {
            return Ok(Vec::new());
        }
        let mut by_shape: HashMap<String, Vec<usize>> = HashMap::new();
        for (i, row) in rows.iter().enumerate() {
            let key = serialized_object_key_set(row)?;
            by_shape.entry(key).or_default().push(i);
        }
        let mut out = Vec::new();
        for indices in by_shape.values() {
            let batch: Vec<T> = indices.iter().map(|&i| rows[i].clone()).collect();
            let mut part = self.insert_bulk(table, &batch).await?;
            out.append(&mut part);
        }
        Ok(out)
    }

    /// Upserts multiple items in a single bulk operation (insert or update on conflict)
    pub async fn upsert_bulk<T>(&mut self, table: &str, data: &[T]) -> Result<Vec<T>>
    where
        T: for<'de> serde::Deserialize<'de> + serde::Serialize,
    {
        let client = self.get_client()?;

        let json_data = serde_json::to_string(data)?;

        let response = client
            .from(table)
            .upsert(&json_data)
            .on_conflict("id")
            .execute()
            .await?;

        let body = response.text().await?;

        // Try to parse as the expected type first
        if let Ok(results) = serde_json::from_str::<Vec<T>>(&body) {
            Ok(results)
        } else {
            // If that fails, try to parse as an error response
            if let Ok(error_response) = serde_json::from_str::<serde_json::Value>(&body) {
                if let Some(error_msg) = error_response.get("error") {
                    return Err(anyhow!("Database bulk upsert error: {}", error_msg));
                } else if let Some(message) = error_response.get("message") {
                    return Err(anyhow!("Database bulk upsert message: {}", message));
                } else {
                    return Err(anyhow!(
                        "Database bulk upsert returned unexpected format: {}",
                        body
                    ));
                }
            } else {
                return Err(anyhow!(
                    "Failed to parse database bulk upsert response as JSON: {}",
                    body
                ));
            }
        }
    }

    /// Bulk upsert split by serialized JSON key set (PostgREST needs uniform columns per request).
    pub async fn upsert_bulk_batched_by_shape<T>(
        &mut self,
        table: &str,
        rows: &[T],
    ) -> Result<Vec<T>>
    where
        T: Serialize + Clone + for<'de> Deserialize<'de>,
    {
        if rows.is_empty() {
            return Ok(Vec::new());
        }
        let mut by_shape: HashMap<String, Vec<usize>> = HashMap::new();
        for (i, row) in rows.iter().enumerate() {
            let key = serialized_object_key_set(row)?;
            by_shape.entry(key).or_default().push(i);
        }
        let mut out = Vec::new();
        for indices in by_shape.values() {
            let batch: Vec<T> = indices.iter().map(|&i| rows[i].clone()).collect();
            let mut part = self.upsert_bulk(table, &batch).await?;
            out.append(&mut part);
        }
        Ok(out)
    }

    /// Updates data in a table
    pub async fn update<T>(
        &mut self,
        data: &T,
        filter_builder: impl FnOnce(&Postgrest) -> postgrest::Builder,
    ) -> Result<Vec<T>>
    where
        T: for<'de> serde::Deserialize<'de> + serde::Serialize,
    {
        let client = self.get_client()?;

        let json_data = serde_json::to_string(data)?;

        let builder = filter_builder(client);
        let response = builder.update(&json_data).execute().await?;

        let status = response.status();
        let body = response.text().await?;

        if !status.is_success() {
            if let Ok(v) = serde_json::from_str::<serde_json::Value>(&body) {
                if let Some(m) = v.get("message").and_then(|x| x.as_str()) {
                    return Err(anyhow!("Database update error: {} (HTTP {})", m, status));
                }
            }
            return Err(anyhow!(
                "Database update failed: HTTP {} - {}",
                status,
                body
            ));
        }

        if body.trim().is_empty() {
            return Ok(vec![]);
        }

        if let Ok(results) = serde_json::from_str::<Vec<T>>(&body) {
            return Ok(results);
        }

        if let Ok(error_response) = serde_json::from_str::<serde_json::Value>(&body) {
            if let Some(error_msg) = error_response.get("error") {
                return Err(anyhow!("Database update error: {}", error_msg));
            } else if let Some(message) = error_response.get("message") {
                return Err(anyhow!("Database update message: {}", message));
            }
            return Err(anyhow!(
                "Database update returned unexpected format: {}",
                body
            ));
        }

        Err(anyhow!(
            "Failed to parse database update response as JSON: {}",
            body
        ))
    }

    /// Deletes data from a table
    pub async fn delete(
        &mut self,
        filter_builder: impl FnOnce(&Postgrest) -> postgrest::Builder,
    ) -> Result<()> {
        let client = self.get_client()?;

        let builder = filter_builder(client);
        let response = builder.delete().execute().await?;

        let status = response.status();
        if !status.is_success() {
            let error_text = response.text().await?;
            return Err(anyhow!(
                "Delete operation failed: HTTP {} - {}",
                status,
                error_text
            ));
        }

        Ok(())
    }

    /// Fetches events for the given primary key ids. Returns an empty vector when `ids` is empty.
    pub async fn get_events_by_ids(&mut self, ids: &[i64]) -> Result<Vec<Event>> {
        if ids.is_empty() {
            return Ok(vec![]);
        }
        let id_strings: Vec<String> = ids.iter().map(|id| id.to_string()).collect();
        self.query(move |client| client.from("events").select("*").in_("id", id_strings))
            .await
    }

    /// Fetches artifacts for the given primary key ids. Returns an empty vector when `ids` is empty.
    pub async fn get_artifacts_by_ids(&mut self, ids: &[i64]) -> Result<Vec<Artifact>> {
        if ids.is_empty() {
            return Ok(vec![]);
        }
        let id_strings: Vec<String> = ids.iter().map(|id| id.to_string()).collect();
        self.query(move |client| client.from("artifacts").select("*").in_("id", id_strings))
            .await
    }

    /// Lists active accessories compatible with the product family for a product number.
    pub async fn get_product_compatibilities(
        &mut self,
        primary_product_number: &str,
        role: &str,
    ) -> Result<Vec<ProductCompatibility>> {
        #[derive(Deserialize)]
        struct PrimaryProduct {
            product_id: i64,
        }

        let primary_products: Vec<PrimaryProduct> = self
            .query(|client| {
                client
                    .from("product_numbers")
                    .select(
                        "product_id,products!product_numbers_product_id_fkey!inner(manufacturers!inner(id))",
                    )
                    .eq("product_number", primary_product_number)
                    .eq("lifecycle", "active")
                    .eq("products.lifecycle", "active")
                    .eq("products.manufacturers.lifecycle", "active")
                    .limit(1)
            })
            .await?;
        let Some(primary_product) = primary_products.into_iter().next() else {
            return Ok(Vec::new());
        };

        self.query(|client| {
            client
                .from("product_compatibilities")
                .select(
                    "*,product_numbers!product_compatibilities_accessory_product_number_fkey!inner(products!product_numbers_product_id_fkey!inner(manufacturers!inner(id)))",
                )
                .eq(
                    "primary_product_id",
                    primary_product.product_id.to_string(),
                )
                .eq("role", role)
                .eq("product_numbers.lifecycle", "active")
                .eq("product_numbers.products.lifecycle", "active")
                .eq(
                    "product_numbers.products.manufacturers.lifecycle",
                    "active",
                )
                .order("accessory_product_number")
        })
        .await
    }

    /// Resolves the accessory SKU approved for automatic part creation.
    pub async fn get_auto_assign_accessory_product_number(
        &mut self,
        primary_product_number: &str,
        role: &str,
    ) -> Result<Option<String>> {
        Ok(self
            .get_product_compatibilities(primary_product_number, role)
            .await?
            .into_iter()
            .find(|compatibility| compatibility.auto_assign)
            .map(|compatibility| compatibility.accessory_product_number))
    }

    /// Finds completed sessions whose start and end boundaries are within the tolerance.
    pub async fn match_sessions(
        &mut self,
        device_id: i64,
        timestamp_start: &str,
        timestamp_end: &str,
        tolerance_sec: u32,
    ) -> Result<Vec<Session>> {
        let tolerance_sec = i32::try_from(tolerance_sec)
            .map_err(|_| anyhow!("tolerance_sec exceeds the PostgreSQL integer range"))?;
        self.rpc(
            "match_sessions",
            &serde_json::json!({
                "device_id_caller": device_id,
                "timestamp_start_caller": timestamp_start,
                "timestamp_end_caller": timestamp_end,
                "tolerance_sec_caller": tolerance_sec
            }),
        )
        .await
    }

    /// Returns operator-attributed session time in seconds.
    pub async fn get_user_total_session_time(
        &mut self,
        filters: &UserSessionTimeFilters,
    ) -> Result<f64> {
        self.rpc(
            "get_user_total_session_time",
            &serde_json::json!({ "filters": filters }),
        )
        .await
    }

    /// Connectivity records for an artifact (session or device+time window). Uses RPC get_connectivity_for_artifact.
    pub async fn get_connectivity_for_artifact(
        &mut self,
        artifact_id: i64,
        max_elements: Option<i32>,
    ) -> Result<Vec<ConnectivityWithCoordinates>> {
        self.rpc(
            "get_connectivity_for_artifact",
            &serde_json::json!({
                "artifact_id_caller": artifact_id,
                "max_elements_caller": max_elements.unwrap_or(1000)
            }),
        )
        .await
    }

    /// Connectivity records for an event (device + time window around timestamp_observation). Uses RPC get_connectivity_for_event.
    pub async fn get_connectivity_for_event(
        &mut self,
        event_id: i64,
        max_elements: Option<i32>,
    ) -> Result<Vec<ConnectivityWithCoordinates>> {
        self.rpc(
            "get_connectivity_for_event",
            &serde_json::json!({
                "event_id_caller": event_id,
                "max_elements_caller": max_elements.unwrap_or(1000)
            }),
        )
        .await
    }

    /// Deletes a message from the named pgmq queue (calls public.ack_queue_message).
    /// Used by the GCP video-tagger after processing an embed or tag job.
    pub async fn ack_queue_message(&mut self, queue_name: &str, message_id: i64) -> Result<()> {
        self.rpc_execute(
            "ack_queue_message",
            &serde_json::json!({
                "queue_name": queue_name,
                "message_id": message_id
            }),
        )
        .await
    }
}

impl Drop for ScoutDbClient {
    fn drop(&mut self) {
        self.disconnect();
    }
}