Contrarian: You Don't Need a CS Degree to Be a Senior Engineer in 2026

package main

import (
    "context"
    "crypto/rand"
    "errors"
    "fmt"
    "log"
    "net/http"
    "os"
    "sync"
    "time"

    "github.com/go-redis/redis/v9" // https://github.com/go-redis/redis
)

// RateLimiter implements a sliding window rate limiter using Redis
// Configurable limit and window size, with fallback to in-memory if Redis is unavailable
type RateLimiter struct {
    redisClient *redis.Client
    limit       int           // max requests per window
    window      time.Duration // sliding window duration
    fallbackMap map[string]*slidingWindow // in-memory fallback
    fallbackMu  sync.Mutex    // protects fallbackMap
}

// slidingWindow tracks in-memory request timestamps for fallback mode
type slidingWindow struct {
    timestamps []time.Time
    mu         sync.Mutex
}

// NewRateLimiter initializes a Redis-backed rate limiter with fallback
// Returns error if Redis connection fails after 3 retries
func NewRateLimiter(ctx context.Context, redisAddr string, limit int, window time.Duration) (*RateLimiter, error) {
    if limit <= 0 {
        return nil, errors.New("rate limit must be positive integer")
    }
    if window <= 0 {
        return nil, errors.New("window duration must be positive")
    }

    // Initialize Redis client with retry logic
    var redisClient *redis.Client
    var err error
    for i := 0; i < 3; i++ {
        redisClient = redis.NewClient(&redis.Options{
            Addr:     redisAddr,
            Password: os.Getenv("REDIS_PASSWORD"),
            DB:       0,
        })
        _, err = redisClient.Ping(ctx).Result()
        if err == nil {
            break
        }
        log.Printf("Redis connection attempt %d failed: %v", i+1, err)
        time.Sleep(time.Duration(i+1) * 500 * time.Millisecond)
    }
    if err != nil {
        log.Printf("Falling back to in-memory rate limiting: %v", err)
    }

    return &RateLimiter{
        redisClient: redisClient,
        limit:       limit,
        window:      window,
        fallbackMap: make(map[string]*slidingWindow),
    }, nil
}

// Allow checks if a request from clientID is allowed under the rate limit
// Returns true if allowed, false if rate limited, and error for unexpected failures
func (rl *RateLimiter) Allow(ctx context.Context, clientID string) (bool, error) {
    if rl.redisClient != nil {
        allowed, err := rl.checkRedis(ctx, clientID)
        if err == nil {
            return allowed, nil
        }
        log.Printf("Redis check failed for %s: %v, falling back to in-memory", clientID, err)
    }
    return rl.checkFallback(clientID)
}

// checkRedis implements sliding window rate limiting via Redis sorted sets
func (rl *RateLimiter) checkRedis(ctx context.Context, clientID string) (bool, error) {
    now := time.Now().UnixNano()
    windowStart := now - rl.window.Nanoseconds()

    // Remove timestamps outside the current window
    _, err := rl.redisClient.ZRemRangeByScore(ctx, clientID, "0", fmt.Sprintf("%d", windowStart)).Result()
    if err != nil {
        return false, fmt.Errorf("failed to trim old entries: %w", err)
    }

    // Count current requests in window
    count, err := rl.redisClient.ZCard(ctx, clientID).Result()
    if err != nil {
        return false, fmt.Errorf("failed to count entries: %w", err)
    }

    if int(count) >= rl.limit {
        return false, nil
    }

    // Add current request timestamp
    _, err = rl.redisClient.ZAdd(ctx, clientID, &redis.Z{
        Score:  float64(now),
        Member: fmt.Sprintf("%d-%s", now, generateNonce()),
    }).Result()
    if err != nil {
        return false, fmt.Errorf("failed to add request entry: %w", err)
    }

    // Set key expiry to window duration to avoid stale keys
    rl.redisClient.Expire(ctx, clientID, rl.window)

    return true, nil
}

