listmonk

1. EMAIL CLIENT ─── GET /campaign/{campaignUUID}/{subscriberUUID}/px.png ───▶ ECHO ROUTER
      │
      │  // No auth middleware — public endpoint. No session lookup.
      ▼
2. ECHO ROUTER ─── matches route "/campaign/:campUUID/:subUUID/px.png" ───▶ HANDLER: handleCampaignPixel()
      │
      │  // Parse UUIDs from path params. Validate format (fast regex). No DB lookup yet.
      ▼
3. HANDLER ─── checks privacy.disable_tracking setting ───▶ DECISION POINT
      │
      ├── IF tracking disabled: Return 1x1 PNG immediately. No DB write. O(1).
      │
      ├── IF privacy.individual_tracking = false:
      │     Insert into campaign_views with subscriber_id = NULL (anonymous).
      │     // Still counts the view, but can't attribute to a subscriber.
      │
      └── IF individual tracking enabled:
            │
            ▼
4. DB INSERT ─── INSERT INTO campaign_views (campaign_id, subscriber_id, created_at) ───▶ POSTGRESQL
      │
      │  // campaign_id resolved from UUID via campaigns table lookup (indexed).
      │  // subscriber_id resolved from UUID via subscribers table lookup (indexed).
      │  // Two UUID→ID lookups + one INSERT. Total: 3 queries.
      │  // campaign_views has BIGSERIAL PK — write-optimized append-only table.
      ▼
5. RESPONSE ─── 200 OK, Content-Type: image/png, body: 1x1 transparent PNG (68 bytes) ───▶ EMAIL CLIENT

1. BROWSER ─── GET /link/{linkUUID}/{campaignUUID}/{subscriberUUID} ───▶ ECHO ROUTER
      │
      │  // Public endpoint. No auth. Three UUIDs in path.
      ▼
2. HANDLER ─── handleLinkRedirect()
      │
      ├── Resolve link UUID → link record (get actual URL)
      │     // links table: url TEXT NOT NULL UNIQUE. UUID indexed.
      │
      ├── Resolve campaign UUID → campaign_id
      │
      ├── Resolve subscriber UUID → subscriber_id (if individual tracking on)
      │
      ├── INSERT INTO link_clicks (campaign_id, link_id, subscriber_id)
      │     // BIGSERIAL PK. Indexes on campaign_id, link_id, subscriber_id, DATE.
      │     // 4 index updates per INSERT. Heavier than campaign_views.
      │
      └── 302 REDIRECT → actual URL
            // User sees the destination page. Redirect is instant.

MANAGER GOROUTINE (long-lived, polls DB every ~5s)
      │
      ├── SELECT campaigns WHERE status IN ('running','scheduled')
      │     // status column indexed. Cheap scan — usually 0-5 active campaigns.
      │
      ▼  For each active campaign:
BATCH PRODUCER (one goroutine per campaign)
      │
      │   // Loop until all subscribers processed:
      ├── SELECT subscribers WHERE id > last_subscriber_id
      │     AND id IN (subscriber_lists WHERE list_id IN campaign_lists)
      │     AND status = 'enabled'
      │     ORDER BY id LIMIT 1000  // ← keyset pagination, O(1) via PK index
      │
      │   // For each subscriber in batch:
      ├── TEMPLATE RENDER ─── Go html/template.Execute()
      │     // Inject: subscriber.name, subscriber.attribs, campaign.subject
      │     // Generate: tracking pixel URL, wrapped link URLs
      │     // Sprig functions available. Compiled template cached (sync.Once).
      │     // CPU cost: ~0.1ms per render for simple templates.
      │
      ├── msgChan <- message  // Send to buffered channel. Blocks if channel full (backpressure).
      │
      └── UPDATE campaigns SET last_subscriber_id = ?, sent = ?  // Checkpoint after batch

                          ║ channel (buffer = batch_size)
                          ▼
