package transcode import ( "context" "errors" "fmt" "log/slog" "time" ) // Engine 是 Transcoder 的默认实现:把「格式无关」的那一层(输入字节上限、 // 并发信号量、单次超时、日志)收在一处,把「格式相关」的那一层交给注册进来的 // Codec(design/25 §1.3.2 / §2.2)。 // // 它是普通结构体,由组合根构造并注入;**没有包级全局变量** // (AGENTS §10 红线)。它可安全并发使用。 type Engine struct { limits Limits reg *Registry // sem 是全局并发信号量。容量 = Limits.Concurrency;槽满即 ErrBusy // (快速失败、不入队)。 sem chan struct{} log *slog.Logger // codecs 是选项里攒下的待注册列表,在 NewEngine 末尾统一注册(见 WithCodecs)。 codecs []Codec // regErr 记录注册期的第一个错误(不该发生:DefaultCodecs 是已知良好的)。 // 它不阻塞构造,但绝不静默——组合根可读 RegistrationError 并告警。 regErr error } // Compile-time assertion that *Engine satisfies the Transcoder interface. var _ Transcoder = (*Engine)(nil) // Option 配置 Engine。 type Option func(*Engine) // WithCodecs 把 codecs 交给引擎(在 NewEngine 的所有选项都应用完之后统一注册, // 这样即使调用方按 design/25 §6.1 的顺序先写 WithCodecs 再写 WithLogger, // 注册失败的告警也能落到正确的日志器上)。注册失败(重复 / 不合法)是程序错误: // 记 Error 日志并留在 RegistrationError 里,绝不静默吞掉。 func WithCodecs(codecs ...Codec) Option { return func(e *Engine) { e.codecs = append(e.codecs, codecs...) } } // WithLogger 设置引擎日志。日志一律带 source=capability/transcode // (AGENTS §10:日志必须标来源)。 func WithLogger(l *slog.Logger) Option { return func(e *Engine) { e.log = loggerFor(l) } } // loggerFor 把 nil 日志收敛成一个丢弃日志:本包不因为「没人给日志」而崩, // 也不因为「没人给日志」而少一次调用记录(默认丢弃,责任在组合根)。 func loggerFor(l *slog.Logger) *slog.Logger { if l == nil { return slog.New(slog.DiscardHandler) } return l } // NewEngine 构造一个引擎。limits 的零值字段由 DefaultLimits 补足; // 未给 WithCodecs 时注册表为空 ⇒ Supports 恒 false、Transcode 恒 // ErrUnsupportedFormat(design/25 §8-T1 的空注册表行为)。 func NewEngine(limits Limits, opts ...Option) *Engine { e := &Engine{ limits: limits.withDefaults(), reg: NewRegistry(), sem: make(chan struct{}, limits.withDefaults().Concurrency), log: loggerFor(nil), } for _, opt := range opts { if opt != nil { opt(e) } } e.registerCodecs() return e } // registerCodecs 注册 WithCodecs 攒下的编解码器(在所有选项都生效之后)。 func (e *Engine) registerCodecs() { for _, c := range e.codecs { if err := e.reg.Register(c); err != nil { if e.regErr == nil { e.regErr = err } e.log.Error("transcode codec registration skipped", "error", err) } } e.codecs = nil } // RegistrationError 返回构造期第一个编解码器注册错误(正常为 nil)。 func (e *Engine) RegistrationError() error { return e.regErr } // Limits 返回生效的限额(零值已补默认)。 func (e *Engine) Limits() Limits { return e.limits } // Supports 报告 src→dst 这条路是否可用(能力门)。 func (e *Engine) Supports(src, dst Format) bool { _, ok := e.reg.Lookup(src, dst) return ok } // Formats 报告当前注册的所有格式(发现)。 func (e *Engine) Formats() []Format { return e.reg.Formats() } // Transcode 把 req.Data 从 req.Source 转成 req.Target(design/25 §2.1)。 // // 失败一律返回 errors.go 的哨兵之一(或包裹它)——ctx 截止返回 ErrTimeout, // 调用方主动取消返回 context.Canceled 的包裹(取消是调用方自己的决定, // 与「转码失败」不是一回事)。进入编解码器之前的每一道闸都在**解码前**: // 未知格式、越限、槽满都不会解出一个字节。 // // ctx 为 nil 时按 context.Background() 处理;ctx 超时是硬闸,绝不无界等待, // 也不泄漏 goroutine(本调用全程在调用方 goroutine 里跑完)。 func (e *Engine) Transcode(ctx context.Context, req Request) (Result, error) { start := time.Now() if ctx == nil { ctx = context.Background() } src, err := e.resolveSource(req) if err != nil { return Result{}, e.reject(ctx, req, src, err, start) } if req.Target == "" { return Result{}, e.reject(ctx, req, src, fmt.Errorf("%w: empty target format", ErrUnsupportedFormat), start) } codec, ok := e.reg.Lookup(src, req.Target) if !ok { return Result{}, e.reject(ctx, req, src, fmt.Errorf("%w: no codec registered", ErrUnsupportedFormat), start) } if len(req.Data) > e.limits.MaxInputBytes { return Result{}, e.reject(ctx, req, src, fmt.Errorf("%w: %d bytes > %d", ErrInputTooLarge, len(req.Data), e.limits.MaxInputBytes), start) } if err := ctx.Err(); err != nil { return Result{}, e.reject(ctx, req, src, timeoutError(err), start) } if !e.acquire() { return Result{}, e.reject(ctx, req, src, fmt.Errorf("%w: %d slots busy", ErrBusy, e.limits.Concurrency), start) } defer e.release() callCtx, cancel := context.WithTimeout(ctx, e.limits.Timeout) defer cancel() res, err := codec.Transcode(callCtx, req.Data, e.limits) if err != nil { return Result{}, e.reject(ctx, req, src, classify(callCtx, err), start) } if len(res.Data) == 0 { return Result{}, e.reject(ctx, req, src, fmt.Errorf("%w: codec returned empty output", ErrEncodeFailed), start) } e.log.Info("transcode ok", "source_format", FormatName(res.Source), "target_format", FormatName(res.Format), "in_bytes", len(req.Data), "out_bytes", len(res.Data), "frames", res.Frames, "duration_ms", res.DurationMs, "elapsed_ms", time.Since(start).Milliseconds()) return res, nil } // resolveSource 定出这次请求的实际来源格式:显式给出就用它,留空则按魔数嗅探, // 嗅不出即 ErrUnsupportedFormat(deny-by-default)。 func (e *Engine) resolveSource(req Request) (Format, error) { if req.Source != "" { return req.Source, nil } sniffed, ok := Sniff(req.Data) if !ok { return "", fmt.Errorf("%w: magic not recognized", ErrUnsupportedFormat) } return sniffed, nil } // acquire 非阻塞地取一个并发槽。拿不到即 false——绝不排队。 func (e *Engine) acquire() bool { select { case e.sem <- struct{}{}: return true default: return false } } // release 归还一个并发槽。 func (e *Engine) release() { <-e.sem } // reject 记一次失败日志并把错误包上**路由上下文**后返回(绝不静默、绝不裸返回): // 最终文案形如 `transcode: amr -> wav: unsupported format: no codec registered`。 // 可预期的输入问题(哨兵)记 Warn;不该发生的(非哨兵:编解码器违约)记 Error。 func (e *Engine) reject(ctx context.Context, req Request, src Format, err error, start time.Time) error { level := slog.LevelWarn if !isSentinelError(err) { level = slog.LevelError } e.log.Log(ctx, level, "transcode failed", "source_format", FormatName(src), "target_format", FormatName(req.Target), "in_bytes", len(req.Data), "elapsed_ms", time.Since(start).Milliseconds(), "error", err) return fmt.Errorf("transcode: %s: %w", routeLabel(src, req.Target), err) } // routeLabel 是错误/日志里的路由可读名;格式未知(嗅探失败 / 目标留空)时用 // `` 占位,绝不留一段空白让人猜。 func routeLabel(src, dst Format) string { from, to := FormatName(src), FormatName(dst) if from == "" { from = "" } if to == "" { to = "" } return from + " -> " + to } // classify 把编解码器返回的错误收敛到 design/25 §2.4 的口径:ctx 截止即 // ErrTimeout;调用方取消保持 context.Canceled 的包裹;其余补一层来源后上抛 // (编解码器自己用哨兵,engine 不替它猜,但也不裸返回)。 func classify(callCtx context.Context, err error) error { switch { case errors.Is(err, context.DeadlineExceeded), errors.Is(callCtx.Err(), context.DeadlineExceeded): return timeoutError(context.DeadlineExceeded) case errors.Is(err, context.Canceled), errors.Is(callCtx.Err(), context.Canceled): return fmt.Errorf("canceled: %w", context.Canceled) default: return fmt.Errorf("codec failed: %w", err) } } // timeoutError 把 ctx 截止统一成 ErrTimeout 的包裹,并保留原始原因。 func timeoutError(cause error) error { return fmt.Errorf("%w: %w", ErrTimeout, cause) } // isSentinelError 报告 err 是否属于本包已登记的哨兵(含包内构造期错误)。 // 逐条列举而不是查一个包级切片:本包因此**没有任何包级状态** // (AGENTS §10:禁包级全局变量,除 var ErrXxx)。 func isSentinelError(err error) bool { return errors.Is(err, ErrUnsupportedFormat) || errors.Is(err, ErrInputTooLarge) || errors.Is(err, ErrOutputTooLarge) || errors.Is(err, ErrDecodeFailed) || errors.Is(err, ErrEncodeFailed) || errors.Is(err, ErrTimeout) || errors.Is(err, ErrBusy) || errors.Is(err, ErrNoBackend) || errors.Is(err, ErrDuplicate) || errors.Is(err, ErrInvalidCodec) || errors.Is(err, ErrBadArgs) }