teamcraft/services/

items.rs

//! Items (parts, assemblies, documents, software) and their search.

use chrono::Utc;
use serde_json::json;
use sqlx::{PgConnection, PgPool};
use uuid::Uuid;

use crate::{
    auth::Actor,
    domain::{self, Classification, ItemType, LifecycleState, ProjectVisibility, Role},
    error::{AppError, AppResult},
    models::{Item, ItemSummary, Revision, WhereUsed, WhereUsedLevel},
    services::{
        access, attributes, audit, files, notifications, projects, revisions, simulations, Page,
    },
};

pub(crate) const ITEM_SELECT: &str = "
    SELECT i.id, i.number, i.item_type, i.unit, i.project_id, p.code AS project_code,
           i.created_by, u.display_name AS created_by_name, i.created_at, i.classification,
           i.owner_id, ou.display_name AS owner_name
    FROM items i
    LEFT JOIN projects p ON p.id = i.project_id
    JOIN users u ON u.id = i.created_by
    JOIN users ou ON ou.id = i.owner_id";

#[derive(Clone, Debug, Default)]
pub struct ItemFilter {
    /// Free text matched against number, name and description.
    pub query: Option<String>,
    pub item_type: Option<ItemType>,
    pub project_id: Option<Uuid>,
    /// State of the latest revision.
    pub state: Option<LifecycleState>,
    /// CAD program (`cadformat::SYSTEMS` id): only items whose latest revision
    /// has a file of one of its formats.
    pub system: Option<String>,
    pub classification: Option<Classification>,
    /// Custom attribute key: only items whose latest revision has a value
    /// matching `attribute_value` (see `domain::parse_attribute_condition`).
    pub attribute: Option<String>,
    pub attribute_value: Option<String>,
    /// Simulations whose latest revision has a result value in this state
    /// (see `domain::ResultFilter`), optionally only for values whose name
    /// contains `result_name`.
    pub result: Option<domain::ResultFilter>,
    pub result_name: Option<String>,
}

/// Bind values of an attribute condition: key, LIKE pattern, yes/no, min, max.
type AttributeBinds = (
    Option<String>,
    Option<String>,
    Option<bool>,
    Option<f64>,
    Option<f64>,
);

async fn attribute_binds(pool: &PgPool, filter: &ItemFilter) -> AppResult<AttributeBinds> {
    let Some(key) = filter.attribute.as_deref().filter(|k| !k.trim().is_empty()) else {
        return Ok((None, None, None, None, None));
    };
    let def = attributes::get_by_key(pool, key)
        .await
        .map_err(|_| AppError::BadRequest(format!("Unknown attribute '{}'.", key.trim())))?;
    let data_type =
        domain::AttributeType::parse(&def.data_type).unwrap_or(domain::AttributeType::Text);
    let condition = domain::parse_attribute_condition(
        &def.label,
        data_type,
        filter.attribute_value.as_deref().unwrap_or(""),
    )?;
    let key = Some(def.key);
    Ok(match condition {
        domain::AttributeCondition::Present => (key, None, None, None, None),
        domain::AttributeCondition::Contains(text) => {
            (key, Some(domain::like_pattern(&text)), None, None, None)
        }
        domain::AttributeCondition::Bool(b) => (key, None, Some(b), None, None),
        domain::AttributeCondition::Range(min, max) => (key, None, None, min, max),
    })
}

