TEAMModelOS/TEAMModelOS.SDK/Module/AzureCosmosDBV3/AzureCosmosDBV3Repository.cs

using Microsoft.Azure.Cosmos;
using Microsoft.Azure.Cosmos.Linq;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.IO;
using System.Linq;
using System.Linq.Expressions;
using System.Net;
using System.Reflection;
using System.Text;
using System.Text.Json;
using System.Threading;
using System.Threading.Tasks;
using TEAMModelOS.SDK.Context.Attributes.Azure;
using TEAMModelOS.SDK.Context.Exception;
using TEAMModelOS.SDK.Helper.Common.CollectionHelper;
using TEAMModelOS.SDK.Helper.Common.ReflectorExtensions;
using TEAMModelOS.SDK.Module.AzureCosmosDB.Configuration;

namespace TEAMModelOS.SDK.Module.AzureCosmosDBV3
{
    public class CosmosModelInfo
    {
        public Container container { get; set; }
        public bool cache { get; set; }
        public bool monitor { get; set; } = false;
        public Type type { get; set; }
    }

    public class AzureCosmosDBV3Repository : IAzureCosmosDBV3Repository
    {

        private CosmosClient CosmosClient { get; set; }
        /// <summary>
        /// 线程安全的dict类型
        /// </summary>
        private Dictionary<string, CosmosModelInfo> DocumentCollectionDict { get; set; } = new Dictionary<string, CosmosModelInfo>();

        private string DatabaseId { get; set; }
        private int CollectionThroughput { get; set; }
        private Database database { get; set; }
        int pageSize = 200;
        private const string CacheCosmosPrefix = "cosmos:";
        private string[] ScanModel { get; set; }
        private const int timeoutSeconds = 86400;

        private string leaseId = "AleaseContainer";
        public AzureCosmosDBV3Repository(AzureCosmosDBOptions options, CosmosSerializer cosmosSerializer)
        {
            try
            {
                if (!string.IsNullOrEmpty(options.ConnectionString))
                {
                    CosmosClient = CosmosDBV3ClientSingleton.getInstance(options.ConnectionString, options.ConnectionKey, cosmosSerializer).GetCosmosDBClient();
                }
                else
                {
                    throw new BizException("请设置正确的AzureCosmosDB数据库配置信息！");
                }
                DatabaseId = options.Database;
                CollectionThroughput = options.CollectionThroughput;
                ScanModel = options.ScanModel;
                // InitializeDatabase().GetAwaiter().GetResult();
            }
            catch (CosmosException e)
            {
                // Dispose(true);
                throw new BizException(e.Message, 500, e.StackTrace);
            }
        }
        public AzureCosmosDBV3Repository(AzureCosmosDBOptions options)
        {
            try
            {
                if (!string.IsNullOrEmpty(options.ConnectionString))
                {
                    CosmosClient = CosmosDBV3ClientSingleton.getInstance(options.ConnectionString, options.ConnectionKey, null).GetCosmosDBClient();

                }
                else
                {
                    throw new BizException("请设置正确的AzureCosmosDB数据库配置信息！");
                }
                DatabaseId = options.Database;
                CollectionThroughput = options.CollectionThroughput;
                ScanModel = options.ScanModel;
                // InitializeDatabase().GetAwaiter().GetResult();
            }
            catch (CosmosException e)
            {
                //   Dispose(true);
                throw new BizException(e.Message, 500, e.StackTrace);
            }
        }


