compliance-scanner-agent/compliance-agent/src/pipeline/orchestrator.rs

use std::sync::Arc;

use mongodb::bson::doc;
use tracing::Instrument;

use compliance_core::models::*;
use compliance_core::traits::Scanner;
use compliance_core::AgentConfig;

use crate::database::Database;
use crate::error::AgentError;
use crate::llm::LlmClient;
use crate::pipeline::cve::CveScanner;
use crate::pipeline::git::GitOps;
use crate::pipeline::gitleaks::GitleaksScanner;
use crate::pipeline::lint::LintScanner;
use crate::pipeline::patterns::{GdprPatternScanner, OAuthPatternScanner};
use crate::pipeline::plan::build_scan_plan;
use crate::pipeline::repo_view::RepoView;
use crate::pipeline::sbom::SbomScanner;
use crate::pipeline::semgrep::SemgrepScanner;

/// Context from graph analysis passed to LLM triage for enhanced filtering
#[derive(Debug)]
#[allow(dead_code)]
pub struct GraphContext {
    pub node_count: u32,
    pub edge_count: u32,
    pub community_count: u32,
    pub impacts: Vec<compliance_core::models::graph::ImpactAnalysis>,
}

pub struct PipelineOrchestrator {
    pub(super) config: AgentConfig,
    pub(super) db: Database,
    pub(super) llm: Arc<LlmClient>,
    pub(super) http: reqwest::Client,
}

impl PipelineOrchestrator {
    pub fn new(
        config: AgentConfig,
        db: Database,
        llm: Arc<LlmClient>,
        http: reqwest::Client,
    ) -> Self {
        Self {
            config,
            db,
            llm,
            http,
        }
    }

