Introduction to Apache Cassandra

-- ── SimpleStrategy: development only, single datacenter ─────────────────────
-- Never use in production — it ignores rack and datacenter topology.
-- Replicas are placed on the next N nodes clockwise in the ring,
-- which means two replicas can land on the same physical rack.
CREATE KEYSPACE IF NOT EXISTS forge_dev
  WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1};

-- ── NetworkTopologyStrategy: production standard ─────────────────────────────
-- Rack-aware and datacenter-aware placement.
-- With RF=3 across 3 racks, you can lose an entire rack and still serve queries.
CREATE KEYSPACE IF NOT EXISTS forge_system
  WITH replication = {
    'class': 'NetworkTopologyStrategy',
    'datacenter1': 3
  };

USE forge_system;

-- ── Table design: audit_logs ─────────────────────────────────────────────────
-- Partition key: service_name  → determines which node(s) store this row
-- Clustering column: log_time  → determines sort order within the partition
-- CLUSTERING ORDER BY DESC: most recent logs are read first with no sort overhead
CREATE TABLE IF NOT EXISTS audit_logs (
    service_name text,
    log_time     timestamp,
    request_id   uuid,
    payload      text,
    PRIMARY KEY (service_name, log_time)
) WITH CLUSTERING ORDER BY (log_time DESC)
  AND default_time_to_live = 2592000;  -- 30-day TTL: logs auto-expire

-- ── Writing data: any node accepts the write ────────────────────────────────
INSERT INTO audit_logs (service_name, log_time, request_id, payload)
VALUES ('payment-service', toTimestamp(now()), uuid(), '{"status":"ok"}');

INSERT INTO audit_logs (service_name, log_time, request_id, payload)
VALUES ('payment-service', toTimestamp(now()), uuid(), '{"status":"retry"}');

-- ── Reading data: the query matches the partition key exactly ────────────────
-- This is a single-partition read — one node handles it, response in <5ms
SELECT service_name, log_time, payload
FROM audit_logs
WHERE service_name = 'payment-service'
  AND log_time > '2026-01-01 00:00:00+0000'
LIMIT 100;

-- ── Example 1: Simple partition key — one column ────────────────────────────
-- All events for a user are in one partition.
-- Problem: a user with 10 million events creates a 4GB partition.
CREATE TABLE user_events_v1 (
    user_id   uuid,
    event_time timestamp,
    event_type text,
    metadata  text,
    PRIMARY KEY (user_id, event_time)
) WITH CLUSTERING ORDER BY (event_time DESC);

-- ── Example 2: Composite partition key — time bucketing ──────────────────────
-- Now the partition key is (user_id, bucket).
-- Each day gets its own partition — bounded growth, predictable size.
-- 'bucket' is a date string like '2026-04-18', computed in the application.
-- Trade-off: queries spanning multiple days require multiple partition reads.
CREATE TABLE user_events_v2 (
    user_id    uuid,
    bucket     text,      -- application sets this: '2026-04-18'
    event_time timestamp,
    event_type text,
    metadata   text,
    PRIMARY KEY ((user_id, bucket), event_time)
--                ^-----------^  composite partition key: two columns together
) WITH CLUSTERING ORDER BY (event_time DESC);

-- ── Writing to the bucketed table ─────────────────────────────────────────────
-- Application code computes the bucket from the event timestamp
-- then includes it in every write and read
INSERT INTO user_events_v2 (user_id, bucket, event_time, event_type, metadata)
VALUES (
    a3b4c5d6-e7f8-9012-abcd-ef1234567890,
    '2026-04-18',
    toTimestamp(now()),
    'page_view',
    '{"page":"/dashboard"}'
);

-- ── Reading from a specific bucket ───────────────────────────────────────────
-- This is a single-partition read: fast, predictable, no cluster scan
SELECT event_time, event_type, metadata
FROM user_events_v2
WHERE user_id = a3b4c5d6-e7f8-9012-abcd-ef1234567890
  AND bucket   = '2026-04-18'
  AND event_time > '2026-04-18 08:00:00+0000'
LIMIT 50;

