using System;
using System.Collections.Generic;
using System.Linq;
using Microsoft.Data.SqlClient;
using Newtonsoft.Json;
using Serilog;
using SyncEngine.Access;
using SyncEngine.Configuration;
using SyncEngine.Logging;
using SyncEngine.Metadata;

namespace SyncEngine.SqlServer
{
    public sealed class SqlSchemaProvisioner
    {
        private readonly SqlServerConnectionFactory _factory;
        private readonly SyncOptions _options;
        private readonly SyncMetadataRepository _metadata;
        private readonly ILogger _logger;

        public static readonly string[] SyncColumns =
        {
            "_sync_row_hash VARCHAR(64) NULL",
            "_sync_is_deleted BIT NOT NULL CONSTRAINT [DF_{0}_sync_del] DEFAULT 0",
            "_sync_last_seen_at DATETIME2 NULL",
            "_sync_updated_at DATETIME2 NULL"
        };

        public SqlSchemaProvisioner(
            SqlServerConnectionFactory factory,
            SyncOptions options,
            SyncMetadataRepository metadata,
            ILogger logger)
        {
            _factory = factory;
            _options = options;
            _metadata = metadata;
            _logger = logger;
        }

        public void EnsureDatabase(SqlConnection connection)
        {
            _metadata.EnsureSchema(connection);
        }

        public void ProvisionTables(SqlConnection connection, IList<AccessTableSchema> tables, SyncStatistics stats)
        {
            foreach (var table in tables)
            {
                if (!_options.DryRun && SqlIdentityHelper.NeedsIdentityRebuild(connection, table))
                {
                    RebuildTableForIdentity(connection, table, stats);
                }

                ProvisionTable(connection, table, stats);
                ProvisionColumnDefaults(connection, table, stats);
                TightenNullability(connection, table, stats);
            }
        }

        public void ProvisionKeysAndIndexes(
            SqlConnection connection,
            IList<AccessIndexInfo> indexes,
            IList<AccessTableSchema> tables,
            SyncStatistics stats)
        {
            var seenColumnSignatures = new HashSet<string>(StringComparer.OrdinalIgnoreCase);
            var primaryTables = new HashSet<string>(StringComparer.OrdinalIgnoreCase);
            var columnNullability = BuildColumnNullabilityLookup(tables);

            foreach (var index in indexes.Where(i => i.IsPrimary).OrderBy(i => i.TableName, StringComparer.OrdinalIgnoreCase))
            {
                ProvisionPrimaryKey(connection, index, stats, primaryTables, seenColumnSignatures);
            }

            foreach (var index in indexes.Where(i => !i.IsPrimary).OrderBy(i => i.TableName, StringComparer.OrdinalIgnoreCase))
            {
                ProvisionSecondaryIndex(connection, index, stats, seenColumnSignatures, columnNullability);
            }
        }

        public void ProvisionForeignKeys(SqlConnection connection, IList<ForeignKeyInfo> foreignKeys, SyncStatistics stats)
        {
            foreach (var fk in foreignKeys)
            {
                var constraint = BuildForeignKeyConstraintName(fk);
                if (_options.DryRun)
                {
                    _logger.Information("DRY RUN FK: {Child}->{Parent}", fk.ChildTable, fk.ParentTable);
                    continue;
                }

                if (ForeignKeyExistsWithCascade(connection, constraint, fk))
                {
                    stats.ForeignKeysSkipped++;
                    continue;
                }

                if (ForeignKeyExists(connection, constraint))
                {
                    DropForeignKey(connection, constraint);
                }

                var sql = BuildForeignKeyAlterSql(fk, constraint);
                try
                {
                    using (var cmd = new SqlCommand(sql, connection))
                    {
                        cmd.CommandTimeout = _options.CommandTimeoutSeconds;
                        cmd.ExecuteNonQuery();
                    }

                    stats.ForeignKeysCreated++;
                }
                catch (Exception ex)
                {
                    stats.ForeignKeysSkipped++;
                    _logger.Warning(ex, "Skipped FK {Child}->{Parent}", fk.ChildTable, fk.ParentTable);
                }
            }
        }

        public void ProvisionCheckConstraints(SqlConnection connection, IList<AccessTableSchema> tables, SyncStatistics stats)
        {
            foreach (var table in tables)
            {
                if (!string.IsNullOrWhiteSpace(table.TableValidationRule) &&
                    AccessValidationTranslator.TryTranslate(table.TableValidationRule, null, out var tablePredicate))
                {
                    ProvisionCheckConstraint(
                        connection,
                        table.TableName,
                        AccessValidationTranslator.BuildCheckConstraintName(table.TableName, null),
                        tablePredicate,
                        stats);
                }

                foreach (var col in table.Columns)
                {
                    if (string.IsNullOrWhiteSpace(col.ValidationRule))
                    {
                        continue;
                    }

                    if (!AccessValidationTranslator.TryTranslate(col.ValidationRule, col.Name, out var predicate))
                    {
                        stats.CheckConstraintsSkipped++;
                        _logger.Warning(
                            "Skipped validation rule on {Table}.{Column}: untranslatable",
                            table.TableName,
                            col.Name);
                        continue;
                    }

                    ProvisionCheckConstraint(
                        connection,
                        table.TableName,
                        AccessValidationTranslator.BuildCheckConstraintName(table.TableName, col.Name),
                        predicate,
                        stats);
                }
            }
        }

