package com.foxx.androidcast.sender;

/*********************************************************************
 * CastFanoutPump.java
 * Created at: Sat 16 May 2026 13:35:48 +0200
 * Updated at: Wed 20 May 2026 15:17:13 +0200 by Anton Afanasyeu <a.afanasieff@gmail.com>
 * Commit: 5d8e82d2e60a21fff3138d2a394ee4e8b4c6dcb8
 * class CastFanoutPump
 * Contributors:
 *    - Anton Afanasyeu <a.afanasieff@gmail.com> (4 commits, 193 lines)
 *    - Cursor Agent (project assistant)
 * Digest: SHA256	615e22057e4e8cc6a9c991c10a237fd3bd46e598429c7c1725114f3b8d457abd
**********************************************************************/
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.ArrayList;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * Fan-out send queue to multiple {@link CastSession}s (1:N cast).
 * Audio and video use separate lanes so video backlog cannot starve audio.
 */
public final class CastFanoutPump {
    private static final String TAG = "CastFanoutPump";
    private static final int MAX_VIDEO_QUEUE = 40;
    private static final int MAX_AUDIO_QUEUE = 96;

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

    public void setSessions(List<CastSession> list) {
        sessions.clear();
        if (list != null) {
            sessions.addAll(list);
        }
    }

    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(List<CastSession> sessionList, Runnable onSendFailed,
            PassThroughNetworkControlPlane statsPlane) {
        setSessions(sessionList);
        this.onSendFailed = onSendFailed;
        this.statsPlane = statsPlane;
        if (!running.compareAndSet(false, true)) {
            return;
        }
        thread = new Thread(this::run, "CastFanoutPump");
        thread.setPriority(Thread.NORM_PRIORITY + 1);
        thread.start();
        Log.i(TAG, "Started fan-out to " + sessions.size() + " peer(s)");
    }

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

    public void enqueueVideo(long ptsUs, boolean keyFrame, byte[] annexBReadyPayload) {
        if (!running.get()) {
            return;
        }
        CastSendPump.Outgoing item = CastSendPump.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;
        }
        CastSendPump.Outgoing item = CastSendPump.Outgoing.raw(type, payload);
        boolean audio = type == CastProtocol.MSG_AUDIO_FRAME || type == CastProtocol.MSG_AUDIO_CONFIG;
        if (audio) {
            if (!audioQueue.offer(item)) {
                CastSendPump.Outgoing dropped = audioQueue.poll();
                if (dropped != null) {
                    droppedAudioFrames++;
                }
                audioQueue.offer(item);
            }
            return;
        }
        while (running.get() && !videoQueue.offer(item)) {
            dropOneVideoPFrame();
        }
    }

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

    private void run() {
        while (running.get()) {
            try {
                CastSendPump.Outgoing item = audioQueue.poll();
                if (item == null) {
                    item = videoQueue.poll(50, TimeUnit.MILLISECONDS);
                }
                if (item == null || sessions.isEmpty()) {
                    continue;
                }
                byte[] payload = item.buildPayload();
                fanOut(item.type, payload, item.keyFrame);
            } catch (InterruptedException e) {
                if (!running.get()) {
                    break;
                }
            } catch (IOException e) {
                Log.w(TAG, "Payload build failed: " + e.getMessage());
                if (onSendFailed != null) {
                    onSendFailed.run();
                }
            }
        }
    }

    private void fanOut(byte type, byte[] payload, boolean keyFrame) {
        int payloadSize = payload != null ? payload.length : 0;
        SessionStatsContext.recordOutboundMessage(type, keyFrame, payloadSize);
        int peer = 0;
        for (CastSession session : sessions) {
            peer++;
            if (session == null) {
                continue;
            }
            try {
                session.send(type, payload);
                if (statsPlane != null) {
                    statsPlane.recordSendBytes(payload.length + 5);
                }
            } catch (IOException e) {
                Log.w(TAG, "Send to peer #" + peer + " failed: " + e.getMessage());
            }
        }
    }
}
