📊 Week 3 — Data & Streams (Days 15–21)

# Create a DB subnet group (required by API, Floci accepts any value)
aws rds create-db-subnet-group \
  --db-subnet-group-name default \
  --db-subnet-group-description "default" \
  --subnet-ids subnet-00000001
 
# Create a PostgreSQL instance
aws rds create-db-instance \
  --db-instance-identifier my-postgres \
  --db-instance-class db.t3.micro \
  --engine postgres \
  --engine-version 15.4 \
  --master-username admin \
  --master-user-password secret123 \
  --db-name appdb \
  --allocated-storage 20 \
  --no-multi-az
 
# Wait for it to be available
aws rds wait db-instance-available \
  --db-instance-identifier my-postgres
 
# Get the endpoint
ENDPOINT=$(aws rds describe-db-instances \
  --db-instance-identifier my-postgres \
  --query 'DBInstances[0].Endpoint.Address' \
  --output text)
 
PORT=$(aws rds describe-db-instances \
  --db-instance-identifier my-postgres \
  --query 'DBInstances[0].Endpoint.Port' \
  --output text)
 
echo "Connect to: $ENDPOINT:$PORT"
 
# Connect directly with psql
psql -h $ENDPOINT -p $PORT -U admin -d appdb

// pkg/database/postgres.go
package database
 
import (
    "context"
    "fmt"
 
    "github.com/jackc/pgx/v5/pgxpool"
)
 
type PostgresConfig struct {
    Host     string
    Port     int
    User     string
    Password string
    DBName   string
    MaxConns int32
}
 
func NewPostgresPool(ctx context.Context, cfg PostgresConfig) (*pgxpool.Pool, error) {
    dsn := fmt.Sprintf(
        "postgresql://%s:%s@%s:%d/%s?sslmode=disable&pool_max_conns=%d",
        cfg.User, cfg.Password, cfg.Host, cfg.Port, cfg.DBName, cfg.MaxConns,
    )
    config, err := pgxpool.ParseConfig(dsn)
    if err != nil {
        return nil, err
    }
    config.MaxConns = cfg.MaxConns
    return pgxpool.NewWithConfig(ctx, config)
}

