torrent-gateway/internal/transcoding/transcoder.go

package transcoding

import (
	"fmt"
	"io"
	"os"
	"os/exec"
	"path/filepath"
	"strconv"
	"strings"
	"sync"
	"time"
)

// Quality represents a transcoding quality profile
type Quality struct {
	Name     string `json:"name"`
	Width    int    `json:"width"`
	Height   int    `json:"height"`
	Bitrate  string `json:"bitrate"`
	Preset   string `json:"preset"`
}

// Job represents a transcoding job
type Job struct {
	ID          string    `json:"id"`
	InputPath   string    `json:"input_path"`
	OutputDir   string    `json:"output_dir"`
	FileHash    string    `json:"file_hash"`
	Qualities   []Quality `json:"qualities"`
	Priority    int       `json:"priority"`
	Status      string    `json:"status"` // "queued", "processing", "completed", "failed"
	Progress    float64   `json:"progress"`
	Error       string    `json:"error,omitempty"`
	StartTime   time.Time `json:"start_time"`
	CompletedAt time.Time `json:"completed_at,omitempty"`
	Callback    func(error) `json:"-"`
}

// Transcoder handles video transcoding operations
type Transcoder struct {
	workDir     string
	ffmpegPath  string
	concurrent  int
	queue       chan Job
	jobs        map[string]*Job // Track job status
	jobsMutex   sync.RWMutex    // Protect jobs map
	enabled     bool
}

// DefaultQualities provides standard quality profiles
var DefaultQualities = []Quality{
	{Name: "1080p", Width: 1920, Height: 1080, Bitrate: "5000k", Preset: "fast"},
	{Name: "720p", Width: 1280, Height: 720, Bitrate: "2500k", Preset: "fast"},
	{Name: "480p", Width: 854, Height: 480, Bitrate: "1000k", Preset: "fast"},
}

// NewTranscoder creates a new transcoder instance
func NewTranscoder(workDir string, concurrent int, enabled bool) (*Transcoder, error) {
	if !enabled {
		return &Transcoder{
			enabled: false,
			jobs:    make(map[string]*Job),
		}, nil
	}

	// Verify FFmpeg is available
	if _, err := exec.LookPath("ffmpeg"); err != nil {
		return nil, fmt.Errorf("ffmpeg not found in PATH: %w", err)
	}

	// Create work directory if it doesn't exist
	if err := os.MkdirAll(workDir, 0755); err != nil {
		return nil, fmt.Errorf("failed to create work directory: %w", err)
	}

	t := &Transcoder{
		workDir:     workDir,
		ffmpegPath:  "ffmpeg",
		concurrent:  concurrent,
		queue:       make(chan Job, 100),
		jobs:        make(map[string]*Job),
		enabled:     true,
	}

	// Start worker goroutines
	for i := 0; i < concurrent; i++ {
		go t.worker()
	}

	return t, nil
}

// IsEnabled returns whether transcoding is enabled
func (t *Transcoder) IsEnabled() bool {
	return t.enabled
}

// worker processes jobs from the queue
func (t *Transcoder) worker() {
	for job := range t.queue {
		t.processJob(job)
	}
}

// SubmitJob adds a job to the transcoding queue
func (t *Transcoder) SubmitJob(job Job) {
	if !t.enabled {
		if job.Callback != nil {
			job.Callback(fmt.Errorf("transcoding is disabled"))
		}
		return
	}

	job.Status = "queued"
	job.StartTime = time.Now()
	t.jobsMutex.Lock()
	t.jobs[job.ID] = &job
	t.jobsMutex.Unlock()
	t.queue <- job
}

// GetJobStatus returns the current status of a job
func (t *Transcoder) GetJobStatus(jobID string) (*Job, bool) {
	t.jobsMutex.RLock()
	job, exists := t.jobs[jobID]
	t.jobsMutex.RUnlock()
	return job, exists
}

// NeedsTranscoding checks if a file needs transcoding for web compatibility
func (t *Transcoder) NeedsTranscoding(filePath string) (bool, error) {
	if !t.enabled {
		return false, nil
	}

	// Use ffprobe to analyze the file
	cmd := exec.Command("ffprobe", "-v", "quiet", "-print_format", "json", "-show_format", "-show_streams", filePath)
	output, err := cmd.Output()
	if err != nil {
		return true, err // Assume needs transcoding if we can't analyze
	}

	outputStr := string(output)

	// Check if already in web-friendly format
	hasH264 := strings.Contains(outputStr, "\"codec_name\": \"h264\"")
	hasAAC := strings.Contains(outputStr, "\"codec_name\": \"aac\"")
	isMP4 := strings.HasSuffix(strings.ToLower(filePath), ".mp4")

	// If it's H.264/AAC in MP4 container, probably doesn't need transcoding
	if hasH264 && hasAAC && isMP4 {
		return false, nil
	}

	return true, nil
}