        public async Task InitializeDatabase()
        {
            try
            {
                database = await CosmosClient.CreateDatabaseIfNotExistsAsync(DatabaseId, CollectionThroughput);
                FeedIterator<ContainerProperties> resultSetIterator = database.GetContainerQueryIterator<ContainerProperties>();
                while (resultSetIterator.HasMoreResults)
                {
                    foreach (ContainerProperties container in await resultSetIterator.ReadNextAsync())
                    {
                        DocumentCollectionDict.TryAdd(container.Id, new CosmosModelInfo { container = database.GetContainer(container.Id), cache = false, monitor = false });
                    }
                }
                bool isMonitor = false;
                //获取数据库所有的表
                List<Type> types = ReflectorExtensions.GetAllTypeAsAttribute<CosmosDBAttribute>(ScanModel);
                foreach (Type type in types)
                {
                    string PartitionKey = GetPartitionKey(type);
                    string CollectionName = "";
                    int RU = 0;
                    bool cache = false;
                    bool monitor = false;
                    IEnumerable<CosmosDBAttribute> attributes = type.GetCustomAttributes<CosmosDBAttribute>(true);
                    if (!string.IsNullOrEmpty(attributes.First<CosmosDBAttribute>().Name))
                    {
                        CollectionName = attributes.First<CosmosDBAttribute>().Name;
                    }
                    else
                    {
                        CollectionName = type.Name;
                    }
                    if (attributes.First<CosmosDBAttribute>().Cache)
                    {
                        cache = attributes.First<CosmosDBAttribute>().Cache;
                    }
                    if (attributes.First<CosmosDBAttribute>().Monitor)
                    {
                        monitor = attributes.First<CosmosDBAttribute>().Monitor;
                        if (monitor)
                        {
                            isMonitor = true;
                        }
                    }
                    //else
                    //{
                    //    cache = false;
                    //}
                    if (attributes.First<CosmosDBAttribute>().RU > 400)
                    {
                        RU = attributes.First<CosmosDBAttribute>().RU;
                    }
                    else
                    {
                        RU = CollectionThroughput;
                    }
                    //如果表存在于数据则检查RU是否变动，如果不存在则执行创建DocumentCollection
                    if (DocumentCollectionDict.TryGetValue(CollectionName, out CosmosModelInfo cosmosModelInfo))
                    { //更新RU

                        cosmosModelInfo.cache = cache;
                        Container container = CosmosClient.GetDatabase(DatabaseId).GetContainer(cosmosModelInfo.container.Id);
                        int? throughputResponse = await container.ReadThroughputAsync();
                        if (throughputResponse < RU)
                        {
                            await CosmosClient.GetDatabase(DatabaseId).GetContainer(cosmosModelInfo.container.Id).ReplaceThroughputAsync(RU);
                        }
                        DocumentCollectionDict[CollectionName] = new CosmosModelInfo { container = container, cache = cache, monitor = monitor, type = type };
                    }
                    else
                    {
                        ContainerProperties containerProperties = new ContainerProperties { Id = CollectionName };

                        if (!string.IsNullOrEmpty(PartitionKey))
                        {
                            containerProperties.PartitionKeyPath = "/" + PartitionKey;
                        }
                        if (RU > CollectionThroughput)
                        {
                            CollectionThroughput = RU;
                        }
                        Container containerWithConsistentIndexing = await database.CreateContainerIfNotExistsAsync(containerProperties, throughput: CollectionThroughput);
                        DocumentCollectionDict.TryAdd(CollectionName, new CosmosModelInfo { container = containerWithConsistentIndexing, cache = cache, monitor = monitor, type = type });
                    }
                }
                if (isMonitor)
                {
                    ContainerProperties leaseProperties = new ContainerProperties { Id = leaseId, PartitionKeyPath = "/id" };
                    Container leaseContainer = await database.CreateContainerIfNotExistsAsync(leaseProperties, throughput: CollectionThroughput);
                    DocumentCollectionDict.TryAdd(leaseId, new CosmosModelInfo { container = leaseContainer, cache = false, monitor = false });
                }
            }
            catch (CosmosException e)
            {
                throw new BizException(e.Message, 500, e.StackTrace);
            }
        }

        public Dictionary<string, CosmosModelInfo> GetCosmosModelInfo()
        {
            return this.DocumentCollectionDict;
        }



        private string GetPartitionKey<T>()
        {
            Type type = typeof(T);
            return GetPartitionKey(type);
        }
        private string GetPartitionKey(Type type)
        {
            PropertyInfo[] properties = type.GetProperties();
            List<PropertyInfo> attrProperties = new List<PropertyInfo>();
            foreach (PropertyInfo property in properties)
            {
                if (property.Name.Equals("PartitionKey"))
                {
                    attrProperties.Add(property);
                    break;
                }
                object[] attributes = property.GetCustomAttributes(true);
                foreach (object attribute in attributes) //2.通过映射，找到成员属性上关联的特性类实例，
                {
                    if (attribute is PartitionKeyAttribute)
                    {
                        attrProperties.Add(property);
                    }
                }
            }
            if (attrProperties.Count <= 0)
            {
                throw new BizException(type.Name + "has no PartitionKey !");
            }
            else
            {
                if (attrProperties.Count == 1)
                {
                    return attrProperties[0].Name;
                }
                else { throw new BizException("PartitionKey can only be single!"); }
            }
        }


        private async Task<CosmosModelInfo> InitializeCollection<T>()
        {
            Type type = typeof(T);
            string partitionKey = GetPartitionKey<T>();
            string CollectionName;
            IEnumerable<CosmosDBAttribute> attributes = type.GetCustomAttributes<CosmosDBAttribute>(true);
            if (!string.IsNullOrEmpty(attributes.First<CosmosDBAttribute>().Name))
            {
                CollectionName = attributes.First<CosmosDBAttribute>().Name;
            }
            else
            {
                CollectionName = type.Name;
            }
            return await InitializeCollection(CollectionName, partitionKey);
        }

        private async Task<CosmosModelInfo> InitializeCollection(string CollectionName, string PartitionKey)
        {
            /////内存中已经存在这个表则直接返回
            if (DocumentCollectionDict.TryGetValue(CollectionName, out CosmosModelInfo cosmosModelInfo))
            {
                return cosmosModelInfo;
            }///如果没有则尝试默认创建
            else
            {
                ContainerProperties containerProperties = new ContainerProperties { Id = CollectionName };
                if (!string.IsNullOrEmpty(PartitionKey))
                {
                    containerProperties.PartitionKeyPath = "/" + PartitionKey;
                }
                Container containerWithConsistentIndexing = await database.CreateContainerIfNotExistsAsync(containerProperties, throughput: CollectionThroughput);
                CosmosModelInfo cosmosModel = new CosmosModelInfo { container = containerWithConsistentIndexing, cache = false };
                DocumentCollectionDict.TryAdd(CollectionName, cosmosModel);
                return cosmosModel;
            }
        }


