monibuca/plugin.go

package m7s

import (
	"bytes"
	"context"
	"crypto/md5"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"net"
	"net/http"
	"net/url"
	"os"
	"path/filepath"
	"reflect"
	"runtime"
	"strconv"
	"strings"
	"time"

	"gopkg.in/yaml.v3"

	"m7s.live/v5/pkg/task"

	"github.com/quic-go/quic-go"

	gatewayRuntime "github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
	myip "github.com/husanpao/ip"
	"google.golang.org/grpc"
	"gorm.io/gorm"

	. "m7s.live/v5/pkg"
	"m7s.live/v5/pkg/config"
	"m7s.live/v5/pkg/db"
	"m7s.live/v5/pkg/util"
)

type (
	DefaultYaml    string
	OnExitHandler  func()
	AuthPublisher  = func(*Publisher) *util.Promise
	AuthSubscriber = func(*Subscriber) *util.Promise

	PluginMeta struct {
		Name                string
		Version             string //插件版本
		Type                reflect.Type
		DefaultYaml         DefaultYaml //默认配置
		ServiceDesc         *grpc.ServiceDesc
		RegisterGRPCHandler func(context.Context, *gatewayRuntime.ServeMux, *grpc.ClientConn) error
		NewPuller           PullerFactory
		NewPusher           PusherFactory
		NewRecorder         RecorderFactory
		NewTransformer      TransformerFactory
		NewPullProxy        PullProxyFactory
		NewPushProxy        PushProxyFactory
		OnExit              OnExitHandler
		OnAuthPub           AuthPublisher
		OnAuthSub           AuthSubscriber
	}

	iPlugin interface {
		nothing()
	}

	IPlugin interface {
		task.IJob
		OnInit() error
		OnStop()
		Pull(string, config.Pull, *config.Publish) (*PullJob, error)
		Push(string, config.Push, *config.Subscribe)
		Transform(*Publisher, config.Transform)
		OnPublish(*Publisher)
	}

	IRegisterHandler interface {
		RegisterHandler() map[string]http.HandlerFunc
	}

	IPullerPlugin interface {
		GetPullableList() []string
	}

	ITCPPlugin interface {
		OnTCPConnect(conn *net.TCPConn) task.ITask
	}

	IUDPPlugin interface {
		OnUDPConnect(conn *net.UDPConn) task.ITask
	}

	IQUICPlugin interface {
		OnQUICConnect(quic.Connection) task.ITask
	}
)

var plugins []PluginMeta

