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

package portal

import (
	"context"
	"encoding/json"
	"errors"
	"strings"
	"time"

	"aigateway.local/core/internal/identity"
	platformid "aigateway.local/core/internal/platform/id"
	"aigateway.local/core/internal/platform/oplog"
	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgconn"
)

func isUniqueViolation(err error) bool {
	var pgError *pgconn.PgError
	return errors.As(err, &pgError) && pgError.Code == "23505"
}

// ResourceRequest 是资源/渠道权限申请(mcp/skill/数字员工/渠道)。
// 审批通过后:市场资源自动安装(use 等级),渠道申请以批准记录本身作为授权凭据。
type ResourceRequest struct {
	ID           string     `json:"id"`
	PortalUserID string     `json:"portal_user_id"`
	UserLogin    string     `json:"user_login,omitempty"`
	ResourceType string     `json:"resource_type"`
	ResourceCode string     `json:"resource_code"`
	Reason       string     `json:"reason"`
	Status       string     `json:"status"`
	DecisionNote string     `json:"decision_note"`
	DecidedAt    *time.Time `json:"decided_at,omitempty"`
	CreatedAt    time.Time  `json:"created_at"`
	UpdatedAt    time.Time  `json:"updated_at"`
}

const resourceRequestSelect = `SELECT r.id::text,r.portal_user_id::text,u.account,r.resource_type,r.resource_code,r.reason,r.status,r.decision_note,r.decided_at,r.created_at,r.updated_at FROM gateway.resource_access_requests r JOIN gateway.portal_users u ON u.id=r.portal_user_id`

func scanResourceRequest(row pgx.Row) (ResourceRequest, error) {
	var item ResourceRequest
	err := row.Scan(&item.ID, &item.PortalUserID, &item.UserLogin, &item.ResourceType, &item.ResourceCode, &item.Reason, &item.Status, &item.DecisionNote, &item.DecidedAt, &item.CreatedAt, &item.UpdatedAt)
	return item, err
}

// resourceTypeSupported 校验申请的资源类型。
func resourceTypeSupported(resourceType string) bool {
	switch strings.ToLower(strings.TrimSpace(resourceType)) {
	case "mcp_server", "skill", "digital_employee", "channel":
		return true
	}
	return false
}

// ResourceRequests 返回当前用户的资源/渠道申请。
func (s *Service) ResourceRequests(ctx context.Context, account identity.Account) ([]ResourceRequest, error) {
	rows, err := s.pool.Query(ctx, resourceRequestSelect+` WHERE r.portal_user_id=$1 ORDER BY r.created_at DESC LIMIT 200`, account.ID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	items := []ResourceRequest{}
	for rows.Next() {
		item, err := scanResourceRequest(rows)
		if err != nil {
			return nil, err
		}
		items = append(items, item)
	}
	return items, rows.Err()
}

// CreateResourceRequest 发起资源/渠道权限申请(每用户每资源至多一个待审项)。
func (s *Service) CreateResourceRequest(ctx context.Context, account identity.Account, resourceType, code, reason string) (ResourceRequest, error) {
	resourceType = strings.ToLower(strings.TrimSpace(resourceType))
	code = strings.ToLower(strings.TrimSpace(code))
	reason = strings.TrimSpace(reason)
	if !resourceTypeSupported(resourceType) {
		return ResourceRequest{}, errors.New("申请类型必须是 mcp_server/skill/digital_employee/channel")
	}
	if code == "" || len(code) > 128 || len(reason) > 4000 {
		return ResourceRequest{}, errors.New("申请内容格式无效")
	}
	// 目标必须真实存在且启用:市场资源须已发布,渠道须已启用。
	switch resourceType {
	case "mcp_server", "skill", "digital_employee":
		if s.market == nil {
			return ResourceRequest{}, errors.New("资源市场服务未配置")
		}
		if _, _, err := s.market.Detail(ctx, resourceType, code); err != nil {
			return ResourceRequest{}, errors.New("资源不存在或未发布")
		}
	case "channel":
		var enabled bool
		if err := s.pool.QueryRow(ctx, `SELECT enabled FROM gateway.channels WHERE code=$1`, code).Scan(&enabled); err != nil || !enabled {
			return ResourceRequest{}, errors.New("渠道不存在或未启用")
		}
	}
	item := ResourceRequest{PortalUserID: account.ID, ResourceType: resourceType, ResourceCode: code, Reason: reason}
	item.ID, _ = platformid.NewUUID()
	eventID, _ := platformid.NewUUID()
	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return ResourceRequest{}, err
	}
	defer func() { _ = tx.Rollback(ctx) }()
	_, err = tx.Exec(ctx, `INSERT INTO gateway.resource_access_requests(id,portal_user_id,resource_type,resource_code,reason) VALUES($1,$2,$3,$4,$5)`, item.ID, account.ID, resourceType, code, reason)
	if err != nil {
		if isUniqueViolation(err) {
			return ResourceRequest{}, errors.New("同类申请已存在，等待管理员审批")
		}
		return ResourceRequest{}, err
	}
	payload, _ := json.Marshal(map[string]any{"request_id": item.ID, "portal_user_id": account.ID, "resource_type": resourceType, "resource_code": code})
	if _, err = tx.Exec(ctx, `INSERT INTO gateway.outbox_events(event_id,event_type,event_version,aggregate_type,aggregate_id,payload) VALUES($1,'resource_access.requested',1,'resource_access_request',$2,$3)`, eventID, item.ID, payload); err != nil {
		return ResourceRequest{}, err
	}
	if err = tx.Commit(ctx); err != nil {
		return ResourceRequest{}, err
	}
	return scanResourceRequest(s.pool.QueryRow(ctx, resourceRequestSelect+` WHERE r.id=$1`, item.ID))
}

