<|start_filename|>AzureFunctions.v3/Extensions.cs<|end_filename|>
﻿using System.Collections.Generic;
using System.Net;
using System.Net.Http;
using Microsoft.WindowsAzure.Storage.Blob;
using Microsoft.WindowsAzure.Storage.Table;

namespace FileValidation
{
    static class StorageExtensions
    {
        public static async System.Threading.Tasks.Task<IEnumerable<T>> ExecuteQueryAsync<T>(this CloudTable table, TableQuery<T> query) where T : ITableEntity, new()
        {
            TableContinuationToken token = null;
            var retVal = new List<T>();
            do
            {
                var results = await table.ExecuteQuerySegmentedAsync(query, token);
                retVal.AddRange(results.Results);
                token = results.ContinuationToken;
            } while (token != null);

            return retVal;
        }


        public static async System.Threading.Tasks.Task<IEnumerable<IListBlobItem>> ListBlobsAsync(this CloudBlobClient blobClient, string prefix)
        {
            BlobContinuationToken token = null;
            var retVal = new List<IListBlobItem>();
            do
            {
                var results = await blobClient.ListBlobsSegmentedAsync(prefix, token);
                retVal.AddRange(results.Results);
                token = results.ContinuationToken;
            } while (token != null);

            return retVal;
        }

    }

    static class HttpExtensions
    {
        public static HttpResponseMessage CreateCompatibleResponse(this HttpRequestMessage _, HttpStatusCode code) => new HttpResponseMessage(code);

        public static HttpResponseMessage CreateCompatibleResponse(this HttpRequestMessage _, HttpStatusCode code, string stringContent) => new HttpResponseMessage(code) { Content = new StringContent(stringContent) };
    }
}


<|start_filename|>AzureFunctions.v3/sample.local.settings.json<|end_filename|>
{
  "IsEncrypted": false,
  "Values": {
    "AzureWebJobsStorage": "UseDevelopmentStorage=true",

    "CustomerBlobStorage": "<blob storage connection string>",
    "ValidateFunctionUrl": "http://localhost:7071/api/Validate"
  }
}

<|start_filename|>AzureFunctions.v3/FunctionValidateFileSet.cs<|end_filename|>
﻿using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Net;
using System.Net.Http;
using System.Threading.Tasks;
using Microsoft.Azure.WebJobs;
using Microsoft.Azure.WebJobs.Extensions.Http;
using Microsoft.Extensions.Logging;
using Microsoft.WindowsAzure.Storage;
using Microsoft.WindowsAzure.Storage.Blob;
using Microsoft.WindowsAzure.Storage.Table;
using Newtonsoft.Json.Linq;

