use super::SqlxStore;
use super::entities::device_code::Model;
use crate::error::record_not_updated;
use crate::model::{self, ActiveRow, SqlxModel};
use crate::pool::{Engine, Exec};
use crate::schema::AuthSchema;
use crate::sql::Sql;
use crate::value::SqlValue;
use alibi_core::error::{AuthError, AuthResult};
use alibi_core::store::DeviceCodeStore;
use alibi_core::types::{CreateDeviceCode, DeviceCode, UpdateDeviceCode};
use async_trait::async_trait;
use uuid::Uuid;
impl<S: AuthSchema + Send + Sync> SqlxStore<S> {
async fn create_device_code_with_connection(
&self,
exec: Exec<'_>,
input: CreateDeviceCode,
fields: serde_json::Map<String, serde_json::Value>,
) -> AuthResult<DeviceCode> {
let mut active = ActiveRow::new();
active.set(
"id",
self.generated_id(self.exec(), "deviceCode", "deviceCode", "id")
.await?
.unwrap_or_else(|| Uuid::new_v4().to_string()),
);
active.set("deviceCode", input.device_code);
active.set("userCode", input.user_code);
active.set("userId", input.user_id);
active.set("expiresAt", input.expires_at);
active.set("status", input.status);
active.set("lastPolledAt", input.last_polled_at);
active.set("pollingInterval", input.polling_interval);
active.set("clientId", input.client_id);
active.set("scope", input.scope);
if fields.is_empty() {
return model::insert::<Model>(exec, &active)
.await
.map(|model| DeviceCode::from(&model));
}
if fields
.keys()
.any(|name| Model::COLUMN_NAMES.contains(&name.as_str()))
{
return Err(AuthError::bad_request(
"Application device fields cannot replace core columns",
));
}
let present = active.present().collect::<Vec<_>>();
let mut sql = Sql::with(exec.engine(), "INSERT INTO ");
sql.ident(Model::TABLE);
sql.push(" (");
for (index, name) in present
.iter()
.map(|(name, _)| *name)
.chain(fields.keys().map(String::as_str))
.enumerate()
{
if index > 0 {
sql.push(", ");
}
sql.ident(name);
}
sql.push(") VALUES (");
for (index, (name, value)) in present.into_iter().enumerate() {
if index > 0 {
sql.push(", ");
}
model::bind_model_value::<Model>(&mut sql, name, value.clone());
}
for value in fields.into_values() {
sql.push(", ");
if value.is_null() {
sql.push("NULL");
} else {
sql.bind(device_field_value(value));
}
}
sql.push(")");
model::returning::<Model>(&mut sql);
exec.fetch_optional::<Model>(sql)
.await?
.map(|model| DeviceCode::from(&model))
.ok_or_else(|| AuthError::internal("Device code insertion returned no row"))
}
async fn find_device_code(&self, column: &str, value: &str) -> AuthResult<Option<DeviceCode>> {
let mut sql = model::select_model::<Model>(self.exec());
sql.push(" WHERE ");
sql.compare(Model::TABLE, column, " = ", value);
model::limit_one(&mut sql);
Ok(self
.exec()
.fetch_optional::<Model>(sql)
.await?
.map(|model| DeviceCode::from(&model)))
}
async fn update_device_codes(
&self,
sets: Vec<(&'static str, SqlValue)>,
guards: Vec<(&'static str, Option<SqlValue>)>,
) -> AuthResult<u64> {
let mut sql = Sql::with(self.exec().engine(), "UPDATE ");
sql.ident(Model::TABLE);
sql.push(" SET ");
for (index, (column, value)) in sets.into_iter().enumerate() {
if index > 0 {
sql.push(", ");
}
sql.assign(column, value);
}
for (index, (column, value)) in guards.into_iter().enumerate() {
sql.push(if index == 0 { " WHERE " } else { " AND " });
sql.column(Model::TABLE, column);
match value {
Some(value) => {
sql.push(" = ");
sql.bind(value);
}
None => {
sql.push(" IS NULL");
}
}
}
self.exec().execute(sql).await
}
}
#[async_trait]
impl<S> DeviceCodeStore for SqlxStore<S>
where
S: AuthSchema + Send + Sync,
{
async fn create_device_code(&self, input: CreateDeviceCode) -> AuthResult<DeviceCode> {
self.create_device_code_with_connection(self.exec(), input, serde_json::Map::new())
.await
}
async fn create_device_code_with_fields(
&self,
input: CreateDeviceCode,
fields: serde_json::Map<String, serde_json::Value>,
) -> AuthResult<DeviceCode> {
self.create_device_code_with_connection(self.exec(), input, fields)
.await
}
async fn device_code_fields(
&self,
id: &str,
) -> AuthResult<serde_json::Map<String, serde_json::Value>> {
let mut query = if self.exec().engine() == Engine::Postgres {
Sql::with(
Engine::Postgres,
"SELECT to_jsonb(t)::text FROM \"deviceCode\" t WHERE \"id\" = ",
)
} else {
#[derive(sqlx::FromRow)]
struct ColumnName {
name: String,
}
let columns = self
.exec()
.fetch_all::<ColumnName>(Sql::with(
Engine::Sqlite,
"SELECT name FROM pragma_table_info('deviceCode')",
))
.await?;
let mut query = Sql::with(Engine::Sqlite, "SELECT json_object(");
for (index, column) in columns.into_iter().enumerate() {
if index > 0 {
query.push(", ");
}
query.bind(column.name.clone());
query.push(", ");
query.ident(&column.name);
}
query.push(") FROM \"deviceCode\" WHERE \"id\" = ");
query
};
query.bind(id);
let json = self.exec().fetch_scalar::<String>(query).await?;
let mut fields = json
.map(|json| serde_json::from_str::<serde_json::Map<String, serde_json::Value>>(&json))
.transpose()?
.unwrap_or_default();
fields.retain(|name, _| !Model::COLUMN_NAMES.contains(&name.as_str()));
Ok(fields)
}
async fn consume_device_code(
&self,
id: &str,
status: &str,
ownership: &serde_json::Map<String, serde_json::Value>,
) -> AuthResult<Option<DeviceCode>> {
let mut sql = Sql::with(
self.exec().engine(),
"DELETE FROM \"deviceCode\" WHERE id = ",
);
sql.bind(id);
sql.push(" AND status = ");
sql.bind(status);
for (field, value) in ownership {
sql.push(" AND ");
sql.ident(field);
if value.is_null() {
sql.push(" IS NULL");
} else {
sql.push(" = ");
sql.bind(device_field_value(value.clone()));
}
}
model::returning::<Model>(&mut sql);
self.exec()
.fetch_optional::<Model>(sql)
.await
.map(|row| row.as_ref().map(DeviceCode::from))
}
async fn get_device_code_by_device_code(
&self,
device_code: &str,
) -> AuthResult<Option<DeviceCode>> {
self.find_device_code("deviceCode", device_code).await
}
async fn get_device_code_by_user_code(
&self,
user_code: &str,
) -> AuthResult<Option<DeviceCode>> {
self.find_device_code("userCode", user_code).await
}
async fn update_device_code(
&self,
id: &str,
update: UpdateDeviceCode,
) -> AuthResult<DeviceCode> {
let mut sql = model::by_id::<Model>(self.exec(), id);
model::limit_one(&mut sql);
let Some(model) = self.exec().fetch_optional::<Model>(sql).await? else {
return Err(AuthError::not_found("Device code not found"));
};
let mut active = model.into_active();
if let Some(status) = update.status {
active.set("status", status);
}
if let Some(user_id) = update.user_id {
active.set("userId", user_id);
}
if let Some(last_polled_at) = update.last_polled_at {
active.set("lastPolledAt", last_polled_at);
}
model::update::<Model>(self.exec(), &active)
.await?
.map(|model| DeviceCode::from(&model))
.ok_or_else(record_not_updated)
}
async fn update_device_code_if_status(
&self,
id: &str,
current_status: &str,
update: UpdateDeviceCode,
) -> AuthResult<bool> {
let mut sets = Vec::new();
if let Some(status) = update.status {
sets.push(("status", status.into()));
}
if let Some(user_id) = update.user_id {
sets.push(("userId", user_id.into()));
}
if let Some(last_polled_at) = update.last_polled_at {
sets.push(("lastPolledAt", last_polled_at.into()));
}
self.update_device_codes(
sets,
vec![
("id", Some(id.into())),
("status", Some(current_status.into())),
],
)
.await
.map(|affected| affected == 1)
}
async fn claim_device_code(&self, id: &str, user_id: &str) -> AuthResult<bool> {
self.update_device_codes(
vec![("userId", user_id.into())],
vec![
("id", Some(id.into())),
("status", Some("pending".into())),
("userId", None),
],
)
.await
.map(|affected| affected == 1)
}
async fn delete_device_code(&self, id: &str) -> AuthResult<()> {
self.exec()
.execute(model::delete_by_id::<Model>(self.exec(), id))
.await
.map(|_| ())
}
async fn delete_device_code_if_status(&self, id: &str, status: &str) -> AuthResult<bool> {
let mut sql = model::delete_by_id::<Model>(self.exec(), id);
sql.push(" AND ");
sql.compare(Model::TABLE, "status", " = ", status);
self.exec().execute(sql).await.map(|affected| affected == 1)
}
}
fn device_field_value(value: serde_json::Value) -> SqlValue {
match value {
serde_json::Value::String(value) => value.into(),
serde_json::Value::Bool(value) => value.into(),
serde_json::Value::Number(value) => value
.as_i64()
.map_or_else(|| value.as_f64().into(), SqlValue::from),
serde_json::Value::Null => SqlValue::Text(None),
value => SqlValue::Json(Some(Box::new(value))),
}
}