// CancelResourceRequest 撤回本人待审申请。
func (s *Service) CancelResourceRequest(ctx context.Context, account identity.Account, id string) error {
	tag, err := s.pool.Exec(ctx, `UPDATE gateway.resource_access_requests SET status='cancelled',updated_at=clock_timestamp() WHERE id=$1 AND portal_user_id=$2 AND status='pending'`, id, account.ID)
	if err != nil {
		return err
	}
	if tag.RowsAffected() == 0 {
		return ErrNotFound
	}
	return nil
}

// AdminResourceRequests 返回全部资源/渠道申请(审批中心)。
func (s *Service) AdminResourceRequests(ctx context.Context, status string) ([]ResourceRequest, error) {
	where, args := " WHERE true", []any{}
	if status != "" {
		args = append(args, status)
		where += " AND r.status=$1"
	}
	rows, err := s.pool.Query(ctx, resourceRequestSelect+where+` ORDER BY r.created_at DESC LIMIT 500`, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	items := []ResourceRequest{}
	for rows.Next() {
		item, err := scanResourceRequest(rows)
		if err != nil {
			return nil, err
		}
		items = append(items, item)
	}
	return items, rows.Err()
}

// DecideResourceRequest 审批资源/渠道申请:通过时自动开通(marketplace 安装 /
// channel_grants 授权)。
//
// 顺序为先开通、后落审批记录:开通方法(Install/Grant)是幂等的且各自独立
// 提交,审批状态更新在同事务内与 outbox 事件一起提交。开通失败时申请保持
// pending,管理员可重试,不会出现"记录已通过但未开通"或"事务内嵌套事务"
// 的中间态;开通成功但状态提交失败时,重试会幂等收敛。
func (s *Service) DecideResourceRequest(ctx context.Context, id, status, note, actorID string) (ResourceRequest, error) {
	if status != "approved" && status != "rejected" {
		return ResourceRequest{}, errors.New("审批状态无效")
	}
	if len(note) > 4000 {
		return ResourceRequest{}, errors.New("审批备注过长")
	}
	var userID, resourceType, resourceCode string
	err := s.pool.QueryRow(ctx, `SELECT portal_user_id::text,resource_type,resource_code FROM gateway.resource_access_requests WHERE id=$1 AND status='pending'`, id).Scan(&userID, &resourceType, &resourceCode)
	if errors.Is(err, pgx.ErrNoRows) {
		return ResourceRequest{}, ErrNotFound
	}
	if err != nil {
		return ResourceRequest{}, err
	}
	if status == "approved" {
		switch resourceType {
		case "mcp_server", "skill", "digital_employee":
			if s.market != nil {
				// 自动安装到申请用户工作区(use 等级,幂等)。
				if _, err = s.market.Install(ctx, resourceType, resourceCode, userID, "use"); err != nil {
					return ResourceRequest{}, err
				}
			}
		case "channel":
			// 渠道审批通过 = 写入 channel_grants 用户级授权(幂等)。
			if s.channels != nil {
				var channelID string
				var enabled bool
				if err = s.pool.QueryRow(ctx, `SELECT id::text,enabled FROM gateway.channels WHERE code=$1`, resourceCode).Scan(&channelID, &enabled); err != nil {
					return ResourceRequest{}, errors.New("渠道不存在，无法开通")
				}
				if !enabled {
					return ResourceRequest{}, errors.New("渠道已停用，无法开通")
				}
				if err = s.channels.Grant(ctx, channelID, userID, actorID, "approval"); err != nil {
					return ResourceRequest{}, err
				}
			}
		}
	}
	eventID, _ := platformid.NewUUID()
	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return ResourceRequest{}, err
	}
	defer func() { _ = tx.Rollback(ctx) }()
	tag, err := tx.Exec(ctx, `UPDATE gateway.resource_access_requests SET status=$2,decision_note=$3,decided_by=$4,decided_at=clock_timestamp(),updated_at=clock_timestamp() WHERE id=$1 AND status='pending'`, id, status, strings.TrimSpace(note), actorID)
	if err != nil {
		return ResourceRequest{}, err
	}
	if tag.RowsAffected() == 0 {
		// 并发审批:后到者失败,开通动作已幂等,无副作用残留。
		return ResourceRequest{}, ErrNotFound
	}
	payload, _ := json.Marshal(map[string]any{"request_id": id, "portal_user_id": userID, "resource_type": resourceType, "resource_code": resourceCode, "status": status, "actor_id": actorID})
	if _, err = tx.Exec(ctx, `INSERT INTO gateway.outbox_events(event_id,event_type,event_version,aggregate_type,aggregate_id,payload) VALUES($1,'resource_access.decided',1,'resource_access_request',$2,$3)`, eventID, id, payload); err != nil {
		return ResourceRequest{}, err
	}
	if err = tx.Commit(ctx); err != nil {
		return ResourceRequest{}, err
	}
	oplog.Record(ctx, s.pool, nil, actorID, "", "approval.decide", resourceType, id, map[string]any{
		"resource_code": resourceCode, "portal_user_id": userID, "status": status, "note": strings.TrimSpace(note),
	})
	return scanResourceRequest(s.pool.QueryRow(ctx, resourceRequestSelect+` WHERE r.id=$1`, id))
}