Case Study: Video Transcoding Service

package main

import (
	"encoding/json"
	"fmt"
	"math"
	"math/rand"
	"net/http"
	"os"
	"os/signal"
	"strings"
	"sync"
	"syscall"
	"time"
)

// ===========================================
// 1. TYPES & CONSTANTS
// ===========================================

type JobStatus string
type SegmentStatus string
type WorkerStatusType string

const (
	JobPending     JobStatus = "pending"
	JobProbing     JobStatus = "probing"
	JobSplitting   JobStatus = "splitting"
	JobTranscoding JobStatus = "transcoding"
	JobMerging     JobStatus = "merging"
	JobCompleted   JobStatus = "completed"
	JobFailed      JobStatus = "failed"
	JobCancelled   JobStatus = "cancelled"

	SegPending    SegmentStatus = "pending"
	SegAssigned   SegmentStatus = "assigned"
	SegProcessing SegmentStatus = "processing"
	SegCompleted  SegmentStatus = "completed"
	SegFailed     SegmentStatus = "failed"

	WorkerIdle     WorkerStatusType = "idle"
	WorkerBusy     WorkerStatusType = "busy"
	WorkerDraining WorkerStatusType = "draining"

	SegmentDuration  = 6
	HeartbeatTimeout = 30 * time.Second
	MaxRetries       = 3
)

type TranscodeProfile struct {
	Name    string `json:"name"`
	Width   int    `json:"width"`
	Height  int    `json:"height"`
	Bitrate int    `json:"bitrate"` // kbps
	Codec   string `json:"codec"`
	Preset  string `json:"preset"`
}

type VideoSegment struct {
	ID             string        `json:"id"`
	JobID          string        `json:"jobId"`
	Index          int           `json:"index"`
	StartTime      float64       `json:"startTime"`
	Duration       float64       `json:"duration"`
	ProfileName    string        `json:"profileName"`
	Status         SegmentStatus `json:"status"`
	AssignedWorker string        `json:"assignedWorker,omitempty"`
	AssignedAt     int64         `json:"assignedAt"`
	CompletedAt    int64         `json:"completedAt"`
	OutputPath     string        `json:"outputPath,omitempty"`
	RetryCount     int           `json:"retryCount"`
}

type VideoMetadata struct {
	Width     int     `json:"width"`
	Height    int     `json:"height"`
	Duration  float64 `json:"duration"`
	Codec     string  `json:"codec"`
	Bitrate   int     `json:"bitrate"`
	FrameRate float64 `json:"frameRate"`
	FileSize  int64   `json:"fileSize"`
}

type TranscodeJob struct {
	ID            string             `json:"id"`
	UserID        string             `json:"userId"`
	SourceURL     string             `json:"sourceUrl"`
	Status        JobStatus          `json:"status"`
	Priority      int                `json:"priority"`
	CreatedAt     int64              `json:"createdAt"`
	StartedAt     int64              `json:"startedAt"`
	CompletedAt   int64              `json:"completedAt"`
	SourceMeta    *VideoMetadata     `json:"sourceMeta,omitempty"`
	Profiles      []TranscodeProfile `json:"profiles"`
	Segments      []*VideoSegment    `json:"segments"`
	WebhookURL    string             `json:"webhookUrl,omitempty"`
	OutputBaseURL string             `json:"outputBaseUrl,omitempty"`
	Error         string             `json:"error,omitempty"`
}

type WorkerInfo struct {
	ID             string           `json:"id"`
	Status         WorkerStatusType `json:"status"`
	CurrentTask    string           `json:"currentTask,omitempty"`
	LastHeartbeat  time.Time        `json:"lastHeartbeat"`
	TasksCompleted int              `json:"tasksCompleted"`
	TasksFailed    int              `json:"tasksFailed"`
	RegisteredAt   time.Time        `json:"registeredAt"`
}

// ===========================================
// 2. DEFAULT PROFILES
// ===========================================

