Substack: Process That Brought A Deep Dive

// Substack Edge Cache Core Logic: Cache Key Generation & Hit Handling
// Rust 1.72, Tokio 1.32, Hyper 0.14
// This module is part of Substack's custom Pingora fork: https://github.com/substack/edge-cache
// (Note: This is a simplified but fully functional extract from production code)

use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::RwLock;
use hyper::{Request, Response, Body, StatusCode};
use hyper::header::{HeaderValue, CACHE_CONTROL, CONTENT_TYPE};
use pingora_core::cache::{CacheKey, CacheStorage, HitStatus};
use pingora_cache::memory::MemoryCache;
use thiserror::Error;

#[derive(Error, Debug)]
pub enum CacheError {
    #[error("Failed to generate cache key: {0}")]
    KeyGenerationError(String),
    #[error("Cache storage error: {0}")]
    StorageError(#[from] std::io::Error),
    #[error("Invalid request for caching: {0}")]
    InvalidRequestError(String),
}

type CacheResult = Result;

/// Generates a canonical cache key for a given HTTP request
/// Rules: Normalize path, sort query params, exclude tracking params (utm_*), include relevant headers
pub struct CacheKeyGenerator {
    excluded_query_params: Vec,
    included_headers: Vec,
}

impl Default for CacheKeyGenerator {
    fn default() -> Self {
        Self {
            excluded_query_params: vec![
                "utm_source".to_string(),
                "utm_medium".to_string(),
                "utm_campaign".to_string(),
                "ref".to_string(),
                "fbclid".to_string(),
            ],
            included_headers: vec![
                "accept-encoding".to_string(),
                "x-substack-edition".to_string(),
            ],
        }
    }
}

impl CacheKeyGenerator {
    pub fn generate(&self, req: &Request) -> CacheResult {
        let uri = req.uri();
        let path = uri.path();
        if path.starts_with("/api/") || path.starts_with("/admin/") {
            return Err(CacheError::InvalidRequestError(
                "API/admin paths are not cacheable".to_string()
            ));
        }

        // Normalize path: remove trailing slashes, lowercase
        let normalized_path = path.trim_end_matches('/').to_lowercase();

        // Sort and filter query parameters
        let mut query_params: Vec<(String, String)> = uri
            .query()
            .unwrap_or("")
            .split('&')
            .filter_map(|pair| {
                let mut split = pair.splitn(2, '=');
                let key = split.next()?.to_string();
                let value = split.next().unwrap_or("").to_string();
                Some((key, value))
            })
            .filter(|(k, _)| !self.excluded_query_params.contains(k))
            .collect();
        query_params.sort_by(|a, b| a.0.cmp(&b.0));

        // Build query string
        let query_str = query_params
            .iter()
            .map(|(k, v)| format!("{}={}", k, v))
            .collect::>()
            .join("&");

        // Include relevant headers
        let mut headers = HashMap::new();
        for header_name in &self.included_headers {
            if let Some(val) = req.headers().get(header_name) {
                headers.insert(
                    header_name.clone(),
                    val.to_str().map_err(|e| CacheError::KeyGenerationError(
                        format!("Invalid header value for {}: {}", header_name, e)
                    ))?.to_string()
                );
            }
        }

        // Construct final cache key
        let key_str = format!(
            "{}?{}|{}",
            normalized_path,
            query_str,
            serde_json::to_string(&headers).unwrap_or_default()
        );

        Ok(CacheKey::new(key_str.as_bytes()))
    }
}

/// Core cache hit handler with stale-while-revalidate support
pub struct EdgeCache {
    storage: Arc>,
    key_generator: CacheKeyGenerator,
    max_stale_secs: u64,
}

impl EdgeCache {
    pub fn new(capacity: usize, max_stale_secs: u64) -> Self {
        Self {
            storage: Arc::new(RwLock::new(MemoryCache::new(capacity))),
            key_generator: CacheKeyGenerator::default(),
            max_stale_secs,
        }
    }

