gb28181/internal/web/api/api.go

package api

import (
	"context"
	"expvar"
	"fmt"
	"log/slog"
	"net/http"
	"net/http/httputil"
	"net/url"
	"path/filepath"
	"runtime"
	"runtime/debug"
	"sort"
	"strings"
	"time"

	"github.com/gin-contrib/cors"
	"github.com/gin-contrib/gzip"
	"github.com/gin-gonic/gin"
	"github.com/gowvp/owl/internal/core/sms"
	"github.com/gowvp/owl/pkg/ota"
	"github.com/gowvp/owl/plugin/stat"
	"github.com/gowvp/owl/plugin/stat/statapi"
	"github.com/ixugo/goddd/domain/version/versionapi"
	"github.com/ixugo/goddd/pkg/system"
	"github.com/ixugo/goddd/pkg/web"
)

var startRuntime = time.Now()

func setupRouter(r *gin.Engine, uc *Usecase) {
	uc.GB28181API.uc = uc
	uc.SMSAPI.uc = uc
	uc.WebHookAPI.uc = uc
	const staticPrefix = "/web"

	go stat.LoadTop(system.Getwd(), func(m map[string]any) {
		_ = m
	})
	r.Use(
		// 格式化输出到控制台，然后记录到日志
		// 此处不做 recover，底层 http.server 也会 recover，但不会输出方便查看的格式
		gin.CustomRecovery(func(c *gin.Context, err any) {
			slog.ErrorContext(c.Request.Context(), "panic", "err", err, "stack", string(debug.Stack()))
			c.AbortWithStatus(http.StatusInternalServerError)
		}),
		web.Metrics(),
		web.Logger(web.IgnorePrefix(staticPrefix),
			web.IgnoreMethod(http.MethodOptions),
			web.IgnorePrefix("/events/image"),
		),
		web.LoggerWithBody(web.DefaultBodyLimit,
			web.IgnoreBool(uc.Conf.Debug),
			web.IgnoreMethod(http.MethodOptions),
			web.IgnorePrefix("/events/image"),
		),
	)
	go web.CountGoroutines(10*time.Minute, 20)

	r.Use(cors.New(cors.Config{
		AllowMethods: []string{"GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"},
		AllowHeaders: []string{
			"Accept", "Content-Length", "Content-Type", "Range", "Accept-Language",
			"Origin", "Authorization",
			"Accept-Encoding",
			"Cache-Control", "Pragma", "X-Requested-With",
			"Sec-Fetch-Mode", "Sec-Fetch-Site", "Sec-Fetch-Dest",
			"Dnt", "X-Forwarded-For", "X-Forwarded-Proto", "X-Forwarded-Host",
			"X-Real-IP", "X-Request-ID", "X-Request-Start", "X-Request-Time",
		},
		AllowCredentials: true,
		MaxAge:           12 * time.Hour,
		AllowOriginFunc: func(_ string) bool {
			return true
		},
	}))

	const staticDir = "www"
	admin := r.Group(staticPrefix, gzip.Gzip(gzip.DefaultCompression))
	admin.Static("/", filepath.Join(system.Getwd(), staticDir))
	r.NoRoute(func(c *gin.Context) {
		// react-router 路由指向前端资源
		if strings.HasPrefix(c.Request.URL.Path, staticPrefix) {
			c.File(filepath.Join(system.Getwd(), staticDir, "index.html"))
			return
		}
		c.JSON(404, gin.H{"msg": "来到了无人的荒漠"})
	})
	// 访问根路径时重定向到前端资源
	r.GET("/", func(ctx *gin.Context) {
		ctx.Redirect(http.StatusPermanentRedirect, staticPrefix+"/"+"index.html")
	})

	auth := web.AuthMiddleware(uc.Conf.Server.HTTP.JwtSecret)
	r.GET("/health", web.WrapH(uc.getHealth))
	r.GET("/app/metrics/api", web.WrapH(uc.getMetricsAPI))
	r.GET("/app/version/check", web.WrapH(uc.checkVersion))
	r.POST("/app/upgrade", auth, uc.upgradeApp)

	versionapi.Register(r, uc.Version, auth)
	statapi.Register(r)
	registerZLMWebhookAPI(r, uc.WebHookAPI)

	registerPushAPI(r, uc.MediaAPI, auth)
	registerGB28181(r, uc.GB28181API, auth)
	registerProxy(r, uc.ProxyAPI, auth)
	registerConfig(r, uc.ConfigAPI, auth)
	registerSms(r, uc.SMSAPI, auth)
	RegisterUser(r, uc.UserAPI, auth)

	// 反向代理流媒体数据
	r.Any("/proxy/sms/*path", uc.proxySMS)

	// 注册 AI 分析服务回调接口
	registerAIWebhookAPI(r, uc.AIWebhookAPI)
	// 启动 AI 任务同步协程，每 5 分钟检测一次数据库与内存状态差异
	uc.AIWebhookAPI.StartAISyncLoop(context.Background(), uc.SMSAPI.smsCore)
	// TODO: 待补充中间件
	RegisterEvent(r, uc.EventAPI)
}