        public async Task<List<IdPk>> DeleteAll<T>(List<IdPk> ids) where T : ID
        {
            CosmosModelInfo container = await InitializeCollection<T>();
            //string partitionKey = GetPartitionKey<T>();
            //await Task.Run(() => Parallel.ForEach(ids, new ParallelOptions { MaxDegreeOfParallelism = 2 }, (item) =>
            //{
            //    Task.WaitAll(DeleteAsync<T>(item.Value, item.Key));
            //}));

            List<IdPk> idPks = new List<IdPk>();
            int pages = (int)Math.Ceiling((double)ids.Count / pageSize);
            Stopwatch stopwatch = Stopwatch.StartNew();
            for (int i = 0; i < pages; i++)
            {
                List<IdPk> lists = ids.Skip((i) * pageSize).Take(pageSize).ToList();
                List<Task> tasks = new List<Task>(lists.Count);
                lists.ForEach(item =>
                {
                    tasks.Add(container.container.DeleteItemStreamAsync(item.id, new PartitionKey(item.pk))
                        .ContinueWith((Task<ResponseMessage> task) =>
                        {
                            using (ResponseMessage response = task.Result)
                            {
                                idPks.Add(new IdPk { id = item.id, pk = item.pk.ToString(), StatusCode = response.StatusCode });
                                //    if (!response.IsSuccessStatusCode)
                                //    {
                                //    }
                            }
                        }
                        ));
                });
                await Task.WhenAll(tasks);
                if (container.cache && RedisHelper.Instance != null)
                {
                    lists.ForEach(async x => {
                        await RedisHelper.HDelAsync(CacheCosmosPrefix + container.container.Id, x.id);
                    });
                }
            }
            stopwatch.Stop();
            return idPks;
        }
        public async Task<List<IdPk>> DeleteAll<T>(Dictionary<string, object> dict) where T : ID
        {
            List<T> list = await FindByDict<T>(dict);
            return await DeleteAll(list);
        }
        public async Task<List<IdPk>> DeleteAll<T>(List<T> enyites) where T : ID
        {

            List<IdPk> idPks = new List<IdPk>();
            CosmosModelInfo container = await InitializeCollection<T>();
            string pk = GetPartitionKey<T>();
            Type type = typeof(T);
            int pages = (int)Math.Ceiling((double)enyites.Count / pageSize);
            Stopwatch stopwatch = Stopwatch.StartNew();
            for (int i = 0; i < pages; i++)
            {
                List<T> lists = enyites.Skip((i) * pageSize).Take(pageSize).ToList();
                List<KeyValuePair<PartitionKey, string>> itemsToInsert = new List<KeyValuePair<PartitionKey, string>>();
                lists.ForEach(x =>
                {
                    object o = type.GetProperty(pk).GetValue(x, null);
                    KeyValuePair<PartitionKey, string> keyValue = new KeyValuePair<PartitionKey, string>(new PartitionKey(o.ToString()), x.id);
                    itemsToInsert.Add(keyValue);
                });

                List<Task> tasks = new List<Task>(lists.Count);
                itemsToInsert.ForEach(item =>
                {
                    tasks.Add(container.container.DeleteItemStreamAsync(item.Value, item.Key)
                    .ContinueWith((Task<ResponseMessage> task) =>
                    {
                        using (ResponseMessage response = task.Result)
                        {

                            idPks.Add(new IdPk { id = item.Value, pk = item.Key.ToString(), StatusCode = response.StatusCode });

                        }
                    }
                    ));
                });
                await Task.WhenAll(tasks); if (container.cache && RedisHelper.Instance != null)
                {
                    lists.ForEach(async x => {
                        await RedisHelper.HDelAsync(CacheCosmosPrefix + container.container.Id, x.id);
                    });
                }
            }
            stopwatch.Stop();
            return idPks;
        }

        public async Task<IdPk> DeleteAsync<T>(IdPk idPk) where T : ID
        {
            return await DeleteAsync<T>(idPk.id, idPk.pk);
        }
        public async Task<IdPk> DeleteAsync<T>(string id, string pk) where T : ID
        {
            CosmosModelInfo container = await InitializeCollection<T>();
            ResponseMessage response = await container.container.DeleteItemStreamAsync(id: id, partitionKey: new PartitionKey(pk));
            if (container.cache && RedisHelper.Instance != null)
            {
                await RedisHelper.HDelAsync(CacheCosmosPrefix + container.container.Id, id);
            }
            return new IdPk { id = id, pk = pk, StatusCode = response.StatusCode };
        }

