using MoreLinq;
using MySql.Data.MySqlClient;
using PetaPoco.Business;
using Sony.Filtr.Contracts.Entities;
using Sony.Filtr.Playlists.Spotify;
using Sony.Filtr.SpotifyAnalytics;
using Sony.Filtr.SpotifyAnalytics.Data;
using Sony.Filtr.SpotifyAnalytics.Models;
using Sony.Filtr.Utility.Extensions;
using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;
using Sony.Filtr.Utility;
using Sony.Filtr.Tasks.Helpers;
using Sony.Filtr.Functional;

namespace Sony.Filtr.Tasks.Tasks.Spotify.Analytics
{
    public class AggregatedPlaylistImportTask : AggregatedImportTaskBase
    {
        private readonly SpotifyAnalyticsManager _spotifyAnalyticsManager;
        private readonly SpotifyPlaylistManager _spotifyPlaylistManager;
        private readonly SpotifyAnalyticsPlaylistAggregation _spotifyAnalyticsPlaylistAggregation;
        private readonly CalculatePlaylistCategoryStreams _calculatePlaylistCategoryStreams;


        private readonly AsyncLogger _logger;
        private readonly AsyncLogger _trackLogger;
        private readonly AsyncLogger _durationLogger;

        private static readonly IEnumerable<SpotifyS3FileType> SupportedFileTypes = new List<SpotifyS3FileType>()
            {
                SpotifyS3FileType.Playlists,
            };

        public AggregatedPlaylistImportTask(SpotifyStreamingAggregatedReportApi aggregatedReportApi,
                                            SpotifyAnalyticsManager spotifyAnalyticsManager,
                                            SpotifyPlaylistManager spotifyPlaylistManager,
                                            SpotifyAnalyticsPlaylistAggregation spotifyAnalyticsPlaylistAggregation,
                                            CalculatePlaylistCategoryStreams calculatePlaylistCategoryStreams)
            : base(aggregatedReportApi)
        {
            _spotifyAnalyticsManager = spotifyAnalyticsManager;
            _spotifyPlaylistManager = spotifyPlaylistManager;
            _spotifyAnalyticsPlaylistAggregation = spotifyAnalyticsPlaylistAggregation;
            _calculatePlaylistCategoryStreams = calculatePlaylistCategoryStreams;
            _logger = AsyncLogger.GetLogger("SpotifyStreamingAggregatedReportApi");
            _trackLogger = AsyncLogger.GetLogger("TrackLogger");
            _durationLogger = AsyncLogger.GetLogger("Duration");
        }

        public override async Task<ScheduledTaskLog> ExecuteAsync(Guid scheduledTaskLogId)
        {
            if (!Directory.Exists(TempFolder))
            {
                Directory.CreateDirectory(TempFolder);
            }

            var recalculateManager = RecalculateManager.Create();

            return recalculateManager.Recalculate ?
                await this.RecalculateFromConfig(recalculateManager) :
                await this.RecalculateFromS3();
        }

        private async Task<ScheduledTaskLog> RecalculateFromConfig(RecalculateManager manager)
        {
            var dates = manager.StartDateIncluding.Value.GetDateRangeTo(manager.EndDateExcluding.Value.AddDays(-1)).Reverse().ToList();

            _logger.InfoAsync(() => $"Recalculating dates only").FireAndForget();

            await this.RecalculateForDatesAsync(dates);

            return new ScheduledTaskLog();
        }

        private static int DaysToLookBack()
        {
            return Maybe.GetAppSettingsIntOrDefault("AggregatedPlaylistImportTask_Days_To_Look_Back", 50);
        }

        private static int GetFilesToProcessCount()
        {
            return Maybe.GetAppSettingsIntOrDefault("AggregatedPlaylistImportTask_Get_Files_To_Process_Block_Count", 2);
        }

        private static int DownloadFilesToProcessCount()
        {
            return Maybe.GetAppSettingsIntOrDefault("AggregatedPlaylistImportTask_Download_Files_To_Process_Block_Count", 4);
        }

        private static int BulkImportCount()
        {
            return Maybe.GetAppSettingsIntOrDefault("AggregatedPlaylistImportTask_Bulk_Import_Block_Count", 3);
        }

