Merge pull request #80 from Nexus-Mods/analyzer-support

analyzer-support

src/NexusMods.MnemonicDB.Abstractions/IAnalyzer.cs

@@ -0,0 +1,22 @@
+namespace NexusMods.MnemonicDB.Abstractions;
+
+/// <summary>
+/// Interface for a transaction analyzer. These can be injected via DI and they will then be fed each database transaction
+/// to analyze and produce a result.
+/// </summary>
+public interface IAnalyzer
+{
+    /// <summary>
+    /// Analyze the database and produce a result.
+    /// </summary>
+    public object Analyze(IDb db);
+}
+
+/// <summary>
+/// Typed version of <see cref="IAnalyzer"/> that specifies the type of the result.
+/// </summary>
+public interface IAnalyzer<out T> : IAnalyzer
+{
+
+}
+

src/NexusMods.MnemonicDB.Abstractions/IConnection.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Collections.Generic;
 using NexusMods.MnemonicDB.Abstractions.IndexSegments;
 using NexusMods.MnemonicDB.Abstractions.Internals;
 
@@ -60,4 +61,9 @@ public interface IConnection
     /// </summary>
     /// <returns></returns>
     public ITransaction BeginTransaction();
+
+    /// <summary>
+    /// The analyzers that are available for this connection
+    /// </summary>
+    public IAnalyzer[] Analyzers { get; }
 }

src/NexusMods.MnemonicDB.Abstractions/IDb.cs

@@ -84,4 +84,10 @@ public interface IDb : IEquatable<IDb>
     /// Returns an index segment of all the datoms that are a reference pointing to the given entity id.
     /// </summary>
     IndexSegment ReferencesTo(EntityId eid);
+
+    /// <summary>
+    /// Get the cached data for the given analyzer.
+    /// </summary>
+    TReturn AnalyzerData<TAnalyzer, TReturn>() 
+        where TAnalyzer : IAnalyzer<TReturn>;
 }

src/NexusMods.MnemonicDB/Connection.cs

@@ -27,34 +27,53 @@ public class Connection : IConnection
 
     private BehaviorSubject<IDb> _dbStream;
     private IDisposable? _dbStreamDisposable;
+    private readonly IAnalyzer[] _analyzers;
 
     /// <summary>
     ///     Main connection class, co-ordinates writes and immutable reads
     /// </summary>
-    public Connection(ILogger<Connection> logger, IDatomStore store, IServiceProvider provider, IEnumerable<IAttribute> declaredAttributes)
+    public Connection(ILogger<Connection> logger, IDatomStore store, IServiceProvider provider, IEnumerable<IAttribute> declaredAttributes, IEnumerable<IAnalyzer> analyzers)
     {
         ServiceProvider = provider;
         _logger = logger;
         _declaredAttributes = declaredAttributes.ToDictionary(a => a.Id);
         _store = store;
         _dbStream = new BehaviorSubject<IDb>(default!);
+        _analyzers = analyzers.ToArray();
         Bootstrap();
     }
 
     /// <summary>
     /// Scrubs the transaction stream so that we only ever move forward and never repeat transactions
     /// </summary>