        public async Task<IdPk> DeleteAsync<T>(T entity) where T : ID
        {
            CosmosModelInfo container = await InitializeCollection<T>();
            string partitionKey = GetPartitionKey<T>();
            Type type = typeof(T);
            object o = type.GetProperty(partitionKey).GetValue(entity, null);
            ResponseMessage response = await container.container.DeleteItemStreamAsync(id: entity.id, partitionKey: new PartitionKey(o.ToString()));
            if (container.cache && RedisHelper.Instance != null)
            {
                await RedisHelper.HDelAsync(CacheCosmosPrefix + container.container.Id, entity.id);
            }
            return new IdPk { id = entity.id, pk = o.ToString(), StatusCode = response.StatusCode };

        }
        //public async Task<T> DeleteAsync<T>(string id) where T : ID
        //{
        //    Container container = await InitializeCollection<T>();
        //    ItemResponse<T> response = await container.DeleteItemAsync<T>(id: id, partitionKey: new PartitionKey(GetPartitionKey<T>()));
        //    return response.Resource;
        //}

        public async Task<List<T>> FindAll<T>(List<string> propertys = null) where T : ID
        {
            CosmosModelInfo container = await InitializeCollection<T>();
            StringBuilder sql;
            sql = SQLHelperParametric.GetSQLSelect(propertys);
            CosmosDbQuery cosmosDbQuery = new CosmosDbQuery { QueryText = sql.ToString() };
            FeedIterator<T> query = container.container.GetItemQueryIterator<T>(queryDefinition: cosmosDbQuery.CosmosQueryDefinition);
            return await ResultsFromFeedIterator(query);

        }
        private async Task<List<T>> ResultsFromFeedIterator<T>(FeedIterator<T> query, int? maxItemCount = null)
        {
            List<T> results = new List<T>();
            while (query.HasMoreResults)
            {
                foreach (T t in await query.ReadNextAsync())
                {
                    results.Add(t);
                    if (results.Count == maxItemCount)
                    {
                        return results;
                    }
                }
            }
            return results;
        }
        private async Task<List<T>> ResultsFromFeedIterator<T>(FeedIterator<T> query, Func<List<T>, Task> batchAction, int itemsPerPage)
        {
            List<T> results = new List<T>();
            while (query.HasMoreResults)
            {
                if (results.Count() >= itemsPerPage)
                {
                    await batchAction(results);
                    results.Clear();
                }

                results.AddRange(await query.ReadNextAsync());
            }

            if (results.Count() > 0)
            {
                await batchAction(results);
                results.Clear();
            }

            return results;
        }
        public async Task<List<dynamic>> FindByDict(string CollectionName, Dictionary<string, object> dict, string partitionKey = null, List<string> propertys = null)
        {
            if (DocumentCollectionDict.TryGetValue(CollectionName, out CosmosModelInfo container))
            {
                //StringBuilder sql = new StringBuilder("select value(c) from c");
                //SQLHelper.GetSQL(dict, ref sql);
                //CosmosDbQuery cosmosDbQuery = new CosmosDbQuery, int itemsPerPage = -1, int? 
                //{
                //    QueryText = sql.ToString()

                //};
                StringBuilder sql;
                sql = SQLHelperParametric.GetSQLSelect(propertys);
                CosmosDbQuery cosmosDbQuery = SQLHelperParametric.GetSQL(dict, sql);
                QueryRequestOptions queryRequestOptions = GetDefaultQueryRequestOptions(itemsPerPage: GetEffectivePageSize(-1, null));
                FeedIterator<dynamic> query = container.container.GetItemQueryIterator<dynamic>(queryDefinition: cosmosDbQuery.CosmosQueryDefinition, requestOptions: queryRequestOptions);
                return await ResultsFromFeedIterator(query);
            }
            else
            {
                throw new BizException("CollectionName named:" + CollectionName + " dose not exsit in Database!");
            }

        }

        public async Task<List<dynamic>> FindCountByDict(string CollectionName, Dictionary<string, object> dict, string partitionKey = null)
        {
            if (DocumentCollectionDict.TryGetValue(CollectionName, out CosmosModelInfo container))
            {
                dict.Remove("@CURRPAGE");
                dict.Remove("@PAGESIZE");
                dict.Remove("@ASC");
                dict.Remove("@DESC");
                StringBuilder sql = new StringBuilder("select  value count(c)  from c");
                CosmosDbQuery cosmosDbQuery = SQLHelperParametric.GetSQL(dict, sql);
                QueryRequestOptions queryRequestOptions = GetDefaultQueryRequestOptions(itemsPerPage: GetEffectivePageSize(-1, null));
                FeedIterator<dynamic> query = container.container.GetItemQueryIterator<dynamic>(queryDefinition: cosmosDbQuery.CosmosQueryDefinition, requestOptions: queryRequestOptions);
                return await ResultsFromFeedIterator(query);
            }
            else
            {
                throw new BizException("CollectionName named:" + CollectionName + " dose not exsit in Database!");
            }
        }

