linearvoid/backend/playout/src/playlist.go

package main

import (
	"bufio"
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"net/http"
	"os"
	"os/exec"
	"path/filepath"
	"runtime"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/config"
	"github.com/aws/aws-sdk-go-v2/service/s3"
	"github.com/aws/aws-sdk-go-v2/service/s3/types"
	"github.com/fsnotify/fsnotify"
)

// VideoFile represents a normalized video entry in the playlist
type VideoFile struct {
	Path       string
	Duration   float64
	Valid      bool
	Validation string
}

var (
	r2Client      *s3.Client
	putObject     = func(client *s3.Client, input *s3.PutObjectInput) (*s3.PutObjectOutput, error) {
		return client.PutObject(context.TODO(), input)
	}
	listObjects = func(client *s3.Client, input *s3.ListObjectsV2Input) (*s3.ListObjectsV2Output, error) {
		return client.ListObjectsV2(context.TODO(), input)
	}
	deleteObjects = func(client *s3.Client, input *s3.DeleteObjectsInput) (*s3.DeleteObjectsOutput, error) {
		return client.DeleteObjects(context.TODO(), input)
	}
	copyObject = func(client *s3.Client, input *s3.CopyObjectInput) (*s3.CopyObjectOutput, error) {
		return client.CopyObject(context.TODO(), input)
	}
	bucket             = os.Getenv("R2_BUCKET_NAME")
	playoutMode        = "starting" // "live", "slate", "degraded"
	restartCount       = 0
	restartWindow      = time.Now()
	slateDir           = "/app/slate"
	hlsDir             = "/app/hls"
	hlsSegmentDuration = envWithDefault("HLS_SEGMENT_DURATION", "2")
	hlsWindowSize      = envWithDefault("HLS_WINDOW_SIZE", "12")

	// Unique segment naming (STREAMING.md contract: r-{runID}/seg-{seq}.m4s)
	runID       string
	segSequence uint64

	// Health endpoint detail (STREAMING.md contract)
	activeItem          string
	playlistRevision    string
	lastSegmentSeq      uint64
	lastR2PubTime       time.Time
	consecutiveFails    int
	lastFFmpegStartTime time.Time
	lastFFmpegExitError error

	// R2 retention cleanup counter
	r2CleanupObjectsTotal int64

	// R2 retention gauges
	r2RunsRetained  int64
	r2BytesRetained int64

	// Continuous playout loop metrics
	loopsTotal      int64
	lastRunDuration time.Duration
	lastRunClean    bool
)

const (
	maxRestartsPerMinute = 5
	slateManifestName    = "playlist.m3u8"
	// Cache-Control header values for R2 uploads.
	segmentCacheControl  = "public, max-age=31536000, immutable"
	manifestCacheControl = "public, max-age=1, s-maxage=2, must-revalidate"

	// R2 retention policy.
	defaultR2KeepRuns          = 2
	defaultR2MaxRetentionBytes = int64(10 * 1024 * 1024 * 1024) // 10 GiB
)

// envWithDefault returns the value of the environment variable or the default if not set.
func envWithDefault(key, defaultVal string) string {
	if v := os.Getenv(key); v != "" {
		return v
	}
	return defaultVal
}

// isR2Configured returns true when the required R2 environment variables are set.
func isR2Configured() bool {
	return os.Getenv("R2_ENDPOINT") != "" &&
		os.Getenv("R2_ACCESS_KEY_ID") != "" &&
		os.Getenv("R2_SECRET_ACCESS_KEY") != "" &&
		os.Getenv("R2_BUCKET_NAME") != ""
}

// initR2Client initializes the R2 (S3-compatible) client.
// Returns an error if required environment variables are missing.
func initR2Client() error {
	if !isR2Configured() {
		return fmt.Errorf("R2 not configured: R2_ENDPOINT, R2_ACCESS_KEY_ID, R2_SECRET_ACCESS_KEY, or R2_BUCKET_NAME is empty")
	}

	cfg, err := config.LoadDefaultConfig(context.TODO(),
		config.WithRegion("auto"),
		config.WithCredentialsProvider(aws.CredentialsProviderFunc(func(ctx context.Context) (aws.Credentials, error) {
			return aws.Credentials{
				AccessKeyID:     os.Getenv("R2_ACCESS_KEY_ID"),
				SecretAccessKey: os.Getenv("R2_SECRET_ACCESS_KEY"),
			}, nil
		})),
	)
	if err != nil {
		return err
	}
	r2Client = s3.NewFromConfig(cfg, func(o *s3.Options) {
		o.BaseEndpoint = aws.String(os.Getenv("R2_ENDPOINT"))
	})
	return nil
}

