[Producers: Booking Web App / Kiosk Sync / Offline Flight Batch]
                              │
                              ▼
        ┌──────────────────────────────────────────────┐
        │  Cloud Storage Landing Bucket (Raw Lake)     │
        │  gs://offvia-raw-lake/bookings/year=.../     │
        └──────────────────────┬───────────────────────┘
                               │
            ┌──────────────────┴──────────────────┐
            ▼                                     ▼
   ┌───────────────────────┐             ┌───────────────────────────┐
   │ BigLake Connection    │             │ Managed Table Ingestion   │
   │ (Secure Staging GCS)  │             │ PARTITION BY DATE()       │
   │ offvia_staging        │             │ CLUSTER BY carrier, route │
   └───────────────────────┘             │ offvia_warehouse          │
                                         └─────────────┬─────────────js
                                                       │
                 ┌─────────────────────────────────────┴─────────────────────────────────────┐
                 ▼                                                                           ▼
   ┌───────────────────────────────┐                                           ┌───────────────────────────────┐
   │ Materialized View             │                                           │ Authorized View               │
   │ (Transparent Query Rewrite)   │                                           │ (PII Masking & Delegation)    │
   │ offvia_analytics              │                                           │ offvia_compliance             │
   └──────────────┬────────────────┘                                           └──────────────┬────────────────┘
                  │                                                                           │
                  ▼                                                                           ▼
     [Live Operational Dashboards]                                                [Aviation Regulatory Auditors]
     (Sub-second response, 0 slot storms)                                         (Aggregates only, zero PII)

# 1. Esporta le variabili di ambiente del deployment
export PROJECT_ID="offvia-prod-data"
export REGION="europe-west1"
export BUCKET_NAME="offvia-raw-lake-${PROJECT_ID}"
export CONNECTION_ID="gcs-secure-conn"

# Imposta il progetto attivo
gcloud config set project ${PROJECT_ID}

# 2. Abilita le API Google Cloud richieste
gcloud services enable \
    storage.googleapis.com \
    bigquery.googleapis.com \
    bigqueryconnection.googleapis.com

# 3. Crea i 4 dataset architetturali in BigQuery
bq --location=${REGION} mk -d \
    --description "Staging and raw external table bindings" \
    ${PROJECT_ID}:offvia_staging

bq --location=${REGION} mk -d \
    --description "Core immutable partitioned & clustered warehouse fact tables" \
    ${PROJECT_ID}:offvia_warehouse

bq --location=${REGION} mk -d \
    --description "Precomputed materialized views and operational analytics" \
    ${PROJECT_ID}:offvia_analytics

bq --location=${REGION} mk -d \
    --description "Governed authorized views for external auditors and compliance" \
    ${PROJECT_ID}:offvia_compliance

gcloud storage buckets create gs://${BUCKET_NAME} \
    --project=${PROJECT_ID} \
    --location=${REGION} \
    --uniform-bucket-level-access

python3 -m venv venv
source venv/bin/activate
pip install pandas pyarrow

import datetime
import os
import random
import uuid
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq

# Seed per riproducibilità
random.seed(42)

CARRIERS = ["LH", "BA", "AF", "FR", "U2"]
ROUTES = ["FRA_LHR", "CDG_BER", "AMS_MAD", "FCO_BCN", "MUC_CDG", "LHR_DUB"]
SEAT_CLASSES = ["ECONOMY", "ECONOMY", "ECONOMY", "PREMIUM_ECONOMY", "BUSINESS"]
STATUSES = ["CONFIRMED", "CONFIRMED", "CONFIRMED", "CANCELLED", "MODIFIED"]

