diff --git a/cmd/server.go b/cmd/server.go index 1555315..3cbcbdb 100644 --- a/cmd/server.go +++ b/cmd/server.go @@ -32,7 +32,7 @@ var ServerCmd = &cobra.Command{ bootstrap.InitProvider, bootstrap.InitOp, bootstrap.InitRtmp, - bootstrap.InitVendor, + bootstrap.InitVendorBackend, bootstrap.InitSetting, ) if !flags.DisableUpdateCheck { diff --git a/go.mod b/go.mod index f88a8bf..db5df66 100644 --- a/go.mod +++ b/go.mod @@ -28,7 +28,7 @@ require ( github.com/sirupsen/logrus v1.9.3 github.com/soheilhy/cmux v0.1.5 github.com/spf13/cobra v1.8.0 - github.com/synctv-org/vendors v0.1.1-0.20231212054308-d2740988c951 + github.com/synctv-org/vendors v0.1.1-0.20231213112456-9743b943a97c github.com/ulule/limiter/v3 v3.11.2 github.com/zencoder/go-dash/v3 v3.0.3 github.com/zijiren233/gencontainer v0.0.0-20231213075414-f7f4c8261dca diff --git a/go.sum b/go.sum index 66aa428..b0297a9 100644 --- a/go.sum +++ b/go.sum @@ -356,6 +356,8 @@ github.com/synctv-org/vendors v0.1.1-0.20231212053257-8f9b1f19d5a3 h1:DbbFcU5wqn github.com/synctv-org/vendors v0.1.1-0.20231212053257-8f9b1f19d5a3/go.mod h1:FX5xPnIKQGcadFONDfq30OPgQcTrpZLbW/9ebbz+2qY= github.com/synctv-org/vendors v0.1.1-0.20231212054308-d2740988c951 h1:xnyL4KYIpNA5TD3c3cL9Kq2FMwpv8r+oGThWO+T9IWM= github.com/synctv-org/vendors v0.1.1-0.20231212054308-d2740988c951/go.mod h1:hzrMvLeO3iEDYiKM6dk4P6uWe3UwkTFBvYnWf/d/w1o= +github.com/synctv-org/vendors v0.1.1-0.20231213112456-9743b943a97c h1:FhIqTq7CKtqn3xvdDAxM6Po2ImcXDw2KaO/Um34KNAs= +github.com/synctv-org/vendors v0.1.1-0.20231213112456-9743b943a97c/go.mod h1:C0ZGPeF8nYsx60gePPhSrgrjrONj6F72XFHW9Mh8+vU= github.com/tv42/httpunix v0.0.0-20150427012821-b75d8614f926/go.mod h1:9ESjWnEqriFuLhtthL60Sar/7RFoluCcXsuvEwTV5KM= github.com/twitchyliquid64/golang-asm v0.15.1 h1:SU5vSMR7hnwNxj24w34ZyCi/FmDZTkS4MhqMhdFk5YI= github.com/twitchyliquid64/golang-asm v0.15.1/go.mod h1:a1lVb/DtPvCB8fslRZhAngC2+aY1QWCk3Cedj/Gdt08= diff --git a/internal/bootstrap/vendor.go b/internal/bootstrap/vendor.go index 8e70927..0536503 100644 --- a/internal/bootstrap/vendor.go +++ b/internal/bootstrap/vendor.go @@ -3,10 +3,19 @@ package bootstrap import ( "context" - "github.com/synctv-org/synctv/internal/conf" + "github.com/synctv-org/synctv/internal/db" "github.com/synctv-org/synctv/internal/vendor" ) -func InitVendor(ctx context.Context) error { - return vendor.Init(&conf.Conf.Vendor) +func InitVendorBackend(ctx context.Context) error { + vb, err := db.GetAllVendorBackend() + if err != nil { + return err + } + b, err := vendor.NewBackends(ctx, vb) + if err != nil { + return err + } + vendor.StoreBackends(b) + return nil } diff --git a/internal/cache/alist.go b/internal/cache/alist.go index dde9860..34ef92a 100644 --- a/internal/cache/alist.go +++ b/internal/cache/alist.go @@ -31,7 +31,7 @@ func NewAlistCache(userID string) *AlistUserCache { func AlistAuthorizationCacheWithConfigInitFunc(host, username, password, backend string) func(ctx context.Context, args ...string) (*AlistUserCacheData, error) { return func(ctx context.Context, args ...string) (*AlistUserCacheData, error) { - cli := vendor.AlistClient(backend) + cli := vendor.LoadAlistClient(backend) if username == "" { _, err := cli.Me(ctx, &alist.MeReq{ Host: host, @@ -86,7 +86,7 @@ func NewAlistMovieCacheInitFunc(user *AlistUserCache, movie *model.Movie) func(c if aucd.Host == "" { return nil, errors.New("not bind alist vendor") } - cli := vendor.AlistClient(movie.Base.VendorInfo.Backend) + cli := vendor.LoadAlistClient(movie.Base.VendorInfo.Backend) fg, err := cli.FsGet(ctx, &alist.FsGetReq{ Host: aucd.Host, Token: aucd.Token, diff --git a/internal/cache/bilibili.go b/internal/cache/bilibili.go index 3881174..1510b00 100644 --- a/internal/cache/bilibili.go +++ b/internal/cache/bilibili.go @@ -46,7 +46,7 @@ func BilibiliSharedMpdCacheInitFunc(ctx context.Context, movie *model.Movie, arg } else { cookies = utils.MapToHttpCookie(vendorInfo.Cookies) } - cli := vendor.BilibiliClient(movie.Base.VendorInfo.Backend) + cli := vendor.LoadBilibiliClient(movie.Base.VendorInfo.Backend) var m, hevcM *mpd.MPD biliInfo := movie.Base.VendorInfo.Bilibili switch { @@ -143,7 +143,7 @@ func BilibiliNoSharedMovieCacheInitFunc(ctx context.Context, id string, movie *m } else { cookies = utils.MapToHttpCookie(vendorInfo.Cookies) } - cli := vendor.BilibiliClient(movie.Base.VendorInfo.Backend) + cli := vendor.LoadBilibiliClient(movie.Base.VendorInfo.Backend) var u string biliInfo := movie.Base.VendorInfo.Bilibili switch { @@ -219,7 +219,7 @@ func initBilibiliSubtitleCache(ctx context.Context, movie *model.Movie, args ... } else { cookies = utils.MapToHttpCookie(vendorInfo.Cookies) } - cli := vendor.BilibiliClient(movie.Base.VendorInfo.Backend) + cli := vendor.LoadBilibiliClient(movie.Base.VendorInfo.Backend) resp, err := cli.GetSubtitles(ctx, &bilibili.GetSubtitlesReq{ Cookies: utils.HttpCookieToMap(cookies), Bvid: biliInfo.Bvid, diff --git a/internal/conf/config.go b/internal/conf/config.go index 88f2a7b..3dab1a6 100644 --- a/internal/conf/config.go +++ b/internal/conf/config.go @@ -22,9 +22,6 @@ type Config struct { // RateLimit RateLimit RateLimitConfig `yaml:"rate_limit"` - - // Vendor - Vendor VendorConfig `yaml:"vendor"` } func (c *Config) Save(file string) error { @@ -50,8 +47,5 @@ func DefaultConfig() *Config { // RateLimit RateLimit: DefaultRateLimitConfig(), - - // Vendor - Vendor: DefaultVendorConfig(), } } diff --git a/internal/conf/vendor.go b/internal/conf/vendor.go deleted file mode 100644 index 9c4eb53..0000000 --- a/internal/conf/vendor.go +++ /dev/null @@ -1,48 +0,0 @@ -package conf - -type VendorConfig struct { - Bilibili map[string]BilibiliConfig `yaml:"bilibili" hc:"default use local vendor"` - Alist map[string]AlistConfig `yaml:"alist" hc:"default use local vendor"` -} - -func DefaultVendorConfig() VendorConfig { - return VendorConfig{ - Bilibili: nil, - } -} - -type Consul struct { - Endpoint string `yaml:"endpoint"` - Token string `yaml:"token,omitempty"` - TokenFile string `yaml:"token_file,omitempty"` - PathPrefix string `yaml:"path_prefix,omitempty"` - Namespace string `yaml:"namespace,omitempty"` - Partition string `yaml:"partition,omitempty"` -} - -type Etcd struct { - Endpoints []string `yaml:"endpoints"` - Username string `yaml:"username,omitempty"` - Password string `yaml:"password,omitempty"` -} - -type VendorBase struct { - ServerName string `yaml:"server_name" hc:"if use tls and grpc, servername must set the cert server name" env:"BILIBILI_SERVER_NAME"` - Endpoint string `yaml:"endpoint" env:"BILIBILI_ENDPOINT"` - JwtSecret string `yaml:"jwt_secret" env:"BILIBILI_JWT_SECRET"` - Scheme string `yaml:"scheme" lc:"grpc | http" env:"BILIBILI_SCHEME"` - Tls bool `yaml:"tls" env:"BILIBILI_TLS"` - CustomCAFile string `yaml:"custom_ca_file,omitempty" env:"BILIBILI_CUSTOM_CA_FILE"` - TimeOut string `yaml:"time_out" env:"BILIBILI_TIME_OUT"` - - Consul Consul `yaml:"consul,omitempty" hc:"if use consul, must set the endpoint"` - Etcd Etcd `yaml:"etcd,omitempty" hc:"if use etcd, must set the endpoints"` -} - -type BilibiliConfig struct { - VendorBase `yaml:",inline"` -} - -type AlistConfig struct { - VendorBase `yaml:",inline"` -} diff --git a/internal/db/db.go b/internal/db/db.go index 92805de..b78055a 100644 --- a/internal/db/db.go +++ b/internal/db/db.go @@ -30,6 +30,7 @@ func Init(d *gorm.DB, t conf.DatabaseType) error { new(model.Movie), new(model.BilibiliVendor), new(model.AlistVendor), + new(model.VendorBackend), ) } diff --git a/internal/db/vendorBackend.go b/internal/db/vendorBackend.go new file mode 100644 index 0000000..2422398 --- /dev/null +++ b/internal/db/vendorBackend.go @@ -0,0 +1,46 @@ +package db + +import ( + "errors" + + "github.com/synctv-org/synctv/internal/model" + "gorm.io/gorm" +) + +func GetAllVendorBackend() ([]*model.VendorBackend, error) { + var backends []*model.VendorBackend + err := db.Find(&backends).Error + return backends, HandleNotFound(err, "backends") +} + +func CreateVendorBackend(backend *model.VendorBackend) error { + return db.Create(backend).Error +} + +func DeleteVendorBackend(endpoint string) error { + return db.Where("backend_endpoint = ?", endpoint).Delete(&model.VendorBackend{}).Error +} + +func DeleteVendorBackends(endpoints []string) error { + return db.Where("backend_endpoint IN ?", endpoints).Delete(&model.VendorBackend{}).Error +} + +func GetVendorBackend(endpoint string) (*model.VendorBackend, error) { + var backend model.VendorBackend + err := db.Where("backend_endpoint = ?", endpoint).First(&backend).Error + return &backend, HandleNotFound(err, "backend") +} + +func CreateOrSaveVendorBackend(backend *model.VendorBackend) (*model.VendorBackend, error) { + return backend, Transactional(func(tx *gorm.DB) error { + if err := tx.Where("backend_endpoint = ?", backend.Backend.Endpoint).First(&model.VendorBackend{}).Error; errors.Is(err, gorm.ErrRecordNotFound) { + return tx.Create(&backend).Error + } else { + return tx.Save(&backend).Error + } + }) +} + +func SaveVendorBackend(backend *model.VendorBackend) error { + return db.Save(backend).Error +} diff --git a/internal/db/vendor.go b/internal/db/vendorRecord.go similarity index 100% rename from internal/db/vendor.go rename to internal/db/vendorRecord.go diff --git a/internal/model/movie.go b/internal/model/movie.go index 8bdee99..804d4db 100644 --- a/internal/model/movie.go +++ b/internal/model/movie.go @@ -47,12 +47,12 @@ type VendorName = string const ( VendorBilibili VendorName = "bilibili" VendorAlist VendorName = "alist" + VendorEmby VendorName = "emby" ) type VendorInfo struct { Vendor VendorName `json:"vendor"` Backend string `json:"backend"` - Shared bool `gorm:"not null;default:false" json:"shared"` Bilibili *BilibiliStreamingInfo `gorm:"embedded;embeddedPrefix:bilibili_" json:"bilibili,omitempty"` Alist *AlistStreamingInfo `gorm:"embedded;embeddedPrefix:alist_" json:"alist,omitempty"` } @@ -62,6 +62,7 @@ type BilibiliStreamingInfo struct { Cid uint64 `json:"cid,omitempty"` Epid uint64 `json:"epid,omitempty"` Quality uint64 `json:"quality,omitempty"` + Shared bool `json:"shared,omitempty"` } func (b *BilibiliStreamingInfo) Validate() error { diff --git a/internal/model/vendorBackend.go b/internal/model/vendorBackend.go new file mode 100644 index 0000000..7b28163 --- /dev/null +++ b/internal/model/vendorBackend.go @@ -0,0 +1,98 @@ +package model + +import ( + "github.com/synctv-org/synctv/utils" + "gorm.io/gorm" +) + +type Consul struct { + ServerName string + Token string + TokenFile string + PathPrefix string + Namespace string + Partition string +} + +type Etcd struct { + ServerName string + Username string + Password string +} + +type Backend struct { + Endpoint string `gorm:"primaryKey" json:"endpoint"` + Comment string `gorm:"type:text" json:"comment"` + Tls bool `gorm:"default:false" json:"tls"` + JwtSecret string `json:"jwtSecret"` + CustomCAFile string `json:"customCaFile"` + TimeOut string `gorm:"default:10s" json:"timeOut"` + + Consul Consul `gorm:"embedded;embeddedPrefix:consul_" json:"consul"` + Etcd Etcd `gorm:"embedded;embeddedPrefix:etcd_" json:"etcd"` +} + +type VendorBackend struct { + Backend Backend `gorm:"embedded;embeddedPrefix:backend_" json:"backend"` + UsedBy BackendUsedBy `gorm:"embedded;embeddedPrefix:used_by_" json:"usedBy"` +} + +type BackendUsedBy struct { + Bilibili bool `gorm:"default:false" json:"bilibili"` + BilibiliBackendName string `json:"bilibiliBackendName"` + Alist bool `gorm:"default:false" json:"alist"` + AlistBackendName string `json:"alistBackendName"` + Emby bool `gorm:"default:false" json:"emby"` + EmbyBackendName string `json:"embyBackendName"` +} + +func (v *VendorBackend) BeforeSave(tx *gorm.DB) error { + key := []byte(v.Backend.Endpoint) + var err error + if v.Backend.JwtSecret != "" { + if v.Backend.JwtSecret, err = utils.CryptoToBase64([]byte(v.Backend.JwtSecret), key); err != nil { + return err + } + } + if v.Backend.Consul.Token != "" { + if v.Backend.Consul.Token, err = utils.CryptoToBase64([]byte(v.Backend.Consul.Token), key); err != nil { + return err + } + } + if v.Backend.Etcd.Password != "" { + if v.Backend.Etcd.Password, err = utils.CryptoToBase64([]byte(v.Backend.Etcd.Password), key); err != nil { + return err + } + } + return nil +} + +func (v *VendorBackend) AfterFind(tx *gorm.DB) error { + key := []byte(v.Backend.Endpoint) + var ( + err error + data []byte + ) + if v.Backend.JwtSecret != "" { + if data, err = utils.DecryptoFromBase64(v.Backend.JwtSecret, key); err != nil { + return err + } else { + v.Backend.JwtSecret = string(data) + } + } + if v.Backend.Consul.Token != "" { + if data, err = utils.DecryptoFromBase64(v.Backend.Consul.Token, key); err != nil { + return err + } else { + v.Backend.Consul.Token = string(data) + } + } + if v.Backend.Etcd.Password != "" { + if data, err = utils.DecryptoFromBase64(v.Backend.Etcd.Password, key); err != nil { + return err + } else { + v.Backend.Etcd.Password = string(data) + } + } + return nil +} diff --git a/internal/model/vendor.go b/internal/model/vendorRecord.go similarity index 100% rename from internal/model/vendor.go rename to internal/model/vendorRecord.go diff --git a/internal/vendor/alist.go b/internal/vendor/alist.go index 2871677..3575181 100644 --- a/internal/vendor/alist.go +++ b/internal/vendor/alist.go @@ -2,256 +2,41 @@ package vendor import ( "context" - "crypto/tls" - "crypto/x509" "errors" - "fmt" - "os" - "time" - "github.com/go-kratos/aegis/circuitbreaker" - "github.com/go-kratos/aegis/circuitbreaker/sre" - consul "github.com/go-kratos/kratos/contrib/registry/consul/v2" - "github.com/go-kratos/kratos/contrib/registry/etcd/v2" - "github.com/go-kratos/kratos/v2/middleware" - "github.com/go-kratos/kratos/v2/middleware/auth/jwt" - kcircuitbreaker "github.com/go-kratos/kratos/v2/middleware/circuitbreaker" "google.golang.org/grpc" - ggrpc "github.com/go-kratos/kratos/v2/transport/grpc" - "github.com/go-kratos/kratos/v2/transport/http" - jwtv4 "github.com/golang-jwt/jwt/v4" - "github.com/hashicorp/consul/api" - log "github.com/sirupsen/logrus" - "github.com/synctv-org/synctv/internal/conf" "github.com/synctv-org/vendors/api/alist" alistService "github.com/synctv-org/vendors/service/alist" - clientv3 "go.etcd.io/etcd/client/v3" ) type AlistInterface = alist.AlistHTTPServer -func AlistClient(name string) AlistInterface { - if name != "" { - if cli, ok := alistClients[name]; ok { - return cli - } +func LoadAlistClient(name string) AlistInterface { + if cli, ok := backends.Load().alist[name]; ok { + return cli } - return alistDefaultClient -} - -func AlistClients() map[string]AlistInterface { - return alistClients + return alistLocalClient } var ( - alistClients map[string]AlistInterface - alistDefaultClient AlistInterface + alistLocalClient AlistInterface ) -func InitAlistVendors(conf map[string]conf.AlistConfig) error { - if alistClients == nil { - alistClients = make(map[string]AlistInterface, len(conf)) - } - for k, vb := range conf { - cli, err := InitAlist(&vb) - if err != nil { - return err - } - if k == "" { - alistDefaultClient = cli - } else { - alistClients[k] = cli - } - } - if alistDefaultClient == nil { - alistDefaultClient = alistService.NewAlistService(nil) - } - return nil +func init() { + alistLocalClient = alistService.NewAlistService(nil) } -func InitAlist(conf *conf.AlistConfig) (AlistInterface, error) { - middlewares := []middleware.Middleware{kcircuitbreaker.Client(kcircuitbreaker.WithCircuitBreaker(func() circuitbreaker.CircuitBreaker { - return sre.NewBreaker( - sre.WithRequest(25), - sre.WithWindow(time.Second*15), - ) - }))} - - if conf.JwtSecret != "" { - key := []byte(conf.JwtSecret) - middlewares = append(middlewares, jwt.Client(func(token *jwtv4.Token) (interface{}, error) { - return key, nil - }, jwt.WithSigningMethod(jwtv4.SigningMethodHS256))) - } - - switch conf.Scheme { - case "grpc": - opts := []ggrpc.ClientOption{} - - opts = append(opts, ggrpc.WithMiddleware(middlewares...)) - - if conf.TimeOut != "" { - timeout, err := time.ParseDuration(conf.TimeOut) - if err != nil { - return nil, err - } - opts = append(opts, ggrpc.WithTimeout(timeout)) - } - - if conf.Endpoint != "" { - opts = append(opts, ggrpc.WithEndpoint(conf.Endpoint)) - log.Infof("alist client init success with endpoint: %s", conf.Endpoint) - } else if conf.Consul.Endpoint != "" { - if conf.ServerName == "" { - return nil, errors.New("alist server name is empty") - } - c := api.DefaultConfig() - c.Address = conf.Consul.Endpoint - client, err := api.NewClient(c) - if err != nil { - return nil, err - } - endpoint := fmt.Sprintf("discovery:///%s", conf.ServerName) - dis := consul.New(client) - opts = append(opts, ggrpc.WithEndpoint(endpoint), ggrpc.WithDiscovery(dis)) - log.Infof("alist client init success with consul: %s", conf.Consul.Endpoint) - } else if len(conf.Etcd.Endpoints) > 0 { - if conf.ServerName == "" { - return nil, errors.New("alist server name is empty") - } - endpoint := fmt.Sprintf("discovery:///%s", conf.ServerName) - cli, err := clientv3.New(clientv3.Config{ - Endpoints: conf.Etcd.Endpoints, - Username: conf.Etcd.Username, - Password: conf.Etcd.Password, - }) - if err != nil { - return nil, err - } - dis := etcd.New(cli) - opts = append(opts, ggrpc.WithEndpoint(endpoint), ggrpc.WithDiscovery(dis)) - log.Infof("alist client init success with etcd: %v", conf.Etcd.Endpoints) - } else { - return nil, errors.New("alist client init failed, endpoint is empty") - } - var ( - con *grpc.ClientConn - err error - ) - if conf.Tls { - var rootCAs *x509.CertPool - rootCAs, err = x509.SystemCertPool() - if err != nil { - return nil, err - } - if conf.CustomCAFile != "" { - b, err := os.ReadFile(conf.CustomCAFile) - if err != nil { - panic(err) - } - rootCAs.AppendCertsFromPEM(b) - } - opts = append(opts, ggrpc.WithTLSConfig(&tls.Config{ - RootCAs: rootCAs, - })) - - con, err = ggrpc.Dial( - context.Background(), - opts..., - ) - } else { - con, err = ggrpc.DialInsecure( - context.Background(), - opts..., - ) - } - if err != nil { - return nil, err - } - return newGrpcAlist(alist.NewAlistClient(con)), nil - case "http": - opts := []http.ClientOption{} - - opts = append(opts, http.WithMiddleware(middlewares...)) - - if conf.TimeOut != "" { - timeout, err := time.ParseDuration(conf.TimeOut) - if err != nil { - return nil, err - } - opts = append(opts, http.WithTimeout(timeout)) - } - - if conf.Tls { - rootCAs, err := x509.SystemCertPool() - if err != nil { - return nil, err - } - if conf.CustomCAFile != "" { - b, err := os.ReadFile(conf.CustomCAFile) - if err != nil { - panic(err) - } - rootCAs.AppendCertsFromPEM(b) - } - opts = append(opts, http.WithTLSConfig(&tls.Config{ - RootCAs: rootCAs, - })) - } +func AlistLocalClient() AlistInterface { + return alistLocalClient +} - if conf.Endpoint != "" { - opts = append(opts, http.WithEndpoint(conf.Endpoint)) - log.Infof("alist client init success with endpoint: %s", conf.Endpoint) - } else if conf.Consul.Endpoint != "" { - if conf.ServerName == "" { - return nil, errors.New("alist server name is empty") - } - c := api.DefaultConfig() - c.Address = conf.Consul.Endpoint - client, err := api.NewClient(c) - if err != nil { - return nil, err - } - c.Token = conf.Consul.Token - c.TokenFile = conf.Consul.TokenFile - c.PathPrefix = conf.Consul.PathPrefix - c.Namespace = conf.Consul.Namespace - c.Partition = conf.Consul.Partition - endpoint := fmt.Sprintf("discovery:///%s", conf.ServerName) - dis := consul.New(client) - opts = append(opts, http.WithEndpoint(endpoint), http.WithDiscovery(dis)) - log.Infof("alist client init success with consul: %s", conf.Consul.Endpoint) - } else if len(conf.Etcd.Endpoints) > 0 { - if conf.ServerName == "" { - return nil, errors.New("alist server name is empty") - } - endpoint := fmt.Sprintf("discovery:///%s", conf.ServerName) - cli, err := clientv3.New(clientv3.Config{ - Endpoints: conf.Etcd.Endpoints, - Username: conf.Etcd.Username, - Password: conf.Etcd.Password, - }) - if err != nil { - return nil, err - } - dis := etcd.New(cli) - opts = append(opts, http.WithEndpoint(endpoint), http.WithDiscovery(dis)) - log.Infof("alist client init success with etcd: %v", conf.Etcd.Endpoints) - } else { - return nil, errors.New("alist client init failed, endpoint is empty") - } - con, err := http.NewClient( - context.Background(), - opts..., - ) - if err != nil { - return nil, err - } - return newHTTPAlist(alist.NewAlistHTTPClient(con)), nil - default: - return nil, errors.New("unknow alist scheme") +func NewAlistGrpcClient(conn *grpc.ClientConn) (AlistInterface, error) { + if conn == nil { + return nil, errors.New("grpc client conn is nil") } + conn.GetState() + return newGrpcAlist(alist.NewAlistClient(conn)), nil } var _ AlistInterface = (*grpcAlist)(nil) @@ -260,7 +45,7 @@ type grpcAlist struct { client alist.AlistClient } -func newGrpcAlist(client alist.AlistClient) *grpcAlist { +func newGrpcAlist(client alist.AlistClient) AlistInterface { return &grpcAlist{ client: client, } @@ -285,35 +70,3 @@ func (a *grpcAlist) Login(ctx context.Context, req *alist.LoginReq) (*alist.Logi func (a *grpcAlist) Me(ctx context.Context, req *alist.MeReq) (*alist.MeResp, error) { return a.client.Me(ctx, req) } - -var _ AlistInterface = (*httpAlist)(nil) - -type httpAlist struct { - client alist.AlistHTTPClient -} - -func newHTTPAlist(client alist.AlistHTTPClient) *httpAlist { - return &httpAlist{ - client: client, - } -} - -func (a *httpAlist) FsGet(ctx context.Context, req *alist.FsGetReq) (*alist.FsGetResp, error) { - return a.client.FsGet(ctx, req) -} - -func (a *httpAlist) FsList(ctx context.Context, req *alist.FsListReq) (*alist.FsListResp, error) { - return a.client.FsList(ctx, req) -} - -func (a *httpAlist) FsOther(ctx context.Context, req *alist.FsOtherReq) (*alist.FsOtherResp, error) { - return a.client.FsOther(ctx, req) -} - -func (a *httpAlist) Login(ctx context.Context, req *alist.LoginReq) (*alist.LoginResp, error) { - return a.client.Login(ctx, req) -} - -func (a *httpAlist) Me(ctx context.Context, req *alist.MeReq) (*alist.MeResp, error) { - return a.client.Me(ctx, req) -} diff --git a/internal/vendor/bilibili.go b/internal/vendor/bilibili.go index db5ffed..74c29b3 100644 --- a/internal/vendor/bilibili.go +++ b/internal/vendor/bilibili.go @@ -2,256 +2,40 @@ package vendor import ( "context" - "crypto/tls" - "crypto/x509" "errors" - "fmt" - "os" - "time" - "github.com/go-kratos/aegis/circuitbreaker" - "github.com/go-kratos/aegis/circuitbreaker/sre" - consul "github.com/go-kratos/kratos/contrib/registry/consul/v2" - "github.com/go-kratos/kratos/contrib/registry/etcd/v2" - "github.com/go-kratos/kratos/v2/middleware" - "github.com/go-kratos/kratos/v2/middleware/auth/jwt" - kcircuitbreaker "github.com/go-kratos/kratos/v2/middleware/circuitbreaker" "google.golang.org/grpc" - ggrpc "github.com/go-kratos/kratos/v2/transport/grpc" - "github.com/go-kratos/kratos/v2/transport/http" - jwtv4 "github.com/golang-jwt/jwt/v4" - "github.com/hashicorp/consul/api" - log "github.com/sirupsen/logrus" - "github.com/synctv-org/synctv/internal/conf" "github.com/synctv-org/vendors/api/bilibili" bilibiliService "github.com/synctv-org/vendors/service/bilibili" - clientv3 "go.etcd.io/etcd/client/v3" ) type BilibiliInterface = bilibili.BilibiliHTTPServer -func BilibiliClient(name string) BilibiliInterface { - if name != "" { - if cli, ok := bilibiliClients[name]; ok { - return cli - } +func LoadBilibiliClient(name string) BilibiliInterface { + if cli, ok := backends.Load().bilibili[name]; ok { + return cli } - return bilibiliDefaultClient -} - -func BilibiliClients() map[string]BilibiliInterface { - return bilibiliClients + return bilibiliLocalClient } var ( - bilibiliClients map[string]BilibiliInterface - bilibiliDefaultClient BilibiliInterface + bilibiliLocalClient BilibiliInterface ) -func InitBilibiliVendors(conf map[string]conf.BilibiliConfig) error { - if bilibiliClients == nil { - bilibiliClients = make(map[string]BilibiliInterface, len(conf)) - } - for k, vb := range conf { - cli, err := InitBilibili(&vb) - if err != nil { - return err - } - if k == "" { - bilibiliDefaultClient = cli - } else { - bilibiliClients[k] = cli - } - } - if bilibiliDefaultClient == nil { - bilibiliDefaultClient = bilibiliService.NewBilibiliService(nil) - } - return nil +func init() { + bilibiliLocalClient = bilibiliService.NewBilibiliService(nil) } -func InitBilibili(conf *conf.BilibiliConfig) (BilibiliInterface, error) { - middlewares := []middleware.Middleware{kcircuitbreaker.Client(kcircuitbreaker.WithCircuitBreaker(func() circuitbreaker.CircuitBreaker { - return sre.NewBreaker( - sre.WithRequest(25), - sre.WithWindow(time.Second*15), - ) - }))} - - if conf.JwtSecret != "" { - key := []byte(conf.JwtSecret) - middlewares = append(middlewares, jwt.Client(func(token *jwtv4.Token) (interface{}, error) { - return key, nil - }, jwt.WithSigningMethod(jwtv4.SigningMethodHS256))) - } - - switch conf.Scheme { - case "grpc": - opts := []ggrpc.ClientOption{} - - opts = append(opts, ggrpc.WithMiddleware(middlewares...)) - - if conf.TimeOut != "" { - timeout, err := time.ParseDuration(conf.TimeOut) - if err != nil { - return nil, err - } - opts = append(opts, ggrpc.WithTimeout(timeout)) - } - - if conf.Endpoint != "" { - opts = append(opts, ggrpc.WithEndpoint(conf.Endpoint)) - log.Infof("bilibili client init success with endpoint: %s", conf.Endpoint) - } else if conf.Consul.Endpoint != "" { - if conf.ServerName == "" { - return nil, errors.New("bilibili server name is empty") - } - c := api.DefaultConfig() - c.Address = conf.Consul.Endpoint - client, err := api.NewClient(c) - if err != nil { - return nil, err - } - endpoint := fmt.Sprintf("discovery:///%s", conf.ServerName) - dis := consul.New(client) - opts = append(opts, ggrpc.WithEndpoint(endpoint), ggrpc.WithDiscovery(dis)) - log.Infof("bilibili client init success with consul: %s", conf.Consul.Endpoint) - } else if len(conf.Etcd.Endpoints) > 0 { - if conf.ServerName == "" { - return nil, errors.New("bilibili server name is empty") - } - endpoint := fmt.Sprintf("discovery:///%s", conf.ServerName) - cli, err := clientv3.New(clientv3.Config{ - Endpoints: conf.Etcd.Endpoints, - Username: conf.Etcd.Username, - Password: conf.Etcd.Password, - }) - if err != nil { - return nil, err - } - dis := etcd.New(cli) - opts = append(opts, ggrpc.WithEndpoint(endpoint), ggrpc.WithDiscovery(dis)) - log.Infof("bilibili client init success with etcd: %v", conf.Etcd.Endpoints) - } else { - return nil, errors.New("bilibili client init failed, endpoint is empty") - } - var ( - con *grpc.ClientConn - err error - ) - if conf.Tls { - var rootCAs *x509.CertPool - rootCAs, err = x509.SystemCertPool() - if err != nil { - return nil, err - } - if conf.CustomCAFile != "" { - b, err := os.ReadFile(conf.CustomCAFile) - if err != nil { - panic(err) - } - rootCAs.AppendCertsFromPEM(b) - } - opts = append(opts, ggrpc.WithTLSConfig(&tls.Config{ - RootCAs: rootCAs, - })) - - con, err = ggrpc.Dial( - context.Background(), - opts..., - ) - } else { - con, err = ggrpc.DialInsecure( - context.Background(), - opts..., - ) - } - if err != nil { - return nil, err - } - return newGrpcBilibili(bilibili.NewBilibiliClient(con)), nil - case "http": - opts := []http.ClientOption{} - - opts = append(opts, http.WithMiddleware(middlewares...)) - - if conf.TimeOut != "" { - timeout, err := time.ParseDuration(conf.TimeOut) - if err != nil { - return nil, err - } - opts = append(opts, http.WithTimeout(timeout)) - } - - if conf.Tls { - rootCAs, err := x509.SystemCertPool() - if err != nil { - return nil, err - } - if conf.CustomCAFile != "" { - b, err := os.ReadFile(conf.CustomCAFile) - if err != nil { - panic(err) - } - rootCAs.AppendCertsFromPEM(b) - } - opts = append(opts, http.WithTLSConfig(&tls.Config{ - RootCAs: rootCAs, - })) - } +func BilibiliLocalClient() BilibiliInterface { + return bilibiliLocalClient +} - if conf.Endpoint != "" { - opts = append(opts, http.WithEndpoint(conf.Endpoint)) - log.Infof("bilibili client init success with endpoint: %s", conf.Endpoint) - } else if conf.Consul.Endpoint != "" { - if conf.ServerName == "" { - return nil, errors.New("bilibili server name is empty") - } - c := api.DefaultConfig() - c.Address = conf.Consul.Endpoint - client, err := api.NewClient(c) - if err != nil { - return nil, err - } - c.Token = conf.Consul.Token - c.TokenFile = conf.Consul.TokenFile - c.PathPrefix = conf.Consul.PathPrefix - c.Namespace = conf.Consul.Namespace - c.Partition = conf.Consul.Partition - endpoint := fmt.Sprintf("discovery:///%s", conf.ServerName) - dis := consul.New(client) - opts = append(opts, http.WithEndpoint(endpoint), http.WithDiscovery(dis)) - log.Infof("bilibili client init success with consul: %s", conf.Consul.Endpoint) - } else if len(conf.Etcd.Endpoints) > 0 { - if conf.ServerName == "" { - return nil, errors.New("bilibili server name is empty") - } - endpoint := fmt.Sprintf("discovery:///%s", conf.ServerName) - cli, err := clientv3.New(clientv3.Config{ - Endpoints: conf.Etcd.Endpoints, - Username: conf.Etcd.Username, - Password: conf.Etcd.Password, - }) - if err != nil { - return nil, err - } - dis := etcd.New(cli) - opts = append(opts, http.WithEndpoint(endpoint), http.WithDiscovery(dis)) - log.Infof("bilibili client init success with etcd: %v", conf.Etcd.Endpoints) - } else { - return nil, errors.New("bilibili client init failed, endpoint is empty") - } - con, err := http.NewClient( - context.Background(), - opts..., - ) - if err != nil { - return nil, err - } - return newHTTPBilibili(bilibili.NewBilibiliHTTPClient(con)), nil - default: - return nil, errors.New("unknow bilibili scheme") +func NewBilibiliGrpcClient(conn *grpc.ClientConn) (BilibiliInterface, error) { + if conn == nil { + return nil, errors.New("grpc client conn is nil") } + return newGrpcBilibili(bilibili.NewBilibiliClient(conn)), nil } var _ BilibiliInterface = (*grpcBilibili)(nil) @@ -260,7 +44,7 @@ type grpcBilibili struct { client bilibili.BilibiliClient } -func newGrpcBilibili(client bilibili.BilibiliClient) *grpcBilibili { +func newGrpcBilibili(client bilibili.BilibiliClient) BilibiliInterface { return &grpcBilibili{ client: client, } @@ -321,71 +105,3 @@ func (g *grpcBilibili) UserInfo(ctx context.Context, in *bilibili.UserInfoReq) ( func (g *grpcBilibili) Match(ctx context.Context, in *bilibili.MatchReq) (*bilibili.MatchResp, error) { return g.client.Match(ctx, in) } - -var _ BilibiliInterface = (*httpBilibili)(nil) - -type httpBilibili struct { - client bilibili.BilibiliHTTPClient -} - -func newHTTPBilibili(client bilibili.BilibiliHTTPClient) *httpBilibili { - return &httpBilibili{ - client: client, - } -} - -func (h *httpBilibili) NewQRCode(ctx context.Context, in *bilibili.Empty) (*bilibili.NewQRCodeResp, error) { - return h.client.NewQRCode(ctx, in) -} - -func (h *httpBilibili) LoginWithQRCode(ctx context.Context, in *bilibili.LoginWithQRCodeReq) (*bilibili.LoginWithQRCodeResp, error) { - return h.client.LoginWithQRCode(ctx, in) -} - -func (h *httpBilibili) NewCaptcha(ctx context.Context, in *bilibili.Empty) (*bilibili.NewCaptchaResp, error) { - return h.client.NewCaptcha(ctx, in) -} - -func (h *httpBilibili) NewSMS(ctx context.Context, in *bilibili.NewSMSReq) (*bilibili.NewSMSResp, error) { - return h.client.NewSMS(ctx, in) -} - -func (h *httpBilibili) LoginWithSMS(ctx context.Context, in *bilibili.LoginWithSMSReq) (*bilibili.LoginWithSMSResp, error) { - return h.client.LoginWithSMS(ctx, in) -} - -func (h *httpBilibili) ParseVideoPage(ctx context.Context, in *bilibili.ParseVideoPageReq) (*bilibili.VideoPageInfo, error) { - return h.client.ParseVideoPage(ctx, in) -} - -func (h *httpBilibili) GetVideoURL(ctx context.Context, in *bilibili.GetVideoURLReq) (*bilibili.VideoURL, error) { - return h.client.GetVideoURL(ctx, in) -} - -func (h *httpBilibili) GetDashVideoURL(ctx context.Context, in *bilibili.GetDashVideoURLReq) (*bilibili.GetDashVideoURLResp, error) { - return h.client.GetDashVideoURL(ctx, in) -} - -func (h *httpBilibili) GetSubtitles(ctx context.Context, in *bilibili.GetSubtitlesReq) (*bilibili.GetSubtitlesResp, error) { - return h.client.GetSubtitles(ctx, in) -} - -func (h *httpBilibili) ParsePGCPage(ctx context.Context, in *bilibili.ParsePGCPageReq) (*bilibili.VideoPageInfo, error) { - return h.client.ParsePGCPage(ctx, in) -} - -func (h *httpBilibili) GetPGCURL(ctx context.Context, in *bilibili.GetPGCURLReq) (*bilibili.VideoURL, error) { - return h.client.GetPGCURL(ctx, in) -} - -func (h *httpBilibili) GetDashPGCURL(ctx context.Context, in *bilibili.GetDashPGCURLReq) (*bilibili.GetDashPGCURLResp, error) { - return h.client.GetDashPGCURL(ctx, in) -} - -func (h *httpBilibili) UserInfo(ctx context.Context, in *bilibili.UserInfoReq) (*bilibili.UserInfoResp, error) { - return h.client.UserInfo(ctx, in) -} - -func (h *httpBilibili) Match(ctx context.Context, in *bilibili.MatchReq) (*bilibili.MatchResp, error) { - return h.client.Match(ctx, in) -} diff --git a/internal/vendor/emby.go b/internal/vendor/emby.go new file mode 100644 index 0000000..1345950 --- /dev/null +++ b/internal/vendor/emby.go @@ -0,0 +1,76 @@ +package vendor + +import ( + "context" + "errors" + + "google.golang.org/grpc" + + "github.com/synctv-org/vendors/api/emby" + embyService "github.com/synctv-org/vendors/service/emby" +) + +type EmbyInterface = emby.EmbyHTTPServer + +func LoadEmbyClient(name string) EmbyInterface { + if cli, ok := backends.Load().emby[name]; ok && cli != nil { + return cli + } + return embyLocalClient +} + +var ( + embyLocalClient EmbyInterface +) + +func init() { + embyLocalClient = embyService.NewEmbyService(nil) +} + +func EmbyLocalClient() EmbyInterface { + return embyLocalClient +} + +func NewEmbyGrpcClient(conn *grpc.ClientConn) (EmbyInterface, error) { + if conn == nil { + return nil, errors.New("grpc client conn is nil") + } + conn.GetState() + return newGrpcEmby(emby.NewEmbyClient(conn)), nil +} + +var _ EmbyInterface = (*grpcEmby)(nil) + +type grpcEmby struct { + client emby.EmbyClient +} + +func newGrpcEmby(client emby.EmbyClient) EmbyInterface { + return &grpcEmby{ + client: client, + } +} + +func (e *grpcEmby) FsList(ctx context.Context, req *emby.FsListReq) (*emby.FsListResp, error) { + return e.client.FsList(ctx, req) +} + +func (e *grpcEmby) GetItem(ctx context.Context, req *emby.GetItemReq) (*emby.Item, error) { + return e.client.GetItem(ctx, req) +} + +func (e *grpcEmby) GetItems(ctx context.Context, req *emby.GetItemsReq) (*emby.GetItemsResp, error) { + return e.client.GetItems(ctx, req) +} + +func (e *grpcEmby) GetSystemInfo(ctx context.Context, req *emby.Empty) (*emby.SystemInfoResp, error) { + return e.client.GetSystemInfo(ctx, req) +} + +func (e *grpcEmby) Login(ctx context.Context, req *emby.LoginReq) (*emby.LoginResp, error) { + return e.client.Login(ctx, req) +} + +func (e *grpcEmby) Me(ctx context.Context, req *emby.MeReq) (*emby.MeResp, error) { + return e.client.Me(ctx, req) +} diff --git a/internal/vendor/vendor.go b/internal/vendor/vendor.go index 4eb5c79..0c2bf18 100644 --- a/internal/vendor/vendor.go +++ b/internal/vendor/vendor.go @@ -1,21 +1,329 @@ package vendor import ( + "context" + "crypto/tls" + "crypto/x509" + "errors" + "fmt" + "os" + "sync/atomic" + "time" + + "github.com/go-kratos/aegis/circuitbreaker" + "github.com/go-kratos/aegis/circuitbreaker/sre" + consul "github.com/go-kratos/kratos/contrib/registry/consul/v2" + "github.com/go-kratos/kratos/contrib/registry/etcd/v2" klog "github.com/go-kratos/kratos/v2/log" + "github.com/go-kratos/kratos/v2/middleware" + "github.com/go-kratos/kratos/v2/middleware/auth/jwt" + kcircuitbreaker "github.com/go-kratos/kratos/v2/middleware/circuitbreaker" "github.com/go-kratos/kratos/v2/selector" "github.com/go-kratos/kratos/v2/selector/wrr" + ggrpc "github.com/go-kratos/kratos/v2/transport/grpc" + "github.com/go-kratos/kratos/v2/transport/http" + jwtv4 "github.com/golang-jwt/jwt/v4" + "github.com/hashicorp/consul/api" log "github.com/sirupsen/logrus" - "github.com/synctv-org/synctv/internal/conf" + "github.com/synctv-org/synctv/internal/model" + clientv3 "go.etcd.io/etcd/client/v3" + "google.golang.org/grpc" ) -func Init(conf *conf.VendorConfig) error { +func init() { klog.SetLogger(klog.NewStdLogger(log.StandardLogger().Writer())) selector.SetGlobalSelector(wrr.NewBuilder()) - if err := InitBilibiliVendors(conf.Bilibili); err != nil { - return err +} + +var backends atomic.Pointer[Backends] + +type BackendConnInfo struct { + Conn *grpc.ClientConn + Info *model.VendorBackend +} + +type Backends struct { + conns map[string]*BackendConnInfo + bilibili map[string]BilibiliInterface + alist map[string]AlistInterface + emby map[string]EmbyInterface +} + +func (b *Backends) Conns() map[string]*BackendConnInfo { + return b.conns +} + +func (b *Backends) BilibiliClients() map[string]BilibiliInterface { + return b.bilibili +} + +func (b *Backends) AlistClients() map[string]AlistInterface { + return b.alist +} + +func (b *Backends) EmbyClients() map[string]EmbyInterface { + return b.emby +} + +func NewBackends(ctx context.Context, conf []*model.VendorBackend) (*Backends, error) { + newConns := make(map[string]*BackendConnInfo, len(conf)) + backends := &Backends{ + conns: newConns, + bilibili: make(map[string]BilibiliInterface), + alist: make(map[string]AlistInterface), + emby: make(map[string]EmbyInterface), + } + for _, vb := range conf { + cc, err := NewGrpcClientConn(ctx, &vb.Backend) + if err != nil { + return nil, err + } + if _, ok := newConns[vb.Backend.Endpoint]; ok { + return nil, fmt.Errorf("duplicate endpoint: %s", vb.Backend.Endpoint) + } + newConns[vb.Backend.Endpoint] = &BackendConnInfo{ + Conn: cc, + Info: vb, + } + if vb.UsedBy.Bilibili { + if _, ok := backends.bilibili[vb.UsedBy.BilibiliBackendName]; ok { + return nil, fmt.Errorf("duplicate bilibili backend name: %s", vb.UsedBy.BilibiliBackendName) + } + cli, err := NewBilibiliGrpcClient(cc) + if err != nil { + return nil, err + } + backends.bilibili[vb.UsedBy.BilibiliBackendName] = cli + } + if vb.UsedBy.Alist { + if _, ok := backends.alist[vb.UsedBy.AlistBackendName]; ok { + return nil, fmt.Errorf("duplicate alist backend name: %s", vb.UsedBy.AlistBackendName) + } + cli, err := NewAlistGrpcClient(cc) + if err != nil { + return nil, err + } + backends.alist[vb.UsedBy.AlistBackendName] = cli + } + if vb.UsedBy.Emby { + if _, ok := backends.emby[vb.UsedBy.EmbyBackendName]; ok { + return nil, fmt.Errorf("duplicate emby backend name: %s", vb.UsedBy.EmbyBackendName) + } + cli, err := NewEmbyGrpcClient(cc) + if err != nil { + return nil, err + } + backends.emby[vb.UsedBy.EmbyBackendName] = cli + } + } + + return backends, nil +} + +func LoadBackends() *Backends { + return backends.Load() +} + +func StoreBackends(b *Backends) { + old := backends.Swap(b) + if old == nil { + return + } + for k, conn := range old.conns { + conn.Conn.Close() + delete(old.conns, k) + } +} + +func NewGrpcClientConn(ctx context.Context, conf *model.Backend) (*grpc.ClientConn, error) { + if conf.Endpoint == "" { + return nil, errors.New("new grpc client failed, endpoint is empty") } - if err := InitAlistVendors(conf.Alist); err != nil { - return err + middlewares := []middleware.Middleware{kcircuitbreaker.Client(kcircuitbreaker.WithCircuitBreaker(func() circuitbreaker.CircuitBreaker { + return sre.NewBreaker( + sre.WithRequest(25), + sre.WithWindow(time.Second*15), + ) + }))} + + if conf.JwtSecret != "" { + key := []byte(conf.JwtSecret) + middlewares = append(middlewares, jwt.Client(func(token *jwtv4.Token) (interface{}, error) { + return key, nil + }, jwt.WithSigningMethod(jwtv4.SigningMethodHS256))) + } + + opts := []ggrpc.ClientOption{ + ggrpc.WithMiddleware(middlewares...), + // ggrpc.WithOptions(grpc.WithBlock()), + } + + if conf.TimeOut != "" { + timeout, err := time.ParseDuration(conf.TimeOut) + if err != nil { + return nil, err + } + opts = append(opts, ggrpc.WithTimeout(timeout)) + } + + if conf.Consul.ServerName != "" { + c := api.DefaultConfig() + c.Address = conf.Endpoint + c.Token = conf.Consul.Token + c.TokenFile = conf.Consul.TokenFile + c.PathPrefix = conf.Consul.PathPrefix + c.Namespace = conf.Consul.Namespace + c.Partition = conf.Consul.Partition + client, err := api.NewClient(c) + if err != nil { + return nil, err + } + endpoint := fmt.Sprintf("discovery:///%s", conf.Consul.ServerName) + dis := consul.New(client) + opts = append(opts, ggrpc.WithEndpoint(endpoint), ggrpc.WithDiscovery(dis)) + log.Infof("new grpc client with consul: %s", conf.Endpoint) + } else if conf.Etcd.ServerName != "" { + endpoint := fmt.Sprintf("discovery:///%s", conf.Etcd.ServerName) + cli, err := clientv3.New(clientv3.Config{ + Endpoints: []string{conf.Endpoint}, + Username: conf.Etcd.Username, + Password: conf.Etcd.Password, + }) + if err != nil { + return nil, err + } + dis := etcd.New(cli) + opts = append(opts, ggrpc.WithEndpoint(endpoint), ggrpc.WithDiscovery(dis)) + log.Infof("new grpc client with etcd: %v", conf.Endpoint) + } else { + opts = append(opts, ggrpc.WithEndpoint(conf.Endpoint)) + log.Infof("new grpc client with endpoint: %s", conf.Endpoint) + } + + var ( + con *grpc.ClientConn + err error + ) + if conf.Tls { + var rootCAs *x509.CertPool + rootCAs, err = x509.SystemCertPool() + if err != nil { + return nil, err + } + if conf.CustomCAFile != "" { + b, err := os.ReadFile(conf.CustomCAFile) + if err != nil { + return nil, err + } + rootCAs.AppendCertsFromPEM(b) + } + opts = append(opts, ggrpc.WithTLSConfig(&tls.Config{ + RootCAs: rootCAs, + })) + + con, err = ggrpc.Dial( + ctx, + opts..., + ) + } else { + con, err = ggrpc.DialInsecure( + ctx, + opts..., + ) + } + if err != nil { + return nil, err + } + return con, nil +} + +func NewHttpClientConn(ctx context.Context, conf *model.Backend) (*http.Client, error) { + if conf.Endpoint == "" { + return nil, errors.New("new http client failed, endpoint is empty") + } + middlewares := []middleware.Middleware{kcircuitbreaker.Client(kcircuitbreaker.WithCircuitBreaker(func() circuitbreaker.CircuitBreaker { + return sre.NewBreaker( + sre.WithRequest(25), + sre.WithWindow(time.Second*15), + ) + }))} + + if conf.JwtSecret != "" { + key := []byte(conf.JwtSecret) + middlewares = append(middlewares, jwt.Client(func(token *jwtv4.Token) (interface{}, error) { + return key, nil + }, jwt.WithSigningMethod(jwtv4.SigningMethodHS256))) + } + + opts := []http.ClientOption{ + http.WithMiddleware(middlewares...), + } + + if conf.TimeOut != "" { + timeout, err := time.ParseDuration(conf.TimeOut) + if err != nil { + return nil, err + } + opts = append(opts, http.WithTimeout(timeout)) + } + + if conf.Tls { + rootCAs, err := x509.SystemCertPool() + if err != nil { + return nil, err + } + if conf.CustomCAFile != "" { + b, err := os.ReadFile(conf.CustomCAFile) + if err != nil { + return nil, err + } + rootCAs.AppendCertsFromPEM(b) + } + opts = append(opts, http.WithTLSConfig(&tls.Config{ + RootCAs: rootCAs, + })) + } + + if conf.Consul.ServerName != "" { + c := api.DefaultConfig() + c.Address = conf.Endpoint + c.Token = conf.Consul.Token + c.TokenFile = conf.Consul.TokenFile + c.PathPrefix = conf.Consul.PathPrefix + c.Namespace = conf.Consul.Namespace + c.Partition = conf.Consul.Partition + client, err := api.NewClient(c) + if err != nil { + return nil, err + } + endpoint := fmt.Sprintf("discovery:///%s", conf.Consul.ServerName) + dis := consul.New(client) + opts = append(opts, http.WithEndpoint(endpoint), http.WithDiscovery(dis)) + log.Infof("new http client with consul: %s", conf.Endpoint) + } else if conf.Etcd.ServerName != "" { + endpoint := fmt.Sprintf("discovery:///%s", conf.Etcd.ServerName) + cli, err := clientv3.New(clientv3.Config{ + Endpoints: []string{conf.Endpoint}, + Username: conf.Etcd.Username, + Password: conf.Etcd.Password, + }) + if err != nil { + return nil, err + } + dis := etcd.New(cli) + opts = append(opts, http.WithEndpoint(endpoint), http.WithDiscovery(dis)) + log.Infof("new http client with etcd: %v", conf.Endpoint) + } else { + opts = append(opts, http.WithEndpoint(conf.Endpoint)) + log.Infof("new http client with endpoint: %s", conf.Endpoint) + } + + con, err := http.NewClient( + ctx, + opts..., + ) + if err != nil { + return nil, err } - return nil + return con, nil } diff --git a/server/handlers/admin.go b/server/handlers/admin.go index 0d0444d..59508b6 100644 --- a/server/handlers/admin.go +++ b/server/handlers/admin.go @@ -10,6 +10,7 @@ import ( dbModel "github.com/synctv-org/synctv/internal/model" "github.com/synctv-org/synctv/internal/op" "github.com/synctv-org/synctv/internal/settings" + "github.com/synctv-org/synctv/internal/vendor" "github.com/synctv-org/synctv/server/model" "gorm.io/gorm" ) @@ -699,3 +700,123 @@ func AdminRoomPassword(ctx *gin.Context) { ctx.Status(http.StatusNoContent) } + +func AdminGetVendorBackends(ctx *gin.Context) { + // user := ctx.MustGet("user").(*op.User) + + conns := vendor.LoadBackends().Conns() + resp := make([]*model.GetVendorBackendResp, 0, len(conns)) + for _, conn := range conns { + resp = append(resp, &model.GetVendorBackendResp{ + Status: conn.Conn.GetState(), + Info: conn.Info, + }) + } + + ctx.JSON(http.StatusOK, model.NewApiDataResp(resp)) +} + +func AdminAddVendorBackends(ctx *gin.Context) { + // user := ctx.MustGet("user").(*op.User) + + var req model.AddVendorBackendReq + if err := model.Decode(ctx, &req); err != nil { + ctx.AbortWithStatusJSON(http.StatusBadRequest, model.NewApiErrorResp(err)) + return + } + + vb, err := db.GetAllVendorBackend() + if err != nil { + ctx.AbortWithStatusJSON(http.StatusInternalServerError, model.NewApiErrorResp(err)) + return + } + + vb = append(vb, (*dbModel.VendorBackend)(&req)) + + backends, err := vendor.NewBackends(ctx, vb) + if err != nil { + ctx.AbortWithStatusJSON(http.StatusBadRequest, model.NewApiErrorResp(err)) + return + } + + err = db.CreateVendorBackend((*dbModel.VendorBackend)(&req)) + if err != nil { + ctx.AbortWithStatusJSON(http.StatusBadRequest, model.NewApiErrorResp(err)) + return + } + + vendor.StoreBackends(backends) + + ctx.Status(http.StatusNoContent) +} + +func AdminDeleteVendorBackends(ctx *gin.Context) { + // user := ctx.MustGet("user").(*op.User) + + var req model.DeleteVendorBackendsReq + if err := model.Decode(ctx, &req); err != nil { + ctx.AbortWithStatusJSON(http.StatusBadRequest, model.NewApiErrorResp(err)) + return + } + + err := db.DeleteVendorBackends(req.Endpoints) + if err != nil { + ctx.AbortWithStatusJSON(http.StatusBadRequest, model.NewApiErrorResp(err)) + return + } + + vb, err := db.GetAllVendorBackend() + if err != nil { + ctx.AbortWithStatusJSON(http.StatusInternalServerError, model.NewApiErrorResp(err)) + return + } + + backends, err := vendor.NewBackends(ctx, vb) + if err != nil { + ctx.AbortWithStatusJSON(http.StatusBadRequest, model.NewApiErrorResp(err)) + return + } + + vendor.StoreBackends(backends) + + ctx.Status(http.StatusNoContent) +} + +func AdminUpdateVendorBackends(ctx *gin.Context) { + // user := ctx.MustGet("user").(*op.User) + + var req model.AddVendorBackendReq + if err := model.Decode(ctx, &req); err != nil { + ctx.AbortWithStatusJSON(http.StatusBadRequest, model.NewApiErrorResp(err)) + return + } + + vb, err := db.GetAllVendorBackend() + if err != nil { + ctx.AbortWithStatusJSON(http.StatusInternalServerError, model.NewApiErrorResp(err)) + return + } + + for i, vb2 := range vb { + if vb2.Backend.Endpoint == req.Backend.Endpoint { + vb[i] = (*dbModel.VendorBackend)(&req) + break + } + } + + backends, err := vendor.NewBackends(ctx, vb) + if err != nil { + ctx.AbortWithStatusJSON(http.StatusBadRequest, model.NewApiErrorResp(err)) + return + } + + err = db.SaveVendorBackend((*dbModel.VendorBackend)(&req)) + if err != nil { + ctx.AbortWithStatusJSON(http.StatusBadRequest, model.NewApiErrorResp(err)) + return + } + + vendor.StoreBackends(backends) + + ctx.Status(http.StatusNoContent) +} diff --git a/server/handlers/init.go b/server/handlers/init.go index 4cb7cfa..3d12165 100644 --- a/server/handlers/init.go +++ b/server/handlers/init.go @@ -38,6 +38,14 @@ func Init(e *gin.Engine) { admin.POST("/settings", EditAdminSettings) + admin.GET("/vendors", AdminGetVendorBackends) + + admin.POST("/vendors", AdminAddVendorBackends) + + admin.PUT("/vendors", AdminUpdateVendorBackends) + + admin.DELETE("/vendors", AdminDeleteVendorBackends) + { user := admin.Group("/user") diff --git a/server/handlers/movie.go b/server/handlers/movie.go index 7249675..cb2d369 100644 --- a/server/handlers/movie.go +++ b/server/handlers/movie.go @@ -732,11 +732,6 @@ func proxyVendorMovie(ctx *gin.Context, movie *op.Movie) { } func parse2VendorMovie(ctx context.Context, user *op.User, room *op.Room, movie *dbModel.Movie) (err error) { - userID := user.ID - if movie.Base.VendorInfo.Shared { - userID = movie.CreatorID - } - switch movie.Base.VendorInfo.Vendor { case dbModel.VendorBilibili: opM, err := room.GetMovieByID(movie.ID) @@ -748,6 +743,10 @@ func parse2VendorMovie(ctx context.Context, user *op.User, room *op.Room, movie if err != nil { return err } + userID := user.ID + if movie.Base.VendorInfo.Bilibili.Shared { + userID = movie.CreatorID + } s, err := bmc.NoSharedMovie.LoadOrStore(ctx, userID) if err != nil { return err diff --git a/server/handlers/vendors/vendorAlist/login.go b/server/handlers/vendors/vendorAlist/login.go index b37eb9d..91058d5 100644 --- a/server/handlers/vendors/vendorAlist/login.go +++ b/server/handlers/vendors/vendorAlist/login.go @@ -42,7 +42,7 @@ func Login(ctx *gin.Context) { return } - cli := vendor.AlistClient("") + cli := vendor.LoadAlistClient("") if req.Username == "" { _, err := cli.Me(ctx, &alist.MeReq{ diff --git a/server/handlers/vendors/vendorAlist/me.go b/server/handlers/vendors/vendorAlist/me.go index c3729b4..25dfcde 100644 --- a/server/handlers/vendors/vendorAlist/me.go +++ b/server/handlers/vendors/vendorAlist/me.go @@ -17,7 +17,7 @@ type AlistMeResp = model.VendorMeResp[*alist.MeResp] func Me(ctx *gin.Context) { user := ctx.MustGet("user").(*op.User) - cli := vendor.AlistClient(ctx.Query("backend")) + cli := vendor.LoadAlistClient(ctx.Query("backend")) aucd, err := user.AlistCache().Get(ctx, ctx.Query("backend")) if err != nil { diff --git a/server/handlers/vendors/vendorAlist/parse.go b/server/handlers/vendors/vendorAlist/parse.go index d7be632..4e593d3 100644 --- a/server/handlers/vendors/vendorAlist/parse.go +++ b/server/handlers/vendors/vendorAlist/parse.go @@ -39,7 +39,7 @@ func List(ctx *gin.Context) { return } - var cli = vendor.AlistClient(ctx.Query("backend")) + var cli = vendor.LoadAlistClient(ctx.Query("backend")) aucd, err := user.AlistCache().Get(ctx, ctx.Query("backend")) if err != nil { if errors.Is(err, db.ErrNotFound("vendor")) { diff --git a/server/handlers/vendors/vendorBilibili/bilibili.go b/server/handlers/vendors/vendorBilibili/bilibili.go index 4225948..de2aeb1 100644 --- a/server/handlers/vendors/vendorBilibili/bilibili.go +++ b/server/handlers/vendors/vendorBilibili/bilibili.go @@ -39,7 +39,7 @@ func Parse(ctx *gin.Context) { return } - var cli = vendor.BilibiliClient(ctx.Query("backend")) + var cli = vendor.LoadBilibiliClient(ctx.Query("backend")) resp, err := cli.Match(ctx, &bilibili.MatchReq{ Url: req.URL, diff --git a/server/handlers/vendors/vendorBilibili/login.go b/server/handlers/vendors/vendorBilibili/login.go index a149831..0e376a0 100644 --- a/server/handlers/vendors/vendorBilibili/login.go +++ b/server/handlers/vendors/vendorBilibili/login.go @@ -15,7 +15,7 @@ import ( ) func NewQRCode(ctx *gin.Context) { - r, err := vendor.BilibiliClient("").NewQRCode(ctx, &bilibili.Empty{}) + r, err := vendor.LoadBilibiliClient("").NewQRCode(ctx, &bilibili.Empty{}) if err != nil { ctx.AbortWithStatusJSON(http.StatusInternalServerError, model.NewApiErrorResp(err)) return @@ -47,7 +47,7 @@ func LoginWithQR(ctx *gin.Context) { return } - resp, err := vendor.BilibiliClient("").LoginWithQRCode(ctx, &bilibili.LoginWithQRCodeReq{ + resp, err := vendor.LoadBilibiliClient("").LoginWithQRCode(ctx, &bilibili.LoginWithQRCodeReq{ Key: req.Key, }) if err != nil { @@ -89,7 +89,7 @@ func LoginWithQR(ctx *gin.Context) { } func NewCaptcha(ctx *gin.Context) { - r, err := vendor.BilibiliClient("").NewCaptcha(ctx, &bilibili.Empty{}) + r, err := vendor.LoadBilibiliClient("").NewCaptcha(ctx, &bilibili.Empty{}) if err != nil { ctx.AbortWithStatusJSON(http.StatusInternalServerError, model.NewApiErrorResp(err)) return @@ -131,7 +131,7 @@ func NewSMS(ctx *gin.Context) { return } - r, err := vendor.BilibiliClient("").NewSMS(ctx, &bilibili.NewSMSReq{ + r, err := vendor.LoadBilibiliClient("").NewSMS(ctx, &bilibili.NewSMSReq{ Phone: req.Telephone, Token: req.Token, Challenge: req.Challenge, @@ -176,7 +176,7 @@ func LoginWithSMS(ctx *gin.Context) { return } - c, err := vendor.BilibiliClient("").LoginWithSMS(ctx, &bilibili.LoginWithSMSReq{ + c, err := vendor.LoadBilibiliClient("").LoginWithSMS(ctx, &bilibili.LoginWithSMSReq{ Phone: req.Telephone, CaptchaKey: req.CaptchaKey, Code: req.Code, diff --git a/server/handlers/vendors/vendorBilibili/me.go b/server/handlers/vendors/vendorBilibili/me.go index f130e1e..3579450 100644 --- a/server/handlers/vendors/vendorBilibili/me.go +++ b/server/handlers/vendors/vendorBilibili/me.go @@ -33,7 +33,7 @@ func Me(ctx *gin.Context) { })) return } - resp, err := vendor.BilibiliClient("").UserInfo(ctx, &bilibili.UserInfoReq{ + resp, err := vendor.LoadBilibiliClient("").UserInfo(ctx, &bilibili.UserInfoReq{ Cookies: v.Cookies, }) if err != nil { diff --git a/server/handlers/vendors/vendors.go b/server/handlers/vendors/vendors.go index 0c12d8e..7c4fa0c 100644 --- a/server/handlers/vendors/vendors.go +++ b/server/handlers/vendors/vendors.go @@ -14,11 +14,13 @@ func Backends(ctx *gin.Context) { var backends []string switch ctx.Param("vendor") { case dbModel.VendorBilibili: - backends = maps.Keys(vendor.BilibiliClients()) + backends = maps.Keys(vendor.LoadBackends().BilibiliClients()) case dbModel.VendorAlist: - backends = maps.Keys(vendor.AlistClients()) + backends = maps.Keys(vendor.LoadBackends().AlistClients()) + case dbModel.VendorEmby: + backends = maps.Keys(vendor.LoadBackends().EmbyClients()) default: - ctx.AbortWithStatusJSON(http.StatusBadRequest, model.NewApiErrorStringResp("invalid vendor")) + ctx.AbortWithStatusJSON(http.StatusBadRequest, model.NewApiErrorStringResp("invalid vendor name")) return } ctx.JSON(http.StatusOK, model.NewApiDataResp(backends)) diff --git a/server/model/admin.go b/server/model/admin.go index 8a1769d..edf4469 100644 --- a/server/model/admin.go +++ b/server/model/admin.go @@ -5,7 +5,9 @@ import ( "github.com/gin-gonic/gin" json "github.com/json-iterator/go" + "github.com/synctv-org/synctv/internal/model" dbModel "github.com/synctv-org/synctv/internal/model" + "google.golang.org/grpc/connectivity" ) var ( @@ -128,3 +130,36 @@ func (aur *AdminRoomPasswordReq) Validate() error { func (aur *AdminRoomPasswordReq) Decode(ctx *gin.Context) error { return json.NewDecoder(ctx.Request.Body).Decode(aur) } + +type GetVendorBackendResp struct { + Info *dbModel.VendorBackend `json:"info"` + Status connectivity.State `json:"status"` +} + +type AddVendorBackendReq model.VendorBackend + +func (avbr *AddVendorBackendReq) Validate() error { + if avbr.Backend.Endpoint == "" { + return errors.New("endpoint is empty") + } + return nil +} + +func (avbr *AddVendorBackendReq) Decode(ctx *gin.Context) error { + return json.NewDecoder(ctx.Request.Body).Decode(avbr) +} + +type DeleteVendorBackendsReq struct { + Endpoints []string `json:"endpoints"` +} + +func (dvbr *DeleteVendorBackendsReq) Validate() error { + if len(dvbr.Endpoints) == 0 { + return errors.New("endpoints is empty") + } + return nil +} + +func (dvbr *DeleteVendorBackendsReq) Decode(ctx *gin.Context) error { + return json.NewDecoder(ctx.Request.Body).Decode(dvbr) +}