DEV Community: ahmet gedik

Monitoring and Alerting for Video Platform Infrastructure with Go

ahmet gedik — Mon, 11 May 2026 22:00:01 +0000

Why Custom Monitoring

Off-the-shelf monitoring tools are powerful but heavy. For a video platform like DailyWatch running on shared LiteSpeed hosting, we needed a lightweight Go binary that checks endpoint health, tracks uptime, and exposes Prometheus metrics -- without installing a full monitoring stack.

Health Check Configuration

Define what to monitor:

package monitor

import "time"

type CheckConfig struct {
    Name        string        `json:"name"`
    URL         string        `json:"url"`
    Method      string        `json:"method"`
    Interval    time.Duration `json:"interval"`
    Timeout     time.Duration `json:"timeout"`
    ExpectCode  int           `json:"expect_code"`
    ExpectBody  string        `json:"expect_body"`
}

type CheckResult struct {
    Name       string        `json:"name"`
    URL        string        `json:"url"`
    Status     string        `json:"status"` // "up" or "down"
    StatusCode int           `json:"status_code"`
    Latency    time.Duration `json:"latency"`
    Error      string        `json:"error,omitempty"`
    CheckedAt  time.Time     `json:"checked_at"`
}

var DefaultChecks = []CheckConfig{
    {
        Name:       "homepage",
        URL:        "https://dailywatch.video",
        Method:     "GET",
        Interval:   5 * time.Minute,
        Timeout:    10 * time.Second,
        ExpectCode: 200,
    },
    {
        Name:       "search_api",
        URL:        "https://dailywatch.video/search?q=test&format=json",
        Method:     "GET",
        Interval:   10 * time.Minute,
        Timeout:    15 * time.Second,
        ExpectCode: 200,
        ExpectBody: "results",
    },
    {
        Name:       "sitemap",
        URL:        "https://dailywatch.video/sitemap.xml",
        Method:     "GET",
        Interval:   30 * time.Minute,
        Timeout:    10 * time.Second,
        ExpectCode: 200,
        ExpectBody: "<urlset",
    },
}

The Health Checker

Perform checks concurrently:

package monitor

import (
    "fmt"
    "io"
    "net/http"
    "strings"
    "sync"
    "time"
)

type HealthChecker struct {
    client  *http.Client
    results map[string]*CheckResult
    history map[string][]CheckResult
    mu      sync.RWMutex
}

func NewHealthChecker() *HealthChecker {
    return &HealthChecker{
        client:  &http.Client{},
        results: make(map[string]*CheckResult),
        history: make(map[string][]CheckResult),
    }
}

func (hc *HealthChecker) RunCheck(cfg CheckConfig) CheckResult {
    hc.client.Timeout = cfg.Timeout
    start := time.Now()

    req, err := http.NewRequest(cfg.Method, cfg.URL, nil)
    if err != nil {
        return hc.recordResult(cfg.Name, cfg.URL, CheckResult{
            Name: cfg.Name, URL: cfg.URL, Status: "down",
            Error: fmt.Sprintf("request creation failed: %v", err),
            CheckedAt: time.Now(),
        })
    }
    req.Header.Set("User-Agent", "DailyWatch-Monitor/1.0")

    resp, err := hc.client.Do(req)
    latency := time.Since(start)

    if err != nil {
        return hc.recordResult(cfg.Name, cfg.URL, CheckResult{
            Name: cfg.Name, URL: cfg.URL, Status: "down",
            Latency: latency, Error: err.Error(), CheckedAt: time.Now(),
        })
    }
    defer resp.Body.Close()

    result := CheckResult{
        Name:       cfg.Name,
        URL:        cfg.URL,
        StatusCode: resp.StatusCode,
        Latency:    latency,
        CheckedAt:  time.Now(),
    }

    // Check status code
    if cfg.ExpectCode > 0 && resp.StatusCode != cfg.ExpectCode {
        result.Status = "down"
        result.Error = fmt.Sprintf("expected %d, got %d", cfg.ExpectCode, resp.StatusCode)
        return hc.recordResult(cfg.Name, cfg.URL, result)
    }

    // Check body content
    if cfg.ExpectBody != "" {
        body, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) // 1MB max
        if !strings.Contains(string(body), cfg.ExpectBody) {
            result.Status = "down"
            result.Error = fmt.Sprintf("body missing expected string: %s", cfg.ExpectBody)
            return hc.recordResult(cfg.Name, cfg.URL, result)
        }
    }

    result.Status = "up"
    return hc.recordResult(cfg.Name, cfg.URL, result)
}

func (hc *HealthChecker) recordResult(name, url string, result CheckResult) CheckResult {
    hc.mu.Lock()
    defer hc.mu.Unlock()
    hc.results[name] = &result
    hc.history[name] = append(hc.history[name], result)
    // Keep last 100 results
    if len(hc.history[name]) > 100 {
        hc.history[name] = hc.history[name][len(hc.history[name])-100:]
    }
    return result
}

func (hc *HealthChecker) GetUptime(name string, window time.Duration) float64 {
    hc.mu.RLock()
    defer hc.mu.RUnlock()
    cutoff := time.Now().Add(-window)
    total, up := 0, 0
    for _, r := range hc.history[name] {
        if r.CheckedAt.After(cutoff) {
            total++
            if r.Status == "up" {
                up++
            }
        }
    }
    if total == 0 {
        return 0
    }
    return float64(up) / float64(total) * 100
}

Prometheus Metrics Endpoint

Expose metrics in Prometheus format for scraping:

package main

import (
    "fmt"
    "net/http"
    "time"
    "myapp/monitor"
)

func metricsHandler(checker *monitor.HealthChecker, checks []monitor.CheckConfig) http.HandlerFunc {
    return func(w http.ResponseWriter, r *http.Request) {
        w.Header().Set("Content-Type", "text/plain; version=0.0.4")

        for _, cfg := range checks {
            result := checker.RunCheck(cfg)
            upVal := 0
            if result.Status == "up" {
                upVal = 1
            }
            uptime24h := checker.GetUptime(cfg.Name, 24*time.Hour)

            fmt.Fprintf(w, "# HELP video_platform_up Whether the endpoint is up (1) or down (0)\n")
            fmt.Fprintf(w, "video_platform_up{name=%q,url=%q} %d\n", cfg.Name, cfg.URL, upVal)
            fmt.Fprintf(w, "video_platform_latency_ms{name=%q} %.1f\n", cfg.Name, float64(result.Latency.Milliseconds()))
            fmt.Fprintf(w, "video_platform_uptime_24h{name=%q} %.2f\n", cfg.Name, uptime24h)
        }
    }
}

func main() {
    checker := monitor.NewHealthChecker()
    checks := monitor.DefaultChecks

    // Run checks on schedule
    go func() {
        for {
            for _, cfg := range checks {
                result := checker.RunCheck(cfg)
                if result.Status == "down" {
                    fmt.Printf("ALERT: %s is DOWN - %s\n", result.Name, result.Error)
                }
            }
            time.Sleep(5 * time.Minute)
        }
    }()

    http.HandleFunc("/metrics", metricsHandler(checker, checks))
    http.HandleFunc("/health", func(w http.ResponseWriter, r *http.Request) {
        w.Write([]byte("OK"))
    })

    fmt.Println("Monitor running on :9090")
    http.ListenAndServe(":9090", nil)
}

Scrape with Prometheus

Add to your prometheus.yml:

scrape_configs:
  - job_name: 'dailywatch'
    scrape_interval: 5m
    static_configs:
      - targets: ['localhost:9090']

This monitoring setup keeps us informed about the health of DailyWatch without any external dependencies. The single Go binary uses under 10MB of RAM and catches downtime within minutes. When an endpoint goes down, we see it in Grafana dashboards and can set up Alertmanager rules for Slack or email notifications.

This article is part of the Building DailyWatch series. Check out DailyWatch to see these techniques in action.

Monitoring and Alerting for Video Platform Infrastructure with Go

ahmet gedik — Mon, 11 May 2026 14:00:01 +0000

Monitoring a Multi-Region Video Vault

ViralVidVault serves viral video content from 7 European regions. When the Polish trending page goes down or the Swedish cron job fails, we need to know immediately -- not when a user complains. A custom Go monitor gives us exactly the health visibility we need.

Configuration

package monitor

import "time"

type Target struct {
    Name       string        `json:"name"`
    URL        string        `json:"url"`
    Interval   time.Duration `json:"interval"`
    Timeout    time.Duration `json:"timeout"`
    ExpectCode int           `json:"expect_code"`
    ExpectBody string        `json:"expect_body"`
    Region     string        `json:"region"`
}

type Result struct {
    Name       string        `json:"name"`
    Region     string        `json:"region"`
    Status     string        `json:"status"`
    StatusCode int           `json:"status_code"`
    Latency    time.Duration `json:"latency"`
    Error      string        `json:"error,omitempty"`
    Timestamp  time.Time     `json:"timestamp"`
}