var defaultProfiles = []TranscodeProfile{
	{Name: "1080p", Width: 1920, Height: 1080, Bitrate: 5000, Codec: "h264", Preset: "medium"},
	{Name: "720p", Width: 1280, Height: 720, Bitrate: 2800, Codec: "h264", Preset: "medium"},
	{Name: "480p", Width: 854, Height: 480, Bitrate: 1400, Codec: "h264", Preset: "fast"},
	{Name: "360p", Width: 640, Height: 360, Bitrate: 800, Codec: "h264", Preset: "fast"},
}

// ===========================================
// 3. FAIR PRIORITY QUEUE
// ===========================================

type FairPriorityQueue struct {
	mu              sync.Mutex
	queues          map[string][]*TranscodeJob
	roundRobinIndex int
}

func NewFairPriorityQueue() *FairPriorityQueue {
	return &FairPriorityQueue{queues: make(map[string][]*TranscodeJob)}
}

func (q *FairPriorityQueue) Enqueue(job *TranscodeJob) {
	q.mu.Lock()
	defer q.mu.Unlock()

	userQueue := q.queues[job.UserID]
	inserted := false
	for i, existing := range userQueue {
		if job.Priority < existing.Priority {
			userQueue = append(userQueue[:i], append([]*TranscodeJob{job}, userQueue[i:]...)...)
			inserted = true
			break
		}
	}
	if !inserted {
		userQueue = append(userQueue, job)
	}
	q.queues[job.UserID] = userQueue
}

func (q *FairPriorityQueue) Dequeue() *TranscodeJob {
	q.mu.Lock()
	defer q.mu.Unlock()

	var userIDs []string
	for uid, queue := range q.queues {
		if len(queue) > 0 {
			userIDs = append(userIDs, uid)
		}
	}
	if len(userIDs) == 0 {
		return nil
	}

	q.roundRobinIndex = q.roundRobinIndex % len(userIDs)
	uid := userIDs[q.roundRobinIndex]
	q.roundRobinIndex++

	queue := q.queues[uid]
	job := queue[0]
	q.queues[uid] = queue[1:]
	if len(q.queues[uid]) == 0 {
		delete(q.queues, uid)
	}
	return job
}

func (q *FairPriorityQueue) Remove(jobID string) bool {
	q.mu.Lock()
	defer q.mu.Unlock()

	for uid, queue := range q.queues {
		for i, j := range queue {
			if j.ID == jobID {
				q.queues[uid] = append(queue[:i], queue[i+1:]...)
				if len(q.queues[uid]) == 0 {
					delete(q.queues, uid)
				}
				return true
			}
		}
	}
	return false
}

func (q *FairPriorityQueue) Size() int {
	q.mu.Lock()
	defer q.mu.Unlock()
	total := 0
	for _, queue := range q.queues {
		total += len(queue)
	}
	return total
}

// ===========================================
// 4. PROGRESS TRACKER
// ===========================================

type JobProgress struct {
	Total     int     `json:"total"`
	Completed int     `json:"completed"`
	Failed    int     `json:"failed"`
	Percent   float64 `json:"percent"`
	EtaMs     int64   `json:"eta"`
}

type ProgressTracker struct {
	mu   sync.RWMutex
	jobs map[string]*TranscodeJob
}

func NewProgressTracker() *ProgressTracker {
	return &ProgressTracker{jobs: make(map[string]*TranscodeJob)}
}

func (pt *ProgressTracker) RegisterJob(job *TranscodeJob) {
	pt.mu.Lock()
	defer pt.mu.Unlock()
	pt.jobs[job.ID] = job
}

func (pt *ProgressTracker) GetJob(jobID string) *TranscodeJob {
	pt.mu.RLock()
	defer pt.mu.RUnlock()
	return pt.jobs[jobID]
}

func (pt *ProgressTracker) GetAllJobs() []*TranscodeJob {
	pt.mu.RLock()
	defer pt.mu.RUnlock()
	result := make([]*TranscodeJob, 0, len(pt.jobs))
	for _, j := range pt.jobs {
		result = append(result, j)
	}
	return result
}

func (pt *ProgressTracker) UpdateSegment(jobID, segmentID string, status SegmentStatus, outputPath string) {
	pt.mu.Lock()
	defer pt.mu.Unlock()
	job := pt.jobs[jobID]
	if job == nil {
		return
	}
	for _, seg := range job.Segments {
		if seg.ID == segmentID {
			seg.Status = status
			if status == SegCompleted {
				seg.CompletedAt = time.Now().UnixMilli()
				seg.OutputPath = outputPath
			}
			return
		}
	}
}

