// MongoDB shell — insert & query documents
// A flexible event document for an e-commerce app

db.events.insertOne({
  user_id: "u_1234",
  event_type: "add_to_cart",
  timestamp: new Date(),
  product: { id: "p_99", name: "Sneakers", price: 3499 },
  metadata: { device: "android", app_version: "2.4.1" }
});

// Simple query: recent add-to-cart events
db.events.find({
  event_type: "add_to_cart",
  timestamp: { $gte: new Date(Date.now() - 24 * 60 * 60 * 1000) }
}).limit(10);

// Filter inside nested objects
db.events.find({ "product.id": "p_99" });

// MongoDB — indexes for query performance
// Single-field index
db.events.createIndex({ timestamp: -1 });

// Compound index for common query patterns
db.events.createIndex({ event_type: 1, timestamp: -1 });

// Index inside nested objects
db.events.createIndex({ "product.id": 1 });

// TTL index to auto-expire old events (30 days)
db.events.createIndex(
  { timestamp: 1 },
  { expireAfterSeconds: 60 * 60 * 24 * 30 }
);

// Check index usage
db.events.find(
  { event_type: "add_to_cart" }
).explain("executionStats");

// MongoDB aggregation — daily revenue by category
// A classic pre-aggregation for analytics dashboards

db.orders.aggregate([
  // Stage 1 — filter last 30 days
  { $match: {
      status: "completed",
      created_at: { $gte: new Date(Date.now() - 30*24*60*60*1000) }
  }},

  // Stage 2 — unwind items array
  { $unwind: "$items" },

  // Stage 3 — group by day and category
  { $group: {
      _id: {
        day: { $dateToString: { format: "%Y-%m-%d", date: "$created_at" } },
        category: "$items.category"
      },
      revenue: { $sum: { $multiply: ["$items.price", "$items.qty"] } },
      orders:  { $sum: 1 }
  }},

  // Stage 4 — sort newest first
  { $sort: { "_id.day": -1, revenue: -1 } },

  // Stage 5 — reshape for the dashboard
  { $project: {
      _id: 0,
      day: "$_id.day",
      category: "$_id.category",
      revenue: 1,
      orders: 1
  }}
]);

# Python — CDC with MongoDB change streams
from pymongo import MongoClient
from datetime import datetime

client = MongoClient("mongodb://localhost:27017")
db = client["shop"]

# Watch inserts/updates on orders
with db.orders.watch(
    [{"$match": {"operationType": {"$in": ["insert", "update"]}}}]
) as stream:

    for change in stream:
        op  = change["operationType"]
        doc = change["fullDocument"]

        print(f"[{datetime.utcnow()}] {op} order {doc['_id']}")

        # Push to a downstream system — e.g. Kafka, warehouse,
        # or a pre-aggregation collection
        # kafka_producer.send("orders", doc)

# Change streams use the oplog — no polling, low overhead,
# and it works across replica set members.

// E-commerce catalog — embed what you read together
{
  "_id": "p_99",
  "name": "Running Shoes",
  "brand": "Acme",
  "category": "Footwear",
  "variants": [
    { "sku": "p_99-bk-9", "color": "black", "size": 9,
      "price": 3499, "stock": 14 },
    { "sku": "p_99-bk-10", "color": "black", "size": 10,
      "price": 3499, "stock": 8  },
    { "sku": "p_99-wh-9", "color": "white", "size": 9,
      "price": 3299, "stock": 22 }
  ],
  "images": ["img1.jpg", "img2.jpg"],
  "rating": { "avg": 4.4, "count": 312 },
  "updated_at": ISODate("2026-08-28T10:00:00Z")
}

// One document = one page view.
// No joins. No ORM gymnastics. Fast reads.

// CDC to warehouse — pipeline sketch
// 1) Application writes to MongoDB
// 2) Change stream watches inserts/updates
// 3) Data engineer writes a small service that:
//    - normalises the document
//    - appends a schema_version field
//    - writes to a warehouse (BigQuery, Snowflake, Redshift)
//    - stores the last resume token for recovery

// Order matters:
//   resume_token → checkpoint to a state store
//   batch writes → warehouse
//   metrics → monitoring

// This is a standard, interview-tested data engineering pattern.
// You are bridging operational (MongoDB) and analytical (warehouse) systems.