package com.foxx.androidcast.sender;

/*********************************************************************
 * CastSendPump.java
 * Created at: Fri 15 May 2026 18:37:53 +0200
 * Updated at: Wed 20 May 2026 15:17:13 +0200 by Anton Afanasyeu <a.afanasieff@gmail.com>
 * Commit: 5d8e82d2e60a21fff3138d2a394ee4e8b4c6dcb8
 * class CastSendPump
 * Contributors:
 *    - Anton Afanasyeu <a.afanasieff@gmail.com> (7 commits, 195 lines)
 *    - Cursor Agent (project assistant)
 * Digest: SHA256	e6147251c99139c388f2f5a1820528544cffd3322eb24b3d1c5c170e71e86a37
**********************************************************************/
import android.util.Log;

import com.foxx.androidcast.network.CastProtocol;
import com.foxx.androidcast.network.CastSession;
import com.foxx.androidcast.network.control.PassThroughNetworkControlPlane;
import com.foxx.androidcast.stats.SessionStatsContext;

import java.io.IOException;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

/** Background send queue: audio has its own lane so video backlog cannot starve it. */
public final class CastSendPump {
    private static final String TAG = "CastSendPump";
    private static final int MAX_VIDEO_QUEUE = 40;
    private static final int MAX_AUDIO_QUEUE = 96;

    private final LinkedBlockingQueue<Outgoing> videoQueue = new LinkedBlockingQueue<>(MAX_VIDEO_QUEUE);
    private final LinkedBlockingQueue<Outgoing> audioQueue = new LinkedBlockingQueue<>(MAX_AUDIO_QUEUE);
    private final AtomicBoolean running = new AtomicBoolean(false);
    private Thread thread;
    private CastSession session;
    private Runnable onSendFailed;
    private PassThroughNetworkControlPlane statsPlane;
    private volatile int droppedPFrames;
    private volatile int droppedAudioFrames;

    public int getDroppedPFrames() {
        return droppedPFrames;
    }

    public int getDroppedAudioFrames() {
        return droppedAudioFrames;
    }

    public int getQueueDepth() {
        return videoQueue.size() + audioQueue.size();
    }

    /** Drops stale video frames (e.g. after capture resize / rotation). */
    public void flushVideoQueue() {
        videoQueue.clear();
    }

    public void start(CastSession session, Runnable onSendFailed) {
        start(session, onSendFailed, null);
    }

    public void start(CastSession session, Runnable onSendFailed, PassThroughNetworkControlPlane statsPlane) {
        this.session = session;
        this.onSendFailed = onSendFailed;
        this.statsPlane = statsPlane;
        if (!running.compareAndSet(false, true)) {
            return;
        }
        thread = new Thread(this::run, "CastSendPump");
        thread.setPriority(Thread.NORM_PRIORITY + 1);
        thread.start();
    }

    public void stop() {
        running.set(false);
        videoQueue.clear();
        audioQueue.clear();
        if (thread != null) {
            thread.interrupt();
            try {
                thread.join(500);
            } catch (InterruptedException ignored) {
                Thread.currentThread().interrupt();
            }
            thread = null;
        }
        session = null;
    }

    public void enqueueVideo(long ptsUs, boolean keyFrame, byte[] annexBReadyPayload) {
        if (!running.get()) {
            return;
        }
        Outgoing item = Outgoing.video(ptsUs, keyFrame, annexBReadyPayload);
        if (!videoQueue.offer(item)) {
            if (!keyFrame) {
                dropOneVideoPFrame();
            }
            if (!videoQueue.offer(item) && !keyFrame) {
                droppedPFrames++;
            }
        }
    }

    public void enqueue(byte type, byte[] payload) {
        if (!running.get()) {
            return;
        }
        Outgoing item = Outgoing.raw(type, payload);
        boolean audio = type == CastProtocol.MSG_AUDIO_FRAME || type == CastProtocol.MSG_AUDIO_CONFIG;
        if (audio) {
            if (!audioQueue.offer(item)) {
                Outgoing dropped = audioQueue.poll();
                if (dropped != null) {
                    droppedAudioFrames++;
                }
                audioQueue.offer(item);
            }
            return;
        }
        while (running.get() && !videoQueue.offer(item)) {
            dropOneVideoPFrame();
        }
    }

    private void dropOneVideoPFrame() {
        for (Outgoing o : videoQueue) {
            if (o.type == CastProtocol.MSG_VIDEO_FRAME && !o.keyFrame) {
                videoQueue.remove(o);
                droppedPFrames++;
                return;
            }
        }
        Outgoing head = videoQueue.poll();
        if (head != null && head.type == CastProtocol.MSG_VIDEO_FRAME && !head.keyFrame) {
            droppedPFrames++;
        }
    }

    private void run() {
        while (running.get()) {
            try {
                Outgoing item = audioQueue.poll();
                if (item == null) {
                    item = videoQueue.poll(50, TimeUnit.MILLISECONDS);
                }
                if (item == null || session == null) {
                    continue;
                }
                byte[] payload = item.buildPayload();
                session.send(item.type, payload);
                com.foxx.androidcast.dev.StreamDumpWriter.recordFrame(item.type, payload);
                int payloadSize = payload != null ? payload.length : 0;
                SessionStatsContext.recordOutboundMessage(item.type, item.keyFrame, payloadSize);
                if (statsPlane != null) {
                    statsPlane.recordSendBytes(payload.length + 5);
                }
            } catch (InterruptedException e) {
                if (!running.get()) {
                    break;
                }
            } catch (IOException e) {
                Log.w(TAG, "Send failed: " + e.getMessage());
                if (onSendFailed != null) {
                    onSendFailed.run();
                }
            }
        }
    }

    static final class Outgoing {
        final byte type;
        final byte[] payload;
        final long ptsUs;
        final boolean keyFrame;

        Outgoing(byte type, byte[] payload, long ptsUs, boolean keyFrame) {
            this.type = type;
            this.payload = payload;
            this.ptsUs = ptsUs;
            this.keyFrame = keyFrame;
        }

        public static Outgoing raw(byte type, byte[] payload) {
            return new Outgoing(type, payload, 0, false);
        }

        public static Outgoing video(long ptsUs, boolean keyFrame, byte[] payload) {
            return new Outgoing(CastProtocol.MSG_VIDEO_FRAME, payload, ptsUs, keyFrame);
        }

        public byte[] buildPayload() throws IOException {
            if (type != CastProtocol.MSG_VIDEO_FRAME) {
                return payload;
            }
            return CastProtocol.videoFramePayload(ptsUs, keyFrame, payload);
        }
    }
}
