Mini Project — Build a Cloud-Native Data Platform

db.createCollection("users", {
  validator: {
    $jsonSchema: {
      bsonType: "object",
      required: ["userId", "email", "createdAt"],
      properties: {
        userId:    { bsonType: "string" },
        email:     { bsonType: "string", pattern: "^.+@.+\\..+$" },
        createdAt: { bsonType: "date" },
        // Additional fields are allowed without declaration (flexible schema)
      }
    }
  },
  validationAction: "error"   // reject invalid docs, don't just warn
});

// Partial index: enforce email uniqueness only on verified accounts
// Unverified accounts (email pending confirmation) can share email temporarily
db.users.createIndex(
  { email: 1 },
  {
    unique: true,
    partialFilterExpression: { emailVerified: true },
    name: "unique_verified_email"
  }
);

// Insert a full user profile — required + optional fields together
db.users.insertOne({
  userId:           "usr_8821",
  email:            "alex@example.com",
  createdAt:        new Date(),
  emailVerified:    true,
  // Optional fields added freely — no schema change required
  streamingQuality: "1080p",
  paymentMethods:   [{ type: "card", last4: "4242", default: true }],
  addressBook:      [{ label: "home", city: "London", postcode: "EC1A" }],
  preferences:      { notifications: true, darkMode: false }
});

-- Primary table: query by session (access pattern 1)
CREATE TABLE shopstream.session_events (
    session_id  TEXT,
    occurred_at TIMESTAMP,
    event_id    UUID,
    user_id     TEXT,
    event_type  TEXT,     -- join | leave | chat | product_view | purchase
    payload     TEXT,     -- JSON blob — flexible per event type
    PRIMARY KEY (session_id, occurred_at, event_id)
) WITH CLUSTERING ORDER BY (occurred_at ASC)
  AND default_time_to_live = 7776000;   -- 90 days in seconds

-- Materialized view: query by user (access pattern 2)
-- Same data, different partition key — Cassandra maintains sync automatically
CREATE MATERIALIZED VIEW shopstream.user_events AS
    SELECT * FROM shopstream.session_events
    WHERE user_id IS NOT NULL
      AND session_id IS NOT NULL
      AND occurred_at IS NOT NULL
      AND event_id IS NOT NULL
    PRIMARY KEY (user_id, occurred_at, session_id, event_id)
    WITH CLUSTERING ORDER BY (occurred_at DESC);  -- newest first for user feed

-- Insert a high-volume event batch (no TTL override — uses table default)
INSERT INTO shopstream.session_events
    (session_id, occurred_at, event_id, user_id, event_type, payload)
    VALUES ('sess_live_882', toTimestamp(now()), uuid(),
            'usr_8821', 'product_view',
            '{"productId":"prod_001","dwellMs":4200}');

import boto3
from botocore.exceptions import ClientError
import uuid, datetime

dynamodb = boto3.client("dynamodb", region_name="us-east-1")

def place_order(user_id: str, product_id: str, amount_cents: int) -> dict:
    order_id = str(uuid.uuid4())

    try:
        dynamodb.transact_write_items(
            TransactItems=[

                # Operation 1: decrement inventory (fails if stock = 0)
                {
                    "Update": {
                        "TableName": "Inventory",
                        "Key": {"productId": {"S": product_id}},
                        "UpdateExpression": "SET stock = stock - :one",
                        "ConditionExpression": "stock > :zero",  # prevent oversell
                        "ExpressionAttributeValues": {
                            ":one":  {"N": "1"},
                            ":zero": {"N": "0"}
                        }
                    }
                },

                # Operation 2: create order record
                {
                    "Put": {
                        "TableName": "Orders",
                        "Item": {
                            "userId":    {"S": user_id},
                            "SK":        {"S": f"ORDER#{order_id}"},
                            "productId": {"S": product_id},
                            "amountCents": {"N": str(amount_cents)},
                            "status":    {"S": "PENDING"},
                            "createdAt": {"S": datetime.datetime.utcnow().isoformat()}
                        },
                        "ConditionExpression": "attribute_not_exists(SK)"  # idempotent
                    }
                },

                # Operation 3: write payment intent
                {
                    "Put": {
                        "TableName": "Payments",
                        "Item": {
                            "orderId":     {"S": order_id},
                            "SK":          {"S": "INTENT"},
                            "amountCents": {"N": str(amount_cents)},
                            "status":      {"S": "INITIATED"}
                        }
                    }
                }
            ]
        )
        return {"ok": True, "orderId": order_id}

    except ClientError as e:
        if e.response["Error"]["Code"] == "TransactionCanceledException":
            return {"ok": False, "reason": "Out of stock or duplicate order"}
        raise

import boto3, json, os
from pymongo import MongoClient

sqs     = boto3.client("sqs")
mongo   = MongoClient(os.environ["MONGO_URI"])
users   = mongo["shopstream"]["users"]

# Neptune/Neo4j client omitted for brevity — same pattern applies

def handler(event, context):
    for record in event["Records"]:

        # Only process new order inserts — ignore updates and deletes
        if record["eventName"] != "INSERT":
            continue

        new = record["dynamodb"]["NewImage"]
        order_id   = new["SK"]["S"].replace("ORDER#", "")
        user_id    = new["userId"]["S"]
        product_id = new["productId"]["S"]
        amount     = int(new["amountCents"]["N"])

        # 1. Update MongoDB user profile — $push is idempotent-safe with $ne
        users.update_one(
            {"userId": user_id},
            {"$addToSet": {"orderHistory": order_id}}  # addToSet = no duplicates
        )

        # 2. Publish fulfilment event to SQS
        # MessageDeduplicationId makes SQS FIFO queue idempotent on retry
        sqs.send_message(
            QueueUrl=os.environ["FULFILMENT_QUEUE"],
            MessageBody=json.dumps({
                "orderId":   order_id,
                "userId":    user_id,
                "productId": product_id,
                "amount":    amount
            }),
            MessageGroupId="fulfilment",
            MessageDeduplicationId=order_id   # SQS FIFO deduplication window: 5min
        )

        # 3. Neptune graph edge — MERGE prevents duplicate edges
        # (Neptune Gremlin query simplified)
        # g.V(user_id).addE('PURCHASED').to(g.V(product_id))
        #   with .property(id, order_id)  — idempotent via MERGE semantics