package portal import ( "bytes" "context" "crypto/rand" "crypto/sha256" "crypto/subtle" "encoding/hex" "encoding/json" "errors" "net/http" "net/http/httptest" "regexp" "strings" "time" "aigateway.local/core/internal/gateway" "aigateway.local/core/internal/identity" platformid "aigateway.local/core/internal/platform/id" "github.com/jackc/pgx/v5" ) func regexpMust(pattern string) *regexp.Regexp { return regexp.MustCompile(pattern) } // PersonalChannel 是门户用户自建的 webhook 渠道,绑定已批准模型。 // 入站消息经用户运行时凭据应答,用量归属用户自己的 Key。 type PersonalChannel struct { ID string `json:"id"` Code string `json:"code"` Name string `json:"name"` Kind string `json:"kind"` ProviderCode string `json:"provider_code"` Model string `json:"model"` Enabled bool `json:"enabled"` LastUsedAt *time.Time `json:"last_used_at,omitempty"` CreatedAt time.Time `json:"created_at"` UpdatedAt time.Time `json:"updated_at"` } const personalChannelSelect = `SELECT id::text,code,name,kind,provider_code,model,enabled,last_used_at,created_at,updated_at FROM gateway.personal_channels` func scanPersonalChannel(row pgx.Row) (PersonalChannel, error) { var c PersonalChannel err := row.Scan(&c.ID, &c.Code, &c.Name, &c.Kind, &c.ProviderCode, &c.Model, &c.Enabled, &c.LastUsedAt, &c.CreatedAt, &c.UpdatedAt) return c, err } // PersonalChannels 返回当前用户的个人渠道。 func (s *Service) PersonalChannels(ctx context.Context, account identity.Account) ([]PersonalChannel, error) { rows, err := s.pool.Query(ctx, personalChannelSelect+` WHERE portal_user_id=$1 ORDER BY created_at DESC`, account.ID) if err != nil { return nil, err } defer rows.Close() items := []PersonalChannel{} for rows.Next() { item, err := scanPersonalChannel(rows) if err != nil { return nil, err } items = append(items, item) } return items, rows.Err() } // CreatePersonalChannel 创建 webhook 个人渠道并返回入站令牌(仅此一次显示)。 func (s *Service) CreatePersonalChannel(ctx context.Context, account identity.Account, code, name, providerCode, model string) (PersonalChannel, string, error) { code = strings.ToLower(strings.TrimSpace(code)) name = strings.TrimSpace(name) if !personalChannelCodePattern.MatchString(code) || name == "" || len(name) > 128 { return PersonalChannel{}, "", errors.New("渠道代码或名称无效") } if _, err := s.approvedModel(ctx, account, providerCode, model); err != nil { return PersonalChannel{}, "", errors.New("模型未批准或不可用,请先在「模型权限」申请") } // 聊天依赖用户运行时凭据,先确保开通。 if _, err := s.ensureChatCredential(ctx, account); err != nil { return PersonalChannel{}, "", err } token, err := randomToken(32) if err != nil { return PersonalChannel{}, "", err } id, err := platformid.NewUUID() if err != nil { return PersonalChannel{}, "", err } hash := channelTokenHash(token) _, err = s.pool.Exec(ctx, `INSERT INTO gateway.personal_channels(id,portal_user_id,code,name,kind,inbound_token_hash,provider_code,model) VALUES($1,$2,$3,$4,'webhook',$5,$6,$7)`, id, account.ID, code, name, hash, providerCode, model) if err != nil { if isUniqueViolation(err) { return PersonalChannel{}, "", errors.New("渠道代码已存在") } return PersonalChannel{}, "", err } item, err := scanPersonalChannel(s.pool.QueryRow(ctx, personalChannelSelect+` WHERE id=$1`, id)) return item, token, err } // RegenerateToken 轮换入站令牌(旧令牌立即失效)。 func (s *Service) RegenerateToken(ctx context.Context, account identity.Account, id string) (string, error) { token, err := randomToken(32) if err != nil { return "", err } tag, err := s.pool.Exec(ctx, `UPDATE gateway.personal_channels SET inbound_token_hash=$3,updated_at=clock_timestamp() WHERE id=$1 AND portal_user_id=$2`, id, account.ID, channelTokenHash(token)) if err != nil { return "", err } if tag.RowsAffected() == 0 { return "", ErrNotFound } return token, nil } // DeletePersonalChannel 删除个人渠道(仅本人)。 func (s *Service) DeletePersonalChannel(ctx context.Context, account identity.Account, id string) error { tag, err := s.pool.Exec(ctx, `DELETE FROM gateway.personal_channels WHERE id=$1 AND portal_user_id=$2`, id, account.ID) if err != nil { return err } if tag.RowsAffected() == 0 { return ErrNotFound } return nil } // HandlePersonalInbound 处理个人渠道入站消息:令牌校验 → 用户运行时凭据应答。 func (s *Service) HandlePersonalInbound(ctx context.Context, code, presentedToken, text string) (string, error) { code = strings.ToLower(strings.TrimSpace(code)) text = strings.TrimSpace(text) if code == "" || presentedToken == "" { return "", errors.New("渠道代码或令牌缺失") } if text == "" || len(text) > 100000 { return "", errors.New("消息为空或过长") } var id, userID, providerCode, model, tokenHash string err := s.pool.QueryRow(ctx, `SELECT id::text,portal_user_id::text,provider_code,model,inbound_token_hash FROM gateway.personal_channels WHERE code=$1 AND enabled`, code).Scan(&id, &userID, &providerCode, &model, &tokenHash) if errors.Is(err, pgx.ErrNoRows) { // 与令牌无效同一文案,不泄露渠道存在性。 return "", errors.New("入站令牌无效") } if err != nil { return "", err } if subtle.ConstantTimeCompare([]byte(channelTokenHash(presentedToken)), []byte(tokenHash)) != 1 { return "", errors.New("入站令牌无效") } if s.credentials == nil || s.gateway == nil { return "", errors.New("渠道服务未配置") } secret, _, err := s.credentials.UserSecret(ctx, userID) if err != nil { return "", err } if secret == "" { return "", errors.New("用户运行时凭据未开通") } payload, _ := json.Marshal(map[string]any{"model": model, "messages": []map[string]any{{"role": "user", "content": text}}, "stream": false}) request := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(payload)).WithContext(gateway.WithRequestID(ctx, "personal-channel-"+time.Now().UTC().Format("20060102150405.000000000"))) request.Header.Set("Authorization", "Bearer "+secret) request.Header.Set("Content-Type", "application/json") recorder := httptest.NewRecorder() s.gateway.ServeHTTP(recorder, request) var response map[string]any if err := json.Unmarshal(recorder.Body.Bytes(), &response); err != nil { return "", errors.New("模型响应无法解析") } if recorder.Code < 200 || recorder.Code >= 300 { message := "模型调用失败" if value, ok := response["error"].(map[string]any); ok { if text, ok := value["message"].(string); ok { message = text } } return "", errors.New(message) } choices, _ := response["choices"].([]any) if len(choices) == 0 { return "", errors.New("模型未返回回答") } choice, _ := choices[0].(map[string]any) message, _ := choice["message"].(map[string]any) answer, _ := message["content"].(string) if strings.TrimSpace(answer) == "" { return "", errors.New("模型未返回文本回答") } _, _ = s.pool.Exec(ctx, `UPDATE gateway.personal_channels SET last_used_at=clock_timestamp() WHERE id=$1`, id) return answer, nil } func channelTokenHash(token string) string { sum := sha256.Sum256([]byte(token)) return hex.EncodeToString(sum[:]) } func randomToken(size int) (string, error) { buf := make([]byte, size) if _, err := rand.Read(buf); err != nil { return "", err } return hex.EncodeToString(buf), nil } var personalChannelCodePattern = regexpMust(`^[a-z][a-z0-9_-]{2,63}$`)