        private static int MarkAsProcessedCount()
        {
            return Maybe.GetAppSettingsIntOrDefault("AggregatedPlaylistImportTask_Mark_Processed_Block_Count", 2);
        }

        private static bool MarkFileAsProcessedInDB()
        {
            return Maybe.GetAppSettingsBooleanOrDefault("AggregatedPlaylistImportTask_Mark_FileAs_Processed_In_DB", true);
        }

        private static bool BulkImportRawStreams()
        {
            return Maybe.GetAppSettingsBooleanOrDefault("AggregatedPlaylistImportTask_Bulk_Import_Raw_Streams", true);
        }

        private static string NewPlaylistsFolderPath()
        {
            var path = Maybe.GetAppSettingsStringOrDefault("AggregatedPlaylistImportTask_New_Playlist_Ids_File_Path", "");

            if (String.IsNullOrEmpty(path)) { path = Environment.CurrentDirectory; }

            return path;
        }

        private async Task<ScheduledTaskLog> RecalculateFromS3()
        {
            HashSet<string> knownPlaylistIds = await GetKnownPlaylistIds();

            bool removeLocalS3File = RemoveLocalS3Files;

            Func<DateTime, Task<IEnumerable<SpotifyS3File>>> getFilesToProcessForDate = async d => await GetFilesToImportFromS3(d, SupportedFileTypes);

            TransformManyBlock<DateTime, Result<FileToProcess>> getFilesToProcessBlock =
                getFilesToProcessForDate
                .Timeout(TimeSpan.FromSeconds(30))
                .Retry(1)
                .Duration((dt, files, duration) => _durationLogger.DebugAsync(() => $"Files ({files.Count()}) for date {dt.ToShortDateString()} in {duration.TotalMilliseconds}ms"))
                .Map<DateTime, IEnumerable<FileToProcess>, DateTime, IEnumerable<SpotifyS3File>>(
                    inputTransform: dt => dt,
                    outputTransform: (dt, files, dt2) => files.Select((f, i) => new FileToProcess(dt, f, i)).ToList()
                )
                .TryCatch()
                .OnFailure((dt, filesResult) => this._logger.ErrorAsync(filesResult.Exception, () => $"Could not load files for date {dt.ToShortDateString()}"))
                .OnSuccess((dt, filesResult) =>
                {
                    if (filesResult.Value.Any()) { this._logger.InfoAsync(() => $"Loaded {filesResult.Value.Count()} files for date {dt.ToShortDateString()}"); }
                    else { this._logger.InfoAsync(() => $"No new files to process for date {dt.ToShortDateString()}"); }
                })
                .SelectMany()
                .AsTransformManyBlock(new ExecutionDataflowBlockOptions() { MaxDegreeOfParallelism = GetFilesToProcessCount() });

            Func<SpotifyS3File, Task<IEnumerable<SpotifyPlaylistAggregatedStreamV2>>> downloadAndParseFileFromS3 = async s3File =>
            {
                var localPath = await DownloadFileFromS3Async(s3File);

                var streams = ReadNJsonData<SpotifyPlaylistAggregatedStreamV2>(localPath, s3File.DistributorId).ToList();

                if (removeLocalS3File)
                {
                    File.Delete(localPath);
                }

                return streams;
            };

            TransformBlock<Result<FileToProcess>, Result<StreamsToProcess>> downloadFileBlock =
                downloadAndParseFileFromS3
                .Timeout(TimeSpan.FromSeconds(90))
                .Retry(3)
                .Duration((file, streams, duration) => _durationLogger.DebugAsync(() => $"Streams ({streams.Count()}) for file {file.FilePath} downloaded and parsed in {duration.TotalMilliseconds}ms"))
                .Map<FileToProcess, StreamsToProcess, SpotifyS3File, IEnumerable<SpotifyPlaylistAggregatedStreamV2>>(
                    inputTransform: fileToProcess => fileToProcess.File,
                    outputTransform: (s3File, streams, fileToProcess) => new StreamsToProcess(streams, s3File, fileToProcess.Date, fileToProcess.Index)
                )
                .TryCatch()
                .OnFailure((fileToProcess, streamsResult) => _logger.ErrorAsync(streamsResult.Exception, () => $"Could not download or parse S3 file: '{fileToProcess.File.FilePath}'."))
                .OnSuccess((fileToProcess, streamsResult) => _logger.DebugAsync(() => $"Read {streamsResult.Value.Streams.Count()} playlists streams from {fileToProcess.File.FilePath}"))
                .AcceptResult()
                .AsTransformBlock(new ExecutionDataflowBlockOptions() { MaxDegreeOfParallelism = DownloadFilesToProcessCount(), BoundedCapacity = 1000 });

            TransformBlock<Result<StreamsToProcess>, Result<StreamsToProcess>> fixStreamPlaylistIdsBlock = new TransformBlock<Result<StreamsToProcess>, Result<StreamsToProcess>>(
                dto =>
                {
                    foreach (var stream in dto.Value.Streams)
                    {
                        stream.PlaylistUri = new SpotifyLink(stream.PlaylistUri).ExtractPlaylistID();
                    }

                    return dto;
                }
                , new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 1 });

