package com.sonic.common.client.impl;

import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Maps;
import com.sonic.common.AppRuntime;
import com.sonic.common.client.JobmanClient;
import com.sonic.common.enums.AppEnv;
import lombok.AccessLevel;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.core.task.TaskExecutor;
import org.springframework.data.redis.core.RedisTemplate;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.sql.Date;
import java.text.ParseException;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;

@Slf4j
public class JobmanClientImpl implements JobmanClient, DisposableBean {
    private static final int MAX_LOGS_SIZE = 1_000;

    @Setter(AccessLevel.PROTECTED)
    private String hostIp;
    @Setter(AccessLevel.PROTECTED)
    private TaskExecutor executor;
    @Setter(AccessLevel.PROTECTED)
    private AppRuntime appRuntime;
    @Setter(AccessLevel.PROTECTED)
    private RedisTemplate<String, String> redisTemplate;
    @Setter(AccessLevel.PROTECTED)
    private JobmanClient.LockPolicy globalLockPolicy;

    /**
     * jobName对应的status. 用于加锁保持run与destroy互斥
     */
    private final Map<String, JobStatus> statusMap = Maps.newConcurrentMap();

    @Override
    public void run(String jobName, LockPolicy lockPolicy, Function<JobContext, JobResult> runner) {
        Objects.requireNonNull(lockPolicy);

        //只在第一次没有的jobName对应的Status情况下才build新的Context
        statusMap.putIfAbsent(jobName, JobStatus.buildContext(jobName, lockPolicy));
        JobStatus status = statusMap.get(jobName);
        synchronized (status) {
            if (status.isDestroyed() || !lockPolicy.tryLock()) {
                return;
            }

            JobContext jobContext = JobContext.builder()
                    .jobName(jobName)
                    .startTime(LocalDateTime.now())
                    .build();
            JobResult jobResult;
            try {
                jobResult = runner.apply(jobContext);
            } catch (Exception ex) {
                jobResult = JobResult.fail();
                jobContext.log("=== run job error ===, name = {}, appRuntime = {}, context = {}",
                        jobName, appRuntime.toJson().toJSONString(), JSONObject.toJSONString(jobContext), ex);
            }

            jobContext.setEndTime(LocalDateTime.now());
            JobResult finalJobResult = jobResult;
            executor.execute(() -> {
                Map<String, Object> params = ImmutableMap.<String, Object>builder()
                        .put("jobName", jobContext.getJobName())
                        .put("status", finalJobResult.getJobStatus())
                        .put("startTime", jobContext.getStartTime())
                        .put("endTime", jobContext.getEndTime())
                        .put("ip", Optional.ofNullable(hostIp).orElse(StringUtils.EMPTY))
                        .put("loadCount", finalJobResult.getLoadCount())
                        .put("logs", jobContext.getLogs().subList(0, Math.min(MAX_LOGS_SIZE, jobContext.getLogs().size())))
                        .put("ext", Optional.ofNullable(finalJobResult.getExt()).orElse(ImmutableMap.of()))
                        .build();
                log.info(JSONObject.toJSONString(params));
            });
        }
    }

    @Override
    public void run(String jobName, Long lockDurationSeconds, Function<JobContext, JobResult> runner) {
        LockPolicy lockPolicy = Optional.ofNullable(globalLockPolicy)
                .orElse(RedisLockPolicy.builder()
                        .appRuntime(appRuntime)
                        .redisTemplate(redisTemplate)
                        .lockDurationSeconds(lockDurationSeconds)
                        .jobName(jobName)
                        .build());
        run(jobName, lockPolicy, runner);
    }

    @Override
    public boolean isCronSatisfied(String cornExpression, LocalDateTime dateTime) {
        CronExpression expression;
        try {
            expression = new CronExpression(cornExpression);
        } catch (ParseException e) {
            throw new RuntimeException(String.format("invalid corn expression, expression = %s", cornExpression), e);
        }
        return expression.isSatisfiedBy(Date.from(dateTime.atZone(ZoneId.systemDefault()).toInstant()));
    }

    @Override
    public void destroy() {
        //由于destroy方法有jobStatus的锁. 因此并行调用. 该方法只会由容器调用一次. 不需要考虑性能
        statusMap.forEach((key, value) -> value.destroy());
    }

    public static class RedisLockPolicy implements LockPolicy {
        private final String jobName;
        private final long lockDurationSeconds;
        private final AppRuntime appRuntime;
        private final RedisTemplate<String, String> redisTemplate;

