loopflow 0.12.26

//! Verified Linear webhook ingestion into Task control.
//!
//! Linear pushes each issue/comment change to a Loopflow receiver. This module
//! is the receiver's brain: verify the signature, parse the event, and map it
//! onto the durable Task input spine — a title/description edit or a participant
//! comment becomes an ordered Steer. Exactly-once lives in the store
//! ([`crate::ops::linear_observe`] + the `task_linear_*` tables); this module
//! never bypasses them, so a redelivered or out-of-order webhook is a no-op.
//!
//! The HTTP glue ([`router`]/[`serve`]) is intentionally thin — all judgement is
//! in [`verify_signature`], [`parse_event`], and [`ingest_event`], which are
//! pure of the socket and unit-tested.

use std::net::SocketAddr;
use std::sync::Arc;

use axum::body::Bytes;
use axum::extract::State;
use axum::http::{HeaderMap, StatusCode};
use axum::routing::post;
use axum::Router;
use hmac::{Hmac, Mac};
use serde::Deserialize;
use sha2::Sha256;
use time::OffsetDateTime;

use crate::ops::linear_observe::reconcile_linear_observation;
use crate::pm::IssueObservation;
use crate::store::{Store, StoreError};

type HmacSha256 = Hmac<Sha256>;

/// Linear signs each delivery with the hex HMAC-SHA256 of the raw body in this
/// header, keyed by the webhook signing secret.
pub const SIGNATURE_HEADER: &str = "linear-signature";

/// How far the body's `webhookTimestamp` may sit from now before the delivery is
/// rejected as a replay. Linear recommends one minute.
pub const REPLAY_TOLERANCE_MS: i64 = 60_000;

#[derive(Debug, thiserror::Error, PartialEq, Eq)]
pub enum WebhookError {
    #[error("signature is not valid hex")]
    MalformedSignature,
    #[error("signature does not match the body")]
    BadSignature,
    #[error("payload is not valid JSON: {0}")]
    Malformed(String),
}

/// Verify the hex HMAC-SHA256 signature over the raw request body. The compare is
/// constant-time (`Mac::verify_slice`), so a mismatch leaks no timing signal.
pub fn verify_signature(
    secret: &[u8],
    raw_body: &[u8],
    signature_hex: &str,
) -> Result<(), WebhookError> {
    let expected =
        hex::decode(signature_hex.trim()).map_err(|_| WebhookError::MalformedSignature)?;
    let mut mac = HmacSha256::new_from_slice(secret).expect("HMAC accepts any key length");
    mac.update(raw_body);
    mac.verify_slice(&expected)
        .map_err(|_| WebhookError::BadSignature)
}

/// True when a delivery's `webhookTimestamp` (ms) is close enough to now to not
/// be a replay of an old, captured request.
pub fn within_replay_window(webhook_timestamp_ms: i64, now: OffsetDateTime) -> bool {
    let now_ms = now.unix_timestamp_nanos() / 1_000_000;
    (i128::from(webhook_timestamp_ms) - now_ms).unsigned_abs() <= REPLAY_TOLERANCE_MS as u128
}

/// The only Linear changes that carry Task direction. Everything else — status,
/// assignment, labels, removals — decodes to `Ignored`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum WebhookEvent {
    /// A title and/or description edit. `revision` is the issue `updatedAt`.
    IssueEdit {
        issue_id: String,
        title: String,
        description: String,
        revision: String,
        actor_id: Option<String>,
    },
    /// A new comment. `author_id` distinguishes a participant from Loopflow's own
    /// writeback and from an integration/bot with no backing user.
    Comment {
        issue_id: String,
        comment_id: String,
        revision: Option<String>,
        body: String,
        author_id: Option<String>,
        author_name: Option<String>,
    },
    Ignored,
}

#[derive(Deserialize)]
struct RawWebhook {
    #[serde(default)]
    action: String,
    #[serde(rename = "type", default)]
    kind: String,
    #[serde(default)]
    data: serde_json::Value,
    #[serde(rename = "updatedFrom", default)]
    updated_from: serde_json::Value,
    #[serde(default)]
    actor: serde_json::Value,
    #[serde(rename = "webhookTimestamp", default)]
    webhook_timestamp: i64,
}