// migrations/000001_create_orders.up.sql
CREATE TABLE orders (
    id          UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    user_id     TEXT NOT NULL,
    total       DECIMAL(12,2) NOT NULL,
    status      TEXT NOT NULL DEFAULT 'pending',
    created_at  TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at  TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
 
CREATE INDEX idx_orders_user_id ON orders(user_id);
CREATE INDEX idx_orders_status ON orders(status);
CREATE INDEX idx_orders_created_at ON orders(created_at DESC);

# Run migrations
migrate -path ./migrations \
  -database "postgresql://admin:secret123@$ENDPOINT:$PORT/appdb?sslmode=disable" \
  up

# Create a manual snapshot
aws rds create-db-snapshot \
  --db-instance-identifier my-postgres \
  --db-snapshot-identifier my-postgres-snap-1
 
# Restore from snapshot to a new instance
aws rds restore-db-instance-from-db-snapshot \
  --db-instance-identifier my-postgres-restored \
  --db-snapshot-identifier my-postgres-snap-1

# Create a replication group (this starts a real Redis container)
aws elasticache create-replication-group \
  --replication-group-id my-redis \
  --description "dev redis" \
  --engine redis \
  --cache-node-type cache.t3.micro \
  --num-cache-clusters 1
 
# Wait for it
aws elasticache wait replication-group-available \
  --replication-group-id my-redis
 
# Get the endpoint
aws elasticache describe-replication-groups \
  --replication-group-id my-redis \
  --query 'ReplicationGroups[0].NodeGroups[0].PrimaryEndpoint'

// pkg/cache/redis.go
package cache
 
import (
    "context"
    "encoding/json"
    "time"
 
    "github.com/redis/go-redis/v9"
)
 
type RedisCache struct {
    client *redis.Client
}
 
func NewRedisCache(addr string) *RedisCache {
    return &RedisCache{
        client: redis.NewClient(&redis.Options{
            Addr:     addr,
            Password: "",
            DB:       0,
        }),
    }
}
 
// Generic typed cache methods
func Set[T any](ctx context.Context, c *RedisCache, key string, value T, ttl time.Duration) error {
    data, err := json.Marshal(value)
    if err != nil {
        return err
    }
    return c.client.Set(ctx, key, data, ttl).Err()
}
 
func Get[T any](ctx context.Context, c *RedisCache, key string) (T, bool, error) {
    var zero T
    data, err := c.client.Get(ctx, key).Bytes()
    if err == redis.Nil {
        return zero, false, nil
    }
    if err != nil {
        return zero, false, err
    }
    var value T
    return value, true, json.Unmarshal(data, &value)
}
 
// Rate limiter using Redis INCR + EXPIRE
func (c *RedisCache) RateLimit(ctx context.Context, key string, limit int, window time.Duration) (bool, error) {
    pipe := c.client.Pipeline()
    incr := pipe.Incr(ctx, key)
    pipe.Expire(ctx, key, window)
    _, err := pipe.Exec(ctx)
    if err != nil {
        return false, err
    }
    return incr.Val() <= int64(limit), nil
}
 
// Session storage
func (c *RedisCache) SetSession(ctx context.Context, sessionID string, data map[string]interface{}, ttl time.Duration) error {
    return c.client.HSet(ctx, "session:"+sessionID, data).Err()
}

// pkg/service/order_service.go — cache-aside
func (s *OrderService) GetOrder(ctx context.Context, orderID string) (*Order, error) {
    cacheKey := "order:" + orderID
 
    // 1. Try cache first
    if order, found, _ := cache.Get[Order](ctx, s.cache, cacheKey); found {
        return &order, nil
    }
 
    // 2. Cache miss — fetch from DB
    order, err := s.repo.GetByID(ctx, orderID)
    if err != nil {
        return nil, err
    }
 
    // 3. Populate cache
    cache.Set(ctx, s.cache, cacheKey, order, 5*time.Minute)
 
    return order, nil
}

# Create MSK cluster (Floci starts real Redpanda/Kafka)
CLUSTER_ARN=$(aws kafka create-cluster \
  --cluster-name dev-kafka \
  --kafka-version "3.5.1" \
  --number-of-broker-nodes 1 \
  --broker-node-group-info '{
    "instanceType":"kafka.t3.small",
    "clientSubnets":["subnet-00000001"]
  }' \
  --query ClusterArn --output text)
 
# Wait for cluster to be active
aws kafka wait cluster-active --cluster-arn $CLUSTER_ARN
 
# Get bootstrap brokers
BROKERS=$(aws kafka get-bootstrap-brokers \
  --cluster-arn $CLUSTER_ARN \
  --query BootstrapBrokerString --output text)
 
echo "Kafka brokers: $BROKERS"

// pkg/streaming/kafka_producer.go
package streaming
 
import (
    "context"
    "encoding/json"
 
    "github.com/IBM/sarama"
)
 
type KafkaProducer struct {
    producer sarama.SyncProducer
    topic    string
}
 
func NewKafkaProducer(brokers []string, topic string) (*KafkaProducer, error) {
    cfg := sarama.NewConfig()
    cfg.Producer.RequiredAcks = sarama.WaitForAll  // all replicas must ack
    cfg.Producer.Retry.Max = 5
    cfg.Producer.Return.Successes = true
    cfg.Producer.Idempotent = true
    cfg.Net.MaxOpenRequests = 1
 
    producer, err := sarama.NewSyncProducer(brokers, cfg)
    return &KafkaProducer{producer: producer, topic: topic}, err
}
 
func (p *KafkaProducer) Publish(ctx context.Context, key string, payload interface{}) error {
    data, err := json.Marshal(payload)
    if err != nil {
        return err
    }
    _, _, err = p.producer.SendMessage(&sarama.ProducerMessage{
        Topic: p.topic,
        Key:   sarama.StringEncoder(key),
        Value: sarama.ByteEncoder(data),
    })
    return err
}
 
// pkg/streaming/kafka_consumer.go
type KafkaConsumer struct {
    group   sarama.ConsumerGroup
    topics  []string
    handler sarama.ConsumerGroupHandler
}
 
func (c *KafkaConsumer) Start(ctx context.Context) error {
    for {
        if err := c.group.Consume(ctx, c.topics, c.handler); err != nil {
            return err
        }
        if ctx.Err() != nil {
            return nil
        }
    }
}
 
// Implement ConsumerGroupHandler:
type OrderEventHandler struct{}
 
func (h *OrderEventHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
    for msg := range claim.Messages() {
        var event OrderEvent
        json.Unmarshal(msg.Value, &event)
        // process...
        sess.MarkMessage(msg, "")  // commit offset
    }
    return nil
}

API (Go/ECS)
  └── POST /orders → Write to RDS PostgreSQL
                    └── Publish to MSK (Kafka topic: orders)
 
Inventory Service (Go/Lambda)
  └── Kafka consumer: orders topic
       └── Update inventory in DynamoDB
 
Notification Service (Go/Lambda)
  └── SQS consumer (SNS fan-out)
       └── Template → SES email
 
Cache Layer
  └── ElastiCache Redis — session store + order cache
 
Secrets
  └── All DB credentials via Secrets Manager