using System;
using System.Collections.Generic;
using System.Data.SqlClient;
using System.Diagnostics;
using System.Linq;
using Microsoft.WindowsAzure.Storage;
using Microsoft.WindowsAzure.Storage.Table;
using SharedStorage;

namespace TableExample
{
    internal class Program
    {
        private static void Main(string[] args)
        {
            var account = CloudStorageAccount.Parse(StorageConst.StorageConnectionString);
            var client = account.CreateCloudTableClient();
            var table = client.GetTableReference(StorageConst.TableName);
            table.CreateIfNotExists();
            var batchOperation = new TableBatchOperation();
            var bachInsert = new List<Model>();
            
            GetFromDb(ref bachInsert);

            UploadToTableStorage(bachInsert, batchOperation, table);
            Console.ReadKey();
        }

        private static void UploadToTableStorage(List<Model> bachInsert, TableBatchOperation batchOperation, CloudTable table)
        {
            var groupedByPartition = bachInsert.GroupBy(partition => partition.PartitionKey);
            foreach (var groupedKey in groupedByPartition)
            {
                foreach (var value in groupedKey)
                {
                    TableOperation operation = TableOperation.Insert(value);
                    batchOperation.Add(operation);
                }
                table.ExecuteBatch(batchOperation);
                batchOperation.Clear();
                Console.WriteLine("Inserted");
            }
        }

        private static void GetFromDb(ref List<Model> bachInsert)
        {
            using (
                var connection =
                    new SqlConnection("Server=LukaszSzulc;Database=AdventureWorksDW2008R2_TP;Integrated Security = SSPI")
                )
            {
                connection.Open();
                var command =
                    new SqlCommand(
                        "SELECT  EmailAddress,FirstName,LastName,BirthDate,OrderQuantity,UnitPrice,SalesAmount FROM DimCustomer JOIN FactInternetSales ON DimCustomer.CustomerKey = FactInternetSales.CustomerKey",
                        connection);
                var reader = command.ExecuteReader();
                var counter = 0;
                while (reader.Read())
                {
                    var nosqlModel = new Model();

                    nosqlModel.PartitionKey = (string) reader[0];
                    nosqlModel.FirstName = reader[1].ToString();
                    nosqlModel.LastName = reader[2].ToString();
                    nosqlModel.BirthDate = reader[3].ToString();
                    nosqlModel.OrderQuantity = int.Parse(reader[4].ToString());
                    nosqlModel.SalesAmmount = double.Parse(reader[6].ToString());
                    nosqlModel.UnitPrice = double.Parse(reader[5].ToString());

                    bachInsert.Add(nosqlModel);
                }
            }
        }
    }
}