backbone_payroll/infrastructure/persistence/

payroll_entry_repository.rs

//! Repository for PayrollEntry entities
//!
//! Originally generated by metaphor-schema; now **user-owned** — this exact path is declared under
//! `user_owned` in `metaphor.codegen.yaml`, so the generator skips it wholesale. The custom methods
//! below hold the hand-written PayrollEntry SQL — the run lifecycle and the transition gate that makes
//! a run post to the GL at most once (4-layer rule: services orchestrate, repositories hold the SQL).
//!
//! Thin newtype over `backbone_orm::GenericCrudRepository<PayrollEntry, backbone_orm::SoftDelete>`.
//! All standard CRUD methods are available via `Deref`.

use anyhow::Result;
use chrono::{DateTime, Utc};
use rust_decimal::Decimal;
use sqlx::{PgPool, Row};
use uuid::Uuid;

use backbone_orm::org_scope;
// The scalar read twin lives only in the legacy `company_scope` module. Its connection discipline
// is what this adapter needs — request-dedicated connection when the composing service bound one,
// plain pool otherwise. The helper's legacy task-local branch is never taken: this module sets no
// legacy scope of its own (ADR-0029).
use backbone_orm::company_scope::fetch_one_scalar_scoped;

use crate::domain::entity::PayrollEntry;

/// Table name for PayrollEntry entities
pub const TABLE_NAME: &str = "payroll.payroll_entries";

/// Repository for PayrollEntry entities.
///
/// All standard CRUD, soft-delete, pagination, and bulk methods are
/// provided automatically via `Deref` to `backbone_orm::GenericCrudRepository`.
pub struct PayrollEntryRepository(
    backbone_orm::GenericCrudRepository<PayrollEntry, backbone_orm::SoftDelete>,
);

impl std::ops::Deref for PayrollEntryRepository {
    type Target = backbone_orm::GenericCrudRepository<PayrollEntry, backbone_orm::SoftDelete>;
    fn deref(&self) -> &Self::Target { &self.0 }
}

impl PayrollEntryRepository {
    /// Create a new repository instance.
    pub fn new(pool: PgPool) -> Self {
        Self(backbone_orm::GenericCrudRepository::new(pool, TABLE_NAME))
    }
}

/// The exact row a run open writes. Mirrors the raw column shape, not the `PayrollEntry` entity: the
/// insert hard-codes `draft` status and seeds all three totals at 0, so none of those are parameters.
pub struct NewPayrollEntryRow {
    pub id: Uuid,
    pub period_year: i32,
    pub period_month: i32,
    /// Non-calendar bounds (both or neither; the CHECK constrains order).
    pub period_start: Option<chrono::NaiveDate>,
    pub period_end: Option<chrono::NaiveDate>,
    pub salary_expense_account_id: Uuid,
    pub salary_payable_account_id: Uuid,
}

/// A live run's lifecycle state — what the slip path reads before it writes.
pub struct RunStateRow {
    pub status: String,
}

/// The computed-slip orchestrator's run read: state + the period (which statutory parameter set is
/// effective) + the expense account overtime/THR earning lines book against.
pub struct RunPeriodRow {
    pub status: String,
    pub period_year: i32,
    pub period_month: i32,
    /// Non-calendar bounds (both or neither).
    pub period_start: Option<chrono::NaiveDate>,
    pub period_end: Option<chrono::NaiveDate>,
    pub salary_expense_account_id: Option<Uuid>,
}

/// What the GL post path reads before it builds the journal. The account columns are nullable in the
/// schema, so they come back as `Option` for the caller to reject; `journal_id`/`accounting_post_id`
/// are Some only once posted, which is what makes a re-post return the original journal.
pub struct RunPostingRow {
    pub status: String,
    pub salary_expense_account_id: Option<Uuid>,
    pub salary_payable_account_id: Option<Uuid>,
    pub total_gross: Decimal,
    pub total_deductions: Decimal,
    pub total_net: Decimal,
    pub journal_id: Option<Uuid>,
    pub accounting_post_id: Option<Uuid>,
}

