diff --git a/server/internal/bootstrap/bootstrap.go b/server/internal/bootstrap/bootstrap.go index e914318..25add98 100644 --- a/server/internal/bootstrap/bootstrap.go +++ b/server/internal/bootstrap/bootstrap.go @@ -49,7 +49,9 @@ type App struct { Cache cache.Cache KolProfiles repository.KolProfileRepository KolPackages repository.KolPackageRepository + KolMarketplace repository.KolMarketplaceRepository KolPrompts repository.KolPromptRepository + KolSubscriptions repository.KolSubscriptionRepository KolAssists repository.KolAssistRepository KolPromptAsset *tenantapp.KolPromptAsset } @@ -105,7 +107,9 @@ func New(configPath string) (*App, error) { appCache := cache.New(cfg.Cache.Driver, rdb) kolProfiles := repository.NewKolProfileRepository(pool) kolPackages := repository.NewKolPackageRepository(pool) + kolMarketplace := repository.NewKolMarketplaceRepository(pool) kolPrompts := repository.NewKolPromptRepository(pool) + kolSubscriptions := repository.NewKolSubscriptionRepository(pool) kolAssists := repository.NewKolAssistRepository(pool) kolPromptAsset := tenantapp.NewKolPromptAsset(objectStorageClient) @@ -163,7 +167,9 @@ func New(configPath string) (*App, error) { Cache: appCache, KolProfiles: kolProfiles, KolPackages: kolPackages, + KolMarketplace: kolMarketplace, KolPrompts: kolPrompts, + KolSubscriptions: kolSubscriptions, KolAssists: kolAssists, KolPromptAsset: kolPromptAsset, }, nil diff --git a/server/internal/tenant/app/kol_marketplace_service.go b/server/internal/tenant/app/kol_marketplace_service.go new file mode 100644 index 0000000..e461f01 --- /dev/null +++ b/server/internal/tenant/app/kol_marketplace_service.go @@ -0,0 +1,282 @@ +package app + +import ( + "context" + "time" + + "github.com/geo-platform/tenant-api/internal/shared/auth" + "github.com/geo-platform/tenant-api/internal/shared/response" + "github.com/geo-platform/tenant-api/internal/tenant/repository" +) + +var ( + errKolMarketplacePackageNotFound = response.ErrNotFound(40471, "kol_marketplace_package_not_found", "package not found") + errKolSubscriptionNotFound = response.ErrNotFound(40472, "kol_subscription_not_found", "subscription not found") + errKolSubscriptionExists = response.ErrConflict(40971, "kol_subscription_exists", "subscription already exists") + errKolSubscriptionNotPending = response.ErrConflict(40972, "kol_subscription_not_pending", "subscription is not pending") +) + +type MarketFilter struct { + Industry *string + Keyword *string + Offset int + Limit int +} + +type KolMarketplacePackageResponse struct { + ID int64 `json:"id"` + KolProfileID int64 `json:"kol_profile_id"` + KolDisplayName string `json:"kol_display_name"` + KolAvatarURL *string `json:"kol_avatar_url"` + Name string `json:"name"` + Description *string `json:"description"` + CoverURL *string `json:"cover_url"` + Industry *string `json:"industry"` + Tags []string `json:"tags"` + PriceNote *string `json:"price_note"` + Status string `json:"status"` + PromptCount int `json:"prompt_count"` + SubscriberCount int `json:"subscriber_count"` + CreatedAt string `json:"created_at"` + UpdatedAt string `json:"updated_at"` +} + +type KolMarketplacePromptResponse struct { + ID int64 `json:"id"` + Name string `json:"name"` + PlatformHint *string `json:"platform_hint"` + Status string `json:"status"` + PublishedRevisionNo *int `json:"published_revision_no"` + CreatedAt string `json:"created_at"` +} + +type KolMarketplacePackageDetailResponse struct { + KolMarketplacePackageResponse + KolBio *string `json:"kol_bio"` + Prompts []KolMarketplacePromptResponse `json:"prompts"` + Subscription *KolSubscriptionResponse `json:"subscription"` +} + +type KolSubscriptionResponse struct { + ID int64 `json:"id"` + PackageID int64 `json:"package_id"` + Status string `json:"status"` + StartAt *string `json:"start_at"` + EndAt *string `json:"end_at"` + CreatedAt string `json:"created_at"` +} + +type KolSubscriptionPromptCardResponse struct { + ID int64 `json:"id"` + PackageID int64 `json:"package_id"` + PackageName string `json:"package_name"` + PromptID int64 `json:"prompt_id"` + PromptName string `json:"prompt_name"` + PlatformHint *string `json:"platform_hint"` + KolDisplayName string `json:"kol_display_name"` + GrantedAt string `json:"granted_at"` +} + +type KolMarketplaceService struct { + marketplaceRepo repository.KolMarketplaceRepository + subRepo repository.KolSubscriptionRepository +} + +func NewKolMarketplaceService( + marketplaceRepo repository.KolMarketplaceRepository, + subRepo repository.KolSubscriptionRepository, +) *KolMarketplaceService { + return &KolMarketplaceService{ + marketplaceRepo: marketplaceRepo, + subRepo: subRepo, + } +} + +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) + 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) + if err != nil { + return nil, response.ErrInternal(50090, "kol_marketplace_query_failed", err.Error()) + } + if pkg == 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) + 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 +} + +func (s *KolMarketplaceService) Subscribe(ctx context.Context, actor auth.Actor, packageID int64) (*KolSubscriptionResponse, error) { + pkg, err := s.marketplaceRepo.GetPublished(ctx, packageID) + if err != nil { + return nil, response.ErrInternal(50090, "kol_marketplace_query_failed", err.Error()) + } + if pkg == nil { + return nil, errKolMarketplacePackageNotFound + } + + existing, err := s.subRepo.GetActiveForTenant(ctx, actor.TenantID, packageID) + if err != nil { + return nil, response.ErrInternal(50093, "kol_subscription_query_failed", err.Error()) + } + if existing != nil { + return nil, errKolSubscriptionExists + } + + sub, err := s.subRepo.Create(ctx, actor.TenantID, packageID) + if err != nil { + if isUniqueConstraintError(err, "uk_kol_subscription_active") { + return nil, errKolSubscriptionExists + } + return nil, response.ErrInternal(50094, "kol_subscription_create_failed", err.Error()) + } + return newKolSubscriptionResponse(sub), nil +} + +func (s *KolMarketplaceService) ListMySubscriptions(ctx context.Context, actor auth.Actor) ([]KolSubscriptionResponse, error) { + subs, err := s.subRepo.ListByTenant(ctx, actor.TenantID) + if err != nil { + return nil, response.ErrInternal(50093, "kol_subscription_query_failed", err.Error()) + } + + result := make([]KolSubscriptionResponse, 0, len(subs)) + for _, sub := range subs { + if resp := newKolSubscriptionResponse(&sub); resp != nil { + result = append(result, *resp) + } + } + return result, nil +} + +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) + } + 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) (*KolMarketplacePackageResponse, error) { + tags, err := decodeKolStringList(pkg.Tags) + if err != nil { + return nil, err + } + + return &KolMarketplacePackageResponse{ + ID: pkg.ID, + KolProfileID: pkg.KolProfileID, + KolDisplayName: pkg.KolDisplayName, + KolAvatarURL: pkg.KolAvatarURL, + Name: pkg.Name, + Description: pkg.Description, + CoverURL: pkg.CoverURL, + Industry: pkg.Industry, + Tags: tags, + PriceNote: pkg.PriceNote, + Status: pkg.Status, + PromptCount: int(pkg.PromptCount), + SubscriberCount: int(pkg.SubscriberCount), + CreatedAt: pkg.CreatedAt.Format(time.RFC3339), + UpdatedAt: pkg.UpdatedAt.Format(time.RFC3339), + }, nil +} + +func newKolSubscriptionResponse(sub *repository.KolSubscription) *KolSubscriptionResponse { + if sub == nil { + return nil + } + return &KolSubscriptionResponse{ + ID: sub.ID, + PackageID: sub.PackageID, + Status: sub.Status, + StartAt: timeStringPtr(sub.StartAt), + EndAt: timeStringPtr(sub.EndAt), + CreatedAt: sub.CreatedAt.Format(time.RFC3339), + } +} + +func normalizeMarketFilter(filter MarketFilter) MarketFilter { + if filter.Offset < 0 { + filter.Offset = 0 + } + if filter.Limit <= 0 { + filter.Limit = 20 + } + if filter.Limit > 100 { + filter.Limit = 100 + } + filter.Industry = normalizeKolOptionalString(filter.Industry) + filter.Keyword = normalizeKolOptionalString(filter.Keyword) + return filter +} diff --git a/server/internal/tenant/app/kol_prompt_service.go b/server/internal/tenant/app/kol_prompt_service.go index 30810e9..fe3548a 100644 --- a/server/internal/tenant/app/kol_prompt_service.go +++ b/server/internal/tenant/app/kol_prompt_service.go @@ -69,6 +69,7 @@ type KolPromptService struct { profileSvc *KolProfileService pkgRepo repository.KolPackageRepository promptRepo repository.KolPromptRepository + subRepo repository.KolSubscriptionRepository asset *KolPromptAsset } @@ -77,6 +78,7 @@ func NewKolPromptService( profileSvc *KolProfileService, pkgRepo repository.KolPackageRepository, promptRepo repository.KolPromptRepository, + subRepo repository.KolSubscriptionRepository, asset *KolPromptAsset, ) *KolPromptService { return &KolPromptService{ @@ -84,6 +86,7 @@ func NewKolPromptService( profileSvc: profileSvc, pkgRepo: pkgRepo, promptRepo: promptRepo, + subRepo: subRepo, asset: asset, } } @@ -251,7 +254,7 @@ func (s *KolPromptService) SaveDraft(ctx context.Context, actor auth.Actor, inpu } func (s *KolPromptService) Publish(ctx context.Context, actor auth.Actor, promptID int64) error { - _, prompt, _, err := s.loadOwnedPrompt(ctx, actor, promptID) + _, prompt, pkg, err := s.loadOwnedPrompt(ctx, actor, promptID) if err != nil { return err } @@ -265,10 +268,28 @@ func (s *KolPromptService) Publish(ctx context.Context, actor auth.Actor, prompt } defer tx.Rollback(ctx) - if err := repository.NewKolPromptRepository(tx).SetPublishedRevision(ctx, actor.TenantID, promptID, *prompt.DraftRevisionNo, actor.UserID); err != nil { + promptTx := repository.NewKolPromptRepository(tx) + subTx := repository.NewKolSubscriptionRepository(tx) + + if err := promptTx.SetPublishedRevision(ctx, actor.TenantID, promptID, *prompt.DraftRevisionNo, actor.UserID); err != nil { return response.ErrInternal(50080, "kol_prompt_publish_failed", err.Error()) } + subscribers, err := subTx.ListActiveSubscribersForPackage(ctx, pkg.ID) + if err != nil { + return response.ErrInternal(50083, "kol_subscription_query_failed", err.Error()) + } + for _, sub := range subscribers { + if err := subTx.CreateAccessRow(ctx, repository.CreateAccessRowInput{ + TenantID: sub.TenantID, + SubscriptionID: sub.ID, + PackageID: pkg.ID, + PromptID: promptID, + }); err != nil { + return response.ErrInternal(50084, "kol_subscription_access_create_failed", err.Error()) + } + } + if err := tx.Commit(ctx); err != nil { return response.ErrInternal(50081, "kol_prompt_publish_commit_failed", err.Error()) } diff --git a/server/internal/tenant/app/kol_subscription_admin_service.go b/server/internal/tenant/app/kol_subscription_admin_service.go new file mode 100644 index 0000000..d3b2433 --- /dev/null +++ b/server/internal/tenant/app/kol_subscription_admin_service.go @@ -0,0 +1,254 @@ +package app + +import ( + "context" + "encoding/json" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + + "github.com/geo-platform/tenant-api/internal/shared/auditlog" + "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 +} + +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) 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") + } + + 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") + } + + 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") + } + + 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, + }) +}