package backend

import (
	"bytes"
	"context"
	"database/sql"
	"errors"
	"fmt"
	"io"
	"os/exec"
	"strings"
	"time"

	"github.com/pingcap/failpoint"

	"github.com/mem9-ai/drive9/pkg/datastore"
)

// VideoExtractRequest is the input to a pluggable video->text extractor.
type VideoExtractRequest struct {
	TenantID    string
	FileID      string
	Path        string
	ContentType string
	Data        []byte
	// BeforeProviderCall runs immediately before the extractor sends the
	// external Vision request, after local media preparation succeeds.
	BeforeProviderCall func()
	// SourcePath points at the task-scoped staged source when available. It
	// keeps large videos out of heap memory; extractors must not retain it.
	SourcePath string
}

// VideoExtractUsage reports the resource consumption of one video visual extraction call.
type VideoExtractUsage struct {
	PromptTokens     int
	CompletionTokens int
	FramesExtracted  int
}

// TotalTokens returns the sum of prompt and completion tokens.
func (u VideoExtractUsage) TotalTokens() int { return u.PromptTokens + u.CompletionTokens }

// VideoTextExtractor extracts searchable text from video bytes by sampling
// frames and describing them via a Vision model.
type VideoTextExtractor interface {
	ExtractVideoText(ctx context.Context, req VideoExtractRequest) (string, VideoExtractUsage, error)
}

// SegmentedVideoTextExtractor exposes the local duration probe, bounded visual
// segment extraction, and final text-only synthesis needed for durable
// long-video progress. VideoTextExtractor remains supported for existing
// providers and tests that perform one whole-video request.
type SegmentedVideoTextExtractor interface {
	ProbeVideoDuration(ctx context.Context, req VideoExtractRequest) (float64, error)
	ExtractVideoSegment(ctx context.Context, req VideoExtractRequest, segment VideoVisualSegment) (string, VideoExtractUsage, error)
	SynthesizeVideoText(ctx context.Context, req VideoExtractRequest, summaries string) (string, VideoExtractUsage, error)
}

// VideoVisualSegment is one bounded range on the complete video timeline.
type VideoVisualSegment struct {
	Index        int
	StartSeconds float64
	EndSeconds   float64
}

// VideoExtractRuntime is the effective tenant-scoped runtime for one video
// task. The worker resolves it immediately before execution.
type VideoExtractRuntime struct {
	Checkpoint      func(VideoExtractCheckpoint) error
	CheckpointState VideoExtractCheckpoint
	FinalAttempt    bool
	// BeforeExtract is forwarded to the extractor and runs at the external
	// provider-call boundary, after local video preprocessing succeeds.
	BeforeExtract       func()
	Extractor           VideoTextExtractor
	Timeout             time.Duration
	MaxSourceBytes      int64
	MaxExtractTextBytes int
	Workspace           *MediaExtractWorkspace
	AudioRuntime        *AudioExtractRuntime
}

// VideoExtractCheckpoint keeps completed modality output in the existing
// lease-fenced semantic task payload so retries do not repeat provider calls.
type VideoExtractCheckpoint struct {
	VisualDone              bool
	VisualText              string
	VisualCompletedSegments int
	VisualSegmentCount      int
	VisualSegmentsText      string
	AudioDone               bool
	AudioPresent            bool
	AudioCompletedSegments  int
	AudioText               string
	VisualTags              map[string]string
	AudioTags               map[string]string
}

const DefaultVideoExtractMaxTextBytes = 256 << 10

// VideoExtractTaskSpec carries the revision-scoped inputs needed to extract
// video visual text for one file version.
type VideoExtractTaskSpec struct {
	FileID      string
	Path        string
	ContentType string
	Revision    int64
}

// VideoExtractResult reports the outcome of one video extraction attempt.
type VideoExtractResult string