/// Searches items the actor may read; every row carries its latest revision.
///
/// The latest revision of an item is looked up per item, so the query never
/// does that for the whole vault: a search text, a state, a CAD program or
/// an attribute condition first narrows the items to candidates found
/// set-based (any revision matches – a superset, the exact conditions on
/// the latest revision are checked afterwards), and the total comes from the
/// same query (`COUNT(*) OVER ()`). Without such a filter only the requested
/// page of items (by number) is looked up, and the total is a plain count.
pub async fn search(
    pool: &PgPool,
    actor: &Actor,
    filter: &ItemFilter,
    page: Page,
) -> AppResult<(Vec<ItemSummary>, i64)> {
    // $1 = LIKE pattern, $5 = prefix tsquery, $6 = formats the query names
    // ("SolidWorks", "NX", "STEP"), $7 = formats of the CAD program filter,
    // $8/$9 = limit/offset, $10 = the actor's hidden project groups,
    // $11 = classification, $12–$16 = attribute key, LIKE pattern, yes/no,
    // min, max, $17 = result filter, $18 = LIKE pattern of a result name.
    // A match is any of: item number, name/description/material or a custom
    // attribute value (substring or full-text), a file name, the extracted
    // text of the current version of a file of the latest revision, or a file
    // of a format the query names.
    const JOINS: &str = "
        LEFT JOIN projects p ON p.id = i.project_id
        CROSS JOIN LATERAL (
            SELECT r.id, r.code, r.state, r.name, r.description, r.material, r.search_vector,
                   r.attributes
            FROM revisions r WHERE r.item_id = i.id ORDER BY r.seq DESC LIMIT 1
        ) lr";
    // A file of the latest revision whose name, content or format matches.
    // In the condition it is checked last (only for items that match
    // nothing else); the file name is only looked up for the rows of the
    // page (PostgreSQL computes such sub-selects after ORDER BY / LIMIT).
    const FILE_MATCH: &str = "
        FROM files f
        JOIN file_versions fv ON fv.file_id = f.id AND fv.version = f.current_version
        WHERE $1::text IS NOT NULL AND f.revision_id = lr.id
          AND (f.filename ILIKE $1
               OR ($5::text IS NOT NULL AND fv.content_vector @@ to_tsquery('simple', $5))
               OR f.format = ANY($6::text[]))";
    // A result value of the revision: its name and its state against the limits.
    const RESULT_MATCH: &str = "($18::text IS NULL OR sr.name ILIKE $18)
          AND ($17::text IS NULL
               OR ($17 = 'failed' AND (RESULT_FAILED))
               OR ($17 = 'passed' AND (sr.lower_limit IS NOT NULL OR sr.upper_limit IS NOT NULL)
                   AND NOT (RESULT_FAILED))
               OR ($17 = 'above_limit' AND sr.upper_limit IS NOT NULL AND sr.value > sr.upper_limit)
               OR ($17 = 'below_limit' AND sr.lower_limit IS NOT NULL AND sr.value < sr.lower_limit))";
    let result_match = RESULT_MATCH.replace("RESULT_FAILED", simulations::RESULT_FAILED_SQL);
    let visible = access::item_visible("i", 10);
    // conditions on the item itself (also applied to the page of items)
    let item_conditions = format!(
        "($2::text IS NULL OR i.item_type = $2)
          AND ($3::uuid IS NULL OR i.project_id = $3)
          AND ($11::text IS NULL OR i.classification = $11)
          AND {visible}"
    );
    let where_ = format!(
        "($1::text IS NULL
               OR i.number ILIKE $1 OR lr.name ILIKE $1 OR lr.description ILIKE $1 OR lr.material ILIKE $1
               OR jsonb_path_query_array(lr.attributes, '$.*')::text ILIKE $1
               OR ($5::text IS NOT NULL AND lr.search_vector @@ to_tsquery('simple', $5))
               OR EXISTS (SELECT 1 {FILE_MATCH}))
          AND {item_conditions}
          AND ($4::text IS NULL OR lr.state = $4)
          AND ($7::text[] IS NULL OR EXISTS (
                SELECT 1 FROM files sf WHERE sf.revision_id = lr.id AND sf.format = ANY($7::text[])))
          AND ($12::text IS NULL OR (lr.attributes ? $12
               AND ($13::text IS NULL OR lr.attributes ->> $12 ILIKE $13)
               AND ($14::boolean IS NULL OR lr.attributes -> $12 = to_jsonb($14::boolean))
               AND ($15::float8 IS NULL OR CASE WHEN jsonb_typeof(lr.attributes -> $12) = 'number'
                    THEN (lr.attributes ->> $12)::float8 >= $15 ELSE FALSE END)
               AND ($16::float8 IS NULL OR CASE WHEN jsonb_typeof(lr.attributes -> $12) = 'number'
                    THEN (lr.attributes ->> $12)::float8 <= $16 ELSE FALSE END)))
          AND ($17::text IS NULL AND $18::text IS NULL OR EXISTS (
                SELECT 1 FROM simulation_results sr WHERE sr.revision_id = lr.id AND {result_match}))"
    );
    let columns = format!(
        "i.id, i.number, i.item_type, i.unit, p.code AS project_code, i.created_at,
         lr.id AS latest_revision_id, lr.code AS latest_code, lr.state AS latest_state, lr.name,
         (SELECT rr.code FROM revisions rr WHERE rr.item_id = i.id AND rr.state = 'released'
          ORDER BY rr.seq DESC LIMIT 1) AS released_code,
         (SELECT f.filename {FILE_MATCH} ORDER BY f.filename LIMIT 1) AS matched_file,
         i.classification, lr.attributes"
    );
    const ORDER: &str = "ORDER BY CASE WHEN $5::text IS NULL THEN 0
                       ELSE ts_rank(lr.search_vector, to_tsquery('simple', $5)) END DESC,
                  i.number";
    // Items with a revision or file that matches the text (in any revision).
    const TEXT_CANDIDATES: &str = "
        SELECT ci.id FROM items ci WHERE ci.number ILIKE $1
        UNION
        SELECT cr.item_id FROM revisions cr
        WHERE cr.name ILIKE $1 OR cr.description ILIKE $1 OR cr.material ILIKE $1
           OR jsonb_path_query_array(cr.attributes, '$.*')::text ILIKE $1
           OR ($5::text IS NOT NULL AND cr.search_vector @@ to_tsquery('simple', $5))
        UNION
        SELECT cr.item_id FROM files cf JOIN revisions cr ON cr.id = cf.revision_id
        WHERE cf.filename ILIKE $1 OR cf.format = ANY($6::text[])
        UNION
        SELECT cr.item_id FROM file_versions cv
        JOIN files cf ON cf.id = cv.file_id AND cv.version = cf.current_version
        JOIN revisions cr ON cr.id = cf.revision_id
        WHERE $5::text IS NOT NULL AND cv.content_vector @@ to_tsquery('simple', $5)";
    const STATE_CANDIDATES: &str = "SELECT cr.item_id FROM revisions cr WHERE cr.state = $4";
    const SYSTEM_CANDIDATES: &str = "
        SELECT cr.item_id FROM files cf JOIN revisions cr ON cr.id = cf.revision_id
        WHERE cf.format = ANY($7::text[])";
    const ATTRIBUTE_CANDIDATES: &str =
        "SELECT cr.item_id FROM revisions cr WHERE cr.attributes ? $12";
    let result_candidates = format!(
        "SELECT cr.item_id FROM simulation_results sr JOIN revisions cr ON cr.id = sr.revision_id
         WHERE {result_match}"
    );

    let classification = filter.classification.map(Classification::as_str);
    let (attr_key, attr_pattern, attr_bool, attr_min, attr_max) =
        attribute_binds(pool, filter).await?;
    let query = filter
        .query
        .as_deref()
        .map(str::trim)
        .filter(|q| !q.is_empty());
    let pattern = query.map(domain::like_pattern);
    let tsquery = query.and_then(crate::textindex::prefix_query);
    let result_state = filter.result.map(domain::ResultFilter::as_str);
    let result_pattern = filter
        .result_name
        .as_deref()
        .map(str::trim)
        .filter(|n| !n.is_empty())
        .map(domain::like_pattern);
    let item_type = filter.item_type.map(ItemType::as_str);
    let state = filter.state.map(LifecycleState::as_str);
    let named_formats = query
        .map(crate::cadformat::ids_named_by)
        .unwrap_or_default();
    let system_formats = filter
        .system
        .as_deref()
        .filter(|s| !s.is_empty())
        .map(crate::cadformat::ids_of_system);

    let mut narrow: Vec<&str> = Vec::new();
    if pattern.is_some() {
        narrow.push(TEXT_CANDIDATES);
    }
    if state.is_some() {
        narrow.push(STATE_CANDIDATES);
    }
    if system_formats.is_some() {
        narrow.push(SYSTEM_CANDIDATES);
    }
    if attr_key.is_some() {
        narrow.push(ATTRIBUTE_CANDIDATES);
    }
    if result_state.is_some() || result_pattern.is_some() {
        narrow.push(&result_candidates);
    }
    let candidates: String = narrow
        .iter()
        .map(|c| format!(" AND i.id IN ({c})"))
        .collect();

    #[derive(sqlx::FromRow)]
    struct Row {
        #[sqlx(flatten)]
        item: ItemSummary,
        total: i64,
    }
    let rows_sql = if narrow.is_empty() {
        // Only the page of (visible) items is looked up; every item has at
        // least one revision (created together, revisions are never deleted
        // alone), so the latest-revision conditions cannot drop any item here.
        format!(
            "WITH page AS (
                 SELECT i.id FROM items i
                 WHERE {item_conditions}
                 ORDER BY i.number LIMIT $8 OFFSET $9
             )
             SELECT {columns}, 0::bigint AS total
             FROM page JOIN items i ON i.id = page.id {JOINS}
             WHERE {where_}
             {ORDER}"
        )
    } else {
        format!(
            "SELECT {columns}, COUNT(*) OVER () AS total
             FROM items i {JOINS}
             WHERE {where_} {candidates}
             {ORDER}
             LIMIT $8 OFFSET $9"
        )
    };
    let rows = sqlx::query_as::<_, Row>(&rows_sql)
        .bind(&pattern)
        .bind(item_type)
        .bind(filter.project_id)
        .bind(state)
        .bind(&tsquery)
        .bind(&named_formats)
        .bind(&system_formats)
        .bind(page.limit)
        .bind(page.offset)
        .bind(&actor.hidden_projects)
        .bind(classification)
        .bind(&attr_key)
        .bind(&attr_pattern)
        .bind(attr_bool)
        .bind(attr_min)
        .bind(attr_max)
        .bind(result_state)
        .bind(&result_pattern)
        .fetch_all(pool)
        .await?;
    // The window total is there whenever the page has rows; a page behind
    // the last one (and the unfiltered listing) needs the count query.
    let total = match rows.first() {
        _ if narrow.is_empty() => {
            sqlx::query_scalar::<_, i64>(&format!(
                "SELECT COUNT(*) FROM items i
                 WHERE ($1::text IS NULL OR i.item_type = $1) AND ($2::uuid IS NULL OR i.project_id = $2)
                   AND ($4::text IS NULL OR i.classification = $4)
                   AND {}",
                access::item_visible("i", 3)
            ))
            .bind(item_type)
            .bind(filter.project_id)
            .bind(&actor.hidden_projects)
            .bind(classification)
            .fetch_one(pool)
            .await?
        }
        Some(row) => row.total,
        None if page.offset == 0 => 0,
        None => {
            // same parameter positions as the page query ($8/$9 unused)
            sqlx::query_scalar::<_, i64>(&format!(
                "SELECT COUNT(*) FROM items i {JOINS} WHERE {where_} {candidates}
                   AND $8::bigint IS NOT NULL AND $9::bigint IS NOT NULL"
            ))
            .bind(&pattern)
            .bind(item_type)
            .bind(filter.project_id)
            .bind(state)
            .bind(&tsquery)
            .bind(&named_formats)
            .bind(&system_formats)
            .bind(page.limit)
            .bind(page.offset)
            .bind(&actor.hidden_projects)
            .bind(classification)
            .bind(&attr_key)
            .bind(&attr_pattern)
            .bind(attr_bool)
            .bind(attr_min)
            .bind(attr_max)
        .bind(result_state)
        .bind(&result_pattern)
            .fetch_one(pool)
            .await?
        }
    };
    Ok((rows.into_iter().map(|r| r.item).collect(), total))
}

