package com.android.commands.monkey.ape.jev;

import java.io.ByteArrayOutputStream;
import java.io.Closeable;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.HttpURLConnection;
import java.net.SocketTimeoutException;
import java.net.URL;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Semaphore;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import org.json.*;
import java.util.ArrayList;
import java.util.List;

/** A bounded asynchronous transport. Only the caller can install a validated result. */
public final class JevClient implements Closeable {
    private static final String STARTLUX_HOST = "211.71.15.47";
    public static final class Completion {
        public final JevProtocol.Request request;
        public final JevProtocol.Result result;
        public final String error;
        public final long sentNanos, completedNanos;
        public final JSONArray attempts;
        private Completion(JevProtocol.Request r, JevProtocol.Result result, String error, long sent, JSONArray attempts) {
            this.attempts=attempts;
            request = r; this.result = result; this.error = error; sentNanos = sent; completedNanos = System.nanoTime();
        }
        public long latencyNanos() { return completedNanos - sentNanos; }
    }
    public static final class Sent {
        public final JevProtocol.Request request;
        public final long sentNanos;
        private Sent(JevProtocol.Request r, long sent) { request = r; sentNanos = sent; }
    }
    private final URL endpoint;
    private final String token;
    private final int timeoutMs;
    private final ArrayBlockingQueue<JevProtocol.Request> waiting;
    private final ArrayBlockingQueue<Completion> completed;
    private final ArrayBlockingQueue<Sent> sent;
    private final Semaphore admission;
    private final Timer deadlines = new Timer("ape-jev-deadline", true);
    private final Thread[] workers;
    private volatile boolean closed;
    private final ConcurrentHashMap<Job,Boolean> active=new ConcurrentHashMap<Job,Boolean>();

    public JevClient(String endpoint, String token, int timeoutMs, int queueCapacity) throws IOException {
        this(endpoint,token,timeoutMs,queueCapacity,1);
    }
    public JevClient(String endpoint, String token, int timeoutMs, int queueCapacity, int workerCount) throws IOException {
        if (timeoutMs <= 0 || queueCapacity < 1 || workerCount<1 || workerCount>16 || queueCapacity>64) throw new IllegalArgumentException("invalid Jev client bounds");
        this.endpoint = new URL(endpoint);
        if (!("https".equals(this.endpoint.getProtocol()) || ("http".equals(this.endpoint.getProtocol())
                && ("127.0.0.1".equals(this.endpoint.getHost()) || "localhost".equals(this.endpoint.getHost())
                    || STARTLUX_HOST.equals(this.endpoint.getHost())))))
            throw new IllegalArgumentException("Jev endpoint requires HTTPS (HTTP allowed only for loopback or StartLux server 47)");
        this.token = token == null ? "" : token; this.timeoutMs = timeoutMs;
        waiting = new ArrayBlockingQueue<JevProtocol.Request>(queueCapacity);
        completed = new ArrayBlockingQueue<Completion>(queueCapacity + workerCount);
        sent = new ArrayBlockingQueue<Sent>(queueCapacity + workerCount);
        admission = new Semaphore(queueCapacity + workerCount);
        workers=new Thread[workerCount];
        for(int i=0;i<workerCount;i++) {
            workers[i] = new Thread(new Runnable() { public void run() { work(); } }, "ape-jev-http-"+i);
            workers[i].setDaemon(true);workers[i].start();
        }
    }
    /** Never waits for queue capacity or network activity. False must be logged as rejection. */
    public boolean canSubmit() {
        return !closed && admission.availablePermits() > 0 && waiting.remainingCapacity() > 0;
    }
    public synchronized boolean submit(JevProtocol.Request request) {
        if (request == null) throw new IllegalArgumentException("missing request");
        if (closed || !admission.tryAcquire()) return false;
        if (closed || !waiting.offer(request)) { admission.release(); return false; }
        return true;
    }
    /** Called at an APE main-thread safe point. */
    public Completion poll() {
        Completion result = completed.poll();
        if (result != null) admission.release();
        return result;
    }
    public Sent pollSent() { return sent.poll(); }

