src/YesSql.Core/Store.cs - main - aprogrammer/yessql

src/YesSql.Core/Store.cs
310 строк12 KB
Mike Alhayek
Add Format and apply rules (#544)
03 май 2024, 03:22
03 май 2024, 03:22c748f4d
using Dapper;
using Microsoft.Extensions.Logging;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Linq.Expressions;
using System.Threading.Tasks;
using YesSql.Commands;
using YesSql.Data;
using YesSql.Indexes;
using YesSql.Services;
using YesSql.Sql;

namespace YesSql
{
    public class Store : IStore
    {
        protected List<IIndexProvider> Indexes;
        protected List<Type> ScopedIndexes;

        public IConfiguration Configuration { get; set; }
        public ISqlDialect Dialect { get; private set; }
        public ITypeService TypeNames { get; private set; }

        internal readonly ConcurrentDictionary<Type, Func<IIndex, object>> GroupMethods = new();

        internal readonly ConcurrentDictionary<string, IEnumerable<IndexDescriptor>> Descriptors = new();

        internal readonly ConcurrentDictionary<Type, IAccessor<long>> IdAccessors = new();

        internal readonly ConcurrentDictionary<Type, IAccessor<long>> VersionAccessors = new();

        internal readonly ConcurrentDictionary<Type, Func<IDescriptor>> DescriptorActivators = new();

        internal readonly ConcurrentDictionary<WorkerQueryKey, Task<object>> Workers = new();

        internal readonly ConcurrentDictionary<long, QueryState> CompiledQueries = new();

        private readonly WorkDispatcher<WorkerQueryKey, object> _dispatcher = new();

        internal const int SmallBufferSize = 128;
        internal const int MediumBufferSize = 512;
        internal const int LargeBufferSize = 1024;

        static Store()
        {
            SqlMapper.ResetTypeHandlers();

            // Databases that don't support DateTimeOffset natively will store these in string columns.
            SqlMapper.AddTypeHandler(new DateTimeOffsetHandler());

            // Required by Sqlite. Guids are stored as text (uniqueidentifier) and are converted back to Guid with this handler.
            SqlMapper.AddTypeHandler(new GuidHandler());

            // Databases that don't support TimeSpan natively will store these in int columns as ticks.
            SqlMapper.AddTypeHandler(new TimeSpanHandler());
        }

        private Store()
        {
            Indexes = [];
            ScopedIndexes = [];
        }

        /// <summary>
        /// Initializes a <see cref="Store"/> instance and its new <see cref="Configuration"/>.
        /// </summary>
        /// <param name="config">An action to execute on the <see cref="Configuration"/> of the new <see cref="Store"/> instance.</param>
        internal Store(Action<IConfiguration> config) : this()
        {
            Configuration = new Configuration();
            config?.Invoke(Configuration);
            Dialect = Configuration.SqlDialect;
        }

        /// <summary>
        /// Initializes a <see cref="Store"/> instance using a specific <see cref="Configuration"/> instance.
        /// </summary>
        /// <param name="configuration">The <see cref="Configuration"/> instance to use.</param>
        internal Store(IConfiguration configuration) : this()
        {
            Configuration = configuration;
            Dialect = Configuration.SqlDialect;
        }

        public async Task InitializeAsync()
        {
            IndexCommand.ResetQueryCache();
            ValidateConfiguration();

            TypeNames = new TypeService();

            if (!string.IsNullOrEmpty(Configuration.Schema))
            {
                await using (var connection = Configuration.ConnectionFactory.CreateConnection())
                {
                    await connection.OpenAsync();

                    await using var transaction = await connection.BeginTransactionAsync(Configuration.IsolationLevel);

                    var builder = new SchemaBuilder(Configuration, transaction);

                    await builder.CreateSchemaAsync(Configuration.Schema);

                    await transaction.CommitAsync();
                }
            }

            // Initialize the Id generator
            await Configuration.IdGenerator.InitializeAsync(this);

            // Pre-initialize the default collection
            await InitializeCollectionAsync(string.Empty);
        }

        public async Task InitializeCollectionAsync(string collection)
        {
            var documentTable = Configuration.TableNameConvention.GetDocumentTable(collection);

            await using var connection = Configuration.ConnectionFactory.CreateConnection();
            await connection.OpenAsync();

            try
            {
                var selectCommand = connection.CreateCommand();

                var selectBuilder = Dialect.CreateBuilder(Configuration.TablePrefix);
                selectBuilder.Select();
                selectBuilder.AddSelector("*");
                selectBuilder.Table(documentTable, null, Configuration.Schema);
                selectBuilder.Take("1");

                selectCommand.CommandText = selectBuilder.ToSqlString();

                Configuration.Logger.LogTrace(selectCommand.CommandText);

                using var result = await selectCommand.ExecuteReaderAsync();
                if (result != null)
                {
                    try
                    {
                        // Check if the Version column exists
                        result.GetOrdinal(nameof(Document.Version));
                    }
                    catch
                    {
                        await result.CloseAsync();
                        await using var migrationTransaction = await connection.BeginTransactionAsync();
                        var migrationBuilder = new SchemaBuilder(Configuration, migrationTransaction);

                        try
                        {
                            await migrationBuilder.AlterTableAsync(documentTable, table => table
                                    .AddColumn<long>(nameof(Document.Version), column => column.WithDefault(0))
                                );

                            await migrationTransaction.CommitAsync();
                        }
                        catch
                        {
                            // Another thread must have altered it
                            await migrationTransaction.RollbackAsync();
                        }
                    }
                    return;
                }
            }
            catch
            {
                await using var transaction = await connection.BeginTransactionAsync();
                var builder = new SchemaBuilder(Configuration, transaction);

                try
                {
                    // The table doesn't exist, create it
                    await builder.CreateTableAsync(documentTable, table => table
                        .Column(Configuration.IdentityColumnSize, nameof(Document.Id), column => column.PrimaryKey().NotNull())
                        .Column<string>(nameof(Document.Type), column => column.NotNull())
                        .Column<string>(nameof(Document.Content), column => column.Unlimited())
                        .Column<long>(nameof(Document.Version), column => column.NotNull().WithDefault(0))
                    );

                    await builder.AlterTableAsync(documentTable, table => table
                        .CreateIndex("IX_" + documentTable + "_Type", "Type")
                    );

                    await transaction.CommitAsync();
                }
                catch
                {
                    // Another thread must have created it
                }
            }
            finally
            {
                await Configuration.IdGenerator.InitializeCollectionAsync(Configuration, collection);
            }
        }

        private void ValidateConfiguration()
        {
            if (Configuration.ConnectionFactory == null)
            {
                throw new Exception("The connection factory should be initialized during configuration.");
            }
        }

        public ISession CreateSession(bool withTracking = true)
            => new Session(this, withTracking);

#pragma warning disable CA1816 // Dispose methods should call SuppressFinalize
        public void Dispose()
#pragma warning restore CA1816 // Dispose methods should call SuppressFinalize
        {
        }

        public IAccessor<long> GetIdAccessor(Type tContainer)
        => IdAccessors.GetOrAdd(tContainer, Configuration.IdentifierAccessorFactory.CreateAccessor<long>);

        public IAccessor<long> GetVersionAccessor(Type tContainer)
            => VersionAccessors.GetOrAdd(tContainer, Configuration.VersionAccessorFactory.CreateAccessor<long>);

        /// <summary>
        /// Returns the available indexers for a specified type
        /// </summary>
        public IEnumerable<IndexDescriptor> Describe(Type target, string collection)
        {
            ArgumentNullException.ThrowIfNull(target);

            var cacheKey = string.IsNullOrEmpty(collection)
                ? target.FullName
                : target.FullName + ":" + collection
                ;

            return Descriptors.GetOrAdd(cacheKey, key => CreateDescriptors(target, collection, Indexes));
        }

        internal IEnumerable<IndexDescriptor> CreateDescriptors(Type target, string collection, IEnumerable<IIndexProvider> indexProviders)
        {
            var activator = DescriptorActivators.GetOrAdd(target, MakeDescriptorActivator);

            var context = activator();

            foreach (var provider in indexProviders)
            {
                if (provider.ForType().IsAssignableFrom(target) &&
                    string.Equals(collection, provider.CollectionName, StringComparison.OrdinalIgnoreCase))
                {
                    provider.Describe(context);
                }
            }

            return context.Describe(new[] { target }).ToList();
        }

        private static Func<IDescriptor> MakeDescriptorActivator(Type type)
        {
            var contextType = typeof(DescribeContext<>).MakeGenericType(type);
            return Expression.Lambda<Func<IDescriptor>>(Expression.New(contextType)).Compile();
        }

        [Obsolete($"Instead, utilize the {nameof(GetNextIdAsync)} method. This current method is slated for removal in upcoming releases.")]
        public long GetNextId(string collection)
            => GetNextIdAsync(collection).GetAwaiter().GetResult();

        public Task<long> GetNextIdAsync(string collection)
            => Configuration.IdGenerator.GetNextIdAsync(collection);

        public IStore RegisterIndexes(IEnumerable<IIndexProvider> indexProviders, string collection = null)
        {
            foreach (var indexProvider in indexProviders)
            {
                indexProvider.CollectionName ??= collection ?? "";
            }

            Indexes.AddRange(indexProviders);
            return this;
        }

        public IStore RegisterScopedIndexes(IEnumerable<Type> indexProviders)
        {
            ScopedIndexes.AddRange(indexProviders);
            return this;
        }

        /// <summary>
        /// Enlists some reusable logic such that not two threads run the same thing.
        /// </summary>
        /// <param name="key">A key identifying the running work.</param>
        /// <param name="work">A function containing the logic to execute.</param>
        /// <returns>The result of the work.</returns>
        internal Task<T> ProduceAsync<T, TState>(WorkerQueryKey key, Func<WorkerQueryKey, TState, Task<T>> work, TState state)
        {
            if (!Configuration.QueryGatingEnabled)
            {
                return work(key, state);
            }

            return ProduceAwaitedAsync(key, work, state);
        }

        internal async Task<T> ProduceAwaitedAsync<T, TState>(WorkerQueryKey key, Func<WorkerQueryKey, TState, Task<T>> work, TState state)
        {
            var content = await _dispatcher.ScheduleAsync(key, state, async (key, state) => await work(key, state));

            return (T)content;
        }
    }
}