fn nested_str(value: &serde_json::Value, path: &[&str]) -> Option<String> {
    let mut cursor = value;
    for key in path {
        cursor = cursor.get(key)?;
    }
    cursor.as_str().map(str::to_string)
}

impl RawWebhook {
    fn comment_author(&self) -> (Option<String>, Option<String>) {
        let actor_id = nested_str(&self.actor, &["id"]);
        let author_id = nested_str(&self.data, &["user", "id"])
            .or_else(|| nested_str(&self.data, &["userId"]))
            .or_else(|| {
                (self.action == "create")
                    .then(|| actor_id.clone())
                    .flatten()
            });
        let author_name = nested_str(&self.data, &["user", "displayName"])
            .or_else(|| nested_str(&self.data, &["user", "name"]))
            .or_else(|| {
                (author_id.is_some() && author_id == actor_id)
                    .then(|| {
                        nested_str(&self.actor, &["displayName"])
                            .or_else(|| nested_str(&self.actor, &["name"]))
                    })
                    .flatten()
            });
        (author_id, author_name)
    }

    fn classify(&self) -> WebhookEvent {
        match (self.kind.as_str(), self.action.as_str()) {
            ("Issue", "update") => {
                // `updatedFrom` names exactly the fields that changed, so a
                // metadata-only edit (status, assignee, labels) carries neither
                // title nor description here and is ignored.
                let content_changed = self.updated_from.get("title").is_some()
                    || self.updated_from.get("description").is_some();
                match (content_changed, nested_str(&self.data, &["id"])) {
                    (true, Some(issue_id)) => WebhookEvent::IssueEdit {
                        issue_id,
                        title: nested_str(&self.data, &["title"]).unwrap_or_default(),
                        description: nested_str(&self.data, &["description"]).unwrap_or_default(),
                        revision: nested_str(&self.data, &["updatedAt"]).unwrap_or_default(),
                        actor_id: nested_str(&self.actor, &["id"]),
                    },
                    _ => WebhookEvent::Ignored,
                }
            }
            ("Comment", "create" | "update") => {
                let (author_id, author_name) = self.comment_author();
                match (
                    nested_str(&self.data, &["id"]),
                    nested_str(&self.data, &["issue", "id"]),
                ) {
                    (Some(comment_id), Some(issue_id)) => WebhookEvent::Comment {
                        issue_id,
                        comment_id,
                        revision: nested_str(&self.data, &["updatedAt"]),
                        body: nested_str(&self.data, &["body"]).unwrap_or_default(),
                        author_id,
                        author_name,
                    },
                    _ => WebhookEvent::Ignored,
                }
            }
            _ => WebhookEvent::Ignored,
        }
    }
}

#[cfg(test)]
#[test]
fn request_authors_do_not_confuse_comment_editors_with_authors() {
    let raw = serde_json::json!({"type":"Comment", "action":"update", "actor":{"id":"editor", "name":"Maya"}, "data":{"id":"comment", "issue":{"id":"issue"}, "body":"request", "userId":"author"}});
    let (event, _) = parse_event(raw.to_string().as_bytes()).unwrap();
    assert!(
        matches!(event, WebhookEvent::Comment { author_id: Some(id), author_name: None, .. } if id == "author")
    );
    let mut named = raw;
    named["data"]["user"] = serde_json::json!({"id":"author", "displayName":"Jack"});
    let (event, _) = parse_event(named.to_string().as_bytes()).unwrap();
    assert!(
        matches!(event, WebhookEvent::Comment { author_name: Some(name), .. } if name == "Jack")
    );
}

/// Parse a raw body into a typed event plus its `webhookTimestamp` (ms).
pub fn parse_event(raw_body: &[u8]) -> Result<(WebhookEvent, i64), WebhookError> {
    let raw: RawWebhook = serde_json::from_slice(raw_body)
        .map_err(|error| WebhookError::Malformed(error.to_string()))?;
    Ok((raw.classify(), raw.webhook_timestamp))
}

