ersatztv/ErsatzTV/Controllers/Api/PlexMediaSourcesController.cs

using System.ComponentModel.DataAnnotations;
using System.Threading.Channels;
using ErsatzTV.Application;
using ErsatzTV.Application.Plex;
using ErsatzTV.Controllers.Api.Requests;
using ErsatzTV.Core;
using ErsatzTV.Core.Api.MediaSources;
using ErsatzTV.Core.Interfaces.Locking;
using ErsatzTV.Core.Interfaces.Plex;
using ErsatzTV.Extensions;
using MediatR;
using Microsoft.AspNetCore.Http;
using Microsoft.AspNetCore.Mvc;

namespace ErsatzTV.Controllers.Api;

[ApiController]
public class PlexMediaSourcesController(
    IMediator mediator,
    IEntityLocker entityLocker,
    IPlexSecretStore plexSecretStore,
    ChannelWriter<IScannerBackgroundServiceRequest> scannerWorkerChannel) : ControllerBase
{
    private const string PlexBusyTitle = "Plex sign-in or sync in progress";

    private const string PlexBusyDetail =
        "A Plex sign-in or synchronization is currently in progress; try again once it completes.";

    private static IActionResult PlexLockedProblem() =>
        ApiResults.ConflictProblem(PlexBusyTitle, PlexBusyDetail);

    [HttpGet("/api/media-sources/plex", Name = "GetPlexState")]
    [Tags("Plex")]
    [EndpointSummary("Get Plex connection state")]
    [EndpointDescription(
        "Returns whether ErsatzTV is authorized with plex.tv, whether a Plex sign-in/sync lock is held, " +
        "and the list of discovered Plex servers. Poll this during the pin flow: authorized && !locked = done.")]
    [EndpointGroupName("general")]
    [ProducesResponseType(typeof(RemoteMediaSourceStateResponseModel), StatusCodes.Status200OK)]
    public async Task<RemoteMediaSourceStateResponseModel> GetState(CancellationToken cancellationToken)
    {
        List<PlexMediaSourceViewModel> servers =
            await mediator.Send(new GetAllPlexMediaSources(), cancellationToken);
        bool isAuthorized = (await plexSecretStore.GetUserAuthTokens()).Count > 0;
        return new RemoteMediaSourceStateResponseModel(
            isAuthorized,
            entityLocker.IsPlexLocked(),
            servers.Map(ToItemResponse).ToList());
    }

    [HttpPost("/api/media-sources/plex/pin-flow", Name = "StartPlexPinFlow")]
    [Tags("Plex")]
    [EndpointSummary("Start the Plex sign-in pin flow")]
    [EndpointDescription(
        "Acquires the Plex lock and starts the OAuth pin flow, returning the plex.tv authorization URL to " +
        "open in a new tab. The lock stays held for the background flow; poll GET /api/media-sources/plex " +
        "until authorized && !locked. Also used to fix credentials for an existing but unauthorized server.")]
    [EndpointGroupName("general")]
    [ProducesResponseType(typeof(PlexPinFlowResponseModel), StatusCodes.Status200OK)]
    [ProducesResponseType(typeof(ProblemDetails), StatusCodes.Status409Conflict)]
    [ProducesResponseType(typeof(ProblemDetails), StatusCodes.Status422UnprocessableEntity)]
    public async Task<IActionResult> StartPinFlow(CancellationToken cancellationToken)
    {
        // The lock IS the running sign-in flow (§3b): fail to acquire = a flow/sync is already active → 409.
        if (!entityLocker.LockPlex())
        {
            return PlexLockedProblem();
        }

        // Only the Right/200 path keeps the lock held (handed off to the background pin flow). Release the
        // controller-acquired lock on the Left branch AND on any thrown exception from the dispatch/enqueue
        // (re-review finding 2 — StartPlexPinFlowHandler awaits a channel WriteAsync that can throw).
        try
        {
            Either<BaseError, string> result = await mediator.Send(new StartPlexPinFlow(), cancellationToken);
            return result.Match(
                Left: error =>
                {
                    entityLocker.UnlockPlex();
                    return error.ToErrorResult();
                },
                Right: url => (IActionResult)new OkObjectResult(new PlexPinFlowResponseModel(url)));
        }
        catch (Exception)
        {
            entityLocker.UnlockPlex();
            throw;
        }
    }

    [HttpPost("/api/media-sources/plex/sign-out", Name = "SignOutOfPlex")]
    [Tags("Plex")]
    [EndpointSummary("Sign out of Plex")]
    [EndpointDescription(
        "Purges all Plex servers, synced Plex content, and stored credentials. Acquires the Plex lock; the " +
        "handler always releases it (even on failure).")]
    [EndpointGroupName("general")]
    [ProducesResponseType(StatusCodes.Status204NoContent)]
    [ProducesResponseType(typeof(ProblemDetails), StatusCodes.Status409Conflict)]
    public async Task<IActionResult> SignOutOfPlex(CancellationToken cancellationToken)
    {
        if (!entityLocker.LockPlex())
        {
            return PlexLockedProblem();
        }

        // SignOutOfPlexHandler is the designated releaser (unconditional finally) — the controller acquires
        // and hands off; it does NOT release here (that would double-release on the handler's success path).
        Either<BaseError, Unit> result = await mediator.Send(new SignOutOfPlex(), cancellationToken);
        return result.ToDeletedResult();
    }

    [HttpGet("/api/media-sources/plex/{id:int}/libraries", Name = "GetPlexLibraries")]
    [Tags("Plex")]
    [EndpointSummary("Get a Plex server's libraries")]
    [EndpointGroupName("general")]
    [ProducesResponseType(typeof(List<RemoteLibraryResponseModel>), StatusCodes.Status200OK)]
    [ProducesResponseType(typeof(ProblemDetails), StatusCodes.Status404NotFound)]
    public async Task<IActionResult> GetLibraries(int id, CancellationToken cancellationToken)
    {
        if (!await PlexSourceExists(id, cancellationToken))
        {
            return ApiResults.NotFoundProblem();
        }

        List<PlexLibraryViewModel> libraries =
            await mediator.Send(new GetPlexLibrariesBySourceId(id), cancellationToken);
        return new OkObjectResult(libraries.Map(ToLibraryResponse).ToList());
    }

    [HttpPut("/api/media-sources/plex/{id:int}/libraries", Name = "ReplacePlexLibraryPreferences")]
    [Tags("Plex")]
    [EndpointSummary("Replace a Plex server's library sync preferences")]
    [EndpointDescription(
        "The body is the complete set of the source's libraries with each shouldSyncItems flag. A row absent " +
        "from the request is left untouched. Every id must belong to this source (no Id=0 rows). Returns the " +
        "reloaded list — library ids change when sync is disabled, so re-key any client draft off the response. " +
        "Enabled libraries are queued for sync (locked libraries are skipped).")]
    [EndpointGroupName("general")]
    [ProducesResponseType(typeof(List<RemoteLibraryResponseModel>), StatusCodes.Status200OK)]
    [ProducesResponseType(typeof(ProblemDetails), StatusCodes.Status404NotFound)]
    [ProducesResponseType(typeof(ProblemDetails), StatusCodes.Status422UnprocessableEntity)]
    public async Task<IActionResult> ReplaceLibraryPreferences(
        int id,
        [Required] [FromBody] ReplaceRemoteLibraryPreferencesRequest request,
        CancellationToken cancellationToken)
    {
        if (!await PlexSourceExists(id, cancellationToken))
        {
            return ApiResults.NotFoundProblem();
        }

        List<RemoteLibraryPreferenceRequest> libraries = request.Libraries ?? [];
        List<PlexLibraryViewModel> owned =
            await mediator.Send(new GetPlexLibrariesBySourceId(id), cancellationToken);
        var ownedIds = owned.Map(l => l.Id).ToHashSet();

        // Source-scope the id set (§C4a / finding 2a): reject Id=0 rows and any id not owned by this source
        // before dispatch — UpdatePlexLibraryPreferences carries no source id, so this is the only guard.
        var invalidIds = libraries.Filter(l => l.Id < 1 || !ownedIds.Contains(l.Id)).Map(l => l.Id).ToList();
        if (invalidIds.Count > 0)
        {
            return BaseError.New($"Library {invalidIds[0]} does not belong to Plex media source {id}")
                .ToErrorResult();
        }

        // Require the complete set of this source's libraries (§C4a; matches Jellyfin/Emby): the PUT is a
        // full flag document, so an omitted owned id would silently keep its old sync state.
        var incomingIds = libraries.Map(l => l.Id).ToList();
        var missingIds = ownedIds.Where(ownedId => !incomingIds.Contains(ownedId)).ToList();
        if (missingIds.Count > 0)
        {
            return BaseError.New(
                    $"The request must include every library for Plex media source {id} " +
                    $"(missing id(s) {string.Join(", ", missingIds)})")
                .ToErrorResult();
        }

        Either<BaseError, Unit> result =
            await mediator.Send(request.ToPlexCommand(), cancellationToken);
        if (result.IsLeft)
        {
            foreach (BaseError error in result.LeftToSeq())
            {
                return error.ToErrorResult();
            }
        }

        // Reload BEFORE enqueueing — ids change on disable (§C4a); the reloaded list carries the fresh ids
        // for both the sync loop and the response (§7 write-path projection: reload via the GET's query).
        List<PlexLibraryViewModel> reloaded =
            await mediator.Send(new GetPlexLibrariesBySourceId(id), cancellationToken);

        await EnqueuePostSaveSync(reloaded, cancellationToken);

        return new OkObjectResult(reloaded.Map(ToLibraryResponse).ToList());
    }

    [HttpGet("/api/media-sources/plex/{id:int}/path-replacements", Name = "GetPlexPathReplacements")]
    [Tags("Plex")]
    [EndpointSummary("Get a Plex server's path replacements")]
    [EndpointGroupName("general")]
    [ProducesResponseType(typeof(List<PathReplacementResponseModel>), StatusCodes.Status200OK)]
    [ProducesResponseType(typeof(ProblemDetails), StatusCodes.Status404NotFound)]
    public async Task<IActionResult> GetPathReplacements(int id, CancellationToken cancellationToken)
    {
        if (!await PlexSourceExists(id, cancellationToken))
        {
            return ApiResults.NotFoundProblem();
        }

        List<PlexPathReplacementViewModel> replacements =
            await mediator.Send(new GetPlexPathReplacementsBySourceId(id), cancellationToken);
        return new OkObjectResult(replacements.Map(ToPathReplacementResponse).ToList());
    }

    [HttpPut("/api/media-sources/plex/{id:int}/path-replacements", Name = "ReplacePlexPathReplacements")]
    [Tags("Plex")]
    [EndpointSummary("Replace a Plex server's path replacements")]
    [EndpointDescription(
        "Replaces the source's path replacements: an existing id updates, id<1 adds, an existing id absent from " +
        "the body is deleted. Every positive id must belong to this source and every row needs a non-empty " +
        "RemotePath and LocalPath (else 422, no partial mutation). Returns the reloaded list.")]
    [EndpointGroupName("general")]
    [ProducesResponseType(typeof(List<PathReplacementResponseModel>), StatusCodes.Status200OK)]
    [ProducesResponseType(typeof(ProblemDetails), StatusCodes.Status404NotFound)]
    [ProducesResponseType(typeof(ProblemDetails), StatusCodes.Status422UnprocessableEntity)]
    public async Task<IActionResult> ReplacePathReplacements(
        int id,
        [Required] [FromBody] ReplacePathReplacementsRequest request,
        CancellationToken cancellationToken)
    {
        if (!await PlexSourceExists(id, cancellationToken))
        {
            return ApiResults.NotFoundProblem();
        }

        // The handler owns the cross-source ownership guard + nonblank/null validation (→ 422, no mutation).
        Either<BaseError, Unit> result =
            await mediator.Send(request.ToPlexCommand(id), cancellationToken);
        if (result.IsLeft)
        {
            foreach (BaseError error in result.LeftToSeq())
            {
                return error.ToErrorResult();
            }
        }

        List<PlexPathReplacementViewModel> reloaded =
            await mediator.Send(new GetPlexPathReplacementsBySourceId(id), cancellationToken);
        return new OkObjectResult(reloaded.Map(ToPathReplacementResponse).ToList());
    }

    [HttpPost("/api/media-sources/plex/{id:int}/refresh-libraries", Name = "RefreshPlexLibraries")]
    [Tags("Plex")]
    [EndpointSummary("Refresh a Plex server's libraries")]
    [EndpointDescription(
        "Queues a synchronization of the server's libraries (fire-and-forget). Returns 409 while a Plex " +
        "sign-in/sync lock is held; duplicate refreshes are accepted (the scan is idempotent).")]
    [EndpointGroupName("general")]
    [ProducesResponseType(StatusCodes.Status202Accepted)]
    [ProducesResponseType(typeof(ProblemDetails), StatusCodes.Status404NotFound)]
    [ProducesResponseType(typeof(ProblemDetails), StatusCodes.Status409Conflict)]
    public async Task<IActionResult> RefreshLibraries(int id, CancellationToken cancellationToken)
    {
        if (!await PlexSourceExists(id, cancellationToken))
        {
            return ApiResults.NotFoundProblem();
        }

        if (entityLocker.IsPlexLocked())
        {
            return PlexLockedProblem();
        }

        await scannerWorkerChannel.WriteAsync(new SynchronizePlexLibraries(id), cancellationToken);
        return new AcceptedResult();
    }

    [HttpPost("/api/media-sources/plex/{id:int}/scan-collections", Name = "ScanPlexCollections")]
    [Tags("Plex")]
    [EndpointSummary("Scan a Plex server's collections")]
    [EndpointDescription(
        "Queues a synchronization of the server's collections (fire-and-forget). Pass ?deep=true for a deep " +
        "scan. Returns 409 while a Plex collections scan is already in progress.")]
    [EndpointGroupName("general")]
    [ProducesResponseType(StatusCodes.Status202Accepted)]
    [ProducesResponseType(typeof(ProblemDetails), StatusCodes.Status404NotFound)]
    [ProducesResponseType(typeof(ProblemDetails), StatusCodes.Status409Conflict)]
    public async Task<IActionResult> ScanCollections(
        int id,
        [FromQuery] bool deep = false,
        CancellationToken cancellationToken = default)
    {
        if (!await PlexSourceExists(id, cancellationToken))
        {
            return ApiResults.NotFoundProblem();
        }

        // The collections lock IS the running scan (§3b): fail to acquire = a scan is already active → 409.
        if (!entityLocker.LockPlexCollections())
        {
            return ApiResults.ConflictProblem(
                "Plex collections scan in progress",
                "A Plex collections scan is already in progress; try again once it completes.");
        }

        try
        {
            await scannerWorkerChannel.WriteAsync(
                new SynchronizePlexCollections(id, true, deep),
                cancellationToken);
        }
        catch
        {
            // the scanner releases the lock when it processes the message; if the enqueue throws after we
            // acquired the lock, release it here (EnqueueWithTraktLock compensating-unlock, §3b)
            entityLocker.UnlockPlexCollections();
            throw;
        }

        return new AcceptedResult();
    }

    private async Task<bool> PlexSourceExists(int id, CancellationToken cancellationToken) =>
        (await mediator.Send(new GetPlexMediaSourceById(id), cancellationToken)).IsSome;

    // Post-save re-sync (§C7 / finding 6): per enabled library, one LockLibrary ⇄ one release carried by
    // the LAST message. Correcting the Blazor bug: the library message runs Unlock:false so the release
    // rides SynchronizePlexNetworks(Unlock:true). Compensating-unlock if the networks enqueue throws.
    private async Task EnqueuePostSaveSync(
        IEnumerable<PlexLibraryViewModel> libraries,
        CancellationToken cancellationToken)
    {
        foreach (PlexLibraryViewModel library in libraries.Filter(l => l.ShouldSyncItems))
        {
            if (!entityLocker.LockLibrary(library.Id))
            {
                continue;
            }

            try
            {
                await scannerWorkerChannel.WriteAsync(
                    new SynchronizePlexLibraryByIdIfNeeded(library.Id, Unlock: false),
                    cancellationToken);
                await scannerWorkerChannel.WriteAsync(
                    new SynchronizePlexNetworks(library.Id, false, Unlock: true),
                    cancellationToken);
            }
            catch
            {
                // the library message carries Unlock:false, so the not-yet-enqueued networks message was
                // the sole releaser — release here since it never ran
                entityLocker.UnlockLibrary(library.Id);
                throw;
            }
        }
    }

    private static RemoteMediaSourceItemResponseModel ToItemResponse(PlexMediaSourceViewModel vm) =>
        new(vm.Id, vm.Name, vm.Address);

    private static RemoteLibraryResponseModel ToLibraryResponse(PlexLibraryViewModel vm) =>
        new(vm.Id, vm.Name, vm.MediaKind, vm.ShouldSyncItems);

    private static PathReplacementResponseModel ToPathReplacementResponse(PlexPathReplacementViewModel vm) =>
        new(vm.Id, vm.PlexPath, vm.LocalPath);
}