Explorer-Workbench/src/Explorer.Hosting/Ipc/WorkbenchPipeClient.cs

using System.Collections.Concurrent;
using System.IO.Pipes;
using System.Text;
using System.Text.Json;
using Explorer.Contracts;
using Explorer.Domain;

namespace Explorer.Hosting.Ipc;

public sealed class WorkbenchPipeClient : IWorkbenchHost, IAsyncDisposable
{
    private readonly NamedPipeClientStream _pipe;
    private readonly StreamWriter _writer;
    private readonly StreamReader _reader;
    private readonly SemaphoreSlim _send = new(1, 1);
    private readonly ConcurrentDictionary<string, TaskCompletionSource<IpcEnvelope>> _pending = new();
    private readonly CancellationTokenSource _cts = new();
    private readonly Task _readLoop;
    private readonly IndexingProxy _indexing;
    private readonly TransferProxy _transfers;

    private WorkbenchPipeClient(NamedPipeClientStream pipe)
    {
        _pipe = pipe;
        _writer = new StreamWriter(pipe, Encoding.UTF8, leaveOpen: true) { AutoFlush = true, NewLine = "\n" };
        _reader = new StreamReader(pipe, Encoding.UTF8, detectEncodingFromByteOrderMarks: false, leaveOpen: true);
        _indexing = new IndexingProxy(this);
        _transfers = new TransferProxy(this);
        _readLoop = ReadLoopAsync(_cts.Token);
    }

    public IIndexingHost Indexing => _indexing;
    public ITransferHost Transfers => _transfers;

    public static async Task<WorkbenchPipeClient> ConnectAsync(
        WorkbenchIpcOptions options,
        TimeSpan timeout,
        CancellationToken cancellationToken = default)
    {
        var deadline = DateTime.UtcNow + timeout;
        Exception? last = null;
        while (DateTime.UtcNow < deadline)
        {
            cancellationToken.ThrowIfCancellationRequested();
            var pipe = new NamedPipeClientStream(
                ".",
                options.PipeName,
                PipeDirection.InOut,
                PipeOptions.Asynchronous);
            try
            {
                var remaining = deadline - DateTime.UtcNow;
                if (remaining < TimeSpan.FromMilliseconds(50))
                {
                    remaining = TimeSpan.FromMilliseconds(50);
                }

                await pipe.ConnectAsync(remaining, cancellationToken).ConfigureAwait(false);
                var client = new WorkbenchPipeClient(pipe);
                await client.CallAsync("Ping", cancellationToken).ConfigureAwait(false);
                return client;
            }
            catch (Exception ex) when (ex is not OperationCanceledException)
            {
                last = ex;
                await pipe.DisposeAsync().ConfigureAwait(false);
                try
                {
                    await Task.Delay(80, cancellationToken).ConfigureAwait(false);
                }
                catch (OperationCanceledException)
                {
                    throw;
                }
            }
        }

        throw new TimeoutException(
            $"Could not connect to Explorer Workbench host pipe '{options.PipeName}'.", last);
    }

    internal IpcEnvelope Call(string op, long? n = null, string? s = null)
        => CallAsync(op, CancellationToken.None, n, s).GetAwaiter().GetResult();

