feat: add reliable OpenList segment uploads

This commit is contained in:
2026-08-01 19:27:56 +08:00
parent 12ba62e2a0
commit 9091c1abc7
30 changed files with 3910 additions and 35 deletions
@@ -10,6 +10,7 @@
<ItemGroup>
<PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.11.1" />
<PackageReference Include="Microsoft.EntityFrameworkCore.InMemory" Version="8.0.4" />
<PackageReference Include="xunit" Version="2.9.2" />
<PackageReference Include="xunit.runner.visualstudio" Version="2.8.2">
<PrivateAssets>all</PrivateAssets>
@@ -0,0 +1,615 @@
using System.Net;
using System.Text;
using System.Text.Json;
using LiveRecorder.Application.Abstractions.Logging;
using LiveRecorder.Application.Abstractions.Settings;
using LiveRecorder.Application.Models.Logs;
using LiveRecorder.Application.Models.Settings;
using LiveRecorder.Domain.Entities;
using LiveRecorder.Domain.Enums;
using LiveRecorder.Infrastructure.Persistence;
using LiveRecorder.Infrastructure.Services;
using Microsoft.EntityFrameworkCore;
using Microsoft.EntityFrameworkCore.Storage;
namespace LiveRecorder.Tests;
public sealed class OpenListUploadTests
{
[Theory]
[InlineData("https://openlist.example.com/", "https://openlist.example.com")]
[InlineData("https://openlist.example.com/base/dav/archive/file", "https://openlist.example.com/base")]
[InlineData("https://openlist.example.com/dav", "https://openlist.example.com")]
[InlineData("https://openlist.example.com/davinci", "https://openlist.example.com/davinci")]
public void NormalizeBaseUrl_HandlesWebDavUrlsWithoutTruncatingOrdinarySegments(string input, string expected)
{
Assert.Equal(expected, OpenListClient.NormalizeBaseUrl(input));
}
[Fact]
public void OpenListPaths_PreserveUnicodeAndRejectTraversal()
{
Assert.Equal(
"/归档/Douyin/2026/08/01/主播名/分片.mp4",
OpenListClient.CombinePath("/归档/", "Douyin/2026/08/01/主播名/分片.mp4"));
Assert.Throws<InvalidOperationException>(() => OpenListClient.NormalizePath("/归档/../其他"));
Assert.Throws<InvalidOperationException>(() => OpenListClient.CombinePath("/归档", "../其他"));
}
[Fact]
public async Task EnsureDirectory_ReusesExistingLevelsAndCreatesOnlyMissingLevels()
{
var existingDirectories = new HashSet<string>(StringComparer.Ordinal)
{
"/",
"/归档",
"/归档/Douyin"
};
var createdDirectories = new List<string>();
var handler = new StubHttpMessageHandler(async request =>
{
if (request.RequestUri!.AbsolutePath.EndsWith("/api/auth/login", StringComparison.Ordinal))
{
return JsonResponse("""{"code":200,"message":"success","data":{"token":"test-token"}}""");
}
var body = await request.Content!.ReadAsStringAsync();
var path = JsonDocument.Parse(body).RootElement.GetProperty("path").GetString()!;
if (request.RequestUri.AbsolutePath.EndsWith("/api/fs/get", StringComparison.Ordinal))
{
return existingDirectories.Contains(path)
? JsonResponse("""{"code":200,"message":"success","data":{"name":"dir","size":0,"is_dir":true,"hash_info":{}}}""")
: JsonResponse("""{"code":500,"message":"object not found","data":null}""");
}
if (request.RequestUri.AbsolutePath.EndsWith("/api/fs/list", StringComparison.Ordinal))
{
Assert.True(JsonDocument.Parse(body).RootElement.GetProperty("refresh").GetBoolean());
return JsonResponse("""{"code":200,"message":"success","data":{"content":[],"write":true}}""");
}
Assert.EndsWith("/api/fs/mkdir", request.RequestUri.AbsolutePath, StringComparison.Ordinal);
createdDirectories.Add(path);
existingDirectories.Add(path);
return JsonResponse("""{"code":200,"message":"success","data":null}""");
});
var client = CreateClient(handler);
await client.EnsureDirectoryAsync(Connection(), "/归档/Douyin/2026/08/01/主播名");
Assert.Equal(
[
"/归档/Douyin/2026",
"/归档/Douyin/2026/08",
"/归档/Douyin/2026/08/01",
"/归档/Douyin/2026/08/01/主播名"
],
createdDirectories);
}
[Fact]
public async Task TryGetObject_RefreshesParentDirectoryAfterCacheMiss()
{
var getCount = 0;
var refreshedPath = string.Empty;
var handler = new StubHttpMessageHandler(async request =>
{
if (request.RequestUri!.AbsolutePath.EndsWith("/api/auth/login", StringComparison.Ordinal))
{
return JsonResponse("""{"code":200,"message":"success","data":{"token":"test-token"}}""");
}
var body = JsonDocument.Parse(await request.Content!.ReadAsStringAsync()).RootElement;
if (request.RequestUri.AbsolutePath.EndsWith("/api/fs/list", StringComparison.Ordinal))
{
refreshedPath = body.GetProperty("path").GetString();
Assert.True(body.GetProperty("refresh").GetBoolean());
return JsonResponse("""{"code":200,"message":"success","data":{"content":[],"write":true}}""");
}
Assert.EndsWith("/api/fs/get", request.RequestUri.AbsolutePath, StringComparison.Ordinal);
getCount++;
return getCount == 1
? JsonResponse("""{"code":500,"message":"object not found","data":null}""")
: JsonResponse("""{"code":200,"message":"success","data":{"name":"large.flv","size":2792401155,"is_dir":false,"hash_info":{}}}""");
});
var client = CreateClient(handler);
var result = await client.TryGetObjectAsync(Connection(), "/archive/主播/large.flv");
Assert.NotNull(result);
Assert.Equal(2_792_401_155, result.Size);
Assert.Equal(2, getCount);
Assert.Equal("/archive/主播", refreshedPath);
}
[Fact]
public async Task CopyAndTaskPolling_UseOpenListV423Contract()
{
HttpMethod? copyMethod = null;
HttpMethod? taskMethod = null;
string? copyAuthorization = null;
JsonElement copyPayload = default;
var handler = new StubHttpMessageHandler(async request =>
{
var path = request.RequestUri!.AbsolutePath;
if (path.EndsWith("/api/auth/login", StringComparison.Ordinal))
{
return JsonResponse("""{"code":200,"message":"success","data":{"token":"test-token"}}""");
}
if (path.EndsWith("/api/fs/copy", StringComparison.Ordinal))
{
copyMethod = request.Method;
copyAuthorization = request.Headers.Authorization?.ToString();
copyPayload = JsonDocument.Parse(await request.Content!.ReadAsStringAsync()).RootElement.Clone();
return JsonResponse("""{"code":200,"message":"success","data":{"message":"created","tasks":[{"id":"copy-task-1","state":0,"progress":0}]}}""");
}
taskMethod = request.Method;
Assert.EndsWith("/api/task/copy/info", path, StringComparison.Ordinal);
Assert.Equal("copy-task-1", GetQueryValue(request.RequestUri.Query, "tid"));
return JsonResponse("""{"code":200,"message":"success","data":{"id":"copy-task-1","state":2,"progress":100,"status":"done","error":""}}""");
});
var client = CreateClient(handler);
var copy = await client.CopyFileAsync(
Connection(),
"/本地挂载/Douyin/主播/分片.mp4",
"/移动云盘/归档/Douyin/主播/分片.mp4");
var task = await client.TryGetCopyTaskAsync(Connection(), copy.TaskIds.Single());
Assert.Equal(HttpMethod.Post, copyMethod);
Assert.Equal(HttpMethod.Post, taskMethod);
Assert.Equal("test-token", copyAuthorization);
Assert.Equal("/本地挂载/Douyin/主播", copyPayload.GetProperty("src_dir").GetString());
Assert.Equal("/移动云盘/归档/Douyin/主播", copyPayload.GetProperty("dst_dir").GetString());
Assert.Equal("分片.mp4", copyPayload.GetProperty("names")[0].GetString());
Assert.False(copyPayload.GetProperty("overwrite").GetBoolean());
Assert.NotNull(task);
Assert.Equal(2, task.State);
Assert.Equal(100, task.Progress);
}
[Fact]
public void UploadJob_PersistsArtifactProgressAndRetryState()
{
var now = DateTimeOffset.Parse("2026-08-01T10:00:00+08:00");
var job = new RecordUploadJob(
Guid.NewGuid(),
"https://openlist.example.com",
"/source/video.mp4",
"/archive/video.mp4",
100,
"/source/video.xml",
"/archive/video.xml",
20,
deleteLocalFilesAfterUpload: true,
now);
Assert.Equal(RecordArtifactUploadStatus.Queued, job.Status);
job.BeginAttempt(now.AddSeconds(1));
job.TrackExternalTask("task-1", "copy", job.CalculateOverallProgress(50), now.AddSeconds(2));
Assert.Equal(RecordArtifactUploadStatus.Uploading, job.Status);
Assert.Equal(1, job.AttemptCount);
Assert.Equal(41.67, job.ProgressPercent, precision: 2);
job.CompleteCurrentArtifact(now.AddSeconds(3));
Assert.Equal(RecordUploadArtifactStage.Danmaku, job.CurrentArtifact);
Assert.Equal(83.33, job.ProgressPercent, precision: 2);
var retryAt = now.AddMinutes(1);
job.ScheduleRetry("temporary failure", retryAt, now.AddSeconds(4), clearExternalTask: true);
Assert.Equal(RecordArtifactUploadStatus.WaitingRetry, job.Status);
Assert.Equal(retryAt, job.NextAttemptAt);
Assert.Null(job.ExternalTaskId);
job.BeginAttempt(retryAt);
job.CompleteCurrentArtifact(retryAt.AddSeconds(1));
job.MarkSucceeded(retryAt.AddSeconds(2));
Assert.Equal(RecordArtifactUploadStatus.Succeeded, job.Status);
Assert.Equal(RecordUploadArtifactStage.Completed, job.CurrentArtifact);
Assert.Equal(100, job.ProgressPercent);
}
[Fact]
public void UploadJob_RetainsExternalTaskAcrossTransientRetry()
{
var now = DateTimeOffset.Parse("2026-08-01T10:00:00+08:00");
var job = new RecordUploadJob(
Guid.NewGuid(),
"https://openlist.example.com",
"/source/video.mp4",
"/archive/video.mp4",
100,
null,
null,
null,
deleteLocalFilesAfterUpload: false,
now);
job.BeginAttempt(now.AddSeconds(1));
job.TrackExternalTask("copy-task-1", "copy", 25, now.AddSeconds(2));
job.StartVerification(now.AddSeconds(3));
job.ScheduleRetry(
"temporary polling failure",
now.AddMinutes(1),
now.AddSeconds(4),
clearExternalTask: false);
job.BeginAttempt(now.AddMinutes(1));
Assert.Equal(RecordArtifactUploadStatus.Uploading, job.Status);
Assert.Equal(2, job.AttemptCount);
Assert.Equal("copy-task-1", job.ExternalTaskId);
Assert.Equal("copy", job.ExternalTaskType);
Assert.Equal(now.AddSeconds(2), job.ExternalTaskStartedAt);
Assert.Equal(now.AddSeconds(3), job.VerificationStartedAt);
}
[Fact]
public async Task Queue_MapsOutputRelativePathAndTreatsMatchingTargetAsIdempotentSuccess()
{
await using var fixture = await QueueFixture.CreateAsync();
var queued = await fixture.Queue.EnqueueAsync(fixture.RecordTaskId);
var job = await fixture.Context.RecordUploadJobs.AsNoTracking().SingleAsync();
Assert.True(queued.Success);
Assert.Equal(RecordArtifactUploadStatus.Queued, queued.UploadStatus);
Assert.Equal("/source/Douyin/2026/08/01/主播/segment.mp4", job.SourceVideoPath);
Assert.Equal("/destination/Douyin/2026/08/01/主播/segment.mp4", job.TargetVideoPath);
fixture.OpenList.Objects[job.TargetVideoPath] = FileObject("segment.mp4", job.VideoSizeBytes);
Assert.True(await fixture.Queue.ProcessNextAsync());
fixture.Context.ChangeTracker.Clear();
var result = await fixture.Context.RecordResults.SingleAsync();
var completedJob = await fixture.Context.RecordUploadJobs.SingleAsync();
Assert.Equal(RecordArtifactUploadStatus.Succeeded, result.UploadStatus);
Assert.Equal(RecordArtifactUploadStatus.Succeeded, completedJob.Status);
Assert.Equal(job.TargetVideoPath, result.RemoteVideoPath);
Assert.Empty(fixture.OpenList.CopyRequests);
var repeated = await fixture.Queue.EnqueueAsync(fixture.RecordTaskId);
Assert.True(repeated.Success);
Assert.Equal(RecordArtifactUploadStatus.Succeeded, repeated.UploadStatus);
Assert.Equal(1, await fixture.Context.RecordUploadJobs.CountAsync());
}
[Fact]
public async Task Queue_RejectsSameNameConflictWithoutOverwriting()
{
await using var fixture = await QueueFixture.CreateAsync();
await fixture.Queue.EnqueueAsync(fixture.RecordTaskId);
var job = await fixture.Context.RecordUploadJobs.AsNoTracking().SingleAsync();
fixture.OpenList.Objects[job.TargetVideoPath] = FileObject("segment.mp4", job.VideoSizeBytes + 1);
Assert.True(await fixture.Queue.ProcessNextAsync());
fixture.Context.ChangeTracker.Clear();
var failedJob = await fixture.Context.RecordUploadJobs.SingleAsync();
var result = await fixture.Context.RecordResults.SingleAsync();
Assert.Equal(RecordArtifactUploadStatus.Failed, failedJob.Status);
Assert.Equal(RecordArtifactUploadStatus.Failed, result.UploadStatus);
Assert.Contains("内容不一致", failedJob.ErrorMessage);
Assert.Empty(fixture.OpenList.CopyRequests);
}
[Fact]
public async Task Queue_QueuesTwoGiBFileUsingServerSideCopy()
{
const long twoGiB = 2L * 1024 * 1024 * 1024;
await using var fixture = await QueueFixture.CreateAsync();
var result = await fixture.Context.RecordResults.AsNoTracking().SingleAsync();
await using (var stream = new FileStream(result.FilePath, FileMode.Open, FileAccess.Write, FileShare.Read))
{
stream.SetLength(twoGiB);
}
var queued = await fixture.Queue.EnqueueAsync(fixture.RecordTaskId);
var job = await fixture.Context.RecordUploadJobs.AsNoTracking().SingleAsync();
fixture.OpenList.Objects[job.SourceVideoPath] = FileObject("segment.mp4", twoGiB);
Assert.True(queued.Success);
Assert.Equal(twoGiB, job.VideoSizeBytes);
Assert.True(await fixture.Queue.ProcessNextAsync());
Assert.Equal([(job.SourceVideoPath, job.TargetVideoPath)], fixture.OpenList.CopyRequests);
}
[Fact]
public async Task Queue_ResumesPersistedOpenListTaskAfterDbContextRestart()
{
await using var fixture = await QueueFixture.CreateAsync();
await fixture.Queue.EnqueueAsync(fixture.RecordTaskId);
var job = await fixture.Context.RecordUploadJobs.AsNoTracking().SingleAsync();
fixture.OpenList.Objects[job.SourceVideoPath] = FileObject("segment.mp4", job.VideoSizeBytes);
fixture.OpenList.NextCopyResult = new OpenListCopyResult(["copy-task-1"]);
Assert.True(await fixture.Queue.ProcessNextAsync());
fixture.Context.ChangeTracker.Clear();
var runningJob = await fixture.Context.RecordUploadJobs.AsNoTracking().SingleAsync();
Assert.Equal("copy-task-1", runningJob.ExternalTaskId);
Assert.Equal(RecordArtifactUploadStatus.Uploading, runningJob.Status);
await fixture.ReopenContextAsync();
fixture.OpenList.Tasks["copy-task-1"] = new OpenListTaskInfo("copy-task-1", 2, 100, "done", null);
fixture.OpenList.Objects[job.TargetVideoPath] = FileObject("segment.mp4", job.VideoSizeBytes);
Assert.True(await fixture.Queue.ProcessNextAsync());
fixture.Context.ChangeTracker.Clear();
var completedJob = await fixture.Context.RecordUploadJobs.SingleAsync();
var result = await fixture.Context.RecordResults.SingleAsync();
Assert.Equal(RecordArtifactUploadStatus.Succeeded, completedJob.Status);
Assert.Equal(1, completedJob.AttemptCount);
Assert.Equal(RecordArtifactUploadStatus.Succeeded, result.UploadStatus);
Assert.Single(fixture.OpenList.CopyRequests);
}
private static OpenListClient CreateClient(HttpMessageHandler handler) =>
new(new StubHttpClientFactory(handler));
private static OpenListConnectionRequest Connection() => new()
{
BaseUrl = "https://openlist.example.com",
Username = "tester",
Password = "secret"
};
private static OpenListObjectInfo FileObject(string name, long size) =>
new(name, size, false, new Dictionary<string, string>());
private static HttpResponseMessage JsonResponse(string json) => new(HttpStatusCode.OK)
{
Content = new StringContent(json, Encoding.UTF8, "application/json")
};
private static string? GetQueryValue(string query, string key)
{
foreach (var pair in query.TrimStart('?').Split('&', StringSplitOptions.RemoveEmptyEntries))
{
var parts = pair.Split('=', 2);
if (Uri.UnescapeDataString(parts[0]) == key)
{
return parts.Length > 1 ? Uri.UnescapeDataString(parts[1]) : string.Empty;
}
}
return null;
}
private sealed class StubHttpClientFactory : IHttpClientFactory
{
private readonly HttpMessageHandler _handler;
public StubHttpClientFactory(HttpMessageHandler handler)
{
_handler = handler;
}
public HttpClient CreateClient(string name) => new(_handler, disposeHandler: false);
}
private sealed class StubHttpMessageHandler : HttpMessageHandler
{
private readonly Func<HttpRequestMessage, Task<HttpResponseMessage>> _send;
public StubHttpMessageHandler(Func<HttpRequestMessage, Task<HttpResponseMessage>> send)
{
_send = send;
}
protected override Task<HttpResponseMessage> SendAsync(
HttpRequestMessage request,
CancellationToken cancellationToken) => _send(request);
}
private sealed class QueueFixture : IAsyncDisposable
{
private readonly DbContextOptions<LiveRecorderDbContext> _options;
private readonly FixedSettingsService _settingsService;
private readonly string _temporaryRoot;
private QueueFixture(
DbContextOptions<LiveRecorderDbContext> options,
LiveRecorderDbContext context,
FixedSettingsService settingsService,
FakeOpenListClient openList,
string temporaryRoot,
Guid recordTaskId)
{
_options = options;
Context = context;
_settingsService = settingsService;
OpenList = openList;
_temporaryRoot = temporaryRoot;
RecordTaskId = recordTaskId;
Queue = CreateQueue(context);
}
public LiveRecorderDbContext Context { get; private set; }
public OpenListUploadQueueService Queue { get; private set; }
public FakeOpenListClient OpenList { get; }
public Guid RecordTaskId { get; }
public static async Task<QueueFixture> CreateAsync()
{
var temporaryRoot = Path.Combine(Path.GetTempPath(), $"live-recorder-openlist-{Guid.NewGuid():N}");
var recordDirectory = Path.Combine(temporaryRoot, "Douyin", "2026", "08", "01", "主播");
Directory.CreateDirectory(recordDirectory);
var videoPath = Path.Combine(recordDirectory, "segment.mp4");
await File.WriteAllBytesAsync(videoPath, Encoding.UTF8.GetBytes("video-content"));
var databaseRoot = new InMemoryDatabaseRoot();
var options = new DbContextOptionsBuilder<LiveRecorderDbContext>()
.UseInMemoryDatabase($"openlist-{Guid.NewGuid():N}", databaseRoot)
.Options;
var context = new LiveRecorderDbContext(options);
await context.Database.EnsureCreatedAsync();
var now = DateTimeOffset.UtcNow;
var liveRoom = new LiveRoom(
LivePlatformType.Douyin,
"https://live.example/room",
"room-1",
"https://live.example/room",
now);
var session = new RecordSession(
liveRoom.Id,
"origin",
RecordOutputFormat.Mp4,
RecordSaveMode.Segmented,
now);
var task = new RecordTask(
liveRoom.Id,
session.Id,
1,
"origin",
RecordOutputFormat.Mp4,
now);
var result = new RecordResult(
task.Id,
videoPath,
new FileInfo(videoPath).Length,
60,
null,
0,
RecordTaskStatus.Completed,
null,
now);
context.AddRange(liveRoom, session, task, result);
await context.SaveChangesAsync();
var settings = new SystemSettingsDto
{
OutputRoot = temporaryRoot,
EnableFileUpload = true,
EnableAutoUpload = true,
UploadTarget = UploadTargetType.OpenList,
OpenListUpload = new OpenListUploadSettingsDto
{
BaseUrl = "https://openlist.example.com",
Username = "tester",
Password = "secret",
SourcePath = "/source",
DestinationPath = "/destination"
}
};
var settingsService = new FixedSettingsService(settings);
var openList = new FakeOpenListClient();
return new QueueFixture(
options,
context,
settingsService,
openList,
temporaryRoot,
task.Id);
}
public async Task ReopenContextAsync()
{
await Context.DisposeAsync();
Context = new LiveRecorderDbContext(_options);
Queue = CreateQueue(Context);
}
public async ValueTask DisposeAsync()
{
await Context.DisposeAsync();
if (Directory.Exists(_temporaryRoot))
{
Directory.Delete(_temporaryRoot, recursive: true);
}
}
private OpenListUploadQueueService CreateQueue(LiveRecorderDbContext context) =>
new(context, _settingsService, OpenList, new NullSystemLogService());
}
private sealed class FixedSettingsService : ISystemSettingsService
{
private readonly SystemSettingsDto _settings;
public FixedSettingsService(SystemSettingsDto settings)
{
_settings = settings;
}
public Task<SystemSettingsDto> GetAsync(CancellationToken cancellationToken = default) =>
Task.FromResult(_settings);
public Task<SystemSettingsDto> UpdateAsync(
UpdateSystemSettingsRequest request,
CancellationToken cancellationToken = default) => throw new NotSupportedException();
}
private sealed class NullSystemLogService : ISystemLogService
{
public Task WriteAsync(
SystemLogLevel level,
string category,
string message,
string? detail = null,
Guid? liveRoomId = null,
Guid? recordSessionId = null,
Guid? recordTaskId = null,
CancellationToken cancellationToken = default) => Task.CompletedTask;
public Task<IReadOnlyList<SystemLogDto>> ListAsync(
Guid? liveRoomId = null,
Guid? recordSessionId = null,
Guid? recordTaskId = null,
SystemLogLevel? level = null,
string? content = null,
int take = 200,
CancellationToken cancellationToken = default) =>
Task.FromResult<IReadOnlyList<SystemLogDto>>([]);
}
private sealed class FakeOpenListClient : IOpenListClient
{
public Dictionary<string, OpenListObjectInfo> Objects { get; } = new(StringComparer.Ordinal);
public Dictionary<string, OpenListTaskInfo> Tasks { get; } = new(StringComparer.Ordinal);
public List<(string Source, string Target)> CopyRequests { get; } = [];
public OpenListCopyResult NextCopyResult { get; set; } = new([]);
public Task<OpenListConnectionTestDto> TestConnectionAsync(
OpenListConnectionRequest connection,
CancellationToken cancellationToken = default) => throw new NotSupportedException();
public Task<OpenListDirectoryListDto> ListDirectoriesAsync(
OpenListDirectoryRequest request,
CancellationToken cancellationToken = default) => throw new NotSupportedException();
public Task EnsureDirectoryAsync(
OpenListConnectionRequest connection,
string path,
CancellationToken cancellationToken = default) => Task.CompletedTask;
public Task<OpenListObjectInfo?> TryGetObjectAsync(
OpenListConnectionRequest connection,
string path,
CancellationToken cancellationToken = default) =>
Task.FromResult(Objects.GetValueOrDefault(path));
public Task<OpenListCopyResult> CopyFileAsync(
OpenListConnectionRequest connection,
string sourcePath,
string targetPath,
CancellationToken cancellationToken = default)
{
CopyRequests.Add((sourcePath, targetPath));
return Task.FromResult(NextCopyResult);
}
public Task<OpenListTaskInfo?> TryGetCopyTaskAsync(
OpenListConnectionRequest connection,
string taskId,
CancellationToken cancellationToken = default) =>
Task.FromResult(Tasks.GetValueOrDefault(taskId));
}
}