woah

so what are these? auto gc and typed observation (because some flett shit is depending on ts)

Co-authored-by: Copilot <[email protected]>

wakey-control-plane/.sqlx/query-06c45563892b3d688153985aec411f588e80b5f17731f1fce97f9494a1155ffb.json

@@ -0,0 +1,12 @@
+{
+  "db_name": "SQLite",
+  "query": "INSERT INTO agent_device_observation_events\n             (event_id, agent_id, kind, action, mac, ip, hostname, ts_unix)\n             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Right": 8
+    },
+    "nullable": []
+  },
+  "hash": "06c45563892b3d688153985aec411f588e80b5f17731f1fce97f9494a1155ffb"
+}

wakey-control-plane/.sqlx/query-0a7efadab5487c6ebbf159b4fdd842faac795fe24901756ae67303acc8343e19.json

@@ -1,12 +0,0 @@
-{
-  "db_name": "SQLite",
-  "query": "INSERT INTO agent_device_observation_events\n                     (event_id, agent_id, kind, action, mac, ip, hostname, ts_unix)\n                     VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
-  "describe": {
-    "columns": [],
-    "parameters": {
-      "Right": 8
-    },
-    "nullable": []
-  },
-  "hash": "0a7efadab5487c6ebbf159b4fdd842faac795fe24901756ae67303acc8343e19"
-}

wakey-control-plane/.sqlx/query-273bcf6b604fbb48cad93630a1485590b1039149d1dd856db674dcc5584947d0.json

@@ -0,0 +1,12 @@
+{
+  "db_name": "SQLite",
+  "query": "DELETE FROM agent_device_observations\n         WHERE agent_id = ?1\n           AND kind = ?2\n           AND observation_key NOT IN (\n             SELECT observation_key\n             FROM agent_observation_snapshot_keys\n             WHERE agent_id = ?1 AND kind = ?2\n           )",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Right": 2
+    },
+    "nullable": []
+  },
+  "hash": "273bcf6b604fbb48cad93630a1485590b1039149d1dd856db674dcc5584947d0"
+}

wakey-control-plane/.sqlx/query-29f24b83cf513758b6a546e86be8aa7ce4b1f66d799aabacaeac02ff84996d24.json

@@ -0,0 +1,12 @@
+{
+  "db_name": "SQLite",
+  "query": "DELETE FROM agent_observation_snapshot_keys WHERE agent_id = ?1 AND kind = ?2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Right": 2
+    },
+    "nullable": []
+  },
+  "hash": "29f24b83cf513758b6a546e86be8aa7ce4b1f66d799aabacaeac02ff84996d24"
+}

wakey-control-plane/.sqlx/query-52d22161e936b39b57d119aab592426e88f60387ccba58888cc7d440012777ed.json

@@ -1,12 +0,0 @@
-{
-  "db_name": "SQLite",
-  "query": "INSERT INTO agent_device_observations\n                 (observation_key, agent_id, kind, mac, ip, hostname,\n                  first_seen_unix, last_seen_unix, last_action)\n                 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)\n                 ON CONFLICT(observation_key) DO UPDATE SET\n                   mac = excluded.mac,\n                   ip = excluded.ip,\n                   hostname = excluded.hostname,\n                   first_seen_unix = MIN(agent_device_observations.first_seen_unix, excluded.first_seen_unix),\n                   last_seen_unix = MAX(agent_device_observations.last_seen_unix, excluded.last_seen_unix),\n                   last_action = excluded.last_action",
-  "describe": {
-    "columns": [],
-    "parameters": {
-      "Right": 9
-    },
-    "nullable": []
-  },
-  "hash": "52d22161e936b39b57d119aab592426e88f60387ccba58888cc7d440012777ed"
-}

wakey-control-plane/.sqlx/query-8e7964dee0a2c21d377923d83df23454a104b68fc08f334c6e576bee273bc73b.json