/// Hand-written PayrollEntry SQL. Lives here (not in the write service) per the module's 4-layer rule:
/// services orchestrate and own the unit of work, repositories hold the SQL.
impl PayrollEntryRepository {
    /// Open a payroll run for a period, as a draft with zeroed totals.
    ///
    /// A write outside any transaction: takes the pool and runs `execute_scoped` so the composing
    /// service's tenancy RLS fence applies (ADR-0029). The caller relays the ambient org request
    /// scope onto its own transaction first, or runs under HTTP where the request-dedicated
    /// connection already carries it; an undecorated deployment is unfenced by design.
    ///
    /// Returns the raw `sqlx::Error` deliberately: the caller inspects it for a unique violation to
    /// turn a second run on the same (org unit, year, month) into a domain error.
    pub async fn insert_entry(
        &self,
        pool: &PgPool,
        e: &NewPayrollEntryRow,
    ) -> Result<(), sqlx::Error> {
        org_scope::execute_scoped(
            pool,
            sqlx::query(
                r#"INSERT INTO payroll.payroll_entries
                     (id, period_year, period_month, period_start, period_end, status,
                      salary_expense_account_id, salary_payable_account_id,
                      total_gross, total_deductions, total_net)
                   VALUES ($1,$2,$3,$6,$7,'draft'::payroll_status,$4,$5,0,0,0)"#,
            )
            .bind(e.id).bind(e.period_year).bind(e.period_month)
            .bind(e.salary_expense_account_id).bind(e.salary_payable_account_id)
            .bind(e.period_start).bind(e.period_end),
        )
        .await?;
        Ok(())
    }

    /// Read a live run's status — the slip path's lookup. `Ok(None)` = not found in scope.
    ///
    /// ID-only: `fetch_optional_row_scoped` rides a connection carrying the caller's org request
    /// scope (ADR-0029), so another unit's run simply isn't found. The caller relays the same
    /// ambient scope onto its own write transaction.
    pub async fn find_state_by_id(
        &self,
        pool: &PgPool,
        run_id: Uuid,
    ) -> Result<Option<RunStateRow>, sqlx::Error> {
        let row = org_scope::fetch_optional_row_scoped(
            pool,
            sqlx::query(
                r#"SELECT status::text AS status FROM payroll.payroll_entries
                   WHERE id=$1 AND (metadata->>'deleted_at') IS NULL"#,
            )
            .bind(run_id),
        )
        .await?;
        Ok(row.map(|r| RunStateRow { status: r.get("status") }))
    }

    /// The run's period + state, read ID-only under the request scope (same pattern as
    /// [`Self::find_state_by_id`]).
    pub async fn find_period_by_id(
        &self,
        pool: &PgPool,
        run_id: Uuid,
    ) -> Result<Option<RunPeriodRow>, sqlx::Error> {
        let row = org_scope::fetch_optional_row_scoped(
            pool,
            sqlx::query(
                r#"SELECT status::text AS status, period_year, period_month,
                          period_start, period_end, salary_expense_account_id
                   FROM payroll.payroll_entries
                   WHERE id=$1 AND (metadata->>'deleted_at') IS NULL"#,
            )
            .bind(run_id),
        )
        .await?;
        Ok(row.map(|r| RunPeriodRow {
            status: r.get("status"),
            period_year: r.get("period_year"),
            period_month: r.get("period_month"),
            period_start: r.get("period_start"),
            period_end: r.get("period_end"),
            salary_expense_account_id: r.get("salary_expense_account_id"),
        }))
    }