            bool bulkImportRawStreams = BulkImportRawStreams();
            Func<IEnumerable<SpotifyPlaylistAggregatedStreamV2>, Task> bulkImportStreams = async streams =>
            {
                if (bulkImportRawStreams)
                {
                    await PetaPocoRepository.Instance.ImportBulkFileLoaderAsync(streams, MySqlBulkLoaderConflictOption.Replace);
                }
            };

            TransformBlock<Result<StreamsToProcess>, Result<PlaylistIdsToProcess>> bulkImportStreamsBlock =
                bulkImportStreams
                .ToUnit()
                .Timeout(TimeSpan.FromMinutes(5))
                .Retry(3)
                .Duration((streams, unit, duration) => _durationLogger.DebugAsync(() => $"Streams ({streams.Count()}) bulk inserted in {duration.TotalMilliseconds}ms"))
                .Map<StreamsToProcess, PlaylistIdsToProcess, IEnumerable<SpotifyPlaylistAggregatedStreamV2>, Unit>(
                    inputTransform: dto => dto.Streams,
                    outputTransform: (streams, unit, dto) => new PlaylistIdsToProcess(streams.Select(s => s.PlaylistUri), dto.Date, dto.File, dto.FileIndex)
                )
                .TryCatch()
                .OnFailure((dto, result) => _logger.ErrorAsync(result.Exception, () => $"Could not bulk insert streams for date {dto.Date.ToShortDateString()}"))
                .OnSuccess((dto, result) => _logger.DebugAsync(() => $"Done importing playlists streams ({dto.Streams.Count()}) from {dto.File.FilePath} file index: {dto.FileIndex}"))
                .AcceptResult()
                .AsTransformBlock(new ExecutionDataflowBlockOptions() { MaxDegreeOfParallelism = BulkImportCount() });

            Task<bool> TrueTask = Task.FromResult(true);
            bool markFileAsProcessedInDB = MarkFileAsProcessedInDB();
            Func<DateTime, SpotifyS3File, Task> markAsProcessed = (dt, file) =>
            {
                if (markFileAsProcessedInDB)
                {
                    MarkAsProcessed(dt, file);
                }

                return TrueTask;
            };

            TransformBlock<Result<PlaylistIdsToProcess>, Result<PlaylistIdsToProcess>> markFileAsProcessedBlock =
                markAsProcessed
                .Tuple()
                .ToUnit()
                .Timeout(TimeSpan.FromSeconds(10))
                .Retry(1)
                .Duration((tuple, unit, duration) => _durationLogger.DebugAsync(() => $"Marked as processed {tuple.Item2.FilePath} in {duration.TotalMilliseconds}ms"))
                .Map<PlaylistIdsToProcess, PlaylistIdsToProcess, (DateTime, SpotifyS3File), Unit>(
                    inputTransform: dto => (dto.Date, dto.File),
                    outputTransform: (tuple, unit, dto) => dto
                )
                .TryCatch()
                .OnFailure((dto, result) => _logger.ErrorAsync(result.Exception, () => $"Could not mark file as processed '{dto.File.FilePath}'"))
                .OnSuccess((idsToProcess, result) => _logger.InfoAsync(() => $"File marked as processed '{idsToProcess.File.FilePath}' file index: {idsToProcess.FileIndex}"))
                .AcceptResult()
                .AsTransformBlock(new ExecutionDataflowBlockOptions() { MaxDegreeOfParallelism = MarkAsProcessedCount() });

