Add RPC logging

bubelov · bubelov · commit fab6d2acf615 · 2025-12-22T13:12:59.000+07:00
diff --git a/migrations/89.sql b/migrations/89.sql
@@ -0,0 +1,10 @@
+CREATE TABLE IF NOT EXISTS rpc_call (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    user_id INTEGER REFERENCES user(id),
+    ip TEXT NOT NULL,
+    method TEXT NOT NULL,
+    params_json TEXT,
+    created_at TEXT NOT NULL,
+    processed_at TEXT NOT NULL,
+    duration_ns INTEGER NOT NULL
+) STRICT;
diff --git a/src/db/mod.rs b/src/db/mod.rs
@@ -13,6 +13,7 @@ pub mod migration;
 pub mod osm_user;
 pub mod place_submission;
 pub mod report;
+pub mod rpc_call;
 pub mod user;
 use crate::{service::filesystem::data_dir_file_path, Result};
 use deadpool_sqlite::{Config, Hook, Pool, Runtime};
diff --git a/src/db/rpc_call/blocking_queries.rs b/src/db/rpc_call/blocking_queries.rs
@@ -0,0 +1,111 @@
+use crate::db::rpc_call::schema::{self, Columns, RpcCall};
+use crate::Result;
+use geojson::JsonObject;
+use rusqlite::{named_params, Connection};
+use time::format_description::well_known::Rfc3339;
+use time::OffsetDateTime;
+
+pub fn insert(
+    user_id: i64,
+    ip: String,
+    method: String,
+    params: Option<JsonObject>,
+    created_at: OffsetDateTime,
+    processed_at: OffsetDateTime,
+    conn: &Connection,
+) -> Result<RpcCall> {
+    let sql = format!(
+        r#"
+            INSERT INTO {table} (
+                {user_id},
+                {ip},
+                {method},
+                {params_json},
+                {created_at},
+                {processed_at},
+                {duration_ns}
+            ) VALUES (
+                :user_id,
+                :ip,
+                :method,
+                :params_json,
+                :created_at,
+                :processed_at,
+                :duration_ns
+            )
+            RETURNING {projection}
+        "#,
+        table = schema::TABLE_NAME,
+        user_id = Columns::UserId.as_str(),
+        ip = Columns::Ip.as_str(),
+        method = Columns::Method.as_str(),
+        params_json = Columns::ParamsJson.as_str(),
+        created_at = Columns::CreatedAt.as_str(),
+        processed_at = Columns::ProcessedAt.as_str(),
+        duration_ns = Columns::DurationNs.as_str(),
+        projection = RpcCall::projection(),
+    );
+    let params = named_params! {
+        ":user_id" : user_id,
+        ":ip" : ip,
+        ":method" : method,
+        ":params_json" : params.map(|it| serde_json::to_string(&it).unwrap()),
+        ":created_at" : created_at.format(&Rfc3339)?,
+        ":processed_at" : processed_at.format(&Rfc3339)?,
+        ":duration_ns" : (processed_at - created_at).whole_nanoseconds() as i64,
+    };
+    conn.query_row(&sql, params, RpcCall::mapper())
+        .map_err(Into::into)
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::db::test::conn;
+    use serde_json::json;
+    use std::time::Duration;
+    use time::OffsetDateTime;
+
+    #[test]
+    fn insert() {
+        let conn = conn();
+
+        let user_id = 123;
+        let ip = "192.168.1.100".to_string();
+        let method = "get_element".to_string();
+        let params = Some(
+            json!({"id": 456, "name": "test"})
+                .as_object()
+                .unwrap()
+                .clone(),
+        );
+
+        let created_at = OffsetDateTime::now_utc();
+        let processed_at = created_at + Duration::from_millis(150);
+
+        let result = super::insert(
+            user_id,
+            ip.clone(),
+            method.clone(),
+            params.clone(),
+            created_at,
+            processed_at,
+            &conn,
+        );
+
+        assert!(result.is_ok(), "insert should succeed");
+
+        let rpc_call = result.unwrap();
+
+        assert_eq!(rpc_call.user_id, Some(user_id));
+        assert_eq!(rpc_call.ip, ip);
+        assert_eq!(rpc_call.method, method);
+        assert!(rpc_call.params_json.is_some());
+        assert_eq!(rpc_call.params_json, params);
+
+        assert_eq!(rpc_call.created_at, created_at);
+        assert_eq!(rpc_call.processed_at, processed_at);
+
+        let expected_ns = Duration::from_millis(150).as_nanos() as i64;
+        assert_eq!(rpc_call.duration_ns, expected_ns);
+    }
+}
diff --git a/src/db/rpc_call/mod.rs b/src/db/rpc_call/mod.rs
@@ -0,0 +1,3 @@
+pub(super) mod blocking_queries;
+pub mod queries;
+pub mod schema;
diff --git a/src/db/rpc_call/queries.rs b/src/db/rpc_call/queries.rs
@@ -0,0 +1,22 @@
+use super::blocking_queries;
+use crate::{db::rpc_call::schema::RpcCall, Result};
+use deadpool_sqlite::Pool;
+use geojson::JsonObject;
+use time::OffsetDateTime;
+
+pub async fn insert(
+    user_id: i64,
+    ip: String,
+    method: String,
+    params: Option<JsonObject>,
+    created_at: OffsetDateTime,
+    processed_at: OffsetDateTime,
+    pool: &Pool,
+) -> Result<RpcCall> {
+    pool.get()
+        .await?
+        .interact(move |conn| {
+            blocking_queries::insert(user_id, ip, method, params, created_at, processed_at, conn)
+        })
+        .await?
+}
diff --git a/src/db/rpc_call/schema.rs b/src/db/rpc_call/schema.rs
@@ -0,0 +1,96 @@
+use geojson::JsonObject;
+use rusqlite::Row;
+use serde::{Deserialize, Serialize};
+use std::sync::OnceLock;
+use time::OffsetDateTime;
+
+pub const TABLE_NAME: &str = "rpc_call";
+
+pub enum Columns {
+    Id,
+    UserId,
+    Ip,
+    Method,
+    ParamsJson,
+    CreatedAt,
+    ProcessedAt,
+    DurationNs,
+}
+
+impl Columns {
+    pub fn as_str(&self) -> &'static str {
+        match self {
+            Columns::Id => "id",
+            Columns::UserId => "user_id",
+            Columns::Ip => "ip",
+            Columns::Method => "method",
+            Columns::ParamsJson => "params_json",
+            Columns::CreatedAt => "created_at",
+            Columns::ProcessedAt => "processed_at",
+            Columns::DurationNs => "duration_ns",
+        }
+    }
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct RpcCall {
+    pub id: i64,
+    pub user_id: Option<i64>,
+    pub ip: String,
+    pub method: String,
+    pub params_json: Option<JsonObject>,
+    pub created_at: OffsetDateTime,
+    pub processed_at: OffsetDateTime,
+    pub duration_ns: i64,
+}
+
+impl RpcCall {
+    pub fn projection() -> &'static str {
+        static PROJECTION: OnceLock<String> = OnceLock::new();
+        PROJECTION.get_or_init(|| {
+            [
+                Columns::Id,
+                Columns::UserId,
+                Columns::Ip,
+                Columns::Method,
+                Columns::ParamsJson,
+                Columns::CreatedAt,
+                Columns::ProcessedAt,
+                Columns::DurationNs,
+            ]
+            .iter()
+            .map(Columns::as_str)
+            .collect::<Vec<_>>()
+            .join(", ")
+        })
+    }
+
+    pub const fn mapper() -> fn(&Row) -> rusqlite::Result<RpcCall> {
+        |row: &_| {
+            let params_json: Option<String> = row.get(Columns::ParamsJson.as_str())?;
+            let params_json = match params_json {
+                Some(json_str) => match serde_json::from_str(&json_str) {
+                    Ok(json_obj) => Some(json_obj),
+                    Err(e) => {
+                        return Err(rusqlite::Error::FromSqlConversionFailure(
+                            4,
+                            rusqlite::types::Type::Text,
+                            Box::new(e),
+                        ));
+                    }
+                },
+                None => None,
+            };
+            Ok(RpcCall {
+                id: row.get(Columns::Id.as_str())?,
+                user_id: row.get(Columns::UserId.as_str())?,
+                ip: row.get(Columns::Ip.as_str())?,
+                method: row.get(Columns::Method.as_str())?,
+                params_json,
+                created_at: row.get(Columns::CreatedAt.as_str())?,
+                processed_at: row.get(Columns::ProcessedAt.as_str())?,
+                duration_ns: row.get(Columns::DurationNs.as_str())?,
+            })
+        }
+    }
+}
diff --git a/src/rpc/generate_element_icons.rs b/src/rpc/generate_element_icons.rs
@@ -1,6 +1,6 @@
 use crate::{
-    db::{self, conf::schema::Conf, user::schema::User},
-    service::{discord, overpass::OverpassElement},
+    db::{self},
+    service::overpass::OverpassElement,
     Result,
 };
 use deadpool_sqlite::Pool;
