gopeed/pkg/download/engine/ffmpeg/input.go

// Package ffmpeg runs an embedded FFmpeg with host-controlled media inputs.
package ffmpeg

import (
	"context"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"strconv"
	"strings"
	"time"
)

const httpBlockSize = 1024 * 1024

// Input is a file visible to FFmpeg. Only verified HTTP Range sources seek.
type Input interface {
	io.ReadCloser
	Seek(int64, int) (int64, error)
	Size() int64
	Seekable() bool
}

var ErrSeek = errors.New("ffmpeg: input requires seeking but source is sequential")

type HTTPSource struct {
	URL     string            `json:"url"`
	Headers map[string]string `json:"headers"`
}

type httpInput struct {
	ctx                   context.Context
	client                *http.Client
	source                HTTPSource
	body                  io.ReadCloser
	size, pos, blockStart int64
	block                 []byte
	ranged                bool
	etag, modified        string
}

// OpenHTTP combines capability detection with the first media read. A 200
// response remains open and is consumed sequentially without another request.
func OpenHTTP(ctx context.Context, client *http.Client, source HTTPSource) (Input, error) {
	u, err := url.Parse(source.URL)
	if err != nil || (u.Scheme != "http" && u.Scheme != "https") || u.Host == "" {
		return nil, errors.New("ffmpeg: expected an HTTP(S) input URL")
	}
	h := &httpInput{ctx: ctx, client: client, source: source, size: -1}
	resp, err := h.request(0, httpBlockSize-1)
	if err != nil {
		return nil, err
	}
	if resp.StatusCode == http.StatusOK {
		if err = identityEncoding(resp); err != nil {
			resp.Body.Close()
			return nil, err
		}
		h.body, h.size = resp.Body, resp.ContentLength
		return h, nil
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusPartialContent {
		return nil, fmt.Errorf("ffmpeg: HTTP input returned %d", resp.StatusCode)
	}
	start, end, size, err := contentRange(resp.Header.Get("Content-Range"))
	if err != nil || start != 0 || end >= httpBlockSize {
		return nil, errors.New("ffmpeg: invalid initial Content-Range")
	}
	h.size, h.ranged = size, true
	h.etag = resp.Header.Get("ETag")
	if strings.HasPrefix(h.etag, "W/") {
		h.etag = ""
	}
	h.modified = resp.Header.Get("Last-Modified")
	h.block, err = readBlock(resp, end-start+1)
	return h, err
}

func (h *httpInput) request(start, end int64) (*http.Response, error) {
	ctx, cancel := context.WithCancel(h.ctx)
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, h.source.URL, nil)
	if err != nil {
		cancel()
		return nil, err
	}
	for k, v := range h.source.Headers {
		req.Header.Set(k, v)
	}
	// Offsets must refer to the actual media bytes, not a decompressed response.
	req.Header.Set("Accept-Encoding", "identity")
	req.Header.Set("Range", fmt.Sprintf("bytes=%d-%d", start, end))
	if h.etag != "" {
		req.Header.Set("If-Range", h.etag)
	} else if h.modified != "" {
		req.Header.Set("If-Range", h.modified)
	}
	timer := time.AfterFunc(15*time.Second, cancel)
	resp, err := h.client.Do(req)
	timer.Stop()
	if err != nil {
		cancel()
		return nil, err
	}
	resp.Body = &idleMediaBody{ReadCloser: resp.Body, cancel: cancel}
	return resp, nil
}

func identityEncoding(resp *http.Response) error {
	if enc := resp.Header.Get("Content-Encoding"); enc != "" && !strings.EqualFold(enc, "identity") {
		return errors.New("ffmpeg: encoded HTTP responses cannot represent media byte offsets")
	}
	return nil
}

