Merge pull request #69 from Atralupus/feat/diff-block-poller

Implement diff block poller and initializer

Mimir.Worker/Configuration.cs

@@ -4,6 +4,10 @@ public class Configuration
 {
     public string MongoDbConnectionString { get; init; }
 
+    public string SnapshotPath { get; init; }
+
+    public bool EnableInitializing { get; init; } = false;
+
     public string DatabaseName { get; set; }
 
     public Uri HeadlessEndpoint { get; set; }

Mimir.Worker/Constants/CollectionNames.cs

@@ -0,0 +1,59 @@
+using System.Text.RegularExpressions;
+using Libplanet.Crypto;
+using Nekoyume;
+
+namespace Mimir.Worker.Constants
+{
+    public static class CollectionNames
+    {
+        public static Dictionary<Address, string> CollectionMappings =
+            new Dictionary<Address, string>();
+
+        static CollectionNames()
+        {
+            MapAddressToCollectionName("Shop", Addresses.Shop);
+            MapAddressToCollectionName("Ranking", Addresses.Ranking);
+            MapAddressToCollectionName("WeeklyArena", Addresses.WeeklyArena);
+            MapAddressToCollectionName("TableSheet", Addresses.TableSheet);
+            MapAddressToCollectionName("GameConfig", Addresses.GameConfig);
+            MapAddressToCollectionName("RedeemCode", Addresses.RedeemCode);
+            // mongodb already have admin collection
+            // MapAddressToCollectionName("NineChroniclesAdmin", Addresses.Admin);
+            MapAddressToCollectionName("PendingActivation", Addresses.PendingActivation);
+            MapAddressToCollectionName("ActivatedAccount", Addresses.ActivatedAccount);
+            MapAddressToCollectionName("Blacksmith", Addresses.Blacksmith);
+            MapAddressToCollectionName("GoldCurrency", Addresses.GoldCurrency);
+            MapAddressToCollectionName("GoldDistribution", Addresses.GoldDistribution);
+            MapAddressToCollectionName("AuthorizedMiners", Addresses.AuthorizedMiners);
+            MapAddressToCollectionName("Credits", Addresses.Credits);
+            MapAddressToCollectionName("UnlockWorld", Addresses.UnlockWorld);
+            MapAddressToCollectionName("UnlockEquipmentRecipe", Addresses.UnlockEquipmentRecipe);
+            MapAddressToCollectionName("MaterialCost", Addresses.MaterialCost);
+            MapAddressToCollectionName("StageRandomBuff", Addresses.StageRandomBuff);
+            MapAddressToCollectionName("Arena", Addresses.Arena);
+            MapAddressToCollectionName("SuperCraft", Addresses.SuperCraft);
+            MapAddressToCollectionName("EventDungeon", Addresses.EventDungeon);
+            MapAddressToCollectionName("Raid", Addresses.Raid);
+            MapAddressToCollectionName("Rune", Addresses.Rune);
+            MapAddressToCollectionName("Market", Addresses.Market);
+            MapAddressToCollectionName("GarageWallet", Addresses.GarageWallet);
+            MapAddressToCollectionName("AssetMinters", Addresses.AssetMinters);
+            MapAddressToCollectionName("Agent", Addresses.Agent);
+            MapAddressToCollectionName("Avatar", Addresses.Avatar);
+            MapAddressToCollectionName("Inventory", Addresses.Inventory);
+            MapAddressToCollectionName("WorldInformation", Addresses.WorldInformation);
+            MapAddressToCollectionName("QuestList", Addresses.QuestList);
+            MapAddressToCollectionName("Collection", Addresses.Collection);
+            MapAddressToCollectionName("DailyReward", Addresses.DailyReward);
+            MapAddressToCollectionName("ActionPoint", Addresses.ActionPoint);
+            MapAddressToCollectionName("RuneState", Addresses.RuneState);
+        }
+
+        private static void MapAddressToCollectionName(string name, Address address)
+        {
+            // CamelCase to snake_case for MongoDB naming conventions
+            string collectionName = Regex.Replace(name, @"(?<!^)([A-Z])", "_$1").ToLower();
+            CollectionMappings.Add(address, collectionName);
+        }
+    }
+}

Mimir.Worker/DiffBlockPoller.cs

@@ -0,0 +1,71 @@
+using HeadlessGQL;
+using Mimir.Worker.Scrapper;
+using Mimir.Worker.Services;
+
+namespace Mimir.Worker;
+
+public class DiffBlockPoller
+{
+    private readonly DiffMongoDbService _store;
+    private readonly DiffScrapper _diffScrapper;
+    private readonly ILogger<DiffBlockPoller> _logger;
+    private readonly IStateService _stateService;
+
+    public DiffBlockPoller(
+        ILogger<DiffBlockPoller> logger,
+        HeadlessGQLClient headlessGqlClient,
+        IStateService stateService,
+        DiffMongoDbService store
+    )
+    {
+        _logger = logger;
+        _stateService = stateService;
+        _store = store;
+        _diffScrapper = new DiffScrapper(headlessGqlClient, _store);
+    }
+
+    public async Task RunAsync(CancellationToken stoppingToken)
+    {
+        var started = DateTime.UtcNow;
+
+        _logger.LogInformation("Start DiffBlockPoller background service");
+
+        while (!stoppingToken.IsCancellationRequested)
+        {
+            var currentBlockIndex = await _stateService.GetLatestIndex();
+            var syncedBlockIndex = await GetSyncedBlockIndex(currentBlockIndex);
+            var processBlockIndex = syncedBlockIndex + 1;
+
+            _logger.LogInformation(
+                $"Check BlockIndex process: {processBlockIndex}, current: {currentBlockIndex}"
+            );
+
+            if (processBlockIndex >= currentBlockIndex)
+            {
+                await Task.Delay(TimeSpan.FromMilliseconds(5000), stoppingToken);
+                continue;
+            }
+
+            await _diffScrapper.ExecuteAsync(syncedBlockIndex, currentBlockIndex);
+            await _store.UpdateLatestBlockIndex(currentBlockIndex);
+        }
+        _logger.LogInformation(
+            "Finished DiffBlockPoller background service. Elapsed {TotalElapsedMinutes} minutes",
+            DateTime.UtcNow.Subtract(started).Minutes
+        );
+    }
+
+    public async Task<long> GetSyncedBlockIndex(long currentBlockIndex)
+    {
+        try
+        {
+            var syncedBlockIndex = await _store.GetLatestBlockIndex();
+            return syncedBlockIndex;
+        }
+        catch (System.InvalidOperationException)
+        {
+            _logger.LogError($"Failed to get block indexes from db, Set `syncedBlockIndex` {currentBlockIndex} - 1");
+            return currentBlockIndex - 1;
+        }
+    }
+}

Mimir.Worker/Handler/AddressHandlerMappings.cs

@@ -0,0 +1,29 @@
+using Libplanet.Action.State;
+using Libplanet.Crypto;
+using Mimir.Worker.Constants;
+using Mimir.Worker.Models;
+using Nekoyume;
+
+namespace Mimir.Worker.Handler;
+
+public static class AddressHandlerMappings
+{
+    public static Dictionary<Address, IStateHandler<StateData>> HandlerMappings =
+        new Dictionary<Address, IStateHandler<StateData>>();
+
+    static AddressHandlerMappings()
+    {
+        InitializeHandlers();
+
+        HandlerMappings[Addresses.Avatar] = new AvatarStateHandler();
+    }
+
+    private static void InitializeHandlers()
+    {
+        foreach (var address in CollectionNames.CollectionMappings.Keys)
+        {
+            HandlerMappings.Add(address, null);
+        }
+        HandlerMappings.Add(ReservedAddresses.LegacyAccount, null);
+    }
+}

Mimir.Worker/Handler/AvatarStateHandler.cs

@@ -0,0 +1,43 @@
+using Bencodex;
+using Bencodex.Types;
+using Mimir.Worker.Models;
+using Nekoyume.Model.State;
+
+namespace Mimir.Worker.Handler;
+
+public class AvatarStateHandler : IStateHandler<StateData>
+{
+    public StateData ConvertToStateData(IValue rawState)
+    {
+        var avatarState = ConvertToState(rawState);
+        return new StateData(avatarState.address, avatarState);
+    }
+
+    public StateData ConvertToStateData(string rawState)
+    {
+        Codec Codec = new();
+        var state = Codec.Decode(Convert.FromHexString(rawState));
+        var avatarState = ConvertToState(state);
+
+        return new StateData(avatarState.address, avatarState);
+    }
+
+    private AvatarState ConvertToState(IValue state)
+    {
+        if (state is Dictionary dictionary)
+        {
+            return new AvatarState(dictionary);
+        }
+        else if (state is List alist)
+        {
+            return new AvatarState(alist);
+        }
+        else
+        {
+            throw new ArgumentException(
+                "Invalid state type. Expected Dictionary or List.",
+                nameof(state)
+            );
+        }
+    }
+}

Mimir.Worker/Handler/IStateHandler.cs

@@ -0,0 +1,10 @@
+using Bencodex.Types;
+using Mimir.Worker.Models;
+
+namespace Mimir.Worker.Handler;
+
+public interface IStateHandler<T> where T : StateData
+{
+    T ConvertToStateData(string rawState);
+    T ConvertToStateData(IValue rawState);
+}

Mimir.Worker/Mimir.Worker.csproj

@@ -11,6 +11,7 @@
   <ItemGroup>
     <PackageReference Include="Microsoft.Extensions.Hosting" Version="8.0.0" />
     <PackageReference Include="Lib9c" Version="1.8.0-dev.c5644aff871034a92042833edca03b85c2a24a1c" />
+    <PackageReference Include="Libplanet.RocksDBStore" Version="4.4.1" />
     <PackageReference Include="Libplanet" Version="4.4.1" />
     <PackageReference Include="MongoDB.Driver" Version="2.25.0" />
     <PackageReference Include="MongoDB.Driver.GridFS" Version="2.25.0" />

Mimir.Worker/Models/State/StateData.cs

@@ -0,0 +1,17 @@
+using Libplanet.Crypto;
+using Nekoyume.Model.State;
+
+namespace Mimir.Worker.Models;
+
+public class StateData : BaseData
+{
+    public Address Address { get; }
+
+    public State State { get; }
+
+    public StateData(Address address, State state)
+    {
+        Address = address;
+        State = state;
+    }
+}

Mimir.Worker/Program.cs

@@ -1,49 +1,92 @@
 using System.IdentityModel.Tokens.Jwt;
 using System.Net.Http.Headers;
 using System.Text;
+using HeadlessGQL;
+using Microsoft.Extensions.Options;
 using Microsoft.IdentityModel.Tokens;
 using Mimir.Worker;
 using Mimir.Worker.Services;
-using Microsoft.Extensions.Options;
 
 var builder = Host.CreateApplicationBuilder(args);
 
 string configPath = Environment.GetEnvironmentVariable("WORKER_CONFIG_FILE") ?? "appsettings.json";
-builder.Configuration
-    .AddJsonFile(configPath, optional: true, reloadOnChange: true)
+builder
+    .Configuration.AddJsonFile(configPath, optional: true, reloadOnChange: true)
     .AddEnvironmentVariables("WORKER_");
 
 builder.Services.Configure<Configuration>(builder.Configuration.GetSection("Configuration"));
 
 builder.Services.AddSingleton<IStateService, HeadlessStateService>();
-builder.Services.AddHeadlessGQLClient()
-    .ConfigureHttpClient((provider, client) =>
-    {
-        var headlessStateServiceOption = provider.GetRequiredService<IOptions<Configuration>>();
-        client.BaseAddress = headlessStateServiceOption.Value.HeadlessEndpoint;
-
-        if (headlessStateServiceOption.Value.JwtSecretKey is not null && headlessStateServiceOption.Value.JwtIssuer is not null)
+builder
+    .Services.AddHeadlessGQLClient()
+    .ConfigureHttpClient(
+        (provider, client) =>
         {
-            var key = new SymmetricSecurityKey(Encoding.UTF8.GetBytes(headlessStateServiceOption.Value.JwtSecretKey));
-            var creds = new SigningCredentials(key, SecurityAlgorithms.HmacSha256);
+            var headlessStateServiceOption = provider.GetRequiredService<IOptions<Configuration>>();
+            client.BaseAddress = headlessStateServiceOption.Value.HeadlessEndpoint;
 
-            var token = new JwtSecurityToken(
-                issuer: headlessStateServiceOption.Value.JwtIssuer,
-                expires: DateTime.UtcNow.AddMinutes(5),
-                signingCredentials: creds);
+            if (
+                headlessStateServiceOption.Value.JwtSecretKey is not null
+                && headlessStateServiceOption.Value.JwtIssuer is not null
+            )
+            {
+                var key = new SymmetricSecurityKey(
+                    Encoding.UTF8.GetBytes(headlessStateServiceOption.Value.JwtSecretKey)
+                );
+                var creds = new SigningCredentials(key, SecurityAlgorithms.HmacSha256);
 
-            client.DefaultRequestHeaders.Authorization =
-                new AuthenticationHeaderValue("Bearer", new JwtSecurityTokenHandler().WriteToken(token));
+                var token = new JwtSecurityToken(
+                    issuer: headlessStateServiceOption.Value.JwtIssuer,
+                    expires: DateTime.UtcNow.AddMinutes(5),
+                    signingCredentials: creds
+                );
+
+                client.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue(
+                    "Bearer",
+                    new JwtSecurityTokenHandler().WriteToken(token)
+                );
+            }
         }
-    });
+    );
 
 builder.Services.AddSingleton(serviceProvider =>
 {
     var config = serviceProvider.GetRequiredService<IOptions<Configuration>>().Value;
     var logger = serviceProvider.GetRequiredService<ILogger<MongoDbStore>>();
     return new MongoDbStore(logger, config.MongoDbConnectionString, config.DatabaseName);
 });
-builder.Services.AddHostedService<Initializer>();
+builder.Services.AddSingleton(serviceProvider =>
+{
+    var config = serviceProvider.GetRequiredService<IOptions<Configuration>>().Value;
+    var logger = serviceProvider.GetRequiredService<ILogger<DiffMongoDbService>>();
+    return new DiffMongoDbService(
+        logger,
+        config.MongoDbConnectionString,
+        config.DatabaseName + "_diff_test"
+    );
+});
+builder.Services.AddHostedService(serviceProvider =>
+{
+    var config = serviceProvider.GetRequiredService<IOptions<Configuration>>().Value;
+    var logger = serviceProvider.GetRequiredService<ILogger<Worker>>();
+    var blockPollerLogger = serviceProvider.GetRequiredService<ILogger<DiffBlockPoller>>();
+    var initializerLogger = serviceProvider.GetRequiredService<ILogger<SnapshotInitializer>>();
+    var headlessGqlClient = serviceProvider.GetRequiredService<HeadlessGQLClient>();
+    var stateService = serviceProvider.GetRequiredService<IStateService>();
+    var store = serviceProvider.GetRequiredService<DiffMongoDbService>();
+
+    return new Worker(
+        logger,
+        blockPollerLogger,
+        initializerLogger,
+        headlessGqlClient,
+        stateService,
+        store,
+        config.SnapshotPath,
+        config.EnableInitializing
+    );
+});
+// builder.Services.AddHostedService<Initializer>();
 
 var host = builder.Build();
 host.Run();

Mimir.Worker/Queries.graphql

@@ -33,3 +33,21 @@ query GetTransactionSigners($blockIndex: Long!) {
         }
     }
 }
+
+query GetDiffs($baseIndex: Long! $changedIndex: Long!) {
+  diffs(baseIndex: $baseIndex, changedIndex: $changedIndex) {
+    ... on RootStateDiff {
+        path
+        diffs {
+            path
+            baseState
+            changedState
+        }
+    }
+    ... on StateDiff {
+        path
+        baseState
+        changedState
+    }
+  }
+}