func (plugin *PluginMeta) Init(s *Server, userConfig map[string]any) (p *Plugin) {
	instance, ok := reflect.New(plugin.Type).Interface().(IPlugin)
	if !ok {
		panic("plugin must implement IPlugin")
	}
	p = reflect.ValueOf(instance).Elem().FieldByName("Plugin").Addr().Interface().(*Plugin)
	p.handler = instance
	p.Meta = plugin
	p.Server = s
	p.Logger = s.Logger.With("plugin", plugin.Name)
	upperName := strings.ToUpper(plugin.Name)
	if os.Getenv(upperName+"_ENABLE") == "false" {
		p.disable("env")
		p.Warn("disabled by env")
		return
	}
	p.Config.Parse(p.GetCommonConf(), upperName)
	p.Config.Parse(instance, upperName)
	for _, fname := range MergeConfigs {
		if name := strings.ToLower(fname); p.Config.Has(name) {
			p.Config.Get(name).ParseGlobal(s.Config.Get(name))
		}
	}
	if plugin.DefaultYaml != "" {
		var defaultConf map[string]any
		if err := yaml.Unmarshal([]byte(plugin.DefaultYaml), &defaultConf); err != nil {
			p.Error("parsing default config", "error", err)
		} else {
			p.Config.ParseDefaultYaml(defaultConf)
		}
	}
	p.Config.ParseUserFile(userConfig)
	p.SetDescription("version", plugin.Version)
	if userConfig != nil {
		p.SetDescription("userConfig", userConfig)
	}
	finalConfig, _ := yaml.Marshal(p.Config.GetMap())
	p.Logger.Handler().(*MultiLogHandler).SetLevel(ParseLevel(p.config.LogLevel))
	p.Debug("config", "detail", string(finalConfig))
	if userConfig["enable"] == false || (s.DisableAll && userConfig["enable"] != true) {
		p.disable("config")
		return
	}
	p.Info("init", "version", plugin.Version)
	var err error
	if p.config.DSN == s.GetCommonConf().DSN {
		p.DB = s.DB
	} else if p.config.DSN != "" {
		if factory, ok := db.Factory[p.config.DBType]; ok {
			p.DB, err = gorm.Open(factory(p.config.DSN), &gorm.Config{})
			if err != nil {
				s.Error("failed to connect database", "error", err, "dsn", s.config.DSN, "type", s.config.DBType)
				p.disable(fmt.Sprintf("database %v", err))
				return
			}
		}
	}
	if p.DB != nil && p.Meta.NewRecorder != nil {
		if err = p.DB.AutoMigrate(&RecordStream{}); err != nil {
			p.disable(fmt.Sprintf("auto migrate record stream failed %v", err))
			return
		}
		if err = p.DB.AutoMigrate(&EventRecordStream{}); err != nil {
			p.disable(fmt.Sprintf("auto migrate event record stream failed %v", err))
			return
		}
	}
	if err := s.AddTask(instance).WaitStarted(); err != nil {
		p.disable(instance.StopReason().Error())
		return
	}
	var handlers map[string]http.HandlerFunc
	if v, ok := instance.(IRegisterHandler); ok {
		handlers = v.RegisterHandler()
	}
	p.registerHandler(handlers)
	s.Plugins.Add(p)
	return
}

// InstallPlugin 安装插件
func InstallPlugin[C iPlugin](options ...any) error {
	var meta PluginMeta
	for _, option := range options {
		if m, ok := option.(PluginMeta); ok {
			meta = m
		}
	}
	var c *C
	meta.Type = reflect.TypeOf(c).Elem()
	if meta.Name == "" {
		meta.Name = strings.TrimSuffix(meta.Type.Name(), "Plugin")
	}
	_, pluginFilePath, _, _ := runtime.Caller(1)
	configDir := filepath.Dir(pluginFilePath)
	if meta.Version == "" {
		if _, after, found := strings.Cut(configDir, "@"); found {
			meta.Version = after
		} else {
			meta.Version = "dev"
		}
	}
	for _, option := range options {
		switch v := option.(type) {
		case OnExitHandler:
			meta.OnExit = v
		case DefaultYaml:
			meta.DefaultYaml = v
		case PullerFactory:
			meta.NewPuller = v
		case PusherFactory:
			meta.NewPusher = v
		case RecorderFactory:
			meta.NewRecorder = v
		case TransformerFactory:
			meta.NewTransformer = v
		case AuthPublisher:
			meta.OnAuthPub = v
		case AuthSubscriber:
			meta.OnAuthSub = v
		case *grpc.ServiceDesc:
			meta.ServiceDesc = v
		case func(context.Context, *gatewayRuntime.ServeMux, *grpc.ClientConn) error:
			meta.RegisterGRPCHandler = v
		}
	}
	plugins = append(plugins, meta)
	return nil
}

var _ IPlugin = (*Plugin)(nil)

type Plugin struct {
	task.Work
	config.Config
	Disabled           bool
	Meta               *PluginMeta
	PushAddr, PlayAddr []string
	config             config.Common
	handler            IPlugin
	Server             *Server
	DB                 *gorm.DB
}

func (Plugin) nothing() {

}

func (p *Plugin) GetKey() string {
	return p.Meta.Name
}

func (p *Plugin) GetGlobalCommonConf() *config.Common {
	return p.Server.GetCommonConf()
}

