scout_rs 0.95.95

use crate::{
    client::ScoutClient,
    models::{
        connectivity, data, event, sensor_pose, ArtifactLocal, Connectivity, ConnectivityLocal,
        Event, EventLocal, SensorPose, SensorPoseLocal, Session, SessionLocal, Syncable, Tag,
        TagLocal,
    },
    storage::{StorageClient, StorageConfig, UploadProgress},
};
use anyhow::{Error, Result};
use native_db::{Builder, Database, Models, ToInput};
use std::sync::OnceLock;
use tracing::error;

macro_rules! define_models {
    ($models:expr, $($model:ty),+ $(,)?) => {
        $(
            $models
                .define::<$model>()
                .expect(concat!("Failed to define ", stringify!($model)));
        )+
    };
}

static MODELS: OnceLock<Models> = OnceLock::new();

// Superseded versions stay defined so a database written by an older build still opens.
fn models() -> &'static Models {
    MODELS.get_or_init(|| {
        let mut models = Models::new();
        define_models!(
            models,
            data::v9::SessionLocal,
            data::v11::SessionLocal,
            data::v14::SessionLocal,
            SessionLocal,
            event::EventLocalV9,
            event::EventLocalV12,
            event::EventLocalV15,
            event::EventLocalV16,
            EventLocal,
            data::v6::TagLocal,
            data::v8::TagLocal,
            TagLocal,
            data::v1::ConnectivityLocal,
            data::v2::ConnectivityLocal,
            data::v3::ConnectivityLocal,
            data::v4::ConnectivityLocal,
            data::v6::ConnectivityLocal,
            data::v11::ConnectivityLocal,
            data::v14::ConnectivityLocal,
            connectivity::ConnectivityLocalV17,
            ConnectivityLocal,
            data::v2::OperatorLocal,
            data::v9::ArtifactLocal,
            data::v13::ArtifactLocal,
            ArtifactLocal,
            sensor_pose::SensorPoseLocalV1,
            SensorPoseLocal,
        );
        models
    })
}

/// Drains rows left behind by superseded model versions into the current table. native_db keys
/// every table by model version, so rows written by an older build are invisible until moved.
/// Must run before the engine writes anything: `migrate` aborts when two versions of a model
/// hold rows at once.
fn migrate_local_models(database: &Database<'static>) -> Result<()> {
    let rw = database.rw_transaction()?;
    rw.migrate::<SessionLocal>()?;
    rw.migrate::<ConnectivityLocal>()?;
    rw.migrate::<EventLocal>()?;
    rw.migrate::<ArtifactLocal>()?;

    // `Models::define` picks the legacy flag by comparing versions across every registered model
    // rather than only those sharing a native_model id, so the much lower sensor pose version
    // reads as legacy and `migrate` refuses it. Move those rows by hand instead.
    let superseded_poses: Vec<sensor_pose::SensorPoseLocalV1> = rw
        .scan()
        .primary::<sensor_pose::SensorPoseLocalV1>()?
        .all()?
        .collect::<std::result::Result<_, _>>()?;
    for pose in superseded_poses {
        let pose = rw.remove(pose)?;
        rw.insert(SensorPoseLocal::from(pose))?;
    }

    rw.commit()?;
    Ok(())
}

/// SyncEngine handles synchronization between local database and remote Scout server.
///
/// The sync engine maintains a hierarchical sync order:
/// 1. Sessions (parent entities)
/// 2. Connectivity entries (children of sessions)
/// 3. Events (children of sessions)
/// 4. Tags (children of events)
///
/// Features:
/// - Batch operations for efficiency
/// - Automatic ID relationship management
/// - Configurable batch sizes
/// - Resilient error handling with partial failure recovery
pub struct SyncEngine {
    scout_client: ScoutClient,
    db_local_path: String,
    database: Database<'static>,
    max_num_items_per_sync: Option<u64>,
    remove_failed_records: bool,
    storage_client: Option<StorageClient>,
}

pub enum EnumSyncAction {
    Upsert,
    Insert,
    Skip,
}

const DEFAULT_MAX_NUM_ITEMS_PER_SYNC: u64 = 100;

pub struct BatchSync<T: ToInput + Syncable> {
    upsert: Vec<T>,
    insert: Vec<T>,
}

impl<T: ToInput + Syncable> BatchSync<T> {
    fn new() -> Self {
        Self {
            upsert: Vec::new(),
            insert: Vec::new(),
        }
    }

    fn add_upsert_item(&mut self, item: T) {
        self.upsert.push(item);
    }

    fn add_insert_item(&mut self, item: T) {
        self.insert.push(item);
    }
}

impl SyncEngine {
    /// Creates a new SyncEngine with custom configuration.
    ///
    /// # Arguments
    /// * `scout_client` - Client for communicating with Scout server
    /// * `db_local_path` - Path to local database file
    /// * `max_num_items_per_sync` - Maximum items per sync batch (None = unlimited)
    /// * `remove_failed_records` - Whether to remove failed records from the local database
    pub fn new(
        scout_client: ScoutClient,
        db_local_path: String,
        max_num_items_per_sync: Option<u64>,
        remove_failed_records: bool,
    ) -> Result<Self> {
        let database = Builder::new().create(models(), &db_local_path)?;
        if let Err(e) = migrate_local_models(&database) {
            error!("Failed to migrate local models, older rows stay unsynced: {e}");
        }
        Ok(Self {
            scout_client,
            db_local_path,
            database,
            max_num_items_per_sync,
            remove_failed_records,
            storage_client: None,
        })
    }

    /// Creates a default SyncEngine with common settings:
    /// - 100 items per sync batch
    /// - Remove failed records disabled (for safety)
    pub fn with_defaults(scout_client: ScoutClient, db_local_path: String) -> Result<Self> {
        Self::new(
            scout_client,
            db_local_path,
            Some(DEFAULT_MAX_NUM_ITEMS_PER_SYNC),
            false, // Remove failed records disabled by default for safety
        )
    }

    /// Creates a SyncEngine with remove_failed_records enabled:
    /// - 100 items per sync batch
    /// - Remove failed records enabled (removes records with critical errors)
    pub fn with_failed_record_removal(
        scout_client: ScoutClient,
        db_local_path: String,
    ) -> Result<Self> {
        Self::new(
            scout_client,
            db_local_path,
            Some(DEFAULT_MAX_NUM_ITEMS_PER_SYNC),
            true, // Remove failed records enabled
        )
    }

    fn get_batch<T: Syncable + ToInput>(
        &self,
        action_for_items_with_existing_ids: EnumSyncAction,
        action_for_items_without_existing_ids: EnumSyncAction,
    ) -> Result<BatchSync<T>, Error> {
        let r = self.database.r_transaction()?;
        let mut batch: BatchSync<T> = BatchSync::new();

        for raw_item in r.scan().primary::<T>()?.all()? {
            match raw_item {
                Ok(item) => {
                    // handle action for existing remote ids (on remote)
                    if item.id().is_some() {
                        match action_for_items_with_existing_ids {
                            EnumSyncAction::Insert => {
                                batch.add_insert_item(item);
                            }
                            EnumSyncAction::Upsert => {
                                batch.add_upsert_item(item);
                            }
                            EnumSyncAction::Skip => {
                                // Skip items that already have remote IDs
                            }
                        }
                    }
                    // handle action for no remote id (local only)
                    else {
                        match action_for_items_without_existing_ids {
                            EnumSyncAction::Insert => {
                                batch.add_insert_item(item);
                            }
                            EnumSyncAction::Upsert => {
                                batch.add_upsert_item(item);
                            }
                            EnumSyncAction::Skip => {
                                // Skip items without remote IDs (shouldn't happen)
                            }
                        }
                    }
                }
                Err(e) => {
                    tracing::error!("Failed to process item: {}", e);
                }
            }
        }
        Ok(batch)
    }

    /// Flushes all local data to remote server in proper order: sessions -> connectivity -> events -> operators -> tags
    /// Continues with remaining operations even if one fails, but reports all errors
    pub async fn flush(&mut self) -> Result<(), Error> {
        let mut sync_errors = Vec::new();

        // Sync sessions first (they're the parent of everything)
        if let Err(e) = self.flush_sessions().await {
            sync_errors.push(format!("Sessions sync failed: {}", e));
            tracing::error!(
                "Sessions sync failed, continuing with other operations: {}",
                e
            );
        }

        // Sync connectivity (depends on sessions)
        if let Err(e) = self.flush_connectivity().await {
            sync_errors.push(format!("Connectivity sync failed: {}", e));
            tracing::error!(
                "Connectivity sync failed, continuing with other operations: {}",
                e
            );
        }

        // Sync events (depends on sessions)
        if let Err(e) = self.flush_events().await {
            sync_errors.push(format!("Events sync failed: {}", e));
            tracing::error!(
                "Events sync failed, continuing with other operations: {}",
                e
            );
        }

        // Sync operators (depends on sessions)
        if let Err(e) = self.flush_operators().await {
            sync_errors.push(format!("Operators sync failed: {}", e));
            tracing::error!(
                "Operators sync failed, continuing with other operations: {}",
                e
            );
        }

        // Sync tags (depends on events)
        if let Err(e) = self.flush_tags().await {
            sync_errors.push(format!("Tags sync failed: {}", e));
            tracing::error!("Tags sync failed: {}", e);
        }

        // Sync artifacts (depends on sessions and devices)
        if let Err(e) = self.flush_artifacts().await {
            sync_errors.push(format!("Artifacts sync failed: {}", e));
            tracing::error!("Artifacts sync failed: {}", e);
        }

        // Sync sensor poses (children of parts, which are remote entities)
        if let Err(e) = self.flush_sensor_poses().await {
            sync_errors.push(format!("Sensor poses sync failed: {}", e));
            tracing::error!("Sensor poses sync failed: {}", e);
        }

        // Return error if any operations failed
        if !sync_errors.is_empty() {
            return Err(Error::msg(format!(
                "Sync completed with errors: {}",
                sync_errors.join("; ")
            )));
        }

        Ok(())
    }

    /// Syncs sessions to remote server
    async fn flush_sessions(&mut self) -> Result<(), Error> {
        // For sessions, we always upsert because they can be updated (e.g., timestamp_end)
        let sessions_batch: BatchSync<SessionLocal> = self.get_batch::<SessionLocal>(
            EnumSyncAction::Upsert, // Always upsert sessions with remote IDs
            EnumSyncAction::Upsert, // Always upsert sessions without remote IDs (insert)
        )?;

        // Process insert and upsert batches separately to avoid "All object keys must match" errors
        if !sessions_batch.insert.is_empty() {
            self.process_session_batch(sessions_batch.insert).await?;
        }
        if !sessions_batch.upsert.is_empty() {
            self.process_session_batch(sessions_batch.upsert).await?;
        }

        Ok(())
    }

    /// Processes a batch of sessions with fallback to individual processing on bulk failure
    async fn process_session_batch(
        &mut self,
        mut sessions: Vec<SessionLocal>,
    ) -> Result<(), Error> {
        if sessions.is_empty() {
            return Ok(());
        }

        // Apply batch size limit
        if let Some(max_items) = self.max_num_items_per_sync {
            if sessions.len() > max_items as usize {
                sessions.truncate(max_items as usize);
            }
        }

        let sessions_for_upsert: Vec<Session> = sessions
            .iter()
            .map(|local_session| local_session.clone().into())
            .collect();

        // Try bulk upsert first, fallback to individual on key mismatch errors
        let response = match self
            .scout_client
            .upsert_sessions_batch(&sessions_for_upsert)
            .await
        {
            Ok(response) => response,
            Err(e)
                if e.to_string()
                    .to_lowercase()
                    .contains("all object keys must match") =>
            {
                return self.fallback_individual_session_upserts(sessions).await;
            }
            Err(e) => {
                if Self::is_critical_error(&e.to_string()) && self.remove_failed_records {
                    tracing::warn!(
                        "Critical error in sessions batch, removing {} entries from local storage: {}",
                        sessions.len(),
                        e
                    );

                    if let Err(remove_err) = self.remove_items(sessions) {
                        tracing::error!("Failed to remove session entries: {}", remove_err);
                    }
                    return Ok(());
                } else {
                    return Err(e);
                }
            }
        };

        // Process successful bulk response
        if let Some(upserted_sessions) = response.data {
            let updated_locals: Vec<SessionLocal> = upserted_sessions
                .into_iter()
                .zip(sessions.iter())
                .map(|(remote_session, original_local)| {
                    let mut updated_local: SessionLocal = remote_session.into();
                    updated_local.id_local = original_local.id_local.clone();
                    updated_local
                })
                .collect();

            self.upsert_items(updated_locals.clone())?;

            // Update descendants for new sessions - only if parent exists and was newly created
            for (updated, original) in updated_locals.iter().zip(sessions.iter()) {
                if let (Some(new_id), Some(local_id), None) =
                    (updated.id, &original.id_local, original.id)
                {
                    // Validate the session was actually saved before updating descendants
                    if self
                        .validate_session_exists(local_id, new_id)
                        .unwrap_or(false)
                    {
                        if let Err(e) = self.update_session_descendants(local_id, new_id) {
                            tracing::error!(
                                "Failed to update descendants for session {}: {}",
                                local_id,
                                e
                            );
                        }
                    } else {
                        tracing::warn!(
                            "Session {} with remote ID {} not found - skipping descendant updates",
                            local_id,
                            new_id
                        );
                    }
                }
            }
        }
        Ok(())
    }

    /// Fallback to individual session upserts when bulk fails
    async fn fallback_individual_session_upserts(
        &mut self,
        sessions: Vec<SessionLocal>,
    ) -> Result<(), Error> {
        for session in sessions {
            let session_for_upsert: Session = session.clone().into();

            match self
                .scout_client
                .upsert_sessions_batch(&[session_for_upsert])
                .await
            {
                Ok(response) => {
                    if let Some(mut upserted_sessions) = response.data {
                        if let Some(upserted_session) = upserted_sessions.pop() {
                            let mut updated_local: SessionLocal = upserted_session.into();
                            updated_local.id_local = session.id_local.clone();
                            self.upsert_items(vec![updated_local.clone()])?;

                            // Update descendants for new sessions - validate parent exists first
                            if let (Some(new_id), Some(local_id), None) =
                                (updated_local.id, &session.id_local, session.id)
                            {
                                if self
                                    .validate_session_exists(local_id, new_id)
                                    .unwrap_or(false)
                                {
                                    if let Err(e) =
                                        self.update_session_descendants(local_id, new_id)
                                    {
                                        tracing::error!("Failed to update descendants: {}", e);
                                    }
                                } else {
                                    tracing::warn!(
                                        "Session {} with remote ID {} not validated - skipping descendants",
                                        local_id,
                                        new_id
                                    );
                                }
                            }
                        }
                    }
                }
                Err(e) => {
                    let error_message = e.to_string();

                    if Self::is_critical_error(&error_message) && self.remove_failed_records {
                        tracing::warn!(
                            "Critical error detected for session {:?}, removing from local storage: {}",
                            session.id_local,
                            error_message
                        );

                        if let Err(remove_err) = self.remove_items(vec![session]) {
                            tracing::error!(
                                "Failed to remove session from local storage: {}",
                                remove_err
                            );
                        } else {
                            tracing::info!(
                                "Removed session with critical error from local storage"
                            );
                        }
                    } else {
                        tracing::error!("Individual session upsert failed: {}", e);
                        return Err(e);
                    }
                }
            }
        }
        Ok(())
    }

    /// Syncs connectivity entries to remote server
    async fn flush_connectivity(&mut self) -> Result<(), Error> {
        // For connectivity, we only process items without remote IDs (new items to insert)
        let connectivity_batch: BatchSync<ConnectivityLocal> = self
            .get_batch::<ConnectivityLocal>(
                EnumSyncAction::Skip,   // Skip items with remote IDs - they're already synced
                EnumSyncAction::Insert, // Process items without remote IDs
            )?;

        // Only process items without remote IDs (the insert batch)
        let mut all_connectivity = connectivity_batch.insert;

        if let Some(max_items) = self.max_num_items_per_sync {
            if all_connectivity.len() > max_items as usize {
                tracing::info!(
                    "Limiting connectivity sync from {} to {} items",
                    all_connectivity.len(),
                    max_items
                );
                all_connectivity.truncate(max_items as usize);
            }
        }

        if all_connectivity.is_empty() {
            return Ok(());
        }

        // CRITICAL FIX: Update descendants BEFORE sending to remote server
        // Check if any connectivity records have ancestors with remote IDs and update descendants first
        let mut sessions_to_update = std::collections::HashSet::new();
        for connectivity in all_connectivity.iter() {
            if let Some(ancestor_local_id) = &connectivity.ancestor_id_local {
                // Check if the ancestor session has a remote ID
                if let Ok(Some(session)) = self.get_item::<SessionLocal>(ancestor_local_id) {
                    if let Some(_remote_session_id) = session.id {
                        // Session exists and has remote ID, mark for descendant updates
                        sessions_to_update.insert(ancestor_local_id.clone());
                    }
                }
            }
        }

        // Update descendants for all sessions that have remote IDs
        // This ensures connectivity records get their session_id populated BEFORE remote sync
        for session_local_id in sessions_to_update {
            if let Ok(Some(session)) = self.get_item::<SessionLocal>(&session_local_id) {
                if let Some(remote_session_id) = session.id {
                    if let Err(e) =
                        self.update_session_descendants(&session_local_id, remote_session_id)
                    {
                        tracing::error!(
                            "Failed to update descendants for session {} before connectivity sync: {}",
                            session_local_id,
                            e
                        );
                    } else {
                        tracing::debug!(
                            "Updated descendants for session {} before connectivity sync",
                            session_local_id
                        );
                    }
                }
            }
        }

        // NOW re-fetch the connectivity records (they may have been updated with session_id)
        // We need to get the updated versions with populated session_id values
        let mut updated_all_connectivity = Vec::new();
        for conn in all_connectivity.iter() {
            if let Some(local_id) = &conn.id_local {
                if let Ok(Some(updated_conn)) = self.get_item::<ConnectivityLocal>(local_id) {
                    updated_all_connectivity.push(updated_conn);
                } else {
                    // Fallback to original if we can't find the updated version
                    updated_all_connectivity.push(conn.clone());
                }
            } else {
                updated_all_connectivity.push(conn.clone());
            }
        }

        // Now convert the UPDATED connectivity records for remote sync
        let connectivity_for_insert: Vec<Connectivity> = updated_all_connectivity
            .iter()
            .map(|local_connectivity| local_connectivity.clone().into())
            .collect();

        let response = match self
            .scout_client
            .upsert_connectivity_batch(&connectivity_for_insert)
            .await
        {
            Ok(response) => response,
            Err(e) => {
                if Self::is_critical_error(&e.to_string()) && self.remove_failed_records {
                    tracing::warn!(
                        "Critical error in connectivity batch, removing {} entries from local storage: {}",
                        updated_all_connectivity.len(),
                        e
                    );

                    if let Err(remove_err) = self.remove_items(updated_all_connectivity) {
                        tracing::error!("Failed to remove connectivity entries: {}", remove_err);
                    }
                    return Ok(());
                } else {
                    return Err(e);
                }
            }
        };

        if let Some(inserted_connectivity) = response.data {
            let final_connectivity: Vec<ConnectivityLocal> = inserted_connectivity
                .into_iter()
                .zip(updated_all_connectivity.iter())
                .map(|(remote_connectivity, original_local)| {
                    let mut updated_local: ConnectivityLocal = remote_connectivity.into();
                    updated_local.id_local = original_local.id_local.clone();
                    updated_local.ancestor_id_local = original_local.ancestor_id_local.clone();
                    updated_local
                })
                .collect();

            self.upsert_items(final_connectivity)?;
        }

        Ok(())
    }

    /// Flushes local sensor poses to the remote server (insert-only). Sensor poses are
    /// children of parts (remote entities), so `part_id` is set at record time; no
    /// ancestor linking is needed (unlike connectivity's session relationship).
    async fn flush_sensor_poses(&mut self) -> Result<(), Error> {
        let sensor_pose_batch: BatchSync<SensorPoseLocal> = self.get_batch::<SensorPoseLocal>(
            EnumSyncAction::Skip,   // Skip items with remote IDs - already synced
            EnumSyncAction::Insert, // Process items without remote IDs
        )?;

        let mut all_sensor_poses = sensor_pose_batch.insert;

        if let Some(max_items) = self.max_num_items_per_sync {
            if all_sensor_poses.len() > max_items as usize {
                tracing::info!(
                    "Limiting sensor pose sync from {} to {} items",
                    all_sensor_poses.len(),
                    max_items
                );
                all_sensor_poses.truncate(max_items as usize);
            }
        }

        if all_sensor_poses.is_empty() {
            return Ok(());
        }

        let sensor_poses_for_insert: Vec<SensorPose> = all_sensor_poses
            .iter()
            .map(|local| local.clone().into())
            .collect();

        let response = match self
            .scout_client
            .upsert_sensor_pose_batch(&sensor_poses_for_insert)
            .await
        {
            Ok(response) => response,
            Err(e) => {
                if Self::is_critical_error(&e.to_string()) && self.remove_failed_records {
                    tracing::warn!(
                        "Critical error in sensor pose batch, removing {} entries from local storage: {}",
                        all_sensor_poses.len(),
                        e
                    );
                    if let Err(remove_err) = self.remove_items(all_sensor_poses) {
                        tracing::error!("Failed to remove sensor pose entries: {}", remove_err);
                    }
                    return Ok(());
                } else {
                    return Err(e);
                }
            }
        };

        if let Some(inserted_sensor_poses) = response.data {
            let final_sensor_poses: Vec<SensorPoseLocal> = inserted_sensor_poses
                .into_iter()
                .zip(all_sensor_poses.iter())
                .map(|(remote_sensor_pose, original_local)| {
                    let mut updated_local: SensorPoseLocal = remote_sensor_pose.into();
                    updated_local.id_local = original_local.id_local.clone();
                    updated_local
                })
                .collect();

            self.upsert_items(final_sensor_poses)?;
        }

        Ok(())
    }

