package workbench import ( "context" "encoding/json" "errors" "fmt" "sort" "strings" "time" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" ) // MarketItem is the lightweight unified catalog row for a published resource, // regardless of which of the three resource tables it lives in. type MarketItem struct { PermissionLevel string `json:"permission_level,omitempty"` Type string `json:"type"` Code string `json:"code"` Name string `json:"name"` Description string `json:"description"` CategoryID *string `json:"category_id,omitempty"` CategoryName string `json:"category_name"` Tags []string `json:"tags"` DepartmentIDs []string `json:"department_ids"` UpdatedAt time.Time `json:"updated_at"` } // Category is a shared marketplace category, optionally scoped to one resource // type (empty resource_type = global). type Category struct { ID string `json:"id"` Name string `json:"name"` Description string `json:"description"` ResourceType string `json:"resource_type"` SortOrder int `json:"sort_order"` CreatedAt time.Time `json:"created_at"` } // MarketplaceService provides the shared cross-resource surface: categories, // the unified catalog, and portal workspace installations. type MarketplaceService struct { assets *Service mcpServers *MCPServerService skills *SkillService employees *DigitalEmployeeService } func NewMarketplaceService(assets *Service, mcpServers *MCPServerService, skills *SkillService, employees *DigitalEmployeeService) *MarketplaceService { return &MarketplaceService{assets: assets, mcpServers: mcpServers, skills: skills, employees: employees} } // --- 分类 --- func (s *MarketplaceService) ListCategories(ctx context.Context, resourceType string) ([]Category, error) { rows, err := s.assets.pool.Query(ctx, `SELECT id::text,name,description,resource_type,sort_order,created_at FROM gateway.marketplace_categories WHERE resource_type=$1 OR resource_type='' ORDER BY sort_order,created_at`, resourceType) if err != nil { return nil, err } defer rows.Close() items := []Category{} for rows.Next() { var c Category if err = rows.Scan(&c.ID, &c.Name, &c.Description, &c.ResourceType, &c.SortOrder, &c.CreatedAt); err != nil { return nil, err } items = append(items, c) } return items, rows.Err() } func (s *MarketplaceService) CreateCategory(ctx context.Context, input Category, actorID string) (Category, error) { input.Name = strings.TrimSpace(input.Name) input.Description = strings.TrimSpace(input.Description) if input.Name == "" || len(input.Name) > 64 || len(input.Description) > 1000 { return Category{}, errors.New("分类名称或描述格式无效") } switch input.ResourceType { case "", "mcp_server", "skill", "digital_employee": default: return Category{}, errors.New("无效的分类资源类型") } if input.SortOrder < 0 { input.SortOrder = 0 } id, err := newUUID() if err != nil { return Category{}, err } tx, err := s.assets.pool.Begin(ctx) if err != nil { return Category{}, err } defer rollback(ctx, tx) _, err = tx.Exec(ctx, `INSERT INTO gateway.marketplace_categories(id,name,description,resource_type,sort_order,created_by) VALUES($1,$2,$3,$4,$5,$6)`, id, input.Name, input.Description, input.ResourceType, input.SortOrder, actorID) if err != nil { if strings.Contains(err.Error(), "duplicate") { return Category{}, ErrConflict } return Category{}, err } if err = emit(ctx, tx, "marketplace_category.created", "marketplace_category", id, actorID, nil); err != nil { return Category{}, err } if err = tx.Commit(ctx); err != nil { return Category{}, err } input.ID = id input.CreatedAt = time.Now() return input, nil } func (s *MarketplaceService) DeleteCategory(ctx context.Context, id, actorID string) error { tx, err := s.assets.pool.Begin(ctx) if err != nil { return err } defer rollback(ctx, tx) tag, err := tx.Exec(ctx, `DELETE FROM gateway.marketplace_categories WHERE id=$1`, id) if err != nil { return err } if tag.RowsAffected() == 0 { return ErrNotFound } if err = emit(ctx, tx, "marketplace_category.deleted", "marketplace_category", id, actorID, nil); err != nil { return err } return tx.Commit(ctx) } // validCategoryID resolves and verifies an optional category reference. A nil // or blank id yields a nil category; a non-empty id must exist and match the // resource type scope (global categories apply to every type). func validCategoryID(ctx context.Context, pool *pgxpool.Pool, id *string, resourceType string) (*string, error) { if id == nil || strings.TrimSpace(*id) == "" { return nil, nil } var exists bool err := pool.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM gateway.marketplace_categories WHERE id=$1 AND (resource_type='' OR resource_type=$2))`, *id, resourceType).Scan(&exists) if err != nil { return nil, err } if !exists { return nil, errors.New("分类不存在或与资源类型不匹配") } return id, nil } // --- 统一目录 --- // Catalog returns published, enabled resources across all three types, merged // and sorted by update time. Filters are optional and compose with AND. func (s *MarketplaceService) Catalog(ctx context.Context, resourceType, categoryID, tag, query string, limit int) ([]MarketItem, error) { if limit < 1 { limit = 100 } if limit > 200 { limit = 200 } items := make([]MarketItem, 0, limit) types := []string{"mcp_server", "skill", "digital_employee"} if resourceType != "" { types = []string{resourceType} } for _, typ := range types { // Filters are built per table alias so name/description don't clash // with the LEFT JOINed category columns. where, args := marketFilters(aliasFor(typ), categoryID, tag, query) if err := s.catalogTable(ctx, typ, where, args, limit, &items); err != nil { return nil, err } } sort.SliceStable(items, func(i, j int) bool { return items[i].UpdatedAt.After(items[j].UpdatedAt) }) if len(items) > limit { items = items[:limit] } return items, nil } func aliasFor(resourceType string) string { switch resourceType { case "mcp_server": return "m" case "skill": return "s" default: return "d" } } func marketFilters(alias, categoryID, tag, query string) (string, []any) { clauses := []string{} args := []any{} if categoryID != "" { args = append(args, categoryID) clauses = append(clauses, fmt.Sprintf("%s.category_id=$%d", alias, len(args))) } if tag != "" { args = append(args, tag) clauses = append(clauses, fmt.Sprintf("$%d = ANY(%s.tags)", len(args), alias)) } if query != "" { args = append(args, "%"+query+"%") clauses = append(clauses, fmt.Sprintf("(%s.name ILIKE $%d OR %s.code ILIKE $%d OR %s.description ILIKE $%d)", alias, len(args), alias, len(args), alias, len(args))) } where := "status='published' AND enabled" if len(clauses) > 0 { where += " AND " + strings.Join(clauses, " AND ") } return where, args } func (s *MarketplaceService) catalogTable(ctx context.Context, typ, where string, args []any, limit int, out *[]MarketItem) error { // Filters occupy $1..$N in the WHERE clause; the LIMIT placeholder must be // the next position (N+1), not a hardcoded $1 which would collide with a // category filter arg and land a uuid in LIMIT. limitParam := fmt.Sprintf("$%d", len(args)+1) var query string switch typ { case "mcp_server": query = `SELECT 'mcp_server',m.code,m.name,m.description,m.category_id::text,coalesce(c.name,''),m.tags,m.department_ids::text[],m.updated_at FROM gateway.mcp_servers m LEFT JOIN gateway.marketplace_categories c ON c.id=m.category_id WHERE ` + where + ` ORDER BY m.updated_at DESC LIMIT ` + limitParam case "skill": query = `SELECT 'skill',s.code,s.name,s.description,s.category_id::text,coalesce(c.name,''),s.tags,s.department_ids::text[],s.updated_at FROM gateway.skills s LEFT JOIN gateway.marketplace_categories c ON c.id=s.category_id WHERE ` + where + ` ORDER BY s.updated_at DESC LIMIT ` + limitParam case "digital_employee": query = `SELECT 'digital_employee',d.code,d.name,d.description,d.category_id::text,coalesce(c.name,''),d.tags,d.department_ids::text[],d.updated_at FROM gateway.digital_employees d LEFT JOIN gateway.marketplace_categories c ON c.id=d.category_id WHERE ` + where + ` ORDER BY d.updated_at DESC LIMIT ` + limitParam default: return errors.New("未知的资源类型") } queryArgs := append(append([]any{}, args...), limit) rows, err := s.assets.pool.Query(ctx, query, queryArgs...) if err != nil { return err } defer rows.Close() for rows.Next() { var item MarketItem if err = rows.Scan(&item.Type, &item.Code, &item.Name, &item.Description, &item.CategoryID, &item.CategoryName, &item.Tags, &item.DepartmentIDs, &item.UpdatedAt); err != nil { return err } *out = append(*out, item) } return rows.Err() } // Detail returns the full resource payload for a published item by type+code. // MCP servers are returned without their encrypted headers; the consumer calls // a dedicated endpoint to test connectivity. func (s *MarketplaceService) Detail(ctx context.Context, resourceType, code string) (MarketItem, json.RawMessage, error) { var item MarketItem var detail any var err error switch resourceType { case "mcp_server": server, getErr := s.mcpServers.GetPublishedByCode(ctx, code) if getErr != nil { return item, nil, getErr } item = MarketItem{Type: "mcp_server", Code: server.Code, Name: server.Name, Description: server.Description, CategoryID: server.CategoryID, CategoryName: server.CategoryName, Tags: server.Tags, DepartmentIDs: server.DepartmentIDs, UpdatedAt: server.UpdatedAt} // 门户详情不得暴露 endpoint_url 与 has_secret_headers:内网服务拓扑和 // 密钥状态仅管理端可见(运行时列表同样省略该字段)。 raw, err := json.Marshal(server) if err != nil { return item, nil, err } var sanitized map[string]any if err := json.Unmarshal(raw, &sanitized); err != nil { return item, nil, err } delete(sanitized, "endpoint_url") delete(sanitized, "has_secret_headers") detail = sanitized case "skill": skill, getErr := s.skills.GetPublishedByCode(ctx, code) if getErr != nil { return item, nil, getErr } item = MarketItem{Type: "skill", Code: skill.Code, Name: skill.Name, Description: skill.Description, CategoryID: skill.CategoryID, CategoryName: skill.CategoryName, Tags: skill.Tags, DepartmentIDs: skill.DepartmentIDs, UpdatedAt: skill.UpdatedAt} detail = skill case "digital_employee": employee, getErr := s.employees.GetPublishedByCode(ctx, code) if getErr != nil { return item, nil, getErr } item = MarketItem{Type: "digital_employee", Code: employee.Code, Name: employee.Name, Description: employee.Description, CategoryID: employee.CategoryID, CategoryName: employee.CategoryName, Tags: employee.Tags, DepartmentIDs: employee.DepartmentIDs, UpdatedAt: employee.UpdatedAt} detail = employee default: return item, nil, errors.New("未知的资源类型") } raw, err := json.Marshal(detail) if err != nil { return item, nil, err } return item, raw, nil } // --- 安装(工作区绑定 + 权限) --- // Install records a portal user's workspace binding to a published resource. // It is the permission grant that lets a cross-department user invoke a // resource that would otherwise be invisible to them. // Install 记录安装;permissionLevel 为 view/use/manage(默认 use)。 func (s *MarketplaceService) Install(ctx context.Context, resourceType, code, portalUserID, permissionLevel string) (bool, error) { resourceID, err := s.publishedResourceID(ctx, resourceType, code) if err != nil { return false, err } switch permissionLevel { case "", "view", "use", "manage": default: return false, errors.New("权限等级必须是 view/use/manage") } if permissionLevel == "" { permissionLevel = "use" } id, err := newUUID() if err != nil { return false, err } tx, err := s.assets.pool.Begin(ctx) if err != nil { return false, err } defer rollback(ctx, tx) tag, err := tx.Exec(ctx, `INSERT INTO gateway.marketplace_installations(id,resource_type,resource_id,portal_user_id,permission_level) VALUES($1,$2,$3,$4,$5) ON CONFLICT(resource_type,resource_id,portal_user_id) DO UPDATE SET permission_level=$5`, id, resourceType, resourceID, portalUserID, permissionLevel) if err != nil { return false, err } created := tag.RowsAffected() == 1 // Idempotent re-install: only emit the audit event when a row was created. if created { if err = emit(ctx, tx, "marketplace.installed", resourceType, resourceID, "", map[string]any{"code": code, "portal_user_id": portalUserID}); err != nil { return false, err } } if err = tx.Commit(ctx); err != nil { return false, err } return created, nil } func (s *MarketplaceService) Uninstall(ctx context.Context, resourceType, code, portalUserID string) error { // 卸载不受 enabled/status 限制:管理员停用或归档资源后,用户仍能 // 移除自己的安装,否则安装行永久卡死、列表永远显示。 resourceID, err := s.resourceIDByCode(ctx, resourceType, code) if err != nil { return err } tx, err := s.assets.pool.Begin(ctx) if err != nil { return err } defer rollback(ctx, tx) tag, err := tx.Exec(ctx, `DELETE FROM gateway.marketplace_installations WHERE resource_type=$1 AND resource_id=$2 AND portal_user_id=$3`, resourceType, resourceID, portalUserID) if err != nil { return err } if tag.RowsAffected() == 0 { return ErrNotFound } if err = emit(ctx, tx, "marketplace.uninstalled", resourceType, resourceID, "", map[string]any{"code": code, "portal_user_id": portalUserID}); err != nil { return err } return tx.Commit(ctx) } // ListInstalled returns the codes the portal user has installed, grouped by // resource type, joined with the live resource name for display. func (s *MarketplaceService) ListInstalled(ctx context.Context, portalUserID string) ([]MarketItem, error) { rows, err := s.assets.pool.Query(ctx, `SELECT resource_type,resource_id::text FROM gateway.marketplace_installations WHERE portal_user_id=$1 ORDER BY created_at DESC`, portalUserID) if err != nil { return nil, err } type bound struct{ typ, id string } bounds := []bound{} for rows.Next() { var b bound if err = rows.Scan(&b.typ, &b.id); err != nil { rows.Close() return nil, err } bounds = append(bounds, b) } rows.Close() if err = rows.Err(); err != nil { return nil, err } items := make([]MarketItem, 0, len(bounds)) for _, b := range bounds { item, ok, err := s.resourceByID(ctx, b.typ, b.id) if err != nil { return nil, err } if ok { items = append(items, item) } } return items, nil } // Installed reports whether a portal user has an installation for the resource. func (s *MarketplaceService) Installed(ctx context.Context, resourceType, resourceID, portalUserID string) (bool, error) { var exists bool err := s.assets.pool.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM gateway.marketplace_installations WHERE resource_type=$1 AND resource_id=$2 AND portal_user_id=$3)`, resourceType, resourceID, portalUserID).Scan(&exists) return exists, err } // PortalUserForAPIKey resolves the portal user that owns an API key, used to // evaluate installation-based visibility at runtime. func (s *MarketplaceService) PortalUserForAPIKey(ctx context.Context, apiKeyID string) (string, bool, error) { var userID *string err := s.assets.pool.QueryRow(ctx, `SELECT portal_user_id::text FROM gateway.api_keys WHERE id=$1`, apiKeyID).Scan(&userID) if err != nil { if errors.Is(err, pgx.ErrNoRows) { return "", false, nil } return "", false, err } if userID == nil || *userID == "" { return "", false, nil } return *userID, true, nil } func (s *MarketplaceService) publishedResourceID(ctx context.Context, resourceType, code string) (string, error) { var id string var err error switch resourceType { case "mcp_server": err = s.assets.pool.QueryRow(ctx, `SELECT id::text FROM gateway.mcp_servers WHERE code=$1 AND status='published' AND enabled`, code).Scan(&id) case "skill": err = s.assets.pool.QueryRow(ctx, `SELECT id::text FROM gateway.skills WHERE code=$1 AND status='published' AND enabled`, code).Scan(&id) case "digital_employee": err = s.assets.pool.QueryRow(ctx, `SELECT id::text FROM gateway.digital_employees WHERE code=$1 AND status='published' AND enabled`, code).Scan(&id) default: return "", errors.New("未知的资源类型") } return id, mapNotFound(err) } // resourceIDByCode 按 code 解析资源 ID,不限制启用/发布状态(卸载场景使用)。 func (s *MarketplaceService) resourceIDByCode(ctx context.Context, resourceType, code string) (string, error) { var id string var err error switch resourceType { case "mcp_server": err = s.assets.pool.QueryRow(ctx, `SELECT id::text FROM gateway.mcp_servers WHERE code=$1`, code).Scan(&id) case "skill": err = s.assets.pool.QueryRow(ctx, `SELECT id::text FROM gateway.skills WHERE code=$1`, code).Scan(&id) case "digital_employee": err = s.assets.pool.QueryRow(ctx, `SELECT id::text FROM gateway.digital_employees WHERE code=$1`, code).Scan(&id) default: return "", errors.New("未知的资源类型") } return id, mapNotFound(err) } func (s *MarketplaceService) resourceByID(ctx context.Context, resourceType, id string) (MarketItem, bool, error) { var item MarketItem switch resourceType { case "mcp_server": server, err := s.mcpServers.Get(ctx, id) if err != nil { return item, false, nil } item = MarketItem{Type: "mcp_server", Code: server.Code, Name: server.Name, Description: server.Description, CategoryID: server.CategoryID, CategoryName: server.CategoryName, Tags: server.Tags, UpdatedAt: server.UpdatedAt} case "skill": skill, err := s.skills.Get(ctx, id) if err != nil { return item, false, nil } item = MarketItem{Type: "skill", Code: skill.Code, Name: skill.Name, Description: skill.Description, CategoryID: skill.CategoryID, CategoryName: skill.CategoryName, Tags: skill.Tags, UpdatedAt: skill.UpdatedAt} case "digital_employee": employee, err := s.employees.Get(ctx, id) if err != nil { return item, false, nil } item = MarketItem{Type: "digital_employee", Code: employee.Code, Name: employee.Name, Description: employee.Description, CategoryID: employee.CategoryID, CategoryName: employee.CategoryName, Tags: employee.Tags, UpdatedAt: employee.UpdatedAt} default: return item, false, nil } return item, true, nil }