func (p *Plugin) GetCommonConf() *config.Common {
	return &p.config
}

func (p *Plugin) GetHandler() IPlugin {
	return p.handler
}

func (p *Plugin) GetPublicIP(netcardIP string) string {
	if p.config.PublicIP != "" {
		return p.config.PublicIP
	}
	if publicIP := util.GetPublicIP(netcardIP); publicIP != netcardIP {
		return publicIP
	}
	localIp := myip.InternalIPv4()
	if publicIP := util.GetPublicIP(localIp); publicIP != localIp {
		return publicIP
	}
	return localIp
}

func (p *Plugin) disable(reason string) {
	p.Disabled = true
	p.SetDescription("disableReason", reason)
	p.Warn("plugin disabled")
	p.Server.disabledPlugins = append(p.Server.disabledPlugins, p)
}

func (p *Plugin) Start() (err error) {
	s := p.Server
	s.AddTask(&webHookQueueTask)

	if err = p.listen(); err != nil {
		return
	}
	if err = p.handler.OnInit(); err != nil {
		return
	}
	if p.Meta.ServiceDesc != nil && s.grpcServer != nil {
		s.grpcServer.RegisterService(p.Meta.ServiceDesc, p.handler)
		if p.Meta.RegisterGRPCHandler != nil {
			if err = p.Meta.RegisterGRPCHandler(p.Context, s.config.HTTP.GetGRPCMux(), s.grpcClientConn); err != nil {
				p.disable(fmt.Sprintf("grpc %v", err))
				return
			} else {
				p.Info("grpc handler registered")
			}
		}
	}
	if p.config.Hook != nil {
		if hook, ok := p.config.Hook[config.HookOnServerKeepAlive]; ok && hook.Interval > 0 {
			p.AddTask(&ServerKeepAliveTask{plugin: p})
		}
	}
	return
}

func (p *Plugin) Dispose() {
	p.handler.OnStop()
	p.Server.Plugins.Remove(p)
}

func (p *Plugin) listen() (err error) {
	httpConf := &p.config.HTTP

	if httpConf.ListenAddrTLS != "" && (httpConf.ListenAddrTLS != p.Server.config.HTTP.ListenAddrTLS) {
		p.SetDescription("httpTLS", strings.TrimPrefix(httpConf.ListenAddrTLS, ":"))
		p.AddDependTask(CreateHTTPSWork(httpConf, p.Logger))
	}

	if httpConf.ListenAddr != "" && (httpConf.ListenAddr != p.Server.config.HTTP.ListenAddr) {
		p.SetDescription("http", strings.TrimPrefix(httpConf.ListenAddr, ":"))
		p.AddDependTask(CreateHTTPWork(httpConf, p.Logger))
	}

	if tcphandler, ok := p.handler.(ITCPPlugin); ok {
		tcpConf := &p.config.TCP
		if tcpConf.ListenAddr != "" {
			if tcpConf.AutoListen {
				if err = p.AddTask(tcpConf.CreateTCPWork(p.Logger, tcphandler.OnTCPConnect)).WaitStarted(); err != nil {
					return
				}
			}
			p.SetDescription("tcp", strings.TrimPrefix(tcpConf.ListenAddr, ":"))
		}
		if tcpConf.ListenAddrTLS != "" {
			if tcpConf.AutoListen {
				if err = p.AddTask(tcpConf.CreateTCPTLSWork(p.Logger, tcphandler.OnTCPConnect)).WaitStarted(); err != nil {
					return
				}
			}
			p.SetDescription("tcpTLS", strings.TrimPrefix(tcpConf.ListenAddrTLS, ":"))
		}
	}

	if udpHandler, ok := p.handler.(IUDPPlugin); ok {
		udpConf := &p.config.UDP
		if udpConf.ListenAddr != "" {
			if udpConf.AutoListen {
				if err = p.AddTask(udpConf.CreateUDPWork(p.Logger, udpHandler.OnUDPConnect)).WaitStarted(); err != nil {
					return
				}
			}
			p.SetDescription("udp", strings.TrimPrefix(udpConf.ListenAddr, ":"))
		}
	}

	if quicHandler, ok := p.handler.(IQUICPlugin); ok {
		quicConf := &p.config.Quic
		if quicConf.ListenAddr != "" {
			if quicConf.AutoListen {
				if err = p.AddTask(quicConf.CreateQUICWork(p.Logger, quicHandler.OnQUICConnect)).WaitStarted(); err != nil {
					return
				}
			}
			p.SetDescription("quic", strings.TrimPrefix(quicConf.ListenAddr, ":"))
		}
	}
	return
}

