From f5aa377bdcf63f89be0bdb81f8a239e2cc3e995a Mon Sep 17 00:00:00 2001 From: "Haihui.Wang" Date: Thu, 25 Jun 2026 11:06:05 +0800 Subject: [PATCH] Add forward API for third-party services --- api/handler/dataflow.go | 85 ++++++++++++++++++++++--------- api/router/api.go | 18 +++---- api/router/api_forward.go | 40 +++++++++++++++ common/config/config.go | 10 ++++ common/config/config.toml.example | 8 +++ 5 files changed, 125 insertions(+), 36 deletions(-) create mode 100644 api/router/api_forward.go diff --git a/api/handler/dataflow.go b/api/handler/dataflow.go index 866767fd0..5a77ed496 100644 --- a/api/handler/dataflow.go +++ b/api/handler/dataflow.go @@ -14,6 +14,14 @@ import ( "opencsg.com/csghub-server/component" ) +type ForwardType string + +const ( + ForwardTypeDataflow ForwardType = "dataflow" + ForwardTypeAcademy ForwardType = "academy" + ForwardTypeSupplydemand ForwardType = "supplydemand" +) + type DataflowProxyHandler struct { rp proxy.ReverseProxy user component.UserComponent @@ -21,14 +29,38 @@ type DataflowProxyHandler struct { } func NewDataflowProxyHandler(config *config.Config) (*DataflowProxyHandler, error) { + return newThirdPartProxyHandler(config, ForwardTypeDataflow) +} + +func NewAcademyProxyHandler(config *config.Config) (*DataflowProxyHandler, error) { + return newThirdPartProxyHandler(config, ForwardTypeAcademy) +} + +func NewSupplydemandProxyHandler(config *config.Config) (*DataflowProxyHandler, error) { + return newThirdPartProxyHandler(config, ForwardTypeSupplydemand) +} + +func newThirdPartProxyHandler(config *config.Config, forwardType ForwardType) (*DataflowProxyHandler, error) { uc, err := component.NewUserComponent(config) if err != nil { return nil, fmt.Errorf("failed to create user component: %w", err) } - remoteEndpoint := fmt.Sprintf("%s:%d", config.Dataflow.Host, config.Dataflow.Port) + + var remoteEndpoint string + switch forwardType { + case ForwardTypeDataflow: + remoteEndpoint = fmt.Sprintf("%s:%d", config.Dataflow.Host, config.Dataflow.Port) + case ForwardTypeAcademy: + remoteEndpoint = fmt.Sprintf("%s:%d", config.Academy.Host, config.Academy.Port) + case ForwardTypeSupplydemand: + remoteEndpoint = fmt.Sprintf("%s:%d", config.Supplydemand.Host, config.Supplydemand.Port) + default: + return nil, fmt.Errorf("invalid forward type: %s", forwardType) + } + rp, err := proxy.NewReverseProxy(remoteEndpoint) if err != nil { - return nil, fmt.Errorf("failed to create dataflow reverse proxy: %w", err) + return nil, fmt.Errorf("failed to create %s reverse proxy: %w", forwardType, err) } usc := rpc.NewUserSvcHttpClient(fmt.Sprintf("%s:%d", config.User.Host, config.User.Port), rpc.AuthWithApiKey(config.APIToken)) @@ -38,30 +70,33 @@ func NewDataflowProxyHandler(config *config.Config) (*DataflowProxyHandler, erro // Proxy send request to backend service, without change the request path func (h *DataflowProxyHandler) Proxy(ctx *gin.Context) { currentUser := httpbase.GetCurrentUser(ctx) - u, err := h.user.GetUserByName(ctx.Request.Context(), currentUser) - if err != nil { - httpbase.UnauthorizedError(ctx, err) - ctx.Abort() - return - } - token, err := h.usc.GetOrCreateFirstAvaiTokens(ctx.Request.Context(), currentUser, currentUser, string(types.AccessTokenAppGit), "dataflow") - if err != nil { - - httpbase.ServerError(ctx, err) - ctx.Abort() - return - } - if len(token) == 0 { - slog.ErrorContext(ctx.Request.Context(), "fail to get or create user first git access token", slog.Any("user", currentUser), slog.Any("error", err)) - httpbase.ServerError(ctx, errors.New("can not get user first available access token")) - ctx.Abort() - return + if len(currentUser) > 0 { + u, err := h.user.GetUserByName(ctx.Request.Context(), currentUser) + if err != nil { + httpbase.UnauthorizedError(ctx, err) + ctx.Abort() + return + } + token, err := h.usc.GetOrCreateFirstAvaiTokens(ctx.Request.Context(), currentUser, currentUser, + string(types.AccessTokenAppGit), "dataflow") + if err != nil { + httpbase.ServerError(ctx, err) + ctx.Abort() + return + } + if len(token) == 0 { + slog.ErrorContext(ctx.Request.Context(), "failed to get or create user first git access token", + slog.Any("user", currentUser), slog.Any("error", err)) + httpbase.ServerError(ctx, errors.New("can not get user first available access token")) + ctx.Abort() + return + } + ctx.Request.Header.Set("User-Id", fmt.Sprintf("%d", u.ID)) + ctx.Request.Header.Set("User-Name", u.Username) + ctx.Request.Header.Set("User-Token", token) + ctx.Request.Header.Set("User-Email", u.Email) + ctx.Request.Header.Set("isadmin", fmt.Sprintf("%t", u.CanAdmin())) } - ctx.Request.Header.Set("User-Id", fmt.Sprintf("%d", u.ID)) - ctx.Request.Header.Set("User-Name", u.Username) - ctx.Request.Header.Set("User-Token", token) - ctx.Request.Header.Set("User-Email", u.Email) - ctx.Request.Header.Set("isadmin", fmt.Sprintf("%t", u.CanAdmin())) // Log the request URL and header slog.Debug("http request", slog.Any("request", ctx.Request.URL), slog.Any("header", ctx.Request.Header)) // Serve the request using the router diff --git a/api/router/api.go b/api/router/api.go index 735dd2485..48af01952 100644 --- a/api/router/api.go +++ b/api/router/api.go @@ -523,17 +523,12 @@ func NewRouter(config *config.Config, enableSwagger bool) (*gin.Engine, error) { memoryHandler := handler.NewMemoryHandler(memoryComp) createMemoryRoutes(apiGroup, middlewareCollection, memoryHandler) - // dataflow proxy - dataflowHandler, err := handler.NewDataflowProxyHandler(config) - if err != nil { - return nil, fmt.Errorf("error creating data flow proxy handler:%w", err) - } // platform dataflow platformDFHandler, err := handler.NewPlatformDataflowHandler(config) if err != nil { return nil, fmt.Errorf("error creating platform dataflow handler:%w", err) } - createDataflowRoutes(apiGroup, dataflowHandler, platformDFHandler) + createPlatformDataflowRoutes(apiGroup, platformDFHandler) // ClawHub Registry err = createClawHubRoutes(r, apiGroup, config) @@ -589,6 +584,11 @@ func NewRouter(config *config.Config, enableSwagger bool) (*gin.Engine, error) { } createFinetuneRoutes(apiGroup, middlewareCollection, finetuneJobHandler) + err = createForwardRoutes(apiGroup, config) + if err != nil { + return nil, fmt.Errorf("error creating forward routes:%w", err) + } + return r, nil } @@ -1295,11 +1295,7 @@ func createPromptRoutes( promptInfoGrp.GET("/:namespace/:name", promptHandler.PromptDetail) } -func createDataflowRoutes(apiGroup *gin.RouterGroup, dataflowHandler *handler.DataflowProxyHandler, platformDFHandler *handler.PlatformDataflowHandler) { - dataflowGrp := apiGroup.Group("/dataflow") - dataflowGrp.Use(middleware.MustLogin()) - dataflowGrp.Any("/*any", dataflowHandler.Proxy) - +func createPlatformDataflowRoutes(apiGroup *gin.RouterGroup, platformDFHandler *handler.PlatformDataflowHandler) { platformGrp := apiGroup.Group("/platform") platformDFGrp := platformGrp.Group("/dataflow") platformDFGrp.Use(middleware.MustLogin()) diff --git a/api/router/api_forward.go b/api/router/api_forward.go new file mode 100644 index 000000000..bcf8d9f8f --- /dev/null +++ b/api/router/api_forward.go @@ -0,0 +1,40 @@ +package router + +import ( + "fmt" + + "github.com/gin-gonic/gin" + "opencsg.com/csghub-server/api/handler" + "opencsg.com/csghub-server/api/middleware" + "opencsg.com/csghub-server/common/config" +) + +func createForwardRoutes(apiGroup *gin.RouterGroup, config *config.Config) error { + // dataflow proxy + dataflowHandler, err := handler.NewDataflowProxyHandler(config) + if err != nil { + return fmt.Errorf("error creating data flow proxy handler:%w", err) + } + + academyHandler, err := handler.NewAcademyProxyHandler(config) + if err != nil { + return fmt.Errorf("error creating academy proxy handler:%w", err) + } + + supplydemandHandler, err := handler.NewSupplydemandProxyHandler(config) + if err != nil { + return fmt.Errorf("error creating supplydemand proxy handler:%w", err) + } + + dataflowGrp := apiGroup.Group("/dataflow") + dataflowGrp.Use(middleware.MustLogin()) + dataflowGrp.Any("/*any", dataflowHandler.Proxy) + + academyGrp := apiGroup.Group("/academy") + academyGrp.Any("/*any", academyHandler.Proxy) + + supplydemandGrp := apiGroup.Group("/supplydemand") + supplydemandGrp.Any("/*any", supplydemandHandler.Proxy) + + return nil +} diff --git a/common/config/config.go b/common/config/config.go index da2a9357a..d03965432 100644 --- a/common/config/config.go +++ b/common/config/config.go @@ -311,6 +311,16 @@ type Config struct { Port int `env:"OPENCSG_DATAFLOW_SERVER_PORT" default:"8000"` } + Academy struct { + Host string `env:"OPENCSG_ACADEMY_SERVER_HOST" default:"http://127.0.0.1"` + Port int `env:"OPENCSG_ACADEMY_SERVER_PORT" default:"8000"` + } + + Supplydemand struct { + Host string `env:"OPENCSG_SUPPLYDEMAND_SERVER_HOST" default:"http://127.0.0.1"` + Port int `env:"OPENCSG_SUPPLYDEMAND_SERVER_PORT" default:"8000"` + } + // for csghub agents CSGBot struct { Host string `env:"OPENCSG_CSGBOT_SERVER_HOST" default:"http://127.0.0.1"` diff --git a/common/config/config.toml.example b/common/config/config.toml.example index 5deed1f4c..86e275a91 100644 --- a/common/config/config.toml.example +++ b/common/config/config.toml.example @@ -204,6 +204,14 @@ allowCreatePublicDataset = true host = "http://127.0.0.1" port = 8000 +[academy] +host = "http://127.0.0.1" +port = 8000 + +[supplydemand] +host = "http://127.0.0.1" +port = 8000 + [moderation] host = "http://localhost" port = 8089