using Elsa.Caching.Features;
using Elsa.Caching.Models;
using Elsa.Caching.Redis.Options;
using Elsa.EntityFrameworkCore.Modules.Management;
using Elsa.Extensions;
using Elsa.Http.Services;
using Elsa.Workflows.Management.Entities;
using Elsa.Workflows.Runtime.Entities;
using Elsa.Common.Models;
using Microsoft.Extensions.Caching.Memory;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.EntityFrameworkCore;
using Xunit;

namespace Elsa.Caching.UnitTests;

public class RedisFactAttribute : FactAttribute
{
    public RedisFactAttribute()
    {
        if (Environment.GetEnvironmentVariable("REDIS_TEST_CONNECTION") == null)
            Skip = "Set REDIS_TEST_CONNECTION to run integration tests against an isolated Redis server.";
    }
}

public class CacheProviderTests
{
    private static CacheEntryOptions Lifetime => new() { AbsoluteExpirationRelativeToNow = TimeSpan.FromMinutes(1) };
    private static IConfigurationSection Configuration(string provider) => new ConfigurationBuilder()
        .AddInMemoryCollection(new Dictionary<string, string?>
        {
            ["Workflow:Cache:Provider"] = provider,
            ["Workflow:Cache:Duration"] = "00:03:00",
            ["Workflow:Cache:Redis:Database"] = "2",
            ["Workflow:Cache:Redis:KeyPrefix"] = "configured:"
        }).Build().GetSection("Workflow:Cache");

    [Fact]
    public async Task Default_memory_caches_and_keeps_local_objects()
    {
        using var host = CacheTestHost.Create();
        var cache = host.GetRequiredService<ICacheManager>();
        var calls = 0;
        Assert.Equal(1, await cache.GetOrCreateAsync("a", () => Task.FromResult(++calls), Lifetime));
        Assert.Equal(1, await cache.GetOrCreateAsync("a", () => Task.FromResult(++calls), Lifetime));
        var local = host.GetRequiredService<IMemoryCache>();
        var runner = new Func<int>(() => 3);
        local.Set("compiled-script", runner);
        await cache.TriggerTokenAsync("definition");
        Assert.Same(runner, local.Get<Func<int>>("compiled-script"));
        Assert.Equal(2, await cache.GetOrCreateAsync("a", () => Task.FromResult(++calls), Lifetime));
    }

    [Fact]
    public void Configuration_is_overridden_only_by_explicit_fluent_values()
    {
        using var host = CacheTestHost.Create(configure: cache =>
        {
            cache.UseConfiguration(Configuration("Memory"));
            cache.UseRedis(redis => redis.KeyPrefix = "fluent:");
        });
        var options = host.GetRequiredService<RedisCacheOptions>();
        Assert.Equal(2, options.Database);
        Assert.Equal("fluent:", options.KeyPrefix);
        Assert.Equal(TimeSpan.FromMinutes(3), host.GetRequiredService<CacheFeature>().Duration);
        Assert.True(host.GetRequiredService<ICacheProvider>().IsShared);
    }

    [Fact]
    public void Fluent_memory_can_override_configured_redis()
    {
        using var host = CacheTestHost.Create(configure: cache => { cache.UseConfiguration(Configuration("Redis")); cache.UseMemory(); });
        Assert.False(host.GetRequiredService<ICacheProvider>().IsShared);
    }

    [Fact]
    public void Explicit_connection_factory_overrides_configured_connection_string()
    {
        Func<IServiceProvider, StackExchange.Redis.IConnectionMultiplexer> factory = _ => throw new Exception("Not used during registration");
        using var host = CacheTestHost.Create(configure: c => c.UseRedis(o =>
        {
            o.ConnectionString = "configured-host:6379";
            o.ConnectionFactory = factory;
        }));
        Assert.Same(factory, host.GetRequiredService<RedisCacheOptions>().ConnectionFactory);
    }

    [Fact]
    public void Unknown_or_conflicting_providers_fail_clearly()
    {
        Assert.Throws<InvalidOperationException>(() => CacheTestHost.Create(configure: c => c.UseConfiguration(Configuration("Missing"))));
        Assert.Throws<InvalidOperationException>(() => CacheTestHost.Create(configure: c => { c.UseMemory(); c.UseRedis(); }));
    }