    /// Record the rolled-up totals and move `draft → processed`. Returns rows affected: 0 = not draft.
    ///
    /// ID-only: `execute_scoped` rides the request-dedicated connection's org request scope
    /// (ADR-0029). A caller driving its own transaction relays the ambient scope onto it first
    /// (`bind_org_scope_on`); an undecorated deployment is unfenced by design.
    pub async fn mark_processed(
        &self,
        pool: &PgPool,
        run_id: Uuid,
        total_gross: Decimal,
        total_deductions: Decimal,
        total_net: Decimal,
    ) -> Result<u64, sqlx::Error> {
        let done = org_scope::execute_scoped(
            pool,
            sqlx::query(
                r#"UPDATE payroll.payroll_entries
                   SET status='processed'::payroll_status, total_gross=$2, total_deductions=$3, total_net=$4
                   WHERE id=$1 AND status='draft'::payroll_status"#,
            )
            .bind(run_id).bind(total_gross).bind(total_deductions).bind(total_net),
        )
        .await?;
        Ok(done.rows_affected())
    }

    /// Read a live run for GL posting. `Ok(None)` = not found in scope.
    ///
    /// ID-only: under HTTP the request-dedicated connection carries the org request scope
    /// (ADR-0029); a caller driving its own transaction relays the ambient scope onto it first.
    pub async fn find_for_posting(
        &self,
        pool: &PgPool,
        run_id: Uuid,
    ) -> Result<Option<RunPostingRow>, sqlx::Error> {
        let row = org_scope::fetch_optional_row_scoped(
            pool,
            sqlx::query(
                r#"SELECT status::text AS status, salary_expense_account_id, salary_payable_account_id,
                          total_gross, total_deductions, total_net, journal_id, accounting_post_id
                   FROM payroll.payroll_entries WHERE id=$1 AND (metadata->>'deleted_at') IS NULL"#,
            )
            .bind(run_id),
        )
        .await?;
        Ok(row.map(|r| RunPostingRow {
            status: r.get("status"),
            salary_expense_account_id: r.get("salary_expense_account_id"),
            salary_payable_account_id: r.get("salary_payable_account_id"),
            total_gross: r.get("total_gross"), total_deductions: r.get("total_deductions"),
            total_net: r.get("total_net"), journal_id: r.get("journal_id"),
            accounting_post_id: r.get("accounting_post_id"),
        }))
    }

    /// Claim the `processed → posted` transition exactly once, recording the GL's echoed journal + post.
    /// Returns rows affected: 0 = a racing caller already posted it, so this one re-reads that journal.
    /// This gate is what makes a run post AT MOST once.
    ///
    /// A write outside any transaction: it rides the caller's org request scope (ADR-0029), relayed
    /// onto the caller's transaction or carried by the request-dedicated connection.
    pub async fn mark_posted(
        &self,
        pool: &PgPool,
        run_id: Uuid,
        posting_date: DateTime<Utc>,
        journal_id: Uuid,
        post_id: Uuid,
    ) -> Result<u64, sqlx::Error> {
        let done = org_scope::execute_scoped(
            pool,
            sqlx::query(
                r#"UPDATE payroll.payroll_entries
                   SET status='posted'::payroll_status, posting_date=$2, journal_id=$3, accounting_post_id=$4
                   WHERE id=$1 AND status='processed'::payroll_status"#,
            )
            .bind(run_id).bind(posting_date).bind(journal_id).bind(post_id),
        )
        .await?;
        Ok(done.rows_affected())
    }

    /// Re-read the journal the winner recorded, after a lost `mark_posted` race. Rides the caller's
    /// org request scope, as above.
    pub async fn fetch_journal_id(
        &self,
        pool: &PgPool,
        run_id: Uuid,
    ) -> Result<Uuid, sqlx::Error> {
        fetch_one_scalar_scoped(
            pool,
            sqlx::query_scalar("SELECT journal_id FROM payroll.payroll_entries WHERE id=$1")
                .bind(run_id),
        )
        .await
    }
}

backbone_core::impl_crud_repository!(PayrollEntryRepository, PayrollEntry, soft_delete);