// CreateStreamingVersion creates a single web-compatible MP4 version (backward compatibility)
func (t *Transcoder) CreateStreamingVersion(inputPath, outputPath string) error {
	if !t.enabled {
		return fmt.Errorf("transcoding is disabled")
	}

	// Ensure output directory exists
	if err := os.MkdirAll(filepath.Dir(outputPath), 0755); err != nil {
		return fmt.Errorf("failed to create output directory: %w", err)
	}

	args := []string{
		"-i", inputPath,
		"-c:v", "libx264",      // H.264 video codec
		"-c:a", "aac",          // AAC audio codec  
		"-preset", "fast",      // Balance speed/quality
		"-crf", "23",           // Quality (23 is default, lower = better)
		"-maxrate", "2M",       // Max bitrate for streaming
		"-bufsize", "4M",       // Buffer size
		"-movflags", "+faststart", // Enable fast start for web
		"-y",                   // Overwrite output file
		outputPath,
	}

	cmd := exec.Command(t.ffmpegPath, args...)
	output, err := cmd.CombinedOutput()
	if err != nil {
		return fmt.Errorf("transcoding failed: %w\nFFmpeg output: %s", err, string(output))
	}
	return nil
}

// CreateMultiQualityVersions creates multiple quality versions for adaptive streaming
func (t *Transcoder) CreateMultiQualityVersions(inputPath, outputDir string, qualities []Quality) error {
	if !t.enabled {
		return fmt.Errorf("transcoding is disabled")
	}

	// Ensure output directory exists
	if err := os.MkdirAll(outputDir, 0755); err != nil {
		return fmt.Errorf("failed to create output directory: %w", err)
	}

	// Get input video info first to determine which qualities to generate
	inputInfo, err := t.getVideoInfo(inputPath)
	if err != nil {
		return fmt.Errorf("failed to get input video info: %w", err)
	}

	// Filter qualities based on input resolution (don't upscale)
	availableQualities := t.filterQualities(qualities, inputInfo.Height)

	// Generate each quality version
	for _, quality := range availableQualities {
		outputPath := filepath.Join(outputDir, fmt.Sprintf("%s.mp4", quality.Name))
		
		args := []string{
			"-i", inputPath,
			"-c:v", "libx264",
			"-c:a", "aac",
			"-preset", quality.Preset,
			"-b:v", quality.Bitrate,
			"-maxrate", quality.Bitrate,
			"-bufsize", fmt.Sprintf("%dk", parseInt(quality.Bitrate)*2), // 2x bitrate for buffer
			"-vf", fmt.Sprintf("scale=%d:%d", quality.Width, quality.Height),
			"-movflags", "+faststart",
			"-y",
			outputPath,
		}

		cmd := exec.Command(t.ffmpegPath, args...)
		output, err := cmd.CombinedOutput()
		if err != nil {
			return fmt.Errorf("failed to create %s quality: %w\nFFmpeg output: %s", quality.Name, err, string(output))
		}
	}

	// Always create a "stream.mp4" version (highest available quality) for backward compatibility
	if len(availableQualities) > 0 {
		bestQuality := availableQualities[0] // Qualities should be ordered best to worst
		srcPath := filepath.Join(outputDir, fmt.Sprintf("%s.mp4", bestQuality.Name))
		dstPath := filepath.Join(outputDir, "stream.mp4")
		
		// Copy the best quality as stream.mp4
		if err := t.copyFile(srcPath, dstPath); err != nil {
			return fmt.Errorf("failed to create stream.mp4: %w", err)
		}
	}

	return nil
}

// getVideoInfo extracts basic video information
func (t *Transcoder) getVideoInfo(inputPath string) (*VideoInfo, error) {
	cmd := exec.Command("ffprobe", "-v", "quiet", "-print_format", "json", "-show_streams", inputPath)
	output, err := cmd.Output()
	if err != nil {
		return nil, err
	}

	// Parse JSON to get video dimensions
	// This is a simplified version - you might want to use a proper JSON parser
	outputStr := string(output)
	
	// Extract height from JSON (simplified)
	height := 1080 // Default fallback
	if strings.Contains(outputStr, "\"height\":") {
		// Simple regex would be better here, but keeping it simple
		if strings.Contains(outputStr, "\"height\": 720") || strings.Contains(outputStr, "\"height\":720") {
			height = 720
		} else if strings.Contains(outputStr, "\"height\": 480") || strings.Contains(outputStr, "\"height\":480") {
			height = 480
		} else if strings.Contains(outputStr, "\"height\": 1080") || strings.Contains(outputStr, "\"height\":1080") {
			height = 1080
		} else if strings.Contains(outputStr, "\"height\": 2160") || strings.Contains(outputStr, "\"height\":2160") {
			height = 2160
		}
	}

	return &VideoInfo{Height: height}, nil
}

