backbone_support/infrastructure/persistence/

issue_repository.rs

//! Repository for Issue 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 are the home for every hand-written Issue SQL statement — the SLA clock's transitions
//! (4-layer rule: services orchestrate and own the unit of work, repositories hold the SQL).
//!
//! Tenancy (ADR-0029): the SQL here carries no tenant key. The scoped-execute helpers ride the
//! request-dedicated connection when the composing service bound one (its fence variables govern
//! what the RLS layer accepts) and fall back to a plain pool execute otherwise.
//!
//! Thin newtype over `backbone_orm::GenericCrudRepository<Issue, backbone_orm::SoftDelete>`.
//! All standard CRUD methods are available via `Deref`.

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

use backbone_orm::org_scope;

use crate::domain::entity::Issue;

/// Table name for Issue entities
pub const TABLE_NAME: &str = "support.issues";

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

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

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

/// The exact row a raised ticket writes.
///
/// Mirrors the raw column shape rather than the `Issue` entity: `priority` is carried as a free string
/// and cast at the DB (`$5::issue_priority`), so a bad priority fails as a DB error rather than a
/// deserialize panic. The deadlines are already snapshotted by the caller.
pub struct NewIssueRow<'a> {
    pub id: Uuid,
    pub customer_id: Option<Uuid>,
    pub subject: &'a str,
    pub description: Option<&'a str>,
    pub priority: &'a str,
    pub sla_id: Option<Uuid>,
    pub opened_at: DateTime<Utc>,
    pub response_by: Option<DateTime<Utc>>,
    pub resolution_by: Option<DateTime<Utc>>,
}

/// The locked on-hold projection the resume math needs: when the clock stopped, and whether the first
/// response leg is already closed.
pub struct PausedIssueRow {
    pub paused_at: DateTime<Utc>,
    pub first_responded_at: Option<DateTime<Utc>>,
}

/// The pre-flight projection for resolve: the issue's live status (for the friendly
/// on_hold/not-found errors — the SLA verdict itself is judged in `resolve`'s own UPDATE).
pub struct IssueStateRow {
    pub status: String,
}

/// The escalation candidate's projection — the customer the project opens for and the once-only
/// gate (`escalated_project_id`).
pub struct EscalationCandidateRow {
    pub customer_id: Option<Uuid>,
    pub subject: String,
    pub status: String,
    pub escalated_project_id: Option<Uuid>,
}