        @lombok.Builder
        public RedisLockPolicy(String jobName, long lockDurationSeconds, AppRuntime appRuntime, RedisTemplate<String, String> redisTemplate) {
            this.jobName = jobName;
            this.lockDurationSeconds = lockDurationSeconds;
            this.appRuntime = appRuntime;
            this.redisTemplate = redisTemplate;
        }

        @Override
        public boolean tryLock() {
            String lockKey = buildKey();
            redisTemplate.opsForValue().setIfAbsent(lockKey, appRuntime.getInstanceId(), lockDurationSeconds, TimeUnit.SECONDS);

            String locker = redisTemplate.opsForValue().get(lockKey);
            return appRuntime.getInstanceId().equals(locker);
        }

        @Override
        public void destroy() {
            String redisKey = buildKey();
            //如果销毁时是本节点锁定, 则删除锁
            String locker = redisTemplate.opsForValue().get(redisKey);
            if (appRuntime.getInstanceId().equals(locker)) {
                redisTemplate.delete(redisKey);
            }
        }

        private String buildKey() {
            return appRuntime.buildPrefixKey("jobman", jobName, "_locker_");
        }
    }

    @AllArgsConstructor
    private static class JobStatus {
        private final String jobName;
        private final LockPolicy lockPolicy;
        private boolean isAlive;

        public void destroy() {
            synchronized (this) {
                isAlive = false;
                log.info("destroy jobman, destroy by container, jobName = {}, lockPolicy = {}", jobName, lockPolicy.getClass().getName());
                lockPolicy.destroy();
            }
        }

        public boolean isDestroyed() {
            return !isAlive;
        }

        public static JobStatus buildContext(String jobName, LockPolicy lockPolicy) {
            return new JobStatus(jobName, lockPolicy, true);
        }
    }

    @Data
    public static class Builder {
        private TaskExecutor taskExecutor;
        private AppRuntime appRuntime;
        private RedisTemplate<String, String> redisTemplate;
        private LockPolicy lockPolicy;

        public Builder taskExecutor(TaskExecutor taskExecutor) {
            this.taskExecutor = taskExecutor;
            return this;
        }

        public Builder appRuntime(AppRuntime appRuntime) {
            this.appRuntime = appRuntime;
            return this;
        }

        public Builder redisTemplate(RedisTemplate<String, String> redisTemplate) {
            this.redisTemplate = redisTemplate;
            return this;
        }

        public Builder lockPolicy(LockPolicy lockPolicy) {
            this.lockPolicy = lockPolicy;
            return this;
        }

        protected JobmanClient buildService() {
            return new JobmanClientImpl();
        }

        public JobmanClient build() {
            //XXX local环境避免调试等待. 所以直接运行runner
            if (appRuntime.getEnv() == AppEnv.local) {
                return new JobmanClient() {
                    @Override
                    public void run(String jobName, LockPolicy lockPolicy, Function<JobContext, JobResult> runner) {
                        JobContext jobContext = JobContext.builder()
                                .jobName(jobName)
                                .startTime(LocalDateTime.now())
                                .build();
                        JobResult result = runner.apply(jobContext);
                        jobContext.setEndTime(LocalDateTime.now());
                        log.info("job client execute, context = {}, result = {}", jobContext, result);
                    }

                    @Override
                    public void run(String jobName, Long lockDurationSeconds, Function<JobContext, JobResult> runner) {
                        run(jobName, () -> true, runner);
                    }

                    @Override
                    public boolean isCronSatisfied(String cornExpression, LocalDateTime dateTime) {
                        CronExpression expression;
                        try {
                            expression = new CronExpression(cornExpression);
                        } catch (ParseException e) {
                            throw new RuntimeException(String.format("invalid corn expression, expression = %s", cornExpression), e);
                        }
                        return expression.isSatisfiedBy(Date.from(dateTime.atZone(ZoneId.systemDefault()).toInstant()));
                    }
                };
            }

            Objects.requireNonNull(redisTemplate);
            Objects.requireNonNull(taskExecutor);

            JobmanClientImpl client = (JobmanClientImpl) buildService();
            client.setExecutor(taskExecutor);
            client.setRedisTemplate(redisTemplate);

            client.setAppRuntime(appRuntime);
            client.setGlobalLockPolicy(lockPolicy);
            client.setHostIp(getHostIp());

            return client;
        }

        private String getHostIp() {
            String hostIP = "unknown";
            try {
                hostIP = InetAddress.getLocalHost().getHostAddress();
            } catch (UnknownHostException e) {
                log.error("failed to get host ip");
            }
            return hostIP;
        }
    }
}
