feat: multi-cosmos indexing

Directory.Build.props

@@ -0,0 +1,7 @@
+<Project>
+    <PropertyGroup>
+        <LangVersion>10</LangVersion>
+        <Nullable>enable</Nullable>
+        <ImplicitUsings>disable</ImplicitUsings>
+    </PropertyGroup>
+</Project>

SapientFi.Cosmos.sln.DotSettings

@@ -2,6 +2,16 @@
 	<s:String x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/PLACE_ACCESSORHOLDER_ATTRIBUTE_ON_SAME_LINE_EX/@EntryValue">NEVER</s:String>
 	<s:String x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/PLACE_ACCESSOR_ATTRIBUTE_ON_SAME_LINE_EX/@EntryValue">NEVER</s:String>
 	<s:String x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/PLACE_FIELD_ATTRIBUTE_ON_SAME_LINE_EX/@EntryValue">NEVER</s:String>
+	<s:Boolean x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/WRAP_AFTER_DECLARATION_LPAR/@EntryValue">True</s:Boolean>
+	<s:Boolean x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/WRAP_BEFORE_DECLARATION_RPAR/@EntryValue">True</s:Boolean>
+	<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ECSharpKeepExistingMigration/@EntryIndexedValue">True</s:Boolean>
+	<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ECSharpPlaceEmbeddedOnSameLineMigration/@EntryIndexedValue">True</s:Boolean>
+	<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ECSharpUseContinuousIndentInsideBracesMigration/@EntryIndexedValue">True</s:Boolean>
+	<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ESettingsUpgrade_002EMigrateBlankLinesAroundFieldToBlankLinesAroundProperty/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/UserDictionary/Words/=appsettings/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/UserDictionary/Words/=Denom/@EntryIndexedValue">True</s:Boolean>
+	<s:Boolean x:Key="/Default/UserDictionary/Words/=Kujira/@EntryIndexedValue">True</s:Boolean>
+	<s:Boolean x:Key="/Default/UserDictionary/Words/=Querier/@EntryIndexedValue">True</s:Boolean>
+	<s:Boolean x:Key="/Default/UserDictionary/Words/=Queriers/@EntryIndexedValue">True</s:Boolean>
+	<s:Boolean x:Key="/Default/UserDictionary/Words/=Redelegate/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/UserDictionary/Words/=SAPIENTFI/@EntryIndexedValue">True</s:Boolean></wpf:ResourceDictionary>

docker-compose.yml

@@ -2,7 +2,7 @@ version: '3.9'
 
 services:
   db:
-    image: timescale/timescaledb:2.5.0-pg12
+    image: timescale/timescaledb:2.10.0-pg15
     command: ["postgres", "-c", "log_statement=all", "-c", "max_locks_per_transaction=4096"]
     shm_size: 1g
     ports:
@@ -16,7 +16,7 @@ services:
       - .docker/postgres/data:/var/lib/postgresql/data
 
   rabbitmq:
-    image: masstransit/rabbitmq:3.9
+    image: rabbitmq:3.11-management
     environment:
       # default user + pass is "guest"
       RABBITMQ_ERLANG_COOKIE: ${RABBITMQ_ERLANG_COOKIE}
@@ -32,7 +32,7 @@ services:
       - "35433:6379"
 
   logging:
-    image: datalust/seq:5.1
+    image: datalust/seq:latest
     ports:
       - "35435:80"
     environment:

docs/configuration.md

