Implementation smart contract aggregate

backend/Application/Aggregates/Contract/BackgroundServices/ContractJobsBackgroundService.cs

@@ -0,0 +1,96 @@
+using System.Threading;
+using System.Threading.Tasks;
+using Application.Aggregates.Contract.Entities;
+using Application.Aggregates.Contract.Jobs;
+using Application.Api.GraphQL.EfCore;
+using Application.Common.FeatureFlags;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.Extensions.Hosting;
+
+namespace Application.Aggregates.Contract.BackgroundServices;
+
+/// <summary>
+/// Background service which executes background jobs related to Contracts.
+///
+/// When new jobs are added they should be dependency injected and added to the constructor of this service.
+/// </summary>
+internal sealed class ContractJobsBackgroundService : BackgroundService
+{
+    private readonly IContractJobFinder _jobFinder;
+    private readonly IFeatureFlags _featureFlags;
+    private readonly IDbContextFactory<GraphQlDbContext> _dbContextFactory;
+    private readonly ILogger _logger;
+
+    public ContractJobsBackgroundService(
+        IContractJobFinder jobFinder,
+        IFeatureFlags featureFlags, 
+        IDbContextFactory<GraphQlDbContext> dbContextFactory
+    )
+    {
+        _jobFinder = jobFinder;
+        _featureFlags = featureFlags;
+        _dbContextFactory = dbContextFactory;
+        _logger = Log.ForContext<ContractJobsBackgroundService>();
+    }
+
+    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+    {
+        if (!_featureFlags.ConcordiumNodeImportEnabled)
+        {
+            _logger.Information("Import data from Concordium node is disabled. This controller will not run!");
+            return;
+        }
+
+        var jobs = _jobFinder.GetJobs();
+
+        try
+        {
+            await Task.WhenAll(jobs.Select(j => RunJob(j, stoppingToken)));
+
+            _logger.Information($"{nameof(ContractJobsBackgroundService)} done.");
+        }
+        catch (Exception e)
+        {
+            _logger.Error(e, $"{nameof(ContractJobsBackgroundService)} didn't succeed successfully due to exception.");
+        }
+    }
+
+    private async Task RunJob(IContractJob job, CancellationToken token)
+    {
+        try
+        {
+            if (await DoesExistingJobExist(job, token))
+            {
+                return;
+            }
+
+            await job.StartImport(token);
+
+            await SaveSuccessfullyExecutedJob(job, token);
+            _logger.Information($"{job.GetUniqueIdentifier()} finished successfully.");
+        }
+        catch (Exception e)
+        {
+            _logger.Error(e, $"{job.GetUniqueIdentifier()} didn't succeed successfully due to exception.");
+            throw;
+        }
+    }
+
+    internal async Task<bool> DoesExistingJobExist(IContractJob job, CancellationToken token = default)
+    {
+        await using var context = await _dbContextFactory.CreateDbContextAsync(token);
+        var existingJob = await context.ContractJobs
+            .AsNoTracking()
+            .Where(j => j.Job == job.GetUniqueIdentifier())
+            .FirstOrDefaultAsync(token);
+
+        return existingJob != null;
+    }
+
+    internal async Task SaveSuccessfullyExecutedJob(IContractJob job, CancellationToken token = default)
+    {
+        await using var context = await _dbContextFactory.CreateDbContextAsync(token);
+        await context.ContractJobs.AddAsync(new ContractJob(job.GetUniqueIdentifier()), token);
+        await context.SaveChangesAsync(token);
+    }
+}

backend/Application/Aggregates/Contract/BackgroundServices/ContractNodeImportBackgroundService.cs

