This is an automated email from the ASF dual-hosted git repository. hsluoyz pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/casbin-gateway.git
commit 147ebfe25fedec56a3221070fb0f15ea67755bed Author: Ziyang Dong <[email protected]> AuthorDate: Thu Aug 13 13:41:25 2026 +0800 feat: add unified LLM API proxy endpoint /v1/chat/completions (#190) --- controllers/proxy.go | 353 ++++++++++++++++++++++++++++++++++++++++++++++ controllers/proxy_test.go | 260 ++++++++++++++++++++++++++++++++++ object/channel.go | 48 ++++++- routers/filter.go | 2 +- routers/router.go | 3 + 5 files changed, 664 insertions(+), 2 deletions(-) diff --git a/controllers/proxy.go b/controllers/proxy.go new file mode 100644 index 0000000..edb882c --- /dev/null +++ b/controllers/proxy.go @@ -0,0 +1,353 @@ +// Copyright 2023 The casbin Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package controllers + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net" + "net/http" + "strings" + "time" + + "github.com/apache/casbin-gateway/object" + "github.com/beego/beego" +) + +const ( + // proxyResponseHeaderTimeout bounds the wait for the upstream response + // headers. It deliberately does not cover the response body: a completion + // can legitimately take many minutes to finish streaming, so the body is + // bounded by proxyIdleTimeout instead. + proxyResponseHeaderTimeout = 60 * time.Second + // proxyIdleTimeout is the longest gap tolerated between two chunks of the + // upstream response body before the request is aborted. + proxyIdleTimeout = 120 * time.Second +) + +// hopByHopHeaders are connection-scoped, so a proxy must not pass them on. +// See RFC 7230 section 6.1. +var hopByHopHeaders = []string{ + "Connection", + "Keep-Alive", + "Proxy-Authenticate", + "Proxy-Authorization", + "Te", + "Trailer", + "Transfer-Encoding", + "Upgrade", +} + +// chatCompletionsRequest contains only the fields needed for routing. +// All other fields (messages, temperature, etc.) are forwarded as-is to the upstream. +type chatCompletionsRequest struct { + Model string `json:"model"` + Stream bool `json:"stream"` +} + +// openaiErrorResponse follows the OpenAI API error format. +type openaiErrorResponse struct { + Error openaiErrorDetail `json:"error"` +} + +type openaiErrorDetail struct { + Message string `json:"message"` + Type string `json:"type"` +} + +// proxyClient is a shared HTTP client for upstream requests. +// Reusing a single instance allows TCP connection pooling across requests. +// It has no overall Timeout on purpose, see proxyResponseHeaderTimeout. +var proxyClient = &http.Client{ + CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }, + Transport: &http.Transport{ + Proxy: http.ProxyFromEnvironment, + DialContext: (&net.Dialer{ + Timeout: 10 * time.Second, + KeepAlive: 30 * time.Second, + }).DialContext, + MaxIdleConns: 100, + MaxIdleConnsPerHost: 10, + IdleConnTimeout: 90 * time.Second, + TLSHandshakeTimeout: 10 * time.Second, + ExpectContinueTimeout: 1 * time.Second, + ResponseHeaderTimeout: proxyResponseHeaderTimeout, + }, +} + +// ChatCompletions is the OpenAI-compatible chat completions proxy endpoint. +// It matches the upstream channels by model name and forwards the request and +// response body as-is (pass-through), trying the channels in priority order +// until one of them answers. Supports SSE streaming when stream=true in the +// request body. +// This endpoint does NOT require Casdoor authentication (auth deferred to milestone 1.3). +func (c *ApiController) ChatCompletions() { + rawBody := c.Ctx.Input.RequestBody + + // Parse only model and stream; forward everything else as-is. + var req chatCompletionsRequest + if err := json.Unmarshal(rawBody, &req); err != nil { + c.writeOpenAIError(http.StatusBadRequest, "invalid_request_error", "invalid request body") + return + } + if req.Model == "" { + c.writeOpenAIError(http.StatusBadRequest, "invalid_request_error", "model is required") + return + } + + // Match the channels globally, without an owner filter. + channels, err := object.GetChannelsByModel(req.Model) + if err != nil { + if errors.Is(err, object.ErrNoChannelAvailable) { + c.writeOpenAIError(http.StatusBadRequest, "invalid_request_error", err.Error()) + } else { + beego.Error("channel lookup failed:", err) + c.writeOpenAIError(http.StatusBadGateway, "server_error", "channel lookup failed") + } + return + } + + // Drop the channels this proxy cannot talk to before forwarding, so that + // the last usable channel is known and its response can be relayed as-is. + usableChannels := []*object.Channel{} + skipReason := "" + for _, channel := range channels { + if reason := channelUnusableReason(channel); reason != "" { + beego.Error("skipped channel", channel.GetId()+":", reason) + skipReason = reason + continue + } + usableChannels = append(usableChannels, channel) + } + if len(usableChannels) == 0 { + message := fmt.Sprintf("no usable channel for model: %s", req.Model) + if skipReason != "" { + message = skipReason + } + c.writeOpenAIError(http.StatusBadGateway, "server_error", message) + return + } + + // Fail over to the next channel as long as nothing has been written to the + // client yet. The last channel is never retried, so the client gets the + // real upstream answer instead of a synthesized error. + lastStatus, lastMessage := http.StatusBadGateway, "upstream connection failed" + for i, channel := range usableChannels { + if c.Ctx.Request.Context().Err() != nil { + // The client hung up, there is nobody left to fail over for. + return + } + + status, message, written := c.forwardToChannel(channel, rawBody, req.Stream, i == len(usableChannels)-1) + if written { + return + } + lastStatus, lastMessage = status, message + } + + c.writeOpenAIError(lastStatus, "server_error", lastMessage) +} + +// forwardToChannel sends the request to a single channel's upstream. It reports +// whether the client response was already written, and when it was not, the +// status and message describing the failure so that the caller can fail over to +// the next channel. The response of the last channel is always relayed, even +// when its status would otherwise be retried. +func (c *ApiController) forwardToChannel(channel *object.Channel, rawBody []byte, stream bool, isLast bool) (int, string, bool) { + // The path is hardcoded to the OpenAI-compatible /v1/chat/completions. + upstreamUrl := strings.TrimRight(channel.BaseUrl, "/") + "/v1/chat/completions" + + // The context is cancelled when this function returns, which happens only + // after the response body has been relayed to the client. + ctx, cancel := context.WithCancel(c.Ctx.Request.Context()) + defer cancel() + + upstreamReq, err := http.NewRequestWithContext(ctx, http.MethodPost, upstreamUrl, bytes.NewReader(rawBody)) + if err != nil { + return http.StatusBadGateway, "upstream connection failed", false + } + upstreamReq.Header.Set("Content-Type", "application/json") + upstreamReq.Header.Set("Authorization", "Bearer "+channel.ApiKey) + + upstreamResp, err := proxyClient.Do(upstreamReq) + if err != nil { + if c.Ctx.Request.Context().Err() != nil { + // The client hung up mid-request, there is nothing left to answer. + return 0, "", true + } + + beego.Error("upstream request to channel", channel.GetId(), "failed:", err) + var netErr net.Error + if errors.As(err, &netErr) && netErr.Timeout() { + return http.StatusGatewayTimeout, "upstream timeout", false + } + return http.StatusBadGateway, "upstream connection failed", false + } + defer upstreamResp.Body.Close() + + if !isLast && isRetryableStatus(upstreamResp.StatusCode) { + beego.Error("channel", channel.GetId(), "returned a retryable status:", upstreamResp.Status) + // Drain a bounded amount so the connection can be pooled and reused. + _, _ = io.Copy(io.Discard, io.LimitReader(upstreamResp.Body, 4096)) + return http.StatusBadGateway, fmt.Sprintf("upstream returned %s", upstreamResp.Status), false + } + + // Abort the request when the upstream goes silent, rather than capping the + // total duration, which would cut off long but healthy streams. + body := newIdleTimeoutReader(upstreamResp.Body, proxyIdleTimeout, cancel) + defer body.Stop() + + c.relayResponse(upstreamResp, body, stream && isEventStream(upstreamResp)) + return 0, "", true +} + +// relayResponse copies the upstream status code, headers and body to the client +// without any transformation (pass-through mode). When flush is true, every +// chunk is written out as soon as it arrives so that SSE clients receive the +// events in real time. +func (c *ApiController) relayResponse(upstreamResp *http.Response, body io.Reader, flush bool) { + copyResponseHeaders(c.Ctx.ResponseWriter.Header(), upstreamResp.Header) + if flush { + c.Ctx.ResponseWriter.Header().Set("Cache-Control", "no-cache") + } + c.Ctx.ResponseWriter.WriteHeader(upstreamResp.StatusCode) + + if !flush { + if _, err := io.Copy(c.Ctx.ResponseWriter, body); err != nil { + beego.Error("proxy response copy failed:", err) + } + return + } + + buf := make([]byte, 4096) + for { + n, err := body.Read(buf) + if n > 0 { + if _, writeErr := c.Ctx.ResponseWriter.Write(buf[:n]); writeErr != nil { + beego.Error("proxy stream write failed:", writeErr) + return + } + c.Ctx.ResponseWriter.Flush() + } + if err != nil { + if err != io.EOF { + beego.Error("proxy stream read failed:", err) + } + return + } + } +} + +// copyResponseHeaders copies the upstream headers to the client, minus the +// hop-by-hop ones, which belong to the upstream connection and not to the +// response being relayed. +func copyResponseHeaders(dst http.Header, src http.Header) { + for name, values := range src { + if isHopByHopHeader(name) { + continue + } + for _, value := range values { + dst.Add(name, value) + } + } +} + +func isHopByHopHeader(name string) bool { + for _, header := range hopByHopHeaders { + if strings.EqualFold(name, header) { + return true + } + } + return false +} + +// isEventStream reports whether the upstream response really is an SSE stream. +// An upstream that rejects the request answers with a JSON body even when +// stream=true was asked for, and relaying that as text/event-stream would +// leave the client waiting for events that never come. +func isEventStream(upstreamResp *http.Response) bool { + if upstreamResp.StatusCode < 200 || upstreamResp.StatusCode > 299 { + return false + } + return strings.Contains(strings.ToLower(upstreamResp.Header.Get("Content-Type")), "text/event-stream") +} + +// isRetryableStatus reports whether another channel is worth trying. A rate +// limit or an upstream-side error is transient or specific to that channel, +// while a 4xx caused by the request itself would fail the same way everywhere. +func isRetryableStatus(statusCode int) bool { + return statusCode == http.StatusTooManyRequests || statusCode >= 500 +} + +// channelUnusableReason reports why the proxy cannot forward to a channel, or +// an empty string when it can. +func channelUnusableReason(channel *object.Channel) string { + if !object.IsChannelOpenAiCompatible(channel) { + return fmt.Sprintf("the %s channel type is not supported yet in this stage", channel.Type) + } + if channel.BaseUrl == "" { + return "channel base URL is not configured" + } + return "" +} + +// idleTimeoutReader aborts the upstream request when no data arrives for the +// given duration. It takes the place of an overall request timeout, which would +// cut off a long but healthy streaming completion. +type idleTimeoutReader struct { + reader io.Reader + timeout time.Duration + timer *time.Timer +} + +func newIdleTimeoutReader(reader io.Reader, timeout time.Duration, onIdle func()) *idleTimeoutReader { + return &idleTimeoutReader{ + reader: reader, + timeout: timeout, + timer: time.AfterFunc(timeout, onIdle), + } +} + +func (r *idleTimeoutReader) Read(p []byte) (int, error) { + n, err := r.reader.Read(p) + if n > 0 { + r.timer.Reset(r.timeout) + } + return n, err +} + +func (r *idleTimeoutReader) Stop() { + r.timer.Stop() +} + +// writeOpenAIError writes an OpenAI-compatible JSON error response +// with the given HTTP status code, error type, and message. +func (c *ApiController) writeOpenAIError(statusCode int, errType, message string) { + resp := openaiErrorResponse{ + Error: openaiErrorDetail{ + Message: message, + Type: errType, + }, + } + c.Ctx.ResponseWriter.Header().Set("Content-Type", "application/json") + c.Ctx.ResponseWriter.WriteHeader(statusCode) + if err := json.NewEncoder(c.Ctx.ResponseWriter).Encode(resp); err != nil { + beego.Error("proxy error response write failed:", err) + } +} diff --git a/controllers/proxy_test.go b/controllers/proxy_test.go new file mode 100644 index 0000000..ca2c34e --- /dev/null +++ b/controllers/proxy_test.go @@ -0,0 +1,260 @@ +// Copyright 2023 The casbin Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package controllers + +import ( + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/apache/casbin-gateway/object" + beegoContext "github.com/beego/beego/context" +) + +func newTestApiController() (*ApiController, *httptest.ResponseRecorder) { + recorder := httptest.NewRecorder() + ctx := beegoContext.NewContext() + ctx.Reset(recorder, httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader("{}"))) + + c := &ApiController{} + c.Ctx = ctx + return c, recorder +} + +func TestIsEventStream(t *testing.T) { + cases := []struct { + statusCode int + contentType string + expected bool + }{ + {200, "text/event-stream", true}, + {200, "text/event-stream; charset=utf-8", true}, + {200, "application/json", false}, + // An upstream that rejects the request answers with JSON even when + // stream=true was asked for. + {429, "application/json", false}, + {500, "text/event-stream", false}, + } + + for _, tc := range cases { + resp := &http.Response{StatusCode: tc.statusCode, Header: http.Header{"Content-Type": {tc.contentType}}} + if got := isEventStream(resp); got != tc.expected { + t.Errorf("isEventStream(%d, %s) = %v, expected %v", tc.statusCode, tc.contentType, got, tc.expected) + } + } +} + +func TestIsRetryableStatus(t *testing.T) { + for _, statusCode := range []int{429, 500, 502, 503} { + if !isRetryableStatus(statusCode) { + t.Errorf("status %d should be retryable", statusCode) + } + } + for _, statusCode := range []int{200, 400, 401, 404} { + if isRetryableStatus(statusCode) { + t.Errorf("status %d should not be retryable", statusCode) + } + } +} + +func TestChannelUnusableReason(t *testing.T) { + if reason := channelUnusableReason(&object.Channel{Type: "claude", BaseUrl: "https://example.com"}); !strings.Contains(reason, "not supported") { + t.Errorf("the claude channel type should be rejected, got: %s", reason) + } + if reason := channelUnusableReason(&object.Channel{Type: "openai", BaseUrl: ""}); !strings.Contains(reason, "base URL") { + t.Errorf("an empty base URL should be rejected, got: %s", reason) + } + if reason := channelUnusableReason(&object.Channel{Type: "custom", BaseUrl: "https://example.com"}); reason != "" { + t.Errorf("the custom channel type should be usable, got: %s", reason) + } +} + +func TestRelayResponse(t *testing.T) { + c, recorder := newTestApiController() + upstreamResp := &http.Response{ + StatusCode: 429, + Header: http.Header{ + "Content-Type": {"application/json"}, + "Connection": {"keep-alive"}, + "X-Request-Id": {"req-123"}, + "X-Ratelimit-Remaining": {"0"}, + }, + } + c.relayResponse(upstreamResp, strings.NewReader(`{"error":{"message":"slow down"}}`), false) + + if recorder.Code != 429 { + t.Errorf("status code = %d, expected 429", recorder.Code) + } + if header := recorder.Header().Get("Connection"); header != "" { + t.Errorf("the hop-by-hop Connection header was relayed: %s", header) + } + if header := recorder.Header().Get("X-Request-Id"); header != "req-123" { + t.Errorf("X-Request-Id = %s, expected req-123", header) + } + if header := recorder.Header().Get("X-Ratelimit-Remaining"); header != "0" { + t.Errorf("X-Ratelimit-Remaining = %s, expected 0", header) + } + if body := recorder.Body.String(); !strings.Contains(body, "slow down") { + t.Errorf("body = %s", body) + } +} + +func TestRelayResponseStream(t *testing.T) { + c, recorder := newTestApiController() + upstreamResp := &http.Response{ + StatusCode: 200, + Header: http.Header{ + "Content-Type": {"text/event-stream"}, + "X-Request-Id": {"req-abc"}, + "Connection": {"keep-alive"}, + }, + } + c.relayResponse(upstreamResp, strings.NewReader("data: a\n\ndata: [DONE]\n\n"), true) + + if header := recorder.Header().Get("Content-Type"); header != "text/event-stream" { + t.Errorf("Content-Type = %s, expected text/event-stream", header) + } + if header := recorder.Header().Get("Cache-Control"); header != "no-cache" { + t.Errorf("Cache-Control = %s, expected no-cache", header) + } + if header := recorder.Header().Get("X-Request-Id"); header != "req-abc" { + t.Errorf("the upstream headers were dropped, X-Request-Id = %s", header) + } + if header := recorder.Header().Get("Connection"); header != "" { + t.Errorf("the hop-by-hop Connection header was relayed: %s", header) + } + if body := recorder.Body.String(); !strings.Contains(body, "[DONE]") { + t.Errorf("body = %s", body) + } + if !recorder.Flushed { + t.Error("the stream was not flushed") + } +} + +// A stream that is slow but alive must not be cut off, no matter how long it +// lasts in total. Only a stalled upstream is aborted. +func TestIdleTimeoutReader(t *testing.T) { + idleTimeout := 150 * time.Millisecond + aborted := make(chan struct{}) + reader, writer := io.Pipe() + idleReader := newIdleTimeoutReader(reader, idleTimeout, func() { + close(aborted) + _ = writer.CloseWithError(io.ErrUnexpectedEOF) + }) + defer idleReader.Stop() + + chunkCount := 5 + go func() { + // The whole stream takes longer than the idle timeout, while every + // single gap stays below it. + for i := 0; i < chunkCount; i++ { + time.Sleep(idleTimeout * 2 / 3) + if _, err := writer.Write([]byte("data: chunk\n\n")); err != nil { + return + } + } + }() + + buf := make([]byte, 64) + for i := 0; i < chunkCount; i++ { + if _, err := idleReader.Read(buf); err != nil { + t.Fatalf("reading chunk %d failed: %s", i, err.Error()) + } + } + select { + case <-aborted: + t.Fatal("a slow but healthy stream was aborted") + default: + } + + // Nothing is written anymore, so the idle timeout has to fire. + if _, err := idleReader.Read(buf); err == nil { + t.Fatal("the read on a stalled stream returned no error") + } + select { + case <-aborted: + default: + t.Fatal("the idle timeout did not fire on a stalled stream") + } +} + +func TestForwardToChannel(t *testing.T) { + overloadedServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusServiceUnavailable) + _, _ = w.Write([]byte(`{"error":{"message":"overloaded"}}`)) + })) + defer overloadedServer.Close() + + healthyServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if header := r.Header.Get("Authorization"); header != "Bearer sk-good" { + t.Errorf("Authorization = %s, expected Bearer sk-good", header) + } + if r.URL.Path != "/v1/chat/completions" { + t.Errorf("upstream path = %s", r.URL.Path) + } + body, _ := io.ReadAll(r.Body) + if !strings.Contains(string(body), "gpt-4") { + t.Errorf("the request body was not forwarded as-is: %s", body) + } + + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"choices":[]}`)) + })) + defer healthyServer.Close() + + overloadedChannel := &object.Channel{Owner: "admin", Name: "overloaded", Type: "openai", BaseUrl: overloadedServer.URL, ApiKey: "sk-bad"} + healthyChannel := &object.Channel{Owner: "admin", Name: "healthy", Type: "openai", BaseUrl: healthyServer.URL + "/", ApiKey: "sk-good"} + rawBody := []byte(`{"model":"gpt-4","messages":[]}`) + + // A retryable status fails over instead of reaching the client. + c, recorder := newTestApiController() + statusCode, message, written := c.forwardToChannel(overloadedChannel, rawBody, false, false) + if written { + t.Fatal("a retryable status was relayed instead of failing over") + } + if statusCode != http.StatusBadGateway || !strings.Contains(message, "503") { + t.Errorf("statusCode = %d, message = %s", statusCode, message) + } + if recorder.Body.Len() != 0 { + t.Errorf("a body was written before failing over: %s", recorder.Body.String()) + } + + // The last channel is relayed as-is, even with a retryable status, so that + // the client sees the real upstream answer. + c, recorder = newTestApiController() + _, _, written = c.forwardToChannel(overloadedChannel, rawBody, false, true) + if !written || recorder.Code != http.StatusServiceUnavailable || !strings.Contains(recorder.Body.String(), "overloaded") { + t.Errorf("the last channel was not relayed: written = %v, statusCode = %d, body = %s", written, recorder.Code, recorder.Body.String()) + } + + // A healthy channel, with a trailing slash in its base URL. + c, recorder = newTestApiController() + _, _, written = c.forwardToChannel(healthyChannel, rawBody, false, true) + if !written || recorder.Code != http.StatusOK || !strings.Contains(recorder.Body.String(), "choices") { + t.Errorf("the healthy channel failed: written = %v, statusCode = %d, body = %s", written, recorder.Code, recorder.Body.String()) + } + + // stream=true, but the upstream rejected the request: the JSON error must + // not be dressed up as an SSE stream. + c, recorder = newTestApiController() + c.forwardToChannel(overloadedChannel, rawBody, true, true) + if header := recorder.Header().Get("Content-Type"); header != "application/json" { + t.Errorf("Content-Type = %s, expected application/json", header) + } +} diff --git a/object/channel.go b/object/channel.go index 8dffa1f..6762b04 100644 --- a/object/channel.go +++ b/object/channel.go @@ -27,6 +27,12 @@ import ( "github.com/xorm-io/core" ) +// ErrNoChannelAvailable is returned by GetChannelsByModel when no enabled +// channel matches the requested model name. It is a sentinel error so +// callers can distinguish "no match" (client error, HTTP 400) from +// database failures (server error, HTTP 502). +var ErrNoChannelAvailable = errors.New("no available channel") + // ApiKeyMask is what the API returns in place of a stored API key. Sending it // back in an update means "keep the existing key"; sending anything else // (including an empty string) overwrites the stored key. @@ -40,8 +46,18 @@ const ( var ( channelTypes = []string{"openai", "claude", "gemini", "custom"} channelStatuses = []string{"enabled", "disabled"} + // openAiCompatibleChannelTypes are the types whose upstream speaks the + // OpenAI HTTP API, which is the only wire format the gateway can talk so + // far. The claude and gemini types need their own request translation. + openAiCompatibleChannelTypes = []string{"openai", "custom"} ) +// IsChannelOpenAiCompatible reports whether the channel's upstream can be +// reached with an OpenAI-formatted request. +func IsChannelOpenAiCompatible(channel *Channel) bool { + return containsString(openAiCompatibleChannelTypes, channel.Type) +} + func containsString(values []string, value string) bool { for _, v := range values { if v == value { @@ -246,7 +262,7 @@ func TestChannelConnectivity(channel *Channel) (bool, int, string) { return false, 0, "the channel does not exist" } - if stored.Type == "claude" || stored.Type == "gemini" { + if !IsChannelOpenAiCompatible(stored) { return false, 0, fmt.Sprintf("the %s channel type is not supported yet in this stage", stored.Type) } @@ -291,3 +307,33 @@ func TestChannelConnectivity(channel *Channel) (bool, int, string) { _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 4096)) return resp.StatusCode >= 200 && resp.StatusCode < 300, resp.StatusCode, resp.Status } + +// GetChannelsByModel returns every enabled channel that supports the given +// model name, ordered by priority (ascending, so the lowest value comes first) +// so that the caller can fail over from one channel to the next. It queries all +// channels globally (no owner filter) because /v1/chat/completions is an +// unauthenticated public endpoint. +func GetChannelsByModel(model string) ([]*Channel, error) { + channels := []*Channel{} + err := ormer.Engine.Where("status = ?", "enabled").Asc("priority").Find(&channels) + if err != nil { + return nil, fmt.Errorf("channel query failed: %w", err) + } + + // The models are JSON-serialized into a single column, so the match cannot + // be pushed down into the query. + matchedChannels := []*Channel{} + for _, channel := range channels { + for _, channelModel := range channel.Models { + if channelModel == model { + matchedChannels = append(matchedChannels, channel) + break + } + } + } + + if len(matchedChannels) == 0 { + return nil, fmt.Errorf("%w: %s", ErrNoChannelAvailable, model) + } + return matchedChannels, nil +} diff --git a/routers/filter.go b/routers/filter.go index d092a7d..2b9fc91 100644 --- a/routers/filter.go +++ b/routers/filter.go @@ -25,7 +25,7 @@ import ( func TransparentStatic(ctx *context.Context) { urlPath := ctx.Request.URL.Path - if strings.HasPrefix(urlPath, "/api/") { + if strings.HasPrefix(urlPath, "/api/") || strings.HasPrefix(urlPath, "/v1/") { return } diff --git a/routers/router.go b/routers/router.go index 9e6e03f..d42c2c3 100644 --- a/routers/router.go +++ b/routers/router.go @@ -83,4 +83,7 @@ func initAPI() { beego.Router("/api/update-channel", &controllers.ApiController{}, "POST:UpdateChannel") beego.Router("/api/delete-channel", &controllers.ApiController{}, "POST:DeleteChannel") beego.Router("/api/test-channel", &controllers.ApiController{}, "POST:TestChannel") + + // OpenAI-compatible chat completions endpoint for LLM gateway milestone 1.2. + beego.Router("/v1/chat/completions", &controllers.ApiController{}, "POST:ChatCompletions") } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
