Cloud Vehicle — Phase 2 Reference

-- models/gold/fct_orders.sql
-- Gold is the final consumer-facing layer.
-- One row per order. Denormalized for query performance.
-- Optimized for the access pattern: filter by date + region + status.

{{
  config(
    materialized          = 'incremental',
    unique_key            = 'order_id',
    partition_by          = { 'field': 'order_date', 'data_type': 'date', 'granularity': 'day' },
    cluster_by            = ['region', 'order_status', 'customer_tier'],
    incremental_strategy  = 'merge',       -- BQ native MERGE: upsert on unique_key
    on_schema_change      = 'sync_all_columns',
    tags                  = ['gold', 'orders', 'daily'],
  )
}}

WITH

orders AS (
    SELECT * FROM {{ ref('stg_orders') }}
    {% if is_incremental() %}
    -- Only process Silver rows newer than the last Gold run.
    -- _loaded_at is the Bronze ingestion timestamp — always set.
    WHERE _loaded_at > (
        SELECT MAX(_loaded_at) FROM {{ this }}
    )
    {% endif %}
),

customers AS (
    -- Snapshot of customer attributes at query time.
    -- For historical accuracy, use a Type 2 SCD dim_customers table here.
    SELECT
        customer_id,
        customer_sk,
        email      AS customer_email,
        region,
        customer_tier,
        is_active
    FROM {{ ref('stg_customers') }}
),

-- ── Core join ─────────────────────────────────────────────────────────
joined AS (
    SELECT
        o.order_id,
        o.customer_id,
        c.customer_sk,
        o.order_date,
        o.order_total_usd,
        o.order_status,
        o.is_large_order,
        c.customer_email,
        COALESCE(c.region, 'UNKNOWN')        AS region,
        COALESCE(c.customer_tier, 0)         AS customer_tier,
        c.is_active                          AS customer_is_active,
        o._loaded_at,
        o._row_hash
    FROM orders o
    LEFT JOIN customers c USING (customer_id)
    -- LEFT JOIN preserves orders even if customer record is missing.
    -- The orphan test in Silver warns us; Gold doesn't silently drop orders.
),

-- ── Window aggregations added at Gold ─────────────────────────────────
-- These are analytical facts (not definitional) — they belong in Gold.
-- Running totals, lag comparisons, and cohort metrics live here.
enriched AS (
    SELECT
        *,

        -- Customer's running order count at the time of this order
        COUNT(order_id) OVER (
            PARTITION BY customer_id
            ORDER BY order_date
            ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
        )                                    AS customer_order_sequence,

        -- Revenue contribution vs same customer's all-time avg
        ROUND(
            order_total_usd /
            NULLIF(AVG(order_total_usd) OVER (PARTITION BY customer_id), 0),
        2)                                   AS order_vs_customer_avg,

        -- Days since this customer's previous order
        DATE_DIFF(
            order_date,
            LAG(order_date) OVER (
                PARTITION BY customer_id
                ORDER BY order_date
            ),
            DAY
        )                                    AS days_since_last_order

    FROM joined
)

SELECT * FROM enriched

-- ── EXPLAIN plan basics ────────────────────────────────────────────────
-- Run in BQ console or via bq CLI:
--   bq query --use_legacy_sql=false 'EXPLAIN SELECT ...'
--
-- Key stages to look for in the execution plan:

-- STAGE 1: Input  → reads and filters partitions
-- STAGE 2: Join   → hash join or broadcast join
-- STAGE 3: Agg    → GROUP BY / window functions
-- STAGE 4: Output → writes result

-- What you're looking for:
--   "estimated bytes processed": if this is >> what you expected,
--   your partition filter isn't being pushed down.
--
--   "shuffle": data moved between workers. High shuffle = potential skew.
--
--   "max worker time >> avg worker time": DATA SKEW.
--   One worker is processing most of the data.

-- ── Detect partition filter push-down ─────────────────────────────────
-- Good: only 1 partition scanned
SELECT COUNT(*) FROM `project.gold.fct_orders`
WHERE order_date = '2025-04-02'       -- partition filter: BQ reads 1 day

-- Bad: full table scan (no date filter) — NEVER let this reach production
-- SELECT COUNT(*) FROM `project.gold.fct_orders`
-- WHERE order_status = 'COMPLETED'   -- not the partition key: reads ALL

-- ── Detect and fix data skew ───────────────────────────────────────────
-- Symptom: GROUP BY country takes 45min; 1 worker handles 90% of rows.
-- Diagnosis: check the distribution of the GROUP BY key.

