Skip to main content

roomrs_core/
database.rs

1//! Database · 빌더 · 스키마 정의 (명세 §5.4, §10)
2
3use crate::error::{Error, Result};
4use crate::handle::SyncHandle;
5use crate::pool::{ConnectionPool, Pool};
6use rusqlite::Connection;
7use std::path::PathBuf;
8use std::sync::Arc;
9use std::time::Duration;
10
11/// 콜백 타입 별칭
12pub(crate) type ConnCallback = Arc<dyn Fn(&Connection) -> Result<()> + Send + Sync>;
13type QueryLogger = Box<dyn Fn(&str, Duration) + Send + Sync>;
14
15#[cfg(feature = "live")]
16thread_local! {
17    /// 현재 스레드의 중첩 SQL 실행별 preupdate_hook 수집 버퍼.
18    static HOOK_CAPTURES: std::cell::RefCell<Vec<HookCaptureFrame>> =
19        const { std::cell::RefCell::new(Vec::new()) };
20}
21
22/// preupdate_hook 수집 프레임.
23#[cfg(feature = "live")]
24#[derive(Default)]
25pub(crate) struct HookCaptureFrame {
26    pub(crate) tables: std::collections::HashSet<String>,
27    pub(crate) changes: Vec<crate::live::TableChange>,
28}
29
30/// 현재 실행 중인 가장 안쪽 SQL의 행 변경을 기록한다.
31#[cfg(feature = "live")]
32pub(crate) fn record_hook_change(change: crate::live::TableChange) {
33    HOOK_CAPTURES.with(|captures| {
34        if let Some(current) = captures.borrow_mut().last_mut() {
35            current.tables.insert(change.table.clone());
36            current.changes.push(change);
37        }
38    });
39}
40
41/// preupdate_hook capture 프레임을 unwind에서도 제거하는 RAII guard.
42#[cfg(feature = "live")]
43pub(crate) struct HookCapture {
44    depth: usize,
45}
46
47#[cfg(feature = "live")]
48impl Drop for HookCapture {
49    /// 아직 남은 현재 프레임과 그 아래 비정상 중첩 프레임을 제거한다.
50    fn drop(&mut self) {
51        HOOK_CAPTURES.with(|captures| captures.borrow_mut().truncate(self.depth));
52    }
53}
54
55/// 스키마 컬럼 메타를 사용해 preupdate hook을 설치한다.
56#[cfg(feature = "live")]
57pub(crate) fn install_preupdate_hook(
58    conn: &Connection,
59    columns: Arc<std::collections::HashMap<String, Vec<String>>>,
60) -> Result<()> {
61    use rusqlite::hooks::PreUpdateCase;
62    use rusqlite::types::Value;
63
64    conn.preupdate_hook(Some(
65        move |_action, _db: &str, table: &str, change: &PreUpdateCase| {
66            let Some(column_names) = columns.get(&table.to_ascii_lowercase()) else {
67                return;
68            };
69            let read_old = |accessor: &rusqlite::hooks::PreUpdateOldValueAccessor| {
70                let mut row = std::collections::HashMap::new();
71                for (index, name) in column_names.iter().enumerate() {
72                    let Ok(value) = accessor.get_old_column_value(index as i32) else {
73                        return None;
74                    };
75                    row.insert(name.clone(), Value::from(value));
76                }
77                Some(row)
78            };
79            let read_new = |accessor: &rusqlite::hooks::PreUpdateNewValueAccessor| {
80                let mut row = std::collections::HashMap::new();
81                for (index, name) in column_names.iter().enumerate() {
82                    let Ok(value) = accessor.get_new_column_value(index as i32) else {
83                        return None;
84                    };
85                    row.insert(name.clone(), Value::from(value));
86                }
87                Some(row)
88            };
89            let (old, new) = match change {
90                PreUpdateCase::Insert(new) => (None, read_new(new)),
91                PreUpdateCase::Delete(old) => (read_old(old), None),
92                PreUpdateCase::Update {
93                    old_value_accessor,
94                    new_value_accessor,
95                } => (read_old(old_value_accessor), read_new(new_value_accessor)),
96                PreUpdateCase::Unknown => return,
97            };
98            if old.is_some() || new.is_some() {
99                record_hook_change(crate::live::TableChange {
100                    table: table.to_string(),
101                    old,
102                    new,
103                });
104            }
105        },
106    ))?;
107    Ok(())
108}
109
110/// pool 재오픈에 필요한 connection 로컬 설정 소유본.
111#[derive(Clone)]
112struct ConnectionSettings {
113    path: Option<PathBuf>,
114    mem_name: Option<String>,
115    busy_timeout: Duration,
116    on_open: Option<ConnCallback>,
117    #[cfg(feature = "cipher")]
118    encryption_key: Option<String>,
119}
120
121impl ConnectionSettings {
122    /// on_open과 반환 불변식을 적용한다.
123    fn initialize(&self, conn: &Connection) -> Result<()> {
124        if let Some(cb) = &self.on_open {
125            let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| cb(conn)))
126                .map_err(|_| Error::Internal("on_open 콜백 panic".into()))
127                .and_then(|result| result);
128            if !conn.is_autocommit() {
129                conn.execute_batch("ROLLBACK")?;
130            }
131            conn.pragma_update(None, "query_only", "OFF")?;
132            if self.mem_name.is_some() {
133                conn.pragma_update(None, "read_uncommitted", "ON")?;
134            }
135            result?;
136        }
137        Ok(())
138    }
139
140    /// 새 connection을 열고 공통 PRAGMA·선택 callback을 적용한다.
141    fn open(&self, initialize: bool) -> Result<Connection> {
142        let conn = match (&self.mem_name, &self.path) {
143            (Some(uri), _) => {
144                use rusqlite::OpenFlags;
145                Connection::open_with_flags(
146                    uri,
147                    OpenFlags::SQLITE_OPEN_READ_WRITE
148                        | OpenFlags::SQLITE_OPEN_CREATE
149                        | OpenFlags::SQLITE_OPEN_URI
150                        | OpenFlags::SQLITE_OPEN_NO_MUTEX,
151                )?
152            }
153            (None, Some(path)) => Connection::open(path)?,
154            (None, None) => return Err(Error::Config("DB 경로가 설정되지 않았습니다".into())),
155        };
156        #[cfg(feature = "cipher")]
157        if let Some(key) = &self.encryption_key {
158            conn.pragma_update(None, "key", key)?;
159        }
160        conn.busy_timeout(self.busy_timeout)?;
161        if self.mem_name.is_none() {
162            conn.pragma_update(None, "journal_mode", "WAL")?;
163        }
164        conn.pragma_update(None, "foreign_keys", "ON")?;
165        conn.pragma_update(None, "synchronous", "NORMAL")?;
166        conn.pragma_update(None, "query_only", "OFF")?;
167        if self.mem_name.is_some() {
168            conn.pragma_update(None, "read_uncommitted", "ON")?;
169        }
170        if initialize {
171            self.initialize(&conn)?;
172        }
173        Ok(conn)
174    }
175}
176
177/// SQL 식별자 이스케이프 — `"` 배증 후 따옴표로 감싼다 (M-8/M-9)
178pub(crate) fn escape_ident(name: &str) -> String {
179    format!("\"{}\"", name.replace('"', "\"\""))
180}
181
182/// 컬럼 메타 — `#[entity]` 생성, 스냅샷 대조·생성에 사용 (명세 §7)
183#[derive(Debug, Clone, Copy)]
184pub struct ColumnMeta {
185    pub name: &'static str,
186    /// SQLite 타입 (빈 문자열 = typeless)
187    pub sql_type: &'static str,
188    pub not_null: bool,
189    pub pk: bool,
190    /// rename 힌트 (명세 §8.3) — diff 초안 전용
191    pub renamed_from: Option<&'static str>,
192}
193
194/// 테이블 메타 — `#[database]`가 엔티티에서 수집
195pub struct TableMeta {
196    pub name: &'static str,
197    pub columns: &'static [ColumnMeta],
198    /// 테이블·인덱스 DDL
199    pub ddl: &'static [&'static str],
200}
201
202/// 테이블 스키마 정의 — `#[database]` 매크로가 엔티티 메타에서 구성
203pub struct SchemaDef {
204    /// 스키마 버전 (`#[database(version = N)]`)
205    pub version: u32,
206    /// 테이블·인덱스 DDL (실행 순서대로)
207    pub ddl: Vec<&'static str>,
208    /// 테이블·컬럼 메타 — 스냅샷 생성·해시 대조용 (명세 §7.4)
209    pub tables: Vec<TableMeta>,
210}
211
212impl SchemaDef {
213    /// 같은 SQLite 테이블 이름을 가리키는 엔티티 중복을 검증한다.
214    fn validate_unique_tables(&self) -> Result<()> {
215        for (index, table) in self.tables.iter().enumerate() {
216            if self.tables[..index]
217                .iter()
218                .any(|previous| previous.name.eq_ignore_ascii_case(table.name))
219            {
220                return Err(Error::Config(format!(
221                    "database entities에 SQLite 테이블 이름 중복: {}",
222                    table.name
223                )));
224            }
225        }
226        Ok(())
227    }
228
229    /// 엔티티 메타 → 스냅샷 변환 (매크로·런타임 공유 모델, 명세 §3)
230    pub fn to_snapshot(&self) -> roomrs_migrate::SchemaSnapshot {
231        roomrs_migrate::SchemaSnapshot {
232            version: self.version,
233            tables: self
234                .tables
235                .iter()
236                .map(|t| roomrs_migrate::TableSnapshot {
237                    name: t.name.to_string(),
238                    columns: t
239                        .columns
240                        .iter()
241                        .map(|c| roomrs_migrate::ColumnSnapshot {
242                            name: c.name.to_string(),
243                            sql_type: c.sql_type.to_string(),
244                            not_null: c.not_null,
245                            pk: c.pk,
246                            renamed_from: c.renamed_from.map(str::to_string),
247                        })
248                        .collect(),
249                    ddl: t.ddl.iter().map(|d| d.to_string()).collect(),
250                })
251                .collect(),
252        }
253    }
254}
255
256/// Compile-time embedded schema snapshot (spec §8.4, decision 21c).
257///
258/// `#[database]` reads every committed snapshot file
259/// (`migrations/schema/{db}.{version}.json`), compresses it with
260/// miniz_oxide and embeds it into the binary. The full set is exposed via
261/// [`DatabaseSpec::EMBEDDED_SCHEMAS`] in ascending version order.
262#[derive(Debug, Clone, Copy)]
263pub struct EmbeddedSchema {
264    /// Schema version of this snapshot.
265    pub version: u32,
266    /// Compressed snapshot JSON (raw deflate — see
267    /// `roomrs_migrate::compress_snapshot`).
268    pub compressed: &'static [u8],
269}
270
271impl EmbeddedSchema {
272    /// Decompress and parse the embedded snapshot.
273    ///
274    /// Fails with [`Error::Migration`] when the embedded blob is corrupt.
275    pub fn snapshot(&self) -> Result<roomrs_migrate::SchemaSnapshot> {
276        let raw = roomrs_migrate::decompress_snapshot(self.compressed).map_err(|e| {
277            Error::Migration(format!(
278                "내장 스냅샷(v{}) 압축 해제 실패: {e}",
279                self.version
280            ))
281        })?;
282        roomrs_migrate::SchemaSnapshot::from_slice(&raw)
283            .map_err(|e| Error::Migration(format!("내장 스냅샷(v{}) 파스 실패: {e}", self.version)))
284    }
285}
286
287/// `#[database]` 생성물이 구현하는 스펙 trait —
288/// core 빌더가 타입드 DB를 돌려줄 수 있게 한다
289pub trait DatabaseSpec: Sized {
290    /// 스키마 버전
291    const VERSION: u32;
292    /// Snake-case database name — snapshot files are named
293    /// `{DB_NAME}.{version}.json` (spec §7.2, decision 21). The
294    /// `#[database]` macro derives this from the struct identifier.
295    const DB_NAME: &'static str;
296    /// 컴파일 타임에 읽은 현재 버전 스냅샷 파일 해시 — 파일 부재 시 None (명세 §7.4b)
297    const SNAPSHOT_HASH: Option<u64> = None;
298    /// Embedded snapshots in ascending version order (spec §8.4,
299    /// decision 21c). Filled in by the `#[database]` macro; empty when no
300    /// snapshot files exist.
301    const EMBEDDED_SCHEMAS: &'static [EmbeddedSchema] = &[];
302    /// Whether the current-version snapshot file existed when the
303    /// `#[database]` macro expanded (decision 28, D-3b). Defaults to `true`
304    /// so manual `DatabaseSpec` impls are unaffected. When `false`, the
305    /// export test keeps failing even after the file exists and matches,
306    /// until a rebuild re-expands the macro and embeds the snapshot —
307    /// closing the fail-open window where `SNAPSHOT_HASH` and the embedded
308    /// chain silently stay stale.
309    const SNAPSHOT_FILE_SEEN: bool = true;
310    /// 엔티티들의 DDL 수집
311    fn schema() -> SchemaDef;
312    /// core Database를 감싸 사용자 DB 타입 생성
313    fn from_database(db: Database) -> Self;
314}
315
316/// 스냅샷 파일 생성/갱신 — 개발 플로우용 (명세 §7.4a).
317/// 경로: `resolve_schema_dir(manifest_dir)/{DB_NAME}.{VERSION}.json`
318pub fn write_schema_snapshot<T: DatabaseSpec>(manifest_dir: &str) -> Result<std::path::PathBuf> {
319    let schema = T::schema();
320    schema.validate_unique_tables()?;
321    let dir = roomrs_migrate::resolve_schema_dir(manifest_dir);
322    let path = roomrs_migrate::snapshot_path(&dir, T::DB_NAME, T::VERSION);
323    schema
324        .to_snapshot()
325        .write_to(&path)
326        .map_err(|e| Error::Config(format!("스냅샷 저장 실패: {e}")))?;
327    Ok(path)
328}
329
330/// 스냅샷 ↔ 엔티티 메타 일치 검사 — CI/check용 (명세 §7.4a)
331pub fn check_schema_snapshot<T: DatabaseSpec>(manifest_dir: &str) -> Result<()> {
332    let schema = T::schema();
333    schema.validate_unique_tables()?;
334    let dir = roomrs_migrate::resolve_schema_dir(manifest_dir);
335    let path = roomrs_migrate::snapshot_path(&dir, T::DB_NAME, T::VERSION);
336    let file = roomrs_migrate::SchemaSnapshot::read_from(&path)
337        .map_err(|e| Error::SnapshotStale(format!("스냅샷 파일을 읽을 수 없습니다: {e}")))?;
338    let code = schema.to_snapshot();
339    if file.hash() != code.hash() {
340        return Err(Error::SnapshotStale(format!(
341            "스냅샷과 엔티티 정의가 다릅니다 — `write_schema_snapshot`으로 재생성 필요 (파일: {})",
342            path.display()
343        )));
344    }
345    Ok(())
346}
347
348/// Snapshot export/stale-check helper called by the test that
349/// `#[database]` generates (spec §7.4, decisions 21b/28).
350///
351/// Behavior for the current-version snapshot file:
352/// - `ROOMRS_SCHEMA_EXPORT=0` in the environment: no-op, returns `Ok`.
353/// - File missing: write it, then return [`Error::SnapshotStale`] asking to
354///   commit and **rebuild** (`cargo clean -p <crate>` or touching a source
355///   file — cargo does not re-expand the macro for a newly created file,
356///   so `SNAPSHOT_HASH` and the embedded snapshot chain stay stale until
357///   then; decision 28, H-6).
358/// - File exists and matches the entity metadata hash, but
359///   [`DatabaseSpec::SNAPSHOT_FILE_SEEN`] is `false` (the macro expanded
360///   before the file existed): still [`Error::SnapshotStale`] until a
361///   rebuild embeds it (D-3b).
362/// - File matches and was seen at expansion: `Ok`.
363/// - File stale or corrupt: **rewrite** the file, then return
364///   [`Error::SnapshotStale`] so CI fails until the regenerated snapshot is
365///   committed.
366pub fn export_schema_for_test<T: DatabaseSpec>(manifest_dir: &str) -> Result<()> {
367    // 옵트아웃 (명세 §7.4) — CI/리포 설정으로 export 비활성 가능
368    if std::env::var("ROOMRS_SCHEMA_EXPORT").as_deref() == Ok("0") {
369        return Ok(());
370    }
371    let schema = T::schema();
372    schema.validate_unique_tables()?;
373    let code = schema.to_snapshot();
374    let dir = roomrs_migrate::resolve_schema_dir(manifest_dir);
375    let path = roomrs_migrate::snapshot_path(&dir, T::DB_NAME, T::VERSION);
376    // 저장 헬퍼 — 실패는 설정 에러로 승격
377    let write = |snap: &roomrs_migrate::SchemaSnapshot| {
378        snap.write_to(&path)
379            .map_err(|e| Error::Config(format!("스냅샷 저장 실패 ({}): {e}", path.display())))
380    };
381    if !path.exists() {
382        write(&code)?;
383        // 신규 파일은 include_bytes! 의존성에 미등록이라 cargo가 재전개하지
384        // 않는다 — 여기서 성공 처리하면 SNAPSHOT_HASH·내장 체인이 스테일한 채
385        // 남는다. 구체적 복구 행동까지 안내한다 (결정 28, H-6/D-3a)
386        return Err(Error::SnapshotStale(format!(
387            "스냅샷을 생성했습니다 — 커밋 후 `cargo clean -p <크레이트>` 또는 소스 touch 후 재빌드하세요: {}",
388            path.display()
389        )));
390    }
391    match roomrs_migrate::SchemaSnapshot::read_from(&path) {
392        // 파손 파일 = 재생성 후 실패 반환 — 부재와 구분 (M-19)
393        Err(e) => {
394            write(&code)?;
395            Err(Error::SnapshotStale(format!(
396                "스냅샷 파일 파손({e}) — 재생성했습니다, 커밋하세요: {}",
397                path.display()
398            )))
399        }
400        Ok(file) if file.hash() == code.hash() => {
401            // 파일은 일치하지만 매크로 전개 시점엔 없었다(생성 직후 재빌드 전) —
402            // 성공 처리하면 SNAPSHOT_HASH=None·내장 체인 누락이 침묵 통과하는
403            // fail-open 창이 열린다. 재전개 전까지 계속 실패시킨다 (D-3b)
404            if !T::SNAPSHOT_FILE_SEEN {
405                return Err(Error::SnapshotStale(format!(
406                    "스냅샷이 아직 바이너리에 반영되지 않았습니다 — `cargo clean -p <크레이트>` 또는 소스 touch 후 재빌드하세요: {}",
407                    path.display()
408                )));
409            }
410            Ok(())
411        }
412        // 스테일 = 재생성 후 실패 반환 (CI에서 미커밋 차단, 로컬 재생성)
413        Ok(_) => {
414            write(&code)?;
415            Err(Error::SnapshotStale(format!(
416                "스냅샷이 스테일하여 재생성했습니다 — 커밋하세요: {} (반복되면 다른 #[database]와 DB_NAME 충돌 가능 — 구조체명 snake_case가 크레이트 내에서 유일해야 합니다, M-11)",
417                path.display()
418            )))
419        }
420    }
421}
422
423/// 마이그레이션 정책 (명세 §8 — M1은 Auto 최소 동작만, M3에서 완성)
424#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
425pub enum MigrationPolicy {
426    /// 자동: 신규 DB면 DDL 실행, 버전 일치면 통과, 불일치면 에러(M3에서 diff 실행으로 대체)
427    #[default]
428    Auto,
429    /// 검증만: 버전 불일치 시 에러
430    Validate,
431}
432
433/// DB 내부 공유 상태 — 핸들·트랜잭션이 참조
434pub struct DatabaseInner {
435    pub(crate) pool: Pool,
436    query_logger: Option<QueryLogger>,
437    /// 무효화 트래커 + 노티파이어 (feature live, 명세 §9)
438    #[cfg(feature = "live")]
439    pub(crate) tracker: Arc<crate::live::Tracker>,
440    /// preupdate hook이 행 값을 이름으로 복원할 때 사용하는 컬럼 순서.
441    #[cfg(feature = "live")]
442    pub(crate) hook_columns: Arc<std::collections::HashMap<String, Vec<String>>>,
443    /// 노티파이어 스레드 join 핸들 — drop 시 join으로 커넥션 잔류 방지 (M-5)
444    #[cfg(feature = "live")]
445    notifier_join: Option<std::thread::JoinHandle<()>>,
446}
447
448/// 종료 로그 — live 미사용 빌드에도 close 로그를 남긴다 (지시서 logging-log-crate)
449#[cfg(not(feature = "live"))]
450impl Drop for DatabaseInner {
451    /// DB 종료 로그만 수행 (정리할 백그라운드 스레드 없음)
452    fn drop(&mut self) {
453        log::info!("database closed");
454    }
455}
456
457#[cfg(feature = "live")]
458impl Drop for DatabaseInner {
459    /// 노티파이어·MI 폴러 종료 신호 + join — 스레드/커넥션 잔류 방지 (M-5).
460    /// 마지막 Arc가 노티파이어 스레드 자신(구독 콜백 등)에서 drop되면 self-join이
461    /// 교착이므로 그 경우엔 join을 생략하고 분리한다 (H-3)
462    fn drop(&mut self) {
463        log::info!("database closing — shutting down notifier");
464        // 트래커 종료 — 레지스트리 청산으로 대기 중 recv를 깨운다 (M-7)
465        self.tracker.shutdown();
466        if let Some(h) = self.notifier_join.take() {
467            if h.thread().id() == std::thread::current().id() {
468                // 노티파이어 스레드 위에서의 drop — self-join 교착 방지, 분리 (H-3)
469                log::warn!(
470                    "database dropped on notifier thread — detaching notifier instead of joining"
471                );
472            } else {
473                let _ = h.join();
474            }
475        }
476        log::info!("database closed");
477    }
478}
479
480#[cfg(feature = "live")]
481impl DatabaseInner {
482    /// SQL 실행별 hook 수집 프레임을 시작한다.
483    pub(crate) fn begin_hook_capture(&self) -> HookCapture {
484        let depth = HOOK_CAPTURES.with(|captures| {
485            let mut captures = captures.borrow_mut();
486            let depth = captures.len();
487            captures.push(Default::default());
488            depth
489        });
490        HookCapture { depth }
491    }
492
493    /// 훅 수집분 회수
494    pub(crate) fn take_hook_capture(&self) -> HookCaptureFrame {
495        HOOK_CAPTURES.with(|captures| captures.borrow_mut().pop().unwrap_or_default())
496    }
497
498    /// 단문 write 성공 후 무효화 방출 — 문장 파싱 ∪ 훅 (명세 §9.2).
499    /// 확실한 읽기 전용 문장(SELECT/EXPLAIN)은 문장 기반 방출을 하지 않는다
500    /// (L-2). PRAGMA는 상태 변경 여부를 확실히 구분할 수 없어 전체 무효화한다.
501    /// 읽기 전용 문장은 훅 수집분(트리거 write)만 방출한다.
502    pub(crate) fn emit_after_write(&self, sql: &str) {
503        let capture = self.take_hook_capture();
504        let changed_tables: std::collections::HashSet<String> = capture
505            .changes
506            .iter()
507            .map(|change| change.table.clone())
508            .collect();
509        if !capture.changes.is_empty() {
510            self.tracker.invalidate_changes(capture.changes);
511        }
512        match crate::live::extract_write_tables(sql) {
513            crate::live::WriteTables::ReadOnly => {
514                let tables: std::collections::HashSet<String> = capture
515                    .tables
516                    .difference(&changed_tables)
517                    .cloned()
518                    .collect();
519                if !tables.is_empty() {
520                    self.tracker.invalidate(Some(tables));
521                }
522            }
523            crate::live::WriteTables::Tables(mut t) => {
524                t.extend(capture.tables);
525                t.retain(|table| !changed_tables.contains(table));
526                if !t.is_empty() {
527                    self.tracker.invalidate(Some(t));
528                }
529            }
530            // 파싱 실패/DDL = 보수적 전체 무효화
531            crate::live::WriteTables::Unknown => self.tracker.invalidate(None),
532        }
533    }
534}
535
536impl DatabaseInner {
537    /// 쿼리 로거 래핑 실행 — 로거 없으면 오버헤드 없이 통과.
538    /// log 파사드에는 SQL 문자열만 남긴다 — 파라미터 값 금지 (민감정보)
539    pub(crate) fn log_query<R>(&self, sql: &str, f: impl FnOnce() -> Result<R>) -> Result<R> {
540        log::trace!("SQL: {sql}");
541        match &self.query_logger {
542            None => f(),
543            Some(logger) => {
544                let start = std::time::Instant::now();
545                let out = f();
546                logger(sql, start.elapsed());
547                out
548            }
549        }
550    }
551}
552
553/// roomrs 코어 데이터베이스 — 사용자는 `#[database]` 생성 타입으로 감싸 쓴다
554pub struct Database {
555    inner: Arc<DatabaseInner>,
556}
557
558impl Database {
559    /// 동기 핸들 (명세 §5.0)
560    pub fn run_sync(&self) -> SyncHandle<'_> {
561        SyncHandle { inner: &self.inner }
562    }
563
564    /// 내부 상태 Arc — roomrs-async 전용 (직접 사용 금지)
565    #[doc(hidden)]
566    pub fn inner_arc(&self) -> Arc<DatabaseInner> {
567        Arc::clone(&self.inner)
568    }
569}
570
571impl DatabaseInner {
572    /// Arc에서 동기 핸들 구성 — roomrs-async 워커 전용 (직접 사용 금지)
573    #[doc(hidden)]
574    pub fn sync_handle(self: &Arc<Self>) -> SyncHandle<'_> {
575        SyncHandle { inner: self }
576    }
577}
578
579/// 빌더 (명세 §5.4)
580pub struct DatabaseBuilder<T: DatabaseSpec> {
581    path: Option<PathBuf>,
582    in_memory: bool,
583    connections: usize,
584    busy_timeout: Duration,
585    queue_timeout: Option<Duration>,
586    migrate: MigrationPolicy,
587    migrations: Vec<crate::migration::Migration>,
588    auto_migrate: bool,
589    destructive_fallback: bool,
590    #[cfg(feature = "cipher")]
591    encryption_key: Option<String>,
592    on_create: Option<ConnCallback>,
593    on_open: Option<ConnCallback>,
594    query_logger: Option<QueryLogger>,
595    _spec: std::marker::PhantomData<T>,
596}
597
598impl<T: DatabaseSpec> Default for DatabaseBuilder<T> {
599    /// 기본값 — 커넥션 수는 CPU 코어 기반(최대 4), busy_timeout 5초
600    fn default() -> Self {
601        let cores = std::thread::available_parallelism()
602            .map(|n| n.get())
603            .unwrap_or(2);
604        Self {
605            path: None,
606            in_memory: false,
607            connections: cores.clamp(1, 4) + 1,
608            busy_timeout: Duration::from_secs(5),
609            queue_timeout: None,
610            migrate: MigrationPolicy::Auto,
611            migrations: Vec::new(),
612            auto_migrate: false,
613            destructive_fallback: false,
614            #[cfg(feature = "cipher")]
615            encryption_key: None,
616            on_create: None,
617            on_open: None,
618            query_logger: None,
619            _spec: std::marker::PhantomData,
620        }
621    }
622}
623
624impl<T: DatabaseSpec> DatabaseBuilder<T> {
625    /// Sets the SQLite file path. The special `:memory:` path uses the same
626    /// single regular connection setup as [`Self::in_memory`].
627    pub fn sqlite(mut self, path: impl Into<PathBuf>) -> Self {
628        let path = path.into();
629        if path == std::path::Path::new(":memory:") {
630            self.path = None;
631            self.in_memory = true;
632        } else {
633            self.path = Some(path);
634            self.in_memory = false;
635        }
636        self
637    }
638
639    /// In-memory database for tests.
640    ///
641    /// In-memory databases use one regular read/write connection regardless
642    /// of [`Self::connections`]. This serializes transactions and avoids
643    /// SQLite shared-cache `SQLITE_LOCKED` failures. With the `live` feature,
644    /// the notifier still owns its separate internal connection.
645    pub fn in_memory(mut self) -> Self {
646        self.in_memory = true;
647        self.path = None;
648        self
649    }
650
651    /// Sets the number of read/write connections in the unified pool.
652    ///
653    /// In-memory databases always use one regular connection.
654    pub fn connections(mut self, n: usize) -> Self {
655        self.connections = n.max(1);
656        self
657    }
658
659    /// SQLITE_BUSY 대기 — 프로세스 내(통합 풀 동시 write)·외부 프로세스 write 경합 공용 (명세 §10)
660    pub fn busy_timeout(mut self, d: Duration) -> Self {
661        self.busy_timeout = d;
662        self
663    }
664
665    /// 커넥션 풀 대기 타임아웃 — 초과 시 `Error::QueueTimeout`
666    pub fn queue_timeout(mut self, d: Duration) -> Self {
667        self.queue_timeout = Some(d);
668        self
669    }
670
671    /// SQLCipher 암호화 키 (명세 §0 14, feature cipher) — 모든 커넥션 오픈 직후 PRAGMA key
672    #[cfg(feature = "cipher")]
673    pub fn encryption_key(mut self, key: impl Into<String>) -> Self {
674        self.encryption_key = Some(key.into());
675        self
676    }
677
678    /// 마이그레이션 정책
679    pub fn migrate(mut self, policy: MigrationPolicy) -> Self {
680        self.migrate = policy;
681        self
682    }
683
684    /// 마이그레이션 스텝 등록 (명세 §8.2/§8.3) — 세 소스 공통 표현
685    pub fn migration(mut self, m: crate::migration::Migration) -> Self {
686        self.migrations.push(m);
687        self
688    }
689
690    /// 마이그레이션 스텝 일괄 등록 — `migrations_dir!` 산출물 등
691    pub fn migrations(mut self, ms: impl IntoIterator<Item = crate::migration::Migration>) -> Self {
692        self.migrations.extend(ms);
693        self
694    }
695
696    /// Opt-in automatic migration from embedded snapshots (spec §8.4,
697    /// decision 21d, default off).
698    ///
699    /// When enabled, gaps in the registered migration chain are filled by
700    /// diffing consecutive embedded snapshots
701    /// ([`DatabaseSpec::EMBEDDED_SCHEMAS`]). Only **safe** operations are
702    /// executed automatically (CREATE TABLE, nullable ADD COLUMN, valid
703    /// RENAME COLUMN, CREATE INDEX). A gap whose diff contains destructive
704    /// changes fails with a clear [`Error::Migration`] instead — register a
705    /// manual step or use
706    /// [`fallback_to_destructive_migration`](Self::fallback_to_destructive_migration).
707    /// Registered steps always take precedence over synthesized ones.
708    pub fn auto_migrate(mut self, on: bool) -> Self {
709        self.auto_migrate = on;
710        self
711    }
712
713    /// 파괴적 마이그레이션 폴백 (명세 §8, 기본 off) —
714    /// 체인이 불충분하면 **모든 테이블을 삭제**하고 현재 스키마로 재생성한다.
715    pub fn fallback_to_destructive_migration(mut self, enable: bool) -> Self {
716        self.destructive_fallback = enable;
717        self
718    }
719
720    /// 최초 생성 시 1회 콜백 (테이블 생성 직후).
721    ///
722    /// Runs **inside** the schema-creation transaction (L-5): if the callback
723    /// fails, the schema DDL and `user_version` roll back together, so the
724    /// next open retries creation from scratch. Do not manage transactions
725    /// (`BEGIN`/`COMMIT`) inside the callback.
726    pub fn on_create(
727        mut self,
728        f: impl Fn(&Connection) -> Result<()> + Send + Sync + 'static,
729    ) -> Self {
730        self.on_create = Some(Arc::new(f));
731        self
732    }
733
734    /// 오픈 시마다 콜백
735    pub fn on_open(
736        mut self,
737        f: impl Fn(&Connection) -> Result<()> + Send + Sync + 'static,
738    ) -> Self {
739        self.on_open = Some(Arc::new(f));
740        self
741    }
742
743    /// 쿼리 로거 — (sql, 소요시간)
744    pub fn query_logger(mut self, f: impl Fn(&str, Duration) + Send + Sync + 'static) -> Self {
745        self.query_logger = Some(Box::new(f));
746        self
747    }
748
749    /// DB 오픈 — PRAGMA 초기화 · 스냅샷 스테일 검증 · 마이그레이션 · 풀 구성 (명세 §5.4)
750    pub fn build(mut self) -> Result<T> {
751        let schema = T::schema();
752        schema.validate_unique_tables()?;
753
754        // shared-cache 인메모리의 동시 BEGIN IMMEDIATE는 busy_timeout이 처리하지 못하는
755        // SQLITE_LOCKED를 반환하므로 일반 풀을 하나로 고정한다.
756        if self.in_memory {
757            self.connections = 1;
758        }
759
760        // 스냅샷 스테일 런타임 검증 (명세 §7.4b) —
761        // 매크로가 임베드한 스냅샷 파일 해시 vs 엔티티 메타 재계산 해시
762        if let Some(embedded) = T::SNAPSHOT_HASH {
763            let runtime = schema.to_snapshot().hash();
764            if embedded != runtime {
765                return Err(Error::SnapshotStale(format!(
766                    "스냅샷 해시 불일치 (파일={embedded:#x}, 엔티티={runtime:#x}) — \
767                     엔티티 수정 후 스냅샷 재생성이 필요합니다"
768                )));
769            }
770        }
771
772        // 인메모리 공유 이름 — 커넥션 N개가 같은 DB를 보도록 named shared-cache URI 사용
773        let mem_name = if self.in_memory {
774            use std::sync::atomic::{AtomicU64, Ordering};
775            static MEM_SEQ: AtomicU64 = AtomicU64::new(0);
776            Some(format!(
777                "file:roomrs_mem_{}?mode=memory&cache=shared",
778                MEM_SEQ.fetch_add(1, Ordering::Relaxed)
779            ))
780        } else {
781            None
782        };
783
784        // 통합 풀 커넥션 오픈 + 공통 PRAGMA (명세 §10)
785        let first_conn = self.open_conn(mem_name.as_deref(), &schema, false)?;
786
787        // 일반 작업 커넥션 — 모두 read/write 가능 (명세 §10)
788        let mut connections = Vec::with_capacity(self.connections);
789        connections.push(first_conn);
790        for _ in 1..self.connections {
791            let conn = self.open_conn(mem_name.as_deref(), &schema, false)?;
792            connections.push(conn);
793        }
794
795        // 라이브 쿼리 기반 구축 (feature live, 명세 §9) —
796        // 노티파이어 전용 커넥션 + preupdate_hook 행 변경 수집 설치
797        #[cfg(feature = "live")]
798        let hook_columns = Arc::new(
799            schema
800                .tables
801                .iter()
802                .map(|table| {
803                    (
804                        table.name.to_ascii_lowercase(),
805                        table
806                            .columns
807                            .iter()
808                            .map(|column| column.name.to_string())
809                            .collect(),
810                    )
811                })
812                .collect(),
813        );
814        #[cfg(feature = "live")]
815        let (tracker, notifier_join) = {
816            let notifier_conn = self.open_conn(mem_name.as_deref(), &schema, false)?;
817            for conn in &connections {
818                install_preupdate_hook(conn, Arc::clone(&hook_columns))?;
819            }
820            crate::live::Tracker::start(notifier_conn)?
821        };
822
823        let reconnect_settings = ConnectionSettings {
824            path: self.path.clone(),
825            mem_name: mem_name.clone(),
826            busy_timeout: self.busy_timeout,
827            on_open: self.on_open.clone(),
828            #[cfg(feature = "cipher")]
829            encryption_key: self.encryption_key.clone(),
830        };
831        let connection_settings = reconnect_settings;
832        #[cfg(feature = "live")]
833        let connection_hook_columns = Arc::clone(&hook_columns);
834        let connection_reopen: Arc<dyn Fn() -> Result<Connection> + Send + Sync> =
835            Arc::new(move || {
836                let conn = connection_settings.open(false)?;
837                // 사용자 callback이 같은 이름의 함수/hook을 교체할 수 있으므로
838                // roomrs connection-local 상태는 initialize 뒤 마지막에 설치한다.
839                connection_settings.initialize(&conn)?;
840                #[cfg(feature = "live")]
841                {
842                    install_preupdate_hook(&conn, Arc::clone(&connection_hook_columns))?;
843                }
844                Ok(conn)
845            });
846
847        let inner = DatabaseInner {
848            pool: Pool {
849                connections: ConnectionPool::new_with_preservation(
850                    connections,
851                    self.in_memory,
852                    self.in_memory,
853                    self.queue_timeout,
854                    connection_reopen,
855                ),
856            },
857            query_logger: self.query_logger.take(),
858            #[cfg(feature = "live")]
859            tracker,
860            #[cfg(feature = "live")]
861            hook_columns: Arc::clone(&hook_columns),
862            #[cfg(feature = "live")]
863            notifier_join: Some(notifier_join),
864        };
865        let db = Database {
866            inner: Arc::new(inner),
867        };
868
869        // 마이그레이션 — 풀 구성 후 Tx 기반으로 실행 (명세 §8)
870        self.run_migration(&db, &schema)?;
871
872        // 마이그레이션 완료 뒤 모든 풀 연결을 초기화한다.
873        if let Some(cb) = &self.on_open {
874            {
875                db.inner
876                    .pool
877                    .connections
878                    .for_each_idle(|conn| Self::apply_on_open(conn, cb, false, self.in_memory))?;
879            }
880            #[cfg(feature = "live")]
881            db.inner
882                .tracker
883                .initialize(Arc::clone(cb), self.in_memory)?;
884        }
885
886        // on_open이 roomrs hook을 교체했더라도 일반 풀의
887        // connection-local 상태가 최종 승자가 되도록 전부 재설치한다.
888        db.inner.pool.connections.for_each_idle(|_conn| {
889            #[cfg(feature = "live")]
890            {
891                install_preupdate_hook(_conn, Arc::clone(&hook_columns))?;
892            }
893            Ok::<(), Error>(())
894        })?;
895
896        // 오픈 완료 로그 — 경로(인메모리는 ":memory:")와 스키마 버전
897        log::info!(
898            "database opened: path={}, version={}",
899            self.path
900                .as_ref()
901                .map(|p| p.display().to_string())
902                .unwrap_or_else(|| ":memory:".into()),
903            schema.version
904        );
905
906        Ok(T::from_database(db))
907    }
908
909    /// 커넥션 1개 오픈 + PRAGMA 설정
910    fn open_conn(
911        &self,
912        mem_name: Option<&str>,
913        _schema: &SchemaDef,
914        _read_only: bool,
915    ) -> Result<Connection> {
916        let conn = match (mem_name, &self.path) {
917            (Some(uri), _) => {
918                use rusqlite::OpenFlags;
919                Connection::open_with_flags(
920                    uri,
921                    OpenFlags::SQLITE_OPEN_READ_WRITE
922                        | OpenFlags::SQLITE_OPEN_CREATE
923                        | OpenFlags::SQLITE_OPEN_URI
924                        | OpenFlags::SQLITE_OPEN_NO_MUTEX,
925                )?
926            }
927            (None, Some(path)) => Connection::open(path)?,
928            (None, None) => {
929                return Err(Error::Config(
930                    "DB 경로가 설정되지 않았습니다 — .sqlite(path) 또는 .in_memory() 필요".into(),
931                ));
932            }
933        };
934
935        // 암호화 키 — 어떤 접근보다 먼저 적용해야 한다 (feature cipher)
936        #[cfg(feature = "cipher")]
937        if let Some(key) = &self.encryption_key {
938            conn.pragma_update(None, "key", key)?;
939        }
940
941        // busy 핸들러를 다른 PRAGMA보다 먼저 — journal_mode 전환 등도 락 경합이
942        // 있어 동시 오픈 시 SQLITE_BUSY로 실패할 수 있다 (M-4)
943        conn.busy_timeout(self.busy_timeout)?;
944
945        // 공통 PRAGMA (명세 §10) — 인메모리는 WAL 미지원이라 파일 DB에만 적용.
946        // 신규 파일의 WAL 전환은 동시 오픈 시 busy 핸들러가 개입하지 못하는
947        // 락 경합이 있어 짧은 재시도로 흡수한다 (M-4)
948        if mem_name.is_none() {
949            let deadline =
950                std::time::Instant::now() + self.busy_timeout.max(Duration::from_millis(500));
951            loop {
952                match conn.pragma_update(None, "journal_mode", "WAL") {
953                    Ok(()) => break,
954                    Err(rusqlite::Error::SqliteFailure(fe, _))
955                        if fe.code == rusqlite::ErrorCode::DatabaseBusy
956                            && std::time::Instant::now() < deadline =>
957                    {
958                        std::thread::sleep(Duration::from_millis(10));
959                    }
960                    Err(e) => return Err(e.into()),
961                }
962            }
963        }
964        conn.pragma_update(None, "foreign_keys", "ON")?;
965        conn.pragma_update(None, "synchronous", "NORMAL")?;
966
967        conn.pragma_update(None, "query_only", "OFF")?;
968        if mem_name.is_some() {
969            conn.pragma_update(None, "read_uncommitted", "ON")?;
970        }
971        log::debug!("read/write pool connection opened");
972        Ok(conn)
973    }
974
975    /// 사용자 연결 초기화 후 트랜잭션·풀 커넥션 불변식을 복구한다.
976    fn apply_on_open(
977        conn: &Connection,
978        cb: &ConnCallback,
979        read_only: bool,
980        read_uncommitted: bool,
981    ) -> Result<()> {
982        let callback_result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| cb(conn)))
983            .map_err(|_| Error::Internal("on_open 콜백 panic".into()))
984            .and_then(|result| result);
985        if !conn.is_autocommit() {
986            conn.execute_batch("ROLLBACK")?;
987        }
988        let _ = read_only;
989        conn.pragma_update(None, "query_only", "OFF")?;
990        if read_uncommitted {
991            conn.pragma_update(None, "read_uncommitted", "ON")?;
992        }
993        callback_result
994    }
995
996    /// 마이그레이션 러너 (명세 §8) — user_version 기반.
997    /// 0(신규) = DDL 생성 + on_create · 일치 = 통과 ·
998    /// 불일치 = Auto: 스텝 체인 실행(갭이면 destructive 폴백 또는 에러), Validate: 에러.
999    /// 각 트랜잭션(BEGIN IMMEDIATE) 획득 후 user_version을 재확인해
1000    /// 교차 프로세스 동시 마이그레이션 경합을 차단한다 (M-4)
1001    fn run_migration(&self, db: &Database, schema: &SchemaDef) -> Result<()> {
1002        let h = db.run_sync();
1003        let current: u32 =
1004            h.with_connection(|c| Ok(c.query_row("PRAGMA user_version", [], |r| r.get(0))?))?;
1005        let target = schema.version;
1006
1007        // 신규 DB — 스키마 생성 (스텝 없이 최신 DDL로 직행)
1008        if current == 0 {
1009            let created = h.transaction(|tx| {
1010                // 락 확보 후 재확인 — 다른 프로세스가 먼저 생성했으면 스킵 (M-4)
1011                let cur: u32 = tx.query_scalar("PRAGMA user_version", [])?;
1012                if cur != 0 {
1013                    return Ok(false);
1014                }
1015                for ddl in &schema.ddl {
1016                    // 빌드 시점엔 구독자가 존재할 수 없다 — 무효화 수집을 생략해
1017                    // 스테일 전체 무효화가 첫 구독과 경합하지 않게 한다 (H-1 회귀 방지)
1018                    tx.raw_conn().execute_batch(ddl)?;
1019                }
1020                // on_create를 생성 트랜잭션 안에서 실행 — 실패 시 스키마와
1021                // user_version이 함께 롤백돼 다음 오픈이 생성을 재시도한다 (L-5)
1022                if let Some(cb) = &self.on_create {
1023                    cb(tx.raw_conn())?;
1024                }
1025                tx.execute_batch(&format!("PRAGMA user_version = {target}"))?;
1026                Ok(true)
1027            })?;
1028            if created {
1029                log::info!("schema created at version {target}");
1030                return Ok(());
1031            }
1032            // 다른 프로세스가 먼저 생성 — 버전 검증 경로로 재진입 (M-4)
1033            return self.run_migration(db, schema);
1034        }
1035
1036        if current == target {
1037            return Ok(());
1038        }
1039
1040        if self.migrate == MigrationPolicy::Validate {
1041            log::error!(
1042                "migration failed: schema version mismatch (db={current}, code={target}, policy=Validate)"
1043            );
1044            return Err(Error::Migration(format!(
1045                "스키마 버전 불일치: DB={current}, 코드={target} (Validate 정책)"
1046            )));
1047        }
1048
1049        // 자동 마이그레이션(옵트인, 명세 §8.4) — 등록 스텝이 없는 구간을
1050        // 내장 스냅샷 연속 쌍 diff의 안전 연산으로 합성해 메운다
1051        let synthesized = if self.auto_migrate {
1052            synthesize_embedded_steps(T::EMBEDDED_SCHEMAS, &self.migrations, current)?
1053        } else {
1054            SynthesizedSteps::default()
1055        };
1056        let all_steps: Vec<&crate::migration::Migration> = self
1057            .migrations
1058            .iter()
1059            .chain(synthesized.steps.iter())
1060            .collect();
1061
1062        // 스텝 체인 실행 — 스텝별 트랜잭션 + user_version 갱신.
1063        // 파괴적 구간 사전 검사를 먼저 — plan_chain의 일반 갭 에러보다 구체적 안내
1064        let plan_result = check_destructive_gap(&all_steps, &synthesized, current, target)
1065            .and_then(|()| crate::migration::plan_chain(&all_steps, current, target));
1066        match plan_result {
1067            Ok(plan) => {
1068                for step in plan {
1069                    log::info!(
1070                        "migration step: v{}->v{}",
1071                        step.from_version(),
1072                        step.to_version()
1073                    );
1074                    h.transaction(|tx| {
1075                        // 락 확보 후 재확인 — 다른 프로세스가 이미 적용했으면 스킵 (M-4)
1076                        let cur: u32 = tx.query_scalar("PRAGMA user_version", [])?;
1077                        if cur >= step.to_version() {
1078                            return Ok(());
1079                        }
1080                        // 체인 구성이 다른 프로세스의 개입 감지 — 스텝 시작 버전이
1081                        // 실제 버전과 다르면 잘못된 SQL 적용을 차단한다 (M-5)
1082                        if cur != step.from_version() {
1083                            return Err(Error::Migration(format!(
1084                                "동시 마이그레이션 감지: 예상 v{}, 실제 v{cur} — 체인 구성 상이",
1085                                step.from_version()
1086                            )));
1087                        }
1088                        step.run_up(tx)?;
1089                        tx.execute_batch(&format!("PRAGMA user_version = {}", step.to_version()))
1090                    })?;
1091                }
1092                Ok(())
1093            }
1094            Err(e) if self.destructive_fallback => {
1095                // 파괴적 폴백 (옵트인, 명세 §8) — 전부 삭제 후 최신 스키마로 재생성
1096                log::warn!("migration chain insufficient — falling back to destructive migration");
1097                let _ = e;
1098                self.run_destructive(&h, schema)
1099            }
1100            Err(e) => {
1101                log::error!("migration failed: {e}");
1102                Err(e)
1103            }
1104        }
1105    }
1106
1107    /// 파괴적 재생성 — 사용자 객체 전부 drop 후 DDL 재실행
1108    fn run_destructive(&self, h: &SyncHandle<'_>, schema: &SchemaDef) -> Result<()> {
1109        // FK 토글과 DDL은 같은 커넥션에서 실행해야 한다.
1110        h.with_connection(|c| {
1111            c.pragma_update(None, "foreign_keys", "OFF")?;
1112            let result: Result<()> = (|| {
1113                c.execute_batch("BEGIN IMMEDIATE")?;
1114                let migration: Result<()> = (|| {
1115                    // 사용자 객체 수집 (sqlite_* 내부 객체 제외)
1116                    let mut statement = c.prepare(
1117                        "SELECT type, name FROM sqlite_master \
1118                         WHERE name NOT LIKE 'sqlite_%' \
1119                         AND type IN ('trigger','view','index','table')",
1120                    )?;
1121                    let objs = statement
1122                        .query_map([], |row| Ok((row.get(0)?, row.get(1)?)))?
1123                        .collect::<std::result::Result<Vec<(String, String)>, _>>()?;
1124                    drop(statement);
1125                    // 의존 역순: trigger → view → index → table.
1126                    // sqlite_master의 이름은 임의 문자열 — 식별자 이스케이프 필수 (M-9).
1127                    for kind in ["trigger", "view", "index", "table"] {
1128                        for (t, name) in objs.iter().filter(|(t, _)| t == kind) {
1129                            c.execute_batch(&format!(
1130                                "DROP {} {}",
1131                                t.to_uppercase(),
1132                                escape_ident(name)
1133                            ))?;
1134                        }
1135                    }
1136                    for ddl in &schema.ddl {
1137                        c.execute_batch(ddl)?;
1138                    }
1139                    c.execute_batch(&format!("PRAGMA user_version = {}", schema.version))?;
1140                    Ok(())
1141                })();
1142                match migration {
1143                    Ok(()) => c.execute_batch("COMMIT")?,
1144                    Err(error) => {
1145                        let _ = c.execute_batch("ROLLBACK");
1146                        return Err(error);
1147                    }
1148                }
1149                Ok(())
1150            })();
1151            let restore = c.pragma_update(None, "foreign_keys", "ON");
1152            result.and(restore.map_err(Into::into))
1153        })
1154    }
1155}
1156
1157/// 자동 마이그레이션 합성 결과 — 합성 스텝 + 파괴적으로 거부된 구간 기록
1158#[derive(Default)]
1159struct SynthesizedSteps {
1160    /// 안전 연산만으로 합성된 스텝들
1161    steps: Vec<crate::migration::Migration>,
1162    /// from 버전 → (to 버전, 파괴적 항목 요약) — 합성 거부 구간
1163    refused: std::collections::HashMap<u32, (u32, String)>,
1164}
1165
1166/// 내장 스냅샷의 인접 가용 버전 쌍을 diff해 갭 메움 스텝을 합성한다 (명세 §8.4).
1167/// 등록 스텝이 있는 from 버전은 건너뛴다(등록 스텝 우선). 파괴적 변경이 포함된
1168/// 쌍은 합성하지 않고 기록만 남긴다 — 체인이 그 갭에 닿으면 명확한 에러.
1169/// `current` 미만에서 출발하는 쌍은 계획에 쓰일 수 없으므로 건너뛴다 —
1170/// 무관한 옛 스냅샷의 압축 해제(파손 시 실패 포함)를 피한다 (L-1)
1171fn synthesize_embedded_steps(
1172    embedded: &[EmbeddedSchema],
1173    registered: &[crate::migration::Migration],
1174    current: u32,
1175) -> Result<SynthesizedSteps> {
1176    let mut out = SynthesizedSteps::default();
1177    if embedded.len() < 2 {
1178        return Ok(out);
1179    }
1180    // 방어적 정렬 — 매크로는 오름차순 방출하지만 수동 impl 대비
1181    let mut sorted: Vec<&EmbeddedSchema> = embedded.iter().collect();
1182    sorted.sort_by_key(|e| e.version);
1183
1184    for pair in sorted.windows(2) {
1185        let (a, b) = (pair[0], pair[1]);
1186        if a.version == b.version {
1187            continue;
1188        }
1189        // 현재 DB 버전 미만 구간 — 사용 불가, 옛 스냅샷 접근 생략 (L-1)
1190        if a.version < current {
1191            continue;
1192        }
1193        // 등록 스텝 우선 — 같은 from에서 출발하는 스텝이 있으면 합성 생략
1194        if registered.iter().any(|m| m.from_version() == a.version) {
1195            continue;
1196        }
1197        let old = a.snapshot()?;
1198        let new = b.snapshot()?;
1199        let plan = roomrs_migrate::diff_plan(&old, &new);
1200        if plan.destructive.is_empty() {
1201            log::info!(
1202                "auto-migrate synthesized step: v{}->v{}",
1203                a.version,
1204                b.version
1205            );
1206            out.steps.push(crate::migration::Migration::sql(
1207                a.version,
1208                b.version,
1209                plan.safe.join(";\n"),
1210            ));
1211        } else {
1212            out.refused
1213                .insert(a.version, (b.version, plan.destructive.join("; ")));
1214        }
1215    }
1216    Ok(out)
1217}
1218
1219/// 체인을 사전 답사해 파괴적 합성 거부 구간에 닿는지 검사 — 닿으면 실행 전에
1220/// 구체적 에러를 반환한다(일반 갭·형식 오류는 plan_chain이 보고).
1221fn check_destructive_gap(
1222    steps: &[&crate::migration::Migration],
1223    synthesized: &SynthesizedSteps,
1224    current: u32,
1225    target: u32,
1226) -> Result<()> {
1227    if synthesized.refused.is_empty() {
1228        return Ok(());
1229    }
1230    let mut v = current;
1231    while v < target {
1232        match steps.iter().find(|s| s.from_version() == v) {
1233            Some(s) if s.to_version() > v && s.to_version() <= target => v = s.to_version(),
1234            // 역행/오버슈트 스텝 = plan_chain이 보고
1235            Some(_) => return Ok(()),
1236            None => {
1237                if let Some((to, items)) = synthesized.refused.get(&v) {
1238                    return Err(Error::Migration(format!(
1239                        "v{v}->v{to} 자동 마이그레이션 불가 — 파괴적 변경 포함: {items}; \
1240                         수동 스텝을 등록하거나 fallback_to_destructive_migration 사용"
1241                    )));
1242                }
1243                // 일반 갭 = plan_chain이 보고
1244                return Ok(());
1245            }
1246        }
1247    }
1248    Ok(())
1249}
1250
1251#[cfg(test)]
1252mod tests {
1253    use super::*;
1254    use roomrs_migrate::{ColumnSnapshot, SchemaSnapshot, TableSnapshot};
1255
1256    /// 인메모리 DB는 builder 호출 순서와 무관하게 일반 커넥션 하나만 만든다.
1257    #[test]
1258    fn in_memory_uses_one_regular_connection() {
1259        for builder in [
1260            DatabaseBuilder::<DestructiveFkDb>::default()
1261                .in_memory()
1262                .connections(3),
1263            DatabaseBuilder::<DestructiveFkDb>::default()
1264                .connections(3)
1265                .in_memory(),
1266            DatabaseBuilder::<DestructiveFkDb>::default()
1267                .connections(3)
1268                .sqlite(":memory:"),
1269        ] {
1270            let db = builder.build().unwrap().0;
1271            let mut idle = 0;
1272            db.inner
1273                .pool
1274                .connections
1275                .for_each_idle(|_| {
1276                    idle += 1;
1277                    Ok(())
1278                })
1279                .unwrap();
1280            assert_eq!(idle, 1);
1281        }
1282    }
1283
1284    /// 인메모리 동시 트랜잭션은 단일 일반 커넥션에서 직렬 실행된다.
1285    #[test]
1286    fn in_memory_serializes_concurrent_transactions() {
1287        let db = Arc::new(
1288            DatabaseBuilder::<DestructiveFkDb>::default()
1289                .in_memory()
1290                .connections(4)
1291                .build()
1292                .unwrap()
1293                .0,
1294        );
1295        let mut workers = Vec::new();
1296        for worker in 0..4 {
1297            let db = Arc::clone(&db);
1298            workers.push(std::thread::spawn(move || {
1299                for item in 0..25 {
1300                    db.run_sync().transaction(|tx| {
1301                        tx.execute(
1302                            "INSERT INTO parents(id) VALUES (?1)",
1303                            [worker * 25 + item + 1],
1304                        )?;
1305                        Ok(())
1306                    })?;
1307                }
1308                Result::<()>::Ok(())
1309            }));
1310        }
1311        for worker in workers {
1312            worker.join().unwrap().unwrap();
1313        }
1314        let count: i64 = db
1315            .run_sync()
1316            .query_scalar("SELECT COUNT(*) FROM parents", [])
1317            .unwrap();
1318        assert_eq!(count, 100);
1319    }
1320
1321    struct DestructiveFkDb(Database);
1322
1323    impl DatabaseSpec for DestructiveFkDb {
1324        const VERSION: u32 = 1;
1325        const DB_NAME: &'static str = "destructive_fk_db";
1326
1327        fn schema() -> SchemaDef {
1328            SchemaDef {
1329                version: 1,
1330                ddl: vec![
1331                    "CREATE TABLE parents(id INTEGER PRIMARY KEY)",
1332                    "CREATE TABLE children(id INTEGER PRIMARY KEY, parent_id INTEGER NOT NULL REFERENCES parents(id))",
1333                ],
1334                tables: Vec::new(),
1335            }
1336        }
1337
1338        fn from_database(db: Database) -> Self {
1339            Self(db)
1340        }
1341    }
1342
1343    /// 파괴적 재생성은 FK 참조 데이터가 있어도 성공하고 모든 풀 커넥션의 FK를 복구한다.
1344    #[test]
1345    fn destructive_migration_uses_one_connection_and_restores_foreign_keys() {
1346        static FILE_SEQUENCE: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
1347        let file = std::env::temp_dir().join(format!(
1348            "roomrs-destructive-fk-{}-{}.db",
1349            std::process::id(),
1350            FILE_SEQUENCE.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
1351        ));
1352        let db = DatabaseBuilder::<DestructiveFkDb>::default()
1353            .sqlite(&file)
1354            .connections(3)
1355            .build()
1356            .unwrap()
1357            .0;
1358        db.run_sync()
1359            .execute("INSERT INTO parents(id) VALUES (1)", [])
1360            .unwrap();
1361        db.run_sync()
1362            .execute("INSERT INTO children(id, parent_id) VALUES (1, 1)", [])
1363            .unwrap();
1364
1365        let target = SchemaDef {
1366            version: 2,
1367            ddl: vec!["CREATE TABLE replacements(id INTEGER PRIMARY KEY)"],
1368            tables: Vec::new(),
1369        };
1370        DatabaseBuilder::<DestructiveFkDb>::default()
1371            .run_destructive(&db.run_sync(), &target)
1372            .unwrap();
1373
1374        db.inner
1375            .pool
1376            .connections
1377            .for_each_idle(|conn| {
1378                let enabled: i64 = conn.query_row("PRAGMA foreign_keys", [], |row| row.get(0))?;
1379                assert_eq!(enabled, 1);
1380                Ok(())
1381            })
1382            .unwrap();
1383
1384        let invalid_target = SchemaDef {
1385            version: 3,
1386            ddl: vec!["CREATE TABLE invalid("],
1387            tables: Vec::new(),
1388        };
1389        assert!(
1390            DatabaseBuilder::<DestructiveFkDb>::default()
1391                .run_destructive(&db.run_sync(), &invalid_target)
1392                .is_err()
1393        );
1394        db.inner
1395            .pool
1396            .connections
1397            .for_each_idle(|conn| {
1398                let enabled: i64 = conn.query_row("PRAGMA foreign_keys", [], |row| row.get(0))?;
1399                assert_eq!(enabled, 1);
1400                Ok(())
1401            })
1402            .unwrap();
1403        drop(db);
1404        std::fs::remove_file(file).unwrap();
1405    }
1406
1407    /// 다른 Rust path가 같은 SQLite TABLE을 가리키면 schema 단계에서 거부한다.
1408    #[test]
1409    fn duplicate_entity_table_names_are_rejected() {
1410        let schema = SchemaDef {
1411            version: 1,
1412            ddl: Vec::new(),
1413            tables: vec![
1414                TableMeta {
1415                    name: "items",
1416                    columns: &[],
1417                    ddl: &[],
1418                },
1419                TableMeta {
1420                    name: "ITEMS",
1421                    columns: &[],
1422                    ddl: &[],
1423                },
1424            ],
1425        };
1426        assert!(matches!(
1427            schema.validate_unique_tables(),
1428            Err(Error::Config(_))
1429        ));
1430    }
1431
1432    /// 단일 테이블 스냅샷 생성 헬퍼
1433    fn snap(version: u32, cols: Vec<(&str, &str, bool)>) -> SchemaSnapshot {
1434        SchemaSnapshot {
1435            version,
1436            tables: vec![TableSnapshot {
1437                name: "t".into(),
1438                columns: cols
1439                    .into_iter()
1440                    .map(|(name, ty, not_null)| ColumnSnapshot {
1441                        name: name.into(),
1442                        sql_type: ty.into(),
1443                        not_null,
1444                        pk: name == "id",
1445                        renamed_from: None,
1446                    })
1447                    .collect(),
1448                ddl: vec![],
1449            }],
1450        }
1451    }
1452
1453    /// 런타임 스냅샷 → 내장 스냅샷 (테스트 전용 leak)
1454    fn embed(snap: &SchemaSnapshot) -> EmbeddedSchema {
1455        let comp = roomrs_migrate::compress_snapshot(snap.to_json().unwrap().as_bytes());
1456        EmbeddedSchema {
1457            version: snap.version,
1458            compressed: Box::leak(comp.into_boxed_slice()),
1459        }
1460    }
1461
1462    /// 안전 diff 쌍 = 스텝 합성, 구멍([1,3]) 건너 diff
1463    #[test]
1464    fn synthesize_spans_version_holes() {
1465        let v1 = snap(1, vec![("id", "INTEGER", true)]);
1466        let v3 = snap(3, vec![("id", "INTEGER", true), ("a", "TEXT", false)]);
1467        let v4 = snap(
1468            4,
1469            vec![
1470                ("id", "INTEGER", true),
1471                ("a", "TEXT", false),
1472                ("b", "TEXT", false),
1473            ],
1474        );
1475        let embedded = [embed(&v1), embed(&v3), embed(&v4)];
1476        let s = synthesize_embedded_steps(&embedded, &[], 1).unwrap();
1477        assert!(s.refused.is_empty());
1478        let spans: Vec<(u32, u32)> = s
1479            .steps
1480            .iter()
1481            .map(|m| (m.from_version(), m.to_version()))
1482            .collect();
1483        assert_eq!(spans, vec![(1, 3), (3, 4)], "인접 가용 쌍으로 합성");
1484    }
1485
1486    /// 등록 스텝이 있는 from 구간은 합성하지 않는다 (등록 스텝 우선)
1487    #[test]
1488    fn synthesize_skips_registered_from() {
1489        let v1 = snap(1, vec![("id", "INTEGER", true)]);
1490        let v2 = snap(2, vec![("id", "INTEGER", true), ("a", "TEXT", false)]);
1491        let registered = [crate::migration::Migration::sql(1, 2, "SELECT 1")];
1492        let s = synthesize_embedded_steps(&[embed(&v1), embed(&v2)], &registered, 1).unwrap();
1493        assert!(s.steps.is_empty(), "등록 스텝 우선 — 합성 없음");
1494        assert!(s.refused.is_empty());
1495    }
1496
1497    /// 파괴적 diff 쌍 = 합성 거부 + 체인 사전 검사에서 구체적 에러
1498    #[test]
1499    fn synthesize_refuses_destructive_pair() {
1500        let v1 = snap(1, vec![("id", "INTEGER", true), ("c", "TEXT", true)]);
1501        let v2 = snap(2, vec![("id", "INTEGER", true), ("c", "INTEGER", true)]);
1502        let s = synthesize_embedded_steps(&[embed(&v1), embed(&v2)], &[], 1).unwrap();
1503        assert!(s.steps.is_empty());
1504        assert!(s.refused.contains_key(&1), "{:?}", s.refused);
1505
1506        // 체인이 갭에 닿으면 파괴적 안내 에러
1507        match check_destructive_gap(&[], &s, 1, 2) {
1508            Err(Error::Migration(msg)) => {
1509                assert!(msg.contains("v1->v2 자동 마이그레이션 불가"), "{msg}");
1510                assert!(msg.contains("파괴적 변경 포함"), "{msg}");
1511                assert!(msg.contains("fallback_to_destructive_migration"), "{msg}");
1512            }
1513            other => panic!("Migration 에러 기대, 결과: {other:?}"),
1514        }
1515
1516        // 등록 스텝이 그 구간을 이으면 통과 (plan_chain으로 위임)
1517        let manual = crate::migration::Migration::sql(1, 2, "SELECT 1");
1518        assert!(check_destructive_gap(&[&manual], &s, 1, 2).is_ok());
1519    }
1520
1521    /// 내장 스냅샷 파손 = Migration 에러 (한국어 메시지)
1522    #[test]
1523    fn synthesize_corrupt_embedded_errors() {
1524        let v1 = snap(1, vec![("id", "INTEGER", true)]);
1525        let bad = EmbeddedSchema {
1526            version: 2,
1527            compressed: b"\xff\x00\x12corrupt",
1528        };
1529        match synthesize_embedded_steps(&[embed(&v1), bad], &[], 1) {
1530            Err(Error::Migration(msg)) => assert!(msg.contains("내장 스냅샷"), "{msg}"),
1531            Err(other) => panic!("Migration 에러 기대, 결과: {other}"),
1532            Ok(_) => panic!("파손 스냅샷이 통과되면 안 된다"),
1533        }
1534    }
1535
1536    /// 현재 버전 미만 구간은 건너뛴다 — 옛 스냅샷이 파손돼도 접근하지 않는다 (L-1)
1537    #[test]
1538    fn synthesize_skips_pairs_below_current() {
1539        let bad_old = EmbeddedSchema {
1540            version: 1,
1541            compressed: b"\xff\x00\x12corrupt",
1542        };
1543        let v2 = snap(2, vec![("id", "INTEGER", true)]);
1544        let v3 = snap(3, vec![("id", "INTEGER", true), ("a", "TEXT", false)]);
1545        let s = synthesize_embedded_steps(&[bad_old, embed(&v2), embed(&v3)], &[], 2).unwrap();
1546        let spans: Vec<(u32, u32)> = s
1547            .steps
1548            .iter()
1549            .map(|m| (m.from_version(), m.to_version()))
1550            .collect();
1551        assert_eq!(spans, vec![(2, 3)], "v1 파손 스냅샷은 압축 해제하지 않음");
1552        assert!(s.refused.is_empty());
1553    }
1554}