package dev.elaine.distributed;

import org.junit.jupiter.api.Test;

import java.time.Clock;
import java.time.Duration;
import java.time.Instant;
import java.time.ZoneOffset;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

import static dev.elaine.distributed.DistributedLab.*;
import static org.junit.jupiter.api.Assertions.*;

class DistributedMechanismsTest {
    @Test
    void retryStopsAfterSuccessAndUsesBoundedBackoff() {
        AtomicInteger calls = new AtomicInteger();
        RetryPolicy policy = new RetryPolicy(3, Duration.ofMillis(100), Duration.ofMillis(150));
        assertEquals("ok", policy.execute(() -> calls.incrementAndGet() == 3 ? "ok" : "fail", "ok"::equals));
        assertEquals(3, calls.get());
        assertTrue(policy.delayFor(3).toMillis() <= 150);
    }

    @Test
    void snowflakeIdsIncreaseAndQuorumRequiresMajority() {
        Clock clock = Clock.fixed(Instant.ofEpochMilli(1_800_000_000_000L), ZoneOffset.UTC);
        Snowflake generator = new Snowflake(7, clock);
        assertTrue(generator.nextId() < generator.nextId());
        assertFalse(new Quorum(5).canCommit(2));
        assertTrue(new Quorum(5).canCommit(3));
    }

    @Test
    void bindingTablesUseTheSameShardKey() {
        ShardRouter router = new ShardRouter(2, 4);
        assertEquals(router.route(42), router.route(42));
        assertNotEquals(router.route(42), router.route(43));
    }

    @Test
    void outboxAndIdempotentConsumerSurviveDuplicates() {
        Outbox outbox = new Outbox();
        Event event = new Event("evt-1", "order-9", "OrderPaid", "99");
        outbox.saveInSameTransaction(event);
        assertEquals(List.of(event), outbox.unpublished());

        IdempotentConsumer consumer = new IdempotentConsumer();
        assertTrue(consumer.credit(event.id(), "alice", 99));
        assertFalse(consumer.credit(event.id(), "alice", 99));
        assertEquals(99, consumer.balance("alice"));
    }

    @Test
    void sagaCompensatesCompletedStepsInReverseOrder() {
        List<String> state = new java.util.ArrayList<>();
        List<String> trace = runSaga(List.of(
                new SagaStep("库存", () -> state.add("库存已扣"), () -> state.add("库存已还")),
                new SagaStep("支付", () -> { throw new IllegalStateException("支付失败"); }, () -> state.add("退款"))
        ));
        assertEquals(List.of("完成:库存", "补偿:库存"), trace);
        assertEquals(List.of("库存已扣", "库存已还"), state);
    }

    @Test
    void equalMessageKeysStayInOnePartition() {
        assertEquals(partition("order-42", 8), partition("order-42", 8));
    }
}