linearvoid/backend/playout/src/playlist.go

package main

import (
	"bufio"
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"os"
	"os/exec"
	"path/filepath"
	"strconv"
	"strings"
	"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/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)
	}
	bucket        = os.Getenv("R2_BUCKET_NAME")
	playoutMode   = "starting" // "live", "slate", "degraded"
	restartCount  = 0
	restartWindow = time.Now()
	slateDir      = "/app/slate"
	hlsDir        = "/app/hls"

	// 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
)

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"
)

// 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 := "live/" + 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).
func publishManifest(localManifestPath string) error {
	if r2Client == nil {
		if err := initR2Client(); err != nil {
			return err
		}
	}

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

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

	return nil
}

// 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}
	var errors []string

	// Run ffprobe
	cmd := exec.Command("ffprobe", "-v", "quiet", "-print_format", "json", "-show_format", "-show_streams", videoPath)
	output, err := cmd.Output()
	if err != nil {
		v.Validation = "ffprobe failed: " + err.Error()
		return v, fmt.Errorf("ffprobe: %w", err)
	}

	// 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,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
}

// ValidatePlaylist runs normalization validation on all entries
func ValidatePlaylist(playlistPath string) ([]VideoFile, error) {
	videos, err := LoadPlaylist(playlistPath)
	if err != nil {
		return nil, err
	}

	for i := range videos {
		validated, err := ValidateVideo(videos[i].Path)
		videos[i] = validated
		if err != nil {
			fmt.Printf("Warning: %s - %s\n", videos[i].Path, err)
		}
	}

	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"
	}

	resp := map[string]any{
		"status":              status,
		"service":             "playout",
		"mode":                playoutMode,
		"restarts":            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,
		"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"
	if len(os.Args) > 1 {
		playlistPath = os.Args[1]
	}

	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)
	if _, err := ValidatePlaylist(playlistPath); err != nil {
		fmt.Println("ERROR: Playlist validation failed:", err)
		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 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 {
			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()

		if err := resetHLSOutput(hlsDir); err != nil {
			fmt.Println("Failed to reset HLS output:", err)
			time.Sleep(5 * time.Second)
			continue
		}

		// 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)
		}

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

		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", "6",
			"-hls_list_size", "24",
			"-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"),
			filepath.Join(hlsDir, "playlist.m3u8"))

		runErr := cmd.Run()
		if watchCancel != nil {
			watchCancel() // Stop the filesystem watcher
		}

		// 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, publish remaining segments and atomic manifest to R2.
		// Skipped when R2 is not configured.
		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)
			}
		}

		time.Sleep(5 * 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 := "live/" + 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, "live/"+runID+"/init.mp4"); err != nil {
		return err
	}
	fmt.Println("Synced init.mp4 to R2")

	return nil
}