@@ -0,0 +1,12 @@
+{
+  "db_name": "SQLite",
+  "query": "INSERT INTO agent_observation_snapshots\n         (agent_id, kind, last_dump_unix)\n         VALUES (?1, ?2, ?3)\n         ON CONFLICT(agent_id, kind) DO UPDATE SET\n           last_dump_unix = excluded.last_dump_unix",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Right": 3
+    },
+    "nullable": []
+  },
+  "hash": "8e7964dee0a2c21d377923d83df23454a104b68fc08f334c6e576bee273bc73b"
+}

wakey-control-plane/.sqlx/query-a47ba9ef887f67a7715bd919871ddd76f4ee09aa6ac0ee2f36d32c27d8a756d4.json

@@ -0,0 +1,12 @@
+{
+  "db_name": "SQLite",
+  "query": "INSERT INTO agent_device_observations\n         (observation_key, agent_id, kind, mac, ip, hostname,\n          first_seen_unix, last_seen_unix, last_action)\n         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)\n         ON CONFLICT(observation_key) DO UPDATE SET\n           mac = excluded.mac,\n           ip = excluded.ip,\n           hostname = excluded.hostname,\n           first_seen_unix = MIN(agent_device_observations.first_seen_unix, excluded.first_seen_unix),\n           last_seen_unix = MAX(agent_device_observations.last_seen_unix, excluded.last_seen_unix),\n           last_action = excluded.last_action",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Right": 9
+    },
+    "nullable": []
+  },
+  "hash": "a47ba9ef887f67a7715bd919871ddd76f4ee09aa6ac0ee2f36d32c27d8a756d4"
+}

wakey-control-plane/.sqlx/query-aebbe573b428c0aa5ea79d493868c8d9ba06d7a919d955d4c3b96c5fd4b96598.json

@@ -0,0 +1,12 @@
+{
+  "db_name": "SQLite",
+  "query": "INSERT INTO agent_observation_snapshot_keys (agent_id, kind, observation_key)\n             VALUES (?1, ?2, ?3)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Right": 3
+    },
+    "nullable": []
+  },
+  "hash": "aebbe573b428c0aa5ea79d493868c8d9ba06d7a919d955d4c3b96c5fd4b96598"
+}

wakey-control-plane/.sqlx/query-cf1c3bedcc9538d7af97e3e59b705b54dd95e7f219c2a8097187eb386d051e83.json

@@ -0,0 +1,12 @@
+{
+  "db_name": "SQLite",
+  "query": "DELETE FROM agent_device_observations WHERE last_seen_unix < ?1",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Right": 1
+    },
+    "nullable": []
+  },
+  "hash": "cf1c3bedcc9538d7af97e3e59b705b54dd95e7f219c2a8097187eb386d051e83"
+}

wakey-control-plane/.sqlx/query-edfcbcd3d530f59971e3e1c7c293321440dd9702ba6d6ac264fc9cf3d7cc7b5a.json

@@ -0,0 +1,12 @@
+{
+  "db_name": "SQLite",
+  "query": "DELETE FROM agent_device_observation_events WHERE ts_unix < ?1",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Right": 1
+    },
+    "nullable": []
+  },
+  "hash": "edfcbcd3d530f59971e3e1c7c293321440dd9702ba6d6ac264fc9cf3d7cc7b5a"
+}

wakey-control-plane/migrations/20260501000000_observation_snapshots.sql

@@ -0,0 +1,16 @@
+CREATE TABLE agent_observation_snapshots (
+    agent_id TEXT NOT NULL,
+    kind TEXT NOT NULL,
+    last_dump_unix INTEGER NOT NULL,
+    PRIMARY KEY(agent_id, kind)
+);
+
+CREATE TABLE agent_observation_snapshot_keys (
+    agent_id TEXT NOT NULL,
+    kind TEXT NOT NULL,
+    observation_key TEXT NOT NULL,
+    PRIMARY KEY(agent_id, kind, observation_key)
+);
+
+CREATE INDEX agent_observation_snapshot_keys_observation_idx
+    ON agent_observation_snapshot_keys(observation_key);

wakey-control-plane/src/api/control/fleet/mod.rs

