orion-server 1.11.1

//! Package receipts (K14): the storage half of the applied-version
//! immutability rule.
//!
//! A receipt records that a package artifact — a bundle of connectors,
//! workflows and channels the `orion-server package` CLI stages and activates
//! through the per-kind admin APIs — landed here, at which version, with what
//! canonical content hash. The rule this repository enforces is the entity
//! lifecycle one level up: a `staged` receipt may be re-put with new content
//! (only a draft can be updated), an `applied` receipt is immutable — the same
//! version arriving with a different hash is a 409, and the fix is a package
//! version bump.

use async_trait::async_trait;
use sea_query::{Asterisk, Expr, ExprTrait, Order, Query};
use serde::Deserialize;

use crate::errors::OrionError;
use crate::storage::models::{PackageInventory, PackageReceipt, PackageState};
use crate::storage::{DbPool, build_sqlx, schema::Packages};

use super::helpers::{fetch_required_tx, sql_now};

// -- DTOs --

/// Body of `PUT /api/v1/admin/packages/{name}`.
#[derive(Debug, Deserialize, utoipa::ToSchema)]
pub struct PutPackageReceiptRequest {
    /// Package version this receipt records, e.g. `1.4.0`.
    pub version: String,
    /// Canonical content hash of the artifact, e.g. `sha256:…`. Compared for
    /// equality against later PUTs of the same version; never parsed.
    pub content_hash: String,
    /// `staged` before the artifact's entities are activated, `applied` after.
    pub state: PackageState,
    /// What this version carries. Stored on insert and on a staged re-put;
    /// on an applied version it only fills a receipt recorded without one,
    /// since the hash is equal and so is what the version carried.
    #[serde(default)]
    pub inventory: Option<PackageInventory>,
}

impl PutPackageReceiptRequest {
    /// The inventory in its stored form, or `None` when none was sent.
    fn inventory_json(&self) -> Result<Option<String>, OrionError> {
        self.inventory
            .clone()
            .map(|inventory| serde_json::to_string(&inventory.normalized()))
            .transpose()
            .map_err(OrionError::from)
    }
}

// -- Repository trait --

#[async_trait]
pub trait PackageRepository: Send + Sync {
    /// Record (or advance) the receipt for one package version, enforcing
    /// applied-version immutability:
    ///
    /// - no row for `(name, version)` → insert as requested;
    /// - existing `staged` row → update content/state in place;
    /// - existing `applied` row, same hash → touch (idempotent re-apply;
    ///   `staged` is refused — an applied version cannot be demoted);
    /// - existing `applied` row, different hash → `Conflict`.
    ///
    /// Every write carries its state predicate, so two concurrent PUTs cannot
    /// lose an update: the one that finds the row no longer in the state it
    /// read gets a `Conflict` instead of silently overwriting.
    async fn put(
        &self,
        name: &str,
        req: &PutPackageReceiptRequest,
        principal: &str,
    ) -> Result<PackageReceipt, OrionError>;
    /// One page of receipt rows, ordered by name then newest first.
    async fn list(
        &self,
        limit: i64,
        offset: i64,
    ) -> Result<super::helpers::PaginatedResult<PackageReceipt>, OrionError>;
    /// All of one package's receipts, newest first. `NotFound` when none.
    async fn get_by_name(&self, name: &str) -> Result<Vec<PackageReceipt>, OrionError>;
    /// One page of each package's `current` receipt — its newest `applied`
    /// one — ordered by name. A package with only staged receipts has none.
    async fn list_current(
        &self,
        limit: i64,
        offset: i64,
    ) -> Result<super::helpers::PaginatedResult<PackageReceipt>, OrionError>;
}

// -- SQL implementation --

pub struct SqlPackageRepository {
    pool: DbPool,
}

impl SqlPackageRepository {
    pub fn new(pool: DbPool) -> Self {
        Self { pool }
    }
}