var Targets = []Target{
    {Name: "homepage", URL: "https://viralvidvault.com", Interval: 5 * time.Minute, Timeout: 10 * time.Second, ExpectCode: 200},
    {Name: "viral_pl", URL: "https://viralvidvault.com/?region=PL", Interval: 10 * time.Minute, Timeout: 10 * time.Second, ExpectCode: 200, Region: "PL"},
    {Name: "viral_nl", URL: "https://viralvidvault.com/?region=NL", Interval: 10 * time.Minute, Timeout: 10 * time.Second, ExpectCode: 200, Region: "NL"},
    {Name: "viral_se", URL: "https://viralvidvault.com/?region=SE", Interval: 10 * time.Minute, Timeout: 10 * time.Second, ExpectCode: 200, Region: "SE"},
    {Name: "search_api", URL: "https://viralvidvault.com/search?q=viral&format=json", Interval: 15 * time.Minute, Timeout: 15 * time.Second, ExpectCode: 200, ExpectBody: "results"},
    {Name: "sitemap", URL: "https://viralvidvault.com/sitemap.xml", Interval: 30 * time.Minute, Timeout: 10 * time.Second, ExpectCode: 200, ExpectBody: "<urlset"},
}

The Checker Engine

package monitor

import (
    "fmt"
    "io"
    "net/http"
    "strings"
    "sync"
    "time"
)

type Checker struct {
    client  *http.Client
    latest  map[string]*Result
    history map[string][]Result
    mu      sync.RWMutex
    alerts  chan Result
}

func NewChecker() *Checker {
    return &Checker{
        client:  &http.Client{},
        latest:  make(map[string]*Result),
        history: make(map[string][]Result),
        alerts:  make(chan Result, 100),
    }
}

func (c *Checker) Check(t Target) Result {
    c.client.Timeout = t.Timeout
    start := time.Now()

    req, _ := http.NewRequest("GET", t.URL, nil)
    req.Header.Set("User-Agent", "ViralVidVault-Monitor/1.0")

    resp, err := c.client.Do(req)
    latency := time.Since(start)

    result := Result{
        Name:      t.Name,
        Region:    t.Region,
        Latency:   latency,
        Timestamp: time.Now(),
    }

    if err != nil {
        result.Status = "down"
        result.Error = err.Error()
        c.record(t.Name, result)
        return result
    }
    defer resp.Body.Close()

    result.StatusCode = resp.StatusCode

    if t.ExpectCode > 0 && resp.StatusCode != t.ExpectCode {
        result.Status = "down"
        result.Error = fmt.Sprintf("got %d, expected %d", resp.StatusCode, t.ExpectCode)
        c.record(t.Name, result)
        return result
    }

    if t.ExpectBody != "" {
        body, _ := io.ReadAll(io.LimitReader(resp.Body, 512*1024))
        if !strings.Contains(string(body), t.ExpectBody) {
            result.Status = "down"
            result.Error = "missing expected body content"
            c.record(t.Name, result)
            return result
        }
    }

    result.Status = "up"
    c.record(t.Name, result)
    return result
}

func (c *Checker) record(name string, r Result) {
    c.mu.Lock()
    defer c.mu.Unlock()

    // Detect state transitions for alerting
    prev := c.latest[name]
    if prev != nil && prev.Status == "up" && r.Status == "down" {
        c.alerts <- r
    }

    c.latest[name] = &r
    c.history[name] = append(c.history[name], r)
    if len(c.history[name]) > 200 {
        c.history[name] = c.history[name][100:]
    }
}

func (c *Checker) Uptime(name string, window time.Duration) float64 {
    c.mu.RLock()
    defer c.mu.RUnlock()
    cutoff := time.Now().Add(-window)
    var total, up int
    for _, r := range c.history[name] {
        if r.Timestamp.After(cutoff) {
            total++
            if r.Status == "up" {
                up++
            }
        }
    }
    if total == 0 {
        return 100.0
    }
    return float64(up) / float64(total) * 100
}

Prometheus Metrics and Alerting

package main

import (
    "encoding/json"
    "fmt"
    "log"
    "net/http"
    "time"
    "myapp/monitor"
)

