root/geo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: harden competitor create and batch kol stats
Backend CI / Backend (push) Failing after 6m48s
Details

Browse Source
This commit is contained in:
liangxu
2026-05-13 17:31:22 +08:00
parent 7aa786fbf4
commit 18cdbf25f8
5 changed files with 195 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/internal/tenant/app/brand_service.go
+18 -1
View File
@@ -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)
server/internal/tenant/app/kol_dashboard_service.go
+6 -7
View File
@@ -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)),
})
server/internal/tenant/app/kol_package_service.go
+67 -20
View File
@@ -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
server/internal/tenant/repository/kol_prompt_repo.go
+34
View File
@@ -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,
server/internal/tenant/repository/kol_subscription_repo.go
+70
View File
@@ -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),