-- ── What ALLOW FILTERING looks like — and why it is dangerous ─────────────────
-- This forces Cassandra to scan EVERY partition on EVERY node
-- to find rows where event_type = 'purchase'.
-- With 50 nodes and 500GB of data, this query touches all of it.
-- Never run this in production without SAI or a dedicated table.
SELECT * FROM user_events_v2
WHERE event_type = 'purchase'
ALLOW FILTERING;  -- full cluster scan — do not do this

-- ── Demonstrating Cassandra's write semantics ────────────────────────────────
-- INSERT and UPDATE are identical in Cassandra — both are 'upserts'.
-- There is no read-before-write. The write goes to memtable immediately.
INSERT INTO forge_system.audit_logs
    (service_name, log_time, request_id, payload)
VALUES
    ('order-service', '2026-04-18 10:00:00+0000', uuid(), '{"amount":99.99}');

-- UPDATE writes a new cell with a higher timestamp — the old cell still exists
-- in the SSTable until compaction. Cassandra resolves conflicts by timestamp.
UPDATE forge_system.audit_logs
SET payload = '{"amount":99.99,"status":"confirmed"}'
WHERE service_name = 'order-service'
  AND log_time = '2026-04-18 10:00:00+0000';

-- ── Deletes write tombstones, not erasures ───────────────────────────────────
-- This DELETE writes a tombstone marker — it does NOT remove the data immediately.
-- The data will be physically removed after gc_grace_seconds (default: 10 days)
-- AND after a compaction runs AND after repair has propagated the tombstone.
DELETE FROM forge_system.audit_logs
WHERE service_name = 'order-service'
  AND log_time = '2026-04-18 10:00:00+0000';

-- ── Checking SSTable and memtable state via nodetool ─────────────────────────
-- Run from the shell, not from cqlsh:
-- nodetool flush forge_system         → forces memtable to flush to SSTable
-- nodetool compact forge_system       → forces compaction on the keyspace
-- nodetool tablehistograms forge_system audit_logs → read/write latency breakdown

-- ── TTL as an alternative to application-level deletes ───────────────────────
-- When data has a natural expiry, use TTL instead of explicit DELETE.
-- Cassandra handles the tombstone lifecycle automatically.
-- Here, this log entry expires in 86400 seconds (24 hours).
INSERT INTO forge_system.audit_logs
    (service_name, log_time, request_id, payload)
VALUES
    ('cache-service', toTimestamp(now()), uuid(), '{"cache":"hit"}')
USING TTL 86400;

# ── Three-node Cassandra development cluster ─────────────────────────────────
# Pinned to Cassandra 5.0 — do not use 'latest' in development or production.
# Cassandra 5.0 ships with significant storage engine improvements:
#   - Trie-based memtable (better memory efficiency)
#   - Storage Attached Indexes (SAI) replacing legacy SASI
#   - Unified compaction strategy option

services:
  cassandra-node1:
    image: cassandra:5.0
    container_name: forge-cassandra-1
    hostname: cassandra-node1
    ports:
      - "9042:9042"
    environment:
      - CASSANDRA_CLUSTER_NAME=TheCodeForgeCluster
      - CASSANDRA_DC=datacenter1
      - CASSANDRA_RACK=rack1
      - CASSANDRA_SEEDS=cassandra-node1,cassandra-node2
      - CASSANDRA_ENDPOINT_SNITCH=GossipingPropertyFileSnitch
      - MAX_HEAP_SIZE=1G
      - HEAP_NEWSIZE=256M
    volumes:
      - cassandra-data1:/var/lib/cassandra
    healthcheck:
      test: ["CMD-SHELL", "nodetool status | grep -E '^UN'"]
      interval: 30s
      timeout: 10s
      retries: 10
      start_period: 60s

  cassandra-node2:
    image: cassandra:5.0
    container_name: forge-cassandra-2
    hostname: cassandra-node2
    environment:
      - CASSANDRA_CLUSTER_NAME=TheCodeForgeCluster
      - CASSANDRA_DC=datacenter1
      - CASSANDRA_RACK=rack2
      - CASSANDRA_SEEDS=cassandra-node1,cassandra-node2
      - CASSANDRA_ENDPOINT_SNITCH=GossipingPropertyFileSnitch
      - MAX_HEAP_SIZE=1G
      - HEAP_NEWSIZE=256M
    volumes:
      - cassandra-data2:/var/lib/cassandra
    depends_on:
      cassandra-node1:
        condition: service_healthy

  cassandra-node3:
    image: cassandra:5.0
    container_name: forge-cassandra-3
    hostname: cassandra-node3
    environment:
      - CASSANDRA_CLUSTER_NAME=TheCodeForgeCluster
      - CASSANDRA_DC=datacenter1
      - CASSANDRA_RACK=rack3
      - CASSANDRA_SEEDS=cassandra-node1,cassandra-node2
      - CASSANDRA_ENDPOINT_SNITCH=GossipingPropertyFileSnitch
      - MAX_HEAP_SIZE=1G
      - HEAP_NEWSIZE=256M
    volumes:
      - cassandra-data3:/var/lib/cassandra
    depends_on:
      cassandra-node2:
        condition: service_healthy