func (pt *ProgressTracker) GetProgress(jobID string) JobProgress {
	pt.mu.RLock()
	defer pt.mu.RUnlock()

	job := pt.jobs[jobID]
	if job == nil {
		return JobProgress{}
	}

	total := len(job.Segments)
	completed := 0
	failed := 0
	for _, s := range job.Segments {
		if s.Status == SegCompleted {
			completed++
		} else if s.Status == SegFailed {
			failed++
		}
	}

	percent := 0.0
	if total > 0 {
		percent = math.Round(float64(completed) / float64(total) * 100)
	}

	var eta int64
	if completed > 0 && job.StartedAt > 0 {
		elapsed := time.Now().UnixMilli() - job.StartedAt
		avgPerSeg := float64(elapsed) / float64(completed)
		remaining := total - completed - failed
		eta = int64(avgPerSeg * float64(remaining))
	}

	return JobProgress{Total: total, Completed: completed, Failed: failed, Percent: percent, EtaMs: eta}
}

func (pt *ProgressTracker) IsComplete(jobID string) bool {
	pt.mu.RLock()
	defer pt.mu.RUnlock()
	job := pt.jobs[jobID]
	if job == nil {
		return false
	}
	for _, s := range job.Segments {
		if s.Status != SegCompleted && s.Status != SegFailed {
			return false
		}
	}
	return true
}

func (pt *ProgressTracker) HasFailures(jobID string) bool {
	pt.mu.RLock()
	defer pt.mu.RUnlock()
	job := pt.jobs[jobID]
	if job == nil {
		return false
	}
	for _, s := range job.Segments {
		if s.Status == SegFailed && s.RetryCount >= MaxRetries {
			return true
		}
	}
	return false
}

// ===========================================
// 5. WORKER POOL
// ===========================================

type WorkerPool struct {
	mu      sync.Mutex
	workers map[string]*WorkerInfo
}

func NewWorkerPool() *WorkerPool {
	return &WorkerPool{workers: make(map[string]*WorkerInfo)}
}

func (wp *WorkerPool) Register(workerID string) *WorkerInfo {
	wp.mu.Lock()
	defer wp.mu.Unlock()
	w := &WorkerInfo{
		ID:            workerID,
		Status:        WorkerIdle,
		LastHeartbeat: time.Now(),
		RegisteredAt:  time.Now(),
	}
	wp.workers[workerID] = w
	fmt.Printf("[WORKER] Registered worker %s\n", workerID)
	return w
}

func (wp *WorkerPool) Heartbeat(workerID string) bool {
	wp.mu.Lock()
	defer wp.mu.Unlock()
	w := wp.workers[workerID]
	if w == nil {
		return false
	}
	w.LastHeartbeat = time.Now()
	return true
}

func (wp *WorkerPool) AssignTask(workerID, taskID string) bool {
	wp.mu.Lock()
	defer wp.mu.Unlock()
	w := wp.workers[workerID]
	if w == nil || w.Status != WorkerIdle {
		return false
	}
	w.Status = WorkerBusy
	w.CurrentTask = taskID
	return true
}

func (wp *WorkerPool) CompleteTask(workerID string, success bool) {
	wp.mu.Lock()
	defer wp.mu.Unlock()
	w := wp.workers[workerID]
	if w == nil {
		return
	}
	w.Status = WorkerIdle
	w.CurrentTask = ""
	if success {
		w.TasksCompleted++
	} else {
		w.TasksFailed++
	}
}

func (wp *WorkerPool) GetIdleWorkers() []*WorkerInfo {
	wp.mu.Lock()
	defer wp.mu.Unlock()
	var idle []*WorkerInfo
	for _, w := range wp.workers {
		if w.Status == WorkerIdle {
			idle = append(idle, w)
		}
	}
	return idle
}