/// What one verified webhook did — enough for the receiver's log and response.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum WebhookOutcome {
    /// Not an event that carries direction (metadata edit, removal, unknown).
    Ignored,
    /// A real event on an issue that has no Task — nothing to steer.
    NoTarget,
    /// Loopflow's own edit/comment, skipped to avoid a feedback loop.
    SelfAuthored,
    /// A title/description edit; `steer_applied` is false for a duplicate.
    Edit { steer_applied: bool },
    /// A participant comment; `delivered` is false for a duplicate delivery.
    Comment { delivered: bool },
}

/// Map one verified, parsed event onto the durable Task control substrate.
/// Resolves the target Task by Linear issue id; a missing Task or a
/// self-authored change writes nothing.
pub async fn ingest_event(
    store: &Store,
    event: WebhookEvent,
    viewer_id: &str,
    now: OffsetDateTime,
) -> Result<WebhookOutcome, StoreError> {
    let issue_id = match &event {
        WebhookEvent::IssueEdit { issue_id, .. } | WebhookEvent::Comment { issue_id, .. } => {
            issue_id.clone()
        }
        WebhookEvent::Ignored => return Ok(WebhookOutcome::Ignored),
    };
    let Some(task) = store.get_task_by_issue(&issue_id).await? else {
        return Ok(WebhookOutcome::NoTarget);
    };
    match event {
        WebhookEvent::IssueEdit {
            title,
            description,
            revision,
            actor_id,
            ..
        } => {
            if actor_id.as_deref() == Some(viewer_id) {
                return Ok(WebhookOutcome::SelfAuthored);
            }
            let observation = IssueObservation {
                revision,
                title,
                description,
                comments: vec![],
            };
            let outcome =
                reconcile_linear_observation(store, &task, observation, viewer_id, now).await?;
            Ok(WebhookOutcome::Edit {
                steer_applied: outcome.content_steer_applied,
            })
        }
        WebhookEvent::Comment {
            comment_id,
            revision,
            body,
            author_id,
            author_name,
            ..
        } => {
            if !crate::ops::linear_observe::is_direction_comment(&body, author_id.as_deref()) {
                return Ok(WebhookOutcome::SelfAuthored);
            }
            let text = crate::ops::linear_observe::render_comment(
                &comment_id,
                &body,
                author_id.as_deref(),
                author_name.as_deref(),
            );
            let comment_id =
                crate::ops::linear_observe::comment_revision_id(&comment_id, revision.as_deref());
            let created = store
                .apply_linear_comment(&task.id, comment_id, text, now)
                .await?;
            Ok(WebhookOutcome::Comment {
                delivered: created.is_some(),
            })
        }
        WebhookEvent::Ignored => Ok(WebhookOutcome::Ignored),
    }
}

/// Everything the receiver route needs, shared across requests.
#[derive(Clone)]
pub struct WebhookState {
    pub store: Arc<Store>,
    pub secret: Arc<Vec<u8>>,
    pub viewer_id: Arc<String>,
}

/// The receiver router: one signed `POST /linear/webhook`.
pub fn router(state: WebhookState) -> Router {
    Router::new()
        .route("/linear/webhook", post(handle))
        .with_state(state)
}

async fn handle(State(state): State<WebhookState>, headers: HeaderMap, body: Bytes) -> StatusCode {
    let Some(signature) = headers.get(SIGNATURE_HEADER).and_then(|v| v.to_str().ok()) else {
        return StatusCode::UNAUTHORIZED;
    };
    if verify_signature(&state.secret, &body, signature).is_err() {
        return StatusCode::UNAUTHORIZED;
    }
    let (event, webhook_timestamp) = match parse_event(&body) {
        Ok(parsed) => parsed,
        Err(_) => return StatusCode::BAD_REQUEST,
    };
    let now = OffsetDateTime::now_utc();
    if !within_replay_window(webhook_timestamp, now) {
        return StatusCode::UNAUTHORIZED;
    }
    match ingest_event(&state.store, event, &state.viewer_id, now).await {
        // A local store error is ours, not Linear's — 500 asks it to retry.
        Err(_) => StatusCode::INTERNAL_SERVER_ERROR,
        Ok(_) => StatusCode::OK,
    }
}

