1use 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
11pub(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 static HOOK_CAPTURES: std::cell::RefCell<Vec<HookCaptureFrame>> =
19 const { std::cell::RefCell::new(Vec::new()) };
20}
21
22#[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#[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#[cfg(feature = "live")]
43pub(crate) struct HookCapture {
44 depth: usize,
45}
46
47#[cfg(feature = "live")]
48impl Drop for HookCapture {
49 fn drop(&mut self) {
51 HOOK_CAPTURES.with(|captures| captures.borrow_mut().truncate(self.depth));
52 }
53}
54
55#[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#[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 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 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
177pub(crate) fn escape_ident(name: &str) -> String {
179 format!("\"{}\"", name.replace('"', "\"\""))
180}
181
182#[derive(Debug, Clone, Copy)]
184pub struct ColumnMeta {
185 pub name: &'static str,
186 pub sql_type: &'static str,
188 pub not_null: bool,
189 pub pk: bool,
190 pub renamed_from: Option<&'static str>,
192}
193
194pub struct TableMeta {
196 pub name: &'static str,
197 pub columns: &'static [ColumnMeta],
198 pub ddl: &'static [&'static str],
200}
201
202pub struct SchemaDef {
204 pub version: u32,
206 pub ddl: Vec<&'static str>,
208 pub tables: Vec<TableMeta>,
210}
211
212impl SchemaDef {
213 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 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#[derive(Debug, Clone, Copy)]
263pub struct EmbeddedSchema {
264 pub version: u32,
266 pub compressed: &'static [u8],
269}
270
271impl EmbeddedSchema {
272 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
287pub trait DatabaseSpec: Sized {
290 const VERSION: u32;
292 const DB_NAME: &'static str;
296 const SNAPSHOT_HASH: Option<u64> = None;
298 const EMBEDDED_SCHEMAS: &'static [EmbeddedSchema] = &[];
302 const SNAPSHOT_FILE_SEEN: bool = true;
310 fn schema() -> SchemaDef;
312 fn from_database(db: Database) -> Self;
314}
315
316pub 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
330pub 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
348pub fn export_schema_for_test<T: DatabaseSpec>(manifest_dir: &str) -> Result<()> {
367 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 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 return Err(Error::SnapshotStale(format!(
387 "스냅샷을 생성했습니다 — 커밋 후 `cargo clean -p <크레이트>` 또는 소스 touch 후 재빌드하세요: {}",
388 path.display()
389 )));
390 }
391 match roomrs_migrate::SchemaSnapshot::read_from(&path) {
392 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 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 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#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
425pub enum MigrationPolicy {
426 #[default]
428 Auto,
429 Validate,
431}
432
433pub struct DatabaseInner {
435 pub(crate) pool: Pool,
436 query_logger: Option<QueryLogger>,
437 #[cfg(feature = "live")]
439 pub(crate) tracker: Arc<crate::live::Tracker>,
440 #[cfg(feature = "live")]
442 pub(crate) hook_columns: Arc<std::collections::HashMap<String, Vec<String>>>,
443 #[cfg(feature = "live")]
445 notifier_join: Option<std::thread::JoinHandle<()>>,
446}
447
448#[cfg(not(feature = "live"))]
450impl Drop for DatabaseInner {
451 fn drop(&mut self) {
453 log::info!("database closed");
454 }
455}
456
457#[cfg(feature = "live")]
458impl Drop for DatabaseInner {
459 fn drop(&mut self) {
463 log::info!("database closing — shutting down notifier");
464 self.tracker.shutdown();
466 if let Some(h) = self.notifier_join.take() {
467 if h.thread().id() == std::thread::current().id() {
468 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 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 pub(crate) fn take_hook_capture(&self) -> HookCaptureFrame {
495 HOOK_CAPTURES.with(|captures| captures.borrow_mut().pop().unwrap_or_default())
496 }
497
498 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 crate::live::WriteTables::Unknown => self.tracker.invalidate(None),
532 }
533 }
534}
535
536impl DatabaseInner {
537 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
553pub struct Database {
555 inner: Arc<DatabaseInner>,
556}
557
558impl Database {
559 pub fn run_sync(&self) -> SyncHandle<'_> {
561 SyncHandle { inner: &self.inner }
562 }
563
564 #[doc(hidden)]
566 pub fn inner_arc(&self) -> Arc<DatabaseInner> {
567 Arc::clone(&self.inner)
568 }
569}
570
571impl DatabaseInner {
572 #[doc(hidden)]
574 pub fn sync_handle(self: &Arc<Self>) -> SyncHandle<'_> {
575 SyncHandle { inner: self }
576 }
577}
578
579pub 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 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 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 pub fn in_memory(mut self) -> Self {
646 self.in_memory = true;
647 self.path = None;
648 self
649 }
650
651 pub fn connections(mut self, n: usize) -> Self {
655 self.connections = n.max(1);
656 self
657 }
658
659 pub fn busy_timeout(mut self, d: Duration) -> Self {
661 self.busy_timeout = d;
662 self
663 }
664
665 pub fn queue_timeout(mut self, d: Duration) -> Self {
667 self.queue_timeout = Some(d);
668 self
669 }
670
671 #[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 pub fn migrate(mut self, policy: MigrationPolicy) -> Self {
680 self.migrate = policy;
681 self
682 }
683
684 pub fn migration(mut self, m: crate::migration::Migration) -> Self {
686 self.migrations.push(m);
687 self
688 }
689
690 pub fn migrations(mut self, ms: impl IntoIterator<Item = crate::migration::Migration>) -> Self {
692 self.migrations.extend(ms);
693 self
694 }
695
696 pub fn auto_migrate(mut self, on: bool) -> Self {
709 self.auto_migrate = on;
710 self
711 }
712
713 pub fn fallback_to_destructive_migration(mut self, enable: bool) -> Self {
716 self.destructive_fallback = enable;
717 self
718 }
719
720 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 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 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 pub fn build(mut self) -> Result<T> {
751 let schema = T::schema();
752 schema.validate_unique_tables()?;
753
754 if self.in_memory {
757 self.connections = 1;
758 }
759
760 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 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 let first_conn = self.open_conn(mem_name.as_deref(), &schema, false)?;
786
787 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 #[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 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 self.run_migration(&db, &schema)?;
871
872 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 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 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 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 #[cfg(feature = "cipher")]
937 if let Some(key) = &self.encryption_key {
938 conn.pragma_update(None, "key", key)?;
939 }
940
941 conn.busy_timeout(self.busy_timeout)?;
944
945 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 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 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 if current == 0 {
1009 let created = h.transaction(|tx| {
1010 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 tx.raw_conn().execute_batch(ddl)?;
1019 }
1020 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 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 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 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 let cur: u32 = tx.query_scalar("PRAGMA user_version", [])?;
1077 if cur >= step.to_version() {
1078 return Ok(());
1079 }
1080 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 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 fn run_destructive(&self, h: &SyncHandle<'_>, schema: &SchemaDef) -> Result<()> {
1109 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 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 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#[derive(Default)]
1159struct SynthesizedSteps {
1160 steps: Vec<crate::migration::Migration>,
1162 refused: std::collections::HashMap<u32, (u32, String)>,
1164}
1165
1166fn 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 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 if a.version < current {
1191 continue;
1192 }
1193 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
1219fn 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 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 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 #[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 #[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 #[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 #[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 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 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 #[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 #[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)], ®istered, 1).unwrap();
1493 assert!(s.steps.is_empty(), "등록 스텝 우선 — 합성 없음");
1494 assert!(s.refused.is_empty());
1495 }
1496
1497 #[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 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 let manual = crate::migration::Migration::sql(1, 2, "SELECT 1");
1518 assert!(check_destructive_gap(&[&manual], &s, 1, 2).is_ok());
1519 }
1520
1521 #[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 #[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}