func (wp *WorkerPool) GetStaleWorkers() []*WorkerInfo {
	wp.mu.Lock()
	defer wp.mu.Unlock()
	var stale []*WorkerInfo
	for _, w := range wp.workers {
		if w.Status == WorkerBusy && time.Since(w.LastHeartbeat) > HeartbeatTimeout {
			stale = append(stale, w)
		}
	}
	return stale
}

func (wp *WorkerPool) Drain(workerID string) {
	wp.mu.Lock()
	defer wp.mu.Unlock()
	if w := wp.workers[workerID]; w != nil {
		w.Status = WorkerDraining
		fmt.Printf("[WORKER] Draining worker %s\n", workerID)
	}
}

func (wp *WorkerPool) Remove(workerID string) {
	wp.mu.Lock()
	defer wp.mu.Unlock()
	delete(wp.workers, workerID)
	fmt.Printf("[WORKER] Removed worker %s\n", workerID)
}

type PoolStatus struct {
	Total    int `json:"total"`
	Idle     int `json:"idle"`
	Busy     int `json:"busy"`
	Draining int `json:"draining"`
}

func (wp *WorkerPool) GetStatus() PoolStatus {
	wp.mu.Lock()
	defer wp.mu.Unlock()
	s := PoolStatus{Total: len(wp.workers)}
	for _, w := range wp.workers {
		switch w.Status {
		case WorkerIdle:
			s.Idle++
		case WorkerBusy:
			s.Busy++
		case WorkerDraining:
			s.Draining++
		}
	}
	return s
}

func (wp *WorkerPool) GetAll() []*WorkerInfo {
	wp.mu.Lock()
	defer wp.mu.Unlock()
	result := make([]*WorkerInfo, 0, len(wp.workers))
	for _, w := range wp.workers {
		result = append(result, w)
	}
	return result
}

// ===========================================
// 6. SEGMENT SPLITTER
// ===========================================

func createSegments(job *TranscodeJob) []*VideoSegment {
	if job.SourceMeta == nil {
		return nil
	}

	duration := job.SourceMeta.Duration
	segCount := int(math.Ceil(duration / SegmentDuration))
	var segments []*VideoSegment

	for _, profile := range job.Profiles {
		for i := 0; i < segCount; i++ {
			startTime := float64(i) * SegmentDuration
			segDur := math.Min(SegmentDuration, duration-startTime)

			segments = append(segments, &VideoSegment{
				ID:          fmt.Sprintf("%s-%s-seg%04d", job.ID, profile.Name, i),
				JobID:       job.ID,
				Index:       i,
				StartTime:   startTime,
				Duration:    segDur,
				ProfileName: profile.Name,
				Status:      SegPending,
			})
		}
	}
	return segments
}

// ===========================================
// 7. HLS MANIFEST GENERATOR
// ===========================================

func generateMasterPlaylist(job *TranscodeJob) string {
	var b strings.Builder
	b.WriteString("#EXTM3U\n#EXT-X-VERSION:3\n\n")

	for _, profile := range job.Profiles {
		bw := profile.Bitrate * 1000
		fmt.Fprintf(&b, "#EXT-X-STREAM-INF:BANDWIDTH=%d,RESOLUTION=%dx%d,CODECS=\"avc1.640028\"\n",
			bw, profile.Width, profile.Height)
		fmt.Fprintf(&b, "%s/playlist.m3u8\n\n", profile.Name)
	}
	return b.String()
}

func generateVariantPlaylist(job *TranscodeJob, profileName string) string {
	var profileSegs []*VideoSegment
	for _, s := range job.Segments {
		if s.ProfileName == profileName && s.Status == SegCompleted {
			profileSegs = append(profileSegs, s)
		}
	}

	// Sort by index (already ordered, but be safe)
	for i := 1; i < len(profileSegs); i++ {
		for j := i; j > 0 && profileSegs[j].Index < profileSegs[j-1].Index; j-- {
			profileSegs[j], profileSegs[j-1] = profileSegs[j-1], profileSegs[j]
		}
	}

	var b strings.Builder
	b.WriteString("#EXTM3U\n#EXT-X-VERSION:3\n")
	fmt.Fprintf(&b, "#EXT-X-TARGETDURATION:%d\n", SegmentDuration)
	b.WriteString("#EXT-X-MEDIA-SEQUENCE:0\n\n")

	for _, seg := range profileSegs {
		fmt.Fprintf(&b, "#EXTINF:%.3f,\n", seg.Duration)
		fmt.Fprintf(&b, "segment%04d.ts\n", seg.Index)
	}

	b.WriteString("#EXT-X-ENDLIST\n")
	return b.String()
}

