Opt: backend client

internal/bootstrap/vendor.go

@@ -12,10 +12,9 @@ func InitVendorBackend(ctx context.Context) error {
 	if err != nil {
 		return err
 	}
-	b, err := vendor.NewBackends(ctx, vb)
+	bc, err := vendor.NewBackendConns(ctx, vb)
 	if err != nil {
 		return err
 	}
-	vendor.StoreBackends(b)
-	return nil
+	return vendor.StoreConns(bc)
 }

internal/vendor/alist.go

@@ -13,7 +13,7 @@ import (
 type AlistInterface = alist.AlistHTTPServer
 
 func LoadAlistClient(name string) AlistInterface {
-	if cli, ok := backends.Load().alist[name]; ok {
+	if cli, ok := clients.Load().alist[name]; ok {
 		return cli
 	}
 	return alistLocalClient

internal/vendor/bilibili.go

@@ -13,7 +13,7 @@ import (
 type BilibiliInterface = bilibili.BilibiliHTTPServer
 
 func LoadBilibiliClient(name string) BilibiliInterface {
-	if cli, ok := backends.Load().bilibili[name]; ok {
+	if cli, ok := clients.Load().bilibili[name]; ok {
 		return cli
 	}
 	return bilibiliLocalClient

internal/vendor/emby.go

@@ -13,7 +13,7 @@ import (
 type EmbyInterface = emby.EmbyHTTPServer
 
 func LoadEmbyClient(name string) EmbyInterface {
-	if cli, ok := backends.Load().emby[name]; ok && cli != nil {
+	if cli, ok := clients.Load().emby[name]; ok && cli != nil {
 		return cli
 	}
 	return embyLocalClient

internal/vendor/vendor.go

@@ -35,115 +35,131 @@ func init() {
 	selector.SetGlobalSelector(wrr.NewBuilder())
 }
 
-var backends atomic.Pointer[Backends]
+var (
+	conns   atomic.Value
+	clients atomic.Pointer[VendorClients]
+)
 
-type BackendConnInfo struct {
-	Conn *grpc.ClientConn
-	Info *model.VendorBackend
+func LoadClients() *VendorClients {
+	return clients.Load()
 }
 
-type Backends struct {
-	conns    map[string]*BackendConnInfo
-	bilibili map[string]BilibiliInterface
-	alist    map[string]AlistInterface
-	emby     map[string]EmbyInterface
+func storeClients(b *VendorClients) {
+	clients.Store(b)
 }
 
-func (b *Backends) Close() {
-	for _, conn := range b.conns {
-		conn.Conn.Close()
+func LoadConns() map[string]*BackendConn {
+	return conns.Load().(map[string]*BackendConn)
+}
+
+func StoreConns(c map[string]*BackendConn) error {
+	vc, err := newVendorClients(c)
+	if err != nil {
+		return err
 	}
+	conns.Store(c)
+	storeClients(vc)
+	return nil
 }
 
-func (b *Backends) Conns() map[string]*BackendConnInfo {
-	return b.conns
+type BackendConn struct {
+	Conn *grpc.ClientConn
+	Info *model.VendorBackend
 }
 
-func (b *Backends) BilibiliClients() map[string]BilibiliInterface {
+type VendorClients struct {
+	bilibili map[string]BilibiliInterface
+	alist    map[string]AlistInterface
+	emby     map[string]EmbyInterface
+}
+
+func (b *VendorClients) BilibiliClients() map[string]BilibiliInterface {
 	return b.bilibili
 }
 
-func (b *Backends) AlistClients() map[string]AlistInterface {
+func (b *VendorClients) AlistClients() map[string]AlistInterface {
 	return b.alist
 }
 
-func (b *Backends) EmbyClients() map[string]EmbyInterface {
+func (b *VendorClients) EmbyClients() map[string]EmbyInterface {
 	return b.emby
 }
 
-func NewBackends(ctx context.Context, conf []*model.VendorBackend) (backends *Backends, err 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),
+func NewBackendConn(ctx context.Context, conf *model.VendorBackend) (conns *BackendConn, err error) {
+	cc, err := NewGrpcClientConn(ctx, &conf.Backend)
+	if err != nil {
+		return conns, err
 	}
+	return &BackendConn{
+		Conn: cc,
+		Info: conf,
+	}, nil
+}
+
+func NewBackendConns(ctx context.Context, conf []*model.VendorBackend) (conns map[string]*BackendConn, err error) {
+	conns = make(map[string]*BackendConn, len(conf))
 	defer func() {
 		if err != nil {
-			backends.Close()
+			for endpoint, conn := range conns {
+				delete(conns, endpoint)
+				conn.Conn.Close()
+			}
 		}
 	}()
 	for _, vb := range conf {
-		if _, ok := newConns[vb.Backend.Endpoint]; ok {
-			return backends, fmt.Errorf("duplicate endpoint: %s", vb.Backend.Endpoint)
+		if _, ok := conns[vb.Backend.Endpoint]; ok {
+			return conns, fmt.Errorf("duplicate endpoint: %s", vb.Backend.Endpoint)
 		}
-		cc, err := NewGrpcClientConn(ctx, &vb.Backend)
+		cc, err := NewBackendConn(ctx, vb)
 		if err != nil {
-			return backends, err
-		}
-		newConns[vb.Backend.Endpoint] = &BackendConnInfo{
-			Conn: cc,
-			Info: vb,
+			return conns, err
 		}
-		if vb.UsedBy.Bilibili {
-			if _, ok := backends.bilibili[vb.UsedBy.BilibiliBackendName]; ok {
-				return backends, fmt.Errorf("duplicate bilibili backend name: %s", vb.UsedBy.BilibiliBackendName)
+		conns[vb.Backend.Endpoint] = cc
+	}
+
+	return conns, nil
+}
+
+func newVendorClients(conns map[string]*BackendConn) (*VendorClients, error) {
+	clients := &VendorClients{
+		bilibili: make(map[string]BilibiliInterface),
+		alist:    make(map[string]AlistInterface),
+		emby:     make(map[string]EmbyInterface),
+	}
+	for _, conn := range conns {
+		if conn.Info.UsedBy.Bilibili {
+			if _, ok := clients.bilibili[conn.Info.UsedBy.BilibiliBackendName]; ok {
+				return nil, fmt.Errorf("duplicate bilibili backend name: %s", conn.Info.UsedBy.BilibiliBackendName)
 			}
-			cli, err := NewBilibiliGrpcClient(cc)
+			cli, err := NewBilibiliGrpcClient(conn.Conn)
 			if err != nil {
-				return backends, err
+				return nil, err
 			}
-			backends.bilibili[vb.UsedBy.BilibiliBackendName] = cli
+			clients.bilibili[conn.Info.UsedBy.BilibiliBackendName] = cli
 		}
-		if vb.UsedBy.Alist {
-			if _, ok := backends.alist[vb.UsedBy.AlistBackendName]; ok {
-				return backends, fmt.Errorf("duplicate alist backend name: %s", vb.UsedBy.AlistBackendName)
+		if conn.Info.UsedBy.Alist {
+			if _, ok := clients.alist[conn.Info.UsedBy.AlistBackendName]; ok {
+				return nil, fmt.Errorf("duplicate alist backend name: %s", conn.Info.UsedBy.AlistBackendName)
 			}
-			cli, err := NewAlistGrpcClient(cc)
+			cli, err := NewAlistGrpcClient(conn.Conn)
 			if err != nil {
-				return backends, err
+				return nil, err
 			}
-			backends.alist[vb.UsedBy.AlistBackendName] = cli
+			clients.alist[conn.Info.UsedBy.AlistBackendName] = cli
 		}
-		if vb.UsedBy.Emby {
-			if _, ok := backends.emby[vb.UsedBy.EmbyBackendName]; ok {
-				return backends, fmt.Errorf("duplicate emby backend name: %s", vb.UsedBy.EmbyBackendName)
+		if conn.Info.UsedBy.Emby {
+			if _, ok := clients.emby[conn.Info.UsedBy.EmbyBackendName]; ok {
+				return nil, fmt.Errorf("duplicate emby backend name: %s", conn.Info.UsedBy.EmbyBackendName)
 			}
-			cli, err := NewEmbyGrpcClient(cc)
+			cli, err := NewEmbyGrpcClient(conn.Conn)
 			if err != nil {
-				return backends, err
+				return nil, err
 			}
-			backends.emby[vb.UsedBy.EmbyBackendName] = cli
+			clients.emby[conn.Info.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)
-	}
+	return clients, nil
 }
 
 func NewGrpcClientConn(ctx context.Context, conf *model.Backend) (*grpc.ClientConn, error) {

server/handlers/admin.go

@@ -1,9 +1,9 @@
 package handlers
 
 import (
+	"fmt"
 	"net/http"
 	"reflect"
-	"sync"
 
 	"github.com/gin-gonic/gin"
 	"github.com/synctv-org/synctv/internal/bootstrap"
@@ -13,6 +13,7 @@ import (
 	"github.com/synctv-org/synctv/internal/settings"
 	"github.com/synctv-org/synctv/internal/vendor"
 	"github.com/synctv-org/synctv/server/model"
+	"golang.org/x/exp/maps"
 	"gorm.io/gorm"
 )
 
@@ -705,7 +706,7 @@ func AdminRoomPassword(ctx *gin.Context) {
 func AdminGetVendorBackends(ctx *gin.Context) {
 	// user := ctx.MustGet("user").(*op.User)
 
-	conns := vendor.LoadBackends().Conns()
+	conns := vendor.LoadConns()
 	resp := make([]*model.GetVendorBackendResp, 0, len(conns))
 	for _, conn := range conns {
 		resp = append(resp, &model.GetVendorBackendResp{
@@ -717,8 +718,6 @@ func AdminGetVendorBackends(ctx *gin.Context) {
 	ctx.JSON(http.StatusOK, model.NewApiDataResp(resp))
 }
 
-var vendorBackendLock sync.Mutex
-
 func AdminAddVendorBackends(ctx *gin.Context) {
 	// user := ctx.MustGet("user").(*op.User)
 
@@ -728,52 +727,54 @@ func AdminAddVendorBackends(ctx *gin.Context) {
 		return
 	}
 
-	if !vendorBackendLock.TryLock() {
-		ctx.AbortWithStatusJSON(http.StatusConflict, model.NewApiErrorStringResp("vendor backend is updating"))
+	raw := vendor.LoadConns()
+	if _, ok := raw[req.Backend.Endpoint]; ok {
+		ctx.AbortWithStatusJSON(http.StatusBadRequest, model.NewApiErrorStringResp("duplicate endpoint"))
 		return
 	}
-	defer vendorBackendLock.Unlock()
 
-	vb, err := db.GetAllVendorBackend()
+	bc, err := vendor.NewBackendConn(ctx, (*dbModel.VendorBackend)(&req))
 	if err != nil {
-		ctx.AbortWithStatusJSON(http.StatusInternalServerError, model.NewApiErrorResp(err))
+		ctx.AbortWithStatusJSON(http.StatusBadRequest, model.NewApiErrorResp(err))
 		return
 	}
 
-	vb = append(vb, (*dbModel.VendorBackend)(&req))
+	m := maps.Clone(raw)
+	m[req.Backend.Endpoint] = bc
 
-	backends, err := vendor.NewBackends(ctx, vb)
+	err = vendor.StoreConns(m)
 	if err != nil {
+		bc.Conn.Close()
 		ctx.AbortWithStatusJSON(http.StatusBadRequest, model.NewApiErrorResp(err))
 		return
 	}
 
 	err = db.CreateVendorBackend((*dbModel.VendorBackend)(&req))
 	if err != nil {
-		backends.Close()
+		bc.Conn.Close()
 		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
+	var req model.VendorBackendEndpointsReq
 	if err := model.Decode(ctx, &req); err != nil {
 		ctx.AbortWithStatusJSON(http.StatusBadRequest, model.NewApiErrorResp(err))
 		return
 	}
 
-	if !vendorBackendLock.TryLock() {
-		ctx.AbortWithStatusJSON(http.StatusConflict, model.NewApiErrorStringResp("vendor backend is updating"))
-		return
+	raw := vendor.LoadConns()
+	for _, v := range req.Endpoints {
+		if _, ok := raw[v]; !ok {
+			ctx.AbortWithStatusJSON(http.StatusBadRequest, model.NewApiErrorStringResp(fmt.Sprintf("endpoint %s not found", v)))
+			return
+		}
 	}
-	defer vendorBackendLock.Unlock()
 
 	err := db.DeleteVendorBackends(req.Endpoints)
 	if err != nil {
@@ -781,19 +782,24 @@ func AdminDeleteVendorBackends(ctx *gin.Context) {
 		return
 	}
 
-	vb, err := db.GetAllVendorBackend()
-	if err != nil {
-		ctx.AbortWithStatusJSON(http.StatusInternalServerError, model.NewApiErrorResp(err))
-		return
+	m := maps.Clone(raw)
+
+	var deletedConn = make([]*vendor.BackendConn, len(req.Endpoints))
+
+	for i, v := range req.Endpoints {
+		deletedConn[i] = m[v]
+		delete(m, v)
 	}
 
-	backends, err := vendor.NewBackends(ctx, vb)
+	err = vendor.StoreConns(m)
 	if err != nil {
 		ctx.AbortWithStatusJSON(http.StatusBadRequest, model.NewApiErrorResp(err))
 		return
 	}
 
-	vendor.StoreBackends(backends)
+	for _, v := range deletedConn {
+		v.Conn.Close()
+	}
 
 	ctx.Status(http.StatusNoContent)
 }
@@ -807,39 +813,40 @@ func AdminUpdateVendorBackends(ctx *gin.Context) {
 		return
 	}
 
-	if !vendorBackendLock.TryLock() {
-		ctx.AbortWithStatusJSON(http.StatusConflict, model.NewApiErrorStringResp("vendor backend is updating"))
+	var beforeConn *vendor.BackendConn
+
+	raw := vendor.LoadConns()
+	if c, ok := raw[req.Backend.Endpoint]; !ok {
+		ctx.AbortWithStatusJSON(http.StatusBadRequest, model.NewApiErrorStringResp("endpoint not found"))
 		return
+	} else {
+		beforeConn = c
 	}
-	defer vendorBackendLock.Unlock()
 
-	vb, err := db.GetAllVendorBackend()
+	bc, err := vendor.NewBackendConn(ctx, (*dbModel.VendorBackend)(&req))
 	if err != nil {
-		ctx.AbortWithStatusJSON(http.StatusInternalServerError, model.NewApiErrorResp(err))
+		ctx.AbortWithStatusJSON(http.StatusBadRequest, model.NewApiErrorResp(err))
 		return
 	}
 
-	for i, vb2 := range vb {
-		if vb2.Backend.Endpoint == req.Backend.Endpoint {
-			vb[i] = (*dbModel.VendorBackend)(&req)
-			break
-		}
-	}
+	m := maps.Clone(raw)
+	m[req.Backend.Endpoint] = bc
 
-	backends, err := vendor.NewBackends(ctx, vb)
+	err = vendor.StoreConns(m)
 	if err != nil {
+		bc.Conn.Close()
 		ctx.AbortWithStatusJSON(http.StatusBadRequest, model.NewApiErrorResp(err))
 		return
 	}
 
+	beforeConn.Conn.Close()
+
 	err = db.SaveVendorBackend((*dbModel.VendorBackend)(&req))
 	if err != nil {
-		backends.Close()
+		bc.Conn.Close()
 		ctx.AbortWithStatusJSON(http.StatusBadRequest, model.NewApiErrorResp(err))
 		return
 	}
 
-	vendor.StoreBackends(backends)
-
 	ctx.Status(http.StatusNoContent)
 }

server/handlers/vendors/vendors.go

@@ -14,11 +14,11 @@ func Backends(ctx *gin.Context) {
 	var backends []string
 	switch ctx.Param("vendor") {
 	case dbModel.VendorBilibili:
-		backends = maps.Keys(vendor.LoadBackends().BilibiliClients())
+		backends = maps.Keys(vendor.LoadClients().BilibiliClients())
 	case dbModel.VendorAlist:
-		backends = maps.Keys(vendor.LoadBackends().AlistClients())
+		backends = maps.Keys(vendor.LoadClients().AlistClients())
 	case dbModel.VendorEmby:
-		backends = maps.Keys(vendor.LoadBackends().EmbyClients())
+		backends = maps.Keys(vendor.LoadClients().EmbyClients())
 	default:
 		ctx.AbortWithStatusJSON(http.StatusBadRequest, model.NewApiErrorStringResp("invalid vendor name"))
 		return

server/model/admin.go

@@ -149,17 +149,17 @@ func (avbr *AddVendorBackendReq) Decode(ctx *gin.Context) error {
 	return json.NewDecoder(ctx.Request.Body).Decode(avbr)
 }
 
-type DeleteVendorBackendsReq struct {
+type VendorBackendEndpointsReq struct {
 	Endpoints []string `json:"endpoints"`
 }
 
-func (dvbr *DeleteVendorBackendsReq) Validate() error {
+func (dvbr *VendorBackendEndpointsReq) Validate() error {
 	if len(dvbr.Endpoints) == 0 {
 		return errors.New("endpoints is empty")
 	}
 	return nil
 }
 
-func (dvbr *DeleteVendorBackendsReq) Decode(ctx *gin.Context) error {
+func (dvbr *VendorBackendEndpointsReq) Decode(ctx *gin.Context) error {
 	return json.NewDecoder(ctx.Request.Body).Decode(dvbr)
 }