using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Net;
using System.Net.Http;
using System.Threading;
using System.Threading.Channels;
using System.Threading.Tasks;
using MediaBrowser.Controller;
using Microsoft.AspNetCore.Authorization;
using Microsoft.AspNetCore.Mvc;

namespace Jellyfin.Plugin.Enigma2;

/// <summary>Shares one receiver HTTP connection between probing and all consumers of a channel.</summary>
public sealed class LiveStreamRelay : IDisposable
{
    private readonly HttpClient client;
    private readonly int port;
    private readonly Dictionary<string, Source> sources = new();
    private readonly Dictionary<string, Source> tokens = new();
    private readonly object gate = new();
    private int opened;
    private readonly TimeSpan idleGrace;
    private Func<TimeSpan> startupBudget, readBudget;
    private TimeSpan startupTimeout => startupBudget();
    private TimeSpan readTimeout => readBudget();
    public LiveStreamRelay(IServerApplicationHost host) : this(new HttpClient(new SocketsHttpHandler { ConnectTimeout = TimeSpan.FromSeconds(3), AllowAutoRedirect = false }), host.HttpPort) { startupBudget = () => TimeSpan.FromSeconds(StartupBudget.Seconds); readBudget = () => TimeSpan.FromSeconds(StartupBudget.StallSeconds); }
    public LiveStreamRelay(HttpClient client, int port, TimeSpan? idleGrace = null, TimeSpan? startupTimeout = null, TimeSpan? readTimeout = null)
    {
        startupBudget = () => startupTimeout ?? TimeSpan.FromSeconds(20);
        readBudget = () => readTimeout ?? TimeSpan.FromSeconds(8);
        if (this.startupTimeout <= TimeSpan.Zero) throw new ArgumentOutOfRangeException(nameof(startupTimeout));
        if (this.readTimeout <= TimeSpan.Zero) throw new ArgumentOutOfRangeException(nameof(readTimeout));
        this.client = client; client.Timeout = Timeout.InfiniteTimeSpan; this.port = port; this.idleGrace = idleGrace ?? TimeSpan.FromSeconds(15);
    }
    public string Register(string url)
    {
        lock (gate)
        {
            if (!sources.TryGetValue(url, out var source))
            {
                source = new Source(this, url);
                sources.Add(url, source); tokens.Add(source.Id, source);
            }
            return $"http://127.0.0.1:{port}/DecoderIntegration/Live/{source.Id}";
        }
    }
    public Task<Subscription> Subscribe(string id, CancellationToken token)
    {
        Source source;
        lock (gate) if (!tokens.TryGetValue(id, out source)) throw new FileNotFoundException();
        return source.Subscribe(token);
    }
    public object Status()
    {
        lock (gate) return new { UpstreamConnectionsOpened = opened, ActiveSources = sources.Values.Count(s => s.Active), Consumers = sources.Values.Sum(s => s.Consumers), LastErrors = sources.Values.Select(s => s.LastError).Where(s => s != null).ToArray(), Channels = sources.Values.Select(s => new { s.Id, s.ServiceReference, s.Active, s.Consumers }).ToArray() };
    }
    public void Dispose() { lock (gate) foreach (var s in sources.Values) s.Stop(); client.Dispose(); }

    public sealed class Subscription : IAsyncDisposable
    {
        private readonly Source source;
        internal readonly Channel<byte[]> Buffer = Channel.CreateBounded<byte[]>(new BoundedChannelOptions(128) { FullMode = BoundedChannelFullMode.Wait, SingleReader = true });
        internal Subscription(Source source) { this.source = source; }
        public ChannelReader<byte[]> Reader => Buffer.Reader;
        public ValueTask DisposeAsync() { source.Remove(this); return ValueTask.CompletedTask; }
    }

