//! SQLite-backed durable state owned by one dedicated database thread.

use std::{path::Path, thread};

use rusqlite::{Connection, OptionalExtension, Transaction, params};
use tokio::sync::{mpsc, oneshot};

use crate::{
    errors::StateStoreError,
    models::{HealthSnapshot, InvocationCandidate, InvocationRecord, InvocationStatus, SyncState},
};

/// Provides asynchronous access to Lisa's authoritative SQLite state.
#[derive(Clone)]
pub struct StateStore {
    sender: mpsc::UnboundedSender<StateCommand>,
}

enum StateCommand {
    SyncState(oneshot::Sender<Result<SyncState, StateStoreError>>),
    RecordBaseline {
        next_batch: String,
        now_ms: u64,
        response: oneshot::Sender<Result<(), StateStoreError>>,
    },
    IngestBatch {
        next_batch: String,
        room_cursors: Vec<(String, Option<String>)>,
        candidates: Vec<InvocationCandidate>,
        now_ms: u64,
        response: oneshot::Sender<Result<usize, StateStoreError>>,
    },
    ClaimNext {
        room_id: String,
        now_ms: u64,
        lease_until_ms: u64,
        response: oneshot::Sender<Result<Option<InvocationRecord>, StateStoreError>>,
    },
    MarkIgnored {
        event_id: String,
        reason: String,
        now_ms: u64,
        response: oneshot::Sender<Result<(), StateStoreError>>,
    },
    StoreReply {
        event_id: String,
        response_json: String,
        now_ms: u64,
        response: oneshot::Sender<Result<(), StateStoreError>>,
    },
    RecordSent {
        event_id: String,
        sent_event_id: String,
        now_ms: u64,
        response: oneshot::Sender<Result<(), StateStoreError>>,
    },
    IsBotEvent {
        room_id: String,
        event_id: String,
        response: oneshot::Sender<Result<bool, StateStoreError>>,
    },
    ReadyToSend(oneshot::Sender<Result<Vec<InvocationRecord>, StateStoreError>>),
    RequeueUnfinished {
        now_ms: u64,
        response: oneshot::Sender<Result<usize, StateStoreError>>,
    },
    Health {
        now_ms: u64,
        response: oneshot::Sender<Result<HealthSnapshot, StateStoreError>>,
    },
    ReleaseProcessing {
        now_ms: u64,
        response: oneshot::Sender<Result<usize, StateStoreError>>,
    },
}

impl StateStore {
    /// Starts the dedicated SQLite owner thread and runs all migrations.
    pub fn open(path: impl AsRef<Path>) -> Result<Self, StateStoreError> {
        let connection = Connection::open(path)?;
        configure_connection(&connection)?;
        migrate(&connection)?;

        let (sender, mut receiver) = mpsc::unbounded_channel();
        thread::Builder::new()
            .name("lisa-state".to_owned())
            .spawn(move || {
                let connection = connection;
                while let Some(command) = receiver.blocking_recv() {
                    handle_command(&connection, command);
                }
            })
            .map_err(|error| StateStoreError::InvalidData(error.to_string()))?;

        Ok(Self { sender })
    }

    /// Returns the persisted controlled-sync cursor.
    pub async fn syncState(&self) -> Result<SyncState, StateStoreError> {
        self.request(StateCommand::SyncState).await
    }

    /// Records an initial sync cursor while deliberately discarding its timeline.
    pub async fn recordBaseline(
        &self,
        next_batch: String,
        now_ms: u64,
    ) -> Result<(), StateStoreError> {
        self.request(|response| StateCommand::RecordBaseline {
            next_batch,
            now_ms,
            response,
        })
        .await
    }

    /// Atomically records a sync cursor, room cursors, and new candidates.
    pub async fn ingestBatch(
        &self,
        next_batch: String,
        room_cursors: Vec<(String, Option<String>)>,
        candidates: Vec<InvocationCandidate>,
        now_ms: u64,
    ) -> Result<usize, StateStoreError> {
        self.request(|response| StateCommand::IngestBatch {
            next_batch,
            room_cursors,
            candidates,
            now_ms,
            response,
        })
        .await
    }

    /// Leases the oldest eligible invocation for one room.
    pub async fn claimNext(
        &self,
        room_id: String,
        now_ms: u64,
        lease_until_ms: u64,
    ) -> Result<Option<InvocationRecord>, StateStoreError> {
        self.request(|response| StateCommand::ClaimNext {
            room_id,
            now_ms,
            lease_until_ms,
            response,
        })
        .await
    }

