Replace RavenDB Hub/Sink replication with ETL Sink task

- Remove VerityReplicationHub and VerityReplicationSink (pull replication model)
- Remove migration 005_ConfigureHubSink
- Add VeritySinkEtlTask — push-based ETL via RavenDB ETL Sink
- Add migration 005_ConfigureSinkEtl to register the ETL task
- Minor whitespace cleanup in SecEdgarCompanyImporter

Author: Mateosssss

src/RavenDB.Samples.Verity.Model/HubSinks/VerityReplicationHub.cs

@@ -9,12 +9,12 @@ namespace RavenDB.Samples.Verity.Model;
 public static class SecEdgarCompanyImporter
 {
     public record CompanyImportData(
-        string   PaddedCik,
-        string   Name,
-        string?  Sic,
-        string?  SicDescription,
+        string PaddedCik,
+        string Name,
+        string? Sic,
+        string? SicDescription,
         DateTime FiscalYearStart);
-    
+
     private static readonly string[] FirstNames =
     [
         "James", "Mary", "Robert", "Patricia", "Michael",
@@ -37,10 +37,10 @@ public static async Task<CompanyImportData> FetchCompanyDataAsync(
         response.EnsureSuccessStatusCode();
 
         await using var stream = await response.Content.ReadAsStreamAsync(ct);
-        using var doc  = await JsonDocument.ParseAsync(stream, cancellationToken: ct);
-        var       root = doc.RootElement;
+        using var doc = await JsonDocument.ParseAsync(stream, cancellationToken: ct);
+        var root = doc.RootElement;
 
-        var name          = root.GetProperty("name").GetString() ?? paddedCik;
+        var name = root.GetProperty("name").GetString() ?? paddedCik;
         var fiscalYearEnd = root.TryGetProperty("fiscalYearEnd", out var fye) ? fye.GetString() : null;
 
         var fiscalYearStart = new DateTime(1, int.Parse(fiscalYearEnd!.Substring(0, 2)), 1, 0, 0, 0, DateTimeKind.Utc)
@@ -49,9 +49,9 @@ public static async Task<CompanyImportData> FetchCompanyDataAsync(
             fiscalYearStart = fiscalYearStart.AddYears(-1);
 
         return new CompanyImportData(
-            PaddedCik:      paddedCik,
-            Name:           name,
-            Sic:            root.TryGetProperty("sic",            out var sic)     ? sic.GetString()     : null,
+            PaddedCik: paddedCik,
+            Name: name,
+            Sic: root.TryGetProperty("sic", out var sic) ? sic.GetString() : null,
             SicDescription: root.TryGetProperty("sicDescription", out var sicDesc) ? sicDesc.GetString() : null,
             FiscalYearStart: fiscalYearStart);
     }
@@ -66,43 +66,43 @@ public static async Task<Company> StoreCompanyAsync(
         CancellationToken ct = default)
     {
         var rng = Random.Shared;
-        
+
         var companyId = Company.BuildId(data.Name);
 
         if (await session.Advanced.ExistsAsync(companyId, ct))
             return (await session.LoadAsync<Company>(companyId, ct))!;
 
         var company = new Company
         {
-            Id              = companyId,
-            Name            = data.Name,
-            Cik             = data.PaddedCik,
-            Sic             = data.Sic,
-            SicDescription  = data.SicDescription,
+            Id = companyId,
+            Name = data.Name,
+            Cik = data.PaddedCik,
+            Sic = data.Sic,
+            SicDescription = data.SicDescription,
             FiscalYearStart = data.FiscalYearStart,
         };
 
         await session.StoreAsync(company, companyId, ct);
 
         var usedPairs = new HashSet<string>();
-        var domain    = data.Name.Replace(" ", "").Replace(",", "").Replace(".", "").ToLowerInvariant();
+        var domain = data.Name.Replace(" ", "").Replace(",", "").Replace(".", "").ToLowerInvariant();
 
         for (var i = 0; i < 2; i++)
         {
             string firstName, lastName;
             do
             {
                 firstName = FirstNames[rng.Next(FirstNames.Length)];
-                lastName  = LastNames[rng.Next(LastNames.Length)];
+                lastName = LastNames[rng.Next(LastNames.Length)];
             } while (!usedPairs.Add($"{firstName} {lastName}"));
 
             await session.StoreAsync(new User
             {
-                Id        = User.BuildId(data.Name, firstName, lastName),
-                CompanyId = companyId,
-                Name      = firstName,
-                Surname   = lastName,
-                Email     = $"{firstName.ToLower()}{lastName.ToLower()}@{domain}.com"
+                Id = User.BuildId(data.Name, firstName, lastName),
+                CompanyIds = [companyId],
+                Name = firstName,
+                Surname = lastName,
+                Email = $"{firstName.ToLower()}{lastName.ToLower()}@{domain}.com"
             }, ct);
         }
 

src/RavenDB.Samples.Verity.Model/HubSinks/VerityReplicationSink.cs

@@ -0,0 +1,33 @@
+using Raven.Client.Documents.Operations.ETL;
+using RavenDB.Samples.Verity.Model;          // żeby widzieć Company / Report
+
+namespace RavenDB.Samples.Verity.Model.Tasks;
+
+public static class VeritySinkEtlTask
+{
+    public const string ConnectionStringName = "Verity Sink Connection";
+    public const string TaskName = "VeritySinkEtlTask";
+
+    public static RavenEtlConfiguration Create() => new()
+    {
+        Name = TaskName,
+        ConnectionStringName = ConnectionStringName,
+        Disabled = false,
+        Transforms =
+        [
+            new Transformation
+            {
+                Name        = "Companies",
+                Collections = [Company.Collection],
+                Script      = "loadToCompanies(this);"
+            },
+            new Transformation
+            {
+                Name        = "Reports",
+                Collections = [Report.Collection],
+                Script      = "if (this.Summary) loadToReports(this);"
+            }
+        ]
+
+    };
+}

src/RavenDB.Samples.Verity.Model/SecEdgarCompanyImporter.cs

@@ -0,0 +1,40 @@
+using Raven.Client.Documents.Operations.ConnectionStrings;
+using Raven.Client.Documents.Operations.ETL;
+using Raven.Client.ServerWide;
+using Raven.Client.ServerWide.Operations;
+using Raven.Migrations;
+using RavenDB.Samples.Verity.Model.Tasks;
+
+namespace RavenDB.Samples.Verity.Setup.Migrations;
+
+[Migration(5)]
+public sealed class ConfigureSinkEtl(MigrationContext context) : Migration
+{
+    public override void Up()
+    {
+        try
+        {
+            DocumentStore.Maintenance.Server.Send(
+            new CreateDatabaseOperation(new DatabaseRecord(Constants.DatabaseSinkName)));
+        }
+        catch (Exception ex) when (ex.Message.Contains("already exists")) { }
+
+        DocumentStore.Maintenance.Send(new PutConnectionStringOperation<RavenConnectionString>(new RavenConnectionString
+        {
+            Name = VeritySinkEtlTask.ConnectionStringName,
+            Database = Constants.DatabaseSinkName,
+            TopologyDiscoveryUrls = string.IsNullOrEmpty(context.HubServerInternalUrl)
+                ? DocumentStore.Urls
+                : [context.HubServerInternalUrl]
+        }));
+
+        // 3) ETL na źródle (Verity)
+        DocumentStore.Maintenance.Send(new AddEtlOperation<RavenConnectionString>(VeritySinkEtlTask.Create()));
+    }
+
+    public override void Down()
+    {
+        throw new NotSupportedException(
+            "Rolling back the Sink ETL is not supported. Remove the ETL task and connection string via RavenDB Studio.");
+    }
+}