// ===========================================
// 8. WEBHOOK NOTIFIER
// ===========================================

type WebhookNotifier struct {
	mu      sync.Mutex
	pending []webhookEntry
}

type webhookEntry struct {
	URL     string
	Payload interface{}
	Retries int
}

func NewWebhookNotifier() *WebhookNotifier {
	return &WebhookNotifier{}
}

func (wn *WebhookNotifier) Notify(url string, payload interface{}) {
	if url == "" {
		return
	}
	wn.mu.Lock()
	defer wn.mu.Unlock()

	data, _ := json.Marshal(payload)
	preview := string(data)
	if len(preview) > 120 {
		preview = preview[:120] + "..."
	}
	fmt.Printf("[WEBHOOK] Delivered to %s: %s\n", url, preview)
}

// ===========================================
// 9. TRANSCODING SERVICE
// ===========================================

type TranscodingService struct {
	queue             *FairPriorityQueue
	tracker           *ProgressTracker
	pool              *WorkerPool
	notifier          *WebhookNotifier
	mu                sync.Mutex
	processing        bool
	shutdownRequested bool
	nextID            int
}

func NewTranscodingService(workerCount int) *TranscodingService {
	svc := &TranscodingService{
		queue:    NewFairPriorityQueue(),
		tracker:  NewProgressTracker(),
		pool:     NewWorkerPool(),
		notifier: NewWebhookNotifier(),
	}
	for i := 0; i < workerCount; i++ {
		svc.pool.Register(fmt.Sprintf("worker-%d", i))
	}
	return svc
}

func (svc *TranscodingService) genID() string {
	svc.mu.Lock()
	defer svc.mu.Unlock()
	svc.nextID++
	return fmt.Sprintf("job-%06d", svc.nextID)
}

func (svc *TranscodingService) SubmitJob(userID, sourceURL, webhookURL string, priority int) *TranscodeJob {
	job := &TranscodeJob{
		ID:         svc.genID(),
		UserID:     userID,
		SourceURL:  sourceURL,
		Status:     JobPending,
		Priority:   priority,
		CreatedAt:  time.Now().UnixMilli(),
		WebhookURL: webhookURL,
	}
	svc.tracker.RegisterJob(job)
	svc.queue.Enqueue(job)
	fmt.Printf("[JOB] Submitted %s for user %s (priority %d)\n", job.ID, userID, priority)
	return job
}

func (svc *TranscodingService) ProcessNextJob() {
	if svc.shutdownRequested {
		return
	}

	job := svc.queue.Dequeue()
	if job == nil {
		return
	}

	svc.mu.Lock()
	svc.processing = true
	svc.mu.Unlock()

	defer func() {
		svc.mu.Lock()
		svc.processing = false
		svc.mu.Unlock()
	}()

	job.StartedAt = time.Now().UnixMilli()

	// Step 1: Probe
	job.Status = JobProbing
	fmt.Printf("[JOB] Probing %s...\n", job.ID)
	job.SourceMeta = probeVideo(job.SourceURL)

	// Step 2: Select profiles
	job.Status = JobSplitting
	job.Profiles = selectProfiles(job.SourceMeta)
	fmt.Printf("[JOB] Selected %d profiles for %s\n", len(job.Profiles), job.ID)

	// Step 3: Create segments
	job.Segments = createSegments(job)
	fmt.Printf("[JOB] Created %d segments for %s\n", len(job.Segments), job.ID)

	// Step 4: Transcode
	job.Status = JobTranscoding
	svc.transcodeSegments(job)

	// Step 5: Check result
	if svc.tracker.HasFailures(job.ID) {
		job.Status = JobFailed
		job.Error = "Some segments failed after max retries"
		fmt.Printf("[JOB] Job %s FAILED\n", job.ID)
	} else {
		job.Status = JobMerging
		job.OutputBaseURL = fmt.Sprintf("/output/%s", job.ID)
		fmt.Printf("[JOB] Generating manifests for %s\n", job.ID)

		master := generateMasterPlaylist(job)
		fmt.Printf("[MANIFEST] Master playlist:\n%s", master)

		for _, profile := range job.Profiles {
			variant := generateVariantPlaylist(job, profile.Name)
			fmt.Printf("[MANIFEST] Variant %s:\n%s", profile.Name, variant)
		}

		job.Status = JobCompleted
		job.CompletedAt = time.Now().UnixMilli()
		elapsed := float64(job.CompletedAt-job.StartedAt) / 1000.0
		fmt.Printf("[JOB] Job %s COMPLETED in %.1fs\n", job.ID, elapsed)
	}

	// Step 6: Webhook
	svc.notifier.Notify(job.WebhookURL, map[string]interface{}{
		"jobId":         job.ID,
		"status":        job.Status,
		"outputBaseUrl": job.OutputBaseURL,
		"manifestUrl":   job.OutputBaseURL + "/master.m3u8",
		"completedAt":   job.CompletedAt,
	})
}

