timothy/ersatztv
1
0
Fork 0
Code Issues 51 Pull Requests 5 Actions Packages Projects Releases 9 Wiki Activity

feat(api): Plex media-source write API + lock-lifecycle fixes (#202 slice S2)

Browse Source 
New PlexMediaSourcesController (/api/media-sources/plex) P1-P8 wrapping
existing MediatR commands: state GET, pin-flow, sign-out, per-server
libraries/path-replacements GET+PUT, and refresh — VMs projected to the
S0 shared DTOs, ApiResults mapping, 404 controller pre-checks, #215-style
409 lock guards, [EndpointGroupName("general")].

Lock-lifecycle hardening (the tricky part):
- TryCompletePlexPinFlowHandler now releases the Plex lock ONLY on its
  non-handoff exits (timeout-throw, poll exception, enqueue exception, the
  dead return-false) via try/catch — NOT an unconditional finally. On
  success the lock is handed off to SynchronizePlexMediaSources (the sole
  releaser after discovery); a finally would double-release and release
  before discovery, re-opening the finding-5 poll race. Fixes the latent
  leak where an abandoned pin flow wedged Plex locked until restart.
- StartPlexPinFlow controller compensates UnlockPlex on the Left branch AND
  any thrown dispatch/enqueue; only the Right/200 path holds the lock.
- SignOutOfPlexHandler wraps its work in try/finally { UnlockPlex() } — a
  terminal handler with no handoff, so unconditional release is correct.
- Post-save library sync enqueues SynchronizePlexLibraryByIdIfNeeded
  (Unlock:false) then SynchronizePlexNetworks (Unlock:true) — one lock, one
  release on the last message, compensating-unlock if the 2nd enqueue throws
  (corrects the Blazor Unlock-ordering bug, finding 6).

Data-integrity hardening:
- UpdatePlexPathReplacementsHandler rejects (422, no mutation) any positive
  Id not owned by the route source, blank RemotePath/LocalPath, and null
  list/items (findings 2c/8).
- MediaSourceRepository Plex path-replacement UPDATE gains
  AND PlexMediaSourceId = @id (Jellyfin/Emby untouched — slice S3).
- ReplaceLibraryPreferences controller validates the id set against the
  source's libraries (rejects unowned + Id=0), returns the reloaded list
  (ids change on disable).

Tests (NUnit/Shouldly/NSubstitute), 34 new, all green: pin-flow lock
released on thrown-cancellation/poll-throw/enqueue-throw AND held on
success (no double-release); sign-out finally-release under a throwing
dependency; cross-source/nonblank/null path-replacement 422s; library-prefs
id-not-owned 422; post-save enqueue exact messages + Unlock flags; full
controller route/404/409/422 coverage.

Refs #202
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
timothy
2026-07-11 15:42:40 +02:00
parent 295004ab12
commit 4852c36268
7 changed files with 1124 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ErsatzTV/Controllers/Api/PlexMediaSourcesController.cs
+312
View File
@@ -0,0 +1,312 @@
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();
}
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();
}
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);
}