const (
	VideoExtractResultRuntimeNotConfigured VideoExtractResult = "runtime_not_configured"
	VideoExtractResultGetFileError         VideoExtractResult = "get_file_error"
	VideoExtractResultFileNotFound         VideoExtractResult = "file_not_found"
	VideoExtractResultNotConfirmed         VideoExtractResult = "not_confirmed"
	VideoExtractResultNotVideo             VideoExtractResult = "not_video"
	VideoExtractResultStale                VideoExtractResult = "stale"
	VideoExtractResultLoadError            VideoExtractResult = "load_error"
	VideoExtractResultTooLarge             VideoExtractResult = "too_large"
	VideoExtractResultTooLong              VideoExtractResult = "too_long"
	VideoExtractResultExtractError         VideoExtractResult = "extract_error"
	VideoExtractResultEmptyText            VideoExtractResult = "empty_text"
	VideoExtractResultUpdateError          VideoExtractResult = "update_error"
	VideoExtractResultWritten              VideoExtractResult = "written"
	VideoExtractResultPartialSuccess       VideoExtractResult = "partial_success"
	VideoExtractResultBudgetExhausted      VideoExtractResult = "budget_exhausted"
)

// allowedVideoMIME is the closed set of MIME types for video visual extraction.
var allowedVideoMIME = map[string]struct{}{
	"video/mp4":        {},
	"video/quicktime":  {},
	"video/x-msvideo":  {},
	"video/webm":       {},
	"video/x-matroska": {},
}

// videoExtensionMIME maps file extensions to canonical video MIME types.
var videoExtensionMIME = map[string]string{
	".mp4":  "video/mp4",
	".mov":  "video/quicktime",
	".avi":  "video/x-msvideo",
	".webm": "video/webm",
	".mkv":  "video/x-matroska",
}

func isAllowedVideoMIME(ct string) bool {
	_, ok := allowedVideoMIME[stripMIMEParams(ct)]
	return ok
}

func effectiveVideoMIME(path, contentType string) string {
	ct := stripMIMEParams(contentType)
	if isAllowedVideoMIME(ct) {
		return ct
	}
	// When content_type is absent or generic, try path extension.
	switch ct {
	case "", "application/octet-stream", "text/plain":
		path = strings.ToLower(path)
		for ext, mime := range videoExtensionMIME {
			if strings.HasSuffix(path, ext) {
				return mime
			}
		}
	}
	return ""
}

func isVideoContentType(contentType, path string) bool {
	return effectiveVideoMIME(path, contentType) != ""
}

func isSupportedVideoForSemanticTask(path, contentType string) bool {
	return isVideoContentType(contentType, path)
}

// SupportsAsyncVideoExtract reports whether this backend instance has the
// runtime dependencies needed to extract video visual text.
func (b *Dat9Backend) SupportsAsyncVideoExtract() bool {
	return b != nil && b.videoExtractEnabled && b.videoExtractor != nil
}

// SupportsVideoExtractTasks reports whether the backend can execute video
// tasks either with its process default or a tenant runtime resolved later.
func (b *Dat9Backend) SupportsVideoExtractTasks() bool {
	return b != nil && (b.SupportsAsyncVideoExtract() || b.videoExtractCapability)
}

// ProcessVideoExtractTaskWithRuntime runs a video task with an explicit
// tenant-scoped runtime, without requiring the process-default extractor.
func (b *Dat9Backend) ProcessVideoExtractTaskWithRuntime(ctx context.Context, task VideoExtractTaskSpec, rt VideoExtractRuntime) (VideoExtractResult, error) {
	if rt.MaxSourceBytes <= 0 {
		rt.MaxSourceBytes = b.videoExtractMaxSourceSize
	}
	if rt.Extractor == nil || rt.MaxSourceBytes <= 0 {
		return VideoExtractResultRuntimeNotConfigured, fmt.Errorf("async video extract runtime not configured")
	}
	return b.processVideoExtractTask(ctx, task, rt)
}

// ErrVideoExtractSourceTooLarge signals the video exceeds MaxSourceBytes.
var ErrVideoExtractSourceTooLarge = errors.New("video extract source exceeds configured max bytes")

// ProcessVideoExtractTask runs visual extraction for one durable
// video_extract_visual task. Terminal business outcomes return a nil error;
// runtime misconfiguration and transient failures return a retryable error.
func (b *Dat9Backend) ProcessVideoExtractTask(ctx context.Context, task VideoExtractTaskSpec) (VideoExtractResult, error) {
	runtime, ok := b.DefaultVideoExtractRuntime()
	if !ok {
		return VideoExtractResultRuntimeNotConfigured, fmt.Errorf("async video extract runtime not configured")
	}
	return b.processVideoExtractTask(ctx, task, runtime)
}