            var dateWithIdsBroadcastBlock = new BroadcastBlock<Result<PlaylistIdsToProcess>>(dto => dto, new DataflowBlockOptions() { MaxMessagesPerTask = 1 });

            var selectPlaylistIdsBlock = new TransformBlock<Result<PlaylistIdsToProcess>, string[]>(
                dto => dto.Value.PlaylistIds.ToArray(),
                new ExecutionDataflowBlockOptions() { MaxMessagesPerTask = 1 });

            var selectDateBlock = new TransformBlock<Result<PlaylistIdsToProcess>, DateTime>(
                dto => dto.Value.Date,
                new ExecutionDataflowBlockOptions() { MaxMessagesPerTask = 1 });

            HashSet<DateTime> datesToRecalculate = new HashSet<DateTime>();
            ActionBlock<DateTime> storeDateToRecalculateBlock = new ActionBlock<DateTime>(dt =>
            {
                datesToRecalculate.Add(dt);
            }, new ExecutionDataflowBlockOptions() { MaxDegreeOfParallelism = 1 });

            HashSet<string> newPlaylists = new HashSet<string>();
            ActionBlock<string[]> storeNewPlaylistIdsBlock = new ActionBlock<string[]>(
                ids =>
                {
                    foreach (var id in ids)
                    {
                        if (!knownPlaylistIds.Contains(id))
                        {
                            newPlaylists.Add(id);
                        }
                    }
                },
                new ExecutionDataflowBlockOptions() { MaxDegreeOfParallelism = 1 });


            var linkOptions = new DataflowLinkOptions() { PropagateCompletion = true };

            getFilesToProcessBlock.LinkTo(downloadFileBlock, linkOptions, dto => dto.IsOk);
            getFilesToProcessBlock.LinkTo(DataflowBlock.NullTarget<Result<FileToProcess>>());

            downloadFileBlock.LinkTo(fixStreamPlaylistIdsBlock, linkOptions, dto => dto.IsOk);
            downloadFileBlock.LinkTo(DataflowBlock.NullTarget<Result<StreamsToProcess>>());

            fixStreamPlaylistIdsBlock.LinkTo(bulkImportStreamsBlock, linkOptions);

            bulkImportStreamsBlock.LinkTo(markFileAsProcessedBlock, linkOptions, dto => dto.IsOk);
            bulkImportStreamsBlock.LinkTo(DataflowBlock.NullTarget<Result<PlaylistIdsToProcess>>());

            markFileAsProcessedBlock.LinkTo(dateWithIdsBroadcastBlock, linkOptions, dto => dto.IsOk);
            markFileAsProcessedBlock.LinkTo(DataflowBlock.NullTarget<Result<PlaylistIdsToProcess>>());

            dateWithIdsBroadcastBlock.LinkTo(selectDateBlock, linkOptions);
            dateWithIdsBroadcastBlock.LinkTo(selectPlaylistIdsBlock, linkOptions);

            selectDateBlock.LinkTo(storeDateToRecalculateBlock, linkOptions);

            selectPlaylistIdsBlock.LinkTo(storeNewPlaylistIdsBlock, linkOptions);

            var endDate = DateTime.Today;
            var fromDate = endDate.AddDays(-DaysToLookBack());
            var dates = fromDate.GetDateRangeTo(endDate).Reverse().ToList();

            _trackLogger.DebugAsync(() => $"Dates to process: {String.Join(", ", dates.Select(d => d.ToShortDateString()))}").FireAndForget();

            foreach (var date in dates)
            {
                await getFilesToProcessBlock.SendAsync(date);
            }

            getFilesToProcessBlock.Complete();

            await Task.WhenAll(
                storeDateToRecalculateBlock.Completion,
                await storeNewPlaylistIdsBlock.Completion.ContinueWith(async t => await SaveNewPlaylistIdsAsync(newPlaylists)));