    private void work() {
        while (!closed) {
            try {
                JevProtocol.Request request = waiting.take();
                Job job = new Job(request);
                synchronized(this) {
                    if (closed) {job.finish(null,"CLOSED");break;}
                    active.put(job,Boolean.TRUE);
                }
                sent.offer(new Sent(request, job.started));
                try {job.run();}finally{active.remove(job);}
            } catch (InterruptedException e) {
                if (closed) return;
            }
        }
    }
    private final class Job {
        final JevProtocol.Request request;
        final long started = System.nanoTime();
        final AtomicBoolean terminal = new AtomicBoolean();
        final List<JSONObject> attempts=new ArrayList<JSONObject>();
        volatile HttpURLConnection connection;
        volatile String phase="queued";
        volatile int attempt;
        Job(JevProtocol.Request request) { this.request = request; }
        synchronized void finish(JevProtocol.Result result, String error) {
            if (terminal.compareAndSet(false, true)) completed.offer(new Completion(request, result, error, started, new JSONArray(attempts)));
        }
        synchronized void cancel(String error) {
            if(!terminal.get())attempts.add(diagnostic(attempt,phase,0,error,"deadline","logical request deadline exceeded",System.nanoTime()-started));
            finish(null,error);
        }
        void run() {
            TimerTask deadline = new TimerTask() { public void run() { Job.this.cancel("TIMEOUT"); } };
            try {
                deadlines.schedule(deadline, 2L*timeoutMs+5000);
                for(attempt=1;attempt<=2 && !terminal.get() && !closed;attempt++) {
                    long begin=System.nanoTime();String error=null,message="",exception="";int code=0;
                    JevProtocol.Result result=null;boolean retry=false;
                    try {
                        phase="connect";
                        connection=(HttpURLConnection)endpoint.openConnection();
                        connection.setConnectTimeout(timeoutMs);connection.setReadTimeout(timeoutMs);
                        connection.setInstanceFollowRedirects(false);connection.setRequestMethod("POST");
                        connection.setRequestProperty("Content-Type","application/json; charset=utf-8");
                        connection.setRequestProperty("Accept","application/json");
                        connection.setRequestProperty("X-StartLux-Request-Class","probe");
                        // The frozen service uses HTTP/1.0; do not reuse a proxy's idle socket.
                        connection.setRequestProperty("Connection","close");
                        if(!token.isEmpty())connection.setRequestProperty("Authorization","Bearer "+token);
                        connection.setDoOutput(true);
                        byte[] body=request.bodyJson.getBytes(JevProtocol.UTF8);
                        connection.setFixedLengthStreamingMode(body.length);
                        OutputStream out=connection.getOutputStream();phase="write";
                        try {out.write(body);}finally{out.close();}
                        phase="http";code=connection.getResponseCode();
                        if(code!=200) {
                            error="HTTP_"+code;retry=code==429 || code>=500;
                            InputStream in=connection.getErrorStream();
                            if(in!=null)try {
                                String raw=read(in,8192,begin);
                                JSONObject obj=new JSONObject(raw);Object detail=obj.opt("error");
                                message=detail instanceof JSONObject?((JSONObject)detail).optString("message",""):
                                    detail instanceof String?(String)detail:"";
                            }catch(JSONException ignored){}finally{in.close();}
                        } else {
                            phase="read";InputStream in=connection.getInputStream();String raw;
                            try {raw=read(in,Math.max(16384,request.requestBytes),begin);}finally{in.close();}
                            phase="parse";result=JevProtocol.fromProvider(raw,request);
                        }
                    }catch(SocketTimeoutException e){
                        exception=e.getClass().getSimpleName();message=detail(e);
                        if(code==0 || code==200){error="TIMEOUT";retry=true;}
                    }catch(JSONException e){error="INVALID_RESPONSE";exception=e.getClass().getSimpleName();message=detail(e);
                    }catch(IOException e){
                        exception=e.getClass().getSimpleName();message=detail(e);
                        if(code==0 || code==200){error="TRANSPORT_FAILURE";retry=!message.contains("response byte limit exceeded");}
                    }catch(RuntimeException e){error="WORKER_FAILURE";exception=e.getClass().getSimpleName();message=detail(e);
                    }finally{if(connection!=null)connection.disconnect();}
                    synchronized(this) {
                        attempts.add(diagnostic(attempt,phase,code,error,exception,message,System.nanoTime()-begin));
                    }
                    if(error==null || !retry || attempt==2){finish(result,error);return;}
                    if(!closed && !terminal.get())Thread.sleep(5000);
                }
            }catch(InterruptedException e){finish(null,"CLOSED");
            }catch(IllegalStateException e){finish(null,closed?"CLOSED":"WORKER_FAILURE");
            }finally{deadline.cancel();deadlines.purge();}
        }
        String read(InputStream in,int limit,long begin)throws IOException {
            ByteArrayOutputStream bytes=new ByteArrayOutputStream();byte[] buffer=new byte[4096];
            while(!terminal.get() && !closed) {
                long remaining=timeoutMs-(System.nanoTime()-begin)/1000000;
                if(remaining<=0)throw new SocketTimeoutException("attempt deadline exceeded");
                connection.setReadTimeout((int)remaining);
                int n=in.read(buffer);if(n<0)break;
                if(bytes.size()+n>limit)throw new IOException("response byte limit exceeded");
                bytes.write(buffer,0,n);
            }
            return new String(bytes.toByteArray(),JevProtocol.UTF8);
        }
    }
    private static String detail(Throwable e) {
        StringBuilder text=new StringBuilder();
        for(int i=0;e!=null && i<4;i++,e=e.getCause())text.append(e.getClass().getSimpleName()).append(": ").append(e.getMessage()).append("; ");
        return text.toString();
    }
    private JSONObject diagnostic(int attempt,String phase,int code,String error,String exception,String message,long elapsed) {
        if(message==null)message="";
        if(!token.isEmpty())message=message.replace(token,"[REDACTED]");
        message=message.replaceAll("(?i)(bearer\\s+|(?:token|api[_-]?key|authorization)[=:]\\s*)[^\\s,;]+","$1[REDACTED]");
        if(message.length()>512)message=message.substring(0,512);
        try {return new JSONObject().put("attempt",attempt).put("phase",phase).put("http_status",code)
            .put("error",error==null?JSONObject.NULL:error).put("exception",exception).put("message",message).put("elapsed_ns",elapsed);
        }catch(JSONException e){throw new IllegalStateException(e);}
    }

    public synchronized void close() {
        closed = true;for(Thread worker:workers)worker.interrupt();
        deadlines.cancel();
        JevProtocol.Request request;
        while ((request = waiting.poll()) != null) completed.offer(new Completion(request, null, "CLOSED", System.nanoTime(),new JSONArray()));
        // Closing from the APE thread must not wait on platform disconnect locks.
        // Worker I/O has a bounded timeout and is daemonized.
        for(Job job:active.keySet())job.finish(null, "CLOSED");
    }
}