namespace FileValidation
{
    public static class FunctionValidateFileSet
    {
        [FunctionName(@"ValidateFileSet")]
        public static async Task<HttpResponseMessage> Run([HttpTrigger(AuthorizationLevel.Function, @"post", Route = @"Validate")] HttpRequestMessage req, ILogger log)
        {
            log.LogTrace(@"ValidateFileSet run.");
            if (!CloudStorageAccount.TryParse(Environment.GetEnvironmentVariable(@"CustomerBlobStorage"), out var storageAccount))
            {
                throw new Exception(@"Can't create a storage account accessor from app setting connection string, sorry!");
            }

            var payload = JObject.Parse(await req.Content.ReadAsStringAsync());

            var prefix = payload["prefix"].ToString(); // This is the entire path w/ prefix for the file set
            log.LogTrace($@"prefix: {prefix}");

            var filePrefix = prefix.Substring(prefix.LastIndexOf('/') + 1);
            log.LogTrace($@"filePrefix: {filePrefix}");

            var lockTable = await Helpers.GetLockTableAsync();
            if (!await ShouldProceedAsync(lockTable, prefix, filePrefix, log))
            {
                return req.CreateResponse(HttpStatusCode.OK);
            }

            var blobClient = storageAccount.CreateCloudBlobClient();
            var targetBlobs = await blobClient.ListBlobsAsync(WebUtility.UrlDecode(prefix));

            var customerName = filePrefix.Split('_').First().Split('-').Last();

            var errors = new List<string>();
            var filesToProcess = payload["fileTypes"].Values<string>();

            foreach (var blobDetails in targetBlobs)
            {
                var blob = await blobClient.GetBlobReferenceFromServerAsync(blobDetails.StorageUri.PrimaryUri);

                var fileParts = CustomerBlobAttributes.Parse(blob.Uri.AbsolutePath);
                if (!filesToProcess.Contains(fileParts.Filetype, StringComparer.OrdinalIgnoreCase))
                {
                    log.LogTrace($@"{blob.Name} skipped. Isn't in the list of file types to process ({string.Join(", ", filesToProcess)}) for bottler '{customerName}'");
                    continue;
                }

                var lowerFileType = fileParts.Filetype.ToLowerInvariant();
                log.LogInformation($@"Validating {lowerFileType}...");

                uint numColumns = 0;
                switch (lowerFileType)
                {
                    case @"type5":  // salestype
                        numColumns = 2;
                        break;
                    case @"type10": // mixedpack
                    case @"type4":  // shipfrom
                        numColumns = 3;
                        break;
                    case @"type1":  // channel
                    case @"type2":  // customer
                        numColumns = 4;
                        break;
                    case @"type9":  // itemdetail
                    case @"type3": // shipto
                        numColumns = 14;
                        break;
                    case @"type6": // salesdetail
                        numColumns = 15;
                        break;
                    case @"type8":  // product
                        numColumns = 21;
                        break;
                    case @"type7":  // sales
                        numColumns = 23;
                        break;
                    default:
                        throw new ArgumentOutOfRangeException(nameof(prefix), $@"Unhandled file type: {fileParts.Filetype}");
                }

                errors.AddRange(await ValidateCsvStructureAsync(blob, numColumns, lowerFileType));
            }
            try
            {
                await LockTableEntity.UpdateAsync(filePrefix, LockTableEntity.BatchState.Done, lockTable);
            }
            catch (StorageException)
            {
                log.LogWarning($@"That's weird. The lock for prefix {prefix} wasn't there. Shouldn't happen!");
                return req.CreateResponse(HttpStatusCode.OK);
            }

            if (errors.Any())
            {
                log.LogError($@"Errors found in batch {filePrefix}: {string.Join(@", ", errors)}");

                // move files to 'invalid-set' folder
                await MoveBlobsAsync(log, blobClient, targetBlobs, @"invalid-set");

                return req.CreateErrorResponse(HttpStatusCode.BadRequest, string.Join(@", ", errors));
            }
            else
            {
                // move these files to 'valid-set' folder
                await MoveBlobsAsync(log, blobClient, targetBlobs, @"valid-set");

                log.LogInformation($@"Set {filePrefix} successfully validated and queued for further processing.");

                return req.CreateResponse(HttpStatusCode.OK);
            }
        }

        private static async Task<bool> ShouldProceedAsync(CloudTable bottlerFilesTable, string prefix, string filePrefix, ILogger log)
        {
            try
            {
                var lockRecord = await LockTableEntity.GetLockRecordAsync(filePrefix, bottlerFilesTable);
                if (lockRecord?.State == LockTableEntity.BatchState.Waiting)
                {
                    // Update the lock record to mark it as in progress
                    lockRecord.State = LockTableEntity.BatchState.InProgress;
                    await bottlerFilesTable.ExecuteAsync(TableOperation.Replace(lockRecord));
                    return true;
                }
                else
                {
                    log.LogInformation($@"Validate for {prefix} skipped. State was {lockRecord?.State.ToString() ?? @"[null]"}.");
                }
            }
            catch (StorageException)
            {
                log.LogInformation($@"Validate for {prefix} skipped (StorageException. Somebody else picked it up already.");
            }

            return false;
        }

        private static async Task MoveBlobsAsync(ILogger log, CloudBlobClient blobClient, IEnumerable<IListBlobItem> targetBlobs, string folderName)
        {
            foreach (var b in targetBlobs)
            {
                var blobRef = await blobClient.GetBlobReferenceFromServerAsync(b.StorageUri.PrimaryUri);
                var sourceBlob = b.Container.GetBlockBlobReference(blobRef.Name);

                var targetBlob = blobRef.Container
                    .GetDirectoryReference($@"{folderName}")
                    .GetBlockBlobReference(Path.GetFileName(blobRef.Name));

                await targetBlob.StartCopyAsync(sourceBlob);

                while (targetBlob.CopyState.Status == CopyStatus.Pending)
                {
                    ;     // spinlock until the copy completes
                }

                var copySucceeded = targetBlob.CopyState.Status == CopyStatus.Success;
                if (!copySucceeded)
                {
                    log.LogError($@"Error copying {sourceBlob.Name} to {folderName} folder. Retrying once...");

                    await targetBlob.StartCopyAsync(sourceBlob);

                    while (targetBlob.CopyState.Status == CopyStatus.Pending)
                    {
                        ;     // spinlock until the copy completes
                    }

                    copySucceeded = targetBlob.CopyState.Status == CopyStatus.Success;
                    if (!copySucceeded)
                    {
                        log.LogError($@"Error retrying copy of {sourceBlob.Name} to {folderName} folder. File not moved.");
                    }
                }

                if (copySucceeded)
                {
#if DEBUG
                    try
                    {
#endif
                        await sourceBlob.DeleteAsync();
#if DEBUG
                    }
                    catch (StorageException ex)
                    {
                        log.LogError($@"Error deleting blob {sourceBlob.Name}", ex);
                    }
#endif
                }
            }
        }