    [Fact]
    public async Task Shared_provider_requires_durable_consistency()
    {
        using var host = CacheTestHost.Create(configure: c => c.UseRedis());
        await Assert.ThrowsAsync<InvalidOperationException>(() => host.GetRequiredService<ICacheManager>()
            .GetOrCreateAsync("a", () => Task.FromResult(1), Lifetime));
    }

    [Fact]
    public async Task Disabled_cache_always_calls_factory()
    {
        using var host = CacheTestHost.Create(configure: c => c.Enabled = false);
        var cache = host.GetRequiredService<ICacheManager>();
        var calls = 0;
        await cache.GetOrCreateAsync("a", () => Task.FromResult(++calls), Lifetime);
        await cache.GetOrCreateAsync("a", () => Task.FromResult(++calls), Lifetime);
        Assert.Equal(2, calls);
    }

    [Fact]
    public async Task Filling_old_data_during_a_mutation_cannot_resurrect_it()
    {
        using var host = CacheTestHost.Create();
        var cache = host.GetRequiredService<ICacheManager>();
        var captured = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
        var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
        var fill = cache.GetOrCreateAsync("definition", async () => { captured.SetResult(); await release.Task; return "old"; }, Lifetime);
        await captured.Task;
        var mutation = await cache.BeginMutationAsync();
        await cache.CompleteMutationAsync(mutation);
        release.SetResult();
        Assert.Equal("old", await fill); // An already-running request may finish with its captured version.
        Assert.Equal("new", await cache.GetOrCreateAsync("definition", () => Task.FromResult("new"), Lifetime));
    }

    [Fact]
    public void Route_snapshot_replacement_removes_deleted_routes()
    {
        using var host = CacheTestHost.Create();
        var table = ActivatorUtilities.CreateInstance<RouteTable>(host);
        table.Add("/old");
        table.Replace(new[] { new Elsa.Http.HttpRouteData("/new") });
        Assert.Equal("/new", Assert.Single(table).Route);
    }

    [Fact]
    public void Sql_server_migration_generates_the_durable_fence_table()
    {
        var services = new ServiceCollection();
        services.AddDbContextFactory<ManagementElsaDbContext>(options => options.UseSqlServer(
            "Server=(local);Database=workflow-cache-script;Integrated Security=true;TrustServerCertificate=true",
            sql => sql.MigrationsAssembly(typeof(Elsa.EntityFrameworkCore.SqlServer.Migrations.Management.AddCacheMutationFences).Assembly.FullName)));
        using var host = services.BuildServiceProvider();
        using var context = host.GetRequiredService<IDbContextFactory<ManagementElsaDbContext>>().CreateDbContext();
        var migrator = ((Microsoft.EntityFrameworkCore.Infrastructure.IInfrastructure<IServiceProvider>)context.Database).Instance.GetRequiredService<Microsoft.EntityFrameworkCore.Migrations.IMigrator>();
        var script = migrator.GenerateScript("20250818170340_AddExternalUserId", "20261007000000_AddCacheMutationFences");
        Assert.Contains("CREATE TABLE [Elsa].[CacheMutations]", script);
        Assert.Contains("[Scope] nvarchar(32)", script);
        Assert.Contains("IDENTITY", script);
    }

    [RedisFact]
    public async Task Nodes_share_payload_and_cached_null_is_invalidated()
    {
        var path = Path.Combine(Path.GetTempPath(), "elsa-cache-" + Guid.NewGuid().ToString("N") + ".db");
        var prefix = "workflow-tests:" + Guid.NewGuid().ToString("N") + ":";
        using var first = CacheTestHost.Create(path, true, prefix);
        using var second = CacheTestHost.Create(path, true, prefix);
        await CacheTestHost.EnsureCreatedAsync(first);
        var a = first.GetRequiredService<ICacheManager>();
        var b = second.GetRequiredService<ICacheManager>();
        Assert.Equal("node-one", await a.GetOrCreateAsync("definition", () => Task.FromResult("node-one"), Lifetime));
        Assert.Equal("node-one", await b.GetOrCreateAsync<string>("definition", () => throw new Exception("Cache miss on second node"), Lifetime));
        Assert.Null(await a.GetOrCreateAsync<string?>("missing", () => Task.FromResult<string?>(null), Lifetime));
        Assert.Null(await b.GetOrCreateAsync<string?>("missing", () => throw new Exception("Cached null was lost"), Lifetime));
        var mutation = await a.BeginMutationAsync();
        await a.CompleteMutationAsync(mutation);
        Assert.Equal("created", await b.GetOrCreateAsync("missing", () => Task.FromResult("created"), Lifetime));
    }