// DefaultVideoExtractRuntime returns the process-default video dependencies.
func (b *Dat9Backend) DefaultVideoExtractRuntime() (VideoExtractRuntime, bool) {
	if !b.SupportsAsyncVideoExtract() {
		return VideoExtractRuntime{}, false
	}
	return VideoExtractRuntime{
		Extractor: b.videoExtractor, Timeout: b.videoExtractTimeout,
		MaxSourceBytes: b.videoExtractMaxSourceSize, MaxExtractTextBytes: b.maxVideoExtractTextBytes,
		Workspace: b.mediaExtractWorkspace,
		AudioRuntime: func() *AudioExtractRuntime {
			if b.audioExtractor == nil {
				return nil
			}
			return &AudioExtractRuntime{Extractor: b.audioExtractor, Timeout: b.audioExtractTimeout, MaxSourceBytes: b.audioExtractMaxSourceSize, MaxAudioBytes: b.audioExtractMaxSize, MaxExtractTextBytes: b.maxAudioExtractTextBytes, Workspace: b.mediaExtractWorkspace}
		}(),
	}, true
}

func (b *Dat9Backend) processVideoExtractTask(ctx context.Context, task VideoExtractTaskSpec, rt VideoExtractRuntime) (VideoExtractResult, error) {
	start := time.Now()
	var result VideoExtractResult
	defer func() {
		b.recordTenantOperation("video_extract", "process", string(result), time.Since(start))
	}()

	if b.monthlyLLMCostExceededCheck(ctx) {
		b.recordTenantOperation("llm_cost_budget", "process_skip", "budget_exhausted", 0)
		result = VideoExtractResultBudgetExhausted
		return result, nil
	}

	f, err := b.store.GetFile(ctx, task.FileID)
	if err != nil {
		if errors.Is(err, datastore.ErrNotFound) {
			result = VideoExtractResultFileNotFound
			return result, nil
		}
		result = VideoExtractResultGetFileError
		return result, fmt.Errorf("get file: %w", err)
	}
	if f.Status != datastore.StatusConfirmed {
		result = VideoExtractResultNotConfirmed
		return result, nil
	}
	if task.Revision > 0 && f.Revision != task.Revision {
		result = VideoExtractResultStale
		return result, nil
	}

	ct := f.ContentType
	if ct == "" {
		ct = task.ContentType
	}
	resolvedMIME := effectiveVideoMIME(task.Path, ct)
	if resolvedMIME == "" {
		result = VideoExtractResultNotVideo
		return result, nil
	}

	var taskWorkspace *MediaExtractTaskWorkspace
	var sourcePath string
	var data []byte
	if rt.Workspace != nil {
		taskWorkspace, err = rt.Workspace.NewTask(ctx, task.FileID)
		if err != nil {
			result = VideoExtractResultLoadError
			return result, fmt.Errorf("create video extract workspace: %w", err)
		}
		defer func() { _ = taskWorkspace.Close() }()
		sourcePath, _, err = b.stageMediaFile(ctx, taskWorkspace, f, "source"+videoMIMEToExt(resolvedMIME), rt.MaxSourceBytes)
		if err != nil {
			b.recordTenantOperation("video_extract", "workspace_stage", "error", 0)
			if errors.Is(err, ErrMediaExtractSourceTooLarge) {
				result = VideoExtractResultTooLarge
				return result, nil
			}
			result = VideoExtractResultLoadError
			return result, fmt.Errorf("stage video source: %w", err)
		}
	} else {
		maxInMemoryBytes := rt.MaxSourceBytes
		if f.StorageType == datastore.StorageS3 && DefaultVideoExtractMaxProviderPayloadBytes < maxInMemoryBytes {
			maxInMemoryBytes = DefaultVideoExtractMaxProviderPayloadBytes
		}
		data, err = b.loadVideoBytesForExtract(ctx, f, maxInMemoryBytes)
		if err != nil {
			b.recordTenantOperation("video_extract", "source", "load_error", 0)
			if errors.Is(err, ErrVideoExtractSourceTooLarge) && maxInMemoryBytes == rt.MaxSourceBytes {
				result = VideoExtractResultTooLarge
				return result, nil
			}
			if errors.Is(err, ErrVideoExtractSourceTooLarge) {
				result = VideoExtractResultLoadError
				return result, ErrMediaExtractWorkspaceRequired
			}
			result = VideoExtractResultLoadError
			return result, fmt.Errorf("load video bytes: %w", err)
		}
	}

	state := rt.CheckpointState
	state.VisualTags = cloneMediaExtractTags(state.VisualTags)
	state.AudioTags = cloneMediaExtractTags(state.AudioTags)
	normalizeTags := func(visual, audio map[string]string) (map[string]string, map[string]string) {
		normalizedVisual, normalizedAudio, limited := normalizeVideoExtractTagSets(visual, audio)
		if limited {
			b.recordTenantOperation("video_extract", "attr_skip", "attribute_limit", 0)
		}
		return normalizedVisual, normalizedAudio
	}
	text := state.VisualText
	if state.VisualDone {
		// Pre-upgrade checkpoints stored the raw visual provider response in
		// VisualText and had no VisualTags field. Re-parse it opportunistically;
		// normalized checkpoints written by this version make this a no-op.
		for key, value := range prepareImageExtractWriteback(state.VisualText, rt.MaxExtractTextBytes).tags {
			state.VisualTags[key] = value
		}
	}
	state.VisualTags, state.AudioTags = normalizeTags(state.VisualTags, state.AudioTags)
	var visualErr error
	var sourceDurationSeconds float64
	if !state.VisualDone {
		text, visualErr = b.extractVideoVisual(ctx, task, resolvedMIME, sourcePath, data, rt, &state, &sourceDurationSeconds, normalizeTags)
		if errors.Is(visualErr, ErrVideoDurationTooLong) {
			result = VideoExtractResultTooLong
			return result, nil
		}
		if errors.Is(visualErr, ErrVideoProviderPayloadTooLarge) {
			result = VideoExtractResultTooLarge
			return result, nil
		}
	}
	if visualErr != nil {
		b.recordTenantOperation("video_extract", "visual", "error", 0)
	}
	var audioErr error
	audioSucceeded := false
	audioCompletedThisAttempt := false
	audioTextForOutput := state.AudioText
	audioPresentForOutput := state.AudioPresent
	if rt.AudioRuntime != nil && rt.AudioRuntime.Extractor != nil {
		if !state.AudioDone {
			audioRuntime := *rt.AudioRuntime
			audioRuntime.CheckpointTextMaxJSONBytes = videoCheckpointTextMaxJSONBytes
			audioRuntime.CheckpointState = AudioExtractCheckpoint{CompletedSegments: state.AudioCompletedSegments, Text: state.AudioText, Tags: state.AudioTags}
			if rt.Checkpoint != nil {
				audioRuntime.Checkpoint = func(next AudioExtractCheckpoint) error {
					state.AudioCompletedSegments = next.CompletedSegments
					state.AudioText = next.Text
					state.VisualTags, state.AudioTags = normalizeTags(state.VisualTags, next.Tags)
					return b.checkpointVideoExtract(rt.Checkpoint, state)
				}
			}
			audioCtx, cancelAudio := context.WithTimeout(ctx, effectiveAudioExtractTimeout(audioRuntime.Timeout, b.audioExtractTimeout))
			audioText, audioUsage, audioTags, audioPresent, extractErr := b.extractVideoAudioTrack(audioCtx, task, sourcePath, data, resolvedMIME, sourceDurationSeconds, taskWorkspace, audioRuntime)
			cancelAudio()
			state.VisualTags, state.AudioTags = normalizeTags(state.VisualTags, audioTags)
			b.recordAudioExtractUsage(task.FileID, audioUsage)
			audioTextForOutput = audioText
			audioPresentForOutput = audioPresent
			if extractErr != nil {
				if errors.Is(extractErr, ErrAudioExtractDurationTooLong) {
					result = VideoExtractResultTooLong
					return result, nil
				}
				b.recordTenantOperation("video_extract", "audio_ffmpeg", "error", 0)
				audioErr = extractErr
			} else {
				audioCompletedThisAttempt = true
				if rt.Checkpoint == nil || checkpointTextFits(audioText, effectiveAudioCheckpointTextBudget(audioRuntime)) {
					state.AudioDone = true
					state.AudioPresent = audioPresent
					state.AudioText = audioText
					if err := b.checkpointVideoExtract(rt.Checkpoint, state); err != nil {
						result = VideoExtractResultExtractError
						return result, fmt.Errorf("checkpoint video audio phase: %w", err)
					}
				} else {
					b.recordTenantOperation("video_extract", "checkpoint", "payload_oversize", 0)
				}
				if audioPresent {
					b.recordTenantOperation("video_extract", "audio_transcript", "ok", 0)
				}
			}
		}
		audioSucceeded = (state.AudioDone || audioCompletedThisAttempt) && audioPresentForOutput && strings.TrimSpace(audioTextForOutput) != ""
		if audioSucceeded {
			maxAudioText := rt.AudioRuntime.MaxExtractTextBytes
			if maxAudioText <= 0 {
				maxAudioText = b.maxAudioExtractTextBytes
			}
			text += "\n[Audio Transcript]\n" + sanitizeExtractedText(audioTextForOutput, maxAudioText)
		}
	}
	if (visualErr != nil || audioErr != nil) && !rt.FinalAttempt {
		result = VideoExtractResultExtractError
		return result, videoExtractModalityError(visualErr, audioErr)
	}
	if visualErr != nil && !audioSucceeded {
		result = VideoExtractResultExtractError
		return result, videoExtractModalityError(visualErr, audioErr)
	}
	text = sanitizeExtractedText(text, rt.MaxExtractTextBytes)
	emptyText := text == ""
	tags := mergeVideoExtractTags(state.VisualTags, state.AudioTags)

	expectedRevision := task.Revision
	if expectedRevision == 0 {
		expectedRevision = f.Revision
	}

	var updated bool
	err = b.store.InTx(ctx, func(tx *sql.Tx) error {
		if err := injectedVideoExtractWritebackError("videoExtractWritebackUpdateFileSearchTextError"); err != nil {
			return err
		}
		var txErr error
		updated, txErr = b.updateMediaExtractTextTx(tx, task.FileID, expectedRevision, text)
		if txErr != nil {
			return txErr
		}
		if !updated {
			return nil
		}
		if txErr = b.store.DeleteFileTagsBySourceTx(tx, task.FileID, datastore.TagSourceVideoExtract); txErr != nil {
			return txErr
		}
		skipped, txErr := b.store.InsertFileTagsWithSourceTx(tx, task.FileID, datastore.TagSourceVideoExtract, tags)
		if txErr != nil {
			return txErr
		}
		for range skipped {
			b.recordTenantOperation("video_extract", "attr_skip", "user_tag_conflict", 0)
		}
		if emptyText {
			// No replacement embed task follows an empty app-managed result.
			if !b.UsesDatabaseAutoEmbedding() {
				return b.store.ClearFileEmbeddingStateTx(tx, task.FileID)
			}
			return nil
		}
		if b.UsesDatabaseAutoEmbedding() || !b.appSemanticTasksEnabled {
			return nil
		}
		_, txErr = b.store.EnsureSemanticTaskQueuedTx(tx, newEmbedTask(
			b.genID(), task.FileID, expectedRevision, b.appEmbeddingGeneration,
			b.appEmbeddingConfigRevision, b.appEmbeddingTenantManaged, time.Now().UTC(),
		))
		return txErr
	})
	if err != nil {
		result = VideoExtractResultUpdateError
		return result, fmt.Errorf("update file search text: %w", err)
	}
	if !updated {
		result = VideoExtractResultStale
		return result, nil
	}
	if emptyText {
		result = VideoExtractResultEmptyText
		return result, nil
	}
	if audioErr != nil || visualErr != nil {
		b.recordTenantOperation("video_extract", "audio", "partial_failure", 0)
		result = VideoExtractResultPartialSuccess
	} else {
		result = VideoExtractResultWritten
	}
	return result, nil
}