type playOutput struct {
	App    string               `json:"app"`
	Stream string               `json:"stream"`
	Items  []sms.StreamLiveAddr `json:"items"`
}

type getHealthOutput struct {
	Version   string    `json:"version"`
	StartAt   time.Time `json:"start_at"`
	GitBranch string    `json:"git_branch"`
	GitHash   string    `json:"git_hash"`
}

func (uc *Usecase) getHealth(_ *gin.Context, _ *struct{}) (getHealthOutput, error) {
	return getHealthOutput{
		Version:   uc.Conf.BuildVersion,
		GitBranch: strings.Trim(expvar.Get("git_branch").String(), `"`),
		GitHash:   strings.Trim(expvar.Get("git_hash").String(), `"`),
		StartAt:   startRuntime,
	}, nil
}

type getMetricsAPIOutput struct {
	RealTimeRequests int64  `json:"real_time_requests"` // 实时请求数
	TotalRequests    int64  `json:"total_requests"`     // 总请求数
	TotalResponses   int64  `json:"total_responses"`    // 总响应数
	RequestTop10     []KV   `json:"request_top10"`      // 请求TOP10
	StatusCodeTop10  []KV   `json:"status_code_top10"`  // 状态码TOP10
	Goroutines       any    `json:"goroutines"`         // 协程数量
	NumGC            uint32 `json:"num_gc"`             // gc 次数
	SysAlloc         uint64 `json:"sys_alloc"`          // 内存占用
	StartAt          string `json:"start_at"`           // 运行时间
}

func (uc *Usecase) getMetricsAPI(_ *gin.Context, _ *struct{}) (*getMetricsAPIOutput, error) {
	req := expvar.Get("request").(*expvar.Int).Value()
	reqs := expvar.Get("requests").(*expvar.Int).Value()
	resps := expvar.Get("responses").(*expvar.Int).Value()
	urls := expvar.Get(`requestURLs`).(*expvar.Map)
	status := expvar.Get(`statusCodes`).(*expvar.Map)
	u := sortExpvarMap(urls, 10)
	s := sortExpvarMap(status, 10)
	g := expvar.Get("goroutine_num").(expvar.Func)

	var stats runtime.MemStats
	runtime.ReadMemStats(&stats)

	return &getMetricsAPIOutput{
		RealTimeRequests: req,
		TotalRequests:    reqs,
		TotalResponses:   resps,
		RequestTop10:     u,
		StatusCodeTop10:  s,
		Goroutines:       g(),
		NumGC:            stats.NumGC,
		SysAlloc:         stats.Sys,
		StartAt:          startRuntime.Format(time.DateTime),
	}, nil
}

type KV struct {
	Key   string
	Value int64
}

func sortExpvarMap(data *expvar.Map, top int) []KV {
	kvs := make([]KV, 0, 8)
	data.Do(func(kv expvar.KeyValue) {
		kvs = append(kvs, KV{
			Key:   kv.Key,
			Value: kv.Value.(*expvar.Int).Value(),
		})
	})

	sort.Slice(kvs, func(i, j int) bool {
		return kvs[i].Value > kvs[j].Value
	})

	idx := top
	if l := len(kvs); l < top {
		idx = len(kvs)
	}
	return kvs[:idx]
}

const repoName = "gowvp/owl"

type checkVersionOutput struct {
	HasNewVersion  bool   `json:"has_new_version"`
	CurrentVersion string `json:"current_version"`
	NewVersion     string `json:"new_version"`
	Description    string `json:"description"`
}