@@ -0,0 +1,100 @@
+using System.Threading;
+using System.Threading.Tasks;
+using Application.Aggregates.Contract.Configurations;
+using Application.Aggregates.Contract.Jobs;
+using Application.Api.GraphQL.EfCore;
+using Application.Common.FeatureFlags;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Options;
+
+namespace Application.Aggregates.Contract.BackgroundServices;
+
+/// <summary>
+/// Background service which starts contract data import data from nodes.
+/// </summary>
+internal class ContractNodeImportBackgroundService : BackgroundService
+{
+    private readonly IContractJobFinder _jobFinder;
+    private readonly IDbContextFactory<GraphQlDbContext> _dbContextFactory;
+    private readonly IContractRepositoryFactory _repositoryFactory;
+    private readonly IContractNodeClient _client;
+    private readonly IFeatureFlags _featureFlags;
+    private readonly ContractAggregateOptions _options;
+    private readonly ILogger _logger;
+
+    public ContractNodeImportBackgroundService(
+        IContractJobFinder jobFinder,
+        IDbContextFactory<GraphQlDbContext> dbContextFactory,
+        IContractRepositoryFactory repositoryFactory,
+        IContractNodeClient client,
+        IFeatureFlags featureFlags,
+        IOptions<ContractAggregateOptions> options)
+    {
+        _jobFinder = jobFinder;
+        _dbContextFactory = dbContextFactory;
+        _repositoryFactory = repositoryFactory;
+        _client = client;
+        _featureFlags = featureFlags;
+        _options = options.Value;
+        _logger = Log.ForContext<ContractNodeImportBackgroundService>();
+    }
+
+    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+    {
+        if (!_featureFlags.ConcordiumNodeImportEnabled)
+        {
+            _logger.Information("Import data from Concordium node is disabled. This controller will not run!");
+            return;
+        }
+
+        try
+        {
+            await AwaitJobsAsync(stoppingToken);
+
+            var contractAggregate = new ContractAggregate(_repositoryFactory, _options);
+
+            _logger.Information($"{nameof(ContractNodeImportBackgroundService)} started.");
+            await contractAggregate.NodeImportJob(_client, stoppingToken);
+        }
+        catch (Exception e)
+        {
+            _logger.Fatal(e, $"{nameof(ContractNodeImportBackgroundService)} stopped due to exception.");
+        }
+    }
+
+    private async Task AwaitJobsAsync(CancellationToken token = default)
+    {
+        while (!token.IsCancellationRequested)
+        {
+            var jobsToAwait = await GetJobsToAwait(token);
+            if (jobsToAwait.Count == 0)
+            {
+                break;
+            }
+
+            foreach (var job in jobsToAwait)
+            {
+                _logger.Information($"Awaiting job {job}");
+            }
+
+            await Task.Delay(_options.JobDelay, token);
+        }
+    }
+
+    internal async Task<IList<string>> GetJobsToAwait(CancellationToken token = default)
+    {
+        var contractJobs = _jobFinder.GetJobs()
+            .Select(j => j.GetUniqueIdentifier())
+            .ToList();
+        await using var context = await _dbContextFactory.CreateDbContextAsync(token);
+        var doneJobs = await context
+            .ContractJobs
+            .AsNoTracking()
+            .Where(j => contractJobs.Contains(j.Job))
+            .Select(j => j.Job)
+            .ToListAsync(cancellationToken: token);
+
+        return contractJobs.Except(doneJobs).ToList();
+    }
+}

backend/Application/Aggregates/Contract/Configurations/ContractAggregateJobOptions.cs

@@ -0,0 +1,15 @@
+namespace Application.Aggregates.Contract.Configurations;
+
+public class ContractAggregateJobOptions
+{
+    /// <summary>
+    /// Number of tasks which should be used for parallelism. 
+    /// </summary>
+    public int NumberOfTask { get; set; } = 5;
+    /// <summary>
+    /// Each task when processing will load multiple blocks and transaction to avoid databases round trips.
+    ///
+    /// Increasing batch size will increase memory consumption on job since more will be loaded into memory.
+    /// </summary>
+    public int BatchSize { get; set; } = 10_000;
+}

backend/Application/Aggregates/Contract/Configurations/ContractAggregateOptions.cs

@@ -0,0 +1,25 @@
+namespace Application.Aggregates.Contract.Configurations;
+
+public class ContractAggregateOptions
+{
+    /// <summary>
+    /// Set options for jobs related to contracts.
+    ///
+    /// Done as dictionary such that it can be changed from configurations. Key is unique identifier of job and
+    /// it defined within the jobs class.
+    /// </summary>
+    public IDictionary<string, ContractAggregateJobOptions> Jobs { get; set; } =
+        new Dictionary<string, ContractAggregateJobOptions>();
+    /// <summary>
+    /// Delay which is used by the node importer between validation if all jobs has succeeded.
+    /// </summary>
+    public TimeSpan JobDelay { get; set; } = TimeSpan.FromSeconds(10);
+    /// <summary>
+    /// Delay between retries in resilience policies.
+    /// </summary>
+    public TimeSpan DelayBetweenRetries { get; set; } = TimeSpan.FromSeconds(3);
+    /// <summary>
+    /// Number of times to retry.
+    /// </summary>
+    public uint RetryCount { get; set; } = 5;
+}

backend/Application/Aggregates/Contract/Configurations/ContractEntityTypeConfigurations.cs