    /// Syncs events to remote server
    async fn flush_events(&mut self) -> Result<(), Error> {
        // For events, we only process items without remote IDs (new items to insert)
        let events_batch: BatchSync<EventLocal> = self.get_batch::<EventLocal>(
            EnumSyncAction::Skip,   // Skip items with remote IDs - they're already synced
            EnumSyncAction::Insert, // Process items without remote IDs
        )?;

        // Only process items without remote IDs (the insert batch)
        let mut all_events = events_batch.insert;

        if let Some(max_items) = self.max_num_items_per_sync {
            if all_events.len() > max_items as usize {
                tracing::info!(
                    "Limiting events sync from {} to {} items",
                    all_events.len(),
                    max_items
                );
                all_events.truncate(max_items as usize);
            }
        }

        if all_events.is_empty() {
            return Ok(());
        }

        // CRITICAL FIX: Update descendants BEFORE sending to remote server
        // Check if any events have session ancestors with remote IDs and update descendants first
        let mut sessions_to_update = std::collections::HashSet::new();
        for event in all_events.iter() {
            if let Some(ancestor_local_id) = &event.ancestor_id_local {
                // Check if the ancestor session has a remote ID
                if let Ok(Some(session)) = self.get_item::<SessionLocal>(ancestor_local_id) {
                    if let Some(_remote_session_id) = session.id {
                        // Session exists and has remote ID, mark for descendant updates
                        sessions_to_update.insert(ancestor_local_id.clone());
                    }
                }
            }
        }

        // Update descendants for all sessions that have remote IDs
        // This ensures events get their session_id populated BEFORE remote sync
        for session_local_id in sessions_to_update {
            if let Ok(Some(session)) = self.get_item::<SessionLocal>(&session_local_id) {
                if let Some(remote_session_id) = session.id {
                    if let Err(e) =
                        self.update_session_descendants(&session_local_id, remote_session_id)
                    {
                        tracing::error!(
                            "Failed to update descendants for session {} before event sync: {}",
                            session_local_id,
                            e
                        );
                    } else {
                        tracing::debug!(
                            "Updated descendants for session {} before event sync",
                            session_local_id
                        );
                    }
                }
            }
        }

        // NOW re-fetch the events (they may have been updated with session_id)
        // We need to get the updated versions with populated session_id values
        let mut updated_all_events = Vec::new();
        for event in all_events.iter() {
            if let Some(local_id) = &event.id_local {
                if let Ok(Some(updated_event)) = self.get_item::<EventLocal>(local_id) {
                    updated_all_events.push(updated_event);
                } else {
                    // Fallback to original if we can't find the updated version
                    updated_all_events.push(event.clone());
                }
            } else {
                updated_all_events.push(event.clone());
            }
        }

        // Now convert the UPDATED events for remote sync
        let events_for_insert: Vec<Event> = updated_all_events
            .iter()
            .map(|local_event| local_event.clone().into())
            .collect();

        let response = match self
            .scout_client
            .upsert_events_batch(&events_for_insert)
            .await
        {
            Ok(response) => response,
            Err(e) => {
                if Self::is_critical_error(&e.to_string()) && self.remove_failed_records {
                    tracing::warn!(
                        "Critical error in events batch, removing {} entries from local storage: {}",
                        updated_all_events.len(),
                        e
                    );

                    if let Err(remove_err) = self.remove_items(updated_all_events) {
                        tracing::error!("Failed to remove event entries: {}", remove_err);
                    }
                    return Ok(());
                } else {
                    return Err(e);
                }
            }
        };

        if let Some(inserted_events) = response.data {
            let final_events: Vec<EventLocal> = inserted_events
                .into_iter()
                .zip(updated_all_events.iter())
                .map(|(remote_event, original_local)| {
                    let mut updated_local: EventLocal = remote_event.into();
                    updated_local.id_local = original_local.id_local.clone();
                    updated_local.ancestor_id_local = original_local.ancestor_id_local.clone();
                    updated_local
                })
                .collect();

            self.upsert_items(final_events.clone())?;

            // Update tag descendants with new remote event IDs - validate parent exists first
            for (updated_event, original_event) in
                final_events.iter().zip(updated_all_events.iter())
            {
                if let (Some(new_remote_id), Some(local_id)) =
                    (updated_event.id, &original_event.id_local)
                {
                    if original_event.id.is_none() {
                        // Validate the event was actually saved before updating descendants
                        if self
                            .validate_event_exists(local_id, new_remote_id)
                            .unwrap_or(false)
                        {
                            if let Err(e) = self.update_event_descendants(local_id, new_remote_id) {
                                tracing::error!(
                                    "Failed to update descendants for event {}: {}",
                                    local_id,
                                    e
                                );
                            }
                        } else {
                            tracing::warn!(
                                "Event {} with remote ID {} not found - skipping descendant updates",
                                local_id,
                                new_remote_id
                            );
                        }
                    }
                }
            }
        }

        Ok(())
    }

    /// Syncs tags to remote server
    async fn flush_tags(&mut self) -> Result<(), Error> {
        // For tags, we only process items without remote IDs (new items to insert)
        let tags_batch: BatchSync<TagLocal> = self.get_batch::<TagLocal>(
            EnumSyncAction::Skip,   // Skip items with remote IDs - they're already synced
            EnumSyncAction::Insert, // Process items without remote IDs
        )?;

        // Only process items without remote IDs (the insert batch)
        let mut all_tags = tags_batch.insert;

        if let Some(max_items) = self.max_num_items_per_sync {
            if all_tags.len() > max_items as usize {
                tracing::info!(
                    "Limiting tags sync from {} to {} items",
                    all_tags.len(),
                    max_items
                );
                all_tags.truncate(max_items as usize);
            }
        }

        if all_tags.is_empty() {
            return Ok(());
        }

        // CRITICAL FIX: Update descendants BEFORE sending to remote server
        // Check if any tags have event ancestors with remote IDs and update descendants first
        let mut events_to_update = std::collections::HashSet::new();
        let mut sessions_to_update = std::collections::HashSet::new();

        for tag in all_tags.iter() {
            if let Some(ancestor_local_id) = &tag.ancestor_id_local {
                // Check if the ancestor event has a remote ID
                if let Ok(Some(event)) = self.get_item::<EventLocal>(ancestor_local_id) {
                    if let Some(_remote_event_id) = event.id {
                        // Event exists and has remote ID, mark for descendant updates
                        events_to_update.insert(ancestor_local_id.clone());

                        // Also check if the event has a session ancestor
                        if let Some(session_ancestor_id) = &event.ancestor_id_local {
                            if let Ok(Some(session)) =
                                self.get_item::<SessionLocal>(session_ancestor_id)
                            {
                                if let Some(_remote_session_id) = session.id {
                                    sessions_to_update.insert(session_ancestor_id.clone());
                                }
                            }
                        }
                    }
                }
            }
        }

        // Update event descendants first
        for event_local_id in events_to_update {
            if let Ok(Some(event)) = self.get_item::<EventLocal>(&event_local_id) {
                if let Some(remote_event_id) = event.id {
                    if let Err(e) = self.update_event_descendants(&event_local_id, remote_event_id)
                    {
                        tracing::error!(
                            "Failed to update event descendants for event {} before tag sync: {}",
                            event_local_id,
                            e
                        );
                    } else {
                        tracing::debug!(
                            "Updated event descendants for event {} before tag sync",
                            event_local_id
                        );
                    }
                }
            }
        }

        // Update session descendants
        for session_local_id in sessions_to_update {
            if let Ok(Some(session)) = self.get_item::<SessionLocal>(&session_local_id) {
                if let Some(remote_session_id) = session.id {
                    if let Err(e) =
                        self.update_session_descendants(&session_local_id, remote_session_id)
                    {
                        tracing::error!(
                            "Failed to update session descendants for session {} before tag sync: {}",
                            session_local_id,
                            e
                        );
                    } else {
                        tracing::debug!(
                            "Updated session descendants for session {} before tag sync",
                            session_local_id
                        );
                    }
                }
            }
        }

        // NOW re-fetch the tags (they may have been updated with event_id)
        // We need to get the updated versions with populated event_id values
        let mut updated_all_tags = Vec::new();
        for tag in all_tags.iter() {
            if let Some(local_id) = &tag.id_local {
                if let Ok(Some(updated_tag)) = self.get_item::<TagLocal>(local_id) {
                    updated_all_tags.push(updated_tag);
                } else {
                    // Fallback to original if we can't find the updated version
                    updated_all_tags.push(tag.clone());
                }
            } else {
                updated_all_tags.push(tag.clone());
            }
        }

        // Now convert the UPDATED tags for remote sync
        let tags_for_insert: Vec<Tag> = updated_all_tags
            .iter()
            .map(|local_tag| local_tag.clone().into())
            .collect();

        let response = match self.scout_client.upsert_tags_batch(&tags_for_insert).await {
            Ok(response) => response,
            Err(e) => {
                if Self::is_critical_error(&e.to_string()) && self.remove_failed_records {
                    tracing::warn!(
                        "Critical error in tags batch, removing {} entries from local storage: {}",
                        updated_all_tags.len(),
                        e
                    );

                    if let Err(remove_err) = self.remove_items(updated_all_tags) {
                        tracing::error!("Failed to remove tag entries: {}", remove_err);
                    }
                    return Ok(());
                } else {
                    return Err(e);
                }
            }
        };

        if let Some(inserted_tags) = response.data {
            let final_tags: Vec<TagLocal> = inserted_tags
                .into_iter()
                .zip(updated_all_tags.iter())
                .map(|(remote_tag, original_local)| {
                    let mut updated_local: TagLocal = remote_tag.into();
                    updated_local.id_local = original_local.id_local.clone();
                    updated_local.ancestor_id_local = original_local.ancestor_id_local.clone();
                    // Prefer server value; keep local when API omits null optional columns.
                    if updated_local.subject_id.is_none() {
                        updated_local.subject_id = original_local.subject_id;
                    }
                    updated_local
                })
                .collect();

            self.upsert_items(final_tags)?;
        }

        Ok(())
    }

    /// Syncs artifacts to remote server
    ///
    /// Only artifacts with `has_uploaded_file_to_storage = true` will be synced.
    /// This ensures that artifact metadata is only sent to the server after
    /// the actual file has been successfully uploaded to storage.

    async fn flush_artifacts(&mut self) -> Result<(), Error> {
        // For artifacts, we support both upsert (existing items) and insert (new items)
        let artifacts_batch: BatchSync<ArtifactLocal> = self.get_batch::<ArtifactLocal>(
            EnumSyncAction::Upsert, // Process items with remote IDs for updates
            EnumSyncAction::Insert, // Process items without remote IDs for creation
        )?;

        // Process insert and upsert batches separately to ensure consistent field presence
        if !artifacts_batch.insert.is_empty() {
            self.process_artifact_insert_batch(artifacts_batch.insert)
                .await?;
        }
        if !artifacts_batch.upsert.is_empty() {
            self.process_artifact_upsert_batch(artifacts_batch.upsert)
                .await?;
        }

        Ok(())
    }

    /// Processes a batch of artifacts for insertion (new items without remote IDs)
    async fn process_artifact_insert_batch(
        &mut self,
        mut artifacts: Vec<ArtifactLocal>,
    ) -> Result<(), Error> {
        if artifacts.is_empty() {
            return Ok(());
        }

        // Filter to only include artifacts that have uploaded their files to storage
        let total_artifacts = artifacts.len();
        artifacts.retain(|artifact| artifact.has_uploaded_file_to_storage);

        let pending_uploads = total_artifacts - artifacts.len();
        if pending_uploads > 0 {
            tracing::debug!(
                "Skipping {} artifacts without uploaded files (only syncing {} with uploaded files)",
                pending_uploads,
                artifacts.len()
            );
        }

        if let Some(max_items) = self.max_num_items_per_sync {
            if artifacts.len() > max_items as usize {
                tracing::info!(
                    "Limiting artifact inserts from {} to {} items",
                    artifacts.len(),
                    max_items
                );
                artifacts.truncate(max_items as usize);
            }
        }

        if artifacts.is_empty() {
            tracing::debug!("No artifacts with uploaded files found for insertion");
            return Ok(());
        }

        // Update artifacts' session_id if their ancestor sessions have remote IDs
        let mut updated_artifacts = Vec::new();
        for artifact in artifacts.iter() {
            let mut updated_artifact = artifact.clone();
            if let Some(ancestor_local_id) = &artifact.ancestor_id_local {
                if let Ok(Some(session)) = self.get_item::<SessionLocal>(ancestor_local_id) {
                    if let Some(remote_session_id) = session.id {
                        updated_artifact.session_id = Some(remote_session_id);
                    }
                }
            }
            updated_artifacts.push(updated_artifact);
        }

        // Convert to API format for insertion
        let artifacts_for_api: Vec<crate::models::Artifact> = updated_artifacts
            .iter()
            .map(|artifact| {
                let mut api_artifact: crate::models::Artifact = artifact.clone().into();
                // Ensure id is None for inserts
                api_artifact.id = None;
                // Omit created_at and updated_at to rely on database defaults
                api_artifact.created_at = None;
                api_artifact.updated_at = None;
                api_artifact
            })
            .collect();

        tracing::info!("Inserting {} artifacts to remote", artifacts_for_api.len());

        let response = match self
            .scout_client
            .create_artifacts_batch(&artifacts_for_api)
            .await
        {
            Ok(response) => response,
            Err(e) => {
                if Self::is_critical_error(&e.to_string()) && self.remove_failed_records {
                    tracing::warn!(
                        "Critical error in artifacts insert batch, removing {} entries from local storage: {}",
                        updated_artifacts.len(),
                        e
                    );

                    if let Err(remove_err) = self.remove_items(updated_artifacts) {
                        tracing::error!("Failed to remove artifact entries: {}", remove_err);
                    }
                    return Ok(());
                } else {
                    return Err(e);
                }
            }
        };

        if let Some(remote_artifacts) = response.data {
            tracing::info!("Successfully inserted {} artifacts", remote_artifacts.len());

            // Update local records with remote IDs
            let mut updated_locals = Vec::new();
            for (remote_artifact, original_local) in
                remote_artifacts.into_iter().zip(updated_artifacts.iter())
            {
                let mut updated_local: ArtifactLocal = remote_artifact.into();
                updated_local.id_local = original_local.id_local.clone();
                updated_local.ancestor_id_local = original_local.ancestor_id_local.clone();
                updated_local.has_uploaded_file_to_storage =
                    original_local.has_uploaded_file_to_storage;
                updated_local.upload_url = original_local.upload_url.clone();
                updated_local.upload_url_generated_at =
                    original_local.upload_url_generated_at.clone();
                updated_locals.push(updated_local);
            }

            self.upsert_items(updated_locals)?;
        }

        Ok(())
    }

    /// Processes a batch of artifacts for upsert (existing items with remote IDs)
    async fn process_artifact_upsert_batch(
        &mut self,
        mut artifacts: Vec<ArtifactLocal>,
    ) -> Result<(), Error> {
        if artifacts.is_empty() {
            return Ok(());
        }

        // Filter to only include artifacts that have uploaded their files to storage
        let total_artifacts = artifacts.len();
        artifacts.retain(|artifact| artifact.has_uploaded_file_to_storage);

        let pending_uploads = total_artifacts - artifacts.len();
        if pending_uploads > 0 {
            tracing::debug!(
                "Skipping {} artifacts without uploaded files (only syncing {} with uploaded files)",
                pending_uploads,
                artifacts.len()
            );
        }

        if let Some(max_items) = self.max_num_items_per_sync {
            if artifacts.len() > max_items as usize {
                tracing::info!(
                    "Limiting artifact upserts from {} to {} items",
                    artifacts.len(),
                    max_items
                );
                artifacts.truncate(max_items as usize);
            }
        }

        if artifacts.is_empty() {
            tracing::debug!("No artifacts with uploaded files found for upsert");
            return Ok(());
        }

        // Update artifacts' session_id if their ancestor sessions have remote IDs
        let mut updated_artifacts = Vec::new();
        for artifact in artifacts.iter() {
            let mut updated_artifact = artifact.clone();
            if let Some(ancestor_local_id) = &artifact.ancestor_id_local {
                if let Ok(Some(session)) = self.get_item::<SessionLocal>(ancestor_local_id) {
                    if let Some(remote_session_id) = session.id {
                        updated_artifact.session_id = Some(remote_session_id);
                    }
                }
            }
            updated_artifacts.push(updated_artifact);
        }

        // Convert to API format for upsert
        // Filter to ensure all artifacts have remote IDs
        let artifacts_for_api: Vec<crate::models::Artifact> = updated_artifacts
            .iter()
            .filter(|artifact| artifact.id.is_some())
            .map(|artifact| artifact.clone().into())
            .collect();

        tracing::info!("Upserting {} artifacts to remote", artifacts_for_api.len());

        let response = match self
            .scout_client
            .upsert_artifacts_batch(&artifacts_for_api)
            .await
        {
            Ok(response) => response,
            Err(e) => {
                if Self::is_critical_error(&e.to_string()) && self.remove_failed_records {
                    tracing::warn!(
                        "Critical error in artifacts upsert batch, removing {} entries from local storage: {}",
                        updated_artifacts.len(),
                        e
                    );

                    if let Err(remove_err) = self.remove_items(updated_artifacts) {
                        tracing::error!("Failed to remove artifact entries: {}", remove_err);
                    }
                    return Ok(());
                } else {
                    return Err(e);
                }
            }
        };

        if let Some(remote_artifacts) = response.data {
            tracing::info!("Successfully upserted {} artifacts", remote_artifacts.len());

            // Update local records with remote IDs and data
            let mut updated_locals = Vec::new();
            for (remote_artifact, original_local) in
                remote_artifacts.into_iter().zip(updated_artifacts.iter())
            {
                let mut updated_local: ArtifactLocal = remote_artifact.into();
                updated_local.id_local = original_local.id_local.clone();
                updated_local.ancestor_id_local = original_local.ancestor_id_local.clone();
                updated_local.has_uploaded_file_to_storage =
                    original_local.has_uploaded_file_to_storage;
                updated_local.upload_url = original_local.upload_url.clone();
                updated_local.upload_url_generated_at =
                    original_local.upload_url_generated_at.clone();
                updated_locals.push(updated_local);
            }

            self.upsert_items(updated_locals)?;
        }

        Ok(())
    }

    /// Syncs operators to remote server
    async fn flush_operators(&mut self) -> Result<(), Error> {
        // For operators, we only process items without remote IDs (new items to insert)
        let operators_batch: BatchSync<data::v2::OperatorLocal> = self
            .get_batch::<data::v2::OperatorLocal>(
                EnumSyncAction::Skip,   // Skip items with remote IDs - they're already synced
                EnumSyncAction::Insert, // Process items without remote IDs
            )?;

        // Only process items without remote IDs (the insert batch)
        let mut all_operators = operators_batch.insert;

        if let Some(max_items) = self.max_num_items_per_sync {
            if all_operators.len() > max_items as usize {
                tracing::info!(
                    "Limiting operators sync from {} to {} items",
                    all_operators.len(),
                    max_items
                );
                all_operators.truncate(max_items as usize);
            }
        }

        if all_operators.is_empty() {
            return Ok(());
        }

        // CRITICAL FIX: Update descendants BEFORE sending to remote server
        // Check if any operators have session ancestors with remote IDs and update descendants first
        let mut sessions_to_update = std::collections::HashSet::new();
        for operator in all_operators.iter() {
            if let Some(ancestor_local_id) = &operator.ancestor_id_local {
                // Check if the ancestor session has a remote ID
                if let Ok(Some(session)) = self.get_item::<SessionLocal>(ancestor_local_id) {
                    if let Some(_remote_session_id) = session.id {
                        // Session exists and has remote ID, mark for descendant updates
                        sessions_to_update.insert(ancestor_local_id.clone());
                    }
                }
            }
        }

        // Update descendants for all sessions that have remote IDs
        // This ensures operators get their session_id populated BEFORE remote sync
        for session_local_id in sessions_to_update {
            if let Ok(Some(session)) = self.get_item::<SessionLocal>(&session_local_id) {
                if let Some(remote_session_id) = session.id {
                    if let Err(e) =
                        self.update_session_descendants(&session_local_id, remote_session_id)
                    {
                        tracing::error!(
                            "Failed to update descendants for session {} before operator sync: {}",
                            session_local_id,
                            e
                        );
                    } else {
                        tracing::debug!(
                            "Updated descendants for session {} before operator sync",
                            session_local_id
                        );
                    }
                }
            }
        }

        // NOW re-fetch the operators (they may have been updated with session_id)
        // We need to get the updated versions with populated session_id values
        let mut updated_all_operators = Vec::new();
        for operator in all_operators.iter() {
            if let Some(local_id) = &operator.id_local {
                if let Ok(Some(updated_operator)) =
                    self.get_item::<data::v2::OperatorLocal>(local_id)
                {
                    updated_all_operators.push(updated_operator);
                } else {
                    // Fallback to original if we can't find the updated version
                    updated_all_operators.push(operator.clone());
                }
            } else {
                updated_all_operators.push(operator.clone());
            }
        }

        // Now convert the UPDATED operators for remote sync
        let operators_for_insert: Vec<data::v2::Operator> = updated_all_operators
            .iter()
            .map(|local_operator| {
                // Convert OperatorLocal to Operator (removes local-only fields)
                data::v2::Operator::from(local_operator.clone())
            })
            .collect();

        let response = match self
            .scout_client
            .upsert_operators_batch(&operators_for_insert)
            .await
        {
            Ok(response) => response,
            Err(e) => {
                if Self::is_critical_error(&e.to_string()) && self.remove_failed_records {
                    tracing::warn!(
                        "Critical error in operators batch, removing {} entries from local storage: {}",
                        updated_all_operators.len(),
                        e
                    );

                    if let Err(remove_err) = self.remove_items(updated_all_operators) {
                        tracing::error!("Failed to remove operator entries: {}", remove_err);
                    }
                    return Ok(());
                } else {
                    return Err(e);
                }
            }
        };

        if let Some(inserted_operators) = response.data {
            let final_operators: Vec<data::v2::OperatorLocal> = inserted_operators
                .into_iter()
                .zip(updated_all_operators.iter())
                .map(|(remote_operator, original_local)| {
                    let mut updated_local = data::v2::OperatorLocal::from(remote_operator);
                    updated_local.id_local = original_local.id_local.clone();
                    updated_local.ancestor_id_local = original_local.ancestor_id_local.clone();
                    updated_local
                })
                .collect();

            self.upsert_items(final_operators)?;
        }

        Ok(())
    }