@@ -0,0 +1,19 @@
+# Configuration
+
+Keys marked with :exclamation: should be set when deploying.
+
+## General
+
+| Key                                          | Type                 | Default                                                                                                                                                                             | Example         | Description |
+|----------------------------------------------|----------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|-----------------|-------------|
+| **General**                                  |                      |                                                                                                                                                                                     |                 |             |
+| `SAPIENTFI_SERVICE_ROLES_ENABLED`            | comma-separated-list | `API, BACKGROUND_WORKER`                                                                                                                                                            | `Foo, Bar, Baz` |             |
+| `SAPIENTFI_DB_CONNECTION_STRING`             | string               | `User ID=sapientfi_indexer_user;Password=sapientfi_indexer_user_pass;Host=localhost;Port=35432;Database=sapientfi_indexer;Pooling=true;Minimum Pool Size=10;Maximum Pool Size=100;` |                 |             |
+| `SAPIENTFI_DB_RUN_MIGRATIONS_ON_BOOT`        | bool                 | `true`                                                                                                                                                                              |                 |             |
+| `SAPIENTFI_NR_ELEMENTS_DEPOSITS_PR_WALLET`   | int                  | `11`                                                                                                                                                                                |                 |             |
+| `SAPIENTFI_API_ALLOWED_CORS_ORIGINS`         | string               | `http://localhost:3000`                                                                                                                                                             |                 |             |
+| `SAPIENTFI_TRIGGER_GATEWAY_POOL_FULL_RESYNC` | bool                 | `false`                                                                                                                                                                             |                 |             |
+| `SAPIENTFI_TRIGGER_MINE_STAKING_FULL_RESYNC` | bool                 | `false`                                                                                                                                                                             |                 |             |
+| `SAPIENTFI_TRIGGER_MINE_BUYBACK_FULL_RESYNC` | bool                 | `false`                                                                                                                                                                             |                 |             |
+| `SAPIENTFI_MESSAGE_TRANSPORT_URI`            | string               | `amqp://guest:guest@localhost:5672`                                                                                                                                                 |                 |             |
+| `Kujira_TransactionListener_DoEnable`        | bool                 | `true`                                                                                                                                                                              |                 |             |

src/Infrastructure/Cosmos/BusMessages/IRawCosmosTransactionAvailableAnnouncement.cs

@@ -0,0 +1,17 @@
+using System;
+
+namespace SapientFi.Infrastructure.Cosmos.BusMessages;
+
+public interface IRawCosmosTransactionAvailableAnnouncement
+{
+    public string TransactionHash { get; init; }
+    /// <summary>
+    /// Primary key id in the underlying database
+    /// </summary>
+    public long RawEntityId { get; init; }
+
+    /// <summary>
+    /// Timestamp in the entity for it's creation date.
+    /// </summary>
+    public DateTimeOffset CreatedAt { get; init; }
+}

src/Infrastructure/Cosmos/CosmosFactory.cs

@@ -0,0 +1,31 @@
+using System.Text.Json;
+using SapientFi.Infrastructure.Cosmos.Storage;
+using SapientFi.Kernel.IdGeneration;
+using TerraDotnet;
+using TerraDotnet.TerraLcd.Messages;
+
+namespace SapientFi.Infrastructure.Cosmos;
+
+public class CosmosFactory<TRawTransactionEntity>
+    : ICosmosFactory<TRawTransactionEntity>
+    where TRawTransactionEntity : ICosmosRawTransactionEntity, new()
+{
+    protected readonly IdProvider IdProvider;
+
+    public CosmosFactory(IdProvider idProvider)
+    {
+        IdProvider = idProvider;
+    }
+
+    public virtual TRawTransactionEntity NewRawEntity(LcdTxResponse lcdTransaction)
+    {
+        return new()
+        {
+            Id = IdProvider.Snowflake(),
+            Height = lcdTransaction.HeightAsInt,
+            CreatedAt = lcdTransaction.CreatedAt,
+            TxHash = lcdTransaction.TransactionHash,
+            RawTx = JsonSerializer.Serialize(lcdTransaction, TerraJsonSerializerOptions.GetThem())
+        };
+    }
+}

src/Infrastructure/Cosmos/CosmosMarker.cs

@@ -0,0 +1,3 @@
+namespace SapientFi.Infrastructure.Cosmos;
+
+public abstract record CosmosMarker;

src/Infrastructure/Cosmos/ICosmosFactory.cs

@@ -0,0 +1,10 @@
+using SapientFi.Infrastructure.Cosmos.Storage;
+using TerraDotnet.TerraLcd.Messages;
+
+namespace SapientFi.Infrastructure.Cosmos;
+
+public interface ICosmosFactory<out TRawTransactionEntity>
+    where TRawTransactionEntity : ICosmosRawTransactionEntity, new()
+{
+    public TRawTransactionEntity NewRawEntity(LcdTxResponse lcdTransaction);
+}