@@ -90,7 +90,7 @@ pub async fn refresh_fleet_devices(
                 match inventory_result_to_observations(result) {
                     Ok(observations) => match state
                         .store
-                        .upsert_agent_observations(&agent_id, observations)
+                        .upsert_agent_observations_snapshot(&agent_id, "inventory", observations)
                         .await
                     {
                         Ok(accepted) => {

wakey-control-plane/src/api/control/observations.rs

@@ -1,3 +1,5 @@
+use std::collections::BTreeMap;
+
 use axum::Json;
 use axum::extract::{Path as AxumPath, Query, State};
 use axum::http::StatusCode;
@@ -74,38 +76,46 @@ pub async fn upload_agent_observations(
         ));
     }
 
-    let observations = req
-        .observations
-        .into_iter()
-        .map(|observation| AgentDeviceObservationInput {
-            kind: observation.kind,
+    let mut by_kind: BTreeMap<String, Vec<AgentDeviceObservationInput>> = BTreeMap::new();
+    for observation in req.observations {
+        let kind = observation.kind.trim().to_ascii_lowercase();
+        let entry = by_kind.entry(kind.clone()).or_default();
+        entry.push(AgentDeviceObservationInput {
+            kind,
             action: observation.action,
             mac: observation.mac,
             ip: observation.ip,
             hostname: observation.hostname,
             first_seen_unix: observation.first_seen_unix,
             last_seen_unix: observation.last_seen_unix,
-        })
-        .collect();
+        });
+    }
 
-    match state
-        .store
-        .upsert_agent_observations(&req.agent_id, observations)
-        .await
-    {
-        Ok(accepted) => Ok((
-            StatusCode::OK,
-            Json(UploadAgentObservationsResponse { accepted }),
-        )),
-        Err(err) => {
-            warn!(error = %err, agent_id = %req.agent_id, "failed to upload agent observations");
-            Err(json_error(
-                StatusCode::BAD_REQUEST,
-                "upload_observations_failed",
-                &err.to_string(),
-            ))
+    let mut accepted = 0usize;
+    for (kind, observations) in by_kind {
+        match state
+            .store
+            .upsert_agent_observations_snapshot(&req.agent_id, &kind, observations)
+            .await
+        {
+            Ok(written) => {
+                accepted = accepted.saturating_add(written);
+            }
+            Err(err) => {
+                warn!(error = %err, agent_id = %req.agent_id, kind = %kind, "failed to upload agent observations");
+                return Err(json_error(
+                    StatusCode::BAD_REQUEST,
+                    "upload_observations_failed",
+                    &err.to_string(),
+                ));
+            }
         }
     }
+
+    Ok((
+        StatusCode::OK,
+        Json(UploadAgentObservationsResponse { accepted }),
+    ))
 }
 
 pub async fn list_agent_observations(

wakey-control-plane/src/cli.rs

@@ -72,6 +72,9 @@ pub struct ServeArgs {
     #[arg(long)]
     pub enroll_token_ttl_seconds: Option<u64>,
 
+    #[arg(long)]
+    pub observation_retention_seconds: Option<u64>,
+
     #[arg(long)]
     pub pid_file: Option<PathBuf>,
 
@@ -126,6 +129,9 @@ pub struct InitConfigArgs {
     #[arg(long)]
     pub enroll_token_ttl_seconds: Option<u64>,
 
+    #[arg(long)]
+    pub observation_retention_seconds: Option<u64>,
+
     #[arg(long = "bootstrap-enroll-token", visible_alias = "enroll-token")]
     pub bootstrap_enroll_tokens: Vec<String>,
 

wakey-control-plane/src/config/init.rs

@@ -102,6 +102,10 @@ pub fn write_init_config(args: &InitConfigArgs) -> Result<Option<PathBuf>> {
             .or(base.enroll_token_ttl_seconds)
             .unwrap_or(86_400)
             .max(1),
+        observation_retention_seconds: args
+            .observation_retention_seconds
+            .or(base.observation_retention_seconds)
+            .unwrap_or(2_592_000),
         pid_file,
         ui_dist_dir,
         bootstrap_enroll_tokens,
@@ -155,6 +159,7 @@ pub fn bootstrap_config_if_missing(args: &ServeArgs) -> Result<bool> {
         ui_dist_dir: args.ui_dist_dir.clone(),
         command_timeout_ms: args.command_timeout_ms,
         enroll_token_ttl_seconds: args.enroll_token_ttl_seconds,
+        observation_retention_seconds: args.observation_retention_seconds,
         bootstrap_enroll_tokens: args.bootstrap_enroll_tokens.clone(),
         from_config: None,
         telemetry_otlp_endpoint: None,
@@ -219,6 +224,7 @@ json_logs = true
             ui_dist_dir: None,
             command_timeout_ms: Some(999),
             enroll_token_ttl_seconds: None,
+            observation_retention_seconds: None,
             bootstrap_enroll_tokens: Vec::new(),
             telemetry_otlp_endpoint: None,
             telemetry_service_name: None,

wakey-control-plane/src/config/resolve.rs

@@ -66,6 +66,12 @@ impl DaemonConfig {
                 .max(1),
         );
 
+        let observation_retention = Duration::from_secs(
+            args.observation_retention_seconds
+                .or(file.observation_retention_seconds)
+                .unwrap_or(2_592_000),
+        );
+
         let pid_file_raw = args
             .pid_file
             .clone()
@@ -94,6 +100,7 @@ impl DaemonConfig {
             state_file,
             command_timeout,
             enroll_token_ttl,
+            observation_retention,
             pid_file,
             ui_dist_dir,
             bootstrap_enroll_tokens,

wakey-control-plane/src/config/types.rs

@@ -12,6 +12,7 @@ pub struct DaemonConfig {
     pub state_file: PathBuf,
     pub command_timeout: Duration,
     pub enroll_token_ttl: Duration,
+    pub observation_retention: Duration,
     pub pid_file: PathBuf,
     pub ui_dist_dir: PathBuf,
     pub bootstrap_enroll_tokens: Vec<String>,
@@ -43,6 +44,7 @@ pub(crate) struct FileConfig {
     pub(crate) state_file: Option<PathBuf>,
     pub(crate) command_timeout_ms: Option<u64>,
     pub(crate) enroll_token_ttl_seconds: Option<u64>,
+    pub(crate) observation_retention_seconds: Option<u64>,
     pub(crate) pid_file: Option<PathBuf>,
     pub(crate) ui_dist_dir: Option<PathBuf>,
     #[serde(alias = "enroll_tokens")]
@@ -65,6 +67,7 @@ pub(crate) struct WritableConfig {
     pub(crate) state_file: PathBuf,
     pub(crate) command_timeout_ms: u64,
     pub(crate) enroll_token_ttl_seconds: u64,
+    pub(crate) observation_retention_seconds: u64,
     pub(crate) pid_file: PathBuf,
     pub(crate) ui_dist_dir: PathBuf,
     #[serde(skip_serializing_if = "Vec::is_empty")]

wakey-control-plane/src/runtime/mod.rs

@@ -226,6 +226,18 @@ pub async fn serve(daemon: config::DaemonConfig) -> Result<()> {
                         }
                         Err(err) => warn!(error = %err, "periodic gc failed"),
                     }
+                    match app_state
+                        .store
+                        .gc_stale_observations(daemon.observation_retention)
+                        .await
+                    {
+                        Ok(removed) => {
+                            if removed > 0 {
+                                info!(removed, "periodic gc removed stale observations");
+                            }
+                        }
+                        Err(err) => warn!(error = %err, "periodic observation gc failed"),
+                    }
                 }
                 join = &mut server => {
                     let _ = remove_pid_file(&daemon.pid_file);

wakey-control-plane/src/state/store.rs

@@ -533,6 +533,117 @@ mod tests {
         cleanup_dir(&dir);
     }
 
+    #[tokio::test]
+    async fn observation_snapshot_prunes_missing_keys() {
+        let (store, dir) = make_store().await;
+
+        store
+            .upsert_agent_observations(
+                "agent-a",
+                vec![
+                    crate::state::AgentDeviceObservationInput {
+                        kind: "dhcp".into(),
+                        action: "update".into(),
+                        mac: Some("AA:BB:CC:DD:EE:01".into()),
+                        ip: Some("192.168.1.10".into()),
+                        hostname: Some("first".into()),
+                        first_seen_unix: 10,
+                        last_seen_unix: 20,
+                    },
+                    crate::state::AgentDeviceObservationInput {
+                        kind: "dhcp".into(),
+                        action: "update".into(),
+                        mac: Some("AA:BB:CC:DD:EE:02".into()),
+                        ip: Some("192.168.1.11".into()),
+                        hostname: Some("second".into()),
+                        first_seen_unix: 10,
+                        last_seen_unix: 20,
+                    },
+                ],
+            )
+            .await
+            .expect("initial observations should upsert");
+
+        store
+            .upsert_agent_observations_snapshot(
+                "agent-a",
+                "dhcp",
+                vec![crate::state::AgentDeviceObservationInput {
+                    kind: "dhcp".into(),
+                    action: "update".into(),
+                    mac: Some("AA:BB:CC:DD:EE:01".into()),
+                    ip: Some("192.168.1.10".into()),
+                    hostname: Some("first".into()),
+                    first_seen_unix: 10,
+                    last_seen_unix: 30,
+                }],
+            )
+            .await
+            .expect("snapshot upsert should succeed");
+
+        let rows = store
+            .list_agent_observations(Some("agent-a"), 10)
+            .await
+            .expect("observations should list");
+        assert_eq!(rows.len(), 1);
+        assert_eq!(rows[0].mac.as_deref(), Some("aa:bb:cc:dd:ee:01"));
+
+        cleanup_dir(&dir);
+    }
+
+    #[tokio::test]
+    async fn observation_gc_removes_stale_rows() {
+        let (store, dir) = make_store().await;
+
+        let now = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .expect("time should be monotonic")
+            .as_secs();
+        let old = now.saturating_sub(10);
+
+        store
+            .upsert_agent_observations(
+                "agent-a",
+                vec![
+                    crate::state::AgentDeviceObservationInput {
+                        kind: "dhcp".into(),
+                        action: "update".into(),
+                        mac: Some("AA:BB:CC:DD:EE:10".into()),
+                        ip: Some("192.168.1.20".into()),
+                        hostname: Some("old".into()),
+                        first_seen_unix: old,
+                        last_seen_unix: old,
+                    },
+                    crate::state::AgentDeviceObservationInput {
+                        kind: "dhcp".into(),
+                        action: "update".into(),
+                        mac: Some("AA:BB:CC:DD:EE:11".into()),
+                        ip: Some("192.168.1.21".into()),
+                        hostname: Some("fresh".into()),
+                        first_seen_unix: now,
+                        last_seen_unix: now,
+                    },
+                ],
+            )
+            .await
+            .expect("observations should upsert");
+
+        let removed = store
+            .gc_stale_observations(Duration::from_secs(5))
+            .await
+            .expect("gc should succeed");
+        assert!(removed >= 1);
+
+        let rows = store
+            .list_agent_observations(Some("agent-a"), 10)
+            .await
+            .expect("observations should list");
+        assert_eq!(rows.len(), 1);
+        assert_eq!(rows[0].hostname.as_deref(), Some("fresh"));
+
+        cleanup_dir(&dir);
+    }
+
     #[tokio::test]
     async fn agent_observation_views_include_matching_known_device() {
         let (store, dir) = make_store().await;

wakey-control-plane/src/state/store/observations.rs

@@ -1,5 +1,134 @@
+use std::{collections::BTreeSet, ops::DerefMut};
+
 use super::*;
 
+async fn upsert_observation_tx(
+    tx: &mut Transaction<'_, Sqlite>,
+    observation: &AgentDeviceObservation,
+) -> Result<()> {
+    let first_seen_unix = i64::try_from(observation.first_seen_unix)
+        .context("observation first_seen overflow")?;
+    let last_seen_unix = i64::try_from(observation.last_seen_unix)
+        .context("observation last_seen overflow")?;
+    let current = get_observation_current_row(tx, &observation.observation_key)
+        .await
+        .context("failed checking existing observation")?;
+    let append_event = observation_current_changed(
+        current.as_ref(),
+        observation,
+        first_seen_unix,
+        last_seen_unix,
+    );
+    sqlx::query!(
+        "INSERT INTO agent_device_observations
+         (observation_key, agent_id, kind, mac, ip, hostname,
+          first_seen_unix, last_seen_unix, last_action)
+         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)
+         ON CONFLICT(observation_key) DO UPDATE SET
+           mac = excluded.mac,
+           ip = excluded.ip,
+           hostname = excluded.hostname,
+           first_seen_unix = MIN(agent_device_observations.first_seen_unix, excluded.first_seen_unix),
+           last_seen_unix = MAX(agent_device_observations.last_seen_unix, excluded.last_seen_unix),
+           last_action = excluded.last_action",
+        observation.observation_key,
+        observation.agent_id,
+        observation.kind,
+        observation.mac,
+        observation.ip,
+        observation.hostname,
+        first_seen_unix,
+        last_seen_unix,
+        observation.last_action
+    )
+    .execute(tx.deref_mut())
+    .await
+    .context("failed upserting agent device observation")?;
+
+    if append_event {
+        let event_id = format!("ode-{}", Uuid::new_v4());
+        sqlx::query!(
+            "INSERT INTO agent_device_observation_events
+             (event_id, agent_id, kind, action, mac, ip, hostname, ts_unix)
+             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
+            event_id,
+            observation.agent_id,
+            observation.kind,
+            observation.last_action,
+            observation.mac,
+            observation.ip,
+            observation.hostname,
+            last_seen_unix
+        )
+        .execute(tx.deref_mut())
+        .await
+        .context("failed appending agent device observation event")?;
+    }
+    Ok(())
+}
+
+async fn store_observation_snapshot(
+    tx: &mut Transaction<'_, Sqlite>,
+    agent_id: &str,
+    kind: &str,
+    snapshot_unix: u64,
+    keys: &BTreeSet<String>,
+) -> Result<()> {
+    let snapshot_unix = i64::try_from(snapshot_unix).context("snapshot timestamp overflow")?;
+    sqlx::query!(
+        "INSERT INTO agent_observation_snapshots
+         (agent_id, kind, last_dump_unix)
+         VALUES (?1, ?2, ?3)
+         ON CONFLICT(agent_id, kind) DO UPDATE SET
+           last_dump_unix = excluded.last_dump_unix",
+        agent_id,
+        kind,
+        snapshot_unix
+    )
+    .execute(tx.deref_mut())
+    .await
+    .context("failed upserting observation snapshot")?;
+
+    sqlx::query!(
+        "DELETE FROM agent_observation_snapshot_keys WHERE agent_id = ?1 AND kind = ?2",
+        agent_id,
+        kind
+    )
+    .execute(tx.deref_mut())
+    .await
+    .context("failed clearing observation snapshot keys")?;
+
+    for key in keys {
+        sqlx::query!(
+            "INSERT INTO agent_observation_snapshot_keys (agent_id, kind, observation_key)
+             VALUES (?1, ?2, ?3)",
+            agent_id,
+            kind,
+            key
+        )
+        .execute(tx.deref_mut())
+        .await
+        .context("failed inserting observation snapshot key")?;
+    }
+
+    sqlx::query!(
+        "DELETE FROM agent_device_observations
+         WHERE agent_id = ?1
+           AND kind = ?2
+           AND observation_key NOT IN (
+             SELECT observation_key
+             FROM agent_observation_snapshot_keys
+             WHERE agent_id = ?1 AND kind = ?2
+           )",
+        agent_id,
+        kind
+    )
+    .execute(tx.deref_mut())
+    .await
+    .context("failed removing stale observation keys")?;
+    Ok(())
+}
+
 impl Store {
     pub async fn upsert_agent_observations(
         &self,
@@ -14,64 +143,7 @@ impl Store {
         let mut written = 0usize;
         for observation in observations {
             let observation = normalize_agent_observation(agent_id, observation)?;
-            let first_seen_unix = i64::try_from(observation.first_seen_unix)
-                .context("observation first_seen overflow")?;
-            let last_seen_unix = i64::try_from(observation.last_seen_unix)
-                .context("observation last_seen overflow")?;
-            let current = get_observation_current_row(&mut tx, &observation.observation_key)
-                .await
-                .context("failed checking existing observation")?;
-            let append_event = observation_current_changed(
-                current.as_ref(),
-                &observation,
-                first_seen_unix,
-                last_seen_unix,
-            );
-            sqlx::query!(
-                "INSERT INTO agent_device_observations
-                 (observation_key, agent_id, kind, mac, ip, hostname,
-                  first_seen_unix, last_seen_unix, last_action)
-                 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)
-                 ON CONFLICT(observation_key) DO UPDATE SET
-                   mac = excluded.mac,
-                   ip = excluded.ip,
-                   hostname = excluded.hostname,
-                   first_seen_unix = MIN(agent_device_observations.first_seen_unix, excluded.first_seen_unix),
-                   last_seen_unix = MAX(agent_device_observations.last_seen_unix, excluded.last_seen_unix),
-                   last_action = excluded.last_action",
-                observation.observation_key,
-                observation.agent_id,
-                observation.kind,
-                observation.mac,
-                observation.ip,
-                observation.hostname,
-                first_seen_unix,
-                last_seen_unix,
-                observation.last_action
-            )
-            .execute(&mut *tx)
-            .await
-            .context("failed upserting agent device observation")?;
-
-            if append_event {
-                let event_id = format!("ode-{}", Uuid::new_v4());
-                sqlx::query!(
-                    "INSERT INTO agent_device_observation_events
-                     (event_id, agent_id, kind, action, mac, ip, hostname, ts_unix)
-                     VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
-                    event_id,
-                    observation.agent_id,
-                    observation.kind,
-                    observation.last_action,
-                    observation.mac,
-                    observation.ip,
-                    observation.hostname,
-                    last_seen_unix
-                )
-                .execute(&mut *tx)
-                .await
-                .context("failed appending agent device observation event")?;
-            }
+            upsert_observation_tx(&mut tx, &observation).await?;
             written = written.saturating_add(1);
         }
         tx.commit()
@@ -80,7 +152,68 @@ impl Store {
         Ok(written)
     }
 
-    #[cfg_attr(not(test), allow(dead_code))]
+    pub async fn upsert_agent_observations_snapshot(
+        &self,
+        agent_id: &str,
+        kind: &str,
+        observations: Vec<AgentDeviceObservationInput>,
+    ) -> Result<usize> {
+        let mut tx = self
+            .pool
+            .begin()
+            .await
+            .context("failed starting observation snapshot transaction")?;
+        let mut written = 0usize;
+        let mut snapshot_keys = BTreeSet::new();
+        let kind = normalize_required_text(kind, "observation kind")?.to_ascii_lowercase();
+        for observation in observations {
+            let observation = normalize_agent_observation(agent_id, observation)?;
+            if observation.kind != kind {
+                anyhow::bail!(
+                    "observation kind mismatch: expected {kind} got {}",
+                    observation.kind
+                );
+            }
+            snapshot_keys.insert(observation.observation_key.clone());
+            upsert_observation_tx(&mut tx, &observation).await?;
+            written = written.saturating_add(1);
+        }
+
+        let snapshot_unix = now_unix();
+        store_observation_snapshot(&mut tx, agent_id, &kind, snapshot_unix, &snapshot_keys)
+            .await?;
+
+        tx.commit()
+            .await
+            .context("failed committing observation snapshot transaction")?;
+        Ok(written)
+    }
+
+    pub async fn gc_stale_observations(&self, retention: Duration) -> Result<u64> {
+        if retention.as_secs() == 0 {
+            return Ok(0);
+        }
+        let cutoff = now_unix().saturating_sub(retention.as_secs());
+        let cutoff = i64::try_from(cutoff).context("observation retention overflow")?;
+        let removed_observations = sqlx::query!(
+            "DELETE FROM agent_device_observations WHERE last_seen_unix < ?1",
+            cutoff
+        )
+        .execute(&self.pool)
+        .await
+        .context("failed removing stale observations")?
+        .rows_affected();
+        let removed_events = sqlx::query!(
+            "DELETE FROM agent_device_observation_events WHERE ts_unix < ?1",
+            cutoff
+        )
+        .execute(&self.pool)
+        .await
+        .context("failed removing stale observation events")?
+        .rows_affected();
+        Ok(removed_observations.saturating_add(removed_events))
+    }
+
     pub async fn list_agent_observations(
         &self,
         agent_id: Option<&str>,

wakey-control-plane/src/ws.rs

@@ -4,6 +4,7 @@ use axum::extract::ws::{Message, WebSocket, WebSocketUpgrade};
 use axum::response::IntoResponse;
 use futures_util::{SinkExt, StreamExt};
 use serde::Deserialize;
+use std::collections::BTreeMap;
 use std::time::Instant;
 use tokio::sync::mpsc;
 use tracing::{debug, info, info_span, warn};
@@ -259,31 +260,40 @@ async fn process_agent_text(
                 anyhow::bail!("observations for unauthenticated or mismatched agent");
             }
             ensure_current_session(state, &agent_id, connection_id).await?;
-            let inputs = observations
-                .into_iter()
-                .map(|observation| crate::state::AgentDeviceObservationInput {
-                    kind: observation.kind,
-                    action: observation.action,
-                    mac: observation.mac,
-                    ip: observation.ip.map(|ip| ip.to_string()),
-                    hostname: observation.hostname,
-                    first_seen_unix: observation.first_seen_unix,
-                    last_seen_unix: observation.last_seen_unix,
-                })
-                .collect();
-            match state
-                .store
-                .upsert_agent_observations(&agent_id, inputs)
-                .await
-            {
-                Ok(accepted) => {
-                    debug!(agent_id = %agent_id, accepted, "agent websocket observations accepted");
-                }
-                Err(err) => {
-                    warn!(agent_id = %agent_id, error = %err, "failed to store websocket observations");
-                    anyhow::bail!("failed to store observations: {err}");
+            let mut by_kind: BTreeMap<String, Vec<crate::state::AgentDeviceObservationInput>> =
+                BTreeMap::new();
+            for observation in observations {
+                let kind = observation.kind.trim().to_ascii_lowercase();
+                by_kind
+                    .entry(kind.clone())
+                    .or_default()
+                    .push(crate::state::AgentDeviceObservationInput {
+                        kind,
+                        action: observation.action,
+                        mac: observation.mac,
+                        ip: observation.ip.map(|ip| ip.to_string()),
+                        hostname: observation.hostname,
+                        first_seen_unix: observation.first_seen_unix,
+                        last_seen_unix: observation.last_seen_unix,
+                    });
+            }
+            let mut accepted = 0usize;
+            for (kind, inputs) in by_kind {
+                match state
+                    .store
+                    .upsert_agent_observations_snapshot(&agent_id, &kind, inputs)
+                    .await
+                {
+                    Ok(written) => {
+                        accepted = accepted.saturating_add(written);
+                    }
+                    Err(err) => {
+                        warn!(agent_id = %agent_id, error = %err, kind = %kind, "failed to store websocket observations");
+                        anyhow::bail!("failed to store observations: {err}");
+                    }
                 }
             }
+            debug!(agent_id = %agent_id, accepted, "agent websocket observations accepted");
         }
         IncomingClientMessage::Result { request_id, result } => {
             let agent_id = authed_agent_id