// checkFallback implements in-memory sliding window rate limiting
func (rl *RateLimiter) checkFallback(clientID string) (bool, error) {
    rl.fallbackMu.Lock()
    window, exists := rl.fallbackMap[clientID]
    if !exists {
        window = &slidingWindow{
            timestamps: make([]time.Time, 0, rl.limit),
        }
        rl.fallbackMap[clientID] = window
    }
    rl.fallbackMu.Unlock()

    window.mu.Lock()
    defer window.mu.Unlock()

    now := time.Now()
    windowStart := now.Add(-rl.window)

    // Trim old timestamps
    valid := make([]time.Time, 0, len(window.timestamps))
    for _, ts := range window.timestamps {
        if ts.After(windowStart) {
            valid = append(valid, ts)
        }
    }
    window.timestamps = valid

    if len(window.timestamps) >= rl.limit {
        return false, nil
    }

    window.timestamps = append(window.timestamps, now)
    return true, nil
}

// generateNonce creates a random 8-character string to avoid duplicate timestamps
func generateNonce() string {
    nonce := make([]byte, 8)
    rand.Read(nonce)
    return fmt.Sprintf("%x", nonce)
}

// Example usage in an HTTP middleware
func rateLimitMiddleware(rl *RateLimiter, next http.Handler) http.Handler {
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        clientID := r.Header.Get("X-Client-ID")
        if clientID == "" {
            clientID = r.RemoteAddr
        }

        allowed, err := rl.Allow(r.Context(), clientID)
        if err != nil {
            log.Printf("Rate limit check failed: %v", err)
            http.Error(w, "Internal Server Error", http.StatusInternalServerError)
            return
        }

        if !allowed {
            http.Error(w, "Rate limit exceeded", http.StatusTooManyRequests)
            return
        }

        next.ServeHTTP(w, r)
    })
}

func main() {
    ctx := context.Background()
    rl, err := NewRateLimiter(ctx, "localhost:6379", 100, time.Minute)
    if err != nil {
        log.Fatalf("Failed to initialize rate limiter: %v", err)
    }

    mux := http.NewServeMux()
    mux.HandleFunc("/api/resource", func(w http.ResponseWriter, r *http.Request) {
        fmt.Fprintf(w, "Request allowed")
    })

    handler := rateLimitMiddleware(rl, mux)
    log.Println("Server starting on :8080")
    log.Fatal(http.ListenAndServe(":8080", handler))
}

import asyncio
import aiohttp
import asyncpg
import json
import logging
import os
import time
from dataclasses import dataclass
from typing import List, Dict, Optional
from datetime import datetime

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

# Constants
API_BASE_URL = "https://api.example.com/v2/metrics"
DB_DSN = os.getenv("DATABASE_DSN", "postgresql://user:pass@localhost:5432/metrics")
MAX_RETRIES = 3
RETRY_DELAY = 2  # seconds
DEAD_LETTER_QUEUE_TABLE = "dead_letter_queue"

@dataclass
class MetricRecord:
    """Structured metric payload from external API"""
    metric_id: str
    timestamp: datetime
    value: float
    tags: Dict[str, str]

    @classmethod
    def from_api_response(cls, payload: Dict) -> "MetricRecord":
        """Parse API JSON response into MetricRecord, with validation"""
        required_fields = ["metric_id", "timestamp", "value"]
        for field in required_fields:
            if field not in payload:
                raise ValueError(f"Missing required field {field} in API response")

        try:
            timestamp = datetime.fromisoformat(payload["timestamp"].replace("Z", "+00:00"))
        except ValueError as e:
            raise ValueError(f"Invalid timestamp format: {payload['timestamp']}") from e

        if not isinstance(payload["value"], (int, float)):
            raise ValueError(f"Value must be numeric, got {type(payload['value'])}")

        return cls(
            metric_id=payload["metric_id"],
            timestamp=timestamp,
            value=float(payload["value"]),
            tags=payload.get("tags", {})
        )

