feat: now slave server can sync options with master server (close #88)

README.md

@@ -48,26 +48,27 @@ _✨ All in one 的 OpenAI 接口，整合各种 API 访问方式，开箱即用
    + [x] OpenAI 官方通道
    + [x] **Azure OpenAI API**
    + [x] [API2D](https://api2d.com/r/197971)
+   + [x] [OhMyGPT](https://aigptx.top?aff=uFpUl2Kf)
    + [x] [CloseAI](https://console.openai-asia.com)
    + [x] [OpenAI-SB](https://openai-sb.com)
    + [x] [OpenAI Max](https://openaimax.com)
-   + [x] [OhMyGPT](https://www.ohmygpt.com)
    + [x] 自定义渠道：例如使用自行搭建的 OpenAI 代理
 2. 支持通过**负载均衡**的方式访问多个渠道。
 3. 支持 **stream 模式**，可以通过流式传输实现打字机效果。
-4. 支持**令牌管理**，设置令牌的过期时间和使用次数。
-5. 支持**兑换码管理**，支持批量生成和导出兑换码，可使用兑换码为令牌进行充值。
-6. 支持**通道管理**，批量创建通道。
-7. 支持发布公告，设置充值链接，设置新用户初始额度。
-8. 支持丰富的**自定义**设置，
+4. 支持**多机部署**，[详见此处](#多机部署)。
+5. 支持**令牌管理**，设置令牌的过期时间和使用次数。
+6. 支持**兑换码管理**，支持批量生成和导出兑换码，可使用兑换码为令牌进行充值。
+7. 支持**通道管理**，批量创建通道。
+8. 支持发布公告，设置充值链接，设置新用户初始额度。
+9. 支持丰富的**自定义**设置，
    1. 支持自定义系统名称，logo 以及页脚。
    2. 支持自定义首页和关于页面，可以选择使用 HTML & Markdown 代码进行自定义，或者使用一个单独的网页通过 iframe 嵌入。
-9. 支持通过系统访问令牌访问管理 API。
-10. 支持用户管理，支持**多种用户登录注册方式**：
+10. 支持通过系统访问令牌访问管理 API。
+11. 支持用户管理，支持**多种用户登录注册方式**：
     + 邮箱登录注册以及通过邮箱进行密码重置。
     + [GitHub 开放授权](https://github.com/settings/applications/new)。
     + 微信公众号授权（需要额外部署 [WeChat Server](https://github.com/songquanpeng/wechat-server)）。
-11. 未来其他大模型开放 API 后，将第一时间支持，并将其封装成同样的 API 访问方式。
+12. 未来其他大模型开放 API 后，将第一时间支持，并将其封装成同样的 API 访问方式。
 
 ## 部署
 ### 基于 Docker 进行部署
@@ -133,6 +134,14 @@ sudo service nginx restart
 
 更加详细的部署教程[参见此处](https://iamazing.cn/page/how-to-deploy-a-website)。
 
+### 多机部署
+1. 所有服务器 `SESSION_SECRET` 设置一样的值。
+2. 必须设置 `SQL_DSN`，使用 MySQL 数据库而非 SQLite，请执行配置主备数据库同步。
+3. 所有从服务器必须设置 `SYNC_FREQUENCY`，以定期从数据库同步配置。
+4. 从服务器可以选择设置 `FRONTEND_BASE_URL`，以重定向页面请求到主服务器。‘
+
+环境变量的具体使用方法详见[此处](#环境变量)。
+
 ## 配置
 系统本身开箱即用。
 
@@ -157,6 +166,10 @@ sudo service nginx restart
    + 例子：`SESSION_SECRET=random_string`
 3. `SQL_DSN`：设置之后将使用指定数据库而非 SQLite。
    + 例子：`SQL_DSN=root:123456@tcp(localhost:3306)/one-api`
+4. `FRONTEND_BASE_URL`：设置之后将使用指定的前端地址，而非后端地址。
+   + 例子：`FRONTEND_BASE_URL=https://openai.justsong.cn`
+5. `SYNC_FREQUENCY`：设置之后将定期与数据库同步配置，单位为秒，未设置则不进行同步。
+   + 例子：`SYNC_FREQUENCY=60`
 
 ### 命令行参数
 1. `--port <port_number>`: 指定服务器监听的端口号，默认为 `3000`。

controller/channel.go

@@ -201,7 +201,7 @@ func testChannel(channel *model.Channel, request *ChatRequest) error {
 	if err != nil {
 		return err
 	}
-	if response.Error.Type != "" {
+	if response.Error.Message != "" {
 		return errors.New(fmt.Sprintf("type %s, code %s, message %s", response.Error.Type, response.Error.Code, response.Error.Message))
 	}
 	return nil
@@ -265,14 +265,14 @@ var testAllChannelsLock sync.Mutex
 var testAllChannelsRunning bool = false
 
 // disable & notify
-func disableChannel(channelId int, channelName string, err error) {
+func disableChannel(channelId int, channelName string, reason string) {
 	if common.RootUserEmail == "" {
 		common.RootUserEmail = model.GetRootUserEmail()
 	}
 	model.UpdateChannelStatusById(channelId, common.ChannelStatusDisabled)
 	subject := fmt.Sprintf("通道「%s」（#%d）已被禁用", channelName, channelId)
-	content := fmt.Sprintf("通道「%s」（#%d）已被禁用，原因：%s", channelName, channelId, err.Error())
-	err = common.SendEmail(subject, common.RootUserEmail, content)
+	content := fmt.Sprintf("通道「%s」（#%d）已被禁用，原因：%s", channelName, channelId, reason)
+	err := common.SendEmail(subject, common.RootUserEmail, content)
 	if err != nil {
 		common.SysError(fmt.Sprintf("发送邮件失败：%s", err.Error()))
 	}
@@ -312,7 +312,7 @@ func testAllChannels(c *gin.Context) error {
 				if milliseconds > disableThreshold {
 					err = errors.New(fmt.Sprintf("响应时间 %.2fs 超过阈值 %.2fs", float64(milliseconds)/1000.0, float64(disableThreshold)/1000.0))
 				}
-				disableChannel(channel.Id, channel.Name, err)
+				disableChannel(channel.Id, channel.Name, err.Error())
 			}
 			channel.UpdateResponseTime(milliseconds)
 		}

controller/relay.go

@@ -4,7 +4,6 @@ import (
 	"bufio"
 	"bytes"
 	"encoding/json"
-	"errors"
 	"fmt"
 	"github.com/gin-gonic/gin"
 	"github.com/pkoukk/tiktoken-go"
@@ -47,6 +46,11 @@ type OpenAIError struct {
 	Code    string `json:"code"`
 }
 
+type OpenAIErrorWithStatusCode struct {
+	OpenAIError
+	StatusCode int `json:"status_code"`
+}
+
 type TextResponse struct {
 	Usage `json:"usage"`
 	Error OpenAIError `json:"error"`
@@ -71,21 +75,33 @@ func countToken(text string) int {
 func Relay(c *gin.Context) {
 	err := relayHelper(c)
 	if err != nil {
-		c.JSON(http.StatusOK, gin.H{
-			"error": gin.H{
-				"message": err.Error(),
-				"type":    "one_api_error",
-			},
+		c.JSON(err.StatusCode, gin.H{
+			"error": err.OpenAIError,
 		})
-		if common.AutomaticDisableChannelEnabled {
+		channelId := c.GetInt("channel_id")
+		common.SysError(fmt.Sprintf("Relay error (channel #%d): %s", channelId, err.Message))
+		if err.Type != "invalid_request_error" && err.StatusCode != http.StatusTooManyRequests &&
+			common.AutomaticDisableChannelEnabled {
 			channelId := c.GetInt("channel_id")
 			channelName := c.GetString("channel_name")
-			disableChannel(channelId, channelName, err)
+			disableChannel(channelId, channelName, err.Message)
 		}
 	}
 }
 
-func relayHelper(c *gin.Context) error {
+func errorWrapper(err error, code string, statusCode int) *OpenAIErrorWithStatusCode {
+	openAIError := OpenAIError{
+		Message: err.Error(),
+		Type:    "one_api_error",
+		Code:    code,
+	}
+	return &OpenAIErrorWithStatusCode{
+		OpenAIError: openAIError,
+		StatusCode:  statusCode,
+	}
+}
+
+func relayHelper(c *gin.Context) *OpenAIErrorWithStatusCode {
 	channelType := c.GetInt("channel")
 	tokenId := c.GetInt("token_id")
 	consumeQuota := c.GetBool("consume_quota")
@@ -93,15 +109,15 @@ func relayHelper(c *gin.Context) error {
 	if consumeQuota || channelType == common.ChannelTypeAzure {
 		requestBody, err := io.ReadAll(c.Request.Body)
 		if err != nil {
-			return err
+			return errorWrapper(err, "read_request_body_failed", http.StatusBadRequest)
 		}
 		err = c.Request.Body.Close()
 		if err != nil {
-			return err
+			return errorWrapper(err, "close_request_body_failed", http.StatusBadRequest)
 		}
 		err = json.Unmarshal(requestBody, &textRequest)
 		if err != nil {
-			return err
+			return errorWrapper(err, "unmarshal_request_body_failed", http.StatusBadRequest)
 		}
 		// Reset request body
 		c.Request.Body = io.NopCloser(bytes.NewBuffer(requestBody))
@@ -144,12 +160,12 @@ func relayHelper(c *gin.Context) error {
 	if consumeQuota {
 		err := model.PreConsumeTokenQuota(tokenId, preConsumedQuota)
 		if err != nil {
-			return err
+			return errorWrapper(err, "pre_consume_token_quota_failed", http.StatusOK)
 		}
 	}
 	req, err := http.NewRequest(c.Request.Method, fullRequestURL, c.Request.Body)
 	if err != nil {
-		return err
+		return errorWrapper(err, "new_request_failed", http.StatusOK)
 	}
 	if channelType == common.ChannelTypeAzure {
 		key := c.Request.Header.Get("Authorization")
@@ -164,15 +180,15 @@ func relayHelper(c *gin.Context) error {
 	client := &http.Client{}
 	resp, err := client.Do(req)
 	if err != nil {
-		return err
+		return errorWrapper(err, "do_request_failed", http.StatusOK)
 	}
 	err = req.Body.Close()
 	if err != nil {
-		return err
+		return errorWrapper(err, "close_request_body_failed", http.StatusOK)
 	}
 	err = c.Request.Body.Close()
 	if err != nil {
-		return err
+		return errorWrapper(err, "close_request_body_failed", http.StatusOK)
 	}
 	var textResponse TextResponse
 	isStream := strings.HasPrefix(resp.Header.Get("Content-Type"), "text/event-stream")
@@ -257,50 +273,60 @@ func relayHelper(c *gin.Context) error {
 		})
 		err = resp.Body.Close()
 		if err != nil {
-			return err
+			return errorWrapper(err, "close_response_body_failed", http.StatusOK)
 		}
 		return nil
 	} else {
-		for k, v := range resp.Header {
-			c.Writer.Header().Set(k, v[0])
-		}
 		if consumeQuota {
 			responseBody, err := io.ReadAll(resp.Body)
 			if err != nil {
-				return err
+				return errorWrapper(err, "read_response_body_failed", http.StatusOK)
 			}
 			err = resp.Body.Close()
 			if err != nil {
-				return err
+				return errorWrapper(err, "close_response_body_failed", http.StatusOK)
 			}
 			err = json.Unmarshal(responseBody, &textResponse)
 			if err != nil {
-				return err
+				return errorWrapper(err, "unmarshal_response_body_failed", http.StatusOK)
 			}
 			if textResponse.Error.Type != "" {
-				return errors.New(fmt.Sprintf("type %s, code %s, message %s",
-					textResponse.Error.Type, textResponse.Error.Code, textResponse.Error.Message))
+				return &OpenAIErrorWithStatusCode{
+					OpenAIError: textResponse.Error,
+					StatusCode:  resp.StatusCode,
+				}
 			}
 			// Reset response body
 			resp.Body = io.NopCloser(bytes.NewBuffer(responseBody))
 		}
+		// We shouldn't set the header before we parse the response body, because the parse part may fail.
+		// And then we will have to send an error response, but in this case, the header has already been set.
+		// So the client will be confused by the response.
+		// For example, Postman will report error, and we cannot check the response at all.
+		for k, v := range resp.Header {
+			c.Writer.Header().Set(k, v[0])
+		}
+		c.Writer.WriteHeader(resp.StatusCode)
 		_, err = io.Copy(c.Writer, resp.Body)
 		if err != nil {
-			return err
+			return errorWrapper(err, "copy_response_body_failed", http.StatusOK)
 		}
 		err = resp.Body.Close()
 		if err != nil {
-			return err
+			return errorWrapper(err, "close_response_body_failed", http.StatusOK)
 		}
 		return nil
 	}
 }
 
 func RelayNotImplemented(c *gin.Context) {
+	err := OpenAIError{
+		Message: "API not implemented",
+		Type:    "one_api_error",
+		Param:   "",
+		Code:    "api_not_implemented",
+	}
 	c.JSON(http.StatusOK, gin.H{
-		"error": gin.H{
-			"message": "Not Implemented",
-			"type":    "one_api_error",
-		},
+		"error": err,
 	})
 }

main.go

@@ -47,6 +47,13 @@ func main() {
 
 	// Initialize options
 	model.InitOptionMap()
+	if os.Getenv("SYNC_FREQUENCY") != "" {
+		frequency, err := strconv.Atoi(os.Getenv("SYNC_FREQUENCY"))
+		if err != nil {
+			common.FatalLog(err)
+		}
+		go model.SyncOptions(frequency)
+	}
 
 	// Initialize HTTP server
 	server := gin.Default()

model/option.go

@@ -4,6 +4,7 @@ import (
 	"one-api/common"
 	"strconv"
 	"strings"
+	"time"
 )
 
 type Option struct {
@@ -59,6 +60,10 @@ func InitOptionMap() {
 	common.OptionMap["ModelRatio"] = common.ModelRatio2JSONString()
 	common.OptionMap["TopUpLink"] = common.TopUpLink
 	common.OptionMapRWMutex.Unlock()
+	loadOptionsFromDatabase()
+}
+
+func loadOptionsFromDatabase() {
 	options, _ := AllOption()
 	for _, option := range options {
 		err := updateOptionMap(option.Key, option.Value)
@@ -68,6 +73,14 @@ func InitOptionMap() {
 	}
 }
 
+func SyncOptions(frequency int) {
+	for {
+		time.Sleep(time.Duration(frequency) * time.Second)
+		common.SysLog("Syncing options from database")
+		loadOptionsFromDatabase()
+	}
+}
+
 func UpdateOption(key string, value string) error {
 	// Save to database first
 	option := Option{

router/main.go

@@ -2,12 +2,24 @@ package router
 
 import (
 	"embed"
+	"fmt"
 	"github.com/gin-gonic/gin"
+	"net/http"
+	"os"
+	"strings"
 )
 
 func SetRouter(router *gin.Engine, buildFS embed.FS, indexPage []byte) {
 	SetApiRouter(router)
 	SetDashboardRouter(router)
 	SetRelayRouter(router)
-	setWebRouter(router, buildFS, indexPage)
+	frontendBaseUrl := os.Getenv("FRONTEND_BASE_URL")
+	if frontendBaseUrl == "" {
+		SetWebRouter(router, buildFS, indexPage)
+	} else {
+		frontendBaseUrl = strings.TrimSuffix(frontendBaseUrl, "/")
+		router.NoRoute(func(c *gin.Context) {
+			c.Redirect(http.StatusMovedPermanently, fmt.Sprintf("%s%s", frontendBaseUrl, c.Request.RequestURI))
+		})
+	}
 }

router/web-router.go

@@ -10,7 +10,7 @@ import (
 	"one-api/middleware"
 )
 
-func setWebRouter(router *gin.Engine, buildFS embed.FS, indexPage []byte) {
+func SetWebRouter(router *gin.Engine, buildFS embed.FS, indexPage []byte) {
 	router.Use(gzip.Gzip(gzip.DefaultCompression))
 	router.Use(middleware.GlobalWebRateLimit())
 	router.Use(middleware.Cache())