func main() {
    checker := monitor.NewChecker()

    // Background check loop
    go func() {
        for {
            for _, t := range monitor.Targets {
                r := checker.Check(t)
                if r.Status == "down" {
                    log.Printf("DOWN: %s (%s) - %s", r.Name, r.Region, r.Error)
                }
            }
            time.Sleep(5 * time.Minute)
        }
    }()

    // Alert consumer
    go func() {
        for alert := range checker.Alerts() {
            log.Printf("ALERT: %s went DOWN at %s - %s",
                alert.Name, alert.Timestamp.Format(time.RFC3339), alert.Error)
            // Send to Slack, email, etc.
        }
    }()

    // Prometheus metrics
    http.HandleFunc("/metrics", func(w http.ResponseWriter, r *http.Request) {
        w.Header().Set("Content-Type", "text/plain; version=0.0.4")
        for _, t := range monitor.Targets {
            result := checker.Check(t)
            up := 0
            if result.Status == "up" { up = 1 }
            uptime := checker.Uptime(t.Name, 24*time.Hour)
            fmt.Fprintf(w, "vvv_endpoint_up{name=%q,region=%q} %d
", t.Name, t.Region, up)
            fmt.Fprintf(w, "vvv_endpoint_latency_ms{name=%q} %.0f
", t.Name, float64(result.Latency.Milliseconds()))
            fmt.Fprintf(w, "vvv_endpoint_uptime_24h{name=%q} %.2f
", t.Name, uptime)
        }
    })

    // Status JSON endpoint
    http.HandleFunc("/status", func(w http.ResponseWriter, r *http.Request) {
        w.Header().Set("Content-Type", "application/json")
        status := map[string]interface{}{}
        for _, t := range monitor.Targets {
            status[t.Name] = map[string]interface{}{
                "status":  checker.Latest(t.Name).Status,
                "latency": checker.Latest(t.Name).Latency.String(),
                "uptime":  fmt.Sprintf("%.1f%%", checker.Uptime(t.Name, 24*time.Hour)),
            }
        }
        json.NewEncoder(w).Encode(status)
    })

    log.Println("Monitor on :9090")
    log.Fatal(http.ListenAndServe(":9090", nil))
}

This monitor runs as a single Go binary using under 15MB RAM, checking all ViralVidVault endpoints every 5 minutes. The state transition alerting means we only get notified when something changes from up to down -- no alert fatigue from repeated "still down" messages.

This article is part of the Building ViralVidVault series. Check out ViralVidVault to see these techniques in action.

Building a REST API Client Library in Python for Video Data

ahmet gedik — Mon, 11 May 2026 06:00:01 +0000

Why Build a Client Library

When your video platform talks to the YouTube Data API from multiple scripts -- the cron fetcher, the metadata enricher, the broken link checker -- you end up duplicating HTTP logic everywhere. At DailyWatch, we consolidated all YouTube API interactions into a single Python SDK with retry logic, rate limiting, and caching built in.

The Core Client

import time
import hashlib
import json
import requests
from typing import Optional
from dataclasses import dataclass, field
from functools import lru_cache

@dataclass
class APIConfig:
    api_key: str
    base_url: str = "https://www.googleapis.com/youtube/v3"
    max_retries: int = 3
    base_delay: float = 1.0
    timeout: int = 15
    daily_quota: int = 10000
    cache_ttl: int = 3600  # 1 hour

class QuotaExhaustedError(Exception):
    pass

class YouTubeClient:
    """Reusable YouTube Data API client with retry, rate limiting, and caching."""

    def __init__(self, config: APIConfig):
        self.config = config
        self.session = requests.Session()
        self.session.headers.update({"Accept": "application/json"})
        self._quota_used = 0
        self._cache: dict[str, tuple[float, dict]] = {}

    def _cache_key(self, endpoint: str, params: dict) -> str:
        raw = f"{endpoint}:{json.dumps(params, sort_keys=True)}"
        return hashlib.md5(raw.encode()).hexdigest()

    def _get_cached(self, key: str) -> Optional[dict]:
        if key in self._cache:
            timestamp, data = self._cache[key]
            if time.time() - timestamp < self.config.cache_ttl:
                return data
            del self._cache[key]
        return None

    def _request(self, endpoint: str, params: dict, quota_cost: int = 1) -> dict:
        """Make an API request with retry, backoff, caching, and quota tracking."""
        # Check quota
        if self._quota_used + quota_cost > self.config.daily_quota:
            raise QuotaExhaustedError(
                f"Would exceed daily quota ({self._quota_used}/{self.config.daily_quota})"
            )

        # Check cache
        cache_key = self._cache_key(endpoint, params)
        cached = self._get_cached(cache_key)
        if cached is not None:
            return cached

        url = f"{self.config.base_url}/{endpoint}"
        params["key"] = self.config.api_key
        last_exception = None

        for attempt in range(self.config.max_retries):
            try:
                response = self.session.get(url, params=params, timeout=self.config.timeout)

                if response.status_code == 200:
                    data = response.json()
                    self._quota_used += quota_cost
                    self._cache[cache_key] = (time.time(), data)
                    return data

                if response.status_code == 403:
                    error_reason = response.json().get("error", {}).get("errors", [{}])[0].get("reason", "")
                    if error_reason == "quotaExceeded":
                        raise QuotaExhaustedError("YouTube API quota exceeded")

                if response.status_code in (429, 500, 503):
                    delay = self.config.base_delay * (2 ** attempt)
                    time.sleep(delay)
                    continue

                response.raise_for_status()

            except requests.ConnectionError as e:
                last_exception = e
                delay = self.config.base_delay * (2 ** attempt)
                time.sleep(delay)

        raise last_exception or RuntimeError(f"Failed after {self.config.max_retries} retries")

    @property
    def quota_remaining(self) -> int:
        return self.config.daily_quota - self._quota_used

High-Level Methods

Build domain-specific methods on top of the core:

    def get_trending(self, region: str = "US", max_results: int = 25, category_id: str = "") -> list[dict]:
        """Fetch trending videos for a region."""
        params = {
            "part": "snippet,statistics,contentDetails",
            "chart": "mostPopular",
            "regionCode": region,
            "maxResults": min(max_results, 50),
        }
        if category_id:
            params["videoCategoryId"] = category_id

        data = self._request("videos", params)
        return data.get("items", [])

    def get_video_details(self, video_ids: list[str]) -> list[dict]:
        """Batch fetch video details (up to 50 per call)."""
        results = []
        for i in range(0, len(video_ids), 50):
            batch = video_ids[i:i + 50]
            params = {
                "part": "snippet,statistics,contentDetails",
                "id": ",".join(batch),
            }
            data = self._request("videos", params)
            results.extend(data.get("items", []))
        return results

    def search_videos(self, query: str, region: str = "US", max_results: int = 10) -> list[dict]:
        """Search for videos. Costs 100 quota units per call."""
        params = {
            "part": "snippet",
            "q": query,
            "type": "video",
            "regionCode": region,
            "maxResults": min(max_results, 50),
        }
        data = self._request("search", params, quota_cost=100)
        return data.get("items", [])

    def get_categories(self, region: str = "US") -> dict[str, str]:
        """Get video category ID to name mapping."""
        params = {
            "part": "snippet",
            "regionCode": region,
        }
        data = self._request("videoCategories", params)
        return {
            item["id"]: item["snippet"]["title"]
            for item in data.get("items", [])
        }

Usage in Practice

# Initialize once, use everywhere
config = APIConfig(api_key="AIzaSy...")
client = YouTubeClient(config)

# Fetch trending from multiple regions
for region in ["US", "GB", "DE", "FR", "IN", "BR", "AU", "CA"]:
    videos = client.get_trending(region=region, max_results=25)
    print(f"[{region}] {len(videos)} trending videos (quota left: {client.quota_remaining})")

# Batch details (uses cache if recently fetched)
details = client.get_video_details(["dQw4w9WgXcQ", "jNQXAC9IVRw"])

# Search (expensive -- 100 quota units)
results = client.search_videos("python tutorial", region="US")

Key Design Decisions

Exponential backoff: Each retry doubles the wait time, preventing thundering herd
In-memory cache: Avoids redundant API calls within the same run
Quota tracking: Proactively prevents quota exhaustion instead of discovering it mid-pipeline
Batch-aware: The get_video_details method automatically chunks large ID lists

This client powers every API interaction at DailyWatch. Having retry logic and quota awareness in one place eliminated an entire class of intermittent failures from our pipeline.

This article is part of the Building DailyWatch series. Check out DailyWatch to see these techniques in action.

Monitoring and Alerting for Video Platform Infrastructure with Go

ahmet gedik — Sun, 10 May 2026 23:00:01 +0000

Monitoring Across Time Zones

TrendVidStream serves UAE (+4 UTC), Finland (+2), Czech Republic (+1), Denmark (+1), Belgium (+1), UK (+0), Switzerland (+1), and US (-5 to -8). A cron failure in Finland at 3am local is 1am UTC — monitoring must alert before users notice stale content.

Health Check Endpoint

package main

import (
    "database/sql"
    "encoding/json"
    "net/http"
    "time"
)

type HealthResponse struct {
    Overall string            `json:"overall"`
    Checks  map[string]string `json:"checks"`
    Uptime  string            `json:"uptime"`
    Regions []string          `json:"regions_active"`
}

var (
    startTime     = time.Now()
    activeRegions = []string{"AE", "FI", "CZ", "DK", "BE", "CH", "GB", "US"}
)

func healthHandler(db *sql.DB) http.HandlerFunc {
    return func(w http.ResponseWriter, r *http.Request) {
        checks := map[string]string{}
        overall := "ok"

        if err := db.Ping(); err != nil {
            checks["database"] = "error: " + err.Error()
            overall = "degraded"
        } else {
            checks["database"] = "ok"
        }

        for _, region := range activeRegions {
            var count int
            db.QueryRow(
                "SELECT COUNT(*) FROM videos WHERE region=? AND fetched_at > datetime('now','-8 hours')",
                region,
            ).Scan(&count)
            if count == 0 {
                checks["region_"+region] = "stale"
                overall = "degraded"
            } else {
                checks["region_"+region] = "ok"
            }
        }

        w.Header().Set("Content-Type", "application/json")
        if overall != "ok" {
            w.WriteHeader(http.StatusServiceUnavailable)
        }
        json.NewEncoder(w).Encode(HealthResponse{
            Overall: overall, Checks: checks,
            Uptime: time.Since(startTime).Round(time.Second).String(),
            Regions: activeRegions,
        })
    }
}

Prometheus Exporter

import "github.com/prometheus/client_golang/prometheus"

var (
    fetchedTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
        Name: "tvs_videos_fetched_total",
        Help: "Total videos fetched per region.",
    }, []string{"region"})

    fetchDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{
        Name: "tvs_fetch_duration_seconds", Help: "Fetch duration per region.",
        Buckets: prometheus.DefBuckets,
    }, []string{"region"})

    regionFreshness = prometheus.NewGaugeVec(prometheus.GaugeOpts{
        Name: "tvs_region_freshness_seconds",
        Help: "Seconds since last successful fetch per region.",
    }, []string{"region"})
)

func init() { prometheus.MustRegister(fetchedTotal, fetchDuration, regionFreshness) }

Uptime Monitor with Per-Timezone Context

var probes = []struct {
    URL    string
    Region string
    UTCOff int
}{
    {"https://trendvidstream.com/?region=AE", "AE", +4},
    {"https://trendvidstream.com/?region=FI", "FI", +2},
    {"https://trendvidstream.com/?region=CZ", "CZ", +1},
    {"https://trendvidstream.com/?region=DK", "DK", +1},
    {"https://trendvidstream.com/?region=GB", "GB",  0},
}

func (m *UptimeMonitor) Start(interval time.Duration) {
    for _, p := range probes {
        go func(probe struct{ URL, Region string; UTCOff int }) {
            ticker := time.NewTicker(interval)
            for range ticker.C {
                start := time.Now()
                resp, err := m.client.Get(probe.URL)
                latency := time.Since(start)

                status := 0
                if resp != nil { status = resp.StatusCode; resp.Body.Close() }

                if err != nil || status >= 500 {
                    localHour := (time.Now().UTC().Hour() + probe.UTCOff + 24) % 24
                    msg := fmt.Sprintf(
                        ":red_circle: *%s DOWN* (HTTP %d, %.0fms) — local %02d:xx",
                        probe.Region, status, latency.Seconds()*1000, localHour,
                    )
                    sendSlack(m.slack, msg)
                }
            }
        }(p)
    }
}

Including local time in alerts helps on-call engineers understand whether the affected region is in peak hours — a UAE failure at 21:00 local (prime time) is more urgent than a Finnish failure at 04:00.

Grafana Dashboard Queries

Panel	PromQL
Videos fetched rate	`rate(tvs_videos_fetched_total[5m])` per region
P95 fetch latency	`histogram_quantile(0.95, rate(tvs_fetch_duration_seconds_bucket[5m]))`
Region freshness	`tvs_region_freshness_seconds` — alert if > 28800 (8h)
Quota remaining	`10000 - sum(tvs_api_quota_used)`

The freshness alert is the most valuable metric. When a region goes stale due to API quota exhaustion, Grafana pages on-call before any user reports missing content on TrendVidStream.

This article is part of the Building TrendVidStream series. Check out TrendVidStream to see these techniques in action.

Building a REST API Client Library in Python for Video Data

ahmet gedik — Sun, 10 May 2026 11:00:01 +0000

Centralizing API Logic

When you fetch viral videos from 7 European regions, you end up calling the YouTube Data API from multiple scripts: the main fetcher, the virality scorer, the metadata enricher. At ViralVidVault, we built a Python client library that encapsulates retry logic, quota tracking, and region-aware caching.

The Client Core

import time
import json
import hashlib
import requests
from dataclasses import dataclass
from typing import Optional

@dataclass
class ClientConfig:
    api_keys: list[str]  # Multiple keys for rotation
    base_url: str = "https://www.googleapis.com/youtube/v3"
    max_retries: int = 3
    timeout: int = 15
    quota_per_key: int = 10000
    cache_ttl: int = 1800  # 30 min

class QuotaExhaustedError(Exception):
    pass