func (p *Plugin) OnInit() error {
	return nil
}

func (p *Plugin) OnStop() {

}

type WebHookQueueTask struct {
	task.Work
}

var webHookQueueTask WebHookQueueTask

type WebHookTask struct {
	task.Task
	plugin   *Plugin
	hookType config.HookType
	conf     config.Webhook
	data     any
	jsonData []byte
	alarm    AlarmInfo
}

func (t *WebHookTask) Start() error {
	if t.conf.URL == "" {
		return task.ErrTaskComplete
	}

	// 处理AlarmInfo数据
	if t.data != nil {
		// 获取主机名和IP地址
		hostname, err := os.Hostname()
		if err != nil {
			hostname = "unknown"
		}

		// 获取本机IP地址
		var ipAddr string
		addrs, err := net.InterfaceAddrs()
		if err == nil {
			for _, addr := range addrs {
				if ipnet, ok := addr.(*net.IPNet); ok && !ipnet.IP.IsLoopback() {
					if ipnet.IP.To4() != nil {
						ipAddr = ipnet.IP.String()
						break
					}
				}
			}
		}
		if ipAddr == "" {
			ipAddr = "unknown"
		}

		// 直接使用t.data作为AlarmInfo
		alarmInfo, ok := t.data.(AlarmInfo)
		if !ok {
			return fmt.Errorf("data is not of type AlarmInfo")
		}

		// 更新服务器信息
		if alarmInfo.ServerInfo == "" {
			alarmInfo.ServerInfo = fmt.Sprintf("%s (%s)", hostname, ipAddr)
		}

		// 确保时间戳已设置
		if alarmInfo.CreatedAt.IsZero() {
			alarmInfo.CreatedAt = time.Now()
		}
		if alarmInfo.UpdatedAt.IsZero() {
			alarmInfo.UpdatedAt = time.Now()
		}

		// 将AlarmInfo序列化为JSON
		jsonData, err := json.Marshal(alarmInfo)
		if err != nil {
			return fmt.Errorf("marshal AlarmInfo to json: %w", err)
		}

		t.jsonData = jsonData
		t.alarm = alarmInfo
	}

	t.SetRetry(t.conf.RetryTimes, t.conf.RetryInterval)
	return nil
}