func probeVideo(sourceURL string) *VideoMetadata {
	return &VideoMetadata{
		Width:     1920,
		Height:    1080,
		Duration:  120,
		Codec:     "h264",
		Bitrate:   8000,
		FrameRate: 30,
		FileSize:  120_000_000,
	}
}

func selectProfiles(meta *VideoMetadata) []TranscodeProfile {
	var profiles []TranscodeProfile
	for _, p := range defaultProfiles {
		if p.Height <= meta.Height {
			profiles = append(profiles, p)
		}
	}
	return profiles
}

func (svc *TranscodingService) transcodeSegments(job *TranscodeJob) {
	var wg sync.WaitGroup

	for {
		if svc.shutdownRequested {
			fmt.Printf("[JOB] Shutdown requested, draining for %s\n", job.ID)
			break
		}

		svc.handleStaleWorkers(job)

		pendingCount := 0
		activeCount := 0
		for _, s := range job.Segments {
			if s.Status == SegPending {
				pendingCount++
			} else if s.Status == SegAssigned || s.Status == SegProcessing {
				activeCount++
			}
		}

		if pendingCount == 0 && activeCount == 0 {
			break
		}

		idle := svc.pool.GetIdleWorkers()
		for _, worker := range idle {
			var seg *VideoSegment
			for _, s := range job.Segments {
				if s.Status == SegPending {
					seg = s
					break
				}
			}
			if seg == nil {
				break
			}

			if svc.pool.AssignTask(worker.ID, seg.ID) {
				seg.Status = SegAssigned
				seg.AssignedWorker = worker.ID
				seg.AssignedAt = time.Now().UnixMilli()

				wg.Add(1)
				go func(s *VideoSegment, wID string) {
					defer wg.Done()
					svc.simulateTranscode(job.ID, s, wID)
				}(seg, worker.ID)
			}
		}

		time.Sleep(10 * time.Millisecond)
	}

	wg.Wait()
}

func (svc *TranscodingService) simulateTranscode(jobID string, seg *VideoSegment, workerID string) {
	seg.Status = SegProcessing

	// Simulate transcoding time
	delay := time.Duration(50+rand.Intn(150)) * time.Millisecond
	time.Sleep(delay)

	success := rand.Float64() > 0.05 // 5% failure rate

	if success {
		outputPath := fmt.Sprintf("/output/%s/%s/segment%04d.ts", jobID, seg.ProfileName, seg.Index)
		svc.tracker.UpdateSegment(jobID, seg.ID, SegCompleted, outputPath)
		svc.pool.CompleteTask(workerID, true)
		svc.pool.Heartbeat(workerID)
	} else {
		seg.RetryCount++
		if seg.RetryCount < MaxRetries {
			seg.Status = SegPending
			seg.AssignedWorker = ""
			fmt.Printf("[RETRY] Segment %s retry %d/%d\n", seg.ID, seg.RetryCount, MaxRetries)
		} else {
			svc.tracker.UpdateSegment(jobID, seg.ID, SegFailed, "")
			fmt.Printf("[FAIL] Segment %s permanently failed\n", seg.ID)
		}
		svc.pool.CompleteTask(workerID, false)
	}
}