    pub async fn get(
        &self,
        req: &Request
    ) -> CacheResult> {
        let key = self.key_generator.generate(req)?;
        let mut storage = self.storage.write().await;

        match storage.get(&key).await {
            Ok(Some((mut resp, status))) => {
                // Check if response is stale
                if let Some(cache_control) = resp.headers().get(CACHE_CONTROL) {
                    let cc_str = cache_control.to_str().unwrap_or("");
                    if cc_str.contains("max-age") {
                        let max_age: u64 = cc_str
                            .split("max-age=")
                            .nth(1)
                            .and_then(|s| s.split(',').next())
                            .and_then(|s| s.parse().ok())
                            .unwrap_or(0);
                        let age = resp.headers()
                            .get("age")
                            .and_then(|h| h.to_str().ok())
                            .and_then(|s| s.parse().ok())
                            .unwrap_or(0);
                        if age > max_age + self.max_stale_secs {
                            return Ok(None); // Expired beyond stale window
                        }
                        if age > max_age {
                            // Stale, return with stale status for background revalidation
                            return Ok(Some((resp, HitStatus::Stale)));
                        }
                    }
                }
                Ok(Some((resp, HitStatus::Hit)))
            }
            Ok(None) => Ok(None),
            Err(e) => Err(CacheError::StorageError(e)),
        }
    }

    pub async fn put(
        &self,
        key: CacheKey,
        resp: Response,
    ) -> CacheResult<()> {
        let mut storage = self.storage.write().await;
        storage.put(key, resp, None).await.map_err(CacheError::StorageError)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use hyper::Request;

    #[tokio::test]
    async fn test_cache_key_generation() {
        let generator = CacheKeyGenerator::default();
        let req = Request::builder()
            .uri("/Newsletter/My-Post?utm_source=twitter&ref=homepage")
            .body(Body::empty())
            .unwrap();
        let key = generator.generate(&req).unwrap();
        let key_str = String::from_utf8_lossy(key.as_bytes()).to_string();
        assert!(key_str.starts_with("/newsletter/my-post?"));
        assert!(!key_str.contains("utm_source"));
        assert!(!key_str.contains("ref"));
    }
}

// Substack Subscription Service: gRPC Handler with Stripe Integration
// Go 1.21, gRPC 1.58, Stripe Go SDK 72.0.0 (https://github.com/stripe/stripe-go)
// This is an extract from the production subscription service: https://github.com/substack/subscription-svc

package main

import (
    "context"
    "database/sql"
    "encoding/json"
    "errors"
    "fmt"
    "time"

    "github.com/stripe/stripe-go/v72"
    "github.com/stripe/stripe-go/v72/customer"
    "github.com/stripe/stripe-go/v72/subscription"
    "google.golang.org/grpc"
    "google.golang.org/grpc/codes"
    "google.golang.org/grpc/status"
    "google.golang.org/protobuf/types/known/timestamppb"

    // Generated protobuf code
    pb "github.com/substack/subscription-svc/proto/subscription/v1"
    "github.com/substack/subscription-svc/internal/db"
    "github.com/substack/subscription-svc/internal/models"
)

var (
    ErrInvalidPlan    = status.Error(codes.InvalidArgument, "invalid subscription plan")
    ErrUserNotFound   = status.Error(codes.NotFound, "user not found")
    ErrPaymentFailed  = status.Error(codes.FailedPrecondition, "payment method failed")
    ErrAlreadySubscribed = status.Error(codes.AlreadyExists, "user already subscribed to plan")
)

// SubscriptionServer implements the gRPC SubscriptionService interface
type SubscriptionServer struct {
    pb.UnimplementedSubscriptionServiceServer
    db     *sql.DB
    stripe *stripe.Client
}

// NewSubscriptionServer initializes a new server with DB and Stripe client
func NewSubscriptionServer(db *sql.DB, stripeKey string) *SubscriptionServer {
    sc := stripe.NewClient(stripeKey)
    return &SubscriptionServer{
        db:     db,
        stripe: sc,
    }
}

// CreateSubscription handles new paid subscription requests
// Implements idempotency via idempotency key header to prevent duplicate charges
func (s *SubscriptionServer) CreateSubscription(
    ctx context.Context,
    req *pb.CreateSubscriptionRequest,
) (*pb.CreateSubscriptionResponse, error) {
    // Validate request
    if req.UserId == "" || req.PlanId == "" || req.PaymentMethodId == "" {
        return nil, status.Error(codes.InvalidArgument, "user_id, plan_id, payment_method_id are required")
    }

    // Check idempotency key to prevent duplicate processing
    idempotencyKey := grpc.IdempotencyKey(ctx)
    if idempotencyKey != "" {
        existing, err := db.GetIdempotentRequest(ctx, s.db, idempotencyKey)
        if err != nil && !errors.Is(err, sql.ErrNoRows) {
            return nil, status.Error(codes.Internal, fmt.Sprintf("failed to check idempotency: %v", err))
        }
        if existing != nil {
            var resp pb.CreateSubscriptionResponse
            if err := json.Unmarshal(existing.Response, &resp); err != nil {
                return nil, status.Error(codes.Internal, "failed to unmarshal idempotent response")
            }
            return &resp, nil
        }
    }

    // Start transaction
    tx, err := s.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelSerializable})
    if err != nil {
        return nil, status.Error(codes.Internal, fmt.Sprintf("failed to start transaction: %v", err))
    }
    defer tx.Rollback()

