//! Queues webhook deliveries for repo events.
//!
//! Callers use the `emit_*` helpers, one per event. Each helper builds its payload and hands it
//! to [`emit`], which looks up the active hooks subscribed to the event, inserts a delivery for
//! each, and wakes the worker. Nothing here fails the caller: problems are logged.

use forge_core::webhooks::{Webhook, WebhookEvent};
use forge_core::{Comment, Issue, Label, PullRequest, Repo, Review, User};
use forge_git::browse::CommitInfo;
use serde_json::{Map, Value, json};

use super::payload;
use crate::internal::is_zero_oid;
use crate::state::AppState;

/// Commits read for a push that moves an existing ref.
const MAX_PUSH_COMMITS: usize = 500;
/// Commits read for a push that creates a ref, which reaches the whole history.
const MAX_NEW_REF_COMMITS: usize = 50;

/// Who made a change. An empty `id` is a system change, sent as `"sender": null`.
#[derive(Clone)]
pub struct Sender {
    pub id: String,
    pub username: String,
}

impl Sender {
    pub fn of(user: &User) -> Self {
        Self {
            id: user.id.clone(),
            username: user.username.clone(),
        }
    }

    pub fn system() -> Self {
        Self {
            id: String::new(),
            username: String::new(),
        }
    }
}

/// The sender for an actor id. Empty or unknown ids are system changes.
pub async fn lookup_sender(state: &AppState, actor_id: &str) -> Sender {
    if actor_id.is_empty() {
        return Sender::system();
    }
    match state.store.user_by_id(actor_id).await {
        Ok(Some(user)) => Sender::of(&user),
        Ok(None) => Sender::system(),
        Err(e) => {
            tracing::error!(error = %e, actor_id, "webhooks: sender lookup failed");
            Sender::system()
        }
    }
}

/// Queues `event` for every active hook of `repo` subscribed to it. `payload` is the event's
/// own fields; `action` (when non-empty), `repository` and `sender` are added here.
pub async fn emit(
    state: &AppState,
    repo: &Repo,
    sender: &Sender,
    event: WebhookEvent,
    action: &str,
    payload: Value,
) {
    let hooks = subscribed(state, repo, event).await;
    if hooks.is_empty() {
        return;
    }
    let body = envelope(state, repo, sender, action, payload).await;
    enqueue(state, &hooks, event, action, &body).await;
}

/// The push, create and delete events for one ref update. Only `refs/heads/*` and `refs/tags/*`
/// produce events; other refs are ignored. `old` is the zero oid for a create, and `new` is the
/// zero oid for a delete.
pub async fn emit_ref_update(
    state: &AppState,
    repo: &Repo,
    sender: &Sender,
    refname: &str,
    old: &str,
    new: &str,
) {
    let Some((ref_type, name)) = split_ref(refname) else {
        return;
    };
    if !subscribed(state, repo, WebhookEvent::Push).await.is_empty() {
        let (commits, total) = if is_zero_oid(new) {
            (Vec::new(), 0)
        } else {
            new_commits(state, repo, old, new).await
        };
        let payload = payload::push(
            &state.config.base_url,
            repo,
            refname,
            old,
            new,
            &commits,
            total,
        );
        emit(state, repo, sender, WebhookEvent::Push, "", payload).await;
    }
    if is_zero_oid(old) {
        let payload = payload::ref_change(ref_type, name);
        emit(state, repo, sender, WebhookEvent::Create, "", payload).await;
    }
    if is_zero_oid(new) {
        let payload = payload::ref_change(ref_type, name);
        emit(state, repo, sender, WebhookEvent::Delete, "", payload).await;
    }
}

/// An `issues` event for an issue (not a pull request).
pub async fn emit_issue(
    state: &AppState,
    repo: &Repo,
    sender: &Sender,
    action: &str,
    issue: &Issue,
) {
    let payload = payload::issues(&state.config.base_url, repo, issue);
    emit(state, repo, sender, WebhookEvent::Issues, action, payload).await;
}