    /// Gets an item from the database by local ID and returns a clone
    pub fn get_item<T: ToInput + Syncable + Clone>(
        &self,
        local_id: &str,
    ) -> Result<Option<T>, Error> {
        let r = self.database.r_transaction()?;

        for raw_item in r.scan().primary::<T>()?.all()? {
            if let Ok(item) = raw_item {
                if let Some(item_local_id) = item.id_local() {
                    if item_local_id == local_id {
                        return Ok(Some(item));
                    }
                }
            }
        }

        Ok(None)
    }

    /// Cleans completed sessions and their descendants from local database
    /// Uses safe cleaning: timestamp_end set and all descendants synced
    pub async fn clean(&mut self) -> Result<(), Error> {
        tracing::info!("Starting clean operation for sessions");

        let r = self.database.r_transaction()?;
        let mut sessions_to_clean = Vec::new();

        for raw_session in r.scan().primary::<SessionLocal>()?.all()? {
            if let Ok(session) = raw_session {
                if let (Some(_end_time_str), Some(_remote_id)) =
                    (&session.timestamp_end, session.id)
                {
                    if self.session_descendants_have_remote_ids(&session, &r)? {
                        sessions_to_clean.push(session);
                    }
                }
            }
        }
        drop(r);

        if sessions_to_clean.is_empty() {
            tracing::debug!("No sessions found for cleaning");
            return Ok(());
        }

        tracing::info!("Found {} sessions to clean", sessions_to_clean.len());

        for session in sessions_to_clean {
            self.clean_session_and_descendants(&session).await?;
        }

        Ok(())
    }

    /// Checks if all descendants of a session have remote IDs
    fn session_descendants_have_remote_ids(
        &self,
        session: &SessionLocal,
        r: &native_db::transaction::RTransaction,
    ) -> Result<bool, Error> {
        let session_local_id = match &session.id_local {
            Some(id) => id,
            None => return Ok(false),
        };

        // Check connectivity entries
        for raw_connectivity in r.scan().primary::<ConnectivityLocal>()?.all()? {
            if let Ok(connectivity) = raw_connectivity {
                if connectivity.ancestor_id_local.as_deref() == Some(session_local_id) {
                    if connectivity.id.is_none() {
                        tracing::debug!(
                            "Session {} has connectivity without remote ID",
                            session_local_id
                        );
                        return Ok(false);
                    }
                }
            }
        }

        // Check operators entries
        for raw_operator in r.scan().primary::<data::v2::OperatorLocal>()?.all()? {
            if let Ok(operator) = raw_operator {
                if operator.ancestor_id_local.as_deref() == Some(session_local_id) {
                    if operator.id.is_none() {
                        tracing::debug!(
                            "Session {} has operator without remote ID",
                            session_local_id
                        );
                        return Ok(false);
                    }
                }
            }
        }

        // Check artifacts entries
        for raw_artifact in r.scan().primary::<ArtifactLocal>()?.all()? {
            if let Ok(artifact) = raw_artifact {
                if artifact.ancestor_id_local.as_deref() == Some(session_local_id) {
                    if artifact.id.is_none() {
                        tracing::debug!(
                            "Session {} has artifact without remote ID",
                            session_local_id
                        );
                        return Ok(false);
                    }
                }
            }
        }

        // Check events and their tags
        for raw_event in r.scan().primary::<EventLocal>()?.all()? {
            if let Ok(event) = raw_event {
                if event.ancestor_id_local.as_deref() == Some(session_local_id) {
                    if event.id.is_none() {
                        tracing::debug!("Session {} has event without remote ID", session_local_id);
                        return Ok(false);
                    }

                    // Check tags for this event
                    if let Some(event_local_id) = &event.id_local {
                        for raw_tag in r.scan().primary::<TagLocal>()?.all()? {
                            if let Ok(tag) = raw_tag {
                                if tag.ancestor_id_local.as_deref() == Some(event_local_id) {
                                    if tag.id.is_none() {
                                        tracing::debug!(
                                            "Session {} has tag without remote ID for event {}",
                                            session_local_id,
                                            event_local_id
                                        );
                                        return Ok(false);
                                    }
                                }
                            }
                        }
                    }
                }
            }
        }

        Ok(true)
    }

    /// Removes a session and all its descendants from local database
    async fn clean_session_and_descendants(&mut self, session: &SessionLocal) -> Result<(), Error> {
        let session_local_id = match &session.id_local {
            Some(id) => id.clone(),
            None => return Ok(()),
        };

        tracing::info!("Cleaning session {} and descendants", session_local_id);

        // First, collect all items to remove using read transaction
        let r = self.database.r_transaction()?;

        let mut tags_to_remove = Vec::new();
        let mut events_to_remove = Vec::new();
        let mut connectivity_to_remove = Vec::new();
        let mut operators_to_remove = Vec::new();
        let mut artifacts_to_remove = Vec::new();

        // Collect events for this session
        for raw_event in r.scan().primary::<EventLocal>()?.all()? {
            if let Ok(event) = raw_event {
                if event.ancestor_id_local.as_deref() == Some(&session_local_id) {
                    events_to_remove.push(event);
                }
            }
        }

        // Collect tags for each event
        for event in &events_to_remove {
            if let Some(event_local_id) = &event.id_local {
                for raw_tag in r.scan().primary::<TagLocal>()?.all()? {
                    if let Ok(tag) = raw_tag {
                        if tag.ancestor_id_local.as_deref() == Some(event_local_id) {
                            tags_to_remove.push(tag);
                        }
                    }
                }
            }
        }

        // Collect connectivity entries
        for raw_connectivity in r.scan().primary::<ConnectivityLocal>()?.all()? {
            if let Ok(connectivity) = raw_connectivity {
                if connectivity.ancestor_id_local.as_deref() == Some(&session_local_id) {
                    connectivity_to_remove.push(connectivity);
                }
            }
        }

        // Collect operators entries
        for raw_operator in r.scan().primary::<data::v2::OperatorLocal>()?.all()? {
            if let Ok(operator) = raw_operator {
                if operator.ancestor_id_local.as_deref() == Some(&session_local_id) {
                    operators_to_remove.push(operator);
                }
            }
        }

        // Collect artifacts entries
        for raw_artifact in r.scan().primary::<ArtifactLocal>()?.all()? {
            if let Ok(artifact) = raw_artifact {
                if artifact.ancestor_id_local.as_deref() == Some(&session_local_id) {
                    artifacts_to_remove.push(artifact);
                }
            }
        }

        drop(r); // Close read transaction

        // Now remove all items using write transaction
        let rw = self.database.rw_transaction()?;

        // Remove tags
        let tags_count = tags_to_remove.len();
        for tag in tags_to_remove {
            rw.remove(tag)?;
        }

        // Remove events
        let events_count = events_to_remove.len();
        for event in events_to_remove {
            rw.remove(event)?;
        }

        // Remove connectivity entries
        let connectivity_count = connectivity_to_remove.len();
        for connectivity in connectivity_to_remove {
            rw.remove(connectivity)?;
        }

        // Remove operators entries
        let operators_count = operators_to_remove.len();
        for operator in operators_to_remove {
            rw.remove(operator)?;
        }

        // Remove artifacts entries
        let artifacts_count = artifacts_to_remove.len();
        for artifact in artifacts_to_remove {
            rw.remove(artifact)?;
        }

        // Remove the session itself
        rw.remove(session.clone())?;

        rw.commit()?;

        tracing::info!(
            "Cleaned session {}: removed {} tags, {} events, {} connectivity entries, {} operators, {} artifacts, and 1 session",
            session_local_id,
            tags_count,
            events_count,
            connectivity_count,
            operators_count,
            artifacts_count
        );

        Ok(())
    }

    /// Returns the path to the local database file
    pub fn get_db_path(&self) -> &str {
        &self.db_local_path
    }

    /// Exports all sync engine data to a JSON file
    /// Returns an array where each element is a session with all its descendants
    /// Useful for exporting data to clients that don't support native_db structure
    pub fn export_to_json(&self, output_path: &str) -> Result<(), Error> {
        use serde_json;
        use std::collections::HashMap;
        use std::fs;

        tracing::info!("Exporting sync engine data to {}", output_path);

        let r = self.database.r_transaction()?;

        // Collect all sessions
        let mut sessions = Vec::new();
        for raw_session in r.scan().primary::<SessionLocal>()?.all()? {
            if let Ok(session) = raw_session {
                sessions.push(session);
            }
        }

        // Collect all events and group by session
        let mut events_by_session: HashMap<String, Vec<EventLocal>> = HashMap::new();
        for raw_event in r.scan().primary::<EventLocal>()?.all()? {
            if let Ok(event) = raw_event {
                if let Some(session_id) = &event.ancestor_id_local {
                    events_by_session
                        .entry(session_id.clone())
                        .or_insert_with(Vec::new)
                        .push(event);
                }
            }
        }

        // Collect all tags and group by event
        let mut tags_by_event: HashMap<String, Vec<TagLocal>> = HashMap::new();
        for raw_tag in r.scan().primary::<TagLocal>()?.all()? {
            if let Ok(tag) = raw_tag {
                if let Some(event_id) = &tag.ancestor_id_local {
                    tags_by_event
                        .entry(event_id.clone())
                        .or_insert_with(Vec::new)
                        .push(tag);
                }
            }
        }

        // Collect all connectivity entries and group by session
        let mut connectivity_by_session: HashMap<String, Vec<ConnectivityLocal>> = HashMap::new();
        for raw_connectivity in r.scan().primary::<ConnectivityLocal>()?.all()? {
            if let Ok(conn) = raw_connectivity {
                if let Some(session_id) = &conn.ancestor_id_local {
                    connectivity_by_session
                        .entry(session_id.clone())
                        .or_insert_with(Vec::new)
                        .push(conn);
                }
            }
        }

        // Collect all operators and group by session
        let mut operators_by_session: HashMap<String, Vec<data::v2::OperatorLocal>> =
            HashMap::new();
        for raw_operator in r.scan().primary::<data::v2::OperatorLocal>()?.all()? {
            if let Ok(operator) = raw_operator {
                if let Some(session_id) = &operator.ancestor_id_local {
                    operators_by_session
                        .entry(session_id.clone())
                        .or_insert_with(Vec::new)
                        .push(operator);
                }
            }
        }

        // Collect all artifacts and group by session
        let mut artifacts_by_session: HashMap<String, Vec<ArtifactLocal>> = HashMap::new();
        for raw_artifact in r.scan().primary::<ArtifactLocal>()?.all()? {
            if let Ok(artifact) = raw_artifact {
                if let Some(session_id) = &artifact.ancestor_id_local {
                    artifacts_by_session
                        .entry(session_id.clone())
                        .or_insert_with(Vec::new)
                        .push(artifact);
                }
            }
        }

        drop(r); // Close read transaction

        // Build array of sessions with nested descendants
        let mut export_array = Vec::new();
        for session in sessions {
            let session_local_id = session.id_local.as_deref().unwrap_or("");

            // Get events for this session
            let events = events_by_session
                .get(session_local_id)
                .cloned()
                .unwrap_or_default();

            // Get tags for all events in this session
            let mut tags = Vec::new();
            for event in &events {
                if let Some(event_id) = &event.id_local {
                    if let Some(event_tags) = tags_by_event.get(event_id) {
                        tags.extend(event_tags.clone());
                    }
                }
            }

            // Get connectivity for this session
            let connectivity = connectivity_by_session
                .get(session_local_id)
                .cloned()
                .unwrap_or_default();

            // Get operators for this session
            let operators = operators_by_session
                .get(session_local_id)
                .cloned()
                .unwrap_or_default();

            // Get artifacts for this session
            let artifacts = artifacts_by_session
                .get(session_local_id)
                .cloned()
                .unwrap_or_default();

            // Create session entry with nested descendants
            let session_entry = serde_json::json!({
                "session": session,
                "events": events,
                "tags": tags,
                "connectivity": connectivity,
                "operators": operators,
                "artifacts": artifacts
            });

            export_array.push(session_entry);
        }

        // Write to file
        let json_string = serde_json::to_string_pretty(&export_array)?;
        fs::write(output_path, json_string)?;

        tracing::info!(
            "Exported {} sessions with their descendants",
            export_array.len()
        );

        Ok(())
    }

    /// Wipes data from the sync engine
    /// If session_ids is Some, only wipes the specified sessions and their descendants
    /// If session_ids is None or empty, wipes all data
    /// Removes all items from all tables in dependency order
    pub fn wipe(&mut self, session_ids: Option<Vec<String>>) -> Result<(), Error> {
        let r = self.database.r_transaction()?;

        let mut tags_to_remove = Vec::new();
        let mut events_to_remove = Vec::new();
        let mut connectivity_to_remove = Vec::new();
        let mut operators_to_remove = Vec::new();
        let mut artifacts_to_remove = Vec::new();
        let mut sessions_to_remove = Vec::new();

        // Determine which sessions to wipe
        let session_ids_to_wipe: std::collections::HashSet<String> = if let Some(ids) = session_ids
        {
            if ids.is_empty() {
                // Empty vec means wipe all
                let mut all_ids = std::collections::HashSet::new();
                for raw_session in r.scan().primary::<SessionLocal>()?.all()? {
                    if let Ok(session) = raw_session {
                        if let Some(id) = session.id_local {
                            all_ids.insert(id);
                        }
                    }
                }
                all_ids
            } else {
                ids.into_iter().collect()
            }
        } else {
            // None means wipe all
            let mut all_ids = std::collections::HashSet::new();
            for raw_session in r.scan().primary::<SessionLocal>()?.all()? {
                if let Ok(session) = raw_session {
                    if let Some(id) = session.id_local {
                        all_ids.insert(id);
                    }
                }
            }
            all_ids
        };

        if session_ids_to_wipe.is_empty() {
            tracing::info!("No sessions to wipe");
            return Ok(());
        }

        tracing::info!(
            "Wiping {} session(s) and their descendants",
            session_ids_to_wipe.len()
        );

        // Collect sessions to remove
        for raw_session in r.scan().primary::<SessionLocal>()?.all()? {
            if let Ok(session) = raw_session {
                if let Some(id) = &session.id_local {
                    if session_ids_to_wipe.contains(id) {
                        sessions_to_remove.push(session);
                    }
                }
            }
        }

        // Collect events for specified sessions
        for raw_event in r.scan().primary::<EventLocal>()?.all()? {
            if let Ok(event) = raw_event {
                if let Some(session_id) = &event.ancestor_id_local {
                    if session_ids_to_wipe.contains(session_id) {
                        events_to_remove.push(event);
                    }
                }
            }
        }

        // Collect tags for events in specified sessions
        for event in &events_to_remove {
            if let Some(event_id) = &event.id_local {
                for raw_tag in r.scan().primary::<TagLocal>()?.all()? {
                    if let Ok(tag) = raw_tag {
                        if tag.ancestor_id_local.as_deref() == Some(event_id) {
                            tags_to_remove.push(tag);
                        }
                    }
                }
            }
        }

        // Collect connectivity entries for specified sessions
        for raw_connectivity in r.scan().primary::<ConnectivityLocal>()?.all()? {
            if let Ok(connectivity) = raw_connectivity {
                if let Some(session_id) = &connectivity.ancestor_id_local {
                    if session_ids_to_wipe.contains(session_id) {
                        connectivity_to_remove.push(connectivity);
                    }
                }
            }
        }

        // Collect operators for specified sessions
        for raw_operator in r.scan().primary::<data::v2::OperatorLocal>()?.all()? {
            if let Ok(operator) = raw_operator {
                if let Some(session_id) = &operator.ancestor_id_local {
                    if session_ids_to_wipe.contains(session_id) {
                        operators_to_remove.push(operator);
                    }
                }
            }
        }

        // Collect artifacts for specified sessions
        for raw_artifact in r.scan().primary::<ArtifactLocal>()?.all()? {
            if let Ok(artifact) = raw_artifact {
                if let Some(session_id) = &artifact.ancestor_id_local {
                    if session_ids_to_wipe.contains(session_id) {
                        artifacts_to_remove.push(artifact);
                    }
                }
            }
        }

        drop(r); // Close read transaction

        // Now remove all items using write transaction in dependency order
        let rw = self.database.rw_transaction()?;

        // Remove tags first (depend on events)
        let tags_count = tags_to_remove.len();
        for tag in tags_to_remove {
            rw.remove(tag)?;
        }

        // Remove events (depend on sessions)
        let events_count = events_to_remove.len();
        for event in events_to_remove {
            rw.remove(event)?;
        }

        // Remove connectivity entries (depend on sessions)
        let connectivity_count = connectivity_to_remove.len();
        for connectivity in connectivity_to_remove {
            rw.remove(connectivity)?;
        }

        // Remove operators (depend on sessions)
        let operators_count = operators_to_remove.len();
        for operator in operators_to_remove {
            rw.remove(operator)?;
        }

        // Remove artifacts (depend on sessions)
        let artifacts_count = artifacts_to_remove.len();
        for artifact in artifacts_to_remove {
            rw.remove(artifact)?;
        }

        // Remove sessions last
        let sessions_count = sessions_to_remove.len();
        for session in sessions_to_remove {
            rw.remove(session)?;
        }

        rw.commit()?;

        tracing::info!(
            "Wiped {} session(s): removed {} tags, {} events, {} connectivity, {} operators, {} artifacts, {} sessions",
            session_ids_to_wipe.len(),
            tags_count,
            events_count,
            connectivity_count,
            operators_count,
            artifacts_count,
            sessions_count
        );

        Ok(())
    }

    /// Generates a unique ID using timestamp and table count to avoid race conditions
    pub fn generate_unique_id<T: ToInput>(&self) -> Result<u64, Error> {
        use std::time::{SystemTime, UNIX_EPOCH};

        let timestamp = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map_err(|e| Error::msg(format!("System time error: {}", e)))?
            .as_millis() as u64;

        // Use timestamp as base with table count as offset to ensure uniqueness
        let count = self.get_table_count::<T>()?;
        Ok(timestamp * 1000 + count)
    }

    /// Gets the number of items in a specific table type
    pub fn get_table_count<T: ToInput>(&self) -> Result<u64, Error> {
        let r = self.database.r_transaction()?;
        let count = r.len().primary::<T>();
        match count {
            Ok(count) => Ok(count),
            Err(e) => Err(e.into()),
        }
    }

    /// Removes multiple items from the local database
    pub fn remove_items<T: ToInput>(&mut self, items: Vec<T>) -> Result<(), Error> {
        let rw = self.database.rw_transaction();
        match rw {
            Ok(rw) => {
                for item in items {
                    rw.remove(item)?;
                }
                match rw.commit() {
                    Ok(_) => Ok(()),
                    Err(e) => {
                        error!("Failed to commit items to database: {}", e);
                        Err(e.into())
                    }
                }
            }
            Err(e) => Err(e.into()),
        }
    }

    /// Inserts or updates multiple items in the local database
    pub fn upsert_items<T: ToInput>(&mut self, items: Vec<T>) -> Result<(), Error> {
        let rw = self.database.rw_transaction()?;
        for item in items {
            rw.upsert(item)?;
        }
        rw.commit()?;
        Ok(())
    }

    /// Returns the count of artifacts that are pending file upload
    pub fn get_artifacts_pending_upload_count(&self) -> Result<usize, Error> {
        let r = self.database.r_transaction()?;
        let mut pending_count = 0;

        for raw_artifact in r.scan().primary::<ArtifactLocal>()?.all()? {
            if let Ok(artifact) = raw_artifact {
                if !artifact.has_uploaded_file_to_storage {
                    pending_count += 1;
                }
            }
        }

        Ok(pending_count)
    }

    /// Returns artifacts that are pending file upload
    pub fn get_artifacts_pending_upload(&self) -> Result<Vec<ArtifactLocal>, Error> {
        let r = self.database.r_transaction()?;
        let mut pending_artifacts = Vec::new();

        for raw_artifact in r.scan().primary::<ArtifactLocal>()?.all()? {
            if let Ok(artifact) = raw_artifact {
                if !artifact.has_uploaded_file_to_storage {
                    pending_artifacts.push(artifact);
                }
            }
        }

        Ok(pending_artifacts)
    }

    /// Sets up storage client for artifact uploads
    pub fn with_storage(mut self, storage_config: StorageConfig) -> Result<Self, Error> {
        self.storage_client = Some(StorageClient::new(storage_config)?);
        Ok(self)
    }