    #[tracing::instrument(skip_all, fields(repo_id = repo.name.as_str()))]
    async fn run_pipeline(&self, repo: &RepoView, scan_run_id: &str) -> Result<u32, AgentError> {
        let repo_id = repo.id.as_ref().map(|id| id.to_hex()).unwrap_or_default();

        // Stage 0: Change detection
        tracing::info!("[{repo_id}] Stage 0: Change detection");
        let creds = GitOps::make_repo_credentials(&self.config, repo);
        let git_ops = GitOps::new(&self.config.git_clone_base_path, creds);
        let repo_path = git_ops.clone_or_fetch(&repo.git_url, &repo.name)?;

        if !GitOps::has_new_commits(&repo_path, repo.last_scanned_commit.as_deref())? {
            tracing::info!("[{repo_id}] No new commits, skipping scan");
            return Ok(0);
        }

        let mut all_findings: Vec<Finding> = Vec::new();

        // Stage 1: Semgrep SAST
        tracing::info!("[{repo_id}] Stage 1: Semgrep SAST");
        self.update_phase(scan_run_id, "sast").await;
        match async {
            let semgrep = SemgrepScanner;
            semgrep.scan(&repo_path, &repo_id).await
        }
        .instrument(tracing::info_span!("stage_sast"))
        .await
        {
            Ok(output) => all_findings.extend(output.findings),
            Err(e) => tracing::warn!("[{repo_id}] Semgrep failed: {e}"),
        }

        // Stage 2: SBOM Generation
        tracing::info!("[{repo_id}] Stage 2: SBOM Generation");
        self.update_phase(scan_run_id, "sbom_generation").await;
        let mut sbom_entries = match async {
            let sbom_scanner = SbomScanner;
            sbom_scanner.scan(&repo_path, &repo_id).await
        }
        .instrument(tracing::info_span!("stage_sbom_generation"))
        .await
        {
            Ok(output) => output.sbom_entries,
            Err(e) => {
                tracing::warn!("[{repo_id}] SBOM generation failed: {e}");
                Vec::new()
            }
        };

        // Stage 3: CVE Scanning
        tracing::info!("[{repo_id}] Stage 3: CVE Scanning");
        self.update_phase(scan_run_id, "cve_scanning").await;
        let cve_scanner = CveScanner::new(
            self.http.clone(),
            self.config.searxng_url.clone(),
            self.config.nvd_api_key.as_ref().map(|k| {
                use secrecy::ExposeSecret;
                k.expose_secret().to_string()
            }),
        );
        let cve_alerts = match tokio::time::timeout(
            std::time::Duration::from_secs(600),
            async {
                cve_scanner
                    .scan_dependencies(&repo_id, &mut sbom_entries)
                    .await
            }
            .instrument(tracing::info_span!("stage_cve_scanning")),
        )
        .await
        {
            Ok(Ok(alerts)) => alerts,
            Ok(Err(e)) => {
                tracing::warn!("[{repo_id}] CVE scanning failed: {e}");
                Vec::new()
            }
            Err(_) => {
                tracing::warn!("[{repo_id}] CVE scanning timed out after 10 minutes");
                Vec::new()
            }
        };

        // Stage 4: Pattern Scanning (GDPR + OAuth)
        tracing::info!("[{repo_id}] Stage 4: Pattern Scanning");
        self.update_phase(scan_run_id, "pattern_scanning").await;
        {
            let pattern_findings = async {
                let mut findings = Vec::new();
                let gdpr = GdprPatternScanner::new();
                match gdpr.scan(&repo_path, &repo_id).await {
                    Ok(output) => findings.extend(output.findings),
                    Err(e) => tracing::warn!("[{repo_id}] GDPR pattern scan failed: {e}"),
                }
                let oauth = OAuthPatternScanner::new();
                match oauth.scan(&repo_path, &repo_id).await {
                    Ok(output) => findings.extend(output.findings),
                    Err(e) => tracing::warn!("[{repo_id}] OAuth pattern scan failed: {e}"),
                }
                findings
            }
            .instrument(tracing::info_span!("stage_pattern_scanning"))
            .await;
            all_findings.extend(pattern_findings);
        }

        // Stage 4a: Secret Detection (Gitleaks)
        tracing::info!("[{repo_id}] Stage 4a: Secret Detection");
        self.update_phase(scan_run_id, "secret_detection").await;
        match async {
            let gitleaks = GitleaksScanner;
            gitleaks.scan(&repo_path, &repo_id).await
        }
        .instrument(tracing::info_span!("stage_secret_detection"))
        .await
        {
            Ok(output) => all_findings.extend(output.findings),
            Err(e) => tracing::warn!("[{repo_id}] Gitleaks failed: {e}"),
        }

        // Stage 4b: Lint Scanning
        tracing::info!("[{repo_id}] Stage 4b: Lint Scanning");
        self.update_phase(scan_run_id, "lint_scanning").await;
        match async {
            let lint = LintScanner;
            lint.scan(&repo_path, &repo_id).await
        }
        .instrument(tracing::info_span!("stage_lint_scanning"))
        .await
        {
            Ok(output) => all_findings.extend(output.findings),
            Err(e) => tracing::warn!("[{repo_id}] Lint scanning failed: {e}"),
        }

        // Stage 4.5: Graph Building
        tracing::info!("[{repo_id}] Stage 4.5: Graph Building");
        self.update_phase(scan_run_id, "graph_building").await;
        let graph_context = match async {
            self.build_code_graph(&repo_path, &repo_id, &all_findings)
                .await
        }
        .instrument(tracing::info_span!("stage_graph_building"))
        .await
        {
            Ok(ctx) => Some(ctx),
            Err(e) => {
                tracing::warn!("[{repo_id}] Graph building failed: {e}");
                None
            }
        };

        // Stage 5: LLM Triage (enhanced with graph context)
        tracing::info!(
            "[{repo_id}] Stage 5: LLM Triage ({} findings)",
            all_findings.len()
        );
        self.update_phase(scan_run_id, "llm_triage").await;
        let triaged = crate::llm::triage::triage_findings(
            &self.llm,
            &mut all_findings,
            graph_context.as_ref(),
        )
        .await;
        tracing::info!("[{repo_id}] Triaged: {triaged} findings passed confidence threshold");

        // Dedup against existing findings and insert new ones
        let mut new_count = 0u32;
        let mut new_findings: Vec<Finding> = Vec::new();
        for mut finding in all_findings {
            finding.scan_run_id = Some(scan_run_id.to_string());
            // Check if fingerprint already exists
            let existing = self
                .db
                .findings()
                .find_one(doc! { "fingerprint": &finding.fingerprint })
                .await?;
            if existing.is_none() {
                let result = self.db.findings().insert_one(&finding).await?;
                finding.id = result.inserted_id.as_object_id();
                new_findings.push(finding);
                new_count += 1;
            }
        }

        // Remove stale SBOM entries for this repo before reinserting
        if !sbom_entries.is_empty() {
            self.db
                .sbom_entries()
                .delete_many(doc! { "repo_id": &repo.id })
                .await?;
        }

        // Persist SBOM entries
        for entry in &sbom_entries {
            let filter = doc! {
                "repo_id": &entry.repo_id,
                "name": &entry.name,
                "version": &entry.version,
            };
            let update = mongodb::bson::to_document(entry)
                .map(|d| doc! { "$set": d })
                .unwrap_or_else(|_| doc! {});
            self.db
                .sbom_entries()
                .update_one(filter, update)
                .upsert(true)
                .await?;
        }

        // Persist CVE alerts and create notifications (shared with the PLC path).
        let new_notif_count = self
            .persist_cve_alerts(&repo_id, &repo.name, &cve_alerts)
            .await?;
        if new_notif_count > 0 {
            tracing::info!("[{repo_id}] Created {new_notif_count} CVE notification(s)");
        }

        // Stage 6: Issue Creation
        tracing::info!("[{repo_id}] Stage 6: Issue Creation");
        self.update_phase(scan_run_id, "issue_creation").await;
        if let Err(e) = self
            .create_tracker_issues(repo, &repo_id, &new_findings)
            .await
        {
            tracing::warn!("[{repo_id}] Issue creation failed: {e}");
        }

        // The onboarded target's findings_count and the git artifact's
        // last_scanned_commit watermark are persisted by `finalize_target` after
        // `run_pipeline` returns.

        // Stage 8: DAST (async, optional — only if a DastTarget is configured)
        tracing::info!("[{repo_id}] Stage 8: Checking for DAST targets");
        self.update_phase(scan_run_id, "dast_scanning").await;
        self.maybe_trigger_dast(&repo_id, scan_run_id).await;

        tracing::info!("[{repo_id}] Scan complete: {new_count} new findings");
        Ok(new_count)
    }