/// The `SELECT * WHERE name = ? AND version = ?` single-receipt shape.
fn receipt_select(name: &str, version: &str) -> sea_query::SelectStatement {
    Query::select()
        .column(Asterisk)
        .from(Packages::Table)
        .and_where(Expr::col(Packages::Name).eq(name))
        .and_where(Expr::col(Packages::Version).eq(version))
        .to_owned()
}

/// The immutability refusal, worded for the operator who has to fix it.
fn applied_conflict(name: &str, version: &str) -> OrionError {
    OrionError::Conflict(format!(
        "package '{name}' version '{version}' is already applied here with different \
         content — an applied package version is immutable; bump the package version"
    ))
}

#[async_trait]
impl PackageRepository for SqlPackageRepository {
    async fn put(
        &self,
        name: &str,
        req: &PutPackageReceiptRequest,
        principal: &str,
    ) -> Result<PackageReceipt, OrionError> {
        crate::metrics::timed_db_op("packages.put", async {
            let backend = self.pool.backend();
            let inventory = req.inventory_json()?;
            let mut tx = self.pool.begin_write_tx().await?;

            let (sql, values) =
                build_sqlx(self.pool.backend(), &mut receipt_select(name, &req.version));
            let existing: Option<PackageReceipt> = tx.fetch_optional_as(&sql, values).await?;

            match existing {
                None => {
                    let (sql, values) = build_sqlx(
                        self.pool.backend(),
                        Query::insert()
                            .into_table(Packages::Table)
                            .columns([
                                Packages::Name,
                                Packages::Version,
                                Packages::ContentHash,
                                Packages::State,
                                Packages::Principal,
                                Packages::InventoryJson,
                            ])
                            .values_panic([
                                name.into(),
                                req.version.as_str().into(),
                                req.content_hash.as_str().into(),
                                req.state.as_str().into(),
                                principal.into(),
                                super::helpers::optional_string_value(inventory.as_deref()).into(),
                            ]),
                    );
                    // A concurrent PUT that inserted first surfaces as the
                    // PK collision here — a retry will read its row.
                    tx.execute_query(&sql, values).await.map_err(|e| {
                        super::helpers::map_duplicate(e, || {
                            format!(
                                "package '{name}' version '{}' was recorded by a \
                                 concurrent request — retry to read it",
                                req.version
                            )
                        })
                    })?;
                }
                Some(row) if row.state == PackageState::Applied.as_str() => {
                    if row.content_hash != req.content_hash {
                        return Err(applied_conflict(name, &req.version));
                    }
                    if req.state == PackageState::Staged {
                        return Err(OrionError::Conflict(format!(
                            "package '{name}' version '{}' is already applied — an \
                             applied package version cannot go back to staged",
                            req.version
                        )));
                    }
                    // Idempotent re-apply: touch updated_at so this version
                    // becomes the newest applied receipt again (the rollback
                    // path re-applies an older version). The state and hash
                    // predicates make the touch a no-op — refused, not
                    // absorbed — if the row changed underneath us.
                    let mut update = Query::update();
                    update
                        .table(Packages::Table)
                        .value(Packages::Principal, principal)
                        .value(Packages::UpdatedAt, Expr::cust(sql_now(backend)));
                    // A receipt applied before receipts recorded inventories
                    // gains one here and nowhere else: the hash is equal, so
                    // the inventory is the one it always had. A recorded one
                    // is never replaced — applied stays immutable.
                    if row.inventory_json.is_none()
                        && let Some(inventory) = &inventory
                    {
                        update.value(Packages::InventoryJson, inventory.as_str());
                    }
                    let (sql, values) = build_sqlx(
                        self.pool.backend(),
                        update
                            .and_where(Expr::col(Packages::Name).eq(name))
                            .and_where(Expr::col(Packages::Version).eq(req.version.as_str()))
                            .and_where(
                                Expr::col(Packages::State).eq(PackageState::Applied.as_str()),
                            )
                            .and_where(
                                Expr::col(Packages::ContentHash).eq(req.content_hash.as_str()),
                            ),
                    );
                    if tx.execute_query(&sql, values).await? == 0 {
                        return Err(applied_conflict(name, &req.version));
                    }
                }
                Some(_) => {
                    // Staged: content and state may move — only a draft can be
                    // updated. The state predicate refuses the update if a
                    // concurrent PUT applied this version after our read.
                    // The inventory moves with the content; a PUT without
                    // one leaves what is recorded.
                    let mut update = Query::update();
                    update
                        .table(Packages::Table)
                        .value(Packages::ContentHash, req.content_hash.as_str())
                        .value(Packages::State, req.state.as_str())
                        .value(Packages::Principal, principal)
                        .value(Packages::UpdatedAt, Expr::cust(sql_now(backend)));
                    if let Some(inventory) = &inventory {
                        update.value(Packages::InventoryJson, inventory.as_str());
                    }
                    let (sql, values) = build_sqlx(
                        self.pool.backend(),
                        update
                            .and_where(Expr::col(Packages::Name).eq(name))
                            .and_where(Expr::col(Packages::Version).eq(req.version.as_str()))
                            .and_where(
                                Expr::col(Packages::State).eq(PackageState::Staged.as_str()),
                            ),
                    );
                    if tx.execute_query(&sql, values).await? == 0 {
                        return Err(OrionError::Conflict(format!(
                            "package '{name}' version '{}' was applied by a concurrent \
                             request — re-check its receipt before writing again",
                            req.version
                        )));
                    }
                }
            }

            // D23: the row this PUT produced, read inside the same transaction.
            let (sql, values) =
                build_sqlx(self.pool.backend(), &mut receipt_select(name, &req.version));
            let row = fetch_required_tx(&mut tx, &sql, values, || {
                OrionError::internal(format!(
                    "package receipt '{name}' version '{}' vanished mid-write",
                    req.version
                ))
            })
            .await?;
            tx.commit().await?;
            Ok(row)
        })
        .await
    }

