Pub/Sub & Event-Driven Architecture

# Create a topic
gcloud pubsub topics create order-events \
  --message-retention-duration=7d \
  --labels=team=platform,env=production

# Create a topic with a schema (Avro or Protocol Buffers)
gcloud pubsub topics create user-events \
  --schema=projects/my-project/schemas/user-event-schema \
  --message-encoding=JSON

# Create a pull subscription
gcloud pubsub subscriptions create order-processing \
  --topic=order-events \
  --ack-deadline=60 \
  --message-retention-duration=7d \
  --expiration-period=never \
  --enable-exactly-once-delivery \
  --labels=team=orders,env=production

# Create a push subscription that delivers to Cloud Run
gcloud pubsub subscriptions create order-notifications \
  --topic=order-events \
  --push-endpoint=https://notification-service-abc123.run.app/events \
  --push-auth-service-account=pubsub-push@my-project.iam.gserviceaccount.com \
  --ack-deadline=30 \
  --max-delivery-attempts=5 \
  --dead-letter-topic=projects/my-project/topics/order-events-dlq \
  --min-retry-delay=10s \
  --max-retry-delay=600s

# Create a BigQuery subscription (writes directly to BigQuery)
gcloud pubsub subscriptions create order-analytics \
  --topic=order-events \
  --bigquery-table=my-project:analytics.order_events \
  --write-metadata \
  --use-topic-schema

# Publish a test message
gcloud pubsub topics publish order-events \
  --message='{"order_id":"ORD-001","status":"created","total":99.99}' \
  --attribute='event_type=order.created,source=api'

# Pull messages for testing
gcloud pubsub subscriptions pull order-processing --limit=10 --auto-ack

from google.cloud import pubsub_v1
from concurrent.futures import TimeoutError
import json
import time

project_id = "my-project"
subscription_id = "order-processing"

subscriber = pubsub_v1.SubscriberClient()
subscription_path = subscriber.subscription_path(project_id, subscription_id)

def process_message(message):
    """Process an individual message."""
    print(f"Received message ID: {message.message_id}")
    print(f"Data: {message.data.decode('utf-8')}")
    print(f"Attributes: {dict(message.attributes)}")
    print(f"Publish time: {message.publish_time}")
    print(f"Ordering key: {message.ordering_key}")

    try:
        data = json.loads(message.data.decode("utf-8"))

        # Process the order event
        if message.attributes.get("event_type") == "order.created":
            handle_new_order(data)
        elif message.attributes.get("event_type") == "order.paid":
            handle_payment_confirmation(data)

        # Acknowledge successful processing
        message.ack()
        print(f"Acknowledged message {message.message_id}")

    except json.JSONDecodeError:
        # Bad message format - ack to prevent infinite retry
        print(f"Invalid JSON in message {message.message_id}, acking to discard")
        message.ack()

    except TransientError:
        # Transient error - nack to retry
        print(f"Transient error, nacking message {message.message_id}")
        message.nack()

    except Exception as e:
        # Unexpected error - nack with a delay
        print(f"Unexpected error: {e}")
        message.modify_ack_deadline(30)  # Extend deadline before nack
        message.nack()

# Configure flow control to manage concurrency
flow_control = pubsub_v1.types.FlowControl(
    max_messages=100,          # Max outstanding messages
    max_bytes=10 * 1024 * 1024,  # Max outstanding bytes (10 MB)
    max_lease_duration=600,     # Max time to hold a message lease
)

# Start the subscriber with streaming pull
streaming_pull_future = subscriber.subscribe(
    subscription_path,
    callback=process_message,
    flow_control=flow_control,
)

print(f"Listening for messages on {subscription_path}...")

try:
    # Block the main thread; timeout=None for indefinite listening
    streaming_pull_future.result(timeout=None)
except TimeoutError:
    streaming_pull_future.cancel()
    streaming_pull_future.result()  # Wait for cleanup
except KeyboardInterrupt:
    streaming_pull_future.cancel()
    print("Subscriber shut down.")

# Cloud Run service that receives Pub/Sub push messages
from flask import Flask, request, jsonify
import base64
import json
import logging

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

