fix: on-demand channel progress using next engine (#2985)

CHANGELOG.md

@@ -30,6 +30,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
 - Next engine
   - Fix audio dropout/desync when using QSV accel and loudness normalization with certain content
   - Fix anamorphic content scaling (was incorrectly stretched with older builds)
+  - Fix on-demand channel progress (channels would not save checkpoints and would always start at the same spot)
 
 ## [26.7.1] - 2026-07-31
 ### Changed

ErsatzTV.Application/Playouts/Mapper.cs

@@ -112,4 +112,17 @@ internal static class Mapper
                 return string.Empty;
         }
     }
+
+    internal static Option<PlayoutModeViewModel> ProjectToModeViewModel(Option<Channel> maybeChannel)
+    {
+        foreach (Channel channel in maybeChannel)
+        {
+            foreach (Playout playout in channel.Playouts.HeadOrNone())
+            {
+                return new PlayoutModeViewModel(playout.Id, channel.PlayoutMode);
+            }
+        }
+
+        return Option<PlayoutModeViewModel>.None;
+    }
 }

ErsatzTV.Application/Playouts/PlayoutModeViewModel.cs

@@ -0,0 +1,5 @@
+using ErsatzTV.Core.Domain;
+
+namespace ErsatzTV.Application.Playouts;
+
+public record PlayoutModeViewModel(int PlayoutId, ChannelPlayoutMode PlayoutMode);

ErsatzTV.Application/Playouts/Queries/GetPlayoutModeByChannelNumber.cs

@@ -0,0 +1,3 @@
+namespace ErsatzTV.Application.Playouts;
+
+public record GetPlayoutModeByChannelNumber(string ChannelNumber) : IRequest<Option<PlayoutModeViewModel>>;

ErsatzTV.Application/Playouts/Queries/GetPlayoutModeByChannelNumberHandler.cs

@@ -0,0 +1,22 @@
+using ErsatzTV.Infrastructure.Data;
+using Microsoft.EntityFrameworkCore;
+
+namespace ErsatzTV.Application.Playouts;
+
+public class GetPlayoutModeByChannelNumberHandler(IDbContextFactory<TvContext> dbContextFactory)
+    : IRequestHandler<GetPlayoutModeByChannelNumber, Option<PlayoutModeViewModel>>
+{
+    public async Task<Option<PlayoutModeViewModel>> Handle(
+        GetPlayoutModeByChannelNumber request,
+        CancellationToken cancellationToken)
+    {
+        await using TvContext dbContext = await dbContextFactory.CreateDbContextAsync(cancellationToken);
+
+        return await dbContext.Channels
+            .AsNoTracking()
+            .Include(c => c.Playouts)
+            .SingleOrDefaultAsync(c => c.Number == request.ChannelNumber, cancellationToken)
+            .Map(Optional)
+            .Map(Mapper.ProjectToModeViewModel);
+    }
+}

ErsatzTV.Application/Streaming/NextSessionWorker.cs

@@ -1,6 +1,8 @@
 using System.IO.Abstractions;
 using CliWrap;
+using ErsatzTV.Application.Playouts;
 using ErsatzTV.Core;
+using ErsatzTV.Core.Domain;
 using ErsatzTV.Core.FFmpeg;
 using ErsatzTV.Core.Interfaces.FFmpeg;
 using ErsatzTV.Core.Interfaces.Metadata;
