Bridging the Offline-Online Gap: Ingesting Batch Data to GA4 Server-Side with Cloud Storage, Cloud Run & Pub/Sub

You've mastered the art of building a sophisticated server-side Google Analytics 4 (GA4) pipeline, leveraging Google Tag Manager (GTM) Server Container on Cloud Run to centralize data collection, apply transformations, enrich events, and enforce granular consent. This architecture provides unparalleled control and data quality, forming the backbone of your modern analytics strategy for online interactions.

However, a critical challenge for achieving a truly holistic understanding of your customer journey often remains: integrating offline data. Many businesses generate valuable event data from non-web sources such as in-store purchases, call center interactions, CRM updates, or loyalty program activities. This offline data, when siloed from your online analytics, creates a fragmented view of the customer, hinders accurate attribution, and limits the potential for comprehensive personalization.

The problem is how to reliably and scalably ingest these large volumes of historical or ongoing batch offline data into GA4 (and potentially other marketing platforms) via a serverless pipeline, ensuring data quality, consistent user stitching, and accurate historical timestamps. Manually uploading CSVs to GA4's Data Import API offers limited automation and lacks the robust transformation, validation, and consent enforcement capabilities inherent in a server-side data pipeline.

The Problem: Fragmented Customer Views and Offline Data Silos

Relying solely on online event tracking means you miss crucial parts of the customer journey. Consider these scenarios:

In-store Conversions: A user browses online (tracked by GA4), then completes a purchase in a physical store. How do you attribute that offline purchase to their online browsing behavior?
Call Center Interactions: A customer calls support, and the call center agent records an issue_resolved event. This is a critical touchpoint, but it never reaches GA4.
CRM Lead Lifecycle: A lead is qualified by a sales rep in the CRM. This offline status update is highly valuable but doesn't automatically enrich GA4 user profiles.
Historical Data: You acquire a new dataset of past customer interactions (e.g., from an old loyalty program) that you want to integrate into GA4 for long-term trend analysis.

Without a robust server-side ingestion pipeline, these offline events remain isolated, leading to:

Incomplete Customer 360: You cannot see the full picture of customer behavior across all touchpoints.
Skewed Attribution: Online campaigns might be under-attributed if offline conversions are not linked.
Limited Personalization: Personalization decisions cannot factor in crucial offline signals.
Compliance Gaps: Ensuring PII handling and consent for offline data requires a controlled process.

Why Server-Side for Offline Data Ingestion?

Extending your server-side capabilities to ingest offline data offers significant advantages:

Unified Data Quality & Transformation: Leverage your existing server-side logic for schema validation, PII scrubbing, data normalization, and enrichment for offline data, ensuring consistency with your online events.
Consistent User Stitching: Map offline user_ids to existing online client_ids (using your Identity & Session Resolver) to build a truly unified customer journey.
Accurate Historical Timestamps: Precisely control event timestamps for accurate historical placement in GA4, even for events that occurred weeks or months ago.
Deduplication: Utilize robust event_id generation (as per Server-Side Event Deduplication) to prevent duplicate offline event ingestion.
Scalability & Resilience: Serverless components (Cloud Storage, Cloud Functions, Pub/Sub, Cloud Run) handle varying volumes of batch data, ensuring reliable processing without manual intervention.
Auditability: Log raw and processed offline events to your BigQuery data lake for a complete audit trail.
Cost Efficiency: Pay only for the resources consumed during ingestion and processing.

Our Solution Architecture: A Serverless Offline Ingestion Pipeline

Our solution builds a serverless pipeline on Google Cloud to receive batch offline data, process it, and send it to GA4. This pipeline is triggered by new files landing in Cloud Storage.

graph TD
    subgraph Offline Data Source
        A[CRM Export/POS System/Internal DB] -->|1. Export Batch File (CSV/JSON)| B(Google Cloud Storage: Landing Zone);
    end

    subgraph Serverless Ingestion & Transformation
        B -->|2. GCS File Upload Trigger| C(Cloud Function: File Processor);
        C -->|3. Read File, Parse Rows, Apply Transformations/Validation| C;
        C -->|4. Publish Processed Event JSON to Pub/Sub| D(Google Cloud Pub/Sub Topic: Offline_Events);
    end

    subgraph GA4 Ingestion Service
        D -->|5. Pub/Sub Push Subscription| E(GA4 Offline Ingestion Service on Cloud Run);
        E -->|6. Construct GA4 Measurement Protocol Hit| E;
        E -->|7. Send to GA4 Measurement Protocol| F[Google Analytics 4];
    end

    subgraph Audit & Debugging
        C -->|8. Log Raw/Processed/Failed Rows| G[BigQuery: Offline Events Audit];
        E -->|9. Log GA4 MP Responses| H[Cloud Logging];
    end

    F --> I[GA4 Reports & Insights];

