package job

import (
	"context"
	"errors"
	"fmt"
	"net/url"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"91porn-server/common/constant"
	"91porn-server/common/constant/redisconst"
	"91porn-server/common/hevcpull"
	"91porn-server/common/laosiji"
	"91porn-server/common/log"
	"91porn-server/models/v/hevctaskmod"
	"91porn-server/models/v/moduleconfmod"
	"91porn-server/models/v/modulesectionmod"
	"91porn-server/models/v/modulevidmod"
	"91porn-server/models/v/sysconfmod"
	"91porn-server/models/v/vidmod"
	"91porn-server/skd/skdg"

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

const (
	h265DefaultMaxRunning = 400 // 配置缺失或无效时，最多允许 400 个视频处于云端转码中
	h265CheckBatchSize    = 100 // 每轮最多轮询 100 个已提交的 H265 任务
	h265HomeCandidateSize = 400 // 首页每个模块、每种排序最多保障 400 个视频
	h265PendingTimeout    = 240 * time.Hour
	h265PullURLTTL        = 480 * time.Hour
)

var (
	h265SubmitRunning int32
	h265CheckRunning  int32
	h265ScanRunning   int32

	h265MissingCredentialsLogOnce sync.Once
	h265MissingAppURLLogOnce      sync.Once
	h265InvalidPullSecretLogOnce  sync.Once
)

// h265TranscodingJobEnabled 默认只允许生产环境执行真实云转码。
// 测试环境必须通过 hevc.enableTranscode 显式开启，避免误提交云端任务。
func h265TranscodingJobEnabled() bool {
	if skdg.Conf == nil {
		return false
	}
	log.Info("h265TranscodingJobEnabled ", log.Any("skdg.Conf.Base.Env", skdg.Conf.Base.Env))
	log.Info("h265TranscodingJobEnabled ", log.Any("skdg.Conf.Hevc.EnableTranscode", skdg.Conf.Hevc.EnableTranscode))
	if skdg.Conf.Hevc.EnableTranscode != nil && !*skdg.Conf.Hevc.EnableTranscode {
		return false
	}
	if skdg.Conf.Base.Env != constant.ProdEnv &&
		(skdg.Conf.Hevc.EnableTranscode == nil || !*skdg.Conf.Hevc.EnableTranscode) {
		return false
	}
	if !laosiji.Configured() {
		h265MissingCredentialsLogOnce.Do(func() {
			log.Warn("H265 transcoding jobs disabled: laosiji credentials are incomplete")
		})
		return false
	}
	if _, err := hevcpull.SignURL(
		"https://configuration-check.invalid/source.m3u8",
		skdg.Conf.Hevc.PullSecret,
		time.Now().Add(h265PullURLTTL),
	); err != nil {
		h265InvalidPullSecretLogOnce.Do(func() {
			log.Warn("H265 transcoding jobs disabled: hevc.pullSecret is missing or invalid", log.E(err))
		})
		return false
	}
	if buildH265AppTranscodeM3u8URL(
		skdg.Conf.Url.AppApiUrl,
		"health-check.m3u8",
		skdg.Conf.Hevc.PullSecret,
		time.Now().Add(h265PullURLTTL),
	) == "" {
		h265MissingAppURLLogOnce.Do(func() {
			log.Warn("H265 transcoding jobs disabled: url.appApiUrl is missing or invalid", log.Any("skdg.Conf.Url.AppApiUrl", skdg.Conf.Url.AppApiUrl))
		})
		return false
	}
	return true
}

// SubmitH265TranscodingQueue 把本地等待队列中的任务异步提交到云转码。
func SubmitH265TranscodingQueue() {
	if !h265TranscodingJobEnabled() {
		return
	}
	if !atomic.CompareAndSwapInt32(&h265SubmitRunning, 0, 1) {
		log.Info("SubmitH265TranscodingQueue skip: previous run still running")
		return
	}
	defer atomic.StoreInt32(&h265SubmitRunning, 0)

	ctx := context.Background()
	maxRunning := h265MaxRunning()
	pendingCount, err := vidmod.CountH265Pending()
	if err != nil {
		return
	}
	available := int64(maxRunning) - pendingCount
	if available <= 0 {
		return
	}

	domains, err := laosiji.SystemDomains(ctx)
	if err != nil {
		log.Warn("SubmitH265TranscodingQueue skip: laosiji domains failed", log.E(err))
		return
	}
	videos, err := vidmod.GetQueuedH265Videos(available)
	if err != nil {
		return
	}
	for _, video := range videos {
		if err = submitH265Video(ctx, domains, video, maxRunning); err != nil {
			log.Warn("SubmitH265TranscodingQueue submit failed", log.Any("videoId", video.ID), log.E(err))
			if errors.Is(err, laosiji.ErrTranscodeQueueFull) {
				return
			}
		}
	}
}

// CheckH265TranscodingResults 轮询云端任务并回写 h265Url。
func CheckH265TranscodingResults() {
	if !h265TranscodingJobEnabled() {
		return
	}
	if !atomic.CompareAndSwapInt32(&h265CheckRunning, 0, 1) {
		log.Info("CheckH265TranscodingResults skip: previous run still running")
		return
	}
	defer atomic.StoreInt32(&h265CheckRunning, 0)

	ctx := context.Background()
	domains, err := laosiji.SystemDomains(ctx)
	if err != nil {
		log.Warn("CheckH265TranscodingResults skip: laosiji domains failed", log.E(err))
		return
	}
	videos, err := vidmod.GetPendingH265Videos(h265CheckBatchSize)
	if err != nil || len(videos) == 0 {
		return
	}
	now := time.Now()
	for _, video := range videos {
		if video == nil {
			continue
		}
		if h265PendingTimedOut(video, now) {
			timeoutMsg := fmt.Sprintf("H265 transcode pending timeout after %.0fh", h265PendingTimeout.Hours())
			marked, markErr := vidmod.MarkH265Failed(video.ID, video.SourceURL, video.H265PendingAt)
			if markErr != nil {
				log.Warn("CheckH265TranscodingResults mark timeout failed", log.Any("videoId", video.ID), log.E(markErr))
			} else if marked {
				upsertH265TaskWithError(video, domains, "", hevctaskmod.H265TaskStatusFailed, timeoutMsg)
			} else {
				log.Info("CheckH265TranscodingResults ignored stale timeout", log.Any("videoId", video.ID))
			}
			continue
		}
		result, found, queryErr := queryH265Transcode(ctx, domains, video)
		if queryErr != nil {
			log.Warn("CheckH265TranscodingResults query failed", log.Any("videoId", video.ID), log.E(queryErr))
			// 保留原 processing 审计状态。查询返回和日志已经足够诊断，
			// 此处不落库可避免并发成功后被旧查询错误覆盖。
			continue
		}
		if !found {
			const notFoundMsg = "laosiji transcode task not found"
			log.Warn("CheckH265TranscodingResults task not found, retry as failed", log.Any("videoId", video.ID))
			marked, markErr := vidmod.MarkH265Failed(video.ID, video.SourceURL, video.H265PendingAt)
			if markErr != nil {
				log.Warn("CheckH265TranscodingResults mark not-found failed", log.Any("videoId", video.ID), log.E(markErr))
			} else if marked {
				// Failed 状态仍会进入等待队列，但会累计失败次数，避免云端永久查无任务时无限重提。
				upsertH265TaskWithError(video, domains, "", hevctaskmod.H265TaskStatusFailed, notFoundMsg)
			} else {
				log.Info("CheckH265TranscodingResults ignored stale not-found result", log.Any("videoId", video.ID))
			}
			continue
		}
		applyH265Result(video, result)
	}
}

func h265PendingTimedOut(video *vidmod.VideoModel, now time.Time) bool {
	if video == nil ||
		video.H265Status != vidmod.H265StatusPending ||
		video.H265PendingAt.IsZero() ||
		video.H265PendingAt.After(now) {
		return false
	}
	return now.Sub(video.H265PendingAt) >= h265PendingTimeout
}

// ScanHomeH265TranscodingQueue 将首页各启用视频模块的重点 SP 视频加入等待队列。
func ScanHomeH265TranscodingQueue() {
	if !h265TranscodingJobEnabled() {
		return
	}
	if !atomic.CompareAndSwapInt32(&h265ScanRunning, 0, 1) {
		log.Info("ScanHomeH265TranscodingQueue skip: previous run still running")
		return
	}
	defer atomic.StoreInt32(&h265ScanRunning, 0)

	moduleIDs, moduleIDHexes := h265HomeModuleIDs()
	if len(moduleIDs) == 0 {
		return
	}

	videoIDs := make([]primitive.ObjectID, 0)
	seen := make(map[primitive.ObjectID]struct{})
	for _, moduleIDHex := range moduleIDHexes {
		for _, sortType := range []int{1, 2, 3, 7, 9} {
			videos, err := vidmod.GetVideoListByCond(homeH265Filter(moduleIDHex, sortType), homeH265Options(sortType))
			if err != nil {
				log.Warn("ScanHomeH265TranscodingQueue video query failed",
					log.Any("moduleId", moduleIDHex), log.Any("sortType", sortType), log.E(err))
				continue
			}
			appendUniqueH265VideoIDs(&videoIDs, seen, videos)
		}
	}

	for _, moduleID := range moduleIDs {
		sections, err := modulesectionmod.GetAllBySubModuleID(moduleID)
		if err != nil {
			log.Warn("ScanHomeH265TranscodingQueue section query failed", log.Any("moduleId", moduleID), log.E(err))
			continue
		}
		for _, section := range sections {
			sectionVideos, err := modulevidmod.SectionVideosBySectionID(section.ID, options.Find().
				SetLimit(h265HomeCandidateSize).
				SetSort(bson.D{{Key: "sortCode", Value: -1}, {Key: "videoReviewedAt", Value: -1}}))
			if err != nil {
				log.Warn("ScanHomeH265TranscodingQueue section videos failed", log.Any("sectionId", section.ID), log.E(err))
				continue
			}
			for _, sectionVideo := range sectionVideos {
				if _, exists := seen[sectionVideo.VideoID]; exists {
					continue
				}
				seen[sectionVideo.VideoID] = struct{}{}
				videoIDs = append(videoIDs, sectionVideo.VideoID)
			}
		}
	}

	modified, err := vidmod.QueueH265Transcode(videoIDs)
	if err != nil {
		log.Warn("ScanHomeH265TranscodingQueue enqueue failed", log.E(err))
		return
	}
	if modified > 0 {
		log.Info("ScanHomeH265TranscodingQueue queued videos", log.Any("count", modified))
	}
}

func submitH265Video(ctx context.Context, domains laosiji.SystemDomainsResp, video *vidmod.VideoModel, maxRunning int) error {
	if video == nil || strings.TrimSpace(video.SourceURL) == "" {
		return nil
	}
	claimedAt, claimed, err := vidmod.ClaimH265Pending(video.ID)
	if err != nil || !claimed {
		return err
	}
	video.H265PendingAt = claimedAt
	result, err := submitH265Transcode(ctx, domains, video, maxRunning)
	if err != nil {
		if errors.Is(err, laosiji.ErrTranscodeQueueFull) {
			marked, markErr := vidmod.MarkH265Queued(video.ID, video.SourceURL, video.H265PendingAt)
			if markErr != nil {
				log.Warn("submitH265Video restore queue failed", log.Any("videoId", video.ID), log.E(markErr))
			} else if marked {
				upsertH265TaskWithError(video, domains, "", hevctaskmod.H265TaskStatusQueued, err.Error())
			}
			return err
		}
		marked, markErr := vidmod.MarkH265Failed(video.ID, video.SourceURL, video.H265PendingAt)
		if markErr != nil {
			log.Warn("submitH265Video mark failed", log.Any("videoId", video.ID), log.E(markErr))
		} else if marked {
			upsertH265TaskWithError(video, domains, "", hevctaskmod.H265TaskStatusFailed, err.Error())
		}
		return err
	}
	applyH265Result(video, result)
	return nil
}

func submitH265Transcode(ctx context.Context, domains laosiji.SystemDomainsResp, video *vidmod.VideoModel, maxRunning int) (laosiji.TranscodeResult, error) {
	fileURL := hevcAppTranscodeM3u8URL(video.SourceURL, video.H265PendingAt)
	fileID := h265CloudFileID(video)
	if fileURL == "" || fileID == "" {
		return laosiji.TranscodeResult{}, fmt.Errorf("empty signed app transcode m3u8 url, videoId:%s", video.ID.Hex())
	}
	return laosiji.SubmitH264ToH265Task(ctx, fileID, fileURL, domains, maxRunning)
}

func queryH265Transcode(ctx context.Context, domains laosiji.SystemDomainsResp, video *vidmod.VideoModel) (laosiji.TranscodeResult, bool, error) {
	fileURL := hevcAppTranscodeM3u8URL(video.SourceURL, video.H265PendingAt)
	fileID := h265CloudFileID(video)
	if fileURL == "" || fileID == "" {
		return laosiji.TranscodeResult{}, false, fmt.Errorf("empty signed app transcode m3u8 url, videoId:%s", video.ID.Hex())
	}
	return laosiji.QueryH264ToH265Task(ctx, fileID, fileURL, domains)
}

// hevcAppTranscodeM3u8URL 使用 claim 时间生成稳定、限时且绑定源路径的
// HMAC 拉流 URL。云端 file_id 独立计算，签名或域名变化不会影响轮询。
func hevcAppTranscodeM3u8URL(source string, claimedAt time.Time) string {
	if skdg.Conf == nil {
		return ""
	}
	if claimedAt.IsZero() {
		return ""
	}
	return buildH265AppTranscodeM3u8URL(
		skdg.Conf.Url.AppApiUrl,
		source,
		skdg.Conf.Hevc.PullSecret,
		claimedAt.UTC().Truncate(time.Second).Add(h265PullURLTTL),
	)
}

func buildH265AppTranscodeM3u8URL(baseURL, source, pullSecret string, expiresAt time.Time) string {
	baseURL = strings.TrimRight(strings.TrimSpace(baseURL), "/")
	normalizedSource, sourceErr := hevcpull.NormalizeSource(source)
	if baseURL == "" || sourceErr != nil {
		return ""
	}
	parsed, err := url.Parse(baseURL)
	if err != nil ||
		//parsed.Scheme != "https" ||
		parsed.Host == "" ||
		parsed.User != nil ||
		(parsed.Path != "" && parsed.Path != "/") ||
		parsed.RawPath != "" ||
		parsed.RawQuery != "" ||
		parsed.Fragment != "" {
		return ""
	}
	rawURL := baseURL + "/api/app/vid/transcode/m3u8/" + normalizedSource
	signedURL, err := hevcpull.SignURL(rawURL, pullSecret, expiresAt)
	if err != nil {
		return ""
	}
	return signedURL
}

// h265CloudFileID is independent from the temporary signed fetch URL. It
// changes when the source or retry attempt changes, but not when App domains,
// expiry timestamps, or signing keys rotate during an in-flight attempt.
func h265CloudFileID(video *vidmod.VideoModel) string {
	if video == nil || video.ID.IsZero() {
		return ""
	}
	source, err := hevcpull.NormalizeSource(video.SourceURL)
	if err != nil {
		return ""
	}
	identity := fmt.Sprintf(
		"91porn:h265:v1\x00%s\x00%s\x00%d",
		video.ID.Hex(),
		source,
		video.H265FailCount,
	)
	return laosiji.TranscodeFileID(identity)
}

func applyH265Result(video *vidmod.VideoModel, result laosiji.TranscodeResult) {
	h265URL := ""
	taskStatus := hevctaskmod.H265TaskStatusProcessing
	switch {
	case result.Done:
		h265URL = h265PlayableURL(video, result)
		if h265URL == "" {
			marked, markErr := vidmod.MarkH265Failed(video.ID, video.SourceURL, video.H265PendingAt)
			if markErr != nil {
				log.Warn("applyH265Result mark empty result failed", log.Any("videoId", video.ID), log.E(markErr))
			} else if marked {
				_ = hevctaskmod.UpsertWithError(video.ID, video.Title, result.FileURL, result.FileID, "", hevctaskmod.H265TaskStatusFailed, "empty H265 URL in completed task")
			}
			return
		}
		updated, err := vidmod.MarkPendingH265Success(video.ID, video.SourceURL, video.H265PendingAt, h265URL)
		if err != nil {
			log.Warn("applyH265Result MarkPendingH265Success failed", log.Any("videoId", video.ID), log.E(err))
			return
		}
		if !updated {
			// 当前视频可能已经被老司机回填 H.265，或 sourceURL 已经变化。
			// 对仍处于 pending 且缺少地址的新源重新排队；已有成功地址时该操作自动跳过。
			_, _ = vidmod.MarkH265Queued(video.ID, video.SourceURL, video.H265PendingAt)
			// 不更新 hevc_task：并发轮询的另一个实例可能已经成功落库，
			// 此处覆盖会把正确的 success 审计状态错误回退成 failed。
			log.Info("applyH265Result ignored stale result", log.Any("videoId", video.ID))
			return
		}
		clearH265VideoCache(video.ID)
		taskStatus = hevctaskmod.H265TaskStatusSuccess
	case result.Failed:
		marked, markErr := vidmod.MarkH265Failed(video.ID, video.SourceURL, video.H265PendingAt)
		if markErr != nil {
			log.Warn("applyH265Result MarkH265Failed failed", log.Any("videoId", video.ID), log.E(markErr))
			return
		}
		if !marked {
			log.Info("applyH265Result ignored stale failure", log.Any("videoId", video.ID))
			return
		}
		taskStatus = hevctaskmod.H265TaskStatusFailed
	default:
		// processing 结果不改变视频状态，也没有可用于跨集合原子写入的
		// ModifiedCount。保留现有审计记录，避免旧轮询覆盖并发成功结果。
		return
	}
	if err := hevctaskmod.UpsertWithError(video.ID, video.Title, result.FileURL, result.FileID, h265URL, taskStatus, result.ErrorMsg); err != nil {
		log.Warn("applyH265Result upsert task failed", log.Any("videoId", video.ID), log.E(err))
	}
}

// h265PlayableURL 区分云端“无需转码”和真正生成了新 H265 文件的结果。
// 无需转码时源地址本身已经可直接使用，不能把我方 App 拉流 URL 错存成老司机资源路径。
func h265PlayableURL(video *vidmod.VideoModel, result laosiji.TranscodeResult) string {
	if result.NoNeed {
		if video == nil {
			return ""
		}
		return strings.TrimSpace(video.SourceURL)
	}
	return laosiji.MovieM3u8SourcePath(result.HevcURL)
}

func clearH265VideoCache(videoID primitive.ObjectID) {
	if skdg.Redis == nil || videoID.IsZero() {
		return
	}
	key := fmt.Sprintf(redisconst.VideoInfoKey(), videoID.Hex())
	if _, err := skdg.Redis.Del(key); err != nil {
		log.Warn("clearH265VideoCache failed", log.Any("videoId", videoID), log.E(err))
	}
}

func upsertH265TaskWithError(video *vidmod.VideoModel, domains laosiji.SystemDomainsResp, h265URL string, status hevctaskmod.H265TaskStatus, errorMsg string) {
	if video == nil {
		return
	}
	submitURL, fileID := h265TaskSubmitInfo(video, domains)
	if err := hevctaskmod.UpsertWithError(video.ID, video.Title, submitURL, fileID, h265URL, status, errorMsg); err != nil {
		log.Warn("upsertH265TaskWithError failed", log.Any("videoId", video.ID), log.E(err))
	}
}

func h265TaskSubmitInfo(video *vidmod.VideoModel, _ laosiji.SystemDomainsResp) (string, string) {
	if video == nil {
		return "", ""
	}
	submitURL := strings.TrimSpace(hevcAppTranscodeM3u8URL(video.SourceURL, video.H265PendingAt))
	if submitURL == "" {
		submitURL = strings.TrimSpace(video.SourceURL)
	}
	fileID := h265CloudFileID(video)
	// 按运维要求持久化“可直接访问”的签名 URL，便于从 hevc_task 复制出来核验。
	// 代价：库中会保存有效期内可用的签名拉流地址（含 HMAC），务必收紧 DB 访问权限。
	return submitURL, fileID
}

func h265MaxRunning() int {
	cfg, err := sysconfmod.GetByVCode(sysconfmod.VCodeH265MaxRunning)
	if err != nil {
		return h265DefaultMaxRunning
	}
	maxRunning, err := strconv.Atoi(cfg.Value)
	if err != nil || maxRunning <= 0 {
		return h265DefaultMaxRunning
	}
	return maxRunning
}

func h265HomeModuleIDs() ([]primitive.ObjectID, []string) {
	modules, err := moduleconfmod.GetModuleConfByType(moduleconfmod.HomePage)
	if err != nil {
		log.Warn("h265HomeModuleIDs query homepage modules failed", log.E(err))
		return nil, nil
	}
	moduleIDs := make([]primitive.ObjectID, 0, len(modules))
	moduleIDHexes := make([]string, 0, len(modules))
	for _, module := range modules {
		if module.Status != 1 || module.ID.IsZero() {
			continue
		}
		moduleIDs = append(moduleIDs, module.ID)
		moduleIDHexes = append(moduleIDHexes, module.ID.Hex())
	}
	return moduleIDs, moduleIDHexes
}

func appendUniqueH265VideoIDs(ids *[]primitive.ObjectID, seen map[primitive.ObjectID]struct{}, videos []*vidmod.VideoModel) {
	for _, video := range videos {
		if video == nil {
			continue
		}
		if _, exists := seen[video.ID]; exists {
			continue
		}
		seen[video.ID] = struct{}{}
		*ids = append(*ids, video.ID)
	}
}

func homeH265Filter(moduleID string, sortType int) primitive.M {
	return bson.M{
		"status":    vidmod.CheckPass,
		"newsType":  vidmod.SP,
		"mId":       moduleID,
		"sourceURL": bson.M{"$exists": true, "$ne": ""},
		"deleteAt":  bson.M{"$exists": false},
	}
}

func homeH265Options(sortType int) *options.FindOptions {
	var sort bson.D
	switch sortType {
	case 2:
		sort = bson.D{
			{Key: "liaoBaTopSort", Value: -1},
			{Key: "likeCount", Value: -1},
			{Key: "reviewAt", Value: -1},
		}
	case 3:
		sort = bson.D{{Key: "playCount", Value: -1}, {Key: "reviewAt", Value: -1}}
	case 7:
		sort = bson.D{{Key: "collectCount", Value: -1}, {Key: "reviewAt", Value: -1}}
	case 9:
		sort = bson.D{{Key: "hot", Value: -1}, {Key: "reviewAt", Value: -1}}
	default:
		sort = bson.D{{Key: "reviewAt", Value: -1}}
	}
	return options.Find().SetLimit(h265HomeCandidateSize).SetSort(sort)
}