// initRunID generates a unique run/session ID for globally unique segment naming (STREAMING.md contract).
func initRunID() {
	runID = "r-" + time.Now().UTC().Format("20060102T150405Z")
	segSequence = 0
}

// watchAndPublishSegments watches the HLS output directory for new .m4s segments
// and uploads them to R2 in real time with runID-based keys (STREAMING.md contract).
func watchAndPublishSegments(ctx context.Context, hlsOutputDir string) {
	watcher, err := fsnotify.NewWatcher()
	if err != nil {
		fmt.Println("Warning: fsnotify unavailable, R2 sync falls back to batch:", err)
		return
	}
	defer watcher.Close()

	if err := watcher.Add(hlsOutputDir); err != nil {
		fmt.Println("Warning: cannot watch HLS dir:", err)
		return
	}

	for {
		select {
		case event, ok := <-watcher.Events:
			if !ok {
				return
			}
			if event.Op&fsnotify.Create != 0 && strings.HasSuffix(event.Name, ".m4s") {
				// Wait for FFmpeg to finish writing the segment
				time.Sleep(500 * time.Millisecond)
				segSequence++
				lastSegmentSeq = segSequence
				key := "hls/" + runID + "/" + filepath.Base(event.Name)
				if err := SyncToR2(event.Name, key); err != nil {
					consecutiveFails++
					fmt.Printf("Failed to sync segment %s: %v\n", event.Name, err)
				} else {
					consecutiveFails = 0
					lastR2PubTime = time.Now()
				}
			}
		case err, ok := <-watcher.Errors:
			if !ok {
				return
			}
			fmt.Println("fsnotify error:", err)
		case <-ctx.Done():
			return
		}
	}
}

// publishManifest uploads the manifest atomically: temp key first, then copy to final key (STREAMING.md contract).
// Bare media URIs in the manifest are rewritten to the run-scoped R2 layout
// (r-{runID}/…) so that every referenced segment/init resolves on the media host.
func publishManifest(localManifestPath string) error {
	if r2Client == nil {
		if err := initR2Client(); err != nil {
			return err
		}
	}

	data, err := os.ReadFile(localManifestPath)
	if err != nil {
		return fmt.Errorf("read manifest failed: %w", err)
	}

	// Rewrite bare relative media URIs to the current run folder so the manifest
	// matches the run-scoped segment keys uploaded by watchAndPublishSegments /
	// syncLatestToR2 (candidate (a)). Slate mode is skipped: its manifest references
	// slate assets that are not published under the live run prefix.
	if playoutMode != "slate" {
		data = rewriteManifestToRun(runID, data)
	}

	// Stage the (rewritten) manifest locally so the existing SyncToR2 path is reused.
	staging := localManifestPath + ".r2staging"
	if err := os.WriteFile(staging, data, 0o644); err != nil {
		return fmt.Errorf("write staging manifest failed: %w", err)
	}
	defer os.Remove(staging)

	// 1. Upload to temp key
	if err := SyncToR2(staging, "hls/playlist.m3u8.tmp"); err != nil {
		return fmt.Errorf("temp manifest upload failed: %w", err)
	}

	// 2. Atomic copy to final key
	_, err = copyObject(r2Client, &s3.CopyObjectInput{
		Bucket:     aws.String(bucket),
		CopySource: aws.String(bucket + "/hls/playlist.m3u8.tmp"),
		Key:        aws.String("hls/playlist.m3u8"),
	})
	if err != nil {
		return fmt.Errorf("manifest atomic swap failed: %w", err)
	}

	return nil
}

// isMediaURI reports whether s names an HLS media file (a segment or init file).
func isMediaURI(s string) bool {
	name := s
	if i := strings.IndexByte(name, '?'); i >= 0 {
		name = name[:i]
	}
	switch strings.ToLower(filepath.Ext(name)) {
	case ".m4s", ".ts", ".mp4", ".m4a":
		return true
	}
	return false
}