func (svc *TranscodingService) handleStaleWorkers(job *TranscodeJob) {
	stale := svc.pool.GetStaleWorkers()
	for _, w := range stale {
		fmt.Printf("[HEARTBEAT] Worker %s stale, reassigning\n", w.ID)
		for _, seg := range job.Segments {
			if seg.AssignedWorker == w.ID && (seg.Status == SegAssigned || seg.Status == SegProcessing) {
				seg.Status = SegPending
				seg.AssignedWorker = ""
				seg.RetryCount++
			}
		}
		svc.pool.Remove(w.ID)
		svc.pool.Register(w.ID)
	}
}

func (svc *TranscodingService) CancelJob(jobID string) bool {
	job := svc.tracker.GetJob(jobID)
	if job == nil {
		return false
	}
	if job.Status == JobPending {
		svc.queue.Remove(jobID)
	}
	job.Status = JobCancelled
	fmt.Printf("[JOB] Cancelled %s\n", jobID)
	return true
}

func (svc *TranscodingService) GracefulShutdown() {
	fmt.Println("[SHUTDOWN] Initiating graceful shutdown...")
	svc.shutdownRequested = true

	for _, w := range svc.pool.GetAll() {
		if w.Status == WorkerBusy {
			svc.pool.Drain(w.ID)
		}
	}

	deadline := time.Now().Add(30 * time.Second)
	for {
		svc.mu.Lock()
		active := svc.processing
		svc.mu.Unlock()
		if !active || time.Now().After(deadline) {
			break
		}
		time.Sleep(100 * time.Millisecond)
	}

	fmt.Println("[SHUTDOWN] Graceful shutdown complete")
}

// ===========================================
// 10. HTTP SERVER & MAIN
// ===========================================

