diff --git a/go.mod b/go.mod index 13db5774..f3575db8 100644 --- a/go.mod +++ b/go.mod @@ -3,6 +3,7 @@ module github.com/GopeedLab/gopeed go 1.25.4 require ( + codeberg.org/gruf/go-ffmpreg v0.6.23-0.20260827095416-df1836cb5be0 github.com/GopeedLab/webview_go v0.0.0-20260423085439-7a2f88b6e9b5 github.com/anacrolix/torrent v1.60.1-0.20251217073903-486bcbe758e0 github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5 @@ -21,6 +22,7 @@ require ( github.com/pkg/errors v0.9.1 github.com/rs/dnscache v0.0.0-20230804202142-fc85eb664529 github.com/rs/zerolog v1.31.0 + github.com/tetratelabs/wazero v1.12.0 github.com/xiaoqidun/setft v0.0.0-20220310121541-be86327699ad go.etcd.io/bbolt v1.4.3 golang.org/x/exp v0.0.0-20251219203646-944ab1f22d93 @@ -156,7 +158,7 @@ require ( golang.org/x/crypto v0.48.0 // indirect golang.org/x/mod v0.33.0 // indirect golang.org/x/net v0.50.0 - golang.org/x/sys v0.41.0 // indirect + golang.org/x/sys v0.44.0 // indirect golang.org/x/text v0.34.0 golang.org/x/time v0.15.0 // indirect golang.org/x/tools v0.42.0 // indirect diff --git a/go.sum b/go.sum index 9c4680f4..b1f117c1 100644 --- a/go.sum +++ b/go.sum @@ -14,6 +14,8 @@ cloud.google.com/go/pubsub v1.0.1/go.mod h1:R0Gpsv3s54REJCy4fxDixWD93lHJMoZTyQ2k cloud.google.com/go/pubsub v1.1.0/go.mod h1:EwwdRX2sKPjnvnqCa270oGRyludottCI76h+R3AArQw= cloud.google.com/go/storage v1.0.0/go.mod h1:IhtSnM/ZTZV8YYJWCY8RULGVqBDmpoyjwiyrjsg+URw= cloud.google.com/go/storage v1.5.0/go.mod h1:tpKbwo567HUNpVclU5sGELwQWBDZ8gh0ZeosJ0Rtdos= +codeberg.org/gruf/go-ffmpreg v0.6.23-0.20260827095416-df1836cb5be0 h1:AqNFl60o4xhTTOxmzAvSf7pazXHKBGuV5ATw7sETIRA= +codeberg.org/gruf/go-ffmpreg v0.6.23-0.20260827095416-df1836cb5be0/go.mod h1:XM6eW8CSgAuZRKRynm5nDiSNedm0rav30IychqygOHE= crawshaw.io/iox v0.0.0-20181124134642-c51c3df30797/go.mod h1:sXBiorCo8c46JlQV3oXPKINnZ8mcqnye1EkVkqsectk= crawshaw.io/sqlite v0.3.2/go.mod h1:igAO5JulrQ1DbdZdtVq48mnZUBAPOeFzer7VhDWNtW4= dario.cat/mergo v1.0.0 h1:AGCNq9Evsj31mOgNPcLyXc+4PNABt905YmuqPYYpBWk= @@ -552,6 +554,8 @@ github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4= github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +github.com/tetratelabs/wazero v1.12.0 h1:DuWcpNu/FzgEXgGBDp8J1Spc+CWOvvtvVyjKlaZopYU= +github.com/tetratelabs/wazero v1.12.0/go.mod h1:LvKtzl2RqO4gyF27BiXU+nKAjcV8f38U+kP/q2vgxh0= github.com/tidwall/btree v1.8.1 h1:27ehoXvm5AG/g+1VxLS1SD3vRhp/H7LuEfwNvddEdmA= github.com/tidwall/btree v1.8.1/go.mod h1:jBbTdUWhSZClZWoDg54VnvV7/54modSOzDN7VXftj1A= github.com/tinylib/msgp v1.0.2/go.mod h1:+d+yLhGm8mzTaHzB+wgMYrodPfmZrzkirds8fDWklFE= @@ -732,6 +736,8 @@ golang.org/x/sys v0.12.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.13.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.41.0 h1:Ivj+2Cp/ylzLiEU89QhWblYnOE9zerudt9Ftecq2C6k= golang.org/x/sys v0.41.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= +golang.org/x/sys v0.44.0 h1:ildZl3J4uzeKP07r2F++Op7E9B29JRUy+a27EibtBTQ= +golang.org/x/sys v0.44.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8= golang.org/x/term v0.2.0/go.mod h1:TVmDHMZPmdnySmBfhjOoOdhjzdE1h4u1VwSiw2l1Nuc= diff --git a/internal/protocol/http/config.go b/internal/protocol/http/config.go index fdabb292..7a09e59a 100644 --- a/internal/protocol/http/config.go +++ b/internal/protocol/http/config.go @@ -1,5 +1,8 @@ package http +// DefaultUserAgent is shared by HTTP downloads and extension HTTP requests. +const DefaultUserAgent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/116.0.0.0 Safari/537.36" + type config struct { UserAgent string `json:"userAgent"` Connections int `json:"connections"` diff --git a/internal/protocol/http/fetcher_manager.go b/internal/protocol/http/fetcher_manager.go index 99d9b571..833a4b08 100644 --- a/internal/protocol/http/fetcher_manager.go +++ b/internal/protocol/http/fetcher_manager.go @@ -72,7 +72,7 @@ func (fm *FetcherManager) AutoRename() bool { func (fm *FetcherManager) DefaultConfig() any { return &config{ - UserAgent: "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/116.0.0.0 Safari/537.36", + UserAgent: DefaultUserAgent, Connections: 16, } } diff --git a/pkg/base/model.go b/pkg/base/model.go index 46bcace2..7b49b297 100644 --- a/pkg/base/model.go +++ b/pkg/base/model.go @@ -94,9 +94,6 @@ type Resource struct { } func (r *Resource) Validate() error { - if r.Name == "" { - return fmt.Errorf("invalid resource name") - } if len(r.Files) == 0 { return fmt.Errorf("invalid resource files") } diff --git a/pkg/download/engine/engine.go b/pkg/download/engine/engine.go index fb93662a..147f1f98 100644 --- a/pkg/download/engine/engine.go +++ b/pkg/download/engine/engine.go @@ -9,6 +9,7 @@ import ( "github.com/GopeedLab/gopeed/pkg/base" gojaerror "github.com/GopeedLab/gopeed/pkg/download/engine/inject/error" fetchapi "github.com/GopeedLab/gopeed/pkg/download/engine/inject/fetch" + ffmpegapi "github.com/GopeedLab/gopeed/pkg/download/engine/inject/ffmpeg" "github.com/GopeedLab/gopeed/pkg/download/engine/inject/file" "github.com/GopeedLab/gopeed/pkg/download/engine/inject/stream" "github.com/GopeedLab/gopeed/pkg/download/engine/inject/vm" @@ -159,8 +160,10 @@ func (e *Engine) addCleanup(cleanup func()) { } type Config struct { - ProxyConfig *base.DownloaderProxyConfig - StreamConfig *stream.Config + // HTTPUserAgent is a snapshot of the downloader HTTP default. Nil leaves standalone engines unchanged. + HTTPUserAgent *string + ProxyConfig *base.DownloaderProxyConfig + StreamConfig *stream.Config } func NewEngine(cfg *Config) *Engine { @@ -203,9 +206,17 @@ func NewEngine(cfg *Config) *Engine { if err := stream.Enable(runtime, loop, cfg.StreamConfig); err != nil { return } + if err := ffmpegapi.Enable(runtime, loop, &ffmpegapi.Config{ + ProxyHandler: cfg.ProxyConfig.ToHandler(), + DefaultUserAgent: cfg.HTTPUserAgent, + RegisterCleanup: engine.addCleanup, + }); err != nil { + return + } if err := fetchapi.Enable(runtime, loop, &fetchapi.Config{ - ProxyHandler: cfg.ProxyConfig.ToHandler(), - RegisterCleanup: engine.addCleanup, + ProxyHandler: cfg.ProxyConfig.ToHandler(), + DefaultUserAgent: cfg.HTTPUserAgent, + RegisterCleanup: engine.addCleanup, }); err != nil { return } diff --git a/pkg/download/engine/ffmpeg/README.md b/pkg/download/engine/ffmpeg/README.md new file mode 100644 index 00000000..c7ef41bd --- /dev/null +++ b/pkg/download/engine/ffmpeg/README.md @@ -0,0 +1,128 @@ +# Extension FFmpeg runtime + +`gopeed.runtime.ffmpeg.merge(options)` returns a `ReadableStream`. +It merges one video track and one audio track without re-encoding, using +[go-ffmpreg](https://codeberg.org/gruf/go-ffmpreg)'s embedded FFmpeg 5.1.10 and +wazero. No system FFmpeg installation is required. go-ffmpreg and its embedded +GPL-enabled FFmpeg are distributed under GPLv3; see the pinned dependency's +LICENSE, README and build scripts for its source and included components. + +## HTTP inputs and one downloadable output + +```js +gopeed.events.onResolve(async (ctx) => { + // Resolve these URLs and headers using the site's API in your extension. + const video = { url: videoURL, headers: { Referer: pageURL } }; + const audio = { url: audioURL, headers: { Referer: pageURL } }; + const url = await gopeed.runtime.blob.createObjectURL( + () => gopeed.runtime.ffmpeg.merge({ video, audio }), + { contentType: "video/mp4" } + ); + ctx.res = { + name: "video", + files: [{ name: "video.mp4", req: { url } }] + }; +}); +``` + +The opener creates a new merge on each open. Output size is unknown and range +support must remain disabled. Do not declare the sum of the input sizes as the +output size. The downloader reads the merged output as one file; FFmpeg errors +propagate through the stream instead of completing a partial download. + +## JS inputs, including SABR + +Stream inputs **must** be created by an `inputs({ signal })` callback. It returns +`{ video, audio }`, each an HTTP descriptor or `ReadableStream`, +including mixed inputs. Do not start producers before this callback. SABR/UMP decoding belongs +to the extension: pass the resulting media streams, with initialization data, +not the SABR protocol bytes. Keep each selected track's codec/configuration +stable for the duration of a merge. + +```js +const output = gopeed.runtime.ffmpeg.merge({ + inputs: async ({ signal }) => { + const session = await prepareSession({ signal }); + const { videoStream, audioStream, abort } = await session.openStreams(); + if (signal.aborted) { abort(); throw new Error("Cancelled"); } + signal.addEventListener("abort", abort, { once: true }); + return { video: videoStream, audio: audioStream }; + }, + signal: abortController.signal, + args: ["-shortest", "-metadata", "title=Example"] +}); +``` + +Creating the output does not start requests or acquire input readers until the +output is pulled and a global execution slot is acquired. Queued jobs do not +open HTTP inputs, invoke input factories, or allocate media ring buffers. +The callback runs once per output; its signal aborts on cancellation, engine +shutdown, failure, and completion. A callback must cooperate with cancellation; +returned streams are cancelled even if the callback finishes after cancellation. +A stream input is single-use. To retry a Blob-backed task, +its opener must create a new SABR session and fresh input streams. Arbitrary +output byte ranges / resumable merged output are not supported. + +`gopeed.runtime.ffmpeg.supportsInputFactory === true` identifies this API. +Older builds accept eager streams and must be upgraded before using callbacks. + +## Options + +- `inputs`: callback returning `{ video, audio }`, synchronously or asynchronously. + Required for stream inputs. Cannot be combined with top-level `video`/`audio`. +- `video`, `audio`: HTTP descriptors only when supplied at the top level; both + are required unless `inputs` is supplied. HTTP descriptors have `url` and optional + `headers` (standard `HeadersInit`). HTTP requests use Gopeed's configured + proxy. Supply required cookies/authorization explicitly; browser or `fetch` + cookie jars are not implicitly shared. When `User-Agent` is absent, requests + use the downloader HTTP protocol's configured UA, captured when the extension + engine is created, matching fetch and XHR. Explicit UA values (including an + empty string) take precedence. +- `format`: defaults to `"mp4"`. MP4 output always enables fragmentation + (`frag_keyframe+empty_moov+default_base_moof`) for non-seekable output. The + other accepted muxers are `"mkv"`/`"matroska"`, `"webm"`, and `"mpegts"`. + Codec/container incompatibility is an error, not an automatic conversion. + Choose the matching filename and Blob content type in the extension. +- `args`: optional array of **output options**, not a shell command or a full + FFmpeg invocation. Defaults select `0:v:0`, `1:a:0`, and `-c copy`. Additional + `-map` options are additive. Allowed options are listed in `outputOptions` in + `runtime.go`: metadata, dispositions, bitstream filters, duration/shortest, + fragmentation, timestamp/interleaving options and `-strict`. Codec options + must remain `copy`. Extra inputs, output paths, reports and arbitrary global + options are rejected. Required MP4 fragmentation is enforced by the host. +- `signal`: optional `AbortSignal`. Aborting or cancelling the output stops + network requests and native execution, cancels input readers, and removes + temporary buffers. Await/consume the output inside a live engine or retain + it through a Blob URL; unowned work ends with the extension engine. + +The backend defaults to strict input errors (`-xerror`). Native errors include +the FFmpeg exit code and a bounded stderr tail. No host directories are exposed +to the WASM module; it can access only its two input files and standard output. + +## Input behavior and resource limits + +HTTP inputs start with a 1 MiB Range GET, reusing the bytes for media reads. +A validated 206 response with a known total size enables seeking; a 200 +response is consumed as-is, sequentially. Subsequent ranges verify positions, +lengths, total size and available ETag/Last-Modified validators. Without a +server validator, same-size remote changes cannot be reliably detected. +Content encoding is forced to identity. Authentication/status failures and +malformed range responses are errors. There is no full-file seek fallback. + +Sequential JS inputs are drained independently into memory ring buffers that +grow on demand up to **32 MiB per input**, independent of total media size. +No temporary files are used. This does not enable seeking. Full buffers apply +async backpressure until space is available or the operation is cancelled. +Extensions must propagate backpressure through their upstream queues and +network reads. Independent upstream track requests avoid blocking one track +behind another in a shared response. Buffers are released on close. +Blob output writes also wait asynchronously for bounded downstream capacity, +so a slow consumer does not block the extension event loop. + +At most two WASM merges execute concurrently per process, with a 512 MiB +linear-memory limit per instance. Compilation is shared and lazy; the upstream +embedded package still decompresses its WASM at process initialization. Native +compilation memory, host buffers and JS allocations are additional to the +linear-memory limit. Playback support depends on the selected codecs, not just +the `.mp4` suffix. Platform execution and performance must be tested on target +devices, especially mobile/32-bit devices. diff --git a/pkg/download/engine/ffmpeg/input.go b/pkg/download/engine/ffmpeg/input.go new file mode 100644 index 00000000..74e9c453 --- /dev/null +++ b/pkg/download/engine/ffmpeg/input.go @@ -0,0 +1,227 @@ +// Package ffmpeg runs an embedded FFmpeg with host-controlled media inputs. +package ffmpeg + +import ( + "context" + "errors" + "fmt" + "io" + "net/http" + "net/url" + "strconv" + "strings" +) + +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) { + req, err := http.NewRequestWithContext(h.ctx, http.MethodGet, h.source.URL, nil) + if err != nil { + 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) + } + return h.client.Do(req) +} + +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 +} diff --git a/pkg/download/engine/ffmpeg/input_test.go b/pkg/download/engine/ffmpeg/input_test.go new file mode 100644 index 00000000..63e86c21 --- /dev/null +++ b/pkg/download/engine/ffmpeg/input_test.go @@ -0,0 +1,222 @@ +package ffmpeg + +import ( + "bytes" + "context" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" +) + +func TestHTTPRange(t *testing.T) { + data := bytes.Repeat([]byte("0123456789"), 300000) + var requests atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + count := requests.Add(1) + if r.Header.Get("X-Test") != "secret" { + t.Error("missing headers") + } + if r.Header.Get("Accept-Encoding") != "identity" { + t.Error("missing identity encoding") + } + if count > 1 && r.Header.Get("If-Range") != `"v1"` { + t.Error("missing If-Range") + } + w.Header().Set("ETag", `"v1"`) + http.ServeContent(w, r, "media", time.Time{}, bytes.NewReader(data)) + })) + defer srv.Close() + in, err := OpenHTTP(context.Background(), srv.Client(), HTTPSource{URL: srv.URL, Headers: map[string]string{"X-Test": "secret"}}) + if err != nil { + t.Fatal(err) + } + defer in.Close() + if !in.Seekable() || in.Size() != int64(len(data)) { + t.Fatal("range not detected") + } + for _, off := range []int64{0, int64(len(data) - 16), 12, int64(httpBlockSize + 35)} { + if _, err = in.Seek(off, io.SeekStart); err != nil { + t.Fatal(err) + } + b := make([]byte, 16) + if _, err = io.ReadFull(in, b); err != nil { + t.Fatal(err) + } + if !bytes.Equal(b, data[off:off+16]) { + t.Fatalf("wrong bytes at %d", off) + } + } +} + +func TestHTTPSequentialAndValidation(t *testing.T) { + for _, kind := range []string{"sequential", "bad-range", "changed", "truncated", "unauthorized"} { + t.Run(kind, func(t *testing.T) { + var requests atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + count := requests.Add(1) + switch kind { + case "sequential": + w.Write([]byte("media")) + case "unauthorized": + w.WriteHeader(403) + case "bad-range": + w.Header().Set("Content-Range", "bytes 2-6/10") + w.WriteHeader(206) + w.Write([]byte("media")) + case "truncated": + w.Header().Set("Content-Range", "bytes 0-9/10") + w.WriteHeader(206) + w.Write([]byte("media")) + case "changed": + if count == 1 { + w.Header().Set("Content-Range", "bytes 0-4/10") + w.Header().Set("ETag", `"v1"`) + w.WriteHeader(206) + } + w.Write([]byte("media")) + } + })) + defer srv.Close() + in, err := OpenHTTP(context.Background(), srv.Client(), HTTPSource{URL: srv.URL}) + if kind == "sequential" { + if err != nil { + t.Fatal(err) + } + defer in.Close() + if in.Seekable() { + t.Fatal("unexpected seeking") + } + if _, err = in.Seek(0, 0); err != ErrSeek { + t.Fatal(err) + } + b, err := io.ReadAll(in) + if err != nil || string(b) != "media" || requests.Load() != 1 { + t.Fatalf("%q %v count %d", b, err, requests.Load()) + } + } else if kind == "changed" { + if err != nil { + t.Fatal(err) + } + defer in.Close() + in.Seek(5, 0) + if _, err = in.Read(make([]byte, 2)); err == nil { + t.Fatal("accepted 200 after range") + } + } else if err == nil { + in.Close() + t.Fatal("accepted invalid response") + } + }) + } +} + +func TestStreamRing(t *testing.T) { + s := NewStreamInput(context.Background()) + defer s.Close() + // Start near the boundary to exercise wrap-around without a huge fixture. + s.buffer = make([]byte, 256*1024) + s.read = int64(len(s.buffer)) - 3 + s.written = s.read + if _, err := s.Write([]byte("abcdefgh")); err != nil { + t.Fatal(err) + } + s.End(nil) + b, err := io.ReadAll(s) + if err != nil || string(b) != "abcdefgh" { + t.Fatalf("%q %v", b, err) + } + if err = s.Close(); err != nil { + t.Fatal(err) + } + if s.buffer != nil { + t.Fatal("buffer retained after close") + } +} + +func TestOutputArguments(t *testing.T) { + a, err := Arguments("", []string{"-shortest", "-metadata", "title=test"}) + if err != nil { + t.Fatal(err) + } + if !strings.Contains(strings.Join(a, " "), "-f mp4 pipe:1") { + t.Fatal(a) + } + for _, args := range [][]string{{"-i", "/etc/passwd"}, {"-report"}, {"-c:v", "libx264"}, {"-metadata"}, {"other.mp4"}} { + if _, err := Arguments("mp4", args); err == nil { + t.Fatal(fmt.Sprint(args)) + } + } +} + +func TestStreamBackpressureAndCancel(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + s := NewStreamInput(ctx) + defer s.Close() + s.buffer = make([]byte, StreamBufferLimit) + s.written = StreamBufferLimit + finished := make(chan error, 1) + go func() { _, err := s.Write([]byte{42}); finished <- err }() + select { + case err := <-finished: + t.Fatalf("did not apply backpressure: %v", err) + case <-time.After(10 * time.Millisecond): + } + if _, err := s.Read(make([]byte, 1)); err != nil { + t.Fatal(err) + } + select { + case err := <-finished: + if err != nil { + t.Fatal(err) + } + case <-time.After(time.Second): + t.Fatal("writer did not resume") + } + go func() { _, err := s.Write([]byte{43}); finished <- err }() + cancel() + select { + case err := <-finished: + if err == nil { + t.Fatal("cancelled write succeeded") + } + case <-time.After(time.Second): + t.Fatal("writer did not cancel") + } +} + +func TestStreamMemoryGrowthAndWrap(t *testing.T) { + s := NewStreamInput(context.Background()) + defer s.Close() + if len(s.buffer) != 0 { + t.Fatal("buffer allocated before first write") + } + a := bytes.Repeat([]byte{1}, 200*1024) + b := bytes.Repeat([]byte{2}, 180*1024) + s.Write(a) + consumed := make([]byte, 180*1024) + if _, err := io.ReadFull(s, consumed); err != nil { + t.Fatal(err) + } + s.Write(b) // Wrap in the initial 256 KiB allocation. + c := bytes.Repeat([]byte{3}, 300*1024) + s.Write(c) // Grow while unread data wraps around the old ring. + s.End(nil) + got, err := io.ReadAll(s) + want := append(append(append([]byte{}, a[len(consumed):]...), b...), c...) + if err != nil || !bytes.Equal(got, want) { + t.Fatalf("growth corrupted data: %v", err) + } + if len(s.buffer) > StreamBufferLimit { + t.Fatal("buffer exceeded limit") + } + s.Close() + if s.buffer != nil { + t.Fatal("buffer retained on close") + } +} diff --git a/pkg/download/engine/ffmpeg/runtime.go b/pkg/download/engine/ffmpeg/runtime.go new file mode 100644 index 00000000..2d6b192f --- /dev/null +++ b/pkg/download/engine/ffmpeg/runtime.go @@ -0,0 +1,254 @@ +package ffmpeg + +import ( + "context" + "errors" + "fmt" + "io" + "io/fs" + "strings" + "sync" + + ffembed "codeberg.org/gruf/go-ffmpreg/embed" + "codeberg.org/gruf/go-ffmpreg/wasm" + "github.com/tetratelabs/wazero" + expsys "github.com/tetratelabs/wazero/experimental/sys" + "github.com/tetratelabs/wazero/experimental/sysfs" + "github.com/tetratelabs/wazero/sys" +) + +var shared struct { + once sync.Once + runtime wazero.Runtime + module wazero.CompiledModule + err error +} + +// Limit active instances independently of the number of extension engines. +var slots = make(chan struct{}, 2) + +func initialize() { + shared.once.Do(func() { + defer func() { + if r := recover(); r != nil { + shared.err = fmt.Errorf("ffmpeg initialization: %v", r) + } + }() + cfg := wazero.NewRuntimeConfig().WithCloseOnContextDone(true).WithMemoryLimitPages(8192) + shared.runtime, shared.err = wasm.NewRuntime(context.Background(), cfg) + if shared.err != nil { + return + } + shared.module, shared.err = shared.runtime.CompileModule(context.Background(), ffembed.B()) + if shared.err != nil { + shared.runtime.Close(context.Background()) + return + } + ffembed.Free() + }) +} + +// Output arguments deliberately do not accept extra inputs, output paths or +// global filesystem/logging options. The host owns all I/O and cancellation. +var outputOptions = map[string]int{ + "-c": 1, "-codec": 1, "-c:v": 1, "-c:a": 1, "-codec:v": 1, "-codec:a": 1, + "-map": 1, "-map_metadata": 1, "-map_chapters": 1, "-metadata": 1, "-metadata:s:a:0": 1, "-metadata:s:v:0": 1, + "-movflags": 1, "-frag_duration": 1, "-frag_size": 1, "-min_frag_duration": 1, + "-bsf:v": 1, "-bsf:a": 1, "-strict": 1, "-shortest": 0, "-t": 1, "-to": 1, + "-avoid_negative_ts": 1, "-max_interleave_delta": 1, "-flush_packets": 1, + "-disposition:a:0": 1, "-disposition:v:0": 1, +} + +func Arguments(format string, extra []string) ([]string, error) { + if format == "" { + format = "mp4" + } + if format == "mkv" { + format = "matroska" + } + switch format { + case "mp4", "matroska", "webm", "mpegts": + default: + return nil, fmt.Errorf("ffmpeg: unsupported output format %q", format) + } + for i := 0; i < len(extra); { + count, ok := outputOptions[extra[i]] + if !ok { + return nil, fmt.Errorf("ffmpeg: unsupported output option %q", extra[i]) + } + if i+count >= len(extra) { + return nil, fmt.Errorf("ffmpeg: missing value for %s", extra[i]) + } + if strings.HasPrefix(extra[i], "-c") && extra[i+1] != "copy" { + return nil, errors.New("ffmpeg: merge only supports stream copy") + } + i += count + 1 + } + args := []string{"-hide_banner", "-nostdin", "-xerror", "-loglevel", "error", "-protocol_whitelist", "file,pipe", "-i", "/inputs/video", "-protocol_whitelist", "file,pipe", "-i", "/inputs/audio", "-map", "0:v:0", "-map", "1:a:0", "-c", "copy"} + fragmentFlags := "" + for i := 0; i < len(extra); { + count := outputOptions[extra[i]] + if format == "mp4" && extra[i] == "-movflags" { + if strings.Contains(extra[i+1], "faststart") || strings.Contains(extra[i+1], "global_sidx") { + return nil, errors.New("ffmpeg: these movflags require seekable output") + } + fragmentFlags = extra[i+1] + } else { + args = append(args, extra[i:i+count+1]...) + } + i += count + 1 + } + if format == "mp4" { + args = append(args, "-movflags", fragmentFlags+"+frag_keyframe+empty_moov+default_base_moof") + } + args = append(args, "-f", format, "pipe:1") + return args, nil +} + +// Run shares compilation, but gives every merge its own memory and filesystem. +// A clean stdout EOF is only delivered by the caller after this succeeds. +func Run(ctx context.Context, video, audio Input, out io.Writer, format string, extra []string) error { + _, err := Arguments(format, extra) + if err != nil { + return err + } + release, err := Acquire(ctx) + if err != nil { + return err + } + defer release() + return RunAcquired(ctx, video, audio, out, format, extra) +} + +// Acquire reserves execution capacity before input factories or HTTP requests start. +func Acquire(ctx context.Context) (func(), error) { + select { + case slots <- struct{}{}: + case <-ctx.Done(): + return nil, ctx.Err() + } + var once sync.Once + return func() { once.Do(func() { <-slots }) }, nil +} + +// RunAcquired requires a reservation held until this call returns. +func RunAcquired(ctx context.Context, video, audio Input, out io.Writer, format string, extra []string) error { + args, err := Arguments(format, extra) + if err != nil { + return err + } + ready := make(chan struct{}) + go func() { initialize(); close(ready) }() + select { + case <-ready: + case <-ctx.Done(): + return ctx.Err() + } + if shared.err != nil { + return shared.err + } + files := &inputFS{files: map[string]Input{"video": video, "audio": audio}} + fsconfig := wazero.NewFSConfig().(sysfs.FSConfig).WithSysFSMount(files, "/inputs") + logs := &tailLog{} + rc, err := wasm.Run(ctx, shared.runtime, shared.module, wasm.Args{ + Name: "ffmpeg", Args: args, Stdout: out, Stderr: logs, + Config: func(c wazero.ModuleConfig) wazero.ModuleConfig { return c.WithFSConfig(fsconfig) }, + }) + if ctx.Err() != nil { + return ctx.Err() + } + if files.err != nil { + return fmt.Errorf("ffmpeg input: %w (%s)", files.err, logs.String()) + } + if err != nil { + return fmt.Errorf("ffmpeg runtime: %w (%s)", err, logs.String()) + } + if rc != 0 { + return fmt.Errorf("ffmpeg exited with code %d: %s", rc, logs.String()) + } + return nil +} + +type tailLog struct{ data []byte } + +func (l *tailLog) Write(p []byte) (int, error) { + n := len(p) + if len(p) > 8192 { + p = p[len(p)-8192:] + } + l.data = append(l.data, p...) + if len(l.data) > 8192 { + l.data = l.data[len(l.data)-8192:] + } + return n, nil +} +func (l *tailLog) String() string { return strings.TrimSpace(string(l.data)) } + +type inputFS struct { + expsys.UnimplementedFS + files map[string]Input + err error +} + +func (f *inputFS) OpenFile(path string, flag expsys.Oflag, perm fs.FileMode) (expsys.File, expsys.Errno) { + if flag&(expsys.O_WRONLY|expsys.O_RDWR|expsys.O_CREAT|expsys.O_TRUNC) != 0 { + return nil, expsys.EROFS + } + path = strings.TrimPrefix(path, "/") + if path == "." || path == "" { + return &inputFile{owner: f}, 0 + } + in := f.files[path] + if in == nil { + return nil, expsys.ENOENT + } + return &inputFile{owner: f, in: in}, 0 +} +func (f *inputFS) Stat(path string) (sys.Stat_t, expsys.Errno) { + file, err := f.OpenFile(path, 0, 0) + if err != 0 { + return sys.Stat_t{}, err + } + return file.Stat() +} + +type inputFile struct { + expsys.UnimplementedFile + owner *inputFS + in Input +} + +func (f *inputFile) IsDir() (bool, expsys.Errno) { return f.in == nil, 0 } +func (f *inputFile) Stat() (sys.Stat_t, expsys.Errno) { + if f.in == nil { + return sys.Stat_t{Mode: fs.ModeDir | 0555}, 0 + } + mode := fs.FileMode(0444) + if !f.in.Seekable() { + mode |= fs.ModeNamedPipe + } + return sys.Stat_t{Mode: mode, Size: max(0, f.in.Size()), Nlink: 1}, 0 +} +func (f *inputFile) Read(p []byte) (int, expsys.Errno) { + if f.in == nil { + return 0, expsys.EISDIR + } + n, err := f.in.Read(p) + if err != nil && err != io.EOF { + f.owner.err = err + return n, expsys.EIO + } + return n, 0 +} +func (f *inputFile) Seek(off int64, whence int) (int64, expsys.Errno) { + if f.in == nil { + return 0, expsys.EISDIR + } + n, err := f.in.Seek(off, whence) + // Capability probes may legitimately fail; a failed seek alone is not a + // media error. FFmpeg decides whether it can continue sequentially. + if err != nil { + return 0, expsys.ENOSYS + } + return n, 0 +} diff --git a/pkg/download/engine/ffmpeg/runtime_test.go b/pkg/download/engine/ffmpeg/runtime_test.go new file mode 100644 index 00000000..ee57c180 --- /dev/null +++ b/pkg/download/engine/ffmpeg/runtime_test.go @@ -0,0 +1,215 @@ +package ffmpeg + +import ( + "bytes" + "context" + "encoding/binary" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "testing" + "time" + + "codeberg.org/gruf/go-ffmpreg/wasm" + "github.com/tetratelabs/wazero" +) + +// Generate tiny deterministic media with the exact embedded build under test; +// no system FFmpeg, network media or third-party fixture licensing is needed. +func fixture(t *testing.T, audio bool) []byte { + t.Helper() + initialize() + if shared.err != nil { + t.Fatal(shared.err) + } + args := []string{"-hide_banner", "-loglevel", "error", "-f", "lavfi", "-i", "color=c=black:s=32x32:r=10", "-t", "0.3", "-c:v", "libx264", "-preset", "ultrafast", "-movflags", "frag_keyframe+empty_moov", "-f", "mp4", "pipe:1"} + if audio { + args = []string{"-hide_banner", "-loglevel", "error", "-f", "lavfi", "-i", "anullsrc=r=48000:cl=mono", "-t", "0.3", "-c:a", "aac", "-movflags", "frag_keyframe+empty_moov", "-f", "mp4", "pipe:1"} + } + var out, logs bytes.Buffer + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + rc, err := wasm.Run(ctx, shared.runtime, shared.module, wasm.Args{Name: "ffmpeg", Args: args, Stdout: &out, Stderr: &logs}) + if err != nil || rc != 0 { + t.Fatalf("fixture: %d %v %s", rc, err, logs.String()) + } + return out.Bytes() +} + +func TestTailMoovNeedsRange(t *testing.T) { + initialize() + if shared.err != nil { + t.Fatal(shared.err) + } + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + dir := t.TempDir() + var logs bytes.Buffer + rc, err := wasm.Run(ctx, shared.runtime, shared.module, wasm.Args{Name: "ffmpeg", Args: []string{ + "-v", "error", "-f", "lavfi", "-i", "color=c=black:s=32x32:r=10", "-t", "0.3", "-c:v", "libx264", "-preset", "ultrafast", "/out/video.mp4", + }, Stderr: &logs, Config: func(c wazero.ModuleConfig) wazero.ModuleConfig { + return c.WithFSConfig(wazero.NewFSConfig().WithDirMount(dir, "/out")) + }}) + if rc != 0 || err != nil { + t.Fatalf("fixture %d %v %s", rc, err, logs.String()) + } + raw, err := os.ReadFile(filepath.Join(dir, "video.mp4")) + if err != nil { + t.Fatal(err) + } + var video []byte + for offset := 0; offset < len(raw); { + length := int(binary.BigEndian.Uint32(raw[offset : offset+4])) + if string(raw[offset+4:offset+8]) == "moov" { + free := make([]byte, 2*httpBlockSize) + binary.BigEndian.PutUint32(free, uint32(len(free))) + copy(free[4:], "free") + video = append(video, free...) + } + video = append(video, raw[offset:offset+length]...) + offset += length + } + audio := fixture(t, true) + for _, ranged := range []bool{true, false} { + t.Run(fmtBool(ranged), func(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + data := video + if r.URL.Path == "/audio" { + data = audio + } + if ranged { + http.ServeContent(w, r, "media", time.Time{}, bytes.NewReader(data)) + } else { + w.Write(data) + } + })) + defer srv.Close() + v, err := OpenHTTP(ctx, srv.Client(), HTTPSource{URL: srv.URL}) + if err != nil { + t.Fatal(err) + } + defer v.Close() + a, err := OpenHTTP(ctx, srv.Client(), HTTPSource{URL: srv.URL + "/audio"}) + if err != nil { + t.Fatal(err) + } + defer a.Close() + err = Run(ctx, v, a, io.Discard, "mp4", nil) + if ranged && err != nil { + t.Fatal(err) + } + if !ranged && err == nil { + t.Fatal("non-seekable tail-moov input unexpectedly succeeded") + } + }) + } +} + +func fmtBool(b bool) string { + if b { + return "range" + } + return "sequential" +} + +func TestWebMToMP4(t *testing.T) { + initialize() + if shared.err != nil { + t.Fatal(shared.err) + } + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + inputs := make([]*StreamInput, 2) + for i, name := range []string{"video.webm", "audio.webm"} { + data, err := os.ReadFile(filepath.Join("..", "testdata", "ffmpeg", name)) + if err != nil { + t.Fatal(err) + } + inputs[i] = NewStreamInput(ctx) + defer inputs[i].Close() + if _, err = inputs[i].Write(data); err != nil { + t.Fatal(err) + } + inputs[i].End(nil) + } + + var output, probe, logs bytes.Buffer + if err := Run(ctx, inputs[0], inputs[1], &output, "mp4", nil); err != nil { + t.Fatal(err) + } + rc, err := wasm.Run(ctx, shared.runtime, shared.module, wasm.Args{Name: "ffprobe", Args: []string{"-v", "error", "-show_entries", "stream=codec_name", "-of", "json", "pipe:0"}, Stdin: bytes.NewReader(output.Bytes()), Stdout: &probe, Stderr: &logs}) + if rc != 0 || err != nil { + t.Fatalf("probe %d %v %s", rc, err, logs.String()) + } + if !bytes.Contains(probe.Bytes(), []byte(`"vp9"`)) || !bytes.Contains(probe.Bytes(), []byte(`"opus"`)) { + t.Fatalf("codecs changed: %s", probe.String()) + } +} + +func TestWASMMerge(t *testing.T) { + video, audio := fixture(t, false), fixture(t, true) + for _, ranged := range []bool{false, true} { + t.Run(map[bool]string{false: "sequential", true: "range"}[ranged], func(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + data := video + if r.URL.Path == "/audio" { + data = audio + } + if ranged { + http.ServeContent(w, r, "media", time.Time{}, bytes.NewReader(data)) + } else { + w.Write(data) + } + })) + defer srv.Close() + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + v, err := OpenHTTP(ctx, srv.Client(), HTTPSource{URL: srv.URL + "/video"}) + if err != nil { + t.Fatal(err) + } + defer v.Close() + a, err := OpenHTTP(ctx, srv.Client(), HTTPSource{URL: srv.URL + "/audio"}) + if err != nil { + t.Fatal(err) + } + defer a.Close() + var output bytes.Buffer + if err = Run(ctx, v, a, &output, "", nil); err != nil { + t.Fatal(err) + } + if !bytes.Contains(output.Bytes(), []byte("moof")) { + t.Fatal("output is not fragmented MP4") + } + // Probe the actual result, rather than just checking its filename/header. + var probe, logs bytes.Buffer + rc, err := wasm.Run(ctx, shared.runtime, shared.module, wasm.Args{Name: "ffprobe", Args: []string{"-v", "error", "-show_streams", "-of", "json", "pipe:0"}, Stdin: bytes.NewReader(output.Bytes()), Stdout: &probe, Stderr: &logs}) + if rc != 0 || err != nil { + t.Fatalf("probe: %d %v %s", rc, err, logs.String()) + } + var result struct { + Streams []struct { + CodecType string `json:"codec_type"` + CodecName string `json:"codec_name"` + } + } + if err = json.Unmarshal(probe.Bytes(), &result); err != nil { + t.Fatal(err) + } + if len(result.Streams) != 2 || result.Streams[0].CodecName != "h264" || result.Streams[1].CodecName != "aac" { + t.Fatalf("unexpected streams %s", probe.String()) + } + }) + } + // A malformed input must produce an error, not successful empty output. + s := NewStreamInput(context.Background()) + defer s.Close() + s.Write([]byte("invalid")) + s.End(nil) + if err := Run(context.Background(), s, s, io.Discard, "", nil); err == nil { + t.Fatal("accepted invalid media") + } +} diff --git a/pkg/download/engine/ffmpeg/stream.go b/pkg/download/engine/ffmpeg/stream.go new file mode 100644 index 00000000..7db44f4f --- /dev/null +++ b/pkg/download/engine/ffmpeg/stream.go @@ -0,0 +1,163 @@ +package ffmpeg + +import ( + "context" + "errors" + "io" + "sync" +) + +// A bounded memory ring lets both JS inputs drain independently, including when +// they share a SABR producer. It does not make a sequential input seekable. +// A full ring waits for consumption or cancellation. Producers must propagate +// backpressure and allow each track to make progress independently. +const StreamBufferLimit = 32 * 1024 * 1024 + +type StreamInput struct { + ctx context.Context + mu sync.Mutex + buffer []byte + read, written int64 + ended, closed bool + err error + wake chan struct{} + space chan struct{} +} + +func NewStreamInput(ctx context.Context) *StreamInput { + return &StreamInput{ctx: ctx, wake: make(chan struct{}, 1), space: make(chan struct{}, 1)} +} + +func (s *StreamInput) Write(p []byte) (int, error) { + s.mu.Lock() + defer s.mu.Unlock() + if len(p) > StreamBufferLimit { + return 0, errors.New("ffmpeg: input chunk exceeds buffer capacity") + } + for int64(len(p)) > StreamBufferLimit-(s.written-s.read) && !s.closed && !s.ended { + s.mu.Unlock() + var err error + select { + case <-s.space: + case <-s.ctx.Done(): + err = s.ctx.Err() + } + s.mu.Lock() + if err != nil { + return 0, err + } + } + if s.closed || s.ended { + return 0, io.ErrClosedPipe + } + if err := s.ctx.Err(); err != nil { + return 0, err + } + if len(p) == 0 { + return 0, nil + } + // Grow only as needed. Rebase unread bytes when the ring grows. + need := int(s.written-s.read) + len(p) + if need > len(s.buffer) { + capacity := min(StreamBufferLimit, max(256*1024, max(need, len(s.buffer)*2))) + buffer := make([]byte, capacity) + unread := int(s.written - s.read) + if unread > 0 { + off := int(s.read % int64(len(s.buffer))) + first := min(unread, len(s.buffer)-off) + copy(buffer, s.buffer[off:off+first]) + copy(buffer[first:], s.buffer[:unread-first]) + } + s.buffer = buffer + s.read, s.written = 0, int64(unread) + } + off := int(s.written % int64(len(s.buffer))) + first := min(len(p), len(s.buffer)-off) + copy(s.buffer[off:], p[:first]) + copy(s.buffer, p[first:]) + s.written += int64(len(p)) + select { + case s.wake <- struct{}{}: + default: + } + return len(p), nil +} + +func (s *StreamInput) End(err error) { + s.mu.Lock() + s.ended = true + if s.err == nil { + s.err = err + } + s.mu.Unlock() + select { + case s.space <- struct{}{}: + default: + } + select { + case s.wake <- struct{}{}: + default: + } +} + +func (s *StreamInput) Read(p []byte) (int, error) { + if len(p) == 0 { + return 0, nil + } + for { + if err := s.ctx.Err(); err != nil { + return 0, err + } + s.mu.Lock() + if s.closed { + s.mu.Unlock() + return 0, io.ErrClosedPipe + } + if s.err != nil { + err := s.err + s.mu.Unlock() + return 0, err + } + if s.read < s.written { + off := s.read % int64(len(s.buffer)) + n := min(int64(len(p)), s.written-s.read, int64(len(s.buffer))-off) + count := copy(p, s.buffer[off:off+n]) + s.read += int64(count) + s.mu.Unlock() + select { + case s.space <- struct{}{}: + default: + } + return count, nil + } + ended := s.ended + s.mu.Unlock() + if ended { + return 0, io.EOF + } + select { + case <-s.wake: + case <-s.ctx.Done(): + return 0, s.ctx.Err() + } + } +} + +func (s *StreamInput) Close() error { + s.mu.Lock() + defer s.mu.Unlock() + s.closed = true + select { + case s.space <- struct{}{}: + default: + } + select { + case s.wake <- struct{}{}: + default: + } + s.buffer = nil + return nil +} +func (*StreamInput) Seek(int64, int) (int64, error) { return 0, ErrSeek } +func (*StreamInput) Size() int64 { return -1 } +func (*StreamInput) Seekable() bool { return false } diff --git a/pkg/download/engine/ffmpeg_integration_test.go b/pkg/download/engine/ffmpeg_integration_test.go new file mode 100644 index 00000000..29c988ce --- /dev/null +++ b/pkg/download/engine/ffmpeg_integration_test.go @@ -0,0 +1,321 @@ +package engine + +import ( + "bytes" + "context" + "fmt" + media "github.com/GopeedLab/gopeed/pkg/download/engine/ffmpeg" + "github.com/GopeedLab/gopeed/pkg/download/engine/inject/stream" + "io" + "net/http" + "net/http/httptest" + "os" + "strings" + "sync/atomic" + "testing" + "time" +) + +func TestFFmpegJSStreams(t *testing.T) { + video, err := os.ReadFile("testdata/ffmpeg/video.mp4") + if err != nil { + t.Fatal(err) + } + audio, err := os.ReadFile("testdata/ffmpeg/audio.mp4") + if err != nil { + t.Fatal(err) + } + for _, mode := range []string{"streams", "http", "mixed"} { + t.Run(mode, func(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + b := video + if r.URL.Path == "/audio" { + b = audio + } + http.ServeContent(w, r, "media", time.Time{}, bytes.NewReader(b)) + })) + defer srv.Close() + e := NewEngine(nil) + defer e.Close() + // Input data is deliberately interleaved through one JS producer. + // Native push/read operations must never block the event loop. + e.Runtime.Set("videoBytes", e.Runtime.NewArrayBuffer(video)) + e.Runtime.Set("audioBytes", e.Runtime.NewArrayBuffer(audio)) + e.Runtime.Set("sourceURL", srv.URL) + e.Runtime.Set("inputMode", mode) + result, err := e.RunString(`(async()=>{ + let vc, ac, offset=0; + const v=new Uint8Array(videoBytes), a=new Uint8Array(audioBytes); + const video=new ReadableStream({start(c){vc=c;}}); + const audio=new ReadableStream({start(c){ac=c;}}); + const timer=setInterval(()=>{ + if(offset=Math.max(v.length,a.length)){clearInterval(timer);vc.close();ac.close();} + },1); + const stream=__gopeed_ffmpeg.merge({ + inputs: () => ({ video:inputMode==='http'?{url:sourceURL+'/video'}:video, + audio:inputMode==='streams'?audio:{url:sourceURL+'/audio'} }), + args:['-metadata','title=gopeed-test'] + }); + const bytes=await new Response(stream).arrayBuffer(); + return new Uint8Array(bytes); + })()`) + if err != nil { + t.Fatal(err) + } + out, ok := result.([]byte) + if !ok { + t.Fatalf("output %T", result) + } + if !bytes.Contains(out, []byte("moof")) || !bytes.Contains(out, []byte("gopeed-test")) { + t.Fatalf("invalid output (%d bytes)", len(out)) + } + }) + } +} + +func TestFFmpegJSErrorAndCancel(t *testing.T) { + for _, scenario := range []string{"input-error", "abort", "output-cancel", "invalid-args"} { + t.Run(scenario, func(t *testing.T) { + e := NewEngine(nil) + defer e.Close() + e.Runtime.Set("scenario", scenario) + result, err := e.RunString(`(async()=>{ + let cancelled=0; + const source=()=>new ReadableStream({ + pull(c){if(scenario==='input-error')c.error(new Error('upstream failed'));else return new Promise(()=>{});}, + cancel(){cancelled++;} + }); + const controller=new AbortController(); + const output=__gopeed_ffmpeg.merge({inputs:()=>({video:source(),audio:source()}),signal:controller.signal, + args:scenario==='invalid-args'?['-i','/outside']:[]}); + const reader=output.getReader(); + const pending=reader.read().then(()=>'',e=>e.message); + if(scenario==='abort')setTimeout(()=>controller.abort(),20); + if(scenario==='output-cancel')setTimeout(()=>reader.cancel(),20); + const message=await pending; + await new Promise(r=>setTimeout(r,10)); + return {message,cancelled}; + })()`) + if err != nil { + t.Fatal(err) + } + m := result.(map[string]any) + if scenario != "output-cancel" && m["message"] == "" { + t.Fatalf("missing error: %v", m) + } + if scenario == "abort" || scenario == "output-cancel" { + if fmt.Sprint(m["cancelled"]) != "2" { + t.Fatalf("inputs not cancelled: %v", m) + } + } + }) + } +} + +func TestFFmpegMalformedMediaFails(t *testing.T) { + e := NewEngine(nil) + defer e.Close() + _, err := e.RunString(`(async()=>{ + const source=()=>new ReadableStream({start(c){c.enqueue(new Uint8Array([1,2,3]));c.close();}}); + return await new Response(__gopeed_ffmpeg.merge({inputs:()=>({video:source(),audio:source()})})).arrayBuffer(); + })()`) + if err == nil || !strings.Contains(err.Error(), "ffmpeg") { + t.Fatalf("expected FFmpeg error: %v", err) + } +} + +func TestFFmpegBlobOpener(t *testing.T) { + var open stream.ObjectURLOpener + e := NewEngine(&Config{StreamConfig: &stream.Config{CreateObjectURL: func(opts *stream.ObjectURLOptions, opener stream.ObjectURLOpener) (string, error) { + if opts.Range || opts.Size != 0 { + t.Fatal("merged output must not advertise range or a guessed size") + } + open = opener + return "blob:ffmpeg-test", nil + }}}) + defer e.Close() + video, _ := os.ReadFile("testdata/ffmpeg/video.mp4") + audio, _ := os.ReadFile("testdata/ffmpeg/audio.mp4") + e.Runtime.Set("videoBytes", e.Runtime.NewArrayBuffer(video)) + e.Runtime.Set("audioBytes", e.Runtime.NewArrayBuffer(audio)) + _, err := e.RunString(`__gopeed_blob_create_object_url(()=>{ + const source=b=>new ReadableStream({start(c){c.enqueue(new Uint8Array(b));c.close();}}); + return __gopeed_ffmpeg.merge({inputs:()=>({video:source(videoBytes),audio:source(audioBytes)})}); + },{contentType:'video/mp4'})`) + if err != nil { + t.Fatal(err) + } + for i := 0; i < 2; i++ { + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + r, err := open(ctx, stream.ObjectURLOpenRequest{End: -1}) + if err != nil { + cancel() + t.Fatal(err) + } + b, err := io.ReadAll(r) + r.Close() + cancel() + if err != nil || !bytes.Contains(b, []byte("moof")) { + t.Fatalf("Blob read: %v (%d bytes)", err, len(b)) + } + } +} + +func TestFFmpegEngineCloseCancelsRequest(t *testing.T) { + started, closed := make(chan struct{}), make(chan struct{}) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + close(started) + <-r.Context().Done() + close(closed) + })) + defer srv.Close() + e := NewEngine(nil) + defer e.Close() + _, err := e.RunString(fmt.Sprintf(` + globalThis.cancelCount=0; + const audio=new ReadableStream({cancel(){cancelCount++;}}); + __gopeed_ffmpeg.merge({inputs:()=>({video:{url:%q},audio})}).getReader().read().catch(()=>{}); + "started";`, srv.URL)) + if err != nil { + t.Fatal(err) + } + select { + case <-started: + case <-time.After(5 * time.Second): + t.Fatal("HTTP did not start") + } + e.Close() + select { + case <-closed: + case <-time.After(5 * time.Second): + t.Fatal("HTTP leaked after engine shutdown") + } + if e.Runtime.Get("cancelCount").ToInteger() != 1 { + t.Fatal("JS input not cancelled on shutdown") + } +} + +func TestFFmpegInputFactoriesWaitForCapacity(t *testing.T) { + for _, mode := range []string{"factory", "http", "cancel"} { + t.Run(mode, func(t *testing.T) { + release1, _ := media.Acquire(context.Background()) + defer release1() + release2, _ := media.Acquire(context.Background()) + defer release2() + var requests atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { requests.Add(1); http.Error(w, "test", 500) })) + defer srv.Close() + e := NewEngine(nil) + defer e.Close() + e.Runtime.Set("mode", mode) + e.Runtime.Set("sourceURL", srv.URL) + result, err := e.RunString(`(async()=>{ + globalThis.factoryCalls=0; + globalThis.abortQueue=new AbortController(); + const options=mode==='http'?{video:{url:sourceURL},audio:{url:sourceURL}}: + {inputs:()=>{factoryCalls++;throw new Error('factory failed');}}; + options.signal=abortQueue.signal; + globalThis.work=__gopeed_ffmpeg.merge(options).getReader().read().then(()=>'',e=>e.message); + await new Promise(r=>setTimeout(r,50)); + if(mode==='cancel')abortQueue.abort(); + return factoryCalls; + })()`) + if err != nil { + t.Fatal(err) + } + if fmt.Sprint(result) != "0" || requests.Load() != 0 { + t.Fatalf("queued inputs started: %v, requests=%d", result, requests.Load()) + } + if mode == "cancel" { + if _, err = e.RunString(`work`); err != nil { + t.Fatal(err) + } + } + release1() + if _, err = e.RunString(`work`); err != nil { + t.Fatal(err) + } + calls, err := e.RunString(`factoryCalls`) + if err != nil { + t.Fatal(err) + } + if mode == "factory" && fmt.Sprint(calls) != "1" { + t.Fatal(calls) + } + if mode == "cancel" && fmt.Sprint(calls) != "0" { + t.Fatal("cancelled factory ran") + } + if mode == "http" && requests.Load() != 1 { + t.Fatal("HTTP did not start after admission") + } + // Failure/cancellation must return the acquired slot even with the other held. + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + release, err := media.Acquire(ctx) + if err != nil { + t.Fatalf("slot leaked: %v", err) + } + release() + }) + } +} + +func TestFFmpegRequiresFactoryForStreams(t *testing.T) { + e := NewEngine(nil) + defer e.Close() + result, err := e.RunString(`(()=>{ + const source=()=>new ReadableStream({}); + let rejected=0; + for(const opts of [ + {video:source(),audio:source()}, + {video:{url:'https://example.test'},audio:source()}, + {inputs:()=>({}),video:{url:'https://example.test'}}, + {inputs:{video:source(),audio:source()}} + ]) {try{__gopeed_ffmpeg.merge(opts);}catch(e){if(e instanceof TypeError)rejected++;}} + return rejected; + })()`) + if err != nil || fmt.Sprint(result) != "4" { + t.Fatalf("validation: %v %v", result, err) + } +} + +func TestFFmpegCancelPendingFactoryCleansLateInputs(t *testing.T) { + e := NewEngine(nil) + defer e.Close() + result, err := e.RunString(`(async()=>{ + let ready,finish,cancelled=0,signal; + const entered=new Promise(r=>ready=r); + const abort=new AbortController(); + const output=__gopeed_ffmpeg.merge({signal:abort.signal,inputs:async opts=>{ + signal=opts.signal;ready();await new Promise(r=>finish=r); + const source=()=>new ReadableStream({cancel(){cancelled++;}}); + return {video:source(),audio:source()}; + }}); + const work=output.getReader().read().catch(e=>e.name); + await entered;abort.abort();await work;finish(); + await new Promise(r=>setTimeout(r,30)); + return {cancelled,aborted:signal.aborted}; + })()`) + if err != nil { + t.Fatal(err) + } + state := result.(map[string]any) + if fmt.Sprint(state["cancelled"]) != "2" || state["aborted"] != true { + t.Fatal(state) + } + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + first, err := media.Acquire(ctx) + if err != nil { + t.Fatal(err) + } + defer first() + second, err := media.Acquire(ctx) + if err != nil { + t.Fatal("factory cancellation leaked capacity") + } + second() +} diff --git a/pkg/download/engine/inject/fetch/module.go b/pkg/download/engine/inject/fetch/module.go index f882fd7c..49ccf8cb 100644 --- a/pkg/download/engine/inject/fetch/module.go +++ b/pkg/download/engine/inject/fetch/module.go @@ -29,8 +29,9 @@ const FingerprintMagicKey = "__gopeed_fetch_fingerprint" var script string type Config struct { - ProxyHandler func(*http.Request) (*url.URL, error) - RegisterCleanup func(func()) + DefaultUserAgent *string + ProxyHandler func(*http.Request) (*url.URL, error) + RegisterCleanup func(func()) } func Enable(runtime *goja.Runtime, loop *eventloop.EventLoop, cfg *Config) error { @@ -51,6 +52,9 @@ func Enable(runtime *goja.Runtime, loop *eventloop.EventLoop, cfg *Config) error if err != nil { panic(runtime.NewGoError(err)) } + if cfg.DefaultUserAgent != nil && !hasHeader(request.Headers, "User-Agent") { + request.Headers = append(request.Headers, [2]string{"User-Agent", *cfg.DefaultUserAgent}) + } fingerprint := util.SafeGet[string](runtime, FingerprintMagicKey) promise, resolve, reject := runtime.NewPromise() go func() { diff --git a/pkg/download/engine/inject/ffmpeg/ffmpeg.js b/pkg/download/engine/inject/ffmpeg/ffmpeg.js new file mode 100644 index 00000000..49b42f25 --- /dev/null +++ b/pkg/download/engine/inject/ffmpeg/ffmpeg.js @@ -0,0 +1,139 @@ +(function () { + const create = __gopeed_ffmpeg_create; + const read = __gopeed_ffmpeg_read; + const acquire = __gopeed_ffmpeg_acquire; + const start = __gopeed_ffmpeg_start; + const push = __gopeed_ffmpeg_push; + const end = __gopeed_ffmpeg_end; + const cancel = __gopeed_ffmpeg_cancel; + const active = new Set(); + globalThis.__gopeed_ffmpeg_close_all = () => { + for (const stop of Array.from(active)) stop(new Error("extension engine closed")); + }; + + function input(value, name, allowStream) { + if (value && typeof value.getReader === "function") { + if (!allowStream) throw new TypeError("stream inputs must be created by the inputs callback"); + if (value.locked) throw new TypeError(name + " stream is locked"); + return { stream: value, request: null }; + } + if (!value || typeof value.url !== "string" || !/^https?:\/\//i.test(value.url)) { + throw new TypeError(name + " must be an HTTP request or ReadableStream"); + } + const headers = {}; + new Headers(value.headers || {}).forEach((v, k) => { headers[k] = v; }); + return { request: { url: value.url, headers } }; + } + + function merge(options) { + if (!options || typeof options !== "object") throw new TypeError("merge options are required"); + const factory = options.inputs; + if (factory !== undefined && typeof factory !== "function") throw new TypeError("inputs must be a callback"); + if (factory && (options.video !== undefined || options.audio !== undefined)) throw new TypeError("inputs cannot be combined with video/audio"); + let video = factory ? null : input(options.video, "video", false); + let audio = factory ? null : input(options.audio, "audio", false); + const args = options.args === undefined ? [] : options.args; + if (!Array.isArray(args) || args.some(v => typeof v !== "string")) throw new TypeError("args must be a string array"); + const format = options.format === undefined ? "mp4" : options.format; + if (typeof format !== "string") throw new TypeError("format must be a string"); + const signal = options.signal; + let id, controller, stopped = false; + const readers = []; + const lifetime = new AbortController(); + let ownedInputs; + function discardInputs(inputs, reason) { + for (const source of new Set([inputs && inputs.video, inputs && inputs.audio])) { + if (source && typeof source.cancel === "function" && !source.locked) Promise.resolve(source.cancel(reason)).catch(() => {}); + } + } + + function stop(reason) { + if (stopped) return; + stopped = true; + active.delete(shutdown); + lifetime.abort(); + discardInputs(ownedInputs, reason); + if (id) cancel(id); + if (signal) signal.removeEventListener("abort", abort); + for (const reader of readers) { + Promise.resolve(reader.cancel(reason)).catch(() => {}).then(() => { + try { reader.releaseLock(); } catch (_) {} + }); + } + } + function shutdown(reason) { + controller.error(reason); + stop(reason); + } + function abort() { + const error = new Error("FFmpeg operation aborted"); + error.name = "AbortError"; + stop(error); + controller.error(error); + } + async function pump(reader, index) { + try { + while (!stopped) { + const result = await reader.read(); + if (stopped) return; + if (result.done) { end(id, index, ""); return; } + if (!(result.value instanceof Uint8Array)) throw new TypeError("media streams must contain Uint8Array chunks"); + // Keep individual crossings bounded, even if upstream emits a large chunk. + for (let offset = 0; offset < result.value.byteLength && !stopped; offset += 256 * 1024) { + if (!await push(id, index, result.value.subarray(offset, offset + 256 * 1024))) return; + } + } + } catch (error) { + if (!stopped) { + end(id, index, String(error && error.message || error)); + controller.error(error); + stop(error); + } + } + } + return new ReadableStream({ + start(c) { + controller = c; + active.add(shutdown); + if (signal) { + if (signal.aborted) abort(); + else signal.addEventListener("abort", abort); + } + }, + async pull(c) { + if (stopped) return; + try { + if (!id) { + id = create({ format, args }); + await acquire(id); + if (stopped) return; + if (factory) { + const inputs = await factory({ signal: lifetime.signal }); + if (stopped) { discardInputs(inputs, new Error("FFmpeg operation cancelled")); return; } + ownedInputs = inputs; + video = input(inputs && inputs.video, "video", true); + audio = input(inputs && inputs.audio, "audio", true); + } + if (video.stream && video.stream === audio.stream) throw new TypeError("video and audio must be distinct streams"); + for (const source of [video, audio]) { + if (source.stream) readers.push(source.stream.getReader()); + } + start(id, { video: video.request, audio: audio.request, format, args }); + let readerIndex = 0; + for (const [index, source] of [video, audio].entries()) { + if (source.stream) void pump(readers[readerIndex++], index); + } + } + const chunk = await read(id); + if (stopped) return; + if (chunk === null) { stop(); c.close(); } + else c.enqueue(new Uint8Array(chunk)); + } catch (error) { + if (!stopped) { c.error(error); stop(error); } + } + }, + cancel(reason) { stop(reason); } + }); + } + globalThis.__gopeed_ffmpeg = Object.freeze({ merge, supportsInputFactory: true }); +})(); diff --git a/pkg/download/engine/inject/ffmpeg/module.go b/pkg/download/engine/inject/ffmpeg/module.go new file mode 100644 index 00000000..cc73e990 --- /dev/null +++ b/pkg/download/engine/inject/ffmpeg/module.go @@ -0,0 +1,324 @@ +package ffmpeg + +import ( + "context" + _ "embed" + "errors" + "fmt" + "io" + "net/http" + "net/url" + "strings" + "sync" + "sync/atomic" + + "github.com/GopeedLab/gopeed/internal/httpclient" + media "github.com/GopeedLab/gopeed/pkg/download/engine/ffmpeg" + "github.com/dop251/goja" + "github.com/dop251/goja_nodejs/eventloop" +) + +//go:embed ffmpeg.js +var script string + +type Config struct { + DefaultUserAgent *string + ProxyHandler func(*http.Request) (*url.URL, error) + RegisterCleanup func(func()) +} +type options struct { + Video *media.HTTPSource `json:"video"` + Audio *media.HTTPSource `json:"audio"` + Format string `json:"format"` + Args []string `json:"args"` +} +type job struct { + ctx context.Context + cancel context.CancelFunc + output *io.PipeReader + streams [2]*media.StreamInput + mu sync.Mutex + ready chan struct{} + start chan options + started bool +} + +func (j *job) stop() { + j.cancel() + j.output.CloseWithError(context.Canceled) + j.mu.Lock() + defer j.mu.Unlock() + for _, s := range j.streams { + if s != nil { + s.Close() + } + } +} + +func Enable(vm *goja.Runtime, loop *eventloop.EventLoop, cfg *Config) error { + if cfg == nil { + cfg = &Config{} + } + client, err := httpclient.NewClient(httpclient.Options{Transport: httpclient.TransportOptions{Proxy: cfg.ProxyHandler}}) + if err != nil { + return err + } + var mu sync.Mutex + jobs := map[string]*job{} + var next atomic.Uint64 + get := func(id string) *job { mu.Lock(); defer mu.Unlock(); return jobs[id] } + remove := func(id string) { + mu.Lock() + j := jobs[id] + delete(jobs, id) + mu.Unlock() + if j != nil { + j.stop() + } + } + if cfg.RegisterCleanup != nil { + cfg.RegisterCleanup(func() { + done := make(chan struct{}) + if loop.RunOnLoop(func(vm *goja.Runtime) { + defer close(done) + if closeAll, ok := goja.AssertFunction(vm.Get("__gopeed_ffmpeg_close_all")); ok { + _, _ = closeAll(goja.Undefined()) + } + }) { + <-done + } + mu.Lock() + old := jobs + jobs = map[string]*job{} + mu.Unlock() + for _, j := range old { + j.stop() + } + client.CloseIdleConnections() + }) + } + if err := vm.Set("__gopeed_ffmpeg_create", func(call goja.FunctionCall) goja.Value { + var opts options + if err := vm.ExportTo(call.Argument(0), &opts); err != nil { + panic(vm.NewGoError(err)) + } + if _, err := media.Arguments(opts.Format, opts.Args); err != nil { + panic(vm.NewGoError(err)) + } + ctx, cancel := context.WithCancel(context.Background()) + r, w := io.Pipe() + j := &job{ctx: ctx, cancel: cancel, output: r, ready: make(chan struct{}), start: make(chan options, 1)} + id := fmt.Sprintf("ffmpeg-%d", next.Add(1)) + mu.Lock() + jobs[id] = j + mu.Unlock() + go func() { + var inputs [2]media.Input + var runErr error + defer func() { + if v := recover(); v != nil { + runErr = fmt.Errorf("ffmpeg panic: %v", v) + } + cancel() + for _, in := range inputs { + if in != nil { + in.Close() + } + } + w.CloseWithError(runErr) + }() + release, err := media.Acquire(ctx) + if err != nil { + runErr = err + return + } + defer release() + close(j.ready) + var opts options + select { + case opts = <-j.start: + case <-ctx.Done(): + runErr = ctx.Err() + return + } + for i, source := range []*media.HTTPSource{opts.Video, opts.Audio} { + if source == nil { + inputs[i] = j.streams[i] + } else { + inputs[i], runErr = media.OpenHTTP(ctx, client, *source) + if runErr != nil { + return + } + } + } + runErr = media.RunAcquired(ctx, inputs[0], inputs[1], w, opts.Format, opts.Args) + }() + return vm.ToValue(id) + }); err != nil { + return err + } + if err := vm.Set("__gopeed_ffmpeg_acquire", func(id string) goja.Value { + promise, resolve, reject := vm.NewPromise() + j := get(id) + go func() { + var err error + if j == nil { + err = io.ErrClosedPipe + } else { + select { + case <-j.ready: + case <-j.ctx.Done(): + err = j.ctx.Err() + } + if j.ctx.Err() != nil { + err = j.ctx.Err() + } + } + loop.RunOnLoop(func(vm *goja.Runtime) { + if err != nil { + reject(vm.NewGoError(err)) + } else { + resolve(true) + } + }) + }() + return vm.ToValue(promise) + }); err != nil { + return err + } + if err := vm.Set("__gopeed_ffmpeg_start", func(call goja.FunctionCall) goja.Value { + id := call.Argument(0).String() + j := get(id) + if j == nil { + panic(vm.NewGoError(io.ErrClosedPipe)) + } + var opts options + if err := vm.ExportTo(call.Argument(1), &opts); err != nil { + panic(vm.NewGoError(err)) + } + if _, err := media.Arguments(opts.Format, opts.Args); err != nil { + panic(vm.NewGoError(err)) + } + for _, source := range []*media.HTTPSource{opts.Video, opts.Audio} { + if source == nil || cfg.DefaultUserAgent == nil { + continue + } + specified := false + for key := range source.Headers { + if strings.EqualFold(key, "User-Agent") { + specified = true + break + } + } + if !specified { + if source.Headers == nil { + source.Headers = make(map[string]string) + } + source.Headers["User-Agent"] = *cfg.DefaultUserAgent + } + } + + j.mu.Lock() + defer j.mu.Unlock() + if j.started || j.ctx.Err() != nil { + panic(vm.NewGoError(io.ErrClosedPipe)) + } + j.started = true + for i, source := range []*media.HTTPSource{opts.Video, opts.Audio} { + if source == nil { + j.streams[i] = media.NewStreamInput(j.ctx) + } + } + j.start <- opts + return goja.Undefined() + }); err != nil { + return err + } + if err := vm.Set("__gopeed_ffmpeg_read", func(id string) goja.Value { + promise, resolve, reject := vm.NewPromise() + go func() { + var readErr error + buf := make([]byte, 64*1024) + n := 0 + j := get(id) + if j == nil { + readErr = io.ErrClosedPipe + } else { + n, readErr = j.output.Read(buf) + } + if !loop.RunOnLoop(func(vm *goja.Runtime) { + if readErr != nil { + remove(id) + if readErr == io.EOF { + resolve(goja.Null()) + } else { + reject(vm.NewGoError(readErr)) + } + return + } + resolve(vm.ToValue(vm.NewArrayBuffer(buf[:n]))) + }) { + remove(id) + } + }() + return vm.ToValue(promise) + }); err != nil { + return err + } + if err := vm.Set("__gopeed_ffmpeg_push", func(call goja.FunctionCall) goja.Value { + id := call.Argument(0).String() + index := int(call.Argument(1).ToInteger()) + var chunk []byte + switch b := call.Argument(2).Export().(type) { + case goja.ArrayBuffer: + chunk = append([]byte(nil), b.Bytes()...) + case []byte: + chunk = append([]byte(nil), b...) + default: + panic(vm.NewTypeError("ffmpeg input requires bytes")) + } + promise, resolve, reject := vm.NewPromise() + go func() { + var err error + j := get(id) + if j == nil || index < 0 || index > 1 || j.streams[index] == nil { + err = io.ErrClosedPipe + } else { + _, err = j.streams[index].Write(chunk) + } + if err != nil && j != nil && index >= 0 && index < 2 && j.streams[index] != nil { + j.streams[index].End(err) + } + loop.RunOnLoop(func(vm *goja.Runtime) { + if j != nil && j.ctx.Err() != nil { + resolve(false) + return + } + if err != nil { + reject(vm.NewGoError(err)) + } else { + resolve(true) + } + }) + }() + return vm.ToValue(promise) + }); err != nil { + return err + } + if err := vm.Set("__gopeed_ffmpeg_end", func(id string, index int, message string) { + if j := get(id); j != nil && index >= 0 && index < 2 && j.streams[index] != nil { + var err error + if message != "" { + err = errors.New(message) + } + j.streams[index].End(err) + } + }); err != nil { + return err + } + if err := vm.Set("__gopeed_ffmpeg_cancel", remove); err != nil { + return err + } + _, err = vm.RunString(script) + return err +} diff --git a/pkg/download/engine/inject/stream/module.go b/pkg/download/engine/inject/stream/module.go index 5aaa214f..89db11fc 100644 --- a/pkg/download/engine/inject/stream/module.go +++ b/pkg/download/engine/inject/stream/module.go @@ -86,18 +86,23 @@ func Enable(runtime *goja.Runtime, loop *eventloop.EventLoop, cfg *Config) error } if err := runtime.Set("__gopeed_blob_pipe_chunk", func(call goja.FunctionCall) goja.Value { id := call.Argument(0).String() - ok := blobPipes.Push(id, exportBytes(call.Argument(1))) - return runtime.ToValue(ok) + chunk := append([]byte(nil), exportBytes(call.Argument(1))...) + promise, resolve, _ := runtime.NewPromise() + go func() { + ok := blobPipes.Push(id, chunk) + loop.RunOnLoop(func(vm *goja.Runtime) { resolve(ok) }) + }() + return runtime.ToValue(promise) }); err != nil { return err } if err := runtime.Set("__gopeed_blob_pipe_close", func(id string) { - blobPipes.Close(id, nil) + go blobPipes.Close(id, nil) }); err != nil { return err } if err := runtime.Set("__gopeed_blob_pipe_error", func(id string, message string) { - blobPipes.Close(id, errors.New(message)) + go blobPipes.Close(id, errors.New(message)) }); err != nil { return err } diff --git a/pkg/download/engine/inject/stream/module_integration_test.go b/pkg/download/engine/inject/stream/module_integration_test.go index b2c11cec..e9e07411 100644 --- a/pkg/download/engine/inject/stream/module_integration_test.go +++ b/pkg/download/engine/inject/stream/module_integration_test.go @@ -269,3 +269,57 @@ func newStreamTestEngine(t *testing.T, created chan<- createdObjectURL) (*engine }) return e, e.Close } + +// A stalled disk/network consumer must pause the producer without blocking +// the JS event loop, so cancellation and other tracks can still run. +func TestBlobBackpressureLeavesEventLoopResponsive(t *testing.T) { + created := make(chan createdObjectURL, 1) + e, cleanup := newStreamTestEngine(t, created) + defer cleanup() + _, err := e.RunString(` + globalThis.produced = 0; + globalThis.sourceCancelled = false; + __gopeed_blob_create_object_url(() => new ReadableStream({ + pull(c) { produced++; c.enqueue(new Uint8Array(64*1024)); }, + cancel() { sourceCancelled = true; } + })); + `) + if err != nil { + t.Fatal(err) + } + object := <-created + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + reader, err := object.open(ctx, stream.ObjectURLOpenRequest{End: -1}) + if err != nil { + t.Fatal(err) + } + defer reader.Close() + time.Sleep(100 * time.Millisecond) + value, err := e.RunString("produced") + if err != nil { + t.Fatal(err) + } + before := value.(int64) + if before > 10 { + t.Fatalf("unbounded production: %d", before) + } + time.Sleep(50 * time.Millisecond) + after, err := e.RunString("produced") + if err != nil || after != value { + t.Fatalf("producer did not pause: %v -> %v (%v)", value, after, err) + } + if ctx.Err() != nil { + t.Fatal("event loop only resumed after cancellation") + } + if _, err = reader.Read(make([]byte, 64*1024)); err != nil { + t.Fatal(err) + } + time.Sleep(30 * time.Millisecond) + after, err = e.RunString("produced") + if err != nil || after.(int64) <= before { + t.Fatalf("producer did not resume: %v (%v)", after, err) + } + reader.Close() + waitForJSValue(t, e, "String(sourceCancelled)", "true") +} diff --git a/pkg/download/engine/inject/stream/stream.js b/pkg/download/engine/inject/stream/stream.js index ea3bd85b..829a9344 100644 --- a/pkg/download/engine/inject/stream/stream.js +++ b/pkg/download/engine/inject/stream/stream.js @@ -105,6 +105,10 @@ return Promise.resolve({ done: true, value: undefined }); } this._maybePull(); + // pull() may synchronously error before a waiter has been registered. + if (this._errored) { + return Promise.reject(this._errored); + } if (this._queue.length > 0) { return Promise.resolve({ done: false, value: this._queue.shift() }); } @@ -760,11 +764,14 @@ return; } const chunk = value instanceof Uint8Array ? value : new Uint8Array(value); - const buffer = chunk.buffer.slice(chunk.byteOffset, chunk.byteOffset + chunk.byteLength); - if (!globalThis.__gopeed_blob_pipe_chunk(pipeId, buffer)) { - state.cancelRequested = true; - state.cancelReason = "blob pipe closed"; - return; + for (let offset = 0; offset < chunk.byteLength; offset += 256 * 1024) { + const part = chunk.subarray(offset, offset + 256 * 1024); + const buffer = part.buffer.slice(part.byteOffset, part.byteOffset + part.byteLength); + if (!await globalThis.__gopeed_blob_pipe_chunk(pipeId, buffer)) { + state.cancelRequested = true; + state.cancelReason = "blob pipe closed"; + return; + } } await yieldBlobPipeTask(); } diff --git a/pkg/download/engine/testdata/ffmpeg/README.md b/pkg/download/engine/testdata/ffmpeg/README.md new file mode 100644 index 00000000..ba3e5992 --- /dev/null +++ b/pkg/download/engine/testdata/ffmpeg/README.md @@ -0,0 +1,18 @@ +# Synthetic media fixtures + +These files contain 0.3 seconds of a generated 32x32 black image (H.264) and +generated mono silence (AAC). They contain no third-party media. Generated +with the pinned go-ffmpreg FFmpeg binary using the equivalent commands: + +``` +ffmpeg -f lavfi -i color=c=black:s=32x32:r=10 -t 0.3 -c:v libx264 -preset ultrafast -movflags frag_keyframe+empty_moov -f mp4 video.mp4 +ffmpeg -f lavfi -i anullsrc=r=48000:cl=mono -t 0.3 -c:a aac -movflags frag_keyframe+empty_moov -f mp4 audio.mp4 +``` + +The backend tests also generate and probe media using the embedded FFmpeg and +FFprobe at test time, without relying on a system installation. + +`video.webm` and `audio.webm` use the same black/silent sources, encoded once +with native FFmpeg 7.1 using `-c:v libvpx-vp9` and `-c:a libopus`. The committed +fixtures let the tests verify WASM stream-copy remuxing without exercising its +VP9 encoder or requiring native FFmpeg on developer/CI machines. diff --git a/pkg/download/engine/testdata/ffmpeg/audio.mp4 b/pkg/download/engine/testdata/ffmpeg/audio.mp4 new file mode 100644 index 00000000..1a16c3c4 Binary files /dev/null and b/pkg/download/engine/testdata/ffmpeg/audio.mp4 differ diff --git a/pkg/download/engine/testdata/ffmpeg/audio.webm b/pkg/download/engine/testdata/ffmpeg/audio.webm new file mode 100644 index 00000000..6e06a4a2 Binary files /dev/null and b/pkg/download/engine/testdata/ffmpeg/audio.webm differ diff --git a/pkg/download/engine/testdata/ffmpeg/video.mp4 b/pkg/download/engine/testdata/ffmpeg/video.mp4 new file mode 100644 index 00000000..d5d3d385 Binary files /dev/null and b/pkg/download/engine/testdata/ffmpeg/video.mp4 differ diff --git a/pkg/download/engine/testdata/ffmpeg/video.webm b/pkg/download/engine/testdata/ffmpeg/video.webm new file mode 100644 index 00000000..f88aa145 Binary files /dev/null and b/pkg/download/engine/testdata/ffmpeg/video.webm differ diff --git a/pkg/download/extension_engine_session.go b/pkg/download/extension_engine_session.go index 451a2764..4f248c88 100644 --- a/pkg/download/extension_engine_session.go +++ b/pkg/download/extension_engine_session.go @@ -3,9 +3,11 @@ package download import ( "context" "io" + "strings" "sync" internalblob "github.com/GopeedLab/gopeed/internal/blob" + httpProtocol "github.com/GopeedLab/gopeed/internal/protocol/http" "github.com/GopeedLab/gopeed/pkg/download/engine" "github.com/GopeedLab/gopeed/pkg/download/engine/inject/stream" enginewebview "github.com/GopeedLab/gopeed/pkg/download/engine/webview" @@ -119,9 +121,15 @@ func (d *Downloader) newExtensionEngine() (*engine.Engine, *engineSession) { return d.blob.Revoke(url) }, } + httpConfig := struct { + UserAgent string `json:"userAgent"` + }{UserAgent: httpProtocol.DefaultUserAgent} + d.getProtocolConfig("http", &httpConfig) + userAgent := strings.TrimSpace(httpConfig.UserAgent) e := engine.NewEngine(&engine.Config{ - ProxyConfig: d.cfg.Proxy, - StreamConfig: engineCfg, + HTTPUserAgent: &userAgent, + ProxyConfig: d.cfg.Proxy, + StreamConfig: engineCfg, }) session.SetEngine(e) return e, session diff --git a/pkg/download/extension_runtime_webview.go b/pkg/download/extension_runtime_webview.go index c9bb26b1..25d07cad 100644 --- a/pkg/download/extension_runtime_webview.go +++ b/pkg/download/extension_runtime_webview.go @@ -30,6 +30,9 @@ func injectGopeed(vm *goja.Runtime, gopeed *Instance) error { } runtimeObject := vm.NewObject() if gopeed.Runtime != nil { + if err := runtimeObject.Set("ffmpeg", vm.Get("__gopeed_ffmpeg")); err != nil { + return err + } if err := runtimeObject.Set("blob", newJSBlobRuntime(vm)); err != nil { return err } diff --git a/pkg/download/extension_test.go b/pkg/download/extension_test.go index b78380d5..ef5ff8bd 100644 --- a/pkg/download/extension_test.go +++ b/pkg/download/extension_test.go @@ -1484,6 +1484,7 @@ func TestDownloader_ExtensionRuntimeWebViewInjected(t *testing.T) { value, err := runtime.Eval(`(async () => ({ hasRuntime: !!gopeed.runtime, hasBlob: !!(gopeed.runtime && gopeed.runtime.blob), + hasFFmpegMerge: typeof gopeed.runtime.ffmpeg.merge, hasCreateObjectURL: typeof gopeed.runtime.blob.createObjectURL, hasRevokeObjectURL: typeof gopeed.runtime.blob.revokeObjectURL, hasWebView: !!(gopeed.runtime && gopeed.runtime.webview), @@ -1502,6 +1503,9 @@ func TestDownloader_ExtensionRuntimeWebViewInjected(t *testing.T) { if result["hasRuntime"] != true { t.Fatalf("expected runtime injection, got %#v", result) } + if result["hasFFmpegMerge"] != "function" { + t.Fatalf("expected ffmpeg injection, got %#v", result) + } if result["hasBlob"] != true { t.Fatalf("expected blob injection, got %#v", result) } diff --git a/pkg/download/extension_unnamed_resource_test.go b/pkg/download/extension_unnamed_resource_test.go new file mode 100644 index 00000000..7ef765eb --- /dev/null +++ b/pkg/download/extension_unnamed_resource_test.go @@ -0,0 +1,37 @@ +package download + +import ( + "github.com/GopeedLab/gopeed/pkg/base" + "os" + "path/filepath" + "testing" +) + +func TestExtensionUnnamedResource(t *testing.T) { + d, cleanup, err := newTestExtensionEngineDownloader() + if err != nil { + t.Fatal(err) + } + defer cleanup() + dir := t.TempDir() + manifest := `{"name":"unnamed","title":"Unnamed test","author":"test","version":"1.0.0","scripts":[{"event":"onResolve","match":{"urls":["https://example.com/unnamed"]},"entry":"index.js"}]}` + script := `gopeed.events.onResolve(async(ctx)=>{ctx.res={range:false,files:[{name:'video.mp4',req:{url:await gopeed.runtime.blob.createObjectURL(()=>new ReadableStream({start(c){c.enqueue(new Uint8Array([1,2,3]));c.close();}}))}}]};});` + for name, data := range map[string]string{"manifest.json": manifest, "index.js": script} { + if err = os.WriteFile(filepath.Join(dir, name), []byte(data), 0600); err != nil { + t.Fatal(err) + } + } + if _, err = d.InstallExtensionByFolder(dir, true); err != nil { + t.Fatal(err) + } + res, err := d.Resolve(&base.Request{URL: "https://example.com/unnamed"}, nil) + if err != nil { + t.Fatal(err) + } + if res.Res.Name != "" || len(res.Res.Files) != 1 || res.Res.Files[0].Name != "video.mp4" { + t.Fatalf("unexpected resource: %+v", res.Res) + } + if !d.blob.IsURL(res.Res.Files[0].Req.URL) { + t.Fatal("extension result was discarded") + } +} diff --git a/pkg/download/extension_user_agent_test.go b/pkg/download/extension_user_agent_test.go new file mode 100644 index 00000000..627a6f73 --- /dev/null +++ b/pkg/download/extension_user_agent_test.go @@ -0,0 +1,108 @@ +package download + +import ( + "bytes" + "fmt" + httpProtocol "github.com/GopeedLab/gopeed/internal/protocol/http" + "net/http" + "net/http/httptest" + "os" + "strings" + "testing" + "time" +) + +func TestExtensionHTTPUserAgent(t *testing.T) { + video, err := os.ReadFile("engine/testdata/ffmpeg/video.mp4") + if err != nil { + t.Fatal(err) + } + audio, err := os.ReadFile("engine/testdata/ffmpeg/audio.mp4") + if err != nil { + t.Fatal(err) + } + for _, configured := range []string{" Configured-UA/1.0 ", "", ""} { + for _, header := range []string{"default", "custom", "empty"} { + t.Run(fmt.Sprintf("configured=%q/header=%s", configured, header), func(t *testing.T) { + expected := strings.TrimSpace(configured) + if configured == "" { + expected = httpProtocol.DefaultUserAgent + } + headers := `{}` + if header == "custom" { + expected = "Explicit-UA/2.0" + headers = `{"uSeR-aGeNt":"Explicit-UA/2.0"}` + } + if header == "empty" { + expected = "" + headers = `{"user-agent":""}` + } + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.UserAgent() != expected { + t.Errorf("%s UA=%q, want %q", r.URL.Path, r.UserAgent(), expected) + } + if strings.HasPrefix(r.URL.Path, "/redirect/") { + http.Redirect(w, r, strings.TrimPrefix(r.URL.Path, "/redirect"), http.StatusFound) + return + } + data := video + if strings.HasSuffix(r.URL.Path, "audio") { + data = audio + } + http.ServeContent(w, r, "media", time.Time{}, bytes.NewReader(data)) + })) + defer server.Close() + d, cleanup, err := newTestExtensionEngineDownloader() + if err != nil { + t.Fatal(err) + } + defer cleanup() + cfg, err := d.GetConfig() + if err != nil { + t.Fatal(err) + } + cfg.ProtocolConfig["http"] = map[string]any{"userAgent": configured} + if configured == "" { + cfg.ProtocolConfig["http"] = map[string]any{"connections": 4} + } + if err = d.PutConfig(cfg); err != nil { + t.Fatal(err) + } + runtime, err := newTestExtensionEngine(t, d) + if err != nil { + t.Fatal(err) + } + defer runtime.Close() + // The existing engine keeps its UA snapshot when the downloader config changes. + cfg, err = d.GetConfig() + if err != nil { + t.Fatal(err) + } + cfg.ProtocolConfig["http"] = map[string]any{"userAgent": "Next-engine-UA"} + if err = d.PutConfig(cfg); err != nil { + t.Fatal(err) + } + result, err := runtime.Eval(fmt.Sprintf(`(async()=>{ + const headers=%s, url=%q; + await (await fetch(url+'/redirect/fetch',{headers})).arrayBuffer(); + await new Promise((resolve,reject)=>{ + const xhr=new XMLHttpRequest();xhr.open('GET',url+'/redirect/xhr'); + for(const key of Object.keys(headers))xhr.setRequestHeader(key,headers[key]); + xhr.onload=resolve;xhr.onerror=()=>reject(new Error('XHR failed'));xhr.send(); + }); + // Profile defaults must not replace the configured or explicitly empty UA. + __gopeed_setFingerprint('chrome'); + await (await fetch(url+'/fingerprint',{headers})).arrayBuffer(); + const output=gopeed.runtime.ffmpeg.merge({video:{url:url+'/redirect/video',headers},audio:{url:url+'/redirect/audio',headers}}); + return (await new Response(output).arrayBuffer()).byteLength > 0; + })()`, headers, server.URL)) + if err != nil { + t.Fatal(err) + } + if result != true { + t.Fatalf("merge result=%v", result) + } + }) + } + } +} diff --git a/ui/flutter/lib/features/tasks/domain/task_record.dart b/ui/flutter/lib/features/tasks/domain/task_record.dart index 94a94818..ca85a26b 100644 --- a/ui/flutter/lib/features/tasks/domain/task_record.dart +++ b/ui/flutter/lib/features/tasks/domain/task_record.dart @@ -299,8 +299,9 @@ TaskAssetType _assetType(String name, {required bool isFolder, required api_task } String _fileExtension(String name) { - final pathWithoutQuery = name.split(RegExp(r'[?#]')).first; - final extension = path.extension(pathWithoutQuery).replaceFirst('.', '').toLowerCase(); + final uri = Uri.tryParse(name); + final filePath = uri != null && uri.hasScheme && uri.hasAuthority ? uri.path : name; + final extension = path.extension(filePath).replaceFirst('.', '').toLowerCase(); return extension.length <= 10 ? extension : ''; } diff --git a/ui/flutter/test/features/tasks/domain/task_record_test.dart b/ui/flutter/test/features/tasks/domain/task_record_test.dart new file mode 100644 index 00000000..8b5c72ca --- /dev/null +++ b/ui/flutter/test/features/tasks/domain/task_record_test.dart @@ -0,0 +1,20 @@ +import 'package:flutter_test/flutter_test.dart'; +import 'package:gopeed/core/icons/gopeed_icons.dart'; +import 'package:gopeed/features/tasks/domain/task_record.dart'; + +void main() { + test('file names preserve hash and question mark characters for type detection', () { + expect( + taskFileTypeIcon('Slow English Podcast #2 _ Eating, Talking, and Learning Natural English.mp4'), + GopeedIcons.fileVideo, + ); + expect(taskFileTypeIcon('What is this? #2.MP4'), GopeedIcons.fileVideo); + expect(taskFileTypeIcon('Episode #2.mp3'), GopeedIcons.fileAudio); + }); + + test('URL fallback ignores query and fragment when detecting file type', () { + expect(taskFileTypeIcon('https://example.com/video.mp4?token=abc#part'), GopeedIcons.fileVideo); + expect(taskFileTypeIcon('https://example.com/download?name=video.mp4'), GopeedIcons.file); + expect(taskFileTypeIcon('Playlist #2.mp4', isFolder: true), GopeedIcons.folder); + }); +}