huangguo_server/app/service/vidser/free_watch_consume.go

package vidser

import (
	"errors"
	"fmt"
	"strconv"
	"time"

	"91porn-server/app/appg"
	"91porn-server/common/db"
	"91porn-server/common/stderr"
	"91porn-server/common/timeutil"
	"91porn-server/models/v/useractmod"
	"91porn-server/models/v/usermod"
	"91porn-server/models/v/vidmod"

	"go.mongodb.org/mongo-driver/bson/primitive"
	"go.mongodb.org/mongo-driver/mongo"
)

const freeWatchTransactionRetryLimit = 5

type freeWatchConsumeState struct {
	isVIP       bool
	isPublisher bool
	freeArea    bool
	paidVideo   bool
	viewedToday bool
	watchCount  uint64
	totalCount  uint64
}

type freeWatchConsumeDecision struct {
	isCan         bool
	shouldConsume bool
	watchCount    uint64
}

func decideFreeWatchConsume(state freeWatchConsumeState) freeWatchConsumeDecision {
	watchCount := state.watchCount
	if watchCount > state.totalCount {
		watchCount = state.totalCount
	}
	if state.isVIP || state.isPublisher || state.freeArea || state.paidVideo || state.viewedToday {
		return freeWatchConsumeDecision{isCan: true, watchCount: watchCount}
	}
	if watchCount == 0 {
		return freeWatchConsumeDecision{watchCount: 0}
	}
	return freeWatchConsumeDecision{
		isCan:         true,
		shouldConsume: true,
		watchCount:    watchCount,
	}
}

// ConsumeFreeWatch 按现有免费观看视频规则消费次数。
// 同一用户、同一视频在同一自然日最多消费一次。
func ConsumeFreeWatch(uid uint64, oid primitive.ObjectID) (vidmod.WatchConsumeResp, error) {
	at := time.Now()
	day := timeutil.BeginningOfDay(at)
	totalCount := TotalFreeWatchCount()

	user, err := usermod.FindUserByUIDForNoCache(uid)
	if err != nil {
		return vidmod.WatchConsumeResp{}, fmt.Errorf("find user: %w", err)
	}
	if user == nil {
		return vidmod.WatchConsumeResp{}, errors.New("find user: empty user")
	}
	video, err := vidmod.GetVideoInfo(oid.Hex())
	if err != nil {
		return vidmod.WatchConsumeResp{}, fmt.Errorf("find video: %w", err)
	}
	if video.ID.IsZero() || video.Status != vidmod.CheckPass {
		return vidmod.WatchConsumeResp{}, errors.New("video is missing or unavailable")
	}

	state := freeWatchConsumeState{
		isVIP:       !user.VipExpireDate.Before(at),
		isPublisher: video.PublisherID == uid,
		freeArea:    video.FreeArea,
		paidVideo:   video.Coins > 0,
		watchCount:  user.WatchCount,
		totalCount:  totalCount,
	}
	decision := decideFreeWatchConsume(state)
	if decision.isCan && !decision.shouldConsume {
		return watchConsumeResponse(decision, totalCount, false), nil
	}

	viewedToday, err := useractmod.IsViewTodayByNoVipTrans(nil, uid, oid, day)
	if err != nil {
		return vidmod.WatchConsumeResp{}, fmt.Errorf("find daily watch record: %w", err)
	}
	state.viewedToday = viewedToday
	decision = decideFreeWatchConsume(state)
	if !decision.shouldConsume {
		return watchConsumeResponse(decision, totalCount, false), nil
	}

	return consumeFreeWatchTransaction(uid, oid, video, day, at, totalCount)
}