/// A `pull_request` event.
pub async fn emit_pull(
    state: &AppState,
    repo: &Repo,
    sender: &Sender,
    action: &str,
    pr: &PullRequest,
) {
    let payload = payload::pull_request(&state.config.base_url, repo, pr);
    emit(
        state,
        repo,
        sender,
        WebhookEvent::PullRequest,
        action,
        payload,
    )
    .await;
}

/// The `opened`, `closed`, `reopened`, `merged`, `edited` or `synchronize` event for item
/// `number`, reloaded from the store so the payload has its current state. `pull` picks between
/// a pull request and an issue. A missing item sends nothing.
pub async fn emit_item_action(
    state: &AppState,
    repo: &Repo,
    sender: &Sender,
    pull: bool,
    number: i64,
    action: &str,
) {
    if pull {
        match state.store.pull_by_number(&repo.id, number).await {
            Ok(Some(pr)) => emit_pull(state, repo, sender, action, &pr).await,
            Ok(None) => {}
            Err(e) => {
                tracing::error!(error = %e, repo_id = %repo.id, number, "webhooks: loading pull request failed");
            }
        }
    } else {
        match state.store.issue_by_number(&repo.id, number).await {
            Ok(Some(issue)) => emit_issue(state, repo, sender, action, &issue).await,
            Ok(None) => {}
            Err(e) => {
                tracing::error!(error = %e, repo_id = %repo.id, number, "webhooks: loading issue failed");
            }
        }
    }
}

/// `labeled` or `unlabeled` for one label on an item. `pull` picks the `pull_request` event over
/// `issues`.
pub async fn emit_label(
    state: &AppState,
    repo: &Repo,
    sender: &Sender,
    action: &str,
    pull: bool,
    issue: &Issue,
    label: &Label,
) {
    let field = ("label", json!({"name": label.name, "color": label.color}));
    emit_item_field(state, repo, sender, action, pull, issue, field).await;
}

/// `assigned` or `unassigned` for one username on an item. `pull` picks the `pull_request` event
/// over `issues`.
pub async fn emit_assignee(
    state: &AppState,
    repo: &Repo,
    sender: &Sender,
    action: &str,
    pull: bool,
    issue: &Issue,
    username: &str,
) {
    let field = ("assignee", json!({"username": username}));
    emit_item_field(state, repo, sender, action, pull, issue, field).await;
}

/// Sends `labeled` for each label added and `unlabeled` for each removed. Labels are matched by
/// name, so a recolored label is not a change.
pub async fn emit_label_changes(
    state: &AppState,
    repo: &Repo,
    sender: &Sender,
    pull: bool,
    issue: &Issue,
    before: &[Label],
    after: &[Label],
) {
    let has = |list: &[Label], name: &str| list.iter().any(|label| label.name == name);
    for label in after.iter().filter(|label| !has(before, &label.name)) {
        emit_label(state, repo, sender, "labeled", pull, issue, label).await;
    }
    for label in before.iter().filter(|label| !has(after, &label.name)) {
        emit_label(state, repo, sender, "unlabeled", pull, issue, label).await;
    }
}

/// Sends `assigned` for each username added and `unassigned` for each removed.
pub async fn emit_assignee_changes(
    state: &AppState,
    repo: &Repo,
    sender: &Sender,
    pull: bool,
    issue: &Issue,
    before: &[String],
    after: &[String],
) {
    for name in after.iter().filter(|name| !before.contains(name)) {
        emit_assignee(state, repo, sender, "assigned", pull, issue, name).await;
    }
    for name in before.iter().filter(|name| !after.contains(name)) {
        emit_assignee(state, repo, sender, "unassigned", pull, issue, name).await;
    }
}

/// `milestoned` for the milestone `after` and `demilestoned` for the one `before`, given their
/// titles (`None` for no milestone). Nothing is sent when the titles match.
pub async fn emit_milestone_change(
    state: &AppState,
    repo: &Repo,
    sender: &Sender,
    pull: bool,
    issue: &Issue,
    before: Option<&str>,
    after: Option<&str>,
) {
    if before == after {
        return;
    }
    if let Some(title) = before {
        let field = ("milestone", json!({"title": title}));
        emit_item_field(state, repo, sender, "demilestoned", pull, issue, field).await;
    }
    if let Some(title) = after {
        let field = ("milestone", json!({"title": title}));
        emit_item_field(state, repo, sender, "milestoned", pull, issue, field).await;
    }
}