    async fn list(
        &self,
        limit: i64,
        offset: i64,
    ) -> Result<super::helpers::PaginatedResult<PackageReceipt>, OrionError> {
        crate::metrics::timed_db_op("packages.list", async {
            let total = super::helpers::count_where(
                &self.pool,
                Packages::Table,
                sea_query::Condition::all(),
            )
            .await?;
            let (sql, values) = build_sqlx(
                self.pool.backend(),
                Query::select()
                    .column(Asterisk)
                    .from(Packages::Table)
                    .order_by(Packages::Name, Order::Asc)
                    .order_by(Packages::UpdatedAt, Order::Desc)
                    // Unique tiebreaker so OFFSET paging cannot skip or repeat
                    // rows when updated_at collides.
                    .order_by(Packages::Version, Order::Asc)
                    .limit(limit as u64)
                    .offset(offset as u64),
            );
            let data: Vec<PackageReceipt> = self.pool.fetch_all_as(&sql, values).await?;
            Ok(super::helpers::PaginatedResult {
                data,
                total,
                limit,
                offset,
            })
        })
        .await
    }

    async fn get_by_name(&self, name: &str) -> Result<Vec<PackageReceipt>, OrionError> {
        crate::metrics::timed_db_op("packages.get_by_name", async {
            let (sql, values) = build_sqlx(
                self.pool.backend(),
                Query::select()
                    .column(Asterisk)
                    .from(Packages::Table)
                    .and_where(Expr::col(Packages::Name).eq(name))
                    .order_by(Packages::UpdatedAt, Order::Desc)
                    .order_by(Packages::Version, Order::Desc),
            );
            let rows: Vec<PackageReceipt> = self.pool.fetch_all_as(&sql, values).await?;
            if rows.is_empty() {
                return Err(OrionError::NotFound(format!(
                    "Package '{name}' has no receipts"
                )));
            }
            Ok(rows)
        })
        .await
    }