class YouTubeSDK:
    """Multi-key YouTube client with automatic key rotation and caching."""

    def __init__(self, config: ClientConfig):
        self.config = config
        self.session = requests.Session()
        self._key_index = 0
        self._quota_used: dict[str, int] = {k: 0 for k in config.api_keys}
        self._cache: dict[str, tuple[float, dict]] = {}

    @property
    def _current_key(self) -> str:
        return self.config.api_keys[self._key_index]

    def _rotate_key(self) -> bool:
        """Switch to the next available API key. Returns False if all exhausted."""
        original = self._key_index
        while True:
            self._key_index = (self._key_index + 1) % len(self.config.api_keys)
            if self._key_index == original:
                return False  # All keys exhausted
            if self._quota_used[self._current_key] < self.config.quota_per_key:
                return True

    def _cache_get(self, key: str) -> Optional[dict]:
        if key in self._cache:
            ts, data = self._cache[key]
            if time.time() - ts < self.config.cache_ttl:
                return data
            del self._cache[key]
        return None

    def _request(self, endpoint: str, params: dict, cost: int = 1) -> dict:
        cache_key = hashlib.md5(f"{endpoint}:{json.dumps(params, sort_keys=True)}".encode()).hexdigest()
        cached = self._cache_get(cache_key)
        if cached:
            return cached

        last_error = None
        for attempt in range(self.config.max_retries):
            if self._quota_used[self._current_key] + cost > self.config.quota_per_key:
                if not self._rotate_key():
                    raise QuotaExhaustedError("All API keys exhausted")

            params["key"] = self._current_key
            try:
                resp = self.session.get(
                    f"{self.config.base_url}/{endpoint}",
                    params=params,
                    timeout=self.config.timeout,
                )

                if resp.status_code == 200:
                    data = resp.json()
                    self._quota_used[self._current_key] += cost
                    self._cache[cache_key] = (time.time(), data)
                    return data

                if resp.status_code == 403:
                    # Quota exceeded for this key, rotate
                    self._quota_used[self._current_key] = self.config.quota_per_key
                    if not self._rotate_key():
                        raise QuotaExhaustedError("All keys quota exceeded")
                    continue

                if resp.status_code in (429, 500, 503):
                    time.sleep(2 ** attempt)
                    continue

                resp.raise_for_status()

            except requests.ConnectionError as e:
                last_error = e
                time.sleep(2 ** attempt)

        raise last_error or RuntimeError(f"Failed after {self.config.max_retries} attempts")

European Region Methods

Build domain-specific methods for multi-region fetching:

    EU_REGIONS = ["PL", "NL", "SE", "NO", "AT", "GB", "US"]

    def get_trending(self, region: str = "GB", max_results: int = 25) -> list[dict]:
        params = {
            "part": "snippet,statistics,contentDetails",
            "chart": "mostPopular",
            "regionCode": region,
            "maxResults": min(max_results, 50),
        }
        return self._request("videos", params).get("items", [])

    def get_trending_all_regions(self, max_per_region: int = 25) -> dict[str, list]:
        """Fetch trending from all European regions."""
        results = {}
        for region in self.EU_REGIONS:
            try:
                results[region] = self.get_trending(region, max_per_region)
            except QuotaExhaustedError:
                print(f"Quota exhausted at region {region}")
                break
        return results

    def get_video_batch(self, video_ids: list[str]) -> list[dict]:
        """Fetch details for up to 50 videos per call."""
        all_items = []
        for i in range(0, len(video_ids), 50):
            batch = video_ids[i:i + 50]
            params = {
                "part": "snippet,statistics,contentDetails",
                "id": ",".join(batch),
            }
            data = self._request("videos", params)
            all_items.extend(data.get("items", []))
        return all_items

    @property
    def total_quota_remaining(self) -> int:
        return sum(
            self.config.quota_per_key - used
            for used in self._quota_used.values()
        )

    def quota_report(self) -> dict:
        return {
            key[:8] + "...": {
                "used": self._quota_used[key],
                "remaining": self.config.quota_per_key - self._quota_used[key],
            }
            for key in self.config.api_keys
        }

Usage

config = ClientConfig(
    api_keys=["AIzaSy_key1", "AIzaSy_key2", "AIzaSy_key3"],
)
sdk = YouTubeSDK(config)

# Fetch all European regions
all_trending = sdk.get_trending_all_regions(max_per_region=25)
for region, videos in all_trending.items():
    print(f"[{region}] {len(videos)} videos")

# Check quota health
print(sdk.quota_report())
# {'AIzaSy_k...': {'used': 14, 'remaining': 9986}, ...}

Key Design Decisions

Multi-key rotation: When one key hits its quota, automatically switch to the next
Region-aware caching: Same region + same params = cache hit, avoiding duplicate calls
Cost-aware requests: Search costs 100 units vs 1 for videos.list
Graceful degradation: Partial results are better than a crash when quota runs out

This SDK powers every YouTube API interaction at ViralVidVault. The key rotation alone saved us from quota exhaustion during peak viral detection periods.

This article is part of the Building ViralVidVault series. Check out ViralVidVault to see these techniques in action.

Monitoring and Alerting for Video Platform Infrastructure with Go

ahmet gedik — Sun, 10 May 2026 01:00:00 +0000

Why Roll Your Own Monitoring?

Managed monitoring like Datadog is excellent — and expensive. For a self-hosted video platform like TopVideoHub, a 200-line Go binary can cover 90% of what you need: uptime checks, latency histograms, error rate alerts, and a Grafana dashboard wired to Prometheus.

Health Check Endpoints

package main

import (
    "database/sql"
    "encoding/json"
    "net/http"
    "time"
)

type HealthStatus struct {
    Status  string            `json:"status"`
    Checks  map[string]string `json:"checks"`
    Uptime  string            `json:"uptime"`
    Ts      time.Time         `json:"ts"`
}

var startTime = time.Now()

func healthHandler(db *sql.DB) http.HandlerFunc {
    return func(w http.ResponseWriter, r *http.Request) {
        checks := map[string]string{}
        overall := "ok"

        if err := db.Ping(); err != nil {
            checks["database"] = "error: " + err.Error()
            overall = "degraded"
        } else {
            checks["database"] = "ok"
        }

        status := HealthStatus{
            Status: overall,
            Checks: checks,
            Uptime: time.Since(startTime).Round(time.Second).String(),
            Ts:     time.Now(),
        }

        w.Header().Set("Content-Type", "application/json")
        if overall != "ok" {
            w.WriteHeader(http.StatusServiceUnavailable)
        }
        json.NewEncoder(w).Encode(status)
    }
}

Prometheus Metrics Exporter

import "github.com/prometheus/client_golang/prometheus"

var (
    videosFetched = prometheus.NewCounterVec(prometheus.CounterOpts{
        Name: "topvideohub_videos_fetched_total",
        Help: "Total videos fetched.",
    }, []string{"region"})

    fetchDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{
        Name:    "topvideohub_fetch_duration_seconds",
        Help:    "Fetch duration per region.",
        Buckets: prometheus.DefBuckets,
    }, []string{"region"})

    regionFreshness = prometheus.NewGaugeVec(prometheus.GaugeOpts{
        Name: "topvideohub_region_freshness_seconds",
        Help: "Seconds since last successful fetch per region.",
    }, []string{"region"})
)

func init() {
    prometheus.MustRegister(videosFetched, fetchDuration, regionFreshness)
}

Instrument the fetch loop:

func fetchWithMetrics(region string, fn func() (int, error)) {
    start := time.Now()
    count, err := fn()
    duration := time.Since(start).Seconds()

    fetchDuration.WithLabelValues(region).Observe(duration)
    if err == nil {
        videosFetched.WithLabelValues(region).Add(float64(count))
    }
}

Uptime Monitor

type UptimeMonitor struct {
    client *http.Client
    slack  string
}

func (m *UptimeMonitor) Watch(targets []string, interval time.Duration) {
    for _, target := range targets {
        go func(t string) {
            ticker := time.NewTicker(interval)
            for range ticker.C {
                start := time.Now()
                resp, err := m.client.Get(t)
                latency := time.Since(start)

                status := 0
                if resp != nil {
                    status = resp.StatusCode
                    resp.Body.Close()
                }
                if err != nil || status >= 500 {
                    msg := fmt.Sprintf(":red_circle: %s DOWN (HTTP %d, %v)", t, status, latency)
                    sendSlackWebhook(m.slack, msg)
                }
            }
        }(target)
    }
}

Watch all regional URLs for TopVideoHub:

monitor := &UptimeMonitor{client: &http.Client{Timeout: 10 * time.Second}, slack: os.Getenv("SLACK_WEBHOOK")}
monitor.Watch([]string{
    "https://topvideohub.com/health",
    "https://topvideohub.com/?region=JP",
    "https://topvideohub.com/?region=KR",
    "https://topvideohub.com/?region=TW",
}, 60*time.Second)

Grafana Dashboard

prometheus.yml:

scrape_configs:
  - job_name: 'topvideohub'
    static_configs:
      - targets: ['localhost:9090']
    scrape_interval: 30s

Key Grafana panels:

Panel	Query
Videos fetched / region	`rate(topvideohub_videos_fetched_total[5m])`
P95 fetch latency	`histogram_quantile(0.95, rate(topvideohub_fetch_duration_seconds_bucket[5m]))`
Region freshness	`topvideohub_region_freshness_seconds` (alert if > 28800)