SELECT
    region,
    COUNT(*) AS row_count,
    ROUND(100.0 * COUNT(*) / SUM(COUNT(*)) OVER (), 2) AS pct
FROM `project.gold.fct_orders`
GROUP BY region
ORDER BY row_count DESC

-- If one value has >50% of rows, you have skew on that join/group key.

-- Fix 1: Pre-aggregate the heavy key separately, then JOIN
WITH se_orders AS (
    SELECT customer_id, SUM(order_total_usd) AS regional_total
    FROM `project.gold.fct_orders`
    WHERE region = 'Southeast'        -- handle the fat partition alone
    GROUP BY customer_id
),
other_orders AS (
    SELECT customer_id, SUM(order_total_usd) AS regional_total
    FROM `project.gold.fct_orders`
    WHERE region != 'Southeast'
    GROUP BY customer_id
)
SELECT * FROM se_orders
UNION ALL
SELECT * FROM other_orders

-- Fix 2: Salt the join key to spread a hot key across workers
-- (When skew is in a JOIN, not a GROUP BY)
SELECT
    FLOOR(RAND() * 10) AS salt,     -- add random 0-9 suffix
    order_id,
    customer_id
FROM `project.gold.fct_orders`
WHERE region = 'Southeast'

-- ── BigQuery INFORMATION_SCHEMA: operational intelligence ─────────────
-- How much did each job cost? Which queries are the most expensive?

SELECT
    job_id,
    user_email,
    query,
    total_bytes_processed / POW(1024, 4)    AS tb_scanned,
    total_bytes_processed / POW(1024, 4) * 6.25 AS estimated_cost_usd,
    total_slot_ms / 1000                    AS slot_seconds,
    creation_time
FROM `region-us`.INFORMATION_SCHEMA.JOBS_BY_PROJECT
WHERE creation_time >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 24 HOUR)
  AND statement_type = 'SELECT'
ORDER BY total_bytes_processed DESC
LIMIT 20

"""
bq_cost_guard.py — Dry-run any BQ query before execution.

BigQuery dry-run returns bytes_processed without running the query.
Use this in CI/CD: fail the pipeline if a query would scan > threshold.
Use interactively: sanity-check a new query before the analyst runs it.
"""
from google.cloud import bigquery
from google.cloud.bigquery import QueryJobConfig

client = bigquery.Client()
COST_PER_TB_USD = 6.25


def dry_run(sql: str, max_gb: float = 100.0) -> dict:
    """
    Returns bytes estimate. Raises if query would exceed max_gb.
    Does NOT execute the query — zero cost.
    """
    config = QueryJobConfig(dry_run=True, use_query_cache=False)
    job = client.query(sql, job_config=config)

    bytes_est  = job.total_bytes_processed
    gb_est     = bytes_est / 1e9
    tb_est     = bytes_est / 1e12
    cost_est   = tb_est * COST_PER_TB_USD

    result = {
        "bytes":    bytes_est,
        "gb":       round(gb_est, 2),
        "tb":       round(tb_est, 4),
        "cost_usd": round(cost_est, 4),
    }

    if gb_est > max_gb:
        raise ValueError(
            f"Query would scan {gb_est:.1f} GB (limit: {max_gb} GB).\n"
            f"  Estimated cost: ${cost_est:.2f}\n"
            f"  Add a partition filter or review the query before running."
        )

    print(f"  [ok] Dry-run: {gb_est:.2f} GB | ${cost_est:.4f} | safe to run.")
    return result


def set_query_budget(project_id: str, monthly_limit_usd: float = 500.0):
    """
    Sets a project-level BigQuery quota that hard-caps monthly spend.
    Queries exceeding the budget return an error instead of charging.

    Run once per project setup — not on every pipeline run.
    """
    # This requires the BQ Admin API — see:
    # https://cloud.google.com/bigquery/docs/custom-quotas
    print(f"Set via Cloud Console: IAM > Quotas > BigQuery > Query usage per day")
    print(f"Target: {monthly_limit_usd / 30:.0f} USD/day = {monthly_limit_usd:.0f} USD/month")
    print(f"Or via bq CLI: bq update --project_id={project_id} \\")
    print(f"  --set_label=billing_budget:{int(monthly_limit_usd)}")