        public async Task<List<T>> FindByParams<T>(Dictionary<string, object> dict, string partitionKey = null, List<string> propertys = null) where T : ID
        {
            return await FindByDict<T>(dict, partitionKey, propertys);
        }
        public async Task<List<T>> FindByDict<T>(Dictionary<string, object> dict, string partitionKey = null, List<string> propertys = null) where T : ID
        {
            StringBuilder sql;
            sql = SQLHelperParametric.GetSQLSelect(propertys);

            CosmosDbQuery cosmosDbQuery = SQLHelperParametric.GetSQL(dict, sql);
            QueryRequestOptions queryRequestOptions = GetDefaultQueryRequestOptions(itemsPerPage: GetEffectivePageSize(-1, null));
            return await ResultsFromQueryAndOptions<T>(cosmosDbQuery, queryRequestOptions);
        }


        private async Task<List<T>> ResultsFromQueryAndOptions<T>(CosmosDbQuery cosmosDbQuery, QueryRequestOptions queryOptions)
        {
            CosmosModelInfo container = await InitializeCollection<T>();
            FeedIterator<T> query = container.container.GetItemQueryIterator<T>(
                queryDefinition: cosmosDbQuery.CosmosQueryDefinition,
                requestOptions: queryOptions);

            return await ResultsFromFeedIterator(query);
        }
        private int GetEffectivePageSize(int itemsPerPage, int? maxItemCount)
        {
            return itemsPerPage == -1 ? maxItemCount ?? itemsPerPage : Math.Min(maxItemCount ?? itemsPerPage, itemsPerPage);
        }
        private QueryRequestOptions GetDefaultQueryRequestOptions(int? itemsPerPage = null,
        int? maxBufferedItemCount = null,
        int? maxConcurrency = null)
        {
            QueryRequestOptions queryRequestOptions = new QueryRequestOptions
            {
                MaxItemCount = itemsPerPage == -1 ? 1000 : itemsPerPage,
                MaxBufferedItemCount = maxBufferedItemCount ?? 100,
                MaxConcurrency = maxConcurrency ?? 50
            };

            return queryRequestOptions;
        }
        private async Task<List<T>> ResultsFromQueryAndOptions<T>(CosmosDbQuery cosmosDbQuery, Func<List<T>, Task> batchAction, QueryRequestOptions queryOptions)
        {
            CosmosModelInfo container = await InitializeCollection<T>();
            FeedIterator<T> query = container.container.GetItemQueryIterator<T>(
                queryDefinition: cosmosDbQuery.CosmosQueryDefinition,
                requestOptions: queryOptions);

            return await ResultsFromFeedIterator(query, batchAction, queryOptions.MaxItemCount ?? 0);
        }



        private QueryRequestOptions GetQueryRequestOptions(int itemsPerPage)
        {
            QueryRequestOptions queryRequestOptions = new QueryRequestOptions
            {
                MaxItemCount = itemsPerPage
            };

            return queryRequestOptions;
        }
        public async Task<List<T>> FindLinq<T>(Expression<Func<T, bool>> query = null, Expression<Func<T, object>> order = null, bool isDesc = false) where T : ID
        {
            //QueryRequestOptions queryRequestOptions = GetQueryRequestOptions(itemsPerPage);
            QueryRequestOptions queryRequestOptions = GetDefaultQueryRequestOptions(itemsPerPage: GetEffectivePageSize(-1, null));
            FeedIterator<T> feedIterator;
            CosmosModelInfo container = await InitializeCollection<T>();

            if (query == null)
            {
                if (order != null)
                {
                    if (isDesc)
                    {
                        feedIterator = container.container
                       .GetItemLinqQueryable<T>(requestOptions: queryRequestOptions).OrderByDescending(order)
                       .ToFeedIterator();
                    }
                    else
                    {
                        feedIterator = container.container
                       .GetItemLinqQueryable<T>(requestOptions: queryRequestOptions).OrderBy(order)
                       .ToFeedIterator();
                    }
                }
                else
                {
                    feedIterator = container.container
                   .GetItemLinqQueryable<T>(requestOptions: queryRequestOptions)
                   .ToFeedIterator();
                }
            }
            else
            {
                if (order != null)
                {
                    if (isDesc)
                    {
                        feedIterator = container.container
                        .GetItemLinqQueryable<T>(requestOptions: queryRequestOptions)
                        .Where(query).OrderByDescending(order)
                        .ToFeedIterator();
                    }
                    else
                    {
                        feedIterator = container.container
                    .GetItemLinqQueryable<T>(requestOptions: queryRequestOptions)
                    .Where(query).OrderBy(order)
                    .ToFeedIterator();
                    }
                }
                else
                {
                    feedIterator = container.container
                    .GetItemLinqQueryable<T>(requestOptions: queryRequestOptions)
                    .Where(query)
                    .ToFeedIterator();
                }
            }
            return await ResultsFromFeedIterator<T>(feedIterator);
        }

