feat(engine): ConnectionInstanceRegistry with per-id caching + invalidate on delete

src/VolcanicCore.Api/Endpoints/ConnectionsEndpoints.cs

@@ -1,6 +1,7 @@
 using System.Text.Json;
 using VolcanicCore.Api.Auth;
 using VolcanicCore.Api.Contracts;
+using VolcanicCore.Api.Engine;
 using VolcanicCore.Persistence.Repositories;
 
 namespace VolcanicCore.Api.Endpoints;
@@ -46,8 +47,11 @@ public static class ConnectionsEndpoints
             }
         });
 
-        g.MapDelete("/{id}", (string id, ConnectionRepository repo) =>
+        g.MapDelete("/{id}", async (string id, ConnectionRepository repo, ConnectionInstanceRegistry instances) =>
-            repo.Delete(id) ? Results.NoContent() : Results.NotFound());
+        {
+            await instances.InvalidateAsync(id);
+            return repo.Delete(id) ? Results.NoContent() : Results.NotFound();
+        });
     }
 
     private static JsonElement StripSecrets(string configJson, IReadOnlySet<string> secretFields)

src/VolcanicCore.Api/Engine/ConnectionInstanceRegistry.cs

@@ -0,0 +1,69 @@
+using System.Text.Json;
+using VolcanicCore.Api.Registries;
+using VolcanicCore.Persistence.Repositories;
+
+namespace VolcanicCore.Api.Engine;
+
+public sealed class ConnectionInstanceRegistry : IConnectionInstanceRegistry, IAsyncDisposable
+{
+    private sealed record Entry(string ProviderId, object Handle);
+
+    private readonly ConnectionRepository _repo;
+    private readonly ConnectionProviderRegistry _providers;
+    private readonly Dictionary<string, Entry> _cache = new(StringComparer.Ordinal);
+    private readonly SemaphoreSlim _mu = new(1, 1);
+
+    public ConnectionInstanceRegistry(ConnectionRepository repo, ConnectionProviderRegistry providers)
+    {
+        _repo = repo;
+        _providers = providers;
+    }
+
+    public async Task<object> GetAsync(string connectionId, CancellationToken ct)
+    {
+        await _mu.WaitAsync(ct);
+        try
+        {
+            if (_cache.TryGetValue(connectionId, out var hit)) return hit.Handle;
+
+            var row = _repo.Get(connectionId)
+                ?? throw new KeyNotFoundException($"Connection {connectionId} not found.");
+            var provider = _providers.Get(row.ProviderId);
+            using var doc = JsonDocument.Parse(row.ConfigJson);
+            var handle = await provider.OpenAsync(doc.RootElement.Clone(), ct);
+            _cache[connectionId] = new Entry(row.ProviderId, handle);
+            return handle;
+        }
+        finally { _mu.Release(); }
+    }
+
+    public async Task InvalidateAsync(string connectionId)
+    {
+        await _mu.WaitAsync();
+        Entry? removed = null;
+        try
+        {
+            if (_cache.Remove(connectionId, out var e)) removed = e;
+        }
+        finally { _mu.Release(); }
+
+        if (removed is not null)
+            await _providers.Get(removed.ProviderId).CloseAsync(removed.Handle);
+    }
+
+    public async ValueTask DisposeAsync()
+    {
+        Entry[] snapshot;
+        await _mu.WaitAsync();
+        try
+        {
+            snapshot = _cache.Values.ToArray();
+            _cache.Clear();
+        }
+        finally { _mu.Release(); }
+
+        foreach (var e in snapshot)
+            try { await _providers.Get(e.ProviderId).CloseAsync(e.Handle); }
+            catch { /* best-effort on shutdown */ }
+    }
+}

src/VolcanicCore.Api/Program.cs

@@ -2,6 +2,7 @@ using Microsoft.AspNetCore.Authentication.Cookies;
 using VolcanicCore.Api;
 using VolcanicCore.Api.Auth;
 using VolcanicCore.Api.Endpoints;
+using VolcanicCore.Api.Engine;
 using VolcanicCore.Api.Registries;
 using VolcanicCore.Core;
 using VolcanicCore.Persistence;