With these panels, you can spot quota exhaustion before it causes stale data and detect slow region fetches — all without logging into the server.

This article is part of the Building TopVideoHub series. Check out TopVideoHub to see these techniques in action.

Database Sharding Strategies for Growing Video Platforms

ahmet gedik — Sat, 09 May 2026 18:00:01 +0000

When a Single Database Isn't Enough

At DailyWatch, we aggregate trending videos from 8 regions with a cron pipeline running every 2 hours. The videos table grows by thousands of rows daily. At some point, queries slow down, indexes bloat, and backups take too long. That's when you start thinking about sharding.

This article covers three sharding strategies we evaluated, with real PostgreSQL examples.

Strategy 1: Range Partitioning by Date

The most natural fit for time-series video data. PostgreSQL's declarative partitioning handles this natively:

-- Parent table (no data stored directly here)
CREATE TABLE videos (
    id BIGSERIAL,
    video_id VARCHAR(16) NOT NULL,
    title TEXT NOT NULL,
    view_count BIGINT DEFAULT 0,
    region VARCHAR(4) NOT NULL,
    category_id INTEGER,
    fetched_at TIMESTAMP NOT NULL DEFAULT NOW(),
    PRIMARY KEY (id, fetched_at)
) PARTITION BY RANGE (fetched_at);

-- Monthly partitions
CREATE TABLE videos_2026_01 PARTITION OF videos
    FOR VALUES FROM ('2026-01-01') TO ('2026-02-01');

CREATE TABLE videos_2026_02 PARTITION OF videos
    FOR VALUES FROM ('2026-02-01') TO ('2026-03-01');

CREATE TABLE videos_2026_03 PARTITION OF videos
    FOR VALUES FROM ('2026-03-01') TO ('2026-04-01');

-- Each partition gets its own indexes automatically
CREATE INDEX ON videos_2026_01 (video_id);
CREATE INDEX ON videos_2026_02 (video_id);
CREATE INDEX ON videos_2026_03 (video_id);

The query planner automatically prunes partitions:

-- Only scans videos_2026_03, skips others entirely
EXPLAIN ANALYZE
SELECT * FROM videos
WHERE fetched_at >= '2026-03-01' AND fetched_at < '2026-03-15';

Automating Partition Creation

Don't create partitions manually. Use a function:

CREATE OR REPLACE FUNCTION create_monthly_partition(target_date DATE)
RETURNS TEXT AS $$
DECLARE
    partition_name TEXT;
    start_date DATE;
    end_date DATE;
BEGIN
    start_date := DATE_TRUNC('month', target_date);
    end_date := start_date + INTERVAL '1 month';
    partition_name := 'videos_' || TO_CHAR(start_date, 'YYYY_MM');

    -- Check if partition already exists
    IF NOT EXISTS (
        SELECT 1 FROM pg_class WHERE relname = partition_name
    ) THEN
        EXECUTE format(
            'CREATE TABLE %I PARTITION OF videos FOR VALUES FROM (%L) TO (%L)',
            partition_name, start_date, end_date
        );
        EXECUTE format(
            'CREATE INDEX ON %I (video_id)',
            partition_name
        );
        RETURN 'Created: ' || partition_name;
    END IF;

    RETURN 'Already exists: ' || partition_name;
END;
$$ LANGUAGE plpgsql;

-- Call before each month
SELECT create_monthly_partition('2026-04-01');

Strategy 2: List Partitioning by Region

When queries almost always filter by region, partition on that dimension:

CREATE TABLE videos_by_region (
    id BIGSERIAL,
    video_id VARCHAR(16) NOT NULL,
    title TEXT NOT NULL,
    view_count BIGINT DEFAULT 0,
    region VARCHAR(4) NOT NULL,
    fetched_at TIMESTAMP NOT NULL DEFAULT NOW(),
    PRIMARY KEY (id, region)
) PARTITION BY LIST (region);

CREATE TABLE videos_americas PARTITION OF videos_by_region
    FOR VALUES IN ('US', 'BR', 'CA');

CREATE TABLE videos_europe PARTITION OF videos_by_region
    FOR VALUES IN ('GB', 'DE', 'FR');

CREATE TABLE videos_apac PARTITION OF videos_by_region
    FOR VALUES IN ('IN', 'AU');

This shines when your application logic is already region-aware. At DailyWatch, each region's trending page only queries its own partition.

Strategy 3: Application-Level Sharding

When PostgreSQL partitioning isn't enough (different databases on different servers), shard at the application level:

import hashlib
from typing import Dict
import asyncpg

class VideoShardRouter:
    """Routes video queries to the correct database shard."""

    def __init__(self, shard_configs: Dict[str, str]):
        # shard_configs: {"shard_0": "postgres://...", "shard_1": "postgres://...", ...}
        self.shard_count = len(shard_configs)
        self.pools: Dict[str, asyncpg.Pool] = {}
        self.configs = shard_configs

    async def initialize(self):
        for name, dsn in self.configs.items():
            self.pools[name] = await asyncpg.create_pool(dsn, min_size=2, max_size=10)

    def get_shard(self, video_id: str) -> str:
        """Consistent hash-based routing."""
        hash_val = int(hashlib.md5(video_id.encode()).hexdigest(), 16)
        shard_idx = hash_val % self.shard_count
        return f"shard_{shard_idx}"

    async def insert_video(self, video_id: str, title: str, region: str, views: int):
        shard = self.get_shard(video_id)
        pool = self.pools[shard]
        await pool.execute(
            "INSERT INTO videos (video_id, title, region, view_count) VALUES ($1, $2, $3, $4) "
            "ON CONFLICT (video_id) DO UPDATE SET view_count = $4",
            video_id, title, region, views,
        )

    async def get_video(self, video_id: str) -> dict:
        shard = self.get_shard(video_id)
        pool = self.pools[shard]
        row = await pool.fetchrow(
            "SELECT * FROM videos WHERE video_id = $1", video_id
        )
        return dict(row) if row else None

    async def search_all_shards(self, query: str, limit: int = 20) -> list:
        """Fan-out query across all shards, merge results."""
        import asyncio
        tasks = [
            pool.fetch(
                "SELECT * FROM videos WHERE title ILIKE $1 ORDER BY view_count DESC LIMIT $2",
                f"%{query}%", limit,
            )
            for pool in self.pools.values()
        ]
        results = await asyncio.gather(*tasks)
        merged = [dict(row) for shard_rows in results for row in shard_rows]
        merged.sort(key=lambda v: v["view_count"], reverse=True)
        return merged[:limit]

Which Strategy to Choose?

Strategy	Best When	Complexity	DailyWatch Fit
Date partitioning	Time-series queries dominate	Low	Good for archiving
Region partitioning	Region-scoped queries	Low	Great for trending pages
App-level sharding	Multiple DB servers needed	High	Future-proofing

For most video platforms under 10M rows, PostgreSQL native partitioning is the right call. You get automatic partition pruning, simpler ops, and no application code changes. App-level sharding is the nuclear option for when you've outgrown a single server.

Key Takeaway

Start with date-based partitioning. It's the lowest-effort, highest-impact sharding strategy for content platforms where "recent" is always the most-queried dimension. You can always add region partitioning as a sub-partition later.

This article is part of the Building DailyWatch series. Check out DailyWatch to see these techniques in action.

Database Sharding Strategies for Growing Video Platforms

ahmet gedik — Sat, 09 May 2026 09:00:01 +0000

Sharding Is Not Your First Problem

Most video platforms can serve millions of rows from a single PostgreSQL instance with proper indexing. Design your schema to be shardable from day one, but shard only when necessary.

TrendVidStream serves 8 regions — UAE, Finland, Czech Republic, Denmark, Belgium, UK, Switzerland, US. The natural partition keys are region and fetched_at.

Strategy 1: PostgreSQL List Partitioning by Region

CREATE TABLE videos (
    id           BIGSERIAL,
    video_id     TEXT NOT NULL,
    title        TEXT NOT NULL,
    region       TEXT NOT NULL,
    category_id  INT,
    view_count   BIGINT DEFAULT 0,
    fetched_at   TIMESTAMPTZ DEFAULT NOW(),
    embed_broken BOOLEAN DEFAULT FALSE,
    title_lang   TEXT,
    PRIMARY KEY (id, region)
) PARTITION BY LIST (region);

CREATE TABLE videos_ae PARTITION OF videos FOR VALUES IN ('AE');
CREATE TABLE videos_fi PARTITION OF videos FOR VALUES IN ('FI');
CREATE TABLE videos_cz PARTITION OF videos FOR VALUES IN ('CZ');
CREATE TABLE videos_dk PARTITION OF videos FOR VALUES IN ('DK');
CREATE TABLE videos_be PARTITION OF videos FOR VALUES IN ('BE');
CREATE TABLE videos_ch PARTITION OF videos FOR VALUES IN ('CH');
CREATE TABLE videos_gb PARTITION OF videos FOR VALUES IN ('GB');
CREATE TABLE videos_us PARTITION OF videos FOR VALUES IN ('US');
CREATE TABLE videos_other PARTITION OF videos DEFAULT;