volumes:
  cassandra-data1:
  cassandra-data2:
  cassandra-data3:

-- ── Storage Attached Indexes: Cassandra 5.0+ ─────────────────────────────────
-- SAI allows querying on non-primary-key columns without ALLOW FILTERING.
-- SASI is deprecated in Cassandra 5.0 — use SAI for all new indexes.

USE forge_system;

-- Table: product catalog — partition key is category
CREATE TABLE IF NOT EXISTS products (
    category    text,
    product_id  uuid,
    name        text,
    price       decimal,
    in_stock    boolean,
    rating      float,
    PRIMARY KEY (category, product_id)
);

-- ── Create SAI indexes on non-primary-key columns ────────────────────────────
-- Syntax: CREATE INDEX index_name ON table(column) USING 'sai';
CREATE INDEX products_price_idx  ON products(price)   USING 'sai';
CREATE INDEX products_stock_idx  ON products(in_stock) USING 'sai';
CREATE INDEX products_rating_idx ON products(rating)  USING 'sai';

-- ── SAI-enabled queries: no ALLOW FILTERING required ─────────────────────────
-- Range query on price — only possible with SAI or as a partition key
SELECT product_id, name, price
FROM products
WHERE category = 'electronics'
  AND price > 100.00
  AND price < 500.00;

-- Combined filter: in stock AND rating above threshold
-- Both conditions use SAI — Cassandra intersects the index results
SELECT product_id, name, price, rating
FROM products
WHERE category = 'electronics'
  AND in_stock = true
  AND rating > 4.5;

-- ── Vector search with SAI (AI workloads) ────────────────────────────────────
-- Cassandra 5.0 supports ANN (Approximate Nearest Neighbor) search via SAI
-- Useful for semantic search, recommendation systems, embedding lookup
CREATE TABLE IF NOT EXISTS product_embeddings (
    category   text,
    product_id uuid,
    embedding  vector<float, 1536>,  -- 1536-dimensional OpenAI embedding
    PRIMARY KEY (category, product_id)
);

-- SAI index on the vector column for ANN search
CREATE INDEX embeddings_ann_idx
    ON product_embeddings(embedding)
    USING 'sai'
    WITH OPTIONS = {'similarity_function': 'cosine'};

-- ANN query: find the 10 most similar products to a given embedding vector
-- This runs on-node without a cluster scan — orders of magnitude faster than
-- brute-force similarity across all rows
SELECT product_id
FROM product_embeddings
WHERE category = 'electronics'
ORDER BY embedding ANN OF [0.1, 0.2, 0.3, ...] -- 1536 floats from your model
LIMIT 10;

// ── pom.xml dependencies ──────────────────────────────────────────────────────
// <dependency>
//   <groupId>com.datastax.oss</groupId>
//   <artifactId>java-driver-core</artifactId>
//   <version>4.17.0</version>
// </dependency>
// <dependency>
//   <groupId>com.datastax.oss</groupId>
//   <artifactId>java-driver-query-builder</artifactId>
//   <version>4.17.0</version>
// </dependency>

// ── CassandraConfig.java — single CqlSession for the application lifetime ────
package io.thecodeforge.config;