# ── CI/CD integration example ─────────────────────────────────────────────
if __name__ == "__main__":
    test_query = """
        SELECT region, SUM(order_total_usd) AS revenue
        FROM `my-project.gold.fct_orders`
        WHERE order_date BETWEEN '2025-01-01' AND '2025-03-31'
        GROUP BY region
    """
    stats = dry_run(test_query, max_gb=50)
    print(f"Safe to run: {stats}")

# iam.tf — Least-privilege IAM
# Rule: grant the minimum role that allows the task. Nothing more.
# Every binding here has a one-line comment explaining why it exists.

# ── Service Accounts ─────────────────────────────────────────────────────
resource "google_service_account" "agent_runner" {
  account_id   = "${local.prefix}-agent"
  display_name = "Data Agent Runner"
  description  = "Used by GKE Pods via Workload Identity. Reads BQ, writes GCS."
}

resource "google_service_account" "dbt_runner" {
  account_id   = "${local.prefix}-dbt"
  display_name = "dbt Transform Runner"
  description  = "Runs dbt models in CI/CD. BQ job runner + dataset writer."
}

resource "google_service_account" "ingest" {
  account_id   = "${local.prefix}-ingest"
  display_name = "Ingest Service"
  description  = "Writes raw data to Bronze GCS bucket only."
}

# ── Agent runner: read BQ Gold, read/write Silver GCS ────────────────────
resource "google_bigquery_dataset_iam_member" "agent_gold_reader" {
  dataset_id = google_bigquery_dataset.gold.dataset_id
  role       = "roles/bigquery.dataViewer"       # read-only: no DDL, no writes
  member     = "serviceAccount:${google_service_account.agent_runner.email}"
}

resource "google_storage_bucket_iam_member" "agent_silver_reader" {
  bucket = google_storage_bucket.silver.name
  role   = "roles/storage.objectViewer"          # read Silver Parquet files
  member = "serviceAccount:${google_service_account.agent_runner.email}"
}

# ── dbt runner: needs to create tables and run jobs in BQ ────────────────
# roles/bigquery.dataEditor: create/update/delete tables (not drop datasets)
# roles/bigquery.jobUser:    run query jobs (required to execute any SQL)
# These two together = dbt can do its job and nothing else.
resource "google_bigquery_dataset_iam_member" "dbt_silver_editor" {
  dataset_id = google_bigquery_dataset.silver.dataset_id
  role       = "roles/bigquery.dataEditor"
  member     = "serviceAccount:${google_service_account.dbt_runner.email}"
}

resource "google_bigquery_dataset_iam_member" "dbt_gold_editor" {
  dataset_id = google_bigquery_dataset.gold.dataset_id
  role       = "roles/bigquery.dataEditor"
  member     = "serviceAccount:${google_service_account.dbt_runner.email}"
}

resource "google_project_iam_member" "dbt_job_user" {
  project = var.project_id
  role    = "roles/bigquery.jobUser"             # run BQ jobs
  member  = "serviceAccount:${google_service_account.dbt_runner.email}"
}

# ── Ingest service: ONLY write to Bronze bucket ───────────────────────────
resource "google_storage_bucket_iam_member" "ingest_bronze_writer" {
  bucket = google_storage_bucket.bronze.name
  role   = "roles/storage.objectCreator"         # create only — cannot delete/overwrite
  member = "serviceAccount:${google_service_account.ingest.email}"
}
# Ingest service has NO BigQuery access — by design.
# If Bronze ingest is compromised, attacker cannot read/modify BQ data.

# ── Workload Identity: bind GKE KSA to GSA ───────────────────────────────
resource "google_service_account_iam_binding" "agent_workload_identity" {
  service_account_id = google_service_account.agent_runner.name
  role               = "roles/iam.workloadIdentityUser"
  members = [
    "serviceAccount:${var.project_id}.svc.id.goog[agents/agent-sa]"
  ]
}

# variables.tf — all input parameters with descriptions and validation
# This file is the "interface" to your Terraform module.
# Anyone deploying this stack can read variables.tf and know
# exactly what they need to provide — no reading the main.tf required.

variable "project_id" {
  type        = string
  description = "GCP project ID. Must already exist."
  validation {
    condition     = length(var.project_id) > 0
    error_message = "project_id cannot be empty."
  }
}

variable "region" {
  type        = string
  description = "Primary GCP region. All resources deploy here."
  default     = "us-central1"
  validation {
    condition     = contains(["us-central1", "us-east1", "europe-west1", "europe-west4", "asia-northeast1"], var.region)
    error_message = "Must be a supported region. Check your data residency requirements."
  }
}

