package tech.powerjob.server.persistence.storage.impl;

import com.google.common.base.Stopwatch;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.mongodb.ConnectionString;
import com.mongodb.client.MongoClient;
import com.mongodb.client.MongoClients;
import com.mongodb.client.MongoDatabase;
import com.mongodb.client.gridfs.GridFSBucket;
import com.mongodb.client.gridfs.GridFSBuckets;
import com.mongodb.client.gridfs.GridFSDownloadStream;
import com.mongodb.client.gridfs.GridFSFindIterable;
import com.mongodb.client.gridfs.model.GridFSFile;
import com.mongodb.client.model.Filters;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.DateUtils;
import org.bson.conversions.Bson;
import org.bson.types.ObjectId;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Conditional;
import org.springframework.core.env.Environment;
import tech.powerjob.server.extension.dfs.*;
import tech.powerjob.server.persistence.storage.AbstractDFsService;
import tech.powerjob.server.common.spring.condition.PropertyAndOneBeanCondition;

import javax.annotation.Priority;
import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.IOException;
import java.nio.file.Files;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Optional;

/**
 * 使用 MongoDB GridFS 作为底层存储
 * 配置用法：oms.storage.dfs.mongodb.uri=mongodb+srv://zqq:No1Bug2Please3!@cluster0.wie54.gcp.mongodb.net/powerjob_daily?retryWrites=true&w=majority
 *
 * @author tjq
 * @since 2023/7/28
 */
@Slf4j
@Priority(value = Integer.MAX_VALUE - 10)
@Conditional(GridFsService.GridFsCondition.class)
public class GridFsService extends AbstractDFsService {

    private MongoClient mongoClient;
    private MongoDatabase db;
    private final Map<String, GridFSBucket> bucketCache = Maps.newConcurrentMap();
    private static final String TYPE_MONGO = "mongodb";

    private static final String KEY_URI = "uri";

    private static final String SPRING_MONGO_DB_CONFIG_KEY = "spring.data.mongodb.uri";

    @Override
    public void store(StoreRequest storeRequest) throws IOException {
        GridFSBucket bucket = getBucket(storeRequest.getFileLocation().getBucket());
        try (BufferedInputStream bis = new BufferedInputStream(Files.newInputStream(storeRequest.getLocalFile().toPath()))) {
            bucket.uploadFromStream(storeRequest.getFileLocation().getName(), bis);
        }
    }

    @Override
    public void download(DownloadRequest downloadRequest) throws IOException {
        GridFSBucket bucket = getBucket(downloadRequest.getFileLocation().getBucket());
        FileUtils.forceMkdirParent(downloadRequest.getTarget());
        try (GridFSDownloadStream gis = bucket.openDownloadStream(downloadRequest.getFileLocation().getName());
             BufferedOutputStream bos = new BufferedOutputStream(Files.newOutputStream(downloadRequest.getTarget().toPath()))
        ) {
            byte[] buffer = new byte[1024];
            int bytes = 0;
            while ((bytes = gis.read(buffer)) != -1) {
                bos.write(buffer, 0, bytes);
            }
            bos.flush();
        }
    }

    @Override
    public Optional<FileMeta> fetchFileMeta(FileLocation fileLocation) throws IOException {
        GridFSBucket bucket = getBucket(fileLocation.getBucket());
        GridFSFindIterable files = bucket.find(Filters.eq("filename", fileLocation.getName()));
        GridFSFile first = files.first();
        if (first == null) {
            return Optional.empty();
        }
        return Optional.of(new FileMeta()
                .setLength(first.getLength())
                .setLastModifiedTime(first.getUploadDate())
                .setMetaInfo(first.getMetadata()));
    }

    @Override
    public void cleanExpiredFiles(String bucketName, int days) {
        Stopwatch sw = Stopwatch.createStarted();

        Date date = DateUtils.addDays(new Date(), -days);
        GridFSBucket bucket = getBucket(bucketName);
        Bson filter = Filters.lt("uploadDate", date);

        // 循环删除性能很差？我猜你肯定没看过官方实现[狗头]：org.springframework.data.mongodb.gridfs.GridFsTemplate.delete
        bucket.find(filter).forEach(gridFSFile -> {
            ObjectId objectId = gridFSFile.getObjectId();
            try {
                bucket.delete(objectId);
                log.info("[GridFsService] deleted {}#{}", bucketName, objectId);
            }catch (Exception e) {
                log.error("[GridFsService] deleted {}#{} failed.", bucketName, objectId, e);
            }
        });
        log.info("[GridFsService] clean bucket({}) successfully, delete all files before {}, using {}.", bucketName, date, sw.stop());
    }

    private GridFSBucket getBucket(String bucketName) {
        return bucketCache.computeIfAbsent(bucketName, ignore -> GridFSBuckets.create(db, bucketName));
    }

    private String parseMongoUri(Environment environment) {
        // 优先从新的规则读取
        String uri = fetchProperty(environment, TYPE_MONGO, KEY_URI);
        if (StringUtils.isNotEmpty(uri)) {
            return uri;
        }
        // 兼容 4.3.3 前的逻辑，读取 SpringMongoDB 配置
        return environment.getProperty(SPRING_MONGO_DB_CONFIG_KEY);
    }

    void initMongo(String uri) {
        log.info("[GridFsService] mongoDB uri: {}", uri);
        if (StringUtils.isEmpty(uri)) {
            log.warn("[GridFsService] uri is empty, GridFsService is off now!");
            return;
        }

        ConnectionString connectionString = new ConnectionString(uri);
        mongoClient = MongoClients.create(connectionString);

        if (StringUtils.isEmpty(connectionString.getDatabase())) {
            log.warn("[GridFsService] can't find database info from uri, will use [powerjob] as default, please make sure you have created the database 'powerjob'");
        }

        db = mongoClient.getDatabase(Optional.ofNullable(connectionString.getDatabase()).orElse("powerjob"));

        log.info("[GridFsService] initialize MongoDB and GridFS successfully, will use mongodb GridFs as storage layer.");
    }

    @Override
    public void destroy() throws Exception {
        mongoClient.close();
    }

    @Override
    protected void init(ApplicationContext applicationContext) {
        String uri = parseMongoUri(applicationContext.getEnvironment());
        initMongo(uri);

        log.info("[GridFsService] initialize successfully, THIS_WILL_BE_THE_STORAGE_LAYER.");
    }

    public static class GridFsCondition extends PropertyAndOneBeanCondition {
        @Override
        protected List<String> anyConfigKey() {
            return Lists.newArrayList("spring.data.mongodb.uri", "oms.storage.dfs.mongodb.uri");
        }

        @Override
        protected Class<?> beanType() {
            return DFsService.class;
        }
    }
}