Key Flow:

Offline Data Export: Your backend system (CRM, POS, internal database) regularly exports batch files (CSV, JSON) containing offline event data.
Cloud Storage Landing Zone: These files are uploaded to a designated Cloud Storage bucket.
Cloud Function Trigger: A Cloud Function is triggered automatically when a new file lands in the bucket.
File Processing & Pub/Sub Publishing: The Cloud Function reads the file, parses each row as an individual event, applies necessary transformations (e.g., PII hashing, schema mapping, event_id generation, timestamp conversion), and publishes each processed event to a Pub/Sub topic.
GA4 Offline Ingestion Service: A dedicated Cloud Run service subscribes to this Pub/Sub topic. It receives each processed offline event.
GA4 Measurement Protocol Dispatch: The Cloud Run service constructs a valid GA4 Measurement Protocol hit using the event data, ensuring correct user_id, client_id, timestamp_micros, and _eid for accurate historical placement and deduplication. It then sends this hit to GA4.
Audit Trail: Both the Cloud Function and Cloud Run service log their activities and payloads to BigQuery for auditing, debugging, and reconciliation.

Core Components Deep Dive & Implementation Steps

1. Cloud Storage Landing Zone

Create a Cloud Storage bucket to serve as the entry point for your batch offline data.

gcloud storage buckets create gs://your-offline-data-landing-zone --project YOUR_GCP_PROJECT_ID --location YOUR_GCP_REGION

2. Source File Format Example

Assume your incoming CSV files have a header and contain data like this:

offline_event_id,event_name,event_timestamp,user_id,email_address,transaction_id,value,currency,item_ids
OFFLINE001,in_store_purchase,2024-02-20T10:30:00Z,user123,[email protected],TRANSXYZ,50.00,USD,PROD001;PROD002
OFFLINE002,call_center_lead,2024-02-21T14:15:00Z,user456,[email protected],,0.00,USD,

Or JSON:

[
  {
    "offline_event_id": "OFFLINE001",
    "event_name": "in_store_purchase",
    "event_timestamp": "2024-02-20T10:30:00Z",
    "user_id": "user123",
    "email_address": "[email protected]",
    "transaction_id": "TRANSXYZ",
    "value": 50.00,
    "currency": "USD",
    "item_ids": "PROD001;PROD002"
  },
  {
    "offline_event_id": "OFFLINE002",
    "event_name": "call_center_lead",
    "event_timestamp": "2024-02-21T14:15:00Z",
    "user_id": "user456",
    "email_address": "[email protected]",
    "value": 0.00,
    "currency": "USD"
  }
]

3. Cloud Function (GCS Trigger)

This Cloud Function will trigger when a file is uploaded to the landing zone. It will read the file, parse its contents, perform basic transformations (like PII hashing, event_id generation), and publish each row as a message to Pub/Sub.

file-processor-function/main.py:

import os
import json
import csv
import io
import datetime
import uuid
import hashlib
from google.cloud import storage, pubsub_v1, bigquery
import logging

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Google Cloud Clients
storage_client = storage.Client()
pubsub_publisher = pubsub_v1.PublisherClient()
bigquery_client = bigquery.Client()

# Configuration from environment variables
PROJECT_ID = os.environ.get('GCP_PROJECT_ID')
PUBSUB_TOPIC_ID = os.environ.get('PUBSUB_OFFLINE_EVENTS_TOPIC', 'offline-events-topic')
PUBSUB_TOPIC_PATH = pubsub_publisher.topic_path(PROJECT_ID, PUBSUB_TOPIC_ID)