/// Hand-written Issue 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 IssueRepository {
    /// Insert a raised ticket with its snapshotted deadlines.
    ///
    /// A write outside any transaction: takes the pool and runs the scoped helper so it rides the
    /// request-dedicated connection when the composing service bound a scope — under a decorated
    /// deployment the fence's WITH CHECK governs the row; with no scope bound this is a plain insert.
    pub async fn insert_issue(
        &self,
        pool: &PgPool,
        i: &NewIssueRow<'_>,
    ) -> Result<(), sqlx::Error> {
        org_scope::execute_scoped(
            pool,
            sqlx::query(
                r#"INSERT INTO support.issues
                     (id, customer_id, subject, description, priority, sla_id, status,
                      agreement_status, opened_at, response_by, resolution_by, total_paused_mins)
                   VALUES ($1,$2,$3,$4,$5::issue_priority,$6,'open'::issue_status,
                           'first_response_due'::agreement_status,$7,$8,$9,0)"#,
            )
            .bind(i.id)
            .bind(i.customer_id)
            .bind(i.subject)
            .bind(i.description)
            .bind(i.priority)
            .bind(i.sla_id)
            .bind(i.opened_at)
            .bind(i.response_by)
            .bind(i.resolution_by),
        )
        .await?;
        Ok(())
    }

    /// Record the first response and judge the response leg IN THE SAME gated UPDATE, against the row's
    /// live (pause-adjusted) `response_by` — never a stale read. Returns rows affected (0 = the issue is
    /// not awaiting a first response).
    ///
    /// ID-only: rides the request-dedicated connection when the composing service bound a scope, so a
    /// row the deployment's fence excludes is simply not updated; with no scope bound this is a plain
    /// update.
    pub async fn record_first_response(
        &self,
        pool: &PgPool,
        issue_id: Uuid,
        now: DateTime<Utc>,
    ) -> Result<u64, sqlx::Error> {
        let moved = org_scope::execute_scoped(
            pool,
            sqlx::query(
                r#"UPDATE support.issues
                   SET first_responded_at=$2, status='replied'::issue_status,
                       response_breached = (response_by IS NOT NULL AND $2 > response_by),
                       agreement_status = CASE WHEN response_by IS NOT NULL AND $2 > response_by
                                               THEN 'failed'::agreement_status
                                               ELSE 'resolution_due'::agreement_status END
                   WHERE id=$1 AND status='open'::issue_status AND (metadata->>'deleted_at') IS NULL"#,
            )
            .bind(issue_id).bind(now),
        )
        .await?;
        Ok(moved.rows_affected())
    }

    /// Stop the SLA clock. Returns rows affected (0 = not an open/replied issue).
    pub async fn pause(
        &self,
        pool: &PgPool,
        issue_id: Uuid,
        now: DateTime<Utc>,
    ) -> Result<u64, sqlx::Error> {
        let moved = org_scope::execute_scoped(
            pool,
            sqlx::query(
                r#"UPDATE support.issues
                   SET status='on_hold'::issue_status, paused_at=$2
                   WHERE id=$1 AND status IN ('open','replied') AND (metadata->>'deleted_at') IS NULL"#,
            )
            .bind(issue_id).bind(now),
        )
        .await?;
        Ok(moved.rows_affected())
    }

    /// Lock the on-hold row and read what the resume math needs. `Ok(None)` = the issue is not on hold.
    ///
    /// Takes the CALLER'S connection: this `FOR UPDATE` read and the deadline extension that follows are
    /// one unit of work, and the lock must be held across both. The caller has already relayed the
    /// ambient org scope onto that connection — don't re-bind here.
    pub async fn lock_on_hold(
        &self,
        conn: &mut sqlx::PgConnection,
        issue_id: Uuid,
    ) -> Result<Option<PausedIssueRow>, sqlx::Error> {
        let row = sqlx::query(
            r#"SELECT paused_at, first_responded_at FROM support.issues
               WHERE id=$1 AND status='on_hold'::issue_status FOR UPDATE"#,
        )
        .bind(issue_id)
        .fetch_optional(conn)
        .await?;
        Ok(row.map(|r| PausedIssueRow {
            paused_at: r.get("paused_at"),
            first_responded_at: r.get("first_responded_at"),
        }))
    }

    /// Restart the clock: add the paused span back to the outstanding deadlines so a hold never counts
    /// against the SLA, and restore the running status. Extends the response deadline too, but only
    /// while the first response is still outstanding.
    ///
    /// Takes the CALLER'S connection — it must be the one holding the `lock_on_hold` lock.
    pub async fn apply_resume(
        &self,
        conn: &mut sqlx::PgConnection,
        issue_id: Uuid,
        paused_mins: i32,
        running: &str,
    ) -> Result<(), sqlx::Error> {
        sqlx::query(
            r#"UPDATE support.issues
               SET status=$3::issue_status, paused_at=NULL,
                   total_paused_mins = total_paused_mins + $2,
                   resolution_by = resolution_by + ($2 * interval '1 minute'),
                   response_by = CASE WHEN first_responded_at IS NULL
                                      THEN response_by + ($2 * interval '1 minute') ELSE response_by END
               WHERE id=$1"#,
        )
        .bind(issue_id).bind(paused_mins).bind(running)
        .execute(conn)
        .await?;
        Ok(())
    }

    /// Read an issue's live status. Only for the caller's friendly on_hold/not-found errors — the
    /// SLA verdict is NOT computed from this snapshot (see `resolve`).
    pub async fn find_state(
        &self,
        pool: &PgPool,
        issue_id: Uuid,
    ) -> Result<Option<IssueStateRow>, sqlx::Error> {
        let row = org_scope::fetch_optional_row_scoped(
            pool,
            sqlx::query(
                r#"SELECT status::text AS status FROM support.issues
                   WHERE id=$1 AND (metadata->>'deleted_at') IS NULL"#,
            )
            .bind(issue_id),
        )
        .await?;
        Ok(row.map(|r| IssueStateRow {
            status: r.get("status"),
        }))
    }

    /// Resolve a ticket and judge the SLA IN THE SAME gated UPDATE, from the row's LIVE `resolution_by`.
    /// Returns `Some(fulfilled)`, or `None` if the issue is no longer resolvable.
    ///
    /// Judging in the same statement as the transition is load-bearing: a concurrent pause+resume that
    /// commits between a snapshot read and this write extends `resolution_by` and restores an
    /// open/replied status, so a read-then-write split would stamp a stale verdict — a met SLA marked
    /// `failed` (maturity council 2026-07-07). A ticket with no SLA (NULL resolution_by) is always
    /// fulfilled. `fulfilled` requires BOTH legs met: the resolution is within `resolution_by`, AND the
    /// response leg is not breached — either an already-recorded on-time response, or (none recorded
    /// yet) resolving within `response_by` counts as responding in time (completeness council
    /// 2026-07-07).
    pub async fn resolve(
        &self,
        pool: &PgPool,
        issue_id: Uuid,
        now: DateTime<Utc>,
    ) -> Result<Option<bool>, sqlx::Error> {
        let row = org_scope::fetch_optional_row_scoped(
            pool,
            sqlx::query(
                r#"UPDATE support.issues
                   SET status='resolved'::issue_status, resolved_at=$2,
                       agreement_status = CASE
                           WHEN (resolution_by IS NULL OR $2 <= resolution_by)
                                AND NOT response_breached
                                AND (first_responded_at IS NOT NULL OR response_by IS NULL OR $2 <= response_by)
                           THEN 'fulfilled'::agreement_status
                           ELSE 'failed'::agreement_status END
                   WHERE id=$1 AND status IN ('open','replied')
                   RETURNING (agreement_status = 'fulfilled'::agreement_status) AS fulfilled"#,
            )
            .bind(issue_id).bind(now),
        )
        .await?;
        Ok(row.map(|r| r.get("fulfilled")))
    }

    /// Close a resolved ticket (terminal). Returns rows affected (0 = not a resolved issue).
    pub async fn close(&self, pool: &PgPool, issue_id: Uuid) -> Result<u64, sqlx::Error> {
        let moved = org_scope::execute_scoped(
            pool,
            sqlx::query(
                r#"UPDATE support.issues SET status='closed'::issue_status
                   WHERE id=$1 AND status='resolved'::issue_status AND (metadata->>'deleted_at') IS NULL"#,
            )
            .bind(issue_id),
        )
        .await?;
        Ok(moved.rows_affected())
    }

    /// Read what the escalation decision needs, including the once-only gate.
    pub async fn find_escalation_candidate(
        &self,
        pool: &PgPool,
        issue_id: Uuid,
    ) -> Result<Option<EscalationCandidateRow>, sqlx::Error> {
        let row = org_scope::fetch_optional_row_scoped(
            pool,
            sqlx::query(
                r#"SELECT customer_id, subject, status::text AS status, escalated_project_id
                   FROM support.issues WHERE id=$1 AND (metadata->>'deleted_at') IS NULL"#,
            )
            .bind(issue_id),
        )
        .await?;
        Ok(row.map(|r| EscalationCandidateRow {
            customer_id: r.get("customer_id"),
            subject: r.get("subject"),
            status: r.get("status"),
            escalated_project_id: r.get("escalated_project_id"),
        }))
    }

    /// Claim the escalation exactly once (CAS on `escalated_project_id IS NULL`). Returns rows affected
    /// (0 = a concurrent escalation won; the caller re-reads the winner's project id).
    ///
    /// Takes the CALLER'S connection so the claim and the outbox stage commit as one unit — backbone-project
    /// subscribes to IssueEscalated, so a crash between the CAS and the publish must not drop it. The
    /// caller has already relayed the ambient org scope onto that connection — don't re-bind here.
    pub async fn claim_escalation(
        &self,
        conn: &mut sqlx::PgConnection,
        issue_id: Uuid,
        project_id: Uuid,
    ) -> Result<u64, sqlx::Error> {
        let moved = sqlx::query(
            r#"UPDATE support.issues SET escalated_project_id=$2
               WHERE id=$1 AND escalated_project_id IS NULL"#,
        )
        .bind(issue_id)
        .bind(project_id)
        .execute(conn)
        .await?;
        Ok(moved.rows_affected())
    }

    /// Re-read the winner's project id after a losing escalation CAS.
    ///
    /// A read outside the (rolled-back) transaction: rides the request-dedicated connection when the
    /// composing service bound a scope, plain pool otherwise.
    pub async fn fetch_escalated_project_id(
        &self,
        pool: &PgPool,
        issue_id: Uuid,
    ) -> Result<Uuid, sqlx::Error> {
        let row = org_scope::fetch_optional_row_scoped(
            pool,
            sqlx::query("SELECT escalated_project_id FROM support.issues WHERE id=$1")
                .bind(issue_id),
        )
        .await?;
        row.map(|r| r.get::<Uuid, _>("escalated_project_id"))
            .ok_or(sqlx::Error::RowNotFound)
    }
}

backbone_core::impl_crud_repository!(IssueRepository, Issue, soft_delete);