    [RedisFact]
    public async Task Unfinished_durable_fence_bypasses_cache_after_node_restart()
    {
        var path = Path.Combine(Path.GetTempPath(), "elsa-cache-" + Guid.NewGuid().ToString("N") + ".db");
        var prefix = "workflow-tests:" + Guid.NewGuid().ToString("N") + ":";
        using (var first = CacheTestHost.Create(path, true, prefix))
        {
            await CacheTestHost.EnsureCreatedAsync(first);
            var cache = first.GetRequiredService<ICacheManager>();
            await cache.GetOrCreateAsync("definition", () => Task.FromResult("old"), Lifetime);
            await cache.BeginMutationAsync(); // Simulates process death before completion.
        }
        using var restarted = CacheTestHost.Create(path, true, prefix);
        var fresh = restarted.GetRequiredService<ICacheManager>();
        Assert.Equal("sql-current", await fresh.GetOrCreateAsync("definition", () => Task.FromResult("sql-current"), Lifetime));
        Assert.Equal("sql-next", await fresh.GetOrCreateAsync("definition", () => Task.FromResult("sql-next"), Lifetime));
        Assert.True((await fresh.GetConsistencyStateAsync()).HasPendingMutations);
    }

    [RedisFact]
    public async Task Redis_roundtrip_preserves_definition_and_polymorphic_trigger_payload()
    {
        var path = Path.Combine(Path.GetTempPath(), "elsa-cache-" + Guid.NewGuid().ToString("N") + ".db");
        using var host = CacheTestHost.Create(path, true);
        await CacheTestHost.EnsureCreatedAsync(host);
        var cache = host.GetRequiredService<ICacheManager>();
        var definition = new WorkflowDefinition { Id = "v1", DefinitionId = "wf", MaterializerName = "Json", StringData = "{}", Version = 1 };
        definition.Variables.Add(new Elsa.Workflows.Memory.Variable<string>("message", "hello"));
        definition.CustomProperties["payload"] = new Elsa.Http.Bookmarks.HttpEndpointBookmarkPayload("/custom", "GET");
        await cache.GetOrCreateAsync("definition", () => Task.FromResult(definition), Lifetime);
        var roundtrip = await cache.GetOrCreateAsync<WorkflowDefinition>("definition", () => throw new Exception("Miss"), Lifetime);
        Assert.Equal("v1", roundtrip!.Id);
        Assert.Equal("hello", Assert.Single(roundtrip.Variables).Value);
        Assert.IsType<Elsa.Http.Bookmarks.HttpEndpointBookmarkPayload>(roundtrip.CustomProperties["payload"]);
        var trigger = new StoredTrigger { Id = "t", WorkflowDefinitionId = "wf", WorkflowDefinitionVersionId = "v1", ActivityId = "a", Payload = new Elsa.Http.Bookmarks.HttpEndpointBookmarkPayload("/orders", "POST") };
        var page = new Page<StoredTrigger>(new[] { trigger }, 1);
        await cache.GetOrCreateAsync("triggers", () => Task.FromResult(page), Lifetime);
        var result = await cache.GetOrCreateAsync<Page<StoredTrigger>>("triggers", () => throw new Exception("Miss"), Lifetime);
        var payload = Assert.IsType<Elsa.Http.Bookmarks.HttpEndpointBookmarkPayload>(Assert.Single(result!.Items).Payload);
        Assert.Equal("/orders", payload.Path);
    }

