using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Text;
using System.Threading;


using var dedup = new Dedup(inFileName: "list.txt", outFileName: "nodup.txt", encoding: Encoding.ASCII);
dedup.Run(Environment.ProcessorCount);


class Dedup : IDisposable
{
    public Dedup(string inFileName, string outFileName, Encoding encoding)
    {
        _reader = new StreamReader(
            stream: new FileStream(inFileName, FileMode.Open, FileAccess.Read),
            encoding: encoding,
            leaveOpen: false);

        _writer = new StreamWriter(
            stream: new FileStream(outFileName, FileMode.Create, FileAccess.Write),
            encoding: encoding,
            leaveOpen: false)
        {
            NewLine = "\n"
        };
    }

    public void Run(int nofWorkers)
    {
        // spin-up all our threads
        var threadReader = new Thread(ReaderThread)
        {
            Name = "Reader",
            Priority = ThreadPriority.AboveNormal
        };
        threadReader.Start();

        var threadWriter = new Thread(WriterThread)
        {
            Name = "Writer",
            Priority = ThreadPriority.Highest
        };
        threadWriter.Start();

        var workers = Enumerable.Range(0, Math.Max(1, nofWorkers))
            .Select(x => new DedupWorker(x, queueIngress: _queueIngress, queueEgress: _queueEgress))
            .ToArray();

        // wait until reader has read all lines from input file
        threadReader.Join();

        // enqueue death packet into processing queue
        _queueIngress.Enqueue(Enumerable.Repeat(new LineToProcess
        {
            LineNumber = 0
        }, workers.Length));
        // wait until all dedup workers are done
        foreach (var worker in workers)
        {
            worker.Join();
        }

        // enqueue death packet into output queue
        _queueEgress.Enqueue(new ProcessedLine()
        {
            LineNumber = 0
        });
        // wait until writer thread is done
        threadWriter.Join();
    }

    public void Dispose()
    {
        _writer.Dispose();
        _reader.Dispose();
    }

    public void ReaderThread()
    {
        UInt64 currentLine = 1;
        while (_reader.ReadLine() is string line)
        {
            _queueIngress.Enqueue(new LineToProcess
            {
                LineNumber = currentLine++,
                Text = line
            });
        }
    }

    public void WriterThread()
    {
        UInt64 currentLine = 1;
        var reorderDict = new Dictionary<UInt64, ProcessedLine>();
        bool running = true;

        while (running || reorderDict.Any())
        {
            var line = _queueEgress.Dequeue();

            if (line.LineNumber != 0)
            {
                // check if current packet is in sequence
                if (line.LineNumber != currentLine)
                {
                    // nope, keep for later
                    reorderDict.Add(line.LineNumber, line);
                }
                else
                {
                    // yes, write this to file
                    prettyWriter(line);
                    // check if we have the next one(s) already
                    while (reorderDict.TryGetValue(++currentLine, out ProcessedLine nextInLine))
                    {
                        // yup, remove from our list and write it out
                        reorderDict.Remove(nextInLine.LineNumber);
                        prettyWriter(nextInLine);
                    }
                }
            }
            else
            {
                running = false;
            }
        }

        // our DRY writer
        void prettyWriter(ProcessedLine line)
        {
            foreach (var column in line.Columns)
            {
                _writer.Write(column);
                _writer.Write(';');
            }
            _writer.Write('\n');
        }
    }

    private readonly StreamReader _reader;
    private readonly StreamWriter _writer;
    private readonly WaitableQueue<LineToProcess> _queueIngress = new WaitableQueue<LineToProcess>();
    private readonly WaitableQueue<ProcessedLine> _queueEgress = new WaitableQueue<ProcessedLine>();
    private readonly CancellationTokenSource _cts = new CancellationTokenSource();

    record LineToProcess
    {
        public UInt64 LineNumber { get; init; }
        public string Text { get; init; }
    }

    record ProcessedLine
    {
        public UInt64 LineNumber { get; init; }
        public string[] Columns { get; init; }
    }

    class DedupWorker
    {
        public DedupWorker(int no, WaitableQueue<LineToProcess> queueIngress, WaitableQueue<ProcessedLine> queueEgress)
        {
            _queueIngress = queueIngress;
            _queueEgress = queueEgress;

            _thread = new Thread(DedupThread)
            {
                Name = $"Dedup Worker #{no}"
            };
            _thread.Start();
        }

        void DedupThread()
        {
            var known = new Dictionary<string, object>();

            while (true)
            {
                var line = _queueIngress.Dequeue();
                if (line.LineNumber == 0)
                    break;

                _queueEgress.Enqueue(new ProcessedLine()
                {
                    LineNumber = line.LineNumber,
                    Columns = DedupCsvColumns(known, line.Text).ToArray()
                });
            }
        }

        public void Join() => _thread.Join();

        private readonly WaitableQueue<LineToProcess> _queueIngress;
        private readonly WaitableQueue<ProcessedLine> _queueEgress;
        private readonly Thread _thread;
    }

    private static IEnumerable<string> DedupCsvColumns(Dictionary<string, object> known, string line)
    {
        const int NofFixedColumns = 3;
        var columns = line.Split(';');

        known.Clear();

        foreach (var column in columns.Take(NofFixedColumns))
        {
            yield return column;
        }

        foreach (var column in columns.Skip(NofFixedColumns).SkipLast(1))
        {
            yield return known.TryAdd(column, 0) ? column : string.Empty;
        }
    }
}

class WaitableQueue<T>
{
    public void Enqueue(in T value)
    {

        _queue.Enqueue(value);
        _enqueuedSema.Release(1);
    }

    public void Enqueue(IEnumerable<T> values)
    {
        foreach (var value in values)
        {
            Enqueue(value);
        }
    }

    public T Dequeue()
    {
        _enqueuedSema.Wait();
        return _queue.TryDequeue(out T value)
            ? value
            : throw new Exception("nothing to dequeue");
    }

    private readonly SemaphoreSlim _enqueuedSema = new SemaphoreSlim(0);
    private readonly ConcurrentQueue<T> _queue = new ConcurrentQueue<T>();
}