const (
	videoVisualSegmentSummaryMaxJSONBytes = 560
	videoVisualSegmentsMaxJSONBytes       = 16 << 10
)

func (b *Dat9Backend) extractVideoVisual(
	ctx context.Context,
	task VideoExtractTaskSpec,
	contentType, sourcePath string,
	data []byte,
	rt VideoExtractRuntime,
	state *VideoExtractCheckpoint,
	sourceDurationSeconds *float64,
	normalizeTags func(map[string]string, map[string]string) (map[string]string, map[string]string),
) (string, error) {
	timeout := rt.Timeout
	if timeout <= 0 {
		timeout = b.videoExtractTimeout
	}
	request := VideoExtractRequest{
		TenantID: b.tenantID, FileID: task.FileID, Path: task.Path,
		ContentType: contentType, Data: data, SourcePath: sourcePath,
	}
	providerAttemptRecorded := false
	request.BeforeProviderCall = func() {
		if providerAttemptRecorded {
			return
		}
		providerAttemptRecorded = true
		if rt.BeforeExtract != nil {
			rt.BeforeExtract()
		}
	}
	segmented, ok := rt.Extractor.(SegmentedVideoTextExtractor)
	if !ok {
		taskCtx, cancel := context.WithTimeout(ctx, timeout)
		defer cancel()
		text, usage, err := rt.Extractor.ExtractVideoText(taskCtx, request)
		if err != nil {
			return "", err
		}
		return b.completeVideoVisual(task.FileID, text, usage, state, rt, normalizeTags)
	}

	probeCtx, cancelProbe := context.WithTimeout(ctx, timeout)
	duration, err := segmented.ProbeVideoDuration(probeCtx, request)
	cancelProbe()
	if err != nil {
		return "", fmt.Errorf("probe video duration: %w", err)
	}
	*sourceDurationSeconds = duration
	segments, err := videoVisualSegments(duration)
	if err != nil {
		return "", err
	}
	if state.VisualSegmentCount != len(segments) || state.VisualCompletedSegments < 0 || state.VisualCompletedSegments > len(segments) {
		state.VisualCompletedSegments = 0
		state.VisualSegmentCount = len(segments)
		state.VisualSegmentsText = ""
		state.VisualTags = nil
	}

	for i := state.VisualCompletedSegments; i < len(segments); i++ {
		raw, usage, callErr := segmented.ExtractVideoSegment(ctx, request, segments[i])
		if callErr != nil {
			if errors.Is(callErr, errVideoSegmentNoFrames) && strings.TrimSpace(state.VisualSegmentsText) != "" {
				state.VisualCompletedSegments = i + 1
				b.recordTenantOperation("video_extract", "frames", "count_0", 0)
				if err := b.checkpointVideoExtract(rt.Checkpoint, *state); err != nil {
					return "", fmt.Errorf("checkpoint empty video visual segment %d: %w", i, err)
				}
				continue
			}
			if usage.TotalTokens() > 0 {
				b.recordVideoExtractUsage(task.FileID, usage)
			}
			return "", fmt.Errorf("extract visual segment %d: %w", i, callErr)
		}
		b.recordVideoExtractUsage(task.FileID, usage)
		b.recordTenantOperation("video_extract", "frames", fmt.Sprintf("count_%d", minInt(usage.FramesExtracted, 10)), 0)
		writeback := prepareImageExtractWriteback(raw, rt.MaxExtractTextBytes)
		state.VisualTags = mergeMediaExtractTags(state.VisualTags, writeback.tags)
		state.VisualTags, state.AudioTags = normalizeTags(state.VisualTags, state.AudioTags)
		state.VisualSegmentsText = appendVideoVisualSegmentSummary(state.VisualSegmentsText, segments[i], writeback.text)
		state.VisualCompletedSegments = i + 1
		if err := b.checkpointVideoExtract(rt.Checkpoint, *state); err != nil {
			return "", fmt.Errorf("checkpoint video visual segment %d: %w", i, err)
		}
	}

	callCtx, cancel := context.WithTimeout(ctx, timeout)
	raw, usage, err := segmented.SynthesizeVideoText(callCtx, request, state.VisualSegmentsText)
	cancel()
	if err != nil {
		return "", fmt.Errorf("synthesize visual segments: %w", err)
	}
	text, err := b.completeVideoVisual(task.FileID, raw, usage, state, rt, normalizeTags)
	if err != nil {
		return "", err
	}
	return text, nil
}