/// An `issue_comment` event. `pull` says whether the comment is on a pull request.
pub async fn emit_comment(
    state: &AppState,
    repo: &Repo,
    sender: &Sender,
    action: &str,
    pull: bool,
    issue: &Issue,
    comment: &Comment,
) {
    let payload = payload::issue_comment(&state.config.base_url, repo, issue, pull, comment);
    emit(
        state,
        repo,
        sender,
        WebhookEvent::IssueComment,
        action,
        payload,
    )
    .await;
}

/// A `pull_request_review` event with action `submitted`.
pub async fn emit_review(
    state: &AppState,
    repo: &Repo,
    sender: &Sender,
    pr: &PullRequest,
    review: &Review,
) {
    let payload = payload::pull_request_review(&state.config.base_url, repo, pr, review);
    emit(
        state,
        repo,
        sender,
        WebhookEvent::PullRequestReview,
        "submitted",
        payload,
    )
    .await;
}

/// Queues a `ping` for one hook, whether or not it subscribes to pings. Used by the "Test"
/// button and when a hook is created. Fails only when the repo is gone or the delivery cannot be
/// queued.
pub async fn emit_ping(state: &AppState, hook: &Webhook) -> forge_core::Result<()> {
    let repo = state
        .store
        .repo_by_id(&hook.repo_id)
        .await?
        .ok_or(forge_core::Error::NotFound)?;
    let body = envelope(state, &repo, &Sender::system(), "", payload::ping(&hook.id)).await;
    if enqueue(
        state,
        std::slice::from_ref(hook),
        WebhookEvent::Ping,
        "",
        &body,
    )
    .await
        == 0
    {
        return Err(forge_core::Error::Database(
            "could not queue the ping delivery".into(),
        ));
    }
    Ok(())
}

/// Sends an item event (`pull_request` when `pull`, else `issues`) with one extra top-level
/// field, `(key, value)`. A pull request is reloaded from the store so its payload is complete;
/// a missing one sends nothing.
async fn emit_item_field(
    state: &AppState,
    repo: &Repo,
    sender: &Sender,
    action: &str,
    pull: bool,
    issue: &Issue,
    (key, field): (&str, Value),
) {
    let (event, mut payload) = if pull {
        match state.store.pull_by_number(&repo.id, issue.number).await {
            Ok(Some(pr)) => (
                WebhookEvent::PullRequest,
                payload::pull_request(&state.config.base_url, repo, &pr),
            ),
            Ok(None) => return,
            Err(e) => {
                tracing::error!(error = %e, repo_id = %repo.id, number = issue.number, "webhooks: loading pull request failed");
                return;
            }
        }
    } else {
        (
            WebhookEvent::Issues,
            payload::issues(&state.config.base_url, repo, issue),
        )
    };
    if let Value::Object(map) = &mut payload {
        map.insert(key.to_owned(), field);
    }
    emit(state, repo, sender, event, action, payload).await;
}

/// Active hooks of `repo` subscribed to `event`. A lookup error is logged and reads as none.
async fn subscribed(state: &AppState, repo: &Repo, event: WebhookEvent) -> Vec<Webhook> {
    match state.store.hooks_for_event(&repo.id, event).await {
        Ok(hooks) => hooks,
        Err(e) => {
            tracing::error!(error = %e, repo_id = %repo.id, event = event.as_str(), "webhooks: hook lookup failed");
            Vec::new()
        }
    }
}

/// Whether the repo's owner is an organization (`"org"`) or a person (`"user"`). A failed lookup
/// reads as a person, since the payload's owner name is correct either way.
async fn owner_kind(state: &AppState, repo: &Repo) -> &'static str {
    match state.store.user_by_id(&repo.owner_id).await {
        Ok(Some(owner)) if owner.is_org() => "org",
        _ => "user",
    }
}