@app.route("/events", methods=["POST"])
def handle_pubsub_message():
    """Handle incoming Pub/Sub push message."""
    envelope = request.get_json()
    if not envelope:
        return jsonify({"error": "No JSON payload"}), 400

    if "message" not in envelope:
        return jsonify({"error": "No message in envelope"}), 400

    pubsub_message = envelope["message"]

    # Decode the message data
    if "data" in pubsub_message:
        data = base64.b64decode(pubsub_message["data"]).decode("utf-8")
        payload = json.loads(data)
    else:
        payload = {}

    attributes = pubsub_message.get("attributes", {})
    message_id = pubsub_message.get("messageId", "unknown")

    logging.info(f"Processing message {message_id}: {payload}")

    try:
        # Process based on event type
        event_type = attributes.get("event_type")
        if event_type == "order.created":
            send_order_confirmation_email(payload)
        elif event_type == "order.shipped":
            send_shipping_notification(payload)
        else:
            logging.warning(f"Unknown event type: {event_type}")

        # Return 2xx to acknowledge the message
        return jsonify({"status": "processed", "messageId": message_id}), 200

    except Exception as e:
        logging.error(f"Error processing message {message_id}: {e}")
        # Return non-2xx to nack the message (Pub/Sub will retry)
        return jsonify({"error": str(e)}), 500

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8080)

# Create a subscription with exactly-once delivery enabled
gcloud pubsub subscriptions create payment-processing \
  --topic=payment-events \
  --enable-exactly-once-delivery \
  --ack-deadline=60 \
  --message-retention-duration=1d

# Check if exactly-once is enabled on an existing subscription
gcloud pubsub subscriptions describe payment-processing \
  --format='value(enableExactlyOnceDelivery)'

# Update an existing subscription to enable exactly-once
gcloud pubsub subscriptions update payment-processing \
  --enable-exactly-once-delivery

# Create the dead-letter topic
gcloud pubsub topics create order-events-dlq \
  --message-retention-duration=31d

# Create a subscription on the DLT for monitoring/replay
gcloud pubsub subscriptions create order-events-dlq-monitor \
  --topic=order-events-dlq \
  --ack-deadline=120 \
  --message-retention-duration=31d \
  --expiration-period=never

# Update the main subscription to use the dead-letter topic
gcloud pubsub subscriptions update order-processing \
  --dead-letter-topic=projects/my-project/topics/order-events-dlq \
  --max-delivery-attempts=10 \
  --min-retry-delay=10s \
  --max-retry-delay=600s

# Grant Pub/Sub permission to publish to the DLT
# The Pub/Sub service account needs pubsub.publisher on the DLT
PROJECT_NUMBER=$(gcloud projects describe my-project --format='value(projectNumber)')
gcloud pubsub topics add-iam-policy-binding order-events-dlq \
  --member="serviceAccount:service-${PROJECT_NUMBER}@gcp-sa-pubsub.iam.gserviceaccount.com" \
  --role="roles/pubsub.publisher"

# Grant Pub/Sub permission to ack messages on the source subscription
gcloud pubsub subscriptions add-iam-policy-binding order-processing \
  --member="serviceAccount:service-${PROJECT_NUMBER}@gcp-sa-pubsub.iam.gserviceaccount.com" \
  --role="roles/pubsub.subscriber"

# Create an alerting policy for DLT messages
gcloud beta monitoring policies create \
  --display-name="Dead Letter Queue Alert" \
  --condition-display-name="Messages in DLQ" \
  --condition-filter='resource.type="pubsub_subscription"
    AND resource.labels.subscription_id="order-events-dlq-monitor"
    AND metric.type="pubsub.googleapis.com/subscription/num_undelivered_messages"' \
  --condition-threshold-value=1 \
  --condition-threshold-comparison=COMPARISON_GT \
  --condition-threshold-duration=60s \
  --notification-channels=projects/my-project/notificationChannels/CHANNEL_ID

# Trigger Cloud Run when a file is uploaded to Cloud Storage
gcloud eventarc triggers create process-upload \
  --location=us-central1 \
  --destination-run-service=file-processor \
  --destination-run-region=us-central1 \
  --event-filters="type=google.cloud.storage.object.v1.finalized" \
  --event-filters="bucket=my-uploads-bucket" \
  --service-account=eventarc-sa@my-project.iam.gserviceaccount.com

# Trigger Cloud Run on BigQuery job completion via Audit Logs
gcloud eventarc triggers create bq-job-complete \
  --location=us-central1 \
  --destination-run-service=etl-notifier \
  --destination-run-region=us-central1 \
  --event-filters="type=google.cloud.audit.log.v1.written" \
  --event-filters="serviceName=bigquery.googleapis.com" \
  --event-filters="methodName=google.cloud.bigquery.v2.JobService.InsertJob" \
  --service-account=eventarc-sa@my-project.iam.gserviceaccount.com

# Trigger Cloud Functions (2nd gen) on Firestore document writes
gcloud eventarc triggers create user-change-handler \
  --location=us-central1 \
  --destination-function=on-user-change \
  --destination-function-region=us-central1 \
  --event-filters="type=google.cloud.firestore.document.v1.written" \
  --event-filters-path-pattern="database=(default)/documents/users/{userId}" \
  --service-account=eventarc-sa@my-project.iam.gserviceaccount.com

