diff --git a/.gitea/workflows/cd.yml b/.gitea/workflows/cd.yml index e6703a2..c5f7d78 100644 --- a/.gitea/workflows/cd.yml +++ b/.gitea/workflows/cd.yml @@ -1,3 +1,8 @@ +# CD workflow — TEST LAB DEPLOYMENT +# Deploys to test lab host 192.168.88.40:2375. +# Production Hetzner target TBD — switch DOCKER_HOST when provisioned. +# See TODO.md Production Deployment section for cutover plan. + name: CD on: @@ -76,9 +81,9 @@ jobs: # Wait for services to stabilize sleep 20 - # Check chat health + # Check chat health (via docker exec — IP-agnostic, works on any DOCKER_HOST) echo "Checking chat health..." - if ! curl -f http://192.168.88.40:3000/health; then + if ! docker exec linearvoid-chat curl -sf http://localhost:3000/health; then echo "Chat health check failed" docker logs linearvoid-chat --tail 50 exit 1 @@ -86,7 +91,7 @@ jobs: # Check nginx health echo "Checking nginx health..." - if ! curl -f http://192.168.88.40/health/; then + if ! docker exec linearvoid-nginx curl -sf http://localhost/health; then echo "Nginx health check failed" docker logs linearvoid-nginx --tail 50 exit 1 diff --git a/CHANGELOG.md b/CHANGELOG.md index b95ac8f..0d763fc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,51 @@ +## [1.0.7] - 2026-08-06 + +### Added + +- **Emergency slate fallback**: `scripts/generate-slate.sh` creates "We'll Be Right Back" slate video + pre-rendered HLS. Playout detects FFmpeg failures (>5 restarts/60s) and auto-switches to slate manifest. Health endpoint reports `degraded` mode. Auto-recovers when FFmpeg restarts successfully. +- **Prometheus metrics**: `linearvoid_playout_restarts_total`, `linearvoid_playout_mode` (0=live, 1=slate), `linearvoid_chat_dms_total` exposed on playout + chat `/metrics`. +- **Video normalization pipeline**: `scripts/normalize.sh` with env-configurable CRF/preset/audio-bitrate wrapping reference FFmpeg command. +- **Strict ffprobe verification gate**: `ValidateVideo()` in playout parses ffprobe JSON, checks codec (h264/aac), framerate (30/1), resolution (1280x720 or 1920x1080), pix_fmt (yuv420p), sample_rate (48000), keyframe interval (≤2s). `ValidateVideoStrict()` returns `{valid, errors[]}`. +- **Upload normalization enforcement**: `validateMediaNormalization()` runs on every admin upload (single + chunked). Non-conforming files get `needs_normalization` status. Visible in media library + playlist validation warnings. +- **Chat history on room join**: `pushChatHistory()` sends last 50 messages via WebSocket on connect. +- **DM offline delivery**: Private messages persisted to DB, appear in history when recipient reconnects. +- **Online presence indicators**: Green dot per user in chat sidebar, updated via join/leave system messages. +- **Schedule metadata overlay**: "Now Playing" + "Up Next" below viewer player, polling `GET /api/schedule` every 30s. Public `/api/schedule` endpoint added. +- **Playlist drag-to-reorder**: Sortable list with drag-and-drop, per-entry remove buttons, validation warnings (missing files, needs_normalization). +- **Schedule overrides**: `ScheduleOverride` model + `GET/POST/DELETE /api/admin/schedule/overrides` with conflict detection (409 on overlap). Admin UI with form + list. +- **User management actions**: `POST /api/admin/users` with mute/unmute/ban/unban actions. Per-user action buttons in admin users table. +- **Inline moderation**: Mute/Ban buttons on each chat log entry row in admin. +- **CSV export**: `GET /api/admin/chat-log/export` returns CSV download. +- **Analytics time range selector**: 1h, 6h, 24h, 7d dropdown on admin analytics page. +- **Runtime config editor**: `GET/PUT /api/admin/config` for rate limit max + window. Admin system page with config form + restart buttons. +- **Production target documentation**: `DEPLOYMENT.md` updated with two-environment model (test lab 192.168.88.0/24 vs Hetzner production TBD), CX22 specs, firewall requirements, DNS cutover plan. +- **Network diagram**: Mermaid diagram in `SETUP.md` distinguishing test lab from planned production. +- **E2E test script**: `scripts/test-playout-e2e.sh` for runtime validation in test lab. +- **Redis pub/sub stub**: Architecture ready for multi-instance chat (pending implementation). + +### Changed + +- `docker-compose.yml` + `.prod.yml`: added slate volume mount (`./data/slate:/app/slate:ro`), Redis `--maxmemory 256mb --maxmemory-policy allkeys-lru`. +- `backend/playout/Dockerfile`: added `/app/slate` directory. +- `handleConnection`: added 30s presence heartbeat + WebSocket ping, banned/muted user checks, env-configurable rate limits (`CHAT_RATE_LIMIT_MAX`, `CHAT_RATE_LIMIT_WINDOW_S`). +- `handleModerationCommand`: added moderator role check, mute duration parsing, Redis key enforcement. +- `handlePrivateDM`: targeted delivery via `clientMeta` map instead of broadcast. +- `initRedis`: added `redisReconnectLoop` with exponential backoff (2s→60s). +- `handleAdminMedia` + chunked upload: normalization validation runs on every upload, sets `needs_normalization` status. +- nginx: OPTIONS preflight → 204 on all 4 HLS locations. `/health` returns JSON. `/nginx-status` with stub_status (internal-network restricted). +- CD workflow: health checks now IP-agnostic (`docker exec` instead of hardcoded IP curls). +- `docker-compose.yml` + `.prod.yml`, `.gitea/workflows/cd.yml`, `DEPLOYMENT.md`, `SETUP.md`, `docs/architecture/README.md`, `README.md`: all annotated with test-lab vs production distinction, `192.168.88.0/24` flagged as internal test network. +- `docs/architecture/README.md`: CI/CD deployment view updated to note test lab target, production pending. + +### Fixed + +- CD health check no longer hardcodes `192.168.88.40` in curl targets — uses `docker exec` for portability. +- Playlist page no longer single-textarea — now shows sortable list with validation warnings. +- Schedule page shows overrides alongside playlist timeline. + ## [1.0.6] - 2026-08-05 ### Added diff --git a/DEPLOYMENT.md b/DEPLOYMENT.md index eb1d701..a55aa92 100644 --- a/DEPLOYMENT.md +++ b/DEPLOYMENT.md @@ -1,6 +1,65 @@ -# Production Deployment Guide +# Deployment Guide -This guide describes a production-style deployment for Linear Void using Docker Compose, Gitea registry images, and Cloudflare fronting. +## Environments + +Linear Void runs in two separate environments: + +| Environment | Host | Docker Target | Purpose | +| -------------- | -------------------------- | -------------------------- | ---------------------------------------------- | +| **Test Lab** | `192.168.88.40` (internal) | `tcp://192.168.88.40:2375` | Integration testing, staging, CI/CD validation | +| **Production** | Hetzner VPS (TBD) | `tcp://:2375` | Public linearvoid.tv serving real viewers | + +**Current state (2026-08-06):** Test lab is active and receives automated CD deploys. Production Hetzner VPS is **not yet provisioned**. The CD workflow (`cd.yml`) and `docker-compose.prod.yml` currently target the test lab. Switching to production requires only changing `DOCKER_HOST` — see [Production Deployment](#production-deployment-hetzner) below. + +## Production Target (Hetzner) + +**Status: Not provisioned.** The production deployment plan is documented and the pipeline is ready — only the VPS needs to be created. + +### Planned Infrastructure + +| Resource | Spec | +| -------------- | ------------------------------------------------ | +| Provider | Hetzner Cloud | +| Region | Ashburn, VA (us-east) | +| Instance | CX22 (2 vCPU, 4 GB RAM, 40 GB SSD) or equivalent | +| OS | Ubuntu 24.04 LTS or Debian 12 | +| Bandwidth | 20 TB included traffic/month | +| Estimated cost | ~€7–10/month base | + +### Cutover Plan + +1. Provision Hetzner VPS (see `TODO.md` #92) +2. Add production secrets to Gitea (`CI_PROD_*` prefix) +3. Update `DOCKER_HOST` in CD workflow to Hetzner public IP +4. Trigger deploy → validate health checks +5. Update Cloudflare A record `linearvoid.tv` → Hetzner IP +6. Enable HTTPS, WAF, cache rules on production zone +7. Decommission test lab as public target + +### Firewall / Security Group + +| Port | Source | Purpose | +| ---- | --------------------- | ------------------------------ | +| 22 | Admin IPs | SSH management | +| 80 | 0.0.0.0/0 | HTTP (redirect to HTTPS) | +| 443 | 0.0.0.0/0 | HTTPS (nginx) | +| 2375 | Gitea runner IP only | Docker remote API (TLS) | +| 9090 | Admin IPs or internal | Prometheus (or tunnel via SSH) | +| 3001 | Admin IPs or internal | Grafana (or tunnel via SSH) | + +### Pipeline Readiness + +The Gitea Actions CD workflow (`cd.yml`) already supports `DOCKER_HOST` as a configurable variable. To switch targets: + +```bash +# Test lab (current) +DOCKER_HOST=tcp://192.168.88.40:2375 + +# Production (when provisioned) +DOCKER_HOST=tcp://:2375 +``` + +No code changes needed in the pipeline. See `TODO.md` #91 and #92–#99 for full production deployment checklist. ## Target Topology diff --git a/README.md b/README.md index f2bfea8..8ead4a3 100644 --- a/README.md +++ b/README.md @@ -14,7 +14,7 @@ The architecture (detailed in [`docs/architecture/README.md`](docs/architecture/README.md)) uses: -- Hetzner Ashburn VPS for playout (FFmpeg + MediaMTX/SRS) and chat. +- Hetzner Ashburn VPS for playout (FFmpeg + MediaMTX/SRS) and chat *(planned production; currently test lab at 192.168.88.40)*. - Cloudflare R2 + CDN with aggressive immutable caching of HLS `.ts` segments ($0 egress). - Lightweight static frontend and WebSocket chat backend. - Gitea self-hosted CI/CD (Actions) and Container Registry. @@ -112,6 +112,8 @@ This structure directly reflects the building blocks, runtime flows, deployment ## Getting Started +> **Note:** The `docker compose up` below targets local development. The test lab (192.168.88.40) receives automated CD deploys. Production Hetzner VPS is not yet provisioned — see `TODO.md`. + ```bash git clone https://git.allucanget.biz/zwitschi/linearvoid.git cd linearvoid @@ -133,6 +135,7 @@ See the [architecture docs](docs/architecture/README.md) for detailed architectu ## Development - Use Docker Compose for local development (playout + chat + frontend). +- Test lab host: `192.168.88.40` (internal network, receives CD deploys). Production Hetzner VPS TBD. - Public production host model: `https://linearvoid.tv` for viewer/admin/chat and `https://media.linearvoid.tv` for HLS delivery. - Normalize all source videos before adding to playlist (see ADR-001 for exact codec/fps/keyframe requirements). - Monitor cache hit ratio and R2 operations closely — they determine real cost. diff --git a/SETUP.md b/SETUP.md index 6bae224..e26e586 100644 --- a/SETUP.md +++ b/SETUP.md @@ -21,6 +21,42 @@ This guide covers prerequisites, environment setup, running the application loca - `DATABASE_URL`: The production database connection string (e.g., `postgres://linearvoid:@db:5432/linearvoid?sslmode=disable`) - **Hetzner Cloud Account** (optional for production): For the US-East VPS (Ashburn, VA recommended per ADR-001). +## Network Environments + +Linear Void operates in two distinct environments. Know which one you're targeting. + +```mermaid +flowchart LR + subgraph Test_Lab["Test Lab (192.168.88.0/24)"] + TEST_HOST["192.168.88.40
Docker Engine :2375"] + end + + subgraph Prod["Production (Hetzner — NOT provisioned)"] + PROD_HOST["Hetzner CX22 Ashburn, VA
Public IP TBD"] + end + + subgraph Gitea["Gitea CI/CD"] + CI["CI: build + push images"] + CD["CD: deploy via DOCKER_HOST"] + end + + subgraph CF["Cloudflare"] + DNS["linearvoid.tv"] + R2["R2 Bucket"] + end + + CD -->|"DOCKER_HOST=tcp://192.168.88.40:2375"| TEST_HOST + CD -.->|"DOCKER_HOST=tcp://<hetzner-ip>:2375
(when provisioned)"| PROD_HOST + TEST_HOST -->|"test traffic"| DNS + PROD_HOST -.->|"production traffic"| DNS + TEST_HOST --> R2 + PROD_HOST -.-> R2 +``` + +**Test Lab (active):** Internal network `192.168.88.0/24`, host `192.168.88.40`. Receives automated CD deploys from Gitea Actions. Used for integration testing and staging validation. + +**Production (pending):** Hetzner Cloud VPS, Ashburn VA. Not yet provisioned. CD workflow and compose files are ready — only `DOCKER_HOST` needs changing. See `TODO.md` Production Deployment section. + ## Development Prerequisites - **Frontend dependencies**: Run `cd frontend && npm install` before `npm run dev` or `npm run build`. diff --git a/backend/chat/src/main.go b/backend/chat/src/main.go index 866b65f..365045a 100644 --- a/backend/chat/src/main.go +++ b/backend/chat/src/main.go @@ -32,12 +32,22 @@ var ( upgrader = websocket.Upgrader{ CheckOrigin: func(r *http.Request) bool { return true }, } - clients = make(map[*websocket.Conn]bool) - mutex = sync.Mutex{} + clients = make(map[*websocket.Conn]bool) + clientMeta = make(map[*websocket.Conn]clientInfo) + mutex = sync.Mutex{} authSessions = map[string]time.Time{} authSessionsMu = sync.Mutex{} + + // Rate limit config (env-overridable) + rateLimitMax = 10 + rateLimitWindow = 10 * time.Second ) +type clientInfo struct { + Username string + Room string +} + // Message represents a chat message (persisted in SQL DB per ADR-006) type Message struct { ID uint `gorm:"primarykey" json:"id"` @@ -73,6 +83,16 @@ type Media struct { UpdatedAt time.Time `gorm:"autoUpdateTime" json:"updatedAt"` } +// ScheduleOverride represents a scheduled playlist override (ADR-004) +type ScheduleOverride struct { + ID uint `gorm:"primarykey" json:"id"` + MediaID uint `json:"mediaId"` + StartTime time.Time `json:"startTime"` + EndTime time.Time `json:"endTime"` + Reason string `json:"reason"` + CreatedAt time.Time `json:"createdAt"` +} + // Presence tracks online users per room (using Redis per ADR-007) type Presence struct { Username string `json:"username"` @@ -287,11 +307,35 @@ func initRedis() { if err := redisClient.Ping(context.Background()).Err(); err != nil { log.Printf("Redis connection warning: %v (rate limiting and presence will be disabled)", err) redisClient = nil + // Start auto-reconnect goroutine + go redisReconnectLoop(redisAddress, options) } else { log.Println("Redis connected successfully (presence, rate limiting, pub/sub enabled)") } } +// redisReconnectLoop attempts to reconnect to Redis with exponential backoff. +func redisReconnectLoop(addr string, baseOptions *redis.Options) { + backoff := 2 * time.Second + maxBackoff := 60 * time.Second + for { + time.Sleep(backoff) + testClient := redis.NewClient(baseOptions) + if err := testClient.Ping(context.Background()).Err(); err != nil { + log.Printf("Redis reconnect failed: %v (retry in %v)", err, backoff) + backoff *= 2 + if backoff > maxBackoff { + backoff = maxBackoff + } + testClient.Close() + continue + } + redisClient = testClient + log.Println("Redis reconnected successfully") + return + } +} + // initDB initializes configured SQL storage for chat, moderation, and admin metadata (ADR-006) func initDB() { var err error @@ -319,7 +363,7 @@ func initDB() { } databaseDriverName = driverName - db.AutoMigrate(&Message{}, &ModerationLog{}, &Media{}) + db.AutoMigrate(&Message{}, &ModerationLog{}, &Media{}, &ScheduleOverride{}) log.Printf("%s database initialized (chat history, moderation log, media metadata)", strings.Title(driverName)) } @@ -626,7 +670,103 @@ func handleAdminPlaylist(w http.ResponseWriter, r *http.Request) { } } -func handleAdminSchedule(w http.ResponseWriter, r *http.Request) { +// handlePublicSchedule returns playlist schedule without auth (viewer-facing). +func handlePublicSchedule(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + content, err := os.ReadFile(getPlaylistPath()) + if err != nil { + http.Error(w, "schedule unavailable", http.StatusServiceUnavailable) + return + } + entries := parsePlaylistEntries(string(content)) + var mediaRecords []Media + db.Find(&mediaRecords) + mediaByName := make(map[string]Media, len(mediaRecords)) + for _, m := range mediaRecords { + mediaByName[strings.ToLower(m.Filename)] = m + } + type PublicEntry struct { + Index int `json:"index"` + Label string `json:"label"` + Title string `json:"title"` + Status string `json:"status"` + } + output := make([]PublicEntry, 0, len(entries)) + for _, e := range entries { + pe := PublicEntry{Index: e.Index, Label: e.Label, Status: "pending"} + if m, ok := mediaByName[strings.ToLower(filepath.Base(e.Path))]; ok { + pe.Title = m.Title + pe.Status = m.Status + } else { + pe.Title = filepath.Base(e.Path) + } + output = append(output, pe) + } + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]any{ + "entries": output, + "total": len(output), + "generatedAt": time.Now().UTC().Format(time.RFC3339), + }) +} + +// handleAdminScheduleOverrides manages CRUD for schedule overrides. +func handleAdminScheduleOverrides(w http.ResponseWriter, r *http.Request) { + switch r.Method { + case http.MethodGet: + var overrides []ScheduleOverride + db.Order("start_time asc").Find(&overrides) + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(overrides) + + case http.MethodPost: + var ov ScheduleOverride + if err := json.NewDecoder(r.Body).Decode(&ov); err != nil { + http.Error(w, "invalid request body", http.StatusBadRequest) + return + } + if ov.MediaID == 0 || ov.StartTime.IsZero() { + http.Error(w, "mediaId and startTime are required", http.StatusBadRequest) + return + } + // Conflict detection: warn if overlapping + var conflict int64 + db.Model(&ScheduleOverride{}). + Where("start_time < ? AND end_time > ?", ov.EndTime, ov.StartTime). + Count(&conflict) + if conflict > 0 { + http.Error(w, fmt.Sprintf("conflict: %d overlapping override(s)", conflict), http.StatusConflict) + return + } + db.Create(&ov) + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusCreated) + json.NewEncoder(w).Encode(ov) + + case http.MethodDelete: + idStr := r.URL.Query().Get("id") + if idStr == "" { + http.Error(w, "id query param required", http.StatusBadRequest) + return + } + id, err := strconv.ParseUint(idStr, 10, 64) + if err != nil { + http.Error(w, "invalid id", http.StatusBadRequest) + return + } + if err := db.Delete(&ScheduleOverride{}, id).Error; err != nil { + http.Error(w, "failed to delete override", http.StatusInternalServerError) + return + } + w.WriteHeader(http.StatusNoContent) + + default: + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + } +} if r.Method != http.MethodGet { http.Error(w, "method not allowed", http.StatusMethodNotAllowed) return @@ -875,7 +1015,53 @@ func getAdminSystemSummary() AdminSystemSummary { return summary } -func handleAdminSystemSummary(w http.ResponseWriter, r *http.Request) { +// handleAdminConfig returns runtime configuration values for the admin config editor. +func handleAdminConfig(w http.ResponseWriter, r *http.Request) { + switch r.Method { + case http.MethodGet: + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]any{ + "rateLimitMax": rateLimitMax, + "rateLimitWindowSec": int(rateLimitWindow.Seconds()), + "databaseDriver": databaseDriverName, + "redisEnabled": redisClient != nil, + "sessionTTLHours": int(getAdminSessionTTL().Hours()), + }) + case http.MethodPut: + var payload map[string]any + if err := json.NewDecoder(r.Body).Decode(&payload); err != nil { + http.Error(w, "invalid body", http.StatusBadRequest) + return + } + if v, ok := payload["rateLimitMax"].(float64); ok && v > 0 { + rateLimitMax = int(v) + log.Printf("Config updated: rateLimitMax=%d", rateLimitMax) + } + if v, ok := payload["rateLimitWindowSec"].(float64); ok && v > 0 { + rateLimitWindow = time.Duration(v) * time.Second + log.Printf("Config updated: rateLimitWindow=%v", rateLimitWindow) + } + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]string{"status": "ok"}) + default: + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + } +} + +// handleAdminRestart triggers a service restart (placeholder — returns instructions). +func handleAdminRestart(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + service := r.URL.Query().Get("service") + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]string{ + "status": "restart_not_automated", + "service": service, + "message": fmt.Sprintf("Restart %s manually: docker compose -f docker-compose.prod.yml restart %s", service, service), + }) +} if r.Method != http.MethodGet { http.Error(w, "method not allowed", http.StatusMethodNotAllowed) return @@ -886,47 +1072,83 @@ func handleAdminSystemSummary(w http.ResponseWriter, r *http.Request) { } func handleAdminUsers(w http.ResponseWriter, r *http.Request) { - if r.Method != http.MethodGet { - http.Error(w, "method not allowed", http.StatusMethodNotAllowed) - return - } + switch r.Method { + case http.MethodGet: + response := AdminUsersResponse{ + Users: []ActiveUser{}, + RedisEnabled: redisClient != nil, + GeneratedAt: time.Now().UTC().Format(time.RFC3339), + } + if redisClient != nil { + ctx := context.Background() + keys, _, err := redisClient.Scan(ctx, 0, "presence:*", 200).Result() + if err != nil { + http.Error(w, fmt.Sprintf("failed to scan presence keys: %v", err), http.StatusInternalServerError) + return + } + for _, key := range keys { + parts := strings.SplitN(key, ":", 3) + if len(parts) != 3 { + continue + } + response.Users = append(response.Users, ActiveUser{ + Room: parts[1], + Username: parts[2], + Online: true, + }) + } + } + sort.Slice(response.Users, func(i, j int) bool { + if response.Users[i].Room == response.Users[j].Room { + return response.Users[i].Username < response.Users[j].Username + } + return response.Users[i].Room < response.Users[j].Room + }) + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(response) - response := AdminUsersResponse{ - Users: []ActiveUser{}, - RedisEnabled: redisClient != nil, - GeneratedAt: time.Now().UTC().Format(time.RFC3339), - } - - if redisClient != nil { - ctx := context.Background() - keys, _, err := redisClient.Scan(ctx, 0, "presence:*", 200).Result() - if err != nil { - http.Error(w, fmt.Sprintf("failed to scan presence keys: %v", err), http.StatusInternalServerError) + case http.MethodPost: + var payload struct { + Action string `json:"action"` // mute, unmute, ban, unban + Username string `json:"username"` + Duration int `json:"duration"` // minutes, for mute + } + if err := json.NewDecoder(r.Body).Decode(&payload); err != nil { + http.Error(w, "invalid body", http.StatusBadRequest) return } - - for _, key := range keys { - parts := strings.SplitN(key, ":", 3) - if len(parts) != 3 { - continue + if payload.Action == "" || payload.Username == "" { + http.Error(w, "action and username required", http.StatusBadRequest) + return + } + if redisClient == nil { + http.Error(w, "Redis unavailable", http.StatusServiceUnavailable) + return + } + ctx := context.Background() + switch payload.Action { + case "mute": + dur := time.Duration(payload.Duration) * time.Minute + if dur <= 0 { + dur = 10 * time.Minute } - response.Users = append(response.Users, ActiveUser{ - Room: parts[1], - Username: parts[2], - Online: true, - }) + redisClient.Set(ctx, fmt.Sprintf("muted:%s", payload.Username), "1", dur) + case "unmute": + redisClient.Del(ctx, fmt.Sprintf("muted:%s", payload.Username)) + case "ban": + redisClient.Set(ctx, fmt.Sprintf("banned:%s", payload.Username), "1", 0) + case "unban": + redisClient.Del(ctx, fmt.Sprintf("banned:%s", payload.Username)) + default: + http.Error(w, "unknown action", http.StatusBadRequest) + return } + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]string{"status": "ok", "action": payload.Action, "username": payload.Username}) + + default: + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) } - - sort.Slice(response.Users, func(i, j int) bool { - if response.Users[i].Room == response.Users[j].Room { - return response.Users[i].Username < response.Users[j].Username - } - return response.Users[i].Room < response.Users[j].Room - }) - - w.Header().Set("Content-Type", "application/json") - json.NewEncoder(w).Encode(response) } func listMediaFiles() ([]MediaFileEntry, error) { @@ -1139,7 +1361,90 @@ func mediaEntryFromRecord(record Media) MediaFileEntry { } } -func probeMediaDurationSeconds(mediaPath string) (int, error) { +// MediaNormalizationResult holds ffprobe validation output for admin uploads. +type MediaNormalizationResult struct { + Valid bool `json:"valid"` + Errors []string `json:"errors"` +} + +// validateMediaNormalization runs ffprobe against a media file and checks +// conformance with the Linear Void normalization profile (ADR-001). +func validateMediaNormalization(mediaPath string) MediaNormalizationResult { + result := MediaNormalizationResult{Valid: true, Errors: []string{}} + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + out, err := exec.CommandContext(ctx, + "ffprobe", "-v", "quiet", "-print_format", "json", "-show_format", "-show_streams", mediaPath, + ).CombinedOutput() + + if ctx.Err() == context.DeadlineExceeded { + return MediaNormalizationResult{Valid: false, Errors: []string{"ffprobe timed out"}} + } + if err != nil { + return MediaNormalizationResult{Valid: false, Errors: []string{fmt.Sprintf("ffprobe failed: %v (%s)", err, strings.TrimSpace(string(out)))}} + } + + var probe struct { + Streams []struct { + CodecName string `json:"codec_name"` + CodecType string `json:"codec_type"` + Width int `json:"width"` + Height int `json:"height"` + RFrameRate string `json:"r_frame_rate"` + PixFmt string `json:"pix_fmt"` + SampleRate string `json:"sample_rate"` + } `json:"streams"` + } + if err := json.Unmarshal(out, &probe); err != nil { + return MediaNormalizationResult{Valid: false, Errors: []string{"ffprobe JSON parse failed: " + err.Error()}} + } + + var hasVideo, hasAudio bool + for _, s := range probe.Streams { + switch s.CodecType { + case "video": + hasVideo = true + if s.CodecName != "h264" { + result.Errors = append(result.Errors, fmt.Sprintf("video codec must be h264, got %s", s.CodecName)) + } + if s.RFrameRate != "30/1" { + result.Errors = append(result.Errors, fmt.Sprintf("framerate must be 30 fps constant, got %s", s.RFrameRate)) + } + if s.Width != 1280 && s.Width != 1920 { + result.Errors = append(result.Errors, fmt.Sprintf("width must be 1280 or 1920, got %d", s.Width)) + } + if s.Height != 720 && s.Height != 1080 { + result.Errors = append(result.Errors, fmt.Sprintf("height must be 720 or 1080, got %d", s.Height)) + } + if s.PixFmt != "yuv420p" { + result.Errors = append(result.Errors, fmt.Sprintf("pixel format must be yuv420p, got %s", s.PixFmt)) + } + case "audio": + hasAudio = true + if s.CodecName != "aac" { + result.Errors = append(result.Errors, fmt.Sprintf("audio codec must be aac, got %s", s.CodecName)) + } + if s.SampleRate != "48000" { + result.Errors = append(result.Errors, fmt.Sprintf("sample rate must be 48000 Hz, got %s", s.SampleRate)) + } + } + } + + if !hasVideo { + result.Errors = append(result.Errors, "no video stream found") + } + if !hasAudio { + result.Errors = append(result.Errors, "no audio stream found") + } + + if len(result.Errors) > 0 { + result.Valid = false + } + + return result +} ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) defer cancel() @@ -1274,6 +1579,13 @@ func handleAdminMedia(w http.ResponseWriter, r *http.Request) { return } + // Validate normalization profile (ADR-001) + normResult := validateMediaNormalization(targetPath) + status := "ready" + if !normResult.Valid { + status = "needs_normalization" + } + filename := filepath.Base(targetPath) mediaRecord := Media{ Filename: filename, @@ -1281,9 +1593,9 @@ func handleAdminMedia(w http.ResponseWriter, r *http.Request) { Path: targetPath, SizeBytes: fileInfo.Size(), DurationSeconds: durationSeconds, - Status: "ready", + Status: status, Title: strings.TrimSuffix(filename, filepath.Ext(filename)), - Tags: "", + Tags: fmt.Sprintf(`"normalization":%v`, normResult.Valid), } if err := db.Create(&mediaRecord).Error; err != nil { _ = os.Remove(targetPath) @@ -1602,6 +1914,13 @@ func handleAdminMediaUploads(w http.ResponseWriter, r *http.Request) { return } + // Validate normalization profile (ADR-001) + normResult := validateMediaNormalization(targetPath) + status := "ready" + if !normResult.Valid { + status = "needs_normalization" + } + filename := filepath.Base(targetPath) mediaRecord := Media{ Filename: filename, @@ -1609,9 +1928,9 @@ func handleAdminMediaUploads(w http.ResponseWriter, r *http.Request) { Path: targetPath, SizeBytes: fileInfo.Size(), DurationSeconds: durationSeconds, - Status: "ready", + Status: status, Title: strings.TrimSuffix(filename, filepath.Ext(filename)), - Tags: "", + Tags: fmt.Sprintf(`"normalization":%v`, normResult.Valid), } if err := db.Create(&mediaRecord).Error; err != nil { _ = os.Remove(targetPath) @@ -1793,6 +2112,10 @@ func handleMetrics(w http.ResponseWriter, r *http.Request) { fmt.Fprintln(w, "# TYPE linearvoid_chat_messages_total gauge") writePrometheusMetricLine(w, "linearvoid_chat_messages_total", storedMessages) + fmt.Fprintln(w, "# HELP linearvoid_chat_dms_total Total private messages sent.") + fmt.Fprintln(w, "# TYPE linearvoid_chat_dms_total counter") + writePrometheusMetricLine(w, "linearvoid_chat_dms_total", dmsSent) + fmt.Fprintln(w, "# HELP linearvoid_chat_moderation_actions_total Total moderation log entries.") fmt.Fprintln(w, "# TYPE linearvoid_chat_moderation_actions_total gauge") writePrometheusMetricLine(w, "linearvoid_chat_moderation_actions_total", moderationActions) @@ -1841,25 +2164,57 @@ func handleConnection(w http.ResponseWriter, r *http.Request) { log.Println("Upgrade error:", err) return } + + username := r.URL.Query().Get("username") + room := r.URL.Query().Get("room") + if username == "" { + username = fmt.Sprintf("anon_%d", time.Now().UnixNano()%10000) + } + mutex.Lock() clients[conn] = true + clientMeta[conn] = clientInfo{Username: username, Room: room} mutex.Unlock() // Announce presence (ADR-007) - username := r.URL.Query().Get("username") - room := r.URL.Query().Get("room") - if redisClient != nil && username != "" && room != "" { + if redisClient != nil && room != "" { redisClient.Set(context.Background(), fmt.Sprintf("presence:%s:%s", room, username), "online", 60*time.Second) - broadcast(Message{Room: room, Username: "system", Text: username + " joined " + room, Type: "system"}) + broadcast(Message{Room: room, Username: "system", Text: username + " joined " + room, Type: "system", Timestamp: time.Now()}) } + // Push chat history on join (last 50 messages) + pushChatHistory(conn, room, 50) + + // Heartbeat goroutine: refresh presence TTL every 30s + heartbeatStop := make(chan struct{}) + go func() { + ticker := time.NewTicker(30 * time.Second) + defer ticker.Stop() + for { + select { + case <-ticker.C: + if redisClient != nil && room != "" { + redisClient.Set(context.Background(), fmt.Sprintf("presence:%s:%s", room, username), "online", 60*time.Second) + } + // Ping to detect dead connections + if err := conn.WriteMessage(websocket.PingMessage, nil); err != nil { + return + } + case <-heartbeatStop: + return + } + } + }() + defer func() { + close(heartbeatStop) mutex.Lock() delete(clients, conn) + delete(clientMeta, conn) mutex.Unlock() - if redisClient != nil && username != "" && room != "" { + if redisClient != nil && room != "" { redisClient.Del(context.Background(), fmt.Sprintf("presence:%s:%s", room, username)) - broadcast(Message{Room: room, Username: "system", Text: username + " left " + room, Type: "system"}) + broadcast(Message{Room: room, Username: "system", Text: username + " left " + room, Type: "system", Timestamp: time.Now()}) } conn.Close() }() @@ -1873,15 +2228,27 @@ func handleConnection(w http.ResponseWriter, r *http.Request) { msg.Timestamp = time.Now() msg.Username = username // enforce from query param for security - // Rate limiting check using Redis (ADR-007) + // Check if banned before processing + if redisClient != nil { + if isBanned, _ := redisClient.Exists(context.Background(), fmt.Sprintf("banned:%s", username)).Result(); isBanned > 0 { + conn.WriteJSON(Message{Type: "system", Text: "You are banned from chat.", Timestamp: time.Now()}) + continue + } + if isMuted, _ := redisClient.Exists(context.Background(), fmt.Sprintf("muted:%s", username)).Result(); isMuted > 0 { + conn.WriteJSON(Message{Type: "system", Text: "You are muted.", Timestamp: time.Now()}) + continue + } + } + + // Rate limiting check using Redis with env-configurable thresholds if redisClient != nil { key := fmt.Sprintf("rate:%s", msg.Username) count, _ := redisClient.Incr(context.Background(), key).Result() if count == 1 { - redisClient.Expire(context.Background(), key, 10*time.Second) + redisClient.Expire(context.Background(), key, rateLimitWindow) } - if count > 10 { - conn.WriteJSON(Message{Type: "system", Text: "Rate limit exceeded. Please slow down."}) + if count > int64(rateLimitMax) { + conn.WriteJSON(Message{Type: "system", Text: "Rate limit exceeded. Please slow down.", Timestamp: time.Now()}) continue } } @@ -1892,29 +2259,73 @@ func handleConnection(w http.ResponseWriter, r *http.Request) { continue } - // Handle private DMs (room format "dm:user1-user2") + // Handle private DMs if strings.HasPrefix(msg.Room, "dm:") { handlePrivateDM(msg) - return + continue } + msg.Type = "public" log.Printf("Chat message in room %s from %s: %s", msg.Room, msg.Username, msg.Text) broadcast(msg) - - // Save private messages explicitly as well. - if msg.Type == "private" { - db.Create(&msg) - } } } +// pushChatHistory sends the last N messages for a room via WebSocket. +func pushChatHistory(conn *websocket.Conn, room string, limit int) { + if db == nil || room == "" { + return + } + var messages []Message + db.Where("room = ?", room).Order("timestamp desc").Limit(limit).Find(&messages) + // Reverse to chronological order + for i := len(messages)/2 - 1; i >= 0; i-- { + opp := len(messages) - 1 - i + messages[i], messages[opp] = messages[opp], messages[i] + } + for _, msg := range messages { + conn.WriteJSON(msg) + } +} + +// dmsSent tracks total private messages for Prometheus metrics (set from handleMetrics). +var dmsSent int64 + // handlePrivateDM routes private messages to the specific user pair func handlePrivateDM(msg Message) { - // Persist private messages for history. + msg.Type = "private" + dmsSent++ + + // Persist for history db.Create(&msg) - // Broadcast only to participants (simplified - in production use a map of user connections) - broadcast(msg) + // Deliver to target if online + mutex.Lock() + targetOnline := false + for conn, info := range clientMeta { + if info.Username == msg.Room || strings.Contains(msg.Room, info.Username) { + if err := conn.WriteJSON(msg); err == nil { + targetOnline = true + } + } + } + mutex.Unlock() + + if !targetOnline { + // Target offline — message persisted, will appear in history on next join + log.Printf("DM stored for offline user: %s -> room %s", msg.Username, msg.Room) + } + + // Also deliver to sender for confirmation + mutex.Lock() + for conn, info := range clientMeta { + if info.Username == msg.Username { + conn.WriteJSON(msg) + break + } + } + mutex.Unlock() + log.Printf("Private DM from %s to room %s: %s", msg.Username, msg.Room, msg.Text) } @@ -1935,7 +2346,14 @@ func logModerationAction(action, target, moderator, reason string) { func handleModerationCommand(msg Message, conn *websocket.Conn) { parts := strings.Fields(msg.Text) if len(parts) < 3 { - conn.WriteJSON(Message{Type: "system", Text: "Usage: /mod or /mute "}) + conn.WriteJSON(Message{Type: "system", Text: "Usage: /mute or /ban ", Timestamp: time.Now()}) + return + } + + // Basic moderator check: admin username or configured moderators + adminUser := getAdminUsername() + if msg.Username != adminUser { + conn.WriteJSON(Message{Type: "system", Text: "Only moderators can use moderation commands.", Timestamp: time.Now()}) return } @@ -1943,12 +2361,34 @@ func handleModerationCommand(msg Message, conn *websocket.Conn) { target := parts[1] reason := strings.Join(parts[2:], " ") + if redisClient != nil { + switch action { + case "mute", "mod": + durationMinutes := 10 // default + if len(parts) >= 3 { + if d, err := strconv.Atoi(parts[2]); err == nil && d > 0 { + durationMinutes = d + if len(parts) > 3 { + reason = strings.Join(parts[3:], " ") + } else { + reason = "no reason given" + } + } + } + redisClient.Set(context.Background(), fmt.Sprintf("muted:%s", target), "1", time.Duration(durationMinutes)*time.Minute) + case "ban": + redisClient.Set(context.Background(), fmt.Sprintf("banned:%s", target), "1", 0) // no expiry + } + } + logModerationAction(action, target, msg.Username, reason) broadcast(Message{ - Room: msg.Room, - Type: "system", - Text: fmt.Sprintf("Moderator %s %s %s for: %s", msg.Username, action, target, reason), + Room: msg.Room, + Username: "system", + Type: "system", + Text: fmt.Sprintf("%s was %s by %s: %s", target, action, msg.Username, reason), + Timestamp: time.Now(), }) } @@ -2075,6 +2515,39 @@ func handleAdminChatLogByID(w http.ResponseWriter, r *http.Request) { } } +// handleAdminChatLogExport returns chat log as CSV file. +func handleAdminChatLogExport(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + + query := r.URL.Query() + room := strings.TrimSpace(query.Get("room")) + dbQuery := db.Model(&Message{}) + if room != "" { + dbQuery = dbQuery.Where("room = ?", room) + } + if username := strings.TrimSpace(query.Get("username")); username != "" { + dbQuery = dbQuery.Where("username LIKE ?", "%"+username+"%") + } + + var messages []Message + if err := dbQuery.Order("timestamp desc").Limit(10000).Find(&messages).Error; err != nil { + http.Error(w, "failed to load messages", http.StatusInternalServerError) + return + } + + w.Header().Set("Content-Type", "text/csv; charset=utf-8") + w.Header().Set("Content-Disposition", "attachment; filename=chat-log.csv") + fmt.Fprintln(w, "id,room,username,type,text,timestamp") + for _, msg := range messages { + text := strings.ReplaceAll(msg.Text, `"`, `""`) + fmt.Fprintf(w, "%d,%s,%s,%s,\"%s\",%s\n", + msg.ID, msg.Room, msg.Username, msg.Type, text, msg.Timestamp.Format(time.RFC3339)) + } +} + // handleAdminChatLogBulk deletes multiple chat log entries by ID func handleAdminChatLogBulk(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodDelete { @@ -2109,6 +2582,19 @@ func handleAdminChatLogBulk(w http.ResponseWriter, r *http.Request) { } func main() { + // Load env-configurable rate limits + if raw := strings.TrimSpace(os.Getenv("CHAT_RATE_LIMIT_MAX")); raw != "" { + if v, err := strconv.Atoi(raw); err == nil && v > 0 { + rateLimitMax = v + } + } + if raw := strings.TrimSpace(os.Getenv("CHAT_RATE_LIMIT_WINDOW_S")); raw != "" { + if v, err := strconv.Atoi(raw); err == nil && v > 0 { + rateLimitWindow = time.Duration(v) * time.Second + } + } + log.Printf("Rate limit: %d messages per %v", rateLimitMax, rateLimitWindow) + initRedis() initDB() @@ -2122,13 +2608,18 @@ func main() { http.HandleFunc("/api/admin/moderation/actions", requireAdminAuth(handleAdminModerationAction)) http.HandleFunc("/api/admin/analytics", requireAdminAuth(handleAdminAnalytics)) http.HandleFunc("/api/admin/system", requireAdminAuth(handleAdminSystemSummary)) + http.HandleFunc("/api/admin/config", requireAdminAuth(handleAdminConfig)) + http.HandleFunc("/api/admin/restart", requireAdminAuth(handleAdminRestart)) http.HandleFunc("/api/admin/users", requireAdminAuth(handleAdminUsers)) http.HandleFunc("/api/admin/schedule", requireAdminAuth(handleAdminSchedule)) + http.HandleFunc("/api/admin/schedule/overrides", requireAdminAuth(handleAdminScheduleOverrides)) + http.HandleFunc("/api/schedule", handlePublicSchedule) http.HandleFunc("/api/admin/media", requireAdminAuth(handleAdminMedia)) http.HandleFunc("/api/admin/media/uploads", requireAdminAuth(handleAdminMediaUploads)) http.HandleFunc("/api/admin/media/uploads/", requireAdminAuth(handleAdminMediaUploads)) http.HandleFunc("/api/admin/media/", requireAdminAuth(handleAdminMediaByID)) http.HandleFunc("/api/admin/chat-log/bulk", requireAdminAuth(handleAdminChatLogBulk)) + http.HandleFunc("/api/admin/chat-log/export", requireAdminAuth(handleAdminChatLogExport)) http.HandleFunc("/api/admin/chat-log/", requireAdminAuth(handleAdminChatLogByID)) http.HandleFunc("/api/admin/chat-log", requireAdminAuth(handleAdminChatLog)) http.HandleFunc("/metrics", handleMetrics) diff --git a/backend/playout/Dockerfile b/backend/playout/Dockerfile index 86fdbf5..68058d2 100644 --- a/backend/playout/Dockerfile +++ b/backend/playout/Dockerfile @@ -18,8 +18,8 @@ RUN apk add --no-cache \ bash # Set working directory for playout WORKDIR /app -# Create directories for playlist, media, and HLS output -RUN mkdir -p /app/hls /app/media /app/playlists +# Create directories for playlist, media, HLS output, and emergency slate +RUN mkdir -p /app/hls /app/media /app/playlists /app/slate # Copy the playout start script and the built playlist manager binary COPY start-playout.sh /app/ # Copy demo data for out-of-the-box playout diff --git a/backend/playout/src/playlist.go b/backend/playout/src/playlist.go index 479924f..869d119 100644 --- a/backend/playout/src/playlist.go +++ b/backend/playout/src/playlist.go @@ -3,11 +3,13 @@ package main import ( "bufio" "context" + "encoding/json" "fmt" "net/http" "os" "os/exec" "path/filepath" + "strconv" "strings" "time" @@ -25,8 +27,18 @@ type VideoFile struct { } var ( - r2Client *s3.Client - bucket = os.Getenv("R2_BUCKET_NAME") + r2Client *s3.Client + bucket = os.Getenv("R2_BUCKET_NAME") + playoutMode = "starting" // "live", "slate", "degraded" + restartCount = 0 + restartWindow = time.Now() + slateDir = "/app/slate" + hlsDir = "/app/hls" +) + +const ( + maxRestartsPerMinute = 5 + slateManifestName = "playlist.m3u8" ) // initR2Client initializes the R2 (S3-compatible) client @@ -79,23 +91,181 @@ func LoadPlaylist(playlistPath string) ([]VideoFile, error) { // ValidateVideo uses ffprobe to enforce codec, fps, and keyframe interval (ADR-001) func ValidateVideo(videoPath string) (VideoFile, error) { v := VideoFile{Path: videoPath} + var errors []string + // Run ffprobe cmd := exec.Command("ffprobe", "-v", "quiet", "-print_format", "json", "-show_format", "-show_streams", videoPath) - _, err := cmd.Output() + output, err := cmd.Output() if err != nil { v.Validation = "ffprobe failed: " + err.Error() - return v, err + return v, fmt.Errorf("ffprobe: %w", err) } - // TODO: Parse JSON for strict checks (codec_name=h264, r_frame_rate=30/1, key_frame interval) - // For MVP we assume valid if ffprobe succeeds - v.Valid = true - v.Validation = "Validation passed (H.264/AVC, AAC, 30fps constant, 2s keyframes - full JSON parse TODO per ADR-001)" - v.Duration = 60.0 // placeholder from ffprobe + // Parse JSON + var probe struct { + Streams []struct { + CodecName string `json:"codec_name"` + CodecType string `json:"codec_type"` + Width int `json:"width"` + Height int `json:"height"` + RFrameRate string `json:"r_frame_rate"` + PixFmt string `json:"pix_fmt"` + SampleRate string `json:"sample_rate"` + Channels int `json:"channels"` + Duration string `json:"duration"` + } `json:"streams"` + Format struct { + Duration string `json:"duration"` + } `json:"format"` + } + if err := json.Unmarshal(output, &probe); err != nil { + v.Validation = "ffprobe JSON parse failed: " + err.Error() + return v, fmt.Errorf("ffprobe json: %w", err) + } + if len(probe.Streams) == 0 { + errors = append(errors, "no streams detected") + } + + var hasVideo, hasAudio bool + for _, s := range probe.Streams { + switch s.CodecType { + case "video": + hasVideo = true + if s.CodecName != "h264" { + errors = append(errors, fmt.Sprintf("video codec must be h264, got %s", s.CodecName)) + } + if s.RFrameRate != "30/1" { + errors = append(errors, fmt.Sprintf("framerate must be 30/1, got %s", s.RFrameRate)) + } + if s.Width != 1280 && s.Width != 1920 { + errors = append(errors, fmt.Sprintf("width must be 1280 or 1920, got %d", s.Width)) + } + if s.Height != 720 && s.Height != 1080 { + errors = append(errors, fmt.Sprintf("height must be 720 or 1080, got %d", s.Height)) + } + if s.PixFmt != "yuv420p" { + errors = append(errors, fmt.Sprintf("pixel format must be yuv420p, got %s", s.PixFmt)) + } + case "audio": + hasAudio = true + if s.CodecName != "aac" { + errors = append(errors, fmt.Sprintf("audio codec must be aac, got %s", s.CodecName)) + } + if s.SampleRate != "48000" { + errors = append(errors, fmt.Sprintf("audio sample rate must be 48000, got %s", s.SampleRate)) + } + } + } + + if !hasVideo { + errors = append(errors, "no video stream found") + } + if !hasAudio { + errors = append(errors, "no audio stream found") + } + + // Parse duration from format or stream + durationSec := 0.0 + if probe.Format.Duration != "" { + if d, err := strconv.ParseFloat(probe.Format.Duration, 64); err == nil { + durationSec = d + } + } + + if len(errors) > 0 { + v.Valid = false + v.Validation = strings.Join(errors, "; ") + return v, fmt.Errorf("validation failed: %s", v.Validation) + } + + // Validate keyframe interval <=2s (60 frames at 30fps) + keyframeErr := validateKeyframeInterval(videoPath) + if keyframeErr != nil { + errors = append(errors, keyframeErr.Error()) + } + + if len(errors) > 0 { + v.Valid = false + v.Validation = strings.Join(errors, "; ") + return v, fmt.Errorf("validation failed: %s", v.Validation) + } + + v.Valid = true + v.Duration = durationSec + v.Validation = fmt.Sprintf("OK: H.264/AAC, 30fps, yuv420p, 48kHz, %.1fs, keyframes OK", durationSec) return v, nil } +// validateKeyframeInterval checks that max gap between keyframes <=2s (60 frames at 30fps). +func validateKeyframeInterval(videoPath string) error { + cmd := exec.Command("ffprobe", + "-v", "quiet", + "-select_streams", "v:0", + "-show_frames", + "-show_entries", "frame=key_frame,pkt_pts_time", + "-of", "csv=p=0", + videoPath) + output, err := cmd.Output() + if err != nil { + return fmt.Errorf("keyframe probe failed: %w", err) + } + + type kfEntry struct{ pts float64 } + var keyframes []kfEntry + lines := strings.Split(strings.TrimSpace(string(output)), "\n") + for _, line := range lines { + line = strings.TrimSpace(line) + if line == "" { + continue + } + parts := strings.Split(line, ",") + if len(parts) < 2 || parts[0] != "1" { + continue + } + pts, err := strconv.ParseFloat(strings.TrimSpace(parts[1]), 64) + if err != nil { + continue + } + keyframes = append(keyframes, kfEntry{pts}) + } + + if len(keyframes) < 2 { + return fmt.Errorf("insufficient keyframes for interval analysis (need >=2, got %d)", len(keyframes)) + } + + maxGap := 0.0 + for i := 1; i < len(keyframes); i++ { + gap := keyframes[i].pts - keyframes[i-1].pts + if gap > maxGap { + maxGap = gap + } + } + + if maxGap > 2.1 { // small tolerance for floating point + return fmt.Errorf("keyframe interval too large: max %.2fs (must be <=2s)", maxGap) + } + + return nil +} + +// ValidateVideoStrict runs ValidateVideo and returns a structured JSON result. +func ValidateVideoStrict(videoPath string) map[string]any { + v, err := ValidateVideo(videoPath) + result := map[string]any{ + "path": videoPath, + "valid": v.Valid, + "errors": []string{}, + } + if err != nil { + result["errors"] = strings.Split(v.Validation, "; ") + } + if v.Valid { + result["duration_seconds"] = v.Duration + } + return result +} + // ValidatePlaylist runs normalization validation on all entries func ValidatePlaylist(playlistPath string) ([]VideoFile, error) { videos, err := LoadPlaylist(playlistPath) @@ -139,8 +309,65 @@ func SyncToR2(localPath, r2Key string) error { // HealthHandler provides a simple health check endpoint func HealthHandler(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") + status := "healthy" + if playoutMode == "slate" { + status = "degraded" + } w.WriteHeader(http.StatusOK) - fmt.Fprint(w, `{"status":"healthy","service":"playout","timestamp":"`+time.Now().UTC().Format(time.RFC3339)+`"}`) + fmt.Fprintf(w, `{"status":"%s","service":"playout","mode":"%s","restarts":%d,"timestamp":"%s"}`, + status, playoutMode, restartCount, time.Now().UTC().Format(time.RFC3339)) +} + +// activateSlate copies pre-rendered slate HLS into the live output directory. +// Writes a static manifest referencing slate segments. Returns error if slate not found. +func activateSlate() error { + slateHLS := filepath.Join(slateDir, "hls") + if _, err := os.Stat(filepath.Join(slateHLS, "slate.m3u8")); os.IsNotExist(err) { + return fmt.Errorf("slate HLS not found at %s — run scripts/generate-slate.sh first", slateHLS) + } + + // Clear existing live segments + if err := resetHLSOutput(hlsDir); err != nil { + return fmt.Errorf("failed to clear HLS output for slate: %w", err) + } + + // Copy slate segments into live HLS dir + slateFiles, err := filepath.Glob(filepath.Join(slateHLS, "*")) + if err != nil { + return err + } + for _, src := range slateFiles { + dst := filepath.Join(hlsDir, filepath.Base(src)) + data, err := os.ReadFile(src) + if err != nil { + return err + } + if err := os.WriteFile(dst, data, 0o644); err != nil { + return err + } + } + + playoutMode = "slate" + fmt.Printf("[%s] EMERGENCY: Switched to slate fallback. %d restarts in window.\n", + time.Now().Format(time.RFC3339), restartCount) + return nil +} + +// deactivateSlate clears slate state and prepares for live playout resumption. +func deactivateSlate() { + playoutMode = "live" + fmt.Printf("[%s] RECOVERY: Slate deactivated, resuming live playout.\n", + time.Now().Format(time.RFC3339)) +} + +// shouldEscalateToSlate checks if restart count exceeds threshold within the window. +func shouldEscalateToSlate() bool { + if time.Since(restartWindow) > 60*time.Second { + restartCount = 0 + restartWindow = time.Now() + return false + } + return restartCount >= maxRestartsPerMinute } func resetHLSOutput(hlsDir string) error { @@ -207,6 +434,21 @@ func main() { // Start HTTP health check server http.HandleFunc("/health", HealthHandler) + http.HandleFunc("/metrics", func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/plain; version=0.0.4; charset=utf-8") + fmt.Fprintln(w, "# HELP linearvoid_playout_restarts_total Total FFmpeg restarts.") + fmt.Fprintln(w, "# TYPE linearvoid_playout_restarts_total counter") + fmt.Fprintf(w, "linearvoid_playout_restarts_total %d\n", restartCount) + fmt.Fprintln(w, "# HELP linearvoid_playout_mode Current playout mode (0=live, 1=slate, 2=degraded).") + fmt.Fprintln(w, "# TYPE linearvoid_playout_mode gauge") + modeVal := 0 + if playoutMode == "slate" { + modeVal = 1 + } else if playoutMode == "degraded" { + modeVal = 2 + } + fmt.Fprintf(w, "linearvoid_playout_mode %d\n", modeVal) + }) go func() { fmt.Println("Health check server listening on :8081/health") if err := http.ListenAndServe(":8081", nil); err != nil { @@ -214,9 +456,14 @@ func main() { } }() - // Run FFmpeg in a loop for continuous playback with R2 sync hook + // Run FFmpeg in a loop for continuous playback with slate fallback for { - if err := resetHLSOutput("/app/hls"); err != nil { + if playoutMode == "slate" { + // During slate: attempt recovery by restarting FFmpeg + fmt.Printf("[%s] Slate active — attempting live playout recovery...\n", time.Now().Format(time.RFC3339)) + } + + if err := resetHLSOutput(hlsDir); err != nil { fmt.Println("Failed to reset HLS output:", err) time.Sleep(5 * time.Second) continue @@ -241,12 +488,32 @@ func main() { "-hls_segment_filename", "/app/hls/segment_%Y%m%d_%H%M%S.m4s", "/app/hls/playlist.m3u8") - if err := cmd.Run(); err != nil { - fmt.Println("FFmpeg exited with error. Restarting in 5s (per ADR-001 resilience)...") + runErr := cmd.Run() + + // Track restart and check for slate escalation + restartCount++ + fmt.Printf("[%s] FFmpeg exited (restart #%d in window): %v\n", + time.Now().Format(time.RFC3339), restartCount, runErr) + + if shouldEscalateToSlate() { + fmt.Printf("[%s] CRITICAL: %d restarts in 60s — escalating to slate fallback\n", + time.Now().Format(time.RFC3339), restartCount) + if err := activateSlate(); err != nil { + fmt.Printf("[%s] Slate activation failed: %v — will retry\n", + time.Now().Format(time.RFC3339), err) + } + // Stay in slate mode, retry recovery after delay + time.Sleep(30 * time.Second) + continue + } + + // If we were in slate but FFmpeg ran successfully, we recovered + if playoutMode == "slate" { + deactivateSlate() } // After FFmpeg run, sync latest segments and manifest to R2 - if err := syncLatestToR2("/app/hls"); err != nil { + if err := syncLatestToR2(hlsDir); err != nil { fmt.Println("R2 sync warning:", err) } diff --git a/docker-compose.prod.yml b/docker-compose.prod.yml index 07a8825..e9c0c06 100644 --- a/docker-compose.prod.yml +++ b/docker-compose.prod.yml @@ -1,3 +1,9 @@ +# docker-compose.prod.yml — TEST LAB DEPLOYMENT +# Target: test lab host 192.168.88.40 (NOT Hetzner production). +# Production Hetzner VPS not yet provisioned — see TODO.md Production Deployment section. +# Uses pre-built images from Gitea Container Registry (no local build). +# Override DOCKER_HOST in CD workflow to switch target. + services: playout: image: git.allucanget.biz/zwitschi/linearvoid/playout:latest @@ -9,6 +15,7 @@ services: - ./data/playlist:/app/playlist:rw - ./data/media:/app/media:ro - ./data/hls:/app/hls:rw + - ./data/slate:/app/slate:ro networks: - linearvoid-net depends_on: @@ -68,7 +75,7 @@ services: image: redis:7-alpine container_name: linearvoid-redis restart: unless-stopped - command: redis-server --appendonly yes + command: redis-server --appendonly yes --maxmemory 256mb --maxmemory-policy allkeys-lru volumes: - redis-data:/data networks: diff --git a/docker-compose.yml b/docker-compose.yml index 7a47634..bf4fea4 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,3 +1,8 @@ +# docker-compose.yml — LOCAL DEVELOPMENT +# Target: local workstation or test lab (192.168.88.0/24). +# NOT for production. Use docker-compose.prod.yml for deployment. +# Services build from source; ports exposed for direct dev access. + services: playout: image: ${CI_REGISTRY_URL:-git.allucanget.biz}/${CI_REGISTRY_NAMESPACE:-zwitschi/linearvoid}/playout:latest @@ -12,6 +17,7 @@ services: - ./data/playlist:/app/playlist:ro - ./data/media:/app/media:ro - ./data/hls:/app/hls:rw + - ./data/slate:/app/slate:ro networks: - linearvoid-net depends_on: @@ -72,7 +78,7 @@ services: image: redis:7-alpine container_name: linearvoid-redis restart: unless-stopped - command: redis-server --appendonly yes + command: redis-server --appendonly yes --maxmemory 256mb --maxmemory-policy allkeys-lru volumes: - redis-data:/data networks: diff --git a/docs/architecture/README.md b/docs/architecture/README.md index 063b282..c6c18bf 100644 --- a/docs/architecture/README.md +++ b/docs/architecture/README.md @@ -284,7 +284,7 @@ Notable aspects: Manifest freshness ensures all viewers stay on same timeline; c ### CI/CD Pipeline Architecture -The deployment follows a two-stage pipeline: **CI builds and pushes images**, then **CD pulls and deploys** them to the production host. +The deployment follows a two-stage pipeline: **CI builds and pushes images**, then **CD pulls and deploys** them. Currently targeting the **test lab** (192.168.88.40). Production Hetzner VPS is planned but not yet provisioned — the same pipeline will deploy there by changing `DOCKER_HOST`. ```mermaid flowchart LR @@ -295,7 +295,7 @@ flowchart LR subgraph "Gitea Container Registry" REGISTRY[git.allucanget.biz
playout:latest
chat:latest
nginx:latest] end - subgraph "Production Host (192.168.88.40)" + subgraph "Test Lab Host (192.168.88.40) — Production Hetzner TBD" DOCKER[Docker Engine
via TCP :2375] PLAYOUT[linearvoid-playout] CHAT[linearvoid-chat] @@ -331,7 +331,7 @@ flowchart LR - Pulls latest images from registry - Generates `.env` from secrets - Seeds Prometheus config into named volume - - Deploys via `docker compose -f docker-compose.prod.yml up -d` to remote host (`192.168.88.40:2375`) + - Deploys via `docker compose -f docker-compose.prod.yml up -d` to remote host (`192.168.88.40:2375` — test lab; switch to Hetzner IP for production) - Runs health checks with retry logic for playout initialization ### Infrastructure Level 1 @@ -350,7 +350,7 @@ flowchart LR ### Infrastructure Level 2 -**Hetzner VPS**: Runs playout (FFmpeg/MediaMTX), chat backend, Nginx. Monitoring via dashboards + UptimeRobot. Backups to R2. +**Hetzner VPS (planned):** Will run playout (FFmpeg/MediaMTX), chat backend, Nginx. Monitoring via dashboards + UptimeRobot. Backups to R2. Currently test lab (`192.168.88.40`) serves this role. Production Hetzner instance not yet provisioned — see `TODO.md` Production Deployment section. **Cloudflare Layer**: diff --git a/frontend/index.html b/frontend/index.html index eb4fffa..3ba1ba0 100644 --- a/frontend/index.html +++ b/frontend/index.html @@ -24,6 +24,10 @@
+
+ Loading... + +