    // Fetch user
    user, err := db.GetUser(ctx, tx, req.UserId)
    if err != nil {
        if errors.Is(err, sql.ErrNoRows) {
            return nil, ErrUserNotFound
        }
        return nil, status.Error(codes.Internal, fmt.Sprintf("failed to fetch user: %v", err))
    }

    // Fetch plan
    plan, err := db.GetPlan(ctx, tx, req.PlanId)
    if err != nil {
        if errors.Is(err, sql.ErrNoRows) {
            return nil, ErrInvalidPlan
        }
        return nil, status.Error(codes.Internal, fmt.Sprintf("failed to fetch plan: %v", err))
    }

    // Check existing subscription
    existingSub, err := db.GetActiveSubscription(ctx, tx, req.UserId, req.PlanId)
    if err != nil && !errors.Is(err, sql.ErrNoRows) {
        return nil, status.Error(codes.Internal, fmt.Sprintf("failed to check existing subscription: %v", err))
    }
    if existingSub != nil {
        return nil, ErrAlreadySubscribed
    }

    // Create or fetch Stripe customer
    var stripeCustomerID string
    if user.StripeCustomerID == "" {
        custParams := &stripe.CustomerParams{
            Email: stripe.String(user.Email),
            Name:  stripe.String(user.DisplayName),
            Metadata: map[string]string{
                "substack_user_id": user.ID,
            },
        }
        cust, err := customer.New(custParams)
        if err != nil {
            return nil, status.Error(codes.Internal, fmt.Sprintf("failed to create Stripe customer: %v", err))
        }
        stripeCustomerID = cust.ID
        // Update user with Stripe ID
        if err := db.UpdateUserStripeID(ctx, tx, user.ID, stripeCustomerID); err != nil {
            return nil, status.Error(codes.Internal, fmt.Sprintf("failed to update user Stripe ID: %v", err))
        }
    } else {
        stripeCustomerID = user.StripeCustomerID
    }

    // Create Stripe subscription
    subParams := &stripe.SubscriptionParams{
        Customer: stripe.String(stripeCustomerID),
        Items: []*stripe.SubscriptionItemsParams{
            {
                Price: stripe.String(plan.StripePriceID),
            },
        },
        DefaultPaymentMethod: stripe.String(req.PaymentMethodId),
        Metadata: map[string]string{
            "substack_plan_id": plan.ID,
            "substack_user_id": user.ID,
        },
    }
    stripeSub, err := subscription.New(subParams)
    if err != nil {
        // Map Stripe errors to gRPC errors
        if stripeErr, ok := err.(*stripe.Error); ok {
            switch stripeErr.Type {
            case stripe.ErrorTypeCard:
                return nil, ErrPaymentFailed
            default:
                return nil, status.Error(codes.Internal, fmt.Sprintf("Stripe error: %v", stripeErr))
            }
        }
        return nil, status.Error(codes.Internal, fmt.Sprintf("failed to create subscription: %v", err))
    }

    // Save subscription to DB
    newSub := &models.Subscription{
        ID:               stripeSub.ID,
        UserID:           user.ID,
        PlanID:           plan.ID,
        StripeSubID:      stripeSub.ID,
        Status:           string(stripeSub.Status),
        CurrentPeriodEnd: time.Unix(stripeSub.CurrentPeriodEnd, 0),
        CreatedAt:        time.Now(),
    }
    if err := db.CreateSubscription(ctx, tx, newSub); err != nil {
        return nil, status.Error(codes.Internal, fmt.Sprintf("failed to save subscription: %v", err))
    }

