use rust_decimal::Decimal;
use uuid::Uuid;
use crate::infrastructure::persistence::{NewDraftInvoiceRow, NewInvoiceItemRow, NewTenderRow};
use super::pos_compute::ComputeLineIn;
use super::pos_events::{PosEvent, PosTenderCompleted};
use super::pos_ports::{BillingPort, PaymentPort, PosTaxComputePort, PosTaxDocumentType};
use super::pos_write_service::{
legacy_company_echo, map_invoice_dup, money, relay_ambient_scope, NewSyncSale, PosError,
PosWriteService, SyncAction, SyncOutcome, TicketTotals,
};
impl PosWriteService {
pub async fn sync_from_ui(
&self,
s: NewSyncSale,
tax: &dyn PosTaxComputePort,
billing: &dyn BillingPort,
payment: &dyn PaymentPort,
) -> Result<SyncOutcome, PosError> {
if let Some(existing) = self.invoices.find_by_client_uuid(&self.rpool(), s.client_uuid).await? {
if existing.status != "draft" {
return Ok(finalized(&existing));
}
return self.sync_update_draft(s, existing, tax).await;
}
if let Some(parent_uuid) = s.refund_of_client_uuid {
let manager = s.manager.as_ref().ok_or(PosError::ManagerAuthRequired)?;
self.verify_manager_internal(manager, s.source_ip.as_deref()).await?;
let parent = self.invoices
.find_by_client_uuid(&self.rpool(), parent_uuid)
.await?
.ok_or(PosError::RefundParentNotFound(parent_uuid))?;
let out = self
.return_sale(parent.id, billing, payment, None)
.await?;
let action = if parent.status == "paid" { SyncAction::Created } else { SyncAction::ReplayFinalized };
return Ok(SyncOutcome {
pos_invoice_id: out.return_ticket_id,
action,
totals: TicketTotals {
net_total: parent.net_total,
tax_total: parent.tax_total,
grand_total: parent.grand_total,
rounding_adjustment: Decimal::ZERO,
rounded_total: parent.rounded_total,
paid_total: parent.rounded_total,
change_due: Decimal::ZERO,
},
});
}
self.sync_create_draft(s, tax).await
}
async fn sync_create_draft(&self, s: NewSyncSale, tax: &dyn PosTaxComputePort) -> Result<SyncOutcome, PosError> {
let mut tx = self.rpool().begin().await?;
relay_ambient_scope(&mut tx).await?;
let session = self.resolve_session_for_create(&mut *tx, &s).await?;
self.validate_tender_methods(&mut *tx, &s).await?;
if let Some(table) = s.pos_table_id {
if self.tables.fetch_table(&mut *tx, table).await?.is_none() {
return Err(PosError::TableNotFound(table));
}
if let Some(occupant) = self
.invoices
.find_draft_on_table(&mut *tx, table, None)
.await?
{
return Err(PosError::TableOccupied { pos_table_id: table, draft_invoice_id: occupant });
}
}
let order_discount = self
.resolve_order_discount(&mut tx, s.pos_profile_id, s.discount_id)
.await?;
let mut compute_lines: Vec<ComputeLineIn> = s
.lines
.iter()
.map(|l| ComputeLineIn {
item_id: l.item_id,
revenue_account_id: l.revenue_account_id,
description: l.description.clone(),
course: l.course,
quantity: l.quantity,
unit_price: l.unit_price,
discount_amount: l.discount_amount,
})
.collect();
if let Some(d) = &order_discount {
Self::fold_order_discount(&mut compute_lines, d.percentage)?;
}
let computed = self
.compute_ticket(
&mut *tx,
s.pos_profile_id,
s.posting_at.date(),
PosTaxDocumentType::Invoice,
compute_lines,
tax,
)
.await?;
let paid = s.tenders.iter().map(|t| t.amount).sum::<Decimal>();
let totals = Self::totals_with_tenders(&computed, paid);
let receipt_number = format!(
"SYNC-{}-{}",
&legacy_company_echo().simple().to_string()[..4],
&s.client_uuid.simple().to_string()[..12]
);
let id = Uuid::new_v4();
let r = self.invoices.insert_draft(&mut tx, &NewDraftInvoiceRow {
id,
client_uuid: Some(s.client_uuid),
pos_profile_id: s.pos_profile_id,
opening_entry_id: session,
branch_id: s.branch_id,
customer_id: s.customer_id,
pos_table_id: s.pos_table_id,
receipt_number: &receipt_number,
posting_at: s.posting_at,
net_total: computed.net_total,
tax_total: computed.tax_total,
grand_total: computed.grand_total,
rounding_adjustment: computed.rounding_adjustment,
rounded_total: computed.rounded_total,
}).await;
if let Err(e) = r {
return Err(map_invoice_dup(
e, &receipt_number, Some(s.client_uuid), s.pos_table_id,
&self.invoices, &self.rpool(), None,
).await);
}
check_replay_change_is_cash(&s, computed.rounded_total)?;
self.sync_write_children(&mut tx, id, &computed, &s).await?;
let paid_total = self.payments.sum_paid_total_on(&mut tx, id).await?;
let change_due = if paid_total > computed.rounded_total {
paid_total - computed.rounded_total
} else {
Decimal::ZERO
};
self.invoices.update_tender_totals(&mut tx, id, paid_total, change_due).await?;
let crossed = crosses_full_payment(false, computed.rounded_total, paid_total);
if crossed {
self.stage_tender_completed_in_tx(&mut tx, id).await?;
}
tx.commit().await?;
if crossed {
self.emit_tender_completed(id);
}
Ok(SyncOutcome {
pos_invoice_id: id,
action: SyncAction::Created,
totals: TicketTotals { paid_total, change_due, ..totals },
})
}
async fn sync_update_draft(
&self,
s: NewSyncSale,
existing: crate::infrastructure::persistence::SyncLookupRow,
tax: &dyn PosTaxComputePort,
) -> Result<SyncOutcome, PosError> {
let mut tx = self.rpool().begin().await?;
relay_ambient_scope(&mut tx).await?;
let held = self.invoices
.lock_for_sync_on(&mut tx, existing.id).await?
.ok_or(PosError::InvoiceNotFound(existing.id))?;
if held.status != "draft" {
drop(tx);
let now = self.invoices
.find_by_client_uuid(&self.rpool(), s.client_uuid).await?
.ok_or(PosError::InvoiceNotFound(existing.id))?;
return Ok(finalized(&now));
}
let was_full = held.rounded_total > Decimal::ZERO && held.paid_total >= held.rounded_total;
if existing.customer_id != s.customer_id {
return Err(PosError::SyncPartnerMismatch);
}
if existing.is_return || existing.return_against.is_some() || s.refund_of_client_uuid.is_some() {
return Err(PosError::RefundLineageConflict);
}
let session = self.resolve_session_for_update(&mut *tx, &s, &existing).await?;
self.validate_tender_methods(&mut *tx, &s).await?;
if let Some(table) = s.pos_table_id {
if self.tables.fetch_table(&mut *tx, table).await?.is_none() {
return Err(PosError::TableNotFound(table));
}
if Some(table) != existing.pos_table_id {
if let Some(occupant) = self
.invoices
.find_draft_on_table(&mut *tx, table, Some(existing.id))
.await?
{
return Err(PosError::TableOccupied { pos_table_id: table, draft_invoice_id: occupant });
}
}
}
let order_discount = self
.resolve_order_discount(&mut tx, existing.pos_profile_id, s.discount_id)
.await?;
let mut compute_lines: Vec<ComputeLineIn> = s
.lines
.iter()
.map(|l| ComputeLineIn {
item_id: l.item_id,
revenue_account_id: l.revenue_account_id,
description: l.description.clone(),
course: l.course,
quantity: l.quantity,
unit_price: l.unit_price,
discount_amount: l.discount_amount,
})
.collect();
if let Some(d) = &order_discount {
Self::fold_order_discount(&mut compute_lines, d.percentage)?;
}
let computed = self
.compute_ticket(
&mut *tx,
existing.pos_profile_id,
s.posting_at.date(),
PosTaxDocumentType::Invoice,
compute_lines,
tax,
)
.await?;
let paid = s.tenders.iter().map(|t| t.amount).sum::<Decimal>();
let totals = Self::totals_with_tenders(&computed, paid);
check_replay_change_is_cash(&s, computed.rounded_total)?;
let id = existing.id;
let now = chrono::Utc::now();
self.items.soft_delete_lines_for_ticket(&mut tx, id, now).await?;
self.payments.soft_delete_tenders_for_ticket(&mut tx, id, now).await?;
let r = self.invoices
.update_draft_from_sync(
&mut tx, id, session, s.branch_id, s.customer_id, s.pos_table_id,
computed.net_total, computed.tax_total, computed.grand_total,
computed.rounding_adjustment, computed.rounded_total,
totals.paid_total, totals.change_due,
)
.await;
match r {
Ok(1) => {}
Ok(_) => return Err(PosError::NotDraft),
Err(e) => {
return Err(map_invoice_dup(
e, "", Some(s.client_uuid), s.pos_table_id,
&self.invoices, &self.rpool(), Some(id),
).await);
}
}
self.sync_write_children(&mut tx, id, &computed, &s).await?;
let crossed = crosses_full_payment(was_full, computed.rounded_total, totals.paid_total);
if crossed {
self.stage_tender_completed_in_tx(&mut tx, id).await?;
}
tx.commit().await?;
if crossed {
self.emit_tender_completed(id);
}
Ok(SyncOutcome { pos_invoice_id: id, action: SyncAction::Updated, totals })
}
async fn sync_write_children(
&self,
tx: &mut sqlx::PgConnection,
pos_invoice_id: Uuid,
computed: &super::pos_compute::ComputedTicket,
s: &NewSyncSale,
) -> Result<(), PosError> {
for (l, src) in computed.lines.iter().zip(&s.lines) {
self.items.insert_line(tx, &NewInvoiceItemRow {
id: Uuid::new_v4(),
pos_invoice_id,
client_uuid: Some(src.client_uuid),
item_id: l.item_id,
description: l.description.as_deref(),
course: l.course,
quantity: l.quantity,
unit_price: l.unit_price,
discount_amount: l.discount_amount,
net_amount: l.net_amount,
revenue_account_id: l.revenue_account_id,
}).await?;
}
for t in &s.tenders {
if t.amount <= Decimal::ZERO {
return Err(PosError::NegativeAmount);
}
self.payments.insert_tender(tx, &NewTenderRow {
id: Uuid::new_v4(),
pos_invoice_id,
client_uuid: Some(t.client_uuid),
payment_method: &t.method,
amount: money(t.amount),
reference_no: t.reference_no.as_deref(),
}).await?;
}
Ok(())
}
async fn stage_tender_completed_in_tx(
&self,
tx: &mut sqlx::PgConnection,
pos_invoice_id: Uuid,
) -> Result<(), PosError> {
if let Some(schema) = &self.outbox_schema {
let rec = super::pos_tender::tender_completed_outbox_record(pos_invoice_id, legacy_company_echo(), chrono::Utc::now());
backbone_outbox::outbox::stage(&mut *tx, schema, &rec)
.await
.map_err(|e| PosError::Db(sqlx::Error::Protocol(e.to_string())))?;
}
Ok(())
}
fn emit_tender_completed(&self, pos_invoice_id: Uuid) {
self.sink.publish(PosEvent::PosTenderCompleted(PosTenderCompleted {
pos_invoice_id, company_id: legacy_company_echo(),
}));
}
async fn resolve_session_for_create(&self, conn: &mut sqlx::PgConnection, s: &NewSyncSale) -> Result<Uuid, PosError> {
let named = self.openings
.fetch_status_on(&mut *conn, s.opening_entry_id)
.await?
.ok_or(PosError::SessionNotFound(s.opening_entry_id))?;
if named == "open" {
return self.same_register_open_session(&mut *conn, s.opening_entry_id, s.pos_profile_id).await;
}
if let Some(rescue) = s.rescue_opening_entry_id {
return self.same_register_open_session(&mut *conn, rescue, s.pos_profile_id).await;
}
Err(PosError::SessionClosedRescueRequired(s.opening_entry_id))
}
async fn resolve_session_for_update(
&self,
conn: &mut sqlx::PgConnection,
s: &NewSyncSale,
existing: &crate::infrastructure::persistence::SyncLookupRow,
) -> Result<Uuid, PosError> {
if s.opening_entry_id == existing.opening_entry_id {
let st = self.openings
.fetch_status_on(&mut *conn, existing.opening_entry_id)
.await?
.ok_or(PosError::SessionNotFound(existing.opening_entry_id))?;
if st != "open" {
if let Some(rescue) = s.rescue_opening_entry_id {
return self.same_register_open_session(&mut *conn, rescue, existing.pos_profile_id).await;
}
return Err(PosError::SessionClosedRescueRequired(existing.opening_entry_id));
}
return Ok(existing.opening_entry_id);
}
let orig = self.openings
.fetch_status_on(&mut *conn, existing.opening_entry_id)
.await?
.ok_or(PosError::SessionNotFound(existing.opening_entry_id))?;
if orig == "open" {
return Err(PosError::SyncSessionMismatch);
}
let target = s.rescue_opening_entry_id.unwrap_or(s.opening_entry_id);
self.same_register_open_session(&mut *conn, target, existing.pos_profile_id).await
}
async fn same_register_open_session(
&self,
conn: &mut sqlx::PgConnection,
session: Uuid,
pos_profile_id: Uuid,
) -> Result<Uuid, PosError> {
let st = self.openings
.fetch_status_on(&mut *conn, session)
.await?
.ok_or(PosError::SessionNotFound(session))?;
if st != "open" {
return Err(PosError::SessionClosedRescueRequired(session));
}
let profile = self.openings
.fetch_profile_id_on(&mut *conn, session)
.await?
.ok_or(PosError::SessionNotFound(session))?;
if profile != pos_profile_id {
return Err(PosError::SessionRegisterMismatch);
}
Ok(session)
}
async fn validate_tender_methods(&self, conn: &mut sqlx::PgConnection, s: &NewSyncSale) -> Result<(), PosError> {
let valid = self.payments.valid_methods(&mut *conn).await?;
for t in &s.tenders {
if !valid.iter().any(|m| m == &t.method) {
return Err(PosError::InvalidTenderMethod(t.method.clone()));
}
}
Ok(())
}
}
fn finalized(t: &crate::infrastructure::persistence::SyncLookupRow) -> SyncOutcome {
SyncOutcome {
pos_invoice_id: t.id,
action: SyncAction::ReplayFinalized,
totals: TicketTotals {
net_total: t.net_total,
tax_total: t.tax_total,
grand_total: t.grand_total,
rounding_adjustment: t.rounding_adjustment,
rounded_total: t.rounded_total,
paid_total: t.paid_total,
change_due: t.change_due,
},
}
}
fn crosses_full_payment(was_full: bool, rounded_total: Decimal, paid_total: Decimal) -> bool {
!was_full && rounded_total > Decimal::ZERO && paid_total >= rounded_total
}
fn check_replay_change_is_cash(s: &NewSyncSale, rounded_total: Decimal) -> Result<(), PosError> {
let non_cash = s.tenders.iter().filter(|t| t.method != "cash").map(|t| money(t.amount)).sum::<Decimal>();
if non_cash > rounded_total.max(Decimal::ZERO) {
let method = s.tenders.iter().find(|t| t.method != "cash").map(|t| t.method.clone()).unwrap_or_default();
return Err(PosError::NonCashOverpayment { method, remaining: rounded_total.max(Decimal::ZERO) });
}
Ok(())
}
#[cfg(test)]
mod crossing_tests {
use super::crosses_full_payment;
use rust_decimal::Decimal;
#[test]
fn only_the_replay_that_completes_payment_triggers_recognition() {
let d = |s: &str| s.parse::<Decimal>().unwrap();
assert!(crosses_full_payment(false, d("100"), d("100")));
assert!(!crosses_full_payment(true, d("100"), d("120")), "already paid up: no re-trigger");
assert!(!crosses_full_payment(false, d("100"), d("99.99")));
assert!(!crosses_full_payment(false, d("0"), d("0")), "a zero ticket is not a sale to recognise");
}
}