class ETLPipeline:
    """Async ETL pipeline for ingesting metrics from external API to PostgreSQL"""

    def __init__(self, api_token: str, batch_size: int = 100):
        self.api_token = api_token
        self.batch_size = batch_size
        self.session: Optional[aiohttp.ClientSession] = None
        self.db_pool: Optional[asyncpg.Pool] = None
        self.dead_letter_queue: List[MetricRecord] = []

    async def __aenter__(self):
        """Initialize resources on context entry"""
        timeout = aiohttp.ClientTimeout(total=10)
        self.session = aiohttp.ClientSession(
            headers={"Authorization": f"Bearer {self.api_token}"},
            timeout=timeout
        )

        self.db_pool = await asyncpg.create_pool(
            dsn=DB_DSN,
            min_size=2,
            max_size=10
        )

        # Ensure dead letter queue table exists
        async with self.db_pool.acquire() as conn:
            await conn.execute(f"""
                CREATE TABLE IF NOT EXISTS {DEAD_LETTER_QUEUE_TABLE} (
                    id SERIAL PRIMARY KEY,
                    metric_id VARCHAR(255) NOT NULL,
                    payload JSONB NOT NULL,
                    error TEXT NOT NULL,
                    created_at TIMESTAMP NOT NULL DEFAULT NOW()
                )
            """)
        return self

    async def __aexit__(self, exc_type, exc, tb):
        """Clean up resources on context exit"""
        if self.session:
            await self.session.close()
        if self.db_pool:
            await self.db_pool.close()

    async def fetch_metrics_page(self, page: int) -> List[Dict]:
        """Fetch a single page of metrics from the API with retry logic"""
        url = f"{API_BASE_URL}?page={page}&per_page={self.batch_size}"
        for attempt in range(MAX_RETRIES):
            try:
                async with self.session.get(url) as resp:
                    if resp.status == 200:
                        return await resp.json()
                    elif resp.status == 429:
                        retry_after = int(resp.headers.get("Retry-After", RETRY_DELAY))
                        logger.warning(f"Rate limited, retrying after {retry_after}s")
                        await asyncio.sleep(retry_after)
                    else:
                        raise aiohttp.ClientResponseError(
                            resp.request_info,
                            resp.history,
                            status=resp.status,
                            message=await resp.text()
                        )
            except (aiohttp.ClientError, asyncio.TimeoutError) as e:
                logger.warning(f"Attempt {attempt+1} failed for page {page}: {e}")
                if attempt < MAX_RETRIES - 1:
                    await asyncio.sleep(RETRY_DELAY * (attempt + 1))
                else:
                    raise RuntimeError(f"Failed to fetch page {page} after {MAX_RETRIES} attempts") from e
        return []

    async def process_batch(self, raw_records: List[Dict]) -> None:
        """Process a batch of raw API records: parse, validate, insert to DB"""
        parsed_records: List[MetricRecord] = []
        for raw in raw_records:
            try:
                record = MetricRecord.from_api_response(raw)
                parsed_records.append(record)
            except ValueError as e:
                logger.error(f"Failed to parse record {raw.get('metric_id', 'unknown')}: {e}")
                # Write to dead letter queue
                async with self.db_pool.acquire() as conn:
                    await conn.execute(
                        f"INSERT INTO {DEAD_LETTER_QUEUE_TABLE} (metric_id, payload, error) VALUES ($1, $2, $3)",
                        raw.get("metric_id", "unknown"),
                        json.dumps(raw),
                        str(e)
                    )

        if not parsed_records:
            return

        # Bulk insert to PostgreSQL
        async with self.db_pool.acquire() as conn:
            await conn.executemany("""
                INSERT INTO metrics (metric_id, timestamp, value, tags)
                VALUES ($1, $2, $3, $4)
                ON CONFLICT (metric_id, timestamp) DO UPDATE SET
                    value = EXCLUDED.value,
                    tags = EXCLUDED.tags
            """, [
                (r.metric_id, r.timestamp, r.value, json.dumps(r.tags))
                for r in parsed_records
            ])
        logger.info(f"Inserted {len(parsed_records)} records to metrics table")

    async def run(self, start_page: int = 1, max_pages: int = 10) -> None:
        """Run the full ETL pipeline for specified number of pages"""
        page = start_page
        while page <= max_pages:
            logger.info(f"Fetching page {page}")
            try:
                raw_records = await self.fetch_metrics_page(page)
                if not raw_records:
                    logger.info("No more records to fetch")
                    break
                await self.process_batch(raw_records)
                page += 1
                await asyncio.sleep(0.5)  # Respect API rate limits
            except Exception as e:
                logger.error(f"Pipeline failed on page {page}: {e}")
                raise