    /// Generates upload URLs for the provided artifacts
    ///
    /// This will update artifacts in-place with upload URLs and timestamps.
    /// Existing URLs within 24 hours will be reused.
    pub async fn generate_upload_urls(
        &mut self,
        artifacts: &mut Vec<ArtifactLocal>,
    ) -> Result<(), Error> {
        let storage_client = self.storage_client.as_ref().ok_or_else(|| {
            Error::msg("Storage client not configured. Call with_storage() first.")
        })?;

        storage_client.generate_upload_urls(artifacts).await?;

        // Update the artifacts in the database
        self.upsert_items(artifacts.clone())?;

        Ok(())
    }

    /// Upload a single artifact to storage using spawned task
    /// Returns a tuple of (task handle, progress receiver). Consumer must handle updating database.
    ///
    /// # Arguments
    /// * `artifact` - The artifact to upload
    /// * `chunk_size` - Optional chunk size in bytes (default: 1MB)
    /// * `max_retries` - Optional maximum number of retries for expired upload URLs (default: 2)
    pub fn spawn_upload_artifact(
        &self,
        artifact: ArtifactLocal,
        chunk_size: Option<usize>,
        max_retries: Option<u32>,
    ) -> Result<
        (
            tokio::task::JoinHandle<Result<(ArtifactLocal, String)>>,
            tokio::sync::broadcast::Receiver<UploadProgress>,
        ),
        Error,
    > {
        let storage_client = self.storage_client.as_ref().ok_or_else(|| {
            Error::msg("Storage client not configured. Call with_storage() first.")
        })?;

        Ok(storage_client.spawn_upload_artifact(artifact, chunk_size, max_retries))
    }

    /// Get artifacts that need upload URLs
    pub fn get_artifacts_needing_upload_urls(&self) -> Result<Vec<ArtifactLocal>, Error> {
        let storage_client = self.storage_client.as_ref().ok_or_else(|| {
            Error::msg("Storage client not configured. Call with_storage() first.")
        })?;

        // Get all artifacts from database
        let r = self.database.r_transaction()?;
        let mut all_artifacts = Vec::new();

        for raw_artifact in r.scan().primary::<ArtifactLocal>()?.all()? {
            if let Ok(artifact) = raw_artifact {
                all_artifacts.push(artifact);
            }
        }

        Ok(storage_client.get_artifacts_needing_urls(&all_artifacts))
    }

    /// Get all artifacts from the database
    pub fn get_all_artifacts(&self) -> Result<Vec<ArtifactLocal>, Error> {
        let r = self.database.r_transaction()?;
        let mut all_artifacts = Vec::new();

        for raw_artifact in r.scan().primary::<ArtifactLocal>()?.all()? {
            if let Ok(artifact) = raw_artifact {
                all_artifacts.push(artifact);
            }
        }

        Ok(all_artifacts)
    }

    /// Get artifacts that have upload URLs but haven't been uploaded yet
    pub fn get_artifacts_ready_for_upload(&self) -> Result<Vec<ArtifactLocal>, Error> {
        let r = self.database.r_transaction()?;
        let mut ready_artifacts = Vec::new();

        for raw_artifact in r.scan().primary::<ArtifactLocal>()?.all()? {
            if let Ok(artifact) = raw_artifact {
                if !artifact.has_uploaded_file_to_storage && artifact.upload_url.is_some() {
                    ready_artifacts.push(artifact);
                }
            }
        }

        Ok(ready_artifacts)
    }

    /// Get artifacts by their upload status
    pub fn get_artifacts_by_upload_status(
        &self,
        uploaded: bool,
    ) -> Result<Vec<ArtifactLocal>, Error> {
        let r = self.database.r_transaction()?;
        let mut filtered_artifacts = Vec::new();

        for raw_artifact in r.scan().primary::<ArtifactLocal>()?.all()? {
            if let Ok(artifact) = raw_artifact {
                if artifact.has_uploaded_file_to_storage == uploaded {
                    filtered_artifacts.push(artifact);
                }
            }
        }

        Ok(filtered_artifacts)
    }

    /// Get a specific artifact by its local ID
    pub fn get_artifact_by_local_id(&self, local_id: &str) -> Result<Option<ArtifactLocal>, Error> {
        let r = self.database.r_transaction()?;

        for raw_artifact in r.scan().primary::<ArtifactLocal>()?.all()? {
            if let Ok(artifact) = raw_artifact {
                if artifact.id_local.as_deref() == Some(local_id) {
                    return Ok(Some(artifact));
                }
            }
        }

        Ok(None)
    }

    /// Updates all descendants of a session with the new remote session ID
    fn update_session_descendants(
        &mut self,
        session_local_id: &str,
        new_remote_session_id: i64,
    ) -> Result<(), Error> {
        // Update connectivity entries
        self.update_connectivity_session_id(session_local_id, new_remote_session_id)?;

        // Update events that belong to this session
        self.update_events_session_id(session_local_id, new_remote_session_id)?;

        // Update operators that belong to this session
        self.update_operators_session_id(session_local_id, new_remote_session_id)?;

        tracing::info!(
            "Updated descendants for session {} with remote ID {}",
            session_local_id,
            new_remote_session_id
        );
        Ok(())
    }

    /// Updates connectivity entries to reference the new remote session ID
    fn update_connectivity_session_id(
        &mut self,
        session_local_id: &str,
        new_remote_session_id: i64,
    ) -> Result<(), Error> {
        let r = self.database.r_transaction()?;

        // Find all connectivity entries that reference this session's local ID
        let mut connectivity_to_update = Vec::new();
        for raw_connectivity in r.scan().primary::<ConnectivityLocal>()?.all()? {
            if let Ok(mut connectivity) = raw_connectivity {
                if connectivity.ancestor_id_local.as_deref() == Some(session_local_id) {
                    // Validate: if session_id is already set, ensure it matches
                    if connectivity.session_id.is_some()
                        && connectivity.session_id != Some(new_remote_session_id)
                    {
                        tracing::warn!(
                            "Connectivity {} has conflicting session_id {:?} vs expected {}",
                            connectivity.id_local.as_deref().unwrap_or("unknown"),
                            connectivity.session_id,
                            new_remote_session_id
                        );
                        continue; // Skip this entry to prevent wrong linkage
                    }

                    // Convert to hybrid connectivity: keep device_id and add session_id
                    connectivity.session_id = Some(new_remote_session_id);
                    // Ensure device_id is set if not already present
                    if connectivity.device_id.is_none() {
                        // This should not happen in v2, but handle gracefully
                        tracing::warn!(
                            "Connectivity {} missing device_id, this may cause RLS issues",
                            connectivity.id_local.as_deref().unwrap_or("unknown")
                        );
                    }
                    // Keep ancestor_id_local as metadata showing original relationship
                    connectivity_to_update.push(connectivity);
                }
            }
        }

        drop(r); // Close read transaction before opening write transaction

        if !connectivity_to_update.is_empty() {
            let count = connectivity_to_update.len();
            self.upsert_items(connectivity_to_update)?;
            tracing::debug!(
                "Updated {} connectivity entries for session {}",
                count,
                session_local_id
            );
        }

        Ok(())
    }

    /// Updates events to reference the new remote session ID
    fn update_events_session_id(
        &mut self,
        session_local_id: &str,
        new_remote_session_id: i64,
    ) -> Result<(), Error> {
        let r = self.database.r_transaction()?;

        // Find all events that reference this session's local ID
        let mut events_to_update = Vec::new();
        for raw_event in r.scan().primary::<EventLocal>()?.all()? {
            if let Ok(mut event) = raw_event {
                if event.ancestor_id_local.as_deref() == Some(session_local_id) {
                    // Validate: if session_id is already set, ensure it matches
                    if let Some(existing_session_id) = event.session_id {
                        if existing_session_id != new_remote_session_id {
                            tracing::warn!(
                                "Event {} has conflicting session_id {} vs expected {}",
                                event.id_local.as_deref().unwrap_or("unknown"),
                                existing_session_id,
                                new_remote_session_id
                            );
                            continue; // Skip this entry to prevent wrong linkage
                        }
                    }

                    event.session_id = Some(new_remote_session_id);
                    // Keep ancestor_id_local as metadata showing original relationship
                    events_to_update.push(event);
                }
            }
        }

        drop(r); // Close read transaction before opening write transaction

        if !events_to_update.is_empty() {
            let count = events_to_update.len();
            self.upsert_items(events_to_update)?;
            tracing::debug!("Updated {} events for session {}", count, session_local_id);
        }

        Ok(())
    }

    /// Updates all descendants of an event with the new remote event ID
    fn update_event_descendants(
        &mut self,
        event_local_id: &str,
        new_remote_event_id: i64,
    ) -> Result<(), Error> {
        // Update tags that belong to this event
        self.update_tags_event_id(event_local_id, new_remote_event_id)?;

        tracing::info!(
            "Updated descendants for event {} with remote ID {}",
            event_local_id,
            new_remote_event_id
        );
        Ok(())
    }

    /// Updates tags to reference the new remote event ID
    fn update_tags_event_id(
        &mut self,
        event_local_id: &str,
        new_remote_event_id: i64,
    ) -> Result<(), Error> {
        let r = self.database.r_transaction()?;

        // Find all tags that reference this event's local ID
        let mut tags_to_update = Vec::new();
        for raw_tag in r.scan().primary::<TagLocal>()?.all()? {
            if let Ok(mut tag) = raw_tag {
                if tag.ancestor_id_local.as_deref() == Some(event_local_id) {
                    if let Some(eid) = tag.event_id {
                        if eid != new_remote_event_id {
                            tracing::warn!(
                                "Tag {} has conflicting event_id {} vs expected {}",
                                tag.id_local.as_deref().unwrap_or("unknown"),
                                eid,
                                new_remote_event_id
                            );
                            continue;
                        }
                    }
                    tag.event_id = Some(new_remote_event_id);
                    // Keep ancestor_id_local as metadata showing original relationship
                    tags_to_update.push(tag);
                }
            }
        }

        drop(r); // Close read transaction before opening write transaction

        if !tags_to_update.is_empty() {
            let count = tags_to_update.len();
            self.upsert_items(tags_to_update)?;
            tracing::debug!("Updated {} tags for event {}", count, event_local_id);
        }

        Ok(())
    }

    /// Updates operators to reference the new remote session ID
    fn update_operators_session_id(
        &mut self,
        session_local_id: &str,
        new_remote_session_id: i64,
    ) -> Result<(), Error> {
        let r = self.database.r_transaction()?;

        // Find all operators that reference this session's local ID
        let mut operators_to_update = Vec::new();
        for raw_operator in r.scan().primary::<data::v2::OperatorLocal>()?.all()? {
            if let Ok(mut operator) = raw_operator {
                if operator.ancestor_id_local.as_deref() == Some(session_local_id) {
                    // Validate: if session_id is already set, ensure it matches
                    if let Some(existing_session_id) = operator.session_id {
                        if existing_session_id != new_remote_session_id {
                            tracing::warn!(
                                "Operator {} has conflicting session_id {} vs expected {}",
                                operator.id_local.as_deref().unwrap_or("unknown"),
                                existing_session_id,
                                new_remote_session_id
                            );
                            continue; // Skip this entry to prevent wrong linkage
                        }
                    }

                    operator.session_id = Some(new_remote_session_id);
                    // Keep ancestor_id_local as metadata showing original relationship
                    operators_to_update.push(operator);
                }
            }
        }

        drop(r); // Close read transaction before opening write transaction

        if !operators_to_update.is_empty() {
            let count = operators_to_update.len();
            self.upsert_items(operators_to_update)?;
            tracing::debug!(
                "Updated {} operators for session {}",
                count,
                session_local_id
            );
        }

        Ok(())
    }

    /// Validates that a session exists in local database with given local_id and remote_id
    fn validate_session_exists(&self, local_id: &str, remote_id: i64) -> Result<bool, Error> {
        let r = self.database.r_transaction()?;

        for raw_session in r.scan().primary::<SessionLocal>()?.all()? {
            if let Ok(session) = raw_session {
                if session.id_local.as_deref() == Some(local_id) && session.id == Some(remote_id) {
                    return Ok(true);
                }
            }
        }

        Ok(false)
    }

    /// Validates that an event exists in local database with given local_id and remote_id
    fn validate_event_exists(&self, local_id: &str, remote_id: i64) -> Result<bool, Error> {
        let r = self.database.r_transaction()?;

        for raw_event in r.scan().primary::<EventLocal>()?.all()? {
            if let Ok(event) = raw_event {
                if event.id_local.as_deref() == Some(local_id) && event.id == Some(remote_id) {
                    return Ok(true);
                }
            }
        }

        Ok(false)
    }

    /// Log information about each table in the local database
    /// Displays table name, count, and all rows for each table
    pub fn log(&self) -> Result<(), Error> {
        println!("=== Database Tables Log ===");

        // Log SessionLocal table
        self.log_table::<SessionLocal>("SessionLocal")?;

        // Log EventLocal table
        self.log_table::<EventLocal>("EventLocal")?;

        // Log TagLocal table
        self.log_table::<TagLocal>("TagLocal")?;

        // Log v1 ConnectivityLocal table
        self.log_table::<data::v1::ConnectivityLocal>("ConnectivityLocal (v1)")?;

        // Log v2 ConnectivityLocal table
        self.log_table::<data::v2::ConnectivityLocal>("ConnectivityLocal (v2)")?;

        // Log Operator table
        self.log_table::<data::v2::OperatorLocal>("OperatorLocal")?;

        println!("=== End Database Tables Log ===");
        Ok(())
    }

    /// Helper method to log a specific table
    fn log_table<T: ToInput + std::fmt::Debug>(&self, table_name: &str) -> Result<(), Error> {
        let r = self.database.r_transaction()?;
        let count = r.len().primary::<T>().unwrap_or(0);

        println!("\n--- Table: {} ---", table_name);
        println!("Count: {}", count);

        if count > 0 {
            println!("Rows:");
            let mut row_num = 1;
            for raw_item in r.scan().primary::<T>()?.all()? {
                match raw_item {
                    Ok(item) => {
                        println!("  {}: {:?}", row_num, item);
                        row_num += 1;
                    }
                    Err(e) => {
                        println!("  Error reading row {}: {:?}", row_num, e);
                        row_num += 1;
                    }
                }
            }
        } else {
            println!("No rows found");
        }

        Ok(())
    }

