From 18cdbf25f8ea923fbed574f2670ec848e84bfeba Mon Sep 17 00:00:00 2001 From: liangxu Date: Wed, 13 May 2026 17:31:22 +0800 Subject: [PATCH] fix: harden competitor create and batch kol stats --- server/internal/tenant/app/brand_service.go | 19 +++- .../tenant/app/kol_dashboard_service.go | 13 ++- .../tenant/app/kol_package_service.go | 87 ++++++++++++++----- .../tenant/repository/kol_prompt_repo.go | 34 ++++++++ .../repository/kol_subscription_repo.go | 70 +++++++++++++++ 5 files changed, 195 insertions(+), 28 deletions(-) diff --git a/server/internal/tenant/app/brand_service.go b/server/internal/tenant/app/brand_service.go index a51e0dc..a53b3b5 100644 --- a/server/internal/tenant/app/brand_service.go +++ b/server/internal/tenant/app/brand_service.go @@ -639,6 +639,20 @@ func (s *BrandService) ListCompetitors(ctx context.Context, brandID int64) ([]Co func (s *BrandService) CreateCompetitor(ctx context.Context, brandID int64, req CompetitorRequest) (*CompetitorResponse, error) { actor := auth.MustActor(ctx) + req.Name = strings.TrimSpace(req.Name) + req.Website = normalizeOptionalString(req.Website) + req.Description = normalizeOptionalString(req.Description) + if req.Name == "" { + return nil, response.ErrBadRequest(40001, "invalid_params", "name is required") + } + exists, err := s.brandExists(ctx, actor.TenantID, brandID) + if err != nil { + return nil, err + } + if !exists { + return nil, response.ErrNotFound(40420, "brand_not_found", "brand not found") + } + var plJSON []byte if req.ProductLinesJSON != nil { plJSON = *req.ProductLinesJSON @@ -646,11 +660,14 @@ func (s *BrandService) CreateCompetitor(ctx context.Context, brandID int64, req var id int64 var ca interface{} - err := s.pool.QueryRow(ctx, ` + err = s.pool.QueryRow(ctx, ` INSERT INTO competitors (tenant_id, brand_id, name, website, description, product_lines_json) VALUES ($1, $2, $3, $4, $5, $6) RETURNING id, created_at `, actor.TenantID, brandID, req.Name, req.Website, req.Description, plJSON).Scan(&id, &ca) if err != nil { + if isForeignKeyConstraintError(err) { + return nil, response.ErrNotFound(40420, "brand_not_found", "brand not found") + } return nil, response.ErrInternal(50010, "create_failed", "failed to create competitor") } invalidateBrandCaches(ctx, s.cache, actor.TenantID, brandID) diff --git a/server/internal/tenant/app/kol_dashboard_service.go b/server/internal/tenant/app/kol_dashboard_service.go index 4f10fa3..00b8921 100644 --- a/server/internal/tenant/app/kol_dashboard_service.go +++ b/server/internal/tenant/app/kol_dashboard_service.go @@ -111,19 +111,18 @@ func (s *KolDashboardService) PerPackage(ctx context.Context, actor auth.Actor) usageByPackage[stat.PackageID] = stat } + subscriberCounts, err := s.subRepo.CountActiveSubscriberTenantsByPackages(ctx, packageIDs) + if err != nil { + return nil, response.ErrInternal(50123, "kol_dashboard_subscriber_query_failed", err.Error()) + } + result := make([]DashboardPackageStat, 0, len(packages)) for _, pkg := range packages { - subscribers, err := s.subRepo.ListActiveSubscribersForPackage(ctx, pkg.ID) - if err != nil { - return nil, response.ErrInternal(50123, "kol_dashboard_subscriber_query_failed", err.Error()) - } - - subscriberTenants := uniqueSubscriberTenants(subscribers) usage := usageByPackage[pkg.ID] result = append(result, DashboardPackageStat{ PackageID: pkg.ID, PackageName: pkg.Name, - SubscriberTenants: subscriberTenants, + SubscriberTenants: subscriberCounts[pkg.ID], UsageCount: usage.Total, FailureRate: float64(usage.Failed) / float64(maxDashboardBase(usage.Total)), }) diff --git a/server/internal/tenant/app/kol_package_service.go b/server/internal/tenant/app/kol_package_service.go index 768dec1..fc28c95 100644 --- a/server/internal/tenant/app/kol_package_service.go +++ b/server/internal/tenant/app/kol_package_service.go @@ -80,9 +80,19 @@ func (s *KolPackageService) List(ctx context.Context, actor auth.Actor) ([]KolPa return nil, response.ErrInternal(50062, "kol_package_query_failed", err.Error()) } + packageIDs := kolPackageIDs(packages) + promptCounts, err := s.packagePromptCounts(ctx, actor.TenantID, packageIDs) + if err != nil { + return nil, err + } + selfSubscriptions, err := s.packageSelfSubscriptions(ctx, actor.TenantID, packageIDs) + if err != nil { + return nil, err + } + result := make([]KolPackageResponse, 0, len(packages)) for _, pkg := range packages { - item, err := s.buildPackageResponse(ctx, profile, &pkg) + item, err := s.buildPackageResponseWithStats(profile, &pkg, promptCounts[pkg.ID], selfSubscriptions[pkg.ID]) if err != nil { return nil, err } @@ -299,30 +309,31 @@ func (s *KolPackageService) buildPackageResponse( ctx context.Context, profile *repository.KolProfile, pkg *repository.KolPackage, +) (*KolPackageResponse, error) { + promptCounts, err := s.packagePromptCounts(ctx, pkg.TenantID, []int64{pkg.ID}) + if err != nil { + return nil, err + } + + selfSubscriptions, err := s.packageSelfSubscriptions(ctx, pkg.TenantID, []int64{pkg.ID}) + if err != nil { + return nil, err + } + + return s.buildPackageResponseWithStats(profile, pkg, promptCounts[pkg.ID], selfSubscriptions[pkg.ID]) +} + +func (s *KolPackageService) buildPackageResponseWithStats( + profile *repository.KolProfile, + pkg *repository.KolPackage, + promptCount int, + selfSubscription *repository.KolSubscription, ) (*KolPackageResponse, error) { tags, err := decodeKolStringList(pkg.Tags) if err != nil { return nil, response.ErrInternal(50068, "kol_package_decode_failed", err.Error()) } - promptCount := 0 - if s.promptRepo != nil { - prompts, err := s.promptRepo.ListByPackage(ctx, pkg.TenantID, pkg.ID) - if err != nil { - return nil, response.ErrInternal(50069, "kol_prompt_query_failed", err.Error()) - } - promptCount = len(prompts) - } - - var selfSubscription *KolSubscriptionResponse - if s.subRepo != nil { - sub, err := s.subRepo.GetActiveForTenant(ctx, pkg.TenantID, pkg.ID) - if err != nil { - return nil, response.ErrInternal(50093, "kol_subscription_query_failed", err.Error()) - } - selfSubscription = newKolSubscriptionResponse(sub) - } - return &KolPackageResponse{ ID: pkg.ID, KolProfileID: pkg.KolProfileID, @@ -337,12 +348,48 @@ func (s *KolPackageService) buildPackageResponse( Status: pkg.Status, PromptCount: promptCount, SubscriberCount: 0, - SelfSubscription: selfSubscription, + SelfSubscription: newKolSubscriptionResponse(selfSubscription), CreatedAt: pkg.CreatedAt, UpdatedAt: pkg.UpdatedAt, }, nil } +func (s *KolPackageService) packagePromptCounts(ctx context.Context, tenantID int64, packageIDs []int64) (map[int64]int, error) { + if s.promptRepo == nil { + return map[int64]int{}, nil + } + counts, err := s.promptRepo.CountByPackages(ctx, tenantID, packageIDs) + if err != nil { + return nil, response.ErrInternal(50069, "kol_prompt_query_failed", err.Error()) + } + return counts, nil +} + +func (s *KolPackageService) packageSelfSubscriptions(ctx context.Context, tenantID int64, packageIDs []int64) (map[int64]*repository.KolSubscription, error) { + result := make(map[int64]*repository.KolSubscription, len(packageIDs)) + if s.subRepo == nil { + return result, nil + } + + subscriptions, err := s.subRepo.ListActiveForTenantByPackages(ctx, tenantID, packageIDs) + if err != nil { + return nil, response.ErrInternal(50093, "kol_subscription_query_failed", err.Error()) + } + for packageID, sub := range subscriptions { + subscription := sub + result[packageID] = &subscription + } + return result, nil +} + +func kolPackageIDs(packages []repository.KolPackage) []int64 { + ids := make([]int64, 0, len(packages)) + for _, pkg := range packages { + ids = append(ids, pkg.ID) + } + return ids +} + func normalizeKolOptionalString(value *string) *string { if value == nil { return nil diff --git a/server/internal/tenant/repository/kol_prompt_repo.go b/server/internal/tenant/repository/kol_prompt_repo.go index 6e8c972..cd3f081 100644 --- a/server/internal/tenant/repository/kol_prompt_repo.go +++ b/server/internal/tenant/repository/kol_prompt_repo.go @@ -86,6 +86,7 @@ type KolPromptRepository interface { Create(ctx context.Context, input CreateKolPromptInput) (*KolPrompt, error) GetByID(ctx context.Context, tenantID, id int64) (*KolPrompt, error) ListByPackage(ctx context.Context, tenantID, packageID int64) ([]KolPrompt, error) + CountByPackages(ctx context.Context, tenantID int64, packageIDs []int64) (map[int64]int, error) ListActiveByPackage(ctx context.Context, tenantID, packageID int64) ([]ActiveKolPrompt, error) UpdateMetadata(ctx context.Context, input UpdateKolPromptMetadataInput) error Save(ctx context.Context, input SaveKolPromptInput) error @@ -269,6 +270,39 @@ ORDER BY sort_order ASC, created_at ASC return result, nil } +func (r *kolPromptRepository) CountByPackages(ctx context.Context, tenantID int64, packageIDs []int64) (map[int64]int, error) { + counts := make(map[int64]int, len(packageIDs)) + if len(packageIDs) == 0 { + return counts, nil + } + + rows, err := r.db.Query(ctx, ` +SELECT package_id, COUNT(*)::int +FROM kol_prompts +WHERE tenant_id = $1 + AND package_id = ANY($2::bigint[]) + AND deleted_at IS NULL +GROUP BY package_id +`, tenantID, packageIDs) + if err != nil { + return nil, err + } + defer rows.Close() + + for rows.Next() { + var packageID int64 + var count int + if err := rows.Scan(&packageID, &count); err != nil { + return nil, err + } + counts[packageID] = count + } + if err := rows.Err(); err != nil { + return nil, err + } + return counts, nil +} + func (r *kolPromptRepository) ListActiveByPackage(ctx context.Context, tenantID, packageID int64) ([]ActiveKolPrompt, error) { rows, err := r.q.ListActiveKolPromptsByPackage(ctx, generated.ListActiveKolPromptsByPackageParams{ PackageID: packageID, diff --git a/server/internal/tenant/repository/kol_subscription_repo.go b/server/internal/tenant/repository/kol_subscription_repo.go index 9bbcf8d..2c23bc8 100644 --- a/server/internal/tenant/repository/kol_subscription_repo.go +++ b/server/internal/tenant/repository/kol_subscription_repo.go @@ -60,8 +60,10 @@ type KolSubscriptionRepository interface { GetByID(ctx context.Context, tenantID, id int64) (*KolSubscription, error) GetByIDAny(ctx context.Context, id int64) (*KolSubscription, error) GetActiveForTenant(ctx context.Context, tenantID, packageID int64) (*KolSubscription, error) + ListActiveForTenantByPackages(ctx context.Context, tenantID int64, packageIDs []int64) (map[int64]KolSubscription, error) ListByTenant(ctx context.Context, tenantID int64) ([]KolSubscription, error) ListActiveSubscribersForPackage(ctx context.Context, packageID int64) ([]KolSubscription, error) + CountActiveSubscriberTenantsByPackages(ctx context.Context, packageIDs []int64) (map[int64]int64, error) Approve(ctx context.Context, id, tenantID, approvedBy int64, endAt *time.Time) error RevokeByTenant(ctx context.Context, id, tenantID int64) error CreateAccessRow(ctx context.Context, input CreateAccessRowInput) error @@ -362,6 +364,41 @@ func (r *kolSubscriptionRepository) GetActiveForTenant(ctx context.Context, tena return &sub, nil } +func (r *kolSubscriptionRepository) ListActiveForTenantByPackages(ctx context.Context, tenantID int64, packageIDs []int64) (map[int64]KolSubscription, error) { + result := make(map[int64]KolSubscription, len(packageIDs)) + if len(packageIDs) == 0 { + return result, nil + } + + rows, err := r.db.Query(ctx, ` +SELECT DISTINCT ON (package_id) + id, tenant_id, package_id, status, approved_by, start_at, end_at, created_at, updated_at +FROM kol_subscriptions +WHERE tenant_id = $1 + AND package_id = ANY($2::bigint[]) + AND status IN ('pending', 'active') +ORDER BY package_id, created_at DESC, id DESC +`, tenantID, packageIDs) + if err != nil { + return nil, err + } + defer rows.Close() + + for rows.Next() { + sub, err := scanKolSubscriptionRaw(rows) + if err != nil { + return nil, err + } + if sub != nil { + result[sub.PackageID] = *sub + } + } + if err := rows.Err(); err != nil { + return nil, err + } + return result, nil +} + func (r *kolSubscriptionRepository) ListByTenant(ctx context.Context, tenantID int64) ([]KolSubscription, error) { rows, err := r.q.ListKolSubscriptions(ctx, tenantID) if err != nil { @@ -386,6 +423,39 @@ func (r *kolSubscriptionRepository) ListActiveSubscribersForPackage(ctx context. return result, nil } +func (r *kolSubscriptionRepository) CountActiveSubscriberTenantsByPackages(ctx context.Context, packageIDs []int64) (map[int64]int64, error) { + counts := make(map[int64]int64, len(packageIDs)) + if len(packageIDs) == 0 { + return counts, nil + } + + rows, err := r.db.Query(ctx, ` +SELECT package_id, COUNT(DISTINCT tenant_id)::bigint +FROM kol_subscriptions +WHERE package_id = ANY($1::bigint[]) + AND status = 'active' + AND (end_at IS NULL OR end_at > NOW()) +GROUP BY package_id +`, packageIDs) + if err != nil { + return nil, err + } + defer rows.Close() + + for rows.Next() { + var packageID int64 + var count int64 + if err := rows.Scan(&packageID, &count); err != nil { + return nil, err + } + counts[packageID] = count + } + if err := rows.Err(); err != nil { + return nil, err + } + return counts, nil +} + func (r *kolSubscriptionRepository) Approve(ctx context.Context, id, tenantID, approvedBy int64, endAt *time.Time) error { return r.q.ApproveKolSubscription(ctx, generated.ApproveKolSubscriptionParams{ ApprovedBy: pgInt8(&approvedBy),