    /// Unified entry point (behind `UNIFIED_PIPELINE`): run a scan for an
    /// `OnboardedTarget`. Mirrors [`Self::run`] but sources the target from
    /// `onboarded_targets` and dispatches by the scan plan.
    #[tracing::instrument(skip_all, fields(target_id = %target_id, trigger = ?trigger))]
    pub async fn run_target(
        &self,
        target_id: &str,
        trigger: ScanTrigger,
    ) -> Result<(), AgentError> {
        let oid = mongodb::bson::oid::ObjectId::parse_str(target_id)
            .map_err(|e| AgentError::Other(e.to_string()))?;
        let target = self
            .db
            .onboarded_targets()
            .find_one(doc! { "_id": oid })
            .await?
            .ok_or_else(|| AgentError::Other(format!("Onboarded target {target_id} not found")))?;

        let scan_run = ScanRun::new(target_id.to_string(), trigger);
        let insert = self.db.scan_runs().insert_one(&scan_run).await?;
        let scan_run_id = insert
            .inserted_id
            .as_object_id()
            .map(|id| id.to_hex())
            .unwrap_or_default();

        let result = self.run_target_pipeline(&target, &scan_run_id).await;
        match &result {
            Ok(count) => {
                self.db
                    .scan_runs()
                    .update_one(
                        doc! { "_id": &insert.inserted_id },
                        doc! { "$set": {
                            "status": "completed",
                            "current_phase": "completed",
                            "new_findings_count": *count as i64,
                            "completed_at": mongodb::bson::DateTime::now(),
                        } },
                    )
                    .await?;
                // Refresh the target's cached findings count. The shared pipeline
                // (Stage 7) increments `repositories`, which the unified path does
                // not use, so set the accurate total on the target itself.
                let total = self
                    .db
                    .findings()
                    .count_documents(doc! { "repo_id": target_id })
                    .await
                    .unwrap_or(*count as u64);
                self.db
                    .onboarded_targets()
                    .update_one(
                        doc! { "_id": oid },
                        doc! { "$set": {
                            "findings_count": total as i64,
                            "updated_at": mongodb::bson::DateTime::now(),
                        } },
                    )
                    .await?;
            }
            Err(e) => {
                tracing::error!(target_id, error = %e, "Unified scan pipeline failed");
                self.db
                    .scan_runs()
                    .update_one(
                        doc! { "_id": &insert.inserted_id },
                        doc! { "$set": {
                            "status": "failed",
                            "error_message": e.to_string(),
                            "completed_at": mongodb::bson::DateTime::now(),
                        } },
                    )
                    .await?;
            }
        }
        result.map(|_| ())
    }