func (t *WebHookTask) Go() error {
	// 检查是否需要保存告警到数据库
	var dbID uint
	if t.conf.SaveAlarm && t.plugin.DB != nil {
		// 默认 IsSent 为 false
		t.alarm.IsSent = false
		if err := t.plugin.DB.Create(&t.alarm).Error; err != nil {
			t.plugin.Error("保存告警到数据库失败", "error", err)
		} else {
			dbID = t.alarm.ID
			t.plugin.Info(""+
				"", "id", dbID)
		}
	}

	req, err := http.NewRequest(t.conf.Method, t.conf.URL, bytes.NewBuffer(t.jsonData))
	if err != nil {
		return err
	}

	req.Header.Set("Content-Type", "application/json")
	for k, v := range t.conf.Headers {
		req.Header.Set(k, v)
	}

	client := &http.Client{
		Timeout: time.Duration(t.conf.TimeoutSeconds) * time.Second,
	}

	resp, err := client.Do(req)
	if err != nil {
		t.plugin.Error("webhook请求失败", "error", err)
		return err
	}
	defer resp.Body.Close()

	// 如果发送成功且已保存到数据库，则更新IsSent字段为true
	if resp.StatusCode >= 200 && resp.StatusCode < 300 && t.conf.SaveAlarm && t.plugin.DB != nil && dbID > 0 {
		t.alarm.IsSent = true
		if err := t.plugin.DB.Model(&AlarmInfo{}).Where("id = ?", dbID).Update("is_sent", true).Error; err != nil {
			t.plugin.Error("更新告警发送状态失败", "error", err)
		} else {
			t.plugin.Info("告警发送状态已更新", "id", dbID, "is_sent", true)
		}
		return task.ErrTaskComplete
	}

	if resp.StatusCode >= 200 && resp.StatusCode < 300 {
		return task.ErrTaskComplete
	}

	err = fmt.Errorf("webhook请求失败，状态码：%d", resp.StatusCode)
	t.plugin.Error("webhook响应错误", "状态码", resp.StatusCode)
	return err
}

func (p *Plugin) SendWebhook(conf config.Webhook, data any) *task.Task {
	webhookTask := &WebHookTask{
		plugin: p,
		conf:   conf,
		data:   data,
	}
	return webHookQueueTask.AddTask(webhookTask)
}

// TODO: use alias stream
func (p *Plugin) OnPublish(pub *Publisher) {
	onPublish := p.config.OnPub
	if p.Meta.NewPusher != nil {
		for r, pushConf := range onPublish.Push {
			if pushConf.URL = r.Replace(pub.StreamPath, pushConf.URL); pushConf.URL != "" {
				p.Push(pub.StreamPath, pushConf, nil)
			}
		}
	}
	if p.Meta.NewRecorder != nil {
		for r, recConf := range onPublish.Record {
			if recConf.FilePath = r.Replace(pub.StreamPath, recConf.FilePath); recConf.FilePath != "" {
				p.Record(pub, recConf, nil)
			}
		}
	}
	var owner = pub.Value(Owner)
	var isTransformer bool
	if owner != nil {
		_, isTransformer = owner.(ITransformer)
	}
	if p.Meta.NewTransformer != nil && !isTransformer {
		for r, tranConf := range onPublish.Transform {
			if group := r.FindStringSubmatch(pub.StreamPath); group != nil {
				for j, to := range tranConf.Output {
					for i, g := range group {
						to.Target = strings.ReplaceAll(to.Target, fmt.Sprintf("$%d", i), g)
					}
					targetUrl, err := url.Parse(to.Target)
					if err == nil {
						to.StreamPath = strings.TrimPrefix(targetUrl.Path, "/")
					}
					tranConf.Output[j] = to
				}
				p.Transform(pub, tranConf)
			}
		}
	}
}

func (p *Plugin) auth(streamPath string, key string, secret string, expire string) (err error) {
	if unixTime, err := strconv.ParseInt(expire, 16, 64); err != nil || time.Now().Unix() > unixTime {
		return fmt.Errorf("auth failed expired")
	}
	if len(secret) != 32 {
		return fmt.Errorf("auth failed secret length must be 32")
	}
	trueSecret := md5.Sum([]byte(key + streamPath + expire))
	if secret == hex.EncodeToString(trueSecret[:]) {
		return nil
	}
	return fmt.Errorf("auth failed invalid secret")
}