func (b *Dat9Backend) completeVideoVisual(
	taskID, raw string,
	usage VideoExtractUsage,
	state *VideoExtractCheckpoint,
	rt VideoExtractRuntime,
	normalizeTags func(map[string]string, map[string]string) (map[string]string, map[string]string),
) (string, error) {
	next := *state
	writeback := prepareImageExtractWriteback(raw, rt.MaxExtractTextBytes)
	next.VisualTags = mergeMediaExtractTags(next.VisualTags, writeback.tags)
	next.VisualTags, next.AudioTags = normalizeTags(next.VisualTags, next.AudioTags)
	b.recordVideoExtractUsage(taskID, usage)
	if usage.FramesExtracted > 0 {
		b.recordTenantOperation("video_extract", "frames", fmt.Sprintf("count_%d", minInt(usage.FramesExtracted, 10)), 0)
	}
	text := writeback.text
	if strings.TrimSpace(text) != "" {
		text = "[Visual Summary]\n" + text
	}
	if rt.Checkpoint != nil && !checkpointTextFits(text, videoCheckpointTextMaxJSONBytes) {
		b.recordTenantOperation("video_extract", "checkpoint", "payload_oversize", 0)
		state.VisualTags = next.VisualTags
		state.AudioTags = next.AudioTags
		return text, nil
	}
	next.VisualDone = true
	next.VisualText = text
	next.VisualCompletedSegments = 0
	next.VisualSegmentCount = 0
	next.VisualSegmentsText = ""
	published, err := b.publishVideoExtractCheckpoint(rt.Checkpoint, next)
	if err != nil {
		return "", fmt.Errorf("checkpoint video visual phase: %w", err)
	}
	if published {
		*state = next
	} else {
		state.VisualTags = next.VisualTags
		state.AudioTags = next.AudioTags
	}
	return text, nil
}