// rewriteManifestToRun prefixes bare relative media URIs in an HLS manifest with the
// run folder, e.g. "seg-000001.m4s" -> "r-20260903T…/seg-000001.m4s" and
// #EXT-X-MAP:URI="init.mp4" -> URI="r-20260903T…/init.mp4". This keeps the published
// manifest consistent with the run-scoped R2 layout (STREAMING.md contract). Directives,
// comments, blank lines, and URIs that already contain a path are left unchanged.
// Returns data unchanged when runID is empty.
func rewriteManifestToRun(runID string, data []byte) []byte {
	if runID == "" {
		return data
	}
	prefix := runID + "/"
	lines := strings.Split(string(data), "\n")
	out := make([]string, 0, len(lines))
	for _, line := range lines {
		trimmed := strings.TrimSpace(line)
		switch {
		case strings.HasPrefix(trimmed, "#EXT-X-MAP:URI="):
			if open := strings.Index(trimmed, `URI="`); open >= 0 {
				rest := trimmed[open+len(`URI="`):]
				if closeIdx := strings.IndexByte(rest, '"'); closeIdx > 0 {
					uri := rest[:closeIdx]
					if uri != "" && !strings.Contains(uri, "/") {
						line = strings.Replace(line, `URI="`+uri+`"`, `URI="`+prefix+uri+`"`, 1)
					}
				}
			}
			out = append(out, line)
		case trimmed != "" && !strings.HasPrefix(trimmed, "#") && !strings.Contains(trimmed, "/") && isMediaURI(trimmed):
			out = append(out, prefix+trimmed)
		default:
			out = append(out, line)
		}
	}
	return []byte(strings.Join(out, "\n"))
}

// LoadPlaylist reads a playlist file and returns a list of VideoFile entries
func LoadPlaylist(playlistPath string) ([]VideoFile, error) {
	file, err := os.Open(playlistPath)
	if err != nil {
		return nil, err
	}
	defer file.Close()

	var videos []VideoFile
	scanner := bufio.NewScanner(file)
	for scanner.Scan() {
		line := strings.TrimSpace(scanner.Text())
		if line == "" || strings.HasPrefix(line, "#") {
			continue
		}

		if strings.HasPrefix(line, "file ") {
			line = strings.TrimSpace(strings.TrimPrefix(line, "file "))
			line = strings.Trim(line, "'\"")
		}

		videos = append(videos, VideoFile{Path: line})
	}

	return videos, scanner.Err()
}

// ValidateVideo uses ffprobe to enforce codec, fps, and keyframe interval (ADR-001)
func ValidateVideo(videoPath string) (VideoFile, error) {
	v := VideoFile{Path: videoPath}

	// Run the stream/format probe and the keyframe probe concurrently: they are
	// independent, and the keyframe probe (-show_frames) is the expensive one.
	var streamOut []byte
	var streamErr error
	var keyframeErr error
	var wg sync.WaitGroup
	wg.Add(2)
	go func() {
		defer wg.Done()
		cmd := exec.Command("ffprobe", "-v", "quiet", "-print_format", "json", "-show_format", "-show_streams", videoPath)
		streamOut, streamErr = cmd.Output()
	}()
	go func() {
		defer wg.Done()
		keyframeErr = validateKeyframeInterval(videoPath)
	}()
	wg.Wait()

	if streamErr != nil {
		v.Validation = "ffprobe failed: " + streamErr.Error()
		return v, fmt.Errorf("ffprobe: %w", streamErr)
	}

	var errors []string

	// 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(streamOut, &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)
	}

	// Keyframe interval check (probe already completed concurrently above).
	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,best_effort_timestamp_time,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
		}
		var pts float64
		var found bool
		for _, raw := range parts[1:] {
			if value, err := strconv.ParseFloat(strings.TrimSpace(raw), 64); err == nil {
				pts = value
				found = true
				break
			}
		}
		if !found {
			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
}

// validateWorkers returns the number of concurrent validation workers.
func validateWorkers() int {
	n := runtime.NumCPU()
	if v := os.Getenv("PLAYOUT_VALIDATE_WORKERS"); v != "" {
		if parsed, err := strconv.Atoi(v); err == nil && parsed > 0 {
			n = parsed
		}
	}
	return n
}

// ValidatePlaylist runs normalization validation on all entries in parallel,
// deduplicating repeated paths so each unique file is probed once.
func ValidatePlaylist(playlistPath string) ([]VideoFile, error) {
	videos, err := LoadPlaylist(playlistPath)
	if err != nil {
		return nil, err
	}

	// Collect unique paths in order of first appearance.
	var order []string
	seen := make(map[string]bool, len(videos))
	for _, v := range videos {
		if !seen[v.Path] {
			seen[v.Path] = true
			order = append(order, v.Path)
		}
	}

	// Validate unique paths concurrently with a bounded worker pool.
	results := make(map[string]VideoFile, len(order))
	var mu sync.Mutex
	var wg sync.WaitGroup
	sem := make(chan struct{}, validateWorkers())
	for _, path := range order {
		wg.Add(1)
		sem <- struct{}{}
		go func(p string) {
			defer wg.Done()
			defer func() { <-sem }()
			vf, err := ValidateVideo(p)
			if err != nil {
				fmt.Printf("Warning: %s - %s\n", p, err)
			}
			mu.Lock()
			results[p] = vf
			mu.Unlock()
		}(path)
	}
	wg.Wait()

	// Map results back onto the playlist (preserving order and duplicates).
	for i := range videos {
		videos[i] = results[videos[i].Path]
	}

	return videos, nil
}

