package com.yd.oss.api.service.impl;

import com.aliyun.oss.OSS;
import com.aliyun.oss.model.*;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.yd.common.exception.BusinessException;
import com.yd.common.result.Result;
import com.yd.oss.api.service.ApiChunkedUploadContextService;
import com.yd.oss.service.config.OssClientFactory;
import com.yd.oss.service.dao.ChunkedUploadContextMapper;
import com.yd.oss.service.model.ChunkedUploadContext;
import com.yd.oss.service.model.OssProvider;
import com.yd.oss.service.service.IOssProviderService;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.multipart.MultipartFile;
import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.TimeUnit;

@Slf4j
@Service
public class ApiChunkedUploadContextServiceImpl implements ApiChunkedUploadContextService {

    @Resource
    private ChunkedUploadContextMapper contextMapper;

    @Resource
    private ObjectMapper objectMapper;

    @Resource
    private OssClientFactory ossClientFactory;

    @Resource
    private IOssProviderService ossProviderService;

    @Resource
    private RedissonClient redissonClient;

    private static final String LOCK_KEY_PREFIX = "chunked:upload:lock:";

    /**
     * 上传单个分片（使用分布式锁保证上下文唯一性）
     * @param taskId         录制任务ID
     * @param chunkIndex     分片索引（从0开始）
     * @param chunk          分片文件
     * @param projectBizId   项目业务ID（为空时使用默认OSS）
     * @param source         来源标识（如 'recording'）
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void uploadChunk(String taskId, Integer chunkIndex, MultipartFile chunk,
                            String projectBizId, String source) {
        // ---------- 前置准备 ----------
        log.info("【uploadChunk】开始处理, taskId={}, chunkIndex={}, projectBizId={}, source={}, fileSize={}",
                taskId, chunkIndex, projectBizId, source, chunk != null ? chunk.getSize() : 0);

        // 获取 OSS 服务商（不依赖上下文，提前获取）
        OssProvider provider = ossProviderService.getProviderByProjectId(projectBizId);
        if (provider == null) {
            log.error("【uploadChunk】未找到OSS服务商, projectBizId={}", projectBizId);
            throw new BusinessException("未找到对应的OSS服务商配置");
        }
        log.info("【uploadChunk】获取到OSS服务商: name={}, bucket={}", provider.getName(), provider.getBucketName());

        // 创建 OSS 客户端（提前创建，供后续使用）
        OSS ossClient = ossClientFactory.createOssClient(provider);
        log.info("【uploadChunk】OSS客户端创建成功");

        // ---------- 分布式锁（保护整个方法，确保同一taskId操作串行） ----------
        String lockKey = LOCK_KEY_PREFIX + taskId;
        RLock lock = redissonClient.getLock(lockKey);
        boolean locked = false;

        try {
            // 尝试获取锁，等待5秒，锁持有时间30秒（足够处理上传+DB更新）
            locked = lock.tryLock(5, 30, TimeUnit.SECONDS);
            if (!locked) {
                log.error("【uploadChunk】获取分布式锁失败, taskId={}", taskId);
                throw new BusinessException("系统繁忙，请稍后重试");
            }
            log.debug("【uploadChunk】获取锁成功, taskId={}", taskId);

            // ---------- 1. 查询或创建上传上下文（原子操作） ----------
            ChunkedUploadContext context = contextMapper.selectOne(
                    new LambdaQueryWrapper<ChunkedUploadContext>()
                            .eq(ChunkedUploadContext::getTaskId, taskId)
                            .last(" limit 1 ")
            );

            if (context == null) {
                log.info("【uploadChunk】上下文不存在，将初始化新的分片上传, taskId={}", taskId);
                context = new ChunkedUploadContext();
                context.setTaskId(taskId);

                // 生成 OSS 对象 Key
                String objectKey = String.format("sharding/" + source + "/%tY/%tm/%s_%d.webm",
                        new Date(), new Date(), taskId, System.currentTimeMillis());
                context.setObjectKey(objectKey);
                log.info("【uploadChunk】生成 objectKey: {}", objectKey);

                // 调用 OSS 初始化分片上传
                InitiateMultipartUploadRequest initRequest =
                        new InitiateMultipartUploadRequest(provider.getBucketName(), objectKey);
                InitiateMultipartUploadResult initResult = ossClient.initiateMultipartUpload(initRequest);
                context.setUploadId(initResult.getUploadId());
                context.setPartEtagsJson("[]");
                context.setStatus(1); // 上传中

                // 插入数据库（此时其他线程被锁阻挡，不会重复插入）
                int insertResult = contextMapper.insert(context);
                log.info("【uploadChunk】初始化分片上传并入库, taskId={}, uploadId={}, objectKey={}, insertResult={}",
                        taskId, context.getUploadId(), objectKey, insertResult);
            } else {
                log.info("【uploadChunk】找到已存在的上下文, taskId={}, uploadId={}, status={}, partCount={}",
                        taskId, context.getUploadId(), context.getStatus(),
                        context.getPartEtagsJson() != null ? context.getPartEtagsJson().length() : 0);
            }

            // ---------- 2. 上传当前分片 ----------
            int partNumber = chunkIndex + 1;
            log.info("【uploadChunk】开始上传分片, taskId={}, partNumber={}, chunkSize={}",
                    taskId, partNumber, chunk.getSize());

            UploadPartRequest uploadPartRequest = new UploadPartRequest();
            uploadPartRequest.setBucketName(provider.getBucketName());
            uploadPartRequest.setKey(context.getObjectKey());
            uploadPartRequest.setUploadId(context.getUploadId());
            uploadPartRequest.setPartNumber(partNumber);
            uploadPartRequest.setInputStream(chunk.getInputStream());
            uploadPartRequest.setPartSize(chunk.getSize());
            UploadPartResult uploadResult = ossClient.uploadPart(uploadPartRequest);

            // ---------- 3. 更新 ETag 列表（在锁保护下，不会并发覆盖） ----------
            PartETag partETag = new PartETag(uploadResult.getPartNumber(), uploadResult.getETag());
            String json = context.getPartEtagsJson();
            List<PartETag> partETags = objectMapper.readValue(json, new TypeReference<List<PartETag>>() {});
            partETags.add(partETag);
            partETags.sort(Comparator.comparingInt(PartETag::getPartNumber));
            context.setPartEtagsJson(objectMapper.writeValueAsString(partETags));
            context.setUpdateTime(LocalDateTime.now());
            contextMapper.updateById(context);

            log.info("【uploadChunk】分片上传成功, taskId={}, partNumber={}, ETag={}, 当前总分片数={}",
                    taskId, partNumber, uploadResult.getETag(), partETags.size());

        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.error("【uploadChunk】获取锁被中断, taskId={}", taskId, e);
            throw new BusinessException("系统中断，请重试");
        } catch (Exception e) {
            log.error("【uploadChunk】处理失败, taskId={}, chunkIndex={}", taskId, chunkIndex, e);
            throw new BusinessException("分片上传失败: " + e.getMessage());
        } finally {
            // 释放锁（仅当当前线程持有锁时）
            if (locked && lock.isHeldByCurrentThread()) {
                lock.unlock();
                log.debug("【uploadChunk】释放分布式锁, taskId={}", taskId);
            }
        }
    }

    /**
     * 完成分片上传（合并）
     * @param taskId
     * @param projectBizId
     * @return
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public Result<Map<String, Object>> finishChunks(String taskId, String projectBizId) {
        // ... 原有逻辑不变（无需加锁，因为合并时上下文已存在且不再变化）
        // 但为了安全，也可以加锁确保合并时上下文不会被更新，但此时上传已完成，不需要锁。
        log.info("【finishChunks】开始完成分片上传, taskId={}, projectBizId={}", taskId, projectBizId);

        // 1. 查询上下文
        ChunkedUploadContext context = contextMapper.selectOne(
                new LambdaQueryWrapper<ChunkedUploadContext>()
                        .eq(ChunkedUploadContext::getTaskId, taskId)
        );
        if (context == null) {
            log.error("【finishChunks】未找到分片上传上下文, taskId={}", taskId);
            throw new BusinessException("未找到分片上传上下文，请检查 taskId 是否正确");
        }
        log.info("【finishChunks】找到上下文, taskId={}, uploadId={}, status={}, objectKey={}",
                taskId, context.getUploadId(), context.getStatus(), context.getObjectKey());

        // 2. 解析分片列表
        String json = context.getPartEtagsJson();
        List<PartETag> partETags;
        try {
            partETags = objectMapper.readValue(json, new TypeReference<List<PartETag>>() {});
        } catch (Exception e) {
            log.error("【finishChunks】解析分片 ETag 列表失败, taskId={}, json={}", taskId, json, e);
            throw new BusinessException("数据异常，无法合并");
        }
        if (partETags.isEmpty()) {
            log.error("【finishChunks】没有分片数据, taskId={}", taskId);
            throw new BusinessException("没有分片数据，无法合并");
        }
        log.info("【finishChunks】解析到 {} 个分片, taskId={}", partETags.size(), taskId);

        // 3. 获取 OSS 服务商
        OssProvider provider = ossProviderService.getProviderByProjectId(projectBizId);
        if (provider == null) {
            log.error("【finishChunks】未找到OSS服务商, projectBizId={}", projectBizId);
            throw new BusinessException("未找到对应的OSS服务商配置");
        }
        log.info("【finishChunks】获取到OSS服务商: name={}, bucket={}", provider.getName(), provider.getBucketName());

        OSS ossClient = ossClientFactory.createOssClient(provider);

        // 4. 执行合并
        CompleteMultipartUploadRequest completeRequest = new CompleteMultipartUploadRequest(
                provider.getBucketName(),
                context.getObjectKey(),
                context.getUploadId(),
                partETags
        );
        CompleteMultipartUploadResult completeResult = ossClient.completeMultipartUpload(completeRequest);
        log.info("【finishChunks】合并成功, taskId={}, location={}", taskId, completeResult.getLocation());

        // 5. 构建文件 URL 和大小
        String fileUrl = String.format("https://%s.%s/%s",
                provider.getBucketName(),
                provider.getEndpoint().replace("https://", ""),
                context.getObjectKey()
        );
        ObjectMetadata metadata = ossClient.getObjectMetadata(provider.getBucketName(), context.getObjectKey());
        long fileSize = metadata.getContentLength();

        // 6. 更新上下文状态
        context.setStatus(2); // 已完成
        contextMapper.updateById(context);
        log.info("【finishChunks】上下文状态更新为已完成, taskId={}", taskId);

        // 7. 返回结果
        Map<String, Object> result = new HashMap<>();
        result.put("taskId", taskId);
        result.put("fileUrl", fileUrl);
        result.put("fileKey", context.getObjectKey());
        result.put("fileSize", fileSize);

        log.info("【finishChunks】完成, taskId={}, fileUrl={}", taskId, fileUrl);
        return Result.success(result);
    }
}