variable "environment" {
  type        = string
  description = "Deployment environment. Controls resource naming and deletion protection."
  validation {
    condition     = contains(["dev", "staging", "prod"], var.environment)
    error_message = "environment must be dev, staging, or prod."
  }
}

variable "subnet_cidr" {
  type        = string
  description = "Primary subnet CIDR. Must not overlap with on-prem or other VPCs."
  default     = "10.0.0.0/20"
}

variable "pods_cidr" {
  type        = string
  description = "GKE Pod IP range. Needs to accommodate max_pods_per_node * max_nodes."
  default     = "10.1.0.0/16"   # 65,536 pod IPs — safe for most deployments
}

variable "services_cidr" {
  type        = string
  description = "GKE Service IP range."
  default     = "10.2.0.0/20"   # 4,096 service IPs
}

variable "authorized_cidr" {
  type        = string
  description = "CIDR block allowed to reach the GKE API server. Your VPN or bastion range."
  default     = "10.0.0.0/8"
}

---
# terraform.tfvars — actual values (NOT committed to git for prod)
# Use separate .tfvars files per environment:
#   dev.tfvars, staging.tfvars, prod.tfvars
# In CI/CD: terraform apply -var-file=prod.tfvars

project_id      = "my-client-project-prod"
region          = "us-central1"
environment     = "prod"
subnet_cidr     = "10.10.0.0/20"
pods_cidr       = "10.20.0.0/16"
services_cidr   = "10.30.0.0/20"
authorized_cidr = "10.0.0.0/8"

# backend.tf — Remote state configuration
# Run this BEFORE anything else in a new project:
#   1. Create the GCS bucket manually (or via bootstrap script)
#   2. Add this file
#   3. terraform init   (migrates local state to GCS)
#
# Why remote state?
#   - Local state breaks team workflows (who has the latest state.tfstate?)
#   - GCS provides versioning + automatic locking
#   - State is encrypted at rest by default in GCS
#
# WARNING: the state file contains secrets. Enable:
#   - Uniform bucket-level access (no public ACLs)
#   - Customer-managed encryption (CMEK) for regulated clients
#   - Object versioning (recover from accidental corruption)

terraform {
  backend "gcs" {
    bucket  = "my-project-terraform-state"   # created manually before init
    prefix  = "envs/prod"                    # separate folder per workspace
    # GCS provides object-level locking natively — no separate lock table needed
    # (unlike S3, which requires DynamoDB for locking)
  }
}

# ── Bootstrap script: create the state bucket ────────────────────────────
# Run this once, manually, before the first terraform init.
# The state bucket itself is NOT managed by Terraform (chicken-and-egg).
#
# gcloud storage buckets create gs://my-project-terraform-state \
#   --location=us-central1 \
#   --uniform-bucket-level-access \
#   --public-access-prevention
#
# gcloud storage buckets update gs://my-project-terraform-state \
#   --versioning
#
# gcloud storage buckets update gs://my-project-terraform-state \
#   --default-encryption-key=projects/my-project/locations/us-central1/keyRings/main/cryptoKeys/terraform

# ── Team workflow ─────────────────────────────────────────────────────────
# Developer A:
#   terraform plan   → acquires lock, generates plan, releases lock
#   terraform apply  → acquires lock, applies, releases lock
#
# Developer B (while A is applying):
#   terraform apply  → BLOCKED with "state is locked by process X"
#   → This is correct behavior. Wait for A to finish.
#
# Emergency unlock (if a process died holding the lock):
#   terraform force-unlock LOCK_ID
#   → Only use this if you're certain no other process is running.

"""
pubsub/publisher.py — Production Pub/Sub publisher.

Key patterns:
  - OrderingKey: guarantees in-order delivery per customer
  - Batching: reduce API calls for high-volume publishing
  - Future callbacks: non-blocking publish with error handling
  - Exponential backoff: retry transient failures
"""
import json
import logging
from datetime import datetime, timezone
from concurrent.futures import TimeoutError
from google.cloud import pubsub_v1
from google.api_core.exceptions import GoogleAPIError

logger = logging.getLogger(__name__)

PROJECT_ID = "my-project"
TOPIC_ID   = "prod-orders"


def get_publisher() -> pubsub_v1.PublisherClient:
    """
    Publisher with batching enabled.
    Pub/Sub batches messages before sending — reduces API calls
    and increases throughput at the cost of a small latency increase.
    """
    batch_settings = pubsub_v1.types.BatchSettings(
        max_messages  = 100,      # send when 100 messages queued
        max_bytes     = 1_000_000, # or when batch hits 1MB
        max_latency   = 0.01,     # or after 10ms — whichever first
    )
    return pubsub_v1.PublisherClient(batch_settings=batch_settings)


