""" Repository for Bill and invoice sequence operations. """ from datetime import UTC, datetime from decimal import Decimal from uuid import UUID from sqlalchemy import and_, func, or_, select, text from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload from app.models.billing import Bill, BillStatus from app.models.order import VisitOrder class BillRepository: """Repository handling database operations for Bill and invoice sequence.""" async def get_next_invoice_number(self, db: AsyncSession, lab_id: UUID) -> str: """ Generate year-aware sequential invoice number for a lab. Format: INV{YY}{6-digit-seq} e.g. INV26000001 (no separators — easy to type). Uses a dedicated PostgreSQL sequence per lab and year synchronized via transactional advisory locks (SELECT pg_advisory_xact_lock) to prevent race conditions and duplicate invoice numbers under concurrent checkouts. """ year_short = datetime.now(UTC).strftime("%y") prefix = f"INV{year_short}" lab_hex = lab_id.hex seq_name = f"seq_inv_{lab_hex}_{year_short}" lock_key = f"inv_seq_{lab_hex}_{year_short}" bind = db.get_bind() is_postgres = bool(bind and bind.dialect.name == "postgresql") if is_postgres: # 1. Acquire transaction-level advisory lock to serialize sequence creation / calibration await db.execute( text("SELECT pg_advisory_xact_lock(hashtext(:lock_key))"), {"lock_key": lock_key}, ) # 2. Check if sequence already exists in pg_class seq_check = await db.execute( text("SELECT 1 FROM pg_class WHERE relkind = 'S' AND relname = :seq_name"), {"seq_name": seq_name}, ) if not seq_check.scalar_one_or_none(): # Find maximum sequence number from existing bills for this lab and year stmt = ( select(Bill.bill_number) .where( Bill.lab_id == lab_id, Bill.bill_number.like(f"{prefix}%"), ) .order_by(Bill.bill_number.desc()) .limit(1) ) res = await db.execute(stmt) last_num = res.scalar_one_or_none() start_val = 1 if last_num: try: start_val = int(last_num[len(prefix) :]) + 1 except (ValueError, IndexError): start_val = 1 # Create the sequence starting at start_val await db.execute(text(f'CREATE SEQUENCE IF NOT EXISTS "{seq_name}" START WITH {start_val}')) # 3. Atomically advance and fetch next sequence number next_res = await db.execute(text(f"SELECT nextval('\"{seq_name}\"')")) next_seq = next_res.scalar_one() return f"{prefix}{next_seq:06d}" # Fallback for non-PostgreSQL / SQLite test environments stmt = ( select(Bill.bill_number) .where( Bill.lab_id == lab_id, Bill.bill_number.like(f"{prefix}%"), ) .order_by(Bill.bill_number.desc()) .limit(1) ) res = await db.execute(stmt) last_num = res.scalar_one_or_none() next_seq = 1 if last_num: try: next_seq = int(last_num[len(prefix) :]) + 1 except (ValueError, IndexError): next_seq = 1 return f"{prefix}{next_seq:06d}" async def get_by_id(self, db: AsyncSession, bill_id: UUID, lab_id: UUID | None = None) -> Bill | None: """Fetch a bill by ID with relationships eager loaded.""" stmt = ( select(Bill) .execution_options(populate_existing=True) .where(Bill.id == bill_id, Bill.deleted_at.is_(None)) .options( selectinload(Bill.order).selectinload(VisitOrder.samples), selectinload(Bill.patient), selectinload(Bill.discount_rule), selectinload(Bill.payments), ) ) if lab_id: stmt = stmt.where(Bill.lab_id == lab_id) res = await db.execute(stmt) return res.scalar_one_or_none() async def get_by_order_id( self, db: AsyncSession, order_id: UUID, lab_id: UUID | None = None ) -> Bill | None: """Fetch active bill for a visit order (1:1 relation).""" stmt = ( select(Bill) .execution_options(populate_existing=True) .where(Bill.order_id == order_id, Bill.deleted_at.is_(None)) .options( selectinload(Bill.order).selectinload(VisitOrder.samples), selectinload(Bill.patient), selectinload(Bill.discount_rule), selectinload(Bill.payments), ) ) if lab_id: stmt = stmt.where(Bill.lab_id == lab_id) res = await db.execute(stmt) return res.scalar_one_or_none() async def create(self, db: AsyncSession, bill: Bill) -> Bill: """Persist a new bill.""" db.add(bill) await db.commit() await db.refresh(bill) return bill async def list_bills( self, db: AsyncSession, lab_id: UUID, status: BillStatus | None = None, patient_id: UUID | None = None, order_id: UUID | None = None, has_due: bool | None = None, is_abroad: bool | None = None, destination_country: str | None = None, search: str | None = None, offset: int = 0, limit: int = 50, created_from: datetime | None = None, created_to: datetime | None = None, ) -> tuple[list[Bill], int]: """List bills for a lab branch with filters and pagination count.""" stmt = ( select(Bill) .where(Bill.lab_id == lab_id, Bill.deleted_at.is_(None)) .options( selectinload(Bill.order), selectinload(Bill.patient), selectinload(Bill.discount_rule), ) ) count_stmt = select(func.count(Bill.id)).where(Bill.lab_id == lab_id, Bill.deleted_at.is_(None)) if status: stmt = stmt.where(Bill.status == status) count_stmt = count_stmt.where(Bill.status == status) if patient_id: stmt = stmt.where(Bill.patient_id == patient_id) count_stmt = count_stmt.where(Bill.patient_id == patient_id) if order_id: stmt = stmt.where(Bill.order_id == order_id) count_stmt = count_stmt.where(Bill.order_id == order_id) if has_due is True: stmt = stmt.where(Bill.due_amount > Decimal("0.00")) count_stmt = count_stmt.where(Bill.due_amount > Decimal("0.00")) elif has_due is False: stmt = stmt.where(Bill.due_amount <= Decimal("0.00")) count_stmt = count_stmt.where(Bill.due_amount <= Decimal("0.00")) if is_abroad is True: abroad_filter = Bill.order.has( and_(VisitOrder.destination_country.isnot(None), VisitOrder.destination_country != "") ) stmt = stmt.where(abroad_filter) count_stmt = count_stmt.where(abroad_filter) elif is_abroad is False: not_abroad_filter = Bill.order.has( or_(VisitOrder.destination_country.is_(None), VisitOrder.destination_country == "") ) stmt = stmt.where(not_abroad_filter) count_stmt = count_stmt.where(not_abroad_filter) if destination_country: dest_filter = Bill.order.has( VisitOrder.destination_country.ilike(f"%{destination_country.strip()}%") ) stmt = stmt.where(dest_filter) count_stmt = count_stmt.where(dest_filter) if created_from: stmt = stmt.where(Bill.created_at >= created_from) count_stmt = count_stmt.where(Bill.created_at >= created_from) if created_to: stmt = stmt.where(Bill.created_at < created_to) count_stmt = count_stmt.where(Bill.created_at < created_to) if search: pattern = f"%{search.strip()}%" stmt = stmt.where(Bill.bill_number.ilike(pattern)) count_stmt = count_stmt.where(Bill.bill_number.ilike(pattern)) stmt = stmt.order_by(Bill.created_at.desc()).offset(offset).limit(limit) total_res = await db.execute(count_stmt) total = total_res.scalar_one() items_res = await db.execute(stmt) items = list(items_res.scalars().all()) return items, total async def update(self, db: AsyncSession, bill: Bill, notes: str | None) -> Bill: """Update bill notes.""" if notes is not None: bill.notes = notes await db.commit() await db.refresh(bill) return bill async def void_bill(self, db: AsyncSession, bill: Bill, reason: str) -> Bill: """Void an unpaid bill with timestamped audit note.""" bill.status = BillStatus.VOID timestamp = datetime.now(UTC).strftime("%Y-%m-%d %H:%M UTC") void_note = f"[VOIDED at {timestamp}: {reason.strip()}]" bill.notes = f"{bill.notes}\n{void_note}".strip() if bill.notes else void_note await db.commit() await db.refresh(bill) return bill