async def main():
    api_token = os.getenv("API_TOKEN")
    if not api_token:
        raise ValueError("API_TOKEN environment variable is required")

    async with ETLPipeline(api_token=api_token, batch_size=100) as pipeline:
        await pipeline.run(start_page=1, max_pages=10)

if __name__ == "__main__":
    asyncio.run(main())

import axios, { AxiosError, AxiosRequestConfig } from "axios";
import { z } from "zod"; // https://github.com/colinhacks/zod
import { RateLimiter } from "limiter"; // https://github.com/nickpoorman/node-rate-limiter

// Schema definitions for type-safe API responses
const UserSchema = z.object({
  id: z.string().uuid(),
  email: z.string().email(),
  role: z.enum(["admin", "editor", "viewer"]),
  createdAt: z.string().datetime(),
});

const PaginatedUsersSchema = z.object({
  data: z.array(UserSchema),
  total: z.number().int().positive(),
  page: z.number().int().min(1),
  perPage: z.number().int().min(1).max(100),
});

// Inferred TypeScript types from schemas
export type User = z.infer;
export type PaginatedUsers = z.infer;

// Custom error class for API client errors
export class APIClientError extends Error {
  constructor(
    message: string,
    public readonly status?: number,
    public readonly code?: string,
    public readonly cause?: Error
  ) {
    super(message);
    this.name = "APIClientError";
    Object.setPrototypeOf(this, APIClientError.prototype);
  }
}

// Configuration for the API client
export interface APIClientConfig {
  baseURL: string;
  apiKey: string;
  maxRetries: number;
  rateLimitPerSecond: number;
}

export class TypeSafeAPIClient {
  private client: axios.AxiosInstance;
  private rateLimiter: RateLimiter;
  private maxRetries: number;

  constructor(config: APIClientConfig) {
    this.maxRetries = config.maxRetries;
    this.rateLimiter = new RateLimiter({
      tokensPerInterval: config.rateLimitPerSecond,
      interval: "second",
    });

    this.client = axios.create({
      baseURL: config.baseURL,
      headers: {
        "Authorization": `Bearer ${config.apiKey}`,
        "Content-Type": "application/json",
      },
      timeout: 10_000, // 10 second timeout
    });

    // Add response interceptor for global error handling
    this.client.interceptors.response.use(
      (response) => response,
      (error: AxiosError) => {
        const status = error.response?.status;
        const message = error.response?.data?.message || error.message;
        return Promise.reject(
          new APIClientError(
            `API request failed: ${message}`,
            status,
            error.code,
            error
          )
        );
      }
    );
  }

  /**
   * Execute a request with rate limiting and retry logic
   */
  private async executeRequest(
    config: AxiosRequestConfig,
    schema: z.ZodSchema,
    attempt: number = 1
  ): Promise {
    // Wait for rate limiter token
    await this.rateLimiter.removeTokens(1);

    try {
      const response = await this.client.request(config);
      // Validate response against schema
      const validated = schema.parse(response.data);
      return validated;
    } catch (error) {
      // Handle retryable errors (network errors, 429, 500s)
      const isRetryable =
        error instanceof APIClientError &&
        (error.status === 429 ||
          error.status >= 500 ||
          error.cause instanceof AxiosError);

      if (isRetryable && attempt <= this.maxRetries) {
        const delay = Math.min(2 ** attempt * 1000, 10_000); // Exponential backoff, max 10s
        console.warn(`Retrying request (attempt ${attempt}/${this.maxRetries}) after ${delay}ms`);
        await new Promise((resolve) => setTimeout(resolve, delay));
        return this.executeRequest(config, schema, attempt + 1);
      }

      // Re-throw non-retryable errors
      if (error instanceof z.ZodError) {
        throw new APIClientError(
          `Response validation failed: ${error.message}`,
          200,
          "VALIDATION_ERROR",
          error
        );
      }
      throw error;
    }
  }