pub(crate) async fn get_unchecked(pool: impl sqlx::PgExecutor<'_>, id: Uuid) -> AppResult<Item> {
    sqlx::query_as::<_, Item>(&format!("{ITEM_SELECT} WHERE i.id = $1"))
        .bind(id)
        .fetch_optional(pool)
        .await?
        .ok_or_else(|| AppError::not_found("Item not found."))
}

pub(crate) async fn get_by_number_unchecked(pool: &PgPool, number: &str) -> AppResult<Item> {
    let number = number.trim().to_ascii_uppercase();
    sqlx::query_as::<_, Item>(&format!("{ITEM_SELECT} WHERE i.number = $1"))
        .bind(&number)
        .fetch_optional(pool)
        .await?
        .ok_or_else(|| AppError::not_found(format!("Item '{number}' not found.")))
}

/// All revisions of an item, newest first.
pub(crate) async fn revisions_unchecked(pool: &PgPool, item_id: Uuid) -> AppResult<Vec<Revision>> {
    Ok(sqlx::query_as::<_, Revision>(&format!(
        "{} WHERE r.item_id = $1 ORDER BY r.seq DESC",
        revisions::REVISION_SELECT
    ))
    .bind(item_id)
    .fetch_all(pool)
    .await?)
}

pub(crate) async fn latest_revision_unchecked(pool: &PgPool, item_id: Uuid) -> AppResult<Revision> {
    sqlx::query_as::<_, Revision>(&format!(
        "{} WHERE r.item_id = $1 ORDER BY r.seq DESC LIMIT 1",
        revisions::REVISION_SELECT
    ))
    .bind(item_id)
    .fetch_optional(pool)
    .await?
    // every item has a revision: none means no item, and the message must
    // be the one of a hidden item (`latest_revision`)
    .ok_or_else(access::not_found_item)
}

/// The in-work or in-review revision of an item, if any.
pub(crate) async fn open_revision_unchecked(
    pool: &PgPool,
    item_id: Uuid,
) -> AppResult<Option<Revision>> {
    Ok(sqlx::query_as::<_, Revision>(&format!(
        "{} WHERE r.item_id = $1 AND r.state IN ('in_work', 'in_review')",
        revisions::REVISION_SELECT
    ))
    .bind(item_id)
    .fetch_optional(pool)
    .await?)
}