        public async Task<List<T>> FindSQL<T>(string sql, Dictionary<string, object> Parameters = null) where T : ID
        {
            CosmosModelInfo container = await InitializeCollection<T>();
            QueryRequestOptions queryOptions = GetQueryRequestOptions(GetEffectivePageSize(-1, null));
            if (Parameters != null)
            {
                CosmosDbQuery cosmosDbQuery = new CosmosDbQuery
                {
                    QueryText = sql,
                    Parameters = Parameters
                };
                FeedIterator<T> feedIterator = container.container
                .GetItemQueryIterator<T>(cosmosDbQuery.CosmosQueryDefinition, requestOptions: queryOptions);
                return await ResultsFromFeedIterator(feedIterator);
            }
            else
            {
                QueryDefinition queryDefinition = new QueryDefinition(sql);
                return await ResultsFromFeedIterator<T>(container.container.GetItemQueryIterator<T>(queryDefinition));
            }
        }



        public async Task<T> Save<T>(T entity) where T : ID
        {
            try
            {
                CosmosModelInfo container = await InitializeCollection<T>();
                ItemResponse<T> response = await container.container.CreateItemAsync<T>(entity);
                if (container.cache && RedisHelper.Instance != null)
                {
                    if (!RedisHelper.Exists(CacheCosmosPrefix + container.container.Id))
                    {
                        await RedisHelper.HSetAsync(CacheCosmosPrefix + container.container.Id, entity.id, entity);
                    }
                    else
                    {
                        await RedisHelper.HSetAsync(CacheCosmosPrefix + container.container.Id, entity.id, entity);
                        await RedisHelper.ExpireAsync(CacheCosmosPrefix + container.container.Id, timeoutSeconds);
                    }

                }
                return response.Resource;
            }
            catch (Exception e)
            {
                throw new BizException(e.Message);
            }
        }

        public async Task<List<T>> SaveAll<T>(List<T> enyites) where T : ID
        {
            int pages = (int)Math.Ceiling((double)enyites.Count / pageSize);
            CosmosModelInfo container = await InitializeCollection<T>();
            bool flag = false;
            if (RedisHelper.Exists(CacheCosmosPrefix + container.container.Id))
            {
                flag = true;
            }
            string pk = GetPartitionKey<T>();
            Type type = typeof(T);
            Stopwatch stopwatch = Stopwatch.StartNew();
            for (int i = 0; i < pages; i++)
            {
                List<T> lists = enyites.Skip((i) * pageSize).Take(pageSize).ToList();
                List<KeyValuePair<PartitionKey, Stream>> itemsToInsert = new List<KeyValuePair<PartitionKey, Stream>>();
                lists.ForEach(async x =>
                {
                    MemoryStream stream = new MemoryStream();
                    await JsonSerializer.SerializeAsync(stream, x, new JsonSerializerOptions { IgnoreNullValues = true });
                    object o = type.GetProperty(pk).GetValue(x, null);
                    KeyValuePair<PartitionKey, Stream> keyValue = new KeyValuePair<PartitionKey, Stream>(new PartitionKey(o.ToString()), stream);
                    itemsToInsert.Add(keyValue);
                });
                List<Task> tasks = new List<Task>(lists.Count);
                itemsToInsert.ForEach(item =>
                {
                    tasks.Add(container.container.CreateItemStreamAsync(item.Value, item.Key)
                        .ContinueWith((Task<ResponseMessage> task) =>
                        {
                            using (ResponseMessage response = task.Result)
                            {
                                if (!response.IsSuccessStatusCode)
                                {
                                }
                            }
                        }
                        ));
                });
                await Task.WhenAll(tasks);
                if (container.cache && RedisHelper.Instance != null)
                {
                    lists.ForEach(async x => {
                        await RedisHelper.HSetAsync(CacheCosmosPrefix + container.container.Id, x.id, x);
                    });
                }
            }
            if (container.cache && RedisHelper.Instance != null && !flag)
            {
                await RedisHelper.ExpireAsync(CacheCosmosPrefix + container.container.Id, timeoutSeconds);
            }
            stopwatch.Stop();
            return enyites;
        }
        public async Task<T> SaveOrUpdate<T>(T entity) where T : ID
        {
            CosmosModelInfo container = await InitializeCollection<T>();
            ItemResponse<T> response = await container.container.UpsertItemAsync(item: entity);
            if (container.cache && RedisHelper.Instance != null)
            {
                if (!RedisHelper.Exists(CacheCosmosPrefix + container.container.Id))
                {
                    await RedisHelper.HSetAsync(CacheCosmosPrefix + container.container.Id, entity.id, entity);
                }
                else
                {
                    await RedisHelper.HSetAsync(CacheCosmosPrefix + container.container.Id, entity.id, entity);
                    await RedisHelper.ExpireAsync(CacheCosmosPrefix + container.container.Id, timeoutSeconds);
                }
            }
            return response.Resource;
        }
        public async Task<List<T>> SaveOrUpdateAll<T>(List<T> enyites) where T : ID
        {
            //await Task.Run(() => Parallel.ForEach(entities, new ParallelOptions { MaxDegreeOfParallelism = 2 }, (item) =>
            //{
            //    Task.WaitAll(Update(item));
            //}));

            int pages = (int)Math.Ceiling((double)enyites.Count / pageSize);
            CosmosModelInfo container = await InitializeCollection<T>();
            bool flag = false;
            if (RedisHelper.Exists(CacheCosmosPrefix + container.container.Id))
            {
                flag = true;
            }
            string pk = GetPartitionKey<T>();
            Type type = typeof(T);
            Stopwatch stopwatch = Stopwatch.StartNew();
            for (int i = 0; i < pages; i++)
            {
                List<T> lists = enyites.Skip((i) * pageSize).Take(pageSize).ToList();
                List<KeyValuePair<PartitionKey, Stream>> itemsToInsert = new List<KeyValuePair<PartitionKey, Stream>>();
                lists.ForEach(async x =>
                {
                    MemoryStream stream = new MemoryStream();
                    await JsonSerializer.SerializeAsync(stream, x, new JsonSerializerOptions { IgnoreNullValues = true });
                    object o = type.GetProperty(pk).GetValue(x, null);
                    KeyValuePair<PartitionKey, Stream> keyValue = new KeyValuePair<PartitionKey, Stream>(new PartitionKey(o.ToString()), stream);
                    itemsToInsert.Add(keyValue);
                });
                List<Task> tasks = new List<Task>(lists.Count);
                itemsToInsert.ForEach(item =>
                {
                    tasks.Add(container.container.UpsertItemStreamAsync(item.Value, item.Key)
                        .ContinueWith((Task<ResponseMessage> task) =>
                        {
                            //using (ResponseMessage response = task.Result)
                            //{
                            //    if (!response.IsSuccessStatusCode)
                            //    {
                            //    }
                            //}
                        }
                        ));
                });
                await Task.WhenAll(tasks);
                if (container.cache && RedisHelper.Instance != null)
                {
                    lists.ForEach(async x => {
                        await RedisHelper.HSetAsync(CacheCosmosPrefix + container.container.Id, x.id, x);
                    });
                }
            }
            if (container.cache && RedisHelper.Instance != null && !flag)
            {
                await RedisHelper.ExpireAsync(CacheCosmosPrefix + container.container.Id, timeoutSeconds);
            }
            stopwatch.Stop();
            return enyites;
        }