func consumeFreeWatchTransaction(
	uid uint64,
	oid primitive.ObjectID,
	video vidmod.VideoModel,
	day time.Time,
	at time.Time,
	totalCount uint64,
) (vidmod.WatchConsumeResp, error) {
	if appg.VideoDB == nil {
		return vidmod.WatchConsumeResp{}, errors.New("video database is unavailable")
	}

	var resp vidmod.WatchConsumeResp
	transaction := func(t *db.MongoTool) error {
		// The transaction callback may be retried, so reset its result each time.
		resp = vidmod.WatchConsumeResp{TotalWatchCount: totalCount}

		user, err := usermod.FindUserByUIDTrans(t, uid)
		if err != nil {
			return fmt.Errorf("find user in transaction: %w", err)
		}
		if user == nil {
			return errors.New("find user in transaction: empty user")
		}
		viewedToday, err := useractmod.IsViewTodayByNoVipTrans(t, uid, oid, day)
		if err != nil {
			return fmt.Errorf("find daily watch record in transaction: %w", err)
		}
		decision := decideFreeWatchConsume(freeWatchConsumeState{
			isVIP:       !user.VipExpireDate.Before(at),
			isPublisher: video.PublisherID == uid,
			freeArea:    video.FreeArea,
			paidVideo:   video.Coins > 0,
			viewedToday: viewedToday,
			watchCount:  user.WatchCount,
			totalCount:  totalCount,
		})
		if !decision.shouldConsume {
			resp = watchConsumeResponse(decision, totalCount, false)
			return nil
		}

		if err := useractmod.UserActInsertTrans(t, useractmod.UserAct{
			VID:        oid,
			PlayWay:    useractmod.IsNoVip,
			UID:        uid,
			DailyDate:  day,
			CreatedAt:  at,
			ConsumeKey: freeWatchConsumeKey(uid, oid, day),
		}); err != nil {
			return err
		}

		remaining := decision.watchCount - 1
		updated, err := usermod.UpdateTrans(t, uid, usermod.UserSelector{WatchCount: &remaining})
		if err != nil || updated == nil {
			return fmt.Errorf("decrement watch count: %w", err)
		}
		resp = vidmod.WatchConsumeResp{
			IsCan:           true,
			WatchCount:      remaining,
			TotalWatchCount: totalCount,
			Consumed:        true,
		}
		return nil
	}

	err := runFreeWatchTransactionWithRetry(func() error {
		return appg.VideoDB.Trans(transaction)
	})
	if err == nil {
		if resp.Consumed {
			// UpdateTrans clears cache inside the transaction; clear once more after
			// commit so a concurrent cache refill cannot retain the old count.
			usermod.RefreshCache(uid)
		}
		return resp, nil
	}
	if !stderr.IsEqual(err, stderr.InsertExistError) {
		return vidmod.WatchConsumeResp{}, err
	}

	// A concurrent request inserted the same daily marker first.
	user, findErr := usermod.FindUserByUIDForNoCache(uid)
	if findErr != nil {
		return vidmod.WatchConsumeResp{}, fmt.Errorf("find user after duplicate consume: %w", findErr)
	}
	if user == nil {
		return vidmod.WatchConsumeResp{}, errors.New("find user after duplicate consume: empty user")
	}
	decision := freeWatchConsumeDecision{
		isCan:      true,
		watchCount: clampWatchCount(user.WatchCount, totalCount),
	}
	return watchConsumeResponse(decision, totalCount, false), nil
}

func runFreeWatchTransactionWithRetry(run func() error) error {
	var err error
	for attempt := 0; attempt < freeWatchTransactionRetryLimit; attempt++ {
		err = run()
		if err == nil ||
			stderr.IsEqual(err, stderr.InsertExistError) ||
			!isRetryableFreeWatchTransactionError(err) {
			return err
		}
		// Each run starts a fresh Mongo session. The shared transaction helper
		// cannot reliably retry wrapped transient errors in-place.
		time.Sleep(time.Duration(attempt+1) * 10 * time.Millisecond)
	}
	return err
}

func isRetryableFreeWatchTransactionError(err error) bool {
	if err == nil {
		return false
	}
	var serverErr mongo.ServerError
	if !errors.As(err, &serverErr) {
		return false
	}
	return serverErr.HasErrorLabel("TransientTransactionError") ||
		serverErr.HasErrorCode(112) || // WriteConflict
		serverErr.HasErrorCode(244) || // TransactionAborted
		serverErr.HasErrorCode(251) // NoSuchTransaction
}

func watchConsumeResponse(decision freeWatchConsumeDecision, totalCount uint64, consumed bool) vidmod.WatchConsumeResp {
	return vidmod.WatchConsumeResp{
		IsCan:           decision.isCan,
		WatchCount:      decision.watchCount,
		TotalWatchCount: totalCount,
		Consumed:        consumed,
	}
}

func clampWatchCount(watchCount, totalCount uint64) uint64 {
	if watchCount > totalCount {
		return totalCount
	}
	return watchCount
}

func freeWatchConsumeKey(uid uint64, oid primitive.ObjectID, day time.Time) string {
	return strconv.FormatUint(uid, 10) + ":" + oid.Hex() + ":" + strconv.FormatInt(day.Unix(), 10)
}