From 7a7fdecf10454229db0795f09810a5f9970aa1f5 Mon Sep 17 00:00:00 2001 From: hitalin Date: Wed, 22 Jul 2026 14:25:43 +0900 Subject: [PATCH 1/2] =?UTF-8?q?feat:=20server=5Finfo=20SWR=20=E3=82=AD?= =?UTF-8?q?=E3=83=A3=E3=83=83=E3=82=B7=E3=83=A5=E3=82=92=E5=B0=8E=E5=85=A5?= =?UTF-8?q?=E3=81=97=20servers=20=E3=82=92=20server=5Fdetections=20?= =?UTF-8?q?=E3=81=AB=E7=BD=AE=E6=8F=9B=20(notedeck#782)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit フロント (Pinia store) が持っていた「メモリ → DB → ネットワーク」の stale-while-revalidate・TTL 判定・in-flight dedup を ServerInfoService に集約。 保存するのは生の検出結果 (nodeinfo software + /api/meta 生 JSON) で、 フォーク解決と feature 判定はアプリ側が読取時に行う — 判定ロジックの更新が 古いキャッシュに埋まる鮮度問題 (旧 servers テーブル) を構造的に解消する。 - V5 migration: server_detections 追加、旧 servers は削除 (24h キャッシュの ため旧データ損失は軽微) - StoredServer / load_servers / get_server / upsert_server は削除 (β方針) - plan_for (TTL 判定) は純関数、SWR 経路は wiremock + temp DB でテスト Co-Authored-By: Claude Fable 5 --- migrations/V5__server_detections.sql | 18 ++ src/api.rs | 35 ++- src/commands/auth.rs | 16 +- src/commands/doctor.rs | 51 ++++- src/commands/mod.rs | 14 +- src/commands/notes.rs | 38 +--- src/db.rs | 108 +++++---- src/http_server.rs | 67 +++--- src/lib.rs | 6 +- src/main.rs | 5 +- src/models.rs | 36 ++- src/server_info.rs | 316 +++++++++++++++++++++++++++ src/streaming.rs | 96 +++++--- 13 files changed, 627 insertions(+), 179 deletions(-) create mode 100644 migrations/V5__server_detections.sql create mode 100644 src/server_info.rs diff --git a/migrations/V5__server_detections.sql b/migrations/V5__server_detections.sql new file mode 100644 index 0000000..8f40b01 --- /dev/null +++ b/migrations/V5__server_detections.sql @@ -0,0 +1,18 @@ +-- V5: servers (アプリ側で計算した software/features のキャッシュ) を +-- server_detections (生の検出結果キャッシュ) に置き換える。 +-- +-- 旧 servers は「フロントで解決した software 名 + features 判定結果」を保存して +-- いたため、判定ロジックの更新が既存キャッシュに反映されない鮮度問題があった +-- (notedeck#782)。生の nodeinfo / meta を保存し、解決はアプリ側で読取時に行う。 +-- 旧データは捨てる (次回アクセス時に再検出される 24h キャッシュのため損失は軽微)。 + +CREATE TABLE IF NOT EXISTS server_detections ( + host TEXT PRIMARY KEY, + software_name TEXT NOT NULL, + software_version TEXT NOT NULL, + software_repository TEXT, + meta_json TEXT NOT NULL, + updated_at INTEGER NOT NULL +); + +DROP TABLE IF EXISTS servers; diff --git a/src/api.rs b/src/api.rs index dbd0c9e..55af8e2 100644 --- a/src/api.rs +++ b/src/api.rs @@ -8,11 +8,11 @@ use serde_json::{json, Value}; use crate::error::NoteDeckError; use crate::models::{ - Antenna, AuthResult, Channel, ChatMessage, ChatUser, Clip, CreateNoteParams, + Antenna, AuthResult, Channel, ChatMessage, ChatUser, Clip, CreateNoteParams, MutedWordsResult, NormalizedDriveFile, NormalizedNote, NormalizedNoteReaction, NormalizedNotification, - MutedWordsResult, NormalizedUser, NormalizedUserDetail, RawCreateNoteResponse, RawDriveFile, - RawEmojisResponse, RawMiAuthResponse, RawNote, RawNoteReaction, RawNotification, RawUser, - RawUserDetail, SearchOptions, ServerEmoji, TimelineOptions, TimelineType, UserList, + NormalizedUser, NormalizedUserDetail, RawCreateNoteResponse, RawDriveFile, RawEmojisResponse, + RawMiAuthResponse, RawNote, RawNoteReaction, RawNotification, RawUser, RawUserDetail, + SearchOptions, ServerEmoji, TimelineOptions, TimelineType, UserList, }; /// Maximum response body size (50 MB) to prevent memory exhaustion from malicious servers. @@ -251,7 +251,12 @@ impl MisskeyClient { antenna_id: &str, ) -> Result { let data = self - .request(host, token, "antennas/show", json!({ "antennaId": antenna_id })) + .request( + host, + token, + "antennas/show", + json!({ "antennaId": antenna_id }), + ) .await?; let antenna: Antenna = serde_json::from_value(data)?; Ok(antenna) @@ -2901,7 +2906,10 @@ mod tests { let role = notifs[0].role.as_ref().expect("role present"); assert_eq!(role.name, "Active"); assert_eq!(role.color.as_deref(), Some("#ff0000")); - assert_eq!(role.icon_url.as_deref(), Some("https://example.com/role.png")); + assert_eq!( + role.icon_url.as_deref(), + Some("https://example.com/role.png") + ); } #[tokio::test] @@ -3596,7 +3604,9 @@ mod tests { // withReplies のみ指定 → notify は body に含まれないこと Mock::given(method("POST")) .and(path("/api/following/update")) - .and(body_partial_json(json!({ "userId": "u1", "withReplies": false }))) + .and(body_partial_json( + json!({ "userId": "u1", "withReplies": false }), + )) .respond_with(ResponseTemplate::new(200).set_body_json(json!({}))) .mount(&server) .await; @@ -3613,7 +3623,9 @@ mod tests { let server = MockServer::start().await; Mock::given(method("POST")) .and(path("/api/users/update-memo")) - .and(body_partial_json(json!({ "userId": "u1", "memo": "friend" }))) + .and(body_partial_json( + json!({ "userId": "u1", "memo": "friend" }), + )) .respond_with(ResponseTemplate::new(200).set_body_json(json!({}))) .mount(&server) .await; @@ -3700,10 +3712,11 @@ mod tests { let server = MockServer::start().await; Mock::given(method("POST")) .and(path("/api/notes/search")) - .and(body_partial_json(json!({ "query": "rust", "userId": "u1" }))) + .and(body_partial_json( + json!({ "query": "rust", "userId": "u1" }), + )) .respond_with( - ResponseTemplate::new(200) - .set_body_json(json!([raw_note_json("n1", "rust note")])), + ResponseTemplate::new(200).set_body_json(json!([raw_note_json("n1", "rust note")])), ) .mount(&server) .await; diff --git a/src/commands/auth.rs b/src/commands/auth.rs index 597061b..d5de2fc 100644 --- a/src/commands/auth.rs +++ b/src/commands/auth.rs @@ -52,11 +52,7 @@ pub fn run_accounts(db: &Database, fmt: OutputFormat) -> Result<(), NoteDeckErro Ok(()) } -pub async fn run_login( - db: &Database, - host: &str, - fmt: OutputFormat, -) -> Result<(), NoteDeckError> { +pub async fn run_login(db: &Database, host: &str, fmt: OutputFormat) -> Result<(), NoteDeckError> { let client = MisskeyClient::new()?; let session_id = uuid::Uuid::new_v4().to_string(); @@ -89,9 +85,8 @@ pub async fn run_login( ]; let permission_str = permissions.join(","); let scheme = crate::insecure::http_scheme(host); - let auth_url = format!( - "{scheme}://{host}/miauth/{session_id}?name=notecli&permission={permission_str}" - ); + let auth_url = + format!("{scheme}://{host}/miauth/{session_id}?name=notecli&permission={permission_str}"); match fmt { OutputFormat::Json | OutputFormat::Jsonl => { @@ -108,7 +103,10 @@ pub async fn run_login( println!(); println!(" {}", theme::link(&auth_url)); println!(); - println!("{}", theme::muted("認証が完了したらEnterを押してください...")); + println!( + "{}", + theme::muted("認証が完了したらEnterを押してください...") + ); } } diff --git a/src/commands/doctor.rs b/src/commands/doctor.rs index a241f9f..93f241b 100644 --- a/src/commands/doctor.rs +++ b/src/commands/doctor.rs @@ -35,10 +35,22 @@ pub struct Check { impl Check { fn env(name: &str, status: Status, message: String) -> Self { - Self { name: name.into(), status, message, account: None, fix: None } + Self { + name: name.into(), + status, + message, + account: None, + fix: None, + } } fn acc(account: &str, name: &str, status: Status, message: String) -> Self { - Self { name: name.into(), status, message, account: Some(account.into()), fix: None } + Self { + name: name.into(), + status, + message, + account: Some(account.into()), + fix: None, + } } fn with_fix(mut self, fix: impl Into) -> Self { self.fix = Some(fix.into()); @@ -78,7 +90,10 @@ pub async fn diagnose( } let targets: Vec<&Account> = match account_spec { - Some(spec) => accounts.iter().filter(|a| account_matches(a, spec)).collect(), + Some(spec) => accounts + .iter() + .filter(|a| account_matches(a, spec)) + .collect(), None => accounts.iter().collect(), }; if let Some(spec) = account_spec { @@ -119,7 +134,11 @@ pub async fn run_doctor( fn check_database(db: &Database, path: &Path) -> Check { match db.load_accounts() { - Ok(_) => Check::env("database", Status::Ok, format!("{} (readable)", path.display())), + Ok(_) => Check::env( + "database", + Status::Ok, + format!("{} (readable)", path.display()), + ), Err(e) => Check::env( "database", Status::Fail, @@ -157,7 +176,12 @@ async fn check_account(client: &MisskeyClient, a: &Account, checks: &mut Vec 0 { println!("{}", theme::error(&format!("{fails} check(s) failed"))); } else if warns > 0 { - println!("{}", theme::badge(&format!("all checks passed ({warns} warning(s))"))); + println!( + "{}", + theme::badge(&format!("all checks passed ({warns} warning(s))")) + ); } else { println!("{}", theme::success("all checks passed")); } diff --git a/src/commands/mod.rs b/src/commands/mod.rs index da1af40..2acd9b1 100644 --- a/src/commands/mod.rs +++ b/src/commands/mod.rs @@ -63,7 +63,15 @@ pub async fn run_cli( reply_to, local_only, } => { - notes::run_post(&ctx, text, cw.as_deref(), visibility, reply_to.as_deref(), *local_only).await + notes::run_post( + &ctx, + text, + cw.as_deref(), + visibility, + reply_to.as_deref(), + *local_only, + ) + .await } Commands::Timeline { r#type, limit } => notes::run_timeline(&ctx, r#type, *limit).await, Commands::Search { query, limit } => notes::run_search(&ctx, query, *limit).await, @@ -71,9 +79,7 @@ pub async fn run_cli( Commands::Replies { id, limit } => notes::run_replies(&ctx, id, *limit).await, Commands::Thread { id, limit } => notes::run_thread(&ctx, id, *limit).await, Commands::Delete { id } => notes::run_delete(&ctx, id).await, - Commands::Update { id, text, cw } => { - notes::run_update(&ctx, id, text, cw.as_deref()).await - } + Commands::Update { id, text, cw } => notes::run_update(&ctx, id, text, cw.as_deref()).await, Commands::React { note_id, reaction } => notes::run_react(&ctx, note_id, reaction).await, Commands::Unreact { note_id } => notes::run_unreact(&ctx, note_id).await, Commands::Renote { note_id } => notes::run_renote(&ctx, note_id).await, diff --git a/src/commands/notes.rs b/src/commands/notes.rs index 6e93854..847df7f 100644 --- a/src/commands/notes.rs +++ b/src/commands/notes.rs @@ -43,7 +43,11 @@ pub async fn run_post( Ok(()) } -pub async fn run_timeline(ctx: &CmdContext, tl_type: &str, limit: i64) -> Result<(), NoteDeckError> { +pub async fn run_timeline( + ctx: &CmdContext, + tl_type: &str, + limit: i64, +) -> Result<(), NoteDeckError> { let notes = ctx .client .get_timeline( @@ -58,11 +62,7 @@ pub async fn run_timeline(ctx: &CmdContext, tl_type: &str, limit: i64) -> Result Ok(()) } -pub async fn run_search( - ctx: &CmdContext, - query: &str, - limit: i64, -) -> Result<(), NoteDeckError> { +pub async fn run_search(ctx: &CmdContext, query: &str, limit: i64) -> Result<(), NoteDeckError> { let notes = ctx .client .search_notes( @@ -93,11 +93,7 @@ pub async fn run_note(ctx: &CmdContext, id: &str) -> Result<(), NoteDeckError> { Ok(()) } -pub async fn run_replies( - ctx: &CmdContext, - id: &str, - limit: i64, -) -> Result<(), NoteDeckError> { +pub async fn run_replies(ctx: &CmdContext, id: &str, limit: i64) -> Result<(), NoteDeckError> { let notes = ctx .client .get_note_children(&ctx.host, &ctx.token, &ctx.account.id, id, limit as u32) @@ -106,11 +102,7 @@ pub async fn run_replies( Ok(()) } -pub async fn run_thread( - ctx: &CmdContext, - id: &str, - limit: i64, -) -> Result<(), NoteDeckError> { +pub async fn run_thread(ctx: &CmdContext, id: &str, limit: i64) -> Result<(), NoteDeckError> { let notes = ctx .client .get_note_conversation(&ctx.host, &ctx.token, &ctx.account.id, id, limit as u32) @@ -280,24 +272,14 @@ pub async fn run_unfavorite(ctx: &CmdContext, note_id: &str) -> Result<(), NoteD pub async fn run_favorites(ctx: &CmdContext, limit: i64) -> Result<(), NoteDeckError> { let notes = ctx .client - .get_favorites( - &ctx.host, - &ctx.token, - &ctx.account.id, - limit, - None, - None, - ) + .get_favorites(&ctx.host, &ctx.token, &ctx.account.id, limit, None, None) .await?; print_notes(¬es, ctx.fmt); Ok(()) } pub async fn run_emojis(ctx: &CmdContext) -> Result<(), NoteDeckError> { - let emojis = ctx - .client - .get_server_emojis(&ctx.host, &ctx.token) - .await?; + let emojis = ctx.client.get_server_emojis(&ctx.host, &ctx.token).await?; print_emojis(&emojis, ctx.fmt); Ok(()) } diff --git a/src/db.rs b/src/db.rs index f595a8f..1e81345 100644 --- a/src/db.rs +++ b/src/db.rs @@ -4,7 +4,7 @@ use std::sync::{Mutex, MutexGuard}; use crate::error::NoteDeckError; use crate::models::{ - Account, ChatMessage, ChatMessageReaction, ChatReactionUser, NormalizedNote, StoredServer, + Account, ChatMessage, ChatMessageReaction, ChatReactionUser, NormalizedNote, ServerDetection, }; mod embedded { @@ -391,37 +391,42 @@ impl Database { Ok(count) } - // --- Servers --- + // --- Server detections --- - pub fn load_servers(&self) -> Result, NoteDeckError> { + pub fn load_server_detections(&self) -> Result, NoteDeckError> { let conn = self.lock_read()?; let mut stmt = conn.prepare_cached( - "SELECT host, software, version, features_json, updated_at FROM servers", + "SELECT host, software_name, software_version, software_repository, meta_json, updated_at FROM server_detections", )?; let rows = stmt.query_map([], |row| { - Ok(StoredServer { + Ok(ServerDetection { host: row.get(0)?, - software: row.get(1)?, - version: row.get(2)?, - features_json: row.get(3)?, - updated_at: row.get(4)?, + software_name: row.get(1)?, + software_version: row.get(2)?, + software_repository: row.get(3)?, + meta_json: row.get(4)?, + updated_at: row.get(5)?, }) })?; Ok(rows.collect::>>()?) } - pub fn get_server(&self, host: &str) -> Result, NoteDeckError> { + pub fn get_server_detection( + &self, + host: &str, + ) -> Result, NoteDeckError> { let conn = self.lock_read()?; let mut stmt = conn.prepare_cached( - "SELECT host, software, version, features_json, updated_at FROM servers WHERE host = ?1", + "SELECT host, software_name, software_version, software_repository, meta_json, updated_at FROM server_detections WHERE host = ?1", )?; let mut rows = stmt.query_map(params![host], |row| { - Ok(StoredServer { + Ok(ServerDetection { host: row.get(0)?, - software: row.get(1)?, - version: row.get(2)?, - features_json: row.get(3)?, - updated_at: row.get(4)?, + software_name: row.get(1)?, + software_version: row.get(2)?, + software_repository: row.get(3)?, + meta_json: row.get(4)?, + updated_at: row.get(5)?, }) })?; match rows.next() { @@ -900,22 +905,25 @@ impl Database { Ok(()) } - pub fn upsert_server(&self, server: &StoredServer) -> Result<(), NoteDeckError> { + pub fn upsert_server_detection(&self, det: &ServerDetection) -> Result<(), NoteDeckError> { let conn = self.lock_write()?; conn.execute( - "INSERT INTO servers (host, software, version, features_json, updated_at) - VALUES (?1, ?2, ?3, ?4, ?5) + "INSERT INTO server_detections + (host, software_name, software_version, software_repository, meta_json, updated_at) + VALUES (?1, ?2, ?3, ?4, ?5, ?6) ON CONFLICT(host) DO UPDATE SET - software = excluded.software, - version = excluded.version, - features_json = excluded.features_json, + software_name = excluded.software_name, + software_version = excluded.software_version, + software_repository = excluded.software_repository, + meta_json = excluded.meta_json, updated_at = excluded.updated_at", params![ - server.host, - server.software, - server.version, - server.features_json, - server.updated_at, + det.host, + det.software_name, + det.software_version, + det.software_repository, + det.meta_json, + det.updated_at, ], )?; Ok(()) @@ -1289,7 +1297,7 @@ impl Database { #[cfg(test)] mod tests { use super::*; - use crate::models::{Account, NormalizedNote, NormalizedUser, StoredServer}; + use crate::models::{Account, NormalizedNote, NormalizedUser, ServerDetection}; use std::collections::HashMap; fn temp_db() -> (tempfile::TempDir, Database) { @@ -1316,7 +1324,9 @@ mod tests { .unwrap(); assert!(tables.contains(&"accounts".to_string())); - assert!(tables.contains(&"servers".to_string())); + assert!(tables.contains(&"server_detections".to_string())); + // V5 で旧 servers テーブルは削除済み + assert!(!tables.contains(&"servers".to_string())); assert!(tables.contains(&"notes_cache".to_string())); assert!(tables.contains(&"ogp_cache".to_string())); assert!(tables.contains(&"chat_messages_cache".to_string())); @@ -1465,35 +1475,45 @@ mod tests { // --- Server CRUD tests --- - fn sample_server() -> StoredServer { - StoredServer { + fn sample_detection() -> ServerDetection { + ServerDetection { host: "misskey.io".to_string(), - software: "misskey".to_string(), - version: "2025.3.0".to_string(), - features_json: "{}".to_string(), + software_name: "misskey".to_string(), + software_version: "2025.3.0".to_string(), + software_repository: Some("https://github.com/misskey-dev/misskey".to_string()), + meta_json: "{}".to_string(), updated_at: 1700000000, } } #[test] - fn server_upsert_and_load() { + fn server_detection_upsert_and_load() { let (_dir, db) = temp_db(); - db.upsert_server(&sample_server()).unwrap(); + db.upsert_server_detection(&sample_detection()).unwrap(); - let servers = db.load_servers().unwrap(); - assert_eq!(servers.len(), 1); - assert_eq!(servers[0].host, "misskey.io"); + let dets = db.load_server_detections().unwrap(); + assert_eq!(dets.len(), 1); + assert_eq!(dets[0].host, "misskey.io"); } #[test] - fn server_get_by_host() { + fn server_detection_get_by_host_and_update() { let (_dir, db) = temp_db(); - db.upsert_server(&sample_server()).unwrap(); + db.upsert_server_detection(&sample_detection()).unwrap(); + + let d = db.get_server_detection("misskey.io").unwrap().unwrap(); + assert_eq!(d.software_version, "2025.3.0"); - let s = db.get_server("misskey.io").unwrap().unwrap(); - assert_eq!(s.version, "2025.3.0"); + // upsert は同 host を上書きする + let mut newer = sample_detection(); + newer.software_version = "2025.4.0".to_string(); + newer.updated_at = 1700001000; + db.upsert_server_detection(&newer).unwrap(); + let d = db.get_server_detection("misskey.io").unwrap().unwrap(); + assert_eq!(d.software_version, "2025.4.0"); + assert_eq!(d.updated_at, 1700001000); - assert!(db.get_server("nonexistent").unwrap().is_none()); + assert!(db.get_server_detection("nonexistent").unwrap().is_none()); } // --- Notes cache tests --- diff --git a/src/http_server.rs b/src/http_server.rs index 8608c86..117a395 100644 --- a/src/http_server.rs +++ b/src/http_server.rs @@ -9,12 +9,12 @@ use axum::{ routing::get, Json, Router, }; -use subtle::ConstantTimeEq; use futures_util::stream::Stream; use serde::Deserialize; use serde_json::{json, Value}; use std::net::SocketAddr; use std::sync::Arc; +use subtle::ConstantTimeEq; use tokio_stream::wrappers::BroadcastStream; use tokio_stream::StreamExt; use tower_http::cors::CorsLayer; @@ -239,7 +239,11 @@ fn core_openapi_router() -> OpenApiRouter { .routes(routes!(get_note, delete_note)) .routes(routes!(get_note_children)) .routes(routes!(get_note_conversation)) - .routes(routes!(get_note_reactions, create_reaction, delete_reaction)) + .routes(routes!( + get_note_reactions, + create_reaction, + delete_reaction + )) .routes(routes!(get_user)) .routes(routes!(get_user_notes)) .routes(routes!(search_notes)) @@ -254,7 +258,10 @@ fn core_openapi_router() -> OpenApiRouter { /// own spec. pub fn build_core_routes(state: AppState) -> OpenApiRouter { core_openapi_router() - .layer(middleware::from_fn_with_state(state.clone(), auth_middleware)) + .layer(middleware::from_fn_with_state( + state.clone(), + auth_middleware, + )) .layer(CorsLayer::permissive()) .with_state(state) } @@ -293,8 +300,14 @@ pub fn endpoints_from_spec(openapi: &utoipa::openapi::OpenApi) -> Vec { } } out.sort_by(|a, b| { - let ka = (a["path"].as_str().unwrap_or(""), a["method"].as_str().unwrap_or("")); - let kb = (b["path"].as_str().unwrap_or(""), b["method"].as_str().unwrap_or("")); + let ka = ( + a["path"].as_str().unwrap_or(""), + a["method"].as_str().unwrap_or(""), + ); + let kb = ( + b["path"].as_str().unwrap_or(""), + b["method"].as_str().unwrap_or(""), + ); ka.cmp(&kb) }); out @@ -622,7 +635,14 @@ async fn get_note_reactions( let limit = opts.limit.unwrap_or(20); let reactions = state .client - .get_note_reactions(&h, &token, ¬e_id, opts.r#type.as_deref(), limit, opts.until_id.as_deref()) + .get_note_reactions( + &h, + &token, + ¬e_id, + opts.r#type.as_deref(), + limit, + opts.until_id.as_deref(), + ) .await?; Ok(Json(reactions)) } @@ -674,10 +694,7 @@ async fn delete_reaction( ) -> Result { let account_id = state.account_id_for_host(&host)?; let (h, token) = crate::get_credentials(&state.db, &account_id)?; - state - .client - .delete_reaction(&h, &token, ¬e_id) - .await?; + state.client.delete_reaction(&h, &token, ¬e_id).await?; Ok(StatusCode::NO_CONTENT) } @@ -788,22 +805,20 @@ async fn sse_events( .r#type .map(|t| t.split(',').map(|s| s.trim().to_string()).collect()); - let stream = BroadcastStream::new(rx).filter_map(move |result| { - match result { - Ok(sse_event) => { - if let Some(ref filter) = type_filter { - if !filter.iter().any(|f| sse_event.event_type.starts_with(f)) { - return None; - } + let stream = BroadcastStream::new(rx).filter_map(move |result| match result { + Ok(sse_event) => { + if let Some(ref filter) = type_filter { + if !filter.iter().any(|f| sse_event.event_type.starts_with(f)) { + return None; } - let event = Event::default() - .event(&sse_event.event_type) - .json_data(&sse_event.data) - .ok()?; - Some(Ok(event)) } - Err(_) => None, + let event = Event::default() + .event(&sse_event.event_type) + .json_data(&sse_event.data) + .ok()?; + Some(Ok(event)) } + Err(_) => None, }); Sse::new(stream).keep_alive(KeepAlive::default()) @@ -824,11 +839,7 @@ struct TimelineQueryParams { impl TimelineQueryParams { fn into_timeline_options(self) -> crate::models::TimelineOptions { - crate::models::TimelineOptions::new( - self.limit.unwrap_or(20), - self.since_id, - self.until_id, - ) + crate::models::TimelineOptions::new(self.limit.unwrap_or(20), self.since_id, self.until_id) } } diff --git a/src/lib.rs b/src/lib.rs index b2406aa..835bbb8 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -9,6 +9,7 @@ pub mod http_server; pub(crate) mod insecure; pub mod keychain; pub mod models; +pub mod server_info; pub mod streaming; use db::Database; @@ -38,10 +39,7 @@ pub fn get_credentials(db: &Database, account_id: &str) -> Result<(String, Strin // Try lazy migration to keychain; verify before clearing DB if keychain::is_persistent() && keychain::store_token(account_id, &db_token).is_ok() - && keychain::get_token(account_id) - .ok() - .flatten() - .is_some() + && keychain::get_token(account_id).ok().flatten().is_some() { let _ = db.clear_token(account_id); } diff --git a/src/main.rs b/src/main.rs index 1a533d4..2a045a1 100644 --- a/src/main.rs +++ b/src/main.rs @@ -58,9 +58,8 @@ async fn run_daemon(port: u16) { let db_path = data_dir.join("notecli.db"); let db = Arc::new(Database::open(&db_path).expect("Failed to open database")); - let client = Arc::new( - notecli::api::MisskeyClient::new().expect("Failed to create HTTP client"), - ); + let client = + Arc::new(notecli::api::MisskeyClient::new().expect("Failed to create HTTP client")); let event_bus = Arc::new(EventBus::new()); diff --git a/src/models.rs b/src/models.rs index ab8fa09..b9320c7 100644 --- a/src/models.rs +++ b/src/models.rs @@ -104,14 +104,24 @@ impl From for AccountPublic { } } +/// サーバー検出結果の生キャッシュ (notedeck#782)。 +/// +/// nodeinfo の software 情報と /api/meta の生 JSON をそのまま保存する。 +/// フォーク解決 (software 名 → ServerSoftware) と feature 判定はアプリ側が +/// 読取時に行う — 判定ロジックの更新が古いキャッシュに埋まらないようにする。 #[derive(Debug, Clone, Serialize, Deserialize)] #[cfg_attr(feature = "specta", derive(specta::Type))] #[serde(rename_all = "camelCase")] -pub struct StoredServer { +pub struct ServerDetection { pub host: String, - pub software: String, - pub version: String, - pub features_json: String, + /// nodeinfo `software.name` (例: "misskey") + pub software_name: String, + /// nodeinfo `software.version` + pub software_version: String, + /// nodeinfo 2.1 `software.repository` (例: "https://github.com/misskey-dev/misskey") + pub software_repository: Option, + /// /api/meta (detail: true) の生 JSON。取得失敗時は "{}" + pub meta_json: String, pub updated_at: i64, } @@ -2348,17 +2358,19 @@ mod tests { } #[test] - fn stored_server_serde_roundtrip() { - let server = StoredServer { + fn server_detection_serde_roundtrip() { + let det = ServerDetection { host: "misskey.io".into(), - software: "misskey".into(), - version: "2024.1.0".into(), - features_json: r#"{"miAuth":true}"#.into(), + software_name: "misskey".into(), + software_version: "2024.1.0".into(), + software_repository: Some("https://github.com/misskey-dev/misskey".into()), + meta_json: r#"{"iconUrl":"/icon.png"}"#.into(), updated_at: 1700000000, }; - let json = serde_json::to_string(&server).unwrap(); - let back: StoredServer = serde_json::from_str(&json).unwrap(); + let json = serde_json::to_string(&det).unwrap(); + assert!(json.contains("softwareName")); + let back: ServerDetection = serde_json::from_str(&json).unwrap(); assert_eq!(back.host, "misskey.io"); - assert_eq!(back.version, "2024.1.0"); + assert_eq!(back.software_version, "2024.1.0"); } } diff --git a/src/server_info.rs b/src/server_info.rs new file mode 100644 index 0000000..ee21dbc --- /dev/null +++ b/src/server_info.rs @@ -0,0 +1,316 @@ +//! サーバー検出結果の SWR キャッシュ (notedeck#782)。 +//! +//! フロント (Pinia store) が持っていた「メモリ → DB → ネットワーク」の +//! stale-while-revalidate・TTL 判定・in-flight dedup を Rust 側へ集約する。 +//! 保存するのは生の検出結果 ([`ServerDetection`]) で、フォーク解決や +//! feature 判定はアプリ側が読取時に行う。 +//! +//! - **fresh** (TTL 内): DB の行をそのまま返す。ネットワークなし +//! - **stale** (TTL 超過): DB の行を即返しつつバックグラウンドで再検出 +//! (オフラインでも stale を維持して壊れない) +//! - **miss**: per-host ロックで dedup してネットワーク検出 → DB 保存 + +use std::collections::{HashMap, HashSet}; +use std::sync::{Arc, Mutex}; +use std::time::{SystemTime, UNIX_EPOCH}; + +use crate::api::MisskeyClient; +use crate::db::Database; +use crate::error::NoteDeckError; +use crate::models::ServerDetection; + +/// 検出結果の TTL。超過した行は stale 扱いで返しつつ再検出する。 +pub const SERVER_DETECTION_TTL_MS: i64 = 24 * 60 * 60 * 1000; + +/// TTL 判定の純粋関数部。`get_or_fetch` の分岐はすべてここに集約する。 +#[derive(Debug, PartialEq, Eq)] +pub enum CachePlan { + /// TTL 内 — そのまま返す + Fresh, + /// TTL 超過 — stale を返しつつバックグラウンド再検出 + StaleRevalidate, + /// 行なし — ネットワーク検出が必要 + Miss, +} + +pub fn plan_for(row: Option<&ServerDetection>, now_ms: i64, ttl_ms: i64) -> CachePlan { + match row { + None => CachePlan::Miss, + Some(det) if now_ms - det.updated_at < ttl_ms => CachePlan::Fresh, + Some(_) => CachePlan::StaleRevalidate, + } +} + +fn now_ms() -> i64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_millis() as i64) + .unwrap_or(0) +} + +pub struct ServerInfoService { + db: Arc, + client: Arc, + /// miss 時の per-host dedup ロック。同一 host への同時要求を直列化し、 + /// 2 本目以降はロック取得後の DB 再読込で検出済みの行を拾う。 + inflight: tokio::sync::Mutex>>>, + /// stale 再検出の実行中 host 集合 (spawn の重複起動防止)。 + revalidating: Mutex>, +} + +impl ServerInfoService { + pub fn new(db: Arc, client: Arc) -> Arc { + Arc::new(Self { + db, + client, + inflight: tokio::sync::Mutex::new(HashMap::new()), + revalidating: Mutex::new(HashSet::new()), + }) + } + + /// SWR 取得。fresh は即返し、stale は返しつつ背景再検出、miss は検出して保存。 + pub async fn get_or_fetch( + self: &Arc, + host: &str, + ) -> Result { + let row = self.db.get_server_detection(host)?; + match plan_for(row.as_ref(), now_ms(), SERVER_DETECTION_TTL_MS) { + CachePlan::Fresh => Ok(row.expect("fresh implies row")), + CachePlan::StaleRevalidate => { + self.spawn_revalidate(host); + Ok(row.expect("stale implies row")) + } + CachePlan::Miss => { + let lock = self.host_lock(host).await; + let _guard = lock.lock().await; + // ロック待機中に先行リクエストが保存した行を拾う (dedup) + if let Some(det) = self.db.get_server_detection(host)? { + if plan_for(Some(&det), now_ms(), SERVER_DETECTION_TTL_MS) == CachePlan::Fresh { + return Ok(det); + } + } + self.detect_and_store(host).await + } + } + } + + /// 強制ネットワーク検出 + DB 保存。ログイン完了直後など、キャッシュを + /// 確実に上書きしたい場面で使う。 + pub async fn detect_and_store(&self, host: &str) -> Result { + let det = self.detect(host).await?; + self.db.upsert_server_detection(&det)?; + Ok(det) + } + + /// nodeinfo (必須) + /api/meta (失敗許容) を並列取得して生の検出結果を作る。 + async fn detect(&self, host: &str) -> Result { + let (nodeinfo, meta) = tokio::join!( + self.client.fetch_nodeinfo(host), + self.client.fetch_server_meta(host), + ); + let nodeinfo = nodeinfo?; + let software = &nodeinfo["software"]; + // meta はオフライン/非公開でも動くよう失敗を握りつぶす (アプリ側は + // favicon フォールバックで表示する) + let meta_json = meta + .map(|v| v.to_string()) + .unwrap_or_else(|_| "{}".to_string()); + Ok(ServerDetection { + host: host.to_string(), + software_name: software["name"].as_str().unwrap_or("").to_string(), + software_version: software["version"].as_str().unwrap_or("").to_string(), + software_repository: software["repository"].as_str().map(|s| s.to_string()), + meta_json, + updated_at: now_ms(), + }) + } + + fn spawn_revalidate(self: &Arc, host: &str) { + { + let mut set = self.revalidating.lock().unwrap_or_else(|e| e.into_inner()); + if !set.insert(host.to_string()) { + return; // 既に再検出中 + } + } + let this = Arc::clone(self); + let host = host.to_string(); + tokio::spawn(async move { + // オフライン等の失敗は無視して stale を維持する + let _ = this.detect_and_store(&host).await; + let mut set = this.revalidating.lock().unwrap_or_else(|e| e.into_inner()); + set.remove(&host); + }); + } + + async fn host_lock(&self, host: &str) -> Arc> { + let mut map = self.inflight.lock().await; + Arc::clone(map.entry(host.to_string()).or_default()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + use tempfile::TempDir; + use wiremock::matchers::{method, path}; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + fn sample(updated_at: i64) -> ServerDetection { + ServerDetection { + host: "misskey.io".into(), + software_name: "misskey".into(), + software_version: "2025.3.0".into(), + software_repository: None, + meta_json: "{}".into(), + updated_at, + } + } + + #[test] + fn plan_fresh_within_ttl() { + let det = sample(1_000); + assert_eq!(plan_for(Some(&det), 1_000 + 99, 100), CachePlan::Fresh); + assert_eq!( + plan_for(Some(&det), 1_000 + 100, 100), + CachePlan::StaleRevalidate + ); + assert_eq!(plan_for(None, 0, 100), CachePlan::Miss); + } + + fn temp_db() -> (TempDir, Arc) { + let dir = TempDir::new().unwrap(); + let db = Database::open(&dir.path().join("test.db")).unwrap(); + (dir, Arc::new(db)) + } + + /// NOTECLI_INSECURE_HOSTS はプロセス共有 env のため、これを触る + /// ネットワーク系テストはこの lock で直列化する。 + static ENV_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); + + fn register_insecure_host(host: &str) { + let merged = match std::env::var("NOTECLI_INSECURE_HOSTS") { + Ok(v) if !v.is_empty() => format!("{v},{host}"), + _ => host.to_string(), + }; + std::env::set_var("NOTECLI_INSECURE_HOSTS", merged); + } + + /// wiremock を Misskey サーバーに見立てる。fetch_nodeinfo は + /// `{scheme}://{host}/...` を直接叩くため、insecure host 登録で + /// http://127.0.0.1:PORT へ向ける。 + async fn mock_misskey() -> (MockServer, String) { + let server = MockServer::start().await; + let host = server.uri().trim_start_matches("http://").to_string(); + register_insecure_host(&host); + + Mock::given(method("GET")) + .and(path("/.well-known/nodeinfo")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "links": [{ + "rel": "http://nodeinfo.diaspora.software/ns/schema/2.1", + "href": format!("http://{host}/nodeinfo/2.1"), + }] + }))) + .mount(&server) + .await; + Mock::given(method("GET")) + .and(path("/nodeinfo/2.1")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "software": { + "name": "misskey", + "version": "2025.4.0", + "repository": "https://github.com/misskey-dev/misskey", + } + }))) + .mount(&server) + .await; + Mock::given(method("POST")) + .and(path("/api/meta")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "iconUrl": "/icon.png", + "themeColor": "#86b300", + }))) + .mount(&server) + .await; + (server, host) + } + + /// ネットワーク系テストは NOTECLI_INSECURE_HOSTS (プロセス共有 env) を + /// 使うため 1 つの async テストにまとめて直列化する。 + #[tokio::test] + async fn swr_miss_then_fresh_then_stale() { + let _env = ENV_LOCK.lock().await; + let (_server, host) = mock_misskey().await; + let (_dir, db) = temp_db(); + let client = Arc::new(MisskeyClient::new().unwrap()); + let svc = ServerInfoService::new(Arc::clone(&db), client); + + // miss: ネットワーク検出して保存 + let det = svc.get_or_fetch(&host).await.unwrap(); + assert_eq!(det.software_name, "misskey"); + assert_eq!(det.software_version, "2025.4.0"); + assert_eq!( + det.software_repository.as_deref(), + Some("https://github.com/misskey-dev/misskey") + ); + assert!(det.meta_json.contains("icon.png")); + assert!(db.get_server_detection(&host).unwrap().is_some()); + + // fresh: DB から返る (同時要求も 1 件に dedup される想定のロック経路) + let again = svc.get_or_fetch(&host).await.unwrap(); + assert_eq!(again.updated_at, det.updated_at); + + // stale: 古い updated_at に書き換えると stale 行が即返る + let mut old = det.clone(); + old.updated_at = 1; // 1970 年 = 確実に TTL 切れ + db.upsert_server_detection(&old).unwrap(); + let stale = svc.get_or_fetch(&host).await.unwrap(); + assert_eq!(stale.updated_at, 1); + // バックグラウンド再検出が走り、いずれ updated_at が更新される + for _ in 0..50 { + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + let cur = db.get_server_detection(&host).unwrap().unwrap(); + if cur.updated_at > 1 { + return; + } + } + panic!("background revalidation did not update the row"); + } + + #[tokio::test] + async fn meta_failure_is_tolerated() { + let _env = ENV_LOCK.lock().await; + let server = MockServer::start().await; + let host = server.uri().trim_start_matches("http://").to_string(); + register_insecure_host(&host); + + Mock::given(method("GET")) + .and(path("/.well-known/nodeinfo")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "links": [{ "rel": "nodeinfo/2.0", "href": format!("http://{host}/nodeinfo/2.0") }] + }))) + .mount(&server) + .await; + Mock::given(method("GET")) + .and(path("/nodeinfo/2.0")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "software": { "name": "misskey", "version": "2025.4.0" } + }))) + .mount(&server) + .await; + // /api/meta は 500 + Mock::given(method("POST")) + .and(path("/api/meta")) + .respond_with(ResponseTemplate::new(500)) + .mount(&server) + .await; + + let (_dir, db) = temp_db(); + let svc = ServerInfoService::new(db, Arc::new(MisskeyClient::new().unwrap())); + let det = svc.detect_and_store(&host).await.unwrap(); + assert_eq!(det.software_name, "misskey"); + assert_eq!(det.meta_json, "{}"); + assert_eq!(det.software_repository, None); + } +} diff --git a/src/streaming.rs b/src/streaming.rs index bfdf5a9..2bbed59 100644 --- a/src/streaming.rs +++ b/src/streaming.rs @@ -405,7 +405,8 @@ impl StreamingManager { } else { StreamConnectionState::Reconnecting }; - self.emitter.emit(StreamEvent::Status(Box::new(StreamStatusEvent { + self.emitter + .emit(StreamEvent::Status(Box::new(StreamStatusEvent { account_id: account_id.to_string(), state, }))); @@ -433,7 +434,10 @@ impl StreamingManager { None } Err(_) => { - tracing::warn!(account_id, "initial connect timed out; retrying in background"); + tracing::warn!( + account_id, + "initial connect timed out; retrying in background" + ); None } }; @@ -483,7 +487,8 @@ impl StreamingManager { }, ); - self.emitter.emit(StreamEvent::Status(Box::new(StreamStatusEvent { + self.emitter + .emit(StreamEvent::Status(Box::new(StreamStatusEvent { account_id: account_id.to_string(), state: if connected { StreamConnectionState::Connected @@ -526,7 +531,8 @@ impl StreamingManager { captured.remove(account_id); drop(captured); - self.emitter.emit(StreamEvent::Status(Box::new(StreamStatusEvent { + self.emitter + .emit(StreamEvent::Status(Box::new(StreamStatusEvent { account_id: account_id.to_string(), state: StreamConnectionState::Disconnected, }))); @@ -568,7 +574,8 @@ impl StreamingManager { let interval = Duration::from_millis(interval_ms.unwrap_or(15_000)); self.start_polling(account_id, host, token, interval).await; - self.emitter.emit(StreamEvent::Status(Box::new(StreamStatusEvent { + self.emitter + .emit(StreamEvent::Status(Box::new(StreamStatusEvent { account_id: account_id.to_string(), state: StreamConnectionState::Connected, }))); @@ -1085,9 +1092,9 @@ async fn connection_task( loop { connected_flag.store(false, Ordering::Relaxed); emitter.emit(StreamEvent::Status(Box::new(StreamStatusEvent { - account_id: account_id.clone(), - state: StreamConnectionState::Reconnecting, - }))); + account_id: account_id.clone(), + state: StreamConnectionState::Reconnecting, + }))); // Wait with backoff, but listen for Shutdown during the wait. // Equal Jitter (sleep in [backoff/2, backoff]) de-syncs reconnects @@ -1128,9 +1135,9 @@ async fn connection_task( connected_flag.store(true, Ordering::Relaxed); emitter.emit(StreamEvent::Status(Box::new(StreamStatusEvent { - account_id: account_id.clone(), - state: StreamConnectionState::Connected, - }))); + account_id: account_id.clone(), + state: StreamConnectionState::Connected, + }))); let reason = run_ws_session( &emitter, @@ -1418,7 +1425,11 @@ async fn handle_ws_message( note_id, update, }; - emit_both(emitter, event_bus, StreamEvent::NoteCaptureUpdated(Box::new(payload))); + emit_both( + emitter, + event_bus, + StreamEvent::NoteCaptureUpdated(Box::new(payload)), + ); } return; } @@ -1501,7 +1512,11 @@ async fn handle_ws_message( note_id, update, }; - emit_both(emitter, event_bus, StreamEvent::NoteUpdated(Box::new(payload))); + emit_both( + emitter, + event_bus, + StreamEvent::NoteUpdated(Box::new(payload)), + ); } else if kind == "main" { if event_type == "notification" { if let Ok(raw) = serde_json::from_value::(event_body) { @@ -1511,7 +1526,11 @@ async fn handle_ws_message( subscription_id: sub_id, notification, }; - emit_both(emitter, event_bus, StreamEvent::Notification(Box::new(payload))); + emit_both( + emitter, + event_bus, + StreamEvent::Notification(Box::new(payload)), + ); } } else if event_type == "mention" || event_type == "reply" { // main-event として emit しつつ、mention としても parse を試みる @@ -1538,7 +1557,11 @@ async fn handle_ws_message( event_type, body: event_body, }; - emit_both(emitter, event_bus, StreamEvent::MainEvent(Box::new(payload))); + emit_both( + emitter, + event_bus, + StreamEvent::MainEvent(Box::new(payload)), + ); } } else if kind == "chat" { if event_type == "message" { @@ -1575,7 +1598,11 @@ async fn handle_ws_message( subscription_id: sub_id, message: msg, }; - emit_both(emitter, event_bus, StreamEvent::ChatMessage(Box::new(payload))); + emit_both( + emitter, + event_bus, + StreamEvent::ChatMessage(Box::new(payload)), + ); } } else if event_type == "deleted" { if let Some(id) = event_body.as_str() { @@ -1596,7 +1623,11 @@ async fn handle_ws_message( subscription_id: sub_id, message_id: id_owned, }; - emit_both(emitter, event_bus, StreamEvent::ChatMessageDeleted(Box::new(payload))); + emit_both( + emitter, + event_bus, + StreamEvent::ChatMessageDeleted(Box::new(payload)), + ); } } else if event_type == "react" || event_type == "unreact" { let is_react = event_type == "react"; @@ -1626,7 +1657,11 @@ async fn handle_ws_message( reaction: body.reaction, user: body.user, }; - emit_both(emitter, event_bus, StreamEvent::ChatMessageReacted(Box::new(payload))); + emit_both( + emitter, + event_bus, + StreamEvent::ChatMessageReacted(Box::new(payload)), + ); } else { let payload = StreamChatMessageUnreactedEvent { account_id: account_id.to_string(), @@ -1635,7 +1670,11 @@ async fn handle_ws_message( reaction: body.reaction, user: body.user, }; - emit_both(emitter, event_bus, StreamEvent::ChatMessageUnreacted(Box::new(payload))); + emit_both( + emitter, + event_bus, + StreamEvent::ChatMessageUnreacted(Box::new(payload)), + ); } } } @@ -1732,7 +1771,11 @@ async fn polling_loop( subscription_id: sub_id.clone(), note, }; - emit_both(emitter.as_ref(), &event_bus, StreamEvent::Note(Box::new(payload))); + emit_both( + emitter.as_ref(), + &event_bus, + StreamEvent::Note(Box::new(payload)), + ); } consecutive_failures = 0; @@ -1841,9 +1884,9 @@ async fn polling_loop( }; emitter.emit(StreamEvent::Status(Box::new(StreamStatusEvent { - account_id: account_id.clone(), - state: StreamConnectionState::Reconnecting, - }))); + account_id: account_id.clone(), + state: StreamConnectionState::Reconnecting, + }))); Duration::from_secs(backoff) } else { @@ -1927,11 +1970,8 @@ mod tests { let dir = tempfile::tempdir().unwrap(); let db = Arc::new(crate::db::Database::open(&dir.path().join("test.db")).unwrap()); let (tx, mut rx) = mpsc::unbounded_channel(); - let manager = StreamingManager::new( - Arc::new(ChannelEmitter(tx)), - Arc::new(EventBus::new()), - db, - ); + let manager = + StreamingManager::new(Arc::new(ChannelEmitter(tx)), Arc::new(EventBus::new()), db); // 127.0.0.1:1 は即 connection refused になる manager From 82cdb75a58b1a8fb84b4ebfc70f647b675942cf3 Mon Sep 17 00:00:00 2001 From: hitalin Date: Wed, 22 Jul 2026 14:26:13 +0900 Subject: [PATCH 2/2] =?UTF-8?q?chore:=20=E7=B7=A8=E9=9B=86=E5=AF=BE?= =?UTF-8?q?=E8=B1=A1=E5=A4=96=E3=83=95=E3=82=A1=E3=82=A4=E3=83=AB=E3=81=AE?= =?UTF-8?q?=20fmt=20=E5=B7=AE=E5=88=86=E3=82=92=E9=99=A4=E5=8E=BB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Fable 5 --- src/api.rs | 35 +++++---------- src/commands/auth.rs | 16 ++++--- src/commands/doctor.rs | 51 ++++------------------ src/commands/mod.rs | 14 ++---- src/commands/notes.rs | 38 ++++++++++++----- src/http_server.rs | 67 ++++++++++++----------------- src/main.rs | 5 ++- src/streaming.rs | 96 ++++++++++++------------------------------ 8 files changed, 119 insertions(+), 203 deletions(-) diff --git a/src/api.rs b/src/api.rs index 55af8e2..dbd0c9e 100644 --- a/src/api.rs +++ b/src/api.rs @@ -8,11 +8,11 @@ use serde_json::{json, Value}; use crate::error::NoteDeckError; use crate::models::{ - Antenna, AuthResult, Channel, ChatMessage, ChatUser, Clip, CreateNoteParams, MutedWordsResult, + Antenna, AuthResult, Channel, ChatMessage, ChatUser, Clip, CreateNoteParams, NormalizedDriveFile, NormalizedNote, NormalizedNoteReaction, NormalizedNotification, - NormalizedUser, NormalizedUserDetail, RawCreateNoteResponse, RawDriveFile, RawEmojisResponse, - RawMiAuthResponse, RawNote, RawNoteReaction, RawNotification, RawUser, RawUserDetail, - SearchOptions, ServerEmoji, TimelineOptions, TimelineType, UserList, + MutedWordsResult, NormalizedUser, NormalizedUserDetail, RawCreateNoteResponse, RawDriveFile, + RawEmojisResponse, RawMiAuthResponse, RawNote, RawNoteReaction, RawNotification, RawUser, + RawUserDetail, SearchOptions, ServerEmoji, TimelineOptions, TimelineType, UserList, }; /// Maximum response body size (50 MB) to prevent memory exhaustion from malicious servers. @@ -251,12 +251,7 @@ impl MisskeyClient { antenna_id: &str, ) -> Result { let data = self - .request( - host, - token, - "antennas/show", - json!({ "antennaId": antenna_id }), - ) + .request(host, token, "antennas/show", json!({ "antennaId": antenna_id })) .await?; let antenna: Antenna = serde_json::from_value(data)?; Ok(antenna) @@ -2906,10 +2901,7 @@ mod tests { let role = notifs[0].role.as_ref().expect("role present"); assert_eq!(role.name, "Active"); assert_eq!(role.color.as_deref(), Some("#ff0000")); - assert_eq!( - role.icon_url.as_deref(), - Some("https://example.com/role.png") - ); + assert_eq!(role.icon_url.as_deref(), Some("https://example.com/role.png")); } #[tokio::test] @@ -3604,9 +3596,7 @@ mod tests { // withReplies のみ指定 → notify は body に含まれないこと Mock::given(method("POST")) .and(path("/api/following/update")) - .and(body_partial_json( - json!({ "userId": "u1", "withReplies": false }), - )) + .and(body_partial_json(json!({ "userId": "u1", "withReplies": false }))) .respond_with(ResponseTemplate::new(200).set_body_json(json!({}))) .mount(&server) .await; @@ -3623,9 +3613,7 @@ mod tests { let server = MockServer::start().await; Mock::given(method("POST")) .and(path("/api/users/update-memo")) - .and(body_partial_json( - json!({ "userId": "u1", "memo": "friend" }), - )) + .and(body_partial_json(json!({ "userId": "u1", "memo": "friend" }))) .respond_with(ResponseTemplate::new(200).set_body_json(json!({}))) .mount(&server) .await; @@ -3712,11 +3700,10 @@ mod tests { let server = MockServer::start().await; Mock::given(method("POST")) .and(path("/api/notes/search")) - .and(body_partial_json( - json!({ "query": "rust", "userId": "u1" }), - )) + .and(body_partial_json(json!({ "query": "rust", "userId": "u1" }))) .respond_with( - ResponseTemplate::new(200).set_body_json(json!([raw_note_json("n1", "rust note")])), + ResponseTemplate::new(200) + .set_body_json(json!([raw_note_json("n1", "rust note")])), ) .mount(&server) .await; diff --git a/src/commands/auth.rs b/src/commands/auth.rs index d5de2fc..597061b 100644 --- a/src/commands/auth.rs +++ b/src/commands/auth.rs @@ -52,7 +52,11 @@ pub fn run_accounts(db: &Database, fmt: OutputFormat) -> Result<(), NoteDeckErro Ok(()) } -pub async fn run_login(db: &Database, host: &str, fmt: OutputFormat) -> Result<(), NoteDeckError> { +pub async fn run_login( + db: &Database, + host: &str, + fmt: OutputFormat, +) -> Result<(), NoteDeckError> { let client = MisskeyClient::new()?; let session_id = uuid::Uuid::new_v4().to_string(); @@ -85,8 +89,9 @@ pub async fn run_login(db: &Database, host: &str, fmt: OutputFormat) -> Result<( ]; let permission_str = permissions.join(","); let scheme = crate::insecure::http_scheme(host); - let auth_url = - format!("{scheme}://{host}/miauth/{session_id}?name=notecli&permission={permission_str}"); + let auth_url = format!( + "{scheme}://{host}/miauth/{session_id}?name=notecli&permission={permission_str}" + ); match fmt { OutputFormat::Json | OutputFormat::Jsonl => { @@ -103,10 +108,7 @@ pub async fn run_login(db: &Database, host: &str, fmt: OutputFormat) -> Result<( println!(); println!(" {}", theme::link(&auth_url)); println!(); - println!( - "{}", - theme::muted("認証が完了したらEnterを押してください...") - ); + println!("{}", theme::muted("認証が完了したらEnterを押してください...")); } } diff --git a/src/commands/doctor.rs b/src/commands/doctor.rs index 93f241b..a241f9f 100644 --- a/src/commands/doctor.rs +++ b/src/commands/doctor.rs @@ -35,22 +35,10 @@ pub struct Check { impl Check { fn env(name: &str, status: Status, message: String) -> Self { - Self { - name: name.into(), - status, - message, - account: None, - fix: None, - } + Self { name: name.into(), status, message, account: None, fix: None } } fn acc(account: &str, name: &str, status: Status, message: String) -> Self { - Self { - name: name.into(), - status, - message, - account: Some(account.into()), - fix: None, - } + Self { name: name.into(), status, message, account: Some(account.into()), fix: None } } fn with_fix(mut self, fix: impl Into) -> Self { self.fix = Some(fix.into()); @@ -90,10 +78,7 @@ pub async fn diagnose( } let targets: Vec<&Account> = match account_spec { - Some(spec) => accounts - .iter() - .filter(|a| account_matches(a, spec)) - .collect(), + Some(spec) => accounts.iter().filter(|a| account_matches(a, spec)).collect(), None => accounts.iter().collect(), }; if let Some(spec) = account_spec { @@ -134,11 +119,7 @@ pub async fn run_doctor( fn check_database(db: &Database, path: &Path) -> Check { match db.load_accounts() { - Ok(_) => Check::env( - "database", - Status::Ok, - format!("{} (readable)", path.display()), - ), + Ok(_) => Check::env("database", Status::Ok, format!("{} (readable)", path.display())), Err(e) => Check::env( "database", Status::Fail, @@ -176,12 +157,7 @@ async fn check_account(client: &MisskeyClient, a: &Account, checks: &mut Vec 0 { println!("{}", theme::error(&format!("{fails} check(s) failed"))); } else if warns > 0 { - println!( - "{}", - theme::badge(&format!("all checks passed ({warns} warning(s))")) - ); + println!("{}", theme::badge(&format!("all checks passed ({warns} warning(s))"))); } else { println!("{}", theme::success("all checks passed")); } diff --git a/src/commands/mod.rs b/src/commands/mod.rs index 2acd9b1..da1af40 100644 --- a/src/commands/mod.rs +++ b/src/commands/mod.rs @@ -63,15 +63,7 @@ pub async fn run_cli( reply_to, local_only, } => { - notes::run_post( - &ctx, - text, - cw.as_deref(), - visibility, - reply_to.as_deref(), - *local_only, - ) - .await + notes::run_post(&ctx, text, cw.as_deref(), visibility, reply_to.as_deref(), *local_only).await } Commands::Timeline { r#type, limit } => notes::run_timeline(&ctx, r#type, *limit).await, Commands::Search { query, limit } => notes::run_search(&ctx, query, *limit).await, @@ -79,7 +71,9 @@ pub async fn run_cli( Commands::Replies { id, limit } => notes::run_replies(&ctx, id, *limit).await, Commands::Thread { id, limit } => notes::run_thread(&ctx, id, *limit).await, Commands::Delete { id } => notes::run_delete(&ctx, id).await, - Commands::Update { id, text, cw } => notes::run_update(&ctx, id, text, cw.as_deref()).await, + Commands::Update { id, text, cw } => { + notes::run_update(&ctx, id, text, cw.as_deref()).await + } Commands::React { note_id, reaction } => notes::run_react(&ctx, note_id, reaction).await, Commands::Unreact { note_id } => notes::run_unreact(&ctx, note_id).await, Commands::Renote { note_id } => notes::run_renote(&ctx, note_id).await, diff --git a/src/commands/notes.rs b/src/commands/notes.rs index 847df7f..6e93854 100644 --- a/src/commands/notes.rs +++ b/src/commands/notes.rs @@ -43,11 +43,7 @@ pub async fn run_post( Ok(()) } -pub async fn run_timeline( - ctx: &CmdContext, - tl_type: &str, - limit: i64, -) -> Result<(), NoteDeckError> { +pub async fn run_timeline(ctx: &CmdContext, tl_type: &str, limit: i64) -> Result<(), NoteDeckError> { let notes = ctx .client .get_timeline( @@ -62,7 +58,11 @@ pub async fn run_timeline( Ok(()) } -pub async fn run_search(ctx: &CmdContext, query: &str, limit: i64) -> Result<(), NoteDeckError> { +pub async fn run_search( + ctx: &CmdContext, + query: &str, + limit: i64, +) -> Result<(), NoteDeckError> { let notes = ctx .client .search_notes( @@ -93,7 +93,11 @@ pub async fn run_note(ctx: &CmdContext, id: &str) -> Result<(), NoteDeckError> { Ok(()) } -pub async fn run_replies(ctx: &CmdContext, id: &str, limit: i64) -> Result<(), NoteDeckError> { +pub async fn run_replies( + ctx: &CmdContext, + id: &str, + limit: i64, +) -> Result<(), NoteDeckError> { let notes = ctx .client .get_note_children(&ctx.host, &ctx.token, &ctx.account.id, id, limit as u32) @@ -102,7 +106,11 @@ pub async fn run_replies(ctx: &CmdContext, id: &str, limit: i64) -> Result<(), N Ok(()) } -pub async fn run_thread(ctx: &CmdContext, id: &str, limit: i64) -> Result<(), NoteDeckError> { +pub async fn run_thread( + ctx: &CmdContext, + id: &str, + limit: i64, +) -> Result<(), NoteDeckError> { let notes = ctx .client .get_note_conversation(&ctx.host, &ctx.token, &ctx.account.id, id, limit as u32) @@ -272,14 +280,24 @@ pub async fn run_unfavorite(ctx: &CmdContext, note_id: &str) -> Result<(), NoteD pub async fn run_favorites(ctx: &CmdContext, limit: i64) -> Result<(), NoteDeckError> { let notes = ctx .client - .get_favorites(&ctx.host, &ctx.token, &ctx.account.id, limit, None, None) + .get_favorites( + &ctx.host, + &ctx.token, + &ctx.account.id, + limit, + None, + None, + ) .await?; print_notes(¬es, ctx.fmt); Ok(()) } pub async fn run_emojis(ctx: &CmdContext) -> Result<(), NoteDeckError> { - let emojis = ctx.client.get_server_emojis(&ctx.host, &ctx.token).await?; + let emojis = ctx + .client + .get_server_emojis(&ctx.host, &ctx.token) + .await?; print_emojis(&emojis, ctx.fmt); Ok(()) } diff --git a/src/http_server.rs b/src/http_server.rs index 117a395..8608c86 100644 --- a/src/http_server.rs +++ b/src/http_server.rs @@ -9,12 +9,12 @@ use axum::{ routing::get, Json, Router, }; +use subtle::ConstantTimeEq; use futures_util::stream::Stream; use serde::Deserialize; use serde_json::{json, Value}; use std::net::SocketAddr; use std::sync::Arc; -use subtle::ConstantTimeEq; use tokio_stream::wrappers::BroadcastStream; use tokio_stream::StreamExt; use tower_http::cors::CorsLayer; @@ -239,11 +239,7 @@ fn core_openapi_router() -> OpenApiRouter { .routes(routes!(get_note, delete_note)) .routes(routes!(get_note_children)) .routes(routes!(get_note_conversation)) - .routes(routes!( - get_note_reactions, - create_reaction, - delete_reaction - )) + .routes(routes!(get_note_reactions, create_reaction, delete_reaction)) .routes(routes!(get_user)) .routes(routes!(get_user_notes)) .routes(routes!(search_notes)) @@ -258,10 +254,7 @@ fn core_openapi_router() -> OpenApiRouter { /// own spec. pub fn build_core_routes(state: AppState) -> OpenApiRouter { core_openapi_router() - .layer(middleware::from_fn_with_state( - state.clone(), - auth_middleware, - )) + .layer(middleware::from_fn_with_state(state.clone(), auth_middleware)) .layer(CorsLayer::permissive()) .with_state(state) } @@ -300,14 +293,8 @@ pub fn endpoints_from_spec(openapi: &utoipa::openapi::OpenApi) -> Vec { } } out.sort_by(|a, b| { - let ka = ( - a["path"].as_str().unwrap_or(""), - a["method"].as_str().unwrap_or(""), - ); - let kb = ( - b["path"].as_str().unwrap_or(""), - b["method"].as_str().unwrap_or(""), - ); + let ka = (a["path"].as_str().unwrap_or(""), a["method"].as_str().unwrap_or("")); + let kb = (b["path"].as_str().unwrap_or(""), b["method"].as_str().unwrap_or("")); ka.cmp(&kb) }); out @@ -635,14 +622,7 @@ async fn get_note_reactions( let limit = opts.limit.unwrap_or(20); let reactions = state .client - .get_note_reactions( - &h, - &token, - ¬e_id, - opts.r#type.as_deref(), - limit, - opts.until_id.as_deref(), - ) + .get_note_reactions(&h, &token, ¬e_id, opts.r#type.as_deref(), limit, opts.until_id.as_deref()) .await?; Ok(Json(reactions)) } @@ -694,7 +674,10 @@ async fn delete_reaction( ) -> Result { let account_id = state.account_id_for_host(&host)?; let (h, token) = crate::get_credentials(&state.db, &account_id)?; - state.client.delete_reaction(&h, &token, ¬e_id).await?; + state + .client + .delete_reaction(&h, &token, ¬e_id) + .await?; Ok(StatusCode::NO_CONTENT) } @@ -805,20 +788,22 @@ async fn sse_events( .r#type .map(|t| t.split(',').map(|s| s.trim().to_string()).collect()); - let stream = BroadcastStream::new(rx).filter_map(move |result| match result { - Ok(sse_event) => { - if let Some(ref filter) = type_filter { - if !filter.iter().any(|f| sse_event.event_type.starts_with(f)) { - return None; + let stream = BroadcastStream::new(rx).filter_map(move |result| { + match result { + Ok(sse_event) => { + if let Some(ref filter) = type_filter { + if !filter.iter().any(|f| sse_event.event_type.starts_with(f)) { + return None; + } } + let event = Event::default() + .event(&sse_event.event_type) + .json_data(&sse_event.data) + .ok()?; + Some(Ok(event)) } - let event = Event::default() - .event(&sse_event.event_type) - .json_data(&sse_event.data) - .ok()?; - Some(Ok(event)) + Err(_) => None, } - Err(_) => None, }); Sse::new(stream).keep_alive(KeepAlive::default()) @@ -839,7 +824,11 @@ struct TimelineQueryParams { impl TimelineQueryParams { fn into_timeline_options(self) -> crate::models::TimelineOptions { - crate::models::TimelineOptions::new(self.limit.unwrap_or(20), self.since_id, self.until_id) + crate::models::TimelineOptions::new( + self.limit.unwrap_or(20), + self.since_id, + self.until_id, + ) } } diff --git a/src/main.rs b/src/main.rs index 2a045a1..1a533d4 100644 --- a/src/main.rs +++ b/src/main.rs @@ -58,8 +58,9 @@ async fn run_daemon(port: u16) { let db_path = data_dir.join("notecli.db"); let db = Arc::new(Database::open(&db_path).expect("Failed to open database")); - let client = - Arc::new(notecli::api::MisskeyClient::new().expect("Failed to create HTTP client")); + let client = Arc::new( + notecli::api::MisskeyClient::new().expect("Failed to create HTTP client"), + ); let event_bus = Arc::new(EventBus::new()); diff --git a/src/streaming.rs b/src/streaming.rs index 2bbed59..bfdf5a9 100644 --- a/src/streaming.rs +++ b/src/streaming.rs @@ -405,8 +405,7 @@ impl StreamingManager { } else { StreamConnectionState::Reconnecting }; - self.emitter - .emit(StreamEvent::Status(Box::new(StreamStatusEvent { + self.emitter.emit(StreamEvent::Status(Box::new(StreamStatusEvent { account_id: account_id.to_string(), state, }))); @@ -434,10 +433,7 @@ impl StreamingManager { None } Err(_) => { - tracing::warn!( - account_id, - "initial connect timed out; retrying in background" - ); + tracing::warn!(account_id, "initial connect timed out; retrying in background"); None } }; @@ -487,8 +483,7 @@ impl StreamingManager { }, ); - self.emitter - .emit(StreamEvent::Status(Box::new(StreamStatusEvent { + self.emitter.emit(StreamEvent::Status(Box::new(StreamStatusEvent { account_id: account_id.to_string(), state: if connected { StreamConnectionState::Connected @@ -531,8 +526,7 @@ impl StreamingManager { captured.remove(account_id); drop(captured); - self.emitter - .emit(StreamEvent::Status(Box::new(StreamStatusEvent { + self.emitter.emit(StreamEvent::Status(Box::new(StreamStatusEvent { account_id: account_id.to_string(), state: StreamConnectionState::Disconnected, }))); @@ -574,8 +568,7 @@ impl StreamingManager { let interval = Duration::from_millis(interval_ms.unwrap_or(15_000)); self.start_polling(account_id, host, token, interval).await; - self.emitter - .emit(StreamEvent::Status(Box::new(StreamStatusEvent { + self.emitter.emit(StreamEvent::Status(Box::new(StreamStatusEvent { account_id: account_id.to_string(), state: StreamConnectionState::Connected, }))); @@ -1092,9 +1085,9 @@ async fn connection_task( loop { connected_flag.store(false, Ordering::Relaxed); emitter.emit(StreamEvent::Status(Box::new(StreamStatusEvent { - account_id: account_id.clone(), - state: StreamConnectionState::Reconnecting, - }))); + account_id: account_id.clone(), + state: StreamConnectionState::Reconnecting, + }))); // Wait with backoff, but listen for Shutdown during the wait. // Equal Jitter (sleep in [backoff/2, backoff]) de-syncs reconnects @@ -1135,9 +1128,9 @@ async fn connection_task( connected_flag.store(true, Ordering::Relaxed); emitter.emit(StreamEvent::Status(Box::new(StreamStatusEvent { - account_id: account_id.clone(), - state: StreamConnectionState::Connected, - }))); + account_id: account_id.clone(), + state: StreamConnectionState::Connected, + }))); let reason = run_ws_session( &emitter, @@ -1425,11 +1418,7 @@ async fn handle_ws_message( note_id, update, }; - emit_both( - emitter, - event_bus, - StreamEvent::NoteCaptureUpdated(Box::new(payload)), - ); + emit_both(emitter, event_bus, StreamEvent::NoteCaptureUpdated(Box::new(payload))); } return; } @@ -1512,11 +1501,7 @@ async fn handle_ws_message( note_id, update, }; - emit_both( - emitter, - event_bus, - StreamEvent::NoteUpdated(Box::new(payload)), - ); + emit_both(emitter, event_bus, StreamEvent::NoteUpdated(Box::new(payload))); } else if kind == "main" { if event_type == "notification" { if let Ok(raw) = serde_json::from_value::(event_body) { @@ -1526,11 +1511,7 @@ async fn handle_ws_message( subscription_id: sub_id, notification, }; - emit_both( - emitter, - event_bus, - StreamEvent::Notification(Box::new(payload)), - ); + emit_both(emitter, event_bus, StreamEvent::Notification(Box::new(payload))); } } else if event_type == "mention" || event_type == "reply" { // main-event として emit しつつ、mention としても parse を試みる @@ -1557,11 +1538,7 @@ async fn handle_ws_message( event_type, body: event_body, }; - emit_both( - emitter, - event_bus, - StreamEvent::MainEvent(Box::new(payload)), - ); + emit_both(emitter, event_bus, StreamEvent::MainEvent(Box::new(payload))); } } else if kind == "chat" { if event_type == "message" { @@ -1598,11 +1575,7 @@ async fn handle_ws_message( subscription_id: sub_id, message: msg, }; - emit_both( - emitter, - event_bus, - StreamEvent::ChatMessage(Box::new(payload)), - ); + emit_both(emitter, event_bus, StreamEvent::ChatMessage(Box::new(payload))); } } else if event_type == "deleted" { if let Some(id) = event_body.as_str() { @@ -1623,11 +1596,7 @@ async fn handle_ws_message( subscription_id: sub_id, message_id: id_owned, }; - emit_both( - emitter, - event_bus, - StreamEvent::ChatMessageDeleted(Box::new(payload)), - ); + emit_both(emitter, event_bus, StreamEvent::ChatMessageDeleted(Box::new(payload))); } } else if event_type == "react" || event_type == "unreact" { let is_react = event_type == "react"; @@ -1657,11 +1626,7 @@ async fn handle_ws_message( reaction: body.reaction, user: body.user, }; - emit_both( - emitter, - event_bus, - StreamEvent::ChatMessageReacted(Box::new(payload)), - ); + emit_both(emitter, event_bus, StreamEvent::ChatMessageReacted(Box::new(payload))); } else { let payload = StreamChatMessageUnreactedEvent { account_id: account_id.to_string(), @@ -1670,11 +1635,7 @@ async fn handle_ws_message( reaction: body.reaction, user: body.user, }; - emit_both( - emitter, - event_bus, - StreamEvent::ChatMessageUnreacted(Box::new(payload)), - ); + emit_both(emitter, event_bus, StreamEvent::ChatMessageUnreacted(Box::new(payload))); } } } @@ -1771,11 +1732,7 @@ async fn polling_loop( subscription_id: sub_id.clone(), note, }; - emit_both( - emitter.as_ref(), - &event_bus, - StreamEvent::Note(Box::new(payload)), - ); + emit_both(emitter.as_ref(), &event_bus, StreamEvent::Note(Box::new(payload))); } consecutive_failures = 0; @@ -1884,9 +1841,9 @@ async fn polling_loop( }; emitter.emit(StreamEvent::Status(Box::new(StreamStatusEvent { - account_id: account_id.clone(), - state: StreamConnectionState::Reconnecting, - }))); + account_id: account_id.clone(), + state: StreamConnectionState::Reconnecting, + }))); Duration::from_secs(backoff) } else { @@ -1970,8 +1927,11 @@ mod tests { let dir = tempfile::tempdir().unwrap(); let db = Arc::new(crate::db::Database::open(&dir.path().join("test.db")).unwrap()); let (tx, mut rx) = mpsc::unbounded_channel(); - let manager = - StreamingManager::new(Arc::new(ChannelEmitter(tx)), Arc::new(EventBus::new()), db); + let manager = StreamingManager::new( + Arc::new(ChannelEmitter(tx)), + Arc::new(EventBus::new()), + db, + ); // 127.0.0.1:1 は即 connection refused になる manager