/// Bind the receiver to `addr` and serve until the process ends. Callers own
/// the secret (from Doppler) and the Loopflow viewer id.
pub async fn serve(
    store: Arc<Store>,
    secret: Vec<u8>,
    viewer_id: String,
    addr: SocketAddr,
) -> anyhow::Result<()> {
    let state = WebhookState {
        store,
        secret: Arc::new(secret),
        viewer_id: Arc::new(viewer_id),
    };
    let listener = tokio::net::TcpListener::bind(addr).await?;
    axum::serve(listener, router(state).into_make_service()).await?;
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::{parse_event, verify_signature, within_replay_window, WebhookEvent};
    use hmac::{Hmac, Mac};
    use serde_json::json;
    use sha2::Sha256;
    use time::OffsetDateTime;

    fn sign(secret: &[u8], body: &[u8]) -> String {
        let mut mac = Hmac::<Sha256>::new_from_slice(secret).unwrap();
        mac.update(body);
        hex::encode(mac.finalize().into_bytes())
    }

    #[test]
    fn a_correct_signature_verifies_and_a_tampered_body_does_not() {
        let secret = b"whsec_example";
        let body = br#"{"action":"update","type":"Issue"}"#;
        let signature = sign(secret, body);
        assert!(verify_signature(secret, body, &signature).is_ok());
        assert!(verify_signature(secret, b"tampered", &signature).is_err());
        assert!(verify_signature(b"wrong-secret", body, &signature).is_err());
        assert!(verify_signature(secret, body, "not-hex-zz").is_err());
    }

    #[test]
    fn replay_window_rejects_a_stale_timestamp() {
        let now = OffsetDateTime::from_unix_timestamp(1_784_160_000).unwrap();
        let now_ms = now.unix_timestamp() * 1000;
        assert!(within_replay_window(now_ms, now));
        assert!(within_replay_window(now_ms - 30_000, now));
        assert!(!within_replay_window(now_ms - 120_000, now));
    }

    #[test]
    fn an_issue_content_edit_parses_but_a_metadata_edit_is_ignored() {
        let edit = json!({
            "action": "update",
            "type": "Issue",
            "data": { "id": "issue-1", "title": "New", "description": "Body", "updatedAt": "2026-07-15T01:00:00.000Z" },
            "updatedFrom": { "title": "Old" },
            "actor": { "id": "user-human" },
            "webhookTimestamp": 1_784_160_000_000i64
        });
        let (event, ts) = parse_event(edit.to_string().as_bytes()).unwrap();
        assert_eq!(ts, 1_784_160_000_000);
        assert_eq!(
            event,
            WebhookEvent::IssueEdit {
                issue_id: "issue-1".into(),
                title: "New".into(),
                description: "Body".into(),
                revision: "2026-07-15T01:00:00.000Z".into(),
                actor_id: Some("user-human".into()),
            }
        );

        let metadata = json!({
            "action": "update",
            "type": "Issue",
            "data": { "id": "issue-1", "title": "New" },
            "updatedFrom": { "stateId": "s-1" },
            "webhookTimestamp": 1i64
        });
        let (event, _) = parse_event(metadata.to_string().as_bytes()).unwrap();
        assert_eq!(event, WebhookEvent::Ignored);
    }

    #[test]
    fn a_comment_create_parses_with_its_author() {
        let comment = json!({
            "action": "create",
            "type": "Comment",
            "data": { "id": "c-1", "body": "please prioritize", "issue": { "id": "issue-1" }, "user": { "id": "user-human" } },
            "webhookTimestamp": 2i64
        });
        let (event, _) = parse_event(comment.to_string().as_bytes()).unwrap();
        assert_eq!(
            event,
            WebhookEvent::Comment {
                author_name: None,
                issue_id: "issue-1".into(),
                comment_id: "c-1".into(),
                revision: None,
                body: "please prioritize".into(),
                author_id: Some("user-human".into()),
            }
        );
    }
}