package com.leanowtech.bloge.book.practices;

import com.leanowtech.bloge.core.context.GraphContext;
import com.leanowtech.bloge.core.dsl.GraphBuilder;
import com.leanowtech.bloge.core.engine.GraphEngine;
import com.leanowtech.bloge.core.exception.RetryableException;
import com.leanowtech.bloge.core.exception.OperatorTimeoutException;
import com.leanowtech.bloge.core.model.NodeStatus;
import com.leanowtech.bloge.core.operator.Idempotency;
import com.leanowtech.bloge.core.operator.Operator;
import com.leanowtech.bloge.core.operator.OperatorContext;
import com.leanowtech.bloge.core.spi.DefaultOperatorRegistry;
import java.time.Duration;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.junit.jupiter.api.Test;
import static org.junit.jupiter.api.Assertions.*;

/** Local provider doubles expose execution responsibility without transferring money. */
class FundingBoundaryTest {
    record FundingRequest(String businessKey, int amount) {}

    static final class Ledger {
        final Map<String, Integer> amounts = new ConcurrentHashMap<>();
        final AtomicInteger calls = new AtomicInteger();
        final AtomicInteger committedEffects = new AtomicInteger();

        String reserve(FundingRequest input, boolean loseFirstResponse) throws Exception {
            int attempt = calls.incrementAndGet();
            Integer previous = amounts.putIfAbsent(input.businessKey(), input.amount());
            if (previous == null) committedEffects.incrementAndGet();
            else if (previous != input.amount()) throw new IllegalArgumentException("key reused with a different amount");
            if (loseFirstResponse && attempt == 1) throw new RetryableException("committed; response lost");
            return "receipt:" + input.businessKey();
        }
    }

    @Test
    void explicitCreditQueryRunsBeforeFundingAndRetryKeepsOneBusinessEffect() {
        var ledger = new Ledger();
        AtomicInteger creditQueries = new AtomicInteger();
        Operator<String, Boolean> credit = (input, ctx) -> {
            creditQueries.incrementAndGet(); return true;
        };
        Operator<FundingRequest, String> funding = new Operator<>() {
            public String execute(FundingRequest input, OperatorContext ctx) throws Exception {
                assertEquals(1, creditQueries.get());
                return ledger.reserve(input, true);
            }
            public Idempotency idempotency() { return Idempotency.IDEMPOTENT; }
        };
        var builder = new GraphBuilder("visible-credit-and-funding");
        var graph = builder.node("credit", credit).input((results, context) -> "applicant-42")
            .node("fund", funding).dependsOn("credit")
            .input((results, context) -> new FundingRequest("LA-42:funding-1", 100000))
            .retry(1, Duration.ZERO).build();
        var registry = new DefaultOperatorRegistry();
        builder.operators().forEach(registry::registerRaw);
        var engine = GraphEngine.builder().registry(registry).build();
        try {
            var result = engine.execute(graph, new GraphContext()).requireSuccess();
            assertEquals("receipt:LA-42:funding-1", result.getOutput("fund", String.class));
            assertEquals(1, creditQueries.get(), "credit is not hidden inside the repeated funding call");
            assertEquals(2, ledger.calls.get(), "one retry adds one call to the initial call");
            assertEquals(1, ledger.committedEffects.get(), "provider-side key handling owns deduplication");
        } finally { engine.shutdownGracefully(Duration.ofSeconds(2)); }
    }