            await ValidateDates(datesToRecalculate)
                .OnFailure(ReportDateValidationError)
                .OnSuccess(ReportDatesToSkip)
                .OnSuccess(async tuples => await RecalculateForDatesAsync(tuples.Where(r => r.IsReadyForProcessing).Select(r => r.Date).ToList()));

            return new ScheduledTaskLog();
        }

        private void ReportDatesToSkip(IEnumerable<(DateTime Date, bool IsReadyForProcessing, int FilesProcessed, double Average)> tuples)
        {
            foreach (var tuple in tuples.Where(r => !r.IsReadyForProcessing))
            {
                _logger.InfoAsync(() => $"SKIPPING DATE {tuple.Date.ToShortDateString()}. Files count: {tuple.FilesProcessed}. Expected average: {tuple.Average}");
            } 
        }

        private void ReportDateValidationError(Exception exception)
        {
            _logger.ErrorAsync(exception, () => $"Could not validate dates for recalculation").FireAndForget();
        }

        private async Task<Result<IEnumerable<(DateTime Date, bool IsReadyForProcessing, int FilesProcessed, double Average)>>> ValidateDates(IEnumerable<DateTime> datesProcessed)
        {
            const int AveragePercentToCompareWith = 95;

            return await GetFilesCountForLastNDays(30, datesProcessed).ToResult()
                .Bind(map => (Map: map, Average: map.Average(p => p.Value)))
                .Bind(tuple => tuple.Map.Select(pair => (
                            Date: pair.Key,
                            IsReadyForProcessing: pair.Value >= tuple.Average.Percentage(AveragePercentToCompareWith),
                            FilesProcessed: pair.Value,
                            Average: tuple.Average.Percentage(AveragePercentToCompareWith))
                        ).ToList())
                .Bind(tuple => tuple.Where(t => datesProcessed.Contains(t.Date)))
                ;
        }

        private async Task<Dictionary<DateTime, int>> GetFilesCountForLastNDays(int daysToLookBack, IEnumerable<DateTime> processedDates)
        {
            return await _spotifyAnalyticsManager.GetFilesProcessedCountForDates(
                DateTime.UtcNow.AddDays(-daysToLookBack).GetDateRangeToNow().Union(processedDates),
                SupportedFileTypes,
                2);
        }

        private async Task SaveNewPlaylistIdsAsync(IEnumerable<string> newPlaylistIds)
        {
            if (!newPlaylistIds.Any()) { return; }

            string folderPath = NewPlaylistsFolderPath();
            if (!Directory.Exists(folderPath))
            {
                _logger.WarnAsync(() => $"Specified folder for new playlist ids file does not exist '{folderPath}'. Defaulting to current folder '{Environment.CurrentDirectory}'").FireAndForget();
                folderPath = Environment.CurrentDirectory;
            }

            string filePath = Path.Combine(folderPath, $"New_Playlist_Ids_{DateTime.Now.ToString("yyyy_MM_dd")}_{Guid.NewGuid().ToString()}.txt");

            using (StreamWriter outputFile = new StreamWriter(filePath))
            {
                await outputFile.WriteAsync(string.Join(Environment.NewLine, newPlaylistIds));
            }

            _logger.InfoAsync(() => $"Saved total {newPlaylistIds.Count()} new playlist ids to file '{filePath}'").FireAndForget();
        }