        public async Task<T> Update<T>(T entity) where T : ID
        {
            CosmosModelInfo container = await InitializeCollection<T>();
            string pk = GetPartitionKey<T>();
            object o = typeof(T).GetProperty(pk).GetValue(entity, null);
            ItemResponse<T> response = await container.container.ReplaceItemAsync(entity, entity.id, new PartitionKey(o.ToString()));
            if (container.cache && RedisHelper.Instance != null)
            {
                if (!RedisHelper.Exists(CacheCosmosPrefix + container.container.Id))
                {
                    await RedisHelper.HSetAsync(CacheCosmosPrefix + container.container.Id, entity.id, entity);
                }
                else
                {
                    await RedisHelper.HSetAsync(CacheCosmosPrefix + container.container.Id, entity.id, entity);
                    await RedisHelper.ExpireAsync(CacheCosmosPrefix + container.container.Id, timeoutSeconds);
                }
            }
            return response.Resource;
        }


        internal class Item
        {
            public string id { get; set; }
            public string pk { get; set; }
            public MemoryStream stream { get; set; }
        }

        public async Task<List<T>> UpdateAll<T>(List<T> enyites) where T : ID
        {
            //await Task.Run(() => Parallel.ForEach(entities, new ParallelOptions { MaxDegreeOfParallelism = 2 }, (item) =>
            //{
            //    Task.WaitAll(Update(item));
            //}));

            int pages = (int)Math.Ceiling((double)enyites.Count / pageSize);
            CosmosModelInfo container = await InitializeCollection<T>();
            bool flag = false;
            if (RedisHelper.Exists(CacheCosmosPrefix + container.container.Id))
            {
                flag = true;
            }
            string pk = GetPartitionKey<T>();
            Type type = typeof(T);
            Stopwatch stopwatch = Stopwatch.StartNew();
            for (int i = 0; i < pages; i++)
            {
                List<T> lists = enyites.Skip((i) * pageSize).Take(pageSize).ToList();
                List<Item> itemsToInsert = new List<Item>();
                lists.ForEach(async x =>
                {
                    MemoryStream stream = new MemoryStream();
                    await JsonSerializer.SerializeAsync(stream, x, new JsonSerializerOptions { IgnoreNullValues = true });
                    object o = type.GetProperty(pk).GetValue(x, null);
                    Item keyValue = new Item { id = x.id, pk = o.ToString(), stream = stream };
                    itemsToInsert.Add(keyValue);
                });
                List<Task> tasks = new List<Task>(lists.Count);
                itemsToInsert.ForEach(item =>
                {
                    tasks.Add(container.container.ReplaceItemStreamAsync(item.stream, item.id, new PartitionKey(item.pk))
                        .ContinueWith((Task<ResponseMessage> task) =>
                        {
                            //using (ResponseMessage response = task.Result)
                            //{
                            //    if (!response.IsSuccessStatusCode)
                            //    {
                            //    }
                            //}
                        }
                        ));
                });
                await Task.WhenAll(tasks);
                if (container.cache && RedisHelper.Instance != null)
                {
                    lists.ForEach(async x => {
                        await RedisHelper.HSetAsync(CacheCosmosPrefix + container.container.Id, x.id, x);
                    });
                }
            }
            if (container.cache && RedisHelper.Instance != null && !flag)
            {
                await RedisHelper.ExpireAsync(CacheCosmosPrefix + container.container.Id, timeoutSeconds);
            }
            stopwatch.Stop();
            return enyites;
        }