    async fn list_current(
        &self,
        limit: i64,
        offset: i64,
    ) -> Result<super::helpers::PaginatedResult<PackageReceipt>, OrionError> {
        crate::metrics::timed_db_op("packages.list_current", async {
            // Every applied row, newest first within a name, and the first
            // per name kept here: receipts are few, and the order is the
            // one `get_by_name` reads `current` from.
            let (sql, values) = build_sqlx(
                self.pool.backend(),
                Query::select()
                    .column(Asterisk)
                    .from(Packages::Table)
                    .and_where(Expr::col(Packages::State).eq(PackageState::Applied.as_str()))
                    .order_by(Packages::Name, Order::Asc)
                    .order_by(Packages::UpdatedAt, Order::Desc)
                    .order_by(Packages::Version, Order::Desc),
            );
            let rows: Vec<PackageReceipt> = self.pool.fetch_all_as(&sql, values).await?;
            let mut current: Vec<PackageReceipt> = Vec::new();
            for row in rows {
                if current.last().is_none_or(|last| last.name != row.name) {
                    current.push(row);
                }
            }
            let total = current.len() as i64;
            let data = current
                .into_iter()
                .skip(Ord::max(offset, 0) as usize)
                .take(Ord::max(limit, 0) as usize)
                .collect();
            Ok(super::helpers::PaginatedResult {
                data,
                total,
                limit,
                offset,
            })
        })
        .await
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    async fn repo() -> SqlPackageRepository {
        SqlPackageRepository::new(crate::storage::test_sqlite_pool().await)
    }

    fn put(version: &str, hash: &str, state: PackageState) -> PutPackageReceiptRequest {
        PutPackageReceiptRequest {
            version: version.to_string(),
            content_hash: hash.to_string(),
            state,
            inventory: None,
        }
    }

    fn with_channels(mut req: PutPackageReceiptRequest, ids: &[&str]) -> PutPackageReceiptRequest {
        req.inventory = Some(PackageInventory {
            channels: ids.iter().map(|id| id.to_string()).collect(),
            ..Default::default()
        });
        req
    }

    fn channels_of(receipt: &PackageReceipt) -> Option<Vec<String>> {
        let json = receipt.inventory_json.as_deref()?;
        let inventory: PackageInventory = serde_json::from_str(json).expect("stored inventory");
        Some(inventory.channels)
    }

    /// The receipt table's whole purpose is the applied-version immutability
    /// rule, and every branch of it lives in one `match` on the existing row.
    /// Each arm gets an assertion here, in the order the trait doc states
    /// them, because the arms differ only in which predicate the write
    /// carries — a difference no type checks.
    #[tokio::test]
    async fn a_new_version_is_inserted_as_requested() {
        let repo = repo().await;
        let receipt = repo
            .put(
                "orders",
                &put("1.0.0", "sha256:aaa", PackageState::Staged),
                "ci",
            )
            .await
            .expect("insert");
        assert_eq!(receipt.version, "1.0.0");
        assert_eq!(receipt.content_hash, "sha256:aaa");
        assert_eq!(receipt.state, PackageState::Staged.as_str());
        assert_eq!(receipt.principal, "ci");
    }

    /// A staged version is a draft: content may still change under it, and it
    /// may be promoted. Neither is a conflict.
    #[tokio::test]
    async fn a_staged_version_is_updated_in_place() {
        let repo = repo().await;
        repo.put(
            "orders",
            &put("1.0.0", "sha256:aaa", PackageState::Staged),
            "ci",
        )
        .await
        .expect("stage");

        let restaged = repo
            .put(
                "orders",
                &put("1.0.0", "sha256:bbb", PackageState::Staged),
                "ci",
            )
            .await
            .expect("re-stage with new content");
        assert_eq!(
            restaged.content_hash, "sha256:bbb",
            "a staged version's content is still mutable"
        );

        let applied = repo
            .put(
                "orders",
                &put("1.0.0", "sha256:bbb", PackageState::Applied),
                "ci",
            )
            .await
            .expect("promote");
        assert_eq!(applied.state, PackageState::Applied.as_str());
    }

    /// Re-applying the identical artifact is the retry a promotion pipeline
    /// performs, so it must be a no-op and not a 409.
    #[tokio::test]
    async fn re_applying_the_same_content_is_idempotent() {
        let repo = repo().await;
        repo.put(
            "orders",
            &put("1.0.0", "sha256:aaa", PackageState::Applied),
            "ci",
        )
        .await
        .expect("apply");
        let again = repo
            .put(
                "orders",
                &put("1.0.0", "sha256:aaa", PackageState::Applied),
                "ci",
            )
            .await
            .expect("re-applying identical content must be accepted");
        assert_eq!(again.state, PackageState::Applied.as_str());
        assert_eq!(again.content_hash, "sha256:aaa");
    }

    /// The rule the table exists for: an applied version's content is fixed.
    /// Different bytes under the same version is the mistake that makes a
    /// promotion receipt worthless, so it is refused rather than recorded.
    #[tokio::test]
    async fn applied_content_cannot_change_under_the_same_version() {
        let repo = repo().await;
        repo.put(
            "orders",
            &put("1.0.0", "sha256:aaa", PackageState::Applied),
            "ci",
        )
        .await
        .expect("apply");

        let err = repo
            .put(
                "orders",
                &put("1.0.0", "sha256:zzz", PackageState::Applied),
                "ci",
            )
            .await
            .expect_err("different content under an applied version must conflict");
        assert!(
            matches!(err, OrionError::Conflict(ref m) if m.contains("immutable")),
            "expected the immutability conflict, got: {err:?}"
        );

        // And the stored row is the one that was applied, not the refused one.
        let stored = repo.get_by_name("orders").await.expect("read back");
        assert_eq!(stored[0].content_hash, "sha256:aaa");
    }

    /// An applied version cannot be demoted back to staged: the receipt is the
    /// record that this content ran here, and a later `staged` PUT would erase
    /// that without changing what is deployed.
    #[tokio::test]
    async fn an_applied_version_cannot_be_demoted_to_staged() {
        let repo = repo().await;
        repo.put(
            "orders",
            &put("1.0.0", "sha256:aaa", PackageState::Applied),
            "ci",
        )
        .await
        .expect("apply");
        assert!(
            repo.put(
                "orders",
                &put("1.0.0", "sha256:aaa", PackageState::Staged),
                "ci"
            )
            .await
            .is_err(),
            "an applied version must not be demotable to staged"
        );
    }

    /// `get_by_name` is newest-first, and a package with no receipts is a
    /// `NotFound` rather than an empty list — the distinction the admin route
    /// turns into 404 versus 200.
    #[tokio::test]
    async fn receipts_read_back_newest_first_and_a_miss_is_not_found() {
        let repo = repo().await;
        for version in ["1.0.0", "1.1.0", "1.2.0"] {
            repo.put(
                "orders",
                &put(version, "sha256:aaa", PackageState::Applied),
                "ci",
            )
            .await
            .expect("apply");
        }

        let receipts = repo.get_by_name("orders").await.expect("read back");
        let versions: Vec<&str> = receipts.iter().map(|r| r.version.as_str()).collect();
        assert_eq!(versions, ["1.2.0", "1.1.0", "1.0.0"]);

        assert!(
            matches!(
                repo.get_by_name("no-such-package").await,
                Err(OrionError::NotFound(_))
            ),
            "a package with no receipts must be NotFound, not an empty list"
        );
    }

    /// Receipts for different packages do not collide on version: the key is
    /// `(name, version)`, and a shared version string is the normal case in an
    /// estate that versions its packages together.
    #[tokio::test]
    async fn the_receipt_key_is_name_and_version_together() {
        let repo = repo().await;
        repo.put(
            "orders",
            &put("1.0.0", "sha256:aaa", PackageState::Applied),
            "ci",
        )
        .await
        .expect("apply orders");
        repo.put(
            "billing",
            &put("1.0.0", "sha256:bbb", PackageState::Applied),
            "ci",
        )
        .await
        .expect("a different package at the same version must not conflict");

        assert_eq!(
            repo.get_by_name("billing").await.expect("read back")[0].content_hash,
            "sha256:bbb"
        );
        let page = repo.list(50, 0).await.expect("list");
        assert_eq!(page.total, 2);
    }

    /// The inventory follows the receipt's own mutability: stored on insert,
    /// replaced while staged, and on an applied version written only where
    /// none was recorded.
    #[tokio::test]
    async fn the_inventory_is_stored_replaced_while_staged_and_frozen_once_applied() {
        let repo = repo().await;
        let staged = repo
            .put(
                "orders",
                &with_channels(
                    put("1.0.0", "sha256:aaa", PackageState::Staged),
                    &["b", "a", "a"],
                ),
                "ci",
            )
            .await
            .expect("stage");
        assert_eq!(
            channels_of(&staged),
            Some(vec!["a".to_string(), "b".to_string()]),
            "stored sorted and de-duplicated"
        );

        let restaged = repo
            .put(
                "orders",
                &with_channels(put("1.0.0", "sha256:bbb", PackageState::Staged), &["c"]),
                "ci",
            )
            .await
            .expect("re-stage");
        assert_eq!(channels_of(&restaged), Some(vec!["c".to_string()]));

        let applied = repo
            .put(
                "orders",
                &put("1.0.0", "sha256:bbb", PackageState::Applied),
                "ci",
            )
            .await
            .expect("apply without an inventory");
        assert_eq!(
            channels_of(&applied),
            Some(vec!["c".to_string()]),
            "a PUT without an inventory leaves the recorded one"
        );

        let touched = repo
            .put(
                "orders",
                &with_channels(put("1.0.0", "sha256:bbb", PackageState::Applied), &["z"]),
                "ci",
            )
            .await
            .expect("re-apply");
        assert_eq!(
            channels_of(&touched),
            Some(vec!["c".to_string()]),
            "an applied version's recorded inventory is never replaced"
        );
    }

    /// A receipt applied before inventories existed gains one on its next
    /// idempotent re-apply — the only write an applied row takes.
    #[tokio::test]
    async fn an_applied_receipt_without_an_inventory_is_backfilled() {
        let repo = repo().await;
        let applied = repo
            .put(
                "orders",
                &put("1.0.0", "sha256:aaa", PackageState::Applied),
                "ci",
            )
            .await
            .expect("apply");
        assert_eq!(channels_of(&applied), None);
        let touched = repo
            .put(
                "orders",
                &with_channels(put("1.0.0", "sha256:aaa", PackageState::Applied), &["a"]),
                "ci",
            )
            .await
            .expect("re-apply");
        assert_eq!(channels_of(&touched), Some(vec!["a".to_string()]));
    }

    /// `list_current` is each package's newest applied receipt, and a
    /// package with only staged receipts has none.
    #[tokio::test]
    async fn list_current_is_the_newest_applied_receipt_per_package() {
        let repo = repo().await;
        for (name, version, state) in [
            ("orders", "1.0.0", PackageState::Applied),
            ("orders", "1.1.0", PackageState::Applied),
            ("orders", "1.2.0", PackageState::Staged),
            ("billing", "2.0.0", PackageState::Applied),
            ("drafts", "0.1.0", PackageState::Staged),
        ] {
            repo.put(name, &put(version, "sha256:aaa", state), "ci")
                .await
                .expect("put");
        }
        let page = repo.list_current(50, 0).await.expect("list_current");
        let current: Vec<(&str, &str)> = page
            .data
            .iter()
            .map(|r| (r.name.as_str(), r.version.as_str()))
            .collect();
        assert_eq!(current, [("billing", "2.0.0"), ("orders", "1.1.0")]);
        assert_eq!(page.total, 2);
        let second = repo.list_current(1, 1).await.expect("second page");
        assert_eq!(second.data[0].name, "orders");
    }
}