-    private static IObservable<Db> ForwardOnly(IObservable<IDb> dbStream)
+    private IObservable<Db> ProcessUpdate(IObservable<IDb> dbStream)
     {
         TxId? prev = null;
 
         return Observable.Create((IObserver<Db> observer) =>
         {
             return dbStream.Subscribe(nextItem =>
             {
+
                 if (prev != null && prev.Value >= nextItem.BasisTxId)
                     return;
 
+                var db = (Db)nextItem;
+                db.Connection = this;
+
+                foreach (var analyzer in _analyzers)
+                {
+                    try
+                    {
+                        var result = analyzer.Analyze(nextItem);
+                        db.AnalyzerData.Add(analyzer.GetType(), result);
+                    }
+                    catch (Exception ex)
+                    {
+                        _logger.LogError(ex, "Failed to analyze with {Analyzer}", analyzer.GetType().Name);
+                    }
+                }
+
                 observer.OnNext((Db)nextItem);
                 prev = nextItem.BasisTxId;
             }, observer.OnError, observer.OnCompleted);
@@ -105,6 +124,9 @@ public ITransaction BeginTransaction()
         return new Transaction(this, _store.Registry);
     }
 
+    /// <inheritdoc />
+    public IAnalyzer[] Analyzers => _analyzers;
+
     /// <inheritdoc />
     public IObservable<IDb> Revisions
     {
@@ -175,12 +197,7 @@ private void Bootstrap()
         {
             AddMissingAttributes(_declaredAttributes.Values);
 
-            _dbStreamDisposable = ForwardOnly(_store.TxLog)
-                .Select(db =>
-                {
-                    db.Connection = this; 
-                    return db;
-                })
+            _dbStreamDisposable = ProcessUpdate(_store.TxLog)
                 .Subscribe(_dbStream);
         }
         catch (Exception ex)

src/NexusMods.MnemonicDB/Db.cs

@@ -30,6 +30,8 @@ internal class Db : IDb
     public IAttributeRegistry Registry => _registry;
 
     public IndexSegment RecentlyAdded { get; }
+
+    internal Dictionary<Type, object> AnalyzerData { get; } = new();
 
     public Db(ISnapshot snapshot, TxId txId, AttributeRegistry registry)
     {
@@ -107,7 +109,14 @@ public IndexSegment ReferencesTo(EntityId id)
     {
         return _cache.GetReferences(id, this);
     }
-
+
+    TReturn IDb.AnalyzerData<TAnalyzer, TReturn>()
+    {
+        if (AnalyzerData.TryGetValue(typeof(TAnalyzer), out var value))
+            return (TReturn)value;
+        throw new KeyNotFoundException($"Analyzer {typeof(TAnalyzer).Name} not found");
+    }
+
     public IndexSegment Datoms<TValue, TLowLevel>(Attribute<TValue, TLowLevel> attribute, TValue value)
     {
         return Datoms(SliceDescriptor.Create(attribute, value, _registry));

tests/NexusMods.MnemonicDB.Storage.Tests/NullConnection.cs

@@ -19,4 +19,6 @@ public ITransaction BeginTransaction()
     {
         throw new NotSupportedException();
     }
+
+    public IAnalyzer[] Analyzers => throw new NotSupportedException();
 }

tests/NexusMods.MnemonicDB.TestModel/Analyzers/AttributesAnalyzer.cs

@@ -0,0 +1,21 @@
+using NexusMods.MnemonicDB.Abstractions;
+
+namespace NexusMods.MnemonicDB.TestModel.Analyzers;
+
+/// <summary>
+/// Records all the attributes in each transaction
+/// </summary>
+public class AttributesAnalyzer : IAnalyzer<HashSet<IAttribute>>
+{
+    public object Analyze(IDb db)
+    {
+        var hashSet = new HashSet<IAttribute>();
+        var registry = db.Registry;
+        foreach (var datom in db.RecentlyAdded)
+        {
+            hashSet.Add(registry.GetAttribute(datom.A));
+        }
+
+        return hashSet;
+    }
+}

tests/NexusMods.MnemonicDB.TestModel/Analyzers/DatomCountAnalyzer.cs

@@ -0,0 +1,14 @@
+using NexusMods.MnemonicDB.Abstractions;
+
+namespace NexusMods.MnemonicDB.TestModel.Analyzers;
+
+/// <summary>
+/// Counts the number of dataoms in each transaction
+/// </summary>
+public class DatomCountAnalyzer : IAnalyzer<int>
+{
+    public object Analyze(IDb db)
+    {
+        return db.RecentlyAdded.Count;
+    }
+}

tests/NexusMods.MnemonicDB.Tests/AMnemonicDBTest.cs

@@ -7,6 +7,7 @@
 using NexusMods.MnemonicDB.TestModel.Helpers;
 using NexusMods.Hashing.xxHash64;
 using NexusMods.MnemonicDB.TestModel;
+using NexusMods.MnemonicDB.TestModel.Analyzers;
 using NexusMods.Paths;
 using Xunit.Sdk;
 using File = NexusMods.MnemonicDB.TestModel.File;
@@ -23,6 +24,7 @@ public class AMnemonicDBTest : IDisposable
     private DatomStore _store;
     protected IConnection Connection;
     protected ILogger Logger;
+    private readonly IAnalyzer[] _analyzers;
 
 
     protected AMnemonicDBTest(IServiceProvider provider)
@@ -40,7 +42,14 @@ protected AMnemonicDBTest(IServiceProvider provider)
         _backend = new Backend(_registry);
 
         _store = new DatomStore(provider.GetRequiredService<ILogger<DatomStore>>(), _registry, Config, _backend);
-        Connection = new Connection(provider.GetRequiredService<ILogger<Connection>>(), _store, provider, _attributes);
+
+        _analyzers =
+        new IAnalyzer[]{
+            new DatomCountAnalyzer(),
+            new AttributesAnalyzer(),
+        };
+
+        Connection = new Connection(provider.GetRequiredService<ILogger<Connection>>(), _store, provider, _attributes, _analyzers);
 
         Logger = provider.GetRequiredService<ILogger<AMnemonicDBTest>>();
     }
@@ -130,7 +139,7 @@ protected async Task RestartDatomStore()
         _registry = new AttributeRegistry(_attributes);
         _store = new DatomStore(_provider.GetRequiredService<ILogger<DatomStore>>(), _registry, Config, _backend);
 
-        Connection = new Connection(_provider.GetRequiredService<ILogger<Connection>>(), _store, _provider, _attributes);
+        Connection = new Connection(_provider.GetRequiredService<ILogger<Connection>>(), _store, _provider, _attributes, _analyzers);
     }
 
 }

tests/NexusMods.MnemonicDB.Tests/DbTests.cs

@@ -8,6 +8,7 @@
 using NexusMods.MnemonicDB.Abstractions.Query;
 using NexusMods.MnemonicDB.Abstractions.TxFunctions;
 using NexusMods.MnemonicDB.TestModel;
+using NexusMods.MnemonicDB.TestModel.Analyzers;
 using NexusMods.Paths;
 using File = NexusMods.MnemonicDB.TestModel.File;
 
@@ -800,6 +801,33 @@ public async Task CanWriteTupleAttributes()
 
         var avet = Connection.Db.Datoms(SliceDescriptor.Create(File.TuplePath, (EntityId.From(0), ""), (EntityId.MaxValueNoPartition, ""), Connection.Db.Registry));
         await VerifyTable(avet.Resolved());
+    }
+
+    [Fact]
+    public async Task CanGetAnalyzerData()
+    {
+        using var tx = Connection.BeginTransaction();
+
+        var loadout1 = new Loadout.New(tx)
+        {
+            Name = "Test Loadout"
+        };
+
+        var mod = new Mod.New(tx)
+        {
+            Name = "Test Mod",
+            Source = new Uri("http://test.com"),
+            LoadoutId = loadout1 
+        };
+
+        var result = await tx.Commit();
+
+        result.Db.Should().Be(Connection.Db);
+
+        var countData = Connection.Db.AnalyzerData<DatomCountAnalyzer, int>();
+        countData.Should().Be(result.Db.RecentlyAdded.Count);
 
+        var attrs = Connection.Db.AnalyzerData<AttributesAnalyzer, HashSet<IAttribute>>();
+        attrs.Should().NotBeEmpty();
     }
 }