ai-gateway-go/internal/channel/http.go

package channel

import (
	"context"
	"crypto/subtle"
	"encoding/json"
	"io"
	"net/http"
	"strings"

	"aigateway.local/core/internal/identity"
	"aigateway.local/core/internal/platform/apiresponse"
)

// HTTPHandler 管理端渠道 CRUD。
type HTTPHandler struct {
	service  *Service
	identity *identity.Service
	mux      *http.ServeMux
}

func NewHTTPHandler(service *Service, identityService *identity.Service) *HTTPHandler {
	h := &HTTPHandler{service: service, identity: identityService, mux: http.NewServeMux()}
	h.mux.HandleFunc("GET /api/v1/admin/channels", h.list)
	h.mux.HandleFunc("POST /api/v1/admin/channels", h.save)
	h.mux.HandleFunc("PUT /api/v1/admin/channels/{id}", h.save)
	h.mux.HandleFunc("DELETE /api/v1/admin/channels/{id}", h.delete)
	h.mux.HandleFunc("POST /api/v1/admin/channels/{id}/test", h.test)
	h.mux.HandleFunc("GET /api/v1/admin/channels/{id}/grants", h.listGrants)
	h.mux.HandleFunc("POST /api/v1/admin/channels/{id}/grants", h.grant)
	h.mux.HandleFunc("DELETE /api/v1/admin/channels/{id}/grants/{user_id}", h.revokeGrant)
	return h
}

func (h *HTTPHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { h.mux.ServeHTTP(w, r) }

func (h *HTTPHandler) require(w http.ResponseWriter, r *http.Request, permission string) (identity.Account, bool) {
	account, err := h.identity.Authenticate(r.Context(), identity.KindAdmin, r.Header.Get("Authorization"))
	if err != nil {
		apiresponse.Error(w, http.StatusUnauthorized, "登录状态无效或已过期")
		return identity.Account{}, false
	}
	if !identity.HasPermission(account, permission) {
		apiresponse.Error(w, http.StatusForbidden, "缺少渠道管理权限")
		return identity.Account{}, false
	}
	return account, true
}

func (h *HTTPHandler) list(w http.ResponseWriter, r *http.Request) {
	if _, ok := h.require(w, r, identity.PermissionChannelRead); !ok {
		return
	}
	items, err := h.service.List(r.Context())
	if err != nil {
		apiresponse.Error(w, http.StatusServiceUnavailable, "渠道查询失败")
		return
	}
	apiresponse.OK(w, items)
}

type channelInput struct {
	Code          string          `json:"code"`
	Name          string          `json:"name"`
	Kind          string          `json:"kind"`
	Config        json.RawMessage `json:"config"`
	ModelBinding  json.RawMessage `json:"model_binding"`
	DepartmentIDs []string        `json:"department_ids"`
	APIKey        string          `json:"api_key"`
	Enabled       *bool           `json:"enabled"`
}

func (h *HTTPHandler) save(w http.ResponseWriter, r *http.Request) {
	actor, ok := h.require(w, r, identity.PermissionChannelManage)
	if !ok {
		return
	}
	var input channelInput
	decoder := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<20))
	decoder.DisallowUnknownFields()
	if decoder.Decode(&input) != nil {
		apiresponse.Error(w, http.StatusBadRequest, "请求格式无效")
		return
	}
	var cfg Config
	if len(input.Config) > 0 {
		if err := json.Unmarshal(input.Config, &cfg); err != nil {
			apiresponse.Error(w, http.StatusBadRequest, "平台配置格式无效")
			return
		}
	}
	enabled := true
	if input.Enabled != nil {
		enabled = *input.Enabled
	}
	item, err := h.service.Save(r.Context(), r.PathValue("id"), input.Code, input.Name, input.Kind, cfg, input.ModelBinding, input.DepartmentIDs, input.APIKey, enabled, actor.ID)
	if err != nil {
		apiresponse.Error(w, http.StatusBadRequest, err.Error())
		return
	}
	apiresponse.OK(w, item)
}

func (h *HTTPHandler) delete(w http.ResponseWriter, r *http.Request) {
	actor, ok := h.require(w, r, identity.PermissionChannelManage)
	if !ok {
		return
	}
	if err := h.service.Delete(r.Context(), r.PathValue("id"), actor.ID); err != nil {
		apiresponse.Error(w, http.StatusBadRequest, err.Error())
		return
	}
	apiresponse.OK(w, map[string]bool{"deleted": true})
}

// test 用渠道绑定模型发送一条测试消息并尝试平台回复。
func (h *HTTPHandler) test(w http.ResponseWriter, r *http.Request) {
	if _, ok := h.require(w, r, identity.PermissionChannelManage); !ok {
		return
	}
	items, err := h.service.List(r.Context())
	if err != nil {
		apiresponse.Error(w, http.StatusServiceUnavailable, "渠道查询失败")
		return
	}
	var target *Channel
	for i := range items {
		if items[i].ID == r.PathValue("id") {
			target = &items[i]
			break
		}
	}
	if target == nil {
		apiresponse.Error(w, http.StatusNotFound, "渠道不存在")
		return
	}
	answer, err := h.service.HandleInbound(r.Context(), *target, InboundMessage{FromUser: "admin-test", Text: "这是一条渠道连通性测试消息,请简短回复。"})
	if err != nil {
		apiresponse.Error(w, http.StatusBadGateway, err.Error())
		return
	}
	cfg, cfgErr := h.service.DecryptConfig(*target)
	if cfgErr == nil {
		_ = h.service.Reply(context.WithoutCancel(r.Context()), *target, cfg, answer)
	}
	apiresponse.OK(w, map[string]any{"answer": answer})
}

