//! SQLite 인덱스 접근 계층. //! //! 아키텍처: **단일 writer 스레드**(crossbeam 채널로 잡 수신, 배치 트랜잭션) //! + 읽기전용 커넥션 풀. WAL 모드라 읽기는 쓰기에 블록되지 않는다. pub mod migrations; use crossbeam_channel::{unbounded, Sender}; use rusqlite::{Connection, OpenFlags}; use std::path::{Path, PathBuf}; use std::sync::Mutex; use std::thread::JoinHandle; #[derive(Debug, thiserror::Error)] pub enum DbError { #[error("sqlite 오류: {0}")] Sqlite(#[from] rusqlite::Error), #[error("DB가 이미 닫혔습니다")] Closed, } pub type Result = std::result::Result; type WriteJob = Box; pub struct Db { path: PathBuf, write_tx: Option>, writer: Option>, readers: Mutex>, pub fts_enabled: bool, } impl Db { /// DB를 열고(없으면 생성) 마이그레이션을 적용한 뒤 writer 스레드를 시작한다. pub fn open(db_path: &Path) -> Result { let mut conn = Connection::open(db_path)?; apply_writer_pragmas(&conn)?; let fts_enabled = migrations::migrate(&mut conn)?; let (tx, rx) = unbounded::(); let writer = std::thread::Builder::new() .name("db-writer".into()) .spawn(move || { while let Ok(job) = rx.recv() { job(&mut conn); } // 종료 정리: 체크포인트 + 통계 갱신 let _ = conn.execute_batch("PRAGMA wal_checkpoint(TRUNCATE); PRAGMA optimize;"); }) .expect("db-writer 스레드 생성 실패"); Ok(Db { path: db_path.to_path_buf(), write_tx: Some(tx), writer: Some(writer), readers: Mutex::new(Vec::new()), fts_enabled, }) } /// writer 스레드에서 클로저를 실행하고 결과를 동기적으로 기다린다. pub fn with_write(&self, f: F) -> Result where T: Send + 'static, F: FnOnce(&mut Connection) -> rusqlite::Result + Send + 'static, { let tx = self.write_tx.as_ref().ok_or(DbError::Closed)?; let (rtx, rrx) = std::sync::mpsc::sync_channel::>(1); tx.send(Box::new(move |conn| { let _ = rtx.send(f(conn)); })) .map_err(|_| DbError::Closed)?; rrx.recv().map_err(|_| DbError::Closed)?.map_err(Into::into) } /// 결과를 기다리지 않는 쓰기(스캔 배치 등 대량 파이프라인용). pub fn write_detached(&self, f: F) -> Result<()> where F: FnOnce(&mut Connection) + Send + 'static, { let tx = self.write_tx.as_ref().ok_or(DbError::Closed)?; tx.send(Box::new(f)).map_err(|_| DbError::Closed) } /// 읽기전용 커넥션으로 쿼리를 실행한다 (풀에서 재사용). pub fn with_read(&self, f: impl FnOnce(&Connection) -> rusqlite::Result) -> Result { let conn = match self.readers.lock().unwrap().pop() { Some(c) => c, None => self.open_reader()?, }; let result = f(&conn); // 풀 크기 상한: 커넥션 4개 let mut pool = self.readers.lock().unwrap(); if pool.len() < 4 { pool.push(conn); } drop(pool); result.map_err(Into::into) } fn open_reader(&self) -> Result { let conn = Connection::open_with_flags( &self.path, OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX, )?; apply_reader_pragmas(&conn)?; Ok(conn) } } impl Drop for Db { fn drop(&mut self) { // 채널을 닫으면 writer 루프가 종료된다 self.write_tx.take(); if let Some(h) = self.writer.take() { let _ = h.join(); } } } fn apply_writer_pragmas(conn: &Connection) -> rusqlite::Result<()> { conn.pragma_update(None, "journal_mode", "WAL")?; conn.pragma_update(None, "synchronous", "NORMAL")?; common_pragmas(conn) } fn apply_reader_pragmas(conn: &Connection) -> rusqlite::Result<()> { // journal_mode는 DB 영속 속성이라 읽기전용 커넥션에서는 건드리지 않는다 common_pragmas(conn) } fn common_pragmas(conn: &Connection) -> rusqlite::Result<()> { conn.pragma_update(None, "temp_store", "MEMORY")?; conn.pragma_update(None, "cache_size", -64000)?; conn.pragma_update(None, "mmap_size", 268_435_456i64)?; conn.pragma_update(None, "foreign_keys", "ON")?; conn.pragma_update(None, "busy_timeout", 5000)?; Ok(()) } #[cfg(test)] mod tests { use super::*; fn open_temp() -> (tempfile::TempDir, Db) { let dir = tempfile::tempdir().unwrap(); let db = Db::open(&dir.path().join("test.db")).unwrap(); (dir, db) } fn insert_fixture_file(db: &Db, name: &str) -> i64 { db.with_write({ let name = name.to_string(); move |conn| { conn.execute( "INSERT OR IGNORE INTO sources(id, kind, name, root) VALUES (1,'local','테스트','C:\\media')", [], )?; conn.execute( "INSERT OR IGNORE INTO folders(id, source_id, path, name) VALUES (1,1,'C:\\media','media')", [], )?; conn.execute( "INSERT INTO files(source_id, folder_id, name, ext, kind, size, mtime_ms) VALUES (1, 1, ?1, 'jpg', 0, 1000, 0)", [&name], )?; Ok(conn.last_insert_rowid()) } }) .unwrap() } /// M0 스파이크 ③: bundled SQLite의 FTS5 trigram 가용성 확인. #[test] fn fts5_trigram_smoke() { let (_dir, db) = open_temp(); assert!(db.fts_enabled, "bundled SQLite에 FTS5 trigram이 없음 — LIKE 폴백 필요"); insert_fixture_file(&db, "가족여행_제주도_2025.jpg"); insert_fixture_file(&db, "회사_워크샵.png"); // 한글 부분 문자열(3자 이상) 매칭 let hits: i64 = db .with_read(|conn| { conn.query_row( "SELECT count(*) FROM files_fts WHERE files_fts MATCH ?1", ["제주도"], |r| r.get(0), ) }) .unwrap(); assert_eq!(hits, 1); } #[test] fn writer_roundtrip_and_read_pool() { let (_dir, db) = open_temp(); let id = insert_fixture_file(&db, "test.jpg"); let name: String = db .with_read(|conn| { conn.query_row("SELECT name FROM files WHERE id = ?1", [id], |r| r.get(0)) }) .unwrap(); assert_eq!(name, "test.jpg"); } #[test] fn fts_sync_triggers() { let (_dir, db) = open_temp(); let id = insert_fixture_file(&db, "바다사진.jpg"); db.with_write(move |conn| { conn.execute("UPDATE files SET name = '산사진.jpg' WHERE id = ?1", [id]) }) .unwrap(); let old_hits: i64 = db .with_read(|conn| { conn.query_row( "SELECT count(*) FROM files_fts WHERE files_fts MATCH '바다사진'", [], |r| r.get(0), ) }) .unwrap(); let new_hits: i64 = db .with_read(|conn| { conn.query_row( "SELECT count(*) FROM files_fts WHERE files_fts MATCH '산사진'", [], |r| r.get(0), ) }) .unwrap(); assert_eq!(old_hits, 0); assert_eq!(new_hits, 1); } }