        private static async Task<IEnumerable<string>> ValidateCsvStructureAsync(ICloudBlob blob, uint requiredNumberOfColumnsPerLine, string filetypeDescription)
        {
            var errs = new List<string>();
            try
            {
                using (var blobReader = new StreamReader(await blob.OpenReadAsync(new AccessCondition(), new BlobRequestOptions(), new OperationContext())))
                {
                    var fileAttributes = CustomerBlobAttributes.Parse(blob.Uri.AbsolutePath);

                    for (var lineNumber = 0; !blobReader.EndOfStream; lineNumber++)
                    {
                        var errorPrefix = $@"{filetypeDescription} file '{fileAttributes.Filename}' Record {lineNumber}";
                        var line = blobReader.ReadLine();
                        var fields = line.Split(',');
                        if (fields.Length != requiredNumberOfColumnsPerLine)
                        {
                            errs.Add($@"{errorPrefix} is malformed. Should have {requiredNumberOfColumnsPerLine} values; has {fields.Length}");
                            continue;
                        }

                        for (var i = 0; i < fields.Length; i++)
                        {
                            errorPrefix = $@"{errorPrefix} Field {i}";
                            var field = fields[i];
                            // each field must be enclosed in double quotes
                            if (field[0] != '"' || field.Last() != '"')
                            {
                                errs.Add($@"{errorPrefix}: value ({field}) is not enclosed in double quotes ("")");
                                continue;
                            }
                        }
                    }

                    // Validate file is UTF-8 encoded
                    if (!blobReader.CurrentEncoding.BodyName.Equals("utf-8", StringComparison.OrdinalIgnoreCase))
                    {
                        errs.Add($@"{blob.Name} is not UTF-8 encoded");
                    }
                }
            }
            catch (StorageException storEx)
            {
                SwallowStorage404(storEx);
            }
            return errs;
        }

        private static void SwallowStorage404(StorageException storEx)
        {
            var webEx = storEx.InnerException as WebException;
            if ((webEx.Response as HttpWebResponse)?.StatusCode == HttpStatusCode.NotFound)
            {
                // Ignore
            }
            else
            {
                throw storEx;
            }
        }
    }
}


<|start_filename|>AzureFunctions.v3/LockTableEntity.cs<|end_filename|>
﻿using System;
using System.Linq;
using System.Threading.Tasks;
using Microsoft.WindowsAzure.Storage;
using Microsoft.WindowsAzure.Storage.Table;

namespace FileValidation
{
    class LockTableEntity : TableEntity
    {
        public LockTableEntity() : base() { }

        public LockTableEntity(string prefix) : base(prefix, prefix) { }

        [IgnoreProperty]
        public string Prefix
        {
            get => this.PartitionKey;
            set
            {
                this.PartitionKey = value;
                this.RowKey = value;
            }
        }

        [IgnoreProperty]
        public BatchState State { get; set; } = BatchState.Waiting;

        public string DbState
        {
            get => this.State.ToString();
            set => this.State = (BatchState)Enum.Parse(typeof(BatchState), value);
        }

        public enum BatchState
        {
            Waiting, InProgress, Done
        }

        public static async Task<LockTableEntity> GetLockRecordAsync(string filePrefix, CloudTable customerFilesTable = null, CloudStorageAccount customerFilesTableStorageAccount = null)
        {
            customerFilesTable = customerFilesTable ?? await Helpers.GetLockTableAsync(customerFilesTableStorageAccount);

            return (await customerFilesTable.ExecuteQueryAsync(
                new TableQuery<LockTableEntity>()
                    .Where(TableQuery.GenerateFilterCondition(@"PartitionKey", QueryComparisons.Equal, filePrefix))))
                .SingleOrDefault();
        }