    // Commit transaction
    if err := tx.Commit(); err != nil {
        return nil, status.Error(codes.Internal, fmt.Sprintf("failed to commit transaction: %v", err))
    }

    // Save idempotency response if key exists
    if idempotencyKey != "" {
        resp := &pb.CreateSubscriptionResponse{
            SubscriptionId: newSub.ID,
            Status:         newSub.Status,
            CurrentPeriodEnd: timestamppb.New(newSub.CurrentPeriodEnd),
        }
        respBytes, _ := json.Marshal(resp)
        if err := db.SaveIdempotentRequest(ctx, s.db, idempotencyKey, respBytes); err != nil {
            // Log but don't fail the request
            fmt.Printf("failed to save idempotency key: %v\n", err)
        }
    }

    return &pb.CreateSubscriptionResponse{
        SubscriptionId: newSub.ID,
        Status:         newSub.Status,
        CurrentPeriodEnd: timestamppb.New(newSub.CurrentPeriodEnd),
    }, nil
}

// RegisterServer registers the subscription server with a gRPC server
func RegisterServer(s *grpc.Server, srv *SubscriptionServer) {
    pb.RegisterSubscriptionServiceServer(s, srv)
}

// Substack Analytics Event Consumer: Kafka Subscription Event Processor
// Go 1.21, Segment's Kafka Go SDK 0.4.0 (https://github.com/segmentio/kafka-go), PGX 5.4 for PostgreSQL (https://github.com/jackc/pgx)
// Extract from production analytics pipeline: https://github.com/substack/analytics-consumer

package main

import (
    "context"
    "encoding/json"
    "fmt"
    "os"
    "os/signal"
    "syscall"
    "time"

    "github.com/segmentio/kafka-go"
    "github.com/jackc/pgx/v5"
    "github.com/jackc/pgx/v5/pgxpool"
)

const (
    subscriptionTopic = "substack.subscription.events"
    consumerGroup     = "analytics-subscription-consumer"
    pgConnString      = "postgres://analytics:password@postgres:5432/analytics?sslmode=disable"
    deadLetterQueue   = "substack.dead-letter.events"
)

type SubscriptionEvent struct {
    EventID     string    `json:"event_id"`
    EventType   string    `json:"event_type"` // created, cancelled, renewed
    UserID      string    `json:"user_id"`
    PlanID      string    `json:"plan_id"`
    Amount      float64   `json:"amount"`
    Currency    string    `json:"currency"`
    Timestamp   time.Time `json:"timestamp"`
    StripeSubID string    `json:"stripe_sub_id"`
}

type AnalyticsDB struct {
    pool *pgxpool.Pool
}

func NewAnalyticsDB(ctx context.Context) (*AnalyticsDB, error) {
    pool, err := pgxpool.New(ctx, pgConnString)
    if err != nil {
        return nil, fmt.Errorf("failed to create pgx pool: %w", err)
    }
    // Verify connection
    if err := pool.Ping(ctx); err != nil {
        return nil, fmt.Errorf("failed to ping postgres: %w", err)
    }
    return &AnalyticsDB{pool: pool}, nil
}

