kefu/im/backend/internal/app/membership_limits.go

package app

import (
	"context"
	"database/sql"
	"fmt"
	"strconv"
	"strings"
)

const defaultFreeDailyActiveChatLimit = 5

type rowQuerier interface {
	QueryRowContext(context.Context, string, ...any) *sql.Row
}

type dailyActiveChatLimitError struct {
	Limit int
}

func (e *dailyActiveChatLimitError) Error() string {
	return fmt.Sprintf("今日主动聊天人数已达上限（%d人），回复收到的消息不受此限制", e.Limit)
}

func (a *App) resolveDailyActiveChatLimit(ctx context.Context, queryer rowQuerier, userID int64) int {
	var limit int
	err := queryer.QueryRowContext(ctx, `SELECT p.daily_active_chat_limit
		FROM subscriptions s JOIN membership_plans p ON p.id=s.plan_id
		WHERE s.user_id=? AND s.status=1 AND s.started_at<=NOW(3) AND s.expires_at>NOW(3)
		ORDER BY p.level DESC,s.expires_at DESC LIMIT 1`, userID).Scan(&limit)
	if err == nil {
		return limit
	}

	var raw string
	if err = queryer.QueryRowContext(ctx, `SELECT config_value FROM system_configs WHERE config_key='membership.free_daily_active_chat_limit'`).Scan(&raw); err == nil {
		if parsed, parseErr := strconv.Atoi(raw); parseErr == nil && parsed >= 0 {
			return parsed
		}
	}
	return defaultFreeDailyActiveChatLimit
}

func dailyActiveChatQuotaView(limit, used int) map[string]any {
	remaining := -1
	unlimited := limit == 0
	if !unlimited {
		remaining = limit - used
		if remaining < 0 {
			remaining = 0
		}
	}
	return map[string]any{
		"limit":     limit,
		"remaining": remaining,
		"unlimited": unlimited,
		"used":      used,
	}
}

func (a *App) dailyActiveChatQuota(ctx context.Context, userID int64) map[string]any {
	limit := a.resolveDailyActiveChatLimit(ctx, a.db, userID)
	var used int
	_ = a.db.QueryRowContext(ctx, `SELECT used_count FROM im_daily_active_chat_usage WHERE user_id=? AND usage_date=CURRENT_DATE()`, userID).Scan(&used)
	return dailyActiveChatQuotaView(limit, used)
}

func (a *App) reserveDailyActiveChat(ctx context.Context, tx *sql.Tx, conversationID, senderID int64) error {
	var user1ID, user2ID int64
	err := tx.QueryRowContext(ctx, `SELECT user1_id,user2_id FROM im_direct_conversations WHERE conversation_id=?`, conversationID).Scan(&user1ID, &user2ID)
	if err == sql.ErrNoRows {
		return nil
	}
	if err != nil {
		return err
	}
	targetUserID := user1ID
	if senderID == user1ID {
		targetUserID = user2ID
	} else if senderID != user2ID {
		return fmt.Errorf("不是会话成员")
	}

	var inboundToday int
	if err = tx.QueryRowContext(ctx, `SELECT EXISTS(
		SELECT 1 FROM im_messages
		WHERE conversation_id=? AND sender_id=?
		AND created_at>=CURRENT_DATE() AND created_at<DATE_ADD(CURRENT_DATE(),INTERVAL 1 DAY)
	)`, conversationID, targetUserID).Scan(&inboundToday); err != nil {
		return err
	}
	if inboundToday == 1 {
		return nil
	}

	if _, err = tx.ExecContext(ctx, `INSERT IGNORE INTO im_daily_active_chat_usage(user_id,usage_date,used_count) VALUES(?,CURRENT_DATE(),0)`, senderID); err != nil {
		return err
	}
	var used int
	if err = tx.QueryRowContext(ctx, `SELECT used_count FROM im_daily_active_chat_usage WHERE user_id=? AND usage_date=CURRENT_DATE() FOR UPDATE`, senderID).Scan(&used); err != nil {
		return err
	}
	var alreadyCounted int
	if err = tx.QueryRowContext(ctx, `SELECT EXISTS(
		SELECT 1 FROM im_daily_active_chat_targets
		WHERE user_id=? AND target_user_id=? AND usage_date=CURRENT_DATE()
	)`, senderID, targetUserID).Scan(&alreadyCounted); err != nil {
		return err
	}
	if alreadyCounted == 1 {
		return nil
	}

	limit := a.resolveDailyActiveChatLimit(ctx, tx, senderID)
	if limit > 0 && used >= limit {
		return &dailyActiveChatLimitError{Limit: limit}
	}
	if _, err = tx.ExecContext(ctx, `INSERT INTO im_daily_active_chat_targets(user_id,target_user_id,usage_date,conversation_id) VALUES(?,?,CURRENT_DATE(),?)`, senderID, targetUserID, conversationID); err != nil {
		return err
	}
	_, err = tx.ExecContext(ctx, `UPDATE im_daily_active_chat_usage SET used_count=used_count+1 WHERE user_id=? AND usage_date=CURRENT_DATE()`, senderID)
	return err
}