    /// Marks a candidate as ignored and clears its raw room content.
    pub async fn markIgnored(
        &self,
        event_id: String,
        reason: String,
        now_ms: u64,
    ) -> Result<(), StateStoreError> {
        self.request(|response| StateCommand::MarkIgnored {
            event_id,
            reason,
            now_ms,
            response,
        })
        .await
    }

    /// Persists an exact reply payload before any Matrix send attempt.
    pub async fn storeReply(
        &self,
        event_id: String,
        response_json: String,
        now_ms: u64,
    ) -> Result<(), StateStoreError> {
        self.request(|response| StateCommand::StoreReply {
            event_id,
            response_json,
            now_ms,
            response,
        })
        .await
    }

    /// Records Matrix acceptance and clears sensitive invocation content.
    pub async fn recordSent(
        &self,
        event_id: String,
        sent_event_id: String,
        now_ms: u64,
    ) -> Result<(), StateStoreError> {
        self.request(|response| StateCommand::RecordSent {
            event_id,
            sent_event_id,
            now_ms,
            response,
        })
        .await
    }

    /// Returns whether Lisa previously sent an event in the given room.
    pub async fn isBotEvent(
        &self,
        room_id: String,
        event_id: String,
    ) -> Result<bool, StateStoreError> {
        self.request(|response| StateCommand::IsBotEvent {
            room_id,
            event_id,
            response,
        })
        .await
    }

    /// Returns persisted reply payloads that are safe to retry after a restart.
    pub async fn readyToSend(&self) -> Result<Vec<InvocationRecord>, StateStoreError> {
        self.request(StateCommand::ReadyToSend).await
    }

    /// Requeues expired processing rows and all persisted sends after restart.
    pub async fn requeueUnfinished(&self, now_ms: u64) -> Result<usize, StateStoreError> {
        self.request(|response| StateCommand::RequeueUnfinished { now_ms, response })
            .await
    }

    /// Returns non-sensitive local queue information for an operator health check.
    pub async fn health(&self, now_ms: u64) -> Result<HealthSnapshot, StateStoreError>
    {
        self.request(|response| StateCommand::Health { now_ms, response }).await
    }

    /// Releases active processing leases during a controlled shutdown.
    pub async fn releaseProcessing(&self, now_ms: u64) -> Result<usize, StateStoreError>
    {
        self.request(|response| StateCommand::ReleaseProcessing { now_ms, response }).await
    }

    async fn request<T, F>(&self, make_command: F) -> Result<T, StateStoreError>
    where
        F: FnOnce(oneshot::Sender<Result<T, StateStoreError>>) -> StateCommand,
    {
        let (response_sender, response_receiver) = oneshot::channel();
        self.sender
            .send(make_command(response_sender))
            .map_err(|_| StateStoreError::WorkerStopped)?;
        response_receiver
            .await
            .map_err(|_| StateStoreError::WorkerStopped)?
    }
}

fn handle_command(connection: &Connection, command: StateCommand) {
    match command {
        StateCommand::SyncState(response) => send(response, read_sync_state(connection)),
        StateCommand::RecordBaseline {
            next_batch,
            now_ms,
            response,
        } => send(response, record_baseline(connection, &next_batch, now_ms)),
        StateCommand::IngestBatch {
            next_batch,
            room_cursors,
            candidates,
            now_ms,
            response,
        } => send(
            response,
            ingest_batch(connection, &next_batch, &room_cursors, &candidates, now_ms),
        ),
        StateCommand::ClaimNext {
            room_id,
            now_ms,
            lease_until_ms,
            response,
        } => send(
            response,
            claim_next(connection, &room_id, now_ms, lease_until_ms),
        ),
        StateCommand::MarkIgnored {
            event_id,
            reason,
            now_ms,
            response,
        } => send(
            response,
            mark_ignored(connection, &event_id, &reason, now_ms),
        ),
        StateCommand::StoreReply {
            event_id,
            response_json,
            now_ms,
            response,
        } => send(
            response,
            store_reply(connection, &event_id, &response_json, now_ms),
        ),
        StateCommand::RecordSent {
            event_id,
            sent_event_id,
            now_ms,
            response,
        } => send(
            response,
            record_sent(connection, &event_id, &sent_event_id, now_ms),
        ),
        StateCommand::IsBotEvent {
            room_id,
            event_id,
            response,
        } => send(response, is_bot_event(connection, &room_id, &event_id)),
        StateCommand::ReadyToSend(response) => send(response, ready_to_send(connection)),
        StateCommand::RequeueUnfinished { now_ms, response } => {
            send(response, requeue_unfinished(connection, now_ms));
        }
        StateCommand::Health { now_ms, response } => send(response, health(connection, now_ms)),
        StateCommand::ReleaseProcessing { now_ms, response } => {
            send(response, release_processing(connection, now_ms));
        }
    }
}