def publish_order_event(
    order: dict,
    event_type: str = "ORDER_CREATED",
) -> str:
    """
    Publish a single order event with ordering key.

    OrderingKey = customer_id:
      All events for the same customer arrive at the subscriber
      in the order they were published. Essential for:
        ORDER_CREATED → ORDER_SHIPPED → ORDER_DELIVERED
      Without ordering: subscriber might process DELIVERED before CREATED.

    Returns the message ID on success.
    """
    publisher  = get_publisher()
    topic_path = publisher.topic_path(PROJECT_ID, TOPIC_ID)

    # Envelope pattern: wrap payload with metadata
    # The subscriber uses event_type to route to the right handler.
    message = {
        "event_type":    event_type,
        "event_version": "1.0",
        "event_time":    datetime.now(timezone.utc).isoformat(),
        "source":        "order-service",
        "payload":       order,
    }

    future = publisher.publish(
        topic_path,
        data          = json.dumps(message).encode("utf-8"),
        ordering_key  = f"customer-{order['customer_id']}",  # per-customer ordering
        event_type    = event_type,          # message attributes (filterable)
        environment   = "prod",
    )

    try:
        message_id = future.result(timeout=30)
        logger.info(f"Published {event_type} for order {order['order_id']}: {message_id}")
        return message_id
    except TimeoutError:
        logger.error(f"Publish timed out for order {order['order_id']}")
        raise
    except GoogleAPIError as e:
        logger.error(f"Pub/Sub API error: {e}")
        raise


def publish_batch(orders: list[dict], event_type: str = "ORDER_CREATED") -> list:
    """
    Publish a batch of orders non-blocking.
    All futures are started, then we wait for results.
    Much faster than sequential publish for bulk ingestion.
    """
    publisher  = get_publisher()
    topic_path = publisher.topic_path(PROJECT_ID, TOPIC_ID)
    futures    = []

    for order in orders:
        message = json.dumps({
            "event_type": event_type,
            "event_time": datetime.now(timezone.utc).isoformat(),
            "payload":    order,
        }).encode("utf-8")

        future = publisher.publish(
            topic_path,
            data         = message,
            ordering_key = f"customer-{order['customer_id']}",
        )
        futures.append((order["order_id"], future))

    # Collect results
    results, errors = [], []
    for order_id, future in futures:
        try:
            msg_id = future.result(timeout=60)
            results.append({"order_id": order_id, "message_id": msg_id})
        except Exception as e:
            errors.append({"order_id": order_id, "error": str(e)})
            logger.error(f"Failed to publish order {order_id}: {e}")

    if errors:
        logger.warning(f"{len(errors)} publish failures — check DLQ")
    logger.info(f"Published {len(results)}/{len(orders)} messages")
    return results

"""
pubsub/subscriber.py — Pull subscriber with correct ACK semantics.

The most common Pub/Sub mistake: ACKing before processing.
If you ACK and then your processing crashes, the message is gone.
If you NACK or let the deadline expire, Pub/Sub redelivers it.
ACK only after you have confirmed the work is done and durable.
"""
import json
import logging
from concurrent.futures import TimeoutError
from google.cloud import pubsub_v1

logger = logging.getLogger(__name__)

PROJECT_ID       = "my-project"
SUBSCRIPTION_ID  = "prod-orders-agent"