    /// Run the applicable scans for a target. For a code target this reuses the
    /// full legacy pipeline over the code artifact (clone → SAST umbrella →
    /// triage → persist → issues → DAST); firmware/PLC/mobile scanners are
    /// follow-ups (#128/#129/#130). Returns the number of new findings.
    async fn run_target_pipeline(
        &self,
        target: &OnboardedTarget,
        scan_run_id: &str,
    ) -> Result<u32, AgentError> {
        let target_id = target.id.map(|id| id.to_hex()).unwrap_or_default();
        let plan = build_scan_plan(target);
        tracing::info!(
            target_id = %target_id,
            target_type = %target.target_type,
            planned_steps = plan.steps.len(),
            "Unified pipeline: scan plan built"
        );

        // Ingest + classify (tramiton for firmware) and store the detected type.
        self.classify_and_store(target, &target_id, scan_run_id)
            .await;
        // Provision a DAST target from a LiveUrl artifact so DAST fires for
        // wizard-created targets, not just migrated ones.
        self.ensure_dast_target(target, &plan).await;

        // PLC/SPS targets: the control-logic scan consumes the PLC source (an
        // uploaded PlcProject *or* a git repo / source archive of PLCopen XML / ST
        // exports), so it takes over the code artifact — we don't also run the
        // SAST pipeline over it. A PLC device is reachable, so DAST still runs
        // against a WebVisu / exposed endpoint when one is provisioned.
        let mut new_count = 0u32;
        let plc = plan.has(ScanType::PlcControlLogic);
        let ics = plan.has(ScanType::IcsProbe);
        if plc {
            new_count += self.run_plc_scan(target, &target_id, scan_run_id).await?;
            // Provision-and-test (#183): with the control logic but no reachable
            // device, instantiate it on an ephemeral soft-PLC and probe that
            // instead of the customer's OT network. Opt-in (needs Docker) and only
            // when there is no live URL to probe directly. Never fails the scan.
            if self.config.plc_runtime.enabled && target.live_url().is_none() {
                match self
                    .run_provisioned_plc_test(target, &target_id, scan_run_id)
                    .await
                {
                    Ok(n) => new_count += n,
                    Err(e) => {
                        tracing::warn!(target_id = %target_id, error = %e, "provision-and-test failed")
                    }
                }
            }
        }
        if ics {
            new_count += self.run_ics_probe(target, &target_id, scan_run_id).await?;
        }
        if plc || ics {
            // PLC/SPS device: also DAST against a WebVisu / exposed endpoint, but
            // only when DAST is actually planned — a device reachable only over an
            // industrial protocol (e.g. modbus://) has no web surface to crawl, and
            // running DAST there just fails at reconnaissance. Gating here (not only
            // at provisioning) also stops a DAST target left over from an earlier
            // run from re-triggering. The control-logic scan already consumed the
            // code artifact, so the SAST pipeline is not re-run.
            if plan.has(ScanType::Dast) {
                self.update_phase(scan_run_id, "dast_scanning").await;
                self.maybe_trigger_dast(&target_id, scan_run_id).await;
            }
            return Ok(new_count);
        }

        match target.code_artifact() {
            Some(code) if code.kind == ArtifactKind::GitRepo => {
                let repo = RepoView::from_target(target, code);
                let n = self.run_pipeline(&repo, scan_run_id).await?;
                self.finalize_target(target, &repo, n).await?;
                new_count += n;
            }
            Some(_) => {
                tracing::warn!(
                    target_id = %target_id,
                    "Unified pipeline: source-archive scanning not yet wired; skipping"
                );
            }
            None => {
                // No code to scan (a migrated DAST target). Firmware/mobile static
                // scanners land in #128/#129; DAST for a running URL works when a
                // DastTarget row exists (provisioned above from a LiveUrl, or from
                // a migrated target).
                tracing::info!(
                    target_id = %target_id,
                    "Unified pipeline: no code artifact; attempting DAST"
                );
                self.update_phase(scan_run_id, "dast_scanning").await;
                self.maybe_trigger_dast(&target_id, scan_run_id).await;
            }
        }
        Ok(new_count)
    }

    /// Analyze a PLC/SPS project (Structured Text / PLCopen XML) for
    /// control-logic security issues and persist the new findings.
    async fn run_plc_scan(
        &self,
        target: &OnboardedTarget,
        target_id: &str,
        scan_run_id: &str,
    ) -> Result<u32, AgentError> {
        tracing::info!(target_id, "[{target_id}] PLC control-logic analysis");
        self.update_phase(scan_run_id, "plc_analysis").await;

        let ctx = crate::ingest::IngestContext::from_config(&self.config, target_id);
        let ingest_set = crate::ingest::ingest_all(target, &ctx)?;
        // Every PLC-source artifact on the target: dedicated PLC projects plus any
        // code artifacts (git repo / source archive) holding PLCopen XML / ST
        // exports. A target can carry several (e.g. one POU export per file).
        let sources: Vec<&Artifact> = target
            .artifacts
            .iter()
            .filter(|a| {
                matches!(
                    a.kind,
                    ArtifactKind::PlcProject | ArtifactKind::GitRepo | ArtifactKind::SourceArchive
                )
            })
            .collect();
        if sources.is_empty() {
            tracing::warn!(target_id, "PLC scan: no PLC source artifact");
            return Ok(0);
        }

        let mut all_findings = Vec::new();
        let mut all_sbom: Vec<SbomEntry> = Vec::new();
        let mut sbom_seen = std::collections::BTreeSet::new();
        for a in &sources {
            let Some(path) = ingest_set.get(&a.id).and_then(|ia| ia.working_path.clone()) else {
                continue;
            };
            all_findings.extend(crate::pipeline::plc::analyze_tree(&path, target_id));
            // Control-application SBOM: CODESYS libraries + runtime from a
            // `.projectarchive` (uploaded, or committed in the working tree).
            let archive = a
                .stored_path
                .clone()
                .unwrap_or_else(|| a.source_ref.clone());
            for e in crate::pipeline::plc::sbom::collect_sbom(
                std::path::Path::new(&archive),
                &path,
                target_id,
            ) {
                if sbom_seen.insert((e.name.clone(), e.version.clone())) {
                    all_sbom.push(e);
                }
            }
        }
        tracing::info!(
            target_id,
            artifacts = sources.len(),
            found = all_findings.len(),
            "PLC control-logic analysis complete"
        );

        let mut new_count = 0u32;
        for mut finding in all_findings {
            finding.scan_run_id = Some(scan_run_id.to_string());
            if self
                .db
                .findings()
                .find_one(doc! { "fingerprint": &finding.fingerprint })
                .await?
                .is_none()
            {
                self.db.findings().insert_one(&finding).await?;
                new_count += 1;
            }
        }

        if !all_sbom.is_empty() {
            if let Err(e) = self
                .persist_control_app_sbom(target_id, &target.name, all_sbom)
                .await
            {
                tracing::warn!(target_id, error = %e, "control-app SBOM persist failed");
            }
        }
        Ok(new_count)
    }

