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

feat(kol): cache marketplace and generation reads with singleflight

Browse Source 
Add cache-aside + singleflight to KOL marketplace list/detail, my-subscription
prompts, and generation schema lookups. Invalidate the shared public-facing
caches when subscriptions are added/revoked (ops) or KOL profile is updated.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
liangxu
2026-05-24 11:33:17 +08:00
parent 09c8fe1e56
commit 8761e47f78
10 changed files with 339 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/internal/tenant/app/kol_marketplace_service.go
+98 -75
View File
@@ -5,8 +5,10 @@ import (
"time"
"github.com/geo-platform/tenant-api/internal/shared/auth"
sharedcache "github.com/geo-platform/tenant-api/internal/shared/cache"
"github.com/geo-platform/tenant-api/internal/shared/response"
"github.com/geo-platform/tenant-api/internal/tenant/repository"
"golang.org/x/sync/singleflight"
)
var (
@@ -82,6 +84,8 @@ type KolSubscriptionPromptCardResponse struct {
type KolMarketplaceService struct {
marketplaceRepo repository.KolMarketplaceRepository
subRepo repository.KolSubscriptionRepository
cache sharedcache.Cache
cacheGroup singleflight.Group
}
func NewKolMarketplaceService(
@@ -94,71 +98,87 @@ func NewKolMarketplaceService(
}
}
func (s *KolMarketplaceService) WithCache(c sharedcache.Cache) *KolMarketplaceService {
s.cache = c
return s
}
func (s *KolMarketplaceService) ListPackages(ctx context.Context, actor auth.Actor, filter MarketFilter) ([]KolMarketplacePackageResponse, error) {
filter = normalizeMarketFilter(filter)
packages, err := s.marketplaceRepo.ListPublished(ctx, repository.ListPublishedFilter{
Industry: filter.Industry,
Keyword: filter.Keyword,
Offset: filter.Offset,
Limit: filter.Limit,
})
if err != nil {
return nil, response.ErrInternal(50090, "kol_marketplace_query_failed", err.Error())
}
result := make([]KolMarketplacePackageResponse, 0, len(packages))
for _, pkg := range packages {
item, err := newKolMarketplacePackageResponse(pkg, actor.TenantID)
return sharedcache.LoadJSON(ctx, s.cache, &s.cacheGroup, kolMarketplacePackageListCacheKey(actor.TenantID, filter), defaultCacheTTL(), func(loadCtx context.Context) ([]KolMarketplacePackageResponse, error) {
packages, err := s.marketplaceRepo.ListPublished(loadCtx, repository.ListPublishedFilter{
Industry: filter.Industry,
Keyword: filter.Keyword,
Offset: filter.Offset,
Limit: filter.Limit,
})
if err != nil {
return nil, response.ErrInternal(50091, "kol_marketplace_decode_failed", err.Error())
return nil, response.ErrInternal(50090, "kol_marketplace_query_failed", err.Error())
}
result = append(result, *item)
}
return result, nil
result := make([]KolMarketplacePackageResponse, 0, len(packages))
for _, pkg := range packages {
item, err := newKolMarketplacePackageResponse(pkg, actor.TenantID)
if err != nil {
return nil, response.ErrInternal(50091, "kol_marketplace_decode_failed", err.Error())
}
result = append(result, *item)
}
return result, nil
})
}
func (s *KolMarketplaceService) GetPackage(ctx context.Context, actor auth.Actor, id int64) (*KolMarketplacePackageDetailResponse, error) {
pkg, err := s.marketplaceRepo.GetPublished(ctx, id)
record, found, err := sharedcache.LoadJSONWithEmpty(ctx, s.cache, &s.cacheGroup, kolMarketplacePackageDetailCacheKey(actor.TenantID, id), defaultCacheTTL(), defaultCacheEmptyTTL(), func(loadCtx context.Context) (*KolMarketplacePackageDetailResponse, bool, error) {
pkg, err := s.marketplaceRepo.GetPublished(loadCtx, id)
if err != nil {
return nil, false, response.ErrInternal(50090, "kol_marketplace_query_failed", err.Error())
}
if pkg == nil {
return nil, false, nil
}
prompts, err := s.marketplaceRepo.ListPublicPrompts(loadCtx, id)
if err != nil {
return nil, false, response.ErrInternal(50092, "kol_marketplace_prompt_query_failed", err.Error())
}
sub, err := s.subRepo.GetActiveForTenant(loadCtx, actor.TenantID, id)
if err != nil {
return nil, false, response.ErrInternal(50093, "kol_subscription_query_failed", err.Error())
}
pkgResp, err := newKolMarketplacePackageResponse(*pkg, actor.TenantID)
if err != nil {
return nil, false, response.ErrInternal(50091, "kol_marketplace_decode_failed", err.Error())
}
resp := &KolMarketplacePackageDetailResponse{
KolMarketplacePackageResponse: *pkgResp,
KolBio: pkg.KolBio,
Prompts: make([]KolMarketplacePromptResponse, 0, len(prompts)),
Subscription: newKolSubscriptionResponse(sub),
}
for _, prompt := range prompts {
resp.Prompts = append(resp.Prompts, KolMarketplacePromptResponse{
ID: prompt.ID,
Name: prompt.Name,
PlatformHint: prompt.PlatformHint,
Status: prompt.Status,
PublishedRevisionNo: prompt.PublishedRevisionNo,
CreatedAt: prompt.CreatedAt.Format(time.RFC3339),
})
}
return resp, true, nil
})
if err != nil {
return nil, response.ErrInternal(50090, "kol_marketplace_query_failed", err.Error())
return nil, err
}
if pkg == nil {
if !found || record == nil {
return nil, errKolMarketplacePackageNotFound
}
prompts, err := s.marketplaceRepo.ListPublicPrompts(ctx, id)
if err != nil {
return nil, response.ErrInternal(50092, "kol_marketplace_prompt_query_failed", err.Error())
}
sub, err := s.subRepo.GetActiveForTenant(ctx, actor.TenantID, id)
if err != nil {
return nil, response.ErrInternal(50093, "kol_subscription_query_failed", err.Error())
}
pkgResp, err := newKolMarketplacePackageResponse(*pkg, actor.TenantID)
if err != nil {
return nil, response.ErrInternal(50091, "kol_marketplace_decode_failed", err.Error())
}
resp := &KolMarketplacePackageDetailResponse{
KolMarketplacePackageResponse: *pkgResp,
KolBio: pkg.KolBio,
Prompts: make([]KolMarketplacePromptResponse, 0, len(prompts)),
Subscription: newKolSubscriptionResponse(sub),
}
for _, prompt := range prompts {
resp.Prompts = append(resp.Prompts, KolMarketplacePromptResponse{
ID: prompt.ID,
Name: prompt.Name,
PlatformHint: prompt.PlatformHint,
Status: prompt.Status,
PublishedRevisionNo: prompt.PublishedRevisionNo,
CreatedAt: prompt.CreatedAt.Format(time.RFC3339),
})
}
return resp, nil
return record, nil
}
func (s *KolMarketplaceService) Subscribe(ctx context.Context, actor auth.Actor, packageID int64) (*KolSubscriptionResponse, error) {
@@ -188,6 +208,7 @@ func (s *KolMarketplaceService) Subscribe(ctx context.Context, actor auth.Actor,
}
return nil, response.ErrInternal(50094, "kol_subscription_create_failed", err.Error())
}
InvalidateKolPublicCaches(ctx, s.cache)
return newKolSubscriptionResponse(sub), nil
}
@@ -207,29 +228,31 @@ func (s *KolMarketplaceService) ListMySubscriptions(ctx context.Context, actor a
}
func (s *KolMarketplaceService) ListMySubscriptionPrompts(ctx context.Context, actor auth.Actor) ([]KolSubscriptionPromptCardResponse, error) {
rows, err := s.subRepo.ListSubscriptionPromptsByTenant(ctx, actor.TenantID)
if err != nil {
return nil, response.ErrInternal(50095, "kol_subscription_prompt_query_failed", err.Error())
}
result := make([]KolSubscriptionPromptCardResponse, 0, len(rows))
for _, row := range rows {
grantedAt := ""
if row.GrantedAt != nil {
grantedAt = row.GrantedAt.Format(time.RFC3339)
return sharedcache.LoadJSON(ctx, s.cache, &s.cacheGroup, kolSubscriptionPromptListCacheKey(actor.TenantID), defaultCacheTTL(), func(loadCtx context.Context) ([]KolSubscriptionPromptCardResponse, error) {
rows, err := s.subRepo.ListSubscriptionPromptsByTenant(loadCtx, actor.TenantID)
if err != nil {
return nil, response.ErrInternal(50095, "kol_subscription_prompt_query_failed", err.Error())
}
result = append(result, KolSubscriptionPromptCardResponse{
ID: row.ID,
PackageID: row.PackageID,
PackageName: row.PackageName,
PromptID: row.PromptID,
PromptName: row.PromptName,
PlatformHint: row.PlatformHint,
KolDisplayName: row.KolDisplayName,
GrantedAt: grantedAt,
})
}
return result, nil
result := make([]KolSubscriptionPromptCardResponse, 0, len(rows))
for _, row := range rows {
grantedAt := ""
if row.GrantedAt != nil {
grantedAt = row.GrantedAt.Format(time.RFC3339)
}
result = append(result, KolSubscriptionPromptCardResponse{
ID: row.ID,
PackageID: row.PackageID,
PackageName: row.PackageName,
PromptID: row.PromptID,
PromptName: row.PromptName,
PlatformHint: row.PlatformHint,
KolDisplayName: row.KolDisplayName,
GrantedAt: grantedAt,
})
}
return result, nil
})
}
func newKolMarketplacePackageResponse(pkg repository.PublishedKolPackage, viewerTenantID int64) (*KolMarketplacePackageResponse, error) {