def generate_flight_records(date_str: str, num_records: int = 15000):
    base_date = datetime.datetime.strptime(date_str, "%Y-%m-%d")
    records = []
    for _ in range(num_records):
        minute_offset = random.randint(0, 1439)
        second_offset = random.randint(0, 59)
        event_time = base_date + datetime.timedelta(minutes=minute_offset, seconds=second_offset)
        # Flight fares in Euro cents (e.g. 149.50 EUR = 14950 cents)
        fare_cents = random.randint(4500, 75000)
        carrier = random.choice(CARRIERS)
        route = random.choice(ROUTES)
        booking_id = str(uuid.uuid4())
        passenger_id = f"PAX_{random.randint(10000, 99999)}"
        passenger_email = f"user_{random.randint(100, 999)}@example-travel.com"
        records.append({
            "booking_id": booking_id,
            "flight_id": f"{carrier}-{random.randint(100, 999)}",
            "carrier_code": carrier,
            "route_id": route,
            "passenger_id": passenger_id,
            "passenger_email": passenger_email,
            "event_time": event_time,
            "fare_amount_cents": fare_cents,
            "currency": "EUR",
            "seat_class": random.choice(SEAT_CLASSES),
            "status": random.choice(STATUSES),
        })
    return pd.DataFrame(records)

# Genera 3 giorni di dati di prenotazione
dates = ["2026-09-28", "2026-09-29", "2026-09-30"]
for d in dates:
    df_day = generate_flight_records(d, num_records=15000)
    year, month, day = d.split("-")
    # Percorso partizione Hive standard
    dir_path = f"raw_data/year={year}/month={month}/day={day}"
    os.makedirs(dir_path, exist_ok=True)
    file_path = os.path.join(dir_path, "bookings_batch_001.parquet")
    table = pa.Table.from_pandas(df_day)
    pq.write_table(table, file_path, compression="SNAPPY")
    print(f"Generati {len(df_day)} record in {file_path}")

python3 generate_bookings.py

# Sincronizza nel bucket Cloud Storage con upload parallelo
gcloud storage rsync -r raw_data/ gs://${BUCKET_NAME}/bookings/

gcloud storage ls --recursive gs://${BUCKET_NAME}/bookings/

gs://offvia-raw-lake-offvia-prod-data/bookings/year=2026/month=09/day=28/bookings_batch_001.parquet
gs://offvia-raw-lake-offvia-prod-data/bookings/year=2026/month=09/day=29/bookings_batch_001.parquet
gs://offvia-raw-lake-offvia-prod-data/bookings/year=2026/month=09/day=30/bookings_batch_001.parquet

bq mk --connection \
    --location=${REGION} \
    --connection_type=CLOUD_RESOURCE \
    ${CONNECTION_ID}

# Recupera e concedi i permessi del service account della connessione sul bucket
CONNECTION_SA=$(bq show --connection ${PROJECT_ID}.${REGION}.${CONNECTION_ID} | grep 'serviceAccountId' | awk -F'"' '{print $4}')

gcloud storage buckets add-iam-policy-binding gs://${BUCKET_NAME} \
    --member="serviceAccount:${CONNECTION_SA}" \
    --role="roles/storage.objectViewer"

CREATE OR REPLACE EXTERNAL TABLE offvia_staging.ext_bookings
WITH CONNECTION `offvia-prod-data.europe-west1.gcs-secure-conn`
OPTIONS (
  format = 'PARQUET',
  uris = ['gs://offvia-raw-lake-offvia-prod-data/bookings/*'],
  hive_partition_uri_prefix = 'gs://offvia-raw-lake-offvia-prod-data/bookings/'
);

