superidou
4563979a15
- 迁移 000035-000037(权限等级/环境变量/渠道) - 新增 internal/channel 渠道抽象层(webhook/企微/钉钉/飞书) - 全部功能端到端验证通过(25 包单测)
443 lines
14 KiB
Go
443 lines
14 KiB
Go
// Package channel 实现多渠道接入:通用 Webhook + 企业微信/钉钉/飞书。
|
|
// 入站消息经绑定模型应答后,按平台协议回发。
|
|
package channel
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"crypto/sha1"
|
|
"crypto/sha256"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"log/slog"
|
|
"net/http"
|
|
"net/url"
|
|
"sort"
|
|
"strings"
|
|
"time"
|
|
|
|
platformid "aigateway.local/core/internal/platform/id"
|
|
"github.com/jackc/pgx/v5"
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
)
|
|
|
|
var (
|
|
ErrNotFound = errors.New("渠道不存在")
|
|
ErrUnavailable = errors.New("渠道服务不可用")
|
|
)
|
|
|
|
// Channel 是一条渠道配置。
|
|
type Channel struct {
|
|
ID string `json:"id"`
|
|
Code string `json:"code"`
|
|
Name string `json:"name"`
|
|
Kind string `json:"kind"`
|
|
Config json.RawMessage `json:"config,omitempty"`
|
|
EncryptedConfig []byte `json:"-"`
|
|
ConfigKEKVersion int `json:"-"`
|
|
ModelBinding json.RawMessage `json:"model_binding"`
|
|
HasAPIKey bool `json:"has_api_key"`
|
|
Enabled bool `json:"enabled"`
|
|
CreatedBy *string `json:"created_by,omitempty"`
|
|
CreatedAt time.Time `json:"created_at"`
|
|
UpdatedAt time.Time `json:"updated_at"`
|
|
}
|
|
|
|
// Config 是渠道的平台配置(明文,仅内部使用)。
|
|
type Config struct {
|
|
// 通用
|
|
InboundToken string `json:"inbound_token,omitempty"` // webhook 鉴权令牌
|
|
// 企业微信(自建应用回调 + 主动发送)
|
|
CorpID string `json:"corp_id,omitempty"`
|
|
Secret string `json:"secret,omitempty"`
|
|
AgentID string `json:"agent_id,omitempty"`
|
|
// 钉钉(机器人 webhook)
|
|
DingRobotToken string `json:"ding_robot_token,omitempty"`
|
|
// 飞书(应用)
|
|
FeishuAppID string `json:"feishu_app_id,omitempty"`
|
|
FeishuAppSecret string `json:"feishu_app_secret,omitempty"`
|
|
}
|
|
|
|
// Service 渠道管理 + 消息收发。
|
|
type Service struct {
|
|
pool *pgxpool.Pool
|
|
gatewayURL string
|
|
client *http.Client
|
|
logger *slog.Logger
|
|
cipher interface {
|
|
Encrypt([]byte) ([]byte, int, error)
|
|
Decrypt([]byte, int) ([]byte, error)
|
|
}
|
|
}
|
|
|
|
// NewService 创建渠道服务;cipher 加密平台配置与 API Key。
|
|
func NewService(pool *pgxpool.Pool, gatewayURL string, cipher interface {
|
|
Encrypt([]byte) ([]byte, int, error)
|
|
Decrypt([]byte, int) ([]byte, error)
|
|
}, logger *slog.Logger) *Service {
|
|
return &Service{
|
|
pool: pool, gatewayURL: strings.TrimRight(gatewayURL, "/"), logger: logger, cipher: cipher,
|
|
client: &http.Client{Timeout: 60 * time.Second},
|
|
}
|
|
}
|
|
|
|
const channelSelect = `SELECT id::text,code,name,kind,encrypted_config,config_kek_version,model_binding,octet_length(encrypted_api_key)>0,enabled,created_by::text,created_at,updated_at FROM gateway.channels`
|
|
|
|
func (s *Service) scan(row pgx.Row) (Channel, error) {
|
|
var c Channel
|
|
var createdBy *string
|
|
err := row.Scan(&c.ID, &c.Code, &c.Name, &c.Kind, &c.EncryptedConfig, &c.ConfigKEKVersion, &c.ModelBinding, &c.HasAPIKey, &c.Enabled, &createdBy, &c.CreatedAt, &c.UpdatedAt)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return Channel{}, ErrNotFound
|
|
}
|
|
if err != nil {
|
|
return Channel{}, err
|
|
}
|
|
c.CreatedBy = createdBy
|
|
return c, nil
|
|
}
|
|
|
|
// List 返回渠道列表(不含敏感配置)。
|
|
func (s *Service) List(ctx context.Context) ([]Channel, error) {
|
|
if s == nil || s.pool == nil {
|
|
return nil, ErrUnavailable
|
|
}
|
|
rows, err := s.pool.Query(ctx, channelSelect+` ORDER BY updated_at DESC`)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []Channel{}
|
|
for rows.Next() {
|
|
c, err := s.scan(rows)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, c)
|
|
}
|
|
return items, rows.Err()
|
|
}
|
|
|
|
// GetByCode 供入站分发使用(无需解密配置)。
|
|
func (s *Service) GetByCode(ctx context.Context, code string) (Channel, error) {
|
|
if s == nil || s.pool == nil {
|
|
return Channel{}, ErrUnavailable
|
|
}
|
|
return s.scan(s.pool.QueryRow(ctx, channelSelect+` WHERE code=$1 AND enabled`, code))
|
|
}
|
|
|
|
// DecryptConfig 解密平台配置。
|
|
func (s *Service) DecryptConfig(c Channel) (Config, error) {
|
|
var cfg Config
|
|
if len(c.EncryptedConfig) == 0 {
|
|
return cfg, nil
|
|
}
|
|
plaintext, err := s.cipher.Decrypt(c.EncryptedConfig, c.ConfigKEKVersion)
|
|
if err != nil {
|
|
return cfg, err
|
|
}
|
|
if err := json.Unmarshal(plaintext, &cfg); err != nil {
|
|
return cfg, err
|
|
}
|
|
return cfg, nil
|
|
}
|
|
|
|
// Save 创建/更新渠道。
|
|
func (s *Service) Save(ctx context.Context, id, code, name, kind string, cfg Config, modelBinding json.RawMessage, apiKey string, enabled bool, actorID string) (Channel, error) {
|
|
if s == nil || s.pool == nil || s.cipher == nil {
|
|
return Channel{}, ErrUnavailable
|
|
}
|
|
code = strings.ToLower(strings.TrimSpace(code))
|
|
name = strings.TrimSpace(name)
|
|
if !codePattern.MatchString(code) || name == "" || len(name) > 128 {
|
|
return Channel{}, errors.New("渠道代码或名称无效")
|
|
}
|
|
switch kind {
|
|
case "webhook", "wecom", "dingtalk", "feishu":
|
|
default:
|
|
return Channel{}, errors.New("渠道类型必须是 webhook/wecom/dingtalk/feishu")
|
|
}
|
|
if id == "" {
|
|
newID, err := platformid.NewUUID()
|
|
if err != nil {
|
|
return Channel{}, err
|
|
}
|
|
id = newID
|
|
}
|
|
encryptedConfig, configVersion, err := s.cipher.Encrypt(mustJSON(cfg))
|
|
if err != nil {
|
|
return Channel{}, err
|
|
}
|
|
var encryptedKey []byte
|
|
var keyVersion int
|
|
if apiKey != "" {
|
|
encryptedKey, keyVersion, err = s.cipher.Encrypt([]byte(apiKey))
|
|
if err != nil {
|
|
return Channel{}, err
|
|
}
|
|
}
|
|
if modelBinding == nil {
|
|
modelBinding = json.RawMessage(`{}`)
|
|
}
|
|
_, err = s.pool.Exec(ctx, `INSERT INTO gateway.channels(id,code,name,kind,encrypted_config,config_kek_version,encrypted_api_key,api_key_kek_version,model_binding,enabled,created_by)
|
|
VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11)
|
|
ON CONFLICT(code) DO UPDATE SET name=$3,kind=$4,encrypted_config=$5,config_kek_version=$6,
|
|
encrypted_api_key=CASE WHEN $7<>'' THEN $7 ELSE gateway.channels.encrypted_api_key END,
|
|
api_key_kek_version=CASE WHEN $7<>'' THEN $8 ELSE gateway.channels.api_key_kek_version END,
|
|
model_binding=$9,enabled=$10,updated_at=clock_timestamp()`,
|
|
id, code, name, kind, encryptedConfig, configVersion, encryptedKey, keyVersion, modelBinding, enabled, actorID)
|
|
if err != nil {
|
|
return Channel{}, err
|
|
}
|
|
return s.scan(s.pool.QueryRow(ctx, channelSelect+` WHERE id=$1`, id))
|
|
}
|
|
|
|
func (s *Service) Delete(ctx context.Context, id string) error {
|
|
if s == nil || s.pool == nil {
|
|
return ErrUnavailable
|
|
}
|
|
tag, err := s.pool.Exec(ctx, `DELETE FROM gateway.channels WHERE id=$1`, id)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if tag.RowsAffected() == 0 {
|
|
return ErrNotFound
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// DecryptAPIKey 解密渠道绑定的网关 API Key(入站调用用)。
|
|
func (s *Service) DecryptAPIKey(ctx context.Context, c Channel) (string, error) {
|
|
if !c.HasAPIKey {
|
|
return "", nil
|
|
}
|
|
var encrypted []byte
|
|
var version int
|
|
if err := s.pool.QueryRow(ctx, `SELECT encrypted_api_key,api_key_kek_version FROM gateway.channels WHERE id=$1`, c.ID).Scan(&encrypted, &version); err != nil {
|
|
return "", err
|
|
}
|
|
plaintext, err := s.cipher.Decrypt(encrypted, version)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
return string(plaintext), nil
|
|
}
|
|
|
|
func mustJSON(value any) []byte {
|
|
raw, _ := json.Marshal(value)
|
|
return raw
|
|
}
|
|
|
|
var codePattern = regexpMust(`^[a-z][a-z0-9_-]{2,63}$`)
|
|
|
|
// InboundMessage 是归一化的入站消息。
|
|
type InboundMessage struct {
|
|
FromUser string
|
|
Text string
|
|
}
|
|
|
|
// HandleInbound 处理入站消息:调用绑定模型,按平台回复。
|
|
func (s *Service) HandleInbound(ctx context.Context, c Channel, msg InboundMessage) (string, error) {
|
|
if s == nil || s.gatewayURL == "" {
|
|
return "", ErrUnavailable
|
|
}
|
|
apiKey, err := s.DecryptAPIKey(ctx, c)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
var binding struct {
|
|
Provider string `json:"provider"`
|
|
Model string `json:"model"`
|
|
}
|
|
_ = json.Unmarshal(c.ModelBinding, &binding)
|
|
if binding.Model == "" {
|
|
binding.Model = "gpt-4o-mini"
|
|
}
|
|
payload, _ := json.Marshal(map[string]any{
|
|
"model": binding.Model,
|
|
"messages": []map[string]any{
|
|
{"role": "user", "content": msg.Text},
|
|
},
|
|
})
|
|
request, err := http.NewRequestWithContext(ctx, http.MethodPost, s.gatewayURL+"/v1/chat/completions", bytes.NewReader(payload))
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
request.Header.Set("Content-Type", "application/json")
|
|
if apiKey != "" {
|
|
request.Header.Set("Authorization", "Bearer "+apiKey)
|
|
}
|
|
response, err := s.client.Do(request)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
defer response.Body.Close()
|
|
raw, _ := io.ReadAll(io.LimitReader(response.Body, 1<<20))
|
|
if response.StatusCode/100 != 2 {
|
|
return "", fmt.Errorf("模型调用失败(HTTP %d)", response.StatusCode)
|
|
}
|
|
var decoded struct {
|
|
Choices []struct {
|
|
Message struct {
|
|
Content string `json:"content"`
|
|
} `json:"message"`
|
|
} `json:"choices"`
|
|
}
|
|
if json.Unmarshal(raw, &decoded) != nil || len(decoded.Choices) == 0 {
|
|
return "", errors.New("模型响应格式无效")
|
|
}
|
|
return decoded.Choices[0].Message.Content, nil
|
|
}
|
|
|
|
// Reply 按平台协议发送回复。webhook 渠道返回同步回复文本。
|
|
func (s *Service) Reply(ctx context.Context, c Channel, cfg Config, text string) error {
|
|
switch c.Kind {
|
|
case "webhook":
|
|
return nil // 同步返回
|
|
case "wecom":
|
|
return s.replyWeCom(ctx, cfg, text)
|
|
case "dingtalk":
|
|
return s.replyDingTalk(ctx, cfg, text)
|
|
case "feishu":
|
|
return s.replyFeishu(ctx, cfg, text)
|
|
default:
|
|
return errors.New("不支持的渠道类型")
|
|
}
|
|
}
|
|
|
|
// replyWeCom 通过企业微信应用消息接口发送。
|
|
func (s *Service) replyWeCom(ctx context.Context, cfg Config, text string) error {
|
|
token, err := s.weComToken(ctx, cfg)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
payload, _ := json.Marshal(map[string]any{
|
|
"touser": "@all",
|
|
"msgtype": "text",
|
|
"agentid": cfg.AgentID,
|
|
"text": map[string]string{"content": text},
|
|
})
|
|
return s.postJSON(ctx, "https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token="+token, payload)
|
|
}
|
|
|
|
func (s *Service) weComToken(ctx context.Context, cfg Config) (string, error) {
|
|
endpoint := fmt.Sprintf("https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid=%s&corpsecret=%s", url.QueryEscape(cfg.CorpID), url.QueryEscape(cfg.Secret))
|
|
response, err := s.client.Get(endpoint)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
defer response.Body.Close()
|
|
raw, _ := io.ReadAll(io.LimitReader(response.Body, 1<<16))
|
|
var decoded struct {
|
|
ErrCode int `json:"errcode"`
|
|
Token string `json:"access_token"`
|
|
}
|
|
if json.Unmarshal(raw, &decoded) != nil || decoded.ErrCode != 0 || decoded.Token == "" {
|
|
return "", errors.New("企业微信 token 获取失败")
|
|
}
|
|
return decoded.Token, nil
|
|
}
|
|
|
|
// replyDingTalk 通过钉钉自定义机器人 webhook 发送。
|
|
func (s *Service) replyDingTalk(ctx context.Context, cfg Config, text string) error {
|
|
payload, _ := json.Marshal(map[string]any{
|
|
"msgtype": "text",
|
|
"text": map[string]string{"content": text},
|
|
})
|
|
endpoint := "https://oapi.dingtalk.com/robot/send?access_token=" + url.QueryEscape(cfg.DingRobotToken)
|
|
return s.postJSON(ctx, endpoint, payload)
|
|
}
|
|
|
|
// replyFeishu 通过飞书机器人消息接口发送。
|
|
func (s *Service) replyFeishu(ctx context.Context, cfg Config, text string) error {
|
|
token, err := s.feishuToken(ctx, cfg)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
payload, _ := json.Marshal(map[string]any{
|
|
"receive_id": "@all",
|
|
"msg_type": "text",
|
|
"content": map[string]string{"text": text},
|
|
})
|
|
return s.postJSON(ctx, "https://open.feishu.cn/open-apis/im/v1/messages?receive_id_type=chat_id", payload, func(r *http.Request) {
|
|
r.Header.Set("Authorization", "Bearer "+token)
|
|
})
|
|
}
|
|
|
|
func (s *Service) feishuToken(ctx context.Context, cfg Config) (string, error) {
|
|
payload, _ := json.Marshal(map[string]string{"app_id": cfg.FeishuAppID, "app_secret": cfg.FeishuAppSecret})
|
|
response, err := s.client.Post("https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal", "application/json", bytes.NewReader(payload))
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
defer response.Body.Close()
|
|
raw, _ := io.ReadAll(io.LimitReader(response.Body, 1<<16))
|
|
var decoded struct {
|
|
Code int `json:"code"`
|
|
Token string `json:"tenant_access_token"`
|
|
}
|
|
if json.Unmarshal(raw, &decoded) != nil || decoded.Code != 0 || decoded.Token == "" {
|
|
return "", errors.New("飞书 token 获取失败")
|
|
}
|
|
return decoded.Token, nil
|
|
}
|
|
|
|
func (s *Service) postJSON(ctx context.Context, endpoint string, payload []byte, options ...func(*http.Request)) error {
|
|
request, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(payload))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
request.Header.Set("Content-Type", "application/json")
|
|
for _, option := range options {
|
|
option(request)
|
|
}
|
|
response, err := s.client.Do(request)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer response.Body.Close()
|
|
raw, _ := io.ReadAll(io.LimitReader(response.Body, 1<<16))
|
|
if response.StatusCode/100 != 2 {
|
|
return fmt.Errorf("平台接口返回 HTTP %d", response.StatusCode)
|
|
}
|
|
var envelope struct {
|
|
ErrCode int `json:"errcode"`
|
|
Code int `json:"code"`
|
|
}
|
|
_ = json.Unmarshal(raw, &envelope)
|
|
if envelope.ErrCode != 0 || envelope.Code != 0 {
|
|
return fmt.Errorf("平台接口返回错误码 %d/%d", envelope.ErrCode, envelope.Code)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// VerifyWeComSignature 校验企业微信回调签名(URL 参数签名)。
|
|
func VerifyWeComSignature(token, timestamp, nonce, echostr string, values map[string]string) (string, bool) {
|
|
parts := []string{token, timestamp, nonce}
|
|
if values != nil {
|
|
keys := make([]string, 0, len(values))
|
|
for key := range values {
|
|
keys = append(keys, key)
|
|
}
|
|
sort.Strings(keys)
|
|
for _, key := range keys {
|
|
parts = append(parts, key+"="+values[key])
|
|
}
|
|
}
|
|
sort.Strings(parts)
|
|
sum := sha1.Sum([]byte(strings.Join(parts, "")))
|
|
if hex.EncodeToString(sum[:]) != echostr {
|
|
return "", false
|
|
}
|
|
return echostr, true
|
|
}
|
|
|
|
// DingSign 计算钉钉机器人加签(时间戳+密钥)。
|
|
func DingSign(timestamp int64, secret string) string {
|
|
sum := sha256.Sum256([]byte(fmt.Sprintf("%d\n%s", timestamp, secret)))
|
|
return url.QueryEscape(hex.EncodeToString(sum[:]))
|
|
}
|