scout_rs 0.95.95

use anyhow::{anyhow, Result};

use crate::constants::rpc;
use crate::db_client::{parse_postgrest_error_message, DatabaseConfig, ScoutDbClient};
pub use crate::db_client::{ConnectivityWithCoordinates, UserSessionTimeFilters};
use crate::models::*;

/// Response type for `get_tags_for_artifact` / `get_tags_for_event` RPCs (`tags_pretty_location`).
#[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
#[serde(rename_all = "snake_case")]
pub struct TagPrettyLocation {
    pub id: Option<i64>,
    pub inserted_at: Option<String>,
    pub artifact_id: Option<i64>,
    pub event_id: Option<i64>,
    pub model_id: Option<i64>,
    pub width: Option<f64>,
    pub height: Option<f64>,
    pub conf: Option<f64>,
    pub x: Option<f64>,
    pub y: Option<f64>,
    #[serde(rename = "class")]
    pub class: Option<String>,
    pub timestamp_observation: Option<String>,
    pub frame_index: Option<i64>,
    #[serde(default)]
    pub subject_id: Option<i64>,
    pub origin_latitude: Option<f64>,
    pub origin_longitude: Option<f64>,
    pub subject_latitude: Option<f64>,
    pub subject_longitude: Option<f64>,
}

// ===== CLIENT IMPLEMENTATION =====

#[derive(Debug)]
pub struct ScoutClient {
    pub config_db: DatabaseConfig,
    pub device: Option<DevicePrettyLocation>,
    pub herd: Option<Herd>,
    pub parts: Option<Vec<Part>>,
    db_client: Option<ScoutDbClient>,
    is_offline: bool,
}

impl ScoutClient {
    /// Creates a new ScoutClient instance.
    ///
    /// # Arguments
    /// * `api_key` - The API key for the Scout service.
    pub fn new(config_db: DatabaseConfig) -> Self {
        Self {
            config_db,
            device: None,
            herd: None,
            parts: None,
            db_client: None,
            is_offline: false,
        }
    }

    /// Creates a ScoutClient backed by a Supabase service-role key (bypasses RLS).
    /// Device/herd context is not resolved; call DB/CRUD methods directly (no `identify()` needed).
    pub fn with_service_key(
        rest_url: impl Into<String>,
        service_role_key: impl Into<String>,
    ) -> Self {
        Self::new(DatabaseConfig::service_role(rest_url, service_role_key))
    }

    /// True when this client authenticates with a service-role key.
    pub fn is_service_key(&self) -> bool {
        self.config_db.use_service_key && self.config_db.service_role_key.is_some()
    }

    /// Initializes the client in offline mode with default placeholder values
    /// This allows using the sync engine without database connectivity
    pub fn initialize_offline(&mut self) {
        self.is_offline = true;

        // Create default device with placeholder values
        let device = DevicePrettyLocation::default();
        self.device = Some(device);

        // Create default herd with placeholder values
        let herd = Herd::default();
        self.herd = Some(herd);

        self.parts = Some(Vec::new());
    }

    /// Identifies the device and herd, then establishes direct database connection
    /// If in offline mode, sets default values and returns Ok(())
    pub async fn identify(&mut self) -> Result<()> {
        // If already in offline mode, just ensure defaults are set
        if self.is_offline {
            if self.device.is_none() {
                self.device = Some(DevicePrettyLocation::default());
            }
            if self.herd.is_none() {
                self.herd = Some(Herd::default());
            }
            if self.parts.is_none() {
                self.parts = Some(Vec::new());
            }
            return Ok(());
        }

        let mut db_client = ScoutDbClient::new(self.config_db.clone());
        db_client.connect()?;

        self.db_client = Some(db_client);

        let device = self.get_device_from_db().await?;

        let herd = self.get_herd_from_db(device.herd_id).await?;

        let parts = match device.id {
            Some(device_id) => self.get_parts_from_db(device_id).await?,
            None => Vec::new(),
        };

        self.device = Some(device);
        self.herd = Some(herd);
        self.parts = Some(parts);

        Ok(())
    }

    /// Gets device information using get_device_by_api_key function and parsing JSON response
    async fn get_device_from_db(&mut self) -> Result<DevicePrettyLocation> {
        let config_db = self.config_db.clone();
        let api_key = config_db.get_scout_api_key();
        let db_client = self.get_db_client()?;

        let client = db_client.get_client()?;
        let response = client
            .rpc(
                "get_device_by_api_key",
                serde_json::json!({
                    "device_api_key": api_key
                })
                .to_string(),
            )
            .execute()
            .await?;

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

        if !status.is_success() {
            let msg = parse_postgrest_error_message(&body).unwrap_or_else(|| body.clone());
            return Err(anyhow!("Device lookup failed ({}): {}", status, msg));
        }

        if let Some(msg) = parse_postgrest_error_message(&body) {
            return Err(anyhow!("Device lookup failed: {}", msg));
        }

        let device_pretty: DevicePrettyLocation = serde_json::from_str(&body).map_err(|e| {
            anyhow!(
                "Failed to parse device response: {} - Response: {}",
                e,
                body
            )
        })?;

        Ok(device_pretty)
    }

    /// Gets herd information directly from database
    async fn get_herd_from_db(&mut self, herd_id: i64) -> Result<Herd> {
        let db_client = self.get_db_client()?;

        let results = db_client
            .query(|client| {
                client
                    .from("herds")
                    .select("*")
                    .eq("id", herd_id.to_string())
                    .limit(1)
            })
            .await?;

        if results.is_empty() {
            return Err(anyhow!("No herd found for ID: {}", herd_id));
        }

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

    /// Gets the parts owned by a device (one device has many parts).
    async fn get_parts_from_db(&mut self, device_id: i64) -> Result<Vec<Part>> {
        let db_client = self.get_db_client()?;

        db_client
            .query(|client| {
                client
                    .from("parts")
                    .select("*")
                    .eq("device_id", device_id.to_string())
                    .order("created_at.desc")
            })
            .await
    }

    /// Gets the database client, lazily establishing the connection when needed.
    /// This lets service-role clients issue queries without calling `identify()`, and gives
    /// callers a typed escape hatch for tables and functions Scout does not model itself.
    pub fn get_db_client(&mut self) -> Result<&mut ScoutDbClient> {
        if self.is_offline {
            return Err(anyhow!(
                "Database operations not available in offline mode. Call initialize_offline() to use offline mode."
            ));
        }
        if self.db_client.is_none() {
            let mut db_client = ScoutDbClient::new(self.config_db.clone());
            db_client.connect()?;
            self.db_client = Some(db_client);
        }
        self.db_client
            .as_mut()
            .ok_or_else(|| anyhow!("No PostgREST client available"))
    }

    /// Checks if the client has been identified and has a database connection
    /// Returns true if identified normally or if in offline mode
    pub fn is_identified(&self) -> bool {
        (self.db_client.is_some() && self.device.is_some() && self.herd.is_some())
            || self.is_offline
    }

    // ===== HELPER METHODS =====

    /// Checks if a session exists in the database by device_id, start timestamp, and end timestamp
    pub async fn does_session_exist(
        &mut self,
        device_id: i64,
        timestamp_start: &str,
    ) -> Result<bool> {
        let db_client = self.get_db_client()?;

        #[derive(Debug, serde::Deserialize)]
        #[allow(dead_code)]
        struct SessionIdOnly {
            id: i64,
        }

        let results: Vec<SessionIdOnly> = db_client
            .query(|client| {
                client
                    .from("sessions")
                    .select("id")
                    .eq("device_id", device_id.to_string())
                    .eq("timestamp_start", timestamp_start)
                    .limit(1)
            })
            .await?;

        Ok(!results.is_empty())
    }

    /// Convenience method to check if a session exists using a Session object
    pub async fn does_session_exist_from_session(&mut self, session: &Session) -> Result<bool> {
        self.does_session_exist(session.device_id, &session.timestamp_start)
            .await
    }

    /// 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>> {
        self.get_db_client()?
            .match_sessions(device_id, timestamp_start, timestamp_end, 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.get_db_client()?
            .get_user_total_session_time(filters)
            .await
    }

    /// Closest remote session for a completed local session, for deduplication before insert.
    pub async fn find_matching_session(
        &mut self,
        session: &Session,
        tolerance_sec: u32,
    ) -> Result<Option<Session>> {
        let timestamp_end = session
            .timestamp_end
            .as_deref()
            .ok_or_else(|| anyhow!("Cannot match a session without timestamp_end"))?;
        let matches = self
            .match_sessions(
                session.device_id,
                &session.timestamp_start,
                timestamp_end,
                tolerance_sec,
            )
            .await?;
        Ok(matches.into_iter().next())
    }

    /// Helper to create a success response
    fn success_response<T>(data: T) -> ResponseScout<T> {
        ResponseScout::new(ResponseScoutStatus::Success, Some(data))
    }

    /// Helper to create a failure response
    fn failure_response<T>() -> ResponseScout<T> {
        ResponseScout::new(ResponseScoutStatus::Failure, None)
    }

    /// Helper to handle database insert results
    fn handle_insert_result<T>(result: Vec<T>) -> Result<ResponseScout<T>> {
        if result.is_empty() {
            Ok(Self::failure_response())
        } else {
            Ok(Self::success_response(result.into_iter().next().unwrap()))
        }
    }

    /// Helper to handle database query results
    fn handle_query_result<T>(result: Vec<T>) -> ResponseScout<Vec<T>> {
        Self::success_response(result)
    }

    async fn get_rows_by<T>(
        &mut self,
        table: &str,
        column: &str,
        value: String,
        order: &str,
        options: &OperatingContextListOptions,
    ) -> Result<ResponseScout<Vec<T>>>
    where
        T: for<'de> serde::Deserialize<'de>,
    {
        let limit = options.limit.unwrap_or(100);
        if !(1..=500).contains(&limit) {
            return Err(anyhow!("limit must be between 1 and 500"));
        }
        let db_client = self.get_db_client()?;
        let results = db_client
            .query(|client| {
                let mut query = client
                    .from(table)
                    .select("*")
                    .eq(column, value)
                    .order(order)
                    .limit(limit);
                if !options.include_inactive {
                    query = query.eq("lifecycle", "active");
                }
                if let Some(after_id) = options.after_id {
                    query = query.gt("id", after_id.to_string());
                }
                query
            })
            .await?;
        Ok(Self::handle_query_result(results))
    }

    async fn get_row_by_id<T>(&mut self, table: &str, id: i64) -> Result<ResponseScout<T>>
    where
        T: for<'de> serde::Deserialize<'de>,
    {
        let db_client = self.get_db_client()?;
        let results = db_client
            .query(|client| {
                client
                    .from(table)
                    .select("*")
                    .eq("id", id.to_string())
                    .limit(1)
            })
            .await?;
        Self::handle_insert_result(results)
    }

    async fn create_row<T>(&mut self, table: &str, row: &T) -> Result<ResponseScout<T>>
    where
        T: for<'de> serde::Deserialize<'de> + serde::Serialize,
    {
        let db_client = self.get_db_client()?;
        Self::handle_insert_result(db_client.insert(table, row).await?)
    }

    async fn update_row<Input, Output>(
        &mut self,
        table: &str,
        id: i64,
        row: &Input,
    ) -> Result<ResponseScout<Output>>
    where
        Input: serde::Serialize,
        Output: for<'de> serde::Deserialize<'de>,
    {
        let db_client = self.get_db_client()?;
        let client = db_client.get_client()?;
        let value = serde_json::to_value(row)?;
        let fields = value
            .as_object()
            .ok_or_else(|| anyhow!("update payload must serialize as an object"))?;
        if fields.is_empty() {
            return Err(anyhow!("no fields to update"));
        }
        if fields.contains_key("lifecycle_reason") && !fields.contains_key("lifecycle") {
            return Err(anyhow!("lifecycle_reason requires lifecycle"));
        }
        let payload = value.to_string();
        let response = client
            .from(table)
            .eq("id", id.to_string())
            .update(&payload)
            .execute()
            .await?;
        let status = response.status();
        let body = response.text().await?;
        if !status.is_success() {
            let message = parse_postgrest_error_message(&body).unwrap_or(body);
            return Err(anyhow!("Database update failed ({}): {}", status, message));
        }
        let results: Vec<Output> = serde_json::from_str(&body)?;
        Self::handle_insert_result(results)
    }

    // ===== BACKWARD COMPATIBILITY METHODS =====

    /// Gets device information
    pub async fn get_device(&mut self) -> Result<ResponseScout<DevicePrettyLocation>> {
        if let Some(device) = &self.device {
            return Ok(ResponseScout::new(
                ResponseScoutStatus::Success,
                Some(device.clone()),
            ));
        }

        self.identify().await?;

        if let Some(device) = &self.device {
            Ok(ResponseScout::new(
                ResponseScoutStatus::Success,
                Some(device.clone()),
            ))
        } else {
            Ok(ResponseScout::new(ResponseScoutStatus::Failure, None))
        }
    }

    /// Gets the identified device's parts, identifying first if needed.
    pub async fn get_parts(&mut self) -> Result<ResponseScout<Vec<Part>>> {
        if let Some(parts) = &self.parts {
            return Ok(ResponseScout::new(
                ResponseScoutStatus::Success,
                Some(parts.clone()),
            ));
        }

        self.identify().await?;

        if let Some(parts) = &self.parts {
            Ok(ResponseScout::new(
                ResponseScoutStatus::Success,
                Some(parts.clone()),
            ))
        } else {
            Ok(ResponseScout::new(ResponseScoutStatus::Failure, None))
        }
    }

    /// Gets herd information (backward compatibility method)
    pub async fn get_herd(&mut self, herd_id: Option<i64>) -> Result<ResponseScout<Herd>> {
        let herd_id = if let Some(id) = herd_id {
            id
        } else if let Some(device) = &self.device {
            device.herd_id
        } else {
            return Err(anyhow!("No herd_id provided and no device data available"));
        };

        if let Some(herd) = &self.herd {
            if herd.id == Some(herd_id) {
                return Ok(ResponseScout::new(
                    ResponseScoutStatus::Success,
                    Some(herd.clone()),
                ));
            }
        }

        if self.device.is_none() {
            self.identify().await?;
        }

        if let Some(herd) = &self.herd {
            if herd.id == Some(herd_id) {
                Ok(ResponseScout::new(
                    ResponseScoutStatus::Success,
                    Some(herd.clone()),
                ))
            } else {
                Ok(ResponseScout::new(ResponseScoutStatus::Failure, None))
            }
        } else {
            Ok(ResponseScout::new(ResponseScoutStatus::Failure, None))
        }
    }

    // ===== DIRECT DATABASE OPERATIONS =====

    /// Creates an event directly in the database
    pub async fn create_event(&mut self, event: &Event) -> Result<ResponseScout<Event>> {
        let db_client = self.get_db_client()?;
        let result = db_client.insert("events", event).await?;
        Self::handle_insert_result(result)
    }

    /// Creates tags for an event directly in the database
    /// RLS policies and foreign key constraints handle validation automatically
    pub async fn create_tags(
        &mut self,
        event_id: i64,
        tags: &[Tag],
    ) -> Result<ResponseScout<Vec<Tag>>> {
        let db_client = self.get_db_client()?;

        if tags.is_empty() {
            return Ok(ResponseScout::new(
                ResponseScoutStatus::Success,
                Some(Vec::new()),
            ));
        }

        // Prepare tags with event_id for bulk insert
        let tags_with_event_id: Vec<Tag> = tags
            .iter()
            .map(|tag| {
                let mut tag_with_event_id = tag.clone();
                tag_with_event_id.update_event_id(event_id);
                tag_with_event_id
            })
            .collect();

        let result = db_client
            .insert_bulk_batched_by_shape("tags", &tags_with_event_id)
            .await?;
        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(result),
        ))
    }
}