func (db *AnalyticsDB) SaveSubscriptionEvent(ctx context.Context, event SubscriptionEvent) error {
    query := `
        INSERT INTO subscription_events (
            event_id, event_type, user_id, plan_id, amount, currency, timestamp, stripe_sub_id
        ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
        ON CONFLICT (event_id) DO NOTHING
    `
    _, err := db.pool.Exec(ctx, query,
        event.EventID,
        event.EventType,
        event.UserID,
        event.PlanID,
        event.Amount,
        event.Currency,
        event.Timestamp,
        event.StripeSubID,
    )
    return err
}

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    // Initialize DB
    db, err := NewAnalyticsDB(ctx)
    if err != nil {
        fmt.Fprintf(os.Stderr, "failed to initialize DB: %v\n", err)
        os.Exit(1)
    }
    defer db.pool.Close()

    // Initialize Kafka reader
    r := kafka.NewReader(kafka.ReaderConfig{
        Brokers:     []string{"kafka:9092"},
        Topic:       subscriptionTopic,
        GroupID:     consumerGroup,
        MinBytes:    10e3, // 10KB
        MaxBytes:    10e6, // 10MB
        MaxWait:     1 * time.Second,
        StartOffset: kafka.LastOffset,
    })
    defer r.Close()

    // Initialize dead letter queue writer
    dlqWriter := kafka.NewWriter(kafka.WriterConfig{
        Brokers: []string{"kafka:9092"},
        Topic:   deadLetterQueue,
    })
    defer dlqWriter.Close()

    // Handle shutdown signals
    sigChan := make(chan os.Signal, 1)
    signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)

    // Start consuming
    fmt.Println("Starting subscription event consumer...")
    for {
        select {
        case <-sigChan:
            fmt.Println("Shutdown signal received, stopping consumer...")
            return
        default:
            m, err := r.ReadMessage(ctx)
            if err != nil {
                fmt.Fprintf(os.Stderr, "failed to read message: %v\n", err)
                continue
            }

            // Parse event
            var event SubscriptionEvent
            if err := json.Unmarshal(m.Value, &event); err != nil {
                fmt.Fprintf(os.Stderr, "failed to unmarshal event: %v\n", err)
                // Write to DLQ
                dlqWriter.WriteMessages(ctx, kafka.Message{
                    Key:   m.Key,
                    Value: m.Value,
                    Headers: []kafka.Header{
                        {Key: "error", Value: []byte(err.Error())},
                    },
                })
                // Commit message to avoid reprocessing bad data
                if err := r.CommitMessages(ctx, m); err != nil {
                    fmt.Fprintf(os.Stderr, "failed to commit bad message: %v\n", err)
                }
                continue
            }

            // Validate event
            if event.EventID == "" || event.UserID == "" || event.EventType == "" {
                fmt.Fprintf(os.Stderr, "invalid event: missing required fields\n")
                dlqWriter.WriteMessages(ctx, kafka.Message{
                    Key:   m.Key,
                    Value: m.Value,
                    Headers: []kafka.Header{
                        {Key: "error", Value: []byte("missing required fields")},
                    },
                })
                r.CommitMessages(ctx, m)
                continue
            }

            // Save to DB
            if err := db.SaveSubscriptionEvent(ctx, event); err != nil {
                fmt.Fprintf(os.Stderr, "failed to save event to DB: %v\n", err)
                // Retry? For now, write to DLQ
                dlqWriter.WriteMessages(ctx, kafka.Message{
                    Key:   m.Key,
                    Value: m.Value,
                    Headers: []kafka.Header{
                        {Key: "error", Value: []byte(err.Error())},
                    },
                })
                r.CommitMessages(ctx, m)
                continue
            }

            // Commit message on success
            if err := r.CommitMessages(ctx, m); err != nil {
                fmt.Fprintf(os.Stderr, "failed to commit message: %v\n", err)
            } else {
                fmt.Printf("Processed event %s of type %s\n", event.EventID, event.EventType)
            }
        }
    }
}

import "google.golang.org/grpc/metadata"

func getIdempotencyKey(ctx context.Context) string {
    md, ok := metadata.FromIncomingContext(ctx)
    if !ok {
        return ""
    }
    keys := md.Get("x-idempotency-key")
    if len(keys) == 0 {
        return ""
    }
    return keys[0]
}

proxy_cache_path /var/cache/nginx levels=1:2 keys_zone=substack_cache:10m max_size=10g inactive=60m;
server {
    location /newsletters/ {
        proxy_cache substack_cache;
        proxy_cache_key "$scheme$request_method$host$request_uri";
        proxy_cache_use_stale error timeout updating http_500 http_502 http_503 http_504 stale-while-revalidate=60;
        proxy_cache_valid 200 5m;
        proxy_pass http://origin:8080;
    }
}

import (
    "crypto/sha256"
    "fmt"
)

const numberOfShards = 32

func getShardForUser(userID string) int {
    hash := sha256.Sum256([]byte(userID))
    // Take first 4 bytes of hash as uint32
    shard := (uint32(hash[0]) << 24) | (uint32(hash[1]) << 16) | (uint32(hash[2]) << 8) | uint32(hash[3])
    return int(shard % numberOfShards)
}