func appendVideoVisualSegmentSummary(current string, segment VideoVisualSegment, text string) string {
	text = boundCheckpointText(strings.TrimSpace(text), videoVisualSegmentSummaryMaxJSONBytes)
	header := fmt.Sprintf("[Segment %d %.0f-%.0fs]\n", segment.Index+1, segment.StartSeconds, segment.EndSeconds)
	if current != "" {
		current += "\n"
	}
	return boundCheckpointText(current+header+text, videoVisualSegmentsMaxJSONBytes)
}

func videoExtractModalityError(visualErr, audioErr error) error {
	var modalityErrors []error
	if visualErr != nil {
		modalityErrors = append(modalityErrors, fmt.Errorf("visual: %w", visualErr))
	}
	if audioErr != nil {
		modalityErrors = append(modalityErrors, fmt.Errorf("audio: %w", audioErr))
	}
	return fmt.Errorf("extract video modalities: %w", errors.Join(modalityErrors...))
}

func (b *Dat9Backend) checkpointVideoExtract(checkpoint func(VideoExtractCheckpoint) error, state VideoExtractCheckpoint) error {
	_, err := b.publishVideoExtractCheckpoint(checkpoint, state)
	return err
}

func (b *Dat9Backend) publishVideoExtractCheckpoint(checkpoint func(VideoExtractCheckpoint) error, state VideoExtractCheckpoint) (bool, error) {
	if checkpoint == nil {
		return true, nil
	}
	if err := checkpoint(state); err != nil {
		if errors.Is(err, datastore.ErrSemanticTaskPayloadTooLarge) {
			b.recordTenantOperation("video_extract", "checkpoint", "payload_oversize", 0)
			return false, nil
		}
		return false, err
	}
	return true, nil
}