/// The full JSON body: the event fields plus `action`, `repository` and `sender`.
async fn envelope(
    state: &AppState,
    repo: &Repo,
    sender: &Sender,
    action: &str,
    payload: Value,
) -> String {
    let mut map = match payload {
        Value::Object(map) => map,
        _ => Map::new(),
    };
    if !action.is_empty() {
        map.insert("action".to_owned(), json!(action));
    }
    let kind = owner_kind(state, repo).await;
    map.insert(
        "repository".to_owned(),
        payload::repository(&state.config.base_url, repo, kind),
    );
    map.insert(
        "sender".to_owned(),
        payload::sender(&sender.id, &sender.username),
    );
    Value::Object(map).to_string()
}

/// Inserts a delivery per hook and wakes the worker after each one. Returns how many were
/// queued. A failed insert is logged and skipped.
async fn enqueue(
    state: &AppState,
    hooks: &[Webhook],
    event: WebhookEvent,
    action: &str,
    body: &str,
) -> usize {
    // The worker adds the delivery GUID and the signature at send time, so they are not stored.
    let headers = json!({
        "Content-Type": "application/json",
        "X-Forge-Event": event.as_str(),
    })
    .to_string();
    let mut queued = 0;
    for hook in hooks {
        match state
            .store
            .enqueue_delivery(&hook.id, event, action, &headers, body)
            .await
        {
            Ok(_) => {
                queued += 1;
                state.webhook_waker.notify_one();
            }
            Err(e) => {
                tracing::error!(error = %e, hook_id = %hook.id, event = event.as_str(), "webhooks: enqueue failed");
            }
        }
    }
    queued
}

/// `refs/heads/<name>` as `("branch", name)` and `refs/tags/<name>` as `("tag", name)`.
fn split_ref(refname: &str) -> Option<(&'static str, &str)> {
    if let Some(name) = refname.strip_prefix("refs/heads/") {
        Some(("branch", name))
    } else {
        refname.strip_prefix("refs/tags/").map(|name| ("tag", name))
    }
}