// SyncToR2 uploads a file to R2 and returns the key.
// Cache-Control is set automatically based on file extension:
//   - .m4s, .ts, .mp4, .m4a  -> immutable (1 year)
//   - .m3u8                   -> short-lived (2s, live manifest)
//   - everything else          -> no cache header
func SyncToR2(localPath, r2Key string) error {
	if r2Client == nil {
		if err := initR2Client(); err != nil {
			return err
		}
	}

	file, err := os.Open(localPath)
	if err != nil {
		return err
	}
	defer file.Close()

	input := &s3.PutObjectInput{
		Bucket: aws.String(bucket),
		Key:    aws.String(r2Key),
		Body:   file,
	}

	ext := strings.ToLower(filepath.Ext(r2Key))
	switch ext {
	case ".m4s", ".ts", ".mp4", ".m4a":
		input.CacheControl = aws.String(segmentCacheControl)
		input.ContentType = aws.String("video/MP2T")
	case ".m3u8":
		input.CacheControl = aws.String(manifestCacheControl)
		input.ContentType = aws.String("application/vnd.apple.mpegurl")
	}

	_, err = putObject(r2Client, input)
	return err
}

// HealthHandler provides a detailed health check endpoint (STREAMING.md contract)
func HealthHandler(w http.ResponseWriter, r *http.Request) {
	w.Header().Set("Content-Type", "application/json")
	status := "healthy"
	if playoutMode == "slate" || playoutMode == "degraded" {
		status = "degraded"
	}
	// Unhealthy if manifest has not advanced for >2 segment durations (12s)
	if hlsDir != "" && time.Since(lastR2PubTime) > 12*time.Second && playoutMode != "starting" {
		status = "unhealthy"
	}

	lastExit := ""
	if lastFFmpegExitError != nil {
		lastExit = lastFFmpegExitError.Error()
	}
	resp := map[string]any{
		"status":               status,
		"service":              "playout",
		"mode":                 playoutMode,
		"crash_count":          restartCount,
		"run_id":               runID,
		"active_item":          activeItem,
		"playlist_revision":    playlistRevision,
		"last_segment_seq":     lastSegmentSeq,
		"last_r2_pub":          lastR2PubTime.UTC().Format(time.RFC3339),
		"consecutive_failures": consecutiveFails,
		"last_exit_error":      lastExit,
		"timestamp":            time.Now().UTC().Format(time.RFC3339),
	}
	json.NewEncoder(w).Encode(resp)
}

// 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 {
	if err := os.MkdirAll(hlsDir, 0o755); err != nil {
		return err
	}

	matches, err := filepath.Glob(filepath.Join(hlsDir, "segment_*.ts"))
	if err != nil {
		return err
	}

	for _, match := range matches {
		if err := os.Remove(match); err != nil && !os.IsNotExist(err) {
			return err
		}
	}

	fmp4Matches, err := filepath.Glob(filepath.Join(hlsDir, "segment_*.m4s"))
	if err != nil {
		return err
	}

	for _, match := range fmp4Matches {
		if err := os.Remove(match); err != nil && !os.IsNotExist(err) {
			return err
		}
	}

	manifestPath := filepath.Join(hlsDir, "playlist.m3u8")
	if err := os.Remove(manifestPath); err != nil && !os.IsNotExist(err) {
		return err
	}

	initPath := filepath.Join(hlsDir, "init.mp4")
	if err := os.Remove(initPath); err != nil && !os.IsNotExist(err) {
		return err
	}

	return nil
}

