refactor(data): add select query abstraction

crates/aether-data-query/src/lib.rs

@@ -26,6 +26,290 @@ impl SqlDialect {
     }
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct DialectSql<'a> {
+    common: Option<&'a str>,
+    postgres: Option<&'a str>,
+    sqlite: Option<&'a str>,
+    mysql: Option<&'a str>,
+}
+
+impl<'a> DialectSql<'a> {
+    pub const fn common(sql: &'a str) -> Self {
+        Self {
+            common: Some(sql),
+            postgres: None,
+            sqlite: None,
+            mysql: None,
+        }
+    }
+
+    pub const fn dialect(postgres: &'a str, sqlite: &'a str, mysql: &'a str) -> Self {
+        Self {
+            common: None,
+            postgres: Some(postgres),
+            sqlite: Some(sqlite),
+            mysql: Some(mysql),
+        }
+    }
+
+    pub fn with_postgres(mut self, sql: &'a str) -> Self {
+        self.postgres = Some(sql);
+        self
+    }
+
+    pub fn with_sqlite(mut self, sql: &'a str) -> Self {
+        self.sqlite = Some(sql);
+        self
+    }
+
+    pub fn with_mysql(mut self, sql: &'a str) -> Self {
+        self.mysql = Some(sql);
+        self
+    }
+
+    pub fn sql(self, dialect: SqlDialect) -> &'a str {
+        match dialect {
+            SqlDialect::Postgres => self.postgres.or(self.common),
+            SqlDialect::Sqlite => self.sqlite.or(self.common),
+            SqlDialect::Mysql => self.mysql.or(self.common),
+        }
+        .expect("dialect SQL expression is missing for selected dialect")
+    }
+}
+
+impl<'a> From<&'a str> for DialectSql<'a> {
+    fn from(value: &'a str) -> Self {
+        Self::common(value)
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct SelectColumn<'a> {
+    expr: DialectSql<'a>,
+    alias: Option<&'a str>,
+}
+
+impl<'a> SelectColumn<'a> {
+    pub fn expr(expr: impl Into<DialectSql<'a>>) -> Self {
+        Self {
+            expr: expr.into(),
+            alias: None,
+        }
+    }
+
+    pub fn alias(mut self, alias: &'a str) -> Self {
+        self.alias = Some(alias);
+        self
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct SelectQuery<'a> {
+    distinct: bool,
+    columns: Vec<SelectColumn<'a>>,
+    from: DialectSql<'a>,
+    joins: Vec<DialectSql<'a>>,
+}
+
+impl<'a> SelectQuery<'a> {
+    pub fn new(from: impl Into<DialectSql<'a>>) -> Self {
+        Self {
+            distinct: false,
+            columns: Vec::new(),
+            from: from.into(),
+            joins: Vec::new(),
+        }
+    }
+
+    pub fn distinct(mut self) -> Self {
+        self.distinct = true;
+        self
+    }
+
+    pub fn select(mut self, column: SelectColumn<'a>) -> Self {
+        self.columns.push(column);
+        self
+    }
+
+    pub fn select_columns<I>(mut self, columns: I) -> Self
+    where
+        I: IntoIterator<Item = SelectColumn<'a>>,
+    {
+        self.columns.extend(columns);
+        self
+    }
+
+    pub fn join(mut self, join_sql: impl Into<DialectSql<'a>>) -> Self {
+        self.joins.push(join_sql.into());
+        self
+    }
+
+    pub fn render(&self, dialect: SqlDialect) -> String {
+        let mut sql = String::from("SELECT ");
+        if self.distinct {
+            sql.push_str("DISTINCT ");
+        }
+
+        if self.columns.is_empty() {
+            sql.push('*');
+        } else {
+            for (index, column) in self.columns.iter().enumerate() {
+                if index > 0 {
+                    sql.push_str(", ");
+                }
+                sql.push_str(column.expr.sql(dialect));
+                if let Some(alias) = column.alias {
+                    sql.push_str(" AS ");
+                    sql.push_str(&dialect.quote_ident(alias));
+                }
+            }
+        }
+
+        sql.push_str(" FROM ");
+        sql.push_str(self.from.sql(dialect));
+        for join in &self.joins {
+            sql.push(' ');
+            sql.push_str(join.sql(dialect));
+        }
+        sql
+    }
+
+    pub fn statement<'args, DB>(&self, dialect: SqlDialect) -> SelectStatement<'args, DB>
+    where
+        DB: Database,
+    {
+        SelectStatement {
+            dialect,
+            builder: QueryBuilder::<DB>::new(self.render(dialect)),
+            where_clause: WhereClause::new(),
+        }
+    }
+}
+
+pub struct SelectStatement<'args, DB>
+where
+    DB: Database,
+{
+    dialect: SqlDialect,
+    builder: QueryBuilder<'args, DB>,
+    where_clause: WhereClause,
+}
+
+impl<'args, DB> SelectStatement<'args, DB>
+where
+    DB: Database,
+{
+    pub fn where_eq<T>(&mut self, column_sql: &str, value: T) -> &mut Self
+    where
+        T: 'args + Encode<'args, DB> + Type<DB>,
+    {
+        push_eq(&mut self.builder, &mut self.where_clause, column_sql, value);
+        self
+    }
+
+    pub fn where_optional_eq<T>(&mut self, column_sql: &str, value: Option<T>) -> &mut Self
+    where
+        T: 'args + Encode<'args, DB> + Type<DB>,
+    {
+        push_optional_eq(&mut self.builder, &mut self.where_clause, column_sql, value);
+        self
+    }
+
+    pub fn where_in<T>(&mut self, column_sql: &str, values: &[T]) -> &mut Self
+    where
+        T: Clone + 'args + Encode<'args, DB> + Type<DB>,
+    {
+        push_in(
+            &mut self.builder,
+            &mut self.where_clause,
+            column_sql,
+            values,
+        );
+        self
+    }
+
+    pub fn where_ci_contains(&mut self, column_sql: &str, value: &str) -> &mut Self
+    where
+        String: 'args + Encode<'args, DB> + Type<DB>,
+    {
+        push_ci_contains(
+            &mut self.builder,
+            &mut self.where_clause,
+            self.dialect,
+            column_sql,
+            value,
+        );
+        self
+    }
+
+    pub fn where_ci_contains_any(&mut self, column_sqls: &[&str], value: &str) -> &mut Self
+    where
+        String: 'args + Encode<'args, DB> + Type<DB>,
+    {
+        push_ci_contains_any(
+            &mut self.builder,
+            &mut self.where_clause,
+            self.dialect,
+            column_sqls,
+            value,
+        );
+        self
+    }
+
+    pub fn where_raw(&mut self, predicate_sql: &str) -> &mut Self {
+        if !predicate_sql.trim().is_empty() {
+            self.where_clause.push_next(&mut self.builder);
+            self.builder.push(predicate_sql);
+        }
+        self
+    }
+
+    pub fn order_by_sql(&mut self, order_sql: &str) -> &mut Self {
+        if !order_sql.trim().is_empty() {
+            self.builder.push(" ORDER BY ").push(order_sql);
+        }
+        self
+    }
+
+    pub fn order_by(
+        &mut self,
+        requested_key: Option<&str>,
+        direction: SortDirection,
+        allowed: &[OrderByColumn<'_>],
+        default_key: &str,
+    ) -> &mut Self {
+        push_order_by(
+            &mut self.builder,
+            requested_key,
+            direction,
+            allowed,
+            default_key,
+        );
+        self
+    }
+
+    pub fn limit(&mut self, limit: i64) -> &mut Self
+    where
+        i64: 'args + Encode<'args, DB> + Type<DB>,
+    {
+        push_limit(&mut self.builder, limit);
+        self
+    }
+
+    pub fn limit_offset(&mut self, limit: i64, offset: i64) -> &mut Self
+    where
+        i64: 'args + Encode<'args, DB> + Type<DB>,
+    {
+        push_limit_offset(&mut self.builder, limit, offset);
+        self
+    }
+
+    pub fn finish(self) -> QueryBuilder<'args, DB> {
+        self.builder
+    }
+}
+
 #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
 pub struct WhereClause {
     has_clause: bool,
@@ -360,4 +644,54 @@ mod tests {
         assert!(query.sql().contains(" ORDER BY created_at DESC"));
         assert!(query.sql().contains(" LIMIT ? OFFSET ?"));
     }
+
+    #[test]
+    fn select_query_renders_dialect_specific_projection() {
+        let query = SelectQuery::new("providers").select_columns([
+            SelectColumn::expr("id").alias("provider_id"),
+            SelectColumn::expr(DialectSql::dialect(
+                "CAST(monthly_quota_usd AS DOUBLE PRECISION)",
+                "CAST(monthly_quota_usd AS REAL)",
+                "monthly_quota_usd",
+            ))
+            .alias("monthly_quota_usd"),
+        ]);
+
+        assert_eq!(
+            query.render(SqlDialect::Postgres),
+            "SELECT id AS \"provider_id\", CAST(monthly_quota_usd AS DOUBLE PRECISION) AS \"monthly_quota_usd\" FROM providers"
+        );
+        assert_eq!(
+            query.render(SqlDialect::Mysql),
+            "SELECT id AS `provider_id`, monthly_quota_usd AS `monthly_quota_usd` FROM providers"
+        );
+    }
+
+    #[test]
+    fn select_statement_keeps_bind_order_and_dialect_search() {
+        let query = SelectQuery::new("items")
+            .select(SelectColumn::expr("id"))
+            .select(SelectColumn::expr("name"));
+        let mut statement = query.statement::<Postgres>(SqlDialect::Postgres);
+        statement
+            .where_eq("kind", "scheduled".to_string())
+            .where_ci_contains_any(&["name", "description"], "Fetch")
+            .order_by(
+                Some("name"),
+                SortDirection::Asc,
+                &[OrderByColumn {
+                    key: "name",
+                    sql: "name",
+                }],
+                "name",
+            )
+            .limit_offset(20, 40);
+
+        let mut builder = statement.finish();
+        let query = builder.build();
+        assert_eq!(
+            query.sql(),
+            "SELECT id, name FROM items WHERE kind = $1 AND (name ILIKE $2 OR description ILIKE $3) ORDER BY name ASC LIMIT $4 OFFSET $5"
+        );
+    }
 }

crates/aether-data/src/repository/quota/mod.rs

@@ -3,6 +3,8 @@ mod mysql;
 mod postgres;
 mod sqlite;
 
+use aether_data_query::{DialectSql, SelectColumn, SelectQuery};
+
 #[allow(unused_imports)]
 pub(crate) use aether_data_contracts::repository::quota::{
     ProviderQuotaReadRepository, ProviderQuotaRepository, ProviderQuotaWriteRepository,
@@ -12,3 +14,39 @@ pub use memory::InMemoryProviderQuotaRepository;
 pub use mysql::MysqlProviderQuotaRepository;
 pub use postgres::SqlxProviderQuotaRepository;
 pub use sqlite::SqliteProviderQuotaRepository;
+
+fn quota_snapshot_select() -> SelectQuery<'static> {
+    SelectQuery::new("providers").select_columns([
+        SelectColumn::expr("id").alias("provider_id"),
+        SelectColumn::expr(
+            DialectSql::common("billing_type").with_postgres("CAST(billing_type AS TEXT)"),
+        )
+        .alias("billing_type"),
+        SelectColumn::expr(DialectSql::dialect(
+            "CAST(monthly_quota_usd AS DOUBLE PRECISION)",
+            "CAST(monthly_quota_usd AS REAL)",
+            "monthly_quota_usd",
+        ))
+        .alias("monthly_quota_usd"),
+        SelectColumn::expr(DialectSql::dialect(
+            "CAST(COALESCE(monthly_used_usd, 0) AS DOUBLE PRECISION)",
+            "CAST(COALESCE(monthly_used_usd, 0) AS REAL)",
+            "COALESCE(monthly_used_usd, 0)",
+        ))
+        .alias("monthly_used_usd"),
+        SelectColumn::expr("quota_reset_day"),
+        SelectColumn::expr(DialectSql::dialect(
+            "CAST(EXTRACT(EPOCH FROM quota_last_reset_at) AS BIGINT)",
+            "quota_last_reset_at",
+            "quota_last_reset_at",
+        ))
+        .alias("quota_last_reset_at_unix_secs"),
+        SelectColumn::expr(DialectSql::dialect(
+            "CAST(EXTRACT(EPOCH FROM quota_expires_at) AS BIGINT)",
+            "quota_expires_at",
+            "quota_expires_at",
+        ))
+        .alias("quota_expires_at_unix_secs"),
+        SelectColumn::expr("is_active"),
+    ])
+}

crates/aether-data/src/repository/quota/mysql.rs

@@ -1,26 +1,14 @@
 use async_trait::async_trait;
-use sqlx::{mysql::MySqlRow, MySql, QueryBuilder, Row};
+use sqlx::{mysql::MySqlRow, MySql, Row};
 
 use super::{
-    ProviderQuotaReadRepository, ProviderQuotaWriteRepository, StoredProviderQuotaSnapshot,
+    quota_snapshot_select, ProviderQuotaReadRepository, ProviderQuotaWriteRepository,
+    StoredProviderQuotaSnapshot,
 };
 use crate::driver::mysql::MysqlPool;
 use crate::error::SqlResultExt;
 use crate::DataLayerError;
-use aether_data_query::{push_in, WhereClause};
-
-const QUOTA_COLUMNS: &str = r#"
-SELECT
-  id AS provider_id,
-  billing_type,
-  monthly_quota_usd,
-  COALESCE(monthly_used_usd, 0) AS monthly_used_usd,
-  quota_reset_day,
-  quota_last_reset_at AS quota_last_reset_at_unix_secs,
-  quota_expires_at AS quota_expires_at_unix_secs,
-  is_active
-FROM providers
-"#;
+use aether_data_query::SqlDialect;
 
 #[derive(Debug, Clone)]
 pub struct MysqlProviderQuotaRepository {
@@ -39,8 +27,11 @@ impl ProviderQuotaReadRepository for MysqlProviderQuotaRepository {
         &self,
         provider_id: &str,
     ) -> Result<Option<StoredProviderQuotaSnapshot>, DataLayerError> {
-        let row = sqlx::query(&format!("{QUOTA_COLUMNS} WHERE id = ? LIMIT 1"))
-            .bind(provider_id)
+        let mut statement = quota_snapshot_select().statement::<MySql>(SqlDialect::Mysql);
+        statement.where_eq("id", provider_id.to_string()).limit(1);
+        let row = statement
+            .finish()
+            .build()
             .fetch_optional(&self.pool)
             .await
             .map_sql_err()?;
@@ -55,11 +46,16 @@ impl ProviderQuotaReadRepository for MysqlProviderQuotaRepository {
             return Ok(Vec::new());
         }
 
-        let mut builder = QueryBuilder::<MySql>::new(QUOTA_COLUMNS);
-        let mut where_clause = WhereClause::new();
-        push_in(&mut builder, &mut where_clause, "id", provider_ids);
-        builder.push(" ORDER BY id ASC");
-        let rows = builder.build().fetch_all(&self.pool).await.map_sql_err()?;
+        let mut statement = quota_snapshot_select().statement::<MySql>(SqlDialect::Mysql);
+        statement
+            .where_in("id", provider_ids)
+            .order_by_sql("id ASC");
+        let rows = statement
+            .finish()
+            .build()
+            .fetch_all(&self.pool)
+            .await
+            .map_sql_err()?;
         rows.iter().map(map_row).collect()
     }
 }

crates/aether-data/src/repository/quota/postgres.rs

@@ -1,26 +1,12 @@
 use async_trait::async_trait;
-use sqlx::{PgPool, Postgres, QueryBuilder, Row};
+use sqlx::{PgPool, Postgres, Row};
 
 use super::{
-    ProviderQuotaReadRepository, ProviderQuotaWriteRepository, StoredProviderQuotaSnapshot,
+    quota_snapshot_select, ProviderQuotaReadRepository, ProviderQuotaWriteRepository,
+    StoredProviderQuotaSnapshot,
 };
 use crate::{error::SqlxResultExt, DataLayerError};
-use aether_data_query::{push_in, WhereClause};
-
-const FIND_BY_PROVIDER_ID_SQL: &str = r#"
-SELECT
-  id AS provider_id,
-  CAST(billing_type AS TEXT) AS billing_type,
-  CAST(monthly_quota_usd AS DOUBLE PRECISION) AS monthly_quota_usd,
-  CAST(COALESCE(monthly_used_usd, 0) AS DOUBLE PRECISION) AS monthly_used_usd,
-  quota_reset_day,
-  CAST(EXTRACT(EPOCH FROM quota_last_reset_at) AS BIGINT) AS quota_last_reset_at_unix_secs,
-  CAST(EXTRACT(EPOCH FROM quota_expires_at) AS BIGINT) AS quota_expires_at_unix_secs,
-  is_active
-FROM providers
-WHERE id = $1
-LIMIT 1
-"#;
+use aether_data_query::SqlDialect;
 
 const RESET_DUE_SQL: &str = r#"
 UPDATE providers
@@ -54,8 +40,11 @@ impl ProviderQuotaReadRepository for SqlxProviderQuotaRepository {
         &self,
         provider_id: &str,
     ) -> Result<Option<StoredProviderQuotaSnapshot>, DataLayerError> {
-        let row = sqlx::query(FIND_BY_PROVIDER_ID_SQL)
-            .bind(provider_id)
+        let mut statement = quota_snapshot_select().statement::<Postgres>(SqlDialect::Postgres);
+        statement.where_eq("id", provider_id.to_string()).limit(1);
+        let row = statement
+            .finish()
+            .build()
             .fetch_optional(&self.pool)
             .await
             .map_postgres_err()?;
@@ -69,13 +58,13 @@ impl ProviderQuotaReadRepository for SqlxProviderQuotaRepository {
         if provider_ids.is_empty() {
             return Ok(Vec::new());
         }
-        let mut builder = QueryBuilder::<Postgres>::new(
-            "SELECT id AS provider_id, CAST(billing_type AS TEXT) AS billing_type, CAST(monthly_quota_usd AS DOUBLE PRECISION) AS monthly_quota_usd, CAST(COALESCE(monthly_used_usd, 0) AS DOUBLE PRECISION) AS monthly_used_usd, quota_reset_day, CAST(EXTRACT(EPOCH FROM quota_last_reset_at) AS BIGINT) AS quota_last_reset_at_unix_secs, CAST(EXTRACT(EPOCH FROM quota_expires_at) AS BIGINT) AS quota_expires_at_unix_secs, is_active FROM providers",
-        );
-        let mut where_clause = WhereClause::new();
-        push_in(&mut builder, &mut where_clause, "id", provider_ids);
-        builder.push(" ORDER BY id ASC");
-        builder
+
+        let mut statement = quota_snapshot_select().statement::<Postgres>(SqlDialect::Postgres);
+        statement
+            .where_in("id", provider_ids)
+            .order_by_sql("id ASC");
+        statement
+            .finish()
             .build()
             .fetch_all(&self.pool)
             .await

crates/aether-data/src/repository/quota/sqlite.rs

@@ -1,26 +1,14 @@
 use async_trait::async_trait;
-use sqlx::{sqlite::SqliteRow, QueryBuilder, Row, Sqlite};
+use sqlx::{sqlite::SqliteRow, Row, Sqlite};
 
 use super::{
-    ProviderQuotaReadRepository, ProviderQuotaWriteRepository, StoredProviderQuotaSnapshot,
+    quota_snapshot_select, ProviderQuotaReadRepository, ProviderQuotaWriteRepository,
+    StoredProviderQuotaSnapshot,
 };
 use crate::driver::sqlite::{sqlite_optional_real, sqlite_real, SqlitePool};
 use crate::error::SqlResultExt;
 use crate::DataLayerError;
-use aether_data_query::{push_in, WhereClause};
-
-const QUOTA_COLUMNS: &str = r#"
-SELECT
-  id AS provider_id,
-  billing_type,
-  CAST(monthly_quota_usd AS REAL) AS monthly_quota_usd,
-  CAST(COALESCE(monthly_used_usd, 0) AS REAL) AS monthly_used_usd,
-  quota_reset_day,
-  quota_last_reset_at AS quota_last_reset_at_unix_secs,
-  quota_expires_at AS quota_expires_at_unix_secs,
-  is_active
-FROM providers
-"#;
+use aether_data_query::SqlDialect;
 
 #[derive(Debug, Clone)]
 pub struct SqliteProviderQuotaRepository {
@@ -39,8 +27,11 @@ impl ProviderQuotaReadRepository for SqliteProviderQuotaRepository {
         &self,
         provider_id: &str,
     ) -> Result<Option<StoredProviderQuotaSnapshot>, DataLayerError> {
-        let row = sqlx::query(&format!("{QUOTA_COLUMNS} WHERE id = ? LIMIT 1"))
-            .bind(provider_id)
+        let mut statement = quota_snapshot_select().statement::<Sqlite>(SqlDialect::Sqlite);
+        statement.where_eq("id", provider_id.to_string()).limit(1);
+        let row = statement
+            .finish()
+            .build()
             .fetch_optional(&self.pool)
             .await
             .map_sql_err()?;
@@ -55,11 +46,16 @@ impl ProviderQuotaReadRepository for SqliteProviderQuotaRepository {
             return Ok(Vec::new());
         }
 
-        let mut builder = QueryBuilder::<Sqlite>::new(QUOTA_COLUMNS);
-        let mut where_clause = WhereClause::new();
-        push_in(&mut builder, &mut where_clause, "id", provider_ids);
-        builder.push(" ORDER BY id ASC");
-        let rows = builder.build().fetch_all(&self.pool).await.map_sql_err()?;
+        let mut statement = quota_snapshot_select().statement::<Sqlite>(SqlDialect::Sqlite);
+        statement
+            .where_in("id", provider_ids)
+            .order_by_sql("id ASC");
+        let rows = statement
+            .finish()
+            .build()
+            .fetch_all(&self.pool)
+            .await
+            .map_sql_err()?;
         rows.iter().map(map_row).collect()
     }
 }