root
b2605abd6a
- Added `EnableWebSearch` and `KnowledgeGroupIDs` fields to `KolGenerationSubmitRequest`. - Updated `Submit` method in `KolGenerationService` to handle new request fields. - Integrated web search and knowledge group validation based on card configuration. - Introduced caching mechanisms in `KolPackageService`, `KolPromptService`, and `KolSubscriptionAdminService` to improve performance. - Implemented knowledge context resolution in `KolGenerationWorker` to enrich prompts with relevant knowledge snippets. - Added utility functions for handling card configuration in both backend and frontend. - Created tests for knowledge extraction and rendering to ensure accuracy and reliability.
265 lines
8.7 KiB
Go
265 lines
8.7 KiB
Go
package app
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"time"
|
|
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
|
|
"github.com/geo-platform/tenant-api/internal/shared/auditlog"
|
|
sharedcache "github.com/geo-platform/tenant-api/internal/shared/cache"
|
|
"github.com/geo-platform/tenant-api/internal/shared/middleware"
|
|
"github.com/geo-platform/tenant-api/internal/shared/response"
|
|
"github.com/geo-platform/tenant-api/internal/tenant/repository"
|
|
)
|
|
|
|
type ManualBindKolSubscriptionInput struct {
|
|
TenantID int64 `json:"tenant_id" binding:"required"`
|
|
PackageID int64 `json:"package_id" binding:"required"`
|
|
EndAt *time.Time `json:"end_at"`
|
|
}
|
|
|
|
type KolSubscriptionAdminService struct {
|
|
pool *pgxpool.Pool
|
|
auditLogs *auditlog.AsyncWriter
|
|
marketplaceRepo repository.KolMarketplaceRepository
|
|
subRepo repository.KolSubscriptionRepository
|
|
promptRepo repository.KolPromptRepository
|
|
cache sharedcache.Cache
|
|
}
|
|
|
|
func NewKolSubscriptionAdminService(
|
|
pool *pgxpool.Pool,
|
|
auditLogs *auditlog.AsyncWriter,
|
|
marketplaceRepo repository.KolMarketplaceRepository,
|
|
subRepo repository.KolSubscriptionRepository,
|
|
promptRepo repository.KolPromptRepository,
|
|
) *KolSubscriptionAdminService {
|
|
return &KolSubscriptionAdminService{
|
|
pool: pool,
|
|
auditLogs: auditLogs,
|
|
marketplaceRepo: marketplaceRepo,
|
|
subRepo: subRepo,
|
|
promptRepo: promptRepo,
|
|
}
|
|
}
|
|
|
|
func (s *KolSubscriptionAdminService) WithCache(c sharedcache.Cache) *KolSubscriptionAdminService {
|
|
s.cache = c
|
|
return s
|
|
}
|
|
|
|
func (s *KolSubscriptionAdminService) Approve(ctx context.Context, id int64, endAt *time.Time, operatorID int64) (*KolSubscriptionResponse, error) {
|
|
tx, err := s.pool.Begin(ctx)
|
|
if err != nil {
|
|
return nil, response.ErrInternal(50096, "kol_subscription_approve_tx_failed", "failed to begin subscription approval transaction")
|
|
}
|
|
defer tx.Rollback(ctx)
|
|
|
|
subTx := repository.NewKolSubscriptionRepository(tx)
|
|
marketplaceTx := repository.NewKolMarketplaceRepository(tx)
|
|
promptTx := repository.NewKolPromptRepository(tx)
|
|
|
|
sub, err := subTx.GetByIDAny(ctx, id)
|
|
if err != nil {
|
|
return nil, response.ErrInternal(50093, "kol_subscription_query_failed", err.Error())
|
|
}
|
|
if sub == nil {
|
|
return nil, errKolSubscriptionNotFound
|
|
}
|
|
if sub.Status != "pending" {
|
|
return nil, errKolSubscriptionNotPending
|
|
}
|
|
|
|
pkg, err := marketplaceTx.GetPublished(ctx, sub.PackageID)
|
|
if err != nil {
|
|
return nil, response.ErrInternal(50090, "kol_marketplace_query_failed", err.Error())
|
|
}
|
|
if pkg == nil {
|
|
return nil, errKolMarketplacePackageNotFound
|
|
}
|
|
|
|
if err := subTx.Approve(ctx, id, sub.TenantID, operatorID, endAt); err != nil {
|
|
return nil, response.ErrInternal(50097, "kol_subscription_approve_failed", err.Error())
|
|
}
|
|
if err := grantActivePackagePrompts(ctx, subTx, promptTx, sub.TenantID, id, pkg.TenantID, pkg.ID); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
updated, err := subTx.GetByIDAny(ctx, id)
|
|
if err != nil {
|
|
return nil, response.ErrInternal(50093, "kol_subscription_query_failed", err.Error())
|
|
}
|
|
|
|
if err := tx.Commit(ctx); err != nil {
|
|
return nil, response.ErrInternal(50098, "kol_subscription_approve_commit_failed", "failed to commit subscription approval")
|
|
}
|
|
|
|
InvalidateKolPromptCaches(ctx, s.cache)
|
|
s.logAdminAction(ctx, "approve", operatorID, updated.TenantID, updated.ID, map[string]interface{}{
|
|
"id": updated.ID,
|
|
"tenant_id": updated.TenantID,
|
|
"package_id": updated.PackageID,
|
|
"status": updated.Status,
|
|
"end_at": timeStringPtr(endAt),
|
|
})
|
|
return newKolSubscriptionResponse(updated), nil
|
|
}
|
|
|
|
func (s *KolSubscriptionAdminService) ManualBind(ctx context.Context, input ManualBindKolSubscriptionInput, operatorID int64) (*KolSubscriptionResponse, error) {
|
|
if input.TenantID <= 0 {
|
|
return nil, response.ErrBadRequest(40071, "kol_subscription_tenant_required", "tenant_id is required")
|
|
}
|
|
if input.PackageID <= 0 {
|
|
return nil, response.ErrBadRequest(40072, "kol_subscription_package_required", "package_id is required")
|
|
}
|
|
|
|
tx, err := s.pool.Begin(ctx)
|
|
if err != nil {
|
|
return nil, response.ErrInternal(50099, "kol_subscription_bind_tx_failed", "failed to begin manual bind transaction")
|
|
}
|
|
defer tx.Rollback(ctx)
|
|
|
|
subTx := repository.NewKolSubscriptionRepository(tx)
|
|
marketplaceTx := repository.NewKolMarketplaceRepository(tx)
|
|
promptTx := repository.NewKolPromptRepository(tx)
|
|
|
|
pkg, err := marketplaceTx.GetPublished(ctx, input.PackageID)
|
|
if err != nil {
|
|
return nil, response.ErrInternal(50090, "kol_marketplace_query_failed", err.Error())
|
|
}
|
|
if pkg == nil {
|
|
return nil, errKolMarketplacePackageNotFound
|
|
}
|
|
|
|
existing, err := subTx.GetActiveForTenant(ctx, input.TenantID, input.PackageID)
|
|
if err != nil {
|
|
return nil, response.ErrInternal(50093, "kol_subscription_query_failed", err.Error())
|
|
}
|
|
if existing != nil {
|
|
return nil, errKolSubscriptionExists
|
|
}
|
|
|
|
sub, err := subTx.Create(ctx, input.TenantID, input.PackageID)
|
|
if err != nil {
|
|
if isUniqueConstraintError(err, "uk_kol_subscription_active") {
|
|
return nil, errKolSubscriptionExists
|
|
}
|
|
return nil, response.ErrInternal(50100, "kol_subscription_create_failed", err.Error())
|
|
}
|
|
|
|
if err := subTx.Approve(ctx, sub.ID, input.TenantID, operatorID, input.EndAt); err != nil {
|
|
return nil, response.ErrInternal(50097, "kol_subscription_approve_failed", err.Error())
|
|
}
|
|
if err := grantActivePackagePrompts(ctx, subTx, promptTx, input.TenantID, sub.ID, pkg.TenantID, pkg.ID); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
updated, err := subTx.GetByIDAny(ctx, sub.ID)
|
|
if err != nil {
|
|
return nil, response.ErrInternal(50093, "kol_subscription_query_failed", err.Error())
|
|
}
|
|
|
|
if err := tx.Commit(ctx); err != nil {
|
|
return nil, response.ErrInternal(50101, "kol_subscription_bind_commit_failed", "failed to commit manual bind")
|
|
}
|
|
|
|
InvalidateKolPromptCaches(ctx, s.cache)
|
|
s.logAdminAction(ctx, "manual_bind", operatorID, updated.TenantID, updated.ID, map[string]interface{}{
|
|
"id": updated.ID,
|
|
"tenant_id": updated.TenantID,
|
|
"package_id": updated.PackageID,
|
|
"status": updated.Status,
|
|
"end_at": timeStringPtr(input.EndAt),
|
|
})
|
|
return newKolSubscriptionResponse(updated), nil
|
|
}
|
|
|
|
func (s *KolSubscriptionAdminService) Revoke(ctx context.Context, id int64, operatorID int64) error {
|
|
tx, err := s.pool.Begin(ctx)
|
|
if err != nil {
|
|
return response.ErrInternal(50102, "kol_subscription_revoke_tx_failed", "failed to begin revoke transaction")
|
|
}
|
|
defer tx.Rollback(ctx)
|
|
|
|
subTx := repository.NewKolSubscriptionRepository(tx)
|
|
sub, err := subTx.GetByIDAny(ctx, id)
|
|
if err != nil {
|
|
return response.ErrInternal(50093, "kol_subscription_query_failed", err.Error())
|
|
}
|
|
if sub == nil {
|
|
return errKolSubscriptionNotFound
|
|
}
|
|
|
|
if err := subTx.RevokeByTenant(ctx, id, sub.TenantID); err != nil {
|
|
return response.ErrInternal(50103, "kol_subscription_revoke_failed", err.Error())
|
|
}
|
|
if err := subTx.RevokeAccessRowsBySubscription(ctx, id, sub.TenantID); err != nil {
|
|
return response.ErrInternal(50104, "kol_subscription_access_revoke_failed", err.Error())
|
|
}
|
|
|
|
if err := tx.Commit(ctx); err != nil {
|
|
return response.ErrInternal(50105, "kol_subscription_revoke_commit_failed", "failed to commit subscription revoke")
|
|
}
|
|
|
|
InvalidateKolPromptCaches(ctx, s.cache)
|
|
s.logAdminAction(ctx, "revoke", operatorID, sub.TenantID, sub.ID, map[string]interface{}{
|
|
"id": sub.ID,
|
|
"tenant_id": sub.TenantID,
|
|
"package_id": sub.PackageID,
|
|
"status": "revoked",
|
|
})
|
|
return nil
|
|
}
|
|
|
|
func grantActivePackagePrompts(
|
|
ctx context.Context,
|
|
subRepo repository.KolSubscriptionRepository,
|
|
promptRepo repository.KolPromptRepository,
|
|
subscriberTenantID, subscriptionID, creatorTenantID, packageID int64,
|
|
) error {
|
|
prompts, err := promptRepo.ListActiveByPackage(ctx, creatorTenantID, packageID)
|
|
if err != nil {
|
|
return response.ErrInternal(50092, "kol_marketplace_prompt_query_failed", err.Error())
|
|
}
|
|
for _, prompt := range prompts {
|
|
if err := subRepo.CreateAccessRow(ctx, repository.CreateAccessRowInput{
|
|
TenantID: subscriberTenantID,
|
|
SubscriptionID: subscriptionID,
|
|
PackageID: packageID,
|
|
PromptID: prompt.ID,
|
|
}); err != nil {
|
|
return response.ErrInternal(50084, "kol_subscription_access_create_failed", err.Error())
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *KolSubscriptionAdminService) logAdminAction(
|
|
ctx context.Context,
|
|
action string,
|
|
operatorID, targetTenantID, subscriptionID int64,
|
|
after map[string]interface{},
|
|
) {
|
|
if s.auditLogs == nil {
|
|
return
|
|
}
|
|
|
|
afterJSON, _ := json.Marshal(after)
|
|
result := "success"
|
|
resourceType := "kol_subscription"
|
|
requestID := middleware.RequestIDFromContext(ctx)
|
|
s.auditLogs.Log(auditlog.Entry{
|
|
OperatorID: operatorID,
|
|
TenantID: &targetTenantID,
|
|
Module: "kol_admin",
|
|
Action: action,
|
|
ResourceType: &resourceType,
|
|
ResourceID: &subscriptionID,
|
|
RequestID: nilIfEmptyString(requestID),
|
|
AfterJSON: afterJSON,
|
|
Result: &result,
|
|
})
|
|
}
|