/// Bulk-insert tags; groups rows by serialized JSON key set for PostgREST.
pub async fn push_tags_batched(db: &mut ScoutDbClient, tags: &[Tag]) -> Result<()> {
    if tags.is_empty() {
        return Ok(());
    }
    db.insert_bulk_batched_by_shape("tags", tags).await?;
    Ok(())
}

impl ScoutClient {
    /// Creates an event with tags (compatibility method)
    pub async fn create_event_with_tags(
        &mut self,
        event: &Event,
        tags: &[Tag],
        _file_path: Option<&str>,
    ) -> Result<ResponseScout<Event>> {
        let event_response = self.create_event(event).await?;

        if event_response.status != ResponseScoutStatus::Success {
            return Ok(event_response);
        }

        let created_event = event_response.data.unwrap();

        if !tags.is_empty() {
            let tags_response = self.create_tags(created_event.id.unwrap(), tags).await?;
            if tags_response.status != ResponseScoutStatus::Success {
                return Ok(ResponseScout::new(ResponseScoutStatus::Failure, None));
            }
        }

        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(created_event),
        ))
    }

    /// Creates a session directly in the database
    pub async fn create_session(&mut self, session: &Session) -> Result<ResponseScout<Session>> {
        let db_client = self.get_db_client()?;
        let result = db_client.insert("sessions", session).await?;
        Self::handle_insert_result(result)
    }

    /// Creates connectivity data directly from the database
    pub async fn create_connectivity(
        &mut self,
        connectivity: &Connectivity,
    ) -> Result<ResponseScout<Connectivity>> {
        let db_client = self.get_db_client()?;
        let result = db_client.insert("connectivity", connectivity).await?;
        Self::handle_insert_result(result)
    }

    /// Gets sessions for a herd directly from the database
    pub async fn get_sessions_by_herd(
        &mut self,
        herd_id: i64,
    ) -> Result<ResponseScout<Vec<Session>>> {
        let db_client = self.get_db_client()?;
        let results = db_client
            .query(|client| {
                client
                    .from("sessions")
                    .select("*, devices!inner(herd_id)")
                    .eq("devices.herd_id", herd_id.to_string())
                    .order("timestamp_start.desc")
            })
            .await?;
        Ok(Self::handle_query_result(results))
    }

    /// Gets plans for a herd directly from the database
    pub async fn get_plans_by_herd(&mut self, herd_id: i64) -> Result<ResponseScout<Vec<Plan>>> {
        let db_client = self.get_db_client()?;
        let results = db_client
            .query(|client| {
                client
                    .from("plans")
                    .eq("herd_id", herd_id.to_string())
                    .order("inserted_at.desc")
            })
            .await?;

        // Return empty results if no plans found (don't panic)
        Ok(Self::handle_query_result(results))
    }

    /// Gets a specific plan by ID directly from the database
    pub async fn get_plan_by_id(&mut self, plan_id: i64) -> Result<ResponseScout<Plan>> {
        let db_client = self.get_db_client()?;

        let results = db_client
            .query(|client| {
                client
                    .from("plans")
                    .select("*")
                    .eq("id", plan_id.to_string())
                    .limit(1)
            })
            .await?;

        // Return failure status if no plan found (don't panic)
        if results.is_empty() {
            return Ok(ResponseScout::new(ResponseScoutStatus::Failure, None));
        }

        if results.len() > 1 {
            panic!(
                "Multiple plans found with ID: {}. Expected exactly one plan.",
                plan_id
            );
        }

        let plan = results.into_iter().next().unwrap();
        Ok(ResponseScout::new(ResponseScoutStatus::Success, Some(plan)))
    }

    /// Creates a plan directly in the database
    pub async fn create_plan(&mut self, plan: &Plan) -> Result<ResponseScout<Plan>> {
        let db_client = self.get_db_client()?;

        // Create a plan for insertion without ID field
        let plan_for_insert = PlanInsert {
            id: None,          // Will be auto-generated by database
            id_local: None,    // Add if you are upserting to sync engine
            inserted_at: None, // Database will use default value
            name: plan.name.clone(),
            instructions: plan.instructions.clone(),
            herd_id: plan.herd_id,
            plan_type: plan.plan_type.clone(),
        };

        let result = db_client.insert("plans", &plan_for_insert).await?;

        // Convert PlanInsert results back to Plan with generated IDs
        let plans: Vec<Plan> = result
            .into_iter()
            .map(|p| Plan {
                id: Some(p.id.unwrap_or(0)), // Use generated ID or fallback to 0
                id_local: None,
                inserted_at: p.inserted_at,
                name: p.name,
                instructions: p.instructions,
                herd_id: p.herd_id,
                plan_type: p.plan_type,
            })
            .collect();

        Self::handle_insert_result(plans)
    }

    /// Updates a plan directly in the database
    pub async fn update_plan(&mut self, plan_id: i64, plan: &Plan) -> Result<ResponseScout<Plan>> {
        let db_client = self.get_db_client()?;

        let result = db_client
            .update(plan, |client| {
                client.from("plans").eq("id", plan_id.to_string())
            })
            .await?;

        if result.is_empty() {
            return Ok(ResponseScout::new(ResponseScoutStatus::Failure, None));
        }

        let updated_plan = result.into_iter().next().unwrap();
        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(updated_plan),
        ))
    }

    /// Deletes a plan directly from the database
    pub async fn delete_plan(&mut self, plan_id: i64) -> Result<ResponseScout<()>> {
        let db_client = self.get_db_client()?;

        db_client
            .delete(|client| client.from("plans").eq("id", plan_id.to_string()))
            .await?;

        Ok(ResponseScout::new(ResponseScoutStatus::Success, None))
    }

    /// Gets events for a session directly from the database
    pub async fn get_session_events(
        &mut self,
        session_id: i64,
    ) -> Result<ResponseScout<Vec<Event>>> {
        let db_client = self.get_db_client()?;
        let results = db_client
            .query(|client| {
                client
                    .from("events")
                    .eq("session_id", session_id.to_string())
                    .order("timestamp_observation.desc")
            })
            .await?;
        Ok(Self::handle_query_result(results))
    }

    /// Gets connectivity data for a session directly from the database
    pub async fn get_session_connectivity(
        &mut self,
        session_id: i64,
    ) -> Result<ResponseScout<Vec<Connectivity>>> {
        let db_client = self.get_db_client()?;
        let results = db_client
            .query(|client| {
                client
                    .from("connectivity")
                    .eq("session_id", session_id.to_string())
                    .order("timestamp_start.asc")
            })
            .await?;
        Ok(Self::handle_query_result(results))
    }

    /// Updates a session directly in the database
    pub async fn update_session(
        &mut self,
        session_id: i64,
        session: &Session,
    ) -> Result<ResponseScout<Session>> {
        let db_client = self.get_db_client()?;

        let result = db_client
            .update(session, |client| {
                client.from("sessions").eq("id", session_id.to_string())
            })
            .await?;

        if result.is_empty() {
            return Ok(ResponseScout::new(ResponseScoutStatus::Failure, None));
        }

        let updated_session = result.into_iter().next().unwrap();
        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(updated_session),
        ))
    }

    /// Deletes a session directly from the database
    /// Database cascade deletion handles dependent records automatically
    pub async fn delete_session(&mut self, session_id: i64) -> Result<ResponseScout<()>> {
        let db_client = self.get_db_client()?;

        let session_deleted = db_client
            .delete(|client| client.from("sessions").eq("id", session_id.to_string()))
            .await;

        if session_deleted.is_err() {
            return Ok(ResponseScout::new(ResponseScoutStatus::Failure, None));
        }

        Ok(ResponseScout::new(ResponseScoutStatus::Success, None))
    }

    /// Deletes an event directly from the database
    /// Database cascade deletion handles dependent records automatically
    pub async fn delete_event(&mut self, event_id: i64) -> Result<ResponseScout<()>> {
        let db_client = self.get_db_client()?;

        let _event_deleted = db_client
            .delete(|client| client.from("events").eq("id", event_id.to_string()))
            .await?;

        Ok(ResponseScout::new(ResponseScoutStatus::Success, None))
    }

    /// Deletes a tag directly from the database
    pub async fn delete_tag(&mut self, tag_id: i64) -> Result<ResponseScout<()>> {
        let db_client = self.get_db_client()?;

        db_client
            .delete(|client| client.from("tags").eq("id", tag_id.to_string()))
            .await?;

        Ok(ResponseScout::new(ResponseScoutStatus::Success, None))
    }

    /// Deletes connectivity data directly from the database
    pub async fn delete_connectivity(&mut self, connectivity_id: i64) -> Result<ResponseScout<()>> {
        let db_client = self.get_db_client()?;

        db_client
            .delete(|client| {
                client
                    .from("connectivity")
                    .eq("id", connectivity_id.to_string())
            })
            .await?;

        Ok(ResponseScout::new(ResponseScoutStatus::Success, None))
    }

    /// Creates a sensor pose directly in the database.
    pub async fn create_sensor_pose(
        &mut self,
        sensor_pose: &SensorPose,
    ) -> Result<ResponseScout<SensorPose>> {
        let db_client = self.get_db_client()?;
        let result = db_client.insert("sensor_poses", sensor_pose).await?;
        Self::handle_insert_result(result)
    }

    /// Gets sensor poses for a part directly from the database.
    pub async fn get_sensor_poses_by_part(
        &mut self,
        part_id: i64,
    ) -> Result<ResponseScout<Vec<SensorPose>>> {
        let db_client = self.get_db_client()?;
        let results = db_client
            .query(|client| {
                client
                    .from("sensor_poses")
                    .select("*")
                    .eq("part_id", part_id.to_string())
                    .order("timestamp_observation.asc")
            })
            .await?;
        Ok(Self::handle_query_result(results))
    }

    /// Deletes a sensor pose directly from the database.
    pub async fn delete_sensor_pose(&mut self, sensor_pose_id: i64) -> Result<ResponseScout<()>> {
        let db_client = self.get_db_client()?;
        db_client
            .delete(|client| {
                client
                    .from("sensor_poses")
                    .eq("id", sensor_pose_id.to_string())
            })
            .await?;
        Ok(ResponseScout::new(ResponseScoutStatus::Success, None))
    }

    /// Inserts multiple sensor poses in a batch.
    pub async fn create_sensor_pose_batch(
        &mut self,
        sensor_poses: &[SensorPose],
    ) -> Result<ResponseScout<Vec<SensorPose>>> {
        let db_client = self.get_db_client()?;
        if sensor_poses.is_empty() {
            return Ok(ResponseScout::new(
                ResponseScoutStatus::Success,
                Some(Vec::new()),
            ));
        }
        let result = db_client
            .insert_bulk_batched_by_shape("sensor_poses", sensor_poses)
            .await?;
        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(result),
        ))
    }

    /// Upserts multiple sensor poses in a batch (insert or update on conflict). Used by the sync engine.
    pub async fn upsert_sensor_pose_batch(
        &mut self,
        sensor_poses: &[SensorPose],
    ) -> Result<ResponseScout<Vec<SensorPose>>> {
        let db_client = self.get_db_client()?;
        if sensor_poses.is_empty() {
            return Ok(ResponseScout::new(
                ResponseScoutStatus::Success,
                Some(Vec::new()),
            ));
        }
        let result = db_client
            .upsert_bulk_batched_by_shape("sensor_poses", sensor_poses)
            .await?;
        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(result),
        ))
    }

    // ===== LOCALIZATIONS =====

    pub async fn get_localizations(&mut self) -> Result<ResponseScout<Vec<Localization>>> {
        let db_client = self.get_db_client()?;
        let results = db_client
            .query(|client| client.from("localizations").select("*").order("region.asc"))
            .await?;
        Ok(Self::handle_query_result(results))
    }

    pub async fn get_localization_by_id(
        &mut self,
        localization_id: i64,
    ) -> Result<ResponseScout<Localization>> {
        let db_client = self.get_db_client()?;
        let results = db_client
            .query(|client| {
                client
                    .from("localizations")
                    .select("*")
                    .eq("id", localization_id.to_string())
                    .limit(1)
            })
            .await?;
        if results.is_empty() {
            return Ok(ResponseScout::new(ResponseScoutStatus::Failure, None));
        }
        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(results.into_iter().next().unwrap()),
        ))
    }

    pub async fn create_localization(
        &mut self,
        localization: &Localization,
    ) -> Result<ResponseScout<Localization>> {
        let db_client = self.get_db_client()?;
        let result = db_client.insert("localizations", localization).await?;
        Self::handle_insert_result(result)
    }

    pub async fn update_localization(
        &mut self,
        localization_id: i64,
        localization: &Localization,
    ) -> Result<ResponseScout<Localization>> {
        let db_client = self.get_db_client()?;
        let result = db_client
            .update(localization, |client| {
                client
                    .from("localizations")
                    .eq("id", localization_id.to_string())
            })
            .await?;
        Self::handle_insert_result(result)
    }

    pub async fn delete_localization(&mut self, localization_id: i64) -> Result<ResponseScout<()>> {
        let db_client = self.get_db_client()?;
        db_client
            .delete(|client| {
                client
                    .from("localizations")
                    .eq("id", localization_id.to_string())
            })
            .await?;
        Ok(ResponseScout::new(ResponseScoutStatus::Success, None))
    }

    // ===== ISSUERS =====

    pub async fn get_issuers(&mut self) -> Result<ResponseScout<Vec<Issuer>>> {
        let db_client = self.get_db_client()?;
        let results = db_client
            .query(|client| client.from("issuers").select("*").order("name.asc"))
            .await?;
        Ok(Self::handle_query_result(results))
    }

    pub async fn get_issuer_by_id(&mut self, issuer_id: i64) -> Result<ResponseScout<Issuer>> {
        let db_client = self.get_db_client()?;
        let results = db_client
            .query(|client| {
                client
                    .from("issuers")
                    .select("*")
                    .eq("id", issuer_id.to_string())
                    .limit(1)
            })
            .await?;
        if results.is_empty() {
            return Ok(ResponseScout::new(ResponseScoutStatus::Failure, None));
        }
        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(results.into_iter().next().unwrap()),
        ))
    }

    pub async fn create_issuer(&mut self, issuer: &Issuer) -> Result<ResponseScout<Issuer>> {
        let db_client = self.get_db_client()?;
        let result = db_client.insert("issuers", issuer).await?;
        Self::handle_insert_result(result)
    }

    pub async fn update_issuer(
        &mut self,
        issuer_id: i64,
        issuer: &Issuer,
    ) -> Result<ResponseScout<Issuer>> {
        let db_client = self.get_db_client()?;
        let result = db_client
            .update(issuer, |client| {
                client.from("issuers").eq("id", issuer_id.to_string())
            })
            .await?;
        Self::handle_insert_result(result)
    }

    pub async fn delete_issuer(&mut self, issuer_id: i64) -> Result<ResponseScout<()>> {
        let db_client = self.get_db_client()?;
        db_client
            .delete(|client| client.from("issuers").eq("id", issuer_id.to_string()))
            .await?;
        Ok(ResponseScout::new(ResponseScoutStatus::Success, None))
    }

    // ===== DOCUMENT TEMPLATES =====

    pub async fn get_document_templates(&mut self) -> Result<ResponseScout<Vec<DocumentTemplate>>> {
        let db_client = self.get_db_client()?;
        let results = db_client
            .query(|client| {
                client
                    .from("document_templates")
                    .select("*")
                    .order("name.asc")
            })
            .await?;
        Ok(Self::handle_query_result(results))
    }

    pub async fn get_document_template_by_id(
        &mut self,
        template_id: i64,
    ) -> Result<ResponseScout<DocumentTemplate>> {
        let db_client = self.get_db_client()?;
        let results = db_client
            .query(|client| {
                client
                    .from("document_templates")
                    .select("*")
                    .eq("id", template_id.to_string())
                    .limit(1)
            })
            .await?;
        if results.is_empty() {
            return Ok(ResponseScout::new(ResponseScoutStatus::Failure, None));
        }
        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(results.into_iter().next().unwrap()),
        ))
    }

    pub async fn get_document_templates_by_system_tag(
        &mut self,
        system_tag: &str,
    ) -> Result<ResponseScout<Vec<DocumentTemplate>>> {
        let db_client = self.get_db_client()?;
        let results = db_client
            .query(|client| {
                client
                    .from("document_templates")
                    .select("*")
                    .eq("system_tag", system_tag)
                    .order("name.asc")
            })
            .await?;
        Ok(Self::handle_query_result(results))
    }

    pub async fn create_document_template(
        &mut self,
        template: &DocumentTemplate,
    ) -> Result<ResponseScout<DocumentTemplate>> {
        let db_client = self.get_db_client()?;
        let result = db_client.insert("document_templates", template).await?;
        Self::handle_insert_result(result)
    }

    pub async fn update_document_template(
        &mut self,
        template_id: i64,
        template: &DocumentTemplatePatch,
    ) -> Result<ResponseScout<DocumentTemplate>> {
        self.update_row("document_templates", template_id, template)
            .await
    }

    pub async fn delete_document_template(
        &mut self,
        template_id: i64,
    ) -> Result<ResponseScout<()>> {
        let db_client = self.get_db_client()?;
        db_client
            .delete(|client| {
                client
                    .from("document_templates")
                    .eq("id", template_id.to_string())
            })
            .await?;
        Ok(ResponseScout::new(ResponseScoutStatus::Success, None))
    }

    // ===== CALIBRATIONS =====

    pub async fn get_calibrations(&mut self) -> Result<ResponseScout<Vec<Calibration>>> {
        let db_client = self.get_db_client()?;
        let results = db_client
            .query(|client| client.from("calibrations").select("*").order("id.asc"))
            .await?;
        Ok(Self::handle_query_result(results))
    }

    pub async fn get_calibration_by_id(
        &mut self,
        calibration_id: i64,
    ) -> Result<ResponseScout<Calibration>> {
        let db_client = self.get_db_client()?;
        let results = db_client
            .query(|client| {
                client
                    .from("calibrations")
                    .select("*")
                    .eq("id", calibration_id.to_string())
                    .limit(1)
            })
            .await?;
        if results.is_empty() {
            return Ok(ResponseScout::new(ResponseScoutStatus::Failure, None));
        }
        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(results.into_iter().next().unwrap()),
        ))
    }

    pub async fn create_calibration(
        &mut self,
        calibration: &Calibration,
    ) -> Result<ResponseScout<Calibration>> {
        let db_client = self.get_db_client()?;
        let result = db_client.insert("calibrations", calibration).await?;
        Self::handle_insert_result(result)
    }

    pub async fn update_calibration(
        &mut self,
        calibration_id: i64,
        calibration: &Calibration,
    ) -> Result<ResponseScout<Calibration>> {
        let db_client = self.get_db_client()?;
        let result = db_client
            .update(calibration, |client| {
                client
                    .from("calibrations")
                    .eq("id", calibration_id.to_string())
            })
            .await?;
        Self::handle_insert_result(result)
    }

    pub async fn delete_calibration(&mut self, calibration_id: i64) -> Result<ResponseScout<()>> {
        let db_client = self.get_db_client()?;
        db_client
            .delete(|client| {
                client
                    .from("calibrations")
                    .eq("id", calibration_id.to_string())
            })
            .await?;
        Ok(ResponseScout::new(ResponseScoutStatus::Success, None))
    }

    // ===== DOCUMENTS =====

    pub async fn get_documents_by_herd(
        &mut self,
        herd_id: i64,
    ) -> Result<ResponseScout<Vec<Document>>> {
        let db_client = self.get_db_client()?;
        let results = db_client
            .query(|client| {
                client
                    .from("documents")
                    .select("*")
                    .eq("herd_id", herd_id.to_string())
                    .order("created_at.desc")
            })
            .await?;
        Ok(Self::handle_query_result(results))
    }

    pub async fn get_documents_by_device(
        &mut self,
        device_id: i64,
    ) -> Result<ResponseScout<Vec<Document>>> {
        let db_client = self.get_db_client()?;
        let results = db_client
            .query(|client| {
                client
                    .from("documents")
                    .select("*")
                    .eq("device_id", device_id.to_string())
                    .order("created_at.desc")
            })
            .await?;
        Ok(Self::handle_query_result(results))
    }

    pub async fn get_documents_by_product(
        &mut self,
        product_id: i64,
    ) -> Result<ResponseScout<Vec<Document>>> {
        let db_client = self.get_db_client()?;
        let results = db_client
            .query(|client| {
                client
                    .from("documents")
                    .select("*")
                    .eq("product_id", product_id.to_string())
                    .order("created_at.desc")
            })
            .await?;
        Ok(Self::handle_query_result(results))
    }

    pub async fn get_documents_by_operating_context(
        &mut self,
        operating_context_id: i64,
    ) -> Result<ResponseScout<Vec<Document>>> {
        let db_client = self.get_db_client()?;
        let results = db_client
            .query(|client| {
                client
                    .from("documents")
                    .select("*")
                    .eq("operating_context_id", operating_context_id.to_string())
                    .order("created_at.desc")
            })
            .await?;
        Ok(Self::handle_query_result(results))
    }

    pub async fn get_documents_by_user(
        &mut self,
        user_id: &str,
    ) -> Result<ResponseScout<Vec<Document>>> {
        let db_client = self.get_db_client()?;
        let uid = user_id.to_string();
        let results = db_client
            .query(|client| {
                client
                    .from("documents")
                    .select("*")
                    .eq("user_id", uid)
                    .order("created_at.desc")
            })
            .await?;
        Ok(Self::handle_query_result(results))
    }

    /// Status-only view of a user's documents; visible with `user_documents.status.read`.
    pub async fn get_user_document_statuses_by_user(
        &mut self,
        user_id: &str,
    ) -> Result<ResponseScout<Vec<UserDocumentStatus>>> {
        let db_client = self.get_db_client()?;
        let uid = user_id.to_string();
        let results = db_client
            .query(|client| {
                client
                    .from("user_document_statuses")
                    .select("*")
                    .eq("user_id", uid)
                    .order("created_at.desc")
            })
            .await?;
        Ok(Self::handle_query_result(results))
    }

    pub async fn get_document_by_id(
        &mut self,
        document_id: i64,
    ) -> Result<ResponseScout<Document>> {
        let db_client = self.get_db_client()?;
        let results = db_client
            .query(|client| {
                client
                    .from("documents")
                    .select("*")
                    .eq("id", document_id.to_string())
                    .limit(1)
            })
            .await?;
        if results.is_empty() {
            return Ok(ResponseScout::new(ResponseScoutStatus::Failure, None));
        }
        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(results.into_iter().next().unwrap()),
        ))
    }

    pub async fn create_document(
        &mut self,
        document: &Document,
    ) -> Result<ResponseScout<Document>> {
        let db_client = self.get_db_client()?;
        let result = db_client.insert("documents", document).await?;
        Self::handle_insert_result(result)
    }

    pub async fn update_document(
        &mut self,
        document_id: i64,
        document: &DocumentPatch,
    ) -> Result<ResponseScout<Document>> {
        self.update_row("documents", document_id, document).await
    }

    pub async fn delete_document(&mut self, document_id: i64) -> Result<ResponseScout<()>> {
        let db_client = self.get_db_client()?;
        db_client
            .delete(|client| client.from("documents").eq("id", document_id.to_string()))
            .await?;
        Ok(ResponseScout::new(ResponseScoutStatus::Success, None))
    }

    // ===== MAINTENANCE REQUESTS =====

    pub async fn get_maintenance_requests_by_device(
        &mut self,
        device_id: i64,
    ) -> Result<ResponseScout<Vec<MaintenanceRequest>>> {
        let db_client = self.get_db_client()?;
        let results = db_client
            .query(|client| {
                client
                    .from("maintenance_requests")
                    .select("*")
                    .eq("device_id", device_id.to_string())
                    .order("created_at.desc")
            })
            .await?;
        Ok(Self::handle_query_result(results))
    }

    pub async fn get_maintenance_requests_by_part(
        &mut self,
        part_id: i64,
    ) -> Result<ResponseScout<Vec<MaintenanceRequest>>> {
        let db_client = self.get_db_client()?;
        let results = db_client
            .query(|client| {
                client
                    .from("maintenance_requests")
                    .select("*")
                    .eq("part_id", part_id.to_string())
                    .order("created_at.desc")
            })
            .await?;
        Ok(Self::handle_query_result(results))
    }

    pub async fn get_maintenance_request_by_id(
        &mut self,
        maintenance_request_id: i64,
    ) -> Result<ResponseScout<MaintenanceRequest>> {
        let db_client = self.get_db_client()?;
        let results = db_client
            .query(|client| {
                client
                    .from("maintenance_requests")
                    .select("*")
                    .eq("id", maintenance_request_id.to_string())
                    .limit(1)
            })
            .await?;
        if results.is_empty() {
            return Ok(ResponseScout::new(ResponseScoutStatus::Failure, None));
        }
        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(results.into_iter().next().unwrap()),
        ))
    }

    pub async fn create_maintenance_request(
        &mut self,
        maintenance_request: &MaintenanceRequest,
    ) -> Result<ResponseScout<MaintenanceRequest>> {
        let db_client = self.get_db_client()?;
        let result = db_client
            .insert("maintenance_requests", maintenance_request)
            .await?;
        Self::handle_insert_result(result)
    }

    pub async fn update_maintenance_request(
        &mut self,
        maintenance_request_id: i64,
        maintenance_request: &MaintenanceRequest,
    ) -> Result<ResponseScout<MaintenanceRequest>> {
        let db_client = self.get_db_client()?;
        let result = db_client
            .update(maintenance_request, |client| {
                client
                    .from("maintenance_requests")
                    .eq("id", maintenance_request_id.to_string())
            })
            .await?;
        Self::handle_insert_result(result)
    }

    pub async fn patch_maintenance_request(
        &mut self,
        maintenance_request_id: i64,
        patch: &MaintenanceRequestPatch,
    ) -> Result<ResponseScout<MaintenanceRequest>> {
        self.update_row("maintenance_requests", maintenance_request_id, patch)
            .await
    }

    pub async fn delete_maintenance_request(
        &mut self,
        maintenance_request_id: i64,
    ) -> Result<ResponseScout<()>> {
        let db_client = self.get_db_client()?;
        db_client
            .delete(|client| {
                client
                    .from("maintenance_requests")
                    .eq("id", maintenance_request_id.to_string())
            })
            .await?;
        Ok(ResponseScout::new(ResponseScoutStatus::Success, None))
    }

    // ===== USER LAST ACTIVITY =====

    pub async fn get_user_last_activity_by_user_id(
        &mut self,
        user_id: &str,
    ) -> Result<ResponseScout<UserLastActivity>> {
        let db_client = self.get_db_client()?;
        let results = db_client
            .query(|client| {
                client
                    .from("user_last_activity")
                    .select("*")
                    .eq("user_id", user_id)
                    .limit(1)
            })
            .await?;
        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            results.into_iter().next(),
        ))
    }

    // ===== USER EARTHRANGER IDS =====

    pub async fn get_user_earthranger_ids_by_herd(
        &mut self,
        herd_id: i64,
    ) -> Result<ResponseScout<Vec<UserEarthrangerId>>> {
        let db_client = self.get_db_client()?;
        let results = db_client
            .query(|client| {
                client
                    .from("user_earthranger_ids")
                    .select("*")
                    .eq("herd_id", herd_id.to_string())
            })
            .await?;
        Ok(Self::handle_query_result(results))
    }

    /// Resolves the EarthRanger subject id for a user in a herd: the per-herd row, else
    /// `users.earthranger_id`. `None` when neither is set or the caller may not see it.
    pub async fn get_user_earthranger_id(
        &mut self,
        user_id: &str,
        herd_id: i64,
    ) -> Result<Option<String>> {
        self.get_db_client()?
            .rpc(
                rpc::USER_EARTHRANGER_ID,
                &serde_json::json!({ "p_user_id": user_id, "p_herd_id": herd_id }),
            )
            .await
    }

    // ===== PLATFORM DEFAULT ROLES =====

    /// Catalog of system roles seeded into new herds; readable by platform superadmins.
    pub async fn get_platform_default_roles(
        &mut self,
    ) -> Result<ResponseScout<Vec<PlatformDefaultRole>>> {
        let db_client = self.get_db_client()?;
        let results = db_client
            .query(|client| {
                client
                    .from("platform_default_roles")
                    .select("*")
                    .order("system_name.asc")
            })
            .await?;
        Ok(Self::handle_query_result(results))
    }

    // ===== SESSION INCIDENTS =====

    pub async fn get_session_incidents_by_session(
        &mut self,
        session_id: i64,
    ) -> Result<ResponseScout<Vec<SessionIncident>>> {
        let db_client = self.get_db_client()?;
        let results = db_client
            .query(|client| {
                client
                    .from("session_incidents")
                    .select("*")
                    .eq("session_id", session_id.to_string())
                    .order("created_at.desc")
            })
            .await?;
        Ok(Self::handle_query_result(results))
    }

    pub async fn get_session_incident_by_id(
        &mut self,
        session_incident_id: i64,
    ) -> Result<ResponseScout<SessionIncident>> {
        let db_client = self.get_db_client()?;
        let results = db_client
            .query(|client| {
                client
                    .from("session_incidents")
                    .select("*")
                    .eq("id", session_incident_id.to_string())
                    .limit(1)
            })
            .await?;
        if results.is_empty() {
            return Ok(ResponseScout::new(ResponseScoutStatus::Failure, None));
        }
        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(results.into_iter().next().unwrap()),
        ))
    }

    pub async fn create_session_incident(
        &mut self,
        session_incident: &SessionIncident,
    ) -> Result<ResponseScout<SessionIncident>> {
        let db_client = self.get_db_client()?;
        let result = db_client
            .insert("session_incidents", session_incident)
            .await?;
        Self::handle_insert_result(result)
    }

    pub async fn update_session_incident(
        &mut self,
        session_incident_id: i64,
        session_incident: &SessionIncident,
    ) -> Result<ResponseScout<SessionIncident>> {
        let db_client = self.get_db_client()?;
        let result = db_client
            .update(session_incident, |client| {
                client
                    .from("session_incidents")
                    .eq("id", session_incident_id.to_string())
            })
            .await?;
        Self::handle_insert_result(result)
    }

    pub async fn delete_session_incident(
        &mut self,
        session_incident_id: i64,
    ) -> Result<ResponseScout<()>> {
        let db_client = self.get_db_client()?;
        db_client
            .delete(|client| {
                client
                    .from("session_incidents")
                    .eq("id", session_incident_id.to_string())
            })
            .await?;
        Ok(ResponseScout::new(ResponseScoutStatus::Success, None))
    }

    // ===== CONTACTS =====

    pub async fn get_contacts_by_herd(
        &mut self,
        herd_id: i64,
    ) -> Result<ResponseScout<Vec<Contact>>> {
        let db_client = self.get_db_client()?;
        let results = db_client
            .query(|client| {
                client
                    .from("contacts")
                    .select("*")
                    .eq("herd_id", herd_id.to_string())
                    .order("name.asc")
            })
            .await?;
        Ok(Self::handle_query_result(results))
    }

    pub async fn get_contact_by_id(&mut self, contact_id: i64) -> Result<ResponseScout<Contact>> {
        let db_client = self.get_db_client()?;
        let results = db_client
            .query(|client| {
                client
                    .from("contacts")
                    .select("*")
                    .eq("id", contact_id.to_string())
                    .limit(1)
            })
            .await?;
        if results.is_empty() {
            return Ok(ResponseScout::new(ResponseScoutStatus::Failure, None));
        }
        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(results.into_iter().next().unwrap()),
        ))
    }

    pub async fn create_contact(&mut self, contact: &Contact) -> Result<ResponseScout<Contact>> {
        let db_client = self.get_db_client()?;
        let result = db_client.insert("contacts", contact).await?;
        Self::handle_insert_result(result)
    }

    pub async fn update_contact(
        &mut self,
        contact_id: i64,
        contact: &Contact,
    ) -> Result<ResponseScout<Contact>> {
        let db_client = self.get_db_client()?;
        let result = db_client
            .update(contact, |client| {
                client.from("contacts").eq("id", contact_id.to_string())
            })
            .await?;
        Self::handle_insert_result(result)
    }

    pub async fn delete_contact(&mut self, contact_id: i64) -> Result<ResponseScout<()>> {
        let db_client = self.get_db_client()?;
        db_client
            .delete(|client| client.from("contacts").eq("id", contact_id.to_string()))
            .await?;
        Ok(ResponseScout::new(ResponseScoutStatus::Success, None))
    }

    // ===== OPERATING CONTEXTS =====

    pub async fn get_document_conditions_by_localization(
        &mut self,
        localization_id: i64,
        options: &OperatingContextListOptions,
    ) -> Result<ResponseScout<Vec<DocumentCondition>>> {
        self.get_rows_by(
            "document_conditions",
            "localization_id",
            localization_id.to_string(),
            "id.asc",
            options,
        )
        .await
    }

    pub async fn create_document_condition(
        &mut self,
        row: &DocumentCondition,
    ) -> Result<ResponseScout<DocumentCondition>> {
        self.create_row("document_conditions", row).await
    }

    pub async fn update_document_condition(
        &mut self,
        id: i64,
        row: &DocumentConditionPatch,
    ) -> Result<ResponseScout<DocumentCondition>> {
        self.update_row("document_conditions", id, row).await
    }

    pub async fn get_operating_context_types_by_localization(
        &mut self,
        localization_id: i64,
        options: &OperatingContextListOptions,
    ) -> Result<ResponseScout<Vec<OperatingContextType>>> {
        self.get_rows_by(
            "operating_context_types",
            "localization_id",
            localization_id.to_string(),
            "id.asc",
            options,
        )
        .await
    }

    pub async fn create_operating_context_type(
        &mut self,
        row: &OperatingContextType,
    ) -> Result<ResponseScout<OperatingContextType>> {
        self.create_row("operating_context_types", row).await
    }

    pub async fn update_operating_context_type(
        &mut self,
        id: i64,
        row: &OperatingContextCatalogPatch,
    ) -> Result<ResponseScout<OperatingContextType>> {
        self.update_row("operating_context_types", id, row).await
    }

    pub async fn get_operating_context_roles_by_localization(
        &mut self,
        localization_id: i64,
        options: &OperatingContextListOptions,
    ) -> Result<ResponseScout<Vec<OperatingContextRole>>> {
        self.get_rows_by(
            "operating_context_roles",
            "localization_id",
            localization_id.to_string(),
            "id.asc",
            options,
        )
        .await
    }

    pub async fn create_operating_context_role(
        &mut self,
        row: &OperatingContextRole,
    ) -> Result<ResponseScout<OperatingContextRole>> {
        self.create_row("operating_context_roles", row).await
    }

    pub async fn update_operating_context_role(
        &mut self,
        id: i64,
        row: &OperatingContextCatalogPatch,
    ) -> Result<ResponseScout<OperatingContextRole>> {
        self.update_row("operating_context_roles", id, row).await
    }

    pub async fn get_operating_context_equipment_categories_by_localization(
        &mut self,
        localization_id: i64,
        options: &OperatingContextListOptions,
    ) -> Result<ResponseScout<Vec<OperatingContextEquipmentCategory>>> {
        self.get_rows_by(
            "operating_context_equipment_categories",
            "localization_id",
            localization_id.to_string(),
            "id.asc",
            options,
        )
        .await
    }

    pub async fn create_operating_context_equipment_category(
        &mut self,
        row: &OperatingContextEquipmentCategory,
    ) -> Result<ResponseScout<OperatingContextEquipmentCategory>> {
        self.create_row("operating_context_equipment_categories", row)
            .await
    }

    pub async fn update_operating_context_equipment_category(
        &mut self,
        id: i64,
        row: &OperatingContextCatalogPatch,
    ) -> Result<ResponseScout<OperatingContextEquipmentCategory>> {
        self.update_row("operating_context_equipment_categories", id, row)
            .await
    }

    pub async fn get_operating_context_equipment_items_by_category(
        &mut self,
        equipment_category_id: i64,
        options: &OperatingContextListOptions,
    ) -> Result<ResponseScout<Vec<OperatingContextEquipmentItem>>> {
        self.get_rows_by(
            "operating_context_equipment_items",
            "equipment_category_id",
            equipment_category_id.to_string(),
            "id.asc",
            options,
        )
        .await
    }

    pub async fn create_operating_context_equipment_item(
        &mut self,
        row: &OperatingContextEquipmentItem,
    ) -> Result<ResponseScout<OperatingContextEquipmentItem>> {
        self.create_row("operating_context_equipment_items", row)
            .await
    }

    pub async fn update_operating_context_equipment_item(
        &mut self,
        id: i64,
        row: &OperatingContextEquipmentItemPatch,
    ) -> Result<ResponseScout<OperatingContextEquipmentItem>> {
        self.update_row("operating_context_equipment_items", id, row)
            .await
    }

    pub async fn get_operating_contexts_by_herd(
        &mut self,
        herd_id: i64,
        options: &OperatingContextListOptions,
    ) -> Result<ResponseScout<Vec<OperatingContext>>> {
        self.get_rows_by(
            "operating_contexts",
            "herd_id",
            herd_id.to_string(),
            "id.asc",
            options,
        )
        .await
    }

    pub async fn get_operating_context_by_id(
        &mut self,
        id: i64,
    ) -> Result<ResponseScout<OperatingContext>> {
        self.get_row_by_id("operating_contexts", id).await
    }

    pub async fn create_operating_context(
        &mut self,
        row: &OperatingContext,
    ) -> Result<ResponseScout<OperatingContext>> {
        self.create_row("operating_contexts", row).await
    }

    pub async fn update_operating_context(
        &mut self,
        id: i64,
        row: &OperatingContextPatch,
    ) -> Result<ResponseScout<OperatingContext>> {
        self.update_row("operating_contexts", id, row).await
    }

    pub async fn get_operating_context_payloads_by_operating_context(
        &mut self,
        operating_context_id: i64,
        options: &OperatingContextListOptions,
    ) -> Result<ResponseScout<Vec<OperatingContextPayload>>> {
        self.get_rows_by(
            "operating_context_payloads",
            "operating_context_id",
            operating_context_id.to_string(),
            "id.asc",
            options,
        )
        .await
    }

    pub async fn create_operating_context_payload(
        &mut self,
        row: &OperatingContextPayload,
    ) -> Result<ResponseScout<OperatingContextPayload>> {
        self.create_row("operating_context_payloads", row).await
    }

    pub async fn update_operating_context_payload(
        &mut self,
        id: i64,
        row: &OperatingContextPayloadPatch,
    ) -> Result<ResponseScout<OperatingContextPayload>> {
        self.update_row("operating_context_payloads", id, row).await
    }

    pub async fn get_operating_context_role_assignments_by_operating_context(
        &mut self,
        operating_context_id: i64,
        options: &OperatingContextListOptions,
    ) -> Result<ResponseScout<Vec<OperatingContextRoleAssignment>>> {
        self.get_rows_by(
            "operating_context_role_assignments",
            "operating_context_id",
            operating_context_id.to_string(),
            "id.asc",
            options,
        )
        .await
    }

    pub async fn create_operating_context_role_assignment(
        &mut self,
        row: &OperatingContextRoleAssignment,
    ) -> Result<ResponseScout<OperatingContextRoleAssignment>> {
        self.create_row("operating_context_role_assignments", row)
            .await
    }

    pub async fn update_operating_context_role_assignment(
        &mut self,
        id: i64,
        row: &OperatingContextRoleAssignmentPatch,
    ) -> Result<ResponseScout<OperatingContextRoleAssignment>> {
        self.update_row("operating_context_role_assignments", id, row)
            .await
    }

    pub async fn get_operating_context_equipment_requirements_by_operating_context(
        &mut self,
        operating_context_id: i64,
        options: &OperatingContextListOptions,
    ) -> Result<ResponseScout<Vec<OperatingContextEquipmentRequirement>>> {
        self.get_rows_by(
            "operating_context_equipment_requirements",
            "operating_context_id",
            operating_context_id.to_string(),
            "id.asc",
            options,
        )
        .await
    }

    pub async fn create_operating_context_equipment_requirement(
        &mut self,
        row: &OperatingContextEquipmentRequirement,
    ) -> Result<ResponseScout<OperatingContextEquipmentRequirement>> {
        self.create_row("operating_context_equipment_requirements", row)
            .await
    }

    pub async fn update_operating_context_equipment_requirement(
        &mut self,
        id: i64,
        row: &OperatingContextEquipmentRequirementPatch,
    ) -> Result<ResponseScout<OperatingContextEquipmentRequirement>> {
        self.update_row("operating_context_equipment_requirements", id, row)
            .await
    }

    pub async fn get_operating_context_regions_by_operating_context(
        &mut self,
        operating_context_id: i64,
        options: &OperatingContextListOptions,
    ) -> Result<ResponseScout<Vec<OperatingContextRegion>>> {
        self.get_rows_by(
            "operating_context_regions",
            "operating_context_id",
            operating_context_id.to_string(),
            "id.asc",
            options,
        )
        .await
    }

    pub async fn create_operating_context_region(
        &mut self,
        row: &OperatingContextRegion,
    ) -> Result<ResponseScout<OperatingContextRegion>> {
        self.create_row("operating_context_regions", row).await
    }

    pub async fn update_operating_context_region(
        &mut self,
        id: i64,
        row: &OperatingContextRegionPatch,
    ) -> Result<ResponseScout<OperatingContextRegion>> {
        self.update_row("operating_context_regions", id, row).await
    }

    pub async fn get_operating_context_risks_by_operating_context(
        &mut self,
        operating_context_id: i64,
        options: &OperatingContextListOptions,
    ) -> Result<ResponseScout<Vec<OperatingContextRisk>>> {
        self.get_rows_by(
            "operating_context_risks",
            "operating_context_id",
            operating_context_id.to_string(),
            "id.asc",
            options,
        )
        .await
    }

    pub async fn create_operating_context_risk(
        &mut self,
        row: &OperatingContextRisk,
    ) -> Result<ResponseScout<OperatingContextRisk>> {
        self.create_row("operating_context_risks", row).await
    }

    pub async fn update_operating_context_risk(
        &mut self,
        id: i64,
        row: &OperatingContextRiskPatch,
    ) -> Result<ResponseScout<OperatingContextRisk>> {
        self.update_row("operating_context_risks", id, row).await
    }

    pub async fn get_operating_context_weather_snapshots(
        &mut self,
        operating_context_id: i64,
        limit: i32,
        cursor: Option<(&str, i64)>,
    ) -> Result<ResponseScout<Vec<OperatingContextWeatherSnapshot>>> {
        let db_client = self.get_db_client()?;
        let (cursor_collected_at, cursor_id) = match cursor {
            Some((collected_at, id)) => (
                Some(serde_json::Value::String(collected_at.to_string())),
                Some(serde_json::Value::Number(id.into())),
            ),
            None => (None, None),
        };
        let results = db_client
            .query(|client| {
                client.rpc(
                    "get_operating_context_weather_snapshots",
                    serde_json::json!({
                        "operating_context_id_caller": operating_context_id,
                        "limit_caller": limit,
                        "cursor_collected_at": cursor_collected_at,
                        "cursor_id": cursor_id
                    })
                    .to_string(),
                )
            })
            .await?;
        Ok(Self::handle_query_result(results))
    }

    // ===== ADDITIONAL OPERATIONS =====

    /// Gets all devices for a herd directly from the database
    pub async fn get_devices_by_herd(
        &mut self,
        herd_id: i64,
    ) -> Result<ResponseScout<Vec<Device>>> {
        let db_client = self.get_db_client()?;

        let results = db_client
            .query(|client| {
                client
                    .from("devices")
                    .eq("herd_id", herd_id.to_string())
                    .order("inserted_at.desc")
            })
            .await?;

        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(results),
        ))
    }

    /// Gets all providers for a herd
    pub async fn get_providers_by_herd(
        &mut self,
        herd_id: i64,
    ) -> Result<ResponseScout<Vec<Provider>>> {
        let db_client = self.get_db_client()?;

        let results = db_client
            .query(|client| {
                client
                    .from("providers")
                    .select("*")
                    .eq("herd_id", herd_id.to_string())
                    .order("created_at.desc")
            })
            .await?;

        Ok(Self::handle_query_result(results))
    }

    /// Gets a specific provider by ID
    pub async fn get_provider_by_id(
        &mut self,
        provider_id: i64,
    ) -> Result<ResponseScout<Provider>> {
        let db_client = self.get_db_client()?;

        let results = db_client
            .query(|client| {
                client
                    .from("providers")
                    .select("*")
                    .eq("id", provider_id.to_string())
                    .limit(1)
            })
            .await?;

        if results.is_empty() {
            return Ok(ResponseScout::new(ResponseScoutStatus::Failure, None));
        }

        let provider = results.into_iter().next().unwrap();
        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(provider),
        ))
    }

    /// Gets the catalog row for a specific product number.
    pub async fn get_product_number(
        &mut self,
        product_number: &str,
    ) -> Result<ResponseScout<ProductNumber>> {
        let db_client = self.get_db_client()?;
        let pn = product_number.to_string();

        let results: Vec<ProductNumber> = db_client
            .query(|client| {
                client
                    .from("product_numbers")
                    .select("*")
                    .eq("product_number", pn)
                    .limit(1)
            })
            .await?;

        if results.is_empty() {
            return Ok(ResponseScout::new(ResponseScoutStatus::Failure, None));
        }
        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(results.into_iter().next().unwrap()),
        ))
    }

    /// Gets the product line and manufacturer associated with a specific part
    /// (resolved via parts.product_number -> product_numbers -> products -> manufacturers).
    pub async fn get_product_and_manufacturer_for_part(
        &mut self,
        part: &Part,
    ) -> Result<ResponseScout<PartProductInfo>> {
        #[derive(serde::Deserialize)]
        struct ProductWithManufacturer {
            #[serde(flatten)]
            product: Product,
            manufacturers: Option<Manufacturer>,
        }
        #[derive(serde::Deserialize)]
        struct ProductNumberJoin {
            products: Option<ProductWithManufacturer>,
        }

        let db_client = self.get_db_client()?;
        let pn = part.product_number.clone();

        let results: Vec<ProductNumberJoin> = db_client
            .query(|client| {
                client
                    .from("product_numbers")
                    .select("products!product_numbers_product_id_fkey(*,manufacturers(*))")
                    .eq("product_number", pn)
                    .limit(1)
            })
            .await?;

        let info = match results.into_iter().next().and_then(|r| r.products) {
            Some(pwm) => PartProductInfo {
                product: Some(pwm.product),
                manufacturer: pwm.manufacturers,
            },
            None => PartProductInfo {
                product: None,
                manufacturer: None,
            },
        };
        Ok(ResponseScout::new(ResponseScoutStatus::Success, Some(info)))
    }

    /// Gets the product line and manufacturer for each active part of a device, in one query.
    /// Does not require `identify()`, so it works with service-role clients enumerating many devices.
    pub async fn get_product_and_manufacturer_for_device(
        &mut self,
        device_id: i64,
    ) -> Result<ResponseScout<Vec<PartProductInfo>>> {
        #[derive(serde::Deserialize)]
        struct ProductWithManufacturer {
            #[serde(flatten)]
            product: Product,
            manufacturers: Option<Manufacturer>,
        }
        #[derive(serde::Deserialize)]
        struct ProductNumberJoin {
            products: Option<ProductWithManufacturer>,
        }
        #[derive(serde::Deserialize)]
        struct PartJoin {
            product_numbers: Option<ProductNumberJoin>,
        }

        let db_client = self.get_db_client()?;
        let rows: Vec<PartJoin> = db_client
            .query(|client| {
                client
                    .from("parts")
                    .select(
                        "product_numbers!parts_product_number_fkey(products!product_numbers_product_id_fkey(*,manufacturers(*)))",
                    )
                    .eq("device_id", device_id.to_string())
                    .eq("lifecycle", "active")
            })
            .await?;

        let infos = rows
            .into_iter()
            .filter_map(|r| r.product_numbers.and_then(|pn| pn.products))
            .map(|pwm| PartProductInfo {
                product: Some(pwm.product),
                manufacturer: pwm.manufacturers,
            })
            .collect();

        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(infos),
        ))
    }

    /// Gets all devices that the current user/device has permission to view.
    /// RLS policies will automatically filter results based on permissions:
    /// - Devices can see themselves and other devices in the same herd
    /// - Users can see devices in herds they have view access to
    pub async fn get_all_devices(&mut self) -> Result<ResponseScout<Vec<Device>>> {
        let db_client = self.get_db_client()?;

        let results = db_client
            .query(|client| client.from("devices").order("inserted_at.desc"))
            .await?;

        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(results),
        ))
    }

    /// Gets devices in the same herd as the current device (peer devices).
    /// This method requires the client to be identified first.
    /// Returns an error if the device or herd is not set.
    pub async fn get_peer_devices(&mut self) -> Result<ResponseScout<Vec<Device>>> {
        // Check if client is identified with device info
        let device = self
            .device
            .as_ref()
            .ok_or_else(|| anyhow!("Device not identified - call identify() first"))?;

        // Check if device has a herd_id
        let herd_id = device.herd_id;
        if herd_id <= 0 {
            return Err(anyhow!(
                "Device herd_id is invalid - ensure device is properly configured"
            ));
        }

        // Get all devices in the same herd using existing method
        self.get_devices_by_herd(herd_id).await
    }

    /// Gets a specific event by ID directly from the database
    pub async fn get_event_by_id(&mut self, event_id: i64) -> Result<ResponseScout<Event>> {
        let db_client = self.get_db_client()?;

        let results = db_client
            .query(|client| {
                client
                    .from("events")
                    .select("*")
                    .eq("id", event_id.to_string())
                    .limit(1)
            })
            .await?;

        if results.is_empty() {
            return Ok(ResponseScout::new(ResponseScoutStatus::Failure, None));
        }

        let event = results.into_iter().next().unwrap();
        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(event),
        ))
    }

    /// Gets a specific device by ID directly from the database
    pub async fn get_device_by_id(&mut self, device_id: i64) -> Result<ResponseScout<Device>> {
        let db_client = self.get_db_client()?;

        let results = db_client
            .query(|client| {
                client
                    .from("devices")
                    .select("*")
                    .eq("id", device_id.to_string())
                    .limit(1)
            })
            .await?;

        if results.is_empty() {
            return Ok(ResponseScout::new(ResponseScoutStatus::Failure, None));
        }

        let device = results.into_iter().next().unwrap();
        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(device),
        ))
    }

    /// Gets a specific herd by ID directly from the database
    pub async fn get_herd_by_id(&mut self, herd_id: i64) -> Result<ResponseScout<Herd>> {
        let db_client = self.get_db_client()?;

        let results = db_client
            .query(|client| {
                client
                    .from("herds")
                    .select("*")
                    .eq("id", herd_id.to_string())
                    .limit(1)
            })
            .await?;

        if results.is_empty() {
            return Ok(ResponseScout::new(ResponseScoutStatus::Failure, None));
        }

        let device = results.into_iter().next().unwrap();
        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(device),
        ))
    }

    /// Gets media usage snapshots for a herd, newest first. Optional limit.
    pub async fn get_herd_media_usage(
        &mut self,
        herd_id: i64,
        limit: Option<u32>,
    ) -> Result<ResponseScout<Vec<HerdMediaUsage>>> {
        let db_client = self.get_db_client()?;
        let results = db_client
            .query(|client| {
                let q = client
                    .from("herd_media_usage")
                    .select("*")
                    .eq("herd_id", herd_id.to_string())
                    .order("inserted_at.desc");
                match limit {
                    Some(n) => q.limit(n as usize),
                    None => q,
                }
            })
            .await?;
        Ok(Self::handle_query_result(results))
    }

    pub async fn create_herd_media_usage(
        &mut self,
        usage: &HerdMediaUsage,
    ) -> Result<ResponseScout<HerdMediaUsage>> {
        let db_client = self.get_db_client()?;
        let result = db_client.insert("herd_media_usage", usage).await?;
        Self::handle_insert_result(result)
    }

    /// Gets all events for a device directly from the database
    pub async fn get_device_events(&mut self, device_id: i64) -> Result<ResponseScout<Vec<Event>>> {
        let db_client = self.get_db_client()?;

        let results = db_client
            .query(|client| {
                client
                    .from("events")
                    .eq("device_id", device_id.to_string())
                    .order("timestamp_observation.desc")
            })
            .await?;

        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(results),
        ))
    }

    /// Gets all parts for a device directly from the database (one-to-many).
    pub async fn get_device_parts(&mut self, device_id: i64) -> Result<ResponseScout<Vec<Part>>> {
        let parts = self.get_parts_from_db(device_id).await?;
        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(parts),
        ))
    }

    /// Gets events with tags for a device directly from the database
    pub async fn get_device_events_with_tags(
        &mut self,
        device_id: i64,
    ) -> Result<ResponseScout<Vec<Event>>> {
        let db_client = self.get_db_client()?;

        let results = db_client
            .query(|client| {
                client
                    .from("events")
                    .select("*, tags(*)")
                    .eq("device_id", device_id.to_string())
                    .order("timestamp_observation.desc")
            })
            .await?;

        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(results),
        ))
    }

    /// Gets events with tags for a device using the database function
    pub async fn get_device_events_with_tags_via_function(
        &mut self,
        device_id: i64,
        limit: i64,
    ) -> Result<ResponseScout<Vec<Event>>> {
        let db_client = self.get_db_client()?;

        let results = db_client
            .query(|client| {
                client.rpc(
                    "get_events_and_tags_for_device",
                    serde_json::json!({
                        "device_id_caller": device_id,
                        "limit_caller": limit
                    })
                    .to_string(),
                )
            })
            .await?;

        Ok(Self::handle_query_result(results))
    }

    /// Gets events within a time range directly from the database
    pub async fn get_events_in_timerange(
        &mut self,
        start_time: &str,
        end_time: &str,
    ) -> Result<ResponseScout<Vec<Event>>> {
        let db_client = self.get_db_client()?;

        let results = db_client
            .query(|client| {
                client
                    .from("events")
                    .gte("timestamp_observation", start_time)
                    .lte("timestamp_observation", end_time)
                    .order("timestamp_observation.desc")
            })
            .await?;

        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(results),
        ))
    }

    /// Gets events within a geographic area directly from the database
    pub async fn get_events_in_area(
        &mut self,
        min_lat: f64,
        max_lat: f64,
        min_lon: f64,
        max_lon: f64,
    ) -> Result<ResponseScout<Vec<Event>>> {
        let db_client = self.get_db_client()?;

        let results = db_client
            .query(|client| {
                client
                    .from("events")
                    .select("*")
                    .gte("latitude", min_lat.to_string())
                    .lte("latitude", max_lat.to_string())
                    .gte("longitude", min_lon.to_string())
                    .lte("longitude", max_lon.to_string())
                    .order("timestamp_observation.desc")
            })
            .await?;

        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(results),
        ))
    }

    /// Creates multiple events in a batch directly in the database
    pub async fn create_events_batch(
        &mut self,
        events: &[Event],
    ) -> Result<ResponseScout<Vec<Event>>> {
        let db_client = self.get_db_client()?;

        if events.is_empty() {
            return Ok(ResponseScout::new(
                ResponseScoutStatus::Success,
                Some(Vec::new()),
            ));
        }

        let result = db_client
            .insert_bulk_batched_by_shape("events", events)
            .await?;
        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(result),
        ))
    }

    /// Creates multiple sessions in a batch directly in the database
    pub async fn create_sessions_batch(
        &mut self,
        sessions: &[Session],
    ) -> Result<ResponseScout<Vec<Session>>> {
        let db_client = self.get_db_client()?;

        if sessions.is_empty() {
            return Ok(ResponseScout::new(
                ResponseScoutStatus::Success,
                Some(Vec::new()),
            ));
        }

        let result = db_client
            .insert_bulk_batched_by_shape("sessions", sessions)
            .await?;
        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(result),
        ))
    }

    /// Creates multiple connectivity entries in a batch directly in the database
    pub async fn create_connectivity_batch(
        &mut self,
        connectivity_entries: &[Connectivity],
    ) -> Result<ResponseScout<Vec<Connectivity>>> {
        let db_client = self.get_db_client()?;

        if connectivity_entries.is_empty() {
            return Ok(ResponseScout::new(
                ResponseScoutStatus::Success,
                Some(Vec::new()),
            ));
        }

        let result = db_client
            .insert_bulk_batched_by_shape("connectivity", connectivity_entries)
            .await?;
        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(result),
        ))
    }

    /// Upserts multiple sessions in a batch (insert or update on conflict)
    pub async fn upsert_sessions_batch(
        &mut self,
        sessions: &[Session],
    ) -> Result<ResponseScout<Vec<Session>>> {
        let db_client = self.get_db_client()?;

        if sessions.is_empty() {
            return Ok(ResponseScout::new(
                ResponseScoutStatus::Success,
                Some(Vec::new()),
            ));
        }

        let result = db_client
            .upsert_bulk_batched_by_shape("sessions", sessions)
            .await?;
        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(result),
        ))
    }

    /// Upserts multiple connectivity entries in a batch (insert or update on conflict)
    pub async fn upsert_connectivity_batch(
        &mut self,
        connectivity_entries: &[Connectivity],
    ) -> Result<ResponseScout<Vec<Connectivity>>> {
        let db_client = self.get_db_client()?;

        if connectivity_entries.is_empty() {
            return Ok(ResponseScout::new(
                ResponseScoutStatus::Success,
                Some(Vec::new()),
            ));
        }

        let result = db_client
            .upsert_bulk_batched_by_shape("connectivity", connectivity_entries)
            .await?;
        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(result),
        ))
    }

    /// Upserts multiple events in a batch (insert or update on conflict)
    pub async fn upsert_events_batch(
        &mut self,
        events: &[Event],
    ) -> Result<ResponseScout<Vec<Event>>> {
        let db_client = self.get_db_client()?;

        if events.is_empty() {
            return Ok(ResponseScout::new(
                ResponseScoutStatus::Success,
                Some(Vec::new()),
            ));
        }

        let result = db_client
            .upsert_bulk_batched_by_shape("events", events)
            .await?;
        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(result),
        ))
    }

    /// Upserts multiple tags in a batch (insert or update on conflict)
    pub async fn upsert_tags_batch(&mut self, tags: &[Tag]) -> Result<ResponseScout<Vec<Tag>>> {
        let db_client = self.get_db_client()?;

        if tags.is_empty() {
            return Ok(ResponseScout::new(
                ResponseScoutStatus::Success,
                Some(Vec::new()),
            ));
        }

        let result = db_client.upsert_bulk_batched_by_shape("tags", tags).await?;
        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(result),
        ))
    }

    /// Upserts multiple operators in a batch (insert or update on conflict)
    pub async fn upsert_operators_batch(
        &mut self,
        operators: &[data::v2::Operator],
    ) -> Result<ResponseScout<Vec<data::v2::Operator>>> {
        let db_client = self.get_db_client()?;

        if operators.is_empty() {
            return Ok(ResponseScout::new(
                ResponseScoutStatus::Success,
                Some(Vec::new()),
            ));
        }

        let result = db_client
            .upsert_bulk_batched_by_shape("operators", operators)
            .await?;
        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(result),
        ))
    }

    /// Updates an event directly in the database
    pub async fn update_event(
        &mut self,
        event_id: i64,
        event: &Event,
    ) -> Result<ResponseScout<Event>> {
        let db_client = self.get_db_client()?;

        let result = db_client
            .update(event, |client| {
                client.from("events").eq("id", event_id.to_string())
            })
            .await?;

        if result.is_empty() {
            return Ok(ResponseScout::new(ResponseScoutStatus::Failure, None));
        }

        let updated_event = result.into_iter().next().unwrap();
        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(updated_event),
        ))
    }

    /// Updates connectivity data directly in the database
    pub async fn update_connectivity(
        &mut self,
        connectivity_id: i64,
        connectivity: &Connectivity,
    ) -> Result<ResponseScout<Connectivity>> {
        let db_client = self.get_db_client()?;

        let result = db_client
            .update(connectivity, |client| {
                client
                    .from("connectivity")
                    .eq("id", connectivity_id.to_string())
            })
            .await?;

        if result.is_empty() {
            return Err(anyhow!(
                "Failed to update connectivity entry - no data returned"
            ));
        }

        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(result.into_iter().next().unwrap()),
        ))
    }

    /// Gets connectivity data with coordinates directly from the database
    pub async fn get_connectivity_with_coordinates(
        &mut self,
        session_id: i64,
    ) -> Result<ResponseScout<Vec<Connectivity>>> {
        let db_client = self.get_db_client()?;

        let results = db_client
            .query(|client| {
                client
                    .from("connectivity")
                    .eq("session_id", session_id.to_string())
                    .order("timestamp_start.asc")
            })
            .await?;

        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(results),
        ))
    }

    /// Gets connectivity records for an artifact (uses session when set, else device + time window).
    pub async fn get_connectivity_for_artifact(
        &mut self,
        artifact_id: i64,
        max_elements: Option<i32>,
    ) -> Result<ResponseScout<Vec<ConnectivityWithCoordinates>>> {
        let db_client = self.get_db_client()?;
        let client = db_client.get_client()?;
        let max = max_elements.unwrap_or(1000);

        let response = client
            .rpc(
                "get_connectivity_for_artifact",
                serde_json::json!({
                    "artifact_id_caller": artifact_id,
                    "max_elements_caller": max
                })
                .to_string(),
            )
            .execute()
            .await?;

        let status = response.status();
        let body = response.text().await?;
        if !status.is_success() {
            return Ok(ResponseScout::new(
                ResponseScoutStatus::Failure,
                Some(serde_json::from_str(&body).unwrap_or_else(|_| vec![])),
            ));
        }

        let results: Vec<ConnectivityWithCoordinates> =
            serde_json::from_str(&body).map_err(|e| {
                anyhow!(
                    "Failed to parse get_connectivity_for_artifact response: {} - {}",
                    e,
                    body
                )
            })?;

        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(results),
        ))
    }

    /// Gets artifact IDs for a session (same device, overlapping session interval).
    pub async fn get_artifact_ids_for_session(
        &mut self,
        session_id: i64,
    ) -> Result<ResponseScout<Vec<i64>>> {
        let db_client = self.get_db_client()?;
        let client = db_client.get_client()?;

        let response = client
            .rpc(
                "get_artifact_ids_for_session",
                serde_json::json!({ "session_id_caller": session_id }).to_string(),
            )
            .execute()
            .await?;

        let status = response.status();
        let body = response.text().await?;
        if !status.is_success() {
            return Ok(ResponseScout::new(
                ResponseScoutStatus::Failure,
                Some(vec![]),
            ));
        }

        let results: Vec<i64> = serde_json::from_str(&body).map_err(|e| {
            anyhow!(
                "Failed to parse get_artifact_ids_for_session response: {} - {}",
                e,
                body
            )
        })?;

        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(results),
        ))
    }

    /// Gets tags for an artifact with optional timestamp range and max items.
    pub async fn get_tags_for_artifact(
        &mut self,
        artifact_id: i64,
        start_timestamp: Option<&str>,
        end_timestamp: Option<&str>,
        max_elements: Option<i32>,
    ) -> Result<ResponseScout<Vec<TagPrettyLocation>>> {
        let db_client = self.get_db_client()?;
        let client = db_client.get_client()?;
        let max = max_elements.unwrap_or(1000);

        let mut params = serde_json::json!({
            "artifact_id_caller": artifact_id,
            "max_elements_caller": max
        });
        if let Some(s) = start_timestamp {
            params["start_timestamp_caller"] = serde_json::Value::String(s.to_string());
        }
        if let Some(e) = end_timestamp {
            params["end_timestamp_caller"] = serde_json::Value::String(e.to_string());
        }

        let response = client
            .rpc("get_tags_for_artifact", params.to_string())
            .execute()
            .await?;

        let status = response.status();
        let body = response.text().await?;
        if !status.is_success() {
            return Ok(ResponseScout::new(
                ResponseScoutStatus::Failure,
                Some(serde_json::from_str(&body).unwrap_or_else(|_| vec![])),
            ));
        }

        let results: Vec<TagPrettyLocation> = serde_json::from_str(&body).map_err(|e| {
            anyhow!(
                "Failed to parse get_tags_for_artifact response: {} - {}",
                e,
                body
            )
        })?;

        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(results),
        ))
    }

    /// Gets tags for an event with optional timestamp range and max items (same RPC shape as [`Self::get_tags_for_artifact`]).
    pub async fn get_tags_for_event(
        &mut self,
        event_id: i64,
        start_timestamp: Option<&str>,
        end_timestamp: Option<&str>,
        max_elements: Option<i32>,
    ) -> Result<ResponseScout<Vec<TagPrettyLocation>>> {
        let db_client = self.get_db_client()?;
        let client = db_client.get_client()?;
        let max = max_elements.unwrap_or(1000);

        let mut params = serde_json::json!({
            "event_id_caller": event_id,
            "max_elements_caller": max
        });
        if let Some(s) = start_timestamp {
            params["start_timestamp_caller"] = serde_json::Value::String(s.to_string());
        }
        if let Some(e) = end_timestamp {
            params["end_timestamp_caller"] = serde_json::Value::String(e.to_string());
        }

        let response = client
            .rpc("get_tags_for_event", params.to_string())
            .execute()
            .await?;

        let status = response.status();
        let body = response.text().await?;
        if !status.is_success() {
            return Ok(ResponseScout::new(
                ResponseScoutStatus::Failure,
                Some(serde_json::from_str(&body).unwrap_or_else(|_| vec![])),
            ));
        }

        let results: Vec<TagPrettyLocation> = serde_json::from_str(&body).map_err(|e| {
            anyhow!(
                "Failed to parse get_tags_for_event response: {} - {}",
                e,
                body
            )
        })?;

        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(results),
        ))
    }

    /// Ends a session by updating its timestamp_end directly in the database
    pub async fn end_session(
        &mut self,
        session_id: i64,
        timestamp_end: u64,
    ) -> Result<ResponseScout<()>> {
        let mut session = match self.get_session_statistics(session_id).await?.data {
            Some(session) => session,
            None => return Ok(ResponseScout::new(ResponseScoutStatus::Failure, None)),
        };

        session.end(
            timestamp_end,
            SessionStatus::Finalized,
            SessionEndReason::Completed,
        );

        let response = self.update_session(session_id, &session).await?;
        if response.status == ResponseScoutStatus::Success {
            Ok(ResponseScout::new(ResponseScoutStatus::Success, None))
        } else {
            Ok(ResponseScout::new(ResponseScoutStatus::Failure, None))
        }
    }

    /// Gets statistics for a session directly from the database
    pub async fn get_session_statistics(
        &mut self,
        session_id: i64,
    ) -> Result<ResponseScout<Session>> {
        let db_client = self.get_db_client()?;

        let results = db_client
            .query(|client| {
                client
                    .from("sessions")
                    .select("*")
                    .eq("id", session_id.to_string())
                    .limit(1)
            })
            .await?;

        if results.is_empty() {
            return Ok(ResponseScout::new(ResponseScoutStatus::Failure, None));
        }

        let session = results.into_iter().next().unwrap();
        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(session),
        ))
    }

    // ===== COMPATIBILITY METHODS =====

    /// Compatibility method for post_events_batch
    pub async fn post_events_batch(
        &mut self,
        events_and_files: &[(Event, Vec<Tag>, String)],
        _batch_size: usize,
    ) -> Result<ResponseScout<Vec<Event>>> {
        let mut created_events = Vec::new();

        for (event, tags, _file_path) in events_and_files {
            let event_response = self.create_event(event).await?;
            if event_response.status != ResponseScoutStatus::Success {
                return Ok(ResponseScout::new(ResponseScoutStatus::Failure, None));
            }

            let created_event = event_response.data.unwrap();
            if !tags.is_empty() {
                let tags_response = self.create_tags(created_event.id.unwrap(), tags).await?;
                if tags_response.status != ResponseScoutStatus::Success {
                    return Ok(ResponseScout::new(ResponseScoutStatus::Failure, None));
                }
            }
            created_events.push(created_event);
        }

        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(created_events),
        ))
    }

    /// Gets zones and actions for a herd directly from the database
    pub async fn get_zones_and_actions_by_herd(
        &mut self,
        herd_id: i64,
        limit: i64,
        offset: i64,
    ) -> Result<ResponseScout<Vec<Zone>>> {
        let db_client = self.get_db_client()?;

        let results = db_client
            .query(|client| {
                client
                    .from("zones_and_actions")
                    .eq("herd_id", herd_id.to_string())
                    .order("inserted_at.desc")
                    .range(offset as usize, (offset + limit - 1) as usize)
            })
            .await?;

        Ok(Self::handle_query_result(results))
    }

    // ===== ARTIFACT OPERATIONS =====

    /// Creates an artifact directly in the database
    pub async fn create_artifact(
        &mut self,
        artifact: &Artifact,
    ) -> Result<ResponseScout<Artifact>> {
        let db_client = self.get_db_client()?;
        let result = db_client.insert("artifacts", artifact).await?;
        Self::handle_insert_result(result)
    }

    /// Gets artifacts for a session directly from the database
    pub async fn get_artifacts_by_session(
        &mut self,
        session_id: i64,
    ) -> Result<ResponseScout<Vec<Artifact>>> {
        let db_client = self.get_db_client()?;

        let results = db_client
            .query(|client| {
                client
                    .from("artifacts")
                    .eq("session_id", session_id.to_string())
                    .order("created_at.desc")
            })
            .await?;

        Ok(Self::handle_query_result(results))
    }

    /// Gets all artifacts for a herd (via sessions) directly from the database
    pub async fn get_artifacts_by_herd(
        &mut self,
        herd_id: i64,
    ) -> Result<ResponseScout<Vec<Artifact>>> {
        let db_client = self.get_db_client()?;

        let client = db_client.get_client()?;
        let response = client
            .rpc(
                "get_artifacts_for_herd",
                serde_json::json!({
                    "herd_id_caller": herd_id,
                    "limit_caller": 1000,
                    "offset_caller": 0
                })
                .to_string(),
            )
            .execute()
            .await?;

        let body = response.text().await?;
        let results: Vec<Artifact> = serde_json::from_str(&body).map_err(|e| {
            anyhow!(
                "Failed to parse artifacts response: {} - Response: {}",
                e,
                body
            )
        })?;

        Ok(Self::handle_query_result(results))
    }

    /// Updates an artifact directly in the database
    pub async fn update_artifact(
        &mut self,
        artifact_id: i64,
        artifact: &Artifact,
    ) -> Result<ResponseScout<Artifact>> {
        let db_client = self.get_db_client()?;

        let result = db_client
            .update(artifact, |client| {
                client.from("artifacts").eq("id", artifact_id.to_string())
            })
            .await?;

        if result.is_empty() {
            return Ok(ResponseScout::new(ResponseScoutStatus::Failure, None));
        }

        let updated_artifact = result.into_iter().next().unwrap();
        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(updated_artifact),
        ))
    }

    /// Deletes an artifact directly from the database
    pub async fn delete_artifact(&mut self, artifact_id: i64) -> Result<ResponseScout<()>> {
        let db_client = self.get_db_client()?;

        db_client
            .delete(|client| client.from("artifacts").eq("id", artifact_id.to_string()))
            .await?;

        Ok(ResponseScout::new(ResponseScoutStatus::Success, None))
    }

    /// Creates multiple artifacts in a batch directly in the database
    pub async fn create_artifacts_batch(
        &mut self,
        artifacts: &[Artifact],
    ) -> Result<ResponseScout<Vec<Artifact>>> {
        let db_client = self.get_db_client()?;

        if artifacts.is_empty() {
            return Ok(ResponseScout::new(
                ResponseScoutStatus::Success,
                Some(Vec::new()),
            ));
        }

        let result = db_client
            .insert_bulk_batched_by_shape("artifacts", artifacts)
            .await?;
        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(result),
        ))
    }

    /// Upserts multiple artifacts in a batch (insert or update on conflict)
    pub async fn upsert_artifacts_batch(
        &mut self,
        artifacts: &[Artifact],
    ) -> Result<ResponseScout<Vec<Artifact>>> {
        let db_client = self.get_db_client()?;

        if artifacts.is_empty() {
            return Ok(ResponseScout::new(
                ResponseScoutStatus::Success,
                Some(Vec::new()),
            ));
        }

        let result = db_client
            .upsert_bulk_batched_by_shape("artifacts", artifacts)
            .await?;
        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(result),
        ))
    }

    /// Inserts Vertex multimodal 001 embeddings into the dedicated embeddings table.
    /// Each row must have either event_id or artifact_id set; optional origin/sensor metadata can be included.
    pub async fn create_embeddings(
        &mut self,
        embeddings: &[EmbeddingVertexMultimodal001],
    ) -> Result<ResponseScout<Vec<EmbeddingVertexMultimodal001>>> {
        let db_client = self.get_db_client()?;

        if embeddings.is_empty() {
            return Ok(ResponseScout::new(
                ResponseScoutStatus::Success,
                Some(Vec::new()),
            ));
        }

        let result = db_client
            .insert_bulk_batched_by_shape("embeddings_vertex_multimodal_001", embeddings)
            .await?;
        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(result),
        ))
    }

    /// Inserts SAM3 segmentation rows into `segmentations_sam3`.
    /// Each row must have either `event_id` or `artifact_id` set; `mask_rle` is required (COCO-style RLE JSON).
    pub async fn create_segmentations_sam3(
        &mut self,
        rows: &[SegmentationSam3],
    ) -> Result<ResponseScout<Vec<SegmentationSam3>>> {
        let db_client = self.get_db_client()?;

        if rows.is_empty() {
            return Ok(ResponseScout::new(
                ResponseScoutStatus::Success,
                Some(Vec::new()),
            ));
        }

        let result = db_client
            .insert_bulk_batched_by_shape("segmentations_sam3", rows)
            .await?;
        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(result),
        ))
    }

    /// Creates a heartbeat record for a device
    pub async fn create_heartbeat(
        &mut self,
        heartbeat: &Heartbeat,
    ) -> Result<ResponseScout<Heartbeat>> {
        let db_client = self.get_db_client()?;
        let result = db_client.insert("heartbeats", heartbeat).await?;
        Self::handle_insert_result(result)
    }

    /// Gets all heartbeats for a specific device
    pub async fn get_heartbeats_by_device(
        &mut self,
        device_id: i64,
    ) -> Result<ResponseScout<Vec<Heartbeat>>> {
        let db_client = self.get_db_client()?;

        let results = db_client
            .query(|client| {
                client
                    .from("heartbeats")
                    .select("*")
                    .eq("device_id", device_id.to_string())
                    .order("timestamp.desc")
            })
            .await?;

        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(results),
        ))
    }

    /// Deletes a heartbeat record by ID
    ///
    /// **Note:** This method is primarily intended for testing and cleanup purposes.
    /// In production, heartbeats are typically append-only for audit trail purposes.
    pub async fn delete_heartbeat(&mut self, heartbeat_id: i64) -> Result<ResponseScout<()>> {
        let db_client = self.get_db_client()?;

        let _heartbeat_deleted = db_client
            .delete(|client| client.from("heartbeats").eq("id", heartbeat_id.to_string()))
            .await?;

        Ok(ResponseScout::new(ResponseScoutStatus::Success, None))
    }

    // ===== HEALTH METRICS =====

    /// Creates a single health metric row (device_id, timestamp, metric_name, value, optional source/unit).
    pub async fn create_health_metric(
        &mut self,
        metric: &HealthMetric,
    ) -> Result<ResponseScout<HealthMetric>> {
        let db_client = self.get_db_client()?;
        let result = db_client.insert("health_metrics", metric).await?;
        Self::handle_insert_result(result)
    }

    /// Creates multiple health metrics in a batch.
    pub async fn create_health_metrics_batch(
        &mut self,
        metrics: &[HealthMetric],
    ) -> Result<ResponseScout<Vec<HealthMetric>>> {
        let db_client = self.get_db_client()?;
        if metrics.is_empty() {
            return Ok(ResponseScout::new(
                ResponseScoutStatus::Success,
                Some(Vec::new()),
            ));
        }
        let result = db_client
            .insert_bulk_batched_by_shape("health_metrics", metrics)
            .await?;
        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(result),
        ))
    }

    /// Gets health metrics for a device, newest first. Optional limit.
    pub async fn get_health_metrics(
        &mut self,
        device_id: i64,
        limit: Option<u32>,
    ) -> Result<ResponseScout<Vec<HealthMetric>>> {
        let db_client = self.get_db_client()?;
        let results = db_client
            .query(|client| {
                let q = client
                    .from("health_metrics")
                    .select("*")
                    .eq("device_id", device_id.to_string())
                    .order("timestamp.desc");
                match limit {
                    Some(n) => q.limit(n as usize),
                    None => q,
                }
            })
            .await?;
        Ok(Self::handle_query_result(results))
    }

    /// Updates a health metric by id (partial update; only non-None fields applied).
    pub async fn update_health_metric(
        &mut self,
        id: i64,
        metric: &HealthMetric,
    ) -> Result<ResponseScout<HealthMetric>> {
        let db_client = self.get_db_client()?;
        let result = db_client
            .update(metric, |client| {
                client.from("health_metrics").eq("id", id.to_string())
            })
            .await?;
        if result.is_empty() {
            return Ok(ResponseScout::new(ResponseScoutStatus::Failure, None));
        }
        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            result.into_iter().next(),
        ))
    }

    /// Deletes a health metric by id.
    pub async fn delete_health_metric(&mut self, id: i64) -> Result<ResponseScout<()>> {
        let db_client = self.get_db_client()?;
        db_client
            .delete(|client| client.from("health_metrics").eq("id", id.to_string()))
            .await?;
        Ok(ResponseScout::new(ResponseScoutStatus::Success, None))
    }

    // ===== ANALYSIS USAGE =====

    pub async fn create_analysis_job(
        &mut self,
        job: &AnalysisJob,
    ) -> Result<ResponseScout<AnalysisJob>> {
        let db_client = self.get_db_client()?;
        let result = db_client.insert("analysis_jobs", job).await?;
        Self::handle_insert_result(result)
    }

    pub async fn create_analysis_jobs_batch(
        &mut self,
        jobs: &[AnalysisJob],
    ) -> Result<ResponseScout<Vec<AnalysisJob>>> {
        let db_client = self.get_db_client()?;
        if jobs.is_empty() {
            return Ok(ResponseScout::new(
                ResponseScoutStatus::Success,
                Some(Vec::new()),
            ));
        }
        let result = db_client
            .insert_bulk_batched_by_shape("analysis_jobs", jobs)
            .await?;
        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(result),
        ))
    }

    pub async fn get_analysis_job(&mut self, id: i64) -> Result<ResponseScout<AnalysisJob>> {
        let db_client = self.get_db_client()?;
        match db_client
            .query_one(|client| {
                client
                    .from("analysis_jobs")
                    .select("*")
                    .eq("id", id.to_string())
            })
            .await
        {
            Ok(row) => Ok(ResponseScout::new(ResponseScoutStatus::Success, Some(row))),
            Err(_) => Ok(ResponseScout::new(ResponseScoutStatus::Failure, None)),
        }
    }

    pub async fn get_analysis_jobs(
        &mut self,
        status: Option<AnalysisWorkStatus>,
        limit: Option<u32>,
    ) -> Result<ResponseScout<Vec<AnalysisJob>>> {
        let db_client = self.get_db_client()?;
        let results = db_client
            .query(|client| {
                let q = client.from("analysis_jobs").select("*");
                let q = match status {
                    Some(st) => q.eq("status", st.as_str()),
                    None => q,
                };
                let q = q.order("inserted_at.desc");
                match limit {
                    Some(n) => q.limit(n as usize),
                    None => q,
                }
            })
            .await?;
        Ok(Self::handle_query_result(results))
    }

    pub async fn update_analysis_job(
        &mut self,
        id: i64,
        job: &AnalysisJob,
    ) -> Result<ResponseScout<AnalysisJob>> {
        let db_client = self.get_db_client()?;
        let result = db_client
            .update(job, |client| {
                client.from("analysis_jobs").eq("id", id.to_string())
            })
            .await?;
        if result.is_empty() {
            return Ok(ResponseScout::new(ResponseScoutStatus::Failure, None));
        }
        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            result.into_iter().next(),
        ))
    }

    pub async fn delete_analysis_job(&mut self, id: i64) -> Result<ResponseScout<()>> {
        let db_client = self.get_db_client()?;
        db_client
            .delete(|client| client.from("analysis_jobs").eq("id", id.to_string()))
            .await?;
        Ok(ResponseScout::new(ResponseScoutStatus::Success, None))
    }

    pub async fn create_analysis_task(
        &mut self,
        task: &AnalysisTask,
    ) -> Result<ResponseScout<AnalysisTask>> {
        let db_client = self.get_db_client()?;
        let result = db_client.insert("analysis_tasks", task).await?;
        Self::handle_insert_result(result)
    }

    pub async fn create_analysis_tasks_batch(
        &mut self,
        tasks: &[AnalysisTask],
    ) -> Result<ResponseScout<Vec<AnalysisTask>>> {
        let db_client = self.get_db_client()?;
        if tasks.is_empty() {
            return Ok(ResponseScout::new(
                ResponseScoutStatus::Success,
                Some(Vec::new()),
            ));
        }
        let result = db_client
            .insert_bulk_batched_by_shape("analysis_tasks", tasks)
            .await?;
        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            Some(result),
        ))
    }

    pub async fn get_analysis_task(&mut self, id: i64) -> Result<ResponseScout<AnalysisTask>> {
        let db_client = self.get_db_client()?;
        match db_client
            .query_one(|client| {
                client
                    .from("analysis_tasks")
                    .select("*")
                    .eq("id", id.to_string())
            })
            .await
        {
            Ok(row) => Ok(ResponseScout::new(ResponseScoutStatus::Success, Some(row))),
            Err(_) => Ok(ResponseScout::new(ResponseScoutStatus::Failure, None)),
        }
    }

    pub async fn get_analysis_tasks_for_job(
        &mut self,
        job_id: i64,
        limit: Option<u32>,
    ) -> Result<ResponseScout<Vec<AnalysisTask>>> {
        let db_client = self.get_db_client()?;
        let results = db_client
            .query(|client| {
                let q = client
                    .from("analysis_tasks")
                    .select("*")
                    .eq("job_id", job_id.to_string())
                    .order("id.desc");
                match limit {
                    Some(n) => q.limit(n as usize),
                    None => q,
                }
            })
            .await?;
        Ok(Self::handle_query_result(results))
    }

    pub async fn update_analysis_task(
        &mut self,
        id: i64,
        task: &AnalysisTask,
    ) -> Result<ResponseScout<AnalysisTask>> {
        let db_client = self.get_db_client()?;
        let result = db_client
            .update(task, |client| {
                client.from("analysis_tasks").eq("id", id.to_string())
            })
            .await?;
        if result.is_empty() {
            return Ok(ResponseScout::new(ResponseScoutStatus::Failure, None));
        }
        Ok(ResponseScout::new(
            ResponseScoutStatus::Success,
            result.into_iter().next(),
        ))
    }

    pub async fn delete_analysis_task(&mut self, id: i64) -> Result<ResponseScout<()>> {
        let db_client = self.get_db_client()?;
        db_client
            .delete(|client| client.from("analysis_tasks").eq("id", id.to_string()))
            .await?;
        Ok(ResponseScout::new(ResponseScoutStatus::Success, None))
    }
}