/// An item the actor may read; hidden items answer like missing ones.
pub async fn get(pool: &PgPool, actor: &Actor, id: Uuid) -> AppResult<Item> {
    let item = get_unchecked(pool, id).await?;
    access::ensure_visible(actor, item.project_id, access::not_found_item())?;
    Ok(item)
}

/// An item by number, if the actor may read it. A hidden item gives exactly
/// the message of an unknown number.
pub async fn get_by_number(pool: &PgPool, actor: &Actor, number: &str) -> AppResult<Item> {
    let item = get_by_number_unchecked(pool, number).await?;
    access::ensure_visible(
        actor,
        item.project_id,
        AppError::not_found(format!(
            "Item '{}' not found.",
            number.trim().to_ascii_uppercase()
        )),
    )?;
    Ok(item)
}

/// All revisions of an item the actor may read, newest first.
pub async fn revisions(pool: &PgPool, actor: &Actor, item_id: Uuid) -> AppResult<Vec<Revision>> {
    let revs = revisions_unchecked(pool, item_id).await?;
    if let Some(first) = revs.first() {
        access::ensure_visible(actor, first.project_id, access::not_found_item())?;
    }
    Ok(revs)
}

/// The latest revision of an item the actor may read.
pub async fn latest_revision(pool: &PgPool, actor: &Actor, item_id: Uuid) -> AppResult<Revision> {
    let rev = latest_revision_unchecked(pool, item_id).await?;
    access::ensure_visible(actor, rev.project_id, access::not_found_item())?;
    Ok(rev)
}

/// The in-work or in-review revision of an item the actor may read.
pub async fn open_revision(
    pool: &PgPool,
    actor: &Actor,
    item_id: Uuid,
) -> AppResult<Option<Revision>> {
    let rev = open_revision_unchecked(pool, item_id).await?;
    if let Some(rev) = &rev {
        access::ensure_visible(actor, rev.project_id, access::not_found_item())?;
    }
    Ok(rev)
}

#[derive(Clone, Debug)]
pub struct NewItem {
    /// Leave empty to generate a number such as `PRT-000042`.
    pub number: Option<String>,
    pub item_type: ItemType,
    pub unit: String,
    pub project_id: Option<Uuid>,
    pub name: String,
    pub description: String,
    pub material: String,
    pub mass_kg: Option<f64>,
    pub unit_cost: Option<f64>,
    /// Make, purchased or standard part.
    pub classification: Classification,
    /// Custom attribute values, `key → value as text` (empty ones are skipped).
    pub attributes: Vec<(String, String)>,
}

fn validate_unit(unit: &str) -> AppResult<String> {
    let unit = unit.trim();
    if unit.is_empty() {
        return Ok("each".into());
    }
    domain::limit_text("Unit", unit, 16)
}

/// Creates an item together with its first revision `A` (in work) and
/// returns that revision.
///
/// # Invariants
///
/// * **Permission:** editor role (403 otherwise); a project group is required
///   (400), it must exist and be visible (400 "Unknown project"), not
///   archived (400) and the actor a member of it (403).
/// * **Locks:** the project group row (`FOR SHARE`, so a visibility change
///   cannot move the counters meanwhile) and the number counter row, which
///   stays locked until the commit.
/// * **Audit:** `item.create`.
/// * **Notifications:** none; the creator only starts watching the item.
/// * **Webhooks:** none.
/// * **Data:** a generated number steps the group's counter (numbers already
///   taken are skipped); the first revision `A` is in work and checked out by
///   the creator. A failed creation rolls the counter back with it.
/// * **Atomic:** one transaction; concurrent creations get distinct numbers
///   (`tests/numbering_and_watches.rs`).
pub async fn create(pool: &PgPool, actor: &Actor, input: NewItem) -> AppResult<Revision> {
    let mut tx = pool.begin().await?;
    let created = create_tx(&mut tx, actor, input).await?;
    tx.commit().await?;
    revisions::get(pool, actor, created.revision_id).await
}

/// The longest idempotency key accepted.
pub const MAX_IDEMPOTENCY_KEY: usize = 255;

/// Validates a client-supplied idempotency key (printable ASCII, 1..=255).
pub fn validate_idempotency_key(key: &str) -> AppResult<&str> {
    let key = key.trim();
    if key.is_empty()
        || key.len() > MAX_IDEMPOTENCY_KEY
        || !key.bytes().all(|b| (0x21..=0x7e).contains(&b))
    {
        return Err(AppError::BadRequest(format!(
            "idempotency_key must be 1 to {MAX_IDEMPOTENCY_KEY} printable ASCII characters without spaces."
        )));
    }
    Ok(key)
}

/// [`create`] with an idempotency key: the first request with a key creates
/// the item; a repetition by the same user with the same `payload_hash`
/// returns that item (second value `true`, nothing is created); the same key
/// with another payload is a conflict (409).
///
/// # Invariants
///
/// * **Permission / locks / audit / data:** as [`create`]; a replay is a
///   read: the original revision is returned only while the actor may read it.
/// * **Atomic:** the key row is inserted first, in the creation's transaction.
///   Two simultaneous requests with one key meet at the primary key: the
///   second waits for the first to commit (replay) or roll back (it then
///   creates the item itself), so exactly one item exists. A refused or
///   failed creation leaves no key behind.
pub async fn create_idempotent(
    pool: &PgPool,
    actor: &Actor,
    input: NewItem,
    key: &str,
    payload_hash: &str,
) -> AppResult<(Revision, bool)> {
    let key = validate_idempotency_key(key)?;
    let mut tx = pool.begin().await?;
    let inserted = sqlx::query(
        "INSERT INTO item_idempotency (user_id, key, payload_hash) VALUES ($1, $2, $3)
         ON CONFLICT (user_id, key) DO NOTHING",
    )
    .bind(actor.id)
    .bind(key)
    .bind(payload_hash)
    .execute(&mut *tx)
    .await?
    .rows_affected();
    if inserted == 0 {
        let (hash, revision_id) = sqlx::query_as::<_, (String, Option<Uuid>)>(
            "SELECT payload_hash, revision_id FROM item_idempotency WHERE user_id = $1 AND key = $2",
        )
        .bind(actor.id)
        .bind(key)
        .fetch_one(&mut *tx)
        .await?;
        drop(tx);
        if hash != payload_hash {
            return Err(AppError::Conflict(
                "This idempotency key was used for a different request; use a new key for a different item.".into(),
            ));
        }
        let revision_id = revision_id.ok_or_else(|| {
            AppError::Conflict("The request with this idempotency key is still running.".into())
        })?;
        return Ok((revisions::get(pool, actor, revision_id).await?, true));
    }
    let created = create_tx(&mut tx, actor, input).await?;
    sqlx::query("UPDATE item_idempotency SET revision_id = $3 WHERE user_id = $1 AND key = $2")
        .bind(actor.id)
        .bind(key)
        .bind(created.revision_id)
        .execute(&mut *tx)
        .await?;
    tx.commit().await?;
    Ok((
        revisions::get(pool, actor, created.revision_id).await?,
        false,
    ))
}