    [RedisFact]
    public async Task Disconnected_client_bypasses_cache_and_cannot_revive_old_data_on_reconnect()
    {
        var path = Path.Combine(Path.GetTempPath(), "elsa-cache-" + Guid.NewGuid().ToString("N") + ".db");
        using var host = CacheTestHost.Create(path, true);
        await CacheTestHost.EnsureCreatedAsync(host);
        var cache = host.GetRequiredService<ICacheManager>();
        var settings = host.GetRequiredService<RedisCacheOptions>();
        var connected = host.GetRequiredService<StackExchange.Redis.IConnectionMultiplexer>();
        using var disconnected = StackExchange.Redis.ConnectionMultiplexer.Connect("127.0.0.1:16380,abortConnect=false,connectTimeout=100,syncTimeout=100,asyncTimeout=100,connectRetry=0");
        await cache.GetOrCreateAsync("definition", () => Task.FromResult("old"), Lifetime);
        settings.ConnectionFactory = _ => disconnected;
        var mutation = await cache.BeginMutationAsync();
        await cache.CompleteMutationAsync(mutation);
        Assert.Equal("sql-current", await cache.GetOrCreateAsync("definition", () => Task.FromResult("sql-current"), Lifetime));
        settings.ConnectionFactory = _ => connected;
        Assert.Equal("sql-current", await cache.GetOrCreateAsync("definition", () => Task.FromResult("sql-current"), Lifetime));
    }

    [RedisFact]
    public async Task Route_only_mutation_keeps_definition_payload_valid()
    {
        var path = Path.Combine(Path.GetTempPath(), "elsa-cache-" + Guid.NewGuid().ToString("N") + ".db");
        using var host = CacheTestHost.Create(path, true);
        await CacheTestHost.EnsureCreatedAsync(host);
        var cache = host.GetRequiredService<ICacheManager>();
        await cache.GetOrCreateAsync("definition", () => Task.FromResult("unchanged"), Lifetime);
        var before = await cache.GetConsistencyStateAsync();
        var mutation = await cache.BeginRouteMutationAsync();
        await cache.CompleteMutationAsync(mutation);
        var after = await cache.GetConsistencyStateAsync();
        Assert.Equal(before.Revision, after.Revision);
        Assert.NotEqual(before.RouteRevision, after.RouteRevision);
        Assert.Equal("unchanged", await cache.GetOrCreateAsync<string>("definition", () => throw new Exception("Bookmark changes invalidated the definition"), Lifetime));
    }

    [RedisFact]
    public async Task Actual_management_features_share_saved_definitions_and_refresh_after_update()
    {
        var path = Path.Combine(Path.GetTempPath(), "elsa-module-" + Guid.NewGuid().ToString("N") + ".db");
        var prefix = "workflow-module-tests:" + Guid.NewGuid().ToString("N") + ":";
        using var first = CreateManagementHost(path, prefix);
        using var second = CreateManagementHost(path, prefix);
        await CacheTestHost.EnsureCreatedAsync(first);
        using var firstScope = first.CreateScope();
        using var secondScope = second.CreateScope();
        var writer = firstScope.ServiceProvider.GetRequiredService<Elsa.Workflows.Management.IWorkflowDefinitionStore>();
        var reader = secondScope.ServiceProvider.GetRequiredService<Elsa.Workflows.Management.IWorkflowDefinitionStore>();
        var definition = new WorkflowDefinition { Id = "version", DefinitionId = "definition", Version = 1, Name = "original", MaterializerName = "Json", StringData = "{}" };
        await writer.SaveAsync(definition);
        var filter = new Elsa.Workflows.Management.Filters.WorkflowDefinitionFilter { Id = "version" };
        Assert.Equal("original", (await reader.FindAsync(filter, new Elsa.Workflows.Management.Filters.WorkflowDefinitionOrder<string>(x => x.Id, Elsa.Common.Entities.OrderDirection.Ascending)))!.Name);
        Assert.Equal("original", (await reader.FindAsync(filter, new Elsa.Workflows.Management.Filters.WorkflowDefinitionOrder<string>(x => x.Id, Elsa.Common.Entities.OrderDirection.Ascending)))!.Name);
        definition.Name = "updated";
        await writer.SaveAsync(definition);
        Assert.Equal("updated", (await reader.FindAsync(filter, new Elsa.Workflows.Management.Filters.WorkflowDefinitionOrder<string>(x => x.Id, Elsa.Common.Entities.OrderDirection.Ascending)))!.Name);
        Assert.True(second.GetRequiredService<ICacheManager>().IsShared);
        Assert.NotNull(second.GetRequiredService<IMemoryCache>());
    }

