[mobile config] track asset ownership (part 1)

mobile_config/Cargo.toml

@@ -60,6 +60,7 @@ task-manager = { path = "../task_manager" }
 tls-init = { path = "../tls_init" }
 
 [dev-dependencies]
+derive_builder = "0.20"
 rand = { workspace = true }
 tokio-stream = { workspace = true, features = ["net"] }
 tempfile = "3"

mobile_config/migrations/20251209173943_drop_mobile_radio_tracker.sql

@@ -0,0 +1 @@
+DROP TABLE IF EXISTS mobile_radio_tracker;

mobile_config/migrations/20251215132218_add_owner_to_gateways.sql

@@ -0,0 +1,3 @@
+ALTER TABLE gateways
+ADD COLUMN IF NOT EXISTS owner VARCHAR(255),
+ADD COLUMN IF NOT EXISTS owner_changed_at TIMESTAMPTZ;

mobile_config/src/cli/mod.rs

@@ -1,5 +1,5 @@
 use crate::{
-    cli::{api::Api, import::Import, server::Server},
+    cli::{api::Api, server::Server},
     settings::Settings,
 };
 use base64::{engine::general_purpose, Engine};
@@ -9,7 +9,6 @@ use std::io::Write;
 use std::{fs::File, path::PathBuf};
 
 pub mod api;
-pub mod import;
 pub mod server;
 
 #[derive(Debug, clap::Parser)]
@@ -55,10 +54,6 @@ impl Cli {
 
                 Ok(())
             }
-            Cmd::Import(import) => {
-                let settings = Settings::new(self.config)?;
-                import.run(&settings).await
-            }
             Cmd::Server(server) => {
                 let settings = Settings::new(self.config)?;
                 server.run(&settings).await
@@ -71,6 +66,5 @@ impl Cli {
 pub enum Cmd {
     Api(Api),
     GenerateKey,
-    Import(Import),
     Server(Server),
 }

mobile_config/src/gateway/db.rs

@@ -95,6 +95,8 @@ pub struct Gateway {
     // When location last changed, set to refreshed_at (updated via SQL query see Gateway::insert)
     pub location_changed_at: Option<DateTime<Utc>>,
     pub location_asserts: Option<u32>,
+    pub owner: Option<String>,
+    pub owner_changed_at: Option<DateTime<Utc>>,
 }
 
 #[derive(Debug)]
@@ -122,7 +124,9 @@ impl Gateway {
                 azimuth,
                 location,
                 location_changed_at,
-                location_asserts
+                location_asserts,
+                owner,
+                owner_changed_at
             ) ",
         );
 
@@ -138,7 +142,9 @@ impl Gateway {
                 .push_bind(g.azimuth.map(|v| v as i64))
                 .push_bind(g.location.map(|v| v as i64))
                 .push_bind(g.location_changed_at)
-                .push_bind(g.location_asserts.map(|v| v as i64));
+                .push_bind(g.location_asserts.map(|v| v as i64))
+                .push_bind(g.owner.as_deref())
+                .push_bind(g.owner_changed_at);
         });
 
         let res = qb.build().execute(pool).await?;
@@ -160,11 +166,13 @@ impl Gateway {
                 azimuth,
                 location,
                 location_changed_at,
-                location_asserts
+                location_asserts,
+                owner,
+                owner_changed_at
             )
             VALUES (
                 $1, $2, $3, $4, $5, $6, $7,
-                $8, $9, $10, $11, $12
+                $8, $9, $10, $11, $12, $13, $14
             )
             "#,
         )
