package com.kxs.transfer.api.config;

import cn.hutool.core.date.LocalDateTimeUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.aliyun.dts.subscribe.clients.metastore.AbstractUserMetaStore;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.kxs.transfer.api.model.KxsCheckpoint;
import com.kxs.transfer.api.service.KxsCheckpointService;

import java.util.Comparator;
import java.util.List;
import java.util.Optional;

/**
 * store the checkpoint data in the shared storage, such us database, shared file storage...
 * this meta store need to be completed by consumer
 * @author mac
 */
public class UserMetaStore extends AbstractUserMetaStore {

    private final KxsCheckpointService kxsCheckpointService;

    public UserMetaStore(KxsCheckpointService kxsCheckpointService) {

        this.kxsCheckpointService = kxsCheckpointService;
    }

    @Override
    protected void saveData(String groupId, String toStoreJson) {

        JSONObject jsonObject = JSON.parseObject(toStoreJson);

        JSONArray streamCheckpoint = jsonObject.getJSONArray("streamCheckpoint");
        String info = streamCheckpoint.get(0).toString();
        JSONObject infoObj = JSON.parseObject(info);

        KxsCheckpoint kxsCheckpoint = new KxsCheckpoint();
        kxsCheckpoint.setGroupId(groupId);
        kxsCheckpoint.setCheckpoint(toStoreJson);
        if(infoObj != null && infoObj.getLong("timestamp") > 0){
            kxsCheckpoint.setUpdateTime(LocalDateTimeUtil.of(infoObj.getLong("timestamp") * 1000));
        }
        kxsCheckpointService.save(kxsCheckpoint);
    }

    @Override
    protected String getData(String groupId) {

        List<KxsCheckpoint> checkpoints = kxsCheckpointService.list(Wrappers.lambdaQuery(KxsCheckpoint.class)
                .eq(KxsCheckpoint::getGroupId, groupId)
                .orderByDesc(KxsCheckpoint::getCreateTime)
                .last("limit 5"));
        Optional<KxsCheckpoint> checkpoint = checkpoints.stream().min(Comparator.comparing(KxsCheckpoint::getCreateTime));

        return checkpoint.map(KxsCheckpoint::getCheckpoint).orElse(null);
    }
}