    /// Provision-and-test (#183): instantiate the target's control logic on an
    /// ephemeral soft-PLC (OpenPLC), start it, probe the provisioned Modbus
    /// endpoint, and tear the instance down. Used when a PLC/SPS target has the
    /// control logic but no reachable live device to probe directly. Guarded by
    /// `plc_runtime.enabled` (needs Docker); persists the same [`ScanType::IcsProbe`]
    /// findings as a live probe.
    async fn run_provisioned_plc_test(
        &self,
        target: &OnboardedTarget,
        target_id: &str,
        scan_run_id: &str,
    ) -> Result<u32, AgentError> {
        self.update_phase(scan_run_id, "plc_provision").await;

        // Locate a loadable control-logic program among the PLC-source artifacts
        // (same selection as the static PLC scan: dedicated PLC projects plus code
        // artifacts holding PLCopen XML / ST exports).
        let ctx = crate::ingest::IngestContext::from_config(&self.config, target_id);
        let ingest_set = crate::ingest::ingest_all(target, &ctx)?;
        let program = target
            .artifacts
            .iter()
            .filter(|a| {
                matches!(
                    a.kind,
                    ArtifactKind::PlcProject | ArtifactKind::GitRepo | ArtifactKind::SourceArchive
                )
            })
            .find_map(|a| {
                let path = ingest_set.get(&a.id).and_then(|ia| ia.working_path.clone())?;
                crate::pipeline::plc::runtime::extract_program(&path)
            });
        let Some(program) = program else {
            tracing::info!(target_id, "provision-and-test: no loadable control-logic program");
            return Ok(0);
        };

        let http = crate::pipeline::plc::runtime::http_client()?;
        let provisioner =
            crate::pipeline::plc::runtime::DockerSoftPlc::new(self.config.plc_runtime.clone());
        let findings = crate::pipeline::plc::runtime::provision_and_test(
            &provisioner,
            &http,
            &self.config.plc_runtime,
            &program,
            target_id,
        )
        .await?;
        tracing::info!(
            target_id,
            found = findings.len(),
            "provision-and-test complete"
        );

        let mut new_count = 0u32;
        for mut finding in findings {
            finding.scan_run_id = Some(scan_run_id.to_string());
            if self
                .db
                .findings()
                .find_one(doc! { "fingerprint": &finding.fingerprint })
                .await?
                .is_none()
            {
                self.db.findings().insert_one(&finding).await?;
                new_count += 1;
            }
        }
        Ok(new_count)
    }

    /// Probe a running PLC/SPS device over industrial protocols (Modbus/TCP, …)
    /// and persist findings for exposed / unauthenticated control access. The
    /// probe is read-only; it targets the Modbus port of the target's live URL.
    async fn run_ics_probe(
        &self,
        target: &OnboardedTarget,
        target_id: &str,
        scan_run_id: &str,
    ) -> Result<u32, AgentError> {
        self.update_phase(scan_run_id, "ics_probe").await;
        let Some(endpoint) = target.live_url().map(|a| a.source_ref.clone()) else {
            tracing::warn!(target_id, "ICS probe: no live URL");
            return Ok(0);
        };
        // Short per-request budget so an unreachable device doesn't stall the scan.
        let budget = std::time::Duration::from_secs(5);
        let findings = crate::pipeline::ics::probe_target(&endpoint, target_id, budget).await;
        tracing::info!(
            target_id,
            endpoint = %endpoint,
            found = findings.len(),
            "ICS probe complete"
        );
        let mut new_count = 0u32;
        for mut finding in findings {
            finding.scan_run_id = Some(scan_run_id.to_string());
            if self
                .db
                .findings()
                .find_one(doc! { "fingerprint": &finding.fingerprint })
                .await?
                .is_none()
            {
                self.db.findings().insert_one(&finding).await?;
                new_count += 1;
            }
        }
        Ok(new_count)
    }