-- Indexes propagate to all partitions automatically
CREATE INDEX idx_videos_views ON videos (view_count DESC);
CREATE INDEX idx_videos_fetched ON videos (fetched_at DESC);

Strategy 2: Sub-Partition by Date for Event Data

CREATE TABLE video_events (
    id       BIGSERIAL,
    video_id TEXT NOT NULL,
    region   TEXT NOT NULL,
    event    TEXT NOT NULL,
    occurred TIMESTAMPTZ DEFAULT NOW()
) PARTITION BY LIST (region);

CREATE TABLE video_events_ae PARTITION OF video_events
    FOR VALUES IN ('AE')
    PARTITION BY RANGE (occurred);

CREATE TABLE video_events_ae_2026q1 PARTITION OF video_events_ae
    FOR VALUES FROM ('2026-01-01') TO ('2026-04-01');

CREATE TABLE video_events_ae_2026q2 PARTITION OF video_events_ae
    FOR VALUES FROM ('2026-04-01') TO ('2026-07-01');

Strategy 3: Application-Level Shard Routing

import asyncpg

SHARD_MAP = {
    'AE': 'postgresql://user:pass@db-mena/trendvidstream',
    'GB': 'postgresql://user:pass@db-eu-west/trendvidstream',
    'BE': 'postgresql://user:pass@db-eu-west/trendvidstream',
    'CH': 'postgresql://user:pass@db-eu-central/trendvidstream',
    'CZ': 'postgresql://user:pass@db-eu-central/trendvidstream',
    'DK': 'postgresql://user:pass@db-eu-north/trendvidstream',
    'FI': 'postgresql://user:pass@db-eu-north/trendvidstream',
    'US': 'postgresql://user:pass@db-us/trendvidstream',
}

_pools: dict[str, asyncpg.Pool] = {}

async def shard_pool(region: str) -> asyncpg.Pool:
    dsn = SHARD_MAP.get(region, SHARD_MAP['US'])
    if dsn not in _pools:
        _pools[dsn] = await asyncpg.create_pool(dsn, min_size=2, max_size=10)
    return _pools[dsn]

async def trending(region: str, limit: int = 50) -> list[dict]:
    pool = await shard_pool(region)
    async with pool.acquire() as conn:
        rows = await conn.fetch(
            'SELECT video_id, title, thumbnail_url, view_count FROM videos '
            'WHERE region = $1 ORDER BY view_count DESC LIMIT $2', region, limit,
        )
    return [dict(r) for r in rows]

Cross-Shard Global Trending

import asyncio

async def global_trending(limit: int = 20) -> list[dict]:
    regions = list(dict.fromkeys(SHARD_MAP.keys()))
    per_shard = await asyncio.gather(*[trending(r, limit) for r in regions],
                                    return_exceptions=True)
    merged = []
    for result in per_shard:
        if isinstance(result, list):
            merged.extend(result)
    merged.sort(key=lambda v: v['view_count'], reverse=True)
    return merged[:limit]

For TrendVidStream, the global trending page completes in 12ms at current scale. The shard routing layer is ready for when that grows.

Migration Checklist

Add region to every table's primary key.
Add PostgreSQL list partitions (ATTACH PARTITION is zero-downtime).
Move hot partitions to read replicas as read load grows.
Introduce shard routing when write throughput demands separate hosts.
Never shard prematurely — each step adds operational complexity.

This article is part of the Building TrendVidStream series. Check out TrendVidStream to see these techniques in action.

Database Sharding Strategies for Growing Video Platforms

ahmet gedik — Sat, 09 May 2026 01:00:01 +0000

Scaling a European Video Database

ViralVidVault stores viral video data from 7 European regions. As the database grows, queries that scan the entire videos table get slower. Sharding -- splitting data across multiple tables or databases -- is how you keep query performance flat as data grows linearly.

Here are three strategies, with concrete PostgreSQL implementations.

Strategy 1: List Partitioning by Region

The most natural fit for a multi-region video vault. Each European region gets its own partition:

CREATE TABLE videos (
    id BIGSERIAL,
    video_id VARCHAR(16) NOT NULL,
    title TEXT NOT NULL,
    view_count BIGINT DEFAULT 0,
    virality_score FLOAT DEFAULT 0,
    region VARCHAR(4) NOT NULL,
    category_id INTEGER,
    fetched_at TIMESTAMP NOT NULL DEFAULT NOW(),
    PRIMARY KEY (id, region)
) PARTITION BY LIST (region);

-- One partition per tracked region
CREATE TABLE videos_pl PARTITION OF videos FOR VALUES IN ('PL');
CREATE TABLE videos_nl PARTITION OF videos FOR VALUES IN ('NL');
CREATE TABLE videos_se PARTITION OF videos FOR VALUES IN ('SE');
CREATE TABLE videos_no PARTITION OF videos FOR VALUES IN ('NO');
CREATE TABLE videos_at PARTITION OF videos FOR VALUES IN ('AT');
CREATE TABLE videos_gb PARTITION OF videos FOR VALUES IN ('GB');
CREATE TABLE videos_us PARTITION OF videos FOR VALUES IN ('US');

-- Default partition for any unexpected regions
CREATE TABLE videos_other PARTITION OF videos DEFAULT;

-- Each partition gets its own index
CREATE INDEX ON videos_pl (virality_score DESC);
CREATE INDEX ON videos_nl (virality_score DESC);
CREATE INDEX ON videos_se (virality_score DESC);
CREATE INDEX ON videos_no (virality_score DESC);
CREATE INDEX ON videos_at (virality_score DESC);
CREATE INDEX ON videos_gb (virality_score DESC);
CREATE INDEX ON videos_us (virality_score DESC);

Now region-scoped queries are lightning fast:

-- Only scans videos_pl, not the other 6 partitions
SELECT * FROM videos
WHERE region = 'PL' AND virality_score >= 85
ORDER BY virality_score DESC
LIMIT 20;

Strategy 2: Composite Partitioning (Region + Date)

For even finer granularity, partition by region first, then by date:

CREATE TABLE videos_partitioned (
    id BIGSERIAL,
    video_id VARCHAR(16) NOT NULL,
    title TEXT NOT NULL,
    virality_score FLOAT DEFAULT 0,
    region VARCHAR(4) NOT NULL,
    fetched_at TIMESTAMP NOT NULL DEFAULT NOW(),
    PRIMARY KEY (id, region, fetched_at)
) PARTITION BY LIST (region);

-- Region partition, further sub-partitioned by month
CREATE TABLE videos_gb_part PARTITION OF videos_partitioned
    FOR VALUES IN ('GB')
    PARTITION BY RANGE (fetched_at);

CREATE TABLE videos_gb_2026_02 PARTITION OF videos_gb_part
    FOR VALUES FROM ('2026-02-01') TO ('2026-03-01');
CREATE TABLE videos_gb_2026_03 PARTITION OF videos_gb_part
    FOR VALUES FROM ('2026-03-01') TO ('2026-04-01');

This is ideal for queries like "show me viral UK videos from this week."

Strategy 3: Application-Level Sharding

When you outgrow a single PostgreSQL instance, shard at the application layer:

import hashlib
from typing import Dict, List
import asyncpg

class RegionShardRouter:
    """Route queries to region-specific database instances."""

    SHARD_MAP = {
        "western": ["GB", "NL", "AT"],   # Western Europe shard
        "nordic": ["SE", "NO"],           # Nordic shard
        "eastern": ["PL"],                # Eastern Europe shard
        "global": ["US"],                 # Global shard
    }

    def __init__(self, shard_dsns: Dict[str, str]):
        self.dsns = shard_dsns
        self.pools: Dict[str, asyncpg.Pool] = {}
        # Build reverse lookup: region -> shard name
        self.region_to_shard = {}
        for shard, regions in self.SHARD_MAP.items():
            for r in regions:
                self.region_to_shard[r] = shard

    async def initialize(self):
        for name, dsn in self.dsns.items():
            self.pools[name] = await asyncpg.create_pool(dsn, min_size=2, max_size=10)

    def _get_pool(self, region: str) -> asyncpg.Pool:
        shard = self.region_to_shard.get(region, "western")
        return self.pools[shard]

    async def get_viral_videos(self, region: str, min_score: float = 85.0, limit: int = 20) -> list:
        pool = self._get_pool(region)
        return await pool.fetch(
            "SELECT * FROM videos WHERE region = $1 AND virality_score >= $2 "
            "ORDER BY virality_score DESC LIMIT $3",
            region, min_score, limit,
        )

    async def search_all_regions(self, query: str, limit: int = 20) -> list:
        """Fan-out search across all shards, merge by virality score."""
        import asyncio
        tasks = [
            pool.fetch(
                "SELECT * FROM videos WHERE title ILIKE $1 ORDER BY virality_score DESC LIMIT $2",
                f"%{query}%", limit,
            )
            for pool in self.pools.values()
        ]
        results = await asyncio.gather(*tasks)
        merged = [dict(r) for shard_rows in results for r in shard_rows]
        merged.sort(key=lambda v: v["virality_score"], reverse=True)
        return merged[:limit]

Choosing the Right Strategy

Strategy	Best For	Complexity	ViralVidVault Fit
Region list partition	Region-scoped queries	Low	Ideal for trending pages
Region + date composite	Time-bounded + region queries	Medium	Great for analytics
App-level shard	Multiple DB servers	High	Future scaling