// checkVersion 检查是否有新版本
// 通过 GitHub API 获取最新 release 信息，与当前版本比较
func (uc *Usecase) checkVersion(_ *gin.Context, _ *struct{}) (checkVersionOutput, error) {
	currentVersion := uc.Conf.BuildVersion
	newVersion, body, err := ota.GetLastVersion(repoName)
	if err != nil {
		return checkVersionOutput{}, err
	}

	hasNew := compareVersion(currentVersion, newVersion) < 0

	return checkVersionOutput{
		HasNewVersion:  hasNew,
		CurrentVersion: currentVersion,
		NewVersion:     newVersion,
		Description:    body,
	}, nil
}

// compareVersion 比较两个版本号
// 返回值: -1 表示 v1 < v2, 0 表示相等, 1 表示 v1 > v2
func compareVersion(v1, v2 string) int {
	v1 = strings.TrimPrefix(v1, "v")
	v2 = strings.TrimPrefix(v2, "v")

	parts1 := strings.Split(v1, ".")
	parts2 := strings.Split(v2, ".")

	maxLen := len(parts1)
	if len(parts2) > maxLen {
		maxLen = len(parts2)
	}

	for i := 0; i < maxLen; i++ {
		var n1, n2 int
		if i < len(parts1) {
			fmt.Sscanf(parts1[i], "%d", &n1)
		}
		if i < len(parts2) {
			fmt.Sscanf(parts2[i], "%d", &n2)
		}
		if n1 < n2 {
			return -1
		}
		if n1 > n2 {
			return 1
		}
	}
	return 0
}

// upgradeApp 执行应用升级
// 通过 SSE 返回下载进度，下载完成后由回调决定如何升级
func (uc *Usecase) upgradeApp(c *gin.Context) {
	c.Header("Content-Type", "text/event-stream")
	c.Header("Cache-Control", "no-cache")
	c.Header("Connection", "keep-alive")
	c.Header("Access-Control-Allow-Origin", "*")

	flusher, ok := c.Writer.(http.Flusher)
	if !ok {
		c.JSON(http.StatusInternalServerError, gin.H{"msg": "不支持 SSE"})
		return
	}

	sendEvent := func(event, data string) {
		fmt.Fprintf(c.Writer, "event: %s\ndata: %s\n\n", event, data)
		flusher.Flush()
	}

	sendEvent("start", `{"msg":"开始下载升级包"}`)

	filename := "linux_amd64"
	if runtime.GOARCH == "arm64" {
		filename = "linux_arm64"
	}

	o := ota.NewOTA(repoName, filename)
	o.SetProgressCallback(func(current, total int64) {
		percent := 0
		if total > 0 {
			percent = int(current * 100 / total)
		}
		sendEvent("progress", fmt.Sprintf(`{"current":%d,"total":%d,"percent":%d}`, current, total, percent))
	})

	if err := o.Download().Error(); err != nil {
		sendEvent("error", fmt.Sprintf(`{"msg":"%s"}`, err.Error()))
		return
	}

	sendEvent("complete", `{"msg":"下载完成，请手动重启服务"}`)
}

func (uc *Usecase) proxySMS(c *gin.Context) {
	defer func() {
		_ = recover()
	}()

	rc := http.NewResponseController(c.Writer)
	exp := time.Now().AddDate(99, 0, 0)
	_ = rc.SetReadDeadline(exp)
	_ = rc.SetWriteDeadline(exp)

	path := c.Param("path")
	addr, err := url.JoinPath(fmt.Sprintf("http://%s:%d", uc.Conf.Media.IP, uc.Conf.Media.HTTPPort), path)
	if err != nil {
		web.Fail(c, err)
		return
	}
	fullAddr, _ := url.Parse(addr)
	c.Request.URL.Path = ""
	proxy := httputil.NewSingleHostReverseProxy(fullAddr)
	proxy.Director = func(req *http.Request) {
		// 设置请求的URL
		req.URL.Scheme = "http"
		req.URL.Host = fmt.Sprintf("%s:%d", uc.Conf.Media.IP, uc.Conf.Media.HTTPPort)
		req.URL.Path = path
	}
	proxy.ModifyResponse = func(r *http.Response) error {
		r.Header.Del("Access-Control-Allow-Credentials")
		r.Header.Del("Access-Control-Allow-Origin")
		if r.StatusCode >= 300 && r.StatusCode < 400 {
			if l := r.Header.Get("Location"); l != "" {
				if !strings.HasPrefix(l, "http") {
					r.Header.Set("Location", "/proxy/sms/"+strings.TrimPrefix(l, "/"))
				}
			}
		}
		return nil
	}
	proxy.ServeHTTP(c.Writer, c.Request)
}