import com.datastax.oss.driver.api.core.CqlSession;
import com.datastax.oss.driver.api.core.config.DefaultDriverOption;
import com.datastax.oss.driver.api.core.config.DriverConfigLoader;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.net.InetSocketAddress;
import java.time.Duration;

@Configuration
public class CassandraConfig {

    // ONE CqlSession per application — not one per request.
    // CqlSession is thread-safe and manages its own connection pool internally.
    // Creating a session per request is one of the most common performance mistakes.
    @Bean(destroyMethod = "close")
    public CqlSession cqlSession() {
        return CqlSession.builder()
                .addContactPoint(new InetSocketAddress("127.0.0.1", 9042))
                .withLocalDatacenter("datacenter1")
                .withKeyspace("forge_system")
                .withConfigLoader(
                    DriverConfigLoader.programmaticBuilder()
                        // Request timeout: fail fast rather than hold threads
                        .withDuration(DefaultDriverOption.REQUEST_TIMEOUT,
                                      Duration.ofMillis(500))
                        // Consistency level: LOCAL_QUORUM for production
                        // Tolerates one replica being down in the local DC
                        .withString(DefaultDriverOption.REQUEST_CONSISTENCY,
                                    "LOCAL_QUORUM")
                        // Token-aware routing: send queries directly to the
                        // replica that owns the partition — avoids coordinator hop
                        .withString(DefaultDriverOption.LOAD_BALANCING_POLICY_CLASS,
                                    "DefaultLoadBalancingPolicy")
                        .build()
                )
                .build();
    }
}

// ── AuditLog.java — simple value object ──────────────────────────────────────
package io.thecodeforge.model;

import java.time.Instant;
import java.util.UUID;

public record AuditLog(
    String  serviceName,
    Instant logTime,
    UUID    requestId,
    String  payload
) {}

// ── AuditLogRepository.java — DataStax driver with prepared statements ────────
package io.thecodeforge.repository;

import com.datastax.oss.driver.api.core.CqlSession;
import com.datastax.oss.driver.api.core.cql.*;
import io.thecodeforge.model.AuditLog;
import org.springframework.stereotype.Repository;
import java.time.Instant;
import java.util.List;
import java.util.UUID;
import java.util.stream.StreamSupport;

@Repository
public class AuditLogRepository {

    private final CqlSession session;

    // Prepared statements: parse once, execute many times.
    // Prepared at construction time — fails fast on startup if CQL is wrong.
    private final PreparedStatement insertStmt;
    private final PreparedStatement selectRecentStmt;

    public AuditLogRepository(CqlSession session) {
        this.session = session;

        this.insertStmt = session.prepare(
            "INSERT INTO audit_logs (service_name, log_time, request_id, payload) "
          + "VALUES (?, ?, ?, ?) USING TTL 2592000"
        );

        this.selectRecentStmt = session.prepare(
            "SELECT service_name, log_time, request_id, payload "
          + "FROM audit_logs "
          + "WHERE service_name = ? AND log_time > ? "
          + "LIMIT 100"
        );
    }

    public void save(AuditLog log) {
        BoundStatement bound = insertStmt.bind(
            log.serviceName(),
            log.logTime(),
            log.requestId(),
            log.payload()
        );
        // execute() is synchronous — use executeAsync() for high-throughput writes
        session.execute(bound);
    }

    public List<AuditLog> findRecent(String serviceName, Instant since) {
        BoundStatement bound = selectRecentStmt.bind(serviceName, since);
        ResultSet rs = session.execute(bound);

        return StreamSupport.stream(rs.spliterator(), false)
                .map(row -> new AuditLog(
                    row.getString("service_name"),
                    row.getInstant("log_time"),
                    row.getUuid("request_id"),
                    row.getString("payload")
                ))
                .toList(); // Java 16+ — returns unmodifiable list
    }

    // ── Async write for high-throughput ingestion paths ───────────────────────
    // Returns CompletableFuture<AsyncResultSet> — caller decides whether to await
    public java.util.concurrent.CompletableFuture<AsyncResultSet> saveAsync(AuditLog log) {
        BoundStatement bound = insertStmt.bind(
            log.serviceName(),
            log.logTime(),
            log.requestId(),
            log.payload()
        );
        return session.executeAsync(bound).toCompletableFuture();
    }
}