        //public void Dispose()
        //{
        //    Dispose(true);
        //}
        //protected virtual void Dispose(bool disposing)
        //{
        //    if (disposing)
        //    {
        //        CosmosClient?.Dispose();
        //    }
        //}
        private async Task<T> FindByIdAsSql<T>(string id) where T : ID
        {
            CosmosModelInfo container = await InitializeCollection<T>();
            CosmosDbQuery cosmosDbQuery = new CosmosDbQuery
            {
                QueryText = @"SELECT *
                            FROM    c 
                            WHERE   c.id = @id",
                Parameters = new Dictionary<string, object>
                 {

                    { "@id",id}
                }
            };
            FeedIterator<T> feedIterator = container.container
               .GetItemQueryIterator<T>(cosmosDbQuery.CosmosQueryDefinition);
            return (await ResultsFromFeedIterator(feedIterator)).SingleOrDefault();
        }


        public async Task<T> FindByIdPk<T>(string id, string pk) where T : ID
        {
            CosmosModelInfo container = await InitializeCollection<T>();
            ItemResponse<T> response = await container.container.ReadItemAsync<T>(id: id, partitionKey: new PartitionKey(pk));
            return response.Resource;
        }

        public async Task<T> FindById<T>(string id) where T : ID
        {
            CosmosModelInfo container = await InitializeCollection<T>();
            if (container.cache && RedisHelper.Instance != null)
            {

                return await RedisHelper.CacheShellAsync(CacheCosmosPrefix + container.container.Id, id, timeoutSeconds, () => { return FindByIdAsSql<T>(id); });

            }
            else
            {
                return await FindByIdAsSql<T>(id);
            }

        }

        public async Task<List<T>> FindByIds<T>(List<string> ids) where T : ID
        {
            CosmosModelInfo container = await InitializeCollection<T>();

            if (container.cache && RedisHelper.Instance != null)
            {
                List<T> list = new List<T>();
                List<string> NotIn = new List<string>();
                foreach (string id in ids)
                {
                    if (!await RedisHelper.HExistsAsync(CacheCosmosPrefix + container.container.Id, id))
                    {
                        NotIn.Add(id);
                    }
                    else
                    {
                        list.Add(await RedisHelper.HGetAsync<T>(CacheCosmosPrefix + container.container.Id, id));
                    }
                }
                if (NotIn.IsNotEmpty())
                {
                    List<T> noInList = await FindByDict<T>(new Dictionary<string, object> { { "id", NotIn.ToArray() } });
                    noInList.ForEach(x => { RedisHelper.HSet(CacheCosmosPrefix + container.container.Id, x.id, x); RedisHelper.Expire(CacheCosmosPrefix + container.container.Id, timeoutSeconds); });
                    list.AddRange(noInList);
                }
                return list;
            }
            else
            {
                return await FindByDict<T>(new Dictionary<string, object> { { "id", ids.ToArray() } });
            }
        }

        public async Task<dynamic> FindById(string CollectionName, string id)
        {
            if (DocumentCollectionDict.TryGetValue(CollectionName, out CosmosModelInfo container))
            {
                if (container.cache && RedisHelper.Instance != null)
                {

                    return await RedisHelper.CacheShellAsync(CacheCosmosPrefix + container.container.Id, id, timeoutSeconds, () => { return FindByDict(CollectionName, new Dictionary<string, object> { { "id", id } }); });

                }
                else
                {
                    return await FindByDict(CollectionName, new Dictionary<string, object> { { "id", id } });
                }
            }
            else
            {
                throw new BizException("CollectionName named:" + CollectionName + " dose not exsit in Database!");
            }
        }

        public async Task<List<dynamic>> FindByIds(string CollectionName, List<string> ids)
        {
            if (DocumentCollectionDict.TryGetValue(CollectionName, out CosmosModelInfo container))
            {
                if (container.cache && RedisHelper.Instance != null)
                {

                    return await RedisHelper.CacheShellAsync(CacheCosmosPrefix + container.container.Id, timeoutSeconds, () => { return FindByDict(CollectionName, new Dictionary<string, object> { { "id", ids.ToArray() } }); });

                }
                else
                {
                    return await FindByDict(CollectionName, new Dictionary<string, object> { { "id", ids.ToArray() } });
                }
            }
            else
            {
                throw new BizException("CollectionName named:" + CollectionName + " dose not exsit in Database!");
            }
        }
    }
}