BQ_AUDIT_DATASET = os.environ.get('BQ_AUDIT_DATASET', 'offline_events_audit')
BQ_AUDIT_TABLE = os.environ.get('BQ_AUDIT_TABLE', 'processed_offline_events')
BQ_AUDIT_TABLE_FULL_ID = f"{PROJECT_ID}.{BQ_AUDIT_DATASET}.{BQ_AUDIT_TABLE}"

# PII fields to hash (example)
PII_FIELDS_TO_HASH = ['email_address', 'phone_number']

def hash_value_sha256(value):
    if value:
        return hashlib.sha256(value.lower().strip().encode('utf-8')).hexdigest()
    return None

def process_file_from_gcs(event, context):
    """Triggered by a change to a Cloud Storage bucket."""
    bucket_name = event['bucket']
    file_name = event['name']
    file_uri = f"gs://{bucket_name}/{file_name}"
    
    logger.info(f"Processing file: {file_uri}.")

    try:
        bucket = storage_client.bucket(bucket_name)
        blob = bucket.blob(file_name)
        file_contents = blob.download_as_text()

        processed_rows = []
        
        if file_name.endswith('.csv'):
            reader = csv.DictReader(io.StringIO(file_contents))
            for row in reader:
                processed_rows.append(process_row(row, file_uri))
        elif file_name.endswith('.json'):
            json_data = json.loads(file_contents)
            if isinstance(json_data, list):
                for row in json_data:
                    processed_rows.append(process_row(row, file_uri))
            elif isinstance(json_data, dict): # Single event in JSON file
                 processed_rows.append(process_row(json_data, file_uri))
            else:
                logger.error(f"Unsupported JSON format in {file_name}. Expected list or single object.")
                return # Stop processing
        else:
            logger.error(f"Unsupported file type: {file_name}. Only CSV and JSON are supported.")
            return # Stop processing

        # Stream audit logs to BigQuery (consider batching for very high volumes)
        if processed_rows:
            errors = bigquery_client.insert_rows_json(BQ_AUDIT_TABLE_FULL_ID, processed_rows)
            if errors:
                logger.error(f"BigQuery audit insert errors: {errors}")
            else:
                logger.info(f"Successfully audited {len(processed_rows)} rows to BigQuery.")

    except Exception as e:
        logger.error(f"Error processing file {file_uri}: {e}", exc_info=True)