@@ -180,6 +188,8 @@ impl Gateway {
         .bind(self.location.map(|v| v as i64))
         .bind(self.location_changed_at)
         .bind(self.location_asserts.map(|v| v as i64))
+        .bind(self.owner.as_deref())
+        .bind(self.owner_changed_at)
         .execute(pool)
         .await?;
 
@@ -205,7 +215,9 @@ impl Gateway {
                 azimuth,
                 location,
                 location_changed_at,
-                location_asserts
+                location_asserts,
+                owner,
+                owner_changed_at
             FROM gateways
             WHERE address = $1
             ORDER BY inserted_at DESC
@@ -240,7 +252,9 @@ impl Gateway {
                 azimuth,
                 location,
                 location_changed_at,
-                location_asserts
+                location_asserts,
+                owner,
+                owner_changed_at
             FROM gateways
             WHERE address = ANY($1)
             ORDER BY address, inserted_at DESC
@@ -273,7 +287,9 @@ impl Gateway {
                 azimuth,
                 location,
                 location_changed_at,
-                location_asserts
+                location_asserts,
+                owner,
+                owner_changed_at
             FROM gateways
             WHERE address = $1
             AND inserted_at <= $2
@@ -311,7 +327,9 @@ impl Gateway {
                 azimuth,
                 location,
                 location_changed_at,
-                location_asserts
+                location_asserts,
+                owner,
+                owner_changed_at
             FROM gateways
             WHERE address = ANY($1)
                 AND last_changed_at >= $2
@@ -346,7 +364,9 @@ impl Gateway {
                     azimuth,
                     location,
                     location_changed_at,
-                    location_asserts
+                    location_asserts,
+                    owner,
+                    owner_changed_at
                 FROM gateways
                 WHERE gateway_type = ANY($1)
                 AND last_changed_at >= $2
@@ -381,7 +401,7 @@ impl Gateway {
                 r#"
                     UPDATE gateways AS g
                     SET location_changed_at = v.location_changed_at
-                    FROM ( 
+                    FROM (
                 "#,
             );
 
@@ -406,6 +426,86 @@ impl Gateway {
 
         Ok(total)
     }
+
+    /// Get gateways where owner is NULL (for post-migration backfill)
+    /// Only returns addresses where the MOST RECENT record has a NULL owner
+    pub async fn get_null_owners(pool: &PgPool) -> anyhow::Result<Vec<Self>> {
+        let gateways = sqlx::query_as::<_, Self>(
+            r#"
+            WITH latest_gateways AS (
+                SELECT DISTINCT ON (address)
+                    address,
+                    gateway_type,
+                    created_at,
+                    inserted_at,
+                    refreshed_at,
+                    last_changed_at,
+                    hash,
+                    antenna,
+                    elevation,
+                    azimuth,
+                    location,
+                    location_changed_at,
+                    location_asserts,
+                    owner,
+                    owner_changed_at
+                FROM gateways
+                ORDER BY address, inserted_at DESC
+            )
+            SELECT * FROM latest_gateways
+            WHERE owner IS NULL
+            "#,
+        )
+        .fetch_all(pool)
+        .await?;
+
+        Ok(gateways)
+    }
+
+    /// Update owner and owner_changed_at for multiple gateways
+    pub async fn update_owners(pool: &PgPool, gateways: &[Gateway]) -> anyhow::Result<u64> {
+        if gateways.is_empty() {
+            return Ok(0);
+        }
+
+        const MAX_ROWS: usize = 20000;
+        let mut total = 0;
+
+        for chunk in gateways.chunks(MAX_ROWS) {
+            let mut qb = QueryBuilder::<Postgres>::new(
+                r#"
+                    UPDATE gateways AS g
+                    SET
+                        owner = v.owner,
+                        owner_changed_at = v.owner_changed_at
+                    FROM (
+                "#,
+            );
+
+            qb.push_values(chunk, |mut b, gw| {
+                b.push_bind(gw.address.as_ref())
+                    .push_bind(gw.owner.as_deref())
+                    .push_bind(gw.owner_changed_at);
+            });
+
+            qb.push(
+                r#"
+                    ) AS v(address, owner, owner_changed_at)
+                    WHERE g.address = v.address
+                    AND g.inserted_at = (
+                        SELECT MAX(inserted_at)
+                        FROM gateways
+                        WHERE address = v.address
+                    )
+                "#,
+            );
+
+            let res = qb.build().execute(pool).await?;
+            total += res.rows_affected();
+        }
+
+        Ok(total)
+    }
 }
 
 impl FromRow<'_, PgRow> for Gateway {
@@ -428,6 +528,8 @@ impl FromRow<'_, PgRow> for Gateway {
             location: to_u64(row.try_get("location")?),
             location_changed_at: row.try_get("location_changed_at")?,
             location_asserts: to_u32(row.try_get("location_asserts")?),
+            owner: row.try_get("owner")?,
+            owner_changed_at: row.try_get("owner_changed_at")?,
         })
     }
 }