WORKER POOL (N = app.concurrency, default 10 goroutines)
      │
      │   // Each worker loops: for msg := range msgChan { ... }
      │
      ├── RATE LIMITER ─── rateLimiter.Wait()
      │     // Token bucket: app.message_rate tokens/sec
      │     // OR sliding window: app.message_sliding_window_rate per window_duration
      │     // Blocks goroutine until token available. Natural throttle.
      │
      ├── MESSENGER.Push(msg) ─── SMTP connection from pool
      │     // Acquires connection from pool (max_conns per SMTP server)
      │     // SMTP EHLO → MAIL FROM → RCPT TO → DATA → message body → QUIT
      │     // Connection reused for next message (persistent TCP). Pipelining.
      │     // On failure: retry up to max_msg_retries times
      │
      ├── atomic.AddInt64(&sent, 1)  // Lock-free counter update
      │
      └── ON ERROR: increment error counter. If errors > max_send_errors → PAUSE campaign

1. BROWSER ─── POST /subscription/form ───▶ ECHO ROUTER
      │
      ├── CAPTCHA VERIFY ─── ALTCHA proof-of-work check OR hCaptcha API call
      │     // ALTCHA: CPU verification, no external API call (~1ms)
      │     // hCaptcha: external HTTP call to verify token (~100-300ms)
      │
      ├── DOMAIN FILTER ─── check email domain against blocklist/allowlist
      │     // privacy.domain_blocklist: ["*.disposable.com", "tempmail.org"]
      │     // In-memory check. O(N) against list but lists are tiny.
      │
      ├── UPSERT SUBSCRIBER ─── INSERT ... ON CONFLICT (email) DO UPDATE
      │     // Case-insensitive: idx_subs_email ON LOWER(email)
      │     // If exists: update name/attribs. If new: create with UUID.
      │
      ├── INSERT subscriber_lists ─── subscribe to requested lists
      │     // Status = 'unconfirmed' for double-optin lists
      │     // Status = 'confirmed' for single-optin lists
      │
      ├── IF double-optin: SEND CONFIRMATION EMAIL
      │     // Render optin template with confirmation link
      │     // Push to SMTP via messenger (same pool as campaigns)
      │     // Confirmation link: /subscription/optin/{subscriberUUID}/{listUUID}
      │
      └── 200 OK ─── render success template page

1. CLIENT ─── GET /api/subscribers?page=1&per_page=50 ───▶ ECHO ROUTER
      │
      │  // Authorization header: "username:api-token" (base64)
      ▼
2. MIDDLEWARE CHAIN
      │
      ├── Auth Middleware ─── validate API token
      │     // Lookup user by username in users table (indexed)
      │     // Verify token (bcrypt compare or direct match for API users)
      │     // Load user role + permissions from roles table
      │     // Set auth context on echo.Context
      │
      ├── Permission Check ─── does user have "subscribers:get_all" permission?
      │     // Check user_role permissions array. If list-scoped, filter by allowed lists.
      │
      └── CORS Middleware ─── check Origin header against security.cors_origins
      ▼
3. HANDLER ─── GetSubscribers()
      │
      ├── Parse query params ─── page, per_page, query, list_id, order_by
      │
      ├── Build SQL ─── dynamic WHERE clause from search/filter params
      │     // If "query" is plain text: ILIKE search on email and name
      │     // If "query" starts with SQL expression: parse as raw WHERE clause
      │     // Permission-filtered: only show subscribers in user's allowed lists
      │
      ├── EXECUTE SQL ─── via prepared statement from goyesql
      │     // Uses sqlx.Select() for struct scanning
      │     // COUNT(*) for total (separate query)
      │     // OFFSET/LIMIT pagination for API (not keyset — acceptable for admin UI)
      │
      └── 200 OK ─── JSON response with data[] + total + per_page + page

1. SES/SENDGRID ─── POST /webhooks/bounce/{type} ───▶ ECHO ROUTER
      │
      ├── Authenticate webhook ─── verify provider-specific auth
      │     // SES: verify SNS signature. SendGrid: verify key. Postmark: basic auth.
      │
      ├── Parse bounce payload ─── extract: email, bounce type, campaign ID
      │     // Normalize across providers into internal bounce_type ENUM
      │
      ├── INSERT INTO bounces ─── (subscriber_id, campaign_id, type, source, meta)
      │     // Subscriber looked up by email. campaign_id nullable (might be unknown).
      │
      ├── CHECK THRESHOLD ─── count bounces for this subscriber by type
      │     // SELECT COUNT(*) FROM bounces WHERE subscriber_id = ? AND type = ?
      │     // Compare against configured actions: hard.count=1, soft.count=2
      │
      └── IF threshold exceeded: UPDATE subscribers SET status = 'blocklisted'
            // Or DELETE subscriber if action = 'delete'
            // Cascading: subscriber_lists entries also cleaned up (FK CASCADE)

