package study.gateway;

import com.sun.net.httpserver.HttpServer;
import java.net.InetSocketAddress;
import java.util.concurrent.Executors;
import java.util.concurrent.ExecutorService;

/** Real loopback HTTP peers with synthetic data; not a registry or business service. */
public class Upstreams implements AutoCloseable {
    final JwtMaterial keys;
    final java.util.Map<String,java.util.concurrent.atomic.AtomicInteger> attempts=new java.util.concurrent.ConcurrentHashMap<>();
    final java.util.Map<String,Integer> health=new java.util.concurrent.ConcurrentHashMap<>();
    final HttpServer a;
    final HttpServer b;
    final ExecutorService workers = Executors.newVirtualThreadPerTaskExecutor();
    public Upstreams(JwtMaterial keys) throws Exception { this.keys=keys; a = start("A"); b = start("B"); }
    private HttpServer start(String id) throws Exception {
        var server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
        server.setExecutor(workers);
        server.createContext("/", exchange -> {
            String subject="anonymous";
            if(exchange.getRequestURI().getPath().equals("/identity")) {
                try { subject=keys.downstreamSubject(exchange.getRequestHeaders().getFirst("Authorization")); }
                catch(Exception invalid) { exchange.sendResponseHeaders(401,-1); exchange.close(); return; }
            }
            if (exchange.getRequestURI().getPath().equals("/actuator/health")) {
                int status=health.getOrDefault(id,200);
                byte[] body="{\"status\":\"DOWN\"}".getBytes(java.nio.charset.StandardCharsets.UTF_8);
                exchange.getResponseHeaders().set("Content-Type","application/json");
                exchange.sendResponseHeaders(status,status==204?-1:body.length);
                try(var out=exchange.getResponseBody()) { if(status!=204)out.write(body); }
                return;
            }
            if (exchange.getRequestURI().getPath().equals("/slow-head")) {
                try { Thread.sleep(350); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
            }
            if (exchange.getRequestURI().getPath().equals("/slow-body")) {
                exchange.sendResponseHeaders(200,0);
                try (var out=exchange.getResponseBody()) {
                    out.write("start".getBytes()); out.flush();
                    try { Thread.sleep(350); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
                    out.write("-end".getBytes());
                }
                return;
            }
            int attempt=attempts.computeIfAbsent(id+exchange.getRequestURI().getPath(), k -> new java.util.concurrent.atomic.AtomicInteger()).incrementAndGet();
            String text = "instance="+id+"\npath="+exchange.getRequestURI().getRawPath()
                    +"\nquery="+exchange.getRequestURI().getRawQuery()
                    +"\nmethod="+exchange.getRequestMethod()
                    +"\nlabel="+exchange.getRequestHeaders().getFirst("X-Lab-Label")
                    +"\nsubject="+subject+"\nuser="+exchange.getRequestHeaders().getFirst("X-User")
                    +"\nverified="+exchange.getRequestHeaders().getFirst("X-Verified-User")
                    +"\nxff="+exchange.getRequestHeaders().getFirst("X-Forwarded-For")
                    +"\nforwarded="+exchange.getRequestHeaders().getFirst("Forwarded")
                    +"\ncookie="+exchange.getRequestHeaders().getFirst("Cookie")
                    +"\npre="+exchange.getRequestHeaders().getFirst("X-Lab-PreTrace");
            text+="\nattempt="+attempt;
            byte[] bytes = text.getBytes(java.nio.charset.StandardCharsets.UTF_8);
            exchange.sendResponseHeaders(exchange.getRequestURI().getPath().startsWith("/retry-fail")?503:200, bytes.length);
            try (var out = exchange.getResponseBody()) { out.write(bytes); }
        });
        server.start(); return server;
    }
    void health(String instance,int code) { health.put(instance,code); }
    java.util.List<org.springframework.cloud.client.ServiceInstance> instances() {
        var first=new org.springframework.cloud.client.DefaultServiceInstance("A","synthetic","127.0.0.1",a.getAddress().getPort(),false);
        var second=new org.springframework.cloud.client.DefaultServiceInstance("B","synthetic","127.0.0.1",b.getAddress().getPort(),false);
        first.getMetadata().put("zone","east"); first.getMetadata().put("hint","blue"); first.getMetadata().put("weight","1");
        second.getMetadata().put("zone","west"); second.getMetadata().put("hint","green"); second.getMetadata().put("weight","3");
        return java.util.List.of(first,second);
    }
    String uri() { return "http://127.0.0.1:"+a.getAddress().getPort(); }
    public void close() { a.stop(0); b.stop(0); workers.close(); System.out.println("U32 webfluxUpstreamsClosed=true count=2"); }
}