For ViralVidVault, region-based list partitioning is the clear winner. Every page on the site is implicitly region-scoped -- "viral in Poland," "trending in Sweden" -- so partition pruning eliminates 80%+ of rows from every query automatically.

Partition Maintenance

Automate monthly partition creation:

CREATE OR REPLACE FUNCTION ensure_monthly_partitions(base_region TEXT, months_ahead INT DEFAULT 3)
RETURNS VOID AS $$
DECLARE
    m INT;
    part_name TEXT;
    start_d DATE;
    end_d DATE;
BEGIN
    FOR m IN 0..months_ahead LOOP
        start_d := DATE_TRUNC('month', NOW() + (m || ' months')::INTERVAL);
        end_d := start_d + INTERVAL '1 month';
        part_name := format('videos_%s_%s', lower(base_region), TO_CHAR(start_d, 'YYYY_MM'));
        IF NOT EXISTS (SELECT 1 FROM pg_class WHERE relname = part_name) THEN
            EXECUTE format(
                'CREATE TABLE %I PARTITION OF videos_%s_part FOR VALUES FROM (%L) TO (%L)',
                part_name, lower(base_region), start_d, end_d
            );
        END IF;
    END LOOP;
END;
$$ LANGUAGE plpgsql;

Start simple, partition early. It's far easier to add partitioning from day one than to migrate a bloated table later.

This article is part of the Building ViralVidVault series. Check out ViralVidVault to see these techniques in action.

Database Sharding Strategies for Growing Video Platforms

ahmet gedik — Fri, 08 May 2026 16:00:01 +0000

When Do You Actually Need Sharding?

Most video platforms reach 10 million rows before sharding becomes necessary. Until then, proper indexing and query optimization outperform any sharding strategy. That said, understanding sharding early helps you design schemas that are shardable when the time comes.

TopVideoHub serves 9 Asia-Pacific regions. The natural partition keys — region and fetched_at — emerge directly from the data shape.

Strategy 1: PostgreSQL Native Partitioning

Declarative partitioning splits one logical table into physical child tables with zero application changes.

CREATE TABLE videos (
    id          BIGSERIAL,
    video_id    TEXT NOT NULL,
    title       TEXT NOT NULL,
    region      TEXT NOT NULL,
    category_id INT,
    view_count  BIGINT DEFAULT 0,
    fetched_at  TIMESTAMPTZ DEFAULT NOW(),
    PRIMARY KEY (id, region)
) PARTITION BY LIST (region);

CREATE TABLE videos_jp PARTITION OF videos FOR VALUES IN ('JP');
CREATE TABLE videos_kr PARTITION OF videos FOR VALUES IN ('KR');
CREATE TABLE videos_tw PARTITION OF videos FOR VALUES IN ('TW');
CREATE TABLE videos_hk PARTITION OF videos FOR VALUES IN ('HK');
CREATE TABLE videos_vn PARTITION OF videos FOR VALUES IN ('VN');
CREATE TABLE videos_th PARTITION OF videos FOR VALUES IN ('TH');
CREATE TABLE videos_sg PARTITION OF videos FOR VALUES IN ('SG');
CREATE TABLE videos_us PARTITION OF videos FOR VALUES IN ('US');
CREATE TABLE videos_gb PARTITION OF videos FOR VALUES IN ('GB');
CREATE TABLE videos_other PARTITION OF videos DEFAULT;

Now SELECT * FROM videos WHERE region = 'JP' only scans videos_jp. The planner prunes all other partitions automatically.

Composite Partition: Region + Date Range

CREATE TABLE video_events (
    id        BIGSERIAL,
    video_id  TEXT NOT NULL,
    region    TEXT NOT NULL,
    event     TEXT NOT NULL,
    occurred  TIMESTAMPTZ DEFAULT NOW()
) PARTITION BY LIST (region);

CREATE TABLE video_events_jp PARTITION OF video_events
    FOR VALUES IN ('JP')
    PARTITION BY RANGE (occurred);

CREATE TABLE video_events_jp_2026q1 PARTITION OF video_events_jp
    FOR VALUES FROM ('2026-01-01') TO ('2026-04-01');

CREATE TABLE video_events_jp_2026q2 PARTITION OF video_events_jp
    FOR VALUES FROM ('2026-04-01') TO ('2026-07-01');

Strategy 2: Application-Level Routing

When you outgrow a single PostgreSQL instance, route queries to different database hosts:

import asyncpg

SHARD_MAP = {
    'JP': 'postgresql://user:pass@db-asia-east/topvideohub',
    'KR': 'postgresql://user:pass@db-asia-east/topvideohub',
    'TW': 'postgresql://user:pass@db-asia-east/topvideohub',
    'HK': 'postgresql://user:pass@db-asia-east/topvideohub',
    'VN': 'postgresql://user:pass@db-asia-south/topvideohub',
    'TH': 'postgresql://user:pass@db-asia-south/topvideohub',
    'SG': 'postgresql://user:pass@db-asia-south/topvideohub',
    'US': 'postgresql://user:pass@db-west/topvideohub',
    'GB': 'postgresql://user:pass@db-eu/topvideohub',
}

_pools: dict[str, asyncpg.Pool] = {}

async def get_pool(region: str) -> asyncpg.Pool:
    dsn = SHARD_MAP.get(region, SHARD_MAP['US'])
    if dsn not in _pools:
        _pools[dsn] = await asyncpg.create_pool(dsn, min_size=2, max_size=10)
    return _pools[dsn]

async def fetch_trending(region: str, limit: int = 50) -> list[dict]:
    pool = await get_pool(region)
    async with pool.acquire() as conn:
        rows = await conn.fetch(
            'SELECT * FROM videos WHERE region=$1 ORDER BY view_count DESC LIMIT $2',
            region, limit,
        )
    return [dict(r) for r in rows]

Strategy 3: Cross-Shard Global Trending

Scatter/Gather:

import asyncio

async def global_trending(limit: int = 20) -> list[dict]:
    tasks = [fetch_trending(r, limit=limit) for r in SHARD_MAP.keys()]
    results = await asyncio.gather(*tasks)
    merged = [v for shard in results for v in shard]
    merged.sort(key=lambda v: v['view_count'], reverse=True)
    return merged[:limit]

Aggregation table (fast, slightly stale):

CREATE TABLE global_trending (
    rank       INT,
    video_id   TEXT,
    region     TEXT,
    view_count BIGINT,
    updated_at TIMESTAMPTZ DEFAULT NOW()
);

-- Refresh every 15 minutes
TRUNCATE global_trending;
INSERT INTO global_trending
    SELECT ROW_NUMBER() OVER (ORDER BY view_count DESC),
           video_id, region, view_count, NOW()
    FROM (
        SELECT video_id, region, view_count FROM videos_jp
        UNION ALL SELECT video_id, region, view_count FROM videos_kr
    ) combined
    ORDER BY view_count DESC
    LIMIT 100;

For TopVideoHub, the aggregation table approach reduced global trending query time from 400ms to 3ms.

Migration Path

Start with a single PostgreSQL instance, partitioned by LIST (region).
Move hot partitions to read replicas as read load grows.
Promote read replicas to shards when write load demands it.
Add the application-level router only at step 3.

Premature sharding is the enemy of developer productivity. Partition early, shard late.

This article is part of the Building TopVideoHub series. Check out TopVideoHub to see these techniques in action.

Python Celery Task Queues for Video Metadata Processing

ahmet gedik — Fri, 08 May 2026 06:00:02 +0000

When Synchronous Processing Breaks Down

As DailyWatch grew past 50,000 videos across 8 regions, our cron pipeline started hitting the wall. Thumbnail validation, metadata enrichment, and broken link detection can't all run in a single sequential script without blowing past timeout limits. The solution: offload heavy work to Celery task queues.

Setting Up Celery with Redis

Install the dependencies:

pip install celery[redis] requests pillow

Configure Celery with sensible defaults:

# celery_app.py
from celery import Celery

app = Celery(
    "dailywatch",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/1",
)

app.conf.update(
    task_serializer="json",
    result_serializer="json",
    accept_content=["json"],
    timezone="UTC",
    task_acks_late=True,             # Re-queue if worker crashes
    worker_prefetch_multiplier=4,    # Don't grab too many tasks at once
    task_soft_time_limit=120,        # Soft limit: 2 minutes
    task_time_limit=180,             # Hard kill: 3 minutes
    task_routes={
        "tasks.validate_thumbnail": {"queue": "thumbnails"},
        "tasks.enrich_metadata": {"queue": "metadata"},
        "tasks.check_broken_links": {"queue": "links"},
    },
)

Task 1: Thumbnail Validation

Detect broken or placeholder thumbnails:

# tasks.py
import requests
from PIL import Image
from io import BytesIO
from celery_app import app