func checkpointTextFits(text string, maxJSONBytes int) bool {
	return boundCheckpointText(text, maxJSONBytes) == text
}

func (b *Dat9Backend) extractVideoAudioTrack(ctx context.Context, task VideoExtractTaskSpec, sourcePath string, data []byte, contentType string, sourceDurationSeconds float64, workspace *MediaExtractTaskWorkspace, rt AudioExtractRuntime) (string, AudioExtractUsage, map[string]string, bool, error) {
	if workspace == nil {
		return "", AudioExtractUsage{}, nil, false, fmt.Errorf("video audio extraction requires a task workspace")
	}
	if sourcePath == "" {
		var err error
		sourcePath, _, err = workspace.StageReader(ctx, "video-audio-source"+videoMIMEToExt(contentType), bytes.NewReader(data), DefaultVideoExtractMaxSourceBytes)
		if err != nil {
			return "", AudioExtractUsage{}, nil, false, err
		}
	}
	present, err := probeMediaHasAudio(ctx, sourcePath)
	if err != nil || !present {
		return "", AudioExtractUsage{}, nil, present, err
	}
	if sourceDurationSeconds <= 0 {
		sourceDurationSeconds, err = probeAudioDuration(ctx, sourcePath, nil, contentType)
		if err != nil {
			return "", AudioExtractUsage{}, nil, true, fmt.Errorf("probe video audio duration: %w", err)
		}
	}
	if sourceDurationSeconds > float64(maxAudioExtractDurationSeconds) {
		return "", AudioExtractUsage{}, nil, true, ErrAudioExtractDurationTooLong
	}
	segments, err := splitAudioWithFFmpeg(ctx, sourcePath, nil, contentType, workspace, rt.MaxAudioBytes)
	if err != nil {
		return "", AudioExtractUsage{}, nil, true, err
	}
	text, usage, tags, _, err := b.extractAudioSegmentsWithTags(ctx,
		AudioExtractTaskSpec{FileID: task.FileID, Path: task.Path + ".audio.mp3", Revision: task.Revision},
		segments, rt.MaxAudioBytes, sourceDurationSeconds, rt, false, false,
	)
	return text, usage, tags, true, err
}