/// Creates an item and stores its first files in the new revision – one
/// transaction, so there is never an item without the files it was created
/// with. The revision stays checked out by the creator.
///
/// # Invariants
///
/// * **Permission / locks / audit / data:** as [`create`], then the new
///   revision is locked (`revisions::lock`) and the files are stored like in
///   [`files::attach`] (`file.upload` per file).
/// * **Notifications:** none; the creator only starts watching the item.
/// * **Webhooks:** `simulation.result_added` when the item is a simulation
///   and result values are read from the uploaded files (see
///   [`files::attach`]); otherwise none.
/// * **Atomic:** one transaction; a refused upload leaves no item behind.
pub async fn create_with_files(
    pool: &PgPool,
    actor: &Actor,
    input: NewItem,
    uploads: &[files::Upload],
) -> AppResult<Revision> {
    let mut tx = pool.begin().await?;
    let created = create_tx(&mut tx, actor, input).await?;
    if !uploads.is_empty() {
        let rev = revisions::lock(&mut tx, created.revision_id).await?;
        files::store_uploads(&mut tx, actor, &rev, uploads, "").await?;
    }
    tx.commit().await?;
    revisions::get(pool, actor, created.revision_id).await
}

/// What [`create_tx`] created.
#[derive(Clone, Debug)]
pub(crate) struct Created {
    pub item_id: Uuid,
    pub revision_id: Uuid,
    pub number: String,
}

/// Creates an item inside the caller's transaction (imports create many items
/// atomically). The first revision is checked out by the actor.
pub(crate) async fn create_tx(
    conn: &mut PgConnection,
    actor: &Actor,
    input: NewItem,
) -> AppResult<Created> {
    actor.require(Role::Editor)?;
    let attrs = revisions::RevisionUpdate {
        name: input.name,
        description: input.description,
        material: input.material,
        mass_kg: input.mass_kg,
        unit_cost: input.unit_cost,
        attributes: input.attributes,
    }
    .validated()?;
    let unit = validate_unit(&input.unit)?;
    let attribute_values = attributes::apply_changes(
        &mut *conn,
        input.item_type.as_str(),
        &serde_json::json!({}),
        &attrs.attributes,
    )
    .await?;
    let manual_number = match input.number.as_deref().map(str::trim) {
        Some(n) if !n.is_empty() => Some(domain::normalize_item_number(n)?),
        _ => None,
    };

    // Every new item belongs to a project group the author is a member of.
    let project_id = input.project_id.ok_or_else(|| {
        AppError::BadRequest("Choose the project group the item belongs to.".into())
    })?;
    let tx = conn;
    let (code, active, scheme, visibility) =
        sqlx::query_as::<_, (String, bool, Option<String>, String)>(
            // FOR SHARE: a visibility change moves the counters (number_counter)
            "SELECT code, active, number_scheme, visibility FROM projects WHERE id = $1 FOR SHARE",
        )
        .bind(project_id)
        .fetch_optional(&mut *tx)
        .await?
        .ok_or_else(access::unknown_project)?;
    access::ensure_visible(actor, Some(project_id), access::unknown_project())?;
    if !active {
        return Err(AppError::BadRequest(format!(
            "The project group {code} is archived."
        )));
    }
    if !actor.can_edit_project(Some(project_id)) {
        return Err(AppError::Forbidden(format!(
            "You are not a member of the project group {code}; ask an admin to add you."
        )));
    }

    let number = match manual_number {
        Some(n) => n,
        None => {
            // The project group's numbering scheme; numbers that were already
            // entered manually (or by another scheme) are skipped.
            let scheme = domain::NumberScheme::of_project(scheme.as_deref())?;
            let ctx = domain::number_context(&code, input.item_type, Utc::now());
            let (counter, seed) = projects::number_counter(project_id, &visibility, &scheme, &ctx);
            let own_scheme = scheme != domain::NumberScheme::default_scheme();
            let mut attempts = 0;
            let mut jumped = false;
            loop {
                let next = projects::next_number(tx, &counter, seed.as_deref()).await?;
                let generated = scheme.format(&ctx, next);
                let candidate = domain::normalize_item_number(&generated).map_err(|_| {
                    AppError::BadRequest(format!(
                        "The numbering scheme of the project group {code} gives the invalid item number '{generated}'; enter a number or change the scheme."
                    ))
                })?;
                // the group of the item that holds the number already
                let taken = sqlx::query_scalar::<_, Option<Uuid>>(
                    "SELECT project_id FROM items WHERE number = $1",
                )
                .bind(&candidate)
                .fetch_optional(&mut *tx)
                .await?;
                let Some(holder) = taken else {
                    break candidate;
                };
                // A scheme of the group's own that runs into the numbers of a
                // group the actor may not see (e.g. one copying a restricted
                // group's prefix) stops instead of skipping: skipping would
                // count the hidden items in one call, follow every new one
                // and use up the hidden group's next numbers. The error says
                // no more than typing that one number in (and the counter
                // stays where it was). The built-in shared counter has
                // counted past every generated number already, so it only
                // meets typed-in numbers and keeps skipping them.
                if own_scheme && !actor.can_see_project(holder) {
                    return Err(AppError::Conflict(format!(
                        "The numbering scheme of the project group {code} gives the item number '{candidate}', which is already in use; change the scheme or enter a number."
                    )));
                }
                if !jumped {
                    // Numbers of this pattern exist already (entered by hand,
                    // imported, or a scheme that continues an old
                    // convention): continue after the highest one at once.
                    // Only items the actor can see count, so the jump tells
                    // nothing about hidden groups (their numbers are met
                    // one by one as above).
                    jumped = true;
                    let (prefix, _) = scheme.around_seq(&ctx);
                    let pattern = format!(
                        "{}%",
                        prefix
                            .replace('\\', "\\\\")
                            .replace('%', "\\%")
                            .replace('_', "\\_")
                    );
                    let highest = sqlx::query_as::<_, (String, Option<Uuid>)>(
                        "SELECT number, project_id FROM items WHERE number LIKE $1 ESCAPE '\\'",
                    )
                    .bind(&pattern)
                    .fetch_all(&mut *tx)
                    .await?
                    .iter()
                    .filter(|(_, project)| actor.can_see_project(*project))
                    .filter_map(|(n, _)| scheme.seq_of(&ctx, n))
                    .max();
                    if let Some(highest) = highest {
                        sqlx::query(
                            "UPDATE counters SET value = GREATEST(value, $2) WHERE name = $1",
                        )
                        .bind(&counter)
                        .bind(highest)
                        .execute(&mut *tx)
                        .await?;
                    }
                    continue;
                }
                attempts += 1;
                if attempts > 1000 {
                    return Err(AppError::Conflict(
                        "Could not allocate an item number.".into(),
                    ));
                }
            }
        }
    };

    let item_id = Uuid::new_v4();
    sqlx::query(
        "INSERT INTO items (id, number, item_type, unit, project_id, created_by, owner_id, classification)
         VALUES ($1, $2, $3, $4, $5, $6, $6, $7)",
    )
    .bind(item_id)
    .bind(&number)
    .bind(input.item_type.as_str())
    .bind(&unit)
    .bind(project_id)
    .bind(actor.id)
    .bind(input.classification.as_str())
    .execute(&mut *tx)
    .await
    .map_err(|e| {
        AppError::unique_violation(e, format!("The item number '{number}' is already in use."))
    })?;

    let rev_id = Uuid::new_v4();
    sqlx::query(
        "INSERT INTO revisions (id, item_id, seq, code, state, name, description, material,
                                mass_kg, unit_cost, created_by, checked_out_by, checked_out_at,
                                attributes)
         VALUES ($1, $2, 1, $3, 'in_work', $4, $5, $6, $7, $8, $9, $9, now(), $10)",
    )
    .bind(rev_id)
    .bind(item_id)
    .bind(domain::revision_code(1))
    .bind(&attrs.name)
    .bind(&attrs.description)
    .bind(&attrs.material)
    .bind(attrs.mass_kg)
    .bind(attrs.unit_cost)
    .bind(actor.id)
    .bind(&attribute_values)
    .execute(&mut *tx)
    .await?;

    audit::record(
        tx,
        Some(actor),
        "item.create",
        "item",
        Some(item_id),
        format!(
            "Created {} {number} – {}",
            input.item_type.label(),
            attrs.name
        ),
        json!({ "revision_id": rev_id, "item_type": input.item_type.as_str() }),
    )
    .await?;
    // The creator follows the item (notifications about checks and releases).
    notifications::add_watch(tx, actor.id, "item", item_id).await?;

    Ok(Created {
        item_id,
        revision_id: rev_id,
        number,
    })
}

