fix: add transaction safety and update comments from code review

This ensures both expanded_license dictionary and sbom_license_expanded
junction table inserts are atomic - if either fails, both roll back.

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>

Author: mrrajan

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

@@ -285,9 +285,9 @@ 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.
+        //   (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.

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

@@ -1,9 +1,9 @@
-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
 ///
-/// This function uses a single SQL statement with CTEs to:
+/// This function uses two SQL statements within a transaction to:
 /// 1. Call expand_license_expression_with_mappings() once per license
 /// 2. Insert distinct expanded texts into the expanded_license dictionary
 /// 3. Populate the sbom_license_expanded junction table
@@ -16,13 +16,23 @@ 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.
+///
+/// **Transaction safety**: Both dictionary and junction table inserts run in a single
+/// transaction to prevent partial state if the second insert fails.
+///
+/// **Note on SQL duplication**: Similar SQL appears in migration m0002120 for backfilling
+/// existing data. The migration processes ALL SBOMs at once, while this function runs
+/// per-SBOM during ingestion. Keep both in sync when updating license expansion logic.
 pub async fn populate_expanded_license(
     sbom_id: Uuid,
-    db: &impl ConnectionTrait,
+    db: &(impl ConnectionTrait + TransactionTrait),
 ) -> Result<(), DbErr> {
+    // Begin transaction to ensure atomicity between dictionary and junction table inserts
+    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 +55,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 +86,8 @@ SET expanded_license_id = EXCLUDED.expanded_license_id
     ))
     .await?;
 
+    // Commit transaction - both inserts succeed or both roll back
+    txn.commit().await?;
+
     Ok(())
 }

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

@@ -134,7 +134,7 @@ 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: ConnectionTrait + sea_orm::TransactionTrait>(
         &self,
         mut sbom: Box<CycloneDx>,
         warnings: &dyn ReportSink,
@@ -283,7 +283,7 @@ impl<'a> Creator<'a> {
     #[instrument(skip(self, db, processors), err(level=tracing::Level::INFO))]
     pub async fn create(
         self,
-        db: &impl ConnectionTrait,
+        db: &(impl ConnectionTrait + sea_orm::TransactionTrait),
         processors: &mut [Box<dyn Processor>],
     ) -> Result<(), Error> {
         let mut purls = PurlCreator::new();

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

@@ -102,7 +102,7 @@ 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: ConnectionTrait + sea_orm::TransactionTrait>(
         &self,
         sbom_data: SPDX,
         warnings: &dyn ReportSink,