    /// Store a control-application SBOM (CODESYS libraries + runtime) for a target
    /// and match it against known CVEs. Scoped to `package_manager = "codesys"` so
    /// it refreshes on re-scan and coexists with any firmware/source SBOM. The
    /// runtime `Cmp*` / `3SLicense` components carry real CODESYS advisories, so
    /// this is where PLC-device CVE coverage comes from.
    async fn persist_control_app_sbom(
        &self,
        target_id: &str,
        target_name: &str,
        mut entries: Vec<SbomEntry>,
    ) -> Result<(), AgentError> {
        if entries.is_empty() {
            return Ok(());
        }
        self.db
            .sbom_entries()
            .delete_many(doc! { "repo_id": target_id, "package_manager": "codesys" })
            .await?;

        let cve_scanner = CveScanner::new(
            self.http.clone(),
            self.config.searxng_url.clone(),
            self.config.nvd_api_key.as_ref().map(|k| {
                use secrecy::ExposeSecret;
                k.expose_secret().to_string()
            }),
        );
        let mut alerts = match tokio::time::timeout(
            std::time::Duration::from_secs(600),
            cve_scanner.scan_dependencies(target_id, &mut entries),
        )
        .await
        {
            Ok(Ok(a)) => a,
            Ok(Err(e)) => {
                tracing::warn!(target_id, error = %e, "control-app CVE scan failed");
                Vec::new()
            }
            Err(_) => {
                tracing::warn!(target_id, "control-app CVE scan timed out");
                Vec::new()
            }
        };
        // OSV can't match `pkg:codesys/*` (no such ecosystem); CODESYS advisories
        // live in NVD keyed by CPE + runtime version. Add those (best-effort).
        if let Ok(codesys) = tokio::time::timeout(
            std::time::Duration::from_secs(120),
            cve_scanner.scan_codesys(target_id, &mut entries),
        )
        .await
        {
            alerts.extend(codesys);
        } else {
            tracing::warn!(target_id, "CODESYS CVE match timed out");
        }

        for entry in &entries {
            let filter = doc! {
                "repo_id": &entry.repo_id,
                "name": &entry.name,
                "version": &entry.version,
            };
            if let Ok(d) = mongodb::bson::to_document(entry) {
                self.db
                    .sbom_entries()
                    .update_one(filter, doc! { "$set": d })
                    .upsert(true)
                    .await?;
            }
        }
        let new_notifs = self
            .persist_cve_alerts(target_id, target_name, &alerts)
            .await?;
        tracing::info!(
            target_id,
            components = entries.len(),
            alerts = alerts.len(),
            notifications = new_notifs,
            "control-app SBOM stored"
        );
        Ok(())
    }

    /// Upsert CVE alerts for a target and create dedup'd CVE notifications;
    /// returns the number of newly-created notifications. Shared by the SAST
    /// pipeline and the PLC control-app SBOM path, so every SBOM source (source,
    /// firmware, CODESYS libraries/runtime) raises the same notifications.
    async fn persist_cve_alerts(
        &self,
        repo_id: &str,
        repo_name: &str,
        alerts: &[CveAlert],
    ) -> Result<u32, AgentError> {
        use compliance_core::models::notification::{parse_severity, CveNotification};

        let mut new_notif = 0u32;
        for alert in alerts {
            let filter = doc! { "cve_id": &alert.cve_id, "repo_id": &alert.repo_id };
            let update = mongodb::bson::to_document(alert)
                .map(|d| doc! { "$set": d })
                .unwrap_or_else(|_| doc! {});
            self.db
                .cve_alerts()
                .update_one(filter, update)
                .upsert(true)
                .await?;

            // Dedup notifications by cve + repo + package + version.
            let notif_filter = doc! {
                "cve_id": &alert.cve_id,
                "repo_id": &alert.repo_id,
                "package_name": &alert.affected_package,
                "package_version": &alert.affected_version,
            };
            let severity = parse_severity(alert.severity.as_deref(), alert.cvss_score);
            let mut notification = CveNotification::new(
                alert.cve_id.clone(),
                repo_id.to_string(),
                repo_name.to_string(),
                alert.affected_package.clone(),
                alert.affected_version.clone(),
                severity,
            );
            notification.cvss_score = alert.cvss_score;
            notification.summary = alert.summary.clone();
            notification.url = Some(format!("https://osv.dev/vulnerability/{}", alert.cve_id));
            let notif_update = doc! {
                "$setOnInsert": mongodb::bson::to_bson(&notification).unwrap_or_default()
            };
            if let Ok(result) = self
                .db
                .cve_notifications()
                .update_one(notif_filter, notif_update)
                .upsert(true)
                .await
            {
                if result.upserted_id.is_some() {
                    new_notif += 1;
                }
            }
        }
        Ok(new_notif)
    }