def process_order_event(message: pubsub_v1.subscriber.message.Message) -> None:
    """
    Callback invoked for each message.

    ACK/NACK contract:
      message.ack()   — processing succeeded. Pub/Sub stops delivering.
      message.nack()  — processing failed. Pub/Sub redelivers after backoff.
      (no response)   — ack_deadline expires, Pub/Sub redelivers.

    Never ACK before your work is committed to a durable store.
    The typical sequence:
      1. Parse and validate the message
      2. Write to DuckDB / BigQuery / GCS   ← durable commit
      3. message.ack()                      ← only now
    """
    try:
        # ── Parse ────────────────────────────────────────────────────────
        envelope = json.loads(message.data.decode("utf-8"))
        event_type = envelope.get("event_type", "UNKNOWN")
        payload    = envelope.get("payload", {})
        order_id   = payload.get("order_id")

        logger.info(f"Processing {event_type} for order {order_id} "
                    f"(delivery #{message.delivery_attempt})")

        # ── Route by event type ──────────────────────────────────────────
        if event_type == "ORDER_CREATED":
            handle_order_created(payload)
        elif event_type == "ORDER_SHIPPED":
            handle_order_shipped(payload)
        elif event_type == "ORDER_CANCELLED":
            handle_order_cancelled(payload)
        else:
            # Unknown event type: ACK to prevent infinite redelivery,
            # but log for investigation.
            logger.warning(f"Unknown event_type '{event_type}' — ACKing to drain")
            message.ack()
            return

        # ── ACK only after successful processing ────────────────────────
        message.ack()
        logger.info(f"ACKed order {order_id}")

    except json.JSONDecodeError as e:
        # Malformed JSON: NACK once to let retry policy try again,
        # but after max_delivery_attempts it goes to DLQ automatically.
        logger.error(f"Malformed message payload: {e}")
        message.nack()

    except Exception as e:
        # Unexpected error: NACK so Pub/Sub redelivers with backoff.
        # After max_delivery_attempts (5), message goes to DLQ.
        logger.error(f"Processing failed for order {order_id}: {e}", exc_info=True)
        message.nack()


def handle_order_created(payload: dict) -> None:
    """Write new order to Bronze Parquet / streaming BQ insert."""
    # Implementation: call ingest.py or BQ streaming insert
    logger.info(f"  → ORDER_CREATED: {payload.get('order_id')}")


def handle_order_shipped(payload: dict) -> None:
    """Update order status in Silver (via BQ MERGE or DuckDB UPDATE)."""
    logger.info(f"  → ORDER_SHIPPED: {payload.get('order_id')}")


def handle_order_cancelled(payload: dict) -> None:
    """Mark order cancelled. Trigger refund event if needed."""
    logger.info(f"  → ORDER_CANCELLED: {payload.get('order_id')}")


def start_subscriber(timeout: float = None) -> None:
    """Start the pull subscriber. Blocks until timeout or KeyboardInterrupt."""
    subscriber = pubsub_v1.SubscriberClient()
    sub_path   = subscriber.subscription_path(PROJECT_ID, SUBSCRIPTION_ID)

    # Flow control: limit in-flight messages to avoid OOM
    flow_control = pubsub_v1.types.FlowControl(
        max_messages       = 50,       # process at most 50 concurrent messages
        max_bytes          = 50_000_000,  # 50MB in flight
    )

    streaming_pull_future = subscriber.subscribe(
        sub_path,
        callback     = process_order_event,
        flow_control = flow_control,
    )

    logger.info(f"Listening for messages on {sub_path}")
    try:
        streaming_pull_future.result(timeout=timeout)
    except TimeoutError:
        streaming_pull_future.cancel()
        streaming_pull_future.result()
    except KeyboardInterrupt:
        streaming_pull_future.cancel()

"""
pubsub/dead_letter.py — Dead Letter Queue monitor.

Messages land here after max_delivery_attempts (5) failed.
This process:
  1. Reads from the DLQ subscription
  2. Categorizes the failure (malformed JSON, missing fields, etc.)
  3. Writes to Bronze quarantine folder for investigation
  4. Sends a Slack/PagerDuty alert
  5. ACKs from DLQ (removes from queue)

Run as a separate Cloud Run service on a slow poll (every 5 minutes).
Not time-critical — DLQ is for investigation, not real-time processing.
"""
import json
import logging
from datetime import datetime, timezone
from pathlib import Path
from google.cloud import pubsub_v1, storage

logger = logging.getLogger(__name__)

PROJECT_ID       = "my-project"
DLQ_SUB_ID       = "prod-orders-dlq-monitor"
QUARANTINE_BUCKET = "prod-bronze"
QUARANTINE_PREFIX = "quarantine/orders"


def categorize_failure(message_data: bytes) -> str:
    """Determine why the message likely failed."""
    try:
        payload = json.loads(message_data.decode("utf-8"))
        if "event_type" not in payload:
            return "missing_event_type"
        if "payload" not in payload:
            return "missing_payload"
        if "order_id" not in payload.get("payload", {}):
            return "missing_order_id"
        return "processing_error"   # parsed fine but processor crashed
    except (json.JSONDecodeError, UnicodeDecodeError):
        return "malformed_json"