    internal sealed class Source
    {
        private readonly LiveStreamRelay owner;
        private readonly string url;
        private readonly object gate = new();
        private readonly HashSet<Subscription> subscribers = new();
        private readonly Queue<byte[]> history = new();
        private int historyBytes;
        private CancellationTokenSource lifetime;
        private CancellationTokenSource idle;
        private Task pump;
        private TaskCompletionSource ready;
        private DateTime failedUntil;
        public string Id { get; } = Guid.NewGuid().ToString("N");
        public string ServiceReference => new Uri(url).AbsolutePath;
        public string LastError { get; private set; }
        public bool Active { get { lock (gate) return pump != null && !pump.IsCompleted; } }
        public int Consumers { get { lock (gate) return subscribers.Count; } }
        public Source(LiveStreamRelay owner, string url) { this.owner = owner; this.url = url; }
        public async Task<Subscription> Subscribe(CancellationToken token)
        {
            Subscription subscription; Task initial;
            lock (gate)
            {
                idle?.Cancel(); idle?.Dispose(); idle = null;
                if (subscribers.Count >= 16) throw new IOException("Channel consumer limit reached.");
                if (pump == null || pump.IsCompleted)
                {
                    if (DateTime.UtcNow < failedUntil) throw new IOException("Decoder source is unavailable.");
                    lifetime?.Dispose(); lifetime = new();
                    history.Clear(); historyBytes = 0;
                    ready = new(TaskCreationOptions.RunContinuationsAsynchronously);
                    pump = Task.Run(() => Pump(lifetime.Token));
                }
                subscription = new(this); subscribers.Add(subscription);
                foreach (var chunk in history) subscription.Buffer.Writer.TryWrite(chunk);
                initial = ready.Task;
            }
            try { await initial.WaitAsync(owner.startupTimeout + TimeSpan.FromSeconds(1), token); return subscription; }
            catch { await subscription.DisposeAsync(); throw; }
        }
        public void Remove(Subscription subscription)
        {
            lock (gate)
            {
                if (!subscribers.Remove(subscription)) return;
                subscription.Buffer.Writer.TryComplete();
                if (subscribers.Count == 0 && pump != null && !pump.IsCompleted)
                {
                    idle?.Cancel(); idle?.Dispose(); idle = new();
                    var timer = idle.Token;
                    var current = lifetime;
                    _ = Task.Run(async () => {
                        try { await Task.Delay(owner.idleGrace, timer); lock (gate) if (subscribers.Count == 0 && lifetime == current) current.Cancel(); }
                        catch (OperationCanceledException) { }
                    });
                }
            }
        }
        public void Stop() { lock (gate) { idle?.Cancel(); lifetime?.Cancel(); } }
        private async Task Pump(CancellationToken token)
        {
            Exception failure = null;
            try
            {
                // Headers and the first MPEG-TS chunk share one startup deadline.
                // A slow tuner may accept HTTP before it starts delivering packets.
                using var opening = CancellationTokenSource.CreateLinkedTokenSource(token); opening.CancelAfter(owner.startupTimeout);
                Interlocked.Increment(ref owner.opened);
                using var response = await owner.client.GetAsync(url, HttpCompletionOption.ResponseHeadersRead, opening.Token);
                response.EnsureSuccessStatusCode();
                using var stream = await response.Content.ReadAsStreamAsync(token);
                var buffer = new byte[188 * 256 + 188]; var carry = 0; var started = false;
                while (true)
                {
                    using var reading = CancellationTokenSource.CreateLinkedTokenSource(token);
                    if (started) reading.CancelAfter(owner.readTimeout);
                    var count = await stream.ReadAtLeastAsync(buffer.AsMemory(carry, buffer.Length - carry), 188 * 256 - carry, throwOnEndOfStream: false, cancellationToken: started ? reading.Token : opening.Token);
                    if (count == 0) throw new IOException("Decoder closed the live stream.");
                    count += carry; var bytes = count / 188 * 188;
                    if (bytes > 0)
                    {
                        if (buffer[0] != 0x47 || (bytes >= 376 && buffer[188] != 0x47)) throw new IOException("Decoder did not return MPEG-TS packets.");
                        started = true;
                        var chunk = buffer.AsSpan(0, bytes).ToArray();
                        lock (gate)
                        {
                            history.Enqueue(chunk); historyBytes += bytes;
                            while (historyBytes > 2 * 1024 * 1024) historyBytes -= history.Dequeue().Length;
                            foreach (var subscriber in subscribers.ToArray())
                                if (!subscriber.Buffer.Writer.TryWrite(chunk)) { subscriber.Buffer.Writer.TryComplete(new IOException("Slow live stream consumer.")); Remove(subscriber); }
                            ready.TrySetResult();
                        }
                    }
                    carry = count - bytes;
                    if (carry > 0) System.Buffer.BlockCopy(buffer, bytes, buffer, 0, carry);
                }
            }
            catch (Exception error) { failure = error; }
            finally
            {
                lock (gate)
                {
                    LastError = token.IsCancellationRequested ? null : failure?.Message;
                    if (!token.IsCancellationRequested) failedUntil = DateTime.UtcNow.AddSeconds(3);
                    ready.TrySetException(new IOException("Decoder stream unavailable.", failure));
                    foreach (var subscriber in subscribers) subscriber.Buffer.Writer.TryComplete(failure);
                    subscribers.Clear(); history.Clear(); historyBytes = 0;
                }
            }
        }
    }
}

[ApiController]
[Route("DecoderIntegration/Live")]
public sealed class LiveRelayController : ControllerBase
{
    private readonly LiveStreamRelay relay;
    public LiveRelayController(LiveStreamRelay relay) { this.relay = relay; }
    [HttpGet("{id}")]
    [AllowAnonymous]
    public async Task Stream(string id, CancellationToken token)
    {
        var remote = HttpContext.Connection.RemoteIpAddress;
        if (remote == null || !(IPAddress.IsLoopback(remote) || (remote.IsIPv4MappedToIPv6 && IPAddress.IsLoopback(remote.MapToIPv4())))) { Response.StatusCode = 403; return; }
        try
        {
            await using var subscription = await relay.Subscribe(id, token);
            Response.ContentType = "video/mp2t"; Response.Headers.CacheControl = "no-store";
            await foreach (var chunk in subscription.Reader.ReadAllAsync(token))
            { await Response.Body.WriteAsync(chunk, token); await Response.Body.FlushAsync(token); }
        }
        catch (OperationCanceledException) when (token.IsCancellationRequested) { }
        catch (Exception)
        {
            if (Response.HasStarted) HttpContext.Abort();
            else Response.StatusCode = 502;
        }
    }
}