type VideoInfo struct {
	Height int
}

// filterQualities removes qualities that would upscale the video
func (t *Transcoder) filterQualities(qualities []Quality, inputHeight int) []Quality {
	var filtered []Quality
	
	for _, quality := range qualities {
		if quality.Height <= inputHeight {
			filtered = append(filtered, quality)
		}
	}
	
	// If no qualities fit (very low resolution input), at least include the lowest
	if len(filtered) == 0 && len(qualities) > 0 {
		// Get the lowest quality
		lowest := qualities[len(qualities)-1]
		filtered = append(filtered, lowest)
	}
	
	return filtered
}

// Helper function to parse bitrate string to int
func parseInt(bitrate string) int {
	// Remove 'k' suffix and convert to int
	if strings.HasSuffix(bitrate, "k") {
		if val := strings.TrimSuffix(bitrate, "k"); val != "" {
			if i, err := strconv.Atoi(val); err == nil {
				return i
			}
		}
	}
	return 2000 // Default fallback
}

// copyFile copies a file from src to dst
func (t *Transcoder) copyFile(src, dst string) error {
	input, err := os.Open(src)
	if err != nil {
		return err
	}
	defer input.Close()

	output, err := os.Create(dst)
	if err != nil {
		return err
	}
	defer output.Close()

	_, err = io.Copy(output, input)
	return err
}

// processJob handles the actual transcoding work
func (t *Transcoder) processJob(job Job) {
	job.Status = "processing"
	t.jobsMutex.Lock()
	t.jobs[job.ID] = &job
	t.jobsMutex.Unlock()

	var err error
	defer func() {
		if err != nil {
			job.Status = "failed"
			job.Error = err.Error()
		} else {
			job.Status = "completed"
			job.Progress = 100.0
		}
		job.CompletedAt = time.Now()
		t.jobsMutex.Lock()
		t.jobs[job.ID] = &job
		t.jobsMutex.Unlock()
		
		if job.Callback != nil {
			job.Callback(err)
		}
	}()

	// Create output directory
	if err = os.MkdirAll(job.OutputDir, 0755); err != nil {
		err = fmt.Errorf("failed to create output directory: %w", err)
		return
	}

	// Create multiple quality versions if qualities are specified
	if len(job.Qualities) > 0 {
		err = t.CreateMultiQualityVersions(job.InputPath, job.OutputDir, job.Qualities)
		if err != nil {
			err = fmt.Errorf("multi-quality transcoding failed: %w", err)
			return
		}
	} else {
		// Fallback to single quality for backward compatibility
		outputPath := filepath.Join(job.OutputDir, "stream.mp4")
		err = t.CreateStreamingVersion(job.InputPath, outputPath)
		if err != nil {
			err = fmt.Errorf("transcoding failed: %w", err)
			return
		}
	}

	job.Progress = 100.0
}

// GetTranscodedPath returns the path to a transcoded file if it exists
func (t *Transcoder) GetTranscodedPath(fileHash string) string {
	if !t.enabled {
		return ""
	}

	streamPath := filepath.Join(t.workDir, fileHash, "stream.mp4")
	if _, err := os.Stat(streamPath); err == nil {
		return streamPath
	}
	return ""
}

// GetQualityPath returns the path to a specific quality version
func (t *Transcoder) GetQualityPath(fileHash, quality string) string {
	if !t.enabled {
		return ""
	}

	qualityPath := filepath.Join(t.workDir, fileHash, fmt.Sprintf("%s.mp4", quality))
	if _, err := os.Stat(qualityPath); err == nil {
		return qualityPath
	}
	return ""
}

// GetAvailableQualities returns a list of available quality versions for a file
func (t *Transcoder) GetAvailableQualities(fileHash string) []Quality {
	if !t.enabled {
		return nil
	}

	var availableQualities []Quality
	transcodedDir := filepath.Join(t.workDir, fileHash)

	// Check which quality files exist
	for _, quality := range DefaultQualities {
		qualityPath := filepath.Join(transcodedDir, fmt.Sprintf("%s.mp4", quality.Name))
		if _, err := os.Stat(qualityPath); err == nil {
			availableQualities = append(availableQualities, quality)
		}
	}

	return availableQualities
}

// GetAllJobs returns information about all jobs (for admin monitoring)
func (t *Transcoder) GetAllJobs() map[string]*Job {
	if !t.enabled {
		return make(map[string]*Job)
	}
	
	t.jobsMutex.RLock()
	defer t.jobsMutex.RUnlock()
	
	// Create a copy of the jobs map
	jobs := make(map[string]*Job)
	for id, job := range t.jobs {
		// Create a copy of the job
		jobCopy := *job
		jobs[id] = &jobCopy
	}
	
	return jobs
}

// Close shuts down the transcoder
func (t *Transcoder) Close() {
	if t.enabled && t.queue != nil {
		close(t.queue)
	}
}