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

import java.io.*;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
import org.json.*;

/** One bounded ordered evidence queue per runtime. Workers only receive copied facts. */
public final class JevOutput implements Closeable {
    interface Write {void run() throws Exception;}
    private final java.util.concurrent.locks.ReentrantLock budgetLock=new java.util.concurrent.locks.ReentrantLock();
    private final ThreadPoolExecutor worker;
    private final JevTiming timing;
    private volatile boolean healthy=true,closed;
    private int peak;
    private final int capacity;
    private final AtomicLong submitted=new AtomicLong(),accepted=new AtomicLong(),completed=new AtomicLong(),writeFailures=new AtomicLong(),rejected=new AtomicLong();
    private final Map<String,Long> rejectedKinds=new TreeMap<String,Long>();
    // A locator update can emit hundreds of nominations before the consumer runs.
    // Keep a bounded burst buffer; never silently discard any evidence class.
    public JevOutput(JevTiming timing){this(timing,1024);}
    JevOutput(JevTiming timing,int capacity) {
        this.timing=timing;this.capacity=capacity;
        worker=new ThreadPoolExecutor(1,1,0,TimeUnit.MILLISECONDS,new ArrayBlockingQueue<Runnable>(capacity),new ThreadFactory(){
            public Thread newThread(Runnable r){Thread t=new Thread(r,"ape-jev-evidence");t.setDaemon(true);return t;}
        });worker.prestartCoreThread();
    }
    public boolean isHealthy(){return healthy;}
    public void fail(Throwable e){boolean first=healthy;healthy=false;if(first)System.err.println("Jev evidence invalid: "+e.getClass().getSimpleName()+": "+e.getMessage());}
    boolean submit(final String segment,final Write write) {return submit(segment,segment,write);}
    boolean submit(final String segment,final String kind,final Write write) {JevDeadline.lock(budgetLock);try {
        submitted.incrementAndGet();
        if(closed){reject(kind,new IOException("evidence queue closed"));return false;}
        final long queued=System.nanoTime();
        Runnable task=new Runnable(){public void run(){
            timing.record("output_queue_wait",System.nanoTime()-queued);long start=System.nanoTime();
            try{write.run();completed.incrementAndGet();}catch(Throwable e){writeFailures.incrementAndGet();fail(e);}finally{timing.record(segment,System.nanoTime()-start);}
        }};
        try {
            long start=System.nanoTime();boolean accepted=(JevDeadline.absolute()!=Long.MAX_VALUE ? worker.getQueue().offer(task) : worker.getQueue().offer(task,100,TimeUnit.MILLISECONDS));
            timing.record("output_backpressure",System.nanoTime()-start);
            peak=Math.max(peak,worker.getQueue().size());
            if(!accepted)reject(kind,new IOException("evidence queue full after bounded backpressure: "+kind));
            else this.accepted.incrementAndGet();
            return accepted;
        } catch(InterruptedException e){Thread.currentThread().interrupt();reject(kind,e);return false;}
    } finally {budgetLock.unlock();}}
    private void reject(String kind,Throwable error) {rejected.incrementAndGet();Long n=rejectedKinds.get(kind);rejectedKinds.put(kind,n==null?1:n+1);fail(error);}
    boolean pendingWrites(){return !worker.getQueue().isEmpty();}
    public JSONObject accounting() {JevDeadline.lock(budgetLock);try {
        return JevRegistry.fields("capacity",capacity,"submitted",submitted.get(),"accepted",accepted.get(),"completed",completed.get(),
            "write_failures",writeFailures.get(),"rejected",rejected.get(),"rejected_by_kind",new JSONObject(rejectedKinds),"queue_peak",peak);
    } finally {budgetLock.unlock();}}
    /** Deep immutable JSON-compatible copy; serialization occurs only on the writer. */
    static Object freeze(Object value)throws JSONException {
        if(value instanceof JSONObject) {
            Map<String,Object> copy=new LinkedHashMap<String,Object>();Iterator<?> keys=((JSONObject)value).keys();
            while(keys.hasNext()){String key=(String)keys.next();copy.put(key,freeze(((JSONObject)value).get(key)));}
            return Collections.unmodifiableMap(copy);
        }
        if(value instanceof JSONArray){List<Object> copy=new ArrayList<Object>();for(int i=0;i<((JSONArray)value).length();i++)copy.add(freeze(((JSONArray)value).get(i)));return Collections.unmodifiableList(copy);}
        if(value==JSONObject.NULL || value==null || value instanceof String || value instanceof Number || value instanceof Boolean)return value;
        throw new JSONException("Unsupported evidence value");
    }
    static Object json(Object value)throws JSONException {
        if(value instanceof Map){JSONObject out=new JSONObject();for(Map.Entry<?,?> e:((Map<?,?>)value).entrySet())out.put((String)e.getKey(),json(e.getValue()));return out;}
        if(value instanceof List){JSONArray out=new JSONArray();for(Object item:(List<?>)value)out.put(json(item));return out;}
        return value;
    }
    public void save(final File path,JSONObject record,final boolean exclusive)throws IOException {
        final Object snapshot;
        try{snapshot=freeze(record);}catch(JSONException e){throw new IOException(e);}
        String name=path.getName();String kind=name.startsWith("region-context-execution-")?"receipt_context_execution":
            name.startsWith("region-context-gate-")?"receipt_context_gate":name.startsWith("transition-")?"receipt_transition":"receipt_other";
        if(!submit("receipt_io",kind,new Write(){public void run()throws Exception{
            if(exclusive && !path.createNewFile())throw new IOException("Receipt already exists: "+path.getName());
            Writer out=new BufferedWriter(new OutputStreamWriter(new FileOutputStream(path),"UTF-8"));
            try{out.write(json(snapshot).toString());out.write('\n');}finally{out.close();}
        }}))throw new IOException("Evidence queue rejected receipt");
    }
    public void drain() {
        final CountDownLatch done=new CountDownLatch(1);
        long deadline=System.nanoTime()+TimeUnit.SECONDS.toNanos(30);
        try{
            JevDeadline.lock(budgetLock);try{
                if(closed){fail(new IOException("evidence queue closed"));return;}
                if(!worker.getQueue().offer(new Runnable(){public void run(){done.countDown();}},30,TimeUnit.SECONDS)){
                    fail(new IOException("Evidence drain barrier timed out"));return;
                }
            } finally {budgetLock.unlock();}
            if(!done.await(Math.max(0,deadline-System.nanoTime()),TimeUnit.NANOSECONDS))fail(new IOException("Evidence drain timed out"));
        }
        catch(InterruptedException e){Thread.currentThread().interrupt();fail(e);}
    }
    public int peak(){return peak;}
    public void close() {
        JevDeadline.lock(budgetLock);try{if(closed)return;closed=true;worker.shutdown();} finally {budgetLock.unlock();}
        try{if(!worker.awaitTermination(30,TimeUnit.SECONDS)){fail(new IOException("Evidence close timed out"));worker.shutdownNow();}}
        catch(InterruptedException e){Thread.currentThread().interrupt();fail(e);worker.shutdownNow();}
    }
}