fn send<T>(
    response: oneshot::Sender<Result<T, StateStoreError>>,
    result: Result<T, StateStoreError>,
) {
    let _ = response.send(result);
}

fn configure_connection(connection: &Connection) -> Result<(), StateStoreError> {
    connection.execute_batch(
        "PRAGMA foreign_keys = ON; PRAGMA journal_mode = WAL; PRAGMA busy_timeout = 5000;",
    )?;
    Ok(())
}

fn migrate(connection: &Connection) -> Result<(), StateStoreError> {
    connection.execute_batch(
        "CREATE TABLE IF NOT EXISTS schema_migrations(
            version INTEGER PRIMARY KEY,
            applied_at_ms INTEGER NOT NULL
        );",
    )?;
    let applied: Option<i64> =
        connection.query_row("SELECT MAX(version) FROM schema_migrations", [], |row| {
            row.get(0)
        })?;
    if applied.unwrap_or_default() >= 1 {
        return Ok(());
    }
    let transaction = connection.unchecked_transaction()?;
    transaction.execute_batch(
        "CREATE TABLE sync_state(
            singleton INTEGER PRIMARY KEY CHECK(singleton = 1),
            next_batch TEXT,
            initialized INTEGER NOT NULL CHECK(initialized IN (0, 1)),
            updated_at_ms INTEGER NOT NULL
        );
        INSERT INTO sync_state(singleton, next_batch, initialized, updated_at_ms)
        VALUES(1, NULL, 0, 0);
        CREATE TABLE room_cursors(
            room_id TEXT PRIMARY KEY,
            last_timeline_event_id TEXT,
            updated_at_ms INTEGER NOT NULL
        );
        CREATE TABLE invocations(
            event_id TEXT PRIMARY KEY,
            room_id TEXT NOT NULL,
            sender_id TEXT NOT NULL,
            received_at_ms INTEGER NOT NULL,
            status TEXT NOT NULL,
            raw_event_json TEXT,
            response_json TEXT,
            matrix_txn_id TEXT NOT NULL UNIQUE,
            sent_event_id TEXT,
            attempt_count INTEGER NOT NULL DEFAULT 0,
            available_at_ms INTEGER NOT NULL,
            last_error_code TEXT,
            updated_at_ms INTEGER NOT NULL
        );
        CREATE INDEX invocations_ready
        ON invocations(status, available_at_ms, received_at_ms);
        CREATE INDEX invocations_room
        ON invocations(room_id, received_at_ms);
        CREATE TABLE bot_events(
            event_id TEXT PRIMARY KEY,
            room_id TEXT NOT NULL,
            invocation_event_id TEXT NOT NULL UNIQUE,
            sent_at_ms INTEGER NOT NULL,
            FOREIGN KEY(invocation_event_id) REFERENCES invocations(event_id)
        );",
    )?;
    transaction.execute(
        "INSERT INTO schema_migrations(version, applied_at_ms) VALUES(1, 0)",
        [],
    )?;
    transaction.commit()?;
    Ok(())
}

fn read_sync_state(connection: &Connection) -> Result<SyncState, StateStoreError> {
    connection
        .query_row(
            "SELECT initialized, next_batch FROM sync_state WHERE singleton = 1",
            [],
            |row| {
                Ok(SyncState {
                    initialized: row.get::<_, i64>(0)? != 0,
                    next_batch: row.get(1)?,
                })
            },
        )
        .map_err(Into::into)
}

fn record_baseline(
    connection: &Connection,
    next_batch: &str,
    now_ms: u64,
) -> Result<(), StateStoreError> {
    connection.execute(
        "UPDATE sync_state SET next_batch = ?1, initialized = 1, updated_at_ms = ?2 WHERE singleton = 1",
        params![next_batch, now_ms],
    )?;
    Ok(())
}

