package dev.elaine.distributed;

import java.time.Clock;
import java.time.Duration;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ThreadLocalRandom;
import java.util.function.Supplier;

public final class DistributedLab {
    private DistributedLab() {}

    public record RetryPolicy(int maxAttempts, Duration initialDelay, Duration maxDelay) {
        public RetryPolicy {
            if (maxAttempts < 1 || initialDelay.isNegative() || maxDelay.isNegative()) {
                throw new IllegalArgumentException("重试参数非法");
            }
        }

        public <T> T execute(Supplier<T> action, java.util.function.Predicate<T> accepted) {
            RuntimeException last = null;
            for (int attempt = 1; attempt <= maxAttempts; attempt++) {
                try {
                    T result = action.get();
                    if (accepted.test(result)) return result;
                    last = new IllegalStateException("结果不可接受");
                } catch (RuntimeException failure) {
                    last = failure;
                }
                // 教学模型不真正休眠，只暴露生产系统应使用的指数上限。
                delayFor(attempt);
            }
            throw last;
        }

        public Duration delayFor(int failedAttempt) {
            long factor = 1L << Math.min(failedAttempt - 1, 30);
            long capped = Math.min(maxDelay.toMillis(), initialDelay.toMillis() * factor);
            long jitter = capped == 0 ? 0 : ThreadLocalRandom.current().nextLong(capped + 1);
            return Duration.ofMillis(jitter);
        }
    }

    public static final class Snowflake {
        private static final long EPOCH = 1_700_000_000_000L;
        private final long nodeId;
        private final Clock clock;
        private long lastMillis = -1;
        private long sequence;

        public Snowflake(long nodeId, Clock clock) {
            if (nodeId < 0 || nodeId > 1023) throw new IllegalArgumentException("nodeId 必须在 0..1023");
            this.nodeId = nodeId;
            this.clock = clock;
        }

        public synchronized long nextId() {
            long now = clock.millis();
            if (now < lastMillis) throw new IllegalStateException("检测到时钟回拨");
            if (now == lastMillis) {
                sequence = (sequence + 1) & 4095;
                if (sequence == 0) throw new IllegalStateException("同一毫秒序号耗尽");
            } else {
                sequence = 0;
                lastMillis = now;
            }
            return ((now - EPOCH) << 22) | (nodeId << 12) | sequence;
        }
    }

    public record Quorum(int members) {
        public Quorum {
            if (members < 1) throw new IllegalArgumentException("成员数必须为正数");
        }
        public int majority() { return members / 2 + 1; }
        public boolean canCommit(int acknowledgements) { return acknowledgements >= majority(); }
    }

    public record Route(int database, int table) {}

    public record ShardRouter(int databaseCount, int tablesPerDatabase) {
        public ShardRouter {
            if (databaseCount < 1 || tablesPerDatabase < 1) throw new IllegalArgumentException("分片数必须为正数");
        }
        public Route route(long key) {
            long slot = Math.floorMod(key, (long) databaseCount * tablesPerDatabase);
            return new Route((int) (slot / tablesPerDatabase), (int) (slot % tablesPerDatabase));
        }
    }

    public record Event(String id, String aggregateId, String type, String payload) {}

    public static final class Outbox {
        private final Map<String, Event> pending = new LinkedHashMap<>();
        public void saveInSameTransaction(Event event) { pending.putIfAbsent(event.id(), event); }
        public List<Event> unpublished() { return List.copyOf(pending.values()); }
        public void markPublished(String id) { pending.remove(id); }
    }

    public static final class IdempotentConsumer {
        private final Set<String> processed = new HashSet<>();
        private final Map<String, Integer> balances = new HashMap<>();
        public boolean credit(String messageId, String accountId, int amount) {
            if (!processed.add(messageId)) return false;
            balances.merge(accountId, amount, Integer::sum);
            return true;
        }
        public int balance(String accountId) { return balances.getOrDefault(accountId, 0); }
    }

    public record SagaStep(String name, Runnable action, Runnable compensation) {}

    public static List<String> runSaga(List<SagaStep> steps) {
        List<SagaStep> completed = new ArrayList<>();
        List<String> trace = new ArrayList<>();
        try {
            for (SagaStep step : steps) {
                step.action().run();
                completed.add(step);
                trace.add("完成:" + step.name());
            }
        } catch (RuntimeException failure) {
            for (int i = completed.size() - 1; i >= 0; i--) {
                SagaStep step = completed.get(i);
                step.compensation().run();
                trace.add("补偿:" + step.name());
            }
        }
        return List.copyOf(trace);
    }

    public static int partition(String messageKey, int partitionCount) {
        if (partitionCount < 1) throw new IllegalArgumentException("分区数必须为正数");
        return Math.floorMod(messageKey.hashCode(), partitionCount);
    }
}