func (p *Plugin) OnSubscribe(streamPath string, args url.Values) {
	//	var avoidTrans bool
	//AVOID:
	//	for trans := range server.Transforms.Range {
	//		for _, output := range trans.Config.Output {
	//			if output.StreamPath == s.StreamPath {
	//				avoidTrans = true
	//				break AVOID
	//			}
	//		}
	//	}
	for reg, conf := range p.config.OnSub.Pull {
		if p.Meta.NewPuller != nil && reg.MatchString(streamPath) {
			conf.Args = config.HTTPValues(args)
			conf.URL = reg.Replace(streamPath, conf.URL)
			p.handler.Pull(streamPath, conf, nil)
		}
	}

	//if !avoidTrans {
	//	for reg, conf := range plugin.GetCommonConf().OnSub.Transform {
	//		if plugin.Meta.Transformer != nil {
	//			if reg.MatchString(s.StreamPath) {
	//				if group := reg.FindStringSubmatch(s.StreamPath); group != nil {
	//					for j, c := range conf.Output {
	//						for i, value := range group {
	//							c.Target = strings.Replace(c.Target, fmt.Sprintf("$%d", i), value, -1)
	//						}
	//						conf.Output[j] = c
	//					}
	//				}
	//				plugin.handler.Transform(s.StreamPath, conf)
	//			}
	//		}
	//	}
	//}
}

func (p *Plugin) PublishWithConfig(ctx context.Context, streamPath string, conf config.Publish) (publisher *Publisher, err error) {
	publisher = createPublisher(p, streamPath, conf)
	if p.config.EnableAuth && publisher.Type == PublishTypeServer {
		onAuthPub := p.Meta.OnAuthPub
		if onAuthPub == nil {
			onAuthPub = p.Server.Meta.OnAuthPub
		}
		if onAuthPub != nil {
			if err = onAuthPub(publisher).Await(); err != nil {
				p.Warn("auth failed", "error", err)
				return
			}
		} else if conf.Key != "" {
			if err = p.auth(publisher.StreamPath, conf.Key, publisher.Args.Get("secret"), publisher.Args.Get("expire")); err != nil {
				p.Warn("auth failed", "error", err)
				return
			}
		}
	}
	err = p.Server.Streams.AddTask(publisher, ctx).WaitStarted()
	if err == nil {
		if sender, webhook := p.getHookSender(config.HookOnPublishEnd); sender != nil {
			publisher.OnDispose(func() {
				alarmInfo := AlarmInfo{
					AlarmName:  string(config.HookOnPublishEnd),
					AlarmDesc:  publisher.StopReason().Error(),
					AlarmType:  config.AlarmPublishOffline,
					StreamPath: publisher.StreamPath,
				}
				sender(webhook, alarmInfo)
			})
		}
		if sender, webhook := p.getHookSender(config.HookOnPublishStart); sender != nil {
			alarmInfo := AlarmInfo{
				AlarmName:  string(config.HookOnPublishStart),
				AlarmType:  config.AlarmPublishRecover,
				StreamPath: publisher.StreamPath,
			}
			sender(webhook, alarmInfo)
		}
	}
	return
}

func (p *Plugin) Publish(ctx context.Context, streamPath string) (publisher *Publisher, err error) {
	return p.PublishWithConfig(ctx, streamPath, p.config.Publish)
}