fn ingest_batch(
    connection: &Connection,
    next_batch: &str,
    room_cursors: &[(String, Option<String>)],
    candidates: &[InvocationCandidate],
    now_ms: u64,
) -> Result<usize, StateStoreError> {
    let transaction = connection.unchecked_transaction()?;
    let mut inserted = 0;
    for candidate in candidates {
        inserted += transaction.execute(
            "INSERT OR IGNORE INTO invocations(
                event_id, room_id, sender_id, received_at_ms, status,
                raw_event_json, matrix_txn_id, available_at_ms, updated_at_ms
            ) VALUES(?1, ?2, ?3, ?4, 'candidate', ?5, ?6, ?7, ?7)",
            params![
                candidate.event_id,
                candidate.room_id,
                candidate.sender_id,
                candidate.received_at_ms,
                candidate.raw_event_json,
                candidate.matrix_txn_id,
                now_ms,
            ],
        )?;
    }
    for (room_id, event_id) in room_cursors {
        transaction.execute(
            "INSERT INTO room_cursors(room_id, last_timeline_event_id, updated_at_ms)
             VALUES(?1, ?2, ?3)
             ON CONFLICT(room_id) DO UPDATE SET
                 last_timeline_event_id = excluded.last_timeline_event_id,
                 updated_at_ms = excluded.updated_at_ms",
            params![room_id, event_id, now_ms],
        )?;
    }
    transaction.execute(
        "UPDATE sync_state SET next_batch = ?1, initialized = 1, updated_at_ms = ?2 WHERE singleton = 1",
        params![next_batch, now_ms],
    )?;
    transaction.commit()?;
    Ok(inserted)
}

fn claim_next(
    connection: &Connection,
    room_id: &str,
    now_ms: u64,
    lease_until_ms: u64,
) -> Result<Option<InvocationRecord>, StateStoreError> {
    let transaction = connection.unchecked_transaction()?;
    let event_id = transaction
        .query_row(
            "SELECT event_id FROM invocations
         WHERE room_id = ?1 AND status = 'candidate' AND available_at_ms <= ?2
         ORDER BY received_at_ms, event_id LIMIT 1",
            params![room_id, now_ms],
            |row| row.get::<_, String>(0),
        )
        .optional()?;
    let Some(event_id) = event_id else {
        transaction.commit()?;
        return Ok(None);
    };
    transaction.execute(
        "UPDATE invocations SET status = 'processing', attempt_count = attempt_count + 1,
         available_at_ms = ?2, updated_at_ms = ?3 WHERE event_id = ?1 AND status = 'candidate'",
        params![event_id, lease_until_ms, now_ms],
    )?;
    let record = read_invocation(&transaction, &event_id)?;
    transaction.commit()?;
    Ok(Some(record))
}

fn mark_ignored(
    connection: &Connection,
    event_id: &str,
    reason: &str,
    now_ms: u64,
) -> Result<(), StateStoreError> {
    let changed = connection.execute(
        "UPDATE invocations SET status = 'ignored', raw_event_json = NULL,
         response_json = NULL, last_error_code = ?2, updated_at_ms = ?3
         WHERE event_id = ?1 AND status IN ('candidate', 'processing')",
        params![event_id, reason, now_ms],
    )?;
    if changed == 0 {
        return Err(StateStoreError::InvalidData(
            "cannot ignore invocation in its current state".to_owned(),
        ));
    }
    Ok(())
}

fn store_reply(
    connection: &Connection,
    event_id: &str,
    response_json: &str,
    now_ms: u64,
) -> Result<(), StateStoreError> {
    let changed = connection.execute(
        "UPDATE invocations SET status = 'ready_to_send', response_json = ?2,
         available_at_ms = ?3, updated_at_ms = ?3
         WHERE event_id = ?1 AND status = 'processing'",
        params![event_id, response_json, now_ms],
    )?;
    if changed == 0 {
        return Err(StateStoreError::InvalidData(
            "cannot store reply for invocation in its current state".to_owned(),
        ));
    }
    Ok(())
}

