add transaction safety and update comments from code review

Signed-off-by: mrrajan <86094767+mrrajan@users.noreply.github.com.>

Author: mrrajan

modules/fundamental/src/license/service/mod.rs

@@ -297,22 +297,29 @@ impl LicenseService {
             .distinct()
             .column_as(expanded_license::Column::ExpandedText, LICENSE_TEXT);
 
-        // Build query for non-expanded licenses: includes both
+        // Build query for licenses not yet linked to any SBOM: includes both
         //   (a) pre-loaded SPDX dictionary entries with no SBOM connection yet, AND
-        //   (b) CycloneDX licenses that exist in sbom_package_license but were never expanded.
-        // A LEFT JOIN on sbom_package_license (instead of INNER JOIN) ensures pre-loaded licenses
-        // with no SBOM attachment are included. Then filtering for sbom_license_expanded IS NULL
-        // removes SPDX licenses that have already been expanded (they appear in spdx_query instead).
+        //   (b) licenses from older SBOMs ingested before license expansion was implemented.
+        // Use NOT EXISTS instead of LEFT JOIN + IS NULL to find licenses without SBOMs.
+        // On large tables, LEFT JOIN scans all rows while NOT EXISTS
+        // uses a Nested Loop Anti Join with index-only scan.
+        let exists_subquery = sea_query::Query::select()
+            .expr(Expr::val(1))
+            .from(sbom_license_expanded::Entity)
+            .and_where(
+                Expr::col((
+                    sbom_license_expanded::Entity,
+                    sbom_license_expanded::Column::LicenseId,
+                ))
+                .equals((license::Entity, license::Column::Id)),
+            )
+            .to_owned();
+
         let mut non_sbom_query = license::Entity::find()
             .select_only()
             .distinct()
             .column_as(license::Column::Text, LICENSE_TEXT)
-            .join(JoinType::LeftJoin, license::Relation::PackageLicense.def())
-            .join(
-                JoinType::LeftJoin,
-                sbom_license_expanded::Relation::License.def().rev(),
-            )
-            .filter(sbom_license_expanded::Column::LicenseId.is_null());
+            .filter(Expr::exists(exists_subquery).not());
 
         // Apply filtering to both queries (without sorting - that's applied to the UNION result)
         let filter_only = Query {

modules/ingestor/src/graph/sbom/common/expanded_license.rs

@@ -1,4 +1,4 @@
-use sea_orm::{ConnectionTrait, DbErr, Statement};
+use sea_orm::{ConnectionTrait, DbErr, Statement, TransactionTrait};
 use uuid::Uuid;
 
 /// Populates expanded_license and sbom_license_expanded tables during SBOM ingestion
@@ -16,13 +16,27 @@ use uuid::Uuid;
 ///
 /// While SeaORM could express this via custom expressions, it would be significantly
 /// more verbose and harder to maintain than the raw SQL.
-pub async fn populate_expanded_license(
-    sbom_id: Uuid,
-    db: &impl ConnectionTrait,
-) -> Result<(), DbErr> {
+///
+/// # Differences from Migration Backfill
+///
+/// The migration in m0002120_normalize_expanded_license/up.sql performs a similar
+/// operation but with key differences:
+/// - Migration: Pre-deduplicates by (text, sbom_id) and uses WHERE NOT EXISTS to skip
+///   already-backfilled SBOMs. Optimized for one-time bulk processing.
+/// - Ingestion: Filters by specific sbom_id parameter for single-SBOM processing.
+///   Uses ON CONFLICT for idempotent re-ingestion of the same SBOM.
+///
+/// Both use the same core logic (expand_license_expression_with_mappings + md5 hash
+/// matching) but optimize for their different use cases.
+pub async fn populate_expanded_license<C>(sbom_id: Uuid, db: &C) -> Result<(), DbErr>
+where
+    C: ConnectionTrait + TransactionTrait,
+{
+    let txn = db.begin().await?;
+
     // Step 1: Insert into expanded_license dictionary
-    db.execute(Statement::from_sql_and_values(
-        db.get_database_backend(),
+    txn.execute(Statement::from_sql_and_values(
+        txn.get_database_backend(),
         r#"
 INSERT INTO expanded_license (expanded_text)
 SELECT DISTINCT expand_license_expression_with_mappings(
@@ -45,8 +59,8 @@ ON CONFLICT (text_hash) DO NOTHING
 
     // Step 2: Insert into sbom_license_expanded junction table
     // Use CTE to call expand_license_expression_with_mappings() only once per (sbom_id, license_id)
-    db.execute(Statement::from_sql_and_values(
-        db.get_database_backend(),
+    txn.execute(Statement::from_sql_and_values(
+        txn.get_database_backend(),
         r#"
 WITH license_expansions AS (
     SELECT DISTINCT
@@ -76,5 +90,7 @@ SET expanded_license_id = EXCLUDED.expanded_license_id
     ))
     .await?;
 
+    txn.commit().await?;
+
     Ok(())
 }

modules/ingestor/src/graph/sbom/cyclonedx.rs

@@ -21,7 +21,7 @@ use sbom_walker::{
     model::sbom::serde_cyclonedx::Sbom,
     report::{ReportSink, check},
 };
-use sea_orm::ConnectionTrait;
+use sea_orm::{ConnectionTrait, TransactionTrait};
 use serde_cyclonedx::cyclonedx::v_1_6::{
     Component, ComponentEvidenceIdentity, CycloneDx, LicenseChoiceUrl, OrganizationalContact,
 };
@@ -138,12 +138,15 @@ impl<'a> From<Information<'a>> for SbomInformation {
 
 impl SbomContext {
     #[instrument(skip(connection, sbom, warnings), err(level=tracing::Level::INFO))]
-    pub async fn ingest_cyclonedx<C: ConnectionTrait>(
+    pub async fn ingest_cyclonedx<C>(
         &self,
         mut sbom: Box<CycloneDx>,
         warnings: &dyn ReportSink,
         connection: &C,
-    ) -> Result<(), Error> {
+    ) -> Result<(), Error>
+    where
+        C: ConnectionTrait + TransactionTrait,
+    {
         // pre-flight checks
 
         check::serde_cyclonedx::all(warnings, &Sbom::V1_6(Cow::Borrowed(&sbom)));
@@ -285,11 +288,10 @@ impl<'a> Creator<'a> {
     }
 
     #[instrument(skip(self, db, processors), err(level=tracing::Level::INFO))]
-    pub async fn create(
-        self,
-        db: &impl ConnectionTrait,
-        processors: &mut [Box<dyn Processor>],
-    ) -> Result<(), Error> {
+    pub async fn create<C>(self, db: &C, processors: &mut [Box<dyn Processor>]) -> Result<(), Error>
+    where
+        C: ConnectionTrait + TransactionTrait,
+    {
         let mut creator = ComponentCreator::new(self.sbom_id, self.components.len());
 
         for comp in self.components {
@@ -576,7 +578,10 @@ impl ComponentCreator {
     // order matters to prevent cross-table deadlocks when running
     // concurrent SBOM ingestions. All SBOM loaders must use the same
     // table insertion order.
-    async fn create(self, db: &impl ConnectionTrait) -> Result<(), Error> {
+    async fn create<C>(self, db: &C) -> Result<(), Error>
+    where
+        C: ConnectionTrait + TransactionTrait,
+    {
         self.licenses.create(db).await?;
         self.purls.create(db).await?;
         self.cpes.create(db).await?;

modules/ingestor/src/graph/sbom/spdx.rs

@@ -16,7 +16,7 @@ use crate::{
     service::Error,
 };
 use sbom_walker::report::{ReportSink, check};
-use sea_orm::ConnectionTrait;
+use sea_orm::{ConnectionTrait, TransactionTrait};
 use spdx_rs::models::{RelationshipType, SPDX};
 use std::collections::HashSet;
 use std::str::FromStr;
@@ -103,12 +103,15 @@ impl<'a> From<Information<'a>> for SbomInformation {
 
 impl SbomContext {
     #[instrument(skip(db, sbom_data, warnings), ret(level=tracing::Level::DEBUG))]
-    pub async fn ingest_spdx<C: ConnectionTrait>(
+    pub async fn ingest_spdx<C>(
         &self,
         sbom_data: SPDX,
         warnings: &dyn ReportSink,
         db: &C,
-    ) -> Result<(), Error> {
+    ) -> Result<(), Error>
+    where
+        C: ConnectionTrait + TransactionTrait,
+    {
         // pre-flight checks
 
         check::spdx::all(warnings, &sbom_data);