package study.iot.command;

import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.time.Clock;
import java.time.Instant;
import java.util.ArrayList;
import java.util.HexFormat;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;

/**
 * 命令服务端状态机与幂等表（教学实现，单进程内存）。
 *
 * <p>commandId 唯一。重复投递命中既有记录时：
 * <ul>
 *   <li>已过期 → 返回 {@link Outcome#DUPLICATE_EXPIRED}，不产生业务效果；</li>
 *   <li>已成功 → 返回 {@link Outcome#DUPLICATE_SUCCEEDED}，不重复执行；</li>
 *   <li>进行中 → 返回 {@link Outcome#DUPLICATE_IN_FLIGHT}。</li>
 * </ul>
 * 业务效果由 {@link EffectLog} 记录，一次 commandId 至多一条。</p>
 */
public final class CommandService {

    public enum Outcome { ACCEPTED, ACKED, EXPIRED, DUPLICATE_EXPIRED, DUPLICATE_SUCCEEDED, DUPLICATE_IN_FLIGHT }

    public record Recorded(Command command, Command.State state, String result, Instant updatedAt) {}

    private final Clock clock;
    private final Map<String, Recorded> table = new LinkedHashMap<>();
    private final EffectLog effects;

    public CommandService(Clock clock, EffectLog effects) {
        this.clock = clock;
        this.effects = effects;
    }

    public synchronized Outcome create(Command command) {
        Recorded existing = table.get(command.commandId());
        if (existing != null) {
            return duplicate(existing);
        }
        if (command.expiredAt(clock.instant())) {
            table.put(command.commandId(), new Recorded(command, Command.State.EXPIRED, "expired-before-publish", clock.instant()));
            return Outcome.EXPIRED;
        }
        table.put(command.commandId(), new Recorded(command, Command.State.CREATED, null, clock.instant()));
        return Outcome.ACCEPTED;
    }

    public synchronized void markPublished(String commandId) {
        transition(commandId, Command.State.CREATED, Command.State.PUBLISHED);
    }

    public synchronized void markDelivered(String commandId) {
        transition(commandId, Command.State.PUBLISHED, Command.State.DELIVERED);
    }

    /**
     * 处理设备回执。三种结局必须分开：
     * <ul>
     *   <li>未过期且正常：DELIVERED → ACKED → SUCCEEDED，写入一次业务效果；</li>
     *   <li>已过期：转 EXPIRED，不写效果，即使设备声称成功；</li>
     *   <li>重复回执：返回相应用例，不重复写效果。</li>
     * </ul>
     */
    public synchronized Outcome acknowledge(String commandId, boolean deviceSucceeded) {
        Recorded recorded = table.get(commandId);
        if (recorded == null) {
            throw new IllegalArgumentException("unknown command " + commandId);
        }
        if (recorded.state() == Command.State.EXPIRED || recorded.state() == Command.State.FAILED) {
            return Outcome.DUPLICATE_EXPIRED;
        }
        if (recorded.state() == Command.State.SUCCEEDED) {
            return Outcome.DUPLICATE_SUCCEEDED;
        }
        Instant now = clock.instant();
        if (recorded.command().expiredAt(now)) {
            table.put(commandId, new Recorded(recorded.command(), Command.State.EXPIRED,
                    "late-ack-after-expiry", now));
            return Outcome.EXPIRED;
        }
        if (!deviceSucceeded) {
            table.put(commandId, new Recorded(recorded.command(), Command.State.FAILED, "device-failed", now));
            return Outcome.ACKED;
        }
        table.put(commandId, new Recorded(recorded.command(), Command.State.ACKED, null, now));
        effects.append(commandId, recorded.command().payload());
        table.put(commandId, new Recorded(recorded.command(), Command.State.SUCCEEDED, "effect-once", now));
        return Outcome.ACKED;
    }

    public synchronized Optional<Recorded> find(String commandId) {
        return Optional.ofNullable(table.get(commandId));
    }

    /** 命令摘要的稳定哈希，用来比较两套编码（Java / Python）是否一致。 */
    /** 契约规定的稳定时间格式：UTC、毫秒精度、Z 结尾，Java 与 Python 两侧必须一致。 */
    public static String contractTime(Instant moment) {
        return CONTRACT_TIME.format(moment);
    }

    private static final java.time.format.DateTimeFormatter CONTRACT_TIME =
            java.time.format.DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'")
                    .withZone(java.time.ZoneOffset.UTC);

    public static String digest(Command command) {
        try {
            MessageDigest sha = MessageDigest.getInstance("SHA-256");
            String canonical = command.commandId() + "|" + command.deviceId() + "|" + command.type() + "|"
                    + command.payload() + "|" + contractTime(command.createdAt()) + "|" + contractTime(command.expireAt());
            return HexFormat.of().formatHex(sha.digest(canonical.getBytes(StandardCharsets.UTF_8)));
        } catch (Exception e) {
            throw new IllegalStateException(e);
        }
    }

    public synchronized List<Map<String, Object>> snapshot() {
        List<Map<String, Object>> rows = new ArrayList<>();
        for (Recorded recorded : table.values()) {
            rows.add(Map.of(
                    "commandId", recorded.command().commandId(),
                    "state", recorded.state().name(),
                    "result", recorded.result() == null ? "" : recorded.result()));
        }
        return rows;
    }

    private Outcome duplicate(Recorded existing) {
        return switch (existing.state()) {
            case EXPIRED, FAILED -> Outcome.DUPLICATE_EXPIRED;
            case SUCCEEDED -> Outcome.DUPLICATE_SUCCEEDED;
            default -> Outcome.DUPLICATE_IN_FLIGHT;
        };
    }

    private void transition(String commandId, Command.State from, Command.State to) {
        Recorded recorded = table.get(commandId);
        if (recorded == null || recorded.state() != from) {
            throw new IllegalStateException("cannot transition " + commandId + " from "
                    + (recorded == null ? "missing" : recorded.state()) + " to " + to);
        }
        table.put(commandId, new Recorded(recorded.command(), to, recorded.result(), clock.instant()));
    }
}