1. YOUR APP ─── POST /api/tx ───▶ ECHO ROUTER
      │
      │  // Body: { "subscriber_email": "...", "template_id": 5, "data": {...} }
      │  // Auth: API token (required)
      ▼
2. HANDLER
      │
      ├── Resolve subscriber ─── lookup by email or ID
      ├── Load TX template ─── from template cache (in-memory)
      ├── Render template ─── with subscriber data + custom data payload
      ├── messenger.Push(msg) ─── synchronous SMTP send
      │     // Blocks until SMTP ACK or error. Caller waits.
      │     // No rate limiting — each TX call = one immediate send.
      └── 200 OK ─── { "data": true }

CASCADE FAILURE SCENARIO — 50K req/sec tracking pixel burst

t=0s    50,000 requests arrive. Go spawns 50,000 goroutines. (~3.5 GB)
          │
          ▼
t=0.001s All 50K goroutines try to acquire a DB connection from pool (max_open=25).
          25 goroutines get connections. 49,975 goroutines BLOCK waiting.
          │
          ▼
t=0.005s Each DB query takes ~5ms. First 25 connections freed. Next 25 goroutines proceed.
          But another 50K requests arrived! Now 99,950 goroutines blocked. (~7 GB)
          │
          ▼
t=1.0s   At 50K/sec inflow, 25 conn pool processes ~5000 req/sec (25 × 200 queries/sec).
          Deficit: 45,000 goroutines/sec accumulating. After 1 second: ~45K blocked.
          │
          ▼
t=10s    ~450K goroutines blocked. Memory: ~30 GB. GC thrashing. Latency spikes.
          │
          ▼
t=20s    ~900K goroutines. OOM killer triggers. Process killed.
          Campaign sending (which shares the same process) dies too.

PRODUCTION FIX: Bounded Concurrency + Async Writes

// 1. Add server-level connection limits
srv := &http.Server{
    Addr:         ":9000",
    ReadTimeout:  10 * time.Second,   // Prevent slow-read attacks
    WriteTimeout: 15 * time.Second,   // Prevent goroutine hang on slow clients
    IdleTimeout:  60 * time.Second,   // Close idle keep-alive connections
    MaxHeaderBytes: 1 << 20,          // 1MB header limit
}

// 2. Add concurrency limiter middleware
sem := make(chan struct{}, 10000)      // Max 10K concurrent requests
e.Use(func(next echo.HandlerFunc) echo.HandlerFunc {
    return func(c echo.Context) error {
        select {
        case sem <- struct{}{}:
            defer func() { <-sem }()
            return next(c)
        default:
            return c.String(503, "server busy")  // Load shedding
        }
    }
})

// 3. Async tracking writes (decouple from request path)
trackChan := make(chan TrackEvent, 100000)  // Buffered channel

// Handler: write to channel, return immediately
func handlePixel(c echo.Context) error {
    select {
    case trackChan <- TrackEvent{campID, subID}:
        // Queued successfully
    default:
        // Channel full — drop event (acceptable for analytics)
    }
    return c.Blob(200, "image/png", pixel1x1)  // Instant response
}

// Background flusher: batch inserts every 100ms
go func() {
    ticker := time.NewTicker(100 * time.Millisecond)
    batch := make([]TrackEvent, 0, 1000)
    for {
        select {
        case ev := <-trackChan:
            batch = append(batch, ev)
            if len(batch) >= 1000 {
                flushBatch(batch)        // COPY INTO campaign_views
                batch = batch[:0]
            }
        case <-ticker.C:
            if len(batch) > 0 {
                flushBatch(batch)
                batch = batch[:0]
            }
        }
    }
}()

// 4. DB call timeouts
ctx, cancel := context.WithTimeout(c.Request().Context(), 3*time.Second)
defer cancel()
db.QueryContext(ctx, query, args...)   // Cancels if DB slow

// 5. Monitor goroutine count
go func() {
    for range time.Tick(5 * time.Second) {
        n := runtime.NumGoroutine()
        metrics.Gauge("goroutines", n)  // Prometheus metric
        if n > 50000 {
            log.Error("goroutine count critical", "count", n)
        }
    }
}()