const (
	messageTypeImage = 2
	messageTypeVoice = 3
)

// messageMembershipGateError names the message kind the reader was blocked on,
// so the client can offer the right upgrade prompt instead of a generic error.
type messageMembershipGateError struct{ Kind string }

func (e *messageMembershipGateError) Error() string {
	return fmt.Sprintf("发送%s消息需要开通会员", e.Kind)
}

// hasActiveMembership uses the same subscription row that /membership/status
// reports as "active", so the app never offers what the server will refuse.
// Every grant path — order, admin, gateway callback — writes that row.
func (a *App) hasActiveMembership(ctx context.Context, userID int64) bool {
	var one int
	return a.db.QueryRowContext(ctx, `SELECT 1 FROM subscriptions WHERE user_id=? AND status=1 AND started_at<=NOW(3) AND expires_at>NOW(3) LIMIT 1`, userID).Scan(&one) == nil
}

// messageMembershipGate reports whether one message type is members-only and
// what to call it. Both switches are edited from the admin console.
func (a *App) messageMembershipGate(ctx context.Context, messageType int) (string, bool) {
	switch messageType {
	case messageTypeImage:
		return "图片", a.configBool(ctx, "membership.image_message_requires_vip", false)
	case messageTypeVoice:
		return "语音", a.configBool(ctx, "membership.voice_message_requires_vip", false)
	}
	return "", false
}

func (a *App) ensureMessageMembership(ctx context.Context, senderID int64, messageType int) error {
	kind, gated := a.messageMembershipGate(ctx, messageType)
	if !gated || a.hasActiveMembership(ctx, senderID) {
		return nil
	}
	return &messageMembershipGateError{Kind: kind}
}

const defaultUnansweredMessageLimit = 3

// unansweredMessageLimitError is the opening-message cap: until the other side
// says something back, a free account may only send so many. It is the mirror of
// the daily active-chat limit — that one bounds how many people you may open a
// conversation with, this one bounds how hard you may knock on one door.
type unansweredMessageLimitError struct{ Limit int }

func (e *unansweredMessageLimitError) Error() string {
	return fmt.Sprintf("对方回复前最多发送 %d 条消息，开通会员后不受此限制", e.Limit)
}

func (a *App) resolveUnansweredMessageLimit(ctx context.Context, queryer rowQuerier) int {
	var raw string
	if err := queryer.QueryRowContext(ctx, `SELECT config_value FROM system_configs WHERE config_key='membership.free_unanswered_message_limit'`).Scan(&raw); err == nil {
		if parsed, parseErr := strconv.Atoi(strings.TrimSpace(raw)); parseErr == nil && parsed >= 0 {
			return parsed
		}
	}
	return defaultUnansweredMessageLimit
}

// reserveUnansweredMessage runs inside the send transaction, so two messages
// racing each other cannot both pass the last slot.
func (a *App) reserveUnansweredMessage(ctx context.Context, tx *sql.Tx, conversationID, senderID int64) error {
	limit := a.resolveUnansweredMessageLimit(ctx, tx)
	if limit <= 0 {
		return nil
	}
	var mine, theirs int
	// Recalled and removed messages still count as knocking: deleting your own
	// message must not hand back a slot.
	if err := tx.QueryRowContext(ctx, `SELECT
		COALESCE(SUM(sender_id=?),0),COALESCE(SUM(sender_id<>?),0)
		FROM im_messages WHERE conversation_id=?`, senderID, senderID, conversationID).Scan(&mine, &theirs); err != nil {
		return err
	}
	if theirs > 0 || mine < limit {
		return nil
	}
	if a.hasActiveMembership(ctx, senderID) {
		return nil
	}
	return &unansweredMessageLimitError{Limit: limit}
}