# List all triggers
gcloud eventarc triggers list --location=us-central1

# Describe a trigger
gcloud eventarc triggers describe process-upload --location=us-central1

from google.cloud import pubsub_v1

publisher = pubsub_v1.PublisherClient()
topic_path = publisher.topic_path("my-project", "order-events")

# Enable message ordering on the publisher
publisher_options = pubsub_v1.types.PublisherOptions(
    enable_message_ordering=True,
)
publisher = pubsub_v1.PublisherClient(publisher_options=publisher_options)

# Publish ordered messages for a specific order
order_id = "ORD-12345"
events = [
    {"event": "order.created", "data": {"total": 99.99}},
    {"event": "order.payment_received", "data": {"payment_id": "PAY-001"}},
    {"event": "order.shipped", "data": {"tracking": "TRACK-789"}},
    {"event": "order.delivered", "data": {"signed_by": "Jane Doe"}},
]

for event in events:
    import json
    data = json.dumps(event).encode("utf-8")
    future = publisher.publish(
        topic_path,
        data,
        ordering_key=order_id,  # Same key ensures in-order delivery
        event_type=event["event"],
    )
    print(f"Published {event['event']} with message ID: {future.result()}")

# IMPORTANT: If a publish with an ordering key fails, all subsequent
# publishes with that key will fail. You must call resume_publish()
# to clear the error state:
try:
    future = publisher.publish(topic_path, b"data", ordering_key="key-1")
    future.result()
except Exception:
    publisher.resume_publish("my-project", "order-events", "key-1")

# Create an Avro schema
gcloud pubsub schemas create order-event-schema \
  --type=AVRO \
  --definition='{
    "type": "record",
    "name": "OrderEvent",
    "fields": [
      {"name": "order_id", "type": "string"},
      {"name": "event_type", "type": {"type": "enum", "name": "EventType",
        "symbols": ["CREATED", "PAID", "SHIPPED", "DELIVERED", "CANCELLED"]}},
      {"name": "timestamp", "type": {"type": "long", "logicalType": "timestamp-millis"}},
      {"name": "total_amount", "type": ["null", "double"], "default": null},
      {"name": "customer_id", "type": "string"},
      {"name": "metadata", "type": {"type": "map", "values": "string"}, "default": {}}
    ]
  }'

# Create a Protocol Buffers schema
gcloud pubsub schemas create user-event-schema \
  --type=PROTOCOL_BUFFER \
  --definition='syntax = "proto3";
message UserEvent {
  string user_id = 1;
  string event_type = 2;
  int64 timestamp = 3;
  string email = 4;
  map<string, string> properties = 5;
}'

# Attach a schema to a topic
gcloud pubsub topics create order-events-validated \
  --schema=order-event-schema \
  --message-encoding=JSON

# Validate a message against a schema (dry run)
gcloud pubsub schemas validate-message \
  --schema-name=order-event-schema \
  --message-encoding=JSON \
  --message='{"order_id":"ORD-001","event_type":"CREATED","timestamp":1700000000000,"customer_id":"CUST-001"}'

# List schema revisions
gcloud pubsub schemas list-revisions order-event-schema

# Commit a new schema revision
gcloud pubsub schemas commit order-event-schema \
  --type=AVRO \
  --definition='...'

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions, StandardOptions
from apache_beam.transforms.window import FixedWindows, SlidingWindows
from apache_beam.transforms.trigger import AfterWatermark, AccumulationMode
import json

class ParseAndValidate(beam.DoFn):
    """Parse JSON messages and validate required fields."""
    def process(self, element):
        try:
            data = json.loads(element.decode("utf-8"))
            if "order_id" not in data or "total" not in data:
                # Send invalid messages to dead-letter output
                yield beam.pvalue.TaggedOutput("dead_letter", element)
                return
            yield data
        except json.JSONDecodeError:
            yield beam.pvalue.TaggedOutput("dead_letter", element)

class EnrichWithCustomerData(beam.DoFn):
    """Enrich order events with customer data from Firestore."""
    def setup(self):
        from google.cloud import firestore
        self.db = firestore.Client()

    def process(self, element):
        customer_ref = self.db.collection("customers").document(
            element["customer_id"]
        )
        customer = customer_ref.get()
        if customer.exists:
            element["customer_name"] = customer.get("name")
            element["customer_tier"] = customer.get("tier")
        yield element