@app.task(bind=True, max_retries=3, default_retry_delay=60)
def validate_thumbnail(self, video_id: str, thumbnail_url: str) -> dict:
    """Check if a thumbnail is valid, not a placeholder, and accessible."""
    try:
        response = requests.get(thumbnail_url, timeout=10, stream=True)
        response.raise_for_status()

        img = Image.open(BytesIO(response.content))
        width, height = img.size

        # YouTube's "no thumbnail" placeholder is 120x90 gray
        is_placeholder = (width == 120 and height == 90)

        # Check for very small or broken images
        is_valid = width >= 320 and height >= 180 and not is_placeholder

        return {
            "video_id": video_id,
            "valid": is_valid,
            "width": width,
            "height": height,
            "is_placeholder": is_placeholder,
            "content_length": len(response.content),
        }

    except requests.RequestException as exc:
        # Retry on network errors
        raise self.retry(exc=exc)
    except Exception as e:
        return {
            "video_id": video_id,
            "valid": False,
            "error": str(e),
        }

Task 2: Metadata Enrichment

Fetch additional data that our initial pipeline skips for speed:

@app.task(bind=True, max_retries=2)
def enrich_metadata(self, video_id: str, api_key: str) -> dict:
    """Fetch extended metadata: tags, default language, topic categories."""
    url = "https://www.googleapis.com/youtube/v3/videos"
    params = {
        "part": "snippet,topicDetails,localizations",
        "id": video_id,
        "key": api_key,
    }

    try:
        resp = requests.get(url, params=params, timeout=15)
        resp.raise_for_status()
        items = resp.json().get("items", [])

        if not items:
            return {"video_id": video_id, "found": False}

        item = items[0]
        snippet = item.get("snippet", {})
        topics = item.get("topicDetails", {})

        return {
            "video_id": video_id,
            "found": True,
            "tags": snippet.get("tags", [])[:20],
            "default_language": snippet.get("defaultLanguage", ""),
            "default_audio_language": snippet.get("defaultAudioLanguage", ""),
            "topic_categories": topics.get("topicCategories", []),
            "localized_titles": {
                lang: loc.get("title", "")
                for lang, loc in item.get("localizations", {}).items()
            },
        }

    except requests.RequestException as exc:
        raise self.retry(exc=exc, countdown=30)

Task 3: Broken Link Checker

Detect videos that have been removed or made private:

@app.task(rate_limit="10/m")  # Don't hammer YouTube
def check_broken_links(video_ids: list[str]) -> dict:
    """Batch check which videos are still accessible."""
    # YouTube oEmbed is free and doesn't need an API key
    results = {"alive": [], "dead": [], "errors": []}

    for vid in video_ids:
        try:
            resp = requests.head(
                f"https://www.youtube.com/oembed?url=https://youtube.com/watch?v={vid}",
                timeout=10,
                allow_redirects=True,
            )
            if resp.status_code == 200:
                results["alive"].append(vid)
            else:
                results["dead"].append(vid)
        except requests.RequestException:
            results["errors"].append(vid)

    return results

Orchestrating from the Cron Pipeline

After the main fetch completes, dispatch background work:

from celery import group, chord
from tasks import validate_thumbnail, enrich_metadata, check_broken_links

def post_fetch_pipeline(new_videos: list[dict], api_key: str):
    """Dispatch background tasks after cron fetch completes."""

    # Fan out thumbnail validation
    thumbnail_jobs = group(
        validate_thumbnail.s(v["video_id"], v["thumbnail_url"])
        for v in new_videos
    )
    thumbnail_jobs.apply_async(queue="thumbnails")

    # Fan out metadata enrichment (smaller batch, costs API quota)
    enrich_jobs = group(
        enrich_metadata.s(v["video_id"], api_key)
        for v in new_videos[:50]  # Limit to preserve quota
    )
    enrich_jobs.apply_async(queue="metadata")

    # Batch broken link check for older videos
    check_broken_links.apply_async(
        args=[get_stale_video_ids(older_than_days=7)],
        queue="links",
    )

    print(f"Dispatched {len(new_videos)} thumbnail + enrichment tasks")

Running the Workers

Start separate workers for each queue so you can scale them independently:

# Terminal 1: thumbnail workers (CPU-bound, fewer)
celery -A celery_app worker --queues=thumbnails --concurrency=4

# Terminal 2: metadata workers (IO-bound, more)
celery -A celery_app worker --queues=metadata --concurrency=8

# Terminal 3: link checker (rate-limited)
celery -A celery_app worker --queues=links --concurrency=2

Monitoring with Flower

pip install flower
celery -A celery_app flower --port=5555

This gives you a web dashboard at localhost:5555 showing task throughput, failure rates, and worker health.

Celery transformed the DailyWatch pipeline from a fragile monolith into a resilient distributed system. Tasks that used to block the cron cycle now run asynchronously, and failures get retried automatically.

This article is part of the Building DailyWatch series. Check out DailyWatch to see these techniques in action.

Python Celery Task Queues for Video Metadata Processing

ahmet gedik — Thu, 07 May 2026 20:00:01 +0000

Background Jobs for a Multilingual Video Library

TrendVidStream fetches videos from UAE, Finland, Czech Republic, Denmark, Belgium, UK, Switzerland, and the US. Content validation — thumbnail health, language detection for Arabic vs. Finnish vs. Czech titles, broken embed cleanup — runs as Celery background tasks so the main fetch loop stays fast.

Project Setup

pip install celery redis requests langdetect

celery_app.py:

from celery import Celery

app = Celery(
    'trendvidstream',
    broker='redis://localhost:6379/0',
    backend='redis://localhost:6379/1',
)

app.conf.update(
    task_serializer='json',
    result_expires=3600,
    task_acks_late=True,
    worker_prefetch_multiplier=1,
    task_routes={
        'tasks.validate_thumbnail': {'queue': 'thumbnails'},
        'tasks.detect_language':    {'queue': 'enrichment'},
        'tasks.check_embed':        {'queue': 'health'},
        'tasks.cleanup_stale':      {'queue': 'maintenance'},
    },
)

Task: Thumbnail Validation

import requests
from celery_app import app
from database import get_db

@app.task(name='tasks.validate_thumbnail', bind=True, max_retries=2, default_retry_delay=60)
def validate_thumbnail(self, video_id: str, thumbnail_url: str, region: str) -> dict:
    try:
        resp = requests.head(thumbnail_url, timeout=6, allow_redirects=True)
        ok = resp.status_code == 200 and 'image' in resp.headers.get('Content-Type', '')
        if not ok:
            db = get_db()
            db.execute('UPDATE videos SET thumbnail_broken=1 WHERE video_id=?', (video_id,))
            db.commit()
        return {'video_id': video_id, 'region': region, 'ok': ok}
    except requests.RequestException as exc:
        raise self.retry(exc=exc)

Task: Multilingual Language Detection

TrendVidStream regions include Arabic (AE), Finnish (FI), Czech (CZ), Dutch (BE), and English (GB):

from langdetect import detect, LangDetectException

EXPECTED_LANGS = {
    'AE': {'ar'}, 'FI': {'fi'}, 'CZ': {'cs'}, 'DK': {'da'},
    'BE': {'nl', 'fr'}, 'CH': {'de', 'fr', 'it'}, 'GB': {'en'}, 'US': {'en'},
}

@app.task(name='tasks.detect_language')
def detect_language(video_id: str, title: str, region: str) -> dict:
    try:
        lang = detect(title)
    except LangDetectException:
        lang = 'und'

    expected = EXPECTED_LANGS.get(region, {'en'})
    is_expected = lang in expected

    db = get_db()
    db.execute('UPDATE videos SET title_lang=?, lang_mismatch=? WHERE video_id=?',
               (lang, 0 if is_expected else 1, video_id))
    db.commit()

    if not is_expected:
        print(f'[LANG MISMATCH] {region}: expected {expected}, got "{lang}" — {title[:50]}')
    return {'video_id': video_id, 'lang': lang, 'expected': is_expected}

Task: Broken Embed Detection

@app.task(name='tasks.check_embed', bind=True, max_retries=1, default_retry_delay=300)
def check_embed(self, video_id: str) -> bool:
    url = f'https://www.youtube.com/oembed?url=https://youtu.be/{video_id}&format=json'
    try:
        resp = requests.get(url, timeout=8)
        live = resp.status_code == 200
        if not live:
            db = get_db()
            db.execute('UPDATE videos SET embed_broken=1 WHERE video_id=?', (video_id,))
            db.commit()
        return live
    except requests.RequestException as exc:
        raise self.retry(exc=exc)

Task: Stale Content Cleanup

from datetime import datetime, timedelta

@app.task(name='tasks.cleanup_stale')
def cleanup_stale(days: int = 30) -> int:
    cutoff = (datetime.utcnow() - timedelta(days=days)).isoformat()
    db = get_db()
    cur = db.execute('DELETE FROM videos WHERE fetched_at < ? AND embed_broken = 1', (cutoff,))
    db.commit()
    return cur.rowcount

Celery Beat Scheduler

celery -A celery_app worker -Q thumbnails,enrichment -c 8 &
celery -A celery_app worker -Q health -c 4 &
celery -A celery_app worker -Q maintenance -c 1 &
celery -A celery_app beat &

Since deploying Celery for TrendVidStream, Arabic language mismatch detection alone flagged over 300 videos incorrectly assigned to European regions, improving content quality across the platform.

This article is part of the Building TrendVidStream series. Check out TrendVidStream to see these techniques in action.