jellyfin-ha-src/tests/Jellyfin.Server.Implementations.Tests/Configuration/RemoteInvalidationApplyTests.cs

using System;
using System.IO;
using System.Threading;
using System.Threading.Tasks;
using Emby.Server.Implementations;
using Emby.Server.Implementations.Configuration;
using Emby.Server.Implementations.Serialization;
using MediaBrowser.Common.Configuration;
using MediaBrowser.Common.Net;
using MediaBrowser.Controller.Configuration;
using Microsoft.Extensions.Logging.Abstractions;
using Xunit;

namespace Jellyfin.Server.Implementations.Tests.Configuration;

/// <summary>
/// Applying an invalidation re-raises the same update events a local save raises, and some consumers of
/// those events answer an update by writing - <c>RecordingsManager</c> creating the recording folders for
/// the <c>livetv</c> key, <c>ApplicationHost</c> clearing <c>IsPortAuthorized</c> on a port change. The
/// instance that did not write must not repeat those writes, and no write it is induced into must reach
/// the bus.
/// </summary>
public sealed class RemoteInvalidationApplyTests : IDisposable
{
    private readonly string _root;

    /// <summary>
    /// Initializes a new instance of the <see cref="RemoteInvalidationApplyTests"/> class.
    /// </summary>
    public RemoteInvalidationApplyTests()
    {
        _root = Path.Combine(Path.GetTempPath(), "jf-config-apply-" + Guid.NewGuid().ToString("N"));
        Directory.CreateDirectory(_root);
    }

    /// <inheritdoc />
    public void Dispose()
    {
        try
        {
            Directory.Delete(_root, true);
        }
        catch (IOException)
        {
        }
    }

    /// <summary>
    /// A consumer that knows nothing about the bus - a plugin, or anything reached transitively from one -
    /// can answer an applied invalidation by writing. That write must not become a notice of its own.
    /// </summary>
    /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
    [Fact]
    public async Task NamedInvalidation_InducingAWriteOnTheReceiver_DoesNotPublishBack()
    {
        var fabric = new FakeInvalidationBusFabric();
        var instanceA = CreateInstance(fabric, "pod-a");
        var instanceB = await CreateSubscribedInstanceAsync(fabric, "pod-b");

        var receivedByA = 0;
        instanceA.InvalidationBus.Subscribe(_ => Interlocked.Increment(ref receivedByA));

        var writesByB = 0;
        instanceB.NamedConfigurationUpdated += (_, e) =>
        {
            // One shot: the induced write raises the event again on this instance.
            if (Interlocked.Increment(ref writesByB) > 1)
            {
                return;
            }

            var configuration = (NetworkConfiguration)instanceB.GetConfiguration(e.Key);
            configuration.PublishedServerUriBySubnet = ["10.0.0.0/8=example"];
            instanceB.SaveConfiguration(e.Key, configuration);
        };

        var updated = instanceA.GetNetworkConfiguration();
        updated.EnableRemoteAccess = false;
        instanceA.SaveConfiguration(NetworkConfigurationStore.StoreKey, updated);

        Assert.Equal(0, receivedByA);

        // The point of the bus still holds: B is not left on its stale copy.
        Assert.False(instanceB.GetNetworkConfiguration().EnableRemoteAccess);
    }

    /// <summary>
    /// The shape of <c>RecordingsManager</c>: a consumer that answers a named configuration update by
    /// writing has to be able to tell that the write was another instance's, and skip it.
    /// </summary>
    /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
    [Fact]
    public async Task NamedInvalidation_WithAWriteTriggeringConsumer_DoesNotDuplicateTheWrite()
    {
        var fabric = new FakeInvalidationBusFabric();
        var instanceA = CreateInstance(fabric, "pod-a");
        var instanceB = await CreateSubscribedInstanceAsync(fabric, "pod-b");

        var receivedByA = 0;
        instanceA.InvalidationBus.Subscribe(_ => Interlocked.Increment(ref receivedByA));

        var writesByB = 0;
        instanceB.NamedConfigurationUpdated += (_, e) =>
        {
            if (ConfigurationInvalidationContext.IsApplyingRemoteInvalidation)
            {
                return;
            }

            Interlocked.Increment(ref writesByB);
        };

        var updated = instanceA.GetNetworkConfiguration();
        updated.EnableRemoteAccess = false;
        instanceA.SaveConfiguration(NetworkConfigurationStore.StoreKey, updated);

        Assert.Equal(0, writesByB);
        Assert.Equal(0, receivedByA);
        Assert.False(instanceB.GetNetworkConfiguration().EnableRemoteAccess);

        // A read-only consumer is still told, which is what the invalidation exists for.
        var refreshes = 0;
        instanceB.NamedConfigurationUpdated += (_, _) => Interlocked.Increment(ref refreshes);

        updated.EnableRemoteAccess = true;
        instanceA.SaveConfiguration(NetworkConfigurationStore.StoreKey, updated);

        Assert.Equal(1, refreshes);
    }

    /// <summary>
    /// The <c>ApplicationHost.IsPortAuthorized</c> class of consumer: the system configuration event is
    /// queued rather than raised inline, so the fix has to survive the hop onto the thread pool.
    /// </summary>
    /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
    [Fact]
    public async Task SystemInvalidation_InducingAQueuedWriteOnTheReceiver_DoesNotPublishBack()
    {
        var fabric = new FakeInvalidationBusFabric();
        var instanceA = CreateInstance(fabric, "pod-a");
        var instanceB = await CreateSubscribedInstanceAsync(fabric, "pod-b");

        var receivedByA = 0;
        instanceA.InvalidationBus.Subscribe(_ => Interlocked.Increment(ref receivedByA));

        var applied = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
        var handled = 0;
        instanceB.ConfigurationUpdated += (_, _) =>
        {
            if (Interlocked.Increment(ref handled) > 1)
            {
                return;
            }

            instanceB.Configuration.IsPortAuthorized = false;
            instanceB.SaveConfiguration();
            applied.TrySetResult();
        };

        instanceA.Configuration.QuickConnectAvailable = false;
        instanceA.SaveConfiguration();

        await applied.Task.WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken);

        Assert.Equal(0, receivedByA);
        Assert.False(instanceB.Configuration.QuickConnectAvailable);
    }

    private async Task<ServerConfigurationManager> CreateSubscribedInstanceAsync(FakeInvalidationBusFabric fabric, string originId)
    {
        var instance = CreateInstance(fabric, originId);
        var subscriber = new ConfigurationInvalidationSubscriber(
            instance.InvalidationBus,
            instance,
            NullLogger<ConfigurationInvalidationSubscriber>.Instance);

        await subscriber.StartAsync(CancellationToken.None);
        return instance;
    }

    private ServerConfigurationManager CreateInstance(FakeInvalidationBusFabric fabric, string originId)
    {
        var paths = new ServerApplicationPaths(
            Ensure("data"),
            Ensure("log"),
            Ensure("config"),
            Ensure("cache"),
            Ensure("web"));

        var manager = new ServerConfigurationManager(paths, NullLoggerFactory.Instance, new MyXmlSerializer())
        {
            InvalidationBus = fabric.Connect(originId)
        };

        manager.AddParts([new NetworkConfigurationFactory()]);
        return manager;
    }

    private string Ensure(string name)
    {
        var path = Path.Combine(_root, name);
        Directory.CreateDirectory(path);
        return path;
    }
}