use super::{
build_in_clause, map_db_error, params_refs, parse_datetime, parse_datetime_opt,
parse_datetime_opt_row, parse_datetime_row, parse_decimal_opt_row, parse_decimal_row,
parse_decimal_strict, parse_enum, parse_enum_row, parse_uuid, parse_uuid_opt,
parse_uuid_opt_row, parse_uuid_row, uuid_params, with_immediate_transaction,
};
use chrono::Utc;
use r2d2::Pool;
use r2d2_sqlite::SqliteConnectionManager;
use rust_decimal::Decimal;
use stateset_core::{
AddWorkOrderMaterial, BatchResult, CommerceError, CreateWorkOrder, CreateWorkOrderTask,
ProductId, Result, TaskStatus, UpdateWorkOrder, UpdateWorkOrderTask, WorkOrder,
WorkOrderFilter, WorkOrderMaterial, WorkOrderPriority, WorkOrderRepository, WorkOrderStatus,
WorkOrderTask, validate_batch_size,
};
use uuid::Uuid;
#[derive(Debug)]
pub struct SqliteWorkOrderRepository {
pool: Pool<SqliteConnectionManager>,
}
impl SqliteWorkOrderRepository {
#[must_use]
pub const fn new(pool: Pool<SqliteConnectionManager>) -> Self {
Self { pool }
}
fn load_tasks(&self, work_order_id: Uuid) -> Result<Vec<WorkOrderTask>> {
let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
let mut stmt = conn
.prepare(
"SELECT id, work_order_id, sequence, task_name, status, estimated_hours,
actual_hours, assigned_to, started_at, completed_at, notes, created_at, updated_at
FROM manufacturing_work_order_tasks WHERE work_order_id = ? ORDER BY sequence",
)
.map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
let rows = stmt
.query_map([work_order_id.to_string()], |row| {
Ok(WorkOrderTask {
id: parse_uuid_row(&row.get::<_, String>(0)?, "work_order_task", "id")?,
work_order_id: parse_uuid_row(
&row.get::<_, String>(1)?,
"work_order_task",
"work_order_id",
)?,
sequence: row.get(2)?,
task_name: row.get(3)?,
status: parse_enum_row(&row.get::<_, String>(4)?, "work_order_task", "status")?,
estimated_hours: parse_decimal_opt_row(
row.get(5)?,
"work_order_task",
"estimated_hours",
)?,
actual_hours: parse_decimal_opt_row(
row.get(6)?,
"work_order_task",
"actual_hours",
)?,
assigned_to: parse_uuid_opt_row(
row.get::<_, Option<String>>(7)?,
"work_order_task",
"assigned_to",
)?,
started_at: parse_datetime_opt_row(
row.get(8)?,
"work_order_task",
"started_at",
)?,
completed_at: parse_datetime_opt_row(
row.get(9)?,
"work_order_task",
"completed_at",
)?,
notes: row.get(10)?,
created_at: parse_datetime_row(
&row.get::<_, String>(11)?,
"work_order_task",
"created_at",
)?,
updated_at: parse_datetime_row(
&row.get::<_, String>(12)?,
"work_order_task",
"updated_at",
)?,
})
})
.map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
let mut tasks = Vec::new();
for row in rows {
tasks.push(row.map_err(|e| CommerceError::DatabaseError(e.to_string()))?);
}
Ok(tasks)
}
fn load_materials(&self, work_order_id: Uuid) -> Result<Vec<WorkOrderMaterial>> {
let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
let mut stmt = conn
.prepare(
"SELECT id, work_order_id, component_id, component_sku, component_name,
reserved_quantity, consumed_quantity, inventory_reservation_id, created_at, updated_at
FROM manufacturing_work_order_materials WHERE work_order_id = ?",
)
.map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
let rows = stmt
.query_map([work_order_id.to_string()], |row| {
Ok(WorkOrderMaterial {
id: parse_uuid_row(&row.get::<_, String>(0)?, "work_order_material", "id")?,
work_order_id: parse_uuid_row(
&row.get::<_, String>(1)?,
"work_order_material",
"work_order_id",
)?,
component_id: parse_uuid_opt_row(
row.get::<_, Option<String>>(2)?,
"work_order_material",
"component_id",
)?,
component_sku: row.get(3)?,
component_name: row.get(4)?,
reserved_quantity: parse_decimal_row(
&row.get::<_, String>(5)?,
"work_order_material",
"reserved_quantity",
)?,
consumed_quantity: parse_decimal_row(
&row.get::<_, String>(6)?,
"work_order_material",
"consumed_quantity",
)?,
inventory_reservation_id: parse_uuid_opt_row(
row.get::<_, Option<String>>(7)?,
"work_order_material",
"inventory_reservation_id",
)?,
created_at: parse_datetime_row(
&row.get::<_, String>(8)?,
"work_order_material",
"created_at",
)?,
updated_at: parse_datetime_row(
&row.get::<_, String>(9)?,
"work_order_material",
"updated_at",
)?,
})
})
.map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
let mut materials = Vec::new();
for row in rows {
materials.push(row.map_err(|e| CommerceError::DatabaseError(e.to_string()))?);
}
Ok(materials)
}
fn get_task_internal(&self, task_id: Uuid) -> Result<WorkOrderTask> {
let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
conn.query_row(
"SELECT id, work_order_id, sequence, task_name, status, estimated_hours,
actual_hours, assigned_to, started_at, completed_at, notes, created_at, updated_at
FROM manufacturing_work_order_tasks WHERE id = ?",
[task_id.to_string()],
|row| {
Ok(WorkOrderTask {
id: parse_uuid_row(&row.get::<_, String>(0)?, "work_order_task", "id")?,
work_order_id: parse_uuid_row(&row.get::<_, String>(1)?, "work_order_task", "work_order_id")?,
sequence: row.get(2)?,
task_name: row.get(3)?,
status: parse_enum_row(&row.get::<_, String>(4)?, "work_order_task", "status")?,
estimated_hours: parse_decimal_opt_row(row.get(5)?, "work_order_task", "estimated_hours")?,
actual_hours: parse_decimal_opt_row(row.get(6)?, "work_order_task", "actual_hours")?,
assigned_to: parse_uuid_opt_row(
row.get::<_, Option<String>>(7)?,
"work_order_task",
"assigned_to",
)?,
started_at: parse_datetime_opt_row(row.get(8)?, "work_order_task", "started_at")?,
completed_at: parse_datetime_opt_row(row.get(9)?, "work_order_task", "completed_at")?,
notes: row.get(10)?,
created_at: parse_datetime_row(&row.get::<_, String>(11)?, "work_order_task", "created_at")?,
updated_at: parse_datetime_row(&row.get::<_, String>(12)?, "work_order_task", "updated_at")?,
})
},
)
.map_err(|e| match e {
rusqlite::Error::QueryReturnedNoRows => CommerceError::NotFound,
_ => CommerceError::DatabaseError(e.to_string()),
})
}
fn get_material_internal(&self, material_id: Uuid) -> Result<WorkOrderMaterial> {
let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
conn.query_row(
"SELECT id, work_order_id, component_id, component_sku, component_name,
reserved_quantity, consumed_quantity, inventory_reservation_id, created_at, updated_at
FROM manufacturing_work_order_materials WHERE id = ?",
[material_id.to_string()],
|row| {
Ok(WorkOrderMaterial {
id: parse_uuid_row(&row.get::<_, String>(0)?, "work_order_material", "id")?,
work_order_id: parse_uuid_row(&row.get::<_, String>(1)?, "work_order_material", "work_order_id")?,
component_id: parse_uuid_opt_row(
row.get::<_, Option<String>>(2)?,
"work_order_material",
"component_id",
)?,
component_sku: row.get(3)?,
component_name: row.get(4)?,
reserved_quantity: parse_decimal_row(&row.get::<_, String>(5)?, "work_order_material", "reserved_quantity")?,
consumed_quantity: parse_decimal_row(&row.get::<_, String>(6)?, "work_order_material", "consumed_quantity")?,
inventory_reservation_id: parse_uuid_opt_row(
row.get::<_, Option<String>>(7)?,
"work_order_material",
"inventory_reservation_id",
)?,
created_at: parse_datetime_row(&row.get::<_, String>(8)?, "work_order_material", "created_at")?,
updated_at: parse_datetime_row(&row.get::<_, String>(9)?, "work_order_material", "updated_at")?,
})
},
)
.map_err(|e| match e {
rusqlite::Error::QueryReturnedNoRows => CommerceError::NotFound,
_ => CommerceError::DatabaseError(e.to_string()),
})
}
}
impl WorkOrderRepository for SqliteWorkOrderRepository {
fn create(&self, input: CreateWorkOrder) -> Result<WorkOrder> {
let id = Uuid::new_v4();
let work_order_number = WorkOrder::generate_work_order_number();
let now = Utc::now();
let priority = input.priority.unwrap_or(WorkOrderPriority::Normal);
{
let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
conn.execute(
"INSERT INTO manufacturing_work_orders (id, work_order_number, product_id, bom_id, work_center_id,
assigned_to, status, priority, quantity_to_build, quantity_completed, scheduled_start, scheduled_end, notes, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, 'planned', ?, ?, '0', ?, ?, ?, ?, ?)",
rusqlite::params![
id.to_string(),
work_order_number,
input.product_id.to_string(),
input.bom_id.map(|u| u.to_string()),
input.work_center_id,
input.assigned_to.map(|u| u.to_string()),
priority.to_string(),
input.quantity_to_build.to_string(),
input.scheduled_start.map(|dt| dt.to_rfc3339()),
input.scheduled_end.map(|dt| dt.to_rfc3339()),
input.notes,
now.to_rfc3339(),
now.to_rfc3339(),
],
)
.map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
}
let mut tasks = Vec::new();
if let Some(task_inputs) = input.tasks {
for task_input in task_inputs {
let task = self.add_task(id, task_input)?;
tasks.push(task);
}
}
Ok(WorkOrder {
id,
work_order_number,
product_id: input.product_id,
bom_id: input.bom_id,
work_center_id: input.work_center_id,
assigned_to: input.assigned_to,
status: WorkOrderStatus::Planned,
priority,
quantity_to_build: input.quantity_to_build,
quantity_completed: Decimal::ZERO,
scheduled_start: input.scheduled_start,
scheduled_end: input.scheduled_end,
actual_start: None,
actual_end: None,
notes: input.notes,
tasks,
materials: vec![],
version: 1,
created_at: now,
updated_at: now,
})
}
fn get(&self, id: Uuid) -> Result<Option<WorkOrder>> {
let wo_data = {
let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
let result = conn.query_row(
"SELECT id, work_order_number, product_id, bom_id, work_center_id, assigned_to,
status, priority, quantity_to_build, quantity_completed, scheduled_start,
scheduled_end, actual_start, actual_end, notes, created_at, updated_at
FROM manufacturing_work_orders WHERE id = ?",
[id.to_string()],
|row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, String>(2)?,
row.get::<_, Option<String>>(3)?,
row.get::<_, Option<String>>(4)?,
row.get::<_, Option<String>>(5)?,
row.get::<_, String>(6)?,
row.get::<_, String>(7)?,
row.get::<_, String>(8)?,
row.get::<_, String>(9)?,
row.get::<_, Option<String>>(10)?,
row.get::<_, Option<String>>(11)?,
row.get::<_, Option<String>>(12)?,
row.get::<_, Option<String>>(13)?,
row.get::<_, Option<String>>(14)?,
row.get::<_, String>(15)?,
row.get::<_, String>(16)?,
))
},
);
match result {
Ok(data) => Some(data),
Err(rusqlite::Error::QueryReturnedNoRows) => None,
Err(e) => return Err(CommerceError::DatabaseError(e.to_string())),
}
};
match wo_data {
Some((
id_str,
work_order_number,
product_id,
bom_id,
work_center_id,
assigned_to,
status,
priority,
quantity_to_build,
quantity_completed,
scheduled_start,
scheduled_end,
actual_start,
actual_end,
notes,
created_at,
updated_at,
)) => {
let wo_id = parse_uuid(&id_str, "work_order", "id")?;
let tasks = self.load_tasks(wo_id)?;
let materials = self.load_materials(wo_id)?;
Ok(Some(WorkOrder {
id: wo_id,
work_order_number,
product_id: ProductId::from(parse_uuid(
&product_id,
"work_order",
"product_id",
)?),
bom_id: parse_uuid_opt(bom_id, "work_order", "bom_id")?,
work_center_id,
assigned_to: parse_uuid_opt(assigned_to, "work_order", "assigned_to")?,
status: parse_enum(&status, "work_order", "status")?,
priority: parse_enum(&priority, "work_order", "priority")?,
quantity_to_build: parse_decimal_strict(
&quantity_to_build,
"work_order",
"quantity_to_build",
)?,
quantity_completed: parse_decimal_strict(
&quantity_completed,
"work_order",
"quantity_completed",
)?,
scheduled_start: parse_datetime_opt(
scheduled_start,
"work_order",
"scheduled_start",
)?,
scheduled_end: parse_datetime_opt(
scheduled_end,
"work_order",
"scheduled_end",
)?,
actual_start: parse_datetime_opt(actual_start, "work_order", "actual_start")?,
actual_end: parse_datetime_opt(actual_end, "work_order", "actual_end")?,
notes,
tasks,
materials,
version: 1, created_at: parse_datetime(&created_at, "work_order", "created_at")?,
updated_at: parse_datetime(&updated_at, "work_order", "updated_at")?,
}))
}
None => Ok(None),
}
}
fn get_by_number(&self, work_order_number: &str) -> Result<Option<WorkOrder>> {
let id_result = {
let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
let result = conn.query_row(
"SELECT id FROM manufacturing_work_orders WHERE work_order_number = ?",
[work_order_number],
|row| row.get::<_, String>(0),
);
match result {
Ok(id_str) => Some(parse_uuid(&id_str, "work_order", "id")?),
Err(rusqlite::Error::QueryReturnedNoRows) => None,
Err(e) => return Err(CommerceError::DatabaseError(e.to_string())),
}
};
match id_result {
Some(id) => self.get(id),
None => Ok(None),
}
}
fn update(&self, id: Uuid, input: UpdateWorkOrder) -> Result<WorkOrder> {
let existing = self.get(id)?.ok_or(CommerceError::NotFound)?;
let now = Utc::now();
let new_status = input.status.unwrap_or(existing.status);
let new_priority = input.priority.unwrap_or(existing.priority);
let new_assigned_to = input.assigned_to.or(existing.assigned_to);
let new_notes = input.notes.or(existing.notes);
let new_work_center_id = input.work_center_id.or(existing.work_center_id);
let new_scheduled_start = input.scheduled_start.or(existing.scheduled_start);
let new_scheduled_end = input.scheduled_end.or(existing.scheduled_end);
{
let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
conn.execute(
"UPDATE manufacturing_work_orders SET status = ?, priority = ?, assigned_to = ?,
work_center_id = ?, scheduled_start = ?, scheduled_end = ?, notes = ?, updated_at = ? WHERE id = ?",
rusqlite::params![
new_status.to_string(),
new_priority.to_string(),
new_assigned_to.map(|u| u.to_string()),
new_work_center_id,
new_scheduled_start.map(|dt| dt.to_rfc3339()),
new_scheduled_end.map(|dt| dt.to_rfc3339()),
new_notes,
now.to_rfc3339(),
id.to_string(),
],
)
.map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
}
self.get(id)?.ok_or(CommerceError::NotFound)
}
fn list(&self, filter: WorkOrderFilter) -> Result<Vec<WorkOrder>> {
let ids = {
let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
let limit = i64::from(super::effective_limit(filter.limit));
let offset = i64::from(filter.offset.unwrap_or(0));
let mut sql = "SELECT id FROM manufacturing_work_orders WHERE 1=1".to_string();
let mut params: Vec<Box<dyn rusqlite::ToSql>> = Vec::new();
if let Some(product_id) = filter.product_id {
sql.push_str(" AND product_id = ?");
params.push(Box::new(product_id.to_string()));
}
if let Some(bom_id) = filter.bom_id {
sql.push_str(" AND bom_id = ?");
params.push(Box::new(bom_id.to_string()));
}
if let Some(status) = filter.status {
sql.push_str(" AND status = ?");
params.push(Box::new(status.to_string()));
}
if let Some(priority) = filter.priority {
sql.push_str(" AND priority = ?");
params.push(Box::new(priority.to_string()));
}
if let Some(assigned_to) = filter.assigned_to {
sql.push_str(" AND assigned_to = ?");
params.push(Box::new(assigned_to.to_string()));
}
if let Some(work_center_id) = filter.work_center_id {
sql.push_str(" AND work_center_id = ?");
params.push(Box::new(work_center_id));
}
if filter.overdue_only.unwrap_or(false) {
sql.push_str(" AND scheduled_end IS NOT NULL AND scheduled_end < ? AND status NOT IN ('completed', 'cancelled')");
params.push(Box::new(Utc::now().to_rfc3339()));
}
if let Some((cursor_created, cursor_id)) = &filter.after_cursor {
sql.push_str(" AND (created_at < ? OR (created_at = ? AND id < ?))");
params.push(Box::new(cursor_created.clone()));
params.push(Box::new(cursor_created.clone()));
params.push(Box::new(cursor_id.clone()));
}
sql.push_str(" ORDER BY created_at DESC, id DESC LIMIT ? OFFSET ?");
params.push(Box::new(limit));
params.push(Box::new(if filter.after_cursor.is_none() { offset } else { 0 }));
let mut stmt =
conn.prepare(&sql).map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
let param_refs: Vec<&dyn rusqlite::ToSql> =
params.iter().map(std::convert::AsRef::as_ref).collect();
let rows = stmt
.query_map(param_refs.as_slice(), |row| row.get::<_, String>(0))
.map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
let mut id_list = Vec::new();
for row in rows {
let id_str = row.map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
id_list.push(parse_uuid(&id_str, "work_order", "id")?);
}
id_list
};
let mut work_orders = Vec::new();
for id in ids {
if let Some(wo) = self.get(id)? {
work_orders.push(wo);
}
}
Ok(work_orders)
}
fn delete(&self, id: Uuid) -> Result<()> {
let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
conn.execute(
"UPDATE manufacturing_work_orders SET status = 'cancelled', updated_at = ? WHERE id = ?",
rusqlite::params![Utc::now().to_rfc3339(), id.to_string()],
)
.map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
Ok(())
}
fn start(&self, id: Uuid) -> Result<WorkOrder> {
let now = Utc::now();
{
let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
conn.execute(
"UPDATE manufacturing_work_orders SET status = 'in_progress', actual_start = ?, updated_at = ? WHERE id = ?",
rusqlite::params![now.to_rfc3339(), now.to_rfc3339(), id.to_string()],
)
.map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
}
self.get(id)?.ok_or(CommerceError::NotFound)
}
fn complete(&self, id: Uuid, quantity_completed: Decimal) -> Result<WorkOrder> {
if quantity_completed <= Decimal::ZERO {
return Err(CommerceError::ValidationError(
"Completed quantity must be greater than zero".to_string(),
));
}
let now = Utc::now();
let id_str = id.to_string();
with_immediate_transaction(&self.pool, |tx| {
let existing: (String, String, Option<String>) = tx
.query_row(
"SELECT quantity_completed, quantity_to_build, actual_end FROM manufacturing_work_orders WHERE id = ?",
[&id_str],
|row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
)
.map_err(|e| match e {
rusqlite::Error::QueryReturnedNoRows => {
rusqlite::Error::ToSqlConversionFailure(Box::new(CommerceError::NotFound))
}
other => other,
})?;
let existing_completed =
parse_decimal_strict(&existing.0, "work_order", "quantity_completed")
.map_err(|e| rusqlite::Error::ToSqlConversionFailure(Box::new(e)))?;
let quantity_to_build =
parse_decimal_strict(&existing.1, "work_order", "quantity_to_build")
.map_err(|e| rusqlite::Error::ToSqlConversionFailure(Box::new(e)))?;
let new_quantity_completed = existing_completed + quantity_completed;
let is_complete = new_quantity_completed >= quantity_to_build;
let new_status = if is_complete { "completed" } else { "partially_completed" };
let new_actual_end = if is_complete { Some(now.to_rfc3339()) } else { existing.2 };
tx.execute(
"UPDATE manufacturing_work_orders
SET quantity_completed = ?, status = ?, actual_end = ?, updated_at = ?
WHERE id = ?",
rusqlite::params![
new_quantity_completed.to_string(),
new_status,
new_actual_end,
now.to_rfc3339(),
id_str,
],
)?;
Ok(())
})?;
self.get(id)?.ok_or(CommerceError::NotFound)
}
fn hold(&self, id: Uuid) -> Result<WorkOrder> {
self.update(
id,
UpdateWorkOrder { status: Some(WorkOrderStatus::OnHold), ..Default::default() },
)
}
fn resume(&self, id: Uuid) -> Result<WorkOrder> {
self.update(
id,
UpdateWorkOrder { status: Some(WorkOrderStatus::InProgress), ..Default::default() },
)
}
fn cancel(&self, id: Uuid) -> Result<WorkOrder> {
self.update(
id,
UpdateWorkOrder { status: Some(WorkOrderStatus::Cancelled), ..Default::default() },
)
}
fn add_task(&self, work_order_id: Uuid, task: CreateWorkOrderTask) -> Result<WorkOrderTask> {
let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
let id = Uuid::new_v4();
let now = Utc::now();
let sequence = task.sequence.unwrap_or(1);
conn.execute(
"INSERT INTO manufacturing_work_order_tasks (id, work_order_id, sequence, task_name, status, estimated_hours, assigned_to, notes, created_at, updated_at)
VALUES (?, ?, ?, ?, 'pending', ?, ?, ?, ?, ?)",
rusqlite::params![
id.to_string(),
work_order_id.to_string(),
sequence,
task.task_name,
task.estimated_hours.map(|h| h.to_string()),
task.assigned_to.map(|u| u.to_string()),
task.notes,
now.to_rfc3339(),
now.to_rfc3339(),
],
)
.map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
Ok(WorkOrderTask {
id,
work_order_id,
sequence,
task_name: task.task_name,
status: TaskStatus::Pending,
estimated_hours: task.estimated_hours,
actual_hours: None,
assigned_to: task.assigned_to,
started_at: None,
completed_at: None,
notes: task.notes,
created_at: now,
updated_at: now,
})
}
fn update_task(&self, task_id: Uuid, task: UpdateWorkOrderTask) -> Result<WorkOrderTask> {
let existing = self.get_task_internal(task_id)?;
let now = Utc::now();
let new_sequence = task.sequence.unwrap_or(existing.sequence);
let new_task_name = task.task_name.unwrap_or(existing.task_name);
let new_status = task.status.unwrap_or(existing.status);
let new_estimated = task.estimated_hours.or(existing.estimated_hours);
let new_actual = task.actual_hours.or(existing.actual_hours);
let new_assigned = task.assigned_to.or(existing.assigned_to);
let new_started = task.started_at.or(existing.started_at);
let new_completed = task.completed_at.or(existing.completed_at);
let new_notes = task.notes.or(existing.notes);
{
let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
conn.execute(
"UPDATE manufacturing_work_order_tasks SET sequence = ?, task_name = ?, status = ?,
estimated_hours = ?, actual_hours = ?, assigned_to = ?, started_at = ?,
completed_at = ?, notes = ?, updated_at = ? WHERE id = ?",
rusqlite::params![
new_sequence,
new_task_name,
new_status.to_string(),
new_estimated.map(|h| h.to_string()),
new_actual.map(|h| h.to_string()),
new_assigned.map(|u| u.to_string()),
new_started.map(|dt| dt.to_rfc3339()),
new_completed.map(|dt| dt.to_rfc3339()),
new_notes,
now.to_rfc3339(),
task_id.to_string(),
],
)
.map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
}
self.get_task_internal(task_id)
}
fn remove_task(&self, task_id: Uuid) -> Result<()> {
let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
conn.execute(
"DELETE FROM manufacturing_work_order_tasks WHERE id = ?",
[task_id.to_string()],
)
.map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
Ok(())
}
fn get_tasks(&self, work_order_id: Uuid) -> Result<Vec<WorkOrderTask>> {
self.load_tasks(work_order_id)
}
fn start_task(&self, task_id: Uuid) -> Result<WorkOrderTask> {
let now = Utc::now();
{
let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
conn.execute(
"UPDATE manufacturing_work_order_tasks SET status = 'in_progress', started_at = ?, updated_at = ? WHERE id = ?",
rusqlite::params![now.to_rfc3339(), now.to_rfc3339(), task_id.to_string()],
)
.map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
}
self.get_task_internal(task_id)
}
fn complete_task(&self, task_id: Uuid, actual_hours: Option<Decimal>) -> Result<WorkOrderTask> {
let now = Utc::now();
{
let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
conn.execute(
"UPDATE manufacturing_work_order_tasks SET status = 'completed', actual_hours = ?, completed_at = ?, updated_at = ? WHERE id = ?",
rusqlite::params![
actual_hours.map(|h| h.to_string()),
now.to_rfc3339(),
now.to_rfc3339(),
task_id.to_string(),
],
)
.map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
}
self.get_task_internal(task_id)
}
fn add_material(
&self,
work_order_id: Uuid,
material: AddWorkOrderMaterial,
) -> Result<WorkOrderMaterial> {
let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
let id = Uuid::new_v4();
let now = Utc::now();
conn.execute(
"INSERT INTO manufacturing_work_order_materials (id, work_order_id, component_id, component_sku, component_name, reserved_quantity, consumed_quantity, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, '0', ?, ?)",
rusqlite::params![
id.to_string(),
work_order_id.to_string(),
material.component_id.map(|u| u.to_string()),
material.component_sku,
material.component_name,
material.quantity.to_string(),
now.to_rfc3339(),
now.to_rfc3339(),
],
)
.map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
Ok(WorkOrderMaterial {
id,
work_order_id,
component_id: material.component_id,
component_sku: material.component_sku,
component_name: material.component_name,
reserved_quantity: material.quantity,
consumed_quantity: Decimal::ZERO,
inventory_reservation_id: None,
created_at: now,
updated_at: now,
})
}
fn consume_material(&self, material_id: Uuid, quantity: Decimal) -> Result<WorkOrderMaterial> {
let existing = self.get_material_internal(material_id)?;
let now = Utc::now();
let new_consumed = existing.consumed_quantity + quantity;
{
let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
conn.execute(
"UPDATE manufacturing_work_order_materials SET consumed_quantity = ?, updated_at = ? WHERE id = ?",
rusqlite::params![
new_consumed.to_string(),
now.to_rfc3339(),
material_id.to_string(),
],
)
.map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
}
self.get_material_internal(material_id)
}
fn get_materials(&self, work_order_id: Uuid) -> Result<Vec<WorkOrderMaterial>> {
self.load_materials(work_order_id)
}
fn count(&self, filter: WorkOrderFilter) -> Result<u64> {
let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
let mut sql = "SELECT COUNT(*) FROM manufacturing_work_orders WHERE 1=1".to_string();
let mut params: Vec<Box<dyn rusqlite::ToSql>> = Vec::new();
if let Some(product_id) = filter.product_id {
sql.push_str(" AND product_id = ?");
params.push(Box::new(product_id.to_string()));
}
if let Some(bom_id) = filter.bom_id {
sql.push_str(" AND bom_id = ?");
params.push(Box::new(bom_id.to_string()));
}
if let Some(status) = filter.status {
sql.push_str(" AND status = ?");
params.push(Box::new(status.to_string()));
}
if let Some(priority) = filter.priority {
sql.push_str(" AND priority = ?");
params.push(Box::new(priority.to_string()));
}
if let Some(assigned_to) = filter.assigned_to {
sql.push_str(" AND assigned_to = ?");
params.push(Box::new(assigned_to.to_string()));
}
if let Some(work_center_id) = filter.work_center_id {
sql.push_str(" AND work_center_id = ?");
params.push(Box::new(work_center_id));
}
if filter.overdue_only.unwrap_or(false) {
sql.push_str(" AND scheduled_end IS NOT NULL AND scheduled_end < ? AND status NOT IN ('completed', 'cancelled')");
params.push(Box::new(Utc::now().to_rfc3339()));
}
let param_refs: Vec<&dyn rusqlite::ToSql> =
params.iter().map(std::convert::AsRef::as_ref).collect();
let count: i64 = conn
.query_row(&sql, param_refs.as_slice(), |row| row.get(0))
.map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
Ok(count as u64)
}
fn create_batch(&self, inputs: Vec<CreateWorkOrder>) -> Result<BatchResult<WorkOrder>> {
validate_batch_size(&inputs)?;
let mut result = BatchResult::with_capacity(inputs.len());
for (index, input) in inputs.into_iter().enumerate() {
match self.create(input) {
Ok(work_order) => result.record_success(work_order),
Err(e) => result.record_failure(index, None, &e),
}
}
Ok(result)
}
fn create_batch_atomic(&self, inputs: Vec<CreateWorkOrder>) -> Result<Vec<WorkOrder>> {
validate_batch_size(&inputs)?;
if inputs.is_empty() {
return Ok(vec![]);
}
let mut conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
let tx = super::begin_immediate(&mut conn).map_err(map_db_error)?;
let mut results = Vec::with_capacity(inputs.len());
for input in inputs {
let id = Uuid::new_v4();
let work_order_number = WorkOrder::generate_work_order_number();
let now = Utc::now();
let priority = input.priority.unwrap_or(WorkOrderPriority::Normal);
tx.execute(
"INSERT INTO manufacturing_work_orders (id, work_order_number, product_id, bom_id, work_center_id,
assigned_to, status, priority, quantity_to_build, quantity_completed, scheduled_start, scheduled_end, notes, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, 'planned', ?, ?, '0', ?, ?, ?, ?, ?)",
rusqlite::params![
id.to_string(),
work_order_number,
input.product_id.to_string(),
input.bom_id.map(|u| u.to_string()),
input.work_center_id,
input.assigned_to.map(|u| u.to_string()),
priority.to_string(),
input.quantity_to_build.to_string(),
input.scheduled_start.map(|dt| dt.to_rfc3339()),
input.scheduled_end.map(|dt| dt.to_rfc3339()),
input.notes,
now.to_rfc3339(),
now.to_rfc3339(),
],
)
.map_err(map_db_error)?;
let mut tasks = Vec::new();
if let Some(task_inputs) = input.tasks {
for task_input in task_inputs {
let task_id = Uuid::new_v4();
let sequence = task_input.sequence.unwrap_or(1);
tx.execute(
"INSERT INTO manufacturing_work_order_tasks (id, work_order_id, sequence, task_name, status, estimated_hours, assigned_to, notes, created_at, updated_at)
VALUES (?, ?, ?, ?, 'pending', ?, ?, ?, ?, ?)",
rusqlite::params![
task_id.to_string(),
id.to_string(),
sequence,
task_input.task_name,
task_input.estimated_hours.map(|h| h.to_string()),
task_input.assigned_to.map(|u| u.to_string()),
task_input.notes,
now.to_rfc3339(),
now.to_rfc3339(),
],
)
.map_err(map_db_error)?;
tasks.push(WorkOrderTask {
id: task_id,
work_order_id: id,
sequence,
task_name: task_input.task_name,
status: TaskStatus::Pending,
estimated_hours: task_input.estimated_hours,
actual_hours: None,
assigned_to: task_input.assigned_to,
started_at: None,
completed_at: None,
notes: task_input.notes,
created_at: now,
updated_at: now,
});
}
}
results.push(WorkOrder {
id,
work_order_number,
product_id: input.product_id,
bom_id: input.bom_id,
work_center_id: input.work_center_id,
assigned_to: input.assigned_to,
status: WorkOrderStatus::Planned,
priority,
quantity_to_build: input.quantity_to_build,
quantity_completed: Decimal::ZERO,
scheduled_start: input.scheduled_start,
scheduled_end: input.scheduled_end,
actual_start: None,
actual_end: None,
notes: input.notes,
tasks,
materials: vec![],
version: 1,
created_at: now,
updated_at: now,
});
}
tx.commit().map_err(map_db_error)?;
Ok(results)
}
fn update_batch(
&self,
updates: Vec<(Uuid, UpdateWorkOrder)>,
) -> Result<BatchResult<WorkOrder>> {
validate_batch_size(&updates)?;
let mut result = BatchResult::with_capacity(updates.len());
for (index, (id, input)) in updates.into_iter().enumerate() {
match self.update(id, input) {
Ok(work_order) => result.record_success(work_order),
Err(e) => result.record_failure(index, Some(id.to_string()), &e),
}
}
Ok(result)
}
fn update_batch_atomic(&self, updates: Vec<(Uuid, UpdateWorkOrder)>) -> Result<Vec<WorkOrder>> {
validate_batch_size(&updates)?;
if updates.is_empty() {
return Ok(vec![]);
}
let mut conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
let tx = super::begin_immediate(&mut conn).map_err(map_db_error)?;
let mut updated_ids = Vec::with_capacity(updates.len());
for (id, input) in updates {
let now = Utc::now();
let existing: (String, String, Option<String>, Option<String>, Option<String>, Option<String>) = tx
.query_row(
"SELECT status, priority, assigned_to, work_center_id, scheduled_start, scheduled_end
FROM manufacturing_work_orders WHERE id = ?",
[id.to_string()],
|row| Ok((
row.get(0)?,
row.get(1)?,
row.get(2)?,
row.get(3)?,
row.get(4)?,
row.get(5)?,
)),
)
.map_err(|e| match e {
rusqlite::Error::QueryReturnedNoRows => CommerceError::NotFound,
_ => map_db_error(e),
})?;
let new_status = input.status.map(|s| s.to_string()).unwrap_or(existing.0);
let new_priority = input.priority.map(|p| p.to_string()).unwrap_or(existing.1);
let new_assigned_to =
input.assigned_to.map(|u| Some(u.to_string())).unwrap_or(existing.2);
let new_work_center_id = input.work_center_id.or(existing.3);
let new_scheduled_start =
input.scheduled_start.map(|dt| Some(dt.to_rfc3339())).unwrap_or(existing.4);
let new_scheduled_end =
input.scheduled_end.map(|dt| Some(dt.to_rfc3339())).unwrap_or(existing.5);
let new_notes = input.notes;
tx.execute(
"UPDATE manufacturing_work_orders SET status = ?, priority = ?, assigned_to = ?,
work_center_id = ?, scheduled_start = ?, scheduled_end = ?, notes = COALESCE(?, notes), updated_at = ? WHERE id = ?",
rusqlite::params![
new_status,
new_priority,
new_assigned_to,
new_work_center_id,
new_scheduled_start,
new_scheduled_end,
new_notes,
now.to_rfc3339(),
id.to_string(),
],
)
.map_err(map_db_error)?;
updated_ids.push(id);
}
tx.commit().map_err(map_db_error)?;
let mut results = Vec::with_capacity(updated_ids.len());
for id in updated_ids {
if let Some(wo) = self.get(id)? {
results.push(wo);
}
}
Ok(results)
}
fn delete_batch(&self, ids: Vec<Uuid>) -> Result<BatchResult<Uuid>> {
validate_batch_size(&ids)?;
let mut result = BatchResult::with_capacity(ids.len());
for (index, id) in ids.into_iter().enumerate() {
match self.delete(id) {
Ok(()) => result.record_success(id),
Err(e) => result.record_failure(index, Some(id.to_string()), &e),
}
}
Ok(result)
}
fn delete_batch_atomic(&self, ids: Vec<Uuid>) -> Result<()> {
validate_batch_size(&ids)?;
if ids.is_empty() {
return Ok(());
}
let mut conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
let tx = super::begin_immediate(&mut conn).map_err(map_db_error)?;
let placeholders = build_in_clause(ids.len());
let now = Utc::now().to_rfc3339();
let mut all_params: Vec<Box<dyn rusqlite::ToSql>> = vec![Box::new(now)];
all_params.extend(uuid_params(&ids));
let all_params_refs = params_refs(&all_params);
let sql = format!(
"UPDATE manufacturing_work_orders SET status = 'cancelled', updated_at = ? WHERE id IN ({placeholders})"
);
tx.execute(&sql, all_params_refs.as_slice()).map_err(map_db_error)?;
tx.commit().map_err(map_db_error)?;
Ok(())
}
fn get_batch(&self, ids: Vec<Uuid>) -> Result<Vec<WorkOrder>> {
validate_batch_size(&ids)?;
if ids.is_empty() {
return Ok(vec![]);
}
let work_order_data = {
let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
let placeholders = build_in_clause(ids.len());
let sql = format!(
"SELECT id, work_order_number, product_id, bom_id, work_center_id, assigned_to,
status, priority, quantity_to_build, quantity_completed, scheduled_start,
scheduled_end, actual_start, actual_end, notes, created_at, updated_at
FROM manufacturing_work_orders WHERE id IN ({placeholders})"
);
let params = uuid_params(&ids);
let param_refs = params_refs(¶ms);
let mut stmt = conn.prepare(&sql).map_err(map_db_error)?;
let rows = stmt
.query_map(param_refs.as_slice(), |row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, String>(2)?,
row.get::<_, Option<String>>(3)?,
row.get::<_, Option<String>>(4)?,
row.get::<_, Option<String>>(5)?,
row.get::<_, String>(6)?,
row.get::<_, String>(7)?,
row.get::<_, String>(8)?,
row.get::<_, String>(9)?,
row.get::<_, Option<String>>(10)?,
row.get::<_, Option<String>>(11)?,
row.get::<_, Option<String>>(12)?,
row.get::<_, Option<String>>(13)?,
row.get::<_, Option<String>>(14)?,
row.get::<_, String>(15)?,
row.get::<_, String>(16)?,
))
})
.map_err(map_db_error)?;
let mut data = Vec::new();
for row in rows {
data.push(row.map_err(map_db_error)?);
}
data
};
let mut work_orders = Vec::with_capacity(work_order_data.len());
for (
id_str,
work_order_number,
product_id,
bom_id,
work_center_id,
assigned_to,
status,
priority,
quantity_to_build,
quantity_completed,
scheduled_start,
scheduled_end,
actual_start,
actual_end,
notes,
created_at,
updated_at,
) in work_order_data
{
let wo_id = parse_uuid(&id_str, "work_order", "id")?;
let tasks = self.load_tasks(wo_id)?;
let materials = self.load_materials(wo_id)?;
work_orders.push(WorkOrder {
id: wo_id,
work_order_number,
product_id: ProductId::from(parse_uuid(&product_id, "work_order", "product_id")?),
bom_id: parse_uuid_opt(bom_id, "work_order", "bom_id")?,
work_center_id,
assigned_to: parse_uuid_opt(assigned_to, "work_order", "assigned_to")?,
status: parse_enum(&status, "work_order", "status")?,
priority: parse_enum(&priority, "work_order", "priority")?,
quantity_to_build: parse_decimal_strict(
&quantity_to_build,
"work_order",
"quantity_to_build",
)?,
quantity_completed: parse_decimal_strict(
&quantity_completed,
"work_order",
"quantity_completed",
)?,
scheduled_start: parse_datetime_opt(
scheduled_start,
"work_order",
"scheduled_start",
)?,
scheduled_end: parse_datetime_opt(scheduled_end, "work_order", "scheduled_end")?,
actual_start: parse_datetime_opt(actual_start, "work_order", "actual_start")?,
actual_end: parse_datetime_opt(actual_end, "work_order", "actual_end")?,
notes,
tasks,
materials,
version: 1,
created_at: parse_datetime(&created_at, "work_order", "created_at")?,
updated_at: parse_datetime(&updated_at, "work_order", "updated_at")?,
});
}
Ok(work_orders)
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::SqliteDatabase;
use rust_decimal_macros::dec;
use stateset_core::{
AddWorkOrderMaterial, CreateWorkOrder, CreateWorkOrderTask, ProductId, TaskStatus,
WorkOrderFilter, WorkOrderRepository, WorkOrderStatus,
};
fn fresh_repo() -> SqliteWorkOrderRepository {
SqliteDatabase::in_memory().expect("in-memory").work_orders()
}
fn make_wo(repo: &SqliteWorkOrderRepository, qty: Decimal) -> WorkOrder {
repo.create(CreateWorkOrder {
product_id: ProductId::new(),
quantity_to_build: qty,
..Default::default()
})
.expect("create wo")
}
#[test]
fn create_wo_starts_in_planned() {
let repo = fresh_repo();
let wo = make_wo(&repo, dec!(10));
assert_eq!(wo.status, WorkOrderStatus::Planned);
assert_eq!(wo.quantity_to_build, dec!(10));
assert!(!wo.work_order_number.is_empty());
}
#[test]
fn create_wo_with_tasks_persists_them() {
let repo = fresh_repo();
let wo = repo
.create(CreateWorkOrder {
product_id: ProductId::new(),
quantity_to_build: dec!(5),
tasks: Some(vec![
CreateWorkOrderTask {
sequence: Some(1),
task_name: "Assembly".into(),
estimated_hours: Some(dec!(2)),
assigned_to: None,
notes: None,
},
CreateWorkOrderTask {
sequence: Some(2),
task_name: "QA".into(),
estimated_hours: Some(dec!(1)),
assigned_to: None,
notes: None,
},
]),
..Default::default()
})
.expect("create");
let tasks = repo.get_tasks(wo.id).expect("tasks");
assert_eq!(tasks.len(), 2);
}
#[test]
fn get_and_get_by_number_round_trip() {
let repo = fresh_repo();
let wo = make_wo(&repo, dec!(1));
let by_id = repo.get(wo.id).expect("ok").expect("found");
assert_eq!(by_id.id, wo.id);
let by_num = repo.get_by_number(&wo.work_order_number).expect("ok").expect("found");
assert_eq!(by_num.id, wo.id);
assert!(repo.get_by_number("missing").expect("ok").is_none());
}
#[test]
fn start_transitions_to_in_progress() {
let repo = fresh_repo();
let wo = make_wo(&repo, dec!(1));
let started = repo.start(wo.id).expect("start");
assert_eq!(started.status, WorkOrderStatus::InProgress);
}
#[test]
fn complete_full_quantity_marks_completed() {
let repo = fresh_repo();
let wo = make_wo(&repo, dec!(5));
repo.start(wo.id).expect("start");
let done = repo.complete(wo.id, dec!(5)).expect("complete");
assert_eq!(done.status, WorkOrderStatus::Completed);
assert_eq!(done.quantity_completed, dec!(5));
}
#[test]
fn complete_partial_quantity_marks_partially_completed() {
let repo = fresh_repo();
let wo = make_wo(&repo, dec!(10));
repo.start(wo.id).expect("start");
let partial = repo.complete(wo.id, dec!(7)).expect("partial");
assert_eq!(partial.status, WorkOrderStatus::PartiallyCompleted);
assert_eq!(partial.quantity_completed, dec!(7));
}
#[test]
fn cancel_transitions_to_cancelled() {
let repo = fresh_repo();
let wo = make_wo(&repo, dec!(1));
let cancelled = repo.cancel(wo.id).expect("cancel");
assert_eq!(cancelled.status, WorkOrderStatus::Cancelled);
}
#[test]
fn list_filters_by_status() {
let repo = fresh_repo();
let planned = make_wo(&repo, dec!(1));
let in_progress = make_wo(&repo, dec!(1));
repo.start(in_progress.id).expect("start");
let pl = repo
.list(WorkOrderFilter { status: Some(WorkOrderStatus::Planned), ..Default::default() })
.expect("planned");
let ip = repo
.list(WorkOrderFilter {
status: Some(WorkOrderStatus::InProgress),
..Default::default()
})
.expect("in_progress");
assert!(pl.iter().any(|w| w.id == planned.id));
assert!(ip.iter().any(|w| w.id == in_progress.id));
}
#[test]
fn add_task_appends_to_work_order() {
let repo = fresh_repo();
let wo = make_wo(&repo, dec!(1));
let task = repo
.add_task(
wo.id,
CreateWorkOrderTask {
sequence: Some(1),
task_name: "Solder".into(),
estimated_hours: Some(dec!(0.5)),
assigned_to: None,
notes: None,
},
)
.expect("add task");
assert_eq!(task.task_name, "Solder");
let tasks = repo.get_tasks(wo.id).expect("tasks");
assert_eq!(tasks.len(), 1);
}
#[test]
fn start_and_complete_task_transitions() {
let repo = fresh_repo();
let wo = make_wo(&repo, dec!(1));
let task = repo
.add_task(
wo.id,
CreateWorkOrderTask {
sequence: Some(1),
task_name: "Wash".into(),
estimated_hours: Some(dec!(1)),
assigned_to: None,
notes: None,
},
)
.expect("add");
let started = repo.start_task(task.id).expect("start");
assert_eq!(started.status, TaskStatus::InProgress);
let completed = repo.complete_task(task.id, Some(dec!(1.5))).expect("complete");
assert_eq!(completed.status, TaskStatus::Completed);
}
#[test]
fn add_and_consume_material() {
let repo = fresh_repo();
let wo = make_wo(&repo, dec!(1));
let mat = repo
.add_material(
wo.id,
AddWorkOrderMaterial {
component_id: None,
component_sku: "PART-A".into(),
component_name: "Resistor".into(),
quantity: dec!(10),
},
)
.expect("add mat");
assert_eq!(mat.component_sku, "PART-A");
let consumed = repo.consume_material(mat.id, dec!(4)).expect("consume");
assert_eq!(consumed.id, mat.id);
let remaining_materials = repo.get_materials(wo.id).expect("materials");
assert_eq!(remaining_materials.len(), 1);
}
#[test]
fn create_batch_returns_per_input_results() {
let repo = fresh_repo();
let result = repo
.create_batch(vec![
CreateWorkOrder {
product_id: ProductId::new(),
quantity_to_build: dec!(1),
..Default::default()
},
CreateWorkOrder {
product_id: ProductId::new(),
quantity_to_build: dec!(2),
..Default::default()
},
])
.expect("batch");
assert_eq!(result.success_count, 2);
assert_eq!(result.failure_count, 0);
}
#[test]
fn get_unknown_id_returns_none() {
let repo = fresh_repo();
assert!(repo.get(Uuid::new_v4()).expect("ok").is_none());
}
#[test]
fn get_batch_returns_only_existing() {
let repo = fresh_repo();
let w1 = make_wo(&repo, dec!(1));
let w2 = make_wo(&repo, dec!(2));
let stranger = Uuid::new_v4();
let fetched = repo.get_batch(vec![w1.id, w2.id, stranger]).expect("ok");
assert_eq!(fetched.len(), 2);
}
#[test]
fn concurrent_completions_are_not_lost() {
use std::sync::{Arc, Barrier};
use std::thread;
let db = Arc::new(SqliteDatabase::in_memory().expect("in-memory"));
let wo = db
.work_orders()
.create(CreateWorkOrder {
product_id: ProductId::new(),
quantity_to_build: dec!(1000),
..Default::default()
})
.expect("create wo");
let thread_count = 10usize;
let barrier = Arc::new(Barrier::new(thread_count));
let mut handles = Vec::new();
for _ in 0..thread_count {
let db = Arc::clone(&db);
let barrier = Arc::clone(&barrier);
let id = wo.id;
handles.push(thread::spawn(move || {
let repo = db.work_orders();
barrier.wait();
repo.complete(id, dec!(1))
}));
}
let results: Vec<_> = handles.into_iter().map(|h| h.join().expect("thread")).collect();
assert!(
results
.iter()
.all(|r| r.is_ok() || format!("{:?}", r.as_ref().unwrap_err()).contains("locked")),
"unexpected non-lock failure: {results:?}"
);
let fetched = db.work_orders().get(wo.id).expect("get").expect("found");
assert_eq!(
fetched.quantity_completed,
Decimal::from(thread_count as u64),
"completions were lost to a race: {results:?}"
);
}
#[test]
fn list_after_cursor_paginates_without_overlap() {
let repo = fresh_repo();
for _ in 0..3 {
make_wo(&repo, dec!(1));
}
let all = repo.list(WorkOrderFilter::default()).expect("list all");
assert_eq!(all.len(), 3);
let first_page =
repo.list(WorkOrderFilter { limit: Some(2), ..Default::default() }).expect("page 1");
assert_eq!(first_page.len(), 2);
assert_eq!(first_page[0].id, all[0].id);
let last = &first_page[1];
let second_page = repo
.list(WorkOrderFilter {
after_cursor: Some((last.created_at.to_rfc3339(), last.id.to_string())),
..Default::default()
})
.expect("page 2");
assert_eq!(second_page.len(), 1);
assert_eq!(second_page[0].id, all[2].id);
}
}