using System.Data.Common; using System.Reflection; using System.Linq.Expressions; using GreenHome.Infrastructure.Data; using Microsoft.Data.SqlClient; using Microsoft.EntityFrameworkCore; using Microsoft.EntityFrameworkCore.Metadata; using Microsoft.Extensions.Configuration; using Npgsql; namespace GreenHome.DataCopy; // ============================================================ // GreenHome.DataCopy — SQL Server -> PostgreSQL data copy tool // // Usage: // dotnet run --project GreenHome.DataCopy (idempotent, resumable) // dotnet run --project GreenHome.DataCopy -- --reset (wipe target tables first) // // Reads from SQL Server (read-only, via EF), writes to PostgreSQL (via EF), // preserving PKs. EF handles all column type conversions (bytea, jsonb, ...). // Safe to re-run after a crash: only missing rows are copied. // Ends with a row-count self-check per table (exit code 1 on mismatch). // ============================================================ internal static class Program { const int BatchSize = 5000; // Generic copy handlers, dispatched per entity type via reflection. static readonly MethodInfo SingleKeyCopy = typeof(Program).GetMethod(nameof(CopySingleIntKeyTable), BindingFlags.NonPublic | BindingFlags.Static)!; static readonly MethodInfo CompositeKeyCopy = typeof(Program).GetMethod(nameof(CopyCompositeKeyTable), BindingFlags.NonPublic | BindingFlags.Static)!; static async Task Main(string[] args) { var config = new ConfigurationBuilder() .SetBasePath(AppContext.BaseDirectory) .AddJsonFile("appsettings.json") .AddEnvironmentVariables() // lets the server pass ConnectionStrings__* env vars .Build(); var sqlServerCs = config.GetConnectionString("SqlServer") ?? throw new InvalidOperationException("Missing ConnectionStrings:SqlServer"); var pgCs = config.GetConnectionString("Postgres") ?? throw new InvalidOperationException("Missing ConnectionStrings:Postgres"); var reset = args.Contains("--reset"); using var source = new GreenHomeDbContext( new DbContextOptionsBuilder().UseSqlServer(sqlServerCs).Options); using var target = new GreenHomeDbContext( new DbContextOptionsBuilder().UseNpgsql(pgCs).Options); target.Database.EnsureCreated(); var entityTypes = target.Model.GetEntityTypes() .Where(e => e.GetTableName() is not null) .Select(e => (Entity: e, Table: e.GetTableName()!)) .ToList(); // --- Topological sort: parents before children (Kahn) --- var deps = entityTypes.ToDictionary( t => t.Table, t => t.Entity.GetForeignKeys() .Select(fk => fk.PrincipalEntityType.GetTableName()!) .Where(pt => pt != t.Table) .ToHashSet()); var order = new List<(IEntityType Entity, string Table)>(); var remaining = deps.ToDictionary(kv => kv.Key, kv => kv.Value.ToHashSet()); while (remaining.Count > 0) { var ready = remaining.Where(kv => kv.Value.Count == 0).Select(kv => kv.Key).OrderBy(t => t).ToList(); if (ready.Count == 0) ready = [remaining.Keys.OrderBy(t => t).First()]; // cycle fallback (shouldn't happen) foreach (var table in ready) { order.Add(entityTypes.First(e => e.Table == table)); remaining.Remove(table); foreach (var kv in remaining) kv.Value.Remove(table); } } // --- Optional reset: children first (reverse dependency order) --- if (reset) { foreach (var (_, table) in order.AsEnumerable().Reverse()) await target.Database.ExecuteSqlRawAsync($"DELETE FROM \"{table}\""); Console.WriteLine("[reset] all tables cleared"); } // --- Copy each table --- foreach (var (entity, table) in order) { var pk = entity.FindPrimaryKey(); if (pk is null) { Console.WriteLine($"[skip ] {table} (no primary key)"); continue; } if (pk.Properties.Count == 1 && pk.Properties[0].ClrType == typeof(int)) await (Task)SingleKeyCopy.MakeGenericMethod(entity.ClrType) .Invoke(null, [source, target, pgCs, entity, table, BatchSize])!; else await (Task)CompositeKeyCopy.MakeGenericMethod(entity.ClrType) .Invoke(null, [source, target, entity, BatchSize])!; } // --- Self-check: row counts must match --- Console.WriteLine(); var problems = 0; await using var srcConn = new SqlConnection(sqlServerCs); await using var pgConn = new NpgsqlConnection(pgCs); await srcConn.OpenAsync(); await pgConn.OpenAsync(); foreach (var (_, table) in order) { // SQL Server COUNT(*) is Int32, Postgres is Int64 — Convert handles both long src = Convert.ToInt64(await ScalarAsync(srcConn, $"SELECT COUNT(*) FROM {table}"))!; long dst = Convert.ToInt64(await ScalarAsync(pgConn, $"SELECT COUNT(*) FROM \"{table}\""))!; if (src != dst) problems++; Console.WriteLine($"{(src == dst ? "OK " : "FAIL ")} {table,-28} source={src,-10} target={dst}"); } Console.WriteLine(problems == 0 ? "\nSelf-check passed: all tables match." : $"\nSelf-check FAILED: {problems} table(s) mismatched."); return problems == 0 ? 0 : 1; } // ============================================================ // Generic copy handlers // ============================================================ static async Task CopySingleIntKeyTable( GreenHomeDbContext source, GreenHomeDbContext target, string pgCs, IEntityType entity, string table, int batchSize) where TEntity : class { var pkName = entity.FindPrimaryKey()!.Properties[0].Name; await using var pg = new NpgsqlConnection(pgCs); await pg.OpenAsync(); // Resume point: only copy rows with PK greater than what target already has var frontier = (int)(await ScalarAsync(pg, $"SELECT COALESCE(MAX(\"{pkName}\"), 0) FROM \"{table}\""))!; long copied = 0; while (true) { var param = Expression.Parameter(typeof(TEntity), "e"); var pkProp = Expression.Property(param, pkName); var where = Expression.Lambda>( Expression.GreaterThan(pkProp, Expression.Constant(frontier)), param); var orderBy = Expression.Lambda>(pkProp, param); var batch = await source.Set().AsNoTracking() .Where(where).OrderBy(orderBy).Take(batchSize).ToListAsync(); if (batch.Count == 0) break; target.Set().AddRange(batch); await target.SaveChangesAsync(); copied += batch.Count; frontier = batch.Max(e => (int)GetProp(e, pkName)!); Console.Write($"\r[copy ] {table}: {copied} rows (pk={frontier}) "); } var max = (int)(await ScalarAsync(pg, $"SELECT COALESCE(MAX(\"{pkName}\"), 0) FROM \"{table}\""))!; if (max > 0) { // Keep the identity sequence above the copied max so new rows don't collide. // pg_get_serial_sequence(text, text) — verified signature. format('%I', ...) // preserves the exact table case EF used; the column is passed as plain text. await target.Database.ExecuteSqlRawAsync( $"SELECT setval(pg_get_serial_sequence(format('%I', '{table}'), '{pkName}'), {max})"); } Console.WriteLine($"\n[done ] {table}: copied={copied}, maxId={max}"); } static async Task CopyCompositeKeyTable( GreenHomeDbContext source, GreenHomeDbContext target, IEntityType entity, int batchSize) where TEntity : class { var pkNames = entity.FindPrimaryKey()!.Properties.Select(p => p.Name).ToArray(); // Composite-PK tables are small (e.g. DeviceUsers): load all source rows, // skip ones already present, write in batches. var existing = new HashSet(); foreach (var e in await target.Set().AsNoTracking().ToListAsync()) existing.Add(KeyOf(e, pkNames)); long copied = 0; foreach (var row in await source.Set().AsNoTracking().ToListAsync()) { if (existing.Contains(KeyOf(row, pkNames))) continue; target.Set().Add(row); existing.Add(KeyOf(row, pkNames)); copied++; if (copied % batchSize == 0) await target.SaveChangesAsync(); } await target.SaveChangesAsync(); Console.WriteLine($"[done ] {typeof(TEntity).Name} (composite PK): copied={copied}"); } // ============================================================ // Plumbing // ============================================================ static async Task ScalarAsync(DbConnection conn, string sql) { await using var cmd = conn.CreateCommand(); cmd.CommandText = sql; return await cmd.ExecuteScalarAsync(); } static object? GetProp(object entity, string name) => entity.GetType().GetProperty(name)?.GetValue(entity); static string KeyOf(object entity, string[] pkNames) => string.Join("|", pkNames.Select(n => GetProp(entity, n)?.ToString() ?? "null")); }