ecat_data/
rdbms.rs

1// Copyright (c) 2026 erik <erik@erik.xyz> — https://erik.xyz
2use async_trait::async_trait;
3
4use crate::dialect::Dialect;
5
6#[derive(Debug, Clone)]
7pub struct Row {
8    columns: Vec<String>,
9    values: Vec<serde_json::Value>,
10}
11
12impl Row {
13    /// Create a new Row with the given columns and values.
14    pub fn new(columns: Vec<String>, values: Vec<serde_json::Value>) -> Self {
15        debug_assert_eq!(
16            columns.len(),
17            values.len(),
18            "columns and values must have the same length"
19        );
20        Self { columns, values }
21    }
22
23    pub fn get(&self, col: &str) -> Option<&serde_json::Value> {
24        self.columns
25            .iter()
26            .position(|c| c == col)
27            .and_then(|i| self.values.get(i))
28    }
29}
30
31/// 事务内部实现。后端（sqlx / tiberius）实现它，`Transaction` 转发调用。
32#[async_trait]
33pub trait TransactionInner: Send {
34    async fn execute(&mut self, sql: &str) -> Result<u64, RdbmsError>;
35    async fn query(&mut self, sql: &str) -> Result<Vec<Row>, RdbmsError>;
36    async fn execute_with(
37        &mut self,
38        sql: &str,
39        params: &[serde_json::Value],
40    ) -> Result<u64, RdbmsError>;
41    async fn query_with(
42        &mut self,
43        sql: &str,
44        params: &[serde_json::Value],
45    ) -> Result<Vec<Row>, RdbmsError>;
46    fn dialect(&self) -> Dialect;
47    async fn commit(&mut self) -> Result<(), RdbmsError>;
48    async fn rollback(&mut self) -> Result<(), RdbmsError>;
49}
50
51/// 无 backing 连接的事务（[`Transaction::new`]）执行 SQL 时的错误文案。
52/// 这类事务只能作为空占位，执行任何语句都是编程错误 —— 必须报错而非
53/// 静默返回 0 行影响，否则写操作会无声丢失。
54const NO_BACKING: &str = "transaction has no backing connection (created via Transaction::new)";
55
56pub struct Transaction {
57    committed: bool,
58    rolled_back: bool,
59    /// 在 `with_inner` 时从 inner 拷贝，避免 `dialect(&self)` 这个同步方法
60    /// 需要等待异步锁。
61    dialect: Dialect,
62    inner: tokio::sync::Mutex<Option<Box<dyn TransactionInner>>>,
63}
64
65impl Transaction {
66    /// 创建一个无 backing 连接的空事务。
67    ///
68    /// 只能作为占位符用于「不执行任何语句」的场景；在其中执行 SQL 会返回错误。
69    /// 需要真正执行语句时用 [`Transaction::with_inner`] 或后端的 `transaction()`。
70    pub fn new() -> Self {
71        Self {
72            committed: false,
73            rolled_back: false,
74            dialect: Dialect::Standard,
75            inner: tokio::sync::Mutex::new(None),
76        }
77    }
78
79    pub fn with_inner(inner: Box<dyn TransactionInner>) -> Self {
80        let dialect = inner.dialect();
81        Self {
82            committed: false,
83            rolled_back: false,
84            dialect,
85            inner: tokio::sync::Mutex::new(Some(inner)),
86        }
87    }
88
89    pub async fn commit(mut self) -> Result<(), RdbmsError> {
90        if let Some(inner) = self.inner.get_mut().as_mut() {
91            inner.commit().await?;
92        }
93        self.committed = true;
94        Ok(())
95    }
96
97    pub async fn rollback(mut self) -> Result<(), RdbmsError> {
98        if let Some(inner) = self.inner.get_mut().as_mut() {
99            inner.rollback().await?;
100        }
101        self.committed = false;
102        self.rolled_back = true;
103        Ok(())
104    }
105}
106
107impl Default for Transaction {
108    fn default() -> Self {
109        Self::new()
110    }
111}
112
113#[async_trait]
114impl SqlExecutor for Transaction {
115    async fn execute(&self, sql: &str) -> Result<u64, RdbmsError> {
116        let mut guard = self.inner.lock().await;
117        match guard.as_mut() {
118            Some(inner) => inner.execute(sql).await,
119            None => Err(RdbmsError::Database(NO_BACKING.into())),
120        }
121    }
122
123    async fn query(&self, sql: &str) -> Result<Vec<Row>, RdbmsError> {
124        let mut guard = self.inner.lock().await;
125        match guard.as_mut() {
126            Some(inner) => inner.query(sql).await,
127            None => Err(RdbmsError::Database(NO_BACKING.into())),
128        }
129    }
130
131    async fn execute_with(
132        &self,
133        sql: &str,
134        params: &[serde_json::Value],
135    ) -> Result<u64, RdbmsError> {
136        let mut guard = self.inner.lock().await;
137        match guard.as_mut() {
138            Some(inner) => inner.execute_with(sql, params).await,
139            None => Err(RdbmsError::Database(NO_BACKING.into())),
140        }
141    }
142
143    async fn query_with(
144        &self,
145        sql: &str,
146        params: &[serde_json::Value],
147    ) -> Result<Vec<Row>, RdbmsError> {
148        let mut guard = self.inner.lock().await;
149        match guard.as_mut() {
150            Some(inner) => inner.query_with(sql, params).await,
151            None => Err(RdbmsError::Database(NO_BACKING.into())),
152        }
153    }
154
155    /// 在**调用方已有的事务**里跑两条语句：不另开事务，也不提交 ——
156    /// 原子边界归调用方，`insert(&tx, …)` 的语义就是「用我这个事务发这两条」。
157    ///
158    /// 全程持锁：若在两条语句之间放开，并发的 `tx.execute(…)`（同一条连接上的
159    /// 另一个 INSERT）会把 `LAST_INSERT_ID()` 的值换成它插的那条 ——
160    /// 正是本方法要挡的静默错值。
161    async fn execute_then_query(
162        &self,
163        first: &str,
164        first_params: &[serde_json::Value],
165        second: &str,
166    ) -> Result<Vec<Row>, RdbmsError> {
167        let mut guard = self.inner.lock().await;
168        match guard.as_mut() {
169            Some(inner) => {
170                inner.execute_with(first, first_params).await?;
171                inner.query(second).await
172            }
173            None => Err(RdbmsError::Database(NO_BACKING.into())),
174        }
175    }
176
177    fn dialect(&self) -> Dialect {
178        self.dialect
179    }
180}
181
182impl Drop for Transaction {
183    fn drop(&mut self) {
184        // 这里只记日志与计数：Drop 里无法执行异步回滚，实际回滚依赖
185        // 底层 sqlx / tiberius 事务在未提交时 Drop 自动回滚。
186        if !self.committed && !self.rolled_back {
187            crate::timeout::TRANSACTIONS_LEAKED.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
188            tracing::warn!("transaction dropped without commit — rolling back");
189        }
190    }
191}
192
193#[async_trait]
194pub trait SqlExecutor: Send + Sync {
195    /// 执行一条 SQL 语句，返回受影响行数。
196    /// 用户提供的值请走 [`SqlExecutor::execute_with`]。
197    async fn execute(&self, sql: &str) -> Result<u64, RdbmsError>;
198    /// 查询多行。用户提供的值请走 [`SqlExecutor::query_with`]。
199    async fn query(&self, sql: &str) -> Result<Vec<Row>, RdbmsError>;
200    /// 参数化执行，防注入。无法绑定参数的后端返回错误。
201    async fn execute_with(
202        &self,
203        _sql: &str,
204        _params: &[serde_json::Value],
205    ) -> Result<u64, RdbmsError> {
206        Err(RdbmsError::Database(
207            "parameterized execute not supported by this backend".into(),
208        ))
209    }
210    /// 参数化查询，防注入。无法绑定参数的后端返回错误。
211    async fn query_with(
212        &self,
213        _sql: &str,
214        _params: &[serde_json::Value],
215    ) -> Result<Vec<Row>, RdbmsError> {
216        Err(RdbmsError::Database(
217            "parameterized query not supported by this backend".into(),
218        ))
219    }
220    /// 写路径且需要返回结果（`INSERT ... RETURNING` / `OUTPUT INSERTED`）。
221    /// 默认委托给 [`SqlExecutor::query_with`]；只有读写分离路由需要覆写，
222    /// 否则写语句会被路由到从库。
223    async fn query_write(
224        &self,
225        sql: &str,
226        params: &[serde_json::Value],
227    ) -> Result<Vec<Row>, RdbmsError> {
228        self.query_with(sql, params).await
229    }
230    /// 在**同一条连接**上先执行 `first`、再执行 `second`（查询），两者原子。
231    ///
232    /// 存在的理由：MySQL 的 `LAST_INSERT_ID()` 是**连接作用域**的 ——
233    /// 池下两次独立取用可能落到不同连接、取回别的会话刚插入的值（静默错值），
234    /// 所以 `insert` 的两步式主键回填必须落在同一条连接上。
235    ///
236    /// 默认实现返回「不支持」。三种覆写：
237    /// - [`Transaction`] —— **直接在自己身上跑两条**（它本来就在一条连接上的事务里）
238    /// - 需要两步式的客户端（如 `ecat-data-sqlx` 的 `SqlxClient`）—— **开一个事务，跑完提交**
239    /// - 一步式后端（MSSQL 的 `OUTPUT INSERTED`、PG/SQLite 的 `RETURNING`）—— 用不到，不覆写
240    async fn execute_then_query(
241        &self,
242        _first: &str,
243        _first_params: &[serde_json::Value],
244        _second: &str,
245    ) -> Result<Vec<Row>, RdbmsError> {
246        Err(RdbmsError::Database(
247            "this backend cannot run two statements atomically on one connection".into(),
248        ))
249    }
250    /// 本执行器背后的数据库方言。
251    fn dialect(&self) -> Dialect;
252}
253
254#[async_trait]
255pub trait RdbmsClient: SqlExecutor {
256    async fn transaction(&self) -> Result<Transaction, RdbmsError>;
257}
258
259#[derive(Debug, thiserror::Error)]
260pub enum RdbmsError {
261    #[error("database error: {0}")]
262    Database(String),
263    #[error("connection error: {0}")]
264    Connection(String),
265    #[error("configuration error: {0}")]
266    Config(String),
267    #[error("timeout: {0}")]
268    Timeout(String),
269    /// 读写分离路由：副本全部不可用，且未开启降级读主
270    /// （[`crate::RdbmsRouting`]）。
271    #[error("no available replica")]
272    NoAvailableReplica,
273}
274
275// 测试独立成文件：本文件紧贴 500 行上限（项目硬规则），内联测试会顶过。
276#[cfg(test)]
277mod tests;
ecat_data/rdbms.rs

ecat_data/
rdbms.rs
