ai-gateway-go/internal/portal/employees.go

package portal

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"net/http"
	"net/http/httptest"
	"strings"
	"time"

	"aigateway.local/core/internal/gateway"
	"aigateway.local/core/internal/identity"
	"github.com/jackc/pgx/v5"
)

// DigitalEmployeeView 是门户可见的数字员工(部门可见或已安装)。
type DigitalEmployeeView struct {
	Code        string `json:"code"`
	Name        string `json:"name"`
	Description string `json:"description"`
	Installed   bool   `json:"installed"`
}

// DigitalEmployeeRun 是用户的数字员工调用记录。
type DigitalEmployeeRun struct {
	EmployeeCode string    `json:"employee_code"`
	EmployeeName string    `json:"employee_name"`
	Status       string    `json:"status"`
	LatencyMS    int64     `json:"latency_ms"`
	Retrieval    int       `json:"retrieval_count"`
	ToolCalls    int       `json:"tool_count"`
	Error        string    `json:"error,omitempty"`
	CreatedAt    time.Time `json:"created_at"`
}

// DigitalEmployees 返回当前用户可用的数字员工。
func (s *Service) DigitalEmployees(ctx context.Context, account identity.Account) ([]DigitalEmployeeView, error) {
	if s.market == nil {
		return []DigitalEmployeeView{}, nil
	}
	items, err := s.market.Catalog(ctx, "digital_employee", "", "", "", 200)
	if err != nil {
		return nil, err
	}
	installedItems, err := s.market.ListInstalled(ctx, account.ID)
	if err != nil {
		return nil, err
	}
	installed := map[string]bool{}
	for _, item := range installedItems {
		if item.Type == "digital_employee" {
			installed[item.Code] = true
		}
	}
	result := []DigitalEmployeeView{}
	for _, item := range items {
		if !visible(item.DepartmentIDs, account.DepartmentID) {
			continue
		}
		result = append(result, DigitalEmployeeView{Code: item.Code, Name: item.Name, Description: item.Description, Installed: installed[item.Code]})
	}
	return result, nil
}

// RunDigitalEmployee 运行一次数字员工对话(复用用户运行时凭据)。
func (s *Service) RunDigitalEmployee(ctx context.Context, account identity.Account, code, message string) (map[string]any, error) {
	code = strings.ToLower(strings.TrimSpace(code))
	message = strings.TrimSpace(message)
	if message == "" || len(message) > 100000 {
		return nil, errors.New("消息为空或过长")
	}
	if s.credentials == nil || s.runtime == nil {
		return nil, errors.New("数字员工服务未配置")
	}
	secret, _, err := s.credentials.EnsureUser(ctx, account.ID, account.DepartmentID, 120, 0)
	if err != nil {
		return nil, err
	}
	payload, _ := json.Marshal(map[string]any{
		"messages":  []map[string]any{{"role": "user", "content": message}},
		"variables": map[string]any{},
	})
	request := httptest.NewRequest(http.MethodPost, "/v1/digital-employees/"+code+"/chat/completions", bytes.NewReader(payload)).WithContext(gateway.WithRequestID(ctx, "portal-employee-"+time.Now().UTC().Format("20060102150405.000000000")))
	request.Header.Set("Authorization", "Bearer "+secret)
	request.Header.Set("Content-Type", "application/json")
	recorder := httptest.NewRecorder()
	s.runtime.ServeHTTP(recorder, request)
	var response map[string]any
	if err := json.Unmarshal(recorder.Body.Bytes(), &response); err != nil {
		return nil, errors.New("数字员工响应无法解析")
	}
	if recorder.Code < 200 || recorder.Code >= 300 {
		message := fmt.Sprintf("数字员工调用失败（HTTP %d）", recorder.Code)
		if value, ok := response["error"].(map[string]any); ok {
			if text, ok := value["message"].(string); ok {
				message = text
			}
		}
		return response, errors.New(message)
	}
	return response, nil
}

// MyEmployeeRuns 返回当前用户的数字员工调用记录(经用户运行时 Key 归属)。
func (s *Service) MyEmployeeRuns(ctx context.Context, account identity.Account, limit int) ([]DigitalEmployeeRun, error) {
	if limit < 1 || limit > 100 {
		limit = 20
	}
	// 运行时 Key 可能尚未开通:此时无记录,直接返回空。
	var apiKeyID string
	err := s.pool.QueryRow(ctx, `SELECT api_key_id::text FROM gateway.portal_user_runtime_credentials WHERE portal_user_id=$1`, account.ID).Scan(&apiKeyID)
	if errors.Is(err, pgx.ErrNoRows) {
		return []DigitalEmployeeRun{}, nil
	}
	if err != nil {
		return nil, err
	}
	rows, err := s.pool.Query(ctx, `SELECT e.code,e.name,r.status,r.latency_ms,r.retrieval_count,r.tool_count,r.error,r.created_at
		FROM gateway.digital_employee_runs r JOIN gateway.digital_employees e ON e.id=r.digital_employee_id
		WHERE r.api_key_id=$1 ORDER BY r.created_at DESC LIMIT $2`, apiKeyID, limit)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	items := []DigitalEmployeeRun{}
	for rows.Next() {
		var item DigitalEmployeeRun
		if err := rows.Scan(&item.EmployeeCode, &item.EmployeeName, &item.Status, &item.LatencyMS, &item.Retrieval, &item.ToolCalls, &item.Error, &item.CreatedAt); err != nil {
			return nil, err
		}
		items = append(items, item)
	}
	return items, rows.Err()
}