KStream: an unbounded sequence of events — each record is an independent fact
KTable:  a materialized view — each record is the latest value for a key (upsert semantics)

// KStream vs KTable mental model
StreamsBuilder builder = new StreamsBuilder();

// KStream: every message is an independent event
// orders topic: [ {orderId: 1, region: "EU", total: 99}, {orderId: 2, region: "US", total: 149}, ... ]
KStream<String, Order> orderStream = builder.stream("orders.created");
// All order events, including duplicates, cancellations, partial events

// KTable: latest value per key (compacted topic — only the latest value per key is kept)
// users topic: [ {userId: "u1", tier: "free"}, {userId: "u1", tier: "premium"}, ... ]
// KTable sees: { "u1": { tier: "premium" } }  — only the latest
KTable<String, User> userTable = builder.table("users",
    Consumed.with(Serdes.String(), userSerde));
// userTable.toStream() emits only when a key changes — changelog semantics

// ✅ Orders per minute per region — tumbling window aggregation
KStream<String, Order> orders = builder.stream("orders.created",
    Consumed.with(Serdes.String(), orderSerde));

KTable<Windowed<String>, Long> ordersPerMinutePerRegion = orders
    // Rekey by region — region becomes the aggregation key
    .selectKey((orderId, order) -> order.region())
    // Tumbling window: non-overlapping, fixed-duration buckets
    .groupByKey(Grouped.with(Serdes.String(), orderSerde))
    .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(1)))
    .count(Materialized.as("orders-per-minute-store"));

// Emit results to a topic — each window result is a Kafka record
ordersPerMinutePerRegion.toStream()
    .map((windowedKey, count) -> KeyValue.pair(
        windowedKey.key(),   // region
        new OrderCount(windowedKey.key(), windowedKey.window().start(), count)
    ))
    .to("orders.metrics.per-minute", Produced.with(Serdes.String(), orderCountSerde));

// Tumbling: fixed buckets, no overlap
.windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(1)))

// Hopping: 1-minute window, advancing every 30 seconds
.windowedBy(TimeWindows.of(Duration.ofMinutes(1)).advanceBy(Duration.ofSeconds(30)))

// Session: close window after 30 minutes of inactivity
.windowedBy(SessionWindows.ofInactivityGapWithNoGrace(Duration.ofMinutes(30)))

// ✅ Stream-table join: enrich every order event with current user tier
KStream<String, Order> orders = builder.stream("orders.created",
    Consumed.with(Serdes.String(), orderSerde));

KTable<String, User> users = builder.table("users",
    Consumed.with(Serdes.String(), userSerde));

// Rekey orders by userId to match the KTable key
KStream<String, Order> ordersByUser = orders
    .selectKey((orderId, order) -> order.customerId());

// Join: for each order, look up the user's current tier from the KTable
KStream<String, EnrichedOrder> enriched = ordersByUser.join(
    users,
    (order, user) -> new EnrichedOrder(order, user.tier(), user.loyaltyPoints()),
    // KTable join: always uses the CURRENT value of the KTable key
    // If user.tier changes mid-stream: new orders get the new tier, old orders are unaffected
    Joined.with(Serdes.String(), orderSerde, userSerde)
);

enriched.to("orders.enriched", Produced.with(Serdes.String(), enrichedOrderSerde));

// ✅ Interactive queries: expose KTable state as a queryable REST endpoint
// Kafka Streams makes the RocksDB state store queryable from within the application

ReadOnlyKeyValueStore<String, Long> orderCountStore = streams.store(
    StoreQueryParameters.fromNameAndType(
        "orders-per-minute-store",
        QueryableStoreTypes.keyValueStore()
    )
);

// Query the materialized store directly — no Kafka round-trip
long count = orderCountStore.get("EU");  // current EU order count from local RocksDB
// Returns in < 1ms from local store — zero network I/O

// ✅ Exactly-once within Kafka Streams topology
Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG,          "order-aggregator");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,       "kafka:9092");
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG,
    StreamsConfig.EXACTLY_ONCE_V2);  // requires Kafka 2.5+
// Effect: input consumption + state update + output production are atomic per task
// A process crash mid-processing: Kafka Streams replays from last committed checkpoint
// Result: each input record produces exactly one output record, state updated exactly once