        private async Task RecalculateForDatesAsync(IEnumerable<DateTime> datesToRecalculate)
        {
            if (!datesToRecalculate.Any()) { return; }

            _logger.InfoAsync(() => $"********** Going to recalculate dates {String.Join(",", datesToRecalculate.Select(d => d.ToShortDateString()))}").FireAndForget();

            var getImportedFilesForDate =
                ((Func<DateTime, List<SpotifyAnalyticsAggregatedDate>>)_spotifyAnalyticsManager.GetImportedFiles)
                .LazyMemoizeThreadSafe()
                .ToTask()
                .Timeout(TimeSpan.FromSeconds(10))
                .Retry(3)
                .TryCatch()
                .OnFailure((dt, result) => _logger.ErrorAsync(result.Exception, () => $"Could not get imported files for date '{dt.ToShortDateString()}'"));

            TransformBlock<Result<DateTime>, Result<DateTime>> aggregatePlaylistStreamsBlock = ((Func<DateTime, Task>)_spotifyAnalyticsPlaylistAggregation.AggregatePlaylistStreamsAsync)
                .ToUnit()
                .Duration((dt, unit, duration) => _durationLogger.InfoAsync(() => $"Aggregate playlist streams duration - {duration.TotalSeconds}sec"))
                .Timeout(TimeSpan.FromHours(3))
                .Retry(2)
                .Map<DateTime, DateTime, DateTime, Unit>(
                    inputTransform: dt => dt,
                    outputTransform: (dt1, unit, dt2) => dt1
                )
                .TryCatch()
                .OnSuccess((dt, result) => _logger.InfoAsync(() => $"tblSpotifyAnalyticsAccountPlaylistStreamInfo aggregated for date '{dt.ToShortDateString()}'"))
                .OnFailure((dt, result) => _logger.ErrorAsync(result.Exception, () => $"Could not calculate aggregation (tblSpotifyAnalyticsAccountPlaylistStreamInfo) for date '{dt.ToShortDateString()}'"))
                .AcceptResult()
                .AsTransformBlock(new ExecutionDataflowBlockOptions() { MaxDegreeOfParallelism = 1 });

            Func<DateTime, Task> setMarketsAsAggregated = async dt =>
            {
                var result = await getImportedFilesForDate(dt);
                if (result.IsOk)
                {
                    await _spotifyAnalyticsManager.SetMarketsAsAggregatedAsync(result.Value, dt);
                }
            };

            ActionBlock<Result<DateTime>> setMarketsAsAggregatedBlock = setMarketsAsAggregated
                .ToUnit()
                .Timeout(TimeSpan.FromMinutes(1))
                .Retry(1)
                .TryCatch()
                .OnFailure((dt, result) => _logger.ErrorAsync(result.Exception, () => $"Could not set markets as aggregated for date '{dt.ToShortDateString()}'"))
                .AcceptResult()
                .AsActionBlock(new ExecutionDataflowBlockOptions() { MaxDegreeOfParallelism = 1 });


            TransformBlock<Result<DateTime>, Result<DateTime>> calculatePlaylistCategoryStreamsBlock = ((Func<DateTime, Task>)_calculatePlaylistCategoryStreams.CalculatePlaylistCategoryStreamsAsync)
                .ToUnit()
                .Duration((dt, unit, duration) => _durationLogger.InfoAsync(() => $"Calculate playlist category streams duration - {duration.TotalSeconds}sec"))
                .Timeout(TimeSpan.FromHours(4))
                .Retry(2)
                .Map<DateTime, DateTime, DateTime, Unit>(
                    inputTransform: dt => dt,
                    outputTransform: (dt, unit, dt2) => dt
                )
                .TryCatch()
                .OnFailure((dt, result) => _logger.ErrorAsync(result.Exception, () => $"Could not calculate playlist category streams for date '{dt.ToShortDateString()}'"))
                .AcceptResult()
                .AsTransformBlock(new ExecutionDataflowBlockOptions() { MaxDegreeOfParallelism = 1 });

            Func<DateTime, Task> setMarketsAsCalculatedCategoryStreams = async dt =>
            {
                var result = await getImportedFilesForDate(dt);
                if (result.IsOk)
                {
                    await _spotifyAnalyticsManager.SetMarketsAsCalculatedCategoryStreamsAsync(result.Value, dt);
                }
            };

            ActionBlock<Result<DateTime>> setMarketsAsCalculatedCategoryStreamsBlock = setMarketsAsCalculatedCategoryStreams
                .ToUnit()
                .Timeout(TimeSpan.FromMinutes(1))
                .Retry(1)
                .TryCatch()
                .OnFailure((dt, result) => _logger.ErrorAsync(result.Exception, () => $"Could not set markets as calculated category streams for date '{dt.ToShortDateString()}'"))
                .OnSuccess((dt, result) => _trackLogger.InfoAsync(() => $"*******************Finished recalculate for date {dt.ToShortDateString()}*******************"))
                .AcceptResult()
                .AsActionBlock(new ExecutionDataflowBlockOptions() { MaxDegreeOfParallelism = 1 });

            var linkOptions = new DataflowLinkOptions() { PropagateCompletion = true };

            var dateBroadcastBlock = new BroadcastBlock<Result<DateTime>>(result => result);
            aggregatePlaylistStreamsBlock.LinkTo(dateBroadcastBlock, linkOptions, dto => dto.IsOk);

            dateBroadcastBlock.LinkTo(setMarketsAsAggregatedBlock, linkOptions);
            dateBroadcastBlock.LinkTo(calculatePlaylistCategoryStreamsBlock, linkOptions);

            calculatePlaylistCategoryStreamsBlock.LinkTo(setMarketsAsCalculatedCategoryStreamsBlock, linkOptions, dto => dto.IsOk);

            var ctSource = new CancellationTokenSource();

            foreach (var date in datesToRecalculate)
            {
                _logger.InfoAsync(() => $"Date to recalculate {date.ToShortDateString()}").FireAndForget();
                await aggregatePlaylistStreamsBlock.SendAsync(date);
            }

            aggregatePlaylistStreamsBlock.Complete();
            await Task.WhenAll(setMarketsAsCalculatedCategoryStreamsBlock.Completion, setMarketsAsAggregatedBlock.Completion);
            ctSource.Cancel();
        }

