package portal import ( "context" "encoding/json" "errors" "strings" "time" "aigateway.local/core/internal/identity" platformid "aigateway.local/core/internal/platform/id" "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 安装)。 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("审批备注过长") } 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 } var userID, resourceType, resourceCode string if err = tx.QueryRow(ctx, `SELECT portal_user_id::text,resource_type,resource_code FROM gateway.resource_access_requests WHERE id=$1`, id).Scan(&userID, &resourceType, &resourceCode); err != nil { return ResourceRequest{}, err } if status == "approved" && resourceType != "channel" && s.market != nil { // 自动安装到申请用户工作区(use 等级)。 if _, err = s.market.Install(ctx, resourceType, resourceCode, userID, "use"); err != nil { return ResourceRequest{}, err } } 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 } return scanResourceRequest(s.pool.QueryRow(ctx, resourceRequestSelect+` WHERE r.id=$1`, id)) }