package com.njcn.gather.detection.util.socket.cilent;

import cn.hutool.core.util.StrUtil;
import com.njcn.gather.detection.util.socket.SocketManager;
import com.njcn.gather.detection.util.socket.communication.handler.ConnectionLifecycleHandler;
import com.njcn.gather.detection.util.socket.communication.handler.HeartbeatMessageStrategy;
import com.njcn.gather.detection.util.socket.communication.model.ConnectionContext;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import lombok.extern.slf4j.Slf4j;

import java.time.LocalDateTime;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

/**
 * Generic Netty heartbeat handler.
 * Stage 4-A moves heartbeat framing behind a strategy interface so the
 * retained transport layer can keep heartbeat capability without embedding
 * detection-specific packet structures.
 *
 * @author cdf
 * @author hongawen
 * @date 2026/04/07
 */
@Slf4j
public class HeartbeatHandler extends SimpleChannelInboundHandler<String> {

    private static final int MAX_HEARTBEAT_MISSES = 3;

    private final ScheduledExecutorService heartbeatExecutor = Executors.newScheduledThreadPool(1);

    private final ConnectionContext connectionContext;

    private final HeartbeatMessageStrategy heartbeatMessageStrategy;

    private final ConnectionLifecycleHandler lifecycleHandler;

    private ScheduledFuture<?> heartbeatFuture;

    private int consecutiveHeartbeatMisses;

    public HeartbeatHandler(ConnectionContext connectionContext, HeartbeatMessageStrategy heartbeatMessageStrategy,
                            ConnectionLifecycleHandler lifecycleHandler) {
        this.connectionContext = connectionContext;
        this.heartbeatMessageStrategy = heartbeatMessageStrategy;
        this.lifecycleHandler = lifecycleHandler == null ? ConnectionLifecycleHandler.NO_OP : lifecycleHandler;
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        scheduleHeartbeat(ctx);
        super.channelActive(ctx);
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        shutdownExecutorGracefully();
        super.channelInactive(ctx);
    }

    private void scheduleHeartbeat(ChannelHandlerContext ctx) {
        if (heartbeatMessageStrategy == null) {
            log.debug("Skip heartbeat scheduling because strategy is null: sessionKey={}", resolveSessionKey());
            return;
        }
        heartbeatFuture = heartbeatExecutor.scheduleAtFixedRate(() -> {
            if (!ctx.channel().isActive()) {
                return;
            }
            try {
                String heartbeatMessage = heartbeatMessageStrategy.buildHeartbeatMessage(connectionContext);
                if (StrUtil.isBlank(heartbeatMessage)) {
                    return;
                }
                // The client pipeline still uses line based frames, so the
                // generic heartbeat writer normalizes the trailing separator.
                if (!heartbeatMessage.endsWith("\n")) {
                    heartbeatMessage = heartbeatMessage + "\n";
                }
                ctx.channel().writeAndFlush(heartbeatMessage);
                consecutiveHeartbeatMisses++;
                log.debug("Send heartbeat packet: sessionKey={}, time={}, misses={}",
                        resolveSessionKey(), LocalDateTime.now(), consecutiveHeartbeatMisses);
                if (consecutiveHeartbeatMisses >= MAX_HEARTBEAT_MISSES) {
                    handleHeartbeatTimeout(ctx);
                }
            } catch (Exception e) {
                log.error("Send heartbeat packet failed: sessionKey={}", resolveSessionKey(), e);
            }
        }, 3, 10, TimeUnit.SECONDS);
    }

    private void handleHeartbeatTimeout(ChannelHandlerContext ctx) {
        log.warn("Heartbeat timeout reached: sessionKey={}, misses={}",
                resolveSessionKey(), consecutiveHeartbeatMisses);
        lifecycleHandler.onIdleTimeout(connectionContext);
        String sessionKey = resolveSessionKey();
        if (StrUtil.isNotBlank(sessionKey)) {
            SocketManager.removeUser(sessionKey);
        }
        consecutiveHeartbeatMisses = 0;
        if (ctx.channel().isActive()) {
            ctx.close();
        }
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
        if (heartbeatMessageStrategy != null && heartbeatMessageStrategy.isHeartbeatResponse(connectionContext, msg)) {
            consecutiveHeartbeatMisses = 0;
            log.debug("Receive heartbeat response: sessionKey={}, time={}", resolveSessionKey(), LocalDateTime.now());
            return;
        }
        ctx.fireChannelRead(msg);
    }

    private String resolveSessionKey() {
        return connectionContext == null ? null : connectionContext.getSessionKey();
    }

    private void shutdownExecutorGracefully() {
        try {
            if (heartbeatFuture != null && !heartbeatFuture.isCancelled()) {
                heartbeatFuture.cancel(false);
            }
            heartbeatExecutor.shutdown();
            if (!heartbeatExecutor.awaitTermination(5, TimeUnit.SECONDS)) {
                heartbeatExecutor.shutdownNow();
            }
        } catch (InterruptedException e) {
            heartbeatExecutor.shutdownNow();
            Thread.currentThread().interrupt();
        }
    }
}