@@ -0,0 +1,40 @@
+using Application.Api.GraphQL.EfCore.Converters.EfCore;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.EntityFrameworkCore.Metadata.Builders;
+
+namespace Application.Aggregates.Contract.Configurations;
+
+public sealed class ContractEntityTypeConfigurations : IEntityTypeConfiguration<Entities.Contract>
+{
+    public void Configure(EntityTypeBuilder<Entities.Contract> builder)
+    {
+        builder.ToTable("graphql_contracts");
+        builder.HasKey(x => new
+        {
+            x.BlockHeight, 
+            x.TransactionIndex,
+            x.EventIndex,
+            x.ContractAddressIndex,
+            x.ContractAddressSubIndex
+        });
+        builder.Property(x => x.BlockHeight)
+            .HasColumnName("block_height");
+        builder.Property(x => x.TransactionHash)
+            .HasColumnName("transaction_hash");
+        builder.Property(x => x.TransactionIndex)
+            .HasColumnName("transaction_index");
+        builder.Property(x => x.EventIndex)
+            .HasColumnName("event_index");
+        builder.Property(x => x.ContractAddressIndex)
+            .HasColumnName("contract_address_index");
+        builder.Property(x => x.ContractAddressSubIndex)
+            .HasColumnName("contract_address_sub_index");
+        builder.Property(x => x.Creator)
+            .HasColumnName("creator")
+            .HasConversion<AccountAddressConverter>();
+        builder.Property(x => x.Source)
+            .HasColumnName("source");
+        builder.Property(x => x.CreatedAt)
+            .HasColumnName("created_at");        
+    }
+}

backend/Application/Aggregates/Contract/Configurations/ContractEventEntityTypeConfigurations.cs

@@ -0,0 +1,42 @@
+using Application.Aggregates.Contract.Entities;
+using Application.Api.GraphQL.EfCore.Converters.EfCore;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.EntityFrameworkCore.Metadata.Builders;
+
+namespace Application.Aggregates.Contract.Configurations;
+
+public sealed class ContractEventEntityTypeConfigurations : IEntityTypeConfiguration<ContractEvent>
+{
+    public void Configure(EntityTypeBuilder<ContractEvent> builder)
+    {
+        builder.ToTable("graphql_contract_events");
+        builder.HasKey(x => new
+        {
+            x.BlockHeight, 
+            x.TransactionIndex,
+            x.EventIndex,
+            x.ContractAddressIndex,
+            x.ContractAddressSubIndex
+        });
+        builder.Property(x => x.BlockHeight)
+            .HasColumnName("block_height");
+        builder.Property(x => x.TransactionHash)
+            .HasColumnName("transaction_hash");
+        builder.Property(x => x.TransactionIndex)
+            .HasColumnName("transaction_index");
+        builder.Property(x => x.EventIndex)
+            .HasColumnName("event_index");
+        builder.Property(x => x.ContractAddressIndex)
+            .HasColumnName("contract_address_index");
+        builder.Property(x => x.ContractAddressSubIndex)
+            .HasColumnName("contract_address_sub_index");
+        builder.Property(x => x.Event)
+            .HasColumnName("event")
+            .HasColumnType("json")
+            .HasConversion<TransactionResultEventToJsonConverter>();
+        builder.Property(x => x.Source)
+            .HasColumnName("source");
+        builder.Property(x => x.CreatedAt)
+            .HasColumnName("created_at");        
+    }
+}

backend/Application/Aggregates/Contract/Configurations/ContractJobEntityTypeConfigurations.cs

@@ -0,0 +1,18 @@
+using Application.Aggregates.Contract.Entities;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.EntityFrameworkCore.Metadata.Builders;
+
+namespace Application.Aggregates.Contract.Configurations;
+
+public sealed class ContractJobEntityTypeConfigurations : IEntityTypeConfiguration<ContractJob>
+{
+    public void Configure(EntityTypeBuilder<ContractJob> builder)
+    {
+        builder.ToTable("graphql_contract_jobs");
+        builder.HasKey(x => x.Job);
+        builder.Property(x => x.Job)
+            .HasColumnName("job");
+        builder.Property(x => x.CreatedAt)
+            .HasColumnName("created_at");
+    }
+}

backend/Application/Aggregates/Contract/Configurations/ContractReadHeightEntityTypeConfigurations.cs

@@ -0,0 +1,22 @@
+using Application.Aggregates.Contract.Entities;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.EntityFrameworkCore.Metadata.Builders;
+
+namespace Application.Aggregates.Contract.Configurations;
+
+public sealed class ContractReadHeightEntityTypeConfigurations : IEntityTypeConfiguration<
+    ContractReadHeight>
+{
+    public void Configure(EntityTypeBuilder<ContractReadHeight> builder)
+    {
+        builder.ToTable("graphql_contract_read_heights");
+        builder.HasKey(x => x.Id);
+        builder.Property(x => x.Id).HasColumnName("id").ValueGeneratedOnAdd();
+        builder.Property(x => x.BlockHeight)
+            .HasColumnName("block_height");
+        builder.Property(x => x.Source)
+            .HasColumnName("source");
+        builder.Property(x => x.CreatedAt)
+            .HasColumnName("created_at");
+    }
+}