@@ -27,25 +27,13 @@ pub struct UpdatedElement {
     pub new_icon: String,
 }
 
-pub async fn run(params: Params, requesting_user: &User, pool: &Pool, conf: &Conf) -> Result<Res> {
+pub async fn run(params: Params, pool: &Pool) -> Result<Res> {
     let started_at = OffsetDateTime::now_utc();
     let updated_elements =
         generate_element_icons(params.from_element_id, params.to_element_id, pool).await?;
-    let time_s = (OffsetDateTime::now_utc() - started_at).as_seconds_f64();
-    discord::send(
-        format!(
-            "{} generated element icons (id range {}..{}, elements affected: {})",
-            requesting_user.name,
-            params.from_element_id,
-            params.to_element_id,
-            updated_elements.len(),
-        ),
-        discord::Channel::Api,
-        conf,
-    );
     Ok(Res {
         updated_elements,
-        time_s,
+        time_s: (OffsetDateTime::now_utc() - started_at).as_seconds_f64(),
     })
 }
 
diff --git a/src/rpc/handler.rs b/src/rpc/handler.rs
@@ -22,6 +22,7 @@ use serde::{de::DeserializeOwned, Deserialize, Serialize};
 use serde_json::{json, Map, Value};
 use std::collections::HashSet;
 use strum::VariantArray;
+use time::OffsetDateTime;
 
 #[derive(Deserialize)]
 pub struct RpcRequest {
@@ -275,6 +276,10 @@ pub async fn handle(
     pool: Data<Pool>,
     conf: Data<Conf>,
 ) -> Result<Json<RpcResponse>> {
+    let conn_info = req.connection_info();
+    let req_ip = conn_info.realip_remote_addr().unwrap_or("0.0.0.0");
+
+    let started_at = OffsetDateTime::now_utc();
     let matrix_client = service::matrix::client(&pool).await;
     let headers = req.headers();
     let Ok(req) = serde_json::from_str::<Map<String, Value>>(&req_body) else {
@@ -342,11 +347,14 @@ pub async fn handle(
         }
         Some(user)
     };
+    let user_id = user.as_ref().map(|it| it.id);
 
     if req.jsonrpc != "2.0" {
         return Ok(Json(RpcResponse::invalid_request(Value::Null)));
     }
 
+    let req_params = req.params.clone();
+
     let res: RpcResponse = match req.method {
         RpcMethod::Whoami => RpcResponse::from(
             req.id.clone(),
@@ -400,8 +408,7 @@ pub async fn handle(
         ),
         RpcMethod::GenerateElementIcons => RpcResponse::from(
             req.id.clone(),
-            super::generate_element_icons::run(params(req.params)?, &user.unwrap(), &pool, &conf)
-                .await?,
+            super::generate_element_icons::run(params(req.params)?, &pool).await?,
         ),
         RpcMethod::GenerateElementCategories => RpcResponse::from(
             req.id.clone(),
@@ -570,6 +577,20 @@ pub async fn handle(
             super::matrix::send_matrix_message::run(params(req.params)?, &matrix_client).await,
         ),
     }?;
+
+    if let Some(user_id) = user_id {
+        db::rpc_call::queries::insert(
+            user_id,
+            req_ip.to_string(),
+            method.as_str().unwrap_or_default().to_string(),
+            req_params.map(|it| it.as_object().map(|it| it.clone()).unwrap()),
+            started_at,
+            OffsetDateTime::now_utc(),
+            &pool,
+        )
+        .await?;
+    }
+
     Ok(Json(res))
 }
 

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,3 @@`
	`1`	`+pub(super) mod blocking_queries;`
	`2`	`+pub mod queries;`
	`3`	`+pub mod schema;`