diff --git a/server/cmd/ops-api/main.go b/server/cmd/ops-api/main.go index 366690f..26f6ba7 100644 --- a/server/cmd/ops-api/main.go +++ b/server/cmd/ops-api/main.go @@ -16,8 +16,10 @@ import ( opsconfig "github.com/geo-platform/tenant-api/internal/ops/config" "github.com/geo-platform/tenant-api/internal/ops/repository" "github.com/geo-platform/tenant-api/internal/ops/transport" + "github.com/geo-platform/tenant-api/internal/shared/cache" "github.com/geo-platform/tenant-api/internal/shared/observability" "github.com/geo-platform/tenant-api/internal/shared/repository/postgres" + "github.com/geo-platform/tenant-api/internal/shared/repository/redis" ) func main() { @@ -50,8 +52,16 @@ func main() { } defer monitoringPool.Close() + rdb, err := redis.NewClient(ctx, cfg.Redis) + if err != nil { + logger.Sugar().Fatalf("ops-api init redis: %v", err) + } + defer func() { _ = rdb.Close() }() + appCache := cache.New(cfg.Cache.Driver, rdb) + accountsRepo := repository.NewAccountRepository(pool) adminUsersRepo := repository.NewAdminUserRepository(pool) + kolSubscriptionsRepo := repository.NewKolSubscriptionRepository(pool) auditsRepo := repository.NewAuditRepository(pool) siteDomainMappingsRepo := repository.NewSiteDomainMappingRepository(monitoringPool) @@ -66,6 +76,7 @@ func main() { authSvc := app.NewAuthService(accountsRepo, auditSvc, issuer) accountSvc := app.NewAccountService(accountsRepo, auditSvc) adminUserSvc := app.NewAdminUserService(adminUsersRepo, auditSvc, cfg.AdminUsers.DefaultPlanCode) + kolSubscriptionSvc := app.NewKolSubscriptionService(kolSubscriptionsRepo, auditSvc).WithCache(appCache) siteDomainMappingSvc := app.NewSiteDomainMappingService(siteDomainMappingsRepo, auditSvc) if err := app.SeedDefaultAdmin(ctx, accountsRepo, app.DefaultAdminSeed{ @@ -91,6 +102,7 @@ func main() { Auth: authSvc, Accounts: accountSvc, AdminUsers: adminUserSvc, + KolSubs: kolSubscriptionSvc, Audits: auditSvc, SiteDomains: siteDomainMappingSvc, }) diff --git a/server/configs/ops-config.yaml b/server/configs/ops-config.yaml index 0cdc960..213cae4 100644 --- a/server/configs/ops-config.yaml +++ b/server/configs/ops-config.yaml @@ -22,6 +22,13 @@ monitoring_database: max_open_conns: 10 max_idle_conns: 2 +redis: + addr: localhost:6379 + db: 0 + +cache: + driver: redis + log: level: info format: json diff --git a/server/internal/ops/app/kol_subscription.go b/server/internal/ops/app/kol_subscription.go new file mode 100644 index 0000000..b5a3eef --- /dev/null +++ b/server/internal/ops/app/kol_subscription.go @@ -0,0 +1,293 @@ +package app + +import ( + "context" + "errors" + "fmt" + "strings" + "time" + + "github.com/geo-platform/tenant-api/internal/ops/repository" + sharedcache "github.com/geo-platform/tenant-api/internal/shared/cache" + "github.com/geo-platform/tenant-api/internal/shared/response" +) + +const ( + ActionKolSubscriptionApprove = "kol_subscription.approve" + ActionKolSubscriptionManualBind = "kol_subscription.manual_bind" + ActionKolSubscriptionRevoke = "kol_subscription.revoke" +) + +type KolSubscriptionService struct { + subs *repository.KolSubscriptionRepository + audits *AuditService + cache sharedcache.Cache +} + +func NewKolSubscriptionService(subs *repository.KolSubscriptionRepository, audits *AuditService) *KolSubscriptionService { + return &KolSubscriptionService{subs: subs, audits: audits} +} + +func (s *KolSubscriptionService) WithCache(c sharedcache.Cache) *KolSubscriptionService { + s.cache = c + return s +} + +type KolSubscriptionListInput struct { + Keyword string + Status string + Page int + Size int +} + +type KolSubscriptionListResult struct { + Items []KolSubscriptionView `json:"items"` + Total int64 `json:"total"` + Page int `json:"page"` + Size int `json:"size"` +} + +type KolSubscriptionView struct { + ID int64 `json:"id"` + TenantID int64 `json:"tenant_id"` + TenantName string `json:"tenant_name"` + SubscriberName *string `json:"subscriber_name"` + SubscriberPhone *string `json:"subscriber_phone"` + PackageID int64 `json:"package_id"` + PackageName string `json:"package_name"` + PackageStatus string `json:"package_status"` + CreatorTenantID int64 `json:"creator_tenant_id"` + CreatorTenantName string `json:"creator_tenant_name"` + KolProfileID int64 `json:"kol_profile_id"` + KolDisplayName string `json:"kol_display_name"` + KolAvatarURL *string `json:"kol_avatar_url"` + Status string `json:"status"` + PromptCount int64 `json:"prompt_count"` + AccessPromptCount int64 `json:"access_prompt_count"` + StartAt *time.Time `json:"start_at"` + EndAt *time.Time `json:"end_at"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` +} + +type KolPackageListInput struct { + Keyword string + Size int +} + +type KolPackageView struct { + ID int64 `json:"id"` + Name string `json:"name"` + Status string `json:"status"` + CreatorTenantID int64 `json:"creator_tenant_id"` + CreatorTenantName string `json:"creator_tenant_name"` + KolProfileID int64 `json:"kol_profile_id"` + KolDisplayName string `json:"kol_display_name"` + KolAvatarURL *string `json:"kol_avatar_url"` + PromptCount int64 `json:"prompt_count"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` +} + +type KolSubscriptionManualBindInput struct { + Phone string + PackageID int64 + EndAt *time.Time +} + +func (s *KolSubscriptionService) List(ctx context.Context, in KolSubscriptionListInput) (*KolSubscriptionListResult, error) { + page := in.Page + if page <= 0 { + page = 1 + } + size := in.Size + if size <= 0 || size > 200 { + size = 20 + } + status := strings.TrimSpace(in.Status) + if status != "" && !isValidKolSubscriptionStatus(status) { + return nil, response.ErrBadRequest(40090, "invalid_subscription_status", "无效的订阅状态") + } + + items, total, err := s.subs.List(ctx, repository.KolSubscriptionListFilter{ + Keyword: strings.TrimSpace(in.Keyword), + Status: status, + Limit: size, + Offset: (page - 1) * size, + }) + if err != nil { + return nil, response.ErrInternal(50090, "kol_subscription_list_failed", "failed to list KOL subscriptions") + } + + views := make([]KolSubscriptionView, len(items)) + for i, item := range items { + views[i] = toKolSubscriptionView(item) + } + return &KolSubscriptionListResult{ + Items: views, + Total: total, + Page: page, + Size: size, + }, nil +} + +func (s *KolSubscriptionService) ListPackages(ctx context.Context, in KolPackageListInput) ([]KolPackageView, error) { + size := in.Size + if size <= 0 || size > 100 { + size = 50 + } + items, err := s.subs.ListPublishedPackages(ctx, strings.TrimSpace(in.Keyword), size) + if err != nil { + return nil, response.ErrInternal(50094, "kol_package_list_failed", "failed to list KOL packages") + } + views := make([]KolPackageView, len(items)) + for i, item := range items { + views[i] = KolPackageView{ + ID: item.ID, + Name: item.Name, + Status: item.Status, + CreatorTenantID: item.CreatorTenantID, + CreatorTenantName: item.CreatorTenantName, + KolProfileID: item.KolProfileID, + KolDisplayName: item.KolDisplayName, + KolAvatarURL: item.KolAvatarURL, + PromptCount: item.PromptCount, + CreatedAt: item.CreatedAt, + UpdatedAt: item.UpdatedAt, + } + } + return views, nil +} + +func (s *KolSubscriptionService) Approve(ctx context.Context, actor *Actor, id int64, endAt *time.Time) (*KolSubscriptionView, error) { + if endAt != nil && endAt.Before(time.Now()) { + return nil, response.ErrBadRequest(40091, "invalid_subscription_end_at", "有效期不能早于当前时间") + } + + item, err := s.subs.Approve(ctx, id, endAt) + if err != nil { + return nil, mapKolSubscriptionError(err) + } + + if actor != nil && s.audits != nil { + _ = s.audits.Append(ctx, actor.audit(ActionKolSubscriptionApprove, "kol_subscription", id, map[string]any{ + "tenant_id": item.TenantID, + "package_id": item.PackageID, + "package_name": item.PackageName, + "end_at": item.EndAt, + })) + } + s.invalidateAccessCaches(ctx, item.TenantID) + + view := toKolSubscriptionView(*item) + return &view, nil +} + +func (s *KolSubscriptionService) ManualBind(ctx context.Context, actor *Actor, in KolSubscriptionManualBindInput) (*KolSubscriptionView, error) { + phone, err := normalizePhone(in.Phone) + if err != nil { + return nil, err + } + if in.PackageID <= 0 { + return nil, response.ErrBadRequest(40094, "invalid_package_id", "请选择模板包") + } + if in.EndAt != nil && in.EndAt.Before(time.Now()) { + return nil, response.ErrBadRequest(40091, "invalid_subscription_end_at", "有效期不能早于当前时间") + } + + item, err := s.subs.BindByPhone(ctx, phone, in.PackageID, in.EndAt) + if err != nil { + return nil, mapKolSubscriptionError(err) + } + + if actor != nil && s.audits != nil { + _ = s.audits.Append(ctx, actor.audit(ActionKolSubscriptionManualBind, "kol_subscription", item.ID, map[string]any{ + "phone": phone, + "tenant_id": item.TenantID, + "package_id": item.PackageID, + "package_name": item.PackageName, + "end_at": item.EndAt, + })) + } + s.invalidateAccessCaches(ctx, item.TenantID) + + view := toKolSubscriptionView(*item) + return &view, nil +} + +func (s *KolSubscriptionService) Revoke(ctx context.Context, actor *Actor, id int64) (*KolSubscriptionView, error) { + item, err := s.subs.Revoke(ctx, id) + if err != nil { + return nil, mapKolSubscriptionError(err) + } + + if actor != nil && s.audits != nil { + _ = s.audits.Append(ctx, actor.audit(ActionKolSubscriptionRevoke, "kol_subscription", id, map[string]any{ + "tenant_id": item.TenantID, + "package_id": item.PackageID, + "package_name": item.PackageName, + })) + } + s.invalidateAccessCaches(ctx, item.TenantID) + + view := toKolSubscriptionView(*item) + return &view, nil +} + +func (s *KolSubscriptionService) invalidateAccessCaches(ctx context.Context, tenantID int64) { + if s.cache == nil || tenantID <= 0 { + return + } + _ = s.cache.Delete(ctx, fmt.Sprintf("workspace:kol_cards:%d", tenantID)) + _ = s.cache.DeletePrefix(ctx, fmt.Sprintf("kol:subscription_prompt:list:%d:", tenantID)) + _ = s.cache.DeletePrefix(ctx, fmt.Sprintf("kol:subscription_prompt:schema:%d:", tenantID)) +} + +func toKolSubscriptionView(item repository.KolSubscriptionRecord) KolSubscriptionView { + return KolSubscriptionView{ + ID: item.ID, + TenantID: item.TenantID, + TenantName: item.TenantName, + SubscriberName: item.SubscriberName, + SubscriberPhone: item.SubscriberPhone, + PackageID: item.PackageID, + PackageName: item.PackageName, + PackageStatus: item.PackageStatus, + CreatorTenantID: item.CreatorTenantID, + CreatorTenantName: item.CreatorTenantName, + KolProfileID: item.KolProfileID, + KolDisplayName: item.KolDisplayName, + KolAvatarURL: item.KolAvatarURL, + Status: item.Status, + PromptCount: item.PromptCount, + AccessPromptCount: item.AccessPromptCount, + StartAt: item.StartAt, + EndAt: item.EndAt, + CreatedAt: item.CreatedAt, + UpdatedAt: item.UpdatedAt, + } +} + +func isValidKolSubscriptionStatus(status string) bool { + switch status { + case "pending", "active", "expired", "revoked": + return true + default: + return false + } +} + +func mapKolSubscriptionError(err error) error { + switch { + case errors.Is(err, repository.ErrKolSubscriptionNotFound): + return response.ErrNotFound(40490, "kol_subscription_not_found", "订阅申请不存在") + case errors.Is(err, repository.ErrKolSubscriptionNotPending): + return response.ErrBadRequest(40092, "kol_subscription_not_pending", "只有审核中的订阅申请可以通过") + case errors.Is(err, repository.ErrKolSubscriptionPackageUnavailable): + return response.ErrBadRequest(40093, "kol_package_unavailable", "订阅包未发布或不可用") + case errors.Is(err, repository.ErrKolSubscriptionUserNotFound): + return response.ErrNotFound(40491, "kol_subscription_user_not_found", "未找到该手机号对应的用户或租户") + default: + return response.ErrInternal(50091, "kol_subscription_update_failed", "failed to update KOL subscription") + } +} diff --git a/server/internal/ops/config/config.go b/server/internal/ops/config/config.go index 5ebe750..438fbe8 100644 --- a/server/internal/ops/config/config.go +++ b/server/internal/ops/config/config.go @@ -15,6 +15,8 @@ type Config struct { Server sharedconfig.ServerConfig `mapstructure:"server"` Database sharedconfig.DatabaseConfig `mapstructure:"database"` MonitoringDatabase sharedconfig.DatabaseConfig `mapstructure:"monitoring_database"` + Redis sharedconfig.RedisConfig `mapstructure:"redis"` + Cache sharedconfig.CacheConfig `mapstructure:"cache"` Log sharedconfig.LogConfig `mapstructure:"log"` JWT JWTConfig `mapstructure:"jwt"` DefaultAdmin DefaultAdminConfig `mapstructure:"default_admin"` @@ -53,6 +55,9 @@ func Load(path string) (*Config, error) { v.SetDefault("server.port", 8090) v.SetDefault("server.mode", "debug") + v.SetDefault("redis.addr", "localhost:6379") + v.SetDefault("redis.db", 0) + v.SetDefault("cache.driver", "redis") v.SetDefault("log.level", "info") v.SetDefault("log.format", "json") v.SetDefault("jwt.access_ttl", 8*time.Hour) diff --git a/server/internal/ops/repository/kol_subscription.go b/server/internal/ops/repository/kol_subscription.go new file mode 100644 index 0000000..375a298 --- /dev/null +++ b/server/internal/ops/repository/kol_subscription.go @@ -0,0 +1,647 @@ +package repository + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" +) + +var ( + ErrKolSubscriptionNotFound = errors.New("kol subscription not found") + ErrKolSubscriptionNotPending = errors.New("kol subscription is not pending") + ErrKolSubscriptionPackageUnavailable = errors.New("kol package is not available") + ErrKolSubscriptionUserNotFound = errors.New("kol subscription user not found") +) + +type KolSubscriptionRepository struct { + pool *pgxpool.Pool +} + +func NewKolSubscriptionRepository(pool *pgxpool.Pool) *KolSubscriptionRepository { + return &KolSubscriptionRepository{pool: pool} +} + +type KolSubscriptionListFilter struct { + Keyword string + Status string + Limit int + Offset int +} + +type KolSubscriptionRecord struct { + ID int64 + TenantID int64 + TenantName string + SubscriberName *string + SubscriberPhone *string + PackageID int64 + PackageName string + PackageStatus string + CreatorTenantID int64 + CreatorTenantName string + KolProfileID int64 + KolDisplayName string + KolAvatarURL *string + Status string + PromptCount int64 + AccessPromptCount int64 + StartAt *time.Time + EndAt *time.Time + CreatedAt time.Time + UpdatedAt time.Time +} + +type KolPackageRecord struct { + ID int64 + Name string + Status string + CreatorTenantID int64 + CreatorTenantName string + KolProfileID int64 + KolDisplayName string + KolAvatarURL *string + PromptCount int64 + CreatedAt time.Time + UpdatedAt time.Time +} + +func (r *KolSubscriptionRepository) List(ctx context.Context, f KolSubscriptionListFilter) ([]KolSubscriptionRecord, int64, error) { + args := []any{} + where := "kp.deleted_at IS NULL AND kprof.deleted_at IS NULL" + + if f.Keyword != "" { + args = append(args, "%"+f.Keyword+"%") + arg := len(args) + where += fmt.Sprintf(` AND ( + t.name ILIKE $%d + OR subscriber.name ILIKE $%d + OR subscriber.phone ILIKE $%d + OR creator_t.name ILIKE $%d + OR kp.name ILIKE $%d + OR kprof.display_name ILIKE $%d + OR s.id::TEXT ILIKE $%d + OR s.tenant_id::TEXT ILIKE $%d + OR s.package_id::TEXT ILIKE $%d + )`, arg, arg, arg, arg, arg, arg, arg, arg, arg) + } + if f.Status != "" { + args = append(args, f.Status) + where += fmt.Sprintf(" AND s.status = $%d", len(args)) + } + + var total int64 + if err := r.pool.QueryRow(ctx, ` +SELECT COUNT(*) +FROM kol_subscriptions s +JOIN tenants t ON t.id = s.tenant_id AND t.deleted_at IS NULL +LEFT JOIN LATERAL ( + SELECT u.name, u.phone + FROM tenant_memberships tm + JOIN users u + ON u.id = tm.user_id + AND u.deleted_at IS NULL + WHERE tm.tenant_id = s.tenant_id + AND tm.deleted_at IS NULL + ORDER BY + CASE WHEN tm.tenant_role = 'tenant_admin' THEN 0 ELSE 1 END, + tm.created_at ASC, + tm.id ASC + LIMIT 1 +) subscriber ON TRUE +JOIN kol_packages kp ON kp.id = s.package_id +JOIN tenants creator_t ON creator_t.id = kp.tenant_id AND creator_t.deleted_at IS NULL +JOIN kol_profiles kprof ON kprof.id = kp.kol_profile_id +WHERE `+where, args...).Scan(&total); err != nil { + return nil, 0, err + } + + limit := f.Limit + if limit <= 0 || limit > 200 { + limit = 20 + } + offset := f.Offset + if offset < 0 { + offset = 0 + } + args = append(args, limit, offset) + limitArg := len(args) - 1 + offsetArg := len(args) + + rows, err := r.pool.Query(ctx, fmt.Sprintf(` +SELECT + s.id, + s.tenant_id, + t.name AS tenant_name, + subscriber.name AS subscriber_name, + subscriber.phone AS subscriber_phone, + s.package_id, + kp.name AS package_name, + kp.status AS package_status, + kp.tenant_id AS creator_tenant_id, + creator_t.name AS creator_tenant_name, + kprof.id AS kol_profile_id, + kprof.display_name AS kol_display_name, + kprof.avatar_url AS kol_avatar_url, + s.status, + ( + SELECT COUNT(*) + FROM kol_prompts p + WHERE p.package_id = s.package_id + AND p.status = 'active' + AND p.published_revision_no IS NOT NULL + AND p.deleted_at IS NULL + ) AS prompt_count, + ( + SELECT COUNT(*) + FROM kol_subscription_prompts sp + WHERE sp.subscription_id = s.id + AND sp.status = 'active' + ) AS access_prompt_count, + s.start_at, + s.end_at, + s.created_at, + s.updated_at +FROM kol_subscriptions s +JOIN tenants t ON t.id = s.tenant_id AND t.deleted_at IS NULL +LEFT JOIN LATERAL ( + SELECT u.name, u.phone + FROM tenant_memberships tm + JOIN users u + ON u.id = tm.user_id + AND u.deleted_at IS NULL + WHERE tm.tenant_id = s.tenant_id + AND tm.deleted_at IS NULL + ORDER BY + CASE WHEN tm.tenant_role = 'tenant_admin' THEN 0 ELSE 1 END, + tm.created_at ASC, + tm.id ASC + LIMIT 1 +) subscriber ON TRUE +JOIN kol_packages kp ON kp.id = s.package_id +JOIN tenants creator_t ON creator_t.id = kp.tenant_id AND creator_t.deleted_at IS NULL +JOIN kol_profiles kprof ON kprof.id = kp.kol_profile_id +WHERE %s +ORDER BY + CASE s.status WHEN 'pending' THEN 0 WHEN 'active' THEN 1 ELSE 2 END, + s.created_at DESC, + s.id DESC +LIMIT $%d OFFSET $%d`, where, limitArg, offsetArg), args...) + if err != nil { + return nil, 0, err + } + defer rows.Close() + + items := make([]KolSubscriptionRecord, 0, limit) + for rows.Next() { + item, err := scanKolSubscription(rows) + if err != nil { + return nil, 0, err + } + items = append(items, *item) + } + return items, total, rows.Err() +} + +func (r *KolSubscriptionRepository) ListPublishedPackages(ctx context.Context, keyword string, limit int) ([]KolPackageRecord, error) { + args := []any{} + where := "kp.deleted_at IS NULL AND kprof.deleted_at IS NULL AND kp.status = 'published'" + + if keyword != "" { + args = append(args, "%"+keyword+"%") + arg := len(args) + where += fmt.Sprintf(` AND ( + kp.name ILIKE $%d + OR kprof.display_name ILIKE $%d + OR creator_t.name ILIKE $%d + OR kp.id::TEXT ILIKE $%d + )`, arg, arg, arg, arg) + } + if limit <= 0 || limit > 100 { + limit = 50 + } + args = append(args, limit) + limitArg := len(args) + + rows, err := r.pool.Query(ctx, fmt.Sprintf(` +SELECT + kp.id, + kp.name, + kp.status, + kp.tenant_id AS creator_tenant_id, + creator_t.name AS creator_tenant_name, + kprof.id AS kol_profile_id, + kprof.display_name AS kol_display_name, + kprof.avatar_url AS kol_avatar_url, + ( + SELECT COUNT(*) + FROM kol_prompts p + WHERE p.package_id = kp.id + AND p.status = 'active' + AND p.published_revision_no IS NOT NULL + AND p.deleted_at IS NULL + ) AS prompt_count, + kp.created_at, + kp.updated_at +FROM kol_packages kp +JOIN tenants creator_t ON creator_t.id = kp.tenant_id AND creator_t.deleted_at IS NULL +JOIN kol_profiles kprof ON kprof.id = kp.kol_profile_id AND kprof.deleted_at IS NULL +WHERE %s +ORDER BY kp.updated_at DESC, kp.id DESC +LIMIT $%d`, where, limitArg), args...) + if err != nil { + return nil, err + } + defer rows.Close() + + items := make([]KolPackageRecord, 0, limit) + for rows.Next() { + var item KolPackageRecord + if err := rows.Scan( + &item.ID, + &item.Name, + &item.Status, + &item.CreatorTenantID, + &item.CreatorTenantName, + &item.KolProfileID, + &item.KolDisplayName, + &item.KolAvatarURL, + &item.PromptCount, + &item.CreatedAt, + &item.UpdatedAt, + ); err != nil { + return nil, err + } + items = append(items, item) + } + return items, rows.Err() +} + +func (r *KolSubscriptionRepository) Approve(ctx context.Context, id int64, endAt *time.Time) (*KolSubscriptionRecord, error) { + tx, err := r.pool.Begin(ctx) + if err != nil { + return nil, err + } + defer tx.Rollback(ctx) + + sub, err := lockKolSubscription(ctx, tx, id) + if err != nil { + return nil, err + } + if sub.Status != "pending" { + return nil, ErrKolSubscriptionNotPending + } + if sub.PackageStatus != "published" { + return nil, ErrKolSubscriptionPackageUnavailable + } + + tag, err := tx.Exec(ctx, ` +UPDATE kol_subscriptions +SET status = 'active', + approved_by = NULL, + start_at = NOW(), + end_at = $2, + updated_at = NOW() +WHERE id = $1 + AND status = 'pending'`, id, endAt) + if err != nil { + return nil, err + } + if tag.RowsAffected() == 0 { + return nil, ErrKolSubscriptionNotPending + } + + if _, err := tx.Exec(ctx, ` +INSERT INTO kol_subscription_prompts (tenant_id, subscription_id, package_id, prompt_id, status) +SELECT s.tenant_id, s.id, s.package_id, p.id, 'active' +FROM kol_subscriptions s +JOIN kol_packages kp + ON kp.id = s.package_id + AND kp.deleted_at IS NULL +JOIN kol_prompts p + ON p.package_id = s.package_id + AND p.tenant_id = kp.tenant_id + AND p.status = 'active' + AND p.published_revision_no IS NOT NULL + AND p.deleted_at IS NULL +WHERE s.id = $1 +ON CONFLICT (subscription_id, prompt_id) DO NOTHING`, id); err != nil { + return nil, err + } + + updated, err := getKolSubscriptionByID(ctx, tx, id) + if err != nil { + return nil, err + } + if err := tx.Commit(ctx); err != nil { + return nil, err + } + return updated, nil +} + +func (r *KolSubscriptionRepository) BindByPhone(ctx context.Context, phone string, packageID int64, endAt *time.Time) (*KolSubscriptionRecord, error) { + tx, err := r.pool.Begin(ctx) + if err != nil { + return nil, err + } + defer tx.Rollback(ctx) + + var tenantID int64 + if err := tx.QueryRow(ctx, ` +SELECT tm.tenant_id +FROM users u +JOIN tenant_memberships tm + ON tm.user_id = u.id + AND tm.deleted_at IS NULL +JOIN tenants t + ON t.id = tm.tenant_id + AND t.deleted_at IS NULL +WHERE u.phone = $1 + AND u.deleted_at IS NULL +ORDER BY + CASE WHEN tm.tenant_role = 'tenant_admin' THEN 0 ELSE 1 END, + tm.created_at ASC, + tm.id ASC +LIMIT 1`, phone).Scan(&tenantID); err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrKolSubscriptionUserNotFound + } + return nil, err + } + + var packageStatus string + if err := tx.QueryRow(ctx, ` +SELECT kp.status +FROM kol_packages kp +JOIN kol_profiles kprof + ON kprof.id = kp.kol_profile_id + AND kprof.deleted_at IS NULL +WHERE kp.id = $1 + AND kp.deleted_at IS NULL`, packageID).Scan(&packageStatus); err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrKolSubscriptionPackageUnavailable + } + return nil, err + } + if packageStatus != "published" { + return nil, ErrKolSubscriptionPackageUnavailable + } + + var subscriptionID int64 + err = tx.QueryRow(ctx, ` +SELECT id +FROM kol_subscriptions +WHERE tenant_id = $1 + AND package_id = $2 + AND status IN ('pending', 'active') +ORDER BY id DESC +LIMIT 1 +FOR UPDATE`, tenantID, packageID).Scan(&subscriptionID) + if err != nil && !errors.Is(err, pgx.ErrNoRows) { + return nil, err + } + + if errors.Is(err, pgx.ErrNoRows) { + if err := tx.QueryRow(ctx, ` +INSERT INTO kol_subscriptions (tenant_id, package_id, status, approved_by, start_at, end_at) +VALUES ($1, $2, 'active', NULL, NOW(), $3) +RETURNING id`, tenantID, packageID, endAt).Scan(&subscriptionID); err != nil { + return nil, err + } + } else { + tag, err := tx.Exec(ctx, ` +UPDATE kol_subscriptions +SET status = 'active', + approved_by = NULL, + start_at = COALESCE(start_at, NOW()), + end_at = $2, + updated_at = NOW() +WHERE id = $1`, subscriptionID, endAt) + if err != nil { + return nil, err + } + if tag.RowsAffected() == 0 { + return nil, ErrKolSubscriptionNotFound + } + } + + if _, err := tx.Exec(ctx, ` +INSERT INTO kol_subscription_prompts (tenant_id, subscription_id, package_id, prompt_id, status) +SELECT $1, $2, $3, p.id, 'active' +FROM kol_prompts p +WHERE p.package_id = $3 + AND p.status = 'active' + AND p.published_revision_no IS NOT NULL + AND p.deleted_at IS NULL +ON CONFLICT (subscription_id, prompt_id) DO UPDATE +SET status = 'active', + revoked_at = NULL, + updated_at = NOW()`, tenantID, subscriptionID, packageID); err != nil { + return nil, err + } + + updated, err := getKolSubscriptionByID(ctx, tx, subscriptionID) + if err != nil { + return nil, err + } + if err := tx.Commit(ctx); err != nil { + return nil, err + } + return updated, nil +} + +func (r *KolSubscriptionRepository) Revoke(ctx context.Context, id int64) (*KolSubscriptionRecord, error) { + tx, err := r.pool.Begin(ctx) + if err != nil { + return nil, err + } + defer tx.Rollback(ctx) + + if _, err := lockKolSubscription(ctx, tx, id); err != nil { + return nil, err + } + + if _, err := tx.Exec(ctx, ` +UPDATE kol_subscriptions +SET status = 'revoked', + updated_at = NOW() +WHERE id = $1`, id); err != nil { + return nil, err + } + + if _, err := tx.Exec(ctx, ` +UPDATE kol_subscription_prompts +SET status = 'revoked', + revoked_at = NOW(), + updated_at = NOW() +WHERE subscription_id = $1 + AND status <> 'revoked'`, id); err != nil { + return nil, err + } + + updated, err := getKolSubscriptionByID(ctx, tx, id) + if err != nil { + return nil, err + } + if err := tx.Commit(ctx); err != nil { + return nil, err + } + return updated, nil +} + +type kolSubscriptionScanner interface { + Scan(dest ...any) error +} + +func scanKolSubscription(row kolSubscriptionScanner) (*KolSubscriptionRecord, error) { + var item KolSubscriptionRecord + err := row.Scan( + &item.ID, + &item.TenantID, + &item.TenantName, + &item.SubscriberName, + &item.SubscriberPhone, + &item.PackageID, + &item.PackageName, + &item.PackageStatus, + &item.CreatorTenantID, + &item.CreatorTenantName, + &item.KolProfileID, + &item.KolDisplayName, + &item.KolAvatarURL, + &item.Status, + &item.PromptCount, + &item.AccessPromptCount, + &item.StartAt, + &item.EndAt, + &item.CreatedAt, + &item.UpdatedAt, + ) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrKolSubscriptionNotFound + } + return nil, err + } + return &item, nil +} + +func lockKolSubscription(ctx context.Context, tx pgx.Tx, id int64) (*KolSubscriptionRecord, error) { + row := tx.QueryRow(ctx, ` +SELECT + s.id, + s.tenant_id, + t.name AS tenant_name, + subscriber.name AS subscriber_name, + subscriber.phone AS subscriber_phone, + s.package_id, + kp.name AS package_name, + kp.status AS package_status, + kp.tenant_id AS creator_tenant_id, + creator_t.name AS creator_tenant_name, + kprof.id AS kol_profile_id, + kprof.display_name AS kol_display_name, + kprof.avatar_url AS kol_avatar_url, + s.status, + ( + SELECT COUNT(*) + FROM kol_prompts p + WHERE p.package_id = s.package_id + AND p.status = 'active' + AND p.published_revision_no IS NOT NULL + AND p.deleted_at IS NULL + ) AS prompt_count, + ( + SELECT COUNT(*) + FROM kol_subscription_prompts sp + WHERE sp.subscription_id = s.id + AND sp.status = 'active' + ) AS access_prompt_count, + s.start_at, + s.end_at, + s.created_at, + s.updated_at +FROM kol_subscriptions s +JOIN tenants t ON t.id = s.tenant_id AND t.deleted_at IS NULL +LEFT JOIN LATERAL ( + SELECT u.name, u.phone + FROM tenant_memberships tm + JOIN users u + ON u.id = tm.user_id + AND u.deleted_at IS NULL + WHERE tm.tenant_id = s.tenant_id + AND tm.deleted_at IS NULL + ORDER BY + CASE WHEN tm.tenant_role = 'tenant_admin' THEN 0 ELSE 1 END, + tm.created_at ASC, + tm.id ASC + LIMIT 1 +) subscriber ON TRUE +JOIN kol_packages kp ON kp.id = s.package_id AND kp.deleted_at IS NULL +JOIN tenants creator_t ON creator_t.id = kp.tenant_id AND creator_t.deleted_at IS NULL +JOIN kol_profiles kprof ON kprof.id = kp.kol_profile_id AND kprof.deleted_at IS NULL +WHERE s.id = $1 +FOR UPDATE OF s`, id) + return scanKolSubscription(row) +} + +func getKolSubscriptionByID(ctx context.Context, tx pgx.Tx, id int64) (*KolSubscriptionRecord, error) { + row := tx.QueryRow(ctx, ` +SELECT + s.id, + s.tenant_id, + t.name AS tenant_name, + subscriber.name AS subscriber_name, + subscriber.phone AS subscriber_phone, + s.package_id, + kp.name AS package_name, + kp.status AS package_status, + kp.tenant_id AS creator_tenant_id, + creator_t.name AS creator_tenant_name, + kprof.id AS kol_profile_id, + kprof.display_name AS kol_display_name, + kprof.avatar_url AS kol_avatar_url, + s.status, + ( + SELECT COUNT(*) + FROM kol_prompts p + WHERE p.package_id = s.package_id + AND p.status = 'active' + AND p.published_revision_no IS NOT NULL + AND p.deleted_at IS NULL + ) AS prompt_count, + ( + SELECT COUNT(*) + FROM kol_subscription_prompts sp + WHERE sp.subscription_id = s.id + AND sp.status = 'active' + ) AS access_prompt_count, + s.start_at, + s.end_at, + s.created_at, + s.updated_at +FROM kol_subscriptions s +JOIN tenants t ON t.id = s.tenant_id AND t.deleted_at IS NULL +LEFT JOIN LATERAL ( + SELECT u.name, u.phone + FROM tenant_memberships tm + JOIN users u + ON u.id = tm.user_id + AND u.deleted_at IS NULL + WHERE tm.tenant_id = s.tenant_id + AND tm.deleted_at IS NULL + ORDER BY + CASE WHEN tm.tenant_role = 'tenant_admin' THEN 0 ELSE 1 END, + tm.created_at ASC, + tm.id ASC + LIMIT 1 +) subscriber ON TRUE +JOIN kol_packages kp ON kp.id = s.package_id AND kp.deleted_at IS NULL +JOIN tenants creator_t ON creator_t.id = kp.tenant_id AND creator_t.deleted_at IS NULL +JOIN kol_profiles kprof ON kprof.id = kp.kol_profile_id AND kprof.deleted_at IS NULL +WHERE s.id = $1`, id) + return scanKolSubscription(row) +} diff --git a/server/internal/ops/transport/kol_subscription_handler.go b/server/internal/ops/transport/kol_subscription_handler.go new file mode 100644 index 0000000..4ce3545 --- /dev/null +++ b/server/internal/ops/transport/kol_subscription_handler.go @@ -0,0 +1,117 @@ +package transport + +import ( + "strconv" + "time" + + "github.com/gin-gonic/gin" + + "github.com/geo-platform/tenant-api/internal/ops/app" + "github.com/geo-platform/tenant-api/internal/shared/response" +) + +type approveKolSubscriptionRequest struct { + EndAt *time.Time `json:"end_at"` +} + +type manualBindKolSubscriptionRequest struct { + Phone string `json:"phone" binding:"required"` + PackageID int64 `json:"package_id" binding:"required"` + EndAt *time.Time `json:"end_at"` +} + +func listKolSubscriptionsHandler(svc *app.KolSubscriptionService) gin.HandlerFunc { + return func(c *gin.Context) { + page, _ := strconv.Atoi(c.DefaultQuery("page", "1")) + size, _ := strconv.Atoi(c.DefaultQuery("size", "20")) + result, err := svc.List(c.Request.Context(), app.KolSubscriptionListInput{ + Keyword: c.Query("keyword"), + Status: c.Query("status"), + Page: page, + Size: size, + }) + if err != nil { + response.Error(c, err) + return + } + response.Success(c, result) + } +} + +func listKolSubscriptionPackagesHandler(svc *app.KolSubscriptionService) gin.HandlerFunc { + return func(c *gin.Context) { + size, _ := strconv.Atoi(c.DefaultQuery("size", "50")) + result, err := svc.ListPackages(c.Request.Context(), app.KolPackageListInput{ + Keyword: c.Query("keyword"), + Size: size, + }) + if err != nil { + response.Error(c, err) + return + } + response.Success(c, result) + } +} + +func manualBindKolSubscriptionHandler(svc *app.KolSubscriptionService) gin.HandlerFunc { + return func(c *gin.Context) { + var body manualBindKolSubscriptionRequest + if err := c.ShouldBindJSON(&body); err != nil { + response.Error(c, response.ErrBadRequest(40000, "invalid_payload", err.Error())) + return + } + + detail, err := svc.ManualBind(c.Request.Context(), actorFromGin(c), app.KolSubscriptionManualBindInput{ + Phone: body.Phone, + PackageID: body.PackageID, + EndAt: body.EndAt, + }) + if err != nil { + response.Error(c, err) + return + } + response.Success(c, detail) + } +} + +func approveKolSubscriptionHandler(svc *app.KolSubscriptionService) gin.HandlerFunc { + return func(c *gin.Context) { + id, err := parseIDParam(c) + if err != nil { + response.Error(c, err) + return + } + + var body approveKolSubscriptionRequest + if c.Request.ContentLength > 0 { + if err := c.ShouldBindJSON(&body); err != nil { + response.Error(c, response.ErrBadRequest(40000, "invalid_payload", err.Error())) + return + } + } + + detail, err := svc.Approve(c.Request.Context(), actorFromGin(c), id, body.EndAt) + if err != nil { + response.Error(c, err) + return + } + response.Success(c, detail) + } +} + +func revokeKolSubscriptionHandler(svc *app.KolSubscriptionService) gin.HandlerFunc { + return func(c *gin.Context) { + id, err := parseIDParam(c) + if err != nil { + response.Error(c, err) + return + } + + detail, err := svc.Revoke(c.Request.Context(), actorFromGin(c), id) + if err != nil { + response.Error(c, err) + return + } + response.Success(c, detail) + } +} diff --git a/server/internal/ops/transport/router.go b/server/internal/ops/transport/router.go index ac0e9e4..0b3f5f9 100644 --- a/server/internal/ops/transport/router.go +++ b/server/internal/ops/transport/router.go @@ -19,6 +19,7 @@ type Deps struct { Auth *app.AuthService Accounts *app.AccountService AdminUsers *app.AdminUserService + KolSubs *app.KolSubscriptionService Audits *app.AuditService SiteDomains *app.SiteDomainMappingService } @@ -81,6 +82,12 @@ func RegisterRoutes(d Deps) { authed.POST("/admin-users/:id/status", setAdminUserStatusHandler(d.AdminUsers)) authed.POST("/admin-users/:id/password", resetAdminUserPasswordHandler(d.AdminUsers)) + authed.GET("/kol/packages", listKolSubscriptionPackagesHandler(d.KolSubs)) + authed.GET("/kol/subscriptions", listKolSubscriptionsHandler(d.KolSubs)) + authed.POST("/kol/subscriptions/manual-bind", manualBindKolSubscriptionHandler(d.KolSubs)) + authed.POST("/kol/subscriptions/:id/approve", approveKolSubscriptionHandler(d.KolSubs)) + authed.POST("/kol/subscriptions/:id/revoke", revokeKolSubscriptionHandler(d.KolSubs)) + authed.GET("/audits", listAuditsHandler(d.Audits)) authed.GET("/site-domain-mappings", listSiteDomainMappingsHandler(d.SiteDomains)) diff --git a/server/internal/tenant/app/kol_subscription_admin_service.go b/server/internal/tenant/app/kol_subscription_admin_service.go deleted file mode 100644 index 232b748..0000000 --- a/server/internal/tenant/app/kol_subscription_admin_service.go +++ /dev/null @@ -1,264 +0,0 @@ -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, - }) -} diff --git a/server/internal/tenant/transport/kol_admin_handler.go b/server/internal/tenant/transport/kol_admin_handler.go deleted file mode 100644 index 8f12b8c..0000000 --- a/server/internal/tenant/transport/kol_admin_handler.go +++ /dev/null @@ -1,116 +0,0 @@ -package transport - -import ( - "time" - - "github.com/gin-gonic/gin" - - "github.com/geo-platform/tenant-api/internal/bootstrap" - "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/app" -) - -var errTenantAdminRequired = response.ErrForbidden(40371, "tenant_admin_required", "tenant admin role is required") - -type KolAdminHandler struct { - svc *app.KolSubscriptionAdminService -} - -type approveKolSubscriptionRequest struct { - EndAt *time.Time `json:"end_at"` -} - -func NewKolAdminHandler(a *bootstrap.App) *KolAdminHandler { - return &KolAdminHandler{ - svc: app.NewKolSubscriptionAdminService( - a.DB, - a.AuditLogs, - a.KolMarketplace, - a.KolSubscriptions, - a.KolPrompts, - ).WithCache(a.Cache), - } -} - -func (h *KolAdminHandler) ApproveSubscription(c *gin.Context) { - id, ok := parseInt64Param(c, "id", "subscription id") - if !ok { - return - } - - actor, ok := requireTenantAdmin(c) - if !ok { - return - } - - var req approveKolSubscriptionRequest - if c.Request.ContentLength > 0 { - if err := c.ShouldBindJSON(&req); err != nil { - response.Error(c, response.ErrBadRequest(40001, "invalid_params", err.Error())) - return - } - } - - data, err := h.svc.Approve(c.Request.Context(), id, req.EndAt, actor.UserID) - if err != nil { - response.Error(c, err) - return - } - - response.Success(c, data) -} - -func (h *KolAdminHandler) ManualBind(c *gin.Context) { - actor, ok := requireTenantAdmin(c) - if !ok { - return - } - - var req app.ManualBindKolSubscriptionInput - if err := c.ShouldBindJSON(&req); err != nil { - response.Error(c, response.ErrBadRequest(40001, "invalid_params", err.Error())) - return - } - - data, err := h.svc.ManualBind(c.Request.Context(), req, actor.UserID) - if err != nil { - response.Error(c, err) - return - } - - response.Success(c, data) -} - -func (h *KolAdminHandler) RevokeSubscription(c *gin.Context) { - id, ok := parseInt64Param(c, "id", "subscription id") - if !ok { - return - } - - actor, ok := requireTenantAdmin(c) - if !ok { - return - } - - if err := h.svc.Revoke(c.Request.Context(), id, actor.UserID); err != nil { - response.Error(c, err) - return - } - - response.Success(c, gin.H{"ok": true}) -} - -// V1 compromise: these admin endpoints live on tenant-api until a dedicated -// platform-api/module exists. The temporary gate is tenant_role=tenant_admin. -func requireTenantAdmin(c *gin.Context) (auth.Actor, bool) { - actor, ok := actorFromRequest(c) - if !ok { - return auth.Actor{}, false - } - if actor.Role != "tenant_admin" { - response.Error(c, errTenantAdminRequired) - return auth.Actor{}, false - } - return actor, true -}