/// Updates item master data. The unit can only change while nothing has
/// been released yet, because released BOMs depend on it.
///
/// # Invariants
///
/// * **Permission:** editor role (403); the item is visible to the actor
///   ("not found" otherwise) and the actor a member of its project group and
///   of the target group (403). A missing target group is a 400. Moving out
///   of a restricted group needs a manager of that group or an admin (403,
///   `ensure_may_take_out`).
/// * **Locks:** the item row (`FOR UPDATE`).
/// * **Audit:** `item.update`.
/// * **Notifications / webhooks:** none.
/// * **Data:** the unit cannot change once a revision is in review, released
///   or obsolete (409).
/// * **Atomic:** one transaction.
pub async fn update_master(
    pool: &PgPool,
    actor: &Actor,
    item_id: Uuid,
    project_id: Option<Uuid>,
    unit: &str,
    classification: Option<Classification>,
) -> AppResult<Item> {
    actor.require(Role::Editor)?;
    let unit = validate_unit(unit)?;
    let mut tx = pool.begin().await?;
    let current =
        sqlx::query_as::<_, Item>(&format!("{ITEM_SELECT} WHERE i.id = $1 FOR UPDATE OF i"))
            .bind(item_id)
            .fetch_optional(&mut *tx)
            .await?
            .ok_or_else(|| AppError::not_found("Item not found."))?;
    access::ensure_visible(actor, current.project_id, access::not_found_item())?;
    // Moving an item requires membership in the old and the new project group.
    if !actor.can_edit_project(current.project_id) {
        return Err(AppError::Forbidden(format!(
            "Only members of the project group {} can change {}.",
            current.project_code.as_deref().unwrap_or("?"),
            current.number
        )));
    }
    let Some(pid) = project_id else {
        return Err(AppError::BadRequest("Choose a project group.".into()));
    };
    let exists =
        sqlx::query_scalar::<_, bool>("SELECT EXISTS(SELECT 1 FROM projects WHERE id = $1)")
            .bind(pid)
            .fetch_one(&mut *tx)
            .await?;
    if !exists {
        return Err(access::unknown_project());
    }
    access::ensure_visible(actor, Some(pid), access::unknown_project())?;
    if !actor.can_edit_project(Some(pid)) {
        return Err(AppError::Forbidden(
            "You can only move items into project groups you are a member of.".into(),
        ));
    }
    // Moving out of a restricted group declassifies the item: only a manager
    // of that group (or an admin) may do it.
    if current.project_id != Some(pid) {
        if let Some(source) = current.project_id {
            ensure_may_take_out(&mut tx, actor, source, "Moving").await?;
        }
    }
    if unit != current.unit {
        let released = sqlx::query_scalar::<_, bool>(
            "SELECT EXISTS(SELECT 1 FROM revisions WHERE item_id = $1 AND state IN ('released', 'obsolete', 'in_review'))",
        )
        .bind(item_id)
        .fetch_one(&mut *tx)
        .await?;
        if released {
            return Err(AppError::Conflict(
                "The unit cannot be changed after a revision was submitted or released.".into(),
            ));
        }
    }
    let classification = classification
        .map(Classification::as_str)
        .unwrap_or(current.classification.as_str());
    sqlx::query("UPDATE items SET project_id = $2, unit = $3, classification = $4 WHERE id = $1")
        .bind(item_id)
        .bind(project_id)
        .bind(&unit)
        .bind(classification)
        .execute(&mut *tx)
        .await?;
    audit::record(
        &mut tx,
        Some(actor),
        "item.update",
        "item",
        Some(item_id),
        format!("Updated master data of {}", current.number),
        json!({ "project_id": project_id, "unit": unit, "classification": classification }),
    )
    .await?;
    tx.commit().await?;
    get(pool, actor, item_id).await
}