        public static async Task UpdateAsync(string filePrefix, BatchState state, CloudTable customerFilesTable = null, CloudStorageAccount customerFilesTableStorageAccount = null)
        {
            var entity = await GetLockRecordAsync(filePrefix, customerFilesTable);
            entity.State = state;

            customerFilesTable = customerFilesTable ?? await Helpers.GetLockTableAsync(customerFilesTableStorageAccount);

            await customerFilesTable.ExecuteAsync(TableOperation.Replace(entity));
        }
    }
}


<|start_filename|>AzureFunctions.v3.DurableEntities/BatchEntity.cs<|end_filename|>
﻿using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
using Microsoft.Azure.WebJobs;
using Microsoft.Azure.WebJobs.Extensions.DurableTask;
using Microsoft.Extensions.Logging;
using Newtonsoft.Json;

namespace FileValidation
{
    [JsonObject(MemberSerialization.OptIn)]
    public class BatchEntity : IBatchEntity
    {
        private readonly string _id;
        private readonly ILogger _logger;

        public BatchEntity(string id, ILogger logger)
        {
            _id = id;
            _logger = logger;
        }

        [JsonProperty]
        public List<string> ReceivedFileTypes { get; set; } = new List<string>();

        [FunctionName(nameof(BatchEntity))]
        public static Task Run([EntityTrigger]IDurableEntityContext ctx, ILogger logger) => ctx.DispatchAsync<BatchEntity>(ctx.EntityKey, logger);

        public async Task NewFile(string fileUri)
        {
            var newCustomerFile = CustomerBlobAttributes.Parse(fileUri);
            _logger.LogInformation($@"Got new file via event: {newCustomerFile.Filename}");
            this.ReceivedFileTypes.Add(newCustomerFile.Filetype);

            _logger.LogTrace($@"Actor '{_id}' got file '{newCustomerFile.Filetype}'");

            var filesStillWaitingFor = Helpers.GetExpectedFilesForCustomer().Except(this.ReceivedFileTypes);
            if (filesStillWaitingFor.Any())
            {
                _logger.LogInformation($@"Still waiting for more files... Still need {string.Join(", ", filesStillWaitingFor)} for customer {newCustomerFile.CustomerName}, batch {newCustomerFile.BatchPrefix}");
            }
            else
            {
                _logger.LogInformation(@"Got all the files! Moving on...");

                // call next step in functions with the prefix so it knows what to go grab
                await Helpers.DoValidationAsync($@"{newCustomerFile.ContainerName}/inbound/{newCustomerFile.BatchPrefix}", _logger);
            }
        }
    }

    public interface IBatchEntity
    {
        Task NewFile(string fileUri);
    }

}


<|start_filename|>AzureFunctions.v3.Durable/DurableContextExtensions.cs<|end_filename|>
﻿using Microsoft.Azure.WebJobs;
using Microsoft.Azure.WebJobs.Extensions.DurableTask;
using Microsoft.Extensions.Logging;
using Newtonsoft.Json.Linq;

namespace FileValidation
{
    static class DurableContextExtensions
    {
#if FUNCTIONS_V1
        public static void Log(this DurableOrchestrationContext context, ILogger log, string messsage, bool onlyIfNotReplaying = true)
        {
            if (!onlyIfNotReplaying || !context.IsReplaying)
            {
                log.LogWarning(messsage);
            }
        }

        public static void Log(this DurableOrchestrationClient _, ILogger log, string messsage) => log.LogWarning(messsage);

        public static JToken GetInputAsJson(this DurableActivityContextBase ctx) => ctx.GetInput<JToken>();

        public static JToken GetInputAsJson(this DurableOrchestrationContextBase ctx) => ctx.GetInput<JToken>();
#else
        public static void Log(this IDurableOrchestrationContext context, ILogger log, string messsage, bool onlyIfNotReplaying = true)
        {
            if (!onlyIfNotReplaying || !context.IsReplaying)
            {
                log.LogWarning(messsage);
            }
        }

        public static void Log(this IDurableClient _, ILogger log, string messsage) => log.LogWarning(messsage);

        public static JToken GetInputAsJson(this IDurableActivityContext ctx) => ctx.GetInput<JToken>();

        public static JToken GetInputAsJson(this IDurableOrchestrationContext ctx) => ctx.GetInput<JToken>();
#endif

    }
}