src/Infrastructure/Cosmos/Indexers/Delegations/CosmosDelegationIndexer.cs

@@ -0,0 +1,205 @@
+using System;
+using System.Text.Json;
+using System.Threading.Tasks;
+using MassTransit;
+using Microsoft.Extensions.Logging;
+using SapientFi.Infrastructure.Cosmos.BusMessages;
+using SapientFi.Infrastructure.Cosmos.Indexers.Delegations.Storage;
+using SapientFi.Infrastructure.Cosmos.Storage;
+using SapientFi.Infrastructure.Indexing;
+using SapientFi.Kernel.IdGeneration;
+using TerraDotnet;
+using TerraDotnet.TerraFcd.Messages;
+using TerraDotnet.TerraLcd.Messages;
+
+namespace SapientFi.Infrastructure.Cosmos.Indexers.Delegations;
+
+public abstract class
+    CosmosDelegationIndexer<
+        TRawTransactionAvailableAnnouncement,
+        TValidatorDelegationLedgerEntity,
+        TRawTransactionEntity>
+    : IIndexer<TRawTransactionAvailableAnnouncement>
+    where TRawTransactionAvailableAnnouncement : class, IRawCosmosTransactionAvailableAnnouncement
+    where TValidatorDelegationLedgerEntity : ICosmosValidatorDelegationLedgerEntity, new()
+    where TRawTransactionEntity : ICosmosRawTransactionEntity, new()
+{
+    protected readonly
+        ILogger<CosmosDelegationIndexer<
+            TRawTransactionAvailableAnnouncement,
+            TValidatorDelegationLedgerEntity,
+            TRawTransactionEntity>> Logger;
+    protected readonly CosmosDelegationsRepository<TValidatorDelegationLedgerEntity> Repository;
+    protected readonly CosmosRawRepository<TRawTransactionEntity> RawRepository;
+    protected readonly IdProvider IdProvider;
+
+    protected CosmosDelegationIndexer(
+        ILogger<CosmosDelegationIndexer<
+            TRawTransactionAvailableAnnouncement,
+            TValidatorDelegationLedgerEntity,
+            TRawTransactionEntity>> logger,
+        CosmosDelegationsRepository<TValidatorDelegationLedgerEntity> repository,
+        CosmosRawRepository<TRawTransactionEntity> rawRepository,
+        IdProvider idProvider
+    )
+    {
+        Logger = logger;
+        Repository = repository;
+        RawRepository = rawRepository;
+        IdProvider = idProvider;
+    }
+
+    public abstract string NameOfBlockChain { get; }
+    public abstract string Id { get; }
+    public abstract string DisplayName { get; }
+
+    public virtual async Task Consume(ConsumeContext<TRawTransactionAvailableAnnouncement> context)
+    {
+        await DefaultConsume(context);
+    }
+
+    protected async Task DefaultConsume(ConsumeContext<TRawTransactionAvailableAnnouncement> context)
+    {
+        var rawTransaction = await RawRepository.GetByIdAndCreateOrDefaultAsync(
+            context.Message.RawEntityId,
+            context.Message.CreatedAt,
+            context.CancellationToken
+        );
+        if (rawTransaction is null)
+        {
+            // TODO write to a custom metric in order to be able to monitor missing raw transactions
+            Logger.LogWarning(
+                "({BlockChain}) Unable to find raw transaction with id={RawEntityId}",
+                NameOfBlockChain,
+                context.Message.RawEntityId
+            );
+            // throw for retry!
+            throw new Exception($"Unable to find raw transaction with id={context.Message.RawEntityId}")
+        }
+
+        var txHash = rawTransaction.TxHash;
+
+        // TODO: deal with re-entrance/re-processing i.e. make sure we do not make duplicate entries if we process the same TX multiple times
+
+        var tx = JsonSerializer.Deserialize<LcdTxResponse>(rawTransaction.RawTx);
+        if (tx == default)
+        {
+            Logger.LogWarning(
+                "({BlockChain}) Could not deserialize transaction with hash={TxHash}",
+                NameOfBlockChain,
+                txHash
+            );
+        }
+
+        foreach (var txMessage in tx!.Transaction.Body.Messages)
+        {
+            var parseSuccessful = TerraMessageParser.TryParse(txMessage, out IMsg? message);
+            if (parseSuccessful)
+            {
+                switch (message)
+                {
+                    case CosmosUndelegateMessage undelegateMsg:
+                    {
+                        if (undelegateMsg.Amount == default)
+                        {
+                            Logger.LogWarning(
+                                "({BlockChain}) Missing amount on undelegate message (txHash={TxHash})",
+                                NameOfBlockChain,
+                                txHash
+                            );
+                            break;
+                        }
+
+                        var undelegateEntity = new TValidatorDelegationLedgerEntity
+                        {
+                            Id = IdProvider.Snowflake(),
+                            TxHash = txHash,
+                            At = rawTransaction.CreatedAt,
+                            DelegatorAddress = undelegateMsg.DelegatorAddress,
+                            ValidatorAddress = undelegateMsg.ValidatorAddress,
+                            Amount = -long.Parse(undelegateMsg.Amount.Amount), // subtracts amount
+                            Denominator = undelegateMsg.Amount.Denominator
+                        };
+
+                        await Repository.SaveAsync(undelegateEntity, context.CancellationToken);
+                        break;
+                    }
+
+                    case CosmosDelegateMessage delegateMsg:
+                    {
+                        if (delegateMsg.Amount == default)
+                        {
+                            Logger.LogWarning(
+                                "({BlockChain}) Missing amount on delegate message (txHash={TxHash})",
+                                NameOfBlockChain,
+                                txHash
+                            );
+                            break;
+                        }
+
+                        var delegateEntity = new TValidatorDelegationLedgerEntity
+                        {
+                            Id = IdProvider.Snowflake(),
+                            TxHash = txHash,
+                            At = rawTransaction.CreatedAt,
+                            DelegatorAddress = delegateMsg.DelegatorAddress,
+                            ValidatorAddress = delegateMsg.ValidatorAddress,
+                            Amount = long.Parse(delegateMsg.Amount.Amount), // adds amount
+                            Denominator = delegateMsg.Amount.Denominator
+                        };
+
+                        await Repository.SaveAsync(delegateEntity, context.CancellationToken);
+                        break;
+                    }
+
+                    case CosmosRedelegateMessage redelegateMsg:
+                    {
+                        if (redelegateMsg.Amount == default)
+                        {
+                            Logger.LogWarning(
+                                "({BlockChain}) Missing amount on redelegate message (txHash={TxHash})",
+                                NameOfBlockChain,
+                                txHash
+                            );
+                            break;
+                        }
+
+                        var entities = new[]
+                        {
+                            // first a ledger entry that removes the delegation
+                            // from the original validator
+                            new TValidatorDelegationLedgerEntity
+                            {
+                                Id = IdProvider.Snowflake(),
+                                TxHash = txHash,
+                                At = rawTransaction.CreatedAt,
+                                DelegatorAddress = redelegateMsg.DelegatorAddress,
+                                ValidatorAddress = redelegateMsg.ValidatorSourceAddress,
+                                Amount = -long.Parse(redelegateMsg.Amount.Amount),  // subtracts amount
+                                Denominator = redelegateMsg.Amount.Denominator
+                            },
+                            // then a ledger entry adding the delegation to the new validator
+                            new TValidatorDelegationLedgerEntity
+                            {
+                                Id = IdProvider.Snowflake(),
+                                TxHash = txHash,
+                                At = rawTransaction.CreatedAt,
+                                DelegatorAddress = redelegateMsg.DelegatorAddress,
+                                ValidatorAddress = redelegateMsg.ValidatorDestinationAddress,
+                                Amount = long.Parse(redelegateMsg.Amount.Amount),  // adds amount
+                                Denominator = redelegateMsg.Amount.Denominator
+                            }
+                        };
+
+                        await Repository.SaveAllAsync(entities, context.CancellationToken);
+                        break;
+                    }
+
+                    default:
+                        // we do not care about this message, so leave it be :)
+                        break;
+                }
+            }
+        }
+    }
+}