    @Test
    void timeoutLeavesACommittedEffectForExplicitReconciliation() throws Exception {
        var ledger = new Ledger();
        var committed = new CountDownLatch(1);
        var releaseResponse = new CountDownLatch(1);
        Operator<FundingRequest, String> funding = (input, ctx) -> {
            String receipt = ledger.reserve(input, false);
            committed.countDown();
            releaseResponse.await();
            return receipt;
        };
        var builder = new GraphBuilder("unknown-funding-response");
        var graph = builder.node("fund", funding).input((results, context) -> new FundingRequest("LA-42:funding-1", 100000))
            .timeout(Duration.ofMillis(100)).build();
        var registry = new DefaultOperatorRegistry();
        builder.operators().forEach(registry::registerRaw);
        var engine = GraphEngine.builder().registry(registry).build();
        try {
            var result = engine.execute(graph, new GraphContext());
            assertTrue(committed.await(1, TimeUnit.SECONDS));
            assertEquals(NodeStatus.FAILED, result.statusMap().get("fund"));
            assertInstanceOf(OperatorTimeoutException.class, result.errors().stream()
                .filter(error -> error.nodeId().equals("fund")).findFirst().orElseThrow().exception());
            assertEquals(1, ledger.committedEffects.get());
            assertEquals(100000, ledger.amounts.get("LA-42:funding-1"), "query by the original key before deciding to retry");
        } finally {
            releaseResponse.countDown();
            engine.shutdownGracefully(Duration.ofSeconds(2));
        }
    }

    @Test
    void businessKeysCannotSilentlyChangeTheOriginalAmount() throws Exception {
        var ledger = new Ledger();
        ledger.reserve(new FundingRequest("LA-42:funding-1", 100000), false);
        assertThrows(IllegalArgumentException.class,
            () -> ledger.reserve(new FundingRequest("LA-42:funding-1", 200000), false));
        assertEquals(1, ledger.committedEffects.get());
        assertEquals(100000, ledger.amounts.get("LA-42:funding-1"));
    }

    @Test
    void pairedExecutionRefactorExposesCreditAndRemovesRetryMultiplication() {
        // The same approved teaching applicant and stable request are used in both versions.
        var facts = Map.<String, Object>of("creditScore", 760, "fraudScore", 10,
            "blacklistHit", false, "incomeStatus", "strong");
        assertEquals("auto_approve", CodingPracticesTest.legacyDecision(facts));
        for (boolean refactored : new boolean[]{false, true}) {
            var ledger = new Ledger();
            var creditCalls = new AtomicInteger();
            Operator<String, Boolean> credit = (input, ctx) -> { creditCalls.incrementAndGet(); return true; };
            Operator<FundingRequest, String> fund = (input, ctx) -> {
                int sdkAttempts = refactored ? 1 : 3;
                for (int attempt = 0; attempt < sdkAttempts; attempt++) {
                    try {
                        ledger.reserve(input, false);
                        throw new RetryableException("committed; response unavailable");
                    } catch (RetryableException lostResponse) {
                        if (attempt == sdkAttempts - 1) throw lostResponse;
                    }
                }
                throw new AssertionError("unreachable");
            };
            var builder = new GraphBuilder(refactored ? "explicit-primary-retry" : "hidden-stacked-retry");
            var request = new FundingRequest("LA-PRACTICE:funding-1", 100000);
            var graph = refactored
                ? builder.node("credit", credit).input((results, ctx) -> "applicant-42")
                    .node("fund", fund).dependsOn("credit").input((results, ctx) -> request)
                    .retry(2, Duration.ZERO).build()
                : builder.node("fund", fund).input((results, ctx) -> {
                    creditCalls.incrementAndGet(); // Hidden external read in input evaluation.
                    return request;
                }).retry(2, Duration.ZERO).build();
            var registry = new DefaultOperatorRegistry();
            builder.operators().forEach(registry::registerRaw);
            var engine = GraphEngine.builder().registry(registry).build();
            try {
                var result = engine.execute(graph, new GraphContext());
                assertEquals(NodeStatus.FAILED, result.statusMap().get("fund"));
                assertEquals(refactored ? 3 : 9, ledger.calls.get());
                assertEquals(1, creditCalls.get());
                assertEquals(refactored, result.statusMap().containsKey("credit"));
                assertEquals(1, ledger.committedEffects.get());
                assertEquals(100000, ledger.amounts.get(request.businessKey()));
            } finally { engine.shutdownGracefully(Duration.ofSeconds(2)); }
        }
    }
}