<|start_filename|>AzureFunctions.v3.Durable/FunctionValidateFileSet.cs<|end_filename|>
﻿using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using Microsoft.Azure.WebJobs;
using Microsoft.Azure.WebJobs.Extensions.DurableTask;
using Microsoft.Extensions.Logging;
using Microsoft.WindowsAzure.Storage;

namespace FileValidation
{
    public static class FunctionValidateFileSet
    {
        [FunctionName(@"ValidateFileSet")]
        public static async Task<bool> Run([ActivityTrigger] FilesetValidationRequest payload, ILogger log)
        {
            log.LogTrace(@"ValidateFileSet run.");
            if (!CloudStorageAccount.TryParse(Environment.GetEnvironmentVariable(@"CustomerBlobStorage"), out _))
            {
                throw new Exception(@"Can't create a storage account accessor from app setting connection string, sorry!");
            }

            var prefix = payload.Prefix; // This is the entire path w/ prefix for the file set

            return await Helpers.DoValidationAsync(prefix, log);
        }

    }

    public class FilesetValidationRequest
    {
        public string Prefix { get; set; }

        public IEnumerable<string> ExpectedFiles { get; set; }
    }

}


<|start_filename|>AzureFunctions.v3/CustomerBlobAttributes.cs<|end_filename|>
﻿using System;
using System.Globalization;
using System.Text.RegularExpressions;

namespace FileValidation
{
    public class CustomerBlobAttributes
    {
        static readonly Regex blobUrlRegexExtract = new Regex(@"^\S*/([^/]+)/inbound/((([^_]+)_([\d]+_[\d]+))_([\w]+))\.csv$", RegexOptions.Compiled | RegexOptions.IgnoreCase | RegexOptions.Singleline);

        static readonly Regex blobUrlRegexExtractWithSubfolder = new Regex(@"^\S*/([^/]+)/([^/]+)/((([^_]+)_([\d]+_[\d]+))_([\w]+))\.csv$", RegexOptions.Compiled | RegexOptions.IgnoreCase | RegexOptions.Singleline);

        public string FullUrl { get; private set; }
        public string Filename { get; private set; }
        public string BatchPrefix { get; private set; }
        public DateTime BatchDateTime { get; private set; }
        public string Filetype { get; private set; }
        public string CustomerName { get; private set; }
        public string ContainerName { get; private set; }
        public string Subfolder { get; private set; }

        public static CustomerBlobAttributes Parse(string fullUri, bool detectSubfolder = false)
        {
            if (detectSubfolder)
            {
                var regexMatch = blobUrlRegexExtractWithSubfolder.Match(fullUri);
                if (regexMatch.Success)
                {
                    return new CustomerBlobAttributes
                    {
                        FullUrl = regexMatch.Groups[0].Value,
                        ContainerName = regexMatch.Groups[1].Value,
                        Subfolder = regexMatch.Groups[2].Value,
                        Filename = regexMatch.Groups[3].Value,
                        BatchPrefix = regexMatch.Groups[4].Value,
                        CustomerName = regexMatch.Groups[5].Value,
                        BatchDateTime = DateTime.ParseExact(regexMatch.Groups[6].Value, @"yyyyMMdd_HHmm", CultureInfo.InvariantCulture),
                        Filetype = regexMatch.Groups[7].Value
                    };
                }
            }
            else
            {
                var regexMatch = blobUrlRegexExtract.Match(fullUri);
                if (regexMatch.Success)
                {
                    return new CustomerBlobAttributes
                    {
                        FullUrl = regexMatch.Groups[0].Value,
                        ContainerName = regexMatch.Groups[1].Value,
                        Filename = regexMatch.Groups[2].Value,
                        BatchPrefix = regexMatch.Groups[3].Value,
                        CustomerName = regexMatch.Groups[4].Value,
                        BatchDateTime = DateTime.ParseExact(regexMatch.Groups[5].Value, @"yyyyMMdd_HHmm", CultureInfo.InvariantCulture),
                        Filetype = regexMatch.Groups[6].Value
                    };
                }
            }

            return null;
        }
    }
}


<|start_filename|>AzureFunctions.Python/local.settings.sample.json<|end_filename|>
{
    "IsEncrypted": false,
    "Values": {
      "AzureWebJobsStorage": "<connection-string>",
      "DataStorage": "<connection-string>",
      "DataContainer": "data",
      "DataSubpath": "subpath",
      "FUNCTIONS_WORKER_RUNTIME": "python"
    }
}