1use std::{
2 cell::RefCell, convert::Infallible, marker::PhantomData, rc::Rc, sync::atomic::AtomicI64,
3};
4
5use rusqlite::ErrorCode;
6use self_cell::{MutBorrow, self_cell};
7
8use crate::{
9 IntoExpr, IntoSelect, Table, TableRow,
10 error::FromConflict,
11 lower::{
12 self, JoinableTableWithId, emit,
13 list_writer::{Alias, ListWriter},
14 ord_rc::OrdRc,
15 },
16 migrate::{Renderable, Schema, check_schema, schema_version, user_version},
17 migration::Config,
18 pool::Pool,
19 private::{IntoJoinable, Reader},
20 query::{OwnedRows, Query, track_stmt},
21 rows::Rows,
22 schema::read::Pragma,
23 scoped_transaction::TransactionScope,
24 value::{DbTyp, OptTable},
25};
26
27pub struct Database<S> {
40 pub(crate) pool: Pool,
41 pub(crate) schema_version: AtomicI64,
42 pub(crate) schema: PhantomData<S>,
43 pub(crate) mut_lock: parking_lot::FairMutex<()>,
44}
45
46impl<S: Schema> Database<S> {
47 pub fn new(config: Config) -> Self {
52 let Some(m) = Self::migrator(config) else {
53 panic!("schema version {}, but got an older version", S::VERSION)
54 };
55 let Some(m) = m.finish() else {
56 panic!("schema version {}, but got a new version", S::VERSION)
57 };
58 m
59 }
60}
61
62use rusqlite::Connection;
63type RTransaction<'x> = Option<rusqlite::Transaction<'x>>;
64
65self_cell!(
66 pub struct OwnedTransaction {
67 owner: MutBorrow<Connection>,
68
69 #[covariant]
70 dependent: RTransaction,
71 }
72);
73
74unsafe impl Send for OwnedTransaction {}
80assert_not_impl_any! {OwnedTransaction: Sync}
81
82thread_local! {
83 pub(crate) static TXN: RefCell<Option<TransactionWithRows>> = const { RefCell::new(None) };
84}
85
86impl OwnedTransaction {
87 pub(crate) fn get(&self) -> &rusqlite::Transaction<'_> {
88 self.borrow_dependent().as_ref().unwrap()
89 }
90
91 pub(crate) fn with(
92 mut self,
93 f: impl FnOnce(rusqlite::Transaction<'_>),
94 ) -> rusqlite::Connection {
95 self.with_dependent_mut(|_, b| f(b.take().unwrap()));
96 self.into_owner().into_inner()
97 }
98}
99
100type OwnedRowsVec<'x> = slab::Slab<OwnedRows<'x>>;
101self_cell!(
102 pub struct TransactionWithRows {
103 owner: OwnedTransaction,
104
105 #[not_covariant]
106 dependent: OwnedRowsVec,
107 }
108);
109
110impl TransactionWithRows {
111 pub(crate) fn new_empty(txn: OwnedTransaction) -> Self {
112 Self::new(txn, |_| slab::Slab::new())
113 }
114
115 pub(crate) fn get(&self) -> &rusqlite::Transaction<'_> {
116 self.borrow_owner().get()
117 }
118}
119
120impl<S: Send + Sync + Schema> Database<S> {
121 #[doc = include_str!("database/transaction.md")]
122 pub fn transaction<R: Send>(&self, f: impl Send + FnOnce(&'static Transaction<S>) -> R) -> R {
123 let res = std::thread::scope(|scope| scope.spawn(|| self.transaction_local(f)).join());
124 match res {
125 Ok(val) => val.unwrap_or_else(|e| e.to_panic()),
126 Err(payload) => std::panic::resume_unwind(payload),
127 }
128 }
129
130 pub(crate) fn transaction_local<R>(
132 &self,
133 f: impl FnOnce(&'static Transaction<S>) -> R,
134 ) -> Result<R, Renderable> {
135 let conn = self.pool.pop();
136
137 let owned = OwnedTransaction::new(MutBorrow::new(conn), |conn| {
138 Some(conn.borrow_mut().transaction().unwrap())
139 });
140
141 let res = f(Transaction::new_checked(owned, &self.schema_version)?);
142
143 let owned = TXN.take().unwrap().into_owner();
144 self.pool.push(owned.into_owner().into_inner());
145
146 Ok(res)
147 }
148
149 #[doc = include_str!("database/transaction_mut.md")]
150 pub fn transaction_mut<O: Send, E: Send>(
151 &self,
152 f: impl Send + FnOnce(&'static mut Transaction<S>) -> Result<O, E>,
153 ) -> Result<O, E> {
154 let join_res =
155 std::thread::scope(|scope| scope.spawn(|| self.transaction_mut_local(f)).join());
156
157 match join_res {
158 Ok(val) => val.unwrap_or_else(|e| e.to_panic()),
159 Err(payload) => std::panic::resume_unwind(payload),
160 }
161 }
162
163 pub(crate) fn transaction_mut_local<O, E>(
164 &self,
165 f: impl FnOnce(&'static mut Transaction<S>) -> Result<O, E>,
166 ) -> Result<Result<O, E>, Renderable> {
167 let guard = self.mut_lock.lock();
171
172 let conn = self.pool.pop();
173
174 let owned = OwnedTransaction::new(MutBorrow::new(conn), |conn| {
175 let txn = conn
176 .borrow_mut()
177 .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)
178 .unwrap();
179 Some(txn)
180 });
181
182 let full_transaction = Transaction::new_checked(owned, &self.schema_version)?;
183 let res = f(full_transaction);
185
186 drop(guard);
189
190 let owned = TXN.take().unwrap().into_owner();
191
192 let conn = if res.is_ok() {
193 owned.with(|x| x.commit().unwrap())
194 } else {
195 owned.with(|x| x.rollback().unwrap())
196 };
197 self.pool.push(conn);
198
199 Ok(res)
200 }
201
202 #[doc = include_str!("database/transaction_mut_ok.md")]
203 pub fn transaction_mut_ok<R: Send>(
204 &self,
205 f: impl Send + FnOnce(&'static mut Transaction<S>) -> R,
206 ) -> R {
207 self.transaction_mut(|txn| Ok::<R, Infallible>(f(txn)))
208 .unwrap()
209 }
210
211 pub fn rusqlite_connection(&self) -> rusqlite::Connection {
224 let conn = self.pool.pop();
225 conn.pragma_update(None, "foreign_keys", "ON").unwrap();
226 conn
227 }
228}
229
230pub struct Transaction<S> {
236 pub(crate) _p2: PhantomData<S>,
237 pub(crate) _local: PhantomData<*const ()>,
238}
239
240impl<S> Transaction<S> {
241 pub(crate) fn new_ref() -> &'static mut Self {
242 const { assert!(size_of::<Self>() == 0) };
244 Box::leak(Box::new(Self {
245 _p2: PhantomData,
246 _local: PhantomData,
247 }))
248 }
249
250 pub(crate) fn pragma(&self) -> &'static Transaction<Pragma> {
251 Transaction::new_ref()
252 }
253
254 pub(crate) fn copy(&self) -> Self {
255 Self {
256 _p2: PhantomData,
257 _local: PhantomData,
258 }
259 }
260}
261
262impl<S: Schema> Transaction<S> {
263 pub(crate) fn new_checked(
265 txn: OwnedTransaction,
266 expected: &AtomicI64,
267 ) -> Result<&'static mut Transaction<S>, Renderable> {
268 let schema_version = schema_version(txn.get());
269 if schema_version != expected.load(std::sync::atomic::Ordering::Relaxed) {
272 if user_version(txn.get()).unwrap() != S::VERSION {
273 panic!("The database user_version changed unexpectedly")
274 }
275
276 TXN.set(Some(TransactionWithRows::new_empty(txn)));
277 check_schema::<S>(Self::new_ref())?;
278 expected.store(schema_version, std::sync::atomic::Ordering::Relaxed);
279 } else {
280 TXN.set(Some(TransactionWithRows::new_empty(txn)));
281 }
282
283 Ok(Transaction::new_ref())
284 }
285}
286
287impl<S: 'static> Transaction<S> {
288 pub fn scope<O>(&mut self, f: impl FnOnce(&mut TransactionScope<S>) -> O) -> O {
290 let mut txn = TransactionScope {
291 _p2: PhantomData,
292 tmp: Default::default(),
293 };
294 f(&mut txn)
295 }
296
297 pub fn query<'t, R>(&'t self, f: impl FnOnce(&mut Query<'t, '_, S>) -> R) -> R {
310 let q = Rows {
314 phantom: PhantomData,
315 ast: Default::default(),
316 _p: PhantomData,
317 };
318 f(&mut Query {
319 q,
320 phantom: PhantomData,
321 })
322 }
323
324 pub fn query_one<O: 'static>(&self, val: impl IntoSelect<'static, S, Out = O>) -> O {
337 let mut query = self.query(|e| e.into_iter(val.into_select()));
338 let res = query.next().unwrap();
339 debug_assert!(query.next().is_none(), "query should return one row");
340 res
341 }
342
343 pub fn lazy<'t, T: OptTable<Schema = S>>(
379 &'t self,
380 val: impl IntoExpr<'static, S, Typ = T>,
381 ) -> T::Lazy<'t> {
382 T::out_to_lazy(self.query_one(val.into_expr()))
383 }
384
385 pub fn lazy_iter<'t, T: Table<Schema = S>>(
390 &'t self,
391 val: impl IntoJoinable<'static, S, Typ = TableRow<T>>,
392 ) -> LazyIter<'t, T> {
393 let val = val.into_joinable();
394 self.query(|rows| {
395 let table = rows.join(val);
396 LazyIter {
397 txn: self,
398 iter: rows.into_iter(table),
399 }
400 })
401 }
402}
403
404pub struct LazyIter<'t, T: Table> {
405 txn: &'t Transaction<T::Schema>,
406 iter: crate::query::Iter<'t, TableRow<T>>,
407}
408
409impl<'t, T: Table> Iterator for LazyIter<'t, T> {
410 type Item = crate::Lazy<'t, T>;
411
412 fn next(&mut self) -> Option<Self::Item> {
413 self.iter.next().map(|x| self.txn.lazy(x))
414 }
415}
416
417impl<S: 'static> Transaction<S> {
418 pub fn insert<T: Table<Schema = S>>(&mut self, val: T) -> Result<TableRow<T>, T::Conflict> {
440 try_insert_private(lower::JoinableTable::Table(T::NAME, None), None, val)
441 }
442
443 pub fn insert_ok<T: Table<Schema = S, Conflict = Infallible>>(
448 &mut self,
449 val: T,
450 ) -> TableRow<T> {
451 let Ok(row) = self.insert(val);
452 row
453 }
454
455 pub fn find_or_insert<T: Table<Schema = S, Conflict = TableRow<T>>>(
475 &mut self,
476 val: T,
477 ) -> TableRow<T> {
478 match self.insert(val) {
479 Ok(row) => row,
480 Err(row) => row,
481 }
482 }
483
484 pub fn downgrade(&'static mut self) -> &'static mut TransactionWeak<S> {
486 Box::leak(Box::new(TransactionWeak { _p: PhantomData }))
487 }
488}
489
490pub struct TransactionWeak<S> {
497 _p: PhantomData<Transaction<S>>,
498}
499
500impl<S: Schema> TransactionWeak<S> {
501 pub fn delete<T: Table<Schema = S>>(&mut self, val: TableRow<T>) -> Result<bool, T::Referer> {
508 let schema = crate::schema::from_macro::Schema::new::<S>();
509
510 let mut checks = vec![];
514 for (&table_name, table) in &schema.tables {
515 for col in table.columns.iter().filter_map(|(col_name, col)| {
516 let col = &col.def;
517 col.fk
518 .as_ref()
519 .is_some_and(|(t, c)| t == T::NAME && c == T::ID)
520 .then_some(col_name)
521 }) {
522 let mut stmt = emit::Stmt::default();
523 stmt.write("SELECT ");
524 stmt.write_param(&OrdRc(Rc::new(val.inner.idx.into())));
525 stmt.write(format_args!(
526 " IN (SELECT {0}.{1} FROM {0})",
527 Alias(table_name),
528 Alias(col)
529 ));
530 checks.push(stmt);
531 }
532 }
533
534 let mut stmt = emit::Stmt::default();
535 stmt.write(format_args!(
536 "DELETE FROM {0} WHERE {0}.{1} = ",
537 Alias(T::NAME),
538 Alias(T::ID)
539 ));
540 stmt.write_param(&OrdRc::new(val.inner.idx));
541
542 TXN.with_borrow(|txn| {
543 let txn = txn.as_ref().unwrap().get();
544
545 for stmt in checks {
546 let mut cached = txn.prepare_cached(&stmt.sql).unwrap();
547 match cached.query_one(rusqlite::params_from_iter(stmt.params), |r| r.get(0)) {
548 Ok(true) => return Err(T::get_referer_unchecked()),
549 Ok(false) => {}
550 Err(err) => panic!("{err:?}"),
551 }
552 }
553
554 let mut cached = txn.prepare_cached(&stmt.sql).unwrap();
555 match cached.execute(rusqlite::params_from_iter(stmt.params)) {
556 Ok(0) => Ok(false),
557 Ok(1) => Ok(true),
558 Ok(n) => {
559 panic!("unexpected number of deletes {n}")
560 }
561 Err(err) => panic!("{err:?}"),
562 }
563 })
564 }
565
566 pub fn delete_ok<T: Table<Referer = Infallible, Schema = S>>(
572 &mut self,
573 val: TableRow<T>,
574 ) -> bool {
575 let Ok(res) = self.delete(val);
576 res
577 }
578
579 pub fn rusqlite_transaction<R>(&mut self, f: impl FnOnce(&rusqlite::Transaction) -> R) -> R {
589 TXN.with_borrow(|txn| f(txn.as_ref().unwrap().get()))
590 }
591}
592
593pub fn try_insert_private<T: Table>(
594 table: lower::JoinableTable,
595 idx: Option<i64>,
596 val: T,
597) -> Result<TableRow<T>, T::Conflict> {
598 let mut reader = Reader::default();
599 T::read(&val, &mut reader);
600 if let Some(idx) = idx {
601 reader.col::<i64>(T::ID, idx);
602 }
603
604 let mut stmt = emit::Stmt::default();
605 stmt.write("INSERT INTO ");
606 table.emit(&mut stmt);
607
608 if reader.builder.is_empty() {
609 stmt.write(" DEFAULT VALUES");
611 } else {
612 let (col_names, col_exprs): (Vec<_>, Vec<_>) = reader.builder.clone().into_iter().collect();
613
614 stmt.write(" (");
615 let mut list = ListWriter::new(&mut stmt, ", ");
616 for col in col_names {
617 list.item().write(Alias(col));
618 }
619 stmt.write(") VALUES (");
620 let mut list = ListWriter::new(&mut stmt, ", ");
621 for val in col_exprs {
622 list.item().write_param(&val);
623 }
624 stmt.write(")");
625 }
626 stmt.write(" RETURNING ").write(T::ID);
627
628 let res = TXN.with_borrow(|txn| {
629 let txn = txn.as_ref().unwrap().get();
630 track_stmt(txn, &stmt.sql, &stmt.params);
631
632 let mut statement = txn.prepare_cached(&stmt.sql).unwrap();
633 let mut res = statement
634 .query_map(rusqlite::params_from_iter(stmt.params), |row| {
635 Ok(TableRow::<T>::from_sql(row.get_ref(T::ID)?)?)
636 })
637 .unwrap();
638
639 res.next().unwrap()
640 });
641
642 match res {
643 Ok(id) => {
644 if let Some(idx) = idx {
645 assert_eq!(idx, id.inner.idx);
646 }
647 Ok(id)
648 }
649 Err(rusqlite::Error::SqliteFailure(kind, Some(msg)))
650 if kind.code == ErrorCode::ConstraintViolation =>
651 {
652 let res = TXN.with_borrow(|txn| {
654 let txn = txn.as_ref().unwrap().get();
655 <T::Conflict as FromConflict>::from_conflict(
656 txn,
657 JoinableTableWithId {
658 name: table,
659 main_column: T::ID,
660 },
661 reader.builder,
662 msg,
663 )
664 });
665 Err(res)
666 }
667 Err(err) => panic!("{err:?}"),
668 }
669}
670
671pub(crate) fn try_update_private<T: Table>(
672 row: TableRow<T>,
673 val: T::Mutable,
674) -> Result<(), T::Conflict> {
675 let val = T::mutable_into_insert(val);
676 let mut reader = Reader::default();
677 T::read(&val, &mut reader);
678
679 let mut stmt = emit::Stmt::default();
680 stmt.write("UPDATE ");
681 lower::JoinableTable::Table(T::NAME, None).emit(&mut stmt);
682
683 stmt.write(" SET ");
684 let mut list = ListWriter::new(&mut stmt, ", ");
685 for (key, val) in &reader.builder {
686 list.item()
687 .write(format_args!("{} = ", Alias(key)))
688 .write_param(val);
689 }
690 list.default(format_args!("{1} = {0}.{1}", Alias(T::NAME), Alias(T::ID)));
691
692 stmt.write(format_args!(
693 " WHERE {}.{} = ",
694 Alias(T::NAME),
695 Alias(T::ID)
696 ));
697 stmt.write_param(&OrdRc(Rc::new(row.inner.idx.into())));
698
699 let res = TXN.with_borrow(|txn| {
700 let txn = txn.as_ref().unwrap().get();
701
702 let mut cached = txn.prepare_cached(&stmt.sql).unwrap();
703 cached.execute(rusqlite::params_from_iter(stmt.params))
704 });
705
706 match res {
707 Ok(1) => Ok(()),
708 Ok(n) => panic!("unexpected number of updates: {n}"),
709 Err(rusqlite::Error::SqliteFailure(kind, Some(msg)))
710 if kind.code == ErrorCode::ConstraintViolation =>
711 {
712 let res = TXN.with_borrow(|txn| {
714 let txn = txn.as_ref().unwrap().get();
715 <T::Conflict as FromConflict>::from_conflict(
716 txn,
717 lower::JoinableTableWithId {
718 name: lower::JoinableTable::Table(T::NAME, None),
719 main_column: T::ID,
720 },
721 reader.builder,
722 msg,
723 )
724 });
725 Err(res)
726 }
727 Err(err) => panic!("{err:?}"),
728 }
729}