//! The Postgres index over `refs/notes/*`.
//!
//! <!-- wiki: mnw-server-git-notes -->
//!
//! Every row here is a projection of a git object. The repository is truth and
//! this table is rebuildable from it, so nothing in this module may be the only
//! copy of anything: see migration 197 and the wiki note's load-bearing rule.
//! The consequence for this file is that every write is a whole-namespace
//! statement of fact rather than an edit, and every read is allowed to be wrong
//! in the direction of "ask the repository instead".
//!
//! Two rules the callers depend on:
//!
//! - **A namespace's rows and its index-state row move together, in one
//!   transaction.** The state row says which notes-ref tip the rows were built
//!   from, and the next reindex diffs against it. A state row ahead of the rows
//!   it describes would make the next diff skip changes it never applied, and
//!   nothing later would notice.
//! - **Visibility is not stored and is not checked here.** These functions take
//!   a repo id and answer about that repository. Whether the person asking may
//!   see it is the caller's question, decided the same way it is for the
//!   repository's other pages.

use std::collections::HashMap;

use chrono::{DateTime, Utc};
use sqlx::PgPool;

use super::GitRepoId;
use crate::error::Result;

/// One note as the reindex hands it over.
///
/// `target_summary` and `target_time` are the annotated commit's, resolved once
/// at index time so the feed does not read a commit header per row. They are
/// empty and `None` for a note on a blob or a tree.
#[derive(Debug, Clone)]
pub struct NoteUpsert {
    pub target_oid: String,
    pub blob_oid: String,
    pub content: String,
    pub target_is_commit: bool,
    pub target_summary: String,
    pub target_time: Option<DateTime<Utc>>,
}

/// One indexed note, as the read paths want it.
#[derive(Debug, Clone, sqlx::FromRow)]
pub struct IndexedNote {
    pub namespace: String,
    pub target_oid: String,
    pub blob_oid: String,
    pub content: String,
    pub target_is_commit: bool,
    pub target_summary: String,
    pub target_time: Option<DateTime<Utc>>,
    pub updated_at: DateTime<Utc>,
    pub updated_by: String,
}

/// The tip a namespace was last indexed from, or `None` when it has never been
/// indexed.
///
/// The distinction is load-bearing on both sides. The reindex uses it as the old
/// side of the diff, so an absent row means "walk the whole namespace". The read
/// paths use it to tell a cold index from a repository with no notes, which an
/// empty result cannot distinguish.
#[tracing::instrument(skip_all)]
pub async fn indexed_tip(
    pool: &PgPool,
    repo_id: GitRepoId,
    namespace: &str,
) -> Result<Option<String>> {
    let tip = sqlx::query_scalar::<_, String>(
        "SELECT indexed_tip FROM git_notes_index_state WHERE repo_id = $1 AND namespace = $2",
    )
    .bind(repo_id)
    .bind(namespace)
    .fetch_optional(pool)
    .await?;
    Ok(tip)
}

/// Every namespace this repository has an index for, whether or not it still has
/// notes in it.
#[tracing::instrument(skip_all)]
pub async fn indexed_namespaces(pool: &PgPool, repo_id: GitRepoId) -> Result<Vec<String>> {
    let names = sqlx::query_scalar::<_, String>(
        "SELECT namespace FROM git_notes_index_state WHERE repo_id = $1 ORDER BY namespace",
    )
    .bind(repo_id)
    .fetch_all(pool)
    .await?;
    Ok(names)
}

/// Whether anything in this repository has been indexed.
///
/// The read paths ask this before trusting an empty result: a repository with no
/// state rows has a cold index, not an unannotated history, and the answer to a
/// cold index is to walk the repository rather than to report nothing.
#[tracing::instrument(skip_all)]
pub async fn is_indexed(pool: &PgPool, repo_id: GitRepoId) -> Result<bool> {
    let found = sqlx::query_scalar::<_, bool>(
        "SELECT EXISTS (SELECT 1 FROM git_notes_index_state WHERE repo_id = $1)",
    )
    .bind(repo_id)
    .fetch_one(pool)
    .await?;
    Ok(found)
}

/// One namespace's diff, and the tip it was computed against.
///
/// A struct rather than eight arguments because every field has to describe the
/// same reindex: a `tip` that did not produce these `upserts` is the one way to
/// corrupt the index silently, and keeping them in one value makes that hard to
/// do by accident.
pub struct NamespaceUpdate<'a> {
    pub namespace: &'a str,
    /// Notes added or changed since the previously indexed tip.
    pub upserts: &'a [NoteUpsert],
    /// Target ids whose note is gone, hex.
    pub removals: &'a [String],
    /// The notes-ref tip this diff brings the index up to, or `None` for a
    /// batch that is part of a larger diff. A batch that claimed the tip before
    /// the rest of the diff was written would make the next reindex skip the
    /// remainder; leaving the old tip in place makes it recompute and redo work
    /// that is idempotent anyway.
    pub tip: Option<&'a str>,
    /// When the annotations were written, and by whom: the notes commit's own
    /// committer, not the person who triggered the reindex.
    pub updated_at: DateTime<Utc>,
    pub updated_by: &'a str,
}