func (p *Plugin) SubscribeWithConfig(ctx context.Context, streamPath string, conf config.Subscribe) (subscriber *Subscriber, err error) {
	subscriber = createSubscriber(p, streamPath, conf)
	if p.config.EnableAuth && subscriber.Type == SubscribeTypeServer {
		onAuthSub := p.Meta.OnAuthSub
		if onAuthSub == nil {
			onAuthSub = p.Server.Meta.OnAuthSub
		}
		if onAuthSub != nil {
			if err = onAuthSub(subscriber).Await(); err != nil {
				p.Warn("auth failed", "error", err)
				return
			}
		} else if conf.Key != "" {
			if err = p.auth(subscriber.StreamPath, conf.Key, subscriber.Args.Get("secret"), subscriber.Args.Get("expire")); err != nil {
				p.Warn("auth failed", "error", err)
				return
			}
		}
	}
	err = p.Server.Streams.AddTask(subscriber, ctx).WaitStarted()
	if err == nil {
		select {
		case <-subscriber.waitPublishDone:
			waitAudio := conf.WaitTrack == "all" || strings.Contains(conf.WaitTrack, "audio")
			waitVideo := conf.WaitTrack == "all" || strings.Contains(conf.WaitTrack, "video")
			err = subscriber.Publisher.WaitTrack(waitAudio, waitVideo)
		case <-subscriber.Done():
			err = subscriber.StopReason()
		}
	}
	if err == nil {
		if sender, webhook := p.getHookSender(config.HookOnSubscribeEnd); sender != nil {
			subscriber.OnDispose(func() {
				alarmInfo := AlarmInfo{
					AlarmName:  string(config.HookOnSubscribeEnd),
					AlarmDesc:  subscriber.StopReason().Error(),
					AlarmType:  config.AlarmSubscribeOffline,
					StreamPath: subscriber.StreamPath,
				}
				sender(webhook, alarmInfo)
			})
		}
		if sender, webhook := p.getHookSender(config.HookOnSubscribeStart); sender != nil {
			alarmInfo := AlarmInfo{
				AlarmName:  string(config.HookOnSubscribeStart),
				AlarmType:  config.AlarmSubscribeRecover,
				StreamPath: subscriber.StreamPath,
			}
			sender(webhook, alarmInfo)
		}
	}
	return
}

func (p *Plugin) Subscribe(ctx context.Context, streamPath string) (subscriber *Subscriber, err error) {
	return p.SubscribeWithConfig(ctx, streamPath, p.config.Subscribe)
}

func (p *Plugin) Pull(streamPath string, conf config.Pull, pubConf *config.Publish) (job *PullJob, err error) {
	puller := p.Meta.NewPuller(conf)
	if puller == nil {
		return nil, ErrNotFound
	}
	job = puller.GetPullJob()
	job.Init(puller, p, streamPath, conf, pubConf)
	return
}

func (p *Plugin) Push(streamPath string, conf config.Push, subConf *config.Subscribe) {
	pusher := p.Meta.NewPusher()
	pusher.GetPushJob().Init(pusher, p, streamPath, conf, subConf)
}

func (p *Plugin) Record(pub *Publisher, conf config.Record, subConf *config.Subscribe) *RecordJob {
	recorder := p.Meta.NewRecorder(conf)
	job := recorder.GetRecordJob().Init(recorder, p, pub.StreamPath, conf, subConf)
	job.Depend(pub)
	return job
}

func (p *Plugin) Transform(pub *Publisher, conf config.Transform) {
	transformer := p.Meta.NewTransformer()
	job := transformer.GetTransformJob().Init(transformer, p, pub, conf)
	job.Depend(pub)
}

func (p *Plugin) registerHandler(handlers map[string]http.HandlerFunc) {
	t := reflect.TypeOf(p.handler)
	v := reflect.ValueOf(p.handler)
	// 注册http响应
	for i, j := 0, t.NumMethod(); i < j; i++ {
		name := t.Method(i).Name
		if name == "ServeHTTP" {
			continue
		}
		switch handler := v.Method(i).Interface().(type) {
		case func(http.ResponseWriter, *http.Request):
			patten := strings.ToLower(strings.ReplaceAll(name, "_", "/"))
			p.handle(patten, http.HandlerFunc(handler))
		}
	}
	for patten, handler := range handlers {
		p.handle(patten, handler)
	}
	if p.config.EnableAuth && p.Server.ServerConfig.Admin.EnableLogin {
		p.handle("/api/secret/{type}/{streamPath...}", http.HandlerFunc(func(rw http.ResponseWriter, r *http.Request) {
			authHeader := r.Header.Get("Authorization")
			if authHeader == "" {
				http.Error(rw, "missing authorization header", http.StatusUnauthorized)
				return
			}

			tokenString := strings.TrimPrefix(authHeader, "Bearer ")
			_, err := p.Server.ValidateToken(tokenString)
			if err != nil {
				http.Error(rw, "invalid token", http.StatusUnauthorized)
				return
			}
			streamPath := r.PathValue("streamPath")
			t := r.PathValue("type")
			expire := r.URL.Query().Get("expire")
			switch t {
			case "publish":
				secret := md5.Sum([]byte(p.config.Publish.Key + streamPath + expire))
				rw.Write([]byte(hex.EncodeToString(secret[:])))
			case "subscribe":
				secret := md5.Sum([]byte(p.config.Subscribe.Key + streamPath + expire))
				rw.Write([]byte(hex.EncodeToString(secret[:])))
			}
		}))
	}
	if rootHandler, ok := p.handler.(http.Handler); ok {
		p.handle("/", rootHandler)
	}
}