fn record_sent(
    connection: &Connection,
    event_id: &str,
    sent_event_id: &str,
    now_ms: u64,
) -> Result<(), StateStoreError> {
    let transaction = connection.unchecked_transaction()?;
    let room_id: String = transaction.query_row(
        "SELECT room_id FROM invocations WHERE event_id = ?1 AND status = 'ready_to_send'",
        [event_id],
        |row| row.get(0),
    )?;
    transaction.execute(
        "INSERT INTO bot_events(event_id, room_id, invocation_event_id, sent_at_ms)
         VALUES(?1, ?2, ?3, ?4)",
        params![sent_event_id, room_id, event_id, now_ms],
    )?;
    transaction.execute(
        "UPDATE invocations SET status = 'sent', raw_event_json = NULL,
         response_json = NULL, sent_event_id = ?2, updated_at_ms = ?3 WHERE event_id = ?1",
        params![event_id, sent_event_id, now_ms],
    )?;
    transaction.commit()?;
    Ok(())
}

fn is_bot_event(
    connection: &Connection,
    room_id: &str,
    event_id: &str,
) -> Result<bool, StateStoreError> {
    connection
        .query_row(
            "SELECT EXISTS(SELECT 1 FROM bot_events WHERE room_id = ?1 AND event_id = ?2)",
            params![room_id, event_id],
            |row| row.get::<_, i64>(0),
        )
        .map(|exists| exists != 0)
        .map_err(Into::into)
}

fn ready_to_send(connection: &Connection) -> Result<Vec<InvocationRecord>, StateStoreError> {
    let mut statement = connection.prepare(
        "SELECT event_id, room_id, sender_id, received_at_ms, status, raw_event_json,
         response_json, matrix_txn_id, sent_event_id, attempt_count, available_at_ms,
         last_error_code FROM invocations WHERE status = 'ready_to_send'
         ORDER BY received_at_ms, event_id",
    )?;
    let rows = statement.query_map([], row_to_invocation)?;
    rows.collect::<Result<Vec<_>, _>>().map_err(Into::into)
}

fn requeue_unfinished(connection: &Connection, now_ms: u64) -> Result<usize, StateStoreError> {
    let changed = connection.execute(
        "UPDATE invocations SET status = 'candidate', available_at_ms = ?1,
         updated_at_ms = ?1 WHERE status = 'processing' AND available_at_ms <= ?1",
        [now_ms],
    )?;
    Ok(changed)
}

fn health(connection: &Connection, now_ms: u64) -> Result<HealthSnapshot, StateStoreError>
{
    let candidate_count = connection.query_row(
        "SELECT COUNT(*) FROM invocations WHERE status = 'candidate'",
        [],
        |row| row.get(0),
    )?;
    let oldest_ready: Option<u64> = connection.query_row(
        "SELECT MIN(updated_at_ms) FROM invocations WHERE status = 'ready_to_send'",
        [],
        |row| row.get(0),
    )?;
    let ready_to_send_count = connection.query_row(
        "SELECT COUNT(*) FROM invocations WHERE status = 'ready_to_send'",
        [],
        |row| row.get(0),
    )?;
    Ok(HealthSnapshot {
        candidate_count,
        ready_to_send_count,
        oldest_ready_age_ms: oldest_ready.map(|timestamp| now_ms.saturating_sub(timestamp)),
    })
}

fn release_processing(connection: &Connection, now_ms: u64) -> Result<usize, StateStoreError>
{
    let changed = connection.execute(
        "UPDATE invocations SET status = 'candidate', available_at_ms = ?1,
         updated_at_ms = ?1 WHERE status = 'processing'",
        [now_ms],
    )?;
    Ok(changed)
}

fn read_invocation(
    transaction: &Transaction<'_>,
    event_id: &str,
) -> Result<InvocationRecord, StateStoreError> {
    transaction
        .query_row(
            "SELECT event_id, room_id, sender_id, received_at_ms, status, raw_event_json,
         response_json, matrix_txn_id, sent_event_id, attempt_count, available_at_ms,
         last_error_code FROM invocations WHERE event_id = ?1",
            [event_id],
            row_to_invocation,
        )
        .map_err(Into::into)
}

fn row_to_invocation(row: &rusqlite::Row<'_>) -> rusqlite::Result<InvocationRecord> {
    let status: String = row.get(4)?;
    let status = InvocationStatus::parse(&status).map_err(|error| {
        rusqlite::Error::FromSqlConversionFailure(
            4,
            rusqlite::types::Type::Text,
            Box::new(std::io::Error::new(std::io::ErrorKind::InvalidData, error)),
        )
    })?;
    Ok(InvocationRecord {
        event_id: row.get(0)?,
        room_id: row.get(1)?,
        sender_id: row.get(2)?,
        received_at_ms: row.get(3)?,
        status,
        raw_event_json: row.get(5)?,
        response_json: row.get(6)?,
        matrix_txn_id: row.get(7)?,
        sent_event_id: row.get(8)?,
        attempt_count: row.get(9)?,
        available_at_ms: row.get(10)?,
        last_error_code: row.get(11)?,
    })
}