@@ -19,6 +21,8 @@ public class NextSessionWorker(
     ILogger<NextSessionWorker> logger)
     : IHlsSessionWorker
 {
+    private readonly TimeSpan _checkpointThreshold = TimeSpan.FromMinutes(5);
+
     private readonly SemaphoreSlim _slim = new(1, 1);
     private CancellationTokenSource _cancellationTokenSource;
     private IServiceScope _serviceScope = serviceScopeFactory.CreateScope();
@@ -26,6 +30,11 @@ public class NextSessionWorker(
     private string _channelNumber;
     private string _workingDirectory;
     private string _heartbeatFileName;
+    private DateTimeOffset _lastTouch;
+    private DateTimeOffset _lastCheckpoint;
+    private ChannelPlayoutMode _channelPlayoutMode = ChannelPlayoutMode.Continuous;
+
+    private IMediator Mediator => _serviceScope.ServiceProvider.GetRequiredService<IMediator>();
 
     void IDisposable.Dispose()
     {
@@ -64,6 +73,8 @@ public class NextSessionWorker(
 
     public void Touch(Option<string> fileName)
     {
+        _lastTouch = DateTimeOffset.Now;
+
         if (!fileSystem.File.Exists(_heartbeatFileName))
         {
             fileSystem.File.WriteAllBytes(_heartbeatFileName, []);
@@ -92,6 +103,12 @@ public class NextSessionWorker(
         CancellationToken incomingCancellationToken)
     {
         _cancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(incomingCancellationToken);
+        using var checkpointCts = CancellationTokenSource.CreateLinkedTokenSource(_cancellationTokenSource.Token);
+        Task checkpointLoop = Task.CompletedTask;
+
+        DateTimeOffset sessionStart = DateTimeOffset.Now;
+        _lastTouch = sessionStart;
+        _lastCheckpoint = _lastTouch;
 
         try
         {
@@ -99,6 +116,30 @@ public class NextSessionWorker(
             _workingDirectory = fileSystem.Path.Combine(FileSystemLayout.TranscodeFolder, _channelNumber);
             _heartbeatFileName = fileSystem.Path.Combine(_workingDirectory, ".heartbeat");
 
+            Option<PlayoutModeViewModel> maybePlayout = await Mediator.Send(
+                new GetPlayoutModeByChannelNumber(_channelNumber),
+                _cancellationTokenSource.Token);
+
+            foreach (PlayoutModeViewModel playout in maybePlayout)
+            {
+                _channelPlayoutMode = playout.PlayoutMode;
+
+                if (_channelPlayoutMode is ChannelPlayoutMode.OnDemand)
+                {
+                    checkpointLoop = CheckpointLoop(checkpointCts.Token);
+
+                    await Mediator.Send(
+                        new TimeShiftOnDemandPlayout(playout.PlayoutId, sessionStart, true),
+                        _cancellationTokenSource.Token);
+
+                    // next reads serialized playout files rather than the database, so ensure it
+                    // sees the time-shifted items before starting the channel process
+                    await Mediator.Send(
+                        new SyncNextPlayout(_channelNumber),
+                        _cancellationTokenSource.Token);
+                }
+            }
+
             List<string> arguments = ["run", "--output-folder", _workingDirectory, "--number", channelNumber, "-"];
 
             string defaultOverlayFile = fileSystem.Path.Combine(
@@ -150,6 +191,25 @@ public class NextSessionWorker(
         }
         finally
         {
+            await checkpointCts.CancelAsync();
+            try
+            {
+                await checkpointLoop;
+            }
+            catch (OperationCanceledException)
+            {
+                // do nothing
+            }
+
+            try
+            {
+                await UpdateOnDemandCheckpoint(CancellationToken.None);
+            }
+            catch
+            {
+                // do nothing
+            }
+
             try
             {
                 localFileSystem.EmptyFolder(_workingDirectory);
@@ -171,4 +231,35 @@ public class NextSessionWorker(
             await Task.Delay(TimeSpan.FromMilliseconds(100), cancellationToken);
         }
     }
+
+    private async Task CheckpointLoop(CancellationToken cancellationToken)
+    {
+        using var timer = new PeriodicTimer(_checkpointThreshold);
+        while (await timer.WaitForNextTickAsync(cancellationToken))
+        {
+            try
+            {
+                if (_lastTouch > _lastCheckpoint)
+                {
+                    await UpdateOnDemandCheckpoint(cancellationToken);
+                }
+            }
+            catch (Exception ex)
+            {
+                logger.LogWarning(ex, "Failed to update on demand checkpoint for channel {Channel}", _channelNumber);
+            }
+        }
+    }
+
+    private async Task UpdateOnDemandCheckpoint(CancellationToken cancellationToken)
+    {
+        if (_channelPlayoutMode is ChannelPlayoutMode.OnDemand)
+        {
+            await Mediator.Send(
+                new UpdateOnDemandCheckpoint(_channelNumber, _lastTouch),
+                cancellationToken);
+        }
+
+        _lastCheckpoint = _lastTouch;
+    }
 }