diff --git a/.github/workflows/build-and-publish.yml b/.github/workflows/build-and-publish.yml index 9485cd8..1210a99 100644 --- a/.github/workflows/build-and-publish.yml +++ b/.github/workflows/build-and-publish.yml @@ -39,9 +39,23 @@ jobs: --health-timeout 5s --health-retries 5 + mysql: + image: mysql:8.4 + env: + MYSQL_ROOT_PASSWORD: test123 + MYSQL_DATABASE: coresync_test + ports: + - 3306:3306 + options: >- + --health-cmd "mysqladmin ping -h 127.0.0.1 -ptest123" + --health-interval 10s + --health-timeout 5s + --health-retries 10 + env: CORE-SYNC_CONNECTION_STRING: "Server=localhost;User Id=sa;Password=CoreSync_Test123!;TrustServerCertificate=True" CORE-SYNC_POSTGRESQL_CONNECTION_STRING: "Host=localhost;Port=5432;Database=coresync_test;Username=coresync;Password=test123" + CORE-SYNC_MYSQL_CONNECTION_STRING: "Server=localhost;Port=3306;Database=coresync_test;User ID=root;Password=test123;GuidFormat=Char36" steps: - uses: actions/checkout@v4 @@ -49,7 +63,7 @@ jobs: - name: Setup .NET uses: actions/setup-dotnet@v4 with: - dotnet-version: '8.0.x' + dotnet-version: '9.0.x' - name: Restore dependencies run: dotnet restore src/CoreSync.sln @@ -71,7 +85,7 @@ jobs: - name: Setup .NET uses: actions/setup-dotnet@v4 with: - dotnet-version: '8.0.x' + dotnet-version: '9.0.x' - name: Extract version from tag id: version diff --git a/README.md b/README.md index 718a286..870417b 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # CoreSync -**CoreSync** is a .NET library for bidirectional data synchronization between databases. It supports SQLite, SQL Server, and PostgreSQL, letting you keep multiple database instances in sync — whether they're on the same machine or communicating over HTTP. +**CoreSync** is a .NET library for bidirectional data synchronization between databases. It supports SQLite, SQL Server, PostgreSQL, and MySQL, letting you keep multiple database instances in sync — whether they're on the same machine or communicating over HTTP. [![Build status](https://ci.appveyor.com/api/projects/status/8cloij4060cbnvfp?svg=true)](https://ci.appveyor.com/project/adospace/coresync) [![NuGet](https://img.shields.io/nuget/v/CoreSync.svg)](https://www.nuget.org/packages/CoreSync/) @@ -8,12 +8,12 @@ ## Key Features -- **Bidirectional sync** between any supported database combination (e.g. SQLite ↔ SQL Server, PostgreSQL ↔ SQLite, etc.) +- **Bidirectional sync** between any supported database combination (e.g. SQLite ↔ SQL Server, PostgreSQL ↔ SQLite, MySQL ↔ SQLite, etc.) - **Automatic change tracking** — detects inserts, updates, and deletes with version-based anchors - **Conflict resolution** — built-in strategies (Skip or ForceWrite) with per-item customization - **Filtered sync** — synchronize a subset of data using parameterized queries - **HTTP transport** — sync over the network with ASP.NET Core server endpoints and a resilient HTTP client (with Polly retries and MessagePack binary format) -- **Multiple providers**: SQLite, SQL Server (custom change tracking), SQL Server CT (native Change Tracking), PostgreSQL +- **Multiple providers**: SQLite, SQL Server (custom change tracking), SQL Server CT (native Change Tracking), PostgreSQL, MySQL ## How It Works @@ -42,6 +42,7 @@ CoreSync uses a **version-based change tracking** approach: | `CoreSync.SqlServer` | SQL Server | Custom trigger-based | | `CoreSync.SqlServerCT` | SQL Server | Native Change Tracking | | `CoreSync.PostgreSQL` | PostgreSQL | Custom trigger-based | +| `CoreSync.MySql` | MySQL | Custom trigger-based | **HTTP packages** for remote sync over the network: @@ -228,6 +229,7 @@ CoreSync (core interfaces and SyncAgent) ├── CoreSync.SqlServer (SQL Server provider - custom CT) ├── CoreSync.SqlServerCT (SQL Server provider - native CT) ├── CoreSync.PostgreSQL (PostgreSQL provider) +├── CoreSync.MySql (MySQL provider) ├── CoreSync.Http (shared HTTP types) ├── CoreSync.Http.Server (ASP.NET Core sync endpoints) └── CoreSync.Http.Client (resilient HTTP sync client) diff --git a/docs/comparison-with-datasync-toolkit.md b/docs/comparison-with-datasync-toolkit.md index 897a6a9..ae7eae8 100644 --- a/docs/comparison-with-datasync-toolkit.md +++ b/docs/comparison-with-datasync-toolkit.md @@ -14,7 +14,7 @@ Both CoreSync and the [Datasync Community Toolkit](https://github.com/CommunityT | **Primary purpose** | Bidirectional database-to-database sync | Client-server offline sync (successor to Azure Mobile Apps) | | **Architecture** | Library — direct provider-to-provider | Client-server — requires ASP.NET Core controllers | | **Sync direction** | Bidirectional (peer-to-peer) | Bidirectional (client-server only) | -| **Server databases** | SQLite, SQL Server, PostgreSQL | SQL Server, PostgreSQL, MySQL, Cosmos DB, MongoDB, LiteDB | +| **Server databases** | SQLite, SQL Server, PostgreSQL, MySQL | SQL Server, PostgreSQL, MySQL, Cosmos DB, MongoDB, LiteDB | | **Client databases** | Any supported provider | SQLite only | | **Change detection** | Database triggers or SQL Server native CT | Timestamp-based (`UpdatedAt` field) | | **Entity requirements** | No changes to your schema | Must implement `ITableData` (Id, UpdatedAt, Version, Deleted) | @@ -117,13 +117,14 @@ Both libraries support per-item conflict resolution. CoreSync uses a simpler mod | SQLite | Yes | Not recommended (timestamp precision) | Yes (only option) | | SQL Server / Azure SQL | Yes (custom triggers + native CT) | Yes | — | | PostgreSQL | Yes | Yes | — | +| MySQL | Yes | Yes | — | | MySQL / MariaDB | — | Yes | — | | Cosmos DB | — | Yes | — | | MongoDB | — | Yes | — | | LiteDB | — | Yes | — | | In-Memory | — | Yes (testing) | — | -CoreSync uniquely supports **SQLite as both a source and target** for sync — not just as a client-side cache. This means you can sync SQLite-to-SQLite, SQLite-to-SQL Server, or SQLite-to-PostgreSQL in any direction. +CoreSync uniquely supports **SQLite as both a source and target** for sync — not just as a client-side cache. This means you can sync SQLite-to-SQLite, SQLite-to-SQL Server, SQLite-to-PostgreSQL, or SQLite-to-MySQL in any direction. The Datasync Toolkit has broader server-side database support (including NoSQL options), but the client is always SQLite. @@ -235,7 +236,7 @@ The Datasync Toolkit has a richer security model since it's designed as a server CoreSync is the right choice when: - **You need database-to-database sync** — no server API required -- **You want any-to-any database combinations** — SQLite ↔ SQL Server ↔ PostgreSQL in any direction +- **You want any-to-any database combinations** — SQLite ↔ SQL Server ↔ PostgreSQL ↔ MySQL in any direction - **You need SQLite as a full sync participant** — not just a client cache - **Your existing schema should stay unchanged** — no base classes or required fields - **You need broad .NET compatibility** — .NET Standard 2.0 supports everything from .NET Framework 4.6.1 to .NET 10 diff --git a/docs/comparison-with-debezium.md b/docs/comparison-with-debezium.md index d759e2a..49f767d 100644 --- a/docs/comparison-with-debezium.md +++ b/docs/comparison-with-debezium.md @@ -15,7 +15,7 @@ Both CoreSync and [Debezium](https://debezium.io/) deal with change data capture | **Language / Ecosystem** | .NET (C#) | Java (JVM / Kafka ecosystem) | | **Sync direction** | Bidirectional | Unidirectional (source to consumers) | | **Infrastructure required** | None beyond your app and databases | Kafka + Kafka Connect (primary mode), or standalone JVM server | -| **Supported databases** | SQLite, SQL Server, PostgreSQL | MySQL, MariaDB, PostgreSQL, SQL Server, Oracle, MongoDB, Db2, Cassandra, Spanner, Vitess, Informix | +| **Supported databases** | SQLite, SQL Server, PostgreSQL, MySQL | MySQL, MariaDB, PostgreSQL, SQL Server, Oracle, MongoDB, Db2, Cassandra, Spanner, Vitess, Informix | | **Change detection method** | Trigger-based or SQL Server native CT | Transaction log reading (binlog, WAL, redo logs) | | **Conflict resolution** | Built-in (Skip / ForceWrite with per-item control) | N/A (unidirectional — no conflicts) | | **Target framework** | .NET Standard 2.0 / .NET 8.0 | Java 11+ | @@ -103,6 +103,7 @@ Debezium's primary deployment (Kafka Connect) requires a running Kafka cluster, | SQLite | Yes | No | | SQL Server | Yes (custom triggers + native CT) | Yes | | PostgreSQL | Yes | Yes | +| MySQL | Yes | Yes | | MySQL / MariaDB | No | Yes | | Oracle | No | Yes | | MongoDB | No | Yes | @@ -200,7 +201,7 @@ CoreSync is the right choice when: - Desktop applications caching data locally - Field service apps that work without connectivity - Multi-site databases that need bidirectional replication -- Any .NET application synchronizing between SQLite, SQL Server, and/or PostgreSQL +- Any .NET application synchronizing between SQLite, SQL Server, PostgreSQL, and/or MySQL ## When to Choose Debezium diff --git a/docs/index.md b/docs/index.md index 5a4ae3b..3a37e7b 100644 --- a/docs/index.md +++ b/docs/index.md @@ -5,7 +5,7 @@ title: CoreSync Documentation # CoreSync -**CoreSync** is a .NET library for bidirectional data synchronization between databases. It supports SQLite, SQL Server, and PostgreSQL, letting you keep multiple database instances in sync — whether they're on the same machine or communicating over HTTP. +**CoreSync** is a .NET library for bidirectional data synchronization between databases. It supports SQLite, SQL Server, PostgreSQL, and MySQL, letting you keep multiple database instances in sync — whether they're on the same machine or communicating over HTTP. ## Pages diff --git a/src/CoreSync.MySql/CoreSync.MySql.csproj b/src/CoreSync.MySql/CoreSync.MySql.csproj new file mode 100644 index 0000000..f358f66 --- /dev/null +++ b/src/CoreSync.MySql/CoreSync.MySql.csproj @@ -0,0 +1,40 @@ + + + + 0.0.1-local + $(APPVEYOR_BUILD_VERSION) + netstandard2.0 + adospace + CoreSync is a .NET standard library that provides data synchronization functions between databases. This is the MySQL provider assembly. + Adolfo Marinucci + https://github.com/adospace/CoreSync + MIT + https://github.com/adospace/CoreSync + mysql mariadb data database sync synchronization .net + true + true + enable + latest + true + RS1036,NU1903,NU1902,NU1901,NU1904 + + + + true + $(NoWarn);CS1591 + + + + + + + + + + + + + + + + diff --git a/src/CoreSync.MySql/MySqlColumn.cs b/src/CoreSync.MySql/MySqlColumn.cs new file mode 100644 index 0000000..a94b1f6 --- /dev/null +++ b/src/CoreSync.MySql/MySqlColumn.cs @@ -0,0 +1,26 @@ +using System; + +namespace CoreSync.MySql +{ + internal class MySqlColumn + { + public MySqlColumn(string name, string dataType, string columnType, bool primaryKey = false) + { + Name = name; + DataType = dataType; + ColumnType = columnType; + IsPrimaryKey = primaryKey; + } + + public string Name { get; } + public string DataType { get; } + public string ColumnType { get; } + public bool IsPrimaryKey { get; } + + public bool IsGuidLike => + (DataType.Equals("char", StringComparison.OrdinalIgnoreCase) || + DataType.Equals("varchar", StringComparison.OrdinalIgnoreCase)) && + (ColumnType.Equals("char(36)", StringComparison.OrdinalIgnoreCase) || + ColumnType.Equals("varchar(36)", StringComparison.OrdinalIgnoreCase)); + } +} diff --git a/src/CoreSync.MySql/MySqlCommandExtensions.cs b/src/CoreSync.MySql/MySqlCommandExtensions.cs new file mode 100644 index 0000000..2eed93f --- /dev/null +++ b/src/CoreSync.MySql/MySqlCommandExtensions.cs @@ -0,0 +1,21 @@ +using MySqlConnector; +using System.Threading; +using System.Threading.Tasks; + +namespace CoreSync.MySql +{ + internal static class MySqlCommandExtensions + { + public static async Task ExecuteLongScalarAsync(this MySqlCommand command, CancellationToken cancellationToken = default) + { + var result = await command.ExecuteScalarAsync(cancellationToken); + + if (result == null || result == System.DBNull.Value) + { + return 0; + } + + return System.Convert.ToInt64(result); + } + } +} diff --git a/src/CoreSync.MySql/MySqlPrimaryColumnType.cs b/src/CoreSync.MySql/MySqlPrimaryColumnType.cs new file mode 100644 index 0000000..f88ee62 --- /dev/null +++ b/src/CoreSync.MySql/MySqlPrimaryColumnType.cs @@ -0,0 +1,9 @@ +namespace CoreSync.MySql +{ + internal enum MySqlPrimaryColumnType + { + Integer, + Text, + Blob + } +} diff --git a/src/CoreSync.MySql/MySqlSyncConfiguration.cs b/src/CoreSync.MySql/MySqlSyncConfiguration.cs new file mode 100644 index 0000000..65884f9 --- /dev/null +++ b/src/CoreSync.MySql/MySqlSyncConfiguration.cs @@ -0,0 +1,18 @@ +using JetBrains.Annotations; + +namespace CoreSync.MySql +{ + public class MySqlSyncConfiguration : SyncConfiguration + { + public string ConnectionString { get; } + + internal MySqlSyncConfiguration([NotNull] string connectionString, [NotNull] MySqlSyncTable[] tables) + : base(tables) + { + Validate.NotNullOrEmptyOrWhiteSpace(connectionString, nameof(connectionString)); + Validate.NotNullOrEmptyArray(tables, nameof(tables)); + + ConnectionString = connectionString; + } + } +} diff --git a/src/CoreSync.MySql/MySqlSyncConfigurationBuilder.cs b/src/CoreSync.MySql/MySqlSyncConfigurationBuilder.cs new file mode 100644 index 0000000..4717d3a --- /dev/null +++ b/src/CoreSync.MySql/MySqlSyncConfigurationBuilder.cs @@ -0,0 +1,118 @@ +using JetBrains.Annotations; +using System; +using System.Collections.Generic; +using System.ComponentModel.DataAnnotations.Schema; +using System.Linq; + +namespace CoreSync.MySql +{ + /// + /// Provides a fluent API for building a that defines which MySQL tables + /// participate in synchronization and how they behave. + /// + public class MySqlSyncConfigurationBuilder + { + private readonly string _connectionString; + private readonly List _tables = []; + + public MySqlSyncConfigurationBuilder([NotNull] string connectionString) + { + Validate.NotNullOrEmptyOrWhiteSpace(connectionString, nameof(connectionString)); + _connectionString = connectionString; + } + + public MySqlSyncConfigurationBuilder Table( + [NotNull] string name, + Type? recordType = null, + SyncDirection syncDirection = SyncDirection.UploadAndDownload, + bool skipInitialSnapshot = false, + string? selectIncrementalQuery = null, + string? customSnapshotQuery = null) + { + Validate.NotNullOrEmptyOrWhiteSpace(name, nameof(name)); + + name = name.Trim(); + if (_tables.Any(_ => string.CompareOrdinal(_.Name, name) == 0)) + { + throw new InvalidOperationException($"Table with name '{name}' already added"); + } + + _tables.Add(new MySqlSyncTable(name, recordType, syncDirection, skipInitialSnapshot, selectIncrementalQuery, customSnapshotQuery)); + return this; + } + + public MySqlSyncConfigurationBuilder Table( + string? name = null, + SyncDirection syncDirection = SyncDirection.UploadAndDownload, + bool skipInitialSnapshot = false, + string? selectIncrementalQuery = null, + string? customSnapshotQuery = null) + { + if (name == null) + { + var tableAttribute = (TableAttribute?)Attribute.GetCustomAttribute(typeof(T), typeof(TableAttribute)); + if (tableAttribute != null) + { + name = tableAttribute.Name; + } + } + + name ??= typeof(T).Name; + + if (_tables.Any(_ => string.CompareOrdinal(_.Name, name) == 0)) + { + throw new InvalidOperationException($"Table with name '{name}' already added"); + } + + return Table(name, typeof(T), syncDirection, skipInitialSnapshot, selectIncrementalQuery, customSnapshotQuery); + } + + public MySqlSyncConfigurationBuilder SkipColumns(params string[] columnNames) + { + var lastTable = _tables.LastOrDefault() + ?? throw new InvalidOperationException("SkipColumns requires a table"); + + lastTable.SkipColumns = columnNames ?? throw new ArgumentNullException(nameof(columnNames)); + return this; + } + + public MySqlSyncConfigurationBuilder SkipColumnsOnInsertOrUpdate(params string[] columnNames) + { + var lastTable = _tables.LastOrDefault() + ?? throw new InvalidOperationException("SkipColumnsOnInsertOrUpdate requires a table"); + + lastTable.SkipColumnsOnInsertOrUpdate = columnNames ?? throw new ArgumentNullException(nameof(columnNames)); + return this; + } + + public MySqlSyncConfigurationBuilder SelectIncrementalQuery(string selectIncrementalQuery) + { + if (string.IsNullOrWhiteSpace(selectIncrementalQuery)) + { + throw new ArgumentException($"'{nameof(selectIncrementalQuery)}' cannot be null or whitespace", nameof(selectIncrementalQuery)); + } + + var lastTable = _tables.LastOrDefault() + ?? throw new InvalidOperationException("SelectIncrementalQuery requires a table"); + + lastTable.SelectIncrementalQuery = selectIncrementalQuery; + return this; + } + + public MySqlSyncConfigurationBuilder CustomSnapshotQuery(string customSnapshotQuery) + { + if (string.IsNullOrWhiteSpace(customSnapshotQuery)) + { + throw new ArgumentException($"'{nameof(customSnapshotQuery)}' cannot be null or whitespace", nameof(customSnapshotQuery)); + } + + var lastTable = _tables.LastOrDefault() + ?? throw new InvalidOperationException("CustomSnapshotQuery requires a table"); + + lastTable.CustomSnapshotQuery = customSnapshotQuery; + return this; + } + + public MySqlSyncConfiguration Build() => new MySqlSyncConfiguration(_connectionString, _tables.ToArray()); + } +} diff --git a/src/CoreSync.MySql/MySqlSyncItem.cs b/src/CoreSync.MySql/MySqlSyncItem.cs new file mode 100644 index 0000000..8e0c583 --- /dev/null +++ b/src/CoreSync.MySql/MySqlSyncItem.cs @@ -0,0 +1,12 @@ +using System.Collections.Generic; + +namespace CoreSync.MySql +{ + internal class MySqlSyncItem : SyncItem + { + public MySqlSyncItem(MySqlSyncTable table, ChangeType changeType, Dictionary values) + : base(table.Name, changeType, values) + { + } + } +} diff --git a/src/CoreSync.MySql/MySqlSyncProvider.cs b/src/CoreSync.MySql/MySqlSyncProvider.cs new file mode 100644 index 0000000..253e657 --- /dev/null +++ b/src/CoreSync.MySql/MySqlSyncProvider.cs @@ -0,0 +1,933 @@ +using JetBrains.Annotations; +using MySqlConnector; +using System; +using System.Collections.Generic; +using System.ComponentModel.DataAnnotations.Schema; +using System.Linq; +using System.Reflection; +using System.Threading; +using System.Threading.Tasks; + +namespace CoreSync.MySql +{ + public class MySqlSyncProvider : ISyncProvider + { + private bool _initialized; + private Guid _storeId; + private readonly ISyncLogger? _logger; + + public MySqlSyncProvider(MySqlSyncConfiguration configuration, ProviderMode providerMode = ProviderMode.Bidirectional, ISyncLogger? logger = null) + { + Configuration = configuration; + ProviderMode = providerMode; + _logger = logger; + + if (configuration.Tables.Any(_ => _.SyncDirection != SyncDirection.UploadAndDownload) && + providerMode == ProviderMode.Bidirectional) + { + throw new InvalidOperationException("One or more table with sync direction different from Bidirectional: please must specify the provider mode to Local or Remote"); + } + } + + public MySqlSyncConfiguration Configuration { get; } + public ProviderMode ProviderMode { get; } + public string[]? SyncTableNames => Configuration.Tables.Select(_ => _.Name).ToArray(); + + public async Task ApplyChangesAsync([NotNull] SyncChangeSet changeSet, [CanBeNull] Func? onConflictFunc = null, CancellationToken cancellationToken = default) + { + Validate.NotNull(changeSet, nameof(changeSet)); + Validate.NotNullAnchor(changeSet.SourceAnchor, nameof(changeSet), nameof(changeSet.SourceAnchor)); + Validate.NotNull(changeSet.TargetAnchor, nameof(changeSet), nameof(changeSet.TargetAnchor)); + + await InitializeStoreAsync(cancellationToken); + + var now = DateTime.Now; + _logger?.Info($"[{_storeId}] Begin ApplyChanges(source={changeSet.SourceAnchor}, target={changeSet.TargetAnchor}, {changeSet.Items.Count} items)"); + + using var connection = new MySqlConnection(Configuration.ConnectionString); + await connection.OpenAsync(cancellationToken); + + using var cmd = connection.CreateCommand(); + using var tr = await connection.BeginTransactionAsync(cancellationToken); + cmd.Transaction = tr; + + try + { + cmd.CommandText = "SELECT MAX(`id`) FROM `__core_sync_ct`"; + cmd.Parameters.Clear(); + var version = await cmd.ExecuteLongScalarAsync(cancellationToken); + + cmd.CommandText = "SELECT MIN(`id`) FROM `__core_sync_ct`"; + cmd.Parameters.Clear(); + var minVersion = await cmd.ExecuteLongScalarAsync(cancellationToken); + + var remainingItems = changeSet.Items.ToList(); + int pass = 0; + + while (remainingItems.Count > 0) + { + pass++; + var failedItems = new List(); + int appliedInPass = 0; + + foreach (var item in remainingItems) + { + var table = (MySqlSyncTable?)Configuration.Tables.FirstOrDefault(_ => _.Name == item.TableName); + if (table == null) + { + continue; + } + + bool syncForceWrite = false; + var itemChangeType = item.ChangeType; + bool itemHandled = false; + + retryWrite: + cmd.Parameters.Clear(); + table.SetupCommand(cmd, itemChangeType, item.Values); + + int affectedRows = 0; + try + { + if (!syncForceWrite && + (itemChangeType == ChangeType.Update || itemChangeType == ChangeType.Delete) && + await HasConflictAsync(connection, tr, table, item, changeSet.TargetAnchor.Version, cancellationToken)) + { + affectedRows = 0; + } + else + { + await ExecuteSavepointCommandAsync(connection, tr, "SAVEPOINT sync_sp", cancellationToken); + affectedRows = await cmd.ExecuteNonQueryAsync(cancellationToken); + await ExecuteSavepointCommandAsync(connection, tr, "RELEASE SAVEPOINT sync_sp", cancellationToken); + + if (affectedRows > 0) + { + _logger?.Trace($"[{_storeId}] Successfully applied {item}"); + } + } + } + catch (OperationCanceledException) + { + throw; + } + catch (MySqlException ex) when (ex.Number == 1451 || ex.Number == 1452) + { + await ExecuteSavepointCommandAsync(connection, tr, "ROLLBACK TO SAVEPOINT sync_sp", cancellationToken); + _logger?.Warning($"[{_storeId}] FK violation applying {itemChangeType} on {item}, skipping: {ex.Message}"); + affectedRows = 0; + } + catch (Exception ex) + { + _logger?.Error($"Unable to {itemChangeType} item {item} to store for table {table}.{Environment.NewLine}{ex}{Environment.NewLine}Generated SQL:{Environment.NewLine}{cmd.CommandText}"); + throw new SynchronizationException($"Unable to {itemChangeType} item {item} to store for table {table}", ex); + } + + if (affectedRows == 0) + { + if (itemChangeType == ChangeType.Insert) + { + cmd.CommandText = table.SelectExistingQuery; + cmd.Parameters.Clear(); + var valueItem = item.Values[table.PrimaryColumnName]; + cmd.Parameters.Add(new MySqlParameter("@PrimaryColumnParameter", table.ConvertPrimaryKeyValue(valueItem.Value))); + if (1 == await cmd.ExecuteLongScalarAsync(cancellationToken) && !syncForceWrite) + { + itemChangeType = ChangeType.Update; + goto retryWrite; + } + + _logger?.Warning($"Unable to {item}: much probably there is a foreign key constraint issue logged before (pass {pass})"); + } + else if (itemChangeType == ChangeType.Update || itemChangeType == ChangeType.Delete) + { + if (syncForceWrite) + { + if (itemChangeType == ChangeType.Delete) + { + _logger?.Trace($"[{_storeId}] Insert on delete conflict occurred for {item}"); + itemHandled = true; + } + else + { + _logger?.Trace($"[{_storeId}] Insert on delete conflict occurred for {item}"); + itemChangeType = ChangeType.Insert; + goto retryWrite; + } + } + else + { + var res = onConflictFunc?.Invoke(item); + if (res.HasValue && res.Value == ConflictResolution.ForceWrite) + { + _logger?.Trace($"[{_storeId}] Force write on conflict occurred for {item}"); + syncForceWrite = true; + goto retryWrite; + } + + _logger?.Warning($"[{_storeId}] Skip conflict for {item}"); + itemHandled = true; + } + } + } + + if (affectedRows > 0) + { + itemHandled = true; + appliedInPass++; + + cmd.CommandText = "SELECT MAX(`id`) FROM `__core_sync_ct`"; + cmd.Parameters.Clear(); + var currentVersion = await cmd.ExecuteLongScalarAsync(cancellationToken); + + cmd.CommandText = "UPDATE `__core_sync_ct` SET `src` = @src WHERE `id` = @id"; + cmd.Parameters.Clear(); + cmd.Parameters.Add(new MySqlParameter("@src", changeSet.SourceAnchor.StoreId.ToString())); + cmd.Parameters.Add(new MySqlParameter("@id", currentVersion)); + await cmd.ExecuteNonQueryAsync(cancellationToken); + } + + if (!itemHandled) + { + failedItems.Add(item); + } + } + + if (failedItems.Count == 0 || appliedInPass == 0) + { + if (failedItems.Count > 0) + { + _logger?.Warning($"[{_storeId}] {failedItems.Count} item(s) could not be applied after {pass} pass(es) (possible unresolvable foreign key constraint)"); + } + + break; + } + + _logger?.Info($"[{_storeId}] Pass {pass}: applied {appliedInPass} item(s), retrying {failedItems.Count} remaining item(s)"); + remainingItems = failedItems; + } + + cmd.CommandText = "UPDATE `__core_sync_remote_anchor` SET `remote_version` = @remoteVersion WHERE `id` = @id"; + cmd.Parameters.Clear(); + cmd.Parameters.Add(new MySqlParameter("@remoteVersion", changeSet.SourceAnchor.Version)); + cmd.Parameters.Add(new MySqlParameter("@id", changeSet.SourceAnchor.StoreId.ToString())); + + if (0 == await cmd.ExecuteNonQueryAsync(cancellationToken)) + { + cmd.CommandText = "INSERT INTO `__core_sync_remote_anchor` (`id`, `remote_version`) VALUES (@id, @remoteVersion)"; + cmd.Parameters.Clear(); + cmd.Parameters.Add(new MySqlParameter("@id", changeSet.SourceAnchor.StoreId.ToString())); + cmd.Parameters.Add(new MySqlParameter("@remoteVersion", changeSet.SourceAnchor.Version)); + await cmd.ExecuteNonQueryAsync(cancellationToken); + } + + await tr.CommitAsync(cancellationToken); + + var resAnchor = new SyncAnchor(_storeId, version); + _logger?.Info($"[{_storeId}] Completed ApplyChanges(resAnchor={resAnchor}) in {(DateTime.Now - now).TotalMilliseconds}ms"); + + return resAnchor; + } + catch (Exception ex) + { + _logger?.Error($"[{_storeId}] Unable to complete the apply changes:{Environment.NewLine}{ex}"); + await tr.RollbackAsync(cancellationToken); + throw; + } + } + + public async Task SaveVersionForStoreAsync(Guid otherStoreId, long version, CancellationToken cancellationToken = default) + { + using var connection = new MySqlConnection(Configuration.ConnectionString); + await connection.OpenAsync(cancellationToken); + + using var cmd = connection.CreateCommand(); + using var tr = await connection.BeginTransactionAsync(cancellationToken); + cmd.Transaction = tr; + + try + { + cmd.CommandText = "UPDATE `__core_sync_remote_anchor` SET `local_version` = @version WHERE `id` = @id"; + cmd.Parameters.Add(new MySqlParameter("@version", version)); + cmd.Parameters.Add(new MySqlParameter("@id", otherStoreId.ToString())); + + if (0 == await cmd.ExecuteNonQueryAsync(cancellationToken)) + { + cmd.CommandText = "INSERT INTO `__core_sync_remote_anchor` (`id`, `local_version`) VALUES (@id, @version)"; + cmd.Parameters.Clear(); + cmd.Parameters.Add(new MySqlParameter("@id", otherStoreId.ToString())); + cmd.Parameters.Add(new MySqlParameter("@version", version)); + await cmd.ExecuteNonQueryAsync(cancellationToken); + } + + await tr.CommitAsync(cancellationToken); + _logger?.Trace($"[{_storeId}] Save version {version} for store {otherStoreId}"); + } + catch + { + await tr.RollbackAsync(cancellationToken); + throw; + } + } + + public async Task GetChangesAsync(Guid otherStoreId, SyncFilterParameter[]? syncFilterParameters = null, SyncDirection syncDirection = SyncDirection.UploadAndDownload, string[]? tables = null, CancellationToken cancellationToken = default) + { + syncFilterParameters ??= []; + + var tablesToSync = Configuration.ResolveTableFilter(tables); + var fromAnchor = await GetLastLocalAnchorForStoreAsync(otherStoreId, cancellationToken); + var now = DateTime.Now; + + _logger?.Info($"[{_storeId}] Begin GetChanges(from={otherStoreId}, syncDirection={syncDirection}, fromVersion={fromAnchor})"); + + using var connection = new MySqlConnection(Configuration.ConnectionString); + await connection.OpenAsync(cancellationToken); + + using var cmd = connection.CreateCommand(); + using var tr = await connection.BeginTransactionAsync(cancellationToken); + cmd.Transaction = tr; + + var items = new List(); + + try + { + cmd.CommandText = "SELECT MAX(`id`) FROM `__core_sync_ct`"; + cmd.Parameters.Clear(); + var version = await cmd.ExecuteLongScalarAsync(cancellationToken); + + cmd.CommandText = "SELECT MIN(`id`) FROM `__core_sync_ct`"; + cmd.Parameters.Clear(); + var minVersion = await cmd.ExecuteLongScalarAsync(cancellationToken); + + if (!fromAnchor.IsNull() && fromAnchor.Version < minVersion - 1) + { + throw new InvalidOperationException($"Unable to get changes, version of data requested ({fromAnchor}) is too old (min valid version {minVersion})"); + } + + foreach (var table in tablesToSync.Cast().Where(_ => _.Columns.Any())) + { + if (table.SyncDirection != SyncDirection.UploadAndDownload && table.SyncDirection != syncDirection) + { + continue; + } + + if (fromAnchor.IsNull() && !table.SkipInitialSnapshot) + { + if (string.IsNullOrWhiteSpace(table.InitialSnapshotQuery)) + { + throw new InvalidOperationException("InitialSnapshotQuery not specified for table"); + } + + cmd.CommandText = table.InitialSnapshotQuery; + cmd.Parameters.Clear(); + AddFilterParameters(cmd, syncFilterParameters); + + using var reader = await cmd.ExecuteReaderAsync(cancellationToken); + while (await reader.ReadAsync(cancellationToken)) + { + var values = Enumerable.Range(0, reader.FieldCount) + .ToDictionary(_ => reader.GetName(_), _ => GetValueFromRecord(table, reader.GetName(_), _, reader)); + items.Add(new MySqlSyncItem(table, ChangeType.Insert, values)); + _logger?.Trace($"[{_storeId}] Initial snapshot {items.Last()}"); + } + } + + if (!fromAnchor.IsNull()) + { + cmd.CommandText = table.IncrementalAddOrUpdatesQuery; + cmd.Parameters.Clear(); + AddFilterParameters(cmd, syncFilterParameters); + cmd.Parameters.Add(new MySqlParameter("@version", fromAnchor.Version)); + cmd.Parameters.Add(new MySqlParameter("@tableName", table.Name)); + cmd.Parameters.Add(new MySqlParameter("@sourceId", otherStoreId.ToString())); + + using (var reader = await cmd.ExecuteReaderAsync(cancellationToken)) + { + while (await reader.ReadAsync(cancellationToken)) + { + var values = Enumerable.Range(0, reader.FieldCount) + .ToDictionary(_ => reader.GetName(_), _ => GetValueFromRecord(table, reader.GetName(_), _, reader)); + + items.Add(new MySqlSyncItem(table, DetectChangeType(values), + values.Where(_ => _.Key != "__op").ToDictionary(_ => _.Key, _ => _.Value == DBNull.Value ? null : _.Value))); + _logger?.Trace($"[{_storeId}] Incremental add or update {items.Last()}"); + } + } + + cmd.CommandText = table.IncrementalDeletesQuery; + cmd.Parameters.Clear(); + cmd.Parameters.Add(new MySqlParameter("@tableName", table.Name)); + cmd.Parameters.Add(new MySqlParameter("@version", fromAnchor.Version)); + cmd.Parameters.Add(new MySqlParameter("@sourceId", otherStoreId.ToString())); + + using (var reader = await cmd.ExecuteReaderAsync(cancellationToken)) + { + while (await reader.ReadAsync(cancellationToken)) + { + var values = Enumerable.Range(0, reader.FieldCount) + .ToDictionary(_ => reader.GetName(_), _ => GetValueFromRecord(table, reader.GetName(_), _, reader)); + items.Add(new MySqlSyncItem(table, ChangeType.Delete, values)); + _logger?.Trace($"[{_storeId}] Incremental delete {items.Last()}"); + } + } + } + } + + await tr.CommitAsync(cancellationToken); + + var resChangeSet = new SyncChangeSet(new SyncAnchor(_storeId, version), await GetLastRemoteAnchorForStoreAsync(otherStoreId, cancellationToken), items); + _logger?.Info($"[{_storeId}] Completed GetChanges(to={version}, {items.Count} items) in {(DateTime.Now - now).TotalMilliseconds}ms"); + + return resChangeSet; + } + catch (Exception ex) + { + _logger?.Error($"[{_storeId}] Unable to complete GetChanges(from={otherStoreId}):{Environment.NewLine}{ex}"); + try { await tr.RollbackAsync(cancellationToken); } catch { } + throw; + } + } + + private async Task GetLastLocalAnchorForStoreAsync(Guid otherStoreId, CancellationToken cancellationToken = default) + { + await InitializeStoreAsync(cancellationToken); + + using var connection = new MySqlConnection(Configuration.ConnectionString); + await connection.OpenAsync(cancellationToken); + + using var cmd = connection.CreateCommand(); + cmd.CommandText = "SELECT `local_version` FROM `__core_sync_remote_anchor` WHERE `id` = @id"; + cmd.Parameters.Add(new MySqlParameter("@id", otherStoreId.ToString())); + + var version = await cmd.ExecuteScalarAsync(cancellationToken); + if (version == null || version == DBNull.Value) + { + return SyncAnchor.Null; + } + + return new SyncAnchor(_storeId, Convert.ToInt64(version)); + } + + private async Task GetLastRemoteAnchorForStoreAsync(Guid otherStoreId, CancellationToken cancellationToken = default) + { + using var connection = new MySqlConnection(Configuration.ConnectionString); + await connection.OpenAsync(cancellationToken); + + using var cmd = connection.CreateCommand(); + cmd.CommandText = "SELECT `remote_version` FROM `__core_sync_remote_anchor` WHERE `id` = @id"; + cmd.Parameters.Add(new MySqlParameter("@id", otherStoreId.ToString())); + + var version = await cmd.ExecuteScalarAsync(cancellationToken); + if (version == null || version == DBNull.Value) + { + return SyncAnchor.Null; + } + + return new SyncAnchor(otherStoreId, Convert.ToInt64(version)); + } + + public async Task GetStoreIdAsync(CancellationToken cancellationToken = default) + { + await InitializeStoreAsync(cancellationToken); + return _storeId; + } + + private static ChangeType DetectChangeType(Dictionary values) + { + if (values.TryGetValue("__op", out var syncChangeOperation)) + { + return syncChangeOperation?.ToString() switch + { + "I" => ChangeType.Insert, + "U" => ChangeType.Update, + "D" => ChangeType.Delete, + _ => throw new NotSupportedException(), + }; + } + + return ChangeType.Insert; + } + + private static object? GetValueFromRecord(MySqlSyncTable table, string columnName, int columnOrdinal, MySqlDataReader reader) + { + if (reader.IsDBNull(columnOrdinal)) + { + return null; + } + + if (table.Columns.TryGetValue(columnName, out var column) && column.IsGuidLike) + { + var value = reader.GetValue(columnOrdinal); + if (value is Guid guidValue) + { + return guidValue.ToString("D").ToUpperInvariant(); + } + + if (Guid.TryParse(value.ToString(), out var parsedGuid)) + { + return parsedGuid.ToString("D").ToUpperInvariant(); + } + } + + if (table.RecordType == null) + { + return reader.GetValue(columnOrdinal); + } + + var property = table.RecordType.GetProperty(columnName); + if (property != null) + { + return GetValueFromRecord(reader, columnOrdinal, property.PropertyType); + } + + property = table.RecordType.GetProperties().FirstOrDefault(_ => + { + var columnAttribute = _.GetCustomAttribute(false); + return columnAttribute?.Name == columnName; + }); + + if (property != null) + { + return GetValueFromRecord(reader, columnOrdinal, property.PropertyType); + } + + return reader.GetValue(columnOrdinal); + } + + private static object GetValueFromRecord(MySqlDataReader reader, int columnOrdinal, Type propertyType) + { + if (propertyType == typeof(string)) + { + return reader.GetString(columnOrdinal); + } + + if (propertyType == typeof(DateTime)) + { + var value = reader.GetValue(columnOrdinal); + return value is DateTime dateTime + ? dateTime + : DateTime.Parse(value.ToString()!); + } + + if (propertyType == typeof(int)) + { + return reader.GetInt32(columnOrdinal); + } + + if (propertyType == typeof(bool)) + { + return reader.GetBoolean(columnOrdinal); + } + + if (propertyType == typeof(byte)) + { + return reader.GetByte(columnOrdinal); + } + + if (propertyType == typeof(char)) + { + return reader.GetChar(columnOrdinal); + } + + if (propertyType == typeof(short)) + { + return reader.GetInt16(columnOrdinal); + } + + if (propertyType == typeof(long)) + { + return reader.GetInt64(columnOrdinal); + } + + if (propertyType == typeof(decimal)) + { + return reader.GetDecimal(columnOrdinal); + } + + if (propertyType == typeof(double)) + { + return reader.GetDouble(columnOrdinal); + } + + if (propertyType == typeof(float)) + { + return reader.GetFloat(columnOrdinal); + } + + if (propertyType == typeof(Guid)) + { + var value = reader.GetValue(columnOrdinal); + if (value is Guid guidValue) + { + return guidValue; + } + + return Guid.Parse(value.ToString()!); + } + + return reader.GetValue(columnOrdinal); + } + + private async Task InitializeStoreAsync(CancellationToken cancellationToken = default) + { + if (_initialized) + { + return; + } + + using var connection = new MySqlConnection(Configuration.ConnectionString); + await connection.OpenAsync(cancellationToken); + + using var cmd = connection.CreateCommand(); + cmd.CommandText = @"CREATE TABLE IF NOT EXISTS `__core_sync_ct` ( +`id` BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY, +`tbl` TEXT NOT NULL, +`op` CHAR(1) NOT NULL, +`pk_integer` BIGINT NULL, +`pk_text` TEXT NULL, +`pk_blob` BLOB NULL, +`src` TEXT NULL +)"; + await cmd.ExecuteNonQueryAsync(cancellationToken); + + await CreateIndexIfMissingAsync(connection, "__core_sync_ct_pk_integer_index", "CREATE INDEX `__core_sync_ct_pk_integer_index` ON `__core_sync_ct`(`pk_integer`)", cancellationToken); + await CreateIndexIfMissingAsync(connection, "__core_sync_ct_pk_text_index", "CREATE INDEX `__core_sync_ct_pk_text_index` ON `__core_sync_ct`(`pk_text`(255))", cancellationToken); + await CreateIndexIfMissingAsync(connection, "__core_sync_ct_pk_blob_index", "CREATE INDEX `__core_sync_ct_pk_blob_index` ON `__core_sync_ct`(`pk_blob`(255))", cancellationToken); + + cmd.CommandText = @"CREATE TABLE IF NOT EXISTS `__core_sync_remote_anchor` ( +`id` CHAR(36) NOT NULL PRIMARY KEY, +`local_version` BIGINT NULL, +`remote_version` BIGINT NULL +)"; + await cmd.ExecuteNonQueryAsync(cancellationToken); + + cmd.CommandText = @"CREATE TABLE IF NOT EXISTS `__core_sync_local_id` ( +`id` CHAR(36) NOT NULL PRIMARY KEY +)"; + await cmd.ExecuteNonQueryAsync(cancellationToken); + + cmd.CommandText = "SELECT `id` FROM `__core_sync_local_id` LIMIT 1"; + var localId = await cmd.ExecuteScalarAsync(cancellationToken); + if (localId == null) + { + localId = Guid.NewGuid().ToString(); + cmd.CommandText = "INSERT INTO `__core_sync_local_id` (`id`) VALUES (@id)"; + cmd.Parameters.Clear(); + cmd.Parameters.Add(new MySqlParameter("@id", localId)); + if (1 != await cmd.ExecuteNonQueryAsync(cancellationToken)) + { + throw new InvalidOperationException(); + } + + cmd.Parameters.Clear(); + } + + _storeId = Guid.Parse(localId.ToString()!); + + foreach (var table in Configuration.Tables.Cast()) + { + cmd.CommandText = @" +SELECT c.COLUMN_NAME, c.DATA_TYPE, c.COLUMN_TYPE, + CASE WHEN pk.COLUMN_NAME IS NOT NULL THEN TRUE ELSE FALSE END AS IS_PRIMARY_KEY +FROM INFORMATION_SCHEMA.COLUMNS c +LEFT JOIN ( + SELECT kcu.COLUMN_NAME + FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS tc + JOIN INFORMATION_SCHEMA.KEY_COLUMN_USAGE kcu + ON tc.CONSTRAINT_NAME = kcu.CONSTRAINT_NAME + AND tc.TABLE_SCHEMA = kcu.TABLE_SCHEMA + AND tc.TABLE_NAME = kcu.TABLE_NAME + WHERE tc.TABLE_SCHEMA = DATABASE() + AND tc.TABLE_NAME = @tableName + AND tc.CONSTRAINT_TYPE = 'PRIMARY KEY' +) pk ON c.COLUMN_NAME = pk.COLUMN_NAME +WHERE c.TABLE_SCHEMA = DATABASE() + AND c.TABLE_NAME = @tableName +ORDER BY c.ORDINAL_POSITION"; + cmd.Parameters.Clear(); + cmd.Parameters.Add(new MySqlParameter("@tableName", table.Name)); + + using var reader = await cmd.ExecuteReaderAsync(cancellationToken); + while (await reader.ReadAsync(cancellationToken)) + { + var colName = reader.GetString(0); + var dataType = reader.GetString(1); + var columnType = reader.GetString(2); + var pk = reader.GetBoolean(3); + + if (string.CompareOrdinal(colName, "__op") == 0) + { + throw new NotSupportedException($"Unable to synchronize table '{table.Name}': one column has a reserved name '__op'"); + } + + table.Columns.Add(colName, new MySqlColumn(colName, dataType, columnType, pk)); + } + + if (table.Columns.Count == 0) + { + throw new InvalidOperationException($"Unable to configure table '{table}': does it exist with at least one column?"); + } + + if (table.Columns.Count(_ => _.Value.IsPrimaryKey) == 0) + { + throw new NotSupportedException($"Unable to configure table '{table}': no primary key defined"); + } + + if (table.Columns.Count(_ => _.Value.IsPrimaryKey) > 1) + { + throw new NotSupportedException($"Unable to configure table '{table}': it has more than one column as primary key"); + } + } + + _initialized = true; + } + + public async Task ApplyProvisionAsync(CancellationToken cancellationToken = default) + { + await InitializeStoreAsync(cancellationToken); + + using var connection = new MySqlConnection(Configuration.ConnectionString); + await connection.OpenAsync(cancellationToken); + + using var cmd = connection.CreateCommand(); + foreach (var table in Configuration.Tables.Cast().Where(_ => _.Columns.Any())) + { + if (table.SyncDirection == SyncDirection.UploadAndDownload || + (table.SyncDirection == SyncDirection.UploadOnly && ProviderMode == ProviderMode.Local) || + (table.SyncDirection == SyncDirection.DownloadOnly && ProviderMode == ProviderMode.Remote)) + { + await SetupTableForFullChangeDetection(table, cmd, cancellationToken); + } + else + { + await SetupTableForUpdatesOrDeletesOnly(table, cmd, cancellationToken); + } + } + } + + private async Task SetupTableForFullChangeDetection(MySqlSyncTable table, MySqlCommand cmd, CancellationToken cancellationToken = default) + { + foreach (var op in new[] { "INSERT", "UPDATE", "DELETE" }) + { + await CreateTriggerAsync(table, cmd, op, cancellationToken); + } + } + + private async Task SetupTableForUpdatesOrDeletesOnly(MySqlSyncTable table, MySqlCommand cmd, CancellationToken cancellationToken = default) + { + foreach (var op in new[] { "UPDATE", "DELETE" }) + { + await CreateTriggerAsync(table, cmd, op, cancellationToken); + } + } + + private async Task CreateTriggerAsync(MySqlSyncTable table, MySqlCommand cmd, string op, CancellationToken cancellationToken) + { + var triggerName = $"__{table.Name}_ct_{op.ToLowerInvariant()}__"; + var rowReference = op == "DELETE" ? "OLD" : "NEW"; + + cmd.CommandText = $"DROP TRIGGER IF EXISTS `{triggerName}`"; + cmd.Parameters.Clear(); + await cmd.ExecuteNonQueryAsync(cancellationToken); + + cmd.CommandText = + $@"CREATE TRIGGER `{triggerName}` +AFTER {op} ON `{table.Name}` +FOR EACH ROW +INSERT INTO `__core_sync_ct` (`tbl`, `op`, `pk_{table.PrimaryColumnType.ToString().ToLowerInvariant()}`) +VALUES ('{table.Name}', '{op[0]}', {rowReference}.`{table.PrimaryColumnName}`)"; + await cmd.ExecuteNonQueryAsync(cancellationToken); + } + + public async Task RemoveProvisionAsync(CancellationToken cancellationToken = default) + { + using var connection = new MySqlConnection(Configuration.ConnectionString); + await connection.OpenAsync(cancellationToken); + + using var cmd = connection.CreateCommand(); + cmd.CommandText = "DROP TABLE IF EXISTS `__core_sync_ct`"; + await cmd.ExecuteNonQueryAsync(cancellationToken); + + cmd.CommandText = "DROP TABLE IF EXISTS `__core_sync_remote_anchor`"; + await cmd.ExecuteNonQueryAsync(cancellationToken); + + cmd.CommandText = "DROP TABLE IF EXISTS `__core_sync_local_id`"; + await cmd.ExecuteNonQueryAsync(cancellationToken); + + foreach (var table in Configuration.Tables.Cast()) + { + await DisableChangeTrackingForTable(cmd, table.Name, cancellationToken); + } + } + + public async Task GetSyncVersionAsync(CancellationToken cancellationToken = default) + { + await InitializeStoreAsync(cancellationToken); + + using var connection = new MySqlConnection(Configuration.ConnectionString); + try + { + await connection.OpenAsync(cancellationToken); + + using var cmd = connection.CreateCommand(); + using var tr = await connection.BeginTransactionAsync(cancellationToken); + cmd.Transaction = tr; + + cmd.CommandText = "SELECT MAX(`id`) FROM `__core_sync_ct`"; + var version = await cmd.ExecuteLongScalarAsync(cancellationToken); + + cmd.CommandText = "SELECT MIN(`id`) FROM `__core_sync_ct`"; + var minVersion = await cmd.ExecuteLongScalarAsync(cancellationToken); + + return new SyncVersion(version, minVersion); + } + catch (Exception ex) + { + throw new InvalidOperationException("Unable to get current/minimum version from store", ex); + } + } + + public async Task ApplyRetentionPolicyAsync(int minVersion, CancellationToken cancellationToken = default) + { + await InitializeStoreAsync(cancellationToken); + + using var connection = new MySqlConnection(Configuration.ConnectionString); + try + { + await connection.OpenAsync(cancellationToken); + + using var cmd = connection.CreateCommand(); + using var tr = await connection.BeginTransactionAsync(cancellationToken); + cmd.Transaction = tr; + + try + { + cmd.CommandText = "DELETE FROM `__core_sync_ct` WHERE `id` < @minVersion"; + cmd.Parameters.Add(new MySqlParameter("@minVersion", minVersion)); + await cmd.ExecuteNonQueryAsync(cancellationToken); + + cmd.CommandText = "SELECT MAX(`id`) FROM `__core_sync_ct`"; + cmd.Parameters.Clear(); + var version = await cmd.ExecuteLongScalarAsync(cancellationToken); + + cmd.CommandText = "SELECT MIN(`id`) FROM `__core_sync_ct`"; + var newMinVersion = await cmd.ExecuteLongScalarAsync(cancellationToken); + + await tr.CommitAsync(cancellationToken); + return new SyncVersion(version, newMinVersion); + } + catch + { + await tr.RollbackAsync(cancellationToken); + throw; + } + } + catch (Exception ex) + { + throw new InvalidOperationException("Unable to apply retention policy to store", ex); + } + } + + public async Task EnableChangeTrackingForTable(string name, CancellationToken cancellationToken = default) + { + Validate.NotNullOrEmptyOrWhiteSpace(name, nameof(name)); + + var table = Configuration.Tables.Cast().FirstOrDefault(_ => _.Name == name); + if (table == null) + { + throw new InvalidOperationException($"Table '{name}' is not in sync configuration"); + } + + using var connection = new MySqlConnection(Configuration.ConnectionString); + await connection.OpenAsync(cancellationToken); + + using var cmd = connection.CreateCommand(); + if (table.SyncDirection == SyncDirection.UploadAndDownload || + (table.SyncDirection == SyncDirection.UploadOnly && ProviderMode == ProviderMode.Local) || + (table.SyncDirection == SyncDirection.DownloadOnly && ProviderMode == ProviderMode.Remote)) + { + await SetupTableForFullChangeDetection(table, cmd, cancellationToken); + } + else + { + await SetupTableForUpdatesOrDeletesOnly(table, cmd, cancellationToken); + } + } + + public async Task DisableChangeTrackingForTable(string name, CancellationToken cancellationToken = default) + { + Validate.NotNullOrEmptyOrWhiteSpace(name, nameof(name)); + + using var connection = new MySqlConnection(Configuration.ConnectionString); + await connection.OpenAsync(cancellationToken); + + using var cmd = connection.CreateCommand(); + await DisableChangeTrackingForTable(cmd, name, cancellationToken); + } + + private async Task DisableChangeTrackingForTable(MySqlCommand cmd, string tableName, CancellationToken cancellationToken) + { + foreach (var op in new[] { "insert", "update", "delete" }) + { + cmd.CommandText = $"DROP TRIGGER IF EXISTS `__{tableName}_ct_{op}__`"; + cmd.Parameters.Clear(); + await cmd.ExecuteNonQueryAsync(cancellationToken); + } + } + + private static void AddFilterParameters(MySqlCommand cmd, SyncFilterParameter[] syncFilterParameters) + { + foreach (var syncFilterParameter in syncFilterParameters) + { + cmd.Parameters.Add(new MySqlParameter(syncFilterParameter.Name, syncFilterParameter.Value)); + } + } + + private static async Task ExecuteSavepointCommandAsync(MySqlConnection connection, MySqlTransaction transaction, string commandText, CancellationToken cancellationToken) + { + using var savepointCmd = connection.CreateCommand(); + savepointCmd.Transaction = transaction; + savepointCmd.CommandText = commandText; + await savepointCmd.ExecuteNonQueryAsync(cancellationToken); + } + + private static async Task HasConflictAsync(MySqlConnection connection, MySqlTransaction transaction, MySqlSyncTable table, SyncItem item, long targetVersion, CancellationToken cancellationToken) + { + if (!item.Values.TryGetValue(table.PrimaryColumnName, out var primaryValue)) + { + return false; + } + + using var conflictCmd = connection.CreateCommand(); + conflictCmd.Transaction = transaction; + conflictCmd.CommandText = + $@"SELECT MAX(`id`) FROM `__core_sync_ct` +WHERE `pk_{table.PrimaryColumnType.ToString().ToLowerInvariant()}` = @primaryKey + AND `tbl` = @tableName"; + conflictCmd.Parameters.Add(new MySqlParameter("@primaryKey", table.ConvertPrimaryKeyValue(primaryValue.Value))); + conflictCmd.Parameters.Add(new MySqlParameter("@tableName", table.Name)); + + return await conflictCmd.ExecuteLongScalarAsync(cancellationToken) > targetVersion; + } + + private static async Task CreateIndexIfMissingAsync(MySqlConnection connection, string indexName, string createSql, CancellationToken cancellationToken) + { + using var existsCmd = connection.CreateCommand(); + existsCmd.CommandText = @" +SELECT COUNT(*) +FROM INFORMATION_SCHEMA.STATISTICS +WHERE TABLE_SCHEMA = DATABASE() + AND TABLE_NAME = '__core_sync_ct' + AND INDEX_NAME = @indexName"; + existsCmd.Parameters.Add(new MySqlParameter("@indexName", indexName)); + if (await existsCmd.ExecuteLongScalarAsync(cancellationToken) > 0) + { + return; + } + + using var createCmd = connection.CreateCommand(); + createCmd.CommandText = createSql; + await createCmd.ExecuteNonQueryAsync(cancellationToken); + } + } +} diff --git a/src/CoreSync.MySql/MySqlSyncTable.cs b/src/CoreSync.MySql/MySqlSyncTable.cs new file mode 100644 index 0000000..c12684d --- /dev/null +++ b/src/CoreSync.MySql/MySqlSyncTable.cs @@ -0,0 +1,145 @@ +using MySqlConnector; +using System; +using System.Collections.Generic; +using System.Linq; + +namespace CoreSync.MySql +{ + public class MySqlSyncTable : SyncTable + { + internal MySqlSyncTable(string name, Type? recordType = null, SyncDirection syncDirection = SyncDirection.UploadAndDownload, bool skipInitialSnapshot = false, string? selectIncrementalQuery = null, string? customSnapshotQuery = null) + : base(name, syncDirection, skipInitialSnapshot, selectIncrementalQuery, customSnapshotQuery) + { + Validate.NotNullOrEmptyOrWhiteSpace(name, nameof(name)); + + RecordType = recordType; + } + + internal string[] SkipColumns { get; set; } = Array.Empty(); + + internal string[] SkipColumnsOnInsertOrUpdate { get; set; } = Array.Empty(); + + public Type? RecordType { get; } + + internal string PrimaryColumnName => Columns.First(_ => _.Value.IsPrimaryKey).Value.Name; + + internal MySqlPrimaryColumnType PrimaryColumnType => GetPrimaryColumnType(Columns[PrimaryColumnName].DataType); + + internal Dictionary Columns { get; set; } = []; + + internal string InitialSnapshotQuery => (CustomSnapshotQuery ?? SelectIncrementalQuery) ?? $"SELECT * FROM `{Name}`"; + + private string SelectQueryWithFilter => SelectIncrementalQuery != null ? $"({SelectIncrementalQuery})" : $"`{Name}`"; + + internal string IncrementalAddOrUpdatesQuery => + $@"SELECT DISTINCT {string.Join(",", Columns.Keys.Except(SkipColumns).Select(_ => "T.`" + _ + "`"))}, CT.op AS __op +FROM {SelectQueryWithFilter} AS T +INNER JOIN `__core_sync_ct` AS CT ON T.`{PrimaryColumnName}` = CT.`pk_{PrimaryColumnType.ToString().ToLowerInvariant()}` +WHERE CT.`id` > @version AND CT.`tbl` = @tableName AND (CT.`src` IS NULL OR CT.`src` != @sourceId)"; + + internal string IncrementalDeletesQuery => + $@"SELECT `pk_{PrimaryColumnType.ToString().ToLowerInvariant()}` AS `{PrimaryColumnName}` +FROM `__core_sync_ct` +WHERE `tbl` = @tableName AND `id` > @version AND `op` = 'D' AND (`src` IS NULL OR `src` != @sourceId)"; + + internal string SelectExistingQuery => + $@"SELECT COUNT(*) FROM `{Name}` +WHERE `{PrimaryColumnName}` = @PrimaryColumnParameter"; + + internal object ConvertPrimaryKeyValue(object? value) => ConvertValueForColumn(PrimaryColumnName, value); + + private static string[] IntegerTypes => + ["tinyint", "smallint", "mediumint", "int", "integer", "bigint"]; + + private MySqlPrimaryColumnType GetPrimaryColumnType(string dataType) + { + if (IntegerTypes.Contains(dataType, StringComparer.OrdinalIgnoreCase)) + { + return MySqlPrimaryColumnType.Integer; + } + + return dataType.ToLowerInvariant() switch + { + "char" or "varchar" or "tinytext" or "text" or "mediumtext" or "longtext" => MySqlPrimaryColumnType.Text, + "binary" or "varbinary" or "tinyblob" or "blob" or "mediumblob" or "longblob" => MySqlPrimaryColumnType.Blob, + _ => throw new NotSupportedException($"Table {Name} primary key type '{dataType}'"), + }; + } + + private object ConvertValueForColumn(string columnName, object? value) + { + if (value == null) + { + return DBNull.Value; + } + + if (Columns.TryGetValue(columnName, out var column)) + { + if (column.IsGuidLike) + { + if (value is Guid guidValue) + { + return guidValue.ToString("D").ToUpperInvariant(); + } + + if (value is string stringValue && Guid.TryParse(stringValue, out var parsedGuid)) + { + return parsedGuid.ToString("D").ToUpperInvariant(); + } + } + + if ((column.DataType.Equals("datetime", StringComparison.OrdinalIgnoreCase) || + column.DataType.Equals("timestamp", StringComparison.OrdinalIgnoreCase) || + column.DataType.Equals("date", StringComparison.OrdinalIgnoreCase)) && + value is string stringDate && + DateTime.TryParse(stringDate, out var parsedDate)) + { + return parsedDate; + } + } + + return value; + } + + internal void SetupCommand(MySqlCommand cmd, ChangeType itemChangeType, Dictionary syncItemValues) + { + var allColumnsExceptSkipColumns = Columns.Keys.Except(SkipColumns.Concat(SkipColumnsOnInsertOrUpdate)).ToArray(); + var valuesForValidColumns = syncItemValues + .Where(value => allColumnsExceptSkipColumns.Any(_ => StringComparer.OrdinalIgnoreCase.Compare(_, value.Key) == 0)) + .ToList(); + + switch (itemChangeType) + { + case ChangeType.Insert: + cmd.CommandText = $@"INSERT IGNORE INTO `{Name}` ({string.Join(", ", valuesForValidColumns.Select(_ => "`" + _.Key + "`"))}) +VALUES ({string.Join(", ", valuesForValidColumns.Select((_, index) => $"@p{index}"))});"; + + foreach (var valueItem in valuesForValidColumns.Select((value, index) => (value, index))) + { + cmd.Parameters.Add(new MySqlParameter($"@p{valueItem.index}", ConvertValueForColumn(valueItem.value.Key, valueItem.value.Value.Value))); + } + break; + + case ChangeType.Update: + cmd.CommandText = $@"UPDATE `{Name}` +SET {string.Join(", ", valuesForValidColumns.Select((_, index) => $"`{_.Key}` = @p{index}"))} +WHERE `{Name}`.`{PrimaryColumnName}` = @PrimaryColumnParameter"; + + foreach (var valueItem in valuesForValidColumns.Select((value, index) => (value, index))) + { + cmd.Parameters.Add(new MySqlParameter($"@p{valueItem.index}", ConvertValueForColumn(valueItem.value.Key, valueItem.value.Value.Value))); + } + + cmd.Parameters.Add(new MySqlParameter("@PrimaryColumnParameter", ConvertValueForColumn(PrimaryColumnName, syncItemValues[PrimaryColumnName].Value))); + break; + + case ChangeType.Delete: + cmd.CommandText = $@"DELETE FROM `{Name}` +WHERE `{Name}`.`{PrimaryColumnName}` = @PrimaryColumnParameter"; + + cmd.Parameters.Add(new MySqlParameter("@PrimaryColumnParameter", ConvertValueForColumn(PrimaryColumnName, syncItemValues[PrimaryColumnName].Value))); + break; + } + } + } +} diff --git a/src/CoreSync.Tests/CoreSync.Tests.csproj b/src/CoreSync.Tests/CoreSync.Tests.csproj index a2efc3e..cb67128 100644 --- a/src/CoreSync.Tests/CoreSync.Tests.csproj +++ b/src/CoreSync.Tests/CoreSync.Tests.csproj @@ -1,7 +1,7 @@  - net8.0 + net9.0 true RS1036,NU1903,NU1902,NU1901,NU1904 enable @@ -21,16 +21,19 @@ + + - + + diff --git a/src/CoreSync.Tests/Data/DatabaseType.cs b/src/CoreSync.Tests/Data/DatabaseType.cs index 09f3c99..9414e43 100644 --- a/src/CoreSync.Tests/Data/DatabaseType.cs +++ b/src/CoreSync.Tests/Data/DatabaseType.cs @@ -10,6 +10,8 @@ public enum DatabaseType Sqlite, - PostgreSQL + PostgreSQL, + + MySql } } diff --git a/src/CoreSync.Tests/Data/DesignTimeMySqlBlogDbContextFactory.cs b/src/CoreSync.Tests/Data/DesignTimeMySqlBlogDbContextFactory.cs new file mode 100644 index 0000000..da9cceb --- /dev/null +++ b/src/CoreSync.Tests/Data/DesignTimeMySqlBlogDbContextFactory.cs @@ -0,0 +1,12 @@ +using Microsoft.EntityFrameworkCore.Design; + +namespace CoreSync.Tests.Data +{ + public class DesignTimeMySqlBlogDbContextFactory : IDesignTimeDbContextFactory + { + public MySqlBlogDbContext CreateDbContext(string[] args) + { + return new MySqlBlogDbContext(IntegrationTests.MySqlConnectionString); + } + } +} diff --git a/src/CoreSync.Tests/Data/MySqlBlogDbContext.cs b/src/CoreSync.Tests/Data/MySqlBlogDbContext.cs new file mode 100644 index 0000000..92ba2a4 --- /dev/null +++ b/src/CoreSync.Tests/Data/MySqlBlogDbContext.cs @@ -0,0 +1,58 @@ +using Microsoft.EntityFrameworkCore; +using System; + +namespace CoreSync.Tests.Data +{ + public class MySqlBlogDbContext : BlogDbContext + { + private static readonly MySqlServerVersion ServerVersion = new(new Version(8, 4, 0)); + + public MySqlBlogDbContext(string connectionString) : base(connectionString) + { + } + + protected override void OnConfiguring(DbContextOptionsBuilder optionsBuilder) + { + optionsBuilder.UseMySql(ConnectionString, ServerVersion); + } + + protected override void OnModelCreating(ModelBuilder modelBuilder) + { + base.OnModelCreating(modelBuilder); + + modelBuilder.Entity(entity => + { + entity.ToTable("Users"); + entity.Property(e => e.Email).HasMaxLength(255); + entity.Property(e => e.Created).HasColumnType("datetime(6)"); + }); + + modelBuilder.Entity(entity => + { + entity.ToTable("Posts"); + entity.Property(e => e.Id).HasColumnType("char(36)"); + entity.Property(e => e.Updated).HasColumnType("datetime(6)"); + entity.Property("AuthorEmail").HasMaxLength(255); + entity.HasOne(e => e.Author) + .WithMany(u => u.Posts) + .OnDelete(DeleteBehavior.Cascade); + }); + + modelBuilder.Entity(entity => + { + entity.ToTable("Comments"); + entity.Property(e => e.Id).HasColumnType("char(36)"); + entity.Property(e => e.Created).HasColumnType("datetime(6)"); + entity.Property("PostId").HasColumnType("char(36)"); + entity.Property("ReplyToId").HasColumnType("char(36)"); + entity.Property("AuthorEmail").HasMaxLength(255); + }); + } + + public override BlogDbContext Refresh() + { + Dispose(); + return new MySqlBlogDbContext(ConnectionString); + } + } +} diff --git a/src/CoreSync.Tests/IntegrationTests.Definitions.cs b/src/CoreSync.Tests/IntegrationTests.Definitions.cs index 5f265d3..f1deda6 100644 --- a/src/CoreSync.Tests/IntegrationTests.Definitions.cs +++ b/src/CoreSync.Tests/IntegrationTests.Definitions.cs @@ -2,6 +2,7 @@ using CoreSync.SqlServer; using CoreSync.SqlServerCT; using CoreSync.PostgreSQL; +using CoreSync.MySql; using CoreSync.Tests.Data; using Microsoft.EntityFrameworkCore; using Microsoft.VisualStudio.TestTools.UnitTesting; @@ -1829,6 +1830,506 @@ public async Task Test_Sqlite_PostgreSQL_TestSynchronizationAfterDisabledChangeT await TestSynchronizationAfterDisabledChangeTrackingForTable(localDb, localSyncProvider, remoteDb, remoteSyncProvider); } + [TestMethod] + public async Task Test_Sqlite_MySql_DeleteParentRecordInRelatedTables() + { + var localDbFile = $"{Path.GetTempPath()}Test_Sqlite_MySql_DeleteParentRecordInRelatedTables_local.sqlite"; + + if (File.Exists(localDbFile)) File.Delete(localDbFile); + + using var localDb = new SqliteBlogDbContext($"Data Source={localDbFile}"); + using var remoteDb = new MySqlBlogDbContext(GetMySqlConnectionString("cs_delparent_sqlite_mysql")); + await localDb.Database.EnsureDeletedAsync(); + await remoteDb.Database.EnsureDeletedAsync(); + + await localDb.Database.MigrateAsync(); + await remoteDb.Database.MigrateAsync(); + + localDb.Database.ExecuteSqlRaw("PRAGMA foreign_keys = ON;"); + + var remoteConfigurationBuilder = + new MySqlSyncConfigurationBuilder(remoteDb.ConnectionString) + .Table("Users") + .Table("Posts") + .Table("Comments"); + + var remoteSyncProvider = new MySqlSyncProvider(remoteConfigurationBuilder.Build(), ProviderMode.Remote, logger: new ConsoleLogger("REM")); + await remoteSyncProvider.ApplyProvisionAsync(); + + var localConfigurationBuilder = + new SqliteSyncConfigurationBuilder(localDb.ConnectionString) + .Table("Users") + .Table("Posts") + .Table("Comments"); + + var localSyncProvider = new SqliteSyncProvider(localConfigurationBuilder.Build(), ProviderMode.Local, logger: new ConsoleLogger("LOC")); + await localSyncProvider.ApplyProvisionAsync(); + + await TestSyncAgentDeleteParentRecordInRelatedTables(localDb, localSyncProvider, remoteDb, remoteSyncProvider); + } + + [TestMethod] + public async Task Test_Sqlite_MySql_DeleteLocalParentRecordInRelatedTablesUpdatedOnServer() + { + var localDbFile = $"{Path.GetTempPath()}Test_Sqlite_MySql_DeleteLocalParentRecordInRelatedTablesUpdatedOnServer.sqlite"; + + if (File.Exists(localDbFile)) File.Delete(localDbFile); + + using var localDb = new SqliteBlogDbContext($"Data Source={localDbFile}"); + using var remoteDb = new MySqlBlogDbContext(GetMySqlConnectionString("cs_dellocalupd_mysql")); + await localDb.Database.EnsureDeletedAsync(); + await remoteDb.Database.EnsureDeletedAsync(); + + await localDb.Database.MigrateAsync(); + await remoteDb.Database.MigrateAsync(); + + localDb.Database.ExecuteSqlRaw("PRAGMA foreign_keys = ON;"); + + var remoteConfigurationBuilder = + new MySqlSyncConfigurationBuilder(remoteDb.ConnectionString) + .Table("Users") + .Table("Posts") + .Table("Comments"); + + var remoteSyncProvider = new MySqlSyncProvider(remoteConfigurationBuilder.Build(), ProviderMode.Remote, logger: new ConsoleLogger("REM")); + await remoteSyncProvider.ApplyProvisionAsync(); + + var localConfigurationBuilder = + new SqliteSyncConfigurationBuilder(localDb.ConnectionString) + .Table("Users") + .Table("Posts") + .Table("Comments"); + + var localSyncProvider = new SqliteSyncProvider(localConfigurationBuilder.Build(), ProviderMode.Local, logger: new ConsoleLogger("LOC")); + await localSyncProvider.ApplyProvisionAsync(); + + await TestDeleteLocalParentRecordInRelatedTablesUpdatedOnServer(localDb, localSyncProvider, remoteDb, remoteSyncProvider); + } + + [TestMethod] + public async Task Test_Sqlite_MySql_TestDeleteLocalParentRecordInRelatedTablesUpdatedOnServerSkipApplyChanges() + { + var localDbFile = $"{Path.GetTempPath()}Test_Sqlite_MySql_DelLocalParentUpdOnSrvSkip.sqlite"; + + if (File.Exists(localDbFile)) File.Delete(localDbFile); + + using var localDb = new SqliteBlogDbContext($"Data Source={localDbFile}"); + using var remoteDb = new MySqlBlogDbContext(GetMySqlConnectionString("cs_dellocalupdskip_mysql")); + await localDb.Database.EnsureDeletedAsync(); + await remoteDb.Database.EnsureDeletedAsync(); + + await localDb.Database.MigrateAsync(); + await remoteDb.Database.MigrateAsync(); + + localDb.Database.ExecuteSqlRaw("PRAGMA foreign_keys = ON;"); + + var remoteConfigurationBuilder = + new MySqlSyncConfigurationBuilder(remoteDb.ConnectionString) + .Table("Users") + .Table("Posts") + .Table("Comments"); + + var remoteSyncProvider = new MySqlSyncProvider(remoteConfigurationBuilder.Build(), ProviderMode.Remote, logger: new ConsoleLogger("REM")); + await remoteSyncProvider.ApplyProvisionAsync(); + + var localConfigurationBuilder = + new SqliteSyncConfigurationBuilder(localDb.ConnectionString) + .Table("Users") + .Table("Posts") + .Table("Comments"); + + var localSyncProvider = new SqliteSyncProvider(localConfigurationBuilder.Build(), ProviderMode.Local, logger: new ConsoleLogger("LOC")); + await localSyncProvider.ApplyProvisionAsync(); + + await TestDeleteLocalParentRecordInRelatedTablesUpdatedOnServerSkipApplyChanges(localDb, localSyncProvider, remoteDb, remoteSyncProvider); + } + + [TestMethod] + public async Task Test1_MySql_MySql() + { + using var localDb = new MySqlBlogDbContext(GetMySqlConnectionString("coresync_mysql_test1_local")); + using var remoteDb = new MySqlBlogDbContext(GetMySqlConnectionString("coresync_mysql_test1_remote")); + await localDb.Database.EnsureDeletedAsync(); + await remoteDb.Database.EnsureDeletedAsync(); + + await localDb.Database.MigrateAsync(); + await remoteDb.Database.MigrateAsync(); + + var remoteConfigurationBuilder = + new MySqlSyncConfigurationBuilder(remoteDb.ConnectionString) + .Table("Users") + .Table("Posts") + .Table("Comments"); + + ISyncProvider remoteSyncProvider = new MySqlSyncProvider(remoteConfigurationBuilder.Build(), logger: new ConsoleLogger("REM")); + await remoteSyncProvider.ApplyProvisionAsync(); + + var localConfigurationBuilder = + new MySqlSyncConfigurationBuilder(localDb.ConnectionString) + .Table("Users") + .Table("Posts") + .Table("Comments"); + + ISyncProvider localSyncProvider = new MySqlSyncProvider(localConfigurationBuilder.Build(), logger: new ConsoleLogger("LOC")); + await localSyncProvider.ApplyProvisionAsync(); + + await Test1(localDb, localSyncProvider, remoteDb, remoteSyncProvider); + } + + [TestMethod] + public async Task Test2_MySql_MySql() + { + using var localDb = new MySqlBlogDbContext(GetMySqlConnectionString("coresync_mysql_test2_local")); + using var remoteDb = new MySqlBlogDbContext(GetMySqlConnectionString("coresync_mysql_test2_remote")); + await localDb.Database.EnsureDeletedAsync(); + await remoteDb.Database.EnsureDeletedAsync(); + + await localDb.Database.MigrateAsync(); + await remoteDb.Database.MigrateAsync(); + + var remoteConfigurationBuilder = + new MySqlSyncConfigurationBuilder(remoteDb.ConnectionString) + .Table("Users") + .Table("Posts") + .Table("Comments"); + + ISyncProvider remoteSyncProvider = new MySqlSyncProvider(remoteConfigurationBuilder.Build(), logger: new ConsoleLogger("REM")); + await remoteSyncProvider.ApplyProvisionAsync(); + + var localConfigurationBuilder = + new MySqlSyncConfigurationBuilder(localDb.ConnectionString) + .Table("Users") + .Table("Posts") + .Table("Comments"); + + ISyncProvider localSyncProvider = new MySqlSyncProvider(localConfigurationBuilder.Build(), logger: new ConsoleLogger("LOC")); + await localSyncProvider.ApplyProvisionAsync(); + + await Test2(localDb, localSyncProvider, remoteDb, remoteSyncProvider); + } + + [TestMethod] + public async Task Test1_MySql_SqlServer() + { + using var localDb = new MySqlBlogDbContext(GetMySqlConnectionString("coresync_mysql_test1_local_sqlserver")); + using var remoteDb = new SqlServerBlogDbContext(ConnectionString + ";Initial Catalog=Test1_Remote_MySql"); + await localDb.Database.EnsureDeletedAsync(); + await remoteDb.Database.EnsureDeletedAsync(); + + await localDb.Database.MigrateAsync(); + await remoteDb.Database.MigrateAsync(); + + var remoteConfigurationBuilder = + new SqlSyncConfigurationBuilder(remoteDb.ConnectionString) + .Table("Users") + .Table("Posts") + .Table("Comments"); + + ISyncProvider remoteSyncProvider = new SqlSyncProvider(remoteConfigurationBuilder.Build(), logger: new ConsoleLogger("REM")); + await remoteSyncProvider.ApplyProvisionAsync(); + + var localConfigurationBuilder = + new MySqlSyncConfigurationBuilder(localDb.ConnectionString) + .Table("Users") + .Table("Posts") + .Table("Comments"); + + ISyncProvider localSyncProvider = new MySqlSyncProvider(localConfigurationBuilder.Build(), logger: new ConsoleLogger("LOC")); + await localSyncProvider.ApplyProvisionAsync(); + + await Test1(localDb, localSyncProvider, remoteDb, remoteSyncProvider); + } + + [TestMethod] + public async Task Test1_Sqlite_MySql() + { + var localDbFile = $"{Path.GetTempPath()}Test1_Sqlite_MySql_local.sqlite"; + + if (File.Exists(localDbFile)) File.Delete(localDbFile); + + using var localDb = new SqliteBlogDbContext($"Data Source={localDbFile}"); + using var remoteDb = new MySqlBlogDbContext(GetMySqlConnectionString("coresync_sqlite_mysql_test1_remote")); + await localDb.Database.EnsureDeletedAsync(); + await remoteDb.Database.EnsureDeletedAsync(); + + await localDb.Database.MigrateAsync(); + await remoteDb.Database.MigrateAsync(); + + var remoteConfigurationBuilder = + new MySqlSyncConfigurationBuilder(remoteDb.ConnectionString) + .Table("Users") + .Table("Posts") + .Table("Comments"); + + ISyncProvider remoteSyncProvider = new MySqlSyncProvider(remoteConfigurationBuilder.Build(), logger: new ConsoleLogger("REM")); + await remoteSyncProvider.ApplyProvisionAsync(); + + var localConfigurationBuilder = + new SqliteSyncConfigurationBuilder(localDb.ConnectionString) + .Table("Users") + .Table("Posts") + .Table("Comments"); + + ISyncProvider localSyncProvider = new SqliteSyncProvider(localConfigurationBuilder.Build(), logger: new ConsoleLogger("LOC")); + await localSyncProvider.ApplyProvisionAsync(); + + await Test1(localDb, localSyncProvider, remoteDb, remoteSyncProvider); + } + + [TestMethod] + public async Task Test2_Sqlite_MySql() + { + var localDbFile = $"{Path.GetTempPath()}Test2_Sqlite_MySql_local.sqlite"; + + if (File.Exists(localDbFile)) File.Delete(localDbFile); + + using var localDb = new SqliteBlogDbContext($"Data Source={localDbFile}"); + using var remoteDb = new MySqlBlogDbContext(GetMySqlConnectionString("coresync_sqlite_mysql_test2_remote")); + await localDb.Database.EnsureDeletedAsync(); + await remoteDb.Database.EnsureDeletedAsync(); + + await localDb.Database.MigrateAsync(); + await remoteDb.Database.MigrateAsync(); + + var remoteConfigurationBuilder = + new MySqlSyncConfigurationBuilder(remoteDb.ConnectionString) + .Table("Users") + .Table("Posts") + .Table("Comments"); + + ISyncProvider remoteSyncProvider = new MySqlSyncProvider(remoteConfigurationBuilder.Build(), logger: new ConsoleLogger("REM")); + await remoteSyncProvider.ApplyProvisionAsync(); + + var localConfigurationBuilder = + new SqliteSyncConfigurationBuilder(localDb.ConnectionString) + .Table("Users") + .Table("Posts") + .Table("Comments"); + + ISyncProvider localSyncProvider = new SqliteSyncProvider(localConfigurationBuilder.Build(), logger: new ConsoleLogger("LOC")); + await localSyncProvider.ApplyProvisionAsync(); + + await Test2(localDb, localSyncProvider, remoteDb, remoteSyncProvider); + } + + [TestMethod] + public async Task Test_MySql_MySql_TestSynchronizationWithFilter() + { + using var localDb = new MySqlBlogDbContext(GetMySqlConnectionString("cs_mysql_filter_l")); + using var remoteDb = new MySqlBlogDbContext(GetMySqlConnectionString("cs_mysql_filter_r")); + + await localDb.Database.EnsureDeletedAsync(); + await remoteDb.Database.EnsureDeletedAsync(); + + await localDb.Database.MigrateAsync(); + await remoteDb.Database.MigrateAsync(); + + var remoteConfigurationBuilder = + new MySqlSyncConfigurationBuilder(remoteDb.ConnectionString) + .Table("Users", selectIncrementalQuery: "SELECT * FROM `Users` WHERE `Email` = @userId") + .Table("Posts", selectIncrementalQuery: "SELECT * FROM `Posts` WHERE `AuthorEmail` = @userId") + .Table("Comments", selectIncrementalQuery: "SELECT c.* FROM `Comments` c INNER JOIN `Posts` p ON c.`PostId` = p.`Id` WHERE p.`AuthorEmail` = @userId"); + + var remoteSyncProvider = new MySqlSyncProvider(remoteConfigurationBuilder.Build(), ProviderMode.Remote, logger: new ConsoleLogger("REM")); + await remoteSyncProvider.ApplyProvisionAsync(); + + var localConfigurationBuilder = + new MySqlSyncConfigurationBuilder(localDb.ConnectionString) + .Table("Users") + .Table("Posts") + .Table("Comments"); + + var localSyncProvider = new MySqlSyncProvider(localConfigurationBuilder.Build(), ProviderMode.Local, logger: new ConsoleLogger("LOC")); + await localSyncProvider.ApplyProvisionAsync(); + + await TestSynchronizationWithFilter(localDb, localSyncProvider, remoteDb, remoteSyncProvider); + } + + [TestMethod] + public async Task Test_MySql_Sqlite_TestSynchronizationWithFilter() + { + var localDbFile = $"{Path.GetTempPath()}Test_MySql_Sqlite_TestSynchronizationWithFilter_local.sqlite"; + + if (File.Exists(localDbFile)) File.Delete(localDbFile); + + using var localDb = new SqliteBlogDbContext($"Data Source={localDbFile}"); + using var remoteDb = new MySqlBlogDbContext(GetMySqlConnectionString("cs_mysql_sqlite_filter_r")); + + await localDb.Database.EnsureDeletedAsync(); + await remoteDb.Database.EnsureDeletedAsync(); + + await localDb.Database.MigrateAsync(); + await remoteDb.Database.MigrateAsync(); + + localDb.Database.ExecuteSqlRaw("PRAGMA foreign_keys = ON;"); + + var remoteConfigurationBuilder = + new MySqlSyncConfigurationBuilder(remoteDb.ConnectionString) + .Table("Users", selectIncrementalQuery: "SELECT * FROM `Users` WHERE `Email` = @userId") + .Table("Posts", selectIncrementalQuery: "SELECT * FROM `Posts` WHERE `AuthorEmail` = @userId") + .Table("Comments", selectIncrementalQuery: "SELECT c.* FROM `Comments` c INNER JOIN `Posts` p ON c.`PostId` = p.`Id` WHERE p.`AuthorEmail` = @userId"); + + var remoteSyncProvider = new MySqlSyncProvider(remoteConfigurationBuilder.Build(), ProviderMode.Remote, logger: new ConsoleLogger("REM")); + await remoteSyncProvider.ApplyProvisionAsync(); + + var localConfigurationBuilder = + new SqliteSyncConfigurationBuilder(localDb.ConnectionString) + .Table("Users") + .Table("Posts") + .Table("Comments"); + + var localSyncProvider = new SqliteSyncProvider(localConfigurationBuilder.Build(), ProviderMode.Local, logger: new ConsoleLogger("LOC")); + await localSyncProvider.ApplyProvisionAsync(); + + await TestSynchronizationWithFilter(localDb, localSyncProvider, remoteDb, remoteSyncProvider); + } + + [TestMethod] + public async Task Test_Sqlite_MySql_TestSynchronizationWithFilter() + { + var remoteDbFile = $"{Path.GetTempPath()}Test_Sqlite_MySql_TestSynchronizationWithFilter_remote.sqlite"; + + if (File.Exists(remoteDbFile)) File.Delete(remoteDbFile); + + using var localDb = new SqliteBlogDbContext($"Data Source={remoteDbFile}"); + using var remoteDb = new MySqlBlogDbContext(GetMySqlConnectionString("cs_sqlite_mysql_filter_r")); + + await localDb.Database.EnsureDeletedAsync(); + await remoteDb.Database.EnsureDeletedAsync(); + + await localDb.Database.MigrateAsync(); + await remoteDb.Database.MigrateAsync(); + + localDb.Database.ExecuteSqlRaw("PRAGMA foreign_keys = ON;"); + + var remoteConfigurationBuilder = + new MySqlSyncConfigurationBuilder(remoteDb.ConnectionString) + .Table("Users", selectIncrementalQuery: "SELECT * FROM `Users` WHERE `Email` = @userId") + .Table("Posts", selectIncrementalQuery: "SELECT * FROM `Posts` WHERE `AuthorEmail` = @userId") + .Table("Comments", selectIncrementalQuery: "SELECT c.* FROM `Comments` c INNER JOIN `Posts` p ON c.`PostId` = p.`Id` WHERE p.`AuthorEmail` = @userId"); + + var remoteSyncProvider = new MySqlSyncProvider(remoteConfigurationBuilder.Build(), ProviderMode.Remote, logger: new ConsoleLogger("REM")); + await remoteSyncProvider.ApplyProvisionAsync(); + + var localConfigurationBuilder = + new SqliteSyncConfigurationBuilder(localDb.ConnectionString) + .Table("Users") + .Table("Posts") + .Table("Comments"); + + var localSyncProvider = new SqliteSyncProvider(localConfigurationBuilder.Build(), ProviderMode.Local, logger: new ConsoleLogger("LOC")); + await localSyncProvider.ApplyProvisionAsync(); + + await TestSynchronizationWithFilter(localDb, localSyncProvider, remoteDb, remoteSyncProvider); + } + + [TestMethod] + public async Task Test_MySql_MySql_TestSynchronizationAfterDisabledChangeTrackingForTable() + { + using var localDb = new MySqlBlogDbContext(GetMySqlConnectionString("cs_mysql_disable_l")); + using var remoteDb = new MySqlBlogDbContext(GetMySqlConnectionString("cs_mysql_disable_r")); + + await localDb.Database.EnsureDeletedAsync(); + await remoteDb.Database.EnsureDeletedAsync(); + + await localDb.Database.MigrateAsync(); + await remoteDb.Database.MigrateAsync(); + + var remoteConfigurationBuilder = + new MySqlSyncConfigurationBuilder(remoteDb.ConnectionString) + .Table("Users") + .Table("Posts") + .Table("Comments"); + + var remoteSyncProvider = new MySqlSyncProvider(remoteConfigurationBuilder.Build(), ProviderMode.Remote, logger: new ConsoleLogger("REM")); + await remoteSyncProvider.ApplyProvisionAsync(); + + var localConfigurationBuilder = + new MySqlSyncConfigurationBuilder(localDb.ConnectionString) + .Table("Users") + .Table("Posts") + .Table("Comments"); + + var localSyncProvider = new MySqlSyncProvider(localConfigurationBuilder.Build(), ProviderMode.Local, logger: new ConsoleLogger("LOC")); + await localSyncProvider.ApplyProvisionAsync(); + + await TestSynchronizationAfterDisabledChangeTrackingForTable(localDb, localSyncProvider, remoteDb, remoteSyncProvider); + } + + [TestMethod] + public async Task Test_MySql_Sqlite_TestSynchronizationAfterDisabledChangeTrackingForTable() + { + var localDbFile = $"{Path.GetTempPath()}Test_MySql_Sqlite_TestSynchronizationAfterDisabledChangeTrackingForTable.sqlite"; + + if (File.Exists(localDbFile)) File.Delete(localDbFile); + + using var localDb = new SqliteBlogDbContext($"Data Source={localDbFile}"); + using var remoteDb = new MySqlBlogDbContext(GetMySqlConnectionString("cs_mysql_sqlite_disable_r")); + + await localDb.Database.EnsureDeletedAsync(); + await remoteDb.Database.EnsureDeletedAsync(); + + await localDb.Database.MigrateAsync(); + await remoteDb.Database.MigrateAsync(); + + var remoteConfigurationBuilder = + new MySqlSyncConfigurationBuilder(remoteDb.ConnectionString) + .Table("Users") + .Table("Posts") + .Table("Comments"); + + var remoteSyncProvider = new MySqlSyncProvider(remoteConfigurationBuilder.Build(), ProviderMode.Remote, logger: new ConsoleLogger("REM")); + await remoteSyncProvider.ApplyProvisionAsync(); + + var localConfigurationBuilder = + new SqliteSyncConfigurationBuilder(localDb.ConnectionString) + .Table("Users") + .Table("Posts") + .Table("Comments"); + + var localSyncProvider = new SqliteSyncProvider(localConfigurationBuilder.Build(), ProviderMode.Local, logger: new ConsoleLogger("LOC")); + await localSyncProvider.ApplyProvisionAsync(); + + await TestSynchronizationAfterDisabledChangeTrackingForTable(localDb, localSyncProvider, remoteDb, remoteSyncProvider); + } + + [TestMethod] + public async Task Test_Sqlite_MySql_TestSynchronizationAfterDisabledChangeTrackingForTable() + { + var remoteDbFile = $"{Path.GetTempPath()}Test_Sqlite_MySql_TestSynchronizationAfterDisabledChangeTrackingForTable.sqlite"; + + if (File.Exists(remoteDbFile)) File.Delete(remoteDbFile); + + using var localDb = new SqliteBlogDbContext($"Data Source={remoteDbFile}"); + using var remoteDb = new MySqlBlogDbContext(GetMySqlConnectionString("cs_sqlite_mysql_disable_r")); + + await localDb.Database.EnsureDeletedAsync(); + await remoteDb.Database.EnsureDeletedAsync(); + + await localDb.Database.MigrateAsync(); + await remoteDb.Database.MigrateAsync(); + + var remoteConfigurationBuilder = + new MySqlSyncConfigurationBuilder(remoteDb.ConnectionString) + .Table("Users") + .Table("Posts") + .Table("Comments"); + + var remoteSyncProvider = new MySqlSyncProvider(remoteConfigurationBuilder.Build(), ProviderMode.Remote, logger: new ConsoleLogger("REM")); + await remoteSyncProvider.ApplyProvisionAsync(); + + var localConfigurationBuilder = + new SqliteSyncConfigurationBuilder(localDb.ConnectionString) + .Table("Users") + .Table("Posts") + .Table("Comments"); + + var localSyncProvider = new SqliteSyncProvider(localConfigurationBuilder.Build(), ProviderMode.Local, logger: new ConsoleLogger("LOC")); + await localSyncProvider.ApplyProvisionAsync(); + + await TestSynchronizationAfterDisabledChangeTrackingForTable(localDb, localSyncProvider, remoteDb, remoteSyncProvider); + } + #region GetChangesWithTableFilter Tests [TestMethod] @@ -1995,6 +2496,70 @@ public async Task TestGetChangesWithTableFilter_Sqlite_PostgreSQL() await TestGetChangesWithTableFilter(localDb, localSyncProvider, remoteDb, remoteSyncProvider); } + [TestMethod] + public async Task TestGetChangesWithTableFilter_MySql_MySql() + { + using var localDb = new MySqlBlogDbContext(GetMySqlConnectionString("cs_mysql_tblfilter_l")); + using var remoteDb = new MySqlBlogDbContext(GetMySqlConnectionString("cs_mysql_tblfilter_r")); + await localDb.Database.EnsureDeletedAsync(); + await remoteDb.Database.EnsureDeletedAsync(); + + await localDb.Database.MigrateAsync(); + await remoteDb.Database.MigrateAsync(); + + var remoteConfigurationBuilder = + new MySqlSyncConfigurationBuilder(remoteDb.ConnectionString) + .Table("Users") + .Table("Posts") + .Table("Comments"); + + ISyncProvider remoteSyncProvider = new MySqlSyncProvider(remoteConfigurationBuilder.Build(), logger: new ConsoleLogger("REM")); + + var localConfigurationBuilder = + new MySqlSyncConfigurationBuilder(localDb.ConnectionString) + .Table("Users") + .Table("Posts") + .Table("Comments"); + + ISyncProvider localSyncProvider = new MySqlSyncProvider(localConfigurationBuilder.Build(), logger: new ConsoleLogger("LOC")); + + await TestGetChangesWithTableFilter(localDb, localSyncProvider, remoteDb, remoteSyncProvider); + } + + [TestMethod] + public async Task TestGetChangesWithTableFilter_Sqlite_MySql() + { + var localDbFile = $"{Path.GetTempPath()}TestGetChangesWithTableFilter_Sqlite_MySql_local.sqlite"; + + if (File.Exists(localDbFile)) File.Delete(localDbFile); + + using var localDb = new SqliteBlogDbContext($"Data Source={localDbFile}"); + using var remoteDb = new MySqlBlogDbContext(GetMySqlConnectionString("cs_sqlite_mysql_tblfilter_r")); + await localDb.Database.EnsureDeletedAsync(); + await remoteDb.Database.EnsureDeletedAsync(); + + await localDb.Database.MigrateAsync(); + await remoteDb.Database.MigrateAsync(); + + var remoteConfigurationBuilder = + new MySqlSyncConfigurationBuilder(remoteDb.ConnectionString) + .Table("Users") + .Table("Posts") + .Table("Comments"); + + ISyncProvider remoteSyncProvider = new MySqlSyncProvider(remoteConfigurationBuilder.Build(), logger: new ConsoleLogger("REM")); + + var localConfigurationBuilder = + new SqliteSyncConfigurationBuilder(localDb.ConnectionString) + .Table("Users") + .Table("Posts") + .Table("Comments"); + + ISyncProvider localSyncProvider = new SqliteSyncProvider(localConfigurationBuilder.Build(), logger: new ConsoleLogger("LOC")); + + await TestGetChangesWithTableFilter(localDb, localSyncProvider, remoteDb, remoteSyncProvider); + } + [TestMethod] public async Task TestGetChangesWithTableFilter_SqlServerCT_SqlServerCT() { @@ -2211,4 +2776,4 @@ public async Task TestSyncAgent_SqlServerCT() } -} \ No newline at end of file +} diff --git a/src/CoreSync.Tests/IntegrationTests.Http.cs b/src/CoreSync.Tests/IntegrationTests.Http.cs index eb00b3f..61524fe 100644 --- a/src/CoreSync.Tests/IntegrationTests.Http.cs +++ b/src/CoreSync.Tests/IntegrationTests.Http.cs @@ -1,6 +1,7 @@ using CoreSync.Sqlite; using CoreSync.SqlServer; using CoreSync.PostgreSQL; +using CoreSync.MySql; using CoreSync.Tests.Data; using Microsoft.EntityFrameworkCore; using Microsoft.VisualStudio.TestTools.UnitTesting; @@ -82,6 +83,70 @@ public async Task TestSyncAgent_Sqlite_Sqlite_MultipleRecordsSameTable_HttpBinar #endregion + #region TestSyncAgent Multiple Records over HTTP — Sqlite_MySql + + [TestMethod] + public async Task TestSyncAgent_Sqlite_MySql_MultipleRecordsSameTable_HttpJson() + { + var localDbFile = $"{Path.GetTempPath()}TestSyncAgent_Sqlite_MySql_MultipleRecordsSameTable_HttpJson.sqlite"; + if (File.Exists(localDbFile)) File.Delete(localDbFile); + + using var localDb = new SqliteBlogDbContext($"Data Source={localDbFile}"); + using var remoteDb = new MySqlBlogDbContext(GetMySqlConnectionString("coresync_http_multi_mysql_json")); + await localDb.Database.EnsureDeletedAsync(); + await remoteDb.Database.EnsureDeletedAsync(); + + await localDb.Database.MigrateAsync(); + await remoteDb.Database.MigrateAsync(); + + var remoteSyncProvider = new MySqlSyncProvider( + new MySqlSyncConfigurationBuilder(remoteDb.ConnectionString) + .Table("Users").Table("Posts").Table("Comments").Build(), + logger: new ConsoleLogger("REM")); + await remoteSyncProvider.ApplyProvisionAsync(); + + var localSyncProvider = new SqliteSyncProvider( + new SqliteSyncConfigurationBuilder(localDb.ConnectionString) + .Table("Users").Table("Posts").Table("Comments").Build(), + logger: new ConsoleLogger("LOC")); + await localSyncProvider.ApplyProvisionAsync(); + + using var server = SyncTestServer.Create(remoteSyncProvider, useBinaryFormat: false); + await TestSyncAgentMultipleRecordsOnSameTable(localDb, localSyncProvider, remoteDb, server.HttpSyncProvider); + } + + [TestMethod] + public async Task TestSyncAgent_Sqlite_MySql_MultipleRecordsSameTable_HttpBinary() + { + var localDbFile = $"{Path.GetTempPath()}TestSyncAgent_Sqlite_MySql_MultipleRecordsSameTable_HttpBinary.sqlite"; + if (File.Exists(localDbFile)) File.Delete(localDbFile); + + using var localDb = new SqliteBlogDbContext($"Data Source={localDbFile}"); + using var remoteDb = new MySqlBlogDbContext(GetMySqlConnectionString("coresync_http_multi_mysql_binary")); + await localDb.Database.EnsureDeletedAsync(); + await remoteDb.Database.EnsureDeletedAsync(); + + await localDb.Database.MigrateAsync(); + await remoteDb.Database.MigrateAsync(); + + var remoteSyncProvider = new MySqlSyncProvider( + new MySqlSyncConfigurationBuilder(remoteDb.ConnectionString) + .Table("Users").Table("Posts").Table("Comments").Build(), + logger: new ConsoleLogger("REM")); + await remoteSyncProvider.ApplyProvisionAsync(); + + var localSyncProvider = new SqliteSyncProvider( + new SqliteSyncConfigurationBuilder(localDb.ConnectionString) + .Table("Users").Table("Posts").Table("Comments").Build(), + logger: new ConsoleLogger("LOC")); + await localSyncProvider.ApplyProvisionAsync(); + + using var server = SyncTestServer.Create(remoteSyncProvider, useBinaryFormat: true); + await TestSyncAgentMultipleRecordsOnSameTable(localDb, localSyncProvider, remoteDb, server.HttpSyncProvider); + } + + #endregion + #region TestSyncAgent Multiple Records over HTTP — Sqlite_SqlServer [TestMethod] @@ -146,6 +211,70 @@ public async Task TestSyncAgent_Sqlite_SqlServer_MultipleRecordsSameTable_HttpBi #endregion + #region TestSyncAgent UpdatedRemoteDeletedLocal over HTTP — Sqlite_MySql + + [TestMethod] + public async Task TestSyncAgent_Sqlite_MySql_UpdatedRemoteDeletedLocal_HttpJson() + { + var localDbFile = $"{Path.GetTempPath()}TestSyncAgent_Sqlite_MySql_UpdatedRemoteDeletedLocal_HttpJson.sqlite"; + if (File.Exists(localDbFile)) File.Delete(localDbFile); + + using var localDb = new SqliteBlogDbContext($"Data Source={localDbFile}"); + using var remoteDb = new MySqlBlogDbContext(GetMySqlConnectionString("coresync_http_urdl_mysql_json")); + await localDb.Database.EnsureDeletedAsync(); + await remoteDb.Database.EnsureDeletedAsync(); + + await localDb.Database.MigrateAsync(); + await remoteDb.Database.MigrateAsync(); + + var remoteSyncProvider = new MySqlSyncProvider( + new MySqlSyncConfigurationBuilder(remoteDb.ConnectionString) + .Table("Users").Table("Posts").Table("Comments").Build(), + ProviderMode.Remote, logger: new ConsoleLogger("REM")); + await remoteSyncProvider.ApplyProvisionAsync(); + + var localSyncProvider = new SqliteSyncProvider( + new SqliteSyncConfigurationBuilder(localDb.ConnectionString) + .Table("Users").Table("Posts").Table("Comments").Build(), + logger: new ConsoleLogger("LOC")); + await localSyncProvider.ApplyProvisionAsync(); + + using var server = SyncTestServer.Create(remoteSyncProvider, useBinaryFormat: false); + await TestSyncAgentWithUpdatedRemoteDeletedLocal(localDb, localSyncProvider, remoteDb, server.HttpSyncProvider); + } + + [TestMethod] + public async Task TestSyncAgent_Sqlite_MySql_UpdatedRemoteDeletedLocal_HttpBinary() + { + var localDbFile = $"{Path.GetTempPath()}TestSyncAgent_Sqlite_MySql_UpdatedRemoteDeletedLocal_HttpBinary.sqlite"; + if (File.Exists(localDbFile)) File.Delete(localDbFile); + + using var localDb = new SqliteBlogDbContext($"Data Source={localDbFile}"); + using var remoteDb = new MySqlBlogDbContext(GetMySqlConnectionString("coresync_http_urdl_mysql_binary")); + await localDb.Database.EnsureDeletedAsync(); + await remoteDb.Database.EnsureDeletedAsync(); + + await localDb.Database.MigrateAsync(); + await remoteDb.Database.MigrateAsync(); + + var remoteSyncProvider = new MySqlSyncProvider( + new MySqlSyncConfigurationBuilder(remoteDb.ConnectionString) + .Table("Users").Table("Posts").Table("Comments").Build(), + ProviderMode.Remote, logger: new ConsoleLogger("REM")); + await remoteSyncProvider.ApplyProvisionAsync(); + + var localSyncProvider = new SqliteSyncProvider( + new SqliteSyncConfigurationBuilder(localDb.ConnectionString) + .Table("Users").Table("Posts").Table("Comments").Build(), + logger: new ConsoleLogger("LOC")); + await localSyncProvider.ApplyProvisionAsync(); + + using var server = SyncTestServer.Create(remoteSyncProvider, useBinaryFormat: true); + await TestSyncAgentWithUpdatedRemoteDeletedLocal(localDb, localSyncProvider, remoteDb, server.HttpSyncProvider); + } + + #endregion + #region TestSyncAgent Multiple Records over HTTP — Sqlite_PostgreSQL [TestMethod] @@ -210,6 +339,70 @@ public async Task TestSyncAgent_Sqlite_PostgreSQL_MultipleRecordsSameTable_HttpB #endregion + #region DeleteWithForeignKeys over HTTP — Sqlite_MySql + + [TestMethod] + public async Task Test_Sqlite_MySql_DeleteWithForeignKeys_HttpJson() + { + var localDbFile = $"{Path.GetTempPath()}Test_Sqlite_MySql_DeleteWithForeignKeys_HttpJson.sqlite"; + if (File.Exists(localDbFile)) File.Delete(localDbFile); + + using var localDb = new SqliteBlogDbContext($"Data Source={localDbFile}"); + using var remoteDb = new MySqlBlogDbContext(GetMySqlConnectionString("coresync_http_delfk_mysql_json")); + await localDb.Database.EnsureDeletedAsync(); + await remoteDb.Database.EnsureDeletedAsync(); + + await localDb.Database.MigrateAsync(); + await remoteDb.Database.MigrateAsync(); + + var remoteSyncProvider = new MySqlSyncProvider( + new MySqlSyncConfigurationBuilder(remoteDb.ConnectionString) + .Table("Users").Table("Posts").Table("Comments").Build(), + ProviderMode.Remote, logger: new ConsoleLogger("REM")); + await remoteSyncProvider.ApplyProvisionAsync(); + + var localSyncProvider = new SqliteSyncProvider( + new SqliteSyncConfigurationBuilder(localDb.ConnectionString) + .Table("Users").Table("Posts").Table("Comments").Build(), + ProviderMode.Local, logger: new ConsoleLogger("LOC")); + await localSyncProvider.ApplyProvisionAsync(); + + using var server = SyncTestServer.Create(remoteSyncProvider, useBinaryFormat: false); + await TestSyncAgentDeleteWithForeignKeys(localDb, localSyncProvider, remoteDb, server.HttpSyncProvider); + } + + [TestMethod] + public async Task Test_Sqlite_MySql_DeleteWithForeignKeys_HttpBinary() + { + var localDbFile = $"{Path.GetTempPath()}Test_Sqlite_MySql_DeleteWithForeignKeys_HttpBinary.sqlite"; + if (File.Exists(localDbFile)) File.Delete(localDbFile); + + using var localDb = new SqliteBlogDbContext($"Data Source={localDbFile}"); + using var remoteDb = new MySqlBlogDbContext(GetMySqlConnectionString("coresync_http_delfk_mysql_binary")); + await localDb.Database.EnsureDeletedAsync(); + await remoteDb.Database.EnsureDeletedAsync(); + + await localDb.Database.MigrateAsync(); + await remoteDb.Database.MigrateAsync(); + + var remoteSyncProvider = new MySqlSyncProvider( + new MySqlSyncConfigurationBuilder(remoteDb.ConnectionString) + .Table("Users").Table("Posts").Table("Comments").Build(), + ProviderMode.Remote, logger: new ConsoleLogger("REM")); + await remoteSyncProvider.ApplyProvisionAsync(); + + var localSyncProvider = new SqliteSyncProvider( + new SqliteSyncConfigurationBuilder(localDb.ConnectionString) + .Table("Users").Table("Posts").Table("Comments").Build(), + ProviderMode.Local, logger: new ConsoleLogger("LOC")); + await localSyncProvider.ApplyProvisionAsync(); + + using var server = SyncTestServer.Create(remoteSyncProvider, useBinaryFormat: true); + await TestSyncAgentDeleteWithForeignKeys(localDb, localSyncProvider, remoteDb, server.HttpSyncProvider); + } + + #endregion + #region TestSyncAgent UpdatedRemoteDeletedLocal over HTTP — Sqlite_Sqlite [TestMethod] @@ -815,4 +1008,72 @@ public async Task Test_Sqlite_PostgreSQL_DeleteParentRecordInRelatedTables_HttpB } #endregion + + #region DeleteParentRecordInRelatedTables over HTTP — Sqlite_MySql + + [TestMethod] + public async Task Test_Sqlite_MySql_DeleteParentRecordInRelatedTables_HttpJson() + { + var localDbFile = $"{Path.GetTempPath()}Test_Sqlite_MySql_DeleteParentRecordInRelatedTables_HttpJson.sqlite"; + if (File.Exists(localDbFile)) File.Delete(localDbFile); + + using var localDb = new SqliteBlogDbContext($"Data Source={localDbFile}"); + using var remoteDb = new MySqlBlogDbContext(GetMySqlConnectionString("coresync_http_delparent_mysql_json")); + await localDb.Database.EnsureDeletedAsync(); + await remoteDb.Database.EnsureDeletedAsync(); + + await localDb.Database.MigrateAsync(); + await remoteDb.Database.MigrateAsync(); + + localDb.Database.ExecuteSqlRaw("PRAGMA foreign_keys = ON;"); + + var remoteSyncProvider = new MySqlSyncProvider( + new MySqlSyncConfigurationBuilder(remoteDb.ConnectionString) + .Table("Users").Table("Posts").Table("Comments").Build(), + ProviderMode.Remote, logger: new ConsoleLogger("REM")); + await remoteSyncProvider.ApplyProvisionAsync(); + + var localSyncProvider = new SqliteSyncProvider( + new SqliteSyncConfigurationBuilder(localDb.ConnectionString) + .Table("Users").Table("Posts").Table("Comments").Build(), + ProviderMode.Local, logger: new ConsoleLogger("LOC")); + await localSyncProvider.ApplyProvisionAsync(); + + using var server = SyncTestServer.Create(remoteSyncProvider, useBinaryFormat: false); + await TestSyncAgentDeleteParentRecordInRelatedTables(localDb, localSyncProvider, remoteDb, server.HttpSyncProvider); + } + + [TestMethod] + public async Task Test_Sqlite_MySql_DeleteParentRecordInRelatedTables_HttpBinary() + { + var localDbFile = $"{Path.GetTempPath()}Test_Sqlite_MySql_DeleteParentRecordInRelatedTables_HttpBinary.sqlite"; + if (File.Exists(localDbFile)) File.Delete(localDbFile); + + using var localDb = new SqliteBlogDbContext($"Data Source={localDbFile}"); + using var remoteDb = new MySqlBlogDbContext(GetMySqlConnectionString("coresync_http_delparent_mysql_binary")); + await localDb.Database.EnsureDeletedAsync(); + await remoteDb.Database.EnsureDeletedAsync(); + + await localDb.Database.MigrateAsync(); + await remoteDb.Database.MigrateAsync(); + + localDb.Database.ExecuteSqlRaw("PRAGMA foreign_keys = ON;"); + + var remoteSyncProvider = new MySqlSyncProvider( + new MySqlSyncConfigurationBuilder(remoteDb.ConnectionString) + .Table("Users").Table("Posts").Table("Comments").Build(), + ProviderMode.Remote, logger: new ConsoleLogger("REM")); + await remoteSyncProvider.ApplyProvisionAsync(); + + var localSyncProvider = new SqliteSyncProvider( + new SqliteSyncConfigurationBuilder(localDb.ConnectionString) + .Table("Users").Table("Posts").Table("Comments").Build(), + ProviderMode.Local, logger: new ConsoleLogger("LOC")); + await localSyncProvider.ApplyProvisionAsync(); + + using var server = SyncTestServer.Create(remoteSyncProvider, useBinaryFormat: true); + await TestSyncAgentDeleteParentRecordInRelatedTables(localDb, localSyncProvider, remoteDb, server.HttpSyncProvider); + } + + #endregion } diff --git a/src/CoreSync.Tests/IntegrationTests.cs b/src/CoreSync.Tests/IntegrationTests.cs index 6ea828b..1ed7b56 100644 --- a/src/CoreSync.Tests/IntegrationTests.cs +++ b/src/CoreSync.Tests/IntegrationTests.cs @@ -1,6 +1,7 @@ using CoreSync.Sqlite; using CoreSync.SqlServer; using CoreSync.PostgreSQL; +using CoreSync.MySql; using CoreSync.Tests.Data; using Microsoft.EntityFrameworkCore; using Microsoft.VisualStudio.TestTools.UnitTesting; @@ -21,6 +22,18 @@ public partial class IntegrationTests public static string PostgreSQLConnectionString => Environment.GetEnvironmentVariable("CORE-SYNC_POSTGRESQL_CONNECTION_STRING") ?? "Host=localhost;Port=5432;Database=coresync_test;Username=coresync;Password=test123"; + public static string MySqlConnectionString => Environment.GetEnvironmentVariable("CORE-SYNC_MYSQL_CONNECTION_STRING") ?? + "Server=localhost;Port=3306;Database=coresync_test;User=root;Password=test123;GuidFormat=Char36"; + + public static string GetMySqlConnectionString(string dbName) + { + var builder = new MySqlConnector.MySqlConnectionStringBuilder(MySqlConnectionString) + { + Database = dbName + }; + return builder.ToString(); + } + private static async Task Test1( BlogDbContext localDb, ISyncProvider localSyncProvider, diff --git a/src/CoreSync.Tests/Migrations/MySqlMigrations/20260707093016_InitialCreate.Designer.cs b/src/CoreSync.Tests/Migrations/MySqlMigrations/20260707093016_InitialCreate.Designer.cs new file mode 100644 index 0000000..bb7a94b --- /dev/null +++ b/src/CoreSync.Tests/Migrations/MySqlMigrations/20260707093016_InitialCreate.Designer.cs @@ -0,0 +1,156 @@ +// +using System; +using CoreSync.Tests.Data; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Infrastructure; +using Microsoft.EntityFrameworkCore.Metadata; +using Microsoft.EntityFrameworkCore.Migrations; +using Microsoft.EntityFrameworkCore.Storage.ValueConversion; + +#nullable disable + +namespace CoreSync.Tests.Migrations.MySqlMigrations +{ + [DbContext(typeof(MySqlBlogDbContext))] + [Migration("20260707093016_InitialCreate")] + partial class InitialCreate + { + /// + protected override void BuildTargetModel(ModelBuilder modelBuilder) + { +#pragma warning disable 612, 618 + modelBuilder + .HasAnnotation("ProductVersion", "9.0.17") + .HasAnnotation("Relational:MaxIdentifierLength", 64); + + MySqlModelBuilderExtensions.AutoIncrementColumns(modelBuilder); + + modelBuilder.Entity("CoreSync.Tests.Data.Comment", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("char(36)"); + + b.Property("AuthorEmail") + .HasMaxLength(255) + .HasColumnType("varchar(255)"); + + b.Property("Content") + .HasColumnType("longtext"); + + b.Property("Created") + .HasColumnType("datetime(6)"); + + b.Property("PostId") + .HasColumnType("char(36)"); + + b.Property("ReplyToId") + .HasColumnType("char(36)"); + + b.HasKey("Id"); + + b.HasIndex("AuthorEmail"); + + b.HasIndex("PostId"); + + b.HasIndex("ReplyToId"); + + b.ToTable("Comments", (string)null); + }); + + modelBuilder.Entity("CoreSync.Tests.Data.Post", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("char(36)"); + + b.Property("AuthorEmail") + .HasMaxLength(255) + .HasColumnType("varchar(255)"); + + b.Property("Claps") + .HasColumnType("int"); + + b.Property("Content") + .HasColumnType("longtext"); + + b.Property("Stars") + .HasColumnType("float"); + + b.Property("Title") + .HasColumnType("longtext"); + + b.Property("Updated") + .HasColumnType("datetime(6)"); + + b.HasKey("Id"); + + b.HasIndex("AuthorEmail"); + + b.ToTable("Posts", (string)null); + }); + + modelBuilder.Entity("CoreSync.Tests.Data.User", b => + { + b.Property("Email") + .HasMaxLength(255) + .HasColumnType("varchar(255)"); + + b.Property("Created") + .HasColumnType("datetime(6)") + .HasColumnName("Date Created(date/$time)"); + + b.Property("Name") + .HasColumnType("longtext"); + + b.HasKey("Email"); + + b.ToTable("Users", (string)null); + }); + + modelBuilder.Entity("CoreSync.Tests.Data.Comment", b => + { + b.HasOne("CoreSync.Tests.Data.User", "Author") + .WithMany("Comments") + .HasForeignKey("AuthorEmail"); + + b.HasOne("CoreSync.Tests.Data.Post", "Post") + .WithMany("Comments") + .HasForeignKey("PostId"); + + b.HasOne("CoreSync.Tests.Data.Comment", "ReplyTo") + .WithMany() + .HasForeignKey("ReplyToId"); + + b.Navigation("Author"); + + b.Navigation("Post"); + + b.Navigation("ReplyTo"); + }); + + modelBuilder.Entity("CoreSync.Tests.Data.Post", b => + { + b.HasOne("CoreSync.Tests.Data.User", "Author") + .WithMany("Posts") + .HasForeignKey("AuthorEmail") + .OnDelete(DeleteBehavior.Cascade); + + b.Navigation("Author"); + }); + + modelBuilder.Entity("CoreSync.Tests.Data.Post", b => + { + b.Navigation("Comments"); + }); + + modelBuilder.Entity("CoreSync.Tests.Data.User", b => + { + b.Navigation("Comments"); + + b.Navigation("Posts"); + }); +#pragma warning restore 612, 618 + } + } +} diff --git a/src/CoreSync.Tests/Migrations/MySqlMigrations/20260707093016_InitialCreate.cs b/src/CoreSync.Tests/Migrations/MySqlMigrations/20260707093016_InitialCreate.cs new file mode 100644 index 0000000..da53aed --- /dev/null +++ b/src/CoreSync.Tests/Migrations/MySqlMigrations/20260707093016_InitialCreate.cs @@ -0,0 +1,128 @@ +using System; +using Microsoft.EntityFrameworkCore.Migrations; + +#nullable disable + +namespace CoreSync.Tests.Migrations.MySqlMigrations +{ + /// + public partial class InitialCreate : Migration + { + /// + protected override void Up(MigrationBuilder migrationBuilder) + { + migrationBuilder.AlterDatabase() + .Annotation("MySql:CharSet", "utf8mb4"); + + migrationBuilder.CreateTable( + name: "Users", + columns: table => new + { + Email = table.Column(type: "varchar(255)", maxLength: 255, nullable: false) + .Annotation("MySql:CharSet", "utf8mb4"), + Name = table.Column(type: "longtext", nullable: true) + .Annotation("MySql:CharSet", "utf8mb4"), + DateCreateddatetime = table.Column(name: "Date Created(date/$time)", type: "datetime(6)", nullable: false) + }, + constraints: table => + { + table.PrimaryKey("PK_Users", x => x.Email); + }) + .Annotation("MySql:CharSet", "utf8mb4"); + + migrationBuilder.CreateTable( + name: "Posts", + columns: table => new + { + Id = table.Column(type: "char(36)", nullable: false, collation: "ascii_general_ci"), + Title = table.Column(type: "longtext", nullable: true) + .Annotation("MySql:CharSet", "utf8mb4"), + Content = table.Column(type: "longtext", nullable: true) + .Annotation("MySql:CharSet", "utf8mb4"), + AuthorEmail = table.Column(type: "varchar(255)", maxLength: 255, nullable: true) + .Annotation("MySql:CharSet", "utf8mb4"), + Updated = table.Column(type: "datetime(6)", nullable: false), + Claps = table.Column(type: "int", nullable: false), + Stars = table.Column(type: "float", nullable: false) + }, + constraints: table => + { + table.PrimaryKey("PK_Posts", x => x.Id); + table.ForeignKey( + name: "FK_Posts_Users_AuthorEmail", + column: x => x.AuthorEmail, + principalTable: "Users", + principalColumn: "Email", + onDelete: ReferentialAction.Cascade); + }) + .Annotation("MySql:CharSet", "utf8mb4"); + + migrationBuilder.CreateTable( + name: "Comments", + columns: table => new + { + Id = table.Column(type: "char(36)", nullable: false, collation: "ascii_general_ci"), + PostId = table.Column(type: "char(36)", nullable: true, collation: "ascii_general_ci"), + AuthorEmail = table.Column(type: "varchar(255)", maxLength: 255, nullable: true) + .Annotation("MySql:CharSet", "utf8mb4"), + Content = table.Column(type: "longtext", nullable: true) + .Annotation("MySql:CharSet", "utf8mb4"), + Created = table.Column(type: "datetime(6)", nullable: false), + ReplyToId = table.Column(type: "char(36)", nullable: true, collation: "ascii_general_ci") + }, + constraints: table => + { + table.PrimaryKey("PK_Comments", x => x.Id); + table.ForeignKey( + name: "FK_Comments_Comments_ReplyToId", + column: x => x.ReplyToId, + principalTable: "Comments", + principalColumn: "Id"); + table.ForeignKey( + name: "FK_Comments_Posts_PostId", + column: x => x.PostId, + principalTable: "Posts", + principalColumn: "Id"); + table.ForeignKey( + name: "FK_Comments_Users_AuthorEmail", + column: x => x.AuthorEmail, + principalTable: "Users", + principalColumn: "Email"); + }) + .Annotation("MySql:CharSet", "utf8mb4"); + + migrationBuilder.CreateIndex( + name: "IX_Comments_AuthorEmail", + table: "Comments", + column: "AuthorEmail"); + + migrationBuilder.CreateIndex( + name: "IX_Comments_PostId", + table: "Comments", + column: "PostId"); + + migrationBuilder.CreateIndex( + name: "IX_Comments_ReplyToId", + table: "Comments", + column: "ReplyToId"); + + migrationBuilder.CreateIndex( + name: "IX_Posts_AuthorEmail", + table: "Posts", + column: "AuthorEmail"); + } + + /// + protected override void Down(MigrationBuilder migrationBuilder) + { + migrationBuilder.DropTable( + name: "Comments"); + + migrationBuilder.DropTable( + name: "Posts"); + + migrationBuilder.DropTable( + name: "Users"); + } + } +} diff --git a/src/CoreSync.Tests/Migrations/MySqlMigrations/MySqlBlogDbContextModelSnapshot.cs b/src/CoreSync.Tests/Migrations/MySqlMigrations/MySqlBlogDbContextModelSnapshot.cs new file mode 100644 index 0000000..f6e6840 --- /dev/null +++ b/src/CoreSync.Tests/Migrations/MySqlMigrations/MySqlBlogDbContextModelSnapshot.cs @@ -0,0 +1,153 @@ +// +using System; +using CoreSync.Tests.Data; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Infrastructure; +using Microsoft.EntityFrameworkCore.Metadata; +using Microsoft.EntityFrameworkCore.Storage.ValueConversion; + +#nullable disable + +namespace CoreSync.Tests.Migrations.MySqlMigrations +{ + [DbContext(typeof(MySqlBlogDbContext))] + partial class MySqlBlogDbContextModelSnapshot : ModelSnapshot + { + protected override void BuildModel(ModelBuilder modelBuilder) + { +#pragma warning disable 612, 618 + modelBuilder + .HasAnnotation("ProductVersion", "9.0.17") + .HasAnnotation("Relational:MaxIdentifierLength", 64); + + MySqlModelBuilderExtensions.AutoIncrementColumns(modelBuilder); + + modelBuilder.Entity("CoreSync.Tests.Data.Comment", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("char(36)"); + + b.Property("AuthorEmail") + .HasMaxLength(255) + .HasColumnType("varchar(255)"); + + b.Property("Content") + .HasColumnType("longtext"); + + b.Property("Created") + .HasColumnType("datetime(6)"); + + b.Property("PostId") + .HasColumnType("char(36)"); + + b.Property("ReplyToId") + .HasColumnType("char(36)"); + + b.HasKey("Id"); + + b.HasIndex("AuthorEmail"); + + b.HasIndex("PostId"); + + b.HasIndex("ReplyToId"); + + b.ToTable("Comments", (string)null); + }); + + modelBuilder.Entity("CoreSync.Tests.Data.Post", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("char(36)"); + + b.Property("AuthorEmail") + .HasMaxLength(255) + .HasColumnType("varchar(255)"); + + b.Property("Claps") + .HasColumnType("int"); + + b.Property("Content") + .HasColumnType("longtext"); + + b.Property("Stars") + .HasColumnType("float"); + + b.Property("Title") + .HasColumnType("longtext"); + + b.Property("Updated") + .HasColumnType("datetime(6)"); + + b.HasKey("Id"); + + b.HasIndex("AuthorEmail"); + + b.ToTable("Posts", (string)null); + }); + + modelBuilder.Entity("CoreSync.Tests.Data.User", b => + { + b.Property("Email") + .HasMaxLength(255) + .HasColumnType("varchar(255)"); + + b.Property("Created") + .HasColumnType("datetime(6)") + .HasColumnName("Date Created(date/$time)"); + + b.Property("Name") + .HasColumnType("longtext"); + + b.HasKey("Email"); + + b.ToTable("Users", (string)null); + }); + + modelBuilder.Entity("CoreSync.Tests.Data.Comment", b => + { + b.HasOne("CoreSync.Tests.Data.User", "Author") + .WithMany("Comments") + .HasForeignKey("AuthorEmail"); + + b.HasOne("CoreSync.Tests.Data.Post", "Post") + .WithMany("Comments") + .HasForeignKey("PostId"); + + b.HasOne("CoreSync.Tests.Data.Comment", "ReplyTo") + .WithMany() + .HasForeignKey("ReplyToId"); + + b.Navigation("Author"); + + b.Navigation("Post"); + + b.Navigation("ReplyTo"); + }); + + modelBuilder.Entity("CoreSync.Tests.Data.Post", b => + { + b.HasOne("CoreSync.Tests.Data.User", "Author") + .WithMany("Posts") + .HasForeignKey("AuthorEmail") + .OnDelete(DeleteBehavior.Cascade); + + b.Navigation("Author"); + }); + + modelBuilder.Entity("CoreSync.Tests.Data.Post", b => + { + b.Navigation("Comments"); + }); + + modelBuilder.Entity("CoreSync.Tests.Data.User", b => + { + b.Navigation("Comments"); + + b.Navigation("Posts"); + }); +#pragma warning restore 612, 618 + } + } +} diff --git a/src/CoreSync.Tests/SelfReferencingForeignKeyTests.Definitions.cs b/src/CoreSync.Tests/SelfReferencingForeignKeyTests.Definitions.cs index d945ef5..660b100 100644 --- a/src/CoreSync.Tests/SelfReferencingForeignKeyTests.Definitions.cs +++ b/src/CoreSync.Tests/SelfReferencingForeignKeyTests.Definitions.cs @@ -2,9 +2,11 @@ using CoreSync.SqlServer; using CoreSync.SqlServerCT; using CoreSync.PostgreSQL; +using CoreSync.MySql; using Microsoft.Data.Sqlite; using Microsoft.Data.SqlClient; using Microsoft.VisualStudio.TestTools.UnitTesting; +using MySqlConnector; using Npgsql; using Shouldly; using System; @@ -21,6 +23,9 @@ public partial class SelfReferencingForeignKeyTests private static string PostgreSQLConnectionString => Environment.GetEnvironmentVariable("CORE-SYNC_POSTGRESQL_CONNECTION_STRING") ?? "Host=localhost;Port=5432;Database=coresync_test;Username=coresync;Password=test123"; + private static string MySqlConnectionString => Environment.GetEnvironmentVariable("CORE-SYNC_MYSQL_CONNECTION_STRING") ?? + "Server=localhost;Port=3306;Database=coresync_test;User=root;Password=test123;GuidFormat=Char36"; + #region SQLite helpers private static async Task CreateSqliteCommentsTable(string connectionString, bool withForeignKey = true) @@ -277,6 +282,108 @@ private static async Task GetPostgreSQLCommentCount(string connectionStrin #endregion + #region MySQL helpers + + private static async Task CreateMySqlDatabase(string dbName) + { + var builder = new MySqlConnectionStringBuilder(MySqlConnectionString) + { + Database = "mysql" + }; + + using var conn = new MySqlConnection(builder.ToString()); + await conn.OpenAsync(); + using var cmd = conn.CreateCommand(); + + cmd.CommandText = $"DROP DATABASE IF EXISTS `{dbName}`"; + await cmd.ExecuteNonQueryAsync(); + cmd.CommandText = $"CREATE DATABASE `{dbName}`"; + await cmd.ExecuteNonQueryAsync(); + } + + private static async Task DropMySqlDatabase(string dbName) + { + var builder = new MySqlConnectionStringBuilder(MySqlConnectionString) + { + Database = "mysql" + }; + + using var conn = new MySqlConnection(builder.ToString()); + await conn.OpenAsync(); + using var cmd = conn.CreateCommand(); + cmd.CommandText = $"DROP DATABASE IF EXISTS `{dbName}`"; + await cmd.ExecuteNonQueryAsync(); + } + + private static string GetMySqlConnectionString(string dbName) + { + var builder = new MySqlConnectionStringBuilder(MySqlConnectionString) + { + Database = dbName + }; + return builder.ToString(); + } + + private static async Task CreateMySqlCommentsTable(string connectionString, bool withForeignKey = true) + { + using var conn = new MySqlConnection(connectionString); + await conn.OpenAsync(); + using var cmd = conn.CreateCommand(); + + var fkClause = withForeignKey + ? ", CONSTRAINT `FK_Comments_Parent` FOREIGN KEY (`ParentId`) REFERENCES `Comments`(`Id`)" + : string.Empty; + cmd.CommandText = $@" + CREATE TABLE `Comments` ( + `Id` INT PRIMARY KEY, + `Content` LONGTEXT NOT NULL, + `ParentId` INT NULL{fkClause} + ) ENGINE=InnoDB"; + await cmd.ExecuteNonQueryAsync(); + } + + private static async Task InsertMySqlComments(string connectionString, params (int id, string content, int? parentId)[] records) + { + using var conn = new MySqlConnection(connectionString); + await conn.OpenAsync(); + using var cmd = conn.CreateCommand(); + + foreach (var (id, content, parentId) in records) + { + cmd.CommandText = "INSERT INTO `Comments` (`Id`, `Content`, `ParentId`) VALUES (@id, @content, @parentId)"; + cmd.Parameters.Clear(); + cmd.Parameters.AddWithValue("@id", id); + cmd.Parameters.AddWithValue("@content", content); + cmd.Parameters.AddWithValue("@parentId", parentId.HasValue ? parentId.Value : DBNull.Value); + await cmd.ExecuteNonQueryAsync(); + } + } + + private static async Task GetMySqlCommentCount(string connectionString) + { + using var conn = new MySqlConnection(connectionString); + await conn.OpenAsync(); + using var cmd = conn.CreateCommand(); + cmd.CommandText = "SELECT COUNT(*) FROM `Comments`"; + return Convert.ToInt64((await cmd.ExecuteScalarAsync())!); + } + + private static async Task<(string content, int? parentId)> GetMySqlComment(string connectionString, int id) + { + using var conn = new MySqlConnection(connectionString); + await conn.OpenAsync(); + using var cmd = conn.CreateCommand(); + cmd.CommandText = "SELECT `Content`, `ParentId` FROM `Comments` WHERE `Id` = @id"; + cmd.Parameters.AddWithValue("@id", id); + using var reader = await cmd.ExecuteReaderAsync(); + await reader.ReadAsync(); + var content = reader.GetString(0); + var parentId = reader.IsDBNull(1) ? (int?)null : reader.GetInt32(1); + return (content, parentId); + } + + #endregion + #region Sqlite -> Sqlite [TestMethod] @@ -683,6 +790,102 @@ await TestDeepChain(remoteSyncProvider, localSyncProvider, async () => #endregion + #region MySQL -> MySQL + + [TestMethod] + public async Task SelfReferencingFK_ChildBeforeParent_MySql_MySql() + { + var remoteDb = "coresync_selfrefk_mysql_remote"; + var localDb = "coresync_selfrefk_mysql_local"; + + try + { + await CreateMySqlDatabase(remoteDb); + await CreateMySqlDatabase(localDb); + + var remoteConnStr = GetMySqlConnectionString(remoteDb); + var localConnStr = GetMySqlConnectionString(localDb); + + await CreateMySqlCommentsTable(remoteConnStr); + await CreateMySqlCommentsTable(localConnStr); + + var remoteConfig = new MySqlSyncConfigurationBuilder(remoteConnStr).Table("Comments").Build(); + ISyncProvider remoteSyncProvider = new MySqlSyncProvider(remoteConfig, logger: new ConsoleLogger("REM")); + await remoteSyncProvider.ApplyProvisionAsync(); + + var localConfig = new MySqlSyncConfigurationBuilder(localConnStr).Table("Comments").Build(); + ISyncProvider localSyncProvider = new MySqlSyncProvider(localConfig, logger: new ConsoleLogger("LOC")); + await localSyncProvider.ApplyProvisionAsync(); + + await InsertMySqlComments(remoteConnStr, + (1, "Parent comment", null), + (2, "Child comment", 1)); + + await TestChildBeforeParent(remoteSyncProvider, localSyncProvider, async () => + { + (await GetMySqlCommentCount(localConnStr)).ShouldBe(2); + var (content1, _) = await GetMySqlComment(localConnStr, 1); + content1.ShouldBe("Parent comment"); + var (content2, parentId2) = await GetMySqlComment(localConnStr, 2); + content2.ShouldBe("Child comment"); + parentId2.ShouldBe(1); + }); + } + finally + { + await DropMySqlDatabase(remoteDb); + await DropMySqlDatabase(localDb); + } + } + + [TestMethod] + public async Task SelfReferencingFK_DeepChain_MySql_MySql() + { + var remoteDb = "coresync_selfrefk_mysql_deep_remote"; + var localDb = "coresync_selfrefk_mysql_deep_local"; + + try + { + await CreateMySqlDatabase(remoteDb); + await CreateMySqlDatabase(localDb); + + var remoteConnStr = GetMySqlConnectionString(remoteDb); + var localConnStr = GetMySqlConnectionString(localDb); + + await CreateMySqlCommentsTable(remoteConnStr); + await CreateMySqlCommentsTable(localConnStr); + + var remoteConfig = new MySqlSyncConfigurationBuilder(remoteConnStr).Table("Comments").Build(); + ISyncProvider remoteSyncProvider = new MySqlSyncProvider(remoteConfig, logger: new ConsoleLogger("REM")); + await remoteSyncProvider.ApplyProvisionAsync(); + + var localConfig = new MySqlSyncConfigurationBuilder(localConnStr).Table("Comments").Build(); + ISyncProvider localSyncProvider = new MySqlSyncProvider(localConfig, logger: new ConsoleLogger("LOC")); + await localSyncProvider.ApplyProvisionAsync(); + + await InsertMySqlComments(remoteConnStr, + (1, "Root", null), + (2, "Child", 1), + (3, "Grandchild", 2)); + + await TestDeepChain(remoteSyncProvider, localSyncProvider, async () => + { + (await GetMySqlCommentCount(localConnStr)).ShouldBe(3); + var (_, parentId3) = await GetMySqlComment(localConnStr, 3); + parentId3.ShouldBe(2); + var (_, parentId2) = await GetMySqlComment(localConnStr, 2); + parentId2.ShouldBe(1); + }); + } + finally + { + await DropMySqlDatabase(remoteDb); + await DropMySqlDatabase(localDb); + } + } + + #endregion + #region Mixed: SqlServer remote -> Sqlite local [TestMethod] @@ -778,4 +981,52 @@ await TestChildBeforeParent(remoteSyncProvider, localSyncProvider, async () => } #endregion + + #region Mixed: MySQL remote -> Sqlite local + + [TestMethod] + public async Task SelfReferencingFK_ChildBeforeParent_MySql_Sqlite() + { + var remoteDb = "coresync_selfrefk_mysql_sqlite_remote"; + var localDbFile = Path.Combine(Path.GetTempPath(), $"SelfRefFK_MySql_Sqlite_local_{Guid.NewGuid()}.sqlite"); + + try + { + await CreateMySqlDatabase(remoteDb); + + var remoteConnStr = GetMySqlConnectionString(remoteDb); + await CreateMySqlCommentsTable(remoteConnStr); + + var remoteConfig = new MySqlSyncConfigurationBuilder(remoteConnStr).Table("Comments").Build(); + ISyncProvider remoteSyncProvider = new MySqlSyncProvider(remoteConfig, logger: new ConsoleLogger("REM")); + await remoteSyncProvider.ApplyProvisionAsync(); + + var (localSyncProvider, localConnStr) = await CreateSqliteProvider(localDbFile, "LOC"); + + await InsertMySqlComments(remoteConnStr, + (1, "Parent comment", null), + (2, "Child comment", 1)); + + await TestChildBeforeParent(remoteSyncProvider, localSyncProvider, async () => + { + using var conn = new SqliteConnection(localConnStr); + await conn.OpenAsync(); + using var cmd = conn.CreateCommand(); + + cmd.CommandText = "SELECT COUNT(*) FROM [Comments]"; + ((long)(await cmd.ExecuteScalarAsync())!).ShouldBe(2); + + cmd.CommandText = "SELECT [ParentId] FROM [Comments] WHERE [Id] = 2"; + ((long)(await cmd.ExecuteScalarAsync())!).ShouldBe(1); + }); + } + finally + { + await DropMySqlDatabase(remoteDb); + SqliteConnection.ClearAllPools(); + if (File.Exists(localDbFile)) File.Delete(localDbFile); + } + } + + #endregion } diff --git a/src/CoreSync.sln b/src/CoreSync.sln index 2b6b23b..76e3b12 100644 --- a/src/CoreSync.sln +++ b/src/CoreSync.sln @@ -21,6 +21,8 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "CoreSync.PostgreSQL", "Core EndProject Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "CoreSync.SqlServerCT", "CoreSync.SqlServerCT\CoreSync.SqlServerCT.csproj", "{C28B11F9-8B81-48ED-9626-8131BA922AFA}" EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "CoreSync.MySql", "CoreSync.MySql\CoreSync.MySql.csproj", "{55BD90FF-A941-449E-869C-1BB9BF15CAF2}" +EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution Debug|Any CPU = Debug|Any CPU @@ -139,6 +141,18 @@ Global {C28B11F9-8B81-48ED-9626-8131BA922AFA}.Release|x64.Build.0 = Release|Any CPU {C28B11F9-8B81-48ED-9626-8131BA922AFA}.Release|x86.ActiveCfg = Release|Any CPU {C28B11F9-8B81-48ED-9626-8131BA922AFA}.Release|x86.Build.0 = Release|Any CPU + {55BD90FF-A941-449E-869C-1BB9BF15CAF2}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {55BD90FF-A941-449E-869C-1BB9BF15CAF2}.Debug|Any CPU.Build.0 = Debug|Any CPU + {55BD90FF-A941-449E-869C-1BB9BF15CAF2}.Debug|x64.ActiveCfg = Debug|Any CPU + {55BD90FF-A941-449E-869C-1BB9BF15CAF2}.Debug|x64.Build.0 = Debug|Any CPU + {55BD90FF-A941-449E-869C-1BB9BF15CAF2}.Debug|x86.ActiveCfg = Debug|Any CPU + {55BD90FF-A941-449E-869C-1BB9BF15CAF2}.Debug|x86.Build.0 = Debug|Any CPU + {55BD90FF-A941-449E-869C-1BB9BF15CAF2}.Release|Any CPU.ActiveCfg = Release|Any CPU + {55BD90FF-A941-449E-869C-1BB9BF15CAF2}.Release|Any CPU.Build.0 = Release|Any CPU + {55BD90FF-A941-449E-869C-1BB9BF15CAF2}.Release|x64.ActiveCfg = Release|Any CPU + {55BD90FF-A941-449E-869C-1BB9BF15CAF2}.Release|x64.Build.0 = Release|Any CPU + {55BD90FF-A941-449E-869C-1BB9BF15CAF2}.Release|x86.ActiveCfg = Release|Any CPU + {55BD90FF-A941-449E-869C-1BB9BF15CAF2}.Release|x86.Build.0 = Release|Any CPU EndGlobalSection GlobalSection(SolutionProperties) = preSolution HideSolutionNode = FALSE