    internal async Task<IpcEnvelope> CallAsync(
        string op,
        CancellationToken cancellationToken,
        long? n = null,
        string? s = null)
    {
        var id = Guid.NewGuid().ToString("N");
        var tcs = new TaskCompletionSource<IpcEnvelope>(TaskCreationOptions.RunContinuationsAsynchronously);
        _pending[id] = tcs;
        var request = new IpcEnvelope { V = WorkbenchIpc.ProtocolVersion, Id = id, Op = op, N = n, S = s };
        var json = JsonSerializer.Serialize(request, WorkbenchIpc.Json);
        await _send.WaitAsync(cancellationToken).ConfigureAwait(false);
        try
        {
            await _writer.WriteLineAsync(json.AsMemory(), cancellationToken).ConfigureAwait(false);
        }
        catch
        {
            _pending.TryRemove(id, out _);
            throw;
        }
        finally
        {
            _send.Release();
        }

        using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(15));
        using var linked = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, timeout.Token, _cts.Token);
        using var cancelReg = linked.Token.Register(() => tcs.TrySetCanceled(linked.Token));
        try
        {
            var reply = await tcs.Task.ConfigureAwait(false);
            if (reply.Ok == false)
            {
                throw new InvalidOperationException(reply.Error ?? "Host call failed.");
            }

            return reply;
        }
        finally
        {
            _pending.TryRemove(id, out _);
        }
    }

    internal void RaiseProgress(ScanProgress progress) => _indexing.Raise(progress);
    internal void RaiseChanged() => _transfers.RaiseChanged();
    internal void RaiseFinished(TransferJob job) => _transfers.RaiseFinished(job);

    private async Task ReadLoopAsync(CancellationToken cancellationToken)
    {
        try
        {
            while (!cancellationToken.IsCancellationRequested)
            {
                var line = await _reader.ReadLineAsync(cancellationToken).ConfigureAwait(false);
                if (line is null)
                {
                    break;
                }

                if (string.IsNullOrWhiteSpace(line))
                {
                    continue;
                }

                IpcEnvelope? envelope;
                try
                {
                    envelope = JsonSerializer.Deserialize<IpcEnvelope>(line, WorkbenchIpc.Json);
                }
                catch (JsonException)
                {
                    continue;
                }

                if (envelope is null)
                {
                    continue;
                }

                if (!string.IsNullOrEmpty(envelope.Evt))
                {
                    switch (envelope.Evt)
                    {
                        case "Indexing.Progress" when envelope.Progress is not null:
                            RaiseProgress(envelope.Progress);
                            break;
                        case "Transfers.Changed":
                            RaiseChanged();
                            break;
                        case "Transfers.JobFinished" when envelope.Job is not null:
                            RaiseFinished(envelope.Job);
                            break;
                    }

                    continue;
                }

                if (envelope.Id is not null && _pending.TryRemove(envelope.Id, out var tcs))
                {
                    tcs.TrySetResult(envelope);
                }
            }
        }
        catch (OperationCanceledException)
        {
            // shutting down
        }
        finally
        {
            foreach (var tcs in _pending.Values)
            {
                tcs.TrySetCanceled(cancellationToken);
            }
        }
    }

    public async ValueTask DisposeAsync()
    {
        await _cts.CancelAsync().ConfigureAwait(false);
        try
        {
            await _writer.DisposeAsync().ConfigureAwait(false);
        }
        catch
        {
            // pipe already closed
        }

        _reader.Dispose();
        await _pipe.DisposeAsync().ConfigureAwait(false);
        try
        {
            await _readLoop.WaitAsync(TimeSpan.FromSeconds(2)).ConfigureAwait(false);
        }
        catch (TimeoutException)
        {
            // reader may still be unwinding after the pipe close
        }
        catch (OperationCanceledException)
        {
            // expected
        }

        _cts.Dispose();
        _send.Dispose();
    }

    private sealed class IndexingProxy : IIndexingHost
    {
        private readonly WorkbenchPipeClient _client;
        public event EventHandler<ScanProgress>? ProgressChanged = delegate { };

        public IndexingProxy(WorkbenchPipeClient client) => _client = client;

        public void EnqueueFullScan(long sourceId) => _client.Call("Indexing.EnqueueFullScan", sourceId);
        public void EnqueueFolderScan(long sourceId, string pathRel)
            => _client.Call("Indexing.EnqueueFolderScan", sourceId, pathRel);
        public void EnqueueReconcile(long sourceId, string pathRel)
            => _client.Call("Indexing.EnqueueReconcile", sourceId, pathRel);
        public void Cancel(long sourceId) => _client.Call("Indexing.Cancel", sourceId);
        public void Raise(ScanProgress progress) => ProgressChanged?.Invoke(this, progress);
    }

    private sealed class TransferProxy : ITransferHost
    {
        private readonly WorkbenchPipeClient _client;
        public event EventHandler? Changed = delegate { };
        public event EventHandler<TransferJob>? JobFinished = delegate { };

        public TransferProxy(WorkbenchPipeClient client) => _client = client;

        public bool IsPaused => _client.Call("Transfers.IsPaused").Paused == true;

        public IReadOnlyList<TransferJob> Snapshot()
            => _client.Call("Transfers.Snapshot").Jobs ?? [];

        public void PauseAll() => _client.Call("Transfers.PauseAll");
        public void ResumeAll() => _client.Call("Transfers.ResumeAll");
        public void Pause(long jobId) => _client.Call("Transfers.Pause", jobId);
        public void Resume(long jobId) => _client.Call("Transfers.Resume", jobId);
        public void Retry(long jobId) => _client.Call("Transfers.Retry", jobId);
        public void Cancel(long jobId) => _client.Call("Transfers.Cancel", jobId);
        public void Dismiss(long jobId) => _client.Call("Transfers.Dismiss", jobId);
        public void ClearFinished() => _client.Call("Transfers.ClearFinished");
        public bool MoveUp(long jobId) => _client.Call("Transfers.MoveUp", jobId).Flag == true;
        public bool MoveDown(long jobId) => _client.Call("Transfers.MoveDown", jobId).Flag == true;
        public void RaiseChanged() => Changed?.Invoke(this, EventArgs.Empty);
        public void RaiseFinished(TransferJob job) => JobFinished?.Invoke(this, job);
    }
}