def run():
    options = PipelineOptions()
    options.view_as(StandardOptions).streaming = True

    with beam.Pipeline(options=options) as p:
        # Read from Pub/Sub
        messages = (
            p
            | "Read from Pub/Sub" >> beam.io.ReadFromPubSub(
                subscription="projects/my-project/subscriptions/order-analytics",
                with_attributes=True,
            )
            | "Extract Data" >> beam.Map(lambda msg: msg.data)
        )

        # Parse and validate
        parsed = messages | "Parse" >> beam.ParDo(
            ParseAndValidate()
        ).with_outputs("dead_letter", main="valid")

        # Process valid messages
        enriched = (
            parsed["valid"]
            | "Enrich" >> beam.ParDo(EnrichWithCustomerData())
        )

        # Window into 5-minute fixed windows for aggregation
        windowed = (
            enriched
            | "Window" >> beam.WindowInto(
                FixedWindows(300),  # 5-minute windows
                trigger=AfterWatermark(),
                accumulation_mode=AccumulationMode.DISCARDING,
            )
        )

        # Aggregate revenue by customer tier
        revenue_by_tier = (
            windowed
            | "Key by Tier" >> beam.Map(
                lambda x: (x.get("customer_tier", "unknown"), x["total"])
            )
            | "Sum Revenue" >> beam.CombinePerKey(sum)
            | "Format" >> beam.Map(
                lambda kv: {"tier": kv[0], "revenue": kv[1]}
            )
        )

        # Write enriched events to BigQuery
        enriched | "Write to BQ" >> beam.io.WriteToBigQuery(
            table="my-project:analytics.enriched_orders",
            schema="order_id:STRING,total:FLOAT,customer_id:STRING,"
                   "customer_name:STRING,customer_tier:STRING",
            write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND,
            create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED,
        )

        # Write dead letters to a separate topic
        parsed["dead_letter"] | "Dead Letter" >> beam.io.WriteToPubSub(
            topic="projects/my-project/topics/order-events-dlq"
        )

if __name__ == "__main__":
    run()

# Grant publisher access to a specific topic
gcloud pubsub topics add-iam-policy-binding order-events \
  --member="serviceAccount:order-api@my-project.iam.gserviceaccount.com" \
  --role="roles/pubsub.publisher"

# Grant subscriber access to a specific subscription
gcloud pubsub subscriptions add-iam-policy-binding order-processing \
  --member="serviceAccount:order-worker@my-project.iam.gserviceaccount.com" \
  --role="roles/pubsub.subscriber"

# Configure Customer-Managed Encryption Keys (CMEK)
gcloud pubsub topics create sensitive-events \
  --topic-encryption-key=projects/my-project/locations/us-central1/keyRings/pubsub-ring/cryptoKeys/pubsub-key

# Grant Pub/Sub access to the KMS key
gcloud kms keys add-iam-policy-binding pubsub-key \
  --location=us-central1 \
  --keyring=pubsub-ring \
  --member="serviceAccount:service-PROJECT_NUMBER@gcp-sa-pubsub.iam.gserviceaccount.com" \
  --role="roles/cloudkms.cryptoKeyEncrypterDecrypter"

# Restrict message storage to specific regions (data residency)
gcloud pubsub topics update order-events \
  --message-storage-policy-allowed-regions=us-central1,us-east1

# Terraform configuration for a complete Pub/Sub setup
resource "google_pubsub_topic" "order_events" {
  name = "order-events"

  message_retention_duration = "604800s" # 7 days

  schema_settings {
    schema   = google_pubsub_schema.order_event.id
    encoding = "JSON"
  }

  labels = {
    team        = "platform"
    environment = "production"
  }
}

resource "google_pubsub_schema" "order_event" {
  name       = "order-event-schema"
  type       = "AVRO"
  definition = file("schemas/order-event.avsc")
}

# Filtered subscription - only receives "order.shipped" events
resource "google_pubsub_subscription" "shipping_notifications" {
  name  = "shipping-notifications"
  topic = google_pubsub_topic.order_events.id

  filter = "attributes.event_type = "order.shipped""

  push_config {
    push_endpoint = google_cloud_run_service.shipping_notifier.status[0].url
    oidc_token {
      service_account_email = google_service_account.pubsub_push.email
      audience              = google_cloud_run_service.shipping_notifier.status[0].url
    }
  }

  retry_policy {
    minimum_backoff = "10s"
    maximum_backoff = "600s"
  }

  dead_letter_policy {
    dead_letter_topic     = google_pubsub_topic.order_events_dlq.id
    max_delivery_attempts = 10
  }

  expiration_policy {
    ttl = ""  # Never expires
  }
}

resource "google_pubsub_topic" "order_events_dlq" {
  name                       = "order-events-dlq"
  message_retention_duration = "2678400s" # 31 days
}

# BigQuery subscription for analytics
resource "google_pubsub_subscription" "order_analytics" {
  name  = "order-analytics-bq"
  topic = google_pubsub_topic.order_events.id

  bigquery_config {
    table            = "${var.project_id}:analytics.order_events"
    write_metadata   = true
    use_topic_schema = true
  }
}