/// Apply one namespace's diff and record the tip it was computed against.
///
/// The upserts and removals are what changed between the previously indexed tip
/// and `update.tip`; everything else in the namespace is left alone, which is
/// what makes a push cost the size of its diff. All of it commits together with
/// the state row, so a failure halfway leaves the namespace exactly as it was
/// and the next reindex recomputes the same diff.
#[tracing::instrument(skip_all)]
pub async fn apply_changes(
    pool: &PgPool,
    repo_id: GitRepoId,
    update: &NamespaceUpdate<'_>,
) -> Result<()> {
    let NamespaceUpdate {
        namespace,
        upserts,
        removals,
        tip,
        updated_at,
        updated_by,
    } = *update;

    let mut tx = pool.begin().await?;

    if !removals.is_empty() {
        sqlx::query(
            "DELETE FROM git_notes
             WHERE repo_id = $1 AND namespace = $2 AND target_oid = ANY($3)",
        )
        .bind(repo_id)
        .bind(namespace)
        .bind(removals)
        .execute(&mut *tx)
        .await?;
    }

    if !upserts.is_empty() {
        // One statement for the batch. A push that annotates a thousand commits
        // is one round trip rather than a thousand, and the arrays keep the
        // parameter count fixed no matter how large the batch is.
        let targets: Vec<&str> = upserts.iter().map(|n| n.target_oid.as_str()).collect();
        let blobs: Vec<&str> = upserts.iter().map(|n| n.blob_oid.as_str()).collect();
        let contents: Vec<&str> = upserts.iter().map(|n| n.content.as_str()).collect();
        let is_commit: Vec<bool> = upserts.iter().map(|n| n.target_is_commit).collect();
        let summaries: Vec<&str> = upserts.iter().map(|n| n.target_summary.as_str()).collect();
        let times: Vec<Option<DateTime<Utc>>> = upserts.iter().map(|n| n.target_time).collect();

        sqlx::query(
            "INSERT INTO git_notes (
                 repo_id, namespace, target_oid, blob_oid, content,
                 target_is_commit, target_summary, target_time, updated_at, updated_by
             )
             SELECT $1, $2, t.target, t.blob, t.content, t.is_commit, t.summary, t.time, $9, $10
             FROM UNNEST($3::text[], $4::text[], $5::text[], $6::bool[], $7::text[], $8::timestamptz[])
                  AS t(target, blob, content, is_commit, summary, time)
             ON CONFLICT (repo_id, namespace, target_oid) DO UPDATE SET
                 blob_oid = EXCLUDED.blob_oid,
                 content = EXCLUDED.content,
                 target_is_commit = EXCLUDED.target_is_commit,
                 target_summary = EXCLUDED.target_summary,
                 target_time = EXCLUDED.target_time,
                 updated_at = EXCLUDED.updated_at,
                 updated_by = EXCLUDED.updated_by",
        )
        .bind(repo_id)
        .bind(namespace)
        .bind(&targets)
        .bind(&blobs)
        .bind(&contents)
        .bind(&is_commit)
        .bind(&summaries)
        .bind(&times)
        .bind(updated_at)
        .bind(updated_by)
        .execute(&mut *tx)
        .await?;
    }

    if let Some(tip) = tip {
        sqlx::query(
            "INSERT INTO git_notes_index_state (repo_id, namespace, indexed_tip, indexed_at)
             VALUES ($1, $2, $3, NOW())
             ON CONFLICT (repo_id, namespace) DO UPDATE SET
                 indexed_tip = EXCLUDED.indexed_tip,
                 indexed_at = EXCLUDED.indexed_at",
        )
        .bind(repo_id)
        .bind(namespace)
        .bind(tip)
        .execute(&mut *tx)
        .await?;
    }

    tx.commit().await?;
    Ok(())
}

/// Forget a namespace: its notes and the fact that it was ever indexed.
///
/// What a deleted `refs/notes/<ns>` produces. The state row goes too, so the
/// namespace reads as cold rather than as empty, and a ref that comes back is
/// walked in full.
#[tracing::instrument(skip_all)]
pub async fn forget_namespace(pool: &PgPool, repo_id: GitRepoId, namespace: &str) -> Result<()> {
    let mut tx = pool.begin().await?;
    sqlx::query("DELETE FROM git_notes WHERE repo_id = $1 AND namespace = $2")
        .bind(repo_id)
        .bind(namespace)
        .execute(&mut *tx)
        .await?;
    sqlx::query("DELETE FROM git_notes_index_state WHERE repo_id = $1 AND namespace = $2")
        .bind(repo_id)
        .bind(namespace)
        .execute(&mut *tx)
        .await?;
    tx.commit().await?;
    Ok(())
}