def process_row(row_data, file_uri):
    """Applies transformations and publishes a single row/event."""
    processed_event = {
        'offline_source_file': file_uri,
        'original_row_data': json.dumps(row_data), # Store original for audit
        'processing_timestamp': datetime.datetime.now(datetime.timezone.utc).isoformat(),
        'status': 'published',
        'errors': []
    }
    
    try:
        # Standardize event_id: Use existing offline_event_id or generate UUID
        event_id = row_data.get('offline_event_id') or str(uuid.uuid4())
        processed_event['event_id'] = event_id

        # Map event_name
        event_name = row_data.get('event_name', 'unknown_offline_event').lower().replace(' ', '_')
        processed_event['event_name'] = event_name

        # Timestamp conversion: Assume ISO format, convert to milliseconds since epoch
        event_timestamp_str = row_data.get('event_timestamp')
        if event_timestamp_str:
            try:
                # Handle various ISO formats, including Z for UTC
                dt_object = datetime.datetime.fromisoformat(event_timestamp_str.replace('Z', '+00:00'))
                # Ensure it's UTC for consistent `timestamp_micros` in GA4 MP
                if dt_object.tzinfo is None:
                    dt_object = pytz.utc.localize(dt_object) # Assume local if no tz, then convert to UTC
                else:
                    dt_object = dt_object.astimezone(pytz.utc)
                    
                event_timestamp_ms = int(dt_object.timestamp() * 1000)
                processed_event['event_timestamp_ms'] = event_timestamp_ms
                processed_event['event_timestamp_utc_iso'] = dt_object.isoformat()
            except ValueError:
                processed_event['errors'].append(f"Invalid event_timestamp format: {event_timestamp_str}")
                logger.warning(f"Event {event_id}: Invalid timestamp format.")
                # Fallback to current time if timestamp is invalid for critical events
                processed_event['event_timestamp_ms'] = int(datetime.datetime.now(datetime.timezone.utc).timestamp() * 1000)
                processed_event['event_timestamp_utc_iso'] = datetime.datetime.now(datetime.timezone.utc).isoformat()
        else:
            processed_event['errors'].append("Missing event_timestamp.")
            logger.warning(f"Event {event_id}: Missing timestamp. Using current time.")
            processed_event['event_timestamp_ms'] = int(datetime.datetime.now(datetime.timezone.utc).timestamp() * 1000)
            processed_event['event_timestamp_utc_iso'] = datetime.datetime.now(datetime.timezone.utc).isoformat()


        # User Identifiers
        processed_event['user_id'] = row_data.get('user_id')
        
        # PII Hashing (e.g., email_address)
        for pii_field in PII_FIELDS_TO_HASH:
            if row_data.get(pii_field):
                hashed_value = hash_value_sha256(row_data[pii_field])
                processed_event[f'{pii_field}_hashed_sha256'] = hashed_value
                logger.debug(f"Hashed {pii_field} for event {event_id}.")
            
        # Common GA4 e-commerce parameters
        if 'value' in row_data:
            try:
                processed_event['value'] = float(row_data['value'])
            except ValueError:
                processed_event['errors'].append(f"Invalid value format: {row_data['value']}")
                processed_event['value'] = 0.0 # Default to 0.0 on error
        
        processed_event['currency'] = row_data.get('currency')
        processed_event['transaction_id'] = row_data.get('transaction_id') # For purchase deduplication

        # Item array processing (simple example for item_ids string)
        if row_data.get('item_ids'):
            item_ids = [i.strip() for i in row_data['item_ids'].split(';') if i.strip()]
            processed_event['items'] = [{'item_id': item_id} for item_id in item_ids]
        
        # Add a flag to identify these as offline events
        processed_event['is_offline_event'] = True

        # --- Publish to Pub/Sub ---
        message_data = json.dumps(processed_event).encode('utf-8')
        future = pubsub_publisher.publish(PUBSUB_TOPIC_PATH, message_data, event_name=event_name, event_id=event_id)
        future.result() # Blocks until publish is complete for immediate feedback/error
        
        processed_event['status'] = 'published'
        
    except Exception as e:
        processed_event['status'] = 'failed'
        processed_event['errors'].append(str(e))
        logger.error(f"Error processing event {event_id}: {e}", exc_info=True)

    return processed_event

# Required for fromisoformat with Z for UTC.
try:
    import pytz # Will be available via requirements.txt
except ImportError:
    # Fallback for local testing without pytz
    pytz = None
    logger.warning("pytz not found. Timestamps might not be handled with full timezone accuracy for non-UTC local times.")

file-processor-function/requirements.txt:

google-cloud-storage
google-cloud-pubsub
google-cloud-bigquery
pytz

Deploy the Cloud Function:

gcloud functions deploy offline-file-processor \\\
    --runtime python311 \\\
    --trigger-bucket YOUR_OFFLINE_DATA_LANDING_ZONE \\\
    --entry-point process_file_from_gcs \\\
    --region YOUR_GCP_REGION \\\
    --set-env-vars GCP_PROJECT_ID=YOUR_GCP_PROJECT_ID,PUBSUB_OFFLINE_EVENTS_TOPIC=offline-events-topic,BQ_AUDIT_DATASET=offline_events_audit,BQ_AUDIT_TABLE=processed_offline_events \\\
    --memory 512MB \\\
    --timeout 300s # 5 minutes for large files

IAM Permissions for Cloud Function: The Cloud Function's service account (e.g., [email protected] or custom) needs:

roles/storage.objectViewer on gs://your-offline-data-landing-zone.
roles/pubsub.publisher on projects/YOUR_GCP_PROJECT_ID/topics/offline-events-topic.
roles/bigquery.dataEditor on projects/YOUR_GCP_PROJECT_ID/datasets/offline_events_audit.

4. BigQuery Audit Table for Offline Events

Create the BigQuery table to log processed/failed offline events, providing an audit trail.

