using System;
using System.Collections.Generic;
using System.Globalization;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Emby.Server.Implementations.Library;
using Jellyfin.Database.Implementations;
using Jellyfin.Database.Implementations.DbConfiguration;
using Jellyfin.Database.Implementations.Entities;
using Jellyfin.Database.Implementations.Locking;
using Jellyfin.Database.Providers.PostgreSQL;
using Jellyfin.Server.Tests.Migrations;
using MediaBrowser.Controller.Configuration;
using MediaBrowser.Model.Configuration;
using MediaBrowser.Model.Entities;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Logging.Abstractions;
using Moq;
using Npgsql;
using Xunit;
using AudioBook = MediaBrowser.Controller.Entities.AudioBook;
namespace Jellyfin.Server.Tests.Library;
///
/// Two independently constructed instances over one PostgreSQL database are the
/// in-process stand-in for two replicas sharing one database: what either of them writes, the other has to
/// see on its very next read, and a read-modify-write on one must not roll back the other's.
///
[Trait("Category", "RequiresDocker")]
public sealed class UserDataManagerReplicaTests : IClassFixture
{
private static readonly long _quarterIn = TimeSpan.FromMinutes(20).Ticks;
private readonly NpgsqlDataSource _dataSource;
public UserDataManagerReplicaTests(DatabaseFixture fixture)
{
_dataSource = fixture.DataSource;
}
///
/// A resume position written by the replica serving the playback tick has to be the position the next
/// request reads, whichever replica it lands on - both through the single item read the write path uses
/// and through the batch read the library pages render from.
///
/// A representing the asynchronous operation.
[Fact]
public async Task ResumePositionWrittenOnOneReplica_IsReadOnAnother()
{
var cancellationToken = TestContext.Current.CancellationToken;
var itemId = Guid.NewGuid();
var user = await CreateUserAndItemAsync(_dataSource, itemId, cancellationToken);
var replicaA = CreateManager(_dataSource);
var replicaB = CreateManager(_dataSource);
var itemOnA = new AudioBook { Id = itemId, Name = "Replica Book" };
var itemOnB = new AudioBook { Id = itemId, Name = "Replica Book" };
var early = replicaA.GetUserData(user, itemOnA)!;
early.PlaybackPositionTicks = TimeSpan.FromMinutes(5).Ticks;
replicaA.SaveUserData(user, itemOnA, early, UserDataSaveReason.PlaybackProgress, cancellationToken);
// Replica B materialised the item before the later tick, so it holds the earlier row in memory.
itemOnB.UserData = await LoadUserDataAsync(_dataSource, itemId, cancellationToken);
var later = replicaA.GetUserData(user, itemOnA)!;
later.PlaybackPositionTicks = _quarterIn;
replicaA.SaveUserData(user, itemOnA, later, UserDataSaveReason.PlaybackProgress, cancellationToken);
Assert.Equal(_quarterIn, replicaB.GetUserData(user, itemOnB)!.PlaybackPositionTicks);
Assert.Equal(_quarterIn, replicaB.GetUserDataBatch([itemOnB], user)[itemId].PlaybackPositionTicks);
}
///
/// The playback tick is a read-modify-write of the whole row, so a tick served by one replica must build
/// on the favourite another replica just recorded instead of writing it back out.
///
/// A representing the asynchronous operation.
[Fact]
public async Task PlaybackTickOnOneReplica_KeepsFavouriteSetOnAnother()
{
var cancellationToken = TestContext.Current.CancellationToken;
var itemId = Guid.NewGuid();
var user = await CreateUserAndItemAsync(_dataSource, itemId, cancellationToken);
var replicaA = CreateManager(_dataSource);
var replicaB = CreateManager(_dataSource);
var itemOnA = new AudioBook { Id = itemId, Name = "Replica Book" };
var itemOnB = new AudioBook { Id = itemId, Name = "Replica Book" };
var seed = replicaA.GetUserData(user, itemOnA)!;
seed.PlaybackPositionTicks = TimeSpan.FromMinutes(5).Ticks;
replicaA.SaveUserData(user, itemOnA, seed, UserDataSaveReason.PlaybackProgress, cancellationToken);
// Replica B is serving the playback session and read the item before the favourite was recorded.
itemOnB.UserData = await LoadUserDataAsync(_dataSource, itemId, cancellationToken);
var favourited = replicaA.GetUserData(user, itemOnA)!;
favourited.IsFavorite = true;
replicaA.SaveUserData(user, itemOnA, favourited, UserDataSaveReason.UpdateUserRating, cancellationToken);
var tick = replicaB.GetUserData(user, itemOnB)!;
tick.PlaybackPositionTicks = _quarterIn;
replicaB.SaveUserData(user, itemOnB, tick, UserDataSaveReason.PlaybackProgress, cancellationToken);
var stored = replicaA.GetUserData(user, itemOnA)!;
Assert.True(stored.IsFavorite);
Assert.Equal(_quarterIn, stored.PlaybackPositionTicks);
}
///
/// A tick that lands on the other replica has to carry the position forward from where the session
/// actually is, not from the position that replica happened to have in memory.
///
/// A representing the asynchronous operation.
[Fact]
public async Task PlaybackTickOnOneReplica_ResumesFromThePositionAnotherWrote()
{
var cancellationToken = TestContext.Current.CancellationToken;
var itemId = Guid.NewGuid();
var user = await CreateUserAndItemAsync(_dataSource, itemId, cancellationToken);
var replicaA = CreateManager(_dataSource);
var replicaB = CreateManager(_dataSource);
var itemOnA = new AudioBook { Id = itemId, Name = "Replica Book" };
var itemOnB = new AudioBook { Id = itemId, Name = "Replica Book" };
var seed = replicaA.GetUserData(user, itemOnA)!;
seed.PlaybackPositionTicks = TimeSpan.FromMinutes(5).Ticks;
replicaA.SaveUserData(user, itemOnA, seed, UserDataSaveReason.PlaybackProgress, cancellationToken);
itemOnB.UserData = await LoadUserDataAsync(_dataSource, itemId, cancellationToken);
// The viewer seeks forward and the tick reporting it lands on replica A.
var seeked = replicaA.GetUserData(user, itemOnA)!;
seeked.PlaybackPositionTicks = _quarterIn;
replicaA.SaveUserData(user, itemOnA, seeked, UserDataSaveReason.PlaybackProgress, cancellationToken);
// The next tick lands on replica B, which adds ten seconds to whatever it reads.
var tick = replicaB.GetUserData(user, itemOnB)!;
tick.PlaybackPositionTicks += TimeSpan.FromSeconds(10).Ticks;
replicaB.SaveUserData(user, itemOnB, tick, UserDataSaveReason.PlaybackProgress, cancellationToken);
var stored = replicaA.GetUserData(user, itemOnA)!;
Assert.Equal(_quarterIn + TimeSpan.FromSeconds(10).Ticks, stored.PlaybackPositionTicks);
}
private static UserDataManager CreateManager(NpgsqlDataSource dataSource)
{
var config = new Mock();
config.SetupGet(c => c.Configuration).Returns(new ServerConfiguration());
return new UserDataManager(config.Object, new DataSourceContextFactory(dataSource));
}
private static async Task> LoadUserDataAsync(NpgsqlDataSource dataSource, Guid itemId, CancellationToken cancellationToken)
{
var context = CreateContext(dataSource);
await using (context.ConfigureAwait(false))
{
return await context.UserData
.AsNoTracking()
.Where(e => e.ItemId.Equals(itemId))
.ToArrayAsync(cancellationToken)
.ConfigureAwait(false);
}
}
private static async Task CreateUserAndItemAsync(NpgsqlDataSource dataSource, Guid itemId, CancellationToken cancellationToken)
{
var context = CreateContext(dataSource);
await using (context.ConfigureAwait(false))
{
var user = new User("replica-user-" + itemId.ToString("N", CultureInfo.InvariantCulture), "provider", "provider");
context.Users.Add(user);
context.BaseItems.Add(new BaseItemEntity { Id = itemId, Type = typeof(AudioBook).FullName! });
await context.SaveChangesAsync(cancellationToken).ConfigureAwait(false);
return user;
}
}
private static JellyfinDbContext CreateContext(NpgsqlDataSource dataSource)
{
var optionsBuilder = new DbContextOptionsBuilder();
var provider = new PostgreSqlDatabaseProvider(dataSource);
provider.Initialise(optionsBuilder, new DatabaseConfigurationOptions { DatabaseType = "PostgreSQL" });
return new JellyfinDbContext(
optionsBuilder.Options,
NullLogger.Instance,
provider,
new NoLockBehavior(NullLogger.Instance));
}
///
/// Hands every its own context over the one shared database, the way the
/// pooled factory does in the server.
///
private sealed class DataSourceContextFactory : IDbContextFactory
{
private readonly NpgsqlDataSource _dataSource;
public DataSourceContextFactory(NpgsqlDataSource dataSource)
{
_dataSource = dataSource;
}
public JellyfinDbContext CreateDbContext() => CreateContext(_dataSource);
}
///
/// Builds the schema once for the whole class. Every test keeps to its own user and item, so one
/// database serves all of them and the shared server is spared three schema builds.
///
public sealed class DatabaseFixture : IAsyncLifetime
{
private PostgreSqlTestServer _server = null!;
public NpgsqlDataSource DataSource { get; private set; } = null!;
///
public async ValueTask InitializeAsync()
{
_server = await PostgreSqlTestServer.StartAsync().ConfigureAwait(false);
var connectionString = await _server.CreateDatabaseAsync("userdata_replica", CancellationToken.None).ConfigureAwait(false);
DataSource = new NpgsqlDataSourceBuilder(connectionString).Build();
var context = CreateContext(DataSource);
await using (context.ConfigureAwait(false))
{
await context.Database.EnsureCreatedAsync(CancellationToken.None).ConfigureAwait(false);
}
}
///
public async ValueTask DisposeAsync()
{
await DataSource.DisposeAsync().ConfigureAwait(false);
await _server.DisposeAsync().ConfigureAwait(false);
}
}
}