/// How many notes are indexed per namespace, for the Notes tab's counts.
#[tracing::instrument(skip_all)]
pub async fn counts_by_namespace(pool: &PgPool, repo_id: GitRepoId) -> Result<Vec<(String, i64)>> {
    let rows = sqlx::query_as::<_, (String, i64)>(
        "SELECT namespace, COUNT(*) FROM git_notes WHERE repo_id = $1
         GROUP BY namespace ORDER BY namespace",
    )
    .bind(repo_id)
    .fetch_all(pool)
    .await?;
    Ok(rows)
}

/// How many namespaces annotate each of `targets`.
///
/// The log-page path: one query for a page of commits, keyed by the hex the
/// caller passed in. Targets with no note are absent from the map rather than
/// present with a zero.
#[tracing::instrument(skip_all)]
pub async fn annotation_counts(
    pool: &PgPool,
    repo_id: GitRepoId,
    targets: &[String],
) -> Result<HashMap<String, i64>> {
    if targets.is_empty() {
        return Ok(HashMap::new());
    }
    let rows = sqlx::query_as::<_, (String, i64)>(
        "SELECT target_oid, COUNT(*) FROM git_notes
         WHERE repo_id = $1 AND target_oid = ANY($2)
         GROUP BY target_oid",
    )
    .bind(repo_id)
    .bind(targets)
    .fetch_all(pool)
    .await?;
    Ok(rows.into_iter().collect())
}

/// The annotation feed: newest annotation first.
///
/// Ordered by when the note was written rather than by the annotated commit's
/// own date, because a feed answers "what has been said lately" and annotating a
/// five-year-old commit is news. `commits_only` is the "annotated commits"
/// filter; `namespace` of `None` reads across all of them.
#[tracing::instrument(skip_all)]
pub async fn feed(
    pool: &PgPool,
    repo_id: GitRepoId,
    namespace: Option<&str>,
    commits_only: bool,
    limit: i64,
    offset: i64,
) -> Result<Vec<IndexedNote>> {
    let rows = sqlx::query_as::<_, IndexedNote>(
        "SELECT namespace, target_oid, blob_oid, content, target_is_commit,
                target_summary, target_time, updated_at, updated_by
         FROM git_notes
         WHERE repo_id = $1
           AND ($2::text IS NULL OR namespace = $2)
           AND (NOT $3::bool OR target_is_commit)
         ORDER BY updated_at DESC, target_oid
         LIMIT $4 OFFSET $5",
    )
    .bind(repo_id)
    .bind(namespace)
    .bind(commits_only)
    .bind(limit)
    .bind(offset)
    .fetch_all(pool)
    .await?;
    Ok(rows)
}

/// Notes matching the feed's filters, for the page count beside the rows.
#[tracing::instrument(skip_all)]
pub async fn count_notes(
    pool: &PgPool,
    repo_id: GitRepoId,
    namespace: Option<&str>,
    commits_only: bool,
) -> Result<i64> {
    let count = sqlx::query_scalar::<_, i64>(
        "SELECT COUNT(*) FROM git_notes
         WHERE repo_id = $1
           AND ($2::text IS NULL OR namespace = $2)
           AND (NOT $3::bool OR target_is_commit)",
    )
    .bind(repo_id)
    .bind(namespace)
    .bind(commits_only)
    .fetch_one(pool)
    .await?;
    Ok(count)
}

/// Full-text search within one repository's notes.
///
/// `websearch_to_tsquery` rather than `plainto_tsquery`: it accepts quoted
/// phrases and `or`/`-` the way a person types them into a search box, and it
/// never raises on malformed input, so a stray quote is a query that finds
/// little rather than a 500.
#[tracing::instrument(skip_all)]
pub async fn search(
    pool: &PgPool,
    repo_id: GitRepoId,
    query: &str,
    namespace: Option<&str>,
    commits_only: bool,
    limit: i64,
) -> Result<Vec<IndexedNote>> {
    let rows = sqlx::query_as::<_, IndexedNote>(
        "SELECT namespace, target_oid, blob_oid, content, target_is_commit,
                target_summary, target_time, updated_at, updated_by
         FROM git_notes
         WHERE repo_id = $1
           AND ($3::text IS NULL OR namespace = $3)
           AND (NOT $4::bool OR target_is_commit)
           AND search_tsv @@ websearch_to_tsquery('english', $2)
         ORDER BY ts_rank(search_tsv, websearch_to_tsquery('english', $2)) DESC,
                  updated_at DESC
         LIMIT $5",
    )
    .bind(repo_id)
    .bind(query)
    .bind(namespace)
    .bind(commits_only)
    .bind(limit)
    .fetch_all(pool)
    .await?;
    Ok(rows)
}