    private static ServiceProvider CreateManagementHost(string path, string prefix)
    {
        var services = new ServiceCollection();
        services.AddLogging();
        services.AddSingleton<StackExchange.Redis.IConnectionMultiplexer>(_ => StackExchange.Redis.ConnectionMultiplexer.Connect(Environment.GetEnvironmentVariable("REDIS_TEST_CONNECTION")!));
        var module = new Elsa.Features.Implementations.Module(services);
        module.UseCaching(cache => cache.UseRedis(redis => { redis.Database = 1; redis.KeyPrefix = prefix; }));
        module.UseWorkflowManagement(management =>
        {
            Elsa.EntityFrameworkCore.Modules.Management.WorkflowManagementFeatureExtensions.UseEntityFrameworkCore(management, ef =>
            {
                ef.RunMigrations = false;
                ef.DbContextOptionsBuilder = (_, options) => options.UseSqlite($"Data Source={path}");
            });
            management.UseCache();
        });
        module.Apply();
        return services.BuildServiceProvider(new ServiceProviderOptions { ValidateScopes = true });
    }

    [Theory]
    [InlineData(true)]
    [InlineData(false)]
    public async Task Ambient_transaction_bypasses_cache_and_keeps_fence_until_commit_or_rollback(bool commit)
    {
        using var host = CacheTestHost.Create();
        var cache = host.GetRequiredService<ICacheManager>();
        await cache.GetOrCreateAsync("definition", () => Task.FromResult("before"), Lifetime);
        using (var transaction = new System.Transactions.TransactionScope(System.Transactions.TransactionScopeAsyncFlowOption.Enabled))
        {
            var mutation = await cache.BeginMutationAsync();
            await cache.CompleteMutationAsync(mutation);
            Assert.True((await cache.GetConsistencyStateAsync()).HasPendingMutations);
            Assert.Equal("transaction-view", await cache.GetOrCreateAsync("definition", () => Task.FromResult("transaction-view"), Lifetime));
            if (commit) transaction.Complete();
        }
        Assert.False((await cache.GetConsistencyStateAsync()).HasPendingMutations);
        Assert.Equal("after", await cache.GetOrCreateAsync("definition", () => Task.FromResult("after"), Lifetime));
    }

    [RedisFact]
    public async Task Completion_invalidates_a_cached_revision_even_if_begin_committed_out_of_identity_order()
    {
        var path = Path.Combine(Path.GetTempPath(), "elsa-late-fence-" + Guid.NewGuid().ToString("N") + ".db");
        using var host = CacheTestHost.Create(path, true);
        await CacheTestHost.EnsureCreatedAsync(host);
        using var scope = host.CreateScope();
        await using var db = await scope.ServiceProvider.GetRequiredService<IDbContextFactory<ManagementElsaDbContext>>().CreateDbContextAsync();
        db.CacheMutations.Add(new CacheMutation { Id = 100, IsCompleted = true });
        await db.SaveChangesAsync();
        var cache = host.GetRequiredService<ICacheManager>();
        await cache.GetOrCreateAsync("definition", () => Task.FromResult("old"), Lifetime);
        // Simulate an earlier allocated identity becoming visible after identity 100 was cached.
        var late = new CacheMutation { Id = 50 };
        db.CacheMutations.Add(late);
        await db.SaveChangesAsync();
        await cache.CompleteMutationAsync(late.MutationId);
        Assert.NotEqual("100", (await cache.GetConsistencyStateAsync()).Revision);
        Assert.Equal("updated", await cache.GetOrCreateAsync("definition", () => Task.FromResult("updated"), Lifetime));
    }

