package study.configbus;

import com.sun.net.httpserver.HttpServer;
import org.eclipse.jgit.api.Git;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.boot.web.server.context.WebServerApplicationContext;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.kafka.test.EmbeddedKafkaKraftBroker;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.URI;
import java.net.http.*;
import java.nio.file.*;
import java.time.Duration;
import java.util.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

/** Each context has its own Environment, RefreshScope, binder and random HTTP port. */
final class ConfigFixture implements AutoCloseable {
    // Config 5.0.6 rejects symlink path components; macOS /var resolves to /private/var.
    final Path directory = Files.createTempDirectory("u34-config-").toRealPath();
    final Path repository = directory.resolve("repository");
    final String password = UUID.randomUUID().toString();
    final String authorization = "Basic " + Base64.getEncoder().encodeToString(("lab:" + password).getBytes());
    final HttpClient http = HttpClient.newBuilder().connectTimeout(Duration.ofSeconds(2)).build();
    final List<ConfigurableApplicationContext> contexts = new ArrayList<>();
    final List<FaultGate> gates = new ArrayList<>();
    Git git;
    EmbeddedKafkaKraftBroker broker;
    ConfigurableApplicationContext server;
    ConfigurableApplicationContext clientA;
    ConfigurableApplicationContext clientB;
    FaultGate gateB;
    String topic = "u34-" + UUID.randomUUID();

    ConfigFixture() throws Exception {
        try {
            Files.createDirectories(repository);
            git = Git.init().setDirectory(repository.toFile()).setInitialBranch("main").call();
            Files.writeString(repository.resolve("application.properties"), "lab.message=shared\n");
            Files.writeString(repository.resolve("application-prod.properties"), "lab.message=shared-profile\n");
            Files.writeString(repository.resolve("catalog.properties"), "lab.message=app-base\n");
            publish("v1", "profile-v1", 4);
            server = start(ServerApp.class, "--spring.application.name=config-server",
                    "--spring.cloud.config.enabled=false", "--spring.cloud.bus.enabled=false",
                    "--spring.cloud.config.server.git.uri=" + repository.toUri(),
                    "--spring.cloud.config.server.git.basedir=" + directory.resolve("server-clone"),
                    "--spring.cloud.config.server.git.default-label=main",
                    "--spring.cloud.config.server.git.refresh-rate=0",
                    "--server.error.include-message=always",
                    "--encrypt.key=" + UUID.randomUUID());
            get(server, "/catalog/prod/main");
            broker = new EmbeddedKafkaKraftBroker(1, 1, topic);
            broker.brokerListProperty("u34.embedded.brokers");
            broker.brokerProperties(Map.of("auto.create.topics.enable", "false"));
            broker.afterPropertiesSet();
            clientA = client("a", base(server), null);
            gateB = new FaultGate(base(server));
            gates.add(gateB);
            clientB = client("b", gateB.base(), null);
        } catch (Throwable failure) {
            close();
            throw failure;
        }
    }

    ConfigurableApplicationContext client(String id, String configUrl, String override) {
        var args = new ArrayList<>(List.of("--spring.application.name=catalog", "--spring.profiles.active=prod",
                "--spring.cloud.config.enabled=true",
                "--spring.config.import=configserver:" + configUrl,
                "--spring.cloud.config.label=main", "--spring.cloud.config.username=lab",
                "--spring.cloud.config.password=" + password,
                "--spring.cloud.config.request-connect-timeout=1000", "--spring.cloud.config.request-read-timeout=1000",
                "--spring.cloud.bus.id=catalog:" + id,
                "--spring.cloud.bus.destination=" + topic,
                "--spring.cloud.stream.kafka.binder.brokers=" + broker.getBrokersAsString(),
                "--spring.cloud.stream.kafka.binder.replication-factor=1",
                "--spring.cloud.stream.kafka.binder.auto-create-topics=false",
                "--spring.cloud.stream.bindings.springCloudBusInput.consumer.max-attempts=1",
                "--management.endpoints.web.exposure.include=health,env,refresh,busrefresh",
                "--management.endpoint.env.show-values=never"));
        if (override != null) args.add("--lab.message=" + override);
        return start(ClientApp.class, args.toArray(String[]::new));
    }