// InboundHTTPHandler 公开入站端点(按渠道 code 分发)。
type InboundHTTPHandler struct {
	service *Service
	mux     *http.ServeMux
}

func NewInboundHTTPHandler(service *Service) *InboundHTTPHandler {
	h := &InboundHTTPHandler{service: service, mux: http.NewServeMux()}
	h.mux.HandleFunc("POST /v1/channels/{code}/inbound", h.inbound)
	return h
}

func (h *InboundHTTPHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { h.mux.ServeHTTP(w, r) }

func (h *InboundHTTPHandler) inbound(w http.ResponseWriter, r *http.Request) {
	c, err := h.service.GetByCode(r.Context(), strings.ToLower(r.PathValue("code")))
	if err != nil {
		// 与令牌无效返回同一错误,不泄露渠道存在性。
		apiresponse.Error(w, http.StatusUnauthorized, "渠道令牌无效或渠道不存在")
		return
	}
	cfg, err := h.service.DecryptConfig(c)
	if err != nil {
		apiresponse.Error(w, http.StatusServiceUnavailable, "渠道配置不可用")
		return
	}
	// 平台签名校验。
	switch c.Kind {
	case "wecom":
		// 企业微信回调验证:echostr 原样返回。
		if echostr := r.URL.Query().Get("echostr"); echostr != "" {
			if _, ok := VerifyWeComSignature(cfg.InboundToken, r.URL.Query().Get("timestamp"), r.URL.Query().Get("nonce"), r.URL.Query().Get("msg_signature"), nil); ok {
				_, _ = w.Write([]byte(echostr))
				return
			}
			apiresponse.Error(w, http.StatusUnauthorized, "签名校验失败")
			return
		}
	case "dingtalk", "feishu":
		// 钉钉/飞书机器人验签由平台侧 access_token/回调令牌控制;此处信任平台。
	case "webhook":
		// 通用 webhook 必须携带入站令牌:未配置令牌的渠道拒绝入站,防止
		// 任意调用者消耗绑定模型的配额与费用。
		presented := strings.TrimSpace(r.Header.Get("X-Inbound-Token"))
		if presented == "" {
			presented = strings.TrimSpace(r.URL.Query().Get("token"))
		}
		if cfg.InboundToken == "" || subtle.ConstantTimeCompare([]byte(presented), []byte(cfg.InboundToken)) != 1 {
			apiresponse.Error(w, http.StatusUnauthorized, "渠道令牌无效或渠道不存在")
			return
		}
	}
	var payload struct {
		Text struct {
			Content string `json:"content"`
		} `json:"text"`
		Content string `json:"content"`
	}
	raw, _ := io.ReadAll(io.LimitReader(r.Body, 1<<20))
	_ = json.Unmarshal(raw, &payload)
	text := payload.Text.Content
	if text == "" {
		text = payload.Content
	}
	if text == "" {
		apiresponse.Error(w, http.StatusBadRequest, "消息内容为空")
		return
	}
	answer, err := h.service.HandleInbound(r.Context(), c, InboundMessage{Text: text})
	if err != nil {
		apiresponse.Error(w, http.StatusBadGateway, err.Error())
		return
	}
	if err := h.service.Reply(r.Context(), c, cfg, answer); err != nil {
		apiresponse.Error(w, http.StatusBadGateway, err.Error())
		return
	}
	// 通用 webhook 同步返回回答;平台渠道返回受理确认。
	if c.Kind == "webhook" {
		apiresponse.OK(w, map[string]any{"reply": answer})
		return
	}
	apiresponse.OK(w, map[string]bool{"accepted": true})
}

// listGrants 渠道用户授权列表。
func (h *HTTPHandler) listGrants(w http.ResponseWriter, r *http.Request) {
	if _, ok := h.require(w, r, identity.PermissionChannelRead); !ok {
		return
	}
	items, err := h.service.ListGrants(r.Context(), r.PathValue("id"))
	if err != nil {
		apiresponse.Error(w, http.StatusServiceUnavailable, "渠道授权查询失败")
		return
	}
	apiresponse.OK(w, items)
}

// grant 直接授予用户渠道使用权限(管理员显式授权,无需走申请流)。
func (h *HTTPHandler) grant(w http.ResponseWriter, r *http.Request) {
	actor, ok := h.require(w, r, identity.PermissionChannelManage)
	if !ok {
		return
	}
	var input struct {
		PortalUserID string `json:"portal_user_id"`
	}
	decoder := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<20))
	decoder.DisallowUnknownFields()
	if decoder.Decode(&input) != nil || strings.TrimSpace(input.PortalUserID) == "" {
		apiresponse.Error(w, http.StatusBadRequest, "请求格式无效")
		return
	}
	if err := h.service.Grant(r.Context(), r.PathValue("id"), input.PortalUserID, actor.ID, "manual"); err != nil {
		apiresponse.Error(w, http.StatusBadRequest, err.Error())
		return
	}
	apiresponse.OK(w, map[string]bool{"granted": true})
}

// revokeGrant 撤销用户的渠道使用权限。
func (h *HTTPHandler) revokeGrant(w http.ResponseWriter, r *http.Request) {
	actor, ok := h.require(w, r, identity.PermissionChannelManage)
	if !ok {
		return
	}
	if err := h.service.RevokeGrant(r.Context(), r.PathValue("id"), r.PathValue("user_id"), actor.ID); err != nil {
		apiresponse.Error(w, http.StatusBadRequest, err.Error())
		return
	}
	apiresponse.OK(w, map[string]bool{"revoked": true})
}