    [RedisFact]
    public async Task Pruning_on_one_node_cannot_delete_a_newer_revision_completed_by_another_node()
    {
        var path = Path.Combine(Path.GetTempPath(), "elsa-pruning-" + Guid.NewGuid().ToString("N") + ".db");
        var prefix = "workflow-prune-tests:" + Guid.NewGuid().ToString("N") + ":";
        var pause = new PauseFirstPruneInterceptor();
        using var first = CacheTestHost.Create(path, true, prefix, interceptor: pause);
        using var second = CacheTestHost.Create(path, true, prefix);
        await CacheTestHost.EnsureCreatedAsync(first);
        var a = first.GetRequiredService<ICacheManager>();
        var b = second.GetRequiredService<ICacheManager>();
        var firstMutation = await a.BeginMutationAsync();
        var completing = a.CompleteMutationAsync(firstMutation);
        await pause.Entered.Task.WaitAsync(TimeSpan.FromSeconds(10));
        string expected;
        try
        {
            var secondMutation = await b.BeginMutationAsync();
            await b.CompleteMutationAsync(secondMutation);
            expected = (await b.GetConsistencyStateAsync()).Revision;
        }
        finally { pause.Release.SetResult(); }
        await completing;
        Assert.Equal(expected, (await a.GetConsistencyStateAsync()).Revision);
    }

    private sealed class PauseFirstPruneInterceptor : Microsoft.EntityFrameworkCore.Diagnostics.DbCommandInterceptor
    {
        private int _paused;
        public TaskCompletionSource Entered { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously);
        public TaskCompletionSource Release { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously);
        public override async ValueTask<Microsoft.EntityFrameworkCore.Diagnostics.InterceptionResult<int>> NonQueryExecutingAsync(
            System.Data.Common.DbCommand command, Microsoft.EntityFrameworkCore.Diagnostics.CommandEventData eventData,
            Microsoft.EntityFrameworkCore.Diagnostics.InterceptionResult<int> result, CancellationToken cancellationToken = default)
        {
            if (command.CommandText.StartsWith("DELETE", StringComparison.OrdinalIgnoreCase)
                && command.CommandText.Contains("CacheMutations") && Interlocked.Exchange(ref _paused, 1) == 0)
            {
                Entered.SetResult();
                await Release.Task.WaitAsync(cancellationToken);
            }
            return result;
        }
    }

    [RedisFact]
    public async Task Disabled_redis_payload_cache_keeps_shared_coordination_active()
    {
        var path = Path.Combine(Path.GetTempPath(), "elsa-disabled-" + Guid.NewGuid().ToString("N") + ".db");
        using var host = CacheTestHost.Create(path, true, configure: cache => cache.Enabled = false);
        await CacheTestHost.EnsureCreatedAsync(host);
        var cache = host.GetRequiredService<ICacheManager>();
        Assert.True(cache.IsShared);
        Assert.False(cache.IsEnabled);
        var calls = 0;
        await cache.GetOrCreateAsync("definition", () => Task.FromResult(++calls), Lifetime);
        await cache.GetOrCreateAsync("definition", () => Task.FromResult(++calls), Lifetime);
        Assert.Equal(2, calls);
        var mutation = await cache.BeginMutationAsync();
        Assert.True((await cache.GetConsistencyStateAsync()).HasPendingMutations);
        await cache.CompleteMutationAsync(mutation);
        Assert.False((await cache.GetConsistencyStateAsync()).HasPendingMutations);
    }

    [RedisFact]
    public async Task Sliding_redis_lifetime_never_extends_the_absolute_deadline()
    {
        using var host = CacheTestHost.Create(redis: true);
        var provider = host.GetRequiredService<ICacheProvider>();
        var options = new CacheEntryOptions { AbsoluteExpirationRelativeToNow = TimeSpan.FromMilliseconds(1500), SlidingExpiration = TimeSpan.FromMilliseconds(700) };
        await provider.SetAsync("lifetime", "payload", options);
        await Task.Delay(400);
        Assert.True((await provider.GetAsync<string>("lifetime")).Found);
        await Task.Delay(400);
        Assert.True((await provider.GetAsync<string>("lifetime")).Found);
        await Task.Delay(800);
        Assert.False((await provider.GetAsync<string>("lifetime")).Found);
    }
}