    ConfigurableApplicationContext start(Class<?> app, String... args) {
        var all = new ArrayList<>(List.of("--server.address=127.0.0.1", "--server.port=0",
                "--spring.main.banner-mode=off", "--logging.level.root=ERROR",
                "--spring.security.user.name=lab", "--spring.security.user.password=" + password,
                "--spring.cloud.config.import-check.enabled=false",
                "--spring.cloud.config.server.health.enabled=false"));
        all.addAll(List.of(args));
        var context = new SpringApplicationBuilder(app).run(all.toArray(String[]::new));
        contexts.add(context);
        return context;
    }

    String publish(String revision, String message, int workers) throws Exception {
        Files.writeString(repository.resolve("catalog-prod.properties"),
                "lab.message=" + message + "\nlab.revision=" + revision + "\nlab.workers=" + workers + "\n");
        git.add().addFilepattern(".").call();
        return git.commit().setMessage("synthetic " + revision).setAuthor("U34", "lab@example.invalid")
                .setCommitter("U34", "lab@example.invalid").call().getName();
    }

    String commitFile(String file, String content) throws Exception {
        Files.writeString(repository.resolve(file), content);
        git.add().addFilepattern(file).call();
        return git.commit().setMessage("synthetic update").setAuthor("U34", "lab@example.invalid")
                .setCommitter("U34", "lab@example.invalid").call().getName();
    }

    static String base(ConfigurableApplicationContext context) {
        return "http://127.0.0.1:" + ((WebServerApplicationContext) context).getWebServer().getPort();
    }
    HttpResponse<String> request(String url, String method, String body, boolean authorized) throws Exception {
        var builder = HttpRequest.newBuilder(URI.create(url)).timeout(Duration.ofSeconds(10))
                .method(method, HttpRequest.BodyPublishers.ofString(body));
        if (authorized) builder.header("Authorization", authorization);
        if (!body.isEmpty()) builder.header("Content-Type", "text/plain");
        return http.send(builder.build(), HttpResponse.BodyHandlers.ofString());
    }
    String get(ConfigurableApplicationContext context, String path) throws Exception {
        var response = request(base(context) + path, "GET", "", true);
        if (response.statusCode() != 200) throw new AssertionError("GET " + path + " status=" + response.statusCode() + " body=" + response.body());
        return response.body();
    }
    int post(ConfigurableApplicationContext context, String path) throws Exception {
        return request(base(context) + path, "POST", "", true).statusCode();
    }
    void awaitState(ConfigurableApplicationContext context, String text) throws Exception {
        var deadline = System.nanoTime() + Duration.ofSeconds(25).toNanos();
        while (System.nanoTime() < deadline) {
            var response = request(base(context) + "/state", "GET", "", true);
            if (response.statusCode() == 200 && response.body().contains(text)) return;
            Thread.sleep(100);
        }
        throw new AssertionError("state did not reach " + text);
    }

    @Override public void close() throws Exception {
        for (int i = contexts.size() - 1; i >= 0; i--) contexts.get(i).close();
        for (var gate : gates) gate.close();
        if (broker != null) broker.destroy();
        http.close();
        if (git != null) git.close();
        try (var paths = Files.walk(directory)) {
            for (var path : paths.sorted(Comparator.reverseOrder()).toList()) Files.deleteIfExists(path);
        }
        System.clearProperty("u34.embedded.brokers");
    }

    /** Synthetic 503 injection for B's Config HTTP path; broker and A stay reachable. */
    final class FaultGate implements AutoCloseable {
        final AtomicBoolean blocked = new AtomicBoolean();
        final AtomicInteger failedRequests = new AtomicInteger();
        final HttpServer gate;
        FaultGate(String upstream) throws IOException {
            gate = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
            gate.createContext("/", exchange -> {
                byte[] body;
                int status;
                try {
                    if (blocked.get()) {
                        failedRequests.incrementAndGet();
                        status = 503; body = "synthetic config-path failure".getBytes();
                    }
                    else {
                        var response = request(upstream + exchange.getRequestURI(), "GET", "", true);
                        status = response.statusCode(); body = response.body().getBytes();
                        exchange.getResponseHeaders().set("Content-Type", "application/json");
                    }
                } catch (Exception failure) { status = 502; body = "synthetic forwarding failure".getBytes(); }
                exchange.sendResponseHeaders(status, body.length);
                try (var stream = exchange.getResponseBody()) { stream.write(body); }
            });
            gate.start();
        }
        String base() { return "http://127.0.0.1:" + gate.getAddress().getPort(); }
        @Override public void close() { gate.stop(0); }
    }
}