func contentRange(value string) (start, end, size int64, err error) {
	err = errors.New("invalid Content-Range")
	if !strings.HasPrefix(value, "bytes ") {
		return
	}
	parts := strings.Split(strings.TrimPrefix(value, "bytes "), "/")
	if len(parts) != 2 {
		return
	}
	bounds := strings.Split(parts[0], "-")
	if len(bounds) != 2 {
		return
	}
	var e error
	if start, e = strconv.ParseInt(bounds[0], 10, 64); e != nil {
		return
	}
	if end, e = strconv.ParseInt(bounds[1], 10, 64); e != nil {
		return
	}
	if size, e = strconv.ParseInt(parts[1], 10, 64); e != nil {
		return
	}
	if start < 0 || end < start || size <= end {
		return
	}
	err = nil
	return
}

func readBlock(resp *http.Response, length int64) ([]byte, error) {
	if err := identityEncoding(resp); err != nil {
		return nil, err
	}
	if resp.ContentLength >= 0 && resp.ContentLength != length {
		return nil, errors.New("ffmpeg: Range response length mismatch")
	}
	b, err := io.ReadAll(io.LimitReader(resp.Body, length+1))
	if err != nil {
		return nil, err
	}
	if int64(len(b)) != length {
		return nil, errors.New("ffmpeg: truncated or oversized Range response")
	}
	return b, nil
}

func (h *httpInput) Read(p []byte) (int, error) {
	if len(p) == 0 {
		return 0, nil
	}
	if err := h.ctx.Err(); err != nil {
		return 0, err
	}
	if !h.ranged {
		return h.body.Read(p)
	}
	if h.pos >= h.size {
		return 0, io.EOF
	}
	if h.pos < h.blockStart || h.pos >= h.blockStart+int64(len(h.block)) {
		end := min(h.pos+httpBlockSize-1, h.size-1)
		resp, err := h.request(h.pos, end)
		if err != nil {
			return 0, err
		}
		defer resp.Body.Close()
		if resp.StatusCode != http.StatusPartialContent {
			return 0, fmt.Errorf("ffmpeg: Range request returned %d; source changed or stopped supporting Range", resp.StatusCode)
		}
		start, actualEnd, size, err := contentRange(resp.Header.Get("Content-Range"))
		if err != nil || start != h.pos || actualEnd > end || size != h.size {
			return 0, errors.New("ffmpeg: inconsistent Content-Range")
		}
		if h.etag != "" && resp.Header.Get("ETag") != h.etag {
			return 0, errors.New("ffmpeg: HTTP input ETag changed")
		}
		if h.etag == "" && h.modified != "" && resp.Header.Get("Last-Modified") != h.modified {
			return 0, errors.New("ffmpeg: HTTP input Last-Modified changed")
		}
		h.block, err = readBlock(resp, actualEnd-start+1)
		if err != nil {
			return 0, err
		}
		h.blockStart = start
	}
	n := copy(p, h.block[h.pos-h.blockStart:])
	h.pos += int64(n)
	return n, nil
}

func (h *httpInput) Seek(offset int64, whence int) (int64, error) {
	if !h.ranged {
		return 0, ErrSeek
	}
	base := int64(0)
	switch whence {
	case io.SeekStart:
	case io.SeekCurrent:
		base = h.pos
	case io.SeekEnd:
		base = h.size
	default:
		return 0, errors.New("invalid seek origin")
	}
	next := base + offset
	if next < 0 || (offset > 0 && next < base) {
		return 0, errors.New("invalid seek offset")
	}
	h.pos = next
	return next, nil
}

func (h *httpInput) Size() int64    { return h.size }
func (h *httpInput) Seekable() bool { return h.ranged }
func (h *httpInput) Close() error {
	if h.body != nil {
		return h.body.Close()
	}
	return nil
}

// Only time spent waiting on the network counts toward this idle deadline.
// Pauses between reads (disk writes or FFmpeg scheduling) do not count.
type idleMediaBody struct {
	io.ReadCloser
	cancel context.CancelFunc
}

func (b *idleMediaBody) Read(p []byte) (int, error) {
	timer := time.AfterFunc(15*time.Second, b.cancel)
	defer timer.Stop()
	return b.ReadCloser.Read(p)
}
func (b *idleMediaBody) Close() error { b.cancel(); return b.ReadCloser.Close() }