CREATE TABLE `your_gcp_project.offline_events_audit.processed_offline_events` (
    event_id STRING NOT NULL,
    event_name STRING NOT NULL,
    user_id STRING,
    email_address_hashed_sha256 STRING,
    event_timestamp_ms INTEGER,
    event_timestamp_utc_iso TIMESTAMP,
    transaction_id STRING,
    value NUMERIC,
    currency STRING,
    items JSON, -- Store processed items as JSON if needed
    is_offline_event BOOLEAN,
    offline_source_file STRING,
    original_row_data JSON, -- Store the original raw row for deep debugging
    processing_timestamp TIMESTAMP,
    status STRING, -- e.g., 'published', 'failed'
    errors ARRAY<STRING>
)
PARTITION BY DATE(event_timestamp_utc_iso)
CLUSTER BY event_name, user_id, event_id
OPTIONS (
    description = 'Audit log of offline events processed by the Cloud Function before sending to GA4.'
);

5. Pub/Sub Topic for Offline Events

Create the Pub/Sub topic that the Cloud Function publishes to.

gcloud pubsub topics create offline-events-topic --project YOUR_GCP_PROJECT_ID

6. GA4 Offline Ingestion Service (Cloud Run)

This Cloud Run service subscribes to the offline-events-topic, reconstructs a GA4 Measurement Protocol hit, and sends it to GA4.

ga4-offline-ingestor/main.py:

import os
import json
import base64
import requests
import datetime
import uuid
import logging

app = Flask(__name__)
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# GA4 Measurement Protocol Configuration (use Secret Manager in production!)
GA4_API_SECRET = os.environ.get('GA4_API_SECRET')
GA4_MEASUREMENT_ID = os.environ.get('GA4_MEASUREMENT_ID')
GA4_ENDPOINT = f"https://www.google-analytics.com/mp/collect?measurement_id={GA4_MEASUREMENT_ID}&api_secret={GA4_API_SECRET}"
DEBUG_GA4_ENDPOINT = f"https://www.google-analytics.com/debug/mp/collect?measurement_id={GA4_MEASUREMENT_ID}&api_secret={GA4_API_SECRET}"