/// The commits a push brought in, oldest first, and their count. A failure is logged and reads
/// as no commits, so the push event still goes out.
async fn new_commits(
    state: &AppState,
    repo: &Repo,
    old: &str,
    new: &str,
) -> (Vec<CommitInfo>, usize) {
    let dir = match forge_git::repo::repo_path(&state.git.data_dir, &repo.id) {
        Ok(dir) => dir,
        Err(e) => {
            tracing::error!(error = %e, repo_id = %repo.id, "webhooks: bad repo path for push commits");
            return (Vec::new(), 0);
        }
    };
    let old = (!is_zero_oid(old)).then_some(old);
    let max = if old.is_some() {
        MAX_PUSH_COMMITS
    } else {
        MAX_NEW_REF_COMMITS
    };
    match forge_git::history::new_commits(&state.git, &dir, old, new, max).await {
        Ok(commits) => {
            let total = commits.len();
            (commits, total)
        }
        Err(e) => {
            tracing::error!(error = %e, repo_id = %repo.id, "webhooks: listing push commits failed");
            (Vec::new(), 0)
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::test_support::{KEY, test_state};
    use forge_core::Visibility;

    /// Three hooks on one repo: an active push hook, an active issues hook, and an inactive push
    /// hook. Returns the state, the repo, and the hook ids in that order.
    async fn fixture(dir: &std::path::Path) -> (AppState, Repo, [String; 3]) {
        let state = test_state(dir).await;
        let owner = state
            .store
            .create_user("alice", None, "unused", false)
            .await
            .expect("user");
        let repo = state
            .store
            .create_repo(&owner.id, "proj", "", Visibility::Public, "main")
            .await
            .expect("repo");
        let sealed = forge_core::secret::seal(&KEY, b"s").expect("seal");
        let mut ids = Vec::new();
        for (events, active) in [
            (vec![WebhookEvent::Push], true),
            (vec![WebhookEvent::Issues], true),
            (vec![WebhookEvent::Push], false),
        ] {
            let hook = state
                .store
                .create_webhook(
                    &repo.id,
                    &owner.id,
                    "https://hooks.example/h",
                    sealed.as_bytes(),
                    &events,
                    active,
                )
                .await
                .expect("hook");
            ids.push(hook.id);
        }
        let ids: [String; 3] = ids.try_into().expect("three hooks");
        (state, repo, ids)
    }

    #[tokio::test]
    async fn emit_queues_only_for_subscribed_active_hooks() {
        let tmp = tempfile::tempdir().expect("tempdir");
        let (state, repo, [push_hook, issues_hook, inactive_push]) = fixture(tmp.path()).await;

        emit(
            &state,
            &repo,
            &Sender::system(),
            WebhookEvent::Push,
            "",
            json!({"ref": "refs/heads/main"}),
        )
        .await;

        assert_eq!(
            state
                .store
                .list_deliveries(&push_hook, 10)
                .await
                .expect("list")
                .len(),
            1
        );
        assert!(
            state
                .store
                .list_deliveries(&issues_hook, 10)
                .await
                .expect("list")
                .is_empty()
        );
        assert!(
            state
                .store
                .list_deliveries(&inactive_push, 10)
                .await
                .expect("list")
                .is_empty()
        );
    }

    #[tokio::test]
    async fn emit_body_has_repository_sender_and_action_only_when_given() {
        let tmp = tempfile::tempdir().expect("tempdir");
        let (state, repo, [_, issues_hook, _]) = fixture(tmp.path()).await;
        let sender = Sender {
            id: "u1".into(),
            username: "alice".into(),
        };

        emit(
            &state,
            &repo,
            &sender,
            WebhookEvent::Issues,
            "opened",
            json!({"issue": {"number": 3}}),
        )
        .await;

        let rows = state
            .store
            .list_deliveries(&issues_hook, 10)
            .await
            .expect("list");
        let row = rows.first().expect("one delivery");
        assert_eq!(row.action, "opened");
        let body: Value = serde_json::from_str(&row.request_body).expect("json");
        assert_eq!(body["action"], "opened");
        assert_eq!(body["issue"]["number"], 3);
        assert_eq!(body["repository"]["full_name"], "alice/proj");
        assert_eq!(body["sender"], json!({"id": "u1", "username": "alice"}));
        let headers: Value = serde_json::from_str(&row.request_headers).expect("json");
        assert_eq!(headers["X-Forge-Event"], "issues");
        assert!(headers.get("X-Forge-Signature-256").is_none());
    }

    #[tokio::test]
    async fn system_changes_send_null_sender_and_no_action() {
        let tmp = tempfile::tempdir().expect("tempdir");
        let (state, repo, [push_hook, _, _]) = fixture(tmp.path()).await;
        emit(
            &state,
            &repo,
            &Sender::system(),
            WebhookEvent::Push,
            "",
            json!({}),
        )
        .await;
        let rows = state
            .store
            .list_deliveries(&push_hook, 10)
            .await
            .expect("list");
        let body: Value = serde_json::from_str(&rows[0].request_body).expect("json");
        assert!(body["sender"].is_null());
        assert!(body.get("action").is_none());
    }

    #[tokio::test]
    async fn ping_reaches_its_hook_even_without_the_subscription() {
        let tmp = tempfile::tempdir().expect("tempdir");
        let (state, _repo, [_, issues_hook, _]) = fixture(tmp.path()).await;
        let hook = state
            .store
            .webhook(&issues_hook)
            .await
            .expect("lookup")
            .expect("hook");

        emit_ping(&state, &hook).await.expect("ping queued");

        let rows = state
            .store
            .list_deliveries(&issues_hook, 10)
            .await
            .expect("list");
        assert_eq!(rows.len(), 1);
        assert_eq!(rows[0].event, WebhookEvent::Ping);
        let body: Value = serde_json::from_str(&rows[0].request_body).expect("json");
        assert_eq!(body["hook_id"], issues_hook.as_str());
    }

    /// A hook on `repo` subscribed to `events`, returned by id.
    async fn hook_for(state: &AppState, repo: &Repo, events: &[WebhookEvent]) -> String {
        let sealed = forge_core::secret::seal(&KEY, b"s").expect("seal");
        state
            .store
            .create_webhook(
                &repo.id,
                &repo.owner_id,
                "https://hooks.example/h",
                sealed.as_bytes(),
                events,
                true,
            )
            .await
            .expect("hook")
            .id
    }

    /// The parsed bodies of a hook's deliveries, newest first, with their event and action.
    async fn bodies(state: &AppState, hook: &str) -> Vec<(WebhookEvent, String, Value)> {
        state
            .store
            .list_deliveries(hook, 50)
            .await
            .expect("list")
            .into_iter()
            .map(|d| {
                let body = serde_json::from_str(&d.request_body).expect("json");
                (d.event, d.action, body)
            })
            .collect()
    }

    #[tokio::test]
    async fn pull_request_label_change_queues_pull_request_labeled() {
        let tmp = tempfile::tempdir().expect("tempdir");
        let state = test_state(tmp.path()).await;
        let owner = state
            .store
            .create_user("alice", None, "unused", false)
            .await
            .expect("user");
        let repo = state
            .store
            .create_repo(&owner.id, "proj", "", Visibility::Public, "main")
            .await
            .expect("repo");
        let zeros = "1".repeat(40);
        let pr = state
            .store
            .create_pull(
                &repo.id, &owner.id, "Title", "", "main", &repo.id, "feat", &zeros, &zeros,
            )
            .await
            .expect("pr");
        let hook = hook_for(
            &state,
            &repo,
            &[WebhookEvent::PullRequest, WebhookEvent::Issues],
        )
        .await;
        let label = Label {
            id: "l1".into(),
            repo_id: repo.id.clone(),
            name: "bug".into(),
            color: "#ff0000".into(),
            description: String::new(),
        };

        emit_label_changes(
            &state,
            &repo,
            &Sender::of(&owner),
            true,
            &pr.issue,
            &[],
            std::slice::from_ref(&label),
        )
        .await;

        let rows = bodies(&state, &hook).await;
        assert_eq!(rows.len(), 1);
        let (event, action, body) = &rows[0];
        assert_eq!(*event, WebhookEvent::PullRequest);
        assert_eq!(action, "labeled");
        assert_eq!(body["action"], "labeled");
        assert_eq!(body["label"], json!({"name": "bug", "color": "#ff0000"}));
        assert_eq!(body["number"], pr.issue.number);
        assert_eq!(body["pull_request"]["title"], "Title");
    }

    #[tokio::test]
    async fn issue_milestone_set_and_clear_queue_milestoned_and_demilestoned() {
        let tmp = tempfile::tempdir().expect("tempdir");
        let state = test_state(tmp.path()).await;
        let owner = state
            .store
            .create_user("alice", None, "unused", false)
            .await
            .expect("user");
        let repo = state
            .store
            .create_repo(&owner.id, "proj", "", Visibility::Public, "main")
            .await
            .expect("repo");
        let issue = state
            .store
            .create_issue(&repo.id, &owner.id, "Broken", "")
            .await
            .expect("issue");
        let hook = hook_for(&state, &repo, &[WebhookEvent::Issues]).await;
        let sender = Sender::of(&owner);

        emit_milestone_change(&state, &repo, &sender, false, &issue, None, Some("v1")).await;
        emit_milestone_change(&state, &repo, &sender, false, &issue, Some("v1"), None).await;
        // Same title on both sides is no change.
        emit_milestone_change(
            &state,
            &repo,
            &sender,
            false,
            &issue,
            Some("v1"),
            Some("v1"),
        )
        .await;

        let rows = bodies(&state, &hook).await;
        assert_eq!(rows.len(), 2);
        // Newest first: the removal, then the set.
        let (event, action, body) = &rows[0];
        assert_eq!(*event, WebhookEvent::Issues);
        assert_eq!(action, "demilestoned");
        assert_eq!(body["milestone"], json!({"title": "v1"}));
        assert_eq!(rows[1].1, "milestoned");
        assert_eq!(rows[1].2["milestone"], json!({"title": "v1"}));
        assert_eq!(rows[1].2["issue"]["number"], issue.number);
    }
}