diff --git a/src/VolcanicCore.Api/Endpoints/ConnectionsEndpoints.cs b/src/VolcanicCore.Api/Endpoints/ConnectionsEndpoints.cs index 97d6f34..8ef62ba 100644 --- a/src/VolcanicCore.Api/Endpoints/ConnectionsEndpoints.cs +++ b/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) => - repo.Delete(id) ? Results.NoContent() : Results.NotFound()); + g.MapDelete("/{id}", async (string id, ConnectionRepository repo, ConnectionInstanceRegistry instances) => + { + await instances.InvalidateAsync(id); + return repo.Delete(id) ? Results.NoContent() : Results.NotFound(); + }); } private static JsonElement StripSecrets(string configJson, IReadOnlySet secretFields) diff --git a/src/VolcanicCore.Api/Engine/ConnectionInstanceRegistry.cs b/src/VolcanicCore.Api/Engine/ConnectionInstanceRegistry.cs new file mode 100644 index 0000000..3f9ac09 --- /dev/null +++ b/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 _cache = new(StringComparer.Ordinal); + private readonly SemaphoreSlim _mu = new(1, 1); + + public ConnectionInstanceRegistry(ConnectionRepository repo, ConnectionProviderRegistry providers) + { + _repo = repo; + _providers = providers; + } + + public async Task 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 */ } + } +} diff --git a/src/VolcanicCore.Api/Program.cs b/src/VolcanicCore.Api/Program.cs index 67a57a1..602111b 100644 --- a/src/VolcanicCore.Api/Program.cs +++ b/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(sp => builder.Services.AddSingleton(sp => new ConnectionProviderRegistry(sp.GetServices())); +builder.Services.AddSingleton(); +builder.Services.AddSingleton(sp => + sp.GetRequiredService()); + var app = builder.Build(); // Apply migrations at boot. diff --git a/tests/VolcanicCore.Tests/ConnectionInstanceRegistryTests.cs b/tests/VolcanicCore.Tests/ConnectionInstanceRegistryTests.cs new file mode 100644 index 0000000..4197e76 --- /dev/null +++ b/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 SecretFields => new HashSet(); + public Task OpenAsync(JsonElement config, CancellationToken ct) + { Opens++; return Task.FromResult(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); + } +}