bq query --use_legacy_sql=false "
CREATE OR REPLACE EXTERNAL TABLE offvia_staging.ext_bookings
WITH CONNECTION \`${PROJECT_ID}.${REGION}.${CONNECTION_ID}\`
OPTIONS (
  format = 'PARQUET',
  uris = ['gs://${BUCKET_NAME}/bookings/*'],
  hive_partition_uri_prefix = 'gs://${BUCKET_NAME}/bookings/'
);"

SELECT 
  carrier_code,
  COUNT(1) AS booking_count,
  ROUND(SUM(fare_amount_cents) / 100.0, 2) AS total_revenue_eur
FROM offvia_staging.ext_bookings
WHERE year = 2026 AND month = 9 AND day = 30
GROUP BY carrier_code
ORDER BY total_revenue_eur DESC;

CREATE OR REPLACE TABLE offvia_warehouse.fct_bookings (
  booking_id STRING NOT NULL OPTIONS(description="Unique UUIDv4 identifier for the booking"),
  flight_id STRING NOT NULL OPTIONS(description="IATA flight number, e.g. LH-402"),
  carrier_code STRING NOT NULL OPTIONS(description="Operating airline code: LH, BA, AF, FR, U2"),
  route_id STRING NOT NULL OPTIONS(description="Route origin and destination, e.g. FRA_LHR"),
  passenger_id STRING NOT NULL OPTIONS(description="Pseudonymized passenger identity token"),
  passenger_email STRING NOT NULL OPTIONS(description="Customer email address (Restricted PII)"),
  event_time TIMESTAMP NOT NULL OPTIONS(description="UTC timestamp when the booking was transacted"),
  ingestion_time TIMESTAMP NOT NULL OPTIONS(description="UTC timestamp when the record landed in the warehouse"),
  fare_amount_cents INT64 NOT NULL OPTIONS(description="Total fare paid in Euro cents"),
  currency STRING(3) NOT NULL OPTIONS(description="ISO-4217 3-letter currency code"),
  seat_class STRING NOT NULL OPTIONS(description="Cabin class: ECONOMY, PREMIUM_ECONOMY, BUSINESS"),
  status STRING NOT NULL OPTIONS(description="Status: CONFIRMED, MODIFIED, CANCELLED")
)
PARTITION BY DATE(event_time)
CLUSTER BY carrier_code, route_id, status
OPTIONS (
  require_partition_filter = true,
  partition_expiration_days = 1095, -- 3 anni di retention automatica
  description = "Production flight booking fact table for Offvia travel platform"
);

bq query --use_legacy_sql=false "
CREATE OR REPLACE TABLE ${PROJECT_ID}:offvia_warehouse.fct_bookings (
  booking_id STRING NOT NULL,
  flight_id STRING NOT NULL,
  carrier_code STRING NOT NULL,
  route_id STRING NOT NULL,
  passenger_id STRING NOT NULL,
  passenger_email STRING NOT NULL,
  event_time TIMESTAMP NOT NULL,
  ingestion_time TIMESTAMP NOT NULL,
  fare_amount_cents INT64 NOT NULL,
  currency STRING NOT NULL,
  seat_class STRING NOT NULL,
  status STRING NOT NULL
)
PARTITION BY DATE(event_time)
CLUSTER BY carrier_code, route_id, status
OPTIONS (
  require_partition_filter = true,
  partition_expiration_days = 1095
);"

bq query --use_legacy_sql=false "
INSERT INTO ${PROJECT_ID}:offvia_warehouse.fct_bookings
SELECT 
  booking_id,
  flight_id,
  carrier_code,
  route_id,
  passenger_id,
  passenger_email,
  event_time,
  CURRENT_TIMESTAMP() AS ingestion_time,
  fare_amount_cents,
  currency,
  seat_class,
  status
FROM ${PROJECT_ID}:offvia_staging.ext_bookings;"

-- Questa query fallirà intenzionalmente
SELECT COUNT(1) FROM offvia_warehouse.fct_bookings WHERE carrier_code = 'LH';

Error: Cannot query over table 'offvia-prod-data.offvia_warehouse.fct_bookings' 
without a filter over column(s) 'event_time' that can be used for partition elimination.

SELECT 
  carrier_code,
  route_id,
  COUNT(1) AS flights_booked,
  ROUND(SUM(fare_amount_cents) / 100.0, 2) AS route_revenue_eur
FROM offvia_warehouse.fct_bookings
WHERE event_time >= TIMESTAMP('2026-09-30 00:00:00')
  AND event_time < TIMESTAMP('2026-10-01 00:00:00')
  AND carrier_code = 'LH'
GROUP BY carrier_code, route_id
ORDER BY route_revenue_eur DESC;

MERGE INTO offvia_warehouse.fct_bookings AS target
USING (
  SELECT 
    '00000000-0000-0000-0000-000000000999' AS booking_id,
    'LH-882' AS flight_id,
    'LH' AS carrier_code,
    'FRA_LHR' AS route_id,
    'PAX_99999' AS passenger_id,
    'vip_pax@example-travel.com' AS passenger_email,
    TIMESTAMP('2026-09-28 14:15:00') AS event_time,
    89000 AS fare_amount_cents,
    'EUR' AS currency,
    'FIRST' AS seat_class,
    'MODIFIED' AS status
) AS source
-- OTTIMIZZAZIONE CRITICA: Vincolare target.event_time forza BigQuery 
-- a scansionare SOLO la partizione 2026-09-28 invece dell'intera tabella!
ON target.event_time >= TIMESTAMP('2026-09-28 00:00:00')
   AND target.event_time < TIMESTAMP('2026-09-29 00:00:00')
   AND target.booking_id = source.booking_id
WHEN MATCHED THEN
  UPDATE SET
    status = source.status,
    seat_class = source.seat_class,
    fare_amount_cents = source.fare_amount_cents,
    ingestion_time = CURRENT_TIMESTAMP()
WHEN NOT MATCHED THEN
  INSERT (
    booking_id, flight_id, carrier_code, route_id, passenger_id, passenger_email,
    event_time, ingestion_time, fare_amount_cents, currency, seat_class, status
  )
  VALUES (
    source.booking_id, source.flight_id, source.carrier_code, source.route_id, source.passenger_id, source.passenger_email,
    source.event_time, CURRENT_TIMESTAMP(), source.fare_amount_cents, source.currency, source.seat_class, source.status
  );

CREATE MATERIALIZED VIEW offvia_analytics.mv_carrier_daily_summary
OPTIONS (
  enable_refresh = true,
  refresh_interval_minutes = 30
)
AS
SELECT 
  DATE(event_time) AS booking_date,
  carrier_code,
  route_id,
  seat_class,
  COUNT(1) AS total_bookings,
  SUM(fare_amount_cents) AS total_revenue_cents,
  AVG(fare_amount_cents) AS avg_fare_cents
FROM offvia_warehouse.fct_bookings
GROUP BY 1, 2, 3, 4;

SELECT 
  carrier_code,
  SUM(fare_amount_cents) / 100.0 AS total_revenue_eur
FROM offvia_warehouse.fct_bookings
WHERE event_time >= TIMESTAMP('2026-09-30 00:00:00')
  AND event_time < TIMESTAMP('2026-10-01 00:00:00')
GROUP BY carrier_code;

UPDATE ${PROJECT_ID}:offvia_warehouse.fct_bookings
SET status = 'CANCELLED'
WHERE DATE(event_time) = '2026-09-30';

-- 1. Crea uno snapshot di recupero temporaneo
CREATE OR REPLACE TABLE offvia_warehouse.fct_bookings_recovery_snapshot AS
SELECT * 
FROM offvia_warehouse.fct_bookings
FOR SYSTEM_TIME AS OF TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 5 MINUTE)
WHERE DATE(event_time) = '2026-09-30';

-- 2. Merge con scope di partizione di ritorno alla tabella principale
MERGE INTO offvia_warehouse.fct_bookings AS target
USING offvia_warehouse.fct_bookings_recovery_snapshot AS source
ON target.event_time >= TIMESTAMP('2026-09-30 00:00:00')
   AND target.event_time < TIMESTAMP('2026-10-01 00:00:00')
   AND target.booking_id = source.booking_id
WHEN MATCHED THEN
  UPDATE SET
    status = source.status,
    seat_class = source.seat_class,
    fare_amount_cents = source.fare_amount_cents;

-- 3. Pulizia snapshot
DROP TABLE offvia_warehouse.fct_bookings_recovery_snapshot;

CREATE OR REPLACE VIEW offvia_compliance.v_audited_flight_metrics
OPTIONS(
  description="Audited aggregated route capacity metrics for regulatory compliance. PII stripped."
)
AS
SELECT 
  DATE(event_time) AS flight_date,
  carrier_code,
  route_id,
  seat_class,
  COUNT(DISTINCT booking_id) AS total_passenger_count,
  ROUND(SUM(fare_amount_cents) / 100.0, 2) AS total_gross_fare_eur,
  ROUND(AVG(fare_amount_cents) / 100.0, 2) AS average_fare_eur
FROM offvia_warehouse.fct_bookings
WHERE status = 'CONFIRMED'
GROUP BY 1, 2, 3, 4;

bq update --dataset --add_view \
    ${PROJECT_ID}:offvia_compliance.v_audited_flight_metrics \
    ${PROJECT_ID}:offvia_warehouse