def quarantine_message(
    message: pubsub_v1.subscriber.message.Message,
    failure_reason: str,
) -> str:
    """Write failed message to Bronze quarantine for investigation."""
    client = storage.Client()
    bucket = client.bucket(QUARANTINE_BUCKET)

    timestamp = datetime.now(timezone.utc).strftime("%Y/%m/%d/%H")
    blob_name = (
        f"{QUARANTINE_PREFIX}/{timestamp}/"
        f"{failure_reason}_{message.message_id}.json"
    )

    quarantine_record = {
        "message_id":        message.message_id,
        "failure_reason":    failure_reason,
        "delivery_attempts": message.delivery_attempt,
        "publish_time":      message.publish_time.isoformat() if message.publish_time else None,
        "quarantined_at":    datetime.now(timezone.utc).isoformat(),
        "attributes":        dict(message.attributes),
        "data_base64":       message.data.hex(),  # store as hex for safety
        "data_text":         message.data.decode("utf-8", errors="replace"),
    }

    blob = bucket.blob(blob_name)
    blob.upload_from_string(
        json.dumps(quarantine_record, indent=2),
        content_type="application/json",
    )

    logger.warning(
        f"Quarantined message {message.message_id} "
        f"(reason: {failure_reason}) → gs://{QUARANTINE_BUCKET}/{blob_name}"
    )
    return f"gs://{QUARANTINE_BUCKET}/{blob_name}"


def send_alert(failure_reason: str, message_id: str, quarantine_path: str) -> None:
    """Send alert to Slack / PagerDuty. Replace with your alerting integration."""
    alert = {
        "severity":       "warning" if failure_reason != "malformed_json" else "error",
        "message":        f"DLQ message quarantined: {failure_reason}",
        "message_id":     message_id,
        "quarantine_path": quarantine_path,
        "action_required": "Inspect quarantine file and determine if replay is needed",
    }
    logger.error(f"ALERT: {json.dumps(alert)}")
    # TODO: post to Slack webhook / PagerDuty API


def process_dlq(max_messages: int = 100) -> dict:
    """Pull and process DLQ messages."""
    subscriber = pubsub_v1.SubscriberClient()
    sub_path   = subscriber.subscription_path(PROJECT_ID, DLQ_SUB_ID)

    response = subscriber.pull(
        request={"subscription": sub_path, "max_messages": max_messages}
    )

    stats = {"processed": 0, "quarantined": 0, "categories": {}}

    for received_message in response.received_messages:
        msg = received_message.message

        failure_reason = categorize_failure(msg.data)
        quarantine_path = quarantine_message(msg, failure_reason)
        send_alert(failure_reason, msg.message_id, quarantine_path)

        # ACK from DLQ — message is now in quarantine (durable)
        subscriber.acknowledge(
            request={"subscription": sub_path, "ack_ids": [received_message.ack_id]}
        )

        stats["processed"] += 1
        stats["quarantined"] += 1
        stats["categories"][failure_reason] = stats["categories"].get(failure_reason, 0) + 1

    if stats["processed"] > 0:
        logger.warning(f"DLQ processed: {json.dumps(stats)}")
    return stats


if __name__ == "__main__":
    result = process_dlq()
    print(f"DLQ run complete: {result}")

# ingress_egress.tf
# ─────────────────────────────────────────────────────────────────────────
# Ingress rule: "Who outside the perimeter can call services inside it?"
# Egress rule:  "Who inside the perimeter can call services outside it?"
#
# Common rules you'll need:
#   Ingress: CI/CD service account (dbt runner from GitHub Actions)
#   Ingress: Looker / BI tool IP calling BigQuery
#   Egress:  Vertex AI calling GCS (model artifacts)
#   Egress:  Cloud Run calling external APIs
# ─────────────────────────────────────────────────────────────────────────