        private static bool RemoveLocalS3Files
        {
            get
            {
                return Maybe.GetAppSettingsBooleanOrDefault("AggregatedPlaylistImportTask_Remove_Local_S3_File", true);
            }
        }

        private async Task<HashSet<string>> GetKnownPlaylistIds()
        {
            return (await _spotifyPlaylistManager.GetAllPlaylistIdsAsync(includeRemoved: true)).ToHashSet();
        }

        private class RecalculateManager
        {
            public readonly bool Recalculate;
            public Nullable<DateTime> StartDateIncluding;
            public Nullable<DateTime> EndDateExcluding;

            public static RecalculateManager Create()
            {
                return new RecalculateManager(
                    Maybe.GetAppSettingsBooleanOrDefault("AggregatedPlaylistImportTask_RecalculateOnly", false),
                    Maybe.GetAppSettingsDateTimeOrDefault("AggregatedPlaylistImportTask_RecalculateOnly_StartDateIncluding", null),
                    Maybe.GetAppSettingsDateTimeOrDefault("AggregatedPlaylistImportTask_RecalculateOnly_EndDateExcluding", null));
            }

            private RecalculateManager(bool recalculate, Nullable<DateTime> start, Nullable<DateTime> end)
            {
                this.Recalculate = recalculate && (start.HasValue && end.HasValue && start.Value < end.Value);
                this.StartDateIncluding = start;
                this.EndDateExcluding = end;
            }
        }

        private struct FileToProcess
        {
            public readonly SpotifyS3File File;
            public readonly DateTime Date;
            public readonly int Index;

            public FileToProcess(DateTime date, SpotifyS3File file, int index)
            {
                this.File = file;
                this.Date = date;
                this.Index = index;
            }
        }

        private struct StreamsToProcess
        {
            public readonly DateTime Date;
            public readonly SpotifyS3File File;
            public readonly int FileIndex;
            public readonly IEnumerable<SpotifyPlaylistAggregatedStreamV2> Streams;

            public StreamsToProcess(IEnumerable<SpotifyPlaylistAggregatedStreamV2> streams, SpotifyS3File file, DateTime date, int index)
            {
                this.Date = date;
                this.Streams = streams;
                this.File = file;
                this.FileIndex = index;
            }
        }

        private struct PlaylistIdsToProcess
        {
            public readonly IEnumerable<string> PlaylistIds;
            public readonly DateTime Date;
            public readonly SpotifyS3File File;
            public readonly int FileIndex;

            public PlaylistIdsToProcess(IEnumerable<string> ids, DateTime date, SpotifyS3File file, int index)
            {
                this.PlaylistIds = ids;
                this.Date = date;
                this.File = file;
                this.FileIndex = index;
            }
        }
    }
}