using System.Buffers.Binary;
using System.IO.Pipes;
using System.Text.Json;
using Lumi.Companion.Protocol;

namespace Lumi.Companion.Transcription;

public sealed class ObsBridgePipe : IAsyncDisposable
{
    private static readonly HashSet<string> AllowedTypes = ["hello", "source_list", "source_state", "obs_state", "health"];
    private readonly string _pipeName;
    private readonly Func<JsonElement, Task> _onMessage;
    private readonly Func<ReadOnlyMemory<byte>, Task> _onAudio;
    private readonly CancellationTokenSource _lifetime = new();
    private Task? _loop;

    public ObsBridgePipe(string userSidHash, Func<JsonElement, Task> onMessage, Func<ReadOnlyMemory<byte>, Task> onAudio)
    {
        _pipeName = $"Lumi.Companion.ObsBridge.v1.{userSidHash}";
        _onMessage = onMessage; _onAudio = onAudio;
    }
    public void Start() => _loop ??= Task.Run(() => ListenAsync(_lifetime.Token));
    private async Task ListenAsync(CancellationToken cancellationToken)
    {
        while (!cancellationToken.IsCancellationRequested)
        {
            await using var pipe = new NamedPipeServerStream(_pipeName, PipeDirection.InOut, 1, PipeTransmissionMode.Byte,
                PipeOptions.Asynchronous | PipeOptions.CurrentUserOnly, 65536, 65536);
            await pipe.WaitForConnectionAsync(cancellationToken);
            try { await ReadConnectionAsync(pipe, cancellationToken); }
            catch (Exception) when (!cancellationToken.IsCancellationRequested) { /* reconnect without taking down the shell */ }
        }
    }
    private async Task ReadConnectionAsync(Stream stream, CancellationToken cancellationToken)
    {
        var lengthBytes = new byte[4];
        while (!cancellationToken.IsCancellationRequested)
        {
            await ReadExactlyAsync(stream, lengthBytes, cancellationToken);
            var length = BinaryPrimitives.ReadUInt32LittleEndian(lengthBytes);
            if (length is 0 or > ProtocolV1.MaxJsonBytes) throw new InvalidDataException("OBS bridge message exceeds its limit.");
            var message = new byte[(int)length];
            await ReadExactlyAsync(stream, message, cancellationToken);
            if (length >= 4 && message.AsSpan(0, 4).SequenceEqual("LACP"u8))
            {
                if (length > ProtocolV1.AudioHeaderBytes + ProtocolV1.MaxAudioPayloadBytes) throw new InvalidDataException("OBS audio frame exceeds its limit.");
                await _onAudio(message); continue;
            }
            using var json = JsonDocument.Parse(message);
            var type = json.RootElement.TryGetProperty("type", out var property) ? property.GetString() : null;
            if (type is null || !AllowedTypes.Contains(type)) throw new InvalidDataException("OBS bridge message type is not allowed.");
            await _onMessage(json.RootElement.Clone());
        }
    }
    public static async Task WriteAsync(Stream stream, object message, CancellationToken cancellationToken)
    {
        var body = JsonSerializer.SerializeToUtf8Bytes(message, ProtocolV1.JsonOptions);
        if (body.Length > ProtocolV1.MaxJsonBytes) throw new InvalidDataException("Companion IPC message exceeds its limit.");
        var prefix = new byte[4]; BinaryPrimitives.WriteUInt32LittleEndian(prefix, (uint)body.Length);
        await stream.WriteAsync(prefix, cancellationToken); await stream.WriteAsync(body, cancellationToken); await stream.FlushAsync(cancellationToken);
    }
    private static async Task ReadExactlyAsync(Stream stream, Memory<byte> buffer, CancellationToken cancellationToken)
    {
        var read = 0;
        while (read < buffer.Length)
        {
            var count = await stream.ReadAsync(buffer[read..], cancellationToken);
            if (count == 0) throw new EndOfStreamException();
            read += count;
        }
    }
    public async ValueTask DisposeAsync() { _lifetime.Cancel(); if (_loop is not null) try { await _loop; } catch (OperationCanceledException) { } _lifetime.Dispose(); }
}