Failed to generate code suggestions for PR

Greptile Overview

Greptile Summary

This PR fixes two critical issues with self-reporting trigger ingestion:

Key Changes:

• Fixed cross-org data leakage: Triggers are now grouped by organization before ingestion, ensuring each org's triggers are ingested separately to the correct org's stream (queues.rs:303-334)
• Fixed alert manager blocking on expired orgs: Added cloud-specific check to skip expired organizations during trigger ingestion, preventing the alert manager from being blocked by expired orgs (queues.rs:310-322)
• Fixed variable reference bug: Corrected usage of local usage_reporting_mode variable instead of cfg.common.usage_reporting_mode in error handling (ingestion.rs:194)

Implementation Details:
The refactored logic moves the per-org trigger ingestion from the additional_reporting_orgs loop to a separate section. It builds a HashMap<String, Vec<json::Value>> to group triggers by org, validates each org's trial period status (cloud only), and ingests each org's triggers independently. This ensures triggers for org1 don't accidentally get ingested to org2's stream.

Confidence Score: 4/5

• Safe to merge with minor considerations - fixes critical data leakage and blocking issues
• The changes address critical production issues (cross-org data leakage and alert manager blocking). The logic is sound: grouping by org prevents data leakage, and skipping expired orgs prevents blocking. Score reduced from 5 to 4 due to the error handling in the cloud-specific check - continuing on error could still cause issues, though it's logged.
• No files require special attention - changes are straightforward bug fixes

Important Files Changed

File Analysis

Sequence Diagram

sequenceDiagram
    participant Queue as ReportingQueue
    participant Ingest as ingest_buffered_data
    participant OrgCheck as organization::is_org_in_free_trial_period
    participant IngestionSvc as ingestion::ingest_reporting_data
    
    Note over Queue,Ingest: Buffered triggers arrive
    Queue->>Ingest: triggers (Vec<json::Value>)
    
    alt additional_reporting_orgs configured
        Note over Ingest: Send to configured orgs (META_ORG, etc.)
        loop for each additional_reporting_org
            Ingest->>IngestionSvc: ingest_reporting_data(triggers.clone(), trigger_stream)
        end
    end
    
    alt usage_report_to_own_org enabled && mode != "remote"
        Note over Ingest: Group triggers by org
        loop for each trigger in triggers
            Ingest->>Ingest: Parse TriggerData from JSON
            alt cloud feature enabled
                Ingest->>OrgCheck: is_org_in_free_trial_period(org_id)
                OrgCheck-->>Ingest: Ok(ongoing) or Err(e)
                alt !ongoing or error
                    Note over Ingest: Skip this trigger (expired org)
                end
            end
            Note over Ingest: Add trigger to per_org_map[org_id]
        end
        
        loop for each (org, values) in per_org_map
            Ingest->>IngestionSvc: ingest_reporting_data(values, trigger_stream)
            Note over Ingest: Each org's triggers ingested separately
        end
    end