/// Fails unless the actor may move or copy data out of the project group
/// `source`: always allowed for internal groups; for restricted groups only
/// for administrators and managers of the group. `what` = "Moving" or
/// "Copying".
pub(crate) async fn ensure_may_take_out(
    conn: &mut PgConnection,
    actor: &Actor,
    source: Uuid,
    what: &str,
) -> AppResult<()> {
    let restricted = sqlx::query_scalar::<_, bool>(
        "SELECT EXISTS(SELECT 1 FROM projects WHERE id = $1 AND visibility = 'restricted')",
    )
    .bind(source)
    .fetch_one(&mut *conn)
    .await?;
    if restricted && !crate::services::projects::can_manage(conn, actor, source).await? {
        return Err(AppError::Forbidden(format!(
            "{what} out of a restricted group requires a manager of that group."
        )));
    }
    Ok(())
}

/// Deletes an item created by mistake. Only allowed while it has a single
/// in-work revision, is not used in any BOM, not part of any change, no
/// simulation basis, no design that units were built to and – in a
/// restricted group – no affected item of a change request.
///
/// # Invariants
///
/// * **Permission:** editor role (403); the item is visible to the actor
///   ("not found" otherwise); only its creator (as a member of the group) or
///   an admin may delete it (403).
/// * **Locks:** the item row (`FOR UPDATE`); the project group row
///   (`FOR SHARE`) when the item has one.
/// * **Audit:** `item.delete`; in an unrestricted group also
///   `change_request.remove_item` for every change request that named the item.
/// * **Notifications:** the people of those change requests (unrestricted
///   groups only).
/// * **Webhooks:** none.
/// * **Refused with 409:** more than one revision or one not in work, used in
///   a BOM, part of a change, basis of a simulation, units were recorded as
///   built to it (a build record names one of its revisions), or
///   (restricted group) an affected item of a change request.
/// * **Atomic:** one transaction.
pub async fn delete(pool: &PgPool, actor: &Actor, item_id: Uuid) -> AppResult<String> {
    actor.require(Role::Editor)?;
    let mut tx = pool.begin().await?;
    let item = sqlx::query_as::<_, Item>(&format!("{ITEM_SELECT} WHERE i.id = $1 FOR UPDATE OF i"))
        .bind(item_id)
        .fetch_optional(&mut *tx)
        .await?
        .ok_or_else(|| AppError::not_found("Item not found."))?;
    access::ensure_visible(actor, item.project_id, access::not_found_item())?;
    if (item.created_by != actor.id || !actor.can_edit_project(item.project_id))
        && !actor.is_admin()
    {
        return Err(AppError::Forbidden(
            "Only the creator (as member of the project group) or an administrator can delete an item.".into(),
        ));
    }
    let not_in_work = sqlx::query_scalar::<_, i64>(
        "SELECT COUNT(*) FROM revisions WHERE item_id = $1 AND (state <> 'in_work' OR seq > 1)",
    )
    .bind(item_id)
    .fetch_one(&mut *tx)
    .await?;
    if not_in_work > 0 {
        return Err(AppError::Conflict(
            "Items that were ever submitted, released or revised cannot be deleted; mark them obsolete instead.".into(),
        ));
    }
    let used = sqlx::query_scalar::<_, bool>(
        "SELECT EXISTS(SELECT 1 FROM bom_lines WHERE child_item_id = $1)",
    )
    .bind(item_id)
    .fetch_one(&mut *tx)
    .await?;
    if used {
        return Err(AppError::Conflict(
            "The item is used in a BOM; remove it from all BOMs first.".into(),
        ));
    }
    let in_change = sqlx::query_scalar::<_, bool>(
        "SELECT EXISTS(SELECT 1 FROM change_items ci JOIN revisions r ON r.id = ci.revision_id WHERE r.item_id = $1)",
    )
    .bind(item_id)
    .fetch_one(&mut *tx)
    .await?;
    if in_change {
        return Err(AppError::Conflict(
            "The item is part of a change; remove it from the change first.".into(),
        ));
    }
    if crate::services::simulations::item_is_basis(&mut tx, item_id).await? {
        return Err(AppError::Conflict(
            "A simulation is based on the item; remove it from the simulation's basis first."
                .into(),
        ));
    }
    if crate::services::as_built::item_is_design(&mut tx, item_id).await? {
        return Err(AppError::Conflict(
            "Units were recorded as built to the item; it stays for their traceability.".into(),
        ));
    }
    // A request is hidden while it names a hidden item; deleting an item of
    // a restricted group (the row would cascade) would silently declassify
    // the request. Items of internal groups hide no request, so their rows
    // simply go: refusing would tell the creator of a public item about a
    // request they cannot see (and cannot edit).
    let restricted = match item.project_id {
        Some(p) => sqlx::query_scalar::<_, String>(
            "SELECT visibility FROM projects WHERE id = $1 FOR SHARE",
        )
        .bind(p)
        .fetch_optional(&mut *tx)
        .await?
        .is_some_and(|v| v == ProjectVisibility::Restricted.as_str()),
        None => false,
    };
    let in_request = restricted
        && sqlx::query_scalar::<_, bool>(
            "SELECT EXISTS(SELECT 1 FROM change_request_items WHERE item_id = $1)",
        )
        .bind(item_id)
        .fetch_one(&mut *tx)
        .await?;
    if in_request {
        return Err(AppError::Conflict(
            "The item is an affected item of a change request; remove it from the request first."
                .into(),
        ));
    }
    if !restricted {
        // the rows go with the item: every request records it in its
        // history and tells its people (only readers of the request see it)
        crate::services::requests::item_deleted(&mut tx, actor, item_id, &item.number).await?;
    }
    sqlx::query("DELETE FROM items WHERE id = $1")
        .bind(item_id)
        .execute(&mut *tx)
        .await?;
    audit::record(
        &mut tx,
        Some(actor),
        "item.delete",
        "item",
        Some(item_id),
        format!("Deleted item {}", item.number),
        json!({}),
    )
    .await?;
    tx.commit().await?;
    Ok(item.number)
}