#[cfg(test)]
mod tests {
    use tempfile::tempdir;

    use super::StateStore;
    use crate::models::{InvocationCandidate, InvocationStatus};

    fn candidate(event_id: &str) -> InvocationCandidate {
        InvocationCandidate {
            event_id: event_id.to_owned(),
            room_id: "!room:example.test".to_owned(),
            sender_id: "@human:example.test".to_owned(),
            received_at_ms: 10,
            raw_event_json: "{}".to_owned(),
            matrix_txn_id: format!("txn_{event_id}"),
        }
    }

    #[tokio::test]
    async fn deduplicates_candidates_and_preserves_cursor() {
        let directory = tempdir().unwrap();
        let store = StateStore::open(directory.path().join("state.sqlite3")).unwrap();
        let first = store
            .ingestBatch("s1".to_owned(), vec![], vec![candidate("$one")], 20)
            .await
            .unwrap();
        let second = store
            .ingestBatch("s2".to_owned(), vec![], vec![candidate("$one")], 30)
            .await
            .unwrap();

        assert_eq!(first, 1);
        assert_eq!(second, 0);
        assert_eq!(
            store.syncState().await.unwrap().next_batch.as_deref(),
            Some("s2")
        );
    }

    #[tokio::test]
    async fn persists_reply_before_recording_single_sent_event() {
        let directory = tempdir().unwrap();
        let store = StateStore::open(directory.path().join("state.sqlite3")).unwrap();
        store
            .ingestBatch("s1".to_owned(), vec![], vec![candidate("$one")], 20)
            .await
            .unwrap();
        let invocation = store
            .claimNext("!room:example.test".to_owned(), 21, 100)
            .await
            .unwrap()
            .unwrap();
        assert_eq!(invocation.status, InvocationStatus::Processing);
        store
            .storeReply("$one".to_owned(), "{\"body\":\"test\"}".to_owned(), 22)
            .await
            .unwrap();
        let ready = store.readyToSend().await.unwrap();
        assert_eq!(ready.len(), 1);
        assert_eq!(ready[0].matrix_txn_id, "txn_$one");
        assert_eq!(
            ready[0].response_json.as_deref(),
            Some("{\"body\":\"test\"}")
        );
        store
            .recordSent("$one".to_owned(), "$reply".to_owned(), 23)
            .await
            .unwrap();

        assert!(
            store
                .isBotEvent("!room:example.test".to_owned(), "$reply".to_owned())
                .await
                .unwrap()
        );
        assert!(
            store
                .recordSent("$one".to_owned(), "$reply".to_owned(), 24)
                .await
                .is_err()
        );
    }

    #[tokio::test]
    async fn requeues_only_an_expired_processing_lease() {
        let directory = tempdir().unwrap();
        let store = StateStore::open(directory.path().join("state.sqlite3")).unwrap();
        store
            .ingestBatch("s1".to_owned(), vec![], vec![candidate("$one")], 20)
            .await
            .unwrap();
        store
            .claimNext("!room:example.test".to_owned(), 21, 100)
            .await
            .unwrap();

        assert_eq!(store.requeueUnfinished(99).await.unwrap(), 0);
        assert_eq!(store.requeueUnfinished(100).await.unwrap(), 1);
        assert!(
            store
                .claimNext("!room:example.test".to_owned(), 101, 200)
                .await
                .unwrap()
                .is_some()
        );
    }

    #[tokio::test]
    async fn health_exposes_counts_without_invocation_content() {
        let directory = tempdir().unwrap();
        let store = StateStore::open(directory.path().join("state.sqlite3")).unwrap();
        store.ingestBatch("s1".to_owned(), vec![], vec![candidate("$one")], 20)
            .await.unwrap();
        let health = store.health(30).await.unwrap();
        assert_eq!(health.candidate_count, 1);
        assert_eq!(health.ready_to_send_count, 0);
        assert_eq!(health.oldest_ready_age_ms, None);
    }
}