    /// Checks if an error indicates a permanent failure that will never succeed
    fn is_critical_error(error_message: &str) -> bool {
        let error_lower = error_message.to_lowercase();
        error_lower.contains("parse error - invalid geometry")
            || error_lower.contains("new row violates row-level security policy")
            || error_lower.contains("all object keys must match")
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{
        db_client::DatabaseConfig,
        models::{AncestorLocal, MediaType, SessionLocal, TagObservationType},
    };

    use crate::constants::env;
    use serde_json;
    use tempfile::tempdir;

    fn test_user_id() -> String {
        std::env::var(env::SCOUT_TEST_USER_ID).unwrap_or_else(|_| {
            panic!(
                "{} must be set to a valid public.users id (see env.example)",
                env::SCOUT_TEST_USER_ID
            )
        })
    }

    fn setup_test_env() {
        dotenv::dotenv().ok();

        // Check for required environment variables and panic if missing
        let missing_vars = vec![
            (
                env::SCOUT_DEVICE_API_KEY,
                std::env::var(env::SCOUT_DEVICE_API_KEY).is_err(),
            ),
            (
                env::SCOUT_DATABASE_REST_URL,
                std::env::var(env::SCOUT_DATABASE_REST_URL).is_err(),
            ),
            (
                env::SCOUT_DEVICE_ID,
                std::env::var(env::SCOUT_DEVICE_ID).is_err(),
            ),
            (
                env::SCOUT_HERD_ID,
                std::env::var(env::SCOUT_HERD_ID).is_err(),
            ),
            (
                env::SCOUT_TEST_USER_ID,
                std::env::var(env::SCOUT_TEST_USER_ID).is_err(),
            ),
        ];

        let missing: Vec<&str> = missing_vars
            .into_iter()
            .filter(|(_, is_missing)| *is_missing)
            .map(|(name, _)| name)
            .collect();

        if !missing.is_empty() {
            panic!(
                "❌ Missing required environment variables: {}. Please check your .env file.",
                missing.join(", ")
            );
        }
    }

    fn create_test_sync_engine() -> Result<SyncEngine> {
        setup_test_env();

        let temp_dir = tempdir()?;
        let timestamp = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos();
        let db_path = temp_dir
            .path()
            .join(format!("test_{}.db", timestamp))
            .to_string_lossy()
            .to_string();
        let database_config = DatabaseConfig::from_env()
            .map_err(|e| Error::msg(format!("System time error: {}", e)))?;
        let scout_client = ScoutClient::new(database_config);
        let sync_engine = SyncEngine::new(scout_client, db_path, None, false)?;

        // Initialize database with a simple transaction to ensure it's properly set up
        {
            let rw = sync_engine.database.rw_transaction()?;
            rw.commit()?;
        }

        Ok(sync_engine)
    }

    async fn create_test_sync_engine_with_identification() -> Result<SyncEngine> {
        setup_test_env();

        // Require API key - tests should fail if not provided
        let _api_key = std::env::var(env::SCOUT_DEVICE_API_KEY)
            .expect("SCOUT_DEVICE_API_KEY environment variable is required for sync tests");

        let temp_dir = tempdir()?;
        let timestamp = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos();
        let db_path = temp_dir
            .path()
            .join(format!("test_{}.db", timestamp))
            .to_string_lossy()
            .to_string();

        // Create and identify scout client - MUST succeed for test to be valid
        let config_db = DatabaseConfig::from_env()?;
        let mut scout_client = ScoutClient::new(config_db);
        scout_client.identify().await.expect(
            "Client identification failed - check SCOUT_DEVICE_API_KEY and database connection",
        );

        let sync_engine = SyncEngine::new(scout_client, db_path, None, false)?;

        // Initialize database with a simple transaction to ensure it's properly set up
        {
            let rw = sync_engine.database.rw_transaction()?;
            rw.commit()?;
        }

        Ok(sync_engine)
    }

    #[tokio::test]
    async fn test_upsert_sessions_and_count() -> Result<()> {
        let mut sync_engine = create_test_sync_engine()?;

        // Check initial count is 0
        let initial_count = sync_engine.get_table_count::<SessionLocal>()?;
        assert_eq!(initial_count, 0);

        let device_id = std::env::var(env::SCOUT_DEVICE_ID)
            .expect("SCOUT_DEVICE_ID required")
            .parse()
            .expect("SCOUT_DEVICE_ID must be valid integer");

        // Create test sessions with proper data
        let mut session1 = SessionLocal::default();
        session1.set_id_local("test_session_1".to_string());
        session1.device_id = device_id;
        session1.timestamp_start = "2023-01-01T00:00:00Z".to_string();

        let mut session2 = SessionLocal::default();
        session2.set_id_local("test_session_2".to_string());
        session2.device_id = device_id;
        session2.timestamp_start = "2023-01-01T01:00:00Z".to_string();

        let mut session3 = SessionLocal::default();
        session3.set_id_local("test_session_3".to_string());
        session3.device_id = device_id;
        session3.timestamp_start = "2023-01-01T02:00:00Z".to_string();

        let sessions = vec![session1, session2, session3];

        // Upsert the sessions
        sync_engine.upsert_items(sessions)?;

        // Check that count is now 3
        let final_count = sync_engine.get_table_count::<SessionLocal>()?;
        assert_eq!(final_count, 3);

        Ok(())
    }

    #[tokio::test]
    async fn test_get_batch() -> Result<()> {
        let mut sync_engine = create_test_sync_engine()?;

        // Create a session with no remote ID (should go to insert batch)
        let device_id = std::env::var(env::SCOUT_DEVICE_ID)
            .expect("SCOUT_DEVICE_ID required")
            .parse()
            .expect("SCOUT_DEVICE_ID must be valid integer");

        let mut session_1 = SessionLocal::default();
        session_1.set_id_local("test_session_1".to_string());
        session_1.device_id = device_id;
        session_1.timestamp_start = "2023-01-01T00:00:00Z".to_string();
        session_1.software_version = "1.0.0".to_string();

        sync_engine.upsert_items::<SessionLocal>(vec![session_1.clone()])?;

        // Verify the session was actually saved
        let count = sync_engine.get_table_count::<SessionLocal>()?;
        assert_eq!(count, 1);

        let batch = sync_engine
            .get_batch::<SessionLocal>(EnumSyncAction::Upsert, EnumSyncAction::Insert)?;

        // The session has no remote ID (id is None), so it should go to insert batch
        assert_eq!(batch.insert.len(), 1);
        assert_eq!(batch.upsert.len(), 0);

        Ok(())
    }

    #[tokio::test]
    async fn test_multiple_upsert_operations() -> Result<()> {
        let mut sync_engine = create_test_sync_engine()?;
        // Create two different sessions
        let device_id = std::env::var(env::SCOUT_DEVICE_ID)
            .expect("SCOUT_DEVICE_ID required")
            .parse()
            .expect("SCOUT_DEVICE_ID must be valid integer");

        let mut session_1 = SessionLocal::default();
        session_1.set_id_local("multi_test_session_1".to_string());
        session_1.device_id = device_id;
        session_1.timestamp_start = "2023-01-01T00:00:00Z".to_string();

        let mut session_2 = SessionLocal::default();
        session_2.set_id_local("multi_test_session_2".to_string());
        session_2.device_id = device_id;
        session_2.timestamp_start = "2023-01-01T01:00:00Z".to_string();

        sync_engine.upsert_items(vec![session_1])?;
        let count_after_first = sync_engine.get_table_count::<SessionLocal>()?;
        assert_eq!(count_after_first, 1);

        // Upsert second session
        sync_engine.upsert_items(vec![session_2])?;
        let count_after_second = sync_engine.get_table_count::<SessionLocal>()?;
        // Count should be 2 since we have two different sessions
        assert_eq!(count_after_second, 2);
        Ok(())
    }

    #[tokio::test]
    async fn test_flush_sessions_without_remote() -> Result<()> {
        let mut sync_engine = create_test_sync_engine_with_identification().await?;

        // Create sessions without remote IDs (they should be inserted to remote)
        let device_id = std::env::var(env::SCOUT_DEVICE_ID)
            .expect("SCOUT_DEVICE_ID required")
            .parse()
            .expect("SCOUT_DEVICE_ID must be valid integer");

        let mut session_1 = SessionLocal::default();
        session_1.set_id_local("flush_test_session_1".to_string());
        session_1.device_id = device_id;
        session_1.timestamp_start = "2023-01-01T10:00:00Z".to_string();
        session_1.software_version = "sync_unit_test_flush_sessions_without_remote_0".to_string();
        session_1.altitude_max = 100.0;
        session_1.altitude_min = 50.0;
        session_1.altitude_average = 75.0;
        session_1.velocity_max = 25.0;
        session_1.velocity_min = 10.0;
        session_1.velocity_average = 15.0;
        session_1.distance_total = 1000.0;
        session_1.distance_max_from_start = 500.0;

        let mut session_2 = SessionLocal::default();
        session_2.set_id_local("flush_test_session_2".to_string());
        session_2.device_id = device_id;
        session_2.timestamp_start = "2023-01-01T11:00:00Z".to_string();
        session_2.software_version = "sync_unit_test_flush_sessions_without_remote_1".to_string();
        session_2.altitude_max = 120.0;
        session_2.altitude_min = 60.0;
        session_2.altitude_average = 90.0;
        session_2.velocity_max = 30.0;
        session_2.velocity_min = 15.0;
        session_2.velocity_average = 20.0;
        session_2.distance_total = 1200.0;
        session_2.distance_max_from_start = 600.0;

        // Insert sessions locally (no remote ID yet)
        sync_engine.upsert_items(vec![session_1, session_2])?;

        // Verify sessions are in local database
        let count_before = sync_engine.get_table_count::<SessionLocal>()?;
        assert_eq!(count_before, 2);

        // Flush MUST succeed - test should fail if remote sync doesn't work
        println!("🚀 Starting session flush to remote...");
        let flush_result = sync_engine.flush().await;

        match &flush_result {
            Ok(_) => println!("✅ Session flush completed successfully!"),
            Err(e) => {
                println!("❌ Session flush failed: {}", e);
                panic!(
                    "Flush operation must succeed - check database connection and API key: {}",
                    e
                );
            }
        }

        flush_result?;

        // Verify sessions are still in database after successful sync
        let count_after = sync_engine.get_table_count::<SessionLocal>()?;
        assert_eq!(count_after, 2);

        // Verify ALL sessions received remote IDs from server
        let r = sync_engine.database.r_transaction()?;
        let mut sessions_with_remote_ids = 0;
        for raw_session in r.scan().primary::<SessionLocal>()?.all()? {
            if let Ok(session) = raw_session {
                if session.id.is_some() {
                    sessions_with_remote_ids += 1;
                }
            }
        }

        // STRICT: All sessions must have remote IDs after successful flush
        assert_eq!(
            sessions_with_remote_ids, 2,
            "All sessions must have remote IDs after successful flush to remote database"
        );

        Ok(())
    }

    #[tokio::test]
    async fn test_flush_with_descendant_updates() -> Result<()> {
        let mut sync_engine = create_test_sync_engine_with_identification().await?;

        let device_id = std::env::var(env::SCOUT_DEVICE_ID)
            .expect("SCOUT_DEVICE_ID required")
            .parse()
            .expect("SCOUT_DEVICE_ID must be valid integer");

        // Create a session without remote ID (will be inserted to remote)
        let mut session = SessionLocal::default();
        session.set_id_local("test_session_with_descendants".to_string());
        session.device_id = device_id;
        session.timestamp_start = "2023-01-01T10:00:00Z".to_string();
        session.software_version = "sync_unit_test_flush_with_descendant_updates_0".to_string();
        session.altitude_max = 100.0;
        session.altitude_min = 50.0;
        session.altitude_average = 75.0;
        session.velocity_max = 25.0;
        session.velocity_min = 10.0;
        session.velocity_average = 15.0;
        session.distance_total = 1000.0;
        session.distance_max_from_start = 500.0;

        // Create connectivity entry that references this session's local ID
        let mut connectivity = ConnectivityLocal::default();
        connectivity.set_id_local("test_connectivity_1".to_string());
        connectivity.session_id = None; // Use device-based connectivity for initial sync
        connectivity.device_id = Some(device_id); // Reference the actual device ID
        connectivity.set_ancestor_id_local("test_session_with_descendants".to_string());
        connectivity.timestamp_start = "2023-01-01T10:05:00Z".to_string();
        connectivity.signal = Some(-70.0);
        connectivity.noise = Some(-90.0);
        connectivity.altitude = 100.0;
        connectivity.heading = 0.0;
        connectivity.location = Some("POINT(-155.15393 19.754824)".to_string());
        connectivity.h14_index = "h14".to_string();
        connectivity.h13_index = "h13".to_string();
        connectivity.h12_index = "h12".to_string();
        connectivity.h11_index = "h11".to_string();

        // Create event that references this session's local ID
        let mut event = EventLocal::default();
        event.set_id_local("test_event_1".to_string());
        event.device_id = device_id;
        event.session_id = None; // Will be updated after session gets remote ID
        event.set_ancestor_id_local("test_session_with_descendants".to_string());
        event.timestamp_observation = "2023-01-01T10:10:00Z".to_string();
        event.message = Some("Test event".to_string());
        event.altitude = 100.0;
        event.heading = 0.0;
        event.media_type = MediaType::Image;

        // Insert all items locally
        sync_engine.upsert_items(vec![session])?;
        sync_engine.upsert_items(vec![connectivity])?;
        sync_engine.upsert_items(vec![event])?;

        // Verify initial state
        let initial_session_count = sync_engine.get_table_count::<SessionLocal>()?;
        let initial_connectivity_count = sync_engine.get_table_count::<ConnectivityLocal>()?;
        let initial_event_count = sync_engine.get_table_count::<EventLocal>()?;
        assert_eq!(initial_session_count, 1);
        assert_eq!(initial_connectivity_count, 1);
        assert_eq!(initial_event_count, 1);

        // Flush MUST succeed - test should fail if remote sync doesn't work
        println!("🚀 Starting descendant update flush to remote...");
        let flush_result = sync_engine.flush().await;

        match &flush_result {
            Ok(_) => println!("✅ Descendant update flush completed successfully!"),
            Err(e) => {
                println!("❌ Descendant update flush failed: {}", e);
                panic!(
                    "Flush operation must succeed - check database connection and API key: {}",
                    e
                );
            }
        }

        flush_result?;

        // Verify all items are still in database after successful sync
        let final_session_count = sync_engine.get_table_count::<SessionLocal>()?;
        let final_connectivity_count = sync_engine.get_table_count::<ConnectivityLocal>()?;
        let final_event_count = sync_engine.get_table_count::<EventLocal>()?;
        assert_eq!(final_session_count, 1);
        assert_eq!(final_connectivity_count, 1);
        assert_eq!(final_event_count, 1);

        // Verify that items received remote IDs and relationships were updated
        let r = sync_engine.database.r_transaction()?;

        // Session MUST have remote ID after successful flush
        let mut session_remote_id = None;
        for raw_session in r.scan().primary::<SessionLocal>()?.all()? {
            if let Ok(session) = raw_session {
                if session.id_local.as_deref() == Some("test_session_with_descendants") {
                    session_remote_id = session.id;
                    break;
                }
            }
        }
        assert!(
            session_remote_id.is_some(),
            "Session must have remote ID after successful flush to remote database"
        );

        let session_id = session_remote_id.unwrap();

        // Verify connectivity entries reference the session's remote ID
        for raw_connectivity in r.scan().primary::<ConnectivityLocal>()?.all()? {
            if let Ok(connectivity) = raw_connectivity {
                if connectivity.ancestor_id_local.as_deref()
                    == Some("test_session_with_descendants")
                {
                    assert_eq!(
                        connectivity.device_id,
                        Some(device_id),
                        "Connectivity must reference the correct device ID"
                    );
                    assert_eq!(
                        connectivity.session_id,
                        Some(session_id),
                        "Connectivity must reference session's remote ID after flush (hybrid mode)"
                    );
                }
            }
        }

        // Verify events reference the session's remote ID
        for raw_event in r.scan().primary::<EventLocal>()?.all()? {
            if let Ok(event) = raw_event {
                if event.ancestor_id_local.as_deref() == Some("test_session_with_descendants") {
                    assert_eq!(
                        event.session_id,
                        Some(session_id),
                        "Event must reference session's remote ID after flush"
                    );
                }
            }
        }

        Ok(())
    }

    #[tokio::test]
    async fn test_clean_completed_sessions() -> Result<()> {
        let mut sync_engine = create_test_sync_engine()?;

        // Create a completed session (with timestamp_end)
        let device_id = std::env::var(env::SCOUT_DEVICE_ID)
            .expect("SCOUT_DEVICE_ID required")
            .parse()
            .expect("SCOUT_DEVICE_ID must be valid integer");

        let mut completed_session = SessionLocal::default();
        completed_session.set_id_local("completed_session".to_string());
        completed_session.id = Some(12345); // Has remote ID
        completed_session.device_id = device_id;
        completed_session.timestamp_start = "2023-01-01T10:00:00Z".to_string();
        completed_session.timestamp_end = Some("2023-01-01T11:00:00Z".to_string()); // Completed
        completed_session.software_version = "1.0.0".to_string();
        completed_session.altitude_max = 100.0;
        completed_session.altitude_min = 50.0;
        completed_session.altitude_average = 75.0;
        completed_session.velocity_max = 25.0;
        completed_session.velocity_min = 10.0;
        completed_session.velocity_average = 15.0;
        completed_session.distance_total = 1000.0;
        completed_session.distance_max_from_start = 500.0;

        // Create an incomplete session (no timestamp_end)
        let mut incomplete_session = SessionLocal::default();
        incomplete_session.set_id_local("incomplete_session".to_string());
        incomplete_session.id = Some(23456); // Has remote ID
        incomplete_session.device_id = device_id;
        incomplete_session.timestamp_start = "2023-01-01T12:00:00Z".to_string();
        // No timestamp_end - should not be cleaned
        incomplete_session.software_version = "1.0.0".to_string();
        incomplete_session.altitude_max = 120.0;
        incomplete_session.altitude_min = 60.0;
        incomplete_session.altitude_average = 90.0;
        incomplete_session.velocity_max = 30.0;
        incomplete_session.velocity_min = 15.0;
        incomplete_session.velocity_average = 22.0;
        incomplete_session.distance_total = 1200.0;
        incomplete_session.distance_max_from_start = 600.0;

        // Create descendants for completed session
        let mut completed_connectivity = ConnectivityLocal::default();
        completed_connectivity.set_id_local("completed_connectivity".to_string());
        completed_connectivity.id = Some(34567); // Has remote ID
        completed_connectivity.session_id = None; // Use device-based connectivity
        completed_connectivity.device_id = Some(device_id);
        completed_connectivity.set_ancestor_id_local("completed_session".to_string());
        completed_connectivity.timestamp_start = "2023-01-01T10:05:00Z".to_string();
        completed_connectivity.signal = Some(-70.0);
        completed_connectivity.noise = Some(-90.0);
        completed_connectivity.altitude = 100.0;
        completed_connectivity.heading = 0.0;
        completed_connectivity.location = Some("POINT(-155.15393 19.754824)".to_string());
        completed_connectivity.h14_index = "h14".to_string();
        completed_connectivity.h13_index = "h13".to_string();
        completed_connectivity.h12_index = "h12".to_string();
        completed_connectivity.h11_index = "h11".to_string();

        let mut completed_event = EventLocal::default();
        completed_event.set_id_local("completed_event".to_string());
        completed_event.id = Some(45678); // Has remote ID
        completed_event.device_id = 1;
        completed_event.session_id = Some(12345);
        completed_event.set_ancestor_id_local("completed_session".to_string());
        completed_event.timestamp_observation = "2023-01-01T10:15:00Z".to_string();
        completed_event.message = Some("Completed event".to_string());
        completed_event.altitude = 100.0;
        completed_event.heading = 0.0;
        completed_event.media_type = MediaType::Image;

        let mut completed_tag = TagLocal::default();
        completed_tag.set_id_local("completed_tag".to_string());
        completed_tag.id = Some(56789); // Has remote ID
        completed_tag.x = 100.0;
        completed_tag.y = 200.0;
        completed_tag.width = 50.0;
        completed_tag.height = 75.0;
        completed_tag.conf = 0.95;
        completed_tag.observation_type = crate::models::TagObservationType::Auto;
        completed_tag.event_id = Some(45678);
        completed_tag.set_ancestor_id_local("completed_event".to_string());
        completed_tag.class = "test_animal".to_string();

        let mut completed_operator = data::v2::OperatorLocal::default();
        completed_operator.set_id_local("completed_operator".to_string());
        completed_operator.id = Some(67890); // Has remote ID
        completed_operator.session_id = Some(12345);
        completed_operator.set_ancestor_id_local("completed_session".to_string());
        completed_operator.user_id = test_user_id();
        completed_operator.action = "test_clean_action".to_string();
        completed_operator.timestamp = Some("2023-01-01T10:20:00Z".to_string());

        // Insert all entities
        sync_engine.upsert_items(vec![completed_session, incomplete_session])?;
        sync_engine.upsert_items(vec![completed_connectivity])?;
        sync_engine.upsert_items(vec![completed_event])?;
        sync_engine.upsert_items(vec![completed_tag])?;
        sync_engine.upsert_items(vec![completed_operator])?;

        // Verify initial state
        assert_eq!(sync_engine.get_table_count::<SessionLocal>()?, 2);
        assert_eq!(sync_engine.get_table_count::<ConnectivityLocal>()?, 1);
        assert_eq!(sync_engine.get_table_count::<EventLocal>()?, 1);
        assert_eq!(sync_engine.get_table_count::<TagLocal>()?, 1);
        assert_eq!(sync_engine.get_table_count::<data::v2::OperatorLocal>()?, 1);

        // Run clean operation
        sync_engine.clean().await?;

        // Verify completed session and descendants are removed
        assert_eq!(sync_engine.get_table_count::<SessionLocal>()?, 1); // Only incomplete remains
        assert_eq!(sync_engine.get_table_count::<ConnectivityLocal>()?, 0); // Removed
        assert_eq!(sync_engine.get_table_count::<EventLocal>()?, 0); // Removed
        assert_eq!(sync_engine.get_table_count::<TagLocal>()?, 0); // Removed
        assert_eq!(sync_engine.get_table_count::<data::v2::OperatorLocal>()?, 0); // Removed

        // Verify the remaining session is the incomplete one
        let r = sync_engine.database.r_transaction()?;
        let remaining_sessions: Vec<SessionLocal> = r
            .scan()
            .primary::<SessionLocal>()?
            .all()?
            .into_iter()
            .collect::<Result<Vec<_>, _>>()?;
        assert_eq!(remaining_sessions.len(), 1);
        assert_eq!(
            remaining_sessions[0].id_local.as_deref(),
            Some("incomplete_session")
        );
        assert!(remaining_sessions[0].timestamp_end.is_none());

        Ok(())
    }

    #[tokio::test]
    async fn test_flush_database_to_remote() -> Result<()> {
        let mut sync_engine = create_test_sync_engine_with_identification().await?;

        // Print diagnostic information
        println!("🔍 Testing full database flush to remote...");
        if let Ok(api_key) = std::env::var(env::SCOUT_DEVICE_API_KEY) {
            println!(
                "📡 Using API key: {}...",
                &api_key[..std::cmp::min(api_key.len(), 8)]
            );
        }
        if let Ok(db_url) = std::env::var(env::SCOUT_DATABASE_REST_URL) {
            println!("🗄️ Database URL: {}", db_url);
        }

        let device_id = std::env::var(env::SCOUT_DEVICE_ID)
            .expect("SCOUT_DEVICE_ID required")
            .parse()
            .expect("SCOUT_DEVICE_ID must be valid integer");

        // Create a complete hierarchy: Session -> Connectivity + Event -> Tag + Operator
        let mut session = SessionLocal::default();
        session.set_id_local("flush_test_session".to_string());
        session.device_id = device_id;
        session.timestamp_start = "2023-01-01T10:00:00Z".to_string();
        session.software_version = "test_flush_database_to_remote".to_string();
        session.altitude_max = 100.0;
        session.altitude_min = 50.0;
        session.altitude_average = 75.0;
        session.velocity_max = 25.0;
        session.velocity_min = 10.0;
        session.velocity_average = 15.0;
        session.distance_total = 1000.0;
        session.distance_max_from_start = 500.0;

        let mut connectivity = ConnectivityLocal::default();
        connectivity.set_id_local("flush_test_connectivity".to_string());
        connectivity.set_ancestor_id_local("flush_test_session".to_string());
        connectivity.session_id = None; // Use device-based connectivity for initial sync
        connectivity.device_id = Some(device_id); // Reference the actual device ID
        connectivity.timestamp_start = "2023-01-01T10:05:00Z".to_string();
        connectivity.signal = Some(-70.0);
        connectivity.noise = Some(-90.0);
        connectivity.altitude = 100.0;
        connectivity.heading = 0.0;
        connectivity.location = Some("POINT(-155.15393 19.754824)".to_string());
        connectivity.h14_index = "h14".to_string();
        connectivity.h13_index = "h13".to_string();
        connectivity.h12_index = "h12".to_string();
        connectivity.h11_index = "h11".to_string();

        let mut event = EventLocal::default();
        event.set_id_local("flush_test_event".to_string());
        event.device_id = device_id;
        event.session_id = None; // Will be updated after session sync
        event.set_ancestor_id_local("flush_test_session".to_string());
        event.timestamp_observation = "2023-01-01T10:10:00Z".to_string();
        event.message = Some("Test flush event".to_string());
        event.altitude = 100.0;
        event.heading = 0.0;
        event.media_type = MediaType::Image;

        let mut tag = TagLocal::default();
        tag.set_id_local("flush_test_tag".to_string());
        tag.event_id = None; // Will be updated after event sync
        tag.set_ancestor_id_local("flush_test_event".to_string());
        tag.class = "test_flush_tag".to_string();
        tag.conf = 0.95;
        tag.observation_type = TagObservationType::Manual;

        let mut operator = data::v2::OperatorLocal::default();
        operator.set_id_local("flush_test_operator".to_string());
        operator.session_id = None; // Will be updated after session sync
        operator.set_ancestor_id_local("flush_test_session".to_string());
        operator.user_id = test_user_id();
        operator.action = "test_flush_action".to_string();
        operator.timestamp = Some("2023-01-01T10:15:00Z".to_string());

        // Insert all items locally
        sync_engine.upsert_items(vec![session])?;
        sync_engine.upsert_items(vec![connectivity])?;
        sync_engine.upsert_items(vec![event])?;
        sync_engine.upsert_items(vec![tag])?;
        sync_engine.upsert_items(vec![operator])?;

        // Verify initial counts
        assert_eq!(sync_engine.get_table_count::<SessionLocal>()?, 1);
        assert_eq!(sync_engine.get_table_count::<ConnectivityLocal>()?, 1);
        assert_eq!(sync_engine.get_table_count::<EventLocal>()?, 1);
        assert_eq!(sync_engine.get_table_count::<TagLocal>()?, 1);
        assert_eq!(sync_engine.get_table_count::<data::v2::OperatorLocal>()?, 1);

        // Perform full database flush to remote - MUST succeed
        println!("🚀 Starting full database flush...");
        let flush_result = sync_engine.flush().await;

        match &flush_result {
            Ok(_) => println!("✅ Flush completed successfully!"),
            Err(e) => {
                println!("❌ Flush failed with error: {}", e);
                println!(
                    "💡 This indicates the test is correctly trying to sync to remote database"
                );
                println!("🔧 Check: 1) Valid SCOUT_DEVICE_API_KEY 2) Database permissions 3) RLS policies");
                panic!(
                    "Full database flush must succeed - check database connection and API key: {}",
                    e
                );
            }
        }

        flush_result?;

        // Verify all items are still in database after successful sync
        assert_eq!(sync_engine.get_table_count::<SessionLocal>()?, 1);
        assert_eq!(sync_engine.get_table_count::<ConnectivityLocal>()?, 1);
        assert_eq!(sync_engine.get_table_count::<EventLocal>()?, 1);
        assert_eq!(sync_engine.get_table_count::<TagLocal>()?, 1);
        assert_eq!(sync_engine.get_table_count::<data::v2::OperatorLocal>()?, 1);

        // Verify the hierarchical sync worked correctly
        let r = sync_engine.database.r_transaction()?;

        // Session MUST have remote ID after successful flush
        let mut session_remote_id = None;
        for raw_session in r.scan().primary::<SessionLocal>()?.all()? {
            if let Ok(session) = raw_session {
                if session.id_local.as_deref() == Some("flush_test_session") {
                    session_remote_id = session.id;
                    break;
                }
            }
        }

        let session_id = session_remote_id
            .expect("Session must have remote ID after successful flush to remote database");

        // Verify connectivity references session remote ID
        // Verify connectivity was properly linked to both device and session (hybrid)
        for raw_connectivity in r.scan().primary::<ConnectivityLocal>()?.all()? {
            if let Ok(connectivity) = raw_connectivity {
                if connectivity.id_local.as_deref() == Some("flush_test_connectivity") {
                    assert_eq!(
                        connectivity.device_id,
                        Some(device_id),
                        "Connectivity must reference the correct device ID"
                    );
                    assert_eq!(
                        connectivity.session_id,
                        Some(session_id),
                        "Connectivity must reference session's remote ID after session sync"
                    );
                }
            }
        }

        // Verify event references session remote ID and has remote ID
        let mut event_remote_id = None;
        for raw_event in r.scan().primary::<EventLocal>()?.all()? {
            if let Ok(event) = raw_event {
                if event.id_local.as_deref() == Some("flush_test_event") {
                    assert_eq!(
                        event.session_id,
                        Some(session_id),
                        "Event must reference session's remote ID after flush"
                    );
                    event_remote_id = event.id;
                    break;
                }
            }
        }

        let event_id = event_remote_id
            .expect("Event must have remote ID after successful flush to remote database");

        // Verify tag references event remote ID and has remote ID
        for raw_tag in r.scan().primary::<TagLocal>()?.all()? {
            if let Ok(tag) = raw_tag {
                if tag.id_local.as_deref() == Some("flush_test_tag") {
                    assert_eq!(
                        tag.event_id,
                        Some(event_id),
                        "Tag must reference event's remote ID after flush"
                    );
                    assert!(
                        tag.id.is_some(),
                        "Tag must have remote ID after successful flush"
                    );
                }
            }
        }

        // Verify operator references session remote ID and has remote ID
        for raw_operator in r.scan().primary::<data::v2::OperatorLocal>()?.all()? {
            if let Ok(operator) = raw_operator {
                if operator.id_local.as_deref() == Some("flush_test_operator") {
                    assert_eq!(
                        operator.session_id,
                        Some(session_id),
                        "Operator must reference session's remote ID after flush"
                    );
                    assert!(
                        operator.id.is_some(),
                        "Operator must have remote ID after successful flush"
                    );
                }
            }
        }

        println!("✅ Full database flush to remote completed successfully!");
        println!("✅ Session synced with remote ID: {}", session_id);
        println!("✅ Event synced with remote ID: {}", event_id);
        println!("✅ Operator synced and linked to session!");
        println!("✅ All relationships updated correctly!");

        Ok(())
    }

    async fn create_test_sync_engine_with_invalid_credentials() -> Result<SyncEngine> {
        let temp_dir = tempdir()?;
        let timestamp = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos();
        let db_path = temp_dir
            .path()
            .join(format!("test_{}.db", timestamp))
            .to_string_lossy()
            .to_string();

        // Create client with invalid credentials - this should fail
        let invalid_config = DatabaseConfig {
            rest_url: "https://invalid.supabase.co/rest/v1".to_string(),
            scout_api_key: "invalid_api_key_12345".to_string(),
            supabase_api_key: "invalid_supabase_key".to_string(),
            service_role_key: None,
            use_service_key: false,
        };
        let mut scout_client = ScoutClient::new(invalid_config);
        scout_client.identify().await?; // This should fail

        let sync_engine = SyncEngine::new(scout_client, db_path, None, false)?;

        // Initialize database with a simple transaction to ensure it's properly set up
        {
            let rw = sync_engine.database.rw_transaction()?;
            rw.commit()?;
        }

        Ok(sync_engine)
    }

    #[tokio::test]
    async fn test_sync_requires_valid_credentials() -> Result<()> {
        println!("🔐 Testing sync failure with invalid credentials...");

        let result = create_test_sync_engine_with_invalid_credentials().await;

        match result {
            Ok(_) => {
                panic!("Sync engine creation should fail with invalid credentials");
            }
            Err(e) => {
                println!("✅ Correctly failed with invalid credentials: {}", e);
                println!("💡 This confirms the sync engine is properly validating credentials");
            }
        }

        Ok(())
    }

    #[tokio::test]
    async fn test_session_lifecycle_insert_update_flush_sequence() -> Result<()> {
        println!(
            "🔄 Testing session lifecycle: insert -> update -> flush -> record another -> flush"
        );
        let mut sync_engine = create_test_sync_engine_with_identification().await?;

        let device_id = std::env::var(env::SCOUT_DEVICE_ID)
            .expect("SCOUT_DEVICE_ID required")
            .parse()
            .expect("SCOUT_DEVICE_ID must be valid integer");

        // PHASE 1: Insert first session
        let mut session1 = SessionLocal::default();
        session1.set_id_local("lifecycle_session_1".to_string());
        session1.device_id = device_id;
        session1.timestamp_start = "2023-01-01T10:00:00Z".to_string();
        session1.software_version = "test_session_lifecycle_v1".to_string();
        session1.altitude_max = 100.0;
        session1.altitude_min = 50.0;
        session1.altitude_average = 75.0;
        session1.velocity_max = 25.0;
        session1.velocity_min = 10.0;
        session1.velocity_average = 15.0;
        session1.distance_total = 1000.0;
        session1.distance_max_from_start = 500.0;

        sync_engine.upsert_items(vec![session1.clone()])?;
        assert_eq!(sync_engine.get_table_count::<SessionLocal>()?, 1);
        println!("✅ Phase 1: First session inserted locally");

        // PHASE 2: Update the same session with new data (e.g., session in progress)
        session1.altitude_max = 150.0; // Updated max altitude
        session1.distance_total = 1500.0; // Updated distance
        session1.timestamp_end = None; // Still in progress

        sync_engine.upsert_items(vec![session1.clone()])?;
        assert_eq!(sync_engine.get_table_count::<SessionLocal>()?, 1); // Still just 1 session
        println!("✅ Phase 2: Session updated with new data");

        // PHASE 3: Flush the session to remote
        println!("🚀 Phase 3: Flushing first session to remote...");
        sync_engine.flush().await?;

        // Verify session got remote ID
        let r = sync_engine.database.r_transaction()?;
        let mut session1_remote_id = None;
        for raw_session in r.scan().primary::<SessionLocal>()?.all()? {
            if let Ok(session) = raw_session {
                if session.id_local.as_deref() == Some("lifecycle_session_1") {
                    session1_remote_id = session.id;
                    break;
                }
            }
        }
        assert!(
            session1_remote_id.is_some(),
            "First session must have remote ID after flush"
        );
        println!(
            "✅ Phase 3: First session flushed with remote ID: {:?}",
            session1_remote_id
        );

        // PHASE 4: Complete the first session
        session1.timestamp_end = Some("2023-01-01T11:30:00Z".to_string());
        session1.altitude_max = 175.0; // Final max altitude
        session1.distance_total = 2000.0; // Final distance

        sync_engine.upsert_items(vec![session1])?;
        println!("✅ Phase 4: First session marked as completed");

        // PHASE 5: Record a completely new session (simulating back-to-back usage)
        let mut session2 = SessionLocal::default();
        session2.set_id_local("lifecycle_session_2".to_string());
        session2.device_id = device_id;
        session2.timestamp_start = "2023-01-01T12:00:00Z".to_string();
        session2.software_version = "test_session_lifecycle_v2".to_string();
        session2.altitude_max = 200.0;
        session2.altitude_min = 80.0;
        session2.altitude_average = 140.0;
        session2.velocity_max = 35.0;
        session2.velocity_min = 20.0;
        session2.velocity_average = 25.0;
        session2.distance_total = 800.0;
        session2.distance_max_from_start = 400.0;

        sync_engine.upsert_items(vec![session2.clone()])?;
        assert_eq!(sync_engine.get_table_count::<SessionLocal>()?, 2); // Now have 2 sessions
        println!("✅ Phase 5: Second session inserted (back-to-back usage)");

        // PHASE 6: Add some events to second session before flushing
        let mut event_for_session2 = EventLocal::default();
        event_for_session2.set_id_local("lifecycle_event_session2".to_string());
        event_for_session2.device_id = device_id;
        event_for_session2.session_id = None; // Will be updated after session sync
        event_for_session2.set_ancestor_id_local("lifecycle_session_2".to_string());
        event_for_session2.timestamp_observation = "2023-01-01T12:15:00Z".to_string();
        event_for_session2.message = Some("Event during second session".to_string());
        event_for_session2.altitude = 150.0;
        event_for_session2.heading = 45.0;
        event_for_session2.media_type = MediaType::Video;

        sync_engine.upsert_items(vec![event_for_session2])?;
        assert_eq!(sync_engine.get_table_count::<EventLocal>()?, 1);
        println!("✅ Phase 6: Event added to second session");

        // PHASE 7: Final flush of everything (simulating critical sync point)
        println!("🚀 Phase 7: Final flush of all data...");
        sync_engine.flush().await?;

        // Verify final state
        assert_eq!(sync_engine.get_table_count::<SessionLocal>()?, 2);
        assert_eq!(sync_engine.get_table_count::<EventLocal>()?, 1);

        // Verify both sessions have remote IDs
        let r = sync_engine.database.r_transaction()?;
        let mut sessions_with_remote_ids = 0;
        let mut session2_remote_id = None;

        for raw_session in r.scan().primary::<SessionLocal>()?.all()? {
            if let Ok(session) = raw_session {
                if session.id.is_some() {
                    sessions_with_remote_ids += 1;
                    if session.id_local.as_deref() == Some("lifecycle_session_2") {
                        session2_remote_id = session.id;
                    }
                }
            }
        }

        assert_eq!(
            sessions_with_remote_ids, 2,
            "Both sessions must have remote IDs"
        );
        assert!(
            session2_remote_id.is_some(),
            "Second session must have remote ID"
        );

        // Verify event references second session's remote ID
        for raw_event in r.scan().primary::<EventLocal>()?.all()? {
            if let Ok(event) = raw_event {
                if event.id_local.as_deref() == Some("lifecycle_event_session2") {
                    assert_eq!(
                        event.session_id, session2_remote_id,
                        "Event must reference second session's remote ID"
                    );
                    assert!(event.id.is_some(), "Event must have remote ID");
                }
            }
        }

        println!("✅ Phase 7: Final state verified - all data synced with relationships intact");
        println!("🎉 Session lifecycle test completed successfully!");

        Ok(())
    }

    #[tokio::test]
    async fn test_session_update_during_recording_with_periodic_flush() -> Result<()> {
        let mut sync_engine = create_test_sync_engine_with_identification().await?;

        let device_id = std::env::var(env::SCOUT_DEVICE_ID)
            .expect("SCOUT_DEVICE_ID required")
            .parse()
            .expect("SCOUT_DEVICE_ID must be valid integer");

        // Start a new session
        let mut active_session = SessionLocal::default();
        active_session.set_id_local("live_recording_session".to_string());
        active_session.device_id = device_id;
        active_session.timestamp_start = "2023-01-01T14:00:00Z".to_string();
        active_session.software_version = "live_recording_test".to_string();
        active_session.altitude_max = 100.0;
        active_session.distance_total = 0.0;

        sync_engine.upsert_items(vec![active_session.clone()])?;

        // Update session during recording
        active_session.altitude_max = 120.0;
        active_session.distance_total = 300.0;
        sync_engine.upsert_items(vec![active_session.clone()])?;

        // Add connectivity data
        let mut connectivity = ConnectivityLocal::default();
        connectivity.set_id_local("live_conn_1".to_string());
        connectivity.set_ancestor_id_local("live_recording_session".to_string());
        connectivity.timestamp_start = "2023-01-01T14:10:00Z".to_string();
        connectivity.signal = Some(-68.0);
        connectivity.altitude = 120.0;
        connectivity.location = Some("POINT(-155.15393 19.754824)".to_string());
        connectivity.h14_index = "h14_live1".to_string();
        connectivity.h13_index = "h13_live1".to_string();
        connectivity.h12_index = "h12_live1".to_string();
        connectivity.h11_index = "h11_live1".to_string();

        sync_engine.upsert_items(vec![connectivity])?;

        // Periodic flush during recording
        sync_engine.flush().await?;

        // Get session remote ID after flush
        let r = sync_engine.database.r_transaction()?;
        let mut session_remote_id = None;
        for raw_session in r.scan().primary::<SessionLocal>()?.all()? {
            if let Ok(session) = raw_session {
                if session.id_local.as_deref() == Some("live_recording_session") {
                    session_remote_id = session.id;
                    assert!(
                        session.timestamp_end.is_none(),
                        "Session should still be active"
                    );
                    break;
                }
            }
        }
        session_remote_id.expect("Session must have remote ID");
        drop(r);

        // Continue recording and add event
        active_session.altitude_max = 140.0;
        active_session.distance_total = 600.0;
        sync_engine.upsert_items(vec![active_session.clone()])?;

        let mut live_event = EventLocal::default();
        live_event.set_id_local("live_event_1".to_string());
        live_event.device_id = device_id;
        live_event.set_ancestor_id_local("live_recording_session".to_string());
        live_event.timestamp_observation = "2023-01-01T14:20:00Z".to_string();
        live_event.message = Some("Live observation".to_string());
        live_event.altitude = 140.0;
        live_event.media_type = MediaType::Image;

        sync_engine.upsert_items(vec![live_event])?;

        // Final flush
        sync_engine.flush().await?;

        // Verify final state
        assert_eq!(sync_engine.get_table_count::<SessionLocal>()?, 1);
        assert_eq!(sync_engine.get_table_count::<ConnectivityLocal>()?, 1);
        assert_eq!(sync_engine.get_table_count::<EventLocal>()?, 1);

        // Complete the session
        active_session.timestamp_end = Some("2023-01-01T14:30:00Z".to_string());
        sync_engine.upsert_items(vec![active_session])?;

        Ok(())
    }

    #[tokio::test]
    async fn test_field_workflow_multiple_sessions_with_strategic_flushing() -> Result<()> {
        let mut sync_engine = create_test_sync_engine_with_identification().await?;

        let device_id = std::env::var(env::SCOUT_DEVICE_ID)
            .expect("SCOUT_DEVICE_ID required")
            .parse()
            .expect("SCOUT_DEVICE_ID must be valid integer");

        // Pre-work session
        let mut pre_work_session = SessionLocal::default();
        pre_work_session.set_id_local("pre_work_session".to_string());
        pre_work_session.device_id = device_id;
        pre_work_session.timestamp_start = "2023-01-01T06:00:00Z".to_string();
        pre_work_session.software_version = "field_workflow_test".to_string();
        pre_work_session.altitude_max = 50.0;
        pre_work_session.distance_total = 200.0;

        sync_engine.upsert_items(vec![pre_work_session.clone()])?;

        // Morning survey with event and connectivity
        let mut morning_survey = SessionLocal::default();
        morning_survey.set_id_local("morning_survey".to_string());
        morning_survey.device_id = device_id;
        morning_survey.timestamp_start = "2023-01-01T08:00:00Z".to_string();
        morning_survey.software_version = "field_workflow_test".to_string();
        morning_survey.altitude_max = 150.0;
        morning_survey.distance_total = 1200.0;

        sync_engine.upsert_items(vec![morning_survey.clone()])?;

        let mut survey_event = EventLocal::default();
        survey_event.set_id_local("survey_obs_1".to_string());
        survey_event.device_id = device_id;
        survey_event.set_ancestor_id_local("morning_survey".to_string());
        survey_event.timestamp_observation = "2023-01-01T08:30:00Z".to_string();
        survey_event.message = Some("Bird observation".to_string());
        survey_event.altitude = 120.0;
        survey_event.media_type = MediaType::Image;

        let mut connectivity = ConnectivityLocal::default();
        connectivity.set_id_local("survey_conn_1".to_string());
        connectivity.set_ancestor_id_local("morning_survey".to_string());
        connectivity.timestamp_start = "2023-01-01T08:15:00Z".to_string();
        connectivity.signal = Some(-68.0);
        connectivity.altitude = 130.0;
        connectivity.location = Some("POINT(-155.15393 19.754824)".to_string());
        connectivity.h14_index = "h14_survey1".to_string();
        connectivity.h13_index = "h13_survey1".to_string();
        connectivity.h12_index = "h12_survey1".to_string();
        connectivity.h11_index = "h11_survey1".to_string();

        sync_engine.upsert_items(vec![survey_event])?;
        sync_engine.upsert_items(vec![connectivity])?;

        // Strategic flush
        sync_engine.flush().await?;

        // Continue with remote area session
        let mut remote_session = SessionLocal::default();
        remote_session.set_id_local("remote_area_session".to_string());
        remote_session.device_id = device_id;
        remote_session.timestamp_start = "2023-01-01T13:00:00Z".to_string();
        remote_session.software_version = "field_workflow_test".to_string();
        remote_session.altitude_max = 200.0;
        remote_session.distance_total = 2500.0;

        sync_engine.upsert_items(vec![remote_session])?;

        // Add two events to remote session
        let mut remote_event1 = EventLocal::default();
        remote_event1.set_id_local("remote_obs_1".to_string());
        remote_event1.device_id = device_id;
        remote_event1.set_ancestor_id_local("remote_area_session".to_string());
        remote_event1.timestamp_observation = "2023-01-01T13:30:00Z".to_string();
        remote_event1.message = Some("Wildlife in remote area".to_string());
        remote_event1.altitude = 200.0;
        remote_event1.media_type = MediaType::Video;

        let mut remote_event2 = EventLocal::default();
        remote_event2.set_id_local("remote_obs_2".to_string());
        remote_event2.device_id = device_id;
        remote_event2.set_ancestor_id_local("remote_area_session".to_string());
        remote_event2.timestamp_observation = "2023-01-01T14:15:00Z".to_string();
        remote_event2.message = Some("Rare species sighting".to_string());
        remote_event2.altitude = 195.0;
        remote_event2.media_type = MediaType::Image;

        sync_engine.upsert_items(vec![remote_event1, remote_event2])?;

        // End of day flush - should now succeed with session fallback
        sync_engine.flush().await?;

        // Verify final state
        assert_eq!(sync_engine.get_table_count::<SessionLocal>()?, 3);
        assert_eq!(sync_engine.get_table_count::<EventLocal>()?, 3);
        assert_eq!(sync_engine.get_table_count::<ConnectivityLocal>()?, 1);

        println!("✅ Test passed: Field workflow completed successfully with session fallback");
        Ok(())
    }
    #[tokio::test]
    async fn test_upsert_same_session_id_no_duplicates() -> Result<()> {
        setup_test_env();
        let mut sync_engine = create_test_sync_engine()?;

        // Check initial count is 0
        let initial_count = sync_engine.get_table_count::<SessionLocal>()?;
        assert_eq!(initial_count, 0);

        let device_id = std::env::var(env::SCOUT_DEVICE_ID)
            .expect("SCOUT_DEVICE_ID required")
            .parse()
            .expect("SCOUT_DEVICE_ID must be valid integer");

        // Create a test session
        let mut session = SessionLocal::default();
        session.set_id_local("duplicate_test_session".to_string());
        session.device_id = device_id;
        session.timestamp_start = "2023-01-01T00:00:00Z".to_string();

        // First upsert - should insert the session
        sync_engine.upsert_items(vec![session.clone()])?;
        let count_after_first = sync_engine.get_table_count::<SessionLocal>()?;
        assert_eq!(count_after_first, 1);

        // Create a modified version of the same session (same id_local but different data)
        let mut updated_session = session.clone();
        updated_session.timestamp_end = Some("2023-01-01T01:00:00Z".to_string());

        // Second upsert with same id_local - should update, not create duplicate
        sync_engine.upsert_items(vec![updated_session])?;
        let count_after_second = sync_engine.get_table_count::<SessionLocal>()?;
        assert_eq!(
            count_after_second, 1,
            "Session count should remain 1 after upserting same ID"
        );

        // Third upsert with the original session again - should still be 1
        sync_engine.upsert_items(vec![session])?;
        let count_after_third = sync_engine.get_table_count::<SessionLocal>()?;
        assert_eq!(
            count_after_third, 1,
            "Session count should remain 1 after upserting same ID again"
        );

        // Test with multiple sessions including duplicates in the same batch
        let mut session2 = SessionLocal::default();
        session2.set_id_local("batch_duplicate_test_session_2".to_string());
        session2.device_id = device_id;
        session2.timestamp_start = "2023-01-01T02:00:00Z".to_string();

        let mut session3 = SessionLocal::default();
        session3.set_id_local("batch_duplicate_test_session_3".to_string());
        session3.device_id = device_id;
        session3.timestamp_start = "2023-01-01T03:00:00Z".to_string();

        // Create duplicate of session2 with different data
        let session2_duplicate = session2.clone();

        // Upsert batch with original and duplicate
        sync_engine.upsert_items(vec![session2, session3, session2_duplicate])?;
        let final_count = sync_engine.get_table_count::<SessionLocal>()?;
        assert_eq!(
            final_count, 3,
            "Should have 3 unique sessions total (1 original + 2 new)"
        );

        Ok(())
    }

    #[tokio::test]
    async fn test_clean_safety_mechanisms() -> Result<()> {
        setup_test_env();
        let mut sync_engine = create_test_sync_engine()?;

        let device_id = std::env::var(env::SCOUT_DEVICE_ID)
            .expect("SCOUT_DEVICE_ID required")
            .parse()
            .expect("SCOUT_DEVICE_ID must be valid integer");

        // Test Case 1: Complete session but no remote ID - should NOT be cleaned
        let mut complete_no_remote = SessionLocal::default();
        complete_no_remote.set_id_local("complete_no_remote".to_string());
        complete_no_remote.id = None; // No remote ID
        complete_no_remote.device_id = device_id;
        complete_no_remote.timestamp_start = "2023-01-01T10:00:00Z".to_string();
        complete_no_remote.timestamp_end = Some("2023-01-01T11:00:00Z".to_string());
        complete_no_remote.software_version = "1.0.0".to_string();
        complete_no_remote.altitude_max = 100.0;
        complete_no_remote.altitude_min = 50.0;
        complete_no_remote.altitude_average = 75.0;
        complete_no_remote.velocity_max = 25.0;
        complete_no_remote.velocity_min = 10.0;
        complete_no_remote.velocity_average = 15.0;
        complete_no_remote.distance_total = 1000.0;
        complete_no_remote.distance_max_from_start = 500.0;

        // Test Case 2: Complete session with remote ID but descendant lacks remote ID
        let mut complete_with_unsynced_descendant = SessionLocal::default();
        complete_with_unsynced_descendant.set_id_local("complete_with_unsynced".to_string());
        complete_with_unsynced_descendant.id = Some(12345); // Has remote ID
        complete_with_unsynced_descendant.device_id = device_id;
        complete_with_unsynced_descendant.timestamp_start = "2023-01-01T12:00:00Z".to_string();
        complete_with_unsynced_descendant.timestamp_end = Some("2023-01-01T13:00:00Z".to_string());
        complete_with_unsynced_descendant.software_version = "1.0.0".to_string();
        complete_with_unsynced_descendant.altitude_max = 120.0;
        complete_with_unsynced_descendant.altitude_min = 60.0;
        complete_with_unsynced_descendant.altitude_average = 90.0;
        complete_with_unsynced_descendant.velocity_max = 30.0;
        complete_with_unsynced_descendant.velocity_min = 15.0;
        complete_with_unsynced_descendant.velocity_average = 22.0;
        complete_with_unsynced_descendant.distance_total = 1200.0;
        complete_with_unsynced_descendant.distance_max_from_start = 600.0;

        // Create event with NO remote ID for the second session
        let mut unsynced_event = EventLocal::default();
        unsynced_event.set_id_local("unsynced_event".to_string());
        unsynced_event.id = None; // No remote ID - this should prevent cleaning
        unsynced_event.device_id = device_id;
        unsynced_event.session_id = Some(12345);
        unsynced_event.set_ancestor_id_local("complete_with_unsynced".to_string());
        unsynced_event.timestamp_observation = "2023-01-01T12:15:00Z".to_string();
        unsynced_event.message = Some("Unsynced event".to_string());
        unsynced_event.altitude = 100.0;
        unsynced_event.heading = 0.0;
        unsynced_event.media_type = MediaType::Image;

        // Test Case 3: Complete session with all descendants synced - SHOULD be cleaned
        let mut complete_fully_synced = SessionLocal::default();
        complete_fully_synced.set_id_local("complete_fully_synced".to_string());
        complete_fully_synced.id = Some(23456); // Has remote ID
        complete_fully_synced.device_id = device_id;
        complete_fully_synced.timestamp_start = "2023-01-01T14:00:00Z".to_string();
        complete_fully_synced.timestamp_end = Some("2023-01-01T15:00:00Z".to_string());
        complete_fully_synced.software_version = "1.0.0".to_string();
        complete_fully_synced.altitude_max = 150.0;
        complete_fully_synced.altitude_min = 80.0;
        complete_fully_synced.altitude_average = 115.0;
        complete_fully_synced.velocity_max = 35.0;
        complete_fully_synced.velocity_min = 20.0;
        complete_fully_synced.velocity_average = 27.0;
        complete_fully_synced.distance_total = 1500.0;
        complete_fully_synced.distance_max_from_start = 750.0;

        // Create fully synced descendants
        let mut synced_connectivity = ConnectivityLocal::default();
        synced_connectivity.set_id_local("synced_connectivity".to_string());
        synced_connectivity.id = Some(34567); // Has remote ID
        synced_connectivity.session_id = None;
        synced_connectivity.device_id = Some(device_id);
        synced_connectivity.set_ancestor_id_local("complete_fully_synced".to_string());
        synced_connectivity.timestamp_start = "2023-01-01T14:05:00Z".to_string();
        synced_connectivity.signal = Some(-70.0);
        synced_connectivity.noise = Some(-90.0);
        synced_connectivity.altitude = 100.0;
        synced_connectivity.heading = 0.0;
        synced_connectivity.location = Some("POINT(-155.15393 19.754824)".to_string());
        synced_connectivity.h14_index = "h14".to_string();
        synced_connectivity.h13_index = "h13".to_string();
        synced_connectivity.h12_index = "h12".to_string();
        synced_connectivity.h11_index = "h11".to_string();

        let mut synced_event = EventLocal::default();
        synced_event.set_id_local("synced_event".to_string());
        synced_event.id = Some(45678); // Has remote ID
        synced_event.device_id = device_id;
        synced_event.session_id = Some(23456);
        synced_event.set_ancestor_id_local("complete_fully_synced".to_string());
        synced_event.timestamp_observation = "2023-01-01T14:15:00Z".to_string();
        synced_event.message = Some("Synced event".to_string());
        synced_event.altitude = 100.0;
        synced_event.heading = 0.0;
        synced_event.media_type = MediaType::Image;

        // Insert all test data
        sync_engine.upsert_items(vec![
            complete_no_remote,
            complete_with_unsynced_descendant,
            complete_fully_synced,
        ])?;
        sync_engine.upsert_items(vec![unsynced_event, synced_event])?;
        sync_engine.upsert_items(vec![synced_connectivity])?;

        // Verify initial state
        assert_eq!(sync_engine.get_table_count::<SessionLocal>()?, 3);
        assert_eq!(sync_engine.get_table_count::<EventLocal>()?, 2);
        assert_eq!(sync_engine.get_table_count::<ConnectivityLocal>()?, 1);

        // Run clean operation
        sync_engine.clean().await?;

        // Verify results:
        // - complete_no_remote should NOT be cleaned (no remote ID)
        // - complete_with_unsynced should NOT be cleaned (descendant lacks remote ID)
        // - complete_fully_synced SHOULD be cleaned (all have remote IDs)
        assert_eq!(
            sync_engine.get_table_count::<SessionLocal>()?,
            2,
            "Should have 2 sessions remaining (2 that couldn't be cleaned)"
        );
        assert_eq!(
            sync_engine.get_table_count::<EventLocal>()?,
            1,
            "Should have 1 event remaining (unsynced_event)"
        );
        assert_eq!(
            sync_engine.get_table_count::<ConnectivityLocal>()?,
            0,
            "Synced connectivity should be cleaned with its session"
        );

        // Verify which sessions remain
        let r = sync_engine.database.r_transaction()?;
        let remaining_sessions: Vec<SessionLocal> = r
            .scan()
            .primary::<SessionLocal>()?
            .all()?
            .into_iter()
            .collect::<Result<Vec<_>, _>>()?;

        let remaining_ids: std::collections::HashSet<&str> = remaining_sessions
            .iter()
            .filter_map(|s| s.id_local.as_deref())
            .collect();

        assert!(
            remaining_ids.contains("complete_no_remote"),
            "Session without remote ID should not be cleaned"
        );
        assert!(
            remaining_ids.contains("complete_with_unsynced"),
            "Session with unsynced descendants should not be cleaned"
        );
        assert!(
            !remaining_ids.contains("complete_fully_synced"),
            "Fully synced session should be cleaned"
        );

        // Verify which events remain
        let remaining_events: Vec<EventLocal> = r
            .scan()
            .primary::<EventLocal>()?
            .all()?
            .into_iter()
            .collect::<Result<Vec<_>, _>>()?;

        assert_eq!(remaining_events.len(), 1);
        assert_eq!(
            remaining_events[0].id_local.as_deref(),
            Some("unsynced_event")
        );

        Ok(())
    }

    #[tokio::test]
    async fn test_descendant_updates_for_late_arriving_children() -> Result<()> {
        let mut sync_engine = create_test_sync_engine_with_identification().await?;

        let device_id = std::env::var(env::SCOUT_DEVICE_ID)
            .expect("SCOUT_DEVICE_ID required")
            .parse()
            .expect("SCOUT_DEVICE_ID must be valid integer");

        // Step 1: Create and sync a session first (gets remote ID)
        let mut session = SessionLocal::default();
        session.set_id_local("session_synced_first".to_string());
        session.device_id = device_id;
        session.timestamp_start = "2023-01-01T10:00:00Z".to_string();
        session.software_version = "test_late_children_0".to_string();
        session.altitude_max = 100.0;
        session.altitude_min = 50.0;
        session.altitude_average = 75.0;
        session.velocity_max = 25.0;
        session.velocity_min = 10.0;
        session.velocity_average = 15.0;
        session.distance_total = 1000.0;
        session.distance_max_from_start = 500.0;

        sync_engine.upsert_items(vec![session])?;

        // Flush session first - it should get a remote ID
        sync_engine.flush_sessions().await?;

        // Verify session has remote ID
        let r = sync_engine.database.r_transaction()?;
        let mut session_remote_id = None;
        for raw_session in r.scan().primary::<SessionLocal>()?.all()? {
            if let Ok(session) = raw_session {
                if session.id_local.as_deref() == Some("session_synced_first") {
                    session_remote_id = session.id;
                    assert!(
                        session_remote_id.is_some(),
                        "Session must have remote ID after first flush"
                    );
                    break;
                }
            }
        }
        drop(r);

        let session_id = session_remote_id.unwrap();

        // Step 2: Now create connectivity records AFTER session has remote ID
        // This simulates the problem: connectivity created during flight after session sync
        let mut connectivity1 = ConnectivityLocal::default();
        connectivity1.set_id_local("late_connectivity_1".to_string());
        connectivity1.session_id = None; // This should get populated by our fix
        connectivity1.device_id = Some(device_id);
        connectivity1.set_ancestor_id_local("session_synced_first".to_string());
        connectivity1.timestamp_start = "2023-01-01T10:05:00Z".to_string();
        connectivity1.signal = Some(-70.0);
        connectivity1.noise = Some(-90.0);
        connectivity1.altitude = 100.0;
        connectivity1.heading = 0.0;
        connectivity1.location = Some("POINT(-155.15393 19.754824)".to_string());
        connectivity1.h14_index = "h14".to_string();
        connectivity1.h13_index = "h13".to_string();
        connectivity1.h12_index = "h12".to_string();
        connectivity1.h11_index = "h11".to_string();

        let mut connectivity2 = ConnectivityLocal::default();
        connectivity2.set_id_local("late_connectivity_2".to_string());
        connectivity2.session_id = None; // This should get populated by our fix
        connectivity2.device_id = Some(device_id);
        connectivity2.set_ancestor_id_local("session_synced_first".to_string());
        connectivity2.timestamp_start = "2023-01-01T10:10:00Z".to_string();
        connectivity2.signal = Some(-75.0);
        connectivity2.noise = Some(-95.0);
        connectivity2.altitude = 105.0;
        connectivity2.heading = 45.0;
        connectivity2.location = Some("POINT(-155.15400 19.754830)".to_string());
        connectivity2.h14_index = "h14".to_string();
        connectivity2.h13_index = "h13".to_string();
        connectivity2.h12_index = "h12".to_string();
        connectivity2.h11_index = "h11".to_string();

        // Insert connectivity records locally
        sync_engine.upsert_items(vec![connectivity1, connectivity2])?;

        // Verify they don't have session_id yet
        let r = sync_engine.database.r_transaction()?;
        for raw_connectivity in r.scan().primary::<ConnectivityLocal>()?.all()? {
            if let Ok(connectivity) = raw_connectivity {
                if connectivity.ancestor_id_local.as_deref() == Some("session_synced_first") {
                    assert_eq!(
                        connectivity.session_id, None,
                        "Connectivity should not have session_id before sync (this is the bug we're fixing)"
                    );
                }
            }
        }
        drop(r);

        // Step 3: Flush connectivity - our fix should populate session_id
        sync_engine.flush_connectivity().await?;

        // Step 4: Verify the fix worked - connectivity records should now have session_id
        let r = sync_engine.database.r_transaction()?;
        let mut connectivity_count_with_session_id = 0;
        for raw_connectivity in r.scan().primary::<ConnectivityLocal>()?.all()? {
            if let Ok(connectivity) = raw_connectivity {
                if connectivity.ancestor_id_local.as_deref() == Some("session_synced_first") {
                    assert_eq!(
                        connectivity.session_id,
                        Some(session_id),
                        "Connectivity must have session_id populated after our fix (connectivity: {})",
                        connectivity.id_local.as_deref().unwrap_or("unknown")
                    );
                    connectivity_count_with_session_id += 1;
                }
            }
        }
        drop(r);

        assert_eq!(
            connectivity_count_with_session_id, 2,
            "Both connectivity records should have session_id populated"
        );

        // Step 5: Test the same scenario with events
        let mut event = EventLocal::default();
        event.set_id_local("late_event_1".to_string());
        event.device_id = device_id;
        event.session_id = None; // Should get populated by our fix
        event.set_ancestor_id_local("session_synced_first".to_string());
        event.timestamp_observation = "2023-01-01T10:15:00Z".to_string();
        event.message = Some("Late arriving event".to_string());
        event.altitude = 100.0;
        event.heading = 0.0;
        event.media_type = MediaType::Image;

        sync_engine.upsert_items(vec![event])?;

        // Flush events - should populate session_id due to our fix
        sync_engine.flush_events().await?;

        // Verify event got session_id populated
        let r = sync_engine.database.r_transaction()?;
        for raw_event in r.scan().primary::<EventLocal>()?.all()? {
            if let Ok(event) = raw_event {
                if event.ancestor_id_local.as_deref() == Some("session_synced_first") {
                    assert_eq!(
                        event.session_id,
                        Some(session_id),
                        "Event must have session_id populated after our fix"
                    );
                }
            }
        }
        drop(r);

        // Step 6: Test the same scenario with operators
        let mut operator = data::v2::OperatorLocal::default();
        operator.set_id_local("late_operator_1".to_string());
        operator.session_id = None; // Should get populated by our fix
        operator.set_ancestor_id_local("session_synced_first".to_string());
        operator.user_id = test_user_id();
        operator.action = "late_test_action".to_string();
        operator.timestamp = Some("2023-01-01T10:20:00Z".to_string());

        sync_engine.upsert_items(vec![operator])?;

        // Flush operators - should populate session_id due to our fix
        sync_engine.flush_operators().await?;

        // Verify operator got session_id populated
        let r = sync_engine.database.r_transaction()?;
        for raw_operator in r.scan().primary::<data::v2::OperatorLocal>()?.all()? {
            if let Ok(operator) = raw_operator {
                if operator.ancestor_id_local.as_deref() == Some("session_synced_first") {
                    assert_eq!(
                        operator.session_id,
                        Some(session_id),
                        "Operator must have session_id populated after our fix"
                    );
                }
            }
        }
        drop(r);

        println!("✅ Test passed: Late arriving children get proper ancestor IDs populated");
        Ok(())
    }

    #[tokio::test]
    async fn test_artifact_upload_filtering() -> Result<()> {
        setup_test_env();
        let mut sync_engine = create_test_sync_engine()?;

        let device_id = std::env::var(env::SCOUT_DEVICE_ID)
            .expect("SCOUT_DEVICE_ID required")
            .parse()
            .expect("SCOUT_DEVICE_ID must be valid integer");

        // Create two artifacts - one with file uploaded, one without
        let mut artifact_uploaded = ArtifactLocal::new(
            "path/to/uploaded_file.jpg".to_string(),
            None,
            device_id,
            Some("image".to_string()),
            None,
        );
        artifact_uploaded.set_id_local("artifact_uploaded".to_string());
        artifact_uploaded.mark_file_uploaded(); // Mark as uploaded

        let mut artifact_pending = ArtifactLocal::new(
            "path/to/pending_file.jpg".to_string(),
            None,
            device_id,
            Some("image".to_string()),
            None,
        );
        artifact_pending.set_id_local("artifact_pending".to_string());
        // Leave as not uploaded (default is false)

        // Insert both artifacts
        sync_engine.upsert_items(vec![artifact_uploaded.clone(), artifact_pending.clone()])?;

        // Verify both are in database
        assert_eq!(sync_engine.get_table_count::<ArtifactLocal>()?, 2);

        // Check pending upload counts
        let pending_count = sync_engine.get_artifacts_pending_upload_count()?;
        assert_eq!(pending_count, 1, "Should have 1 artifact pending upload");

        let pending_artifacts = sync_engine.get_artifacts_pending_upload()?;
        assert_eq!(pending_artifacts.len(), 1);
        assert_eq!(
            pending_artifacts[0].id_local,
            Some("artifact_pending".to_string())
        );

        // Test the filtering in get_batch
        let artifacts_batch: BatchSync<ArtifactLocal> = sync_engine
            .get_batch::<ArtifactLocal>(EnumSyncAction::Upsert, EnumSyncAction::Insert)?;

        let mut all_artifacts = artifacts_batch.upsert;
        all_artifacts.extend(artifacts_batch.insert);

        // Before filtering, we should have 2 artifacts
        assert_eq!(
            all_artifacts.len(),
            2,
            "Should have 2 artifacts before filtering"
        );

        // Apply the same filtering logic as flush_artifacts
        all_artifacts.retain(|artifact| artifact.has_uploaded_file_to_storage);

        // After filtering, we should only have 1 artifact (the uploaded one)
        assert_eq!(
            all_artifacts.len(),
            1,
            "Should have 1 artifact after filtering"
        );
        assert_eq!(
            all_artifacts[0].id_local,
            Some("artifact_uploaded".to_string())
        );

        println!("✅ Test passed: Only artifacts with uploaded files are included in sync");
        Ok(())
    }

    #[test]
    fn test_critical_error_detection() {
        // Test that critical errors are properly detected
        assert!(SyncEngine::is_critical_error(
            "parse error - invalid geometry"
        ));
        assert!(SyncEngine::is_critical_error(
            "Parse Error - Invalid Geometry"
        )); // Case insensitive
        assert!(SyncEngine::is_critical_error(
            "new row violates row-level security policy"
        ));
        assert!(SyncEngine::is_critical_error(
            "New Row Violates Row-Level Security Policy"
        )); // Case insensitive
        assert!(SyncEngine::is_critical_error("all object keys must match"));
        assert!(SyncEngine::is_critical_error("All Object Keys Must Match")); // Case insensitive

        // Test that non-critical errors are not detected as critical
        assert!(!SyncEngine::is_critical_error("network timeout"));
        assert!(!SyncEngine::is_critical_error("connection refused"));
        assert!(!SyncEngine::is_critical_error("invalid json"));
        assert!(!SyncEngine::is_critical_error("server error 500"));

        println!("✅ Test passed: Critical error detection works correctly");
    }

    #[tokio::test]
    async fn test_sync_engine_with_failed_record_removal() -> Result<()> {
        setup_test_env();

        // Test that the constructor with failed record removal works
        let database_config =
            DatabaseConfig::from_env().expect("Failed to create database config from environment");
        let client = ScoutClient::new(database_config);

        let temp_db = format!(
            "/tmp/scout_test_failed_removal_{}.db",
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        );

        let sync_engine = SyncEngine::with_failed_record_removal(client, temp_db.clone())?;

        // Verify the flag is set correctly
        assert_eq!(sync_engine.remove_failed_records, true);

        // Clean up
        let _ = std::fs::remove_file(&temp_db);

        println!("✅ Test passed: SyncEngine with failed record removal constructor works");
        Ok(())
    }

    #[tokio::test]
    async fn test_remove_failed_records_functionality() -> Result<()> {
        setup_test_env();

        let database_config =
            DatabaseConfig::from_env().expect("Failed to create database config from environment");
        let client = ScoutClient::new(database_config);

        let temp_db = format!(
            "/tmp/scout_test_remove_failed_{}.db",
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        );

        // Create sync engine with remove_failed_records enabled for testing
        let mut sync_engine = SyncEngine::new(client, temp_db.clone(), None, true)?;

        let device_id = std::env::var(env::SCOUT_DEVICE_ID)
            .expect("SCOUT_DEVICE_ID required")
            .parse()
            .expect("SCOUT_DEVICE_ID must be valid integer");

        // Create a session that might trigger critical errors
        let mut test_session = SessionLocal::default();
        test_session.set_id_local("test_session_for_removal".to_string());
        test_session.device_id = device_id;
        test_session.timestamp_start = chrono::Utc::now().format("%Y-%m-%dT%H:%M:%SZ").to_string();
        test_session.software_version = "remove_failed_test".to_string();
        test_session.altitude_max = 100.0;
        test_session.altitude_min = 50.0;
        test_session.altitude_average = 75.0;
        test_session.velocity_max = 25.0;
        test_session.velocity_min = 10.0;
        test_session.velocity_average = 15.0;
        test_session.distance_total = 1000.0;
        test_session.distance_max_from_start = 500.0;

        // Insert the session
        sync_engine.upsert_items(vec![test_session])?;

        // Verify session exists
        assert_eq!(sync_engine.get_table_count::<SessionLocal>()?, 1);

        // Test critical error detection
        assert!(SyncEngine::is_critical_error(
            "parse error - invalid geometry"
        ));
        assert!(SyncEngine::is_critical_error(
            "new row violates row-level security policy"
        ));
        assert!(SyncEngine::is_critical_error("all object keys must match"));
        assert!(!SyncEngine::is_critical_error("network timeout"));

        // Verify the sync engine has remove_failed_records enabled
        assert_eq!(sync_engine.remove_failed_records, true);

        // Clean up
        let _ = std::fs::remove_file(&temp_db);

        println!("✅ Test passed: Remove failed records functionality is configured correctly");
        Ok(())
    }

    #[tokio::test]
    async fn test_remove_failed_records_comprehensive() -> Result<()> {
        setup_test_env();

        let database_config =
            DatabaseConfig::from_env().expect("Failed to create database config from environment");
        let client = ScoutClient::new(database_config);

        let temp_db = format!(
            "/tmp/scout_test_comprehensive_remove_{}.db",
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        );

        // Create sync engine with remove_failed_records enabled for testing
        let mut sync_engine = SyncEngine::new(client, temp_db.clone(), None, true)?;

        let device_id = std::env::var(env::SCOUT_DEVICE_ID)
            .expect("SCOUT_DEVICE_ID required")
            .parse()
            .expect("SCOUT_DEVICE_ID must be valid integer");

        // Create test data for all entity types
        let mut test_session = SessionLocal::default();
        test_session.set_id_local("test_session_comprehensive".to_string());
        test_session.device_id = device_id;
        test_session.timestamp_start = chrono::Utc::now().format("%Y-%m-%dT%H:%M:%SZ").to_string();
        test_session.software_version = "comprehensive_test".to_string();
        test_session.altitude_max = 100.0;
        test_session.altitude_min = 50.0;
        test_session.altitude_average = 75.0;
        test_session.velocity_max = 25.0;
        test_session.velocity_min = 10.0;
        test_session.velocity_average = 15.0;
        test_session.distance_total = 1000.0;
        test_session.distance_max_from_start = 500.0;

        let mut test_event = EventLocal::default();
        test_event.set_id_local("test_event_comprehensive".to_string());
        test_event.device_id = device_id;
        test_event.set_ancestor_id_local("test_session_comprehensive".to_string());
        test_event.timestamp_observation =
            chrono::Utc::now().format("%Y-%m-%dT%H:%M:%SZ").to_string();
        test_event.message = Some("Test event".to_string());
        test_event.altitude = 100.0;
        test_event.heading = 0.0;
        test_event.media_type = MediaType::Image;

        let mut test_connectivity = ConnectivityLocal::default();
        test_connectivity.set_id_local("test_conn_comprehensive".to_string());
        test_connectivity.set_ancestor_id_local("test_session_comprehensive".to_string());
        test_connectivity.timestamp_start =
            chrono::Utc::now().format("%Y-%m-%dT%H:%M:%SZ").to_string();
        test_connectivity.signal = Some(-70.0);
        test_connectivity.altitude = 100.0;
        test_connectivity.h14_index = "h14_test".to_string();
        test_connectivity.h13_index = "h13_test".to_string();
        test_connectivity.h12_index = "h12_test".to_string();
        test_connectivity.h11_index = "h11_test".to_string();

        let mut test_tag = TagLocal::default();
        test_tag.set_id_local("test_tag_comprehensive".to_string());
        test_tag.set_ancestor_id_local("test_event_comprehensive".to_string());
        test_tag.event_id = None; // Will be updated when event syncs
        test_tag.class = "test_class_name".to_string();

        let mut test_artifact = ArtifactLocal::new(
            "/test/path/file.jpg".to_string(),
            None,
            device_id,
            Some("image".to_string()),
            None,
        );
        test_artifact.set_id_local("test_artifact_comprehensive".to_string());
        test_artifact.set_ancestor_id_local("test_session_comprehensive".to_string());
        test_artifact.mark_file_uploaded(); // Mark as uploaded so it gets synced

        // Insert all test data
        sync_engine.upsert_items(vec![test_session])?;
        sync_engine.upsert_items(vec![test_event])?;
        sync_engine.upsert_items(vec![test_connectivity])?;
        sync_engine.upsert_items(vec![test_tag])?;
        sync_engine.upsert_items(vec![test_artifact])?;

        // Verify all entities exist before sync
        assert_eq!(sync_engine.get_table_count::<SessionLocal>()?, 1);
        assert_eq!(sync_engine.get_table_count::<EventLocal>()?, 1);
        assert_eq!(sync_engine.get_table_count::<ConnectivityLocal>()?, 1);
        assert_eq!(sync_engine.get_table_count::<TagLocal>()?, 1);
        assert_eq!(sync_engine.get_table_count::<ArtifactLocal>()?, 1);

        // Verify the sync engine has remove_failed_records enabled
        assert_eq!(sync_engine.remove_failed_records, true);

        // Clean up
        let _ = std::fs::remove_file(&temp_db);

        println!("✅ Test passed: Comprehensive remove failed records test completed");
        Ok(())
    }

    #[tokio::test]
    async fn test_export_to_json() -> Result<()> {
        let mut sync_engine = create_test_sync_engine()?;

        let device_id = std::env::var(env::SCOUT_DEVICE_ID)
            .expect("SCOUT_DEVICE_ID required")
            .parse()
            .expect("SCOUT_DEVICE_ID must be valid integer");

        // Create test data for all types
        let mut session = SessionLocal::default();
        session.set_id_local("export_test_session".to_string());
        session.device_id = device_id;
        session.timestamp_start = "2023-01-01T00:00:00Z".to_string();
        session.software_version = "test_export".to_string();

        let mut connectivity = ConnectivityLocal::default();
        connectivity.set_id_local("export_test_connectivity".to_string());
        connectivity.device_id = Some(device_id);
        connectivity.set_ancestor_id_local("export_test_session".to_string());
        connectivity.timestamp_start = "2023-01-01T00:00:00Z".to_string();
        connectivity.location = Some("POINT(-155.15393 19.754824)".to_string());

        let mut event = EventLocal::default();
        event.set_id_local("export_test_event".to_string());
        event.device_id = device_id;
        event.set_ancestor_id_local("export_test_session".to_string());
        event.timestamp_observation = "2023-01-01T10:10:00Z".to_string();
        event.message = Some("Test export event".to_string());
        event.media_type = MediaType::Image;

        let mut tag = TagLocal::default();
        tag.set_id_local("export_test_tag".to_string());
        tag.set_ancestor_id_local("export_test_event".to_string());
        tag.class = "test_export_tag".to_string();
        tag.conf = 0.95;
        tag.observation_type = TagObservationType::Manual;

        let mut operator = data::v2::OperatorLocal::default();
        operator.set_id_local("export_test_operator".to_string());
        operator.set_ancestor_id_local("export_test_session".to_string());
        operator.user_id = test_user_id();
        operator.action = "test_export_action".to_string();

        let mut artifact = ArtifactLocal::default();
        artifact.set_id_local("export_test_artifact".to_string());
        artifact.set_ancestor_id_local("export_test_session".to_string());
        artifact.file_path = "test/path.jpg".to_string();
        artifact.modality = Some("image".to_string());

        // Insert all items
        sync_engine.upsert_items(vec![session])?;
        sync_engine.upsert_items(vec![connectivity])?;
        sync_engine.upsert_items(vec![event])?;
        sync_engine.upsert_items(vec![tag])?;
        sync_engine.upsert_items(vec![operator])?;
        sync_engine.upsert_items(vec![artifact])?;

        // Verify counts before export
        assert_eq!(sync_engine.get_table_count::<SessionLocal>()?, 1);
        assert_eq!(sync_engine.get_table_count::<ConnectivityLocal>()?, 1);
        assert_eq!(sync_engine.get_table_count::<EventLocal>()?, 1);
        assert_eq!(sync_engine.get_table_count::<TagLocal>()?, 1);
        assert_eq!(sync_engine.get_table_count::<data::v2::OperatorLocal>()?, 1);
        assert_eq!(sync_engine.get_table_count::<ArtifactLocal>()?, 1);

        // Create temporary file for export
        let temp_dir = tempdir()?;
        let export_path = temp_dir
            .path()
            .join("export_test.json")
            .to_string_lossy()
            .to_string();

        // Export to JSON
        sync_engine.export_to_json(&export_path)?;

        // Verify file exists
        assert!(std::path::Path::new(&export_path).exists());

        // Read and parse JSON
        let json_content = std::fs::read_to_string(&export_path)?;
        let export_array: Vec<serde_json::Value> = serde_json::from_str(&json_content)?;

        // Verify array structure
        assert_eq!(export_array.len(), 1);

        // Verify session entry structure
        let session_entry = &export_array[0];
        assert!(session_entry.get("session").is_some());
        assert!(session_entry.get("events").is_some());
        assert!(session_entry.get("tags").is_some());
        assert!(session_entry.get("connectivity").is_some());
        assert!(session_entry.get("operators").is_some());
        assert!(session_entry.get("artifacts").is_some());

        // Verify data counts in JSON
        assert_eq!(session_entry["events"].as_array().unwrap().len(), 1);
        assert_eq!(session_entry["tags"].as_array().unwrap().len(), 1);
        assert_eq!(session_entry["connectivity"].as_array().unwrap().len(), 1);
        assert_eq!(session_entry["operators"].as_array().unwrap().len(), 1);
        assert_eq!(session_entry["artifacts"].as_array().unwrap().len(), 1);

        // Verify session data in JSON
        let session_data = &session_entry["session"];
        assert_eq!(
            session_data["id_local"].as_str(),
            Some("export_test_session")
        );

        // Verify event data in JSON
        let event_data = &session_entry["events"][0];
        assert_eq!(event_data["id_local"].as_str(), Some("export_test_event"));

        // Verify tag data in JSON
        let tag_data = &session_entry["tags"][0];
        assert_eq!(tag_data["id_local"].as_str(), Some("export_test_tag"));

        Ok(())
    }

    #[tokio::test]
    async fn test_wipe() -> Result<()> {
        let mut sync_engine = create_test_sync_engine()?;

        let device_id = std::env::var(env::SCOUT_DEVICE_ID)
            .expect("SCOUT_DEVICE_ID required")
            .parse()
            .expect("SCOUT_DEVICE_ID must be valid integer");

        // Create test data for all types
        let mut session = SessionLocal::default();
        session.set_id_local("wipe_test_session".to_string());
        session.device_id = device_id;
        session.timestamp_start = "2023-01-01T00:00:00Z".to_string();

        let mut connectivity = ConnectivityLocal::default();
        connectivity.set_id_local("wipe_test_connectivity".to_string());
        connectivity.device_id = Some(device_id);
        connectivity.set_ancestor_id_local("wipe_test_session".to_string());
        connectivity.timestamp_start = "2023-01-01T00:00:00Z".to_string();

        let mut event = EventLocal::default();
        event.set_id_local("wipe_test_event".to_string());
        event.device_id = device_id;
        event.set_ancestor_id_local("wipe_test_session".to_string());
        event.timestamp_observation = "2023-01-01T10:10:00Z".to_string();
        event.media_type = MediaType::Image;

        let mut tag = TagLocal::default();
        tag.set_id_local("wipe_test_tag".to_string());
        tag.set_ancestor_id_local("wipe_test_event".to_string());
        tag.class = "test_wipe_tag".to_string();
        tag.observation_type = TagObservationType::Manual;

        let mut operator = data::v2::OperatorLocal::default();
        operator.set_id_local("wipe_test_operator".to_string());
        operator.set_ancestor_id_local("wipe_test_session".to_string());
        operator.user_id = test_user_id();
        operator.action = "test_wipe_action".to_string();

        let mut artifact = ArtifactLocal::default();
        artifact.set_id_local("wipe_test_artifact".to_string());
        artifact.set_ancestor_id_local("wipe_test_session".to_string());
        artifact.file_path = "test/path.jpg".to_string();

        // Insert all items
        sync_engine.upsert_items(vec![session])?;
        sync_engine.upsert_items(vec![connectivity])?;
        sync_engine.upsert_items(vec![event])?;
        sync_engine.upsert_items(vec![tag])?;
        sync_engine.upsert_items(vec![operator])?;
        sync_engine.upsert_items(vec![artifact])?;

        // Verify counts before wipe
        assert_eq!(sync_engine.get_table_count::<SessionLocal>()?, 1);
        assert_eq!(sync_engine.get_table_count::<ConnectivityLocal>()?, 1);
        assert_eq!(sync_engine.get_table_count::<EventLocal>()?, 1);
        assert_eq!(sync_engine.get_table_count::<TagLocal>()?, 1);
        assert_eq!(sync_engine.get_table_count::<data::v2::OperatorLocal>()?, 1);
        assert_eq!(sync_engine.get_table_count::<ArtifactLocal>()?, 1);

        // Wipe all data
        sync_engine.wipe(None)?;

        // Verify all counts are 0 after wipe
        assert_eq!(sync_engine.get_table_count::<SessionLocal>()?, 0);
        assert_eq!(sync_engine.get_table_count::<ConnectivityLocal>()?, 0);
        assert_eq!(sync_engine.get_table_count::<EventLocal>()?, 0);
        assert_eq!(sync_engine.get_table_count::<TagLocal>()?, 0);
        assert_eq!(sync_engine.get_table_count::<data::v2::OperatorLocal>()?, 0);
        assert_eq!(sync_engine.get_table_count::<ArtifactLocal>()?, 0);

        Ok(())
    }

    #[tokio::test]
    async fn test_export_to_json_empty_database() -> Result<()> {
        let sync_engine = create_test_sync_engine()?;

        // Create temporary file for export
        let temp_dir = tempdir()?;
        let export_path = temp_dir
            .path()
            .join("export_empty_test.json")
            .to_string_lossy()
            .to_string();

        // Export empty database to JSON
        sync_engine.export_to_json(&export_path)?;

        // Verify file exists
        assert!(std::path::Path::new(&export_path).exists());

        // Read and parse JSON
        let json_content = std::fs::read_to_string(&export_path)?;
        let export_array: Vec<serde_json::Value> = serde_json::from_str(&json_content)?;

        // Verify array is empty
        assert_eq!(export_array.len(), 0);

        Ok(())
    }

    #[tokio::test]
    async fn test_wipe_empty_database() -> Result<()> {
        let mut sync_engine = create_test_sync_engine()?;

        // Verify all counts are 0 initially
        assert_eq!(sync_engine.get_table_count::<SessionLocal>()?, 0);
        assert_eq!(sync_engine.get_table_count::<ConnectivityLocal>()?, 0);
        assert_eq!(sync_engine.get_table_count::<EventLocal>()?, 0);
        assert_eq!(sync_engine.get_table_count::<TagLocal>()?, 0);
        assert_eq!(sync_engine.get_table_count::<data::v2::OperatorLocal>()?, 0);
        assert_eq!(sync_engine.get_table_count::<ArtifactLocal>()?, 0);

        // Wipe empty database (should not error)
        sync_engine.wipe(None)?;

        // Verify all counts are still 0
        assert_eq!(sync_engine.get_table_count::<SessionLocal>()?, 0);
        assert_eq!(sync_engine.get_table_count::<ConnectivityLocal>()?, 0);
        assert_eq!(sync_engine.get_table_count::<EventLocal>()?, 0);
        assert_eq!(sync_engine.get_table_count::<TagLocal>()?, 0);
        assert_eq!(sync_engine.get_table_count::<data::v2::OperatorLocal>()?, 0);
        assert_eq!(sync_engine.get_table_count::<ArtifactLocal>()?, 0);

        Ok(())
    }

    #[test]
    fn test_migrate_local_models_moves_superseded_rows() -> Result<()> {
        let dir = tempdir()?;
        let db_path = dir.path().join("superseded_events.db");

        {
            let database = Builder::new().create(models(), &db_path)?;
            let rw = database.rw_transaction()?;
            rw.insert(event::EventLocalV16 {
                id_local: Some("event-before-bump".to_string()),
                message: Some("written by an older build".to_string()),
                modality: Some("video".to_string()),
                ..Default::default()
            })?;
            rw.insert(data::v13::ArtifactLocal {
                id_local: Some("artifact-before-bump".to_string()),
                file_path: "clip.mp4".to_string(),
                modality: Some("video".to_string()),
                ..Default::default()
            })?;
            rw.insert(data::v14::SessionLocal {
                id_local: Some("session-before-bump".to_string()),
                device_id: 42,
                software_version: "1.2.3".to_string(),
                ..Default::default()
            })?;
            rw.insert(data::v14::ConnectivityLocal {
                id_local: Some("connectivity-before-bump".to_string()),
                device_id: Some(42),
                ..Default::default()
            })?;
            rw.insert(sensor_pose::SensorPoseLocalV1 {
                id_local: Some("pose-before-bump".to_string()),
                part_id: 42,
                timestamp_observation: "2026-01-01T00:00:00Z".to_string(),
                ..Default::default()
            })?;
            rw.commit()?;
        }

        let database = Builder::new().create(models(), &db_path)?;
        let r = database.r_transaction()?;
        assert_eq!(r.scan().primary::<EventLocal>()?.all()?.count(), 0);
        drop(r);

        migrate_local_models(&database)?;

        let r = database.r_transaction()?;
        let events: Vec<EventLocal> = r
            .scan()
            .primary::<EventLocal>()?
            .all()?
            .collect::<std::result::Result<_, _>>()?;
        assert_eq!(events.len(), 1);
        assert_eq!(events[0].id_local.as_deref(), Some("event-before-bump"));
        assert_eq!(
            events[0].message.as_deref(),
            Some("written by an older build")
        );
        assert_eq!(events[0].modality.as_deref(), Some("video"));
        assert_eq!(events[0].connectivity_id, None);
        assert_eq!(events[0].sensor_pose_id, None);
        assert_eq!(
            r.scan().primary::<event::EventLocalV16>()?.all()?.count(),
            0
        );
        let artifacts: Vec<ArtifactLocal> = r
            .scan()
            .primary::<ArtifactLocal>()?
            .all()?
            .collect::<std::result::Result<_, _>>()?;
        assert_eq!(artifacts.len(), 1);
        assert_eq!(
            artifacts[0].id_local.as_deref(),
            Some("artifact-before-bump")
        );
        assert_eq!(artifacts[0].modality.as_deref(), Some("video"));
        assert_eq!(artifacts[0].source, None);
        assert_eq!(
            r.scan()
                .primary::<data::v13::ArtifactLocal>()?
                .all()?
                .count(),
            0
        );
        let sessions: Vec<SessionLocal> = r
            .scan()
            .primary::<SessionLocal>()?
            .all()?
            .collect::<std::result::Result<_, _>>()?;
        assert_eq!(sessions.len(), 1);
        assert_eq!(sessions[0].id_local.as_deref(), Some("session-before-bump"));
        assert_eq!(sessions[0].base_device_id, None);
        assert_eq!(sessions[0].base_device_software_version, None);
        assert_eq!(
            r.scan()
                .primary::<data::v14::SessionLocal>()?
                .all()?
                .count(),
            0
        );
        let connectivity: Vec<ConnectivityLocal> = r
            .scan()
            .primary::<ConnectivityLocal>()?
            .all()?
            .collect::<std::result::Result<_, _>>()?;
        assert_eq!(connectivity.len(), 1);
        assert_eq!(
            connectivity[0].id_local.as_deref(),
            Some("connectivity-before-bump")
        );
        assert_eq!(connectivity[0].frame_center_location, None);
        assert_eq!(connectivity[0].frame_center_altitude, None);
        assert_eq!(connectivity[0].base_device_online, None);
        assert_eq!(connectivity[0].frame_center_h11_index, None);
        assert_eq!(
            r.scan()
                .primary::<data::v14::ConnectivityLocal>()?
                .all()?
                .count(),
            0
        );
        assert_eq!(
            r.scan()
                .primary::<connectivity::ConnectivityLocalV17>()?
                .all()?
                .count(),
            0
        );
        let sensor_poses: Vec<SensorPoseLocal> = r
            .scan()
            .primary::<SensorPoseLocal>()?
            .all()?
            .collect::<std::result::Result<_, _>>()?;
        assert_eq!(sensor_poses.len(), 1);
        assert_eq!(
            sensor_poses[0].id_local.as_deref(),
            Some("pose-before-bump")
        );
        assert_eq!(sensor_poses[0].zoom, None);
        assert_eq!(
            r.scan()
                .primary::<sensor_pose::SensorPoseLocalV1>()?
                .all()?
                .count(),
            0
        );

        Ok(())
    }

    #[tokio::test]
    async fn test_wipe_specific_sessions() -> Result<()> {
        let mut sync_engine = create_test_sync_engine()?;

        let device_id = std::env::var(env::SCOUT_DEVICE_ID)
            .expect("SCOUT_DEVICE_ID required")
            .parse()
            .expect("SCOUT_DEVICE_ID must be valid integer");

        // Create two sessions with their descendants
        let mut session1 = SessionLocal::default();
        session1.set_id_local("wipe_specific_session1".to_string());
        session1.device_id = device_id;
        session1.timestamp_start = "2023-01-01T00:00:00Z".to_string();

        let mut session2 = SessionLocal::default();
        session2.set_id_local("wipe_specific_session2".to_string());
        session2.device_id = device_id;
        session2.timestamp_start = "2023-01-01T01:00:00Z".to_string();

        let mut event1 = EventLocal::default();
        event1.set_id_local("wipe_specific_event1".to_string());
        event1.device_id = device_id;
        event1.set_ancestor_id_local("wipe_specific_session1".to_string());
        event1.timestamp_observation = "2023-01-01T10:10:00Z".to_string();
        event1.media_type = MediaType::Image;

        let mut event2 = EventLocal::default();
        event2.set_id_local("wipe_specific_event2".to_string());
        event2.device_id = device_id;
        event2.set_ancestor_id_local("wipe_specific_session2".to_string());
        event2.timestamp_observation = "2023-01-01T11:10:00Z".to_string();
        event2.media_type = MediaType::Image;

        // Insert all items
        sync_engine.upsert_items(vec![session1, session2])?;
        sync_engine.upsert_items(vec![event1, event2])?;

        // Verify counts before wipe
        assert_eq!(sync_engine.get_table_count::<SessionLocal>()?, 2);
        assert_eq!(sync_engine.get_table_count::<EventLocal>()?, 2);

        // Wipe only session1
        sync_engine.wipe(Some(vec!["wipe_specific_session1".to_string()]))?;

        // Verify session1 and its event are gone, but session2 remains
        assert_eq!(sync_engine.get_table_count::<SessionLocal>()?, 1);
        assert_eq!(sync_engine.get_table_count::<EventLocal>()?, 1);

        // Verify session2 still exists
        let r = sync_engine.database.r_transaction()?;
        let mut found_session2 = false;
        for raw_session in r.scan().primary::<SessionLocal>()?.all()? {
            if let Ok(session) = raw_session {
                if session.id_local.as_deref() == Some("wipe_specific_session2") {
                    found_session2 = true;
                    break;
                }
            }
        }
        assert!(
            found_session2,
            "Session2 should still exist after wiping session1"
        );

        Ok(())
    }
}