func (p *Plugin) logHandler(handler http.Handler) http.Handler {
	return http.HandlerFunc(func(rw http.ResponseWriter, r *http.Request) {
		p.Debug("visit", "path", r.URL.String(), "remote", r.RemoteAddr)
		name := strings.ToLower(p.Meta.Name)
		r.URL.Path = strings.TrimPrefix(r.URL.Path, "/"+name)
		handler.ServeHTTP(rw, r)
	})
}

func (p *Plugin) handle(pattern string, handler http.Handler) {
	if p == nil {
		return
	}
	last := pattern == "/"
	if !strings.HasPrefix(pattern, "/") {
		pattern = "/" + pattern
	}
	handler = p.logHandler(handler)
	p.config.HTTP.Handle(pattern, handler, last)
	if p.Server != p.handler {
		pattern = "/" + strings.ToLower(p.Meta.Name) + pattern
		p.Debug("http handle added to Server", "pattern", pattern)
		p.Server.config.HTTP.Handle(pattern, handler, last)
	}
	p.Server.apiList = append(p.Server.apiList, pattern)
}

func (p *Plugin) getHookSender(hookType config.HookType) (sender func(webhook config.Webhook, data any) *task.Task, conf config.Webhook) {
	if p.config.Hook != nil {
		if _, ok := p.config.Hook[hookType]; ok {
			sender = p.SendWebhook
			conf = p.config.Hook[hookType]
		} else if _, ok := p.config.Hook[config.HookDefault]; ok {
			sender = p.SendWebhook
			conf = p.config.Hook[config.HookDefault]
		} else if p.Server.config.Hook != nil {
			if _, ok := p.Server.config.Hook[hookType]; ok {
				conf = p.config.Hook[hookType]
				sender = p.Server.SendWebhook
			} else if _, ok := p.Server.config.Hook[config.HookDefault]; ok {
				sender = p.Server.SendWebhook
				conf = p.config.Hook[config.HookDefault]
			}
		}
	}
	return
}

type ServerKeepAliveTask struct {
	task.TickTask
	plugin *Plugin
}

func (t *ServerKeepAliveTask) GetTickInterval() time.Duration {
	return time.Duration(t.plugin.config.Hook[config.HookOnServerKeepAlive].Interval) * time.Second
}

func (t *ServerKeepAliveTask) Tick(now any) {
	sender, webhook := t.plugin.getHookSender(config.HookOnServerKeepAlive)
	if sender == nil {
		return
	}
	//s := t.plugin.Server
	alarmInfo := AlarmInfo{
		AlarmName:  string(config.HookOnServerKeepAlive),
		AlarmType:  config.AlarmKeepAliveOnline,
		StreamPath: "",
	}
	sender(webhook, alarmInfo)
	//webhookData := map[string]interface{}{
	//	"event":           config.HookOnServerKeepAlive,
	//	"timestamp":       time.Now().Unix(),
	//	"streams":         s.Streams.Length,
	//	"subscribers":     s.Subscribers.Length,
	//	"publisherCount":  s.Streams.Length,
	//	"subscriberCount": s.Subscribers.Length,
	//	"uptime":          time.Since(s.StartTime).Seconds(),
	//}
	//sender(webhook, webhookData)
}