func main() {
	playlistPath := "/app/playlist/playlist.txt"
	validateOnly := false
	for _, arg := range os.Args[1:] {
		switch arg {
		case "-validate":
			validateOnly = true
		default:
			if !strings.HasPrefix(arg, "-") {
				playlistPath = arg
			}
		}
	}

	if validateOnly {
		runValidateOnly(playlistPath)
		return
	}

	// Load shared HLS runtime config (written by the chat admin) before starting.
	loadSharedHLSConfig()

	fmt.Println("=== Linear Void Playout Service Starting ===")
	fmt.Println("Playlist:", playlistPath)
	fmt.Println("R2 Bucket:", bucket)

	// Initialize run ID for globally unique segment naming (STREAMING.md)
	initRunID()
	fmt.Println("Run ID:", runID)

	// Validate playlist before starting (ADR-001 normalization enforcement).
	// Runs once here; start-playout.sh no longer pre-validates separately.
	videos, err := ValidatePlaylist(playlistPath)
	if err != nil {
		fmt.Println("ERROR: Playlist validation failed:", err)
		os.Exit(1)
	}
	invalid := 0
	for _, v := range videos {
		if !v.Valid {
			fmt.Printf("ERROR: invalid media: %s — %s\n", v.Path, v.Validation)
			invalid++
		}
	}
	if invalid > 0 {
		fmt.Printf("ERROR: Playlist validation failed: %d invalid entries\n", invalid)
		os.Exit(1)
	}

	// Initialize R2 client (skip if not configured — local HLS via nginx still works)
	if err := initR2Client(); err != nil {
		fmt.Println("R2 sync disabled:", err)
	}

	// Start R2 segment retention cleanup (hourly + once at startup)
	if r2Client != nil {
		go runR2CleanupLoop()
	}

	// 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_crashes_total Total FFmpeg crashes (not clean completions).")
		fmt.Fprintln(w, "# TYPE linearvoid_playout_crashes_total counter")
		fmt.Fprintf(w, "linearvoid_playout_crashes_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)
		fmt.Fprintln(w, "# HELP linearvoid_playout_r2_cleanup_objects_total Total R2 objects deleted by retention cleanup.")
		fmt.Fprintln(w, "# TYPE linearvoid_playout_r2_cleanup_objects_total counter")
		fmt.Fprintf(w, "linearvoid_playout_r2_cleanup_objects_total %d\n", r2CleanupObjectsTotal)
		fmt.Fprintln(w, "# HELP linearvoid_playout_r2_runs_retained Number of R2 run folders retained by cleanup.")
		fmt.Fprintln(w, "# TYPE linearvoid_playout_r2_runs_retained gauge")
		fmt.Fprintf(w, "linearvoid_playout_r2_runs_retained %d\n", r2RunsRetained)
		fmt.Fprintln(w, "# HELP linearvoid_playout_r2_bytes_retained Total bytes retained in R2 run folders.")
		fmt.Fprintln(w, "# TYPE linearvoid_playout_r2_bytes_retained gauge")
		fmt.Fprintf(w, "linearvoid_playout_r2_bytes_retained %d\n", r2BytesRetained)
		fmt.Fprintln(w, "# HELP linearvoid_playout_loops_total Total clean playlist completions (loops).")
		fmt.Fprintln(w, "# TYPE linearvoid_playout_loops_total counter")
		fmt.Fprintf(w, "linearvoid_playout_loops_total %d\n", loopsTotal)
		fmt.Fprintln(w, "# HELP linearvoid_playout_run_duration_seconds Duration of the last FFmpeg run in seconds.")
		fmt.Fprintln(w, "# TYPE linearvoid_playout_run_duration_seconds gauge")
		fmt.Fprintf(w, "linearvoid_playout_run_duration_seconds %f\n", lastRunDuration.Seconds())
	})
	go func() {
		fmt.Println("Health check server listening on :8081/health")
		if err := http.ListenAndServe(":8081", nil); err != nil {
			fmt.Println("Health server error:", err)
		}
	}()

	// Run FFmpeg in a loop for continuous playback with slate fallback
	for {
		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))
		}

		// New run ID on each restart for unique R2 segment keys (STREAMING.md)
		initRunID()

		// Remove run folders older than the retention policy (keep the last N runs,
		// cap total bytes) now that the next run has started. This never touches the
		// current or previous run folders.
		if r2Client != nil {
			if _, err := cleanupOldRuns(context.Background()); err != nil {
				fmt.Println("R2 cleanup error:", err)
			}
		}

		// Reset HLS output only when the playlist changed, we are recovering from a
		// crash/slate, or this is the first run. Otherwise continue the segment
		// sequence so the loop is seamless for viewers (no gap on clean loops).
		fingerprint, fpErr := playlistFingerprint(playlistPath)
		if fpErr != nil {
			fmt.Println("Warning: cannot fingerprint playlist:", fpErr)
			fingerprint = ""
		}
		playlistChanged := fingerprint != playlistRevision
		shouldReset := playlistChanged || playoutMode != "live" || !lastRunClean

		windowSize := 24
		if ws, err := strconv.Atoi(hlsWindowSize); err == nil && ws > 0 {
			windowSize = ws
		}

		startNumber := 0
		if shouldReset {
			if err := resetHLSOutput(hlsDir); err != nil {
				fmt.Println("Failed to reset HLS output:", err)
				time.Sleep(5 * time.Second)
				continue
			}
			playlistRevision = fingerprint
		} else {
			if n, err := hlsNextStartNumber(hlsDir, windowSize); err == nil {
				startNumber = n
			}
		}

		// Start filesystem watcher for real-time R2 segment upload (STREAMING.md).
		// Only active when R2 is configured; otherwise the watcher is skipped.
		var watchCtx context.Context
		var watchCancel context.CancelFunc
		if r2Client != nil {
			watchCtx, watchCancel = context.WithCancel(context.Background())
			go watchAndPublishSegments(watchCtx, hlsDir)
		}

		lastFFmpegStartTime = time.Now()
		fmt.Printf("[%s] Starting FFmpeg playout (run %s) with browser-safe H.264/AAC HLS encoding...\n",
			lastFFmpegStartTime.UTC().Format(time.RFC3339), runID)

		cmd := exec.Command("ffmpeg", "-re", "-f", "concat", "-safe", "0", "-i", playlistPath,
			"-map", "0:v:0", "-map", "0:a:0?",
			"-vf", "scale=1280:720:force_original_aspect_ratio=decrease,pad=1280:720:(ow-iw)/2:(oh-ih)/2,fps=30,format=yuv420p",
			"-c:v", "libx264", "-preset", "veryfast", "-pix_fmt", "yuv420p",
			"-profile:v", "high", "-level", "4.0",
			"-r", "30", "-g", "60", "-keyint_min", "60", "-sc_threshold", "0",
			"-c:a", "aac", "-ar", "48000", "-ac", "2", "-b:a", "128k",
			"-f", "hls",
			"-hls_time", hlsSegmentDuration,
			"-hls_list_size", hlsWindowSize,
			"-hls_segment_type", "fmp4",
			"-hls_fmp4_init_filename", "init.mp4",
			"-hls_flags", "delete_segments+independent_segments",
			"-hls_segment_filename", filepath.Join(hlsDir, "seg-%06d.m4s"),
			"-start_number", strconv.Itoa(startNumber),
			filepath.Join(hlsDir, "playlist.m3u8"))

		runErr := cmd.Run()
		runDuration := time.Since(lastFFmpegStartTime)
		lastFFmpegExitError = runErr
		lastRunDuration = runDuration
		lastRunClean = runErr == nil

		if watchCancel != nil {
			watchCancel() // Stop the filesystem watcher
		}

		// After FFmpeg exits, sync remaining segments and publish manifest to R2.
		// Do this before logging/restart logic so R2 is up to date regardless of exit cause.
		if r2Client != nil {
			if err := syncLatestToR2(hlsDir); err != nil {
				fmt.Println("R2 sync warning:", err)
			}

			// Publish manifest atomically (STREAMING.md safe publication)
			manifestPath := filepath.Join(hlsDir, "playlist.m3u8")
			if err := publishManifest(manifestPath); err != nil {
				fmt.Println("Manifest publish warning:", err)
			}
		}

		// Differentiate clean playlist completion from FFmpeg crash.
		// A nil error means FFmpeg completed the playlist normally — not a crash.
		if runErr == nil {
			loopsTotal++
			fmt.Printf("[%s] Playlist completed normally after %s. Looping...\n",
				time.Now().UTC().Format(time.RFC3339), runDuration.Round(time.Second))
			// If we were in slate mode, a successful run means recovery
			if playoutMode == "slate" {
				deactivateSlate()
			}
			// Minimal delay for clean loop — just enough for watcher goroutine cleanup
			time.Sleep(500 * time.Millisecond)
			continue
		}

		// FFmpeg exited with an error — this is a real crash
		restartCount++
		fmt.Printf("[%s] FFmpeg CRASHED after %s (crash #%d in window): %v\n",
			time.Now().UTC().Format(time.RFC3339), runDuration.Round(time.Second), restartCount, runErr)

		if shouldEscalateToSlate() {
			fmt.Printf("[%s] CRITICAL: %d crashes in 60s — escalating to slate fallback\n",
				time.Now().UTC().Format(time.RFC3339), restartCount)
			if err := activateSlate(); err != nil {
				fmt.Printf("[%s] Slate activation failed: %v — will retry\n",
					time.Now().UTC().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 long enough to be a recovery, deactivate slate
		if playoutMode == "slate" && runDuration > 30*time.Second {
			deactivateSlate()
		}

		// Backoff delay after crash before retry
		time.Sleep(2 * time.Second)
	}
}

// syncLatestToR2 uploads new segments to R2 with runID-based keys
func syncLatestToR2(hlsDir string) error {
	files, err := filepath.Glob(filepath.Join(hlsDir, "seg-*.m4s"))
	if err != nil {
		return err
	}

	for _, f := range files {
		key := "hls/" + runID + "/" + filepath.Base(f)
		if err := SyncToR2(f, key); err != nil {
			fmt.Printf("Failed to sync %s: %v\n", f, err)
		} else {
			fmt.Printf("Synced segment to R2: %s\n", key)
		}
	}

	// Sync init segment to run-specific R2 key
	initPath := filepath.Join(hlsDir, "init.mp4")
	if err := SyncToR2(initPath, "hls/"+runID+"/init.mp4"); err != nil {
		return err
	}
	fmt.Println("Synced init.mp4 to R2")

	return nil
}

// r2KeepRuns returns how many of the most recent run folders to keep in R2.
func r2KeepRuns() int {
	n, err := strconv.Atoi(envWithDefault("R2_KEEP_RUNS", "2"))
	if err != nil || n < 1 {
		return defaultR2KeepRuns
	}
	return n
}

// r2MaxRetentionBytes returns the maximum total bytes to retain across run folders.
func r2MaxRetentionBytes() int64 {
	n, err := strconv.ParseInt(envWithDefault("R2_MAX_RETENTION_BYTES", "10737418240"), 10, 64)
	if err != nil || n <= 0 {
		return defaultR2MaxRetentionBytes
	}
	return n
}

// r2RunFromKey extracts the run folder name ("r-...") from an R2 object key.
func r2RunFromKey(key string) (string, bool) {
	rest := strings.TrimPrefix(key, "hls/")
	if rest == key || !strings.HasPrefix(rest, "r-") {
		return "", false
	}
	if idx := strings.IndexByte(rest, '/'); idx > 0 {
		return rest[:idx], true
	}
	return "", false
}

// runObjects groups the R2 object keys and total size for a single run folder.
type runObjects struct {
	keys []string
	size int64
}

// deleteR2Keys deletes the given R2 object keys in batches of up to 1000.
func deleteR2Keys(keys []string) (int, error) {
	deleted := 0
	for i := 0; i < len(keys); i += 1000 {
		end := i + 1000
		if end > len(keys) {
			end = len(keys)
		}
		batch := keys[i:end]
		objects := make([]types.ObjectIdentifier, len(batch))
		for j, key := range batch {
			objects[j] = types.ObjectIdentifier{Key: aws.String(key)}
		}
		if _, err := deleteObjects(r2Client, &s3.DeleteObjectsInput{
			Bucket: aws.String(bucket),
			Delete: &types.Delete{Objects: objects, Quiet: aws.Bool(true)},
		}); err != nil {
			return deleted, err
		}
		for _, key := range batch {
			fmt.Printf("R2 cleanup: deleted %s\n", key)
		}
		deleted += len(batch)
	}
	return deleted, nil
}

// cleanupOldRuns lists run-scoped objects under hls/, keeps the most recent
// R2_KEEP_RUNS runs, and deletes older runs. It also enforces the
// R2_MAX_RETENTION_BYTES cap by removing oldest retained runs first, always
// keeping at least the newest run. The live manifest keys are never deleted.
func cleanupOldRuns(ctx context.Context) (int, error) {
	if r2Client == nil {
		if err := initR2Client(); err != nil {
			return 0, err
		}
	}

	runs := map[string]*runObjects{}
	var token *string
	for {
		output, err := listObjects(r2Client, &s3.ListObjectsV2Input{
			Bucket:            aws.String(bucket),
			Prefix:            aws.String("hls/"),
			ContinuationToken: token,
		})
		if err != nil {
			return 0, err
		}
		for _, obj := range output.Contents {
			key := aws.ToString(obj.Key)
			if key == "hls/playlist.m3u8" || key == "hls/playlist.m3u8.tmp" {
				continue
			}
			run, ok := r2RunFromKey(key)
			if !ok {
				continue
			}
			r, exists := runs[run]
			if !exists {
				r = &runObjects{}
				runs[run] = r
			}
			r.keys = append(r.keys, key)
			if obj.Size != nil {
				r.size += *obj.Size
			}
		}
		if output.IsTruncated == nil || !*output.IsTruncated {
			break
		}
		token = output.NextContinuationToken
	}

	if len(runs) == 0 {
		r2RunsRetained = 0
		r2BytesRetained = 0
		return 0, nil
	}

	// Run IDs are ISO timestamps, so lexicographic order equals chronological order.
	sortedRuns := make([]string, 0, len(runs))
	for run := range runs {
		sortedRuns = append(sortedRuns, run)
	}
	sort.Strings(sortedRuns) // ascending: oldest -> newest

	deleteRun := map[string]bool{}
	keepRuns := r2KeepRuns()

	// Mark all but the newest keepRuns runs for deletion.
	if len(sortedRuns) > keepRuns {
		for _, run := range sortedRuns[:len(sortedRuns)-keepRuns] {
			deleteRun[run] = true
		}
	}

	// Enforce the total size cap on retained runs (oldest retained first), always
	// keeping the newest run even if it alone exceeds the cap.
	maxBytes := r2MaxRetentionBytes()
	retained := make([]string, 0, keepRuns)
	var retainedBytes int64
	for _, run := range sortedRuns {
		if !deleteRun[run] {
			retained = append(retained, run)
			retainedBytes += runs[run].size
		}
	}
	for i := 0; retainedBytes > maxBytes && i < len(retained)-1; i++ {
		run := retained[i]
		deleteRun[run] = true
		retainedBytes -= runs[run].size
	}

	var toDelete []string
	for run, del := range deleteRun {
		if del {
			toDelete = append(toDelete, runs[run].keys...)
		}
	}

	deleted, err := deleteR2Keys(toDelete)
	if err != nil {
		return deleted, err
	}

	keptRuns := int64(0)
	keptBytes := int64(0)
	for run, r := range runs {
		if !deleteRun[run] {
			keptRuns++
			keptBytes += r.size
		}
	}
	r2RunsRetained = keptRuns
	r2BytesRetained = keptBytes

	r2CleanupObjectsTotal += int64(deleted)
	return deleted, nil
}

// runR2CleanupLoop runs retention cleanup once at startup, then hourly.
func runR2CleanupLoop() {
	if _, err := cleanupOldRuns(context.Background()); err != nil {
		fmt.Println("R2 cleanup error:", err)
	}
	ticker := time.NewTicker(time.Hour)
	defer ticker.Stop()
	for range ticker.C {
		if _, err := cleanupOldRuns(context.Background()); err != nil {
			fmt.Println("R2 cleanup error:", err)
		}
	}
}

// playlistFingerprint returns a short hash of the playlist file content, used to
// detect playlist changes between loops.
func playlistFingerprint(playlistPath string) (string, error) {
	data, err := os.ReadFile(playlistPath)
	if err != nil {
		return "", err
	}
	sum := sha256.Sum256(data)
	return hex.EncodeToString(sum[:]), nil
}

// hlsNextStartNumber returns the next segment sequence number to use, and prunes
// stale segments outside the live window so they do not accumulate across loops.
func hlsNextStartNumber(hlsDir string, windowSize int) (int, error) {
	files, err := filepath.Glob(filepath.Join(hlsDir, "seg-*.m4s"))
	if err != nil {
		return 0, err
	}

	seqOf := func(f string) int {
		base := filepath.Base(f)
		num := strings.TrimSuffix(strings.TrimPrefix(base, "seg-"), ".m4s")
		n, err := strconv.Atoi(num)
		if err != nil {
			return -1
		}
		return n
	}

	maxSeq := -1
	for _, f := range files {
		if n := seqOf(f); n > maxSeq {
			maxSeq = n
		}
	}
	next := maxSeq + 1

	// Keep the live window; delete older segments no longer referenced.
	threshold := next - windowSize
	for _, f := range files {
		if n := seqOf(f); n >= 0 && n < threshold {
			if err := os.Remove(f); err != nil && !os.IsNotExist(err) {
				return next, err
			}
		}
	}
	return next, nil
}

// runValidateOnly validates the playlist and exits 0 (valid) or 1 (invalid/missing).
func runValidateOnly(playlistPath string) {
	videos, err := ValidatePlaylist(playlistPath)
	if err != nil {
		fmt.Println("ERROR: Playlist validation failed:", err)
		os.Exit(1)
	}
	valid := true
	for _, v := range videos {
		if v.Valid {
			fmt.Printf("OK: %s (%.1fs)\n", v.Path, v.Duration)
		} else {
			valid = false
			fmt.Printf("INVALID: %s — %s\n", v.Path, v.Validation)
		}
	}
	if !valid {
		os.Exit(1)
	}
	fmt.Println("Playlist valid.")
}

// loadSharedHLSConfig reads HLS settings from the shared runtime config file
// (written by the chat admin service) and overrides the env-derived defaults.
func loadSharedHLSConfig() {
	path := envWithDefault("PLAYOUT_CONFIG_PATH", "/app/playlist/config.json")
	data, err := os.ReadFile(path)
	if err != nil {
		if !os.IsNotExist(err) {
			fmt.Println("Warning: cannot read shared config:", err)
		}
		return
	}
	var cfg map[string]string
	if err := json.Unmarshal(data, &cfg); err != nil {
		fmt.Println("Warning: cannot parse shared config:", err)
		return
	}
	if v := strings.TrimSpace(cfg["hls_segment_duration"]); v != "" {
		if n, err := strconv.Atoi(v); err == nil && n > 0 {
			hlsSegmentDuration = strconv.Itoa(n)
		}
	}
	if v := strings.TrimSpace(cfg["hls_window_size"]); v != "" {
		if n, err := strconv.Atoi(v); err == nil && n > 0 {
			hlsWindowSize = strconv.Itoa(n)
		}
	}
	fmt.Printf("Shared HLS config: segment=%ss window=%s\n", hlsSegmentDuration, hlsWindowSize)
}