From 3d4842e983b5330e60e4bbb0a3b21b7fd80858e0 Mon Sep 17 00:00:00 2001 From: liangxu Date: Fri, 17 Apr 2026 08:17:22 +0800 Subject: [PATCH] feat(kol): repository layer for all KOL tables --- server/internal/tenant/repository/common.go | 8 + .../tenant/repository/kol_assist_repo.go | 118 +++++++ .../tenant/repository/kol_marketplace_repo.go | 145 ++++++++ .../tenant/repository/kol_package_repo.go | 192 +++++++++++ .../tenant/repository/kol_profile_repo.go | 102 ++++++ .../tenant/repository/kol_prompt_repo.go | 323 ++++++++++++++++++ .../repository/kol_subscription_repo.go | 255 ++++++++++++++ .../tenant/repository/kol_usage_repo.go | 198 +++++++++++ 8 files changed, 1341 insertions(+) create mode 100644 server/internal/tenant/repository/kol_assist_repo.go create mode 100644 server/internal/tenant/repository/kol_marketplace_repo.go create mode 100644 server/internal/tenant/repository/kol_package_repo.go create mode 100644 server/internal/tenant/repository/kol_profile_repo.go create mode 100644 server/internal/tenant/repository/kol_prompt_repo.go create mode 100644 server/internal/tenant/repository/kol_subscription_repo.go create mode 100644 server/internal/tenant/repository/kol_usage_repo.go diff --git a/server/internal/tenant/repository/common.go b/server/internal/tenant/repository/common.go index 1aa46c8..ebb7ff2 100644 --- a/server/internal/tenant/repository/common.go +++ b/server/internal/tenant/repository/common.go @@ -58,6 +58,14 @@ func timeFromTimestamp(value pgtype.Timestamptz) time.Time { return value.Time } +func optionalTime(value pgtype.Timestamptz) *time.Time { + if !value.Valid { + return nil + } + t := value.Time + return &t +} + func pgText(value *string) pgtype.Text { if value == nil { return pgtype.Text{} diff --git a/server/internal/tenant/repository/kol_assist_repo.go b/server/internal/tenant/repository/kol_assist_repo.go new file mode 100644 index 0000000..d2f60ed --- /dev/null +++ b/server/internal/tenant/repository/kol_assist_repo.go @@ -0,0 +1,118 @@ +package repository + +import ( + "context" + "errors" + "time" + + "github.com/jackc/pgx/v5" + + "github.com/geo-platform/tenant-api/internal/tenant/repository/generated" +) + +type KolAssistTask struct { + ID string + TenantID int64 + KolProfileID int64 + PromptID *int64 + OperatorID int64 + TaskType string + Status string + RequestJSON []byte + ResultJSON []byte + ErrorMessage *string + StartedAt *time.Time + CompletedAt *time.Time + CreatedAt time.Time + UpdatedAt time.Time +} + +type CreateKolAssistInput struct { + ID string + TenantID int64 + KolProfileID int64 + PromptID *int64 + OperatorID int64 + TaskType string + RequestJSON []byte +} + +type KolAssistRepository interface { + Create(ctx context.Context, input CreateKolAssistInput) error + Get(ctx context.Context, tenantID int64, id string) (*KolAssistTask, error) + MarkStarted(ctx context.Context, tenantID int64, id string) error + MarkCompleted(ctx context.Context, tenantID int64, id string, resultJSON []byte) error + MarkFailed(ctx context.Context, tenantID int64, id string, errorMsg string) error +} + +type kolAssistRepository struct { + q generated.Querier +} + +func NewKolAssistRepository(db generated.DBTX) KolAssistRepository { + return &kolAssistRepository{q: newQuerier(db)} +} + +func (r *kolAssistRepository) Create(ctx context.Context, input CreateKolAssistInput) error { + return r.q.CreateKolAssistTask(ctx, generated.CreateKolAssistTaskParams{ + ID: input.ID, + TenantID: input.TenantID, + KolProfileID: input.KolProfileID, + PromptID: pgInt8(input.PromptID), + OperatorID: input.OperatorID, + TaskType: input.TaskType, + RequestJson: input.RequestJSON, + }) +} + +func (r *kolAssistRepository) Get(ctx context.Context, tenantID int64, id string) (*KolAssistTask, error) { + row, err := r.q.GetKolAssistTask(ctx, generated.GetKolAssistTaskParams{ + ID: id, + TenantID: tenantID, + }) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return nil, nil + } + return nil, err + } + return &KolAssistTask{ + ID: row.ID, + TenantID: row.TenantID, + KolProfileID: row.KolProfileID, + PromptID: nullableInt64(row.PromptID), + OperatorID: row.OperatorID, + TaskType: row.TaskType, + Status: row.Status, + RequestJSON: row.RequestJson, + ResultJSON: row.ResultJson, + ErrorMessage: nullableText(row.ErrorMessage), + StartedAt: optionalTime(row.StartedAt), + CompletedAt: optionalTime(row.CompletedAt), + CreatedAt: timeFromTimestamp(row.CreatedAt), + UpdatedAt: timeFromTimestamp(row.UpdatedAt), + }, nil +} + +func (r *kolAssistRepository) MarkStarted(ctx context.Context, tenantID int64, id string) error { + return r.q.MarkKolAssistStarted(ctx, generated.MarkKolAssistStartedParams{ + ID: id, + TenantID: tenantID, + }) +} + +func (r *kolAssistRepository) MarkCompleted(ctx context.Context, tenantID int64, id string, resultJSON []byte) error { + return r.q.MarkKolAssistCompleted(ctx, generated.MarkKolAssistCompletedParams{ + ResultJson: resultJSON, + ID: id, + TenantID: tenantID, + }) +} + +func (r *kolAssistRepository) MarkFailed(ctx context.Context, tenantID int64, id string, errorMsg string) error { + return r.q.MarkKolAssistFailed(ctx, generated.MarkKolAssistFailedParams{ + ErrorMessage: pgText(&errorMsg), + ID: id, + TenantID: tenantID, + }) +} diff --git a/server/internal/tenant/repository/kol_marketplace_repo.go b/server/internal/tenant/repository/kol_marketplace_repo.go new file mode 100644 index 0000000..baca113 --- /dev/null +++ b/server/internal/tenant/repository/kol_marketplace_repo.go @@ -0,0 +1,145 @@ +package repository + +import ( + "context" + "errors" + "time" + + "github.com/jackc/pgx/v5" + + "github.com/geo-platform/tenant-api/internal/tenant/repository/generated" +) + +type PublishedKolPackage struct { + ID int64 + TenantID int64 + KolProfileID int64 + Name string + Description *string + CoverURL *string + Industry *string + Tags []byte + PriceNote *string + Status string + CreatedAt time.Time + UpdatedAt time.Time + KolDisplayName string + KolAvatarURL *string + KolBio *string + PromptCount int64 + SubscriberCount int64 +} + +type PublicKolPrompt struct { + ID int64 + Name string + PlatformHint *string + Status string + PublishedRevisionNo *int + CreatedAt time.Time +} + +type ListPublishedFilter struct { + Industry *string + Keyword *string + Offset int + Limit int +} + +type KolMarketplaceRepository interface { + ListPublished(ctx context.Context, filter ListPublishedFilter) ([]PublishedKolPackage, error) + GetPublished(ctx context.Context, id int64) (*PublishedKolPackage, error) + ListPublicPrompts(ctx context.Context, packageID int64) ([]PublicKolPrompt, error) +} + +type kolMarketplaceRepository struct { + q generated.Querier +} + +func NewKolMarketplaceRepository(db generated.DBTX) KolMarketplaceRepository { + return &kolMarketplaceRepository{q: newQuerier(db)} +} + +func (r *kolMarketplaceRepository) ListPublished(ctx context.Context, filter ListPublishedFilter) ([]PublishedKolPackage, error) { + limit := filter.Limit + if limit <= 0 { + limit = 20 + } + rows, err := r.q.ListPublishedKolPackages(ctx, generated.ListPublishedKolPackagesParams{ + Industry: pgText(filter.Industry), + Keyword: pgText(filter.Keyword), + Offset: int32(filter.Offset), + Limit: int32(limit), + }) + if err != nil { + return nil, err + } + result := make([]PublishedKolPackage, len(rows)) + for i, row := range rows { + result[i] = PublishedKolPackage{ + ID: row.ID, + TenantID: row.TenantID, + KolProfileID: row.KolProfileID, + Name: row.Name, + Description: nullableText(row.Description), + CoverURL: nullableText(row.CoverUrl), + Industry: nullableText(row.Industry), + Tags: row.Tags, + PriceNote: nullableText(row.PriceNote), + Status: row.Status, + CreatedAt: timeFromTimestamp(row.CreatedAt), + UpdatedAt: timeFromTimestamp(row.UpdatedAt), + KolDisplayName: row.KolDisplayName, + KolAvatarURL: nullableText(row.KolAvatarUrl), + PromptCount: row.PromptCount, + SubscriberCount: row.SubscriberCount, + } + } + return result, nil +} + +func (r *kolMarketplaceRepository) GetPublished(ctx context.Context, id int64) (*PublishedKolPackage, error) { + row, err := r.q.GetPublishedKolPackage(ctx, id) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return nil, nil + } + return nil, err + } + return &PublishedKolPackage{ + ID: row.ID, + TenantID: row.TenantID, + KolProfileID: row.KolProfileID, + Name: row.Name, + Description: nullableText(row.Description), + CoverURL: nullableText(row.CoverUrl), + Industry: nullableText(row.Industry), + Tags: row.Tags, + PriceNote: nullableText(row.PriceNote), + Status: row.Status, + CreatedAt: timeFromTimestamp(row.CreatedAt), + UpdatedAt: timeFromTimestamp(row.UpdatedAt), + KolDisplayName: row.KolDisplayName, + KolAvatarURL: nullableText(row.KolAvatarUrl), + KolBio: nullableText(row.KolBio), + }, nil +} + +func (r *kolMarketplaceRepository) ListPublicPrompts(ctx context.Context, packageID int64) ([]PublicKolPrompt, error) { + rows, err := r.q.ListPublicPromptsForPackage(ctx, packageID) + if err != nil { + return nil, err + } + result := make([]PublicKolPrompt, len(rows)) + for i, row := range rows { + result[i] = PublicKolPrompt{ + ID: row.ID, + Name: row.Name, + PlatformHint: nullableText(row.PlatformHint), + Status: row.Status, + PublishedRevisionNo: nullableIntFromInt4(row.PublishedRevisionNo), + CreatedAt: timeFromTimestamp(row.CreatedAt), + } + } + return result, nil +} diff --git a/server/internal/tenant/repository/kol_package_repo.go b/server/internal/tenant/repository/kol_package_repo.go new file mode 100644 index 0000000..633c3ad --- /dev/null +++ b/server/internal/tenant/repository/kol_package_repo.go @@ -0,0 +1,192 @@ +package repository + +import ( + "context" + "errors" + "time" + + "github.com/jackc/pgx/v5" + + "github.com/geo-platform/tenant-api/internal/tenant/repository/generated" +) + +type KolPackage struct { + ID int64 + TenantID int64 + KolProfileID int64 + Name string + Description *string + CoverURL *string + Industry *string + Tags []byte + PriceNote *string + Status string + CreatedAt time.Time + UpdatedAt time.Time +} + +type CreateKolPackageInput struct { + TenantID int64 + KolProfileID int64 + Name string + Description *string + CoverURL *string + Industry *string + Tags []byte + PriceNote *string +} + +type UpdateKolPackageInput struct { + ID int64 + Name string + Description *string + CoverURL *string + Industry *string + Tags []byte + PriceNote *string +} + +type KolPackageRepository interface { + Create(ctx context.Context, input CreateKolPackageInput) (*KolPackage, error) + GetByID(ctx context.Context, tenantID, id int64) (*KolPackage, error) + ListByProfile(ctx context.Context, tenantID, profileID int64) ([]KolPackage, error) + ListPackageIDsByProfile(ctx context.Context, tenantID, profileID int64) ([]int64, error) + Update(ctx context.Context, tenantID int64, input UpdateKolPackageInput) error + UpdateStatus(ctx context.Context, tenantID, id int64, status string) error + SoftDelete(ctx context.Context, tenantID, id int64) error +} + +type kolPackageRepository struct { + q generated.Querier +} + +func NewKolPackageRepository(db generated.DBTX) KolPackageRepository { + return &kolPackageRepository{q: newQuerier(db)} +} + +func (r *kolPackageRepository) Create(ctx context.Context, input CreateKolPackageInput) (*KolPackage, error) { + row, err := r.q.CreateKolPackage(ctx, generated.CreateKolPackageParams{ + TenantID: input.TenantID, + KolProfileID: input.KolProfileID, + Name: input.Name, + Description: pgText(input.Description), + CoverUrl: pgText(input.CoverURL), + Industry: pgText(input.Industry), + Tags: input.Tags, + PriceNote: pgText(input.PriceNote), + }) + if err != nil { + return nil, err + } + return &KolPackage{ + ID: row.ID, + TenantID: row.TenantID, + KolProfileID: row.KolProfileID, + Name: row.Name, + Description: nullableText(row.Description), + CoverURL: nullableText(row.CoverUrl), + Industry: nullableText(row.Industry), + Tags: row.Tags, + PriceNote: nullableText(row.PriceNote), + Status: row.Status, + CreatedAt: timeFromTimestamp(row.CreatedAt), + UpdatedAt: timeFromTimestamp(row.UpdatedAt), + }, nil +} + +func (r *kolPackageRepository) GetByID(ctx context.Context, tenantID, id int64) (*KolPackage, error) { + row, err := r.q.GetKolPackageByID(ctx, generated.GetKolPackageByIDParams{ + ID: id, + TenantID: tenantID, + }) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return nil, nil + } + return nil, err + } + return &KolPackage{ + ID: row.ID, + TenantID: row.TenantID, + KolProfileID: row.KolProfileID, + Name: row.Name, + Description: nullableText(row.Description), + CoverURL: nullableText(row.CoverUrl), + Industry: nullableText(row.Industry), + Tags: row.Tags, + PriceNote: nullableText(row.PriceNote), + Status: row.Status, + CreatedAt: timeFromTimestamp(row.CreatedAt), + UpdatedAt: timeFromTimestamp(row.UpdatedAt), + }, nil +} + +func (r *kolPackageRepository) ListByProfile(ctx context.Context, tenantID, profileID int64) ([]KolPackage, error) { + rows, err := r.q.ListKolPackagesByProfile(ctx, generated.ListKolPackagesByProfileParams{ + KolProfileID: profileID, + TenantID: tenantID, + }) + if err != nil { + return nil, err + } + result := make([]KolPackage, len(rows)) + for i, row := range rows { + result[i] = KolPackage{ + ID: row.ID, + TenantID: row.TenantID, + KolProfileID: row.KolProfileID, + Name: row.Name, + Description: nullableText(row.Description), + CoverURL: nullableText(row.CoverUrl), + Industry: nullableText(row.Industry), + Tags: row.Tags, + PriceNote: nullableText(row.PriceNote), + Status: row.Status, + CreatedAt: timeFromTimestamp(row.CreatedAt), + UpdatedAt: timeFromTimestamp(row.UpdatedAt), + } + } + return result, nil +} + +func (r *kolPackageRepository) ListPackageIDsByProfile(ctx context.Context, tenantID, profileID int64) ([]int64, error) { + ids, err := r.q.ListKolPackageIDsByProfile(ctx, generated.ListKolPackageIDsByProfileParams{ + KolProfileID: profileID, + TenantID: tenantID, + }) + if err != nil { + return nil, err + } + if ids == nil { + return []int64{}, nil + } + return ids, nil +} + +func (r *kolPackageRepository) Update(ctx context.Context, tenantID int64, input UpdateKolPackageInput) error { + return r.q.UpdateKolPackage(ctx, generated.UpdateKolPackageParams{ + Name: input.Name, + Description: pgText(input.Description), + CoverUrl: pgText(input.CoverURL), + Industry: pgText(input.Industry), + Tags: input.Tags, + PriceNote: pgText(input.PriceNote), + ID: input.ID, + TenantID: tenantID, + }) +} + +func (r *kolPackageRepository) UpdateStatus(ctx context.Context, tenantID, id int64, status string) error { + return r.q.UpdateKolPackageStatus(ctx, generated.UpdateKolPackageStatusParams{ + Status: status, + ID: id, + TenantID: tenantID, + }) +} + +func (r *kolPackageRepository) SoftDelete(ctx context.Context, tenantID, id int64) error { + return r.q.SoftDeleteKolPackage(ctx, generated.SoftDeleteKolPackageParams{ + ID: id, + TenantID: tenantID, + }) +} diff --git a/server/internal/tenant/repository/kol_profile_repo.go b/server/internal/tenant/repository/kol_profile_repo.go new file mode 100644 index 0000000..69532e2 --- /dev/null +++ b/server/internal/tenant/repository/kol_profile_repo.go @@ -0,0 +1,102 @@ +package repository + +import ( + "context" + "errors" + "time" + + "github.com/jackc/pgx/v5" + + "github.com/geo-platform/tenant-api/internal/tenant/repository/generated" +) + +type KolProfile struct { + ID int64 + TenantID int64 + UserID int64 + DisplayName string + Bio *string + AvatarURL *string + MarketEnabled bool + Status string + CreatedAt time.Time + UpdatedAt time.Time +} + +type UpdateKolProfileInput struct { + ID int64 + DisplayName string + Bio *string + AvatarURL *string +} + +type KolProfileRepository interface { + GetByUser(ctx context.Context, tenantID, userID int64) (*KolProfile, error) + GetByID(ctx context.Context, id int64) (*KolProfile, error) + Update(ctx context.Context, tenantID int64, input UpdateKolProfileInput) error +} + +type kolProfileRepository struct { + q generated.Querier +} + +func NewKolProfileRepository(db generated.DBTX) KolProfileRepository { + return &kolProfileRepository{q: newQuerier(db)} +} + +func (r *kolProfileRepository) GetByUser(ctx context.Context, tenantID, userID int64) (*KolProfile, error) { + row, err := r.q.GetKolProfileByUser(ctx, generated.GetKolProfileByUserParams{ + UserID: userID, + TenantID: tenantID, + }) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return nil, nil + } + return nil, err + } + return &KolProfile{ + ID: row.ID, + TenantID: row.TenantID, + UserID: row.UserID, + DisplayName: row.DisplayName, + Bio: nullableText(row.Bio), + AvatarURL: nullableText(row.AvatarUrl), + MarketEnabled: row.MarketEnabled, + Status: row.Status, + CreatedAt: timeFromTimestamp(row.CreatedAt), + UpdatedAt: timeFromTimestamp(row.UpdatedAt), + }, nil +} + +func (r *kolProfileRepository) GetByID(ctx context.Context, id int64) (*KolProfile, error) { + row, err := r.q.GetKolProfileByID(ctx, id) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return nil, nil + } + return nil, err + } + return &KolProfile{ + ID: row.ID, + TenantID: row.TenantID, + UserID: row.UserID, + DisplayName: row.DisplayName, + Bio: nullableText(row.Bio), + AvatarURL: nullableText(row.AvatarUrl), + MarketEnabled: row.MarketEnabled, + Status: row.Status, + CreatedAt: timeFromTimestamp(row.CreatedAt), + UpdatedAt: timeFromTimestamp(row.UpdatedAt), + }, nil +} + +func (r *kolProfileRepository) Update(ctx context.Context, tenantID int64, input UpdateKolProfileInput) error { + return r.q.UpdateKolProfile(ctx, generated.UpdateKolProfileParams{ + DisplayName: input.DisplayName, + Bio: pgText(input.Bio), + AvatarUrl: pgText(input.AvatarURL), + ID: input.ID, + TenantID: tenantID, + }) +} diff --git a/server/internal/tenant/repository/kol_prompt_repo.go b/server/internal/tenant/repository/kol_prompt_repo.go new file mode 100644 index 0000000..2a829f3 --- /dev/null +++ b/server/internal/tenant/repository/kol_prompt_repo.go @@ -0,0 +1,323 @@ +package repository + +import ( + "context" + "errors" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgtype" + + "github.com/geo-platform/tenant-api/internal/tenant/repository/generated" +) + +type KolPrompt struct { + ID int64 + TenantID int64 + PackageID int64 + Name string + PlatformHint *string + Status string + SortOrder int + PublishedRevisionNo *int + DraftRevisionNo *int + CreatedBy int64 + UpdatedBy int64 + CreatedAt time.Time + UpdatedAt time.Time +} + +type KolPromptRevision struct { + ID int64 + TenantID int64 + PromptID int64 + RevisionNo int + PromptAssetKey string + SchemaJSON []byte + CardConfigJSON []byte + CreatedBy int64 + CreatedAt time.Time +} + +type ActiveKolPrompt struct { + ID int64 + TenantID int64 + PackageID int64 + PublishedRevisionNo *int +} + +type CreateKolPromptInput struct { + TenantID int64 + PackageID int64 + Name string + PlatformHint *string + SortOrder int + CreatedBy int64 +} + +type UpdateKolPromptMetadataInput struct { + ID int64 + TenantID int64 + Name string + PlatformHint *string + SortOrder int + UpdatedBy int64 +} + +type CreateKolPromptRevisionInput struct { + TenantID int64 + PromptID int64 + RevisionNo int + PromptAssetKey string + SchemaJSON []byte + CardConfigJSON []byte + CreatedBy int64 +} + +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) + ListActiveByPackage(ctx context.Context, tenantID, packageID int64) ([]ActiveKolPrompt, error) + UpdateMetadata(ctx context.Context, input UpdateKolPromptMetadataInput) error + SetDraftRevision(ctx context.Context, tenantID, promptID int64, revNo int, updatedBy int64) error + SetPublishedRevision(ctx context.Context, tenantID, promptID int64, revNo int, updatedBy int64) error + UpdateStatus(ctx context.Context, tenantID, id int64, status string, updatedBy int64) error + SoftDelete(ctx context.Context, tenantID, id int64) error + CreateRevision(ctx context.Context, input CreateKolPromptRevisionInput) (*KolPromptRevision, error) + GetRevision(ctx context.Context, tenantID, promptID int64, revNo int) (*KolPromptRevision, error) + NextRevisionNo(ctx context.Context, tenantID, promptID int64) (int, error) +} + +type kolPromptRepository struct { + q generated.Querier +} + +func NewKolPromptRepository(db generated.DBTX) KolPromptRepository { + return &kolPromptRepository{q: newQuerier(db)} +} + +func mapKolPrompt(row generated.GetKolPromptByIDRow) *KolPrompt { + return &KolPrompt{ + ID: row.ID, + TenantID: row.TenantID, + PackageID: row.PackageID, + Name: row.Name, + PlatformHint: nullableText(row.PlatformHint), + Status: row.Status, + SortOrder: int(row.SortOrder), + PublishedRevisionNo: nullableIntFromInt4(row.PublishedRevisionNo), + DraftRevisionNo: nullableIntFromInt4(row.DraftRevisionNo), + CreatedBy: row.CreatedBy, + UpdatedBy: row.UpdatedBy, + CreatedAt: timeFromTimestamp(row.CreatedAt), + UpdatedAt: timeFromTimestamp(row.UpdatedAt), + } +} + +func (r *kolPromptRepository) Create(ctx context.Context, input CreateKolPromptInput) (*KolPrompt, error) { + row, err := r.q.CreateKolPrompt(ctx, generated.CreateKolPromptParams{ + TenantID: input.TenantID, + PackageID: input.PackageID, + Name: input.Name, + PlatformHint: pgText(input.PlatformHint), + SortOrder: int32(input.SortOrder), + CreatedBy: input.CreatedBy, + }) + if err != nil { + return nil, err + } + return &KolPrompt{ + ID: row.ID, + TenantID: row.TenantID, + PackageID: row.PackageID, + Name: row.Name, + PlatformHint: nullableText(row.PlatformHint), + Status: row.Status, + SortOrder: int(row.SortOrder), + PublishedRevisionNo: nullableIntFromInt4(row.PublishedRevisionNo), + DraftRevisionNo: nullableIntFromInt4(row.DraftRevisionNo), + CreatedBy: row.CreatedBy, + UpdatedBy: row.UpdatedBy, + CreatedAt: timeFromTimestamp(row.CreatedAt), + UpdatedAt: timeFromTimestamp(row.UpdatedAt), + }, nil +} + +func (r *kolPromptRepository) GetByID(ctx context.Context, tenantID, id int64) (*KolPrompt, error) { + row, err := r.q.GetKolPromptByID(ctx, generated.GetKolPromptByIDParams{ + ID: id, + TenantID: tenantID, + }) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return nil, nil + } + return nil, err + } + return mapKolPrompt(row), nil +} + +func (r *kolPromptRepository) ListByPackage(ctx context.Context, tenantID, packageID int64) ([]KolPrompt, error) { + rows, err := r.q.ListKolPromptsByPackage(ctx, generated.ListKolPromptsByPackageParams{ + PackageID: packageID, + TenantID: tenantID, + }) + if err != nil { + return nil, err + } + result := make([]KolPrompt, len(rows)) + for i, row := range rows { + result[i] = KolPrompt{ + ID: row.ID, + TenantID: row.TenantID, + PackageID: row.PackageID, + Name: row.Name, + PlatformHint: nullableText(row.PlatformHint), + Status: row.Status, + SortOrder: int(row.SortOrder), + PublishedRevisionNo: nullableIntFromInt4(row.PublishedRevisionNo), + DraftRevisionNo: nullableIntFromInt4(row.DraftRevisionNo), + CreatedBy: row.CreatedBy, + UpdatedBy: row.UpdatedBy, + CreatedAt: timeFromTimestamp(row.CreatedAt), + UpdatedAt: timeFromTimestamp(row.UpdatedAt), + } + } + return result, nil +} + +func (r *kolPromptRepository) ListActiveByPackage(ctx context.Context, tenantID, packageID int64) ([]ActiveKolPrompt, error) { + rows, err := r.q.ListActiveKolPromptsByPackage(ctx, generated.ListActiveKolPromptsByPackageParams{ + PackageID: packageID, + TenantID: tenantID, + }) + if err != nil { + return nil, err + } + result := make([]ActiveKolPrompt, len(rows)) + for i, row := range rows { + result[i] = ActiveKolPrompt{ + ID: row.ID, + TenantID: row.TenantID, + PackageID: row.PackageID, + PublishedRevisionNo: nullableIntFromInt4(row.PublishedRevisionNo), + } + } + return result, nil +} + +func (r *kolPromptRepository) UpdateMetadata(ctx context.Context, input UpdateKolPromptMetadataInput) error { + return r.q.UpdateKolPromptMetadata(ctx, generated.UpdateKolPromptMetadataParams{ + Name: input.Name, + PlatformHint: pgText(input.PlatformHint), + SortOrder: int32(input.SortOrder), + UpdatedBy: input.UpdatedBy, + ID: input.ID, + TenantID: input.TenantID, + }) +} + +func (r *kolPromptRepository) SetDraftRevision(ctx context.Context, tenantID, promptID int64, revNo int, updatedBy int64) error { + return r.q.UpdateKolPromptDraftRevision(ctx, generated.UpdateKolPromptDraftRevisionParams{ + RevisionNo: pgtype.Int4{Int32: int32(revNo), Valid: true}, + UpdatedBy: updatedBy, + ID: promptID, + TenantID: tenantID, + }) +} + +func (r *kolPromptRepository) SetPublishedRevision(ctx context.Context, tenantID, promptID int64, revNo int, updatedBy int64) error { + return r.q.UpdateKolPromptPublishedRevision(ctx, generated.UpdateKolPromptPublishedRevisionParams{ + RevisionNo: pgtype.Int4{Int32: int32(revNo), Valid: true}, + UpdatedBy: updatedBy, + ID: promptID, + TenantID: tenantID, + }) +} + +func (r *kolPromptRepository) UpdateStatus(ctx context.Context, tenantID, id int64, status string, updatedBy int64) error { + return r.q.UpdateKolPromptStatus(ctx, generated.UpdateKolPromptStatusParams{ + Status: status, + UpdatedBy: updatedBy, + ID: id, + TenantID: tenantID, + }) +} + +func (r *kolPromptRepository) SoftDelete(ctx context.Context, tenantID, id int64) error { + return r.q.SoftDeleteKolPrompt(ctx, generated.SoftDeleteKolPromptParams{ + ID: id, + TenantID: tenantID, + }) +} + +func (r *kolPromptRepository) CreateRevision(ctx context.Context, input CreateKolPromptRevisionInput) (*KolPromptRevision, error) { + row, err := r.q.CreateKolPromptRevision(ctx, generated.CreateKolPromptRevisionParams{ + TenantID: input.TenantID, + PromptID: input.PromptID, + RevisionNo: int32(input.RevisionNo), + PromptAssetKey: input.PromptAssetKey, + SchemaJson: input.SchemaJSON, + CardConfigJson: input.CardConfigJSON, + CreatedBy: input.CreatedBy, + }) + if err != nil { + return nil, err + } + return &KolPromptRevision{ + ID: row.ID, + TenantID: input.TenantID, + PromptID: row.PromptID, + RevisionNo: int(row.RevisionNo), + PromptAssetKey: row.PromptAssetKey, + SchemaJSON: row.SchemaJson, + CardConfigJSON: row.CardConfigJson, + CreatedBy: row.CreatedBy, + CreatedAt: timeFromTimestamp(row.CreatedAt), + }, nil +} + +func (r *kolPromptRepository) GetRevision(ctx context.Context, tenantID, promptID int64, revNo int) (*KolPromptRevision, error) { + row, err := r.q.GetKolPromptRevision(ctx, generated.GetKolPromptRevisionParams{ + PromptID: promptID, + RevisionNo: int32(revNo), + TenantID: tenantID, + }) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return nil, nil + } + return nil, err + } + return &KolPromptRevision{ + ID: row.ID, + TenantID: tenantID, + PromptID: row.PromptID, + RevisionNo: int(row.RevisionNo), + PromptAssetKey: row.PromptAssetKey, + SchemaJSON: row.SchemaJson, + CardConfigJSON: row.CardConfigJson, + CreatedBy: row.CreatedBy, + CreatedAt: timeFromTimestamp(row.CreatedAt), + }, nil +} + +func (r *kolPromptRepository) NextRevisionNo(ctx context.Context, tenantID, promptID int64) (int, error) { + next, err := r.q.NextKolPromptRevisionNo(ctx, generated.NextKolPromptRevisionNoParams{ + PromptID: promptID, + TenantID: tenantID, + }) + if err != nil { + return 0, err + } + return int(next), nil +} + +func nullableIntFromInt4(value pgtype.Int4) *int { + if !value.Valid { + return nil + } + v := int(value.Int32) + return &v +} diff --git a/server/internal/tenant/repository/kol_subscription_repo.go b/server/internal/tenant/repository/kol_subscription_repo.go new file mode 100644 index 0000000..764bef9 --- /dev/null +++ b/server/internal/tenant/repository/kol_subscription_repo.go @@ -0,0 +1,255 @@ +package repository + +import ( + "context" + "errors" + "time" + + "github.com/jackc/pgx/v5" + + "github.com/geo-platform/tenant-api/internal/tenant/repository/generated" +) + +type KolSubscription struct { + ID int64 + TenantID int64 + PackageID int64 + Status string + ApprovedBy *int64 + StartAt *time.Time + EndAt *time.Time + CreatedAt time.Time + UpdatedAt time.Time +} + +type KolSubscriptionPrompt struct { + ID int64 + TenantID int64 + SubscriptionID int64 + PackageID int64 + PromptID int64 + Status string + GrantedAt *time.Time + RevokedAt *time.Time + PromptName string + PlatformHint *string + PackageName string + PublishedRevisionNo *int + // Fields below only populated by GetSubscriptionPromptByID. + SubscriptionStatus string + SubscriptionEndAt *time.Time + PromptStatus string + PackageStatus string +} + +type CreateAccessRowInput struct { + TenantID int64 + SubscriptionID int64 + PackageID int64 + PromptID int64 +} + +type KolSubscriptionRepository interface { + Create(ctx context.Context, tenantID, packageID int64) (*KolSubscription, error) + GetByID(ctx context.Context, tenantID, id int64) (*KolSubscription, error) + GetActiveForTenant(ctx context.Context, tenantID, packageID int64) (*KolSubscription, error) + ListByTenant(ctx context.Context, tenantID int64) ([]KolSubscription, error) + ListActiveSubscribersForPackage(ctx context.Context, packageID int64) ([]KolSubscription, 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 + ListSubscriptionPromptsByTenant(ctx context.Context, tenantID int64) ([]KolSubscriptionPrompt, error) + GetSubscriptionPromptByID(ctx context.Context, tenantID, id int64) (*KolSubscriptionPrompt, error) + RevokeAccessRowsBySubscription(ctx context.Context, subscriptionID, tenantID int64) error + RevokeAccessRowsByPrompt(ctx context.Context, promptID, tenantID int64) error +} + +type kolSubscriptionRepository struct { + q generated.Querier +} + +func NewKolSubscriptionRepository(db generated.DBTX) KolSubscriptionRepository { + return &kolSubscriptionRepository{q: newQuerier(db)} +} + +func mapKolSubscription(row generated.KolSubscription) KolSubscription { + startAt := optionalTime(row.StartAt) + endAt := optionalTime(row.EndAt) + return KolSubscription{ + ID: row.ID, + TenantID: row.TenantID, + PackageID: row.PackageID, + Status: row.Status, + ApprovedBy: nullableInt64(row.ApprovedBy), + StartAt: startAt, + EndAt: endAt, + CreatedAt: timeFromTimestamp(row.CreatedAt), + UpdatedAt: timeFromTimestamp(row.UpdatedAt), + } +} + +func (r *kolSubscriptionRepository) Create(ctx context.Context, tenantID, packageID int64) (*KolSubscription, error) { + row, err := r.q.CreateKolSubscription(ctx, generated.CreateKolSubscriptionParams{ + TenantID: tenantID, + PackageID: packageID, + }) + if err != nil { + return nil, err + } + sub := mapKolSubscription(row) + return &sub, nil +} + +func (r *kolSubscriptionRepository) GetByID(ctx context.Context, tenantID, id int64) (*KolSubscription, error) { + row, err := r.q.GetKolSubscriptionByID(ctx, generated.GetKolSubscriptionByIDParams{ + ID: id, + TenantID: tenantID, + }) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return nil, nil + } + return nil, err + } + sub := mapKolSubscription(row) + return &sub, nil +} + +func (r *kolSubscriptionRepository) GetActiveForTenant(ctx context.Context, tenantID, packageID int64) (*KolSubscription, error) { + row, err := r.q.GetActiveKolSubscription(ctx, generated.GetActiveKolSubscriptionParams{ + TenantID: tenantID, + PackageID: packageID, + }) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return nil, nil + } + return nil, err + } + sub := mapKolSubscription(row) + return &sub, nil +} + +func (r *kolSubscriptionRepository) ListByTenant(ctx context.Context, tenantID int64) ([]KolSubscription, error) { + rows, err := r.q.ListKolSubscriptions(ctx, tenantID) + if err != nil { + return nil, err + } + result := make([]KolSubscription, len(rows)) + for i, row := range rows { + result[i] = mapKolSubscription(row) + } + return result, nil +} + +func (r *kolSubscriptionRepository) ListActiveSubscribersForPackage(ctx context.Context, packageID int64) ([]KolSubscription, error) { + rows, err := r.q.ListActiveSubscribersForPackage(ctx, packageID) + if err != nil { + return nil, err + } + result := make([]KolSubscription, len(rows)) + for i, row := range rows { + result[i] = mapKolSubscription(row) + } + return result, 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), + EndAt: pgTimestamp(endAt), + ID: id, + TenantID: tenantID, + }) +} + +func (r *kolSubscriptionRepository) RevokeByTenant(ctx context.Context, id, tenantID int64) error { + return r.q.RevokeKolSubscription(ctx, generated.RevokeKolSubscriptionParams{ + ID: id, + TenantID: tenantID, + }) +} + +func (r *kolSubscriptionRepository) CreateAccessRow(ctx context.Context, input CreateAccessRowInput) error { + return r.q.CreateSubscriptionPrompt(ctx, generated.CreateSubscriptionPromptParams{ + TenantID: input.TenantID, + SubscriptionID: input.SubscriptionID, + PackageID: input.PackageID, + PromptID: input.PromptID, + }) +} + +func (r *kolSubscriptionRepository) ListSubscriptionPromptsByTenant(ctx context.Context, tenantID int64) ([]KolSubscriptionPrompt, error) { + rows, err := r.q.ListSubscriptionPromptsByTenant(ctx, tenantID) + if err != nil { + return nil, err + } + result := make([]KolSubscriptionPrompt, len(rows)) + for i, row := range rows { + grantedAt := optionalTime(row.GrantedAt) + revokedAt := optionalTime(row.RevokedAt) + result[i] = KolSubscriptionPrompt{ + ID: row.ID, + TenantID: row.TenantID, + SubscriptionID: row.SubscriptionID, + PackageID: row.PackageID, + PromptID: row.PromptID, + Status: row.Status, + GrantedAt: grantedAt, + RevokedAt: revokedAt, + PromptName: row.PromptName, + PlatformHint: nullableText(row.PlatformHint), + PackageName: row.PackageName, + PublishedRevisionNo: nullableIntFromInt4(row.PublishedRevisionNo), + } + } + return result, nil +} + +func (r *kolSubscriptionRepository) GetSubscriptionPromptByID(ctx context.Context, tenantID, id int64) (*KolSubscriptionPrompt, error) { + row, err := r.q.GetSubscriptionPromptByID(ctx, generated.GetSubscriptionPromptByIDParams{ + ID: id, + TenantID: tenantID, + }) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return nil, nil + } + return nil, err + } + grantedAt := optionalTime(row.GrantedAt) + revokedAt := optionalTime(row.RevokedAt) + subEndAt := optionalTime(row.SubscriptionEndAt) + return &KolSubscriptionPrompt{ + ID: row.ID, + TenantID: row.TenantID, + SubscriptionID: row.SubscriptionID, + PackageID: row.PackageID, + PromptID: row.PromptID, + Status: row.Status, + GrantedAt: grantedAt, + RevokedAt: revokedAt, + PromptName: row.PromptName, + PlatformHint: nullableText(row.PlatformHint), + PackageName: row.PackageName, + PublishedRevisionNo: nullableIntFromInt4(row.PublishedRevisionNo), + SubscriptionStatus: row.SubscriptionStatus, + SubscriptionEndAt: subEndAt, + PromptStatus: row.PromptStatus, + PackageStatus: row.PackageStatus, + }, nil +} + +func (r *kolSubscriptionRepository) RevokeAccessRowsBySubscription(ctx context.Context, subscriptionID, tenantID int64) error { + return r.q.RevokeSubscriptionPromptsBySubscription(ctx, generated.RevokeSubscriptionPromptsBySubscriptionParams{ + SubscriptionID: subscriptionID, + TenantID: tenantID, + }) +} + +func (r *kolSubscriptionRepository) RevokeAccessRowsByPrompt(ctx context.Context, promptID, tenantID int64) error { + return r.q.RevokeSubscriptionPromptsByPrompt(ctx, generated.RevokeSubscriptionPromptsByPromptParams{ + PromptID: promptID, + TenantID: tenantID, + }) +} diff --git a/server/internal/tenant/repository/kol_usage_repo.go b/server/internal/tenant/repository/kol_usage_repo.go new file mode 100644 index 0000000..125176b --- /dev/null +++ b/server/internal/tenant/repository/kol_usage_repo.go @@ -0,0 +1,198 @@ +package repository + +import ( + "context" + "time" + + "github.com/geo-platform/tenant-api/internal/tenant/repository/generated" +) + +type KolUsageLog struct { + ID int64 + TenantID int64 + GenerationTaskID *int64 + Status string + CreatedAt time.Time +} + +type KolUsageByTaskRecord struct { + ID int64 + TenantID int64 + PackageID int64 + PromptID int64 + Status string + CreatedAt time.Time +} + +type CreateKolUsageLogInput struct { + TenantID int64 + SubscriptionID *int64 + SubscriptionPromptID *int64 + PackageID int64 + PromptID int64 + PromptRevisionNo int + GenerationTaskID *int64 + VariablesSnapshot []byte + SchemaSnapshot []byte + RenderedPromptHash *string +} + +type KolDashboardOverview struct { + TotalSubs int64 + MonthSubs int64 + TotalUsage int64 + MonthUsage int64 + FailureCount int64 +} + +type KolDashboardTrendPoint struct { + Day time.Time + Usages int64 +} + +type KolUsagePackageStats struct { + PackageID int64 + Total int64 + Completed int64 + Failed int64 +} + +type KolUsageRepository interface { + Create(ctx context.Context, input CreateKolUsageLogInput) (*KolUsageLog, error) + MarkCompletedByTask(ctx context.Context, tenantID, taskID, articleID int64) error + MarkFailedByTask(ctx context.Context, tenantID, taskID int64, errorMsg string) error + GetByGenerationTask(ctx context.Context, tenantID, taskID int64) (*KolUsageByTaskRecord, error) + DashboardOverview(ctx context.Context, packageIDs []int64) (*KolDashboardOverview, error) + DashboardTrend(ctx context.Context, packageIDs []int64, periodDays int) ([]KolDashboardTrendPoint, error) + PerPackageStats(ctx context.Context, packageIDs []int64) ([]KolUsagePackageStats, error) +} + +type kolUsageRepository struct { + q generated.Querier +} + +func NewKolUsageRepository(db generated.DBTX) KolUsageRepository { + return &kolUsageRepository{q: newQuerier(db)} +} + +func (r *kolUsageRepository) Create(ctx context.Context, input CreateKolUsageLogInput) (*KolUsageLog, error) { + row, err := r.q.CreateKolUsageLog(ctx, generated.CreateKolUsageLogParams{ + TenantID: input.TenantID, + SubscriptionID: pgInt8(input.SubscriptionID), + SubscriptionPromptID: pgInt8(input.SubscriptionPromptID), + PackageID: input.PackageID, + PromptID: input.PromptID, + PromptRevisionNo: int32(input.PromptRevisionNo), + GenerationTaskID: pgInt8(input.GenerationTaskID), + VariablesSnapshot: input.VariablesSnapshot, + SchemaSnapshot: input.SchemaSnapshot, + RenderedPromptHash: pgText(input.RenderedPromptHash), + }) + if err != nil { + return nil, err + } + return &KolUsageLog{ + ID: row.ID, + TenantID: row.TenantID, + GenerationTaskID: nullableInt64(row.GenerationTaskID), + Status: row.Status, + CreatedAt: timeFromTimestamp(row.CreatedAt), + }, nil +} + +func (r *kolUsageRepository) MarkCompletedByTask(ctx context.Context, tenantID, taskID, articleID int64) error { + return r.q.MarkKolUsageCompletedByTask(ctx, generated.MarkKolUsageCompletedByTaskParams{ + ArticleID: pgInt8(&articleID), + GenerationTaskID: pgInt8(&taskID), + TenantID: tenantID, + }) +} + +func (r *kolUsageRepository) MarkFailedByTask(ctx context.Context, tenantID, taskID int64, errorMsg string) error { + return r.q.MarkKolUsageFailedByTask(ctx, generated.MarkKolUsageFailedByTaskParams{ + ErrorMessage: pgText(&errorMsg), + GenerationTaskID: pgInt8(&taskID), + TenantID: tenantID, + }) +} + +func (r *kolUsageRepository) GetByGenerationTask(ctx context.Context, tenantID, taskID int64) (*KolUsageByTaskRecord, error) { + row, err := r.q.GetKolUsageByGenerationTask(ctx, generated.GetKolUsageByGenerationTaskParams{ + GenerationTaskID: pgInt8(&taskID), + TenantID: tenantID, + }) + if err != nil { + // Callers use errors.Is(err, pgx.ErrNoRows) for "not found"; preserve err semantics. + return nil, err + } + return &KolUsageByTaskRecord{ + ID: row.ID, + TenantID: row.TenantID, + PackageID: row.PackageID, + PromptID: row.PromptID, + Status: row.Status, + CreatedAt: timeFromTimestamp(row.CreatedAt), + }, nil +} + +func (r *kolUsageRepository) DashboardOverview(ctx context.Context, packageIDs []int64) (*KolDashboardOverview, error) { + if len(packageIDs) == 0 { + return &KolDashboardOverview{}, nil + } + row, err := r.q.KolDashboardOverview(ctx, packageIDs) + if err != nil { + return nil, err + } + return &KolDashboardOverview{ + TotalSubs: row.TotalSubs, + MonthSubs: row.MonthSubs, + TotalUsage: row.TotalUsage, + MonthUsage: row.MonthUsage, + FailureCount: row.FailureCount, + }, nil +} + +func (r *kolUsageRepository) DashboardTrend(ctx context.Context, packageIDs []int64, periodDays int) ([]KolDashboardTrendPoint, error) { + if len(packageIDs) == 0 { + return []KolDashboardTrendPoint{}, nil + } + rows, err := r.q.KolDashboardTrend(ctx, generated.KolDashboardTrendParams{ + PackageIds: packageIDs, + PeriodDays: int32(periodDays), + }) + if err != nil { + return nil, err + } + result := make([]KolDashboardTrendPoint, len(rows)) + for i, row := range rows { + var day time.Time + if row.D.Valid { + day = row.D.Time + } + result[i] = KolDashboardTrendPoint{ + Day: day, + Usages: row.Usages, + } + } + return result, nil +} + +func (r *kolUsageRepository) PerPackageStats(ctx context.Context, packageIDs []int64) ([]KolUsagePackageStats, error) { + if len(packageIDs) == 0 { + return []KolUsagePackageStats{}, nil + } + rows, err := r.q.KolUsageCountByPackage(ctx, packageIDs) + if err != nil { + return nil, err + } + result := make([]KolUsagePackageStats, len(rows)) + for i, row := range rows { + result[i] = KolUsagePackageStats{ + PackageID: row.PackageID, + Total: row.Total, + Completed: row.Completed, + Failed: row.Failed, + } + } + return result, nil +}