@app.route('/ingest-offline-event', methods=['POST'])
def ingest_offline_event():
    """Receives Pub/Sub push messages, decodes, and sends to GA4 MP."""
    if not request.is_json:
        logger.warning("Ingestor: Request is not JSON. Content-Type: %s", request.headers.get('Content-Type'))
        return jsonify({'error': 'Request must be JSON'}), 400

    try:
        envelope = request.get_json()
        message = envelope['message']
        
        decoded_data = base64.b64decode(message['data']).decode('utf-8')
        processed_offline_event = json.loads(decoded_data)
        
        event_id = processed_offline_event.get('event_id')
        event_name = processed_offline_event.get('event_name')
        event_timestamp_ms = processed_offline_event.get('event_timestamp_ms')
        user_id = processed_offline_event.get('user_id')
        email_hashed = processed_offline_event.get('email_address_hashed_sha256')

        if not event_id or not event_name or not event_timestamp_ms:
            logger.error(f"Offline Event {event_id}: Missing critical fields. Skipping GA4 send.")
            return jsonify({'error': 'Missing critical fields'}), 400 # Pub/Sub will retry

        # --- Reconstruct GA4 Measurement Protocol Payload ---
        # Client ID for offline events: use a consistent placeholder or map to existing client_ids
        # If user_id is known, GA4 can stitch. Otherwise, a generic offline client_id or a unique one per user_id.
        # For simplicity, we can use user_id as client_id, or generate a deterministic client_id from user_id if needed.
        # Alternatively, for truly offline, set client_id as a UUID for unique user_pseudo_id in GA4
        
        # Here's a strategy:
        # If we have a user_id, GA4 can handle stitching. Client_id can be a consistent placeholder.
        # If no user_id, generate a new UUID for client_id for each event, resulting in a new user_pseudo_id in GA4.
        # A more advanced approach would involve a Firestore lookup to map user_id to an existing client_id.
        
        # For this example, we'll make a client_id that is unique per user_id, or truly unique if no user_id.
        # This will create distinct GA4 user_pseudo_ids for offline events unless mapped.
        ga4_client_id = f"offline-{user_id}" if user_id else str(uuid.uuid4())
        
        ga4_event_params = {
            "_eid": event_id, # Crucial for deduplication in GA4
            "is_offline_event": True, # Custom parameter to identify offline events
            "event_origin_offline_timestamp_ms": event_timestamp_ms, # Original timestamp in ms
            "session_id": f"offline_{event_timestamp_ms}", # A consistent session ID for offline events
            "engagement_time_msec": 1, # Minimal for offline events
            # Add other GA4-specific parameters from processed_offline_event
            "transaction_id": processed_offline_event.get('transaction_id'), # For purchase deduplication
            "value": processed_offline_event.get('value'),
            "currency": processed_offline_event.get('currency'),
            # User properties (if any, e.g., email_hashed)
            "user_email_hashed": email_hashed,
            "user_id_from_offline": user_id # Also pass user_id as an event param
        }
        
        # Handle items array for e-commerce events
        if processed_offline_event.get('items') and isinstance(processed_offline_event['items'], list):
            ga4_event_params['items'] = []
            for item in processed_offline_event['items']:
                ga4_event_params['items'].append({
                    "item_id": item.get("item_id"),
                    "item_name": item.get("item_name"), # Enrich this later if needed
                    "price": float(item.get("price", 0)),
                    "quantity": int(item.get("quantity", 1))
                })

        ga4_payload = {
            "client_id": ga4_client_id,
            "user_id": user_id, # Send user_id for GA4's native stitching
            "timestamp_micros": str(event_timestamp_ms * 1000), # Measurement Protocol expects microseconds
            "events": [
                {
                    "name": event_name,
                    "params": ga4_event_params
                }
            ]
        }
        
        # --- Send to GA4 Measurement Protocol ---
        headers = {'Content-Type': 'application/json'}
        mp_endpoint = GA4_ENDPOINT # Use production endpoint
        # mp_endpoint = DEBUG_GA4_ENDPOINT # Use debug endpoint for initial testing

        logger.info(f"Ingestor: Sending offline event '{event_name}' (ID: {event_id}, User: {user_id}) to GA4 MP...")
        
        response = requests.post(mp_endpoint, headers=headers, data=json.dumps(ga4_payload), timeout=10)
        response.raise_for_status() # Raise HTTPError for bad responses (4xx or 5xx)

        logger.info(f"Ingestor: Successfully ingested offline event '{event_name}' (ID: {event_id}) to GA4 MP. Status: {response.status_code}")
        if "debug" in mp_endpoint:
            logger.info(f"GA4 Debug Response: {response.json()}")
        
        return jsonify({'status': 'acknowledged', 'event_id': event_id}), 200

    except requests.exceptions.Timeout:
        logger.error(f"Ingestor: Timeout when sending offline event {event_id} to GA4 MP.", exc_info=True)
        return jsonify({'error': 'GA4 MP send timeout'}), 500 # Pub/Sub will retry
    except requests.exceptions.RequestException as req_e:\
        logger.error(f"Ingestor: Error sending offline event {event_id} to GA4 MP: {req_e}. Response: {req_e.response.text if req_e.response else 'N/A'}", exc_info=True)
        return jsonify({'error': f"GA4 MP send failed: {str(req_e)}"}), 500 # Pub/Sub will retry
    except json.JSONDecodeError as json_e:
        logger.error(f"Ingestor: JSON decoding error for event {event_id}: {json_e}. Raw data: {decoded_data}", exc_info=True)
        return jsonify({'error': 'JSON processing error'}), 500 # Pub/Sub will retry
    except Exception as e:
        logger.error(f"Ingestor: Unexpected error processing event {event_id}: {e}", exc_info=True)
        return jsonify({'error': str(e)}), 500 # Pub/Sub will retry

ga4-offline-ingestor/requirements.txt:

Flask
google-cloud-pubsub
requests

Deploy the Cloud Run Service:

gcloud run deploy ga4-offline-ingestion-service \\\
    --source ./ga4-offline-ingestor \\\
    --platform managed \\\
    --region YOUR_GCP_REGION \\\
    --no-allow-unauthenticated \\\
    --set-env-vars \\\
        GA4_API_SECRET="YOUR_GA4_MP_API_SECRET",\\\
        GA4_MEASUREMENT_ID="G-YOUR_GA4_MEASUREMENT_ID",\\\
        GCP_PROJECT_ID="YOUR_GCP_PROJECT_ID" \\\
    --memory 512Mi \\\
    --cpu 1 \\\
    --timeout 60s # Allow ample time for processing and API calls

IAM Permissions for Cloud Run Service:

The Cloud Run service account needs roles/logging.logWriter.
The Pub/Sub service account ([email protected]) needs roles/run.invoker on this Cloud Run service to push messages.
Security: GA4_API_SECRET should ideally be retrieved from Google Secret Manager at runtime, not stored directly as an environment variable in production. If so, the Cloud Run service account also needs roles/secretmanager.secretAccessor.

Create a Pub/Sub Push Subscription:

gcloud pubsub subscriptions create ga4-offline-ingestion-sub \\\
    --topic offline-events-topic \\\
    --push-endpoint=https://ga4-offline-ingestion-service-YOUR_SERVICE_HASH-YOUR_GCP_REGION.a.run.app/ingest-offline-event \\\
    --ack-deadline=30s \\\
    --message-retention-duration=7d \\\
    --min-duration-per-ack=10s \\\
    --max-duration-per-ack=600s \\\
    --expiration-period=never \\\
    --project YOUR_GCP_PROJECT_ID

7. GA4 Configuration

For these offline events to be useful in GA4:

Custom Dimensions/Metrics: Register any custom parameters (is_offline_event, user_email_hashed, user_id_from_offline) as Event-scoped Custom Dimensions in GA4 to enable segmentation and analysis.
User-ID: By sending the user_id parameter, GA4 will attempt to stitch these offline events with existing online data for the same user_id, creating a unified view.
Deduplication: The _eid parameter (populated with event_id) is crucial for GA4's native deduplication. Ensure each offline event has a globally unique event_id.
Timestamp Accuracy: The timestamp_micros parameter in the Measurement Protocol ensures historical events are placed correctly in your GA4 reports.

Benefits of This Offline Ingestion Pipeline

Unified Customer 360: Seamlessly integrate offline data with your online analytics for a complete view of customer behavior.
Accurate Attribution: Correctly attribute online campaigns by linking to confirmed offline conversions.
Richer Personalization: Drive more effective personalization strategies by incorporating offline customer attributes and interactions.
Improved Data Quality: Apply robust server-side transformations, validation, and PII handling to offline data before it reaches GA4.
Scalable & Resilient: Leverage serverless GCP components to handle any volume of batch data reliably, with built-in retry mechanisms.
Enhanced Auditability: Maintain a detailed audit trail of all processed offline events in BigQuery.
Compliance: Control PII processing and data retention for offline events in a governed environment.

Important Considerations

PII Handling: The pipeline processes raw PII (like email_address) from files. Ensure PII_FIELDS_TO_HASH is comprehensive in your Cloud Function, or integrate Google Cloud DLP for advanced redaction as an additional step. Define strict IAM roles for access to files containing PII.
User ID / Client ID Stitching: While sending user_id helps GA4 stitch, if your offline data only has a user_id and not an existing client_id (from a web session), GA4 will create a new user_pseudo_id for these offline events. For truly deterministic stitching of user_id to client_id across historical events, you might need a separate lookup service (e.g., using Firestore or BigQuery for identity mapping, as discussed in Unlocking Full User Journeys).
Timestamp Accuracy: The event_timestamp in your source files must be accurate. Ensure the Cloud Function correctly parses and converts it to UTC milliseconds for timestamp_micros.
Deduplication: Use a robust offline_event_id from your source system, or generate a UUID in the Cloud Function. GA4's _eid parameter will handle deduplication within GA4 for the replayed events.
Data Volume: For extremely large batch files (e.g., millions of rows), consider replacing the single Cloud Function with a Dataflow pipeline for more efficient and parallelized processing.
Error Handling: Implement dead-letter queues for Pub/Sub subscriptions to catch messages that repeatedly fail processing, allowing for manual inspection and reprocessing.
Cost: Monitor Cloud Storage, Cloud Function, Pub/Sub, and Cloud Run costs, as they scale with data volume. Optimize batch sizes and processing logic.

Conclusion

Bridging the gap between your online and offline data is paramount for a complete and accurate understanding of your customer journeys. By building a robust, serverless ingestion pipeline with Google Cloud Storage, Cloud Functions, Pub/Sub, and Cloud Run, you can seamlessly bring invaluable batch offline data into GA4. This strategic capability not only unifies your customer view and enhances attribution but also leverages your existing server-side data engineering investments for superior data quality and compliance across all customer touchpoints. Embrace this approach to unlock the full potential of your analytics and drive truly data-driven business decisions.