# ── Ingress: CI/CD dbt runner can call BigQuery from outside perimeter ────
# GitHub Actions runs dbt from outside your VPC. Without this rule,
# dbt's BQ calls get 403 PERMISSION_DENIED even with correct IAM.
resource "google_access_context_manager_service_perimeter" "main" {
  # ... (extends the resource from vpc_sc.tf — shown here inline for clarity)

  status {
    ingress_policies {
      ingress_from {
        # Allow from specific service accounts (not IP-based)
        identities = [
          "serviceAccount:${google_service_account.dbt_runner.email}",
        ]
        # From any source (GitHub-hosted runners have variable IPs)
        sources {
          access_level = "*"   # any access level = any source
        }
      }
      ingress_to {
        resources = ["projects/${var.project_number}"]
        operations {
          service_name = "bigquery.googleapis.com"
          method_selectors {
            method = "google.cloud.bigquery.v2.JobService.InsertJob"
          }
          method_selectors {
            method = "google.cloud.bigquery.v2.TableService.InsertTable"
          }
        }
      }
    }

    # ── Ingress: Looker / BI tool by IP ─────────────────────────────────
    ingress_policies {
      ingress_from {
        sources {
          # Looker has fixed outbound IPs — allow them by CIDR via access level
          access_level = google_access_context_manager_access_level.bi_tool.name
        }
        identities = ["serviceAccount:${var.looker_service_account}"]
      }
      ingress_to {
        resources = ["projects/${var.project_number}"]
        operations {
          service_name = "bigquery.googleapis.com"
          method_selectors { method = "*" }   # all BQ methods
        }
      }
    }

    # ── Egress: Vertex AI can read/write GCS for model artifacts ─────────
    egress_policies {
      egress_from {
        identities = [
          "serviceAccount:service-${var.project_number}@gcp-sa-aiplatform.iam.gserviceaccount.com",
        ]
      }
      egress_to {
        resources = ["projects/${var.project_number}"]   # same project = same perimeter, but explicit
        operations {
          service_name = "storage.googleapis.com"
          method_selectors { method = "google.storage.objects.get" }
          method_selectors { method = "google.storage.objects.create" }
        }
      }
    }
  }
}

-- ── Query dry-run VPC SC violations in Cloud Logging ─────────────────────
-- Run in the GCP Logs Explorer after deploying in dry-run mode.
-- This shows every API call that WOULD be blocked by the perimeter.
-- Add ingress/egress rules for all legitimate traffic before enforcing.
--
-- Log filter (Logs Explorer syntax):
protoPayload.methodName="google.identity.accesscontextmanager.v1.AccessContextManager.UpdateServicePerimeter"
resource.type="audited_resource"
severity="WARNING"

-- ── BigQuery query against log exports (if using Log Sink to BQ) ─────────
SELECT
  timestamp,
  protoPayload.authenticationInfo.principalEmail AS caller,
  protoPayload.requestMetadata.callerIp           AS caller_ip,
  protoPayload.methodName                         AS api_method,
  protoPayload.resourceName                       AS resource,
  protoPayload.status.code                        AS status_code,
  protoPayload.status.message                     AS status_message,
  -- VPC SC specific fields
  JSON_VALUE(protoPayload.metadata, '$.violationReason') AS violation_reason,
  JSON_VALUE(protoPayload.metadata, '$.accessLevels')    AS access_levels
FROM `my-project.log_exports.cloudaudit_googleapis_com_data_access`
WHERE
  timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 48 HOUR)
  AND JSON_VALUE(protoPayload.metadata, '$.dryRun') = 'true'  -- dry-run only
ORDER BY timestamp DESC

-- ── Categorize violations for perimeter tuning ────────────────────────────
SELECT
  protoPayload.authenticationInfo.principalEmail AS principal,
  protoPayload.methodName                         AS api_method,
  COUNT(*) AS violation_count,
  -- Determine what ingress/egress rule is needed
  CASE
    WHEN protoPayload.methodName LIKE 'google.cloud.bigquery%'
    THEN 'Add BigQuery ingress rule for this principal'
    WHEN protoPayload.methodName LIKE 'google.storage%'
    THEN 'Add GCS ingress or egress rule'
    WHEN protoPayload.methodName LIKE 'google.cloud.aiplatform%'
    THEN 'Add Vertex AI ingress rule'
    ELSE 'Investigate: ' || protoPayload.methodName
  END AS remediation_action
FROM `my-project.log_exports.cloudaudit_googleapis_com_data_access`
WHERE
  timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 72 HOUR)
  AND JSON_VALUE(protoPayload.metadata, '$.dryRun') = 'true'
GROUP BY principal, api_method
ORDER BY violation_count DESC

-- ── Check: are there any ENFORCED violations (after going live)? ──────────
-- This query should return zero rows in a healthy deployment.
-- Any rows = legitimate traffic being blocked = add an ingress/egress rule.
SELECT
  timestamp,
  protoPayload.authenticationInfo.principalEmail AS blocked_caller,
  protoPayload.methodName                         AS blocked_method,
  protoPayload.resourceName                       AS blocked_resource
FROM `my-project.log_exports.cloudaudit_googleapis_com_data_access`
WHERE
  timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 HOUR)
  AND protoPayload.status.code = 403
  AND JSON_VALUE(protoPayload.metadata, '$.dryRun') IS NULL   -- enforced, not dry-run
ORDER BY timestamp DESC