  /**
   * Fetch paginated list of users
   */
  async getUsers(page: number = 1, perPage: number = 20): Promise {
    if (page < 1) throw new APIClientError("Page must be >= 1");
    if (perPage < 1 || perPage > 100) throw new APIClientError("perPage must be between 1 and 100");

    return this.executeRequest(
      {
        method: "GET",
        url: "/users",
        params: { page, perPage },
      },
      PaginatedUsersSchema
    );
  }

  /**
   * Fetch single user by ID
   */
  async getUser(userId: string): Promise {
    if (!userId) throw new APIClientError("userId is required");

    return this.executeRequest(
      {
        method: "GET",
        url: `/users/${userId}`,
      },
      UserSchema
    );
  }

  /**
   * Update user role (admin only)
   */
  async updateUserRole(userId: string, role: User["role"]): Promise {
    if (!userId) throw new APIClientError("userId is required");
    if (!["admin", "editor", "viewer"].includes(role)) {
      throw new APIClientError("Invalid role");
    }

    return this.executeRequest(
      {
        method: "PATCH",
        url: `/users/${userId}`,
        data: { role },
      },
      UserSchema
    );
  }
}

// Example usage
async function main() {
  const client = new TypeSafeAPIClient({
    baseURL: "https://api.example.com",
    apiKey: process.env.API_KEY || "",
    maxRetries: 3,
    rateLimitPerSecond: 10,
  });

  try {
    const users = await client.getUsers(1, 20);
    console.log(`Fetched ${users.data.length} users (total: ${users.total})`);

    if (users.data.length > 0) {
      const firstUser = await client.getUser(users.data[0].id);
      console.log(`First user: ${firstUser.email} (${firstUser.role})`);
    }
  } catch (error) {
    console.error("Failed to fetch users:", error);
    process.exit(1);
  }
}

main();

// Short Cobra command example for a rate limiter CLI
package cmd

import (
    "github.com/spf13/cobra" // https://github.com/spf13/cobra
    "my-rate-limiter/pkg/limiter"
)

var checkCmd = &cobra.Command{
    Use:   "check [client-id]",
    Short: "Check if a client is rate limited",
    Args:  cobra.ExactArgs(1),
    RunE: func(cmd *cobra.Command, args []string) error {
        clientID := args[0]
        allowed, err := limiter.GlobalLimiter.Allow(cmd.Context(), clientID)
        if err != nil {
            return err
        }
        if allowed {
            cmd.Println("ALLOWED")
        } else {
            cmd.Println("RATE LIMITED")
        }
        return nil
    },
}

func init() {
    rootCmd.AddCommand(checkCmd)
}

// Go benchmark example for rate limiter Allow function
package limiter

import (
    "context"
    "testing"
)

func BenchmarkRateLimiter_Allow(b *testing.B) {
    rl, err := NewRateLimiter(context.Background(), "localhost:6379", 1000, time.Minute)
    if err != nil {
        b.Fatalf("Failed to initialize rate limiter: %v", err)
    }

    b.ResetTimer()
    for i := 0; i < b.N; i++ {
        _, err := rl.Allow(context.Background(), "benchmark-client")
        if err != nil {
            b.Fatalf("Allow failed: %v", err)
        }
    }
}

// Prometheus metric export example for rate limiter
package limiter

import (
    "github.com/prometheus/client_golang/prometheus" // https://github.com/prometheus/client_golang
    "github.com/prometheus/client_golang/prometheus/promauto"
)

var (
    rateLimitChecks = promauto.NewCounterVec(
        prometheus.CounterOpts{
            Name: "rate_limiter_checks_total",
            Help: "Total number of rate limit checks",
        },
        []string{"client_id", "allowed"},
    )
)

func (rl *RateLimiter) AllowWithMetrics(ctx context.Context, clientID string) (bool, error) {
    allowed, err := rl.Allow(ctx, clientID)
    if err != nil {
        return false, err
    }
    rateLimitChecks.WithLabelValues(clientID, fmt.Sprintf("%v", allowed)).Inc()
    return allowed, nil
}