feat: add FFmpeg WASM runtime (#1512)

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() {

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 });
+})();

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
+}

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
 	}

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")
+}

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();
         }