    /// Ingest the target's artifacts, classify (tramiton for firmware/RTOS/Yocto,
    /// heuristics otherwise), and store the detected classification on the target.
    /// Best-effort — never fails the scan.
    async fn classify_and_store(
        &self,
        target: &OnboardedTarget,
        target_id: &str,
        scan_run_id: &str,
    ) {
        self.update_phase(scan_run_id, "classification").await;
        let ctx = crate::ingest::IngestContext::from_config(&self.config, target_id);
        let ingest_set = match crate::ingest::ingest_all(target, &ctx) {
            Ok(set) => set,
            Err(e) => {
                tracing::warn!(target_id, error = %e, "Unified pipeline: ingest for classification failed");
                return;
            }
        };
        let working_paths = ingest_set.working_paths();
        match crate::classify::classify_target(
            target,
            &working_paths,
            &crate::classify::TramitonNative,
        )
        .await
        {
            Ok(classification) => {
                tracing::info!(
                    target_id,
                    suggested = %classification.suggested,
                    "Unified pipeline: classified target"
                );
                if let (Some(oid), Ok(bson)) = (target.id, mongodb::bson::to_bson(&classification))
                {
                    let _ = self
                        .db
                        .onboarded_targets()
                        .update_one(
                            doc! { "_id": oid },
                            doc! { "$set": { "classification": bson } },
                        )
                        .await;
                }
            }
            Err(e) => {
                tracing::warn!(target_id, error = %e, "Unified pipeline: classification failed")
            }
        }

        // Analysis-based firmware SBOM: for embedded targets, derive components
        // (resolved libraries + cross-toolchain) from tramiton's build-plan
        // analysis over the already-ingested source — no build, no binary
        // upload. Best-effort; empty when no build plan forms.
        if crate::pipeline::firmware_sbom::is_firmware_target(target.target_type) {
            if let Some(code) = target.code_artifact() {
                if let Some(path) = working_paths.get(&code.id) {
                    let entries =
                        crate::pipeline::firmware_sbom::firmware_sbom_entries(path, target_id)
                            .await;
                    if !entries.is_empty() {
                        let _ = self
                            .db
                            .sbom_entries()
                            .delete_many(doc! { "repo_id": target_id })
                            .await;
                        for entry in &entries {
                            let filter = doc! {
                                "repo_id": &entry.repo_id,
                                "name": &entry.name,
                                "version": &entry.version,
                            };
                            if let Ok(d) = mongodb::bson::to_document(entry) {
                                let _ = self
                                    .db
                                    .sbom_entries()
                                    .update_one(filter, doc! { "$set": d })
                                    .upsert(true)
                                    .await;
                            }
                        }
                        tracing::info!(
                            target_id,
                            count = entries.len(),
                            "Firmware SBOM: stored components from tramiton analysis"
                        );
                    }
                }
            }
        }
    }

    /// If the target has a `LiveUrl` artifact and DAST is planned, provision a
    /// `DastTarget` (keyed by `repo_id` = target id) so the existing DAST trigger
    /// fires for wizard-created targets. Idempotent.
    async fn ensure_dast_target(
        &self,
        target: &OnboardedTarget,
        plan: &crate::pipeline::plan::ScanPlan,
    ) {
        if !plan.has(ScanType::Dast) {
            return;
        }
        let (Some(url), Some(oid)) = (target.live_url(), target.id) else {
            return;
        };
        let target_id = oid.to_hex();
        if self
            .db
            .dast_targets()
            .find_one(doc! { "repo_id": &target_id })
            .await
            .ok()
            .flatten()
            .is_some()
        {
            return; // already provisioned
        }
        let kind = url
            .web
            .as_ref()
            .map(|w| w.target_kind.clone())
            .unwrap_or(DastTargetType::WebApp);
        let mut dast = DastTarget::new(target.name.clone(), url.source_ref.clone(), kind);
        dast.repo_id = Some(target_id);
        if let Some(web) = &url.web {
            dast.excluded_paths = web.excluded_paths.clone();
            dast.max_crawl_depth = web.max_crawl_depth;
            dast.rate_limit = web.rate_limit;
            dast.allow_destructive = web.allow_destructive;
        }
        if let Some(auth) = &url.auth {
            dast.auth_config = Some(DastAuthConfig {
                method: auth.method.clone(),
                login_url: auth.login_url.clone(),
                username: auth.username.clone(),
                password: None,
                token: auth.secret.clone(),
                headers: auth.headers.clone(),
            });
        }
        if let Err(e) = self.db.dast_targets().insert_one(&dast).await {
            tracing::warn!(error = %e, "Unified pipeline: failed to provision DAST target");
        }
    }

