use std::collections::HashMap;
use std::sync::Arc;
use async_trait::async_trait;
use crate::core::{
Assignment, FieldSchema, FieldType, Filter, ModelSchema, Op, SqlValue, WhereExpr,
};
use crate::sql::Pool;
#[derive(Debug, thiserror::Error)]
pub enum BulkActionError {
#[error("unknown action: {0}")]
UnknownAction(String),
#[error("invalid table or column identifier: {0}")]
InvalidIdent(String),
#[error("invalid primary key: {0}")]
InvalidPk(String),
#[error("database error: {0}")]
Database(String),
}
#[derive(Debug, Clone)]
pub struct BulkActionResult {
pub affected: u64,
pub action: String,
pub table: String,
}
#[derive(Debug, Clone)]
pub struct PkSet {
model: &'static ModelSchema,
keys: Vec<SqlValue>,
}
impl PkSet {
pub const MAX_KEYS: usize = 10_000;
fn capped(
model: &'static ModelSchema,
keys: impl Iterator<Item = Result<SqlValue, BulkActionError>>,
) -> Result<Self, BulkActionError> {
let keys: Vec<SqlValue> = keys.take(Self::MAX_KEYS + 1).collect::<Result<_, _>>()?;
if keys.len() > Self::MAX_KEYS {
return Err(BulkActionError::InvalidPk(format!(
"more than {} keys in one action",
Self::MAX_KEYS
)));
}
Ok(Self { model, keys })
}
pub fn parse<S: AsRef<str>>(
model: &'static ModelSchema,
raw: impl IntoIterator<Item = S>,
) -> Result<Self, BulkActionError> {
let pk = pk_field(model)?;
let keys = raw.into_iter().map(|r| {
crate::forms::parse_pk_string(pk, r.as_ref())
.map_err(|e| BulkActionError::InvalidPk(e.to_string()))
});
Self::capped(model, keys)
}
pub fn new<V: Into<SqlValue>>(
model: &'static ModelSchema,
keys: impl IntoIterator<Item = V>,
) -> Result<Self, BulkActionError> {
let pk = pk_field(model)?;
Self::capped(model, keys.into_iter().map(|k| coerce_key(pk, k.into())))
}
#[must_use]
pub fn model(&self) -> &'static ModelSchema {
self.model
}
#[must_use]
pub fn keys(&self) -> &[SqlValue] {
&self.keys
}
#[must_use]
pub fn is_empty(&self) -> bool {
self.keys.is_empty()
}
fn filter(&self) -> WhereExpr {
let column = self.model.primary_key().map_or("id", |pk| pk.column);
WhereExpr::Predicate(Filter::new(
column,
Op::In,
SqlValue::List(self.keys.clone()),
))
}
fn target(&self) -> &'static ModelSchema {
first_audited(inventory::iter::<crate::core::ModelEntry>, self.model.table)
.unwrap_or(self.model)
}
}
fn pk_field(model: &'static ModelSchema) -> Result<&'static FieldSchema, BulkActionError> {
model
.primary_key()
.ok_or_else(|| BulkActionError::InvalidPk(format!("`{}` has no primary key", model.table)))
}
fn coerce_key(pk: &FieldSchema, v: SqlValue) -> Result<SqlValue, BulkActionError> {
let int = match v {
SqlValue::I16(n) => Some(i64::from(n)),
SqlValue::I32(n) => Some(i64::from(n)),
SqlValue::I64(n) => Some(n),
_ => None,
};
let typed = match pk.ty {
FieldType::I64 => int.map(SqlValue::I64),
FieldType::I32 => int.and_then(|n| i32::try_from(n).ok()).map(SqlValue::I32),
FieldType::I16 => int.and_then(|n| i16::try_from(n).ok()).map(SqlValue::I16),
FieldType::String if matches!(v, SqlValue::String(_)) => Some(v.clone()),
FieldType::Uuid if matches!(v, SqlValue::Uuid(_)) => Some(v.clone()),
_ => None,
};
typed.ok_or_else(|| {
BulkActionError::InvalidPk(format!("`{}` is {}, got {v:?}", pk.name, pk.ty.as_str()))
})
}
#[async_trait]
pub trait BulkAction: Send + Sync + 'static {
fn name(&self) -> &str;
fn label(&self) -> &str;
async fn run(&self, pks: &PkSet, pool: &Pool) -> Result<BulkActionResult, BulkActionError>;
}
pub type BoxedAction = Arc<dyn BulkAction>;
pub struct BulkActionRegistry {
actions: HashMap<String, BoxedAction>,
}
impl Default for BulkActionRegistry {
fn default() -> Self {
Self::new()
}
}
impl BulkActionRegistry {
#[must_use]
pub fn new() -> Self {
Self {
actions: HashMap::new(),
}
}
#[must_use]
pub fn register(mut self, action: BoxedAction) -> Self {
self.actions.insert(action.name().to_owned(), action);
self
}
#[must_use]
pub fn get(&self, name: &str) -> Option<&BoxedAction> {
self.actions.get(name)
}
#[must_use]
pub fn list(&self) -> Vec<(String, String)> {
let mut out: Vec<(String, String)> = self
.actions
.values()
.map(|a| (a.name().to_owned(), a.label().to_owned()))
.collect();
out.sort_by(|a, b| a.0.cmp(&b.0));
out
}
pub async fn run(
&self,
name: &str,
pks: &PkSet,
pool: &Pool,
) -> Result<BulkActionResult, BulkActionError> {
let action = self
.get(name)
.ok_or_else(|| BulkActionError::UnknownAction(name.to_owned()))?;
action.run(pks, pool).await
}
}
fn first_audited<'a>(
entries: impl IntoIterator<Item = &'a crate::core::ModelEntry>,
table: &str,
) -> Option<&'static ModelSchema> {
entries
.into_iter()
.find(|e| e.schema.table == table && e.audited_delete().is_some())
.map(|e| e.schema)
}
async fn set_column(
pool: &Pool,
pks: &PkSet,
column: &'static str,
value: SqlValue,
op: crate::audit::AuditOp,
only_null: bool,
) -> Result<u64, BulkActionError> {
let model = pks.target();
let field = model
.field_by_column(column)
.ok_or_else(|| BulkActionError::InvalidIdent(format!("{}.{column}", model.table)))?;
if pks.is_empty() {
return Ok(0);
}
let mut filter = pks.filter();
filter.push_and(WhereExpr::Predicate(Filter::new(
field.column,
Op::IsNull,
only_null,
)));
let set = vec![Assignment::new(field.column, value)];
let q = crate::core::UpdateQuery::new(model, set, filter);
crate::audit::update_as(pool, &q, op).await.map_err(db_err)
}
fn done(action: &dyn BulkAction, pks: &PkSet, affected: u64) -> BulkActionResult {
BulkActionResult {
affected,
action: action.name().to_owned(),
table: pks.model().table.to_owned(),
}
}
fn db_err(e: crate::sql::ExecError) -> BulkActionError {
BulkActionError::Database(e.to_string())
}
pub struct BulkDeleteAction;
#[async_trait]
impl BulkAction for BulkDeleteAction {
fn name(&self) -> &str {
"delete_selected"
}
fn label(&self) -> &str {
"Delete selected"
}
async fn run(&self, pks: &PkSet, pool: &Pool) -> Result<BulkActionResult, BulkActionError> {
if pks.is_empty() {
return Ok(done(self, pks, 0));
}
let q = crate::core::DeleteQuery::new(pks.target(), pks.filter());
let affected = crate::audit::delete(pool, &q).await.map_err(db_err)?;
Ok(done(self, pks, affected))
}
}
pub struct BulkSoftDeleteAction {
pub column: &'static str,
}
#[async_trait]
impl BulkAction for BulkSoftDeleteAction {
fn name(&self) -> &str {
"soft_delete_selected"
}
fn label(&self) -> &str {
"Soft-delete selected"
}
async fn run(&self, pks: &PkSet, pool: &Pool) -> Result<BulkActionResult, BulkActionError> {
let now = SqlValue::DateTime(chrono::Utc::now());
let op = crate::audit::AuditOp::SoftDelete;
let affected = set_column(pool, pks, self.column, now, op, true).await?;
Ok(done(self, pks, affected))
}
}
pub struct BulkRestoreAction {
pub column: &'static str,
}
#[async_trait]
impl BulkAction for BulkRestoreAction {
fn name(&self) -> &str {
"restore_selected"
}
fn label(&self) -> &str {
"Restore selected"
}
async fn run(&self, pks: &PkSet, pool: &Pool) -> Result<BulkActionResult, BulkActionError> {
let op = crate::audit::AuditOp::Restore;
let affected = set_column(pool, pks, self.column, SqlValue::Null, op, false).await?;
Ok(done(self, pks, affected))
}
}
#[cfg(test)]
mod tests {
use super::*;
struct Dummy {
name: &'static str,
label: &'static str,
}
#[async_trait]
impl BulkAction for Dummy {
fn name(&self) -> &str {
self.name
}
fn label(&self) -> &str {
self.label
}
async fn run(
&self,
pks: &PkSet,
_pool: &Pool,
) -> Result<BulkActionResult, BulkActionError> {
Ok(done(self, pks, 0))
}
}
#[test]
fn registry_starts_empty() {
let r = BulkActionRegistry::new();
assert!(r.list().is_empty());
}
#[test]
fn register_adds_action() {
let r = BulkActionRegistry::new()
.register(Arc::new(Dummy {
name: "a",
label: "A",
}))
.register(Arc::new(Dummy {
name: "b",
label: "B",
}));
let list = r.list();
assert_eq!(list.len(), 2);
assert_eq!(list[0], ("a".to_owned(), "A".to_owned()));
assert_eq!(list[1], ("b".to_owned(), "B".to_owned()));
}
#[test]
fn get_returns_registered_action() {
let r = BulkActionRegistry::new().register(Arc::new(Dummy {
name: "x",
label: "X",
}));
assert!(r.get("x").is_some());
assert!(r.get("nope").is_none());
}
#[test]
fn list_is_alphabetically_sorted() {
let r = BulkActionRegistry::new()
.register(Arc::new(Dummy {
name: "zebra",
label: "Z",
}))
.register(Arc::new(Dummy {
name: "apple",
label: "A",
}))
.register(Arc::new(Dummy {
name: "mango",
label: "M",
}));
let list = r.list();
assert_eq!(list[0].0, "apple");
assert_eq!(list[1].0, "mango");
assert_eq!(list[2].0, "zebra");
}
#[test]
fn re_registering_same_name_replaces() {
let r = BulkActionRegistry::new()
.register(Arc::new(Dummy {
name: "k",
label: "old",
}))
.register(Arc::new(Dummy {
name: "k",
label: "new",
}));
let list = r.list();
assert_eq!(list.len(), 1);
assert_eq!(list[0].1, "new");
}
const fn pk(ty: FieldType) -> FieldSchema {
let mut f = FieldSchema::new("code", "code", ty);
f.primary_key = true;
f
}
static TEXT_FIELDS: [FieldSchema; 1] = [pk(FieldType::String)];
static TEXT_PK: ModelSchema = {
let mut s = ModelSchema::new("Text", "text_pk");
s.fields = &TEXT_FIELDS;
s
};
static I32_FIELDS: [FieldSchema; 1] = [pk(FieldType::I32)];
static I32_PK: ModelSchema = {
let mut s = ModelSchema::new("Small", "small_pk");
s.fields = &I32_FIELDS;
s
};
#[test]
fn pk_set_refuses_keys_of_another_type() {
assert!(matches!(
PkSet::new(&TEXT_PK, [1_i64]),
Err(BulkActionError::InvalidPk(_))
));
let keys = PkSet::new(&TEXT_PK, ["a".to_owned()]).unwrap();
assert_eq!(keys.keys(), [SqlValue::String("a".into())]);
assert!(PkSet::parse(&I32_PK, ["x"]).is_err());
assert!(PkSet::new(&I32_PK, [i64::MAX]).is_err());
let keys = PkSet::new(&I32_PK, [7_i64]).unwrap();
assert_eq!(keys.keys(), [SqlValue::I32(7)]);
}
#[test]
fn pk_set_filters_on_the_schema_pk_column() {
let keys = PkSet::parse(&TEXT_PK, ["a"]).unwrap();
match keys.filter() {
WhereExpr::Predicate(f) => assert_eq!(f.column, "code"),
other => panic!("{other:?}"),
}
}
#[test]
fn pk_set_needs_a_primary_key() {
static NO_PK: ModelSchema = ModelSchema::new("NoPk", "no_pk");
assert!(PkSet::parse(&NO_PK, ["1"]).is_err());
}
#[cfg(feature = "sqlite")]
#[tokio::test]
async fn unknown_action_returns_error() {
let sq = crate::sql::sqlx::SqlitePool::connect_lazy("sqlite::memory:").unwrap();
let pool: Pool = sq.into();
let r = BulkActionRegistry::new();
let keys = PkSet::parse(&TEXT_PK, ["a"]).unwrap();
let err = r.run("nonexistent", &keys, &pool).await.unwrap_err();
assert!(matches!(err, BulkActionError::UnknownAction(_)));
}
fn fake_delete<'a>(
_: &'a Pool,
_: &'a crate::core::DeleteQuery,
) -> std::pin::Pin<
Box<dyn std::future::Future<Output = Result<u64, crate::sql::ExecError>> + Send + 'a>,
> {
Box::pin(async { Ok(0) })
}
#[test]
fn first_audited_skips_an_unaudited_model_on_the_table() {
static PROXY: ModelSchema = ModelSchema::new("Proxy", "shared");
static REAL: ModelSchema = ModelSchema::new("Real", "shared");
let entries = [
crate::core::ModelEntry::new(&PROXY, "t"),
crate::core::ModelEntry::new(&REAL, "t").with_audited(|| None, || Some(fake_delete)),
];
let got = first_audited(&entries, "shared").map(|m| m.name);
assert_eq!(got, Some("Real"));
assert!(first_audited(&entries[..1], "shared").is_none());
}
#[test]
fn builtin_action_names() {
assert_eq!(BulkDeleteAction.name(), "delete_selected");
assert_eq!(
BulkSoftDeleteAction {
column: "deleted_at"
}
.name(),
"soft_delete_selected"
);
assert_eq!(
BulkRestoreAction {
column: "deleted_at"
}
.name(),
"restore_selected"
);
}
}