func probeMediaHasAudio(ctx context.Context, sourcePath string) (bool, error) {
	cmd := exec.CommandContext(ctx, "ffprobe", "-v", "error", "-select_streams", "a:0", "-show_entries", "stream=index", "-of", "csv=p=0", sourcePath)
	out, err := cmd.Output()
	if err != nil {
		if ctx.Err() != nil {
			return false, ctx.Err()
		}
		return false, err
	}
	return strings.TrimSpace(string(out)) != "", nil
}

func injectedVideoExtractWritebackError(name string) error {
	var injected error
	failpoint.Inject(name, func(val failpoint.Value) {
		switch v := val.(type) {
		case string:
			injected = errors.New(v)
		case bool:
			if v {
				injected = fmt.Errorf("injected failpoint %s", name)
			}
		}
	})
	return injected
}

func (b *Dat9Backend) loadVideoBytesForExtract(ctx context.Context, f *datastore.File, maxSize int64) ([]byte, error) {
	if f.ContentLayout == datastore.ContentLayoutAppendLog {
		return nil, ErrAppendLogUnsupported
	}
	if maxSize > 0 && f.SizeBytes > maxSize {
		return nil, ErrVideoExtractSourceTooLarge
	}
	if f.StorageType == datastore.StorageDB9 {
		if maxSize > 0 && int64(len(f.ContentBlob)) > maxSize {
			return nil, ErrVideoExtractSourceTooLarge
		}
		return append([]byte(nil), f.ContentBlob...), nil
	}
	if f.StorageType != datastore.StorageS3 {
		return nil, fmt.Errorf("unsupported storage type: %s", f.StorageType)
	}
	if b.s3 == nil {
		return nil, fmt.Errorf("s3 client not configured")
	}
	rc, err := b.s3.GetObject(ctx, f.StorageRef)
	if err != nil {
		return nil, err
	}
	defer func() { _ = rc.Close() }()

	reader := io.Reader(rc)
	if maxSize > 0 {
		reader = io.LimitReader(rc, maxSize+1)
	}
	data, err := io.ReadAll(reader)
	if err != nil {
		return nil, err
	}
	if maxSize > 0 && int64(len(data)) > maxSize {
		return nil, ErrVideoExtractSourceTooLarge
	}
	return data, nil
}