    /// Sync the onboarded-target document after a scan: bump `findings_count`
    /// and advance the git artifact's `last_scanned_commit` watermark.
    async fn finalize_target(
        &self,
        target: &OnboardedTarget,
        repo: &RepoView,
        new_count: u32,
    ) -> Result<(), AgentError> {
        let oid = match target.id {
            Some(id) => id,
            None => return Ok(()),
        };
        self.db
            .onboarded_targets()
            .update_one(
                doc! { "_id": oid },
                doc! {
                    "$inc": { "findings_count": new_count as i64 },
                    "$set": { "updated_at": mongodb::bson::DateTime::now() },
                },
            )
            .await?;

        let repo_path = std::path::Path::new(&self.config.git_clone_base_path).join(&repo.name);
        if let (Ok(sha), Some(code)) = (GitOps::get_head_sha(&repo_path), target.code_artifact()) {
            self.db
                .onboarded_targets()
                .update_one(
                    doc! { "_id": oid, "artifacts.id": &code.id },
                    doc! { "$set": { "artifacts.$.git.last_scanned_commit": sha } },
                )
                .await?;
        }
        Ok(())
    }

    pub(super) async fn update_phase(&self, scan_run_id: &str, phase: &str) {
        if let Ok(oid) = mongodb::bson::oid::ObjectId::parse_str(scan_run_id) {
            let _ = self
                .db
                .scan_runs()
                .update_one(
                    doc! { "_id": oid },
                    doc! {
                        "$set": { "current_phase": phase },
                        "$push": { "phases_completed": phase },
                    },
                )
                .await;
        }
    }
}

/// Extract the scheme + host from a git URL.
/// e.g. "https://gitea.example.com/owner/repo.git" -> "https://gitea.example.com"
/// e.g. "ssh://git@gitea.example.com:22/owner/repo.git" -> "https://gitea.example.com"
pub(super) fn extract_base_url(git_url: &str) -> Option<String> {
    if let Some(rest) = git_url.strip_prefix("https://") {
        let host = rest.split('/').next()?;
        Some(format!("https://{host}"))
    } else if let Some(rest) = git_url.strip_prefix("http://") {
        let host = rest.split('/').next()?;
        Some(format!("http://{host}"))
    } else if let Some(rest) = git_url.strip_prefix("ssh://") {
        // ssh://git@host:port/path -> extract host
        let after_at = rest.find('@').map(|i| &rest[i + 1..]).unwrap_or(rest);
        let host = after_at.split(&[':', '/'][..]).next()?;
        Some(format!("https://{host}"))
    } else if let Some(at_pos) = git_url.find('@') {
        // SCP-style: git@host:owner/repo.git
        let after_at = &git_url[at_pos + 1..];
        let host = after_at.split(':').next()?;
        Some(format!("https://{host}"))
    } else {
        None
    }
}

#[cfg(test)]
#[allow(clippy::expect_used, clippy::unwrap_used)]
mod tests {
    use super::*;
    use compliance_core::models::{
        Artifact, ArtifactAuth, IssueTrackerConfig, TargetType, TrackerType,
    };

    #[test]
    fn repo_view_preserves_id_git_auth_and_tracker() {
        let mut target = OnboardedTarget::new("acme".to_string(), TargetType::WebApp);
        target.id = Some(mongodb::bson::oid::ObjectId::new());
        target.findings_count = 3;
        target.scan_config.issue_tracker = Some(IssueTrackerConfig {
            tracker_type: Some(TrackerType::Gitea),
            owner: Some("acme".to_string()),
            repo: Some("web".to_string()),
            token: Some("tt".to_string()),
        });

        let mut artifact = Artifact::git_repo("https://git/acme.git", "develop");
        if let Some(git) = artifact.git.as_mut() {
            git.last_scanned_commit = Some("abc123".to_string());
        }
        artifact.auth = Some(ArtifactAuth {
            method: "token".to_string(),
            username: Some("bob".to_string()),
            secret: Some("pat".to_string()),
            ..Default::default()
        });
        target.artifacts.push(artifact);

        let code = target.code_artifact().expect("code artifact");
        let repo = RepoView::from_target(&target, code);

        assert_eq!(repo.id, target.id); // preserved
        assert_eq!(repo.git_url, "https://git/acme.git");
        assert_eq!(repo.default_branch, "develop");
        assert_eq!(repo.last_scanned_commit.as_deref(), Some("abc123"));
        assert_eq!(repo.auth_token.as_deref(), Some("pat"));
        assert_eq!(repo.auth_username.as_deref(), Some("bob"));
        assert_eq!(repo.tracker_type, Some(TrackerType::Gitea));
        assert_eq!(repo.tracker_owner.as_deref(), Some("acme"));
        assert_eq!(repo.findings_count, 3);
    }
}