func main() {
	svc := NewTranscodingService(4)

	// --- HTTP Handlers ---
	http.HandleFunc("/api/transcode", func(w http.ResponseWriter, r *http.Request) {
		if r.Method != http.MethodPost {
			http.Error(w, `{"error":"method not allowed"}`, http.StatusMethodNotAllowed)
			return
		}
		var body struct {
			UserID     string `json:"userId"`
			SourceURL  string `json:"sourceUrl"`
			WebhookURL string `json:"webhookUrl"`
			Priority   int    `json:"priority"`
		}
		if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
			http.Error(w, `{"error":"invalid JSON"}`, http.StatusBadRequest)
			return
		}
		if body.SourceURL == "" || body.UserID == "" {
			http.Error(w, `{"error":"sourceUrl and userId required"}`, http.StatusBadRequest)
			return
		}
		if body.Priority == 0 {
			body.Priority = 5
		}
		job := svc.SubmitJob(body.UserID, body.SourceURL, body.WebhookURL, body.Priority)
		go svc.ProcessNextJob()

		w.Header().Set("Content-Type", "application/json")
		w.WriteHeader(http.StatusCreated)
		json.NewEncoder(w).Encode(map[string]interface{}{
			"jobId": job.ID, "status": job.Status,
		})
	})

	http.HandleFunc("/api/jobs/", func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "application/json")
		path := strings.TrimPrefix(r.URL.Path, "/api/jobs/")
		parts := strings.Split(path, "/")
		jobID := parts[0]

		if jobID == "" {
			http.Error(w, `{"error":"job ID required"}`, http.StatusBadRequest)
			return
		}

		// DELETE /api/jobs/:id
		if r.Method == http.MethodDelete {
			if svc.CancelJob(jobID) {
				json.NewEncoder(w).Encode(map[string]string{"status": "cancelled"})
			} else {
				w.WriteHeader(http.StatusNotFound)
				json.NewEncoder(w).Encode(map[string]string{"error": "Job not found"})
			}
			return
		}

		// GET /api/jobs/:id/manifest
		if len(parts) > 1 && parts[1] == "manifest" {
			job := svc.tracker.GetJob(jobID)
			if job == nil {
				w.WriteHeader(http.StatusNotFound)
				json.NewEncoder(w).Encode(map[string]string{"error": "Job not found"})
				return
			}
			if job.Status != JobCompleted {
				w.WriteHeader(http.StatusConflict)
				json.NewEncoder(w).Encode(map[string]string{"error": "Job not yet completed", "status": string(job.Status)})
				return
			}
			w.Header().Set("Content-Type", "application/vnd.apple.mpegurl")
			fmt.Fprint(w, generateMasterPlaylist(job))
			return
		}

		// GET /api/jobs/:id
		job := svc.tracker.GetJob(jobID)
		if job == nil {
			w.WriteHeader(http.StatusNotFound)
			json.NewEncoder(w).Encode(map[string]string{"error": "Job not found"})
			return
		}
		progress := svc.tracker.GetProgress(jobID)
		profileNames := make([]string, len(job.Profiles))
		for i, p := range job.Profiles {
			profileNames[i] = p.Name
		}
		json.NewEncoder(w).Encode(map[string]interface{}{
			"id":          job.ID,
			"userId":      job.UserID,
			"status":      job.Status,
			"progress":    progress,
			"sourceMeta":  job.SourceMeta,
			"profiles":    profileNames,
			"createdAt":   job.CreatedAt,
			"startedAt":   job.StartedAt,
			"completedAt": job.CompletedAt,
			"error":       job.Error,
		})
	})

	http.HandleFunc("/api/workers", func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "application/json")
		json.NewEncoder(w).Encode(map[string]interface{}{
			"pool":    svc.pool.GetStatus(),
			"workers": svc.pool.GetAll(),
		})
	})

	// --- Demo ---
	fmt.Println("=== Video Transcoding Service Demo ===\n")

	job1 := svc.SubmitJob("user-alice", "/uploads/alice-vacation.mp4", "https://example.com/webhook", 3)
	job2 := svc.SubmitJob("user-bob", "/uploads/bob-tutorial.mp4", "", 5)
	_ = svc.SubmitJob("user-alice", "/uploads/alice-concert.mp4", "https://example.com/webhook", 7)

	fmt.Println("\nQueue: 3 jobs submitted\n")

	fmt.Println("--- Processing Job 1 ---")
	svc.ProcessNextJob()

	progress := svc.tracker.GetProgress(job1.ID)
	fmt.Printf("\nJob 1 progress: %.0f%% (%d/%d segments)\n", progress.Percent, progress.Completed, progress.Total)
	fmt.Printf("Job 1 status: %s\n", job1.Status)

	if job1.Status == JobCompleted {
		fmt.Println("\n--- HLS Master Playlist ---")
		fmt.Print(generateMasterPlaylist(job1))
	}

	fmt.Println("--- Worker Pool ---")
	poolJSON, _ := json.MarshalIndent(svc.pool.GetStatus(), "", "  ")
	fmt.Println(string(poolJSON))

	fmt.Println("\n--- Processing Job 2 ---")
	svc.ProcessNextJob()

	fmt.Printf("\nJob 2 status: %s\n", job2.Status)

	fmt.Println("\n--- Processing Job 3 ---")
	svc.ProcessNextJob()

	fmt.Println("\n=== All jobs processed ===")

	// Start HTTP server with graceful shutdown
	srv := &http.Server{Addr: ":3900"}
	go func() {
		sigCh := make(chan os.Signal, 1)
		signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
		<-sigCh
		svc.GracefulShutdown()
		srv.Close()
	}()

	fmt.Println("\nHTTP server listening on http://localhost:3900")
	fmt.Println("Endpoints:")
	fmt.Println("  POST   /api/transcode        — Submit job")
	fmt.Println("  GET    /api/jobs/:id          — Job status")
	fmt.Println("  GET    /api/jobs/:id/manifest — HLS manifest")
	fmt.Println("  DELETE /api/jobs/:id          — Cancel job")
	fmt.Println("  GET    /api/workers           — Worker pool status")

	if err := srv.ListenAndServe(); err != http.ErrServerClosed {
		fmt.Printf("Server error: %v\n", err)
	}
}