/// Parent revisions whose BOM contains the item, at most `limit` (a
/// standard part can be used in tens of thousands of BOMs) – only parents
/// the actor may read (see [`hidden_parents`] for the rest).
pub async fn where_used(
    pool: &PgPool,
    actor: &Actor,
    item_id: Uuid,
    limit: i64,
) -> AppResult<Vec<WhereUsed>> {
    Ok(sqlx::query_as::<_, WhereUsed>(&format!(
        "SELECT bl.id AS line_id, r.id AS parent_revision_id, i.id AS parent_item_id,
                i.number AS parent_number, r.code AS parent_code, r.name AS parent_name,
                r.state AS parent_state, bl.find_number, bl.quantity, rc.code AS released_child_code
         FROM bom_lines bl
         JOIN revisions r ON r.id = bl.parent_revision_id
         JOIN items i ON i.id = r.item_id
         LEFT JOIN revisions rc ON rc.id = bl.released_child_revision_id
         WHERE bl.child_item_id = $1 AND {}
         ORDER BY i.number, r.seq DESC
         LIMIT $3",
        access::item_visible("i", 2)
    ))
    .bind(item_id)
    .bind(&actor.hidden_projects)
    .bind(limit.max(1))
    .fetch_all(pool)
    .await?)
}

/// Number of distinct parent items that use the item but that the actor may
/// not read ("+N assemblies in groups you cannot see").
pub async fn hidden_parents(pool: &PgPool, actor: &Actor, item_id: Uuid) -> AppResult<i64> {
    if actor.hidden_projects.is_empty() {
        return Ok(0);
    }
    Ok(sqlx::query_scalar::<_, i64>(&format!(
        "SELECT COUNT(DISTINCT i.id)
         FROM bom_lines bl
         JOIN revisions r ON r.id = bl.parent_revision_id
         JOIN items i ON i.id = r.item_id
         WHERE bl.child_item_id = $1 AND NOT {}",
        access::item_visible("i", 2)
    ))
    .bind(item_id)
    .bind(&actor.hidden_projects)
    .fetch_one(pool)
    .await?)
}

/// Most levels a multi-level where-used report walks up.
pub const MAX_WHERE_USED_LEVELS: i32 = 25;
/// Most usages a multi-level where-used report returns.
pub const MAX_WHERE_USED_ROWS: i64 = 2000;

/// Multi-level where-used: every BOM line that uses the item or (up to
/// `levels` levels up) one of its parents. Like the single-level report it
/// looks at the BOMs of all revisions. Each item is reported on the lowest
/// level it is reached on: the walk goes up one level per query (the parents
/// not seen before form the next level), so a part used in thousands of
/// assemblies costs one query per level instead of a recursive expansion
/// of every path. The report stops at [`MAX_WHERE_USED_ROWS`] rows.
pub async fn where_used_levels(
    pool: &PgPool,
    actor: &Actor,
    item_id: Uuid,
    levels: i32,
) -> AppResult<Vec<WhereUsedLevel>> {
    let levels = levels.clamp(1, MAX_WHERE_USED_LEVELS);
    let mut out: Vec<WhereUsedLevel> = Vec::new();
    let mut seen = std::collections::HashSet::from([item_id]);
    let mut frontier = vec![item_id];
    // Only parents the actor may read are reported, and the walk goes up
    // only through them.
    let sql = format!(
        "SELECT $2::int AS level, ci.number AS child_number,
                bl.id AS line_id, r.id AS parent_revision_id, i.id AS parent_item_id,
                i.number AS parent_number, r.code AS parent_code, r.name AS parent_name,
                r.state AS parent_state, bl.find_number, bl.quantity, rc.code AS released_child_code
         FROM bom_lines bl
         JOIN items ci ON ci.id = bl.child_item_id
         JOIN revisions r ON r.id = bl.parent_revision_id
         JOIN items i ON i.id = r.item_id
         LEFT JOIN revisions rc ON rc.id = bl.released_child_revision_id
         WHERE bl.child_item_id = ANY($1) AND {}
         ORDER BY i.number, r.seq DESC, ci.number
         LIMIT $3",
        access::item_visible("i", 4)
    );
    for level in 1..=levels {
        let remaining = MAX_WHERE_USED_ROWS - out.len() as i64;
        if frontier.is_empty() || remaining <= 0 {
            break;
        }
        let rows = sqlx::query_as::<_, WhereUsedLevel>(&sql)
            .bind(&frontier)
            .bind(level)
            .bind(remaining)
            .bind(&actor.hidden_projects)
            .fetch_all(pool)
            .await?;
        // A truncated level ends the report, so the next frontier is only
        // needed when every row of this level was read.
        frontier = rows
            .iter()
            .map(|r| r.usage.parent_item_id)
            .filter(|id| seen.insert(*id))
            .collect();
        out.extend(rows);
    }
    Ok(out)
}