@@ -112,6 +113,10 @@ builder.Services.AddSingleton<NodeRegistry>(sp =>
 builder.Services.AddSingleton<ConnectionProviderRegistry>(sp =>
     new ConnectionProviderRegistry(sp.GetServices<IConnectionProvider>()));
 
+builder.Services.AddSingleton<ConnectionInstanceRegistry>();
+builder.Services.AddSingleton<IConnectionInstanceRegistry>(sp =>
+    sp.GetRequiredService<ConnectionInstanceRegistry>());
+
 var app = builder.Build();
 
 // Apply migrations at boot.

tests/VolcanicCore.Tests/ConnectionInstanceRegistryTests.cs

@@ -0,0 +1,85 @@
+using System.Text.Json;
+using VolcanicCore.Api.Engine;
+using VolcanicCore.Api.Registries;
+using VolcanicCore.Core;
+using VolcanicCore.Persistence;
+using VolcanicCore.Persistence.Encryption;
+using VolcanicCore.Persistence.Migrations;
+using VolcanicCore.Persistence.Repositories;
+using System.Security.Cryptography;
+
+namespace VolcanicCore.Tests;
+
+public class ConnectionInstanceRegistryTests
+{
+    private sealed class Handle : IAsyncDisposable
+    {
+        public bool Disposed;
+        public ValueTask DisposeAsync() { Disposed = true; return default; }
+    }
+
+    private sealed class FakeProvider : IConnectionProvider
+    {
+        public int Opens, Closes;
+        public string Id => "fake";
+        public string DisplayName => "Fake";
+        public JsonSchema ConfigSchema => new(JsonDocument.Parse("{}").RootElement);
+        public IReadOnlySet<string> SecretFields => new HashSet<string>();
+        public Task<object> OpenAsync(JsonElement config, CancellationToken ct)
+        { Opens++; return Task.FromResult<object>(new Handle()); }
+        public Task CloseAsync(object h) { Closes++; return ((IAsyncDisposable)h).DisposeAsync().AsTask(); }
+    }
+
+    private static (ConnectionRepository c, FakeProvider p, ConnectionInstanceRegistry r) Setup()
+    {
+        var path = Path.Combine(Path.GetTempPath(), $"vc-{Guid.NewGuid():N}.db");
+        var f = new SqliteConnectionFactory(path);
+        using (var cn = f.Open())
+            new Migrator(new IMigration[]
+            {
+                new Migration001Initial(), new Migration002Connections(),
+            }).Apply(cn);
+        var key = new byte[32]; RandomNumberGenerator.Fill(key);
+        var enc = new EncryptionService(key);
+        var repo = new ConnectionRepository(f, enc);
+        var provider = new FakeProvider();
+        var providers = new ConnectionProviderRegistry(new IConnectionProvider[] { provider });
+        return (repo, provider, new ConnectionInstanceRegistry(repo, providers));
+    }
+
+    [Fact]
+    public async Task Opens_once_and_returns_cached_handle_on_subsequent_calls()
+    {
+        var (c, p, r) = Setup();
+        var id = c.Create("fake", "n1", "{\"x\":1}");
+        var a = await r.GetAsync(id, CancellationToken.None);
+        var b = await r.GetAsync(id, CancellationToken.None);
+        Assert.Same(a, b);
+        Assert.Equal(1, p.Opens);
+    }
+
+    [Fact]
+    public async Task Invalidate_disposes_current_handle_and_next_get_reopens()
+    {
+        var (c, p, r) = Setup();
+        var id = c.Create("fake", "n1", "{}");
+        var a = await r.GetAsync(id, CancellationToken.None);
+        await r.InvalidateAsync(id);
+        Assert.Equal(1, p.Closes);
+        var b = await r.GetAsync(id, CancellationToken.None);
+        Assert.NotSame(a, b);
+        Assert.Equal(2, p.Opens);
+    }
+
+    [Fact]
+    public async Task DisposeAsync_closes_all_open_handles()
+    {
+        var (c, p, r) = Setup();
+        var id1 = c.Create("fake", "n1", "{}");
+        var id2 = c.Create("fake", "n2", "{}");
+        await r.GetAsync(id1, CancellationToken.None);
+        await r.GetAsync(id2, CancellationToken.None);
+        await r.DisposeAsync();
+        Assert.Equal(2, p.Closes);
+    }
+}