geo/server/internal/ops/repository/kol_subscription.go

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)
}