        public void ProvisionViews(
            SqlConnection connection,
            IList<AccessQuerySchema> queries,
            ViewColumnCatalog viewCatalog,
            SyncStatistics stats)
        {
            foreach (var query in queries.Where(q => q.ShouldProvision).OrderBy(q => q.Name, StringComparer.OrdinalIgnoreCase))
            {
                if (!SqlViewTranslator.IsLikelyTranslatable(query.SqlText))
                {
                    stats.ViewsSkipped++;
                    _logger.Warning("Skipping untranslatable query {Name}", query.Name);
                    continue;
                }

                var translated = SqlViewTranslator.Translate(query.SqlText, viewCatalog);
                if (string.IsNullOrWhiteSpace(translated))
                {
                    stats.ViewsSkipped++;
                    continue;
                }

                var viewName = _factory.Qualify(query.Name);

                if (_options.DryRun)
                {
                    _logger.Information("DRY RUN VIEW: {Name}", query.Name);
                    continue;
                }

                var existed = ObjectExists(connection, viewName, "V");

                try
                {
                    var sql = existed
                        ? string.Format("ALTER VIEW {0} AS {1}", viewName, translated)
                        : string.Format("CREATE VIEW {0} AS {1}", viewName, translated);

                    using (var cmd = new SqlCommand(sql, connection))
                    {
                        cmd.CommandTimeout = _options.CommandTimeoutSeconds;
                        cmd.ExecuteNonQuery();
                    }

                    if (existed)
                    {
                        stats.ViewsUpdated++;
                    }
                    else
                    {
                        stats.ViewsCreated++;
                    }

                    _metadata.SaveSchemaVersion(connection, "view:" + query.Name, translated);
                }
                catch (Exception ex)
                {
                    stats.ViewsSkipped++;
                    _logger.Warning(ex, "Failed to provision view {Name}. Translated SQL: {Sql}", query.Name, translated);
                }
            }
        }

        private void ProvisionCheckConstraint(
            SqlConnection connection,
            string tableName,
            string constraintName,
            string predicate,
            SyncStatistics stats)
        {
            if (_options.DryRun)
            {
                _logger.Information("DRY RUN CHECK: {Table}.{Name}", tableName, constraintName);
                return;
            }

            if (CheckConstraintExists(connection, constraintName))
            {
                stats.CheckConstraintsSkipped++;
                return;
            }

            var qualified = _factory.Qualify(tableName);
            var sql = string.Format(
                "ALTER TABLE {0} ADD CONSTRAINT [{1}] CHECK ({2})",
                qualified,
                constraintName,
                predicate);

            try
            {
                using (var cmd = new SqlCommand(sql, connection))
                {
                    cmd.CommandTimeout = _options.CommandTimeoutSeconds;
                    cmd.ExecuteNonQuery();
                }

                stats.CheckConstraintsCreated++;
            }
            catch (Exception ex)
            {
                stats.CheckConstraintsSkipped++;
                _logger.Warning(ex, "Skipped CHECK {Table}.{Name}", tableName, constraintName);
            }
        }

        private void ProvisionColumnDefaults(SqlConnection connection, AccessTableSchema table, SyncStatistics stats)
        {
            foreach (var col in table.Columns)
            {
                if (string.IsNullOrWhiteSpace(col.DefaultValue))
                {
                    continue;
                }

                if (AccessDefaultTranslator.IsNullDefault(col.DefaultValue))
                {
                    continue;
                }

                if (!AccessDefaultTranslator.TryTranslate(col.DefaultValue, out var defaultExpr))
                {
                    stats.DefaultsSkipped++;
                    _logger.Warning(
                        "Skipped default on {Table}.{Column}: untranslatable ({Expr})",
                        table.TableName,
                        col.Name,
                        col.DefaultValue);
                    continue;
                }

                var constraintName = AccessDefaultTranslator.BuildDefaultConstraintName(table.TableName, col.Name);
                if (_options.DryRun)
                {
                    _logger.Information("DRY RUN DEFAULT: {Table}.{Column}", table.TableName, col.Name);
                    continue;
                }

                if (DefaultConstraintExists(connection, constraintName) ||
                    ColumnHasDefault(connection, table.TableName, col.Name))
                {
                    stats.DefaultsUnchanged++;
                    continue;
                }

                if (!ColumnExists(connection, table.TableName, col.Name))
                {
                    continue;
                }

                var sql = string.Format(
                    "ALTER TABLE {0} ADD CONSTRAINT [{1}] DEFAULT ({2}) FOR [{3}]",
                    _factory.Qualify(table.TableName),
                    constraintName,
                    defaultExpr,
                    col.Name);

                try
                {
                    using (var cmd = new SqlCommand(sql, connection))
                    {
                        cmd.CommandTimeout = _options.CommandTimeoutSeconds;
                        cmd.ExecuteNonQuery();
                    }

                    stats.DefaultsCreated++;
                }
                catch (Exception ex)
                {
                    stats.DefaultsSkipped++;
                    _logger.Warning(ex, "Skipped DEFAULT on {Table}.{Column}", table.TableName, col.Name);
                }
            }
        }

        private void TightenNullability(SqlConnection connection, AccessTableSchema table, SyncStatistics stats)
        {
            foreach (var col in table.Columns)
            {
                if (col.AllowNull || col.IsAutoIncrement)
                {
                    continue;
                }

                if (!ColumnExists(connection, table.TableName, col.Name))
                {
                    continue;
                }

                if (!ColumnIsNullable(connection, table.TableName, col.Name))
                {
                    continue;
                }

                if (ColumnHasNullValues(connection, _factory.Qualify(table.TableName), col.Name))
                {
                    stats.NullabilitySkipped++;
                    _logger.Warning(
                        "Skipped NOT NULL on {Table}.{Column}: column contains NULL values",
                        table.TableName,
                        col.Name);
                    continue;
                }

                if (_options.DryRun)
                {
                    _logger.Information("DRY RUN NOT NULL: {Table}.{Column}", table.TableName, col.Name);
                    continue;
                }

                var notNullType = AccessTypeMapper.ToSqlServerType(
                    col.ClrType,
                    col.ColumnSize,
                    allowNull: false,
                    col.IsAutoIncrement);

                var sql = string.Format(
                    "ALTER TABLE {0} ALTER COLUMN [{1}] {2}",
                    _factory.Qualify(table.TableName),
                    col.Name,
                    notNullType);

                try
                {
                    using (var cmd = new SqlCommand(sql, connection))
                    {
                        cmd.CommandTimeout = _options.CommandTimeoutSeconds;
                        cmd.ExecuteNonQuery();
                    }

                    stats.NullabilityTightened++;
                }
                catch (Exception ex)
                {
                    stats.NullabilitySkipped++;
                    _logger.Warning(ex, "Skipped NOT NULL on {Table}.{Column}", table.TableName, col.Name);
                }
            }
        }

        private void DropConstraintsAndIndexesForIdentityStaging(SqlConnection connection, string stagingTableName)
        {
            var qualified = _factory.Qualify(stagingTableName);
            using (var cmd = new SqlCommand(@"
DECLARE @qualified NVARCHAR(517) = @qualifiedName;
DECLARE @objectId INT = OBJECT_ID(@qualifiedName);
IF @objectId IS NULL RETURN;

DECLARE @sql NVARCHAR(MAX) = N'';

SELECT @sql = @sql + N'ALTER TABLE ' + @qualified + N' DROP CONSTRAINT ' + QUOTENAME(name) + N';'
FROM sys.objects
WHERE parent_object_id = @objectId
  AND type IN ('D', 'C', 'PK', 'UQ');

IF LEN(@sql) > 0 EXEC sp_executesql @sql;

SET @sql = N'';
SELECT @sql = @sql + N'DROP INDEX ' + QUOTENAME(i.name) + N' ON ' + @qualified + N';'
FROM sys.indexes i
WHERE i.object_id = @objectId
  AND i.type > 0
  AND i.is_primary_key = 0
  AND i.is_unique_constraint = 0;

IF LEN(@sql) > 0 EXEC sp_executesql @sql;", connection))
            {
                cmd.Parameters.AddWithValue("@qualifiedName", qualified);
                cmd.CommandTimeout = 120;
                cmd.ExecuteNonQuery();
            }
        }

        private void RebuildTableForIdentity(SqlConnection connection, AccessTableSchema table, SyncStatistics stats)
        {
            var identityCol = SqlIdentityHelper.GetIdentityColumn(table);
            if (identityCol == null)
            {
                return;
            }

            var stagingName = SqlIdentityHelper.GetIdentityStagingTableName(table.TableName);
            var liveExists = SqlIdentityHelper.TableExists(connection, table.TableName);
            var stagingExists = SqlIdentityHelper.TableExists(connection, stagingName);

            if (!liveExists && stagingExists)
            {
                _logger.Information(
                    "Resuming interrupted identity rebuild for {Table} on [{Column}]",
                    table.TableName,
                    identityCol.Name);
                DropForeignKeysForTable(connection, stagingName);
                FinalizeIdentityRebuildFromStaging(connection, table, stats, stagingName, identityCol);
                return;
            }

            if (!liveExists)
            {
                return;
            }

            if (stagingExists)
            {
                _logger.Warning("Dropping orphan staging table {Staging} before identity rebuild", stagingName);
                DropTableIfExists(connection, stagingName);
            }

            _logger.Information(
                "Rebuilding {Table} to add IDENTITY on [{Column}]",
                table.TableName,
                identityCol.Name);

            DropForeignKeysForTable(connection, table.TableName);

            var renameTarget = string.Format("{0}.{1}", _options.SqlExpressSchema, table.TableName);
            using (var rename = new SqlCommand(
                "EXEC sp_rename @objname, @newname, 'OBJECT'",
                connection))
            {
                rename.Parameters.AddWithValue("@objname", renameTarget);
                rename.Parameters.AddWithValue("@newname", stagingName);
                rename.CommandTimeout = _options.CommandTimeoutSeconds;
                rename.ExecuteNonQuery();
            }

            DropConstraintsAndIndexesForIdentityStaging(connection, stagingName);
            FinalizeIdentityRebuildFromStaging(connection, table, stats, stagingName, identityCol);
        }

        private void FinalizeIdentityRebuildFromStaging(
            SqlConnection connection,
            AccessTableSchema table,
            SyncStatistics stats,
            string stagingName,
            AccessColumnSchema identityCol)
        {
            var qualified = _factory.Qualify(table.TableName);
            var stagingQualified = _factory.Qualify(stagingName);

            DropConstraintsAndIndexesForIdentityStaging(connection, stagingName);
            DropOrphanedConstraintsForTable(connection, table);

            var columnDefs = BuildColumnDefinitions(table, includeInlineDefaults: true);
            foreach (var syncCol in SyncColumns)
            {
                columnDefs.Add(string.Format(syncCol, SqlIdentifierHelper.SanitizeConstraintName(table.TableName)));
            }

            if (!SqlIdentityHelper.TableExists(connection, table.TableName))
            {
                var createSql = string.Format(
                    "CREATE TABLE {0} (\n    {1}\n);",
                    qualified,
                    string.Join(",\n    ", columnDefs));

                using (var create = new SqlCommand(createSql, connection))
                {
                    create.CommandTimeout = _options.CommandTimeoutSeconds;
                    create.ExecuteNonQuery();
                }
            }

            var allColumns = table.Columns.Select(c => c.Name)
                .Concat(new[] { "_sync_row_hash", "_sync_is_deleted", "_sync_last_seen_at", "_sync_updated_at" })
                .Select(c => "[" + c + "]")
                .ToList();

            SqlIdentityHelper.SetIdentityInsert(connection, null, qualified, true);
            try
            {
                var insertSql = string.Format(
                    "INSERT INTO {0} ({1}) SELECT {1} FROM {2}",
                    qualified,
                    string.Join(", ", allColumns),
                    stagingQualified);

                using (var insert = new SqlCommand(insertSql, connection))
                {
                    insert.CommandTimeout = _options.CommandTimeoutSeconds;
                    insert.ExecuteNonQuery();
                }
            }
            finally
            {
                SqlIdentityHelper.SetIdentityInsert(connection, null, qualified, false);
            }

            var maxIdentity = SqlIdentityHelper.GetMaxColumnValue(connection, qualified, identityCol.Name);
            SqlIdentityHelper.ReseedIdentity(connection, qualified, maxIdentity ?? 0);

            DropTableIfExists(connection, stagingName);
            stats.TablesIdentityRebuilt++;
        }

        private void DropOrphanedConstraintsForTable(SqlConnection connection, AccessTableSchema table)
        {
            var constraintNames = new List<string>
            {
                string.Format("DF_{0}_sync_del", SqlIdentifierHelper.SanitizeConstraintName(table.TableName))
            };

            foreach (var col in table.Columns)
            {
                if (string.IsNullOrWhiteSpace(col.DefaultValue))
                {
                    continue;
                }

                if (!AccessDefaultTranslator.TryTranslate(col.DefaultValue, out _))
                {
                    continue;
                }

                constraintNames.Add(AccessDefaultTranslator.BuildDefaultConstraintName(table.TableName, col.Name));
            }

            foreach (var name in constraintNames)
            {
                DropDefaultConstraintByName(connection, name);
            }
        }

        private static void DropDefaultConstraintByName(SqlConnection connection, string constraintName)
        {
            using (var cmd = new SqlCommand(@"
DECLARE @sql NVARCHAR(MAX);
SELECT @sql = N'ALTER TABLE '
    + QUOTENAME(OBJECT_SCHEMA_NAME(dc.parent_object_id)) + N'.'
    + QUOTENAME(OBJECT_NAME(dc.parent_object_id))
    + N' DROP CONSTRAINT ' + QUOTENAME(dc.name)
FROM sys.default_constraints dc
WHERE dc.name = @name;
IF @sql IS NOT NULL EXEC sp_executesql @sql;", connection))
            {
                cmd.Parameters.AddWithValue("@name", constraintName);
                cmd.ExecuteNonQuery();
            }
        }

        private void DropTableIfExists(SqlConnection connection, string tableName)
        {
            if (!SqlIdentityHelper.TableExists(connection, tableName))
            {
                return;
            }

            using (var cmd = new SqlCommand(
                string.Format("DROP TABLE {0}", _factory.Qualify(tableName)),
                connection))
            {
                cmd.CommandTimeout = _options.CommandTimeoutSeconds;
                cmd.ExecuteNonQuery();
            }
        }

        private void ProvisionPrimaryKey(
            SqlConnection connection,
            AccessIndexInfo index,
            SyncStatistics stats,
            HashSet<string> primaryTables,
            HashSet<string> seenColumnSignatures)
        {
            if (SqlSchemaKeyBuilder.ShouldSkipIndex(index) || primaryTables.Contains(index.TableName))
            {
                return;
            }

            var signature = index.TableName + "|" + SqlSchemaKeyBuilder.NormalizeColumnKey(index.Columns);
            if (!seenColumnSignatures.Add(signature))
            {
                stats.PrimaryKeysUnchanged++;
                return;
            }

            var qualified = _factory.Qualify(index.TableName);
            var constraintName = SqlSchemaKeyBuilder.BuildPrimaryKeyConstraintName(index.TableName);

            if (_options.DryRun)
            {
                _logger.Information("DRY RUN PK: {Table} ({Columns})", index.TableName, string.Join(", ", index.Columns));
                return;
            }

            if (PrimaryKeyExists(connection, index.TableName) || IndexExistsByName(connection, index.TableName, constraintName))
            {
                stats.PrimaryKeysUnchanged++;
                primaryTables.Add(index.TableName);
                return;
            }

            var sql = SqlSchemaKeyBuilder.BuildPrimaryKeyAlterSql(qualified, constraintName, index.Columns);
            try
            {
                using (var cmd = new SqlCommand(sql, connection))
                {
                    cmd.CommandTimeout = _options.CommandTimeoutSeconds;
                    cmd.ExecuteNonQuery();
                }

                stats.PrimaryKeysCreated++;
                primaryTables.Add(index.TableName);
            }
            catch (Exception ex)
            {
                stats.PrimaryKeysSkipped++;
                _logger.Warning(ex, "Skipped PK on {Table}", index.TableName);
            }
        }

        private void ProvisionSecondaryIndex(
            SqlConnection connection,
            AccessIndexInfo index,
            SyncStatistics stats,
            HashSet<string> seenColumnSignatures,
            IReadOnlyDictionary<string, IReadOnlyDictionary<string, bool>> columnNullability)
        {
            if (SqlSchemaKeyBuilder.ShouldSkipIndex(index))
            {
                return;
            }

            var signature = index.TableName + "|" + SqlSchemaKeyBuilder.NormalizeColumnKey(index.Columns);
            if (!seenColumnSignatures.Add(signature))
            {
                stats.IndexesUnchanged++;
                return;
            }

            var qualified = _factory.Qualify(index.TableName);
            var indexName = SqlSchemaKeyBuilder.BuildSqlIndexName(index);

            if (_options.DryRun)
            {
                _logger.Information(
                    "DRY RUN INDEX: {Table}.{Name} ({Columns})",
                    index.TableName,
                    indexName,
                    string.Join(", ", index.Columns));
                return;
            }

            if (IndexExistsByName(connection, index.TableName, indexName) ||
                IndexExistsByColumns(connection, index.TableName, index.Columns))
            {
                stats.IndexesUnchanged++;
                return;
            }

            var filter = ResolveUniqueIndexFilter(index, columnNullability);
            var sql = SqlSchemaKeyBuilder.BuildCreateIndexSql(
                qualified,
                indexName,
                index.Columns,
                index.IsUnique,
                filter);

            try
            {
                using (var cmd = new SqlCommand(sql, connection))
                {
                    cmd.CommandTimeout = _options.CommandTimeoutSeconds;
                    cmd.ExecuteNonQuery();
                }

                stats.IndexesCreated++;
            }
            catch (Exception ex)
            {
                stats.IndexesSkipped++;
                _logger.Warning(ex, "Skipped index {Table}.{Name}", index.TableName, indexName);
            }
        }

        private static IReadOnlyDictionary<string, IReadOnlyDictionary<string, bool>> BuildColumnNullabilityLookup(
            IList<AccessTableSchema> tables)
        {
            var lookup = new Dictionary<string, IReadOnlyDictionary<string, bool>>(StringComparer.OrdinalIgnoreCase);
            if (tables == null)
            {
                return lookup;
            }

            foreach (var table in tables)
            {
                if (table?.Columns == null || string.IsNullOrWhiteSpace(table.TableName))
                {
                    continue;
                }

                var columns = table.Columns
                    .Where(c => !string.IsNullOrWhiteSpace(c.Name))
                    .ToDictionary(c => c.Name, c => c.AllowNull, StringComparer.OrdinalIgnoreCase);

                lookup[table.TableName] = columns;
            }

            return lookup;
        }

        private static string ResolveUniqueIndexFilter(
            AccessIndexInfo index,
            IReadOnlyDictionary<string, IReadOnlyDictionary<string, bool>> columnNullability)
        {
            if (!index.IsUnique || index.Columns == null || index.Columns.Count != 1)
            {
                return null;
            }

            var columnName = index.Columns[0];
            if (string.IsNullOrWhiteSpace(columnName))
            {
                return null;
            }

            if (columnNullability != null &&
                columnNullability.TryGetValue(index.TableName, out var tableColumns) &&
                tableColumns.TryGetValue(columnName, out var allowNull) &&
                !allowNull)
            {
                return null;
            }

            return SqlSchemaKeyBuilder.BuildNullableUniqueFilter(index.Columns);
        }

        private void ProvisionTable(SqlConnection connection, AccessTableSchema table, SyncStatistics stats)
        {
            var qualified = _factory.Qualify(table.TableName);
            var columnDefs = BuildColumnDefinitions(table, includeInlineDefaults: true);

            foreach (var syncCol in SyncColumns)
            {
                columnDefs.Add(string.Format(syncCol, SqlIdentifierHelper.SanitizeConstraintName(table.TableName)));
            }

            if (_options.DryRun)
            {
                _logger.Information("DRY RUN CREATE: {Table}", table.TableName);
                return;
            }

            var existed = ObjectExists(connection, qualified, "U");
            if (!existed)
            {
                DropOrphanedConstraintsForTable(connection, table);

                var createSql = string.Format(@"
CREATE TABLE {0} (
    {1}
);", qualified, string.Join(",\n    ", columnDefs));

                using (var cmd = new SqlCommand(createSql, connection))
                {
                    cmd.CommandTimeout = _options.CommandTimeoutSeconds;
                    cmd.ExecuteNonQuery();
                }

                stats.TablesCreated++;
            }
            else
            {
                var columnsAdded = AddMissingColumns(connection, table);
                stats.ColumnsAdded += columnsAdded;
                if (columnsAdded > 0)
                {
                    stats.TablesSchemaUpdated++;
                }
                else
                {
                    stats.TablesUnchanged++;
                }
            }

            _metadata.SaveSchemaVersion(connection, "table:" + table.TableName,
                JsonConvert.SerializeObject(table.Columns.Select(c => c.Name)));
        }

        private List<string> BuildColumnDefinitions(AccessTableSchema table, bool includeInlineDefaults)
        {
            var columnDefs = new List<string>();
            foreach (var col in table.Columns)
            {
                columnDefs.Add(BuildColumnDefinition(table, col, includeInlineDefaults));
            }

            return columnDefs;
        }

        private string BuildColumnDefinition(AccessTableSchema table, AccessColumnSchema col, bool includeInlineDefault)
        {
            var typeSql = AccessTypeMapper.ToSqlServerType(col);
            if (!includeInlineDefault || string.IsNullOrWhiteSpace(col.DefaultValue) || col.IsAutoIncrement)
            {
                return string.Format("[{0}] {1}", col.Name, typeSql);
            }

            if (!AccessDefaultTranslator.TryTranslate(col.DefaultValue, out var defaultExpr))
            {
                return string.Format("[{0}] {1}", col.Name, typeSql);
            }

            var dfName = AccessDefaultTranslator.BuildDefaultConstraintName(table.TableName, col.Name);
            return string.Format(
                "[{0}] {1} CONSTRAINT [{2}] DEFAULT ({3})",
                col.Name,
                typeSql,
                dfName,
                defaultExpr);
        }

        private int AddMissingColumns(SqlConnection connection, AccessTableSchema table)
        {
            var qualified = _factory.Qualify(table.TableName);
            var added = 0;

            foreach (var col in table.Columns)
            {
                if (ColumnExists(connection, table.TableName, col.Name))
                {
                    continue;
                }

                var sql = string.Format(
                    "ALTER TABLE {0} ADD {1}",
                    qualified,
                    BuildColumnDefinition(table, col, includeInlineDefault: true));

                using (var cmd = new SqlCommand(sql, connection))
                {
                    cmd.CommandTimeout = _options.CommandTimeoutSeconds;
                    cmd.ExecuteNonQuery();
                }

                added++;
            }

            return added;
        }

        private static string BuildForeignKeyConstraintName(ForeignKeyInfo fk)
        {
            var constraint = string.Format(
                "FK_{0}_{1}_{2}",
                SqlIdentifierHelper.SanitizeConstraintName(fk.ChildTable),
                SqlIdentifierHelper.SanitizeConstraintName(fk.ParentTable),
                SqlIdentifierHelper.SanitizeConstraintName(fk.ChildColumn));
            return constraint.Length > 120 ? constraint.Substring(0, 120) : constraint;
        }

        private string BuildForeignKeyAlterSql(ForeignKeyInfo fk, string constraintName)
        {
            var qualifiedChild = _factory.Qualify(fk.ChildTable);
            var qualifiedParent = _factory.Qualify(fk.ParentTable);
            var updateClause = fk.OnUpdateCascade ? " ON UPDATE CASCADE" : string.Empty;
            var deleteClause = fk.OnDeleteCascade ? " ON DELETE CASCADE" : string.Empty;

            return string.Format(@"
ALTER TABLE {0} ADD CONSTRAINT [{1}]
FOREIGN KEY ([{2}]) REFERENCES {3} ([{4}]){5}{6}",
                qualifiedChild,
                constraintName,
                fk.ChildColumn,
                qualifiedParent,
                fk.ParentColumn,
                updateClause,
                deleteClause);
        }

        private void DropForeignKeysForTable(SqlConnection connection, string tableName)
        {
            var names = new List<string>();
            using (var cmd = new SqlCommand(@"
SELECT fk.name
FROM sys.foreign_keys fk
WHERE fk.parent_object_id = OBJECT_ID(@table) OR fk.referenced_object_id = OBJECT_ID(@table)", connection))
            {
                cmd.Parameters.AddWithValue("@table", tableName);
                using (var reader = cmd.ExecuteReader())
                {
                    while (reader.Read())
                    {
                        names.Add(Convert.ToString(reader["name"]));
                    }
                }
            }

            foreach (var name in names)
            {
                DropForeignKey(connection, name);
            }
        }

        private static void DropForeignKey(SqlConnection connection, string constraintName)
        {
            using (var cmd = new SqlCommand(@"
DECLARE @sql NVARCHAR(MAX) = (
    SELECT 'ALTER TABLE [' + SCHEMA_NAME(t.schema_id) + '].[' + t.name + '] DROP CONSTRAINT [' + fk.name + ']'
    FROM sys.foreign_keys fk
    INNER JOIN sys.tables t ON fk.parent_object_id = t.object_id
    WHERE fk.name = @name
);
IF @sql IS NOT NULL EXEC sp_executesql @sql;", connection))
            {
                cmd.Parameters.AddWithValue("@name", constraintName);
                cmd.ExecuteNonQuery();
            }
        }

        private static bool ForeignKeyExistsWithCascade(SqlConnection connection, string constraintName, ForeignKeyInfo expected)
        {
            using (var cmd = new SqlCommand(@"
SELECT delete_referential_action_desc, update_referential_action_desc
FROM sys.foreign_keys WHERE name = @name", connection))
            {
                cmd.Parameters.AddWithValue("@name", constraintName);
                using (var reader = cmd.ExecuteReader())
                {
                    if (!reader.Read())
                    {
                        return false;
                    }

                    var deleteAction = Convert.ToString(reader["delete_referential_action_desc"]);
                    var updateAction = Convert.ToString(reader["update_referential_action_desc"]);
                    var deleteMatches = expected.OnDeleteCascade
                        ? string.Equals(deleteAction, "CASCADE", StringComparison.OrdinalIgnoreCase)
                        : string.Equals(deleteAction, "NO_ACTION", StringComparison.OrdinalIgnoreCase);
                    var updateMatches = expected.OnUpdateCascade
                        ? string.Equals(updateAction, "CASCADE", StringComparison.OrdinalIgnoreCase)
                        : string.Equals(updateAction, "NO_ACTION", StringComparison.OrdinalIgnoreCase);
                    return deleteMatches && updateMatches;
                }
            }
        }

        private static bool ObjectExists(SqlConnection connection, string qualifiedName, string type)
        {
            using (var cmd = new SqlCommand("SELECT CASE WHEN OBJECT_ID(@name, @type) IS NULL THEN 0 ELSE 1 END", connection))
            {
                cmd.Parameters.AddWithValue("@name", qualifiedName);
                cmd.Parameters.AddWithValue("@type", type);
                return Convert.ToInt32(cmd.ExecuteScalar()) == 1;
            }
        }

        private static bool ColumnExists(SqlConnection connection, string tableName, string columnName)
        {
            using (var cmd = new SqlCommand("SELECT COL_LENGTH(@table, @column)", connection))
            {
                cmd.Parameters.AddWithValue("@table", tableName);
                cmd.Parameters.AddWithValue("@column", columnName);
                var result = cmd.ExecuteScalar();
                return result != null && result != DBNull.Value;
            }
        }

        private static bool ColumnIsNullable(SqlConnection connection, string tableName, string columnName)
        {
            using (var cmd = new SqlCommand(@"
SELECT is_nullable FROM sys.columns
WHERE object_id = OBJECT_ID(@table) AND name = @column", connection))
            {
                cmd.Parameters.AddWithValue("@table", tableName);
                cmd.Parameters.AddWithValue("@column", columnName);
                var result = cmd.ExecuteScalar();
                return result != null && Convert.ToBoolean(result);
            }
        }

        private static bool ColumnHasNullValues(SqlConnection connection, string qualifiedTable, string columnName)
        {
            using (var cmd = new SqlCommand(string.Format(
                "SELECT TOP 1 1 FROM {0} WHERE [{1}] IS NULL",
                qualifiedTable,
                columnName), connection))
            {
                return cmd.ExecuteScalar() != null;
            }
        }

        private static bool ColumnHasDefault(SqlConnection connection, string tableName, string columnName)
        {
            using (var cmd = new SqlCommand(@"
SELECT 1 FROM sys.default_constraints dc
INNER JOIN sys.columns c ON dc.parent_object_id = c.object_id AND dc.parent_column_id = c.column_id
WHERE dc.parent_object_id = OBJECT_ID(@table) AND c.name = @column", connection))
            {
                cmd.Parameters.AddWithValue("@table", tableName);
                cmd.Parameters.AddWithValue("@column", columnName);
                return cmd.ExecuteScalar() != null;
            }
        }

        private static bool DefaultConstraintExists(SqlConnection connection, string constraintName)
        {
            using (var cmd = new SqlCommand("SELECT 1 FROM sys.default_constraints WHERE name = @name", connection))
            {
                cmd.Parameters.AddWithValue("@name", constraintName);
                return cmd.ExecuteScalar() != null;
            }
        }

        private static bool CheckConstraintExists(SqlConnection connection, string constraintName)
        {
            using (var cmd = new SqlCommand("SELECT 1 FROM sys.check_constraints WHERE name = @name", connection))
            {
                cmd.Parameters.AddWithValue("@name", constraintName);
                return cmd.ExecuteScalar() != null;
            }
        }

        private static bool PrimaryKeyExists(SqlConnection connection, string tableName)
        {
            using (var cmd = new SqlCommand(
                "SELECT 1 FROM sys.indexes WHERE object_id = OBJECT_ID(@table) AND is_primary_key = 1",
                connection))
            {
                cmd.Parameters.AddWithValue("@table", tableName);
                return cmd.ExecuteScalar() != null;
            }
        }

        private static bool IndexExistsByName(SqlConnection connection, string tableName, string indexName)
        {
            using (var cmd = new SqlCommand(
                "SELECT 1 FROM sys.indexes WHERE object_id = OBJECT_ID(@table) AND name = @name",
                connection))
            {
                cmd.Parameters.AddWithValue("@table", tableName);
                cmd.Parameters.AddWithValue("@name", indexName);
                return cmd.ExecuteScalar() != null;
            }
        }

        private static bool IndexExistsByColumns(SqlConnection connection, string tableName, IList<string> columns)
        {
            var target = SqlSchemaKeyBuilder.NormalizeColumnKey(columns);
            var signatures = new Dictionary<int, List<string>>();
            using (var cmd = new SqlCommand(@"
SELECT ic.index_id, c.name, ic.key_ordinal
FROM sys.indexes i
INNER JOIN sys.index_columns ic ON i.object_id = ic.object_id AND i.index_id = ic.index_id
INNER JOIN sys.columns c ON ic.object_id = c.object_id AND ic.column_id = c.column_id
WHERE i.object_id = OBJECT_ID(@table) AND ic.is_included_column = 0
ORDER BY ic.index_id, ic.key_ordinal", connection))
            {
                cmd.Parameters.AddWithValue("@table", tableName);
                using (var reader = cmd.ExecuteReader())
                {
                    while (reader.Read())
                    {
                        var indexId = Convert.ToInt32(reader["index_id"]);
                        List<string> list;
                        if (!signatures.TryGetValue(indexId, out list))
                        {
                            list = new List<string>();
                            signatures[indexId] = list;
                        }

                        list.Add(Convert.ToString(reader["name"]));
                    }
                }
            }

            foreach (var signature in signatures.Values)
            {
                if (string.Equals(SqlSchemaKeyBuilder.NormalizeColumnKey(signature), target, StringComparison.OrdinalIgnoreCase))
                {
                    return true;
                }
            }

            return false;
        }

        private static bool ForeignKeyExists(SqlConnection connection, string constraintName)
        {
            using (var cmd = new SqlCommand("SELECT 1 FROM sys.foreign_keys WHERE name = @name", connection))
            {
                cmd.Parameters.AddWithValue("@name", constraintName);
                return cmd.ExecuteScalar() != null;
            }
        }
    }
}