feat(server): implement sealed share lifecycle

Signed-off-by: JustMe <mr.toing@outlook.com>
This commit is contained in:
justme committed 2026-10-02 00:59:56 +08:00
1 parent e02e024605
commit 42b93ae2d8
45 files changed
+6327

No files matched your search

+65
View File
@@ -0,0 +1,65 @@
using System.Text.Json.Serialization;
using WebDrop.Core;
namespace WebDrop.Api;
public sealed class ErrorBody
{
[JsonPropertyName("error")] public required string Error { get; init; }
}
public sealed class GoneLogBody
{
[JsonPropertyName("error")] public string Error { get; init; } = "gone";
[JsonPropertyName("oldest_seq")] public required long OldestSeq { get; init; }
}
public sealed class ServerInfoBody
{
[JsonPropertyName("protocol")] public int Protocol { get; init; } = 1;
[JsonPropertyName("captcha")] public required CaptchaBody Captcha { get; init; }
[JsonPropertyName("transports")] public required TransportsBody Transports { get; init; }
[JsonPropertyName("limits")] public required LimitsBody Limits { get; init; }
}
/// <summary>告诉前端要不要出示验证码、widget 该连哪里(Cap 的 data-cap-api-endpoint)。</summary>
public sealed class CaptchaBody
{
[JsonPropertyName("provider")] public required string Provider { get; init; }
[JsonPropertyName("enabled")] public required bool Enabled { get; init; }
[JsonPropertyName("api_endpoint")] public string? ApiEndpoint { get; init; }
}
public sealed class TransportsBody
{
[JsonPropertyName("websocket")] public bool WebSocket { get; init; } = true;
[JsonPropertyName("sse")] public bool Sse { get; init; } = true;
[JsonPropertyName("http3")] public required bool Http3 { get; init; }
}
public sealed class LimitsBody
{
[JsonPropertyName("max_payload_bytes")] public required int MaxPayloadBytes { get; init; }
[JsonPropertyName("max_frames_per_item")] public required int MaxFramesPerItem { get; init; }
[JsonPropertyName("max_channel_bytes")] public required long MaxChannelBytes { get; init; }
[JsonPropertyName("max_seats")] public required int MaxSeats { get; init; }
[JsonPropertyName("max_ttl_seconds")] public required int MaxTtlSeconds { get; init; }
}
[JsonSerializable(typeof(ErrorBody))]
[JsonSerializable(typeof(GoneLogBody))]
[JsonSerializable(typeof(ServerInfoBody))]
[JsonSerializable(typeof(ChannelInfo))]
[JsonSerializable(typeof(ShareInfo))]
[JsonSerializable(typeof(CreateShareRequest))]
[JsonSerializable(typeof(ShareSettingsPatch))]
[JsonSerializable(typeof(FinishShareRequest))]
[JsonSerializable(typeof(CreateChannelRequest))]
[JsonSerializable(typeof(ChannelSettingsPatch))]
[JsonSerializable(typeof(SeatJoined))]
[JsonSerializable(typeof(List<SeatInfo>))]
[JsonSerializable(typeof(AppendResult))]
[JsonSerializable(typeof(SiteverifyRequest))]
[JsonSerializable(typeof(SiteverifyResponse))]
[JsonSourceGenerationOptions(DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull)]
public sealed partial class ApiJson : JsonSerializerContext;
@@ -0,0 +1,70 @@
using System.Text;
using System.Text.Json;
using System.Text.Json.Serialization;
using WebDrop.Core;
namespace WebDrop.Api;
public sealed class SiteverifyRequest
{
[JsonPropertyName("secret")] public required string Secret { get; init; }
[JsonPropertyName("response")] public required string Response { get; init; }
}
public sealed class SiteverifyResponse
{
[JsonPropertyName("success")] public bool Success { get; init; }
}
/// <summary>
/// 用 Cap(https://github.com/tiagozip/cap)校验创建频道的验证码。
/// 工作量证明的出题、求解、验证全部由 Cap 完成,本项目不实现任何 PoW。
///
/// 流程:浏览器上的 Cap widget 直接向 Cap 取挑战、求解、换得 token,随创建请求放进 X-WD-Cap;
/// 这里只做一件事:把 token 交给 Cap 的 siteverify(与 reCAPTCHA / hCaptcha 兼容的接口)确认。
/// </summary>
public sealed class CapCaptchaVerifier(HttpClient http, CaptchaOptions options, ILogger<CapCaptchaVerifier> logger) : ICaptchaVerifier
{
// token 来自客户端,先限制长度,避免把任意大的内容转发给第三方
private const int MaxTokenLength = 2048;
public bool Enabled => true;
public async Task VerifyAsync(string? token, CancellationToken ct)
{
if (string.IsNullOrEmpty(token) || token.Length > MaxTokenLength)
throw new WebDropException(ErrorCode.CaptchaRequired);
var body = JsonSerializer.Serialize(
new SiteverifyRequest { Secret = options.Secret, Response = token }, ApiJson.Default.SiteverifyRequest);
try
{
using var request = new HttpRequestMessage(HttpMethod.Post, options.SiteverifyUrl)
{
Content = new StringContent(body, Encoding.UTF8, "application/json"),
};
using var response = await http.SendAsync(request, ct).ConfigureAwait(false);
// 非 2xx(站点密钥写错、Cap 出错等)是我们这边的配置或依赖问题,不是用户没通过验证
if (!response.IsSuccessStatusCode)
{
logger.LogError("Cap siteverify 返回了 {Status}", (int)response.StatusCode);
throw new WebDropException(ErrorCode.CaptchaUnavailable);
}
var result = await JsonSerializer.DeserializeAsync(
await response.Content.ReadAsStreamAsync(ct).ConfigureAwait(false),
ApiJson.Default.SiteverifyResponse, ct).ConfigureAwait(false);
if (result is not { Success: true }) throw new WebDropException(ErrorCode.CaptchaRequired);
}
catch (Exception e) when (e is HttpRequestException or TaskCanceledException or JsonException or InvalidOperationException
&& !ct.IsCancellationRequested)
{
// 只记录异常类型:不记录 token、密钥或响应内容
logger.LogError("Cap siteverify 调用失败:{Type}", e.GetType().Name);
throw new WebDropException(ErrorCode.CaptchaUnavailable);
}
}
}
+104
View File
@@ -0,0 +1,104 @@
using WebDrop.Core;
namespace WebDrop.Api;
/// <summary>
/// 进程内的订阅枢纽。只传递「有变化」的信号,订阅者被唤醒后自己从日志里按游标读取,
/// 所以不存在回放与实时之间漏帧的竞态,也不需要为慢客户端缓冲数据。
/// 因此只支持单实例部署(见 architecture.md)。
/// </summary>
public sealed class ChannelHub : IChannelEvents
{
private readonly object _gate = new();
private readonly Dictionary<Id16, Signal> _signals = [];
public Subscription Subscribe(Id16 id)
{
lock (_gate)
{
if (!_signals.TryGetValue(id, out var signal)) _signals[id] = signal = new Signal();
signal.Refs++;
return new Subscription(this, id, signal);
}
}
public void Advanced(Id16 channelId, long seq) => Publish(channelId, s => s.Seq = Math.Max(s.Seq, seq));
public void SettingsChanged(Id16 channelId) => Publish(channelId, s => s.SettingsVersion++);
public void SeatsChanged(Id16 channelId) => Publish(channelId, s => s.SeatsVersion++);
public void ChannelDeleted(Id16 channelId) => Publish(channelId, s => s.Deleted = true);
private void Publish(Id16 id, Action<Signal> mutate)
{
TaskCompletionSource? toComplete;
lock (_gate)
{
if (!_signals.TryGetValue(id, out var signal)) return; // 没有订阅者,无需通知
mutate(signal);
toComplete = signal.Replace();
}
toComplete.TrySetResult(); // 在锁外完成,避免续体在锁内运行
}
private void Release(Id16 id, Signal signal)
{
lock (_gate)
{
if (--signal.Refs == 0) _signals.Remove(id);
}
}
internal sealed class Signal
{
private TaskCompletionSource _tcs = New();
public int Refs;
public long Seq;
public int SettingsVersion;
public int SeatsVersion;
public bool Deleted;
public Task Changed => _tcs.Task;
public TaskCompletionSource Replace()
{
var old = _tcs;
_tcs = New();
return old;
}
private static TaskCompletionSource New() => new(TaskCreationOptions.RunContinuationsAsynchronously);
}
/// <summary>
/// 用法:先取 <see cref="Changed"/>,再读状态,最后等待取到的那个任务。
/// 取任务之后发生的任何发布都会让它完成,不会丢通知。
/// </summary>
public sealed class Subscription : IDisposable
{
private readonly ChannelHub _hub;
private readonly Id16 _id;
private readonly Signal _signal;
private int _disposed;
internal Subscription(ChannelHub hub, Id16 id, Signal signal)
{
_hub = hub;
_id = id;
_signal = signal;
}
public Task Changed
{
get { lock (_hub._gate) return _signal.Changed; }
}
public (int Settings, int Seats, bool Deleted) State
{
get { lock (_hub._gate) return (_signal.SettingsVersion, _signal.SeatsVersion, _signal.Deleted); }
}
public void Dispose()
{
if (Interlocked.Exchange(ref _disposed, 1) == 0) _hub.Release(_id, _signal);
}
}
}
+409
View File
@@ -0,0 +1,409 @@
using System.Buffers;
using System.Diagnostics;
using System.Net.WebSockets;
using System.Text.Json;
using System.Text.Json.Serialization.Metadata;
using WebDrop.Core;
namespace WebDrop.Api;
public static class Endpoints
{
public const string CreatePolicy = "create";
private const string LastSeqHeader = "X-WD-Last-Seq";
private const string OctetStream = "application/octet-stream";
private static readonly TimeSpan MaxLongPoll = TimeSpan.FromSeconds(25);
public static void MapApi(this IEndpointRouteBuilder app, WebDropOptions options, ICaptchaVerifier captcha)
{
var api = app.MapGroup("/api/v1");
api.MapGet("/info", (ServerLimits limits) => Json(new ServerInfoBody
{
Captcha = new CaptchaBody
{
Provider = options.Captcha.Provider,
Enabled = captcha.Enabled,
ApiEndpoint = captcha.Enabled ? options.Captcha.ApiEndpoint : null,
},
Transports = new TransportsBody { Http3 = options.EnableHttp3 },
Limits = new LimitsBody
{
MaxPayloadBytes = limits.MaxPayloadBytes,
MaxFramesPerItem = limits.MaxFramesPerItem,
MaxChannelBytes = limits.MaxChannelBytes,
MaxSeats = limits.MaxSeats,
MaxTtlSeconds = limits.MaxTtlSeconds,
},
}, ApiJson.Default.ServerInfoBody));
// ───────── 频道 ─────────
api.MapPost("/channels", async (HttpContext ctx, ChannelService svc, CancellationToken ct) =>
{
var req = await ReadJson(ctx.Request, ApiJson.Default.CreateChannelRequest, ct);
var endpoint = RequestAuth.ParseEndpointHash(ctx.Request.Headers[RequestAuth.EndpointHeader]);
string token = ctx.Request.Headers[RequestAuth.CaptchaHeader].ToString();
await captcha.VerifyAsync(token.Length > 0 ? token : null, ct);
return Json(svc.CreateChannel(req, endpoint), ApiJson.Default.ChannelInfo, StatusCodes.Status201Created);
}).RequireRateLimiting(CreatePolicy);
api.MapGet("/channels/{id}", (string id, HttpContext ctx, ChannelService svc) =>
Json(svc.GetChannel(ParseId(id), RequestAuth.FromHeaders(ctx.Request, requireEndpoint: false)), ApiJson.Default.ChannelInfo));
api.MapPatch("/channels/{id}/settings", async (string id, HttpContext ctx, ChannelService svc, CancellationToken ct) =>
{
var patch = await ReadJson(ctx.Request, ApiJson.Default.ChannelSettingsPatch, ct);
var info = svc.UpdateSettings(ParseId(id), RequestAuth.FromHeaders(ctx.Request, requireEndpoint: false), patch);
return Json(info, ApiJson.Default.ChannelInfo);
});
api.MapDelete("/channels/{id}", (string id, HttpContext ctx, ChannelService svc) =>
{
svc.DeleteChannel(ParseId(id), RequestAuth.FromHeaders(ctx.Request, requireEndpoint: false));
return Results.NoContent();
});
// ───────── 席位 ─────────
api.MapPost("/channels/{id}/seats", (string id, HttpContext ctx, ChannelService svc) =>
Json(svc.Join(ParseId(id), RequestAuth.FromHeaders(ctx.Request, requireEndpoint: true)), ApiJson.Default.SeatJoined));
api.MapGet("/channels/{id}/seats", (string id, HttpContext ctx, ChannelService svc) =>
{
var seats = svc.ListSeats(ParseId(id), RequestAuth.FromHeaders(ctx.Request, requireEndpoint: false));
return Json([.. seats], ApiJson.Default.ListSeatInfo);
});
api.MapPut("/channels/{id}/seats/me", async (string id, HttpContext ctx, ChannelService svc, ServerLimits limits, CancellationToken ct) =>
{
var creds = RequestAuth.FromHeaders(ctx.Request, requireEndpoint: true);
byte[] label = await ReadBody(ctx.Request, limits.MaxSeatLabelBytes, ct);
svc.SetSeatLabel(ParseId(id), creds, label);
return Results.NoContent();
});
api.MapDelete("/channels/{id}/seats/{endpointHash}", (string id, string endpointHash, HttpContext ctx, ChannelService svc) =>
{
if (!Codec.TryDecodeExact(endpointHash, TokenHash.Size, out var hash)) throw new WebDropException(ErrorCode.BadRequest);
svc.ReleaseSeat(ParseId(id), RequestAuth.FromHeaders(ctx.Request, requireEndpoint: false), hash);
return Results.NoContent();
});
// ───────── 帧与日志 ─────────
api.MapPost("/channels/{id}/frames", async (string id, HttpContext ctx, ChannelService svc, ServerLimits limits, CancellationToken ct) =>
{
var channelId = ParseId(id);
var creds = RequestAuth.FromHeaders(ctx.Request, requireEndpoint: true);
long? maxDownloads = ParseMaxDownloads(ctx.Request.Headers["X-WD-Max-Downloads"].ToString());
string label = ctx.Request.Headers["X-WD-Label"].ToString();
byte[] body = await ReadBody(ctx.Request, Envelope.HeaderSize + limits.MaxPayloadBytes, ct);
var result = await svc.AppendFrameAsync(channelId, creds, body, maxDownloads, label.Length > 0 ? label : null, ct);
return Json(result, ApiJson.Default.AppendResult, result.Created ? StatusCodes.Status201Created : StatusCodes.Status200OK);
});
api.MapGet("/channels/{id}/log", async (string id, HttpContext ctx, ChannelService svc, ChannelHub hub, CancellationToken ct) =>
{
var channelId = ParseId(id);
var creds = RequestAuth.FromHeaders(ctx.Request, requireEndpoint: true);
long from = QueryLong(ctx.Request, "from", 0);
int limit = (int)Math.Clamp(QueryLong(ctx.Request, "limit", 200), 1, 1000);
var wait = TimeSpan.FromSeconds(Math.Clamp(QueryLong(ctx.Request, "wait", 0), 0, 25));
var page = await ReadLogWithWait(svc, hub, channelId, creds, from, limit, wait, ct);
ctx.Response.Headers[LastSeqHeader] = page.LastSeq.ToString();
var all = new byte[page.Records.Sum(r => r.Bytes.Length)];
int offset = 0;
foreach (var r in page.Records)
{
r.Bytes.CopyTo(all, offset);
offset += r.Bytes.Length;
}
return Results.Bytes(all, OctetStream);
});
api.MapGet("/channels/{id}/ws", async (string id, HttpContext ctx, ChannelService svc, ChannelHub hub) =>
{
if (!ctx.WebSockets.IsWebSocketRequest || !ctx.WebSockets.WebSocketRequestedProtocols.Contains("wd.v1"))
throw new WebDropException(ErrorCode.BadRequest);
var channelId = ParseId(id);
var creds = RequestAuth.FromSubprotocols(ctx.WebSockets.WebSocketRequestedProtocols);
long from = QueryLong(ctx.Request, "from", 0);
using var cts = CancellationTokenSource.CreateLinkedTokenSource(ctx.RequestAborted);
WebSocket? ws = null;
try
{
await LogStreamer.RunAsync(svc, hub, channelId, creds, from, async () =>
{
ws = await ctx.WebSockets.AcceptWebSocketAsync("wd.v1");
_ = WatchClose(ws, cts);
return new WebSocketSink(ws);
}, cts.Token);
}
finally
{
await CloseQuietly(ws);
}
return Results.Empty;
});
api.MapGet("/channels/{id}/events", async (string id, HttpContext ctx, ChannelService svc, ChannelHub hub) =>
{
var channelId = ParseId(id);
var creds = RequestAuth.FromHeaders(ctx.Request, requireEndpoint: true);
long from = QueryLong(ctx.Request, "from", -1);
if (from < 0)
from = long.TryParse(ctx.Request.Headers["Last-Event-ID"].ToString(), out long last) && last >= 0 ? last + 1 : 0;
await LogStreamer.RunAsync(svc, hub, channelId, creds, from, async () =>
{
ctx.Response.ContentType = "text/event-stream";
ctx.Response.Headers["X-Accel-Buffering"] = "no";
await ctx.Response.StartAsync();
return new SseSink(ctx.Response);
}, ctx.RequestAborted);
return Results.Empty;
});
// ───────── 条目 ─────────
api.MapGet("/channels/{id}/items/{itemId}/frames", async (string id, string itemId, HttpContext ctx, ChannelService svc, ChannelHub hub, CancellationToken ct) =>
{
var channelId = ParseId(id);
var item = ParseId(itemId);
var creds = RequestAuth.FromHeaders(ctx.Request, requireEndpoint: true);
long from = QueryLong(ctx.Request, "from", 0);
if (from is < 0 or > uint.MaxValue) throw new WebDropException(ErrorCode.BadRequest);
svc.StartItemDownload(channelId, creds, item, (uint)from);
ctx.Response.ContentType = OctetStream;
using var sub = hub.Subscribe(channelId);
uint next = (uint)from;
while (!ct.IsCancellationRequested)
{
var changed = sub.Changed;
var chunk = await svc.ReadItemFramesAsync(channelId, creds, item, next, 16, ct);
foreach (var frame in chunk.Frames) await ctx.Response.Body.WriteAsync(frame, ct);
next = chunk.NextIndex;
if (chunk.Complete) break;
if (chunk.Frames.Count > 0) continue;
// 条目还在上传:等新帧;长时间没有进展就结束本次响应,客户端用 from=next 续传(不重复计数)
if (!await WaitOrTimeout(changed, MaxLongPoll, ct)) break;
}
return Results.Empty;
});
api.MapDelete("/channels/{id}/items/{itemId}", (string id, string itemId, HttpContext ctx, ChannelService svc) =>
{
svc.DeleteItem(ParseId(id), RequestAuth.FromHeaders(ctx.Request, requireEndpoint: true), ParseId(itemId));
return Results.NoContent();
});
// ───────── 便捷分享(独立于频道,无席位和日志) ─────────
api.MapPost("/shares", async (HttpContext ctx, ShareService svc, CancellationToken ct) =>
{
var req = await ReadJson(ctx.Request, ApiJson.Default.CreateShareRequest, ct);
string token = ctx.Request.Headers[RequestAuth.CaptchaHeader].ToString();
await captcha.VerifyAsync(token.Length > 0 ? token : null, ct);
byte[] endpoint = RequestAuth.ParseEndpointHashOptional(ctx.Request.Headers[RequestAuth.EndpointHeader]);
return Json(svc.Create(req, endpoint.Length == 0 ? null : endpoint), ApiJson.Default.ShareInfo, StatusCodes.Status201Created);
}).RequireRateLimiting(CreatePolicy);
api.MapPatch("/shares/{id}/settings", async (string id, HttpContext ctx, ShareService svc, CancellationToken ct) =>
{
var patch = await ReadJson(ctx.Request, ApiJson.Default.ShareSettingsPatch, ct);
return Json(svc.UpdateSettings(ParseId(id), RequestAuth.ShareFromHeaders(ctx.Request, requireEndpoint: false), patch), ApiJson.Default.ShareInfo);
});
api.MapPost("/shares/{id}/frames", async (string id, HttpContext ctx, ShareService svc, ServerLimits limits, CancellationToken ct) =>
{
var creds = RequestAuth.ShareFromHeaders(ctx.Request, requireEndpoint: false);
long? maxDownloads = ParseMaxDownloads(ctx.Request.Headers["X-WD-Max-Downloads"].ToString());
byte[] body = await ReadBody(ctx.Request, Envelope.HeaderSize + limits.MaxPayloadBytes, ct);
var result = await svc.AppendFrameAsync(ParseId(id), creds, body, maxDownloads, ct);
return Json(result, ApiJson.Default.AppendResult, result.Created ? StatusCodes.Status201Created : StatusCodes.Status200OK);
});
api.MapDelete("/shares/{id}/items/{itemId}", (string id, string itemId, HttpContext ctx, ShareService svc) =>
{
svc.DeleteItem(ParseId(id), RequestAuth.ShareFromHeaders(ctx.Request, requireEndpoint: false), ParseId(itemId));
return Results.NoContent();
});
api.MapDelete("/shares/{id}", (string id, HttpContext ctx, ShareService svc) =>
{
svc.DeleteDraft(ParseId(id), RequestAuth.ShareFromHeaders(ctx.Request, requireEndpoint: false));
return Results.NoContent();
});
api.MapPost("/shares/{id}/finish", async (string id, HttpContext ctx, ShareService svc, CancellationToken ct) =>
{
var req = await ReadJson(ctx.Request, ApiJson.Default.FinishShareRequest, ct);
return Json(svc.Finish(ParseId(id), RequestAuth.ShareFromHeaders(ctx.Request, requireEndpoint: false), req.ManifestItemId), ApiJson.Default.ShareInfo);
});
api.MapGet("/shares/{id}", (string id, HttpContext ctx, ShareService svc) =>
Json(svc.GetInfo(ParseId(id), RequestAuth.ShareFromHeaders(ctx.Request, requireEndpoint: false)), ApiJson.Default.ShareInfo));
api.MapGet("/shares/{id}/manifest", async (string id, HttpContext ctx, ShareService svc) =>
{
byte[] manifest = await svc.ReadManifestAsync(ParseId(id), RequestAuth.ShareFromHeaders(ctx.Request, requireEndpoint: true));
return Results.Bytes(manifest, OctetStream);
});
api.MapGet("/shares/{id}/items/{itemId}/frames", async (string id, string itemId, HttpContext ctx, ShareService svc, CancellationToken ct) =>
{
var shareId = ParseId(id);
var parsedItemId = ParseId(itemId);
var creds = RequestAuth.ShareFromHeaders(ctx.Request, requireEndpoint: true);
long from = QueryLong(ctx.Request, "from", 0);
if (from is < 0 or > uint.MaxValue) throw new WebDropException(ErrorCode.BadRequest);
uint next = (uint)from;
svc.StartItemDownload(shareId, creds, parsedItemId, next);
ctx.Response.ContentType = OctetStream;
while (!ct.IsCancellationRequested)
{
var chunk = await svc.ReadItemFramesAsync(shareId, creds, parsedItemId, next, ct);
foreach (var frame in chunk.Frames) await ctx.Response.Body.WriteAsync(frame, ct);
next = chunk.NextIndex;
if (chunk.Complete) break;
if (chunk.Frames.Count == 0) throw new WebDropException(ErrorCode.Gone);
}
return Results.Empty;
});
}
// ───────── 辅助 ─────────
private static IResult Json<T>(T value, JsonTypeInfo<T> info, int? status = null) =>
Results.Json(value, info, statusCode: status);
private static Id16 ParseId(string text) =>
Id16.TryParse(text, out var id) ? id : throw new WebDropException(ErrorCode.BadRequest);
private static long QueryLong(HttpRequest req, string name, long fallback)
{
string value = req.Query[name].ToString();
if (value.Length == 0) return fallback;
return long.TryParse(value, out long n) ? n : throw new WebDropException(ErrorCode.BadRequest);
}
private static long? ParseMaxDownloads(string header)
{
if (header.Length == 0) return null;
return long.TryParse(header, out long n) && n >= 0 ? n : throw new WebDropException(ErrorCode.BadRequest);
}
private static async Task<T> ReadJson<T>(HttpRequest req, JsonTypeInfo<T> info, CancellationToken ct) where T : class
{
try
{
return await JsonSerializer.DeserializeAsync(req.Body, info, ct) ?? throw new WebDropException(ErrorCode.BadRequest);
}
catch (JsonException)
{
throw new WebDropException(ErrorCode.BadRequest);
}
}
/// <summary>读取请求体,超过上限立即拒绝(先看 Content-Length,再按实际读取量)。</summary>
private static async Task<byte[]> ReadBody(HttpRequest req, int max, CancellationToken ct)
{
if (req.ContentLength is { } declared && declared > max) throw new WebDropException(ErrorCode.TooLarge);
using var ms = new MemoryStream(req.ContentLength is { } len ? (int)len : 1024);
byte[] buffer = ArrayPool<byte>.Shared.Rent(16 * 1024);
try
{
int n;
while ((n = await req.Body.ReadAsync(buffer, ct)) > 0)
{
if (ms.Length + n > max) throw new WebDropException(ErrorCode.TooLarge);
ms.Write(buffer, 0, n);
}
}
finally
{
ArrayPool<byte>.Shared.Return(buffer);
}
return ms.ToArray();
}
private static async Task<LogPage> ReadLogWithWait(
ChannelService svc, ChannelHub hub, Id16 id, Credentials creds, long from, int limit, TimeSpan wait, CancellationToken ct)
{
if (wait <= TimeSpan.Zero) return await svc.ReadLogAsync(id, creds, from, limit, ct);
using var sub = hub.Subscribe(id);
var clock = Stopwatch.StartNew();
while (true)
{
var changed = sub.Changed;
var page = await svc.ReadLogAsync(id, creds, from, limit, ct);
var remaining = wait - clock.Elapsed;
if (page.Records.Count > 0 || remaining <= TimeSpan.Zero) return page;
await WaitOrTimeout(changed, remaining, ct);
}
}
/// <summary>等待信号或超时。返回 true 表示信号先到。</summary>
private static async Task<bool> WaitOrTimeout(Task changed, TimeSpan timeout, CancellationToken ct)
{
using var cts = CancellationTokenSource.CreateLinkedTokenSource(ct);
var delay = Task.Delay(timeout, cts.Token);
var winner = await Task.WhenAny(changed, delay);
await cts.CancelAsync();
ct.ThrowIfCancellationRequested();
return winner == changed;
}
/// <summary>WebSocket 的下行由订阅循环负责,这里只负责读取以便察觉客户端关闭。</summary>
private static async Task WatchClose(WebSocket ws, CancellationTokenSource cts)
{
var buffer = new byte[256];
try
{
while (ws.State == WebSocketState.Open)
{
var result = await ws.ReceiveAsync(buffer.AsMemory(), cts.Token);
if (result.MessageType == WebSocketMessageType.Close) break;
}
}
catch (Exception)
{
// 连接异常中断等同于关闭
}
finally
{
try { await cts.CancelAsync(); }
catch (ObjectDisposedException) { }
}
}
private static async Task CloseQuietly(WebSocket? ws)
{
if (ws is null) return;
try
{
if (ws.State is WebSocketState.Open or WebSocketState.CloseReceived)
await ws.CloseOutputAsync(WebSocketCloseStatus.NormalClosure, null, CancellationToken.None);
}
catch (Exception)
{
// 对端已断开
}
finally
{
ws.Dispose();
}
}
}
+119
View File
@@ -0,0 +1,119 @@
using System.Net.WebSockets;
using System.Text;
using WebDrop.Core;
namespace WebDrop.Api;
/// <summary>订阅的下行通道。WebSocket 与 SSE 各自实现,订阅循环共用。</summary>
public interface ILogSink
{
ValueTask SendRecordAsync(LogRecord record, CancellationToken ct);
ValueTask SendControlAsync(string json, CancellationToken ct);
/// <summary>流已建立之后出现的业务错误(如席位被释放),通知客户端并结束。</summary>
ValueTask FailAsync(ErrorCode code, CancellationToken ct);
}
public static class LogStreamer
{
private const int PageSize = 200;
public static readonly TimeSpan Heartbeat = TimeSpan.FromSeconds(20);
/// <summary>
/// 按游标持续读取日志并推送。<paramref name="open"/> 在首次鉴权和读取成功之后才被调用,
/// 因此鉴权失败仍可以返回普通的 HTTP 错误,而不是先升级再断开。
/// </summary>
public static async Task RunAsync(
ChannelService svc, ChannelHub hub, Id16 id, Credentials creds, long from,
Func<Task<ILogSink>> open, CancellationToken ct)
{
using var sub = hub.Subscribe(id);
ILogSink? sink = null;
long cursor = from;
var (settingsV, seatsV, _) = sub.State;
try
{
while (!ct.IsCancellationRequested)
{
var changed = sub.Changed; // 先取任务再读状态:读取期间的发布不会丢
var page = await svc.ReadLogAsync(id, creds, cursor, PageSize, ct).ConfigureAwait(false);
sink ??= await open().ConfigureAwait(false);
foreach (var record in page.Records)
{
await sink.SendRecordAsync(record, ct).ConfigureAwait(false);
cursor = record.Seq + 1;
}
var (settingsNow, seatsNow, deleted) = sub.State;
if (settingsNow != settingsV)
{
settingsV = settingsNow;
await sink.SendControlAsync("""{"t":"settings"}""", ct).ConfigureAwait(false);
}
if (seatsNow != seatsV)
{
seatsV = seatsNow;
await sink.SendControlAsync("""{"t":"seats"}""", ct).ConfigureAwait(false);
}
if (deleted) throw new WebDropException(ErrorCode.NotFound);
if (page.Records.Count >= PageSize) continue; // 还有积压,不用等
using var delayCts = CancellationTokenSource.CreateLinkedTokenSource(ct);
var delay = Task.Delay(Heartbeat, delayCts.Token);
var winner = await Task.WhenAny(changed, delay).ConfigureAwait(false);
await delayCts.CancelAsync().ConfigureAwait(false);
ct.ThrowIfCancellationRequested();
if (winner == delay)
await sink.SendControlAsync($$"""{"t":"hb","seq":{{page.LastSeq}}}""", ct).ConfigureAwait(false);
}
}
catch (OperationCanceledException) when (ct.IsCancellationRequested)
{
// 客户端断开,正常结束
}
catch (WebDropException e) when (sink is not null)
{
await sink.FailAsync(e.Code, CancellationToken.None).ConfigureAwait(false);
}
}
}
public sealed class WebSocketSink(WebSocket ws) : ILogSink
{
public ValueTask SendRecordAsync(LogRecord record, CancellationToken ct) =>
ws.SendAsync(record.Bytes.AsMemory(), WebSocketMessageType.Binary, true, ct);
public ValueTask SendControlAsync(string json, CancellationToken ct) =>
ws.SendAsync(Encoding.UTF8.GetBytes(json).AsMemory(), WebSocketMessageType.Text, true, ct);
public async ValueTask FailAsync(ErrorCode code, CancellationToken ct)
{
if (ws.State is WebSocketState.Open or WebSocketState.CloseReceived)
await ws.CloseAsync(WebSocketCloseStatus.PolicyViolation, code.ToWire(), ct).ConfigureAwait(false);
}
}
public sealed class SseSink(HttpResponse response) : ILogSink
{
public async ValueTask SendRecordAsync(LogRecord record, CancellationToken ct)
{
await response.WriteAsync($"id: {record.Seq}\nevent: log\ndata: {Codec.Encode(record.Bytes)}\n\n", ct).ConfigureAwait(false);
await response.Body.FlushAsync(ct).ConfigureAwait(false);
}
public async ValueTask SendControlAsync(string json, CancellationToken ct)
{
await response.WriteAsync($"event: control\ndata: {json}\n\n", ct).ConfigureAwait(false);
await response.Body.FlushAsync(ct).ConfigureAwait(false);
}
public async ValueTask FailAsync(ErrorCode code, CancellationToken ct)
{
await response.WriteAsync($"event: error\ndata: {{\"error\":\"{code.ToWire()}\"}}\n\n", ct).ConfigureAwait(false);
await response.Body.FlushAsync(ct).ConfigureAwait(false);
}
}
+4
View File
@@ -0,0 +1,4 @@
using WebDrop.Api;
var app = WebDropApp.Build(args);
app.Run();
@@ -0,0 +1,23 @@
{
"$schema": "https://json.schemastore.org/launchsettings.json",
"profiles": {
"http": {
"commandName": "Project",
"dotnetRunMessages": true,
"launchBrowser": true,
"applicationUrl": "http://localhost:5024",
"environmentVariables": {
"ASPNETCORE_ENVIRONMENT": "Development"
}
},
"https": {
"commandName": "Project",
"dotnetRunMessages": true,
"launchBrowser": true,
"applicationUrl": "https://localhost:7212;http://localhost:5024",
"environmentVariables": {
"ASPNETCORE_ENVIRONMENT": "Development"
}
}
}
}
+80
View File
@@ -0,0 +1,80 @@
using WebDrop.Core;
namespace WebDrop.Api;
/// <summary>从请求头(或 WebSocket 子协议)解析凭据。这里只做格式校验,令牌是否正确由服务判断。</summary>
public static class RequestAuth
{
public const string RoleHeader = "X-WD-Role";
public const string EndpointHeader = "X-WD-Endpoint";
public const string CaptchaHeader = "X-WD-Cap";
public static Credentials FromHeaders(HttpRequest req, bool requireEndpoint)
{
var role = ParseRole(req.Headers[RoleHeader].ToString());
byte[]? token = null;
string auth = req.Headers.Authorization.ToString();
if (auth.Length > 0)
{
const string prefix = "Bearer ";
if (!auth.StartsWith(prefix, StringComparison.Ordinal)
|| !Codec.TryDecodeExact(auth[prefix.Length..], TokenHash.Size, out token))
throw new WebDropException(ErrorCode.BadRequest);
}
return new Credentials(role, token, ParseEndpoint(req.Headers[EndpointHeader].ToString(), requireEndpoint));
}
/// <summary>
/// 浏览器的 WebSocket 不能设请求头,凭据放在子协议里:
/// wd.v1, wd.role.&lt;role&gt;, wd.auth.&lt;b64url&gt;, wd.ep.&lt;b64url&gt;
/// </summary>
public static Credentials FromSubprotocols(IEnumerable<string> protocols)
{
string? role = null, auth = null, endpoint = null;
foreach (string p in protocols)
{
if (p.StartsWith("wd.role.", StringComparison.Ordinal)) role = p["wd.role.".Length..];
else if (p.StartsWith("wd.auth.", StringComparison.Ordinal)) auth = p["wd.auth.".Length..];
else if (p.StartsWith("wd.ep.", StringComparison.Ordinal)) endpoint = p["wd.ep.".Length..];
}
byte[]? token = null;
if (auth is not null && !Codec.TryDecodeExact(auth, TokenHash.Size, out token))
throw new WebDropException(ErrorCode.BadRequest);
return new Credentials(ParseRole(role), token, ParseEndpoint(endpoint, requireEndpoint: true));
}
public static byte[] ParseEndpointHash(string? header) => ParseEndpoint(header, requireEndpoint: true);
public static byte[] ParseEndpointHashOptional(string? header) => ParseEndpoint(header, requireEndpoint: false);
public static Credentials ShareFromHeaders(HttpRequest req, bool requireEndpoint)
{
var credentials = FromHeaders(req, requireEndpoint);
if (credentials.ClaimedRole == Role.Write) throw new WebDropException(ErrorCode.BadRequest);
return credentials;
}
private static Role ParseRole(string? value) => value switch
{
"read" => Role.Read,
"write" => Role.Write,
"owner" => Role.Owner,
_ => throw new WebDropException(ErrorCode.BadRequest),
};
/// <summary>终端密钥 S 为 32 字节,服务端只保存并使用 SHA-256(S)。缺省时为空,仅允许不需要席位的操作。</summary>
private static byte[] ParseEndpoint(string? value, bool requireEndpoint)
{
if (string.IsNullOrEmpty(value))
{
if (requireEndpoint) throw new WebDropException(ErrorCode.BadRequest);
return [];
}
if (!Codec.TryDecodeExact(value, 32, out var secret)) throw new WebDropException(ErrorCode.BadRequest);
return TokenHash.Compute(secret);
}
}
+42
View File
@@ -0,0 +1,42 @@
using WebDrop.Core;
namespace WebDrop.Api;
/// <summary>定期清理过期条目、下载达上限的条目、过期频道,并低频清理孤立 blob。</summary>
public sealed class SweepWorker(ChannelService service, ShareService shares, WebDropOptions options, ILogger<SweepWorker> logger) : BackgroundService
{
private const int OrphanSweepEvery = 20;
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
var interval = TimeSpan.FromSeconds(Math.Max(1, options.SweepIntervalSeconds));
using var timer = new PeriodicTimer(interval);
int round = 0;
do
{
try
{
bool orphans = round++ % OrphanSweepEvery == 0;
int removed = await Task.Run(() => service.Sweep(orphans) + shares.Sweep(), stoppingToken).ConfigureAwait(false);
if (removed > 0) logger.LogInformation("清理了 {Count} 个过期条目或频道", removed);
}
catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
{
break;
}
catch (Exception e)
{
// 只记录异常类型,不记录可能含频道 id 的细节
logger.LogError("清理任务失败:{Type}", e.GetType().Name);
}
}
while (await WaitAsync(timer, stoppingToken).ConfigureAwait(false));
}
private static async Task<bool> WaitAsync(PeriodicTimer timer, CancellationToken ct)
{
try { return await timer.WaitForNextTickAsync(ct).ConfigureAwait(false); }
catch (OperationCanceledException) { return false; }
}
}
+17
View File
@@ -0,0 +1,17 @@
<Project Sdk="Microsoft.NET.Sdk.Web">
<PropertyGroup>
<PublishAot>true</PublishAot>
<StripSymbols>true</StripSymbols>
<EnableConfigurationBindingGenerator>true</EnableConfigurationBindingGenerator>
<JsonSerializerIsReflectionEnabledByDefault>false</JsonSerializerIsReflectionEnabledByDefault>
<AssemblyName>webdrop</AssemblyName>
<RootNamespace>WebDrop.Api</RootNamespace>
</PropertyGroup>
<ItemGroup>
<ProjectReference Include="..\WebDrop.Core\WebDrop.Core.csproj" />
<ProjectReference Include="..\WebDrop.Storage\WebDrop.Storage.csproj" />
</ItemGroup>
</Project>
+173
View File
@@ -0,0 +1,173 @@
using System.Threading.RateLimiting;
using WebDrop.Core;
using WebDrop.Storage;
namespace WebDrop.Api;
public static class WebDropApp
{
/// <summary>构建应用。<paramref name="configure"/> 在读取配置之前执行,测试用它覆盖数据目录和端口。</summary>
public static WebApplication Build(string[] args, Action<WebApplicationBuilder>? configure = null)
{
var builder = WebApplication.CreateSlimBuilder(args);
configure?.Invoke(builder);
var options = builder.Configuration.GetSection("WebDrop").Get<WebDropOptions>() ?? new WebDropOptions();
builder.WebHost.ConfigureKestrel(k =>
{
k.AddServerHeader = false;
k.Limits.MaxRequestBodySize = 2 * 1024 * 1024;
});
builder.Services.AddSingleton(options);
builder.Services.AddSingleton(options.Limits);
builder.Services.AddSingleton<IClock, SystemClock>();
builder.Services.AddSingleton<IBlobStore>(_ => new DiskBlobStore(Path.Combine(options.DataDir, "blobs")));
builder.Services.AddSingleton<SqliteChannelStore>(_ => new SqliteChannelStore(Path.Combine(options.DataDir, "webdrop.db")));
builder.Services.AddSingleton<IResourceStore>(sp => sp.GetRequiredService<SqliteChannelStore>());
builder.Services.AddSingleton<ICaptchaVerifier>(sp => CreateCaptchaVerifier(options.Captcha, sp));
builder.Services.AddSingleton<ChannelHub>();
builder.Services.AddSingleton<IChannelEvents>(sp => sp.GetRequiredService<ChannelHub>());
builder.Services.AddSingleton<ChannelService>();
builder.Services.AddSingleton<ShareService>(sp => new ShareService(
sp.GetRequiredService<IResourceStore>(), sp.GetRequiredService<IBlobStore>(), sp.GetRequiredService<IClock>(),
options.Limits, options.ShareDraftTtlSeconds));
builder.Services.AddHostedService<SweepWorker>();
if (options.CorsOrigins.Length > 0)
{
builder.Services.AddCors(c => c.AddDefaultPolicy(p => p
.WithOrigins(options.CorsOrigins)
.AllowAnyMethod()
.WithHeaders("Authorization", "Content-Type", "Last-Event-ID",
"X-WD-Role", "X-WD-Endpoint", "X-WD-Cap", "X-WD-Label", "X-WD-Max-Downloads")
.WithExposedHeaders("X-WD-Last-Seq")
.SetPreflightMaxAge(TimeSpan.FromHours(1))));
}
builder.Services.AddRateLimiter(r =>
{
r.RejectionStatusCode = StatusCodes.Status429TooManyRequests;
// 限流状态只在内存里,按 IP 分区,不落盘
r.GlobalLimiter = PartitionedRateLimiter.Create<HttpContext, string>(ctx =>
RateLimitPartition.GetFixedWindowLimiter(ClientKey(ctx), _ => PerMinute(options.RequestsPerMinute)));
r.AddPolicy(Endpoints.CreatePolicy, ctx =>
RateLimitPartition.GetFixedWindowLimiter(ClientKey(ctx), _ => PerMinute(options.ChannelCreatesPerMinute)));
r.OnRejected = async (context, ct) =>
{
context.HttpContext.Response.StatusCode = StatusCodes.Status429TooManyRequests;
await context.HttpContext.Response.WriteAsJsonAsync(
new ErrorBody { Error = ErrorCode.RateLimited.ToWire() }, ApiJson.Default.ErrorBody, cancellationToken: ct);
};
});
var app = builder.Build();
if (options.TrustForwardedHeaders)
{
app.UseForwardedHeaders(new ForwardedHeadersOptions
{
ForwardedHeaders = Microsoft.AspNetCore.HttpOverrides.ForwardedHeaders.XForwardedFor,
KnownIPNetworks = { },
KnownProxies = { },
ForwardLimit = 1,
});
}
app.Use(SecurityHeaders);
app.Use(HandleErrors);
if (options.CorsOrigins.Length > 0) app.UseCors();
app.UseWebSockets(new WebSocketOptions { KeepAliveInterval = LogStreamer.Heartbeat });
app.UseRateLimiter();
app.MapApi(options, app.Services.GetRequiredService<ICaptchaVerifier>());
return app;
}
/// <summary>配置不完整就在启动时失败,而不是等到第一个用户创建频道时才发现。</summary>
private static ICaptchaVerifier CreateCaptchaVerifier(CaptchaOptions o, IServiceProvider sp)
{
switch (o.Provider.ToLowerInvariant())
{
case "none":
return new NullCaptchaVerifier();
case "cap":
if (!Uri.TryCreate(o.SiteverifyUrl, UriKind.Absolute, out var verifyUri) || verifyUri.Scheme is not ("http" or "https"))
throw new InvalidOperationException("WebDrop:Captcha:SiteverifyUrl 必须是 http(s) 绝对地址");
if (!Uri.TryCreate(o.ApiEndpoint, UriKind.Absolute, out var apiUri) || apiUri.Scheme is not ("http" or "https"))
throw new InvalidOperationException("WebDrop:Captcha:ApiEndpoint 必须是 http(s) 绝对地址(前端 widget 要连它)");
if (string.IsNullOrEmpty(o.Secret))
throw new InvalidOperationException("WebDrop:Captcha:Secret 不能为空");
var http = new HttpClient { Timeout = TimeSpan.FromSeconds(Math.Max(1, o.TimeoutSeconds)) };
return new CapCaptchaVerifier(http, o, sp.GetRequiredService<ILogger<CapCaptchaVerifier>>());
default:
throw new InvalidOperationException($"WebDrop:Captcha:Provider 只能是 none 或 cap,当前为 {o.Provider}");
}
}
private static string ClientKey(HttpContext ctx) => ctx.Connection.RemoteIpAddress?.ToString() ?? "unknown";
private static FixedWindowRateLimiterOptions PerMinute(int permits) => new()
{
PermitLimit = Math.Max(1, permits),
Window = TimeSpan.FromMinutes(1),
QueueLimit = 0,
};
/// <summary>API 不提供任何页面。禁用缓存、禁止嗅探、不发送 Referer、不允许被嵌入。</summary>
private static Task SecurityHeaders(HttpContext ctx, RequestDelegate next)
{
var h = ctx.Response.Headers;
h["Referrer-Policy"] = "no-referrer";
h["X-Content-Type-Options"] = "nosniff";
h["Cache-Control"] = "no-store";
h["Content-Security-Policy"] = "default-src 'none'; frame-ancestors 'none'";
return next(ctx);
}
/// <summary>统一的错误出口:业务错误映射为 {"error": code};未预期的错误只返回 500,不暴露细节。</summary>
private static async Task HandleErrors(HttpContext ctx, RequestDelegate next)
{
try
{
await next(ctx);
}
catch (WebDropException e)
{
await WriteError(ctx, e.Code, e.OldestSeq);
}
catch (BadHttpRequestException e)
{
await WriteError(ctx, e.StatusCode == StatusCodes.Status413PayloadTooLarge ? ErrorCode.TooLarge : ErrorCode.BadRequest, null);
}
catch (OperationCanceledException) when (ctx.RequestAborted.IsCancellationRequested)
{
// 客户端已断开
}
catch (Exception e)
{
ctx.RequestServices.GetRequiredService<ILoggerFactory>().CreateLogger("WebDrop")
.LogError("未处理的异常:{Type}", e.GetType().Name);
if (ctx.Response.HasStarted) { ctx.Abort(); return; }
ctx.Response.StatusCode = StatusCodes.Status500InternalServerError;
await ctx.Response.WriteAsJsonAsync(new ErrorBody { Error = "internal_error" }, ApiJson.Default.ErrorBody);
}
}
private static async Task WriteError(HttpContext ctx, ErrorCode code, long? oldestSeq)
{
if (ctx.Response.HasStarted)
{
ctx.Abort(); // 流已经开始,无法再改状态码,直接断开让客户端续传
return;
}
// 不能调用 Response.Clear():它会把前面中间件设置的安全头和 CORS 头一并清掉,
// 浏览器端就读不到错误体了。响应尚未开始,直接设状态码并写 JSON 即可。
ctx.Response.StatusCode = code.ToStatus();
if (code == ErrorCode.Gone && oldestSeq is { } oldest)
await ctx.Response.WriteAsJsonAsync(new GoneLogBody { OldestSeq = oldest }, ApiJson.Default.GoneLogBody);
else
await ctx.Response.WriteAsJsonAsync(new ErrorBody { Error = code.ToWire() }, ApiJson.Default.ErrorBody);
}
}
+51
View File
@@ -0,0 +1,51 @@
using WebDrop.Core;
namespace WebDrop.Api;
/// <summary>配置节 "WebDrop"。环境变量形式:WebDrop__DataDir、WebDrop__Captcha__Provider ……</summary>
public sealed class WebDropOptions
{
public string DataDir { get; set; } = "data";
/// <summary>允许跨域访问的前端源。前端托管在独立域名,必须在此列出;为空则不发 CORS 头。</summary>
public string[] CorsOrigins { get; set; } = [];
/// <summary>在反向代理之后部署时开启,以便限流按真实客户端 IP 分区。注意代理本身不受信任。</summary>
public bool TrustForwardedHeaders { get; set; }
/// <summary>
/// 声明部署层已启用 HTTP/3(例如由反向代理终结 QUIC),仅用于 /info 的通告。
/// 服务端进程本身不内置 QUIC:CreateSlimBuilder 不含 HTTP/3,需要时再接入。
/// </summary>
public bool EnableHttp3 { get; set; }
public int RequestsPerMinute { get; set; } = 600;
public int ChannelCreatesPerMinute { get; set; } = 30;
public int ShareDraftTtlSeconds { get; set; } = 86_400;
public int SweepIntervalSeconds { get; set; } = 30;
public ServerLimits Limits { get; set; } = new();
public CaptchaOptions Captcha { get; set; } = new();
}
/// <summary>
/// 创建频道的验证码。工作量证明完全交给 Cap(https://github.com/tiagozip/cap),本项目不自己实现。
/// 选 "cap" 时需要部署一个 Cap Standalone(Docker),并在其中建一个站点密钥。
/// </summary>
public sealed class CaptchaOptions
{
/// <summary>"none"(默认,不校验)或 "cap"。</summary>
public string Provider { get; set; } = "none";
/// <summary>浏览器里的 Cap widget 连接的地址,形如 https://cap.example.com/&lt;site_key&gt;/。会通过 /info 下发给前端。</summary>
public string ApiEndpoint { get; set; } = "";
/// <summary>本服务调用 siteverify 的地址,形如 https://cap.example.com/&lt;site_key&gt;/siteverify。可以是内网地址。</summary>
public string SiteverifyUrl { get; set; } = "";
/// <summary>Cap 站点密钥对应的 secret,只在服务端使用,不得下发给前端。</summary>
public string Secret { get; set; } = "";
public int TimeoutSeconds { get; set; } = 5;
}
+26
View File
@@ -0,0 +1,26 @@
{
"Logging": {
"LogLevel": {
"Default": "Information",
"Microsoft.AspNetCore": "Warning",
"Microsoft.AspNetCore.Hosting.Diagnostics": "None"
}
},
"WebDrop": {
"DataDir": "data",
"CorsOrigins": [],
"TrustForwardedHeaders": false,
"EnableHttp3": false,
"RequestsPerMinute": 600,
"ChannelCreatesPerMinute": 30,
"ShareDraftTtlSeconds": 86400,
"SweepIntervalSeconds": 30,
"Captcha": {
"Provider": "none",
"ApiEndpoint": "",
"SiteverifyUrl": "",
"Secret": "",
"TimeoutSeconds": 5
}
}
}
+128
View File
@@ -0,0 +1,128 @@
namespace WebDrop.Core;
public interface IClock
{
long NowMs { get; }
}
public sealed class SystemClock : IClock
{
public long NowMs => DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
}
/// <summary>大帧的字节存储。键由服务生成,实现不得解析。</summary>
public interface IBlobStore
{
ValueTask PutAsync(string key, ReadOnlyMemory<byte> data, CancellationToken ct = default);
ValueTask<byte[]?> ReadAsync(string key, CancellationToken ct = default);
void Delete(string key);
void DeletePrefix(string prefix);
IEnumerable<(string Key, DateTimeOffset LastWrite)> Enumerate();
}
/// <summary>
/// 创建频道前的人机/滥用校验(Cap 的 PoW 验证码)。验证码本身不属于本项目:
/// 校验失败抛 CaptchaRequired,验证服务不可用抛 CaptchaUnavailable(两者对用户的含义不同)。
/// </summary>
public interface ICaptchaVerifier
{
bool Enabled { get; }
Task VerifyAsync(string? token, CancellationToken ct);
}
/// <summary>不启用验证码(开发和小型自托管)。</summary>
public sealed class NullCaptchaVerifier : ICaptchaVerifier
{
public bool Enabled => false;
public Task VerifyAsync(string? token, CancellationToken ct) => Task.CompletedTask;
}
/// <summary>服务提交后发出的通知,传输层据此唤醒订阅者。</summary>
public interface IChannelEvents
{
void Advanced(Id16 channelId, long seq);
void SettingsChanged(Id16 channelId);
void SeatsChanged(Id16 channelId);
void ChannelDeleted(Id16 channelId);
}
public sealed class NullChannelEvents : IChannelEvents
{
public void Advanced(Id16 channelId, long seq) { }
public void SettingsChanged(Id16 channelId) { }
public void SeatsChanged(Id16 channelId) { }
public void ChannelDeleted(Id16 channelId) { }
}
/// <summary>频道和分享资源共用的读写事务入口。Write 串行执行,抛出异常即回滚。</summary>
public interface IResourceStore
{
T Read<T>(Func<IResourceTx, T> action);
T Write<T>(Func<IResourceTx, T> action);
}
/// <summary>单个数据库事务内的频道、分享与帧操作。</summary>
public interface IResourceTx
{
// 便捷分享
Share? GetShare(Id16 id);
void InsertShare(Share share);
void UpdateShare(Share share);
IReadOnlyList<Id16> ListExpiredShares(long now, int limit);
bool HasResourceId(Id16 id);
int CountShareVisitors(Id16 id);
bool HasShareVisitor(Id16 id, ReadOnlySpan<byte> endpointHash);
void InsertShareVisitor(Id16 id, ReadOnlySpan<byte> endpointHash, long joinedAt);
void DeleteShareRows(Id16 id);
IReadOnlyList<ShareItem> ListShareItems(Id16 id);
int CountShareItems(Id16 id);
void AddShareBytes(Id16 id, long delta);
IReadOnlyList<ShareItem> ListExpiredShareItems(long now, int limit);
ShareItem? GetShareItem(Id16 shareId, Id16 itemId);
void InsertShareItem(ShareItem item);
void UpdateShareItem(ShareItem item);
void DeleteShareItemRows(Id16 shareId, Id16 itemId);
ShareFrame? GetShareFrame(Id16 shareId, Id16 itemId, uint index);
void InsertShareFrame(ShareFrame frame);
IReadOnlyList<ShareFrame> ListShareFrames(Id16 shareId, Id16 itemId, uint fromIndex, int limit);
// 频道
Channel? GetChannel(Id16 id);
void InsertChannel(Channel channel);
void UpdateChannel(Id16 id, ChannelSettings settings, long expiresAt);
void TouchChannel(Id16 id, long expiresAt);
void AddChannelBytes(Id16 id, long delta);
void SetCompactedThrough(Id16 id, long seq);
long NextSeq(Id16 id);
void DeleteChannelRows(Id16 id);
IReadOnlyList<Id16> ListExpiredChannels(long now, int limit);
// 席位
Seat? GetSeat(Id16 channelId, ReadOnlySpan<byte> endpointHash);
int CountSeats(Id16 channelId);
void InsertSeat(Id16 channelId, Seat seat);
bool DeleteSeat(Id16 channelId, ReadOnlySpan<byte> endpointHash);
void SetSeatLabel(Id16 channelId, ReadOnlySpan<byte> endpointHash, byte[] label);
IReadOnlyList<Seat> ListSeats(Id16 channelId);
// 条目
Item? GetItem(Id16 channelId, Id16 itemId);
void InsertItem(Item item);
void UpdateItem(Item item);
void DeleteItemRows(Id16 channelId, Id16 itemId);
int CountItems(Id16 channelId);
IReadOnlyList<Item> ListOldestItems(Id16 channelId, int limit);
Item? FindSnapshotItem(Id16 channelId, Id16 excludeItemId);
IReadOnlyList<Item> ListExpiredItems(long now, int limit);
// 帧
StoredFrame? GetFrame(Id16 channelId, Id16 itemId, uint index);
void InsertFrame(StoredFrame frame);
IReadOnlyList<StoredFrame> ListFrames(Id16 channelId, Id16 itemId, uint fromIndex, int limit);
bool BlobKeyExists(string blobKey);
// 日志
void InsertLog(Id16 channelId, LogEntry entry);
IReadOnlyList<LogEntry> ReadLog(Id16 channelId, long fromSeq, int limit);
long PruneTombstones(Id16 channelId, long olderThanMs);
}
+19
View File
@@ -0,0 +1,19 @@
using System.Security.Cryptography;
namespace WebDrop.Core;
/// <summary>服务端只保存令牌的 SHA-256,验证时常量时间比较。</summary>
public static class TokenHash
{
public const int Size = 32;
public static byte[] Compute(ReadOnlySpan<byte> token) => SHA256.HashData(token);
public static bool Matches(ReadOnlySpan<byte> token, byte[]? expectedHash)
{
if (expectedHash is not { Length: Size }) return false;
Span<byte> actual = stackalloc byte[Size];
SHA256.HashData(token, actual);
return CryptographicOperations.FixedTimeEquals(actual, expectedHash);
}
}
+26
View File
@@ -0,0 +1,26 @@
namespace WebDrop.Core;
/// <summary>谁能做什么。规则只写在这里,服务和测试都调用它。</summary>
public static class ChannelPolicy
{
public static bool IsOwner(Principal p) => p.Role == Role.Owner;
public static bool CanRead(Principal p) => p.Role >= Role.Read;
/// <summary>成员写入需要频道允许上传,且服务端持有写令牌哈希。</summary>
public static bool CanWrite(Channel channel, Principal p) =>
p.Role == Role.Owner
|| (p.Role == Role.Write && channel.Settings.MembersCanUpload && channel.WriteHash is not null);
public static bool CanDeleteItem(Channel channel, Item item, Principal p)
{
if (p.Role == Role.Owner) return true;
if (p.Role < Role.Read) return false;
return channel.Settings.DeletePolicy switch
{
DeletePolicy.Anyone => true,
DeletePolicy.Own => item.AuthorHash.AsSpan().SequenceEqual(p.EndpointHash),
_ => false,
};
}
}
+571
View File
@@ -0,0 +1,571 @@
using System.Security.Cryptography;
namespace WebDrop.Core;
/// <summary>
/// 频道的全部业务逻辑。依赖抽象(存储、blob、时钟、事件),不含 HTTP、SQL、文件细节。
/// 约定:所有失败以 <see cref="WebDropException"/> 表达;提交成功后才发事件;blob 先写后提交。
/// </summary>
public sealed class ChannelService(
IResourceStore store,
IBlobStore blobs,
IClock clock,
ServerLimits limits,
IChannelEvents events)
{
private enum SeatRule { None, Required }
// ───────────────────────── 频道 ─────────────────────────
public ChannelInfo CreateChannel(CreateChannelRequest req, byte[] endpointHash)
{
if (!Id16.TryParse(req.Id, out var id)) throw new WebDropException(ErrorCode.BadRequest);
if (req.Mode is not { } mode) throw new WebDropException(ErrorCode.BadRequest);
var settings = req.Settings ?? new ChannelSettings();
settings.Validate(limits);
if (!Codec.TryDecodeExact(req.OwnerHash, TokenHash.Size, out var ownerHash))
throw new WebDropException(ErrorCode.BadRequest);
byte[]? readHash = null;
if (mode == ChannelMode.Enc)
{
if (!Codec.TryDecodeExact(req.ReadHash, TokenHash.Size, out readHash))
throw new WebDropException(ErrorCode.BadRequest);
}
else if (req.ReadHash is not null)
{
throw new WebDropException(ErrorCode.BadRequest); // raw 频道的读权限是 id 本身
}
byte[]? writeHash = null;
if (req.WriteHash is not null && !Codec.TryDecodeExact(req.WriteHash, TokenHash.Size, out writeHash))
throw new WebDropException(ErrorCode.BadRequest);
long now = clock.NowMs;
var channel = new Channel(id, mode, settings, readHash, writeHash, ownerHash,
now, now + settings.TtlSeconds * 1000L, 0, 0, 0);
store.Write(tx =>
{
if (tx.HasResourceId(id)) throw new WebDropException(ErrorCode.Conflict);
tx.InsertChannel(channel);
tx.InsertSeat(id, new Seat(endpointHash, now, null));
return 0;
});
return ToInfo(channel, 1);
}
public ChannelInfo GetChannel(Id16 id, Credentials creds) => store.Read(tx =>
{
var (channel, _) = Authenticate(tx, id, creds, SeatRule.None);
return ToInfo(channel, tx.CountSeats(id));
});
public ChannelInfo UpdateSettings(Id16 id, Credentials creds, ChannelSettingsPatch patch)
{
var info = store.Write(tx =>
{
var (channel, principal) = Authenticate(tx, id, creds, SeatRule.None);
RequireOwner(principal);
var updated = patch.ApplyTo(channel.Settings);
updated.Validate(limits);
long expiresAt = updated.TtlSeconds != channel.Settings.TtlSeconds
? clock.NowMs + updated.TtlSeconds * 1000L
: channel.ExpiresAt;
tx.UpdateChannel(id, updated, expiresAt);
return ToInfo(channel with { Settings = updated, ExpiresAt = expiresAt }, tx.CountSeats(id));
});
events.SettingsChanged(id);
return info;
}
public void DeleteChannel(Id16 id, Credentials creds)
{
store.Write(tx =>
{
var (_, principal) = Authenticate(tx, id, creds, SeatRule.None);
RequireOwner(principal);
tx.DeleteChannelRows(id);
return 0;
});
blobs.DeletePrefix(id.ToHex());
events.ChannelDeleted(id);
}
// ───────────────────────── 席位 ─────────────────────────
public SeatJoined Join(Id16 id, Credentials creds)
{
var (joined, isNew) = store.Write(tx =>
{
var (channel, principal) = Authenticate(tx, id, creds, SeatRule.None);
var existing = tx.GetSeat(id, principal.EndpointHash);
int used = tx.CountSeats(id);
if (existing is not null)
return (new SeatJoined { EndpointHash = Codec.Encode(existing.EndpointHash), JoinedAt = existing.JoinedAt, SeatsUsed = used, SeatsMax = channel.Settings.Seats }, false);
if (used >= channel.Settings.Seats) throw new WebDropException(ErrorCode.SeatsFull);
long now = clock.NowMs;
tx.InsertSeat(id, new Seat(principal.EndpointHash, now, null));
return (new SeatJoined { EndpointHash = Codec.Encode(principal.EndpointHash), JoinedAt = now, SeatsUsed = used + 1, SeatsMax = channel.Settings.Seats }, true);
});
if (isNew) events.SeatsChanged(id);
return joined;
}
public IReadOnlyList<SeatInfo> ListSeats(Id16 id, Credentials creds) => store.Read(tx =>
{
var (_, principal) = Authenticate(tx, id, creds, SeatRule.None);
RequireOwner(principal);
return (IReadOnlyList<SeatInfo>)tx.ListSeats(id).Select(s => new SeatInfo
{
EndpointHash = Codec.Encode(s.EndpointHash),
JoinedAt = s.JoinedAt,
Label = s.Label is null ? null : Codec.Encode(s.Label),
}).ToList();
});
public void SetSeatLabel(Id16 id, Credentials creds, byte[] label)
{
if (label.Length == 0) throw new WebDropException(ErrorCode.BadRequest);
if (label.Length > limits.MaxSeatLabelBytes) throw new WebDropException(ErrorCode.TooLarge);
store.Write(tx =>
{
var (_, principal) = Authenticate(tx, id, creds, SeatRule.Required);
tx.SetSeatLabel(id, principal.EndpointHash, label);
return 0;
});
events.SeatsChanged(id);
}
public void ReleaseSeat(Id16 id, Credentials creds, byte[] endpointHash)
{
store.Write(tx =>
{
var (_, principal) = Authenticate(tx, id, creds, SeatRule.None);
RequireOwner(principal);
if (!tx.DeleteSeat(id, endpointHash)) throw new WebDropException(ErrorCode.NotFound);
return 0;
});
events.SeatsChanged(id);
}
// ───────────────────────── 追加帧 ─────────────────────────
public async Task<AppendResult> AppendFrameAsync(
Id16 channelId, Credentials creds, ReadOnlyMemory<byte> body,
long? maxDownloads, string? label, CancellationToken ct = default)
{
if (!Envelope.TryParse(body.Span, out var env)) throw new WebDropException(ErrorCode.BadRequest);
if (env.PayloadLength > limits.MaxPayloadBytes) throw new WebDropException(ErrorCode.TooLarge);
if (label is not null && System.Text.Encoding.UTF8.GetByteCount(label) > limits.MaxLabelBytes)
throw new WebDropException(ErrorCode.BadRequest);
ValidateShape(env);
byte[] digest = SHA256.HashData(body.Span);
// 1. 预检:鉴权并处理幂等重发,避免为重复帧白写 blob
var existing = store.Read(tx =>
{
var (channel, principal) = Authenticate(tx, channelId, creds, SeatRule.Required);
CheckWritable(channel, principal, env);
return tx.GetFrame(channelId, env.ItemId, env.FrameIndex);
});
if (existing is not null) return Resolve(existing, digest);
// 2. 大帧先写 blob。键含摘要,内容不同的并发写不会互相覆盖
string? blobKey = null;
if (body.Length >= limits.InlineThresholdBytes)
{
blobKey = $"{channelId.ToHex()}/{env.ItemId.ToHex()}/{env.FrameIndex}.{Convert.ToHexStringLower(digest.AsSpan(0, 16))}";
await blobs.PutAsync(blobKey, body, ct).ConfigureAwait(false);
}
// 3. 单事务提交
AppendOutcome outcome;
try
{
outcome = store.Write(tx => AppendInTx(tx, channelId, creds, env, body, digest, blobKey, maxDownloads, label));
}
catch
{
DeleteBlobIfUnreferenced(blobKey);
throw;
}
if (!outcome.Result.Created) DeleteBlobIfUnreferenced(blobKey);
foreach (string prefix in outcome.BlobPrefixesToDelete) blobs.DeletePrefix(prefix);
events.Advanced(channelId, outcome.LastSeq);
return outcome.Result;
}
private sealed class AppendOutcome(AppendResult result, long lastSeq, List<string> prefixes)
{
public AppendResult Result { get; } = result;
public long LastSeq { get; } = lastSeq;
public List<string> BlobPrefixesToDelete { get; } = prefixes;
}
private AppendOutcome AppendInTx(
IResourceTx tx, Id16 channelId, Credentials creds, Envelope env, ReadOnlyMemory<byte> body,
byte[] digest, string? blobKey, long? maxDownloads, string? label)
{
var (channel, principal) = Authenticate(tx, channelId, creds, SeatRule.Required);
CheckWritable(channel, principal, env);
// 并发下预检之后可能已有同一帧
if (tx.GetFrame(channelId, env.ItemId, env.FrameIndex) is { } dup)
return new AppendOutcome(Resolve(dup, digest), dup.Seq, []);
long now = clock.NowMs;
long ttlMs = channel.Settings.TtlSeconds * 1000L;
var item = tx.GetItem(channelId, env.ItemId);
if (env.FrameIndex == 0)
{
if (item is not null) throw new WebDropException(ErrorCode.BadRequest);
long max = env.IsGated ? Math.Clamp(maxDownloads ?? channel.Settings.FileMaxDownloads, 0, 1_000_000) : 0;
item = new Item(channelId, env.ItemId, principal.EndpointHash, env.ItemFlags, max, 0, 0, -1, null, 0,
now, now + ttlMs, null, label);
tx.InsertItem(item);
}
else
{
if (item is null) throw new WebDropException(ErrorCode.BadRequest);
if (item.Flags != env.ItemFlags) throw new WebDropException(ErrorCode.BadRequest);
if (item.CloseDeadline is not null) throw new WebDropException(ErrorCode.Gone);
if (item.FinalIndex is { } fi && env.FrameIndex > fi) throw new WebDropException(ErrorCode.BadRequest);
}
if (env.FrameIndex >= limits.MaxFramesPerItem) throw new WebDropException(ErrorCode.TooLarge);
if (env.IsFinal)
{
if (item.MaxIndex > env.FrameIndex) throw new WebDropException(ErrorCode.BadRequest);
if (item.FinalIndex is { } f && f != env.FrameIndex) throw new WebDropException(ErrorCode.BadRequest);
}
if (channel.Bytes + body.Length > limits.MaxChannelBytes) throw new WebDropException(ErrorCode.TooLarge);
long seq = tx.NextSeq(channelId);
tx.InsertFrame(new StoredFrame(channelId, env.ItemId, env.FrameIndex, seq, body.Length, digest,
blobKey is null ? body.ToArray() : null, blobKey));
if (!env.IsGated)
tx.InsertLog(channelId, new LogEntry(seq, LogKind.Frame, env.ItemId, env.FrameIndex, 0, now));
else if (env.FrameIndex == 0)
tx.InsertLog(channelId, new LogEntry(seq, LogKind.Announce, env.ItemId, 0, 0, now));
tx.UpdateItem(item with
{
FrameCount = item.FrameCount + 1,
MaxIndex = Math.Max(item.MaxIndex, env.FrameIndex),
FinalIndex = env.IsFinal ? env.FrameIndex : item.FinalIndex,
Bytes = item.Bytes + body.Length,
});
tx.AddChannelBytes(channelId, body.Length);
tx.TouchChannel(channelId, Math.Max(channel.ExpiresAt, now + ttlMs));
long lastSeq = seq;
var prefixes = new List<string>();
if (env.FrameIndex == 0)
{
// 新快照取代旧快照
if (env.IsSnapshot && tx.FindSnapshotItem(channelId, env.ItemId) is { } old)
lastSeq = DeleteItemCore(tx, channelId, old, TombstoneReason.Superseded, prefixes);
// 超出条数上限:淘汰最旧的条目
int excess = tx.CountItems(channelId) - channel.Settings.MaxItems;
if (excess > 0)
{
foreach (var victim in tx.ListOldestItems(channelId, excess + 1))
{
if (excess <= 0) break;
if (victim.ItemId == env.ItemId) continue;
lastSeq = DeleteItemCore(tx, channelId, victim, TombstoneReason.Evicted, prefixes);
excess--;
}
}
// 清理过期的墓碑,并推进续传水位
long pruned = tx.PruneTombstones(channelId, now - limits.TombstoneRetentionSeconds * 1000L);
if (pruned > channel.CompactedThrough) tx.SetCompactedThrough(channelId, pruned);
}
return new AppendOutcome(new AppendResult { Seq = seq, Created = true }, lastSeq, prefixes);
}
private static AppendResult Resolve(StoredFrame existing, byte[] digest) =>
existing.Digest.AsSpan().SequenceEqual(digest)
? new AppendResult { Seq = existing.Seq, Created = false }
: throw new WebDropException(ErrorCode.Conflict);
private void ValidateShape(Envelope env)
{
// 非 GATED 条目只能是单帧且必须 FINAL;快照只能是非 GATED 的单帧
if (!env.IsGated && (env.FrameIndex != 0 || !env.IsFinal)) throw new WebDropException(ErrorCode.BadRequest);
if (env.IsSnapshot && env.IsGated) throw new WebDropException(ErrorCode.BadRequest);
}
private static void CheckWritable(Channel channel, Principal principal, Envelope env)
{
if (!ChannelPolicy.CanWrite(channel, principal)) throw new WebDropException(ErrorCode.Forbidden);
if (env.IsEnc != (channel.Mode == ChannelMode.Enc)) throw new WebDropException(ErrorCode.ModeMismatch);
if (env.IsEnc && !PayloadTiers.IsValidEncPayloadLength(env.PayloadLength))
throw new WebDropException(ErrorCode.BadRequest);
}
private void DeleteBlobIfUnreferenced(string? blobKey)
{
if (blobKey is null) return;
if (!store.Read(tx => tx.BlobKeyExists(blobKey))) blobs.Delete(blobKey);
}
// ───────────────────────── 读取 ─────────────────────────
public async Task<LogPage> ReadLogAsync(Id16 id, Credentials creds, long from, int limit, CancellationToken ct = default)
{
limit = Math.Clamp(limit, 1, 1000);
var (pending, lastSeq) = store.Read(tx =>
{
var (channel, principal) = Authenticate(tx, id, creds, SeatRule.Required);
if (!ChannelPolicy.CanRead(principal)) throw new WebDropException(ErrorCode.Forbidden);
// 续传点早于已被清理的墓碑:客户端可能漏掉删除事件,必须从 0 全量同步
if (from > 0 && from < channel.CompactedThrough + 1)
throw new WebDropException(ErrorCode.Gone, channel.CompactedThrough + 1);
var list = new List<(long Seq, byte[]? Bytes, StoredFrame? Frame)>();
foreach (var e in tx.ReadLog(id, from, limit))
{
switch (e.Kind)
{
case LogKind.Frame:
if (tx.GetFrame(id, e.ItemId, e.FrameIndex) is { } f) list.Add((e.Seq, null, f));
break;
case LogKind.Announce:
if (tx.GetItem(id, e.ItemId) is { } it)
list.Add((e.Seq, LogRecord.EncodeAnnounce(e.Seq, e.ItemId, it.Flags, it.CreatedAt), null));
break;
case LogKind.Tombstone:
list.Add((e.Seq, LogRecord.EncodeTombstone(e.Seq, e.ItemId, (TombstoneReason)e.Aux), null));
break;
}
}
return (list, channel.LastSeq);
});
var records = new List<LogRecord>(pending.Count);
foreach (var (seq, bytes, frame) in pending)
{
if (bytes is not null) { records.Add(new LogRecord(seq, bytes)); continue; }
byte[]? env = await LoadFrameBytesAsync(frame!, ct).ConfigureAwait(false);
if (env is not null) records.Add(new LogRecord(seq, LogRecord.Encode(seq, LogKind.Frame, env)));
}
return new LogPage(records, lastSeq);
}
/// <summary>开始拉取条目:鉴权,并按规则计一次下载。<paramref name="fromIndex"/> 为 0 才计数,大于 0 视为续传。</summary>
public void StartItemDownload(Id16 id, Credentials creds, Id16 itemId, uint fromIndex)
{
store.Write(tx =>
{
var (_, principal) = Authenticate(tx, id, creds, SeatRule.Required);
if (!ChannelPolicy.CanRead(principal)) throw new WebDropException(ErrorCode.Forbidden);
var item = tx.GetItem(id, itemId) ?? throw new WebDropException(ErrorCode.NotFound);
long now = clock.NowMs;
if (item.CloseDeadline is { } deadline && (fromIndex == 0 || now > deadline))
throw new WebDropException(ErrorCode.Gone);
if (fromIndex == 0 && item.IsGated)
{
if (item.MaxDownloads > 0 && item.Downloads >= item.MaxDownloads)
throw new WebDropException(ErrorCode.Gone);
long downloads = item.Downloads + 1;
long? closeAt = item.MaxDownloads > 0 && downloads >= item.MaxDownloads
? now + limits.DownloadGraceSeconds * 1000L
: item.CloseDeadline;
tx.UpdateItem(item with { Downloads = downloads, CloseDeadline = closeAt });
}
return 0;
});
}
/// <summary>读取自 <paramref name="fromIndex"/> 起连续的帧。条目未完成时可能返回空批次,调用方等待后重试。</summary>
public async Task<ItemChunk> ReadItemFramesAsync(
Id16 id, Credentials creds, Id16 itemId, uint fromIndex, int maxFrames, CancellationToken ct = default)
{
maxFrames = Math.Clamp(maxFrames, 1, 64);
var (frames, next, complete) = store.Read(tx =>
{
var (_, principal) = Authenticate(tx, id, creds, SeatRule.Required);
if (!ChannelPolicy.CanRead(principal)) throw new WebDropException(ErrorCode.Forbidden);
var item = tx.GetItem(id, itemId) ?? throw new WebDropException(ErrorCode.NotFound);
if (item.CloseDeadline is { } d && clock.NowMs > d) throw new WebDropException(ErrorCode.Gone);
var taken = new List<StoredFrame>();
uint expected = fromIndex;
foreach (var f in tx.ListFrames(id, itemId, fromIndex, maxFrames))
{
if (f.Index != expected) break; // 并行上传时可能有空洞
taken.Add(f);
expected++;
}
return (taken, expected, item.FinalIndex is { } fi && expected > fi);
});
var result = new List<byte[]>(frames.Count);
foreach (var f in frames)
{
byte[]? bytes = await LoadFrameBytesAsync(f, ct).ConfigureAwait(false);
if (bytes is null) throw new WebDropException(ErrorCode.Gone);
result.Add(bytes);
}
return new ItemChunk(result, next, complete);
}
private async ValueTask<byte[]?> LoadFrameBytesAsync(StoredFrame frame, CancellationToken ct) =>
frame.Inline ?? (frame.BlobKey is null ? null : await blobs.ReadAsync(frame.BlobKey, ct).ConfigureAwait(false));
// ───────────────────────── 删除 ─────────────────────────
public void DeleteItem(Id16 id, Credentials creds, Id16 itemId)
{
var (seq, prefixes) = store.Write(tx =>
{
var (channel, principal) = Authenticate(tx, id, creds, SeatRule.Required);
var item = tx.GetItem(id, itemId) ?? throw new WebDropException(ErrorCode.NotFound);
if (!ChannelPolicy.CanDeleteItem(channel, item, principal)) throw new WebDropException(ErrorCode.Forbidden);
var list = new List<string>();
long s = DeleteItemCore(tx, id, item, TombstoneReason.Deleted, list);
return (s, list);
});
foreach (string prefix in prefixes) blobs.DeletePrefix(prefix);
events.Advanced(id, seq);
}
private long DeleteItemCore(IResourceTx tx, Id16 channelId, Item item, TombstoneReason reason, List<string> prefixes)
{
tx.DeleteItemRows(channelId, item.ItemId);
tx.AddChannelBytes(channelId, -item.Bytes);
long seq = tx.NextSeq(channelId);
tx.InsertLog(channelId, new LogEntry(seq, LogKind.Tombstone, item.ItemId, 0, (byte)reason, clock.NowMs));
prefixes.Add($"{channelId.ToHex()}/{item.ItemId.ToHex()}");
return seq;
}
// ───────────────────────── 后台清理 ─────────────────────────
/// <summary>清理过期条目、下载达上限的条目、过期频道;可选清理孤立 blob。返回删除的条目和频道总数。</summary>
public int Sweep(bool includeOrphanBlobs = false)
{
int removed = 0;
long now = clock.NowMs;
foreach (var group in store.Read(tx => tx.ListExpiredItems(now, 500)).GroupBy(i => i.ChannelId))
{
var (lastSeq, prefixes, count) = store.Write(tx =>
{
long last = 0;
int n = 0;
var list = new List<string>();
foreach (var stale in group)
{
// 重新读取,期间条目可能已被删除或续期
if (tx.GetItem(stale.ChannelId, stale.ItemId) is not { } item) continue;
bool closed = item.CloseDeadline is { } d && d <= now;
if (item.ExpiresAt > now && !closed) continue;
last = DeleteItemCore(tx, item.ChannelId, item, closed ? TombstoneReason.DownloadLimit : TombstoneReason.Expired, list);
n++;
}
return (last, list, n);
});
foreach (string prefix in prefixes) blobs.DeletePrefix(prefix);
if (count > 0) events.Advanced(group.Key, lastSeq);
removed += count;
}
foreach (var channelId in store.Read(tx => tx.ListExpiredChannels(now, 200)))
{
bool deleted = store.Write(tx =>
{
if (tx.GetChannel(channelId) is not { } c || c.ExpiresAt > now) return false;
tx.DeleteChannelRows(channelId);
return true;
});
if (!deleted) continue;
blobs.DeletePrefix(channelId.ToHex());
events.ChannelDeleted(channelId);
removed++;
}
if (includeOrphanBlobs) SweepOrphanBlobs(now);
return removed;
}
/// <summary>删除没有任何帧引用、且超过 10 分钟的 blob(崩溃发生在写 blob 与提交事务之间会留下它们)。</summary>
private void SweepOrphanBlobs(long now)
{
var cutoff = DateTimeOffset.FromUnixTimeMilliseconds(now).AddMinutes(-10);
foreach (var (key, lastWrite) in blobs.Enumerate().ToList())
{
if (lastWrite > cutoff) continue;
if (!store.Read(tx => tx.BlobKeyExists(key))) blobs.Delete(key);
}
}
// ───────────────────────── 鉴权 ─────────────────────────
private (Channel Channel, Principal Principal) Authenticate(IResourceTx tx, Id16 id, Credentials creds, SeatRule seat)
{
var channel = tx.GetChannel(id);
if (channel is null || channel.ExpiresAt <= clock.NowMs) throw new WebDropException(ErrorCode.NotFound);
var role = VerifyRole(channel, creds);
if (seat == SeatRule.Required && tx.GetSeat(id, creds.EndpointHash) is null)
throw new WebDropException(ErrorCode.NotJoined);
return (channel, new Principal(role, creds.EndpointHash));
}
private static Role VerifyRole(Channel channel, Credentials creds)
{
switch (creds.ClaimedRole)
{
case Role.Read:
// raw 频道:知道 id 即可读;enc 频道必须出示读令牌
if (channel.Mode == ChannelMode.Raw) return Role.Read;
if (creds.Token is not null && TokenHash.Matches(creds.Token, channel.ReadHash)) return Role.Read;
break;
case Role.Write:
if (creds.Token is not null && TokenHash.Matches(creds.Token, channel.WriteHash)) return Role.Write;
break;
case Role.Owner:
if (creds.Token is not null && TokenHash.Matches(creds.Token, channel.OwnerHash)) return Role.Owner;
break;
}
throw new WebDropException(ErrorCode.Unauthorized);
}
private static void RequireOwner(Principal principal)
{
if (!ChannelPolicy.IsOwner(principal)) throw new WebDropException(ErrorCode.Forbidden);
}
private static ChannelInfo ToInfo(Channel channel, int seatsUsed) => new()
{
Id = channel.Id.ToBase64Url(),
Mode = channel.Mode,
Settings = channel.Settings,
SeatsUsed = seatsUsed,
OldestSeq = channel.CompactedThrough + 1,
LastSeq = channel.LastSeq,
CreatedAt = channel.CreatedAt,
ExpiresAt = channel.ExpiresAt,
};
}
+125
View File
@@ -0,0 +1,125 @@
using System.Buffers.Binary;
using System.Text.Json.Serialization;
namespace WebDrop.Core;
public sealed class CreateChannelRequest
{
[JsonPropertyName("id")] public string? Id { get; init; }
[JsonPropertyName("mode")] public ChannelMode? Mode { get; init; }
[JsonPropertyName("settings")] public ChannelSettings? Settings { get; init; }
[JsonPropertyName("read_hash")] public string? ReadHash { get; init; }
[JsonPropertyName("write_hash")] public string? WriteHash { get; init; }
[JsonPropertyName("owner_hash")] public string? OwnerHash { get; init; }
}
public sealed class CreateShareRequest
{
[JsonPropertyName("id")] public string? Id { get; init; }
[JsonPropertyName("read_hash")] public string? ReadHash { get; init; }
[JsonPropertyName("owner_hash")] public string? OwnerHash { get; init; }
[JsonPropertyName("settings")] public ShareSettings? Settings { get; init; }
}
public sealed class FinishShareRequest
{
[JsonPropertyName("manifest_item_id")] public string? ManifestItemId { get; init; }
}
public sealed class ShareInfo
{
[JsonPropertyName("id")] public required string Id { get; init; }
[JsonPropertyName("state")] public required string State { get; init; }
[JsonPropertyName("expires_at")] public required long ExpiresAt { get; init; }
[JsonPropertyName("created_at")] public required long CreatedAt { get; init; }
[JsonPropertyName("sealed_at")] public long? SealedAt { get; init; }
[JsonPropertyName("manifest_item_id")] public string? ManifestItemId { get; init; }
[JsonPropertyName("settings")] public required ShareSettings Settings { get; init; }
[JsonPropertyName("items")] public required IReadOnlyList<ShareItemInfo> Items { get; init; }
}
public sealed class ShareItemInfo
{
[JsonPropertyName("item_id")] public required string ItemId { get; init; }
[JsonPropertyName("bytes")] public required long Bytes { get; init; }
[JsonPropertyName("frames")] public required int Frames { get; init; }
[JsonPropertyName("max_downloads")] public required long MaxDownloads { get; init; }
[JsonPropertyName("downloads_left")] [JsonIgnore(Condition = JsonIgnoreCondition.Never)] public long? DownloadsLeft { get; init; }
}
public sealed class ChannelInfo
{
[JsonPropertyName("id")] public required string Id { get; init; }
[JsonPropertyName("mode")] public required ChannelMode Mode { get; init; }
[JsonPropertyName("settings")] public required ChannelSettings Settings { get; init; }
[JsonPropertyName("seats_used")] public required int SeatsUsed { get; init; }
/// <summary>从该 seq 起的续传有效;0 &lt; from &lt; oldest_seq 的续传会得到 410,需从 0 全量同步。</summary>
[JsonPropertyName("oldest_seq")] public required long OldestSeq { get; init; }
[JsonPropertyName("last_seq")] public required long LastSeq { get; init; }
[JsonPropertyName("created_at")] public required long CreatedAt { get; init; }
[JsonPropertyName("expires_at")] public required long ExpiresAt { get; init; }
}
public sealed class SeatJoined
{
[JsonPropertyName("endpoint_hash")] public required string EndpointHash { get; init; }
[JsonPropertyName("joined_at")] public required long JoinedAt { get; init; }
[JsonPropertyName("seats_used")] public required int SeatsUsed { get; init; }
[JsonPropertyName("seats_max")] public required int SeatsMax { get; init; }
}
public sealed class SeatInfo
{
[JsonPropertyName("endpoint_hash")] public required string EndpointHash { get; init; }
[JsonPropertyName("joined_at")] public required long JoinedAt { get; init; }
[JsonPropertyName("label")] public string? Label { get; init; }
}
public sealed class AppendResult
{
[JsonPropertyName("seq")] public required long Seq { get; init; }
/// <summary>true 为新追加(201),false 为幂等重发(200)。</summary>
[JsonIgnore] public bool Created { get; init; }
}
/// <summary>一条日志记录的线格式:seq(u64) ‖ kind(u8) ‖ body。</summary>
public sealed record LogRecord(long Seq, byte[] Bytes)
{
public static byte[] Encode(long seq, LogKind kind, ReadOnlySpan<byte> body)
{
var bytes = new byte[9 + body.Length];
BinaryPrimitives.WriteInt64BigEndian(bytes, seq);
bytes[8] = (byte)kind;
body.CopyTo(bytes.AsSpan(9));
return bytes;
}
public static byte[] EncodeTombstone(long seq, Id16 itemId, TombstoneReason reason)
{
Span<byte> body = stackalloc byte[Id16.Size + 1];
itemId.WriteTo(body);
body[Id16.Size] = (byte)reason;
return Encode(seq, LogKind.Tombstone, body);
}
public static byte[] EncodeAnnounce(long seq, Id16 itemId, FrameFlags flags, long createdMs)
{
Span<byte> body = stackalloc byte[Id16.Size + 1 + 8];
itemId.WriteTo(body);
body[Id16.Size] = (byte)flags;
BinaryPrimitives.WriteInt64BigEndian(body[(Id16.Size + 1)..], createdMs);
return Encode(seq, LogKind.Announce, body);
}
}
public sealed record LogPage(IReadOnlyList<LogRecord> Records, long LastSeq);
/// <summary>GATED 条目拉取的一个批次:自 FromIndex 起连续的完整封装。</summary>
public sealed record ItemChunk(IReadOnlyList<byte[]> Frames, uint NextIndex, bool Complete);
[JsonSerializable(typeof(ChannelSettings))]
[JsonSerializable(typeof(ShareSettings))]
public sealed partial class CoreJsonContext : JsonSerializerContext;
+53
View File
@@ -0,0 +1,53 @@
using System.Text.Json.Serialization;
namespace WebDrop.Core;
[JsonConverter(typeof(JsonStringEnumConverter<ChannelMode>))]
public enum ChannelMode : byte
{
[JsonStringEnumMemberName("enc")] Enc = 1,
[JsonStringEnumMemberName("raw")] Raw = 2,
}
[JsonConverter(typeof(JsonStringEnumConverter<DeletePolicy>))]
public enum DeletePolicy : byte
{
[JsonStringEnumMemberName("owner_only")] OwnerOnly = 0,
[JsonStringEnumMemberName("own")] Own = 1,
[JsonStringEnumMemberName("anyone")] Anyone = 2,
}
/// <summary>请求声明并通过验证的角色,数值越大权限越高。</summary>
public enum Role : byte
{
None = 0,
Read = 1,
Write = 2,
Owner = 3,
}
[Flags]
public enum FrameFlags : byte
{
None = 0,
Final = 1,
Snapshot = 2,
Gated = 4,
Enc = 8,
}
public enum LogKind : byte
{
Frame = 1,
Tombstone = 2,
Announce = 3,
}
public enum TombstoneReason : byte
{
Deleted = 0,
Expired = 1,
Evicted = 2,
DownloadLimit = 3,
Superseded = 4,
}
+50
View File
@@ -0,0 +1,50 @@
using System.Buffers.Binary;
namespace WebDrop.Core;
/// <summary>帧封装的 26 字节明文头部,见 protocol.md §4。负载由服务端原样保存,从不解析。</summary>
public readonly record struct Envelope(FrameFlags Flags, Id16 ItemId, uint FrameIndex, int PayloadLength)
{
public const int HeaderSize = 26;
public const byte Version = 1;
public int TotalLength => HeaderSize + PayloadLength;
/// <summary>头部格式合法、保留位为 0、长度与 payload_len 严格一致。</summary>
public static bool TryParse(ReadOnlySpan<byte> data, out Envelope envelope)
{
envelope = default;
if (data.Length < HeaderSize || data[0] != Version) return false;
byte flags = data[1];
if ((flags & 0xF0) != 0) return false;
uint payloadLen = BinaryPrimitives.ReadUInt32BigEndian(data[22..]);
if (payloadLen > int.MaxValue || data.Length != HeaderSize + (int)payloadLen) return false;
envelope = new Envelope(
(FrameFlags)flags,
Id16.FromBytes(data.Slice(2, Id16.Size)),
BinaryPrimitives.ReadUInt32BigEndian(data[18..]),
(int)payloadLen);
return true;
}
public bool IsFinal => (Flags & FrameFlags.Final) != 0;
public bool IsGated => (Flags & FrameFlags.Gated) != 0;
public bool IsSnapshot => (Flags & FrameFlags.Snapshot) != 0;
public bool IsEnc => (Flags & FrameFlags.Enc) != 0;
/// <summary>条目级标志:同一条目的所有帧必须一致。</summary>
public FrameFlags ItemFlags => Flags & (FrameFlags.Snapshot | FrameFlags.Gated);
}
public static class PayloadTiers
{
public const int TagSize = 16;
private static readonly int[] Tiers = [1024, 4096, 16384, 65536, 262144, 1048576];
/// <summary>enc 负载 = 档位明文 + 16 字节 GCM 标签,服务端据此强制 padding。</summary>
public static bool IsValidEncPayloadLength(int payloadLength) =>
payloadLength > TagSize && Array.IndexOf(Tiers, payloadLength - TagSize) >= 0;
}
+75
View File
@@ -0,0 +1,75 @@
namespace WebDrop.Core;
public enum ErrorCode
{
BadRequest,
Unauthorized,
Forbidden,
NotJoined,
SeatsFull,
ModeMismatch,
NotFound,
Conflict,
Gone,
TooLarge,
CaptchaRequired,
CaptchaUnavailable,
RateLimited,
VisitorsFull,
Sealed,
}
/// <summary>预期内的业务失败。消息只含错误码,绝不回显用户提交的内容。</summary>
public sealed class WebDropException : Exception
{
public ErrorCode Code { get; }
/// <summary>仅在 410 日志续传失败时有值,见 api.md。</summary>
public long? OldestSeq { get; }
public WebDropException(ErrorCode code, long? oldestSeq = null) : base(code.ToWire())
{
Code = code;
OldestSeq = oldestSeq;
}
}
public static class ErrorCodes
{
public static string ToWire(this ErrorCode code) => code switch
{
ErrorCode.BadRequest => "bad_request",
ErrorCode.Unauthorized => "unauthorized",
ErrorCode.Forbidden => "forbidden",
ErrorCode.NotJoined => "not_joined",
ErrorCode.SeatsFull => "seats_full",
ErrorCode.ModeMismatch => "mode_mismatch",
ErrorCode.NotFound => "not_found",
ErrorCode.Conflict => "conflict",
ErrorCode.Gone => "gone",
ErrorCode.TooLarge => "too_large",
ErrorCode.CaptchaRequired => "captcha_required",
ErrorCode.CaptchaUnavailable => "captcha_unavailable",
ErrorCode.RateLimited => "rate_limited",
ErrorCode.VisitorsFull => "visitors_full",
ErrorCode.Sealed => "sealed",
_ => "bad_request",
};
public static int ToStatus(this ErrorCode code) => code switch
{
ErrorCode.BadRequest => 400,
ErrorCode.Unauthorized => 401,
ErrorCode.Forbidden or ErrorCode.NotJoined or ErrorCode.SeatsFull or ErrorCode.ModeMismatch => 403,
ErrorCode.NotFound => 404,
ErrorCode.Conflict => 409,
ErrorCode.Sealed => 409,
ErrorCode.Gone => 410,
ErrorCode.TooLarge => 413,
ErrorCode.CaptchaRequired => 428,
ErrorCode.CaptchaUnavailable => 503,
ErrorCode.RateLimited => 429,
ErrorCode.VisitorsFull => 403,
_ => 400,
};
}
+91
View File
@@ -0,0 +1,91 @@
using System.Buffers.Binary;
using System.Buffers.Text;
namespace WebDrop.Core;
/// <summary>16 字节标识(频道 id、条目 id),可作字典键。</summary>
public readonly struct Id16 : IEquatable<Id16>
{
public const int Size = 16;
private const int Base64Length = 22;
private readonly ulong _hi;
private readonly ulong _lo;
private Id16(ulong hi, ulong lo)
{
_hi = hi;
_lo = lo;
}
public static Id16 FromBytes(ReadOnlySpan<byte> bytes)
{
if (bytes.Length != Size) throw new ArgumentException("Id16 需要 16 字节", nameof(bytes));
return new Id16(BinaryPrimitives.ReadUInt64BigEndian(bytes), BinaryPrimitives.ReadUInt64BigEndian(bytes[8..]));
}
/// <summary>只接受规范的 22 字符 base64url,拒绝同一值的非规范编码。</summary>
public static bool TryParse(ReadOnlySpan<char> text, out Id16 id)
{
id = default;
if (text.Length != Base64Length) return false;
Span<byte> buf = stackalloc byte[Size];
if (!Codec.TryDecodeExact(text, buf)) return false;
id = FromBytes(buf);
return true;
}
public void WriteTo(Span<byte> destination)
{
BinaryPrimitives.WriteUInt64BigEndian(destination, _hi);
BinaryPrimitives.WriteUInt64BigEndian(destination[8..], _lo);
}
public byte[] ToArray()
{
var bytes = new byte[Size];
WriteTo(bytes);
return bytes;
}
public string ToBase64Url() => Codec.Encode(ToArray());
/// <summary>小写十六进制。用于文件路径,避免大小写不敏感的文件系统造成不同 id 冲突。</summary>
public string ToHex() => Convert.ToHexStringLower(ToArray());
public bool Equals(Id16 other) => _hi == other._hi && _lo == other._lo;
public override bool Equals(object? obj) => obj is Id16 other && Equals(other);
public override int GetHashCode() => HashCode.Combine(_hi, _lo);
public static bool operator ==(Id16 a, Id16 b) => a.Equals(b);
public static bool operator !=(Id16 a, Id16 b) => !a.Equals(b);
public override string ToString() => ToBase64Url();
}
/// <summary>base64url(无填充)编解码,解码要求规范编码。</summary>
public static class Codec
{
public static string Encode(ReadOnlySpan<byte> bytes) => Base64Url.EncodeToString(bytes);
/// <summary>解码为恰好 destination.Length 字节,且重新编码后必须与输入逐字符一致。</summary>
public static bool TryDecodeExact(ReadOnlySpan<char> text, Span<byte> destination)
{
// 注意:Base64Url.TryDecodeFromChars 遇到非法输入会抛 FormatException,这里用返回状态码的版本
var status = Base64Url.DecodeFromChars(text, destination, out int consumed, out int written, isFinalBlock: true);
if (status != System.Buffers.OperationStatus.Done || consumed != text.Length || written != destination.Length)
return false;
Span<char> check = stackalloc char[Base64Url.GetEncodedLength(destination.Length)];
return Base64Url.TryEncodeToChars(destination, check, out int n)
&& n == text.Length
&& check[..n].SequenceEqual(text);
}
public static bool TryDecodeExact(string? text, int length, out byte[] bytes)
{
bytes = [];
if (string.IsNullOrEmpty(text) || text.Length != Base64Url.GetEncodedLength(length)) return false;
var buf = new byte[length];
if (!TryDecodeExact(text.AsSpan(), buf)) return false;
bytes = buf;
return true;
}
}
+100
View File
@@ -0,0 +1,100 @@
namespace WebDrop.Core;
public sealed record Channel(
Id16 Id,
ChannelMode Mode,
ChannelSettings Settings,
byte[]? ReadHash,
byte[]? WriteHash,
byte[] OwnerHash,
long CreatedAt,
long ExpiresAt,
long LastSeq,
long CompactedThrough,
long Bytes);
public sealed record Share(
Id16 Id,
ShareSettings Settings,
byte[] ReadHash,
byte[]? OwnerHash,
byte[]? CreatorEndpointHash,
long CreatedAt,
long ExpiresAt,
long? SealedAt,
Id16? ManifestItemId,
long Bytes)
{
public bool IsSealed => SealedAt is not null;
}
public sealed record ShareItem(
Id16 ShareId,
Id16 ItemId,
FrameFlags Flags,
long MaxDownloads,
long Downloads,
int FrameCount,
long MaxIndex,
long? FinalIndex,
long Bytes,
long CreatedAt,
long ExpiresAt,
long? CloseDeadline)
{
public bool IsGated => (Flags & FrameFlags.Gated) != 0;
public bool IsSnapshot => (Flags & FrameFlags.Snapshot) != 0;
public bool IsComplete => FinalIndex is { } f && FrameCount == f + 1;
}
public sealed record ShareFrame(
Id16 ShareId,
Id16 ItemId,
uint Index,
long AckSeq,
int Length,
byte[] Digest,
byte[]? Inline,
string? BlobKey);
public sealed record Item(
Id16 ChannelId,
Id16 ItemId,
byte[] AuthorHash,
FrameFlags Flags,
long MaxDownloads,
long Downloads,
int FrameCount,
long MaxIndex,
long? FinalIndex,
long Bytes,
long CreatedAt,
long ExpiresAt,
long? CloseDeadline,
string? Label)
{
public bool IsGated => (Flags & FrameFlags.Gated) != 0;
public bool IsSnapshot => (Flags & FrameFlags.Snapshot) != 0;
public bool IsComplete => FinalIndex is { } f && FrameCount == f + 1;
}
public sealed record Seat(byte[] EndpointHash, long JoinedAt, byte[]? Label);
/// <summary>已存储的一帧(完整封装)。Inline 与 BlobKey 二选一。</summary>
public sealed record StoredFrame(
Id16 ChannelId,
Id16 ItemId,
uint Index,
long Seq,
int Length,
byte[] Digest,
byte[]? Inline,
string? BlobKey);
public sealed record LogEntry(long Seq, LogKind Kind, Id16 ItemId, uint FrameIndex, byte Aux, long CreatedAt);
/// <summary>一次请求验证通过后的身份。</summary>
public sealed record Principal(Role Role, byte[] EndpointHash);
/// <summary>请求携带的凭据,尚未验证。</summary>
public sealed record Credentials(Role ClaimedRole, byte[]? Token, byte[] EndpointHash);
+90
View File
@@ -0,0 +1,90 @@
using System.Text.Json.Serialization;
namespace WebDrop.Core;
/// <summary>服务端限制,由配置绑定。</summary>
public sealed class ServerLimits
{
public int MaxPayloadBytes { get; set; } = (1 << 20) + 16;
public int MaxFramesPerItem { get; set; } = 65_536;
public long MaxChannelBytes { get; set; } = 1L << 30;
public int MaxSeats { get; set; } = 64;
public int MinTtlSeconds { get; set; } = 60;
public int MaxTtlSeconds { get; set; } = 604_800;
public int MaxItemsLimit { get; set; } = 10_000;
public int InlineThresholdBytes { get; set; } = 64 * 1024;
public int DownloadGraceSeconds { get; set; } = 300;
public int TombstoneRetentionSeconds { get; set; } = 604_800;
public int MaxLabelBytes { get; set; } = 255;
public int MaxSeatLabelBytes { get; set; } = 256;
}
public sealed class ChannelSettings
{
[JsonPropertyName("seats")] public int Seats { get; init; } = 2;
[JsonPropertyName("members_can_upload")] public bool MembersCanUpload { get; init; } = true;
[JsonPropertyName("delete_policy")] public DeletePolicy DeletePolicy { get; init; } = DeletePolicy.Own;
[JsonPropertyName("ttl_seconds")] public int TtlSeconds { get; init; } = 86_400;
[JsonPropertyName("max_items")] public int MaxItems { get; init; } = 1000;
[JsonPropertyName("file_max_downloads")] public int FileMaxDownloads { get; init; } = 10;
[JsonPropertyName("ui_restrict_copy_download")] public bool UiRestrictCopyDownload { get; init; }
public void Validate(ServerLimits limits)
{
if (Seats < 1 || Seats > limits.MaxSeats) throw new WebDropException(ErrorCode.BadRequest);
if (TtlSeconds < limits.MinTtlSeconds || TtlSeconds > limits.MaxTtlSeconds) throw new WebDropException(ErrorCode.BadRequest);
if (MaxItems < 1 || MaxItems > limits.MaxItemsLimit) throw new WebDropException(ErrorCode.BadRequest);
if (FileMaxDownloads < 0) throw new WebDropException(ErrorCode.BadRequest);
}
}
public sealed class ChannelSettingsPatch
{
[JsonPropertyName("seats")] public int? Seats { get; init; }
[JsonPropertyName("members_can_upload")] public bool? MembersCanUpload { get; init; }
[JsonPropertyName("delete_policy")] public DeletePolicy? DeletePolicy { get; init; }
[JsonPropertyName("ttl_seconds")] public int? TtlSeconds { get; init; }
[JsonPropertyName("max_items")] public int? MaxItems { get; init; }
[JsonPropertyName("file_max_downloads")] public int? FileMaxDownloads { get; init; }
[JsonPropertyName("ui_restrict_copy_download")] public bool? UiRestrictCopyDownload { get; init; }
public ChannelSettings ApplyTo(ChannelSettings s) => new()
{
Seats = Seats ?? s.Seats,
MembersCanUpload = MembersCanUpload ?? s.MembersCanUpload,
DeletePolicy = DeletePolicy ?? s.DeletePolicy,
TtlSeconds = TtlSeconds ?? s.TtlSeconds,
MaxItems = MaxItems ?? s.MaxItems,
FileMaxDownloads = FileMaxDownloads ?? s.FileMaxDownloads,
UiRestrictCopyDownload = UiRestrictCopyDownload ?? s.UiRestrictCopyDownload,
};
}
public sealed class ShareSettings
{
[JsonPropertyName("ttl_seconds")] public int TtlSeconds { get; init; } = 86_400;
[JsonPropertyName("max_visitors")] public int MaxVisitors { get; init; }
[JsonPropertyName("file_max_downloads")] public int FileMaxDownloads { get; init; } = 10;
public void Validate(ServerLimits limits)
{
if (TtlSeconds < limits.MinTtlSeconds || TtlSeconds > limits.MaxTtlSeconds
|| MaxVisitors < 0 || MaxVisitors > 10_000
|| FileMaxDownloads < 0)
throw new WebDropException(ErrorCode.BadRequest);
}
}
public sealed class ShareSettingsPatch
{
[JsonPropertyName("ttl_seconds")] public int? TtlSeconds { get; init; }
[JsonPropertyName("max_visitors")] public int? MaxVisitors { get; init; }
[JsonPropertyName("file_max_downloads")] public int? FileMaxDownloads { get; init; }
public ShareSettings ApplyTo(ShareSettings s) => new()
{
TtlSeconds = TtlSeconds ?? s.TtlSeconds,
MaxVisitors = MaxVisitors ?? s.MaxVisitors,
FileMaxDownloads = FileMaxDownloads ?? s.FileMaxDownloads,
};
}
+359
View File
@@ -0,0 +1,359 @@
using System.Security.Cryptography;
namespace WebDrop.Core;
/// <summary>便捷分享的草稿、封存、访客和下载规则。与频道共享帧格式及 blob 存储,不共享席位或日志。</summary>
public sealed class ShareService(
IResourceStore store,
IBlobStore blobs,
IClock clock,
ServerLimits limits,
int draftTtlSeconds)
{
public ShareInfo Create(CreateShareRequest req, byte[]? creatorEndpointHash)
{
if (!Id16.TryParse(req.Id, out var id)
|| !Codec.TryDecodeExact(req.ReadHash, TokenHash.Size, out var readHash)
|| !Codec.TryDecodeExact(req.OwnerHash, TokenHash.Size, out var ownerHash))
throw new WebDropException(ErrorCode.BadRequest);
var settings = req.Settings ?? new ShareSettings();
settings.Validate(limits);
long now = clock.NowMs;
var share = new Share(id, settings, readHash, ownerHash, creatorEndpointHash, now,
now + Math.Max(60, draftTtlSeconds) * 1000L, null, null, 0);
store.Write(tx =>
{
if (tx.HasResourceId(id)) throw new WebDropException(ErrorCode.Conflict);
tx.InsertShare(share);
return 0;
});
return ToInfo(share, []);
}
public ShareInfo GetInfo(Id16 id, Credentials creds) => store.Read(tx =>
{
var share = Live(tx.GetShare(id));
if (share.IsSealed)
{
if (creds.ClaimedRole != Role.Read || !TokenHash.Matches(creds.Token ?? [], share.ReadHash))
throw new WebDropException(ErrorCode.Unauthorized);
}
else
{
if (creds.ClaimedRole == Role.Read) throw new WebDropException(ErrorCode.NotFound);
RequireOwner(share, creds);
}
return ToInfo(share, tx.ListShareItems(id));
});
public ShareInfo UpdateSettings(Id16 id, Credentials creds, ShareSettingsPatch patch) => store.Write(tx =>
{
var share = Draft(tx.GetShare(id), creds);
var settings = patch.ApplyTo(share.Settings);
settings.Validate(limits);
share = share with { Settings = settings };
tx.UpdateShare(share);
return ToInfo(share, tx.ListShareItems(id));
});
public async Task<AppendResult> AppendFrameAsync(
Id16 id, Credentials creds, ReadOnlyMemory<byte> body, long? maxDownloads, CancellationToken ct = default)
{
if (!Envelope.TryParse(body.Span, out var env)) throw new WebDropException(ErrorCode.BadRequest);
if (env.PayloadLength > limits.MaxPayloadBytes || body.Length > limits.MaxPayloadBytes + Envelope.HeaderSize)
throw new WebDropException(ErrorCode.TooLarge);
if (!env.IsEnc) throw new WebDropException(ErrorCode.ModeMismatch);
if (!PayloadTiers.IsValidEncPayloadLength(env.PayloadLength)) throw new WebDropException(ErrorCode.BadRequest);
if (env.IsSnapshot) throw new WebDropException(ErrorCode.BadRequest);
if (!env.IsGated && (env.FrameIndex != 0 || !env.IsFinal)) throw new WebDropException(ErrorCode.BadRequest);
if (maxDownloads is < 0) throw new WebDropException(ErrorCode.BadRequest);
var existing = store.Read(tx =>
{
_ = Draft(tx.GetShare(id), creds);
return tx.GetShareFrame(id, env.ItemId, env.FrameIndex);
});
byte[] digest = SHA256.HashData(body.Span);
if (existing is not null) return Resolve(existing, digest);
string? blobKey = null;
if (body.Length >= limits.InlineThresholdBytes)
{
blobKey = $"{id.ToHex()}/{env.ItemId.ToHex()}/{env.FrameIndex}.{Convert.ToHexStringLower(digest.AsSpan(0, 16))}";
await blobs.PutAsync(blobKey, body, ct).ConfigureAwait(false);
}
try
{
var result = store.Write(tx =>
{
var share = Draft(tx.GetShare(id), creds);
if (tx.GetShareFrame(id, env.ItemId, env.FrameIndex) is { } duplicate)
return Resolve(duplicate, digest);
long now = clock.NowMs;
var item = tx.GetShareItem(id, env.ItemId);
if (env.FrameIndex == 0)
{
if (item is not null) throw new WebDropException(ErrorCode.BadRequest);
if (tx.CountShareItems(id) >= limits.MaxItemsLimit) throw new WebDropException(ErrorCode.TooLarge);
long cap = env.IsGated ? maxDownloads ?? share.Settings.FileMaxDownloads : 0;
item = new ShareItem(id, env.ItemId, env.ItemFlags, cap, 0, 0, -1, null,
0, now, share.ExpiresAt, null);
tx.InsertShareItem(item);
}
else
{
if (item is null || item.Flags != env.ItemFlags || item.CloseDeadline is not null
|| item.FinalIndex is { } final && env.FrameIndex > final)
throw new WebDropException(ErrorCode.BadRequest);
}
if (env.FrameIndex >= limits.MaxFramesPerItem) throw new WebDropException(ErrorCode.TooLarge);
if (env.IsFinal && (item!.MaxIndex > env.FrameIndex || item.FinalIndex is { } f && f != env.FrameIndex))
throw new WebDropException(ErrorCode.BadRequest);
if (share.Bytes + body.Length > limits.MaxChannelBytes) throw new WebDropException(ErrorCode.TooLarge);
tx.InsertShareFrame(new ShareFrame(id, env.ItemId, env.FrameIndex, env.FrameIndex, body.Length, digest,
blobKey is null ? body.ToArray() : null, blobKey));
tx.UpdateShareItem(item! with
{
FrameCount = item!.FrameCount + 1,
MaxIndex = Math.Max(item.MaxIndex, env.FrameIndex),
FinalIndex = env.IsFinal ? env.FrameIndex : item.FinalIndex,
Bytes = item.Bytes + body.Length,
});
tx.AddShareBytes(id, body.Length);
return new AppendResult { Seq = env.FrameIndex, Created = true };
});
if (!result.Created) DeleteBlobIfUnreferenced(blobKey);
return result;
}
catch
{
DeleteBlobIfUnreferenced(blobKey);
throw;
}
}
public void DeleteItem(Id16 id, Credentials creds, Id16 itemId)
{
string prefix = $"{id.ToHex()}/{itemId.ToHex()}";
store.Write(tx =>
{
_ = Draft(tx.GetShare(id), creds);
var item = tx.GetShareItem(id, itemId) ?? throw new WebDropException(ErrorCode.NotFound);
tx.DeleteShareItemRows(id, itemId);
tx.AddShareBytes(id, -item.Bytes);
return 0;
});
blobs.DeletePrefix(prefix);
}
public void DeleteDraft(Id16 id, Credentials creds)
{
store.Write(tx =>
{
_ = Draft(tx.GetShare(id), creds);
tx.DeleteShareRows(id);
return 0;
});
blobs.DeletePrefix(id.ToHex());
}
public ShareInfo Finish(Id16 id, Credentials creds, string? manifestItemId)
{
if (!Id16.TryParse(manifestItemId, out var manifestId)) throw new WebDropException(ErrorCode.BadRequest);
return store.Write(tx =>
{
var share = Draft(tx.GetShare(id), creds);
var items = tx.ListShareItems(id);
var manifest = tx.GetShareItem(id, manifestId);
if (manifest is null || manifest.IsGated || manifest.IsSnapshot || !manifest.IsComplete
|| manifest.FrameCount != 1 || manifest.FinalIndex != 0)
throw new WebDropException(ErrorCode.BadRequest);
if (items.Any(i => !i.IsComplete || i.ItemId != manifestId && !i.IsGated))
throw new WebDropException(ErrorCode.BadRequest);
long now = clock.NowMs;
long expiresAt = now + share.Settings.TtlSeconds * 1000L;
var sealedShare = share with
{
OwnerHash = null,
SealedAt = now,
ExpiresAt = expiresAt,
ManifestItemId = manifestId,
};
tx.UpdateShare(sealedShare);
foreach (var item in items) tx.UpdateShareItem(item with { ExpiresAt = expiresAt });
return ToInfo(sealedShare, tx.ListShareItems(id));
});
}
public async Task<byte[]> ReadManifestAsync(Id16 id, Credentials creds)
{
var frame = store.Write(tx =>
{
var share = SealedForRead(tx.GetShare(id), creds);
RegisterVisitor(tx, share, creds.EndpointHash);
var manifestId = share.ManifestItemId ?? throw new WebDropException(ErrorCode.NotFound);
return tx.GetShareFrame(id, manifestId, 0) ?? throw new WebDropException(ErrorCode.Gone);
});
return await LoadFrameAsync(frame).ConfigureAwait(false);
}
public void StartItemDownload(Id16 id, Credentials creds, Id16 itemId, uint fromIndex)
{
store.Write(tx =>
{
var share = SealedForRead(tx.GetShare(id), creds);
RegisterVisitor(tx, share, creds.EndpointHash);
var item = tx.GetShareItem(id, itemId) ?? throw new WebDropException(ErrorCode.NotFound);
if (!item.IsGated) throw new WebDropException(ErrorCode.BadRequest);
long now = clock.NowMs;
if (item.CloseDeadline is { } d && (fromIndex == 0 || now > d)) throw new WebDropException(ErrorCode.Gone);
if (fromIndex == 0)
{
if (item.MaxDownloads > 0 && item.Downloads >= item.MaxDownloads) throw new WebDropException(ErrorCode.Gone);
long count = item.Downloads + 1;
long? closeAt = item.MaxDownloads > 0 && count >= item.MaxDownloads
? now + limits.DownloadGraceSeconds * 1000L
: item.CloseDeadline;
tx.UpdateShareItem(item with { Downloads = count, CloseDeadline = closeAt });
}
return 0;
});
}
public async Task<ItemChunk> ReadItemFramesAsync(Id16 id, Credentials creds, Id16 itemId, uint fromIndex, CancellationToken ct = default)
{
var frames = store.Read(tx =>
{
_ = SealedForRead(tx.GetShare(id), creds);
var item = tx.GetShareItem(id, itemId) ?? throw new WebDropException(ErrorCode.NotFound);
var list = tx.ListShareFrames(id, itemId, fromIndex, 32);
uint expected = fromIndex;
foreach (var frame in list)
{
if (frame.Index != expected++) throw new WebDropException(ErrorCode.Gone);
}
return (Frames: list, Next: expected, Complete: item.FinalIndex is { } f && expected > f);
});
var result = new List<byte[]>(frames.Frames.Count);
foreach (var frame in frames.Frames) result.Add(await LoadFrameAsync(frame).ConfigureAwait(false));
return new ItemChunk(result, frames.Next, frames.Complete);
}
public int Sweep()
{
int removed = 0;
long now = clock.NowMs;
var stale = store.Read(tx => tx.ListExpiredShareItems(now, 500));
foreach (var group in stale.GroupBy(i => i.ShareId))
{
foreach (var item in group)
{
bool didRemove = store.Write(tx =>
{
if (tx.GetShareItem(item.ShareId, item.ItemId) is not { } current) return false;
if (current.ExpiresAt > now && (current.CloseDeadline is null || current.CloseDeadline > now))
return false;
tx.DeleteShareItemRows(item.ShareId, item.ItemId);
tx.AddShareBytes(item.ShareId, -current.Bytes);
return true;
});
if (didRemove)
{
blobs.DeletePrefix($"{group.Key.ToHex()}/{item.ItemId.ToHex()}");
removed++;
}
}
}
foreach (var id in store.Read(tx => tx.ListExpiredShares(now, 200)))
{
store.Write(tx => { tx.DeleteShareRows(id); return 0; });
blobs.DeletePrefix(id.ToHex());
removed++;
}
return removed;
}
private Share SealedForRead(Share? candidate, Credentials creds)
{
var share = Live(candidate);
if (!share.IsSealed || creds.ClaimedRole != Role.Read || !TokenHash.Matches(creds.Token ?? [], share.ReadHash))
throw new WebDropException(share.IsSealed ? ErrorCode.Unauthorized : ErrorCode.NotFound);
return share;
}
private Share Draft(Share? candidate, Credentials creds)
{
var share = Live(candidate);
if (share.IsSealed) throw new WebDropException(ErrorCode.Sealed);
if (creds.ClaimedRole == Role.Read && TokenHash.Matches(creds.Token ?? [], share.ReadHash))
throw new WebDropException(ErrorCode.Forbidden);
RequireOwner(share, creds);
return share;
}
private void RequireOwner(Share share, Credentials creds)
{
if (creds.ClaimedRole != Role.Owner || !TokenHash.Matches(creds.Token ?? [], share.OwnerHash))
throw new WebDropException(ErrorCode.Unauthorized);
}
private Share Live(Share? candidate)
{
if (candidate is null || candidate.ExpiresAt <= clock.NowMs) throw new WebDropException(ErrorCode.NotFound);
return candidate;
}
private void RegisterVisitor(IResourceTx tx, Share share, byte[] endpointHash)
{
if (share.CreatorEndpointHash is { } creator && CryptographicOperations.FixedTimeEquals(creator, endpointHash)) return;
if (tx.HasShareVisitor(share.Id, endpointHash)) return;
if (share.Settings.MaxVisitors > 0 && tx.CountShareVisitors(share.Id) >= share.Settings.MaxVisitors)
throw new WebDropException(ErrorCode.VisitorsFull);
tx.InsertShareVisitor(share.Id, endpointHash, clock.NowMs);
}
private async ValueTask<byte[]> LoadFrameAsync(ShareFrame frame)
{
if (frame.Inline is { } bytes) return bytes;
if (frame.BlobKey is { } key && await blobs.ReadAsync(key).ConfigureAwait(false) is { } blob) return blob;
throw new WebDropException(ErrorCode.Gone);
}
private void DeleteBlobIfUnreferenced(string? key)
{
if (key is not null && !store.Read(tx => tx.BlobKeyExists(key))) blobs.Delete(key);
}
private static AppendResult Resolve(ShareFrame existing, byte[] digest) =>
existing.Digest.AsSpan().SequenceEqual(digest)
? new AppendResult { Seq = existing.AckSeq, Created = false }
: throw new WebDropException(ErrorCode.Conflict);
private static ShareInfo ToInfo(Share share, IReadOnlyList<ShareItem> items) => new()
{
Id = share.Id.ToBase64Url(),
State = share.IsSealed ? "sealed" : "draft",
ExpiresAt = share.ExpiresAt,
CreatedAt = share.CreatedAt,
SealedAt = share.SealedAt,
ManifestItemId = share.ManifestItemId?.ToBase64Url(),
Settings = share.Settings,
Items = items.Where(i => !share.IsSealed || i.ItemId != share.ManifestItemId).Select(i => new ShareItemInfo
{
ItemId = i.ItemId.ToBase64Url(),
Bytes = i.Bytes,
Frames = i.FrameCount,
MaxDownloads = i.MaxDownloads,
DownloadsLeft = i.MaxDownloads == 0 ? null : Math.Max(0, i.MaxDownloads - i.Downloads),
}).ToList(),
};
}
@@ -0,0 +1,9 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net10.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
</PropertyGroup>
</Project>
@@ -0,0 +1,90 @@
using WebDrop.Core;
namespace WebDrop.Storage;
/// <summary>
/// 磁盘 blob 存储。键形如 "channelHex/itemHex/index.digest",直接映射为目录与文件。
/// 键只能由服务生成(十六进制和数字),这里仍做路径逃逸检查作为纵深防御。
/// </summary>
public sealed class DiskBlobStore : IBlobStore
{
private readonly string _root;
public DiskBlobStore(string root)
{
_root = Path.GetFullPath(root);
Directory.CreateDirectory(_root);
}
public async ValueTask PutAsync(string key, ReadOnlyMemory<byte> data, CancellationToken ct = default)
{
string path = Resolve(key);
// 键里含内容摘要:文件已存在就意味着内容相同,无需重写(也避免并发覆盖在 Windows 上冲突)
if (File.Exists(path)) return;
Directory.CreateDirectory(Path.GetDirectoryName(path)!);
// 先写临时文件再重命名,读者永远看不到半个文件
string tmp = path + "." + Guid.NewGuid().ToString("N") + ".tmp";
await using (var fs = new FileStream(tmp, FileMode.CreateNew, FileAccess.Write, FileShare.None, 4096, FileOptions.Asynchronous))
{
await fs.WriteAsync(data, ct).ConfigureAwait(false);
}
try
{
File.Move(tmp, path, overwrite: false);
}
catch (Exception e) when (e is IOException or UnauthorizedAccessException && File.Exists(path))
{
File.Delete(tmp); // 并发写入同一内容,对方先到了
}
}
public async ValueTask<byte[]?> ReadAsync(string key, CancellationToken ct = default)
{
string path = Resolve(key);
try
{
return await File.ReadAllBytesAsync(path, ct).ConfigureAwait(false);
}
catch (FileNotFoundException)
{
return null;
}
catch (DirectoryNotFoundException)
{
return null;
}
}
public void Delete(string key)
{
try { File.Delete(Resolve(key)); }
catch (DirectoryNotFoundException) { }
}
public void DeletePrefix(string prefix)
{
string dir = Resolve(prefix);
try { if (Directory.Exists(dir)) Directory.Delete(dir, recursive: true); }
catch (DirectoryNotFoundException) { }
}
public IEnumerable<(string Key, DateTimeOffset LastWrite)> Enumerate()
{
foreach (string file in Directory.EnumerateFiles(_root, "*", SearchOption.AllDirectories))
{
if (file.EndsWith(".tmp", StringComparison.Ordinal)) continue;
string key = Path.GetRelativePath(_root, file).Replace('\\', '/');
yield return (key, new DateTimeOffset(File.GetLastWriteTimeUtc(file), TimeSpan.Zero));
}
}
private string Resolve(string key)
{
string full = Path.GetFullPath(Path.Combine(_root, key));
if (!full.StartsWith(_root + Path.DirectorySeparatorChar, StringComparison.Ordinal))
throw new ArgumentException("非法 blob 键", nameof(key));
return full;
}
}
@@ -0,0 +1,545 @@
using System.Text.Json;
using Microsoft.Data.Sqlite;
using WebDrop.Core;
namespace WebDrop.Storage;
/// <summary>
/// SQLite 实现:WAL 模式,写入由单个连接加锁串行执行(保证 seq 严格连续),读取使用独立连接。
/// 全部使用手写 SQL 与显式映射,不依赖反射,可 NativeAOT 发布。
/// </summary>
public sealed class SqliteChannelStore : IResourceStore, IDisposable
{
private const int SchemaVersion = 2;
private readonly string _connectionString;
private readonly SqliteConnection _writeConn;
private readonly object _writeLock = new();
public SqliteChannelStore(string dbPath)
{
string? dir = Path.GetDirectoryName(Path.GetFullPath(dbPath));
if (!string.IsNullOrEmpty(dir)) Directory.CreateDirectory(dir);
_connectionString = new SqliteConnectionStringBuilder
{
DataSource = dbPath,
Mode = SqliteOpenMode.ReadWriteCreate,
Cache = SqliteCacheMode.Private,
Pooling = true,
}.ToString();
_writeConn = Open();
Exec(_writeConn, "PRAGMA journal_mode=WAL;");
Migrate();
}
public T Read<T>(Func<IResourceTx, T> action)
{
using var conn = Open();
// 显式事务让一次 Read 内的多条查询看到同一份快照
using var tx = conn.BeginTransaction();
return action(new SqliteTx(conn));
}
public T Write<T>(Func<IResourceTx, T> action)
{
lock (_writeLock)
{
using var tx = _writeConn.BeginTransaction();
try
{
T result = action(new SqliteTx(_writeConn));
tx.Commit();
return result;
}
catch
{
tx.Rollback();
throw;
}
}
}
public void Dispose()
{
_writeConn.Dispose();
SqliteConnection.ClearAllPools();
}
private SqliteConnection Open()
{
var conn = new SqliteConnection(_connectionString);
conn.Open();
Exec(conn, "PRAGMA synchronous=NORMAL; PRAGMA foreign_keys=OFF;");
return conn;
}
private static void Exec(SqliteConnection conn, string sql)
{
using var cmd = conn.CreateCommand();
cmd.CommandText = sql;
cmd.ExecuteNonQuery();
}
private void Migrate()
{
using var cmd = _writeConn.CreateCommand();
cmd.CommandText = "PRAGMA user_version;";
long version = (long)cmd.ExecuteScalar()!;
if (version > SchemaVersion) throw new InvalidOperationException("数据库版本高于本程序支持的版本");
if (version == SchemaVersion) return;
using var tx = _writeConn.BeginTransaction();
if (version == 0) Exec(_writeConn, SchemaV1);
if (version < 2) Exec(_writeConn, SchemaV2);
Exec(_writeConn, $"PRAGMA user_version={SchemaVersion};");
tx.Commit();
}
private const string SchemaV1 = """
CREATE TABLE channels (
id BLOB NOT NULL PRIMARY KEY,
mode INTEGER NOT NULL,
settings TEXT NOT NULL,
read_hash BLOB,
write_hash BLOB,
owner_hash BLOB NOT NULL,
created_at INTEGER NOT NULL,
expires_at INTEGER NOT NULL,
last_seq INTEGER NOT NULL DEFAULT 0,
compacted_through INTEGER NOT NULL DEFAULT 0,
bytes INTEGER NOT NULL DEFAULT 0
);
CREATE INDEX ix_channels_expires ON channels(expires_at);
CREATE TABLE seats (
channel_id BLOB NOT NULL,
endpoint_hash BLOB NOT NULL,
label BLOB,
joined_at INTEGER NOT NULL,
PRIMARY KEY (channel_id, endpoint_hash)
) WITHOUT ROWID;
CREATE TABLE items (
channel_id BLOB NOT NULL,
item_id BLOB NOT NULL,
author_hash BLOB NOT NULL,
flags INTEGER NOT NULL,
max_downloads INTEGER NOT NULL,
downloads INTEGER NOT NULL,
frame_count INTEGER NOT NULL,
max_index INTEGER NOT NULL,
final_index INTEGER,
bytes INTEGER NOT NULL,
created_at INTEGER NOT NULL,
expires_at INTEGER NOT NULL,
close_deadline INTEGER,
label TEXT,
PRIMARY KEY (channel_id, item_id)
);
CREATE INDEX ix_items_channel_created ON items(channel_id, created_at);
CREATE INDEX ix_items_expires ON items(expires_at);
CREATE INDEX ix_items_close ON items(close_deadline) WHERE close_deadline IS NOT NULL;
CREATE TABLE frames (
channel_id BLOB NOT NULL,
item_id BLOB NOT NULL,
frame_index INTEGER NOT NULL,
seq INTEGER NOT NULL,
length INTEGER NOT NULL,
digest BLOB NOT NULL,
inline_blob BLOB,
blob_key TEXT,
PRIMARY KEY (channel_id, item_id, frame_index)
) WITHOUT ROWID;
CREATE INDEX ix_frames_blob ON frames(blob_key) WHERE blob_key IS NOT NULL;
CREATE TABLE log (
channel_id BLOB NOT NULL,
seq INTEGER NOT NULL,
kind INTEGER NOT NULL,
item_id BLOB NOT NULL,
frame_index INTEGER NOT NULL,
aux INTEGER NOT NULL,
created_at INTEGER NOT NULL,
PRIMARY KEY (channel_id, seq)
) WITHOUT ROWID;
CREATE INDEX ix_log_item ON log(channel_id, item_id);
""";
private const string SchemaV2 = """
CREATE TABLE shares (
id BLOB NOT NULL PRIMARY KEY,
settings TEXT NOT NULL,
read_hash BLOB NOT NULL,
owner_hash BLOB,
creator_endpoint_hash BLOB,
created_at INTEGER NOT NULL,
expires_at INTEGER NOT NULL,
sealed_at INTEGER,
manifest_item_id BLOB,
bytes INTEGER NOT NULL DEFAULT 0
);
CREATE INDEX ix_shares_expires ON shares(expires_at);
CREATE TABLE share_visitors (
share_id BLOB NOT NULL,
endpoint_hash BLOB NOT NULL,
joined_at INTEGER NOT NULL,
PRIMARY KEY (share_id, endpoint_hash)
) WITHOUT ROWID;
CREATE TABLE share_items (
share_id BLOB NOT NULL,
item_id BLOB NOT NULL,
flags INTEGER NOT NULL,
max_downloads INTEGER NOT NULL,
downloads INTEGER NOT NULL,
frame_count INTEGER NOT NULL,
max_index INTEGER NOT NULL,
final_index INTEGER,
bytes INTEGER NOT NULL,
created_at INTEGER NOT NULL,
expires_at INTEGER NOT NULL,
close_deadline INTEGER,
PRIMARY KEY (share_id, item_id)
) WITHOUT ROWID;
CREATE INDEX ix_share_items_expires ON share_items(expires_at);
CREATE INDEX ix_share_items_close ON share_items(close_deadline) WHERE close_deadline IS NOT NULL;
CREATE TABLE share_frames (
share_id BLOB NOT NULL,
item_id BLOB NOT NULL,
frame_index INTEGER NOT NULL,
seq INTEGER NOT NULL,
length INTEGER NOT NULL,
digest BLOB NOT NULL,
inline_blob BLOB,
blob_key TEXT,
PRIMARY KEY (share_id, item_id, frame_index)
) WITHOUT ROWID;
CREATE INDEX ix_share_frames_blob ON share_frames(blob_key) WHERE blob_key IS NOT NULL;
""";
private sealed class SqliteTx(SqliteConnection conn) : IResourceTx
{
private const string ChannelCols = "id, mode, settings, read_hash, write_hash, owner_hash, created_at, expires_at, last_seq, compacted_through, bytes";
private const string ItemCols = "channel_id, item_id, author_hash, flags, max_downloads, downloads, frame_count, max_index, final_index, bytes, created_at, expires_at, close_deadline, label";
private const string FrameCols = "channel_id, item_id, frame_index, seq, length, digest, inline_blob, blob_key";
private const string ShareItemCols = "share_id, item_id, flags, max_downloads, downloads, frame_count, max_index, final_index, bytes, created_at, expires_at, close_deadline";
private const string ShareFrameCols = "share_id, item_id, frame_index, seq, length, digest, inline_blob, blob_key";
// ───────── 通用辅助 ─────────
private SqliteCommand Cmd(string sql, (string, object?)[] args)
{
var cmd = conn.CreateCommand();
cmd.CommandText = sql;
foreach (var (name, value) in args) cmd.Parameters.AddWithValue(name, value ?? DBNull.Value);
return cmd;
}
private int Run(string sql, params (string, object?)[] args)
{
using var cmd = Cmd(sql, args);
return cmd.ExecuteNonQuery();
}
private object? Scalar(string sql, params (string, object?)[] args)
{
using var cmd = Cmd(sql, args);
return cmd.ExecuteScalar();
}
private T? One<T>(string sql, Func<SqliteDataReader, T> map, params (string, object?)[] args) where T : class
{
using var cmd = Cmd(sql, args);
using var r = cmd.ExecuteReader();
return r.Read() ? map(r) : null;
}
private List<T> Many<T>(string sql, Func<SqliteDataReader, T> map, params (string, object?)[] args)
{
using var cmd = Cmd(sql, args);
using var r = cmd.ExecuteReader();
var list = new List<T>();
while (r.Read()) list.Add(map(r));
return list;
}
private static byte[] Blob(SqliteDataReader r, int i) => (byte[])r.GetValue(i);
private static byte[]? BlobOrNull(SqliteDataReader r, int i) => r.IsDBNull(i) ? null : (byte[])r.GetValue(i);
private static long? LongOrNull(SqliteDataReader r, int i) => r.IsDBNull(i) ? null : r.GetInt64(i);
private static Id16 IdAt(SqliteDataReader r, int i) => Id16.FromBytes(Blob(r, i));
// ───────── 便捷分享 ─────────
private static Share MapShare(SqliteDataReader r) => new(
IdAt(r, 0),
JsonSerializer.Deserialize(r.GetString(1), CoreJsonContext.Default.ShareSettings)!,
Blob(r, 2), BlobOrNull(r, 3), BlobOrNull(r, 4), r.GetInt64(5), r.GetInt64(6),
LongOrNull(r, 7), r.IsDBNull(8) ? null : IdAt(r, 8), r.GetInt64(9));
public Share? GetShare(Id16 id) => One(
"SELECT id,settings,read_hash,owner_hash,creator_endpoint_hash,created_at,expires_at,sealed_at,manifest_item_id,bytes FROM shares WHERE id=@id",
MapShare, ("@id", id.ToArray()));
public bool HasResourceId(Id16 id) => Scalar(
"SELECT 1 FROM channels WHERE id=@id UNION ALL SELECT 1 FROM shares WHERE id=@id LIMIT 1",
("@id", id.ToArray())) is not null;
public void InsertShare(Share s) => Run(
"INSERT INTO shares (id,settings,read_hash,owner_hash,creator_endpoint_hash,created_at,expires_at,sealed_at,manifest_item_id,bytes) VALUES (@id,@settings,@read,@owner,@creator,@created,@expires,@sealed,@manifest,@bytes)",
("@id", s.Id.ToArray()), ("@settings", JsonSerializer.Serialize(s.Settings, CoreJsonContext.Default.ShareSettings)),
("@read", s.ReadHash), ("@owner", s.OwnerHash), ("@creator", s.CreatorEndpointHash),
("@created", s.CreatedAt), ("@expires", s.ExpiresAt), ("@sealed", s.SealedAt),
("@manifest", s.ManifestItemId?.ToArray()), ("@bytes", s.Bytes));
public void UpdateShare(Share s) => Run(
"UPDATE shares SET settings=@settings,owner_hash=@owner,expires_at=@expires,sealed_at=@sealed,manifest_item_id=@manifest,bytes=@bytes WHERE id=@id",
("@settings", JsonSerializer.Serialize(s.Settings, CoreJsonContext.Default.ShareSettings)), ("@owner", s.OwnerHash),
("@expires", s.ExpiresAt), ("@sealed", s.SealedAt), ("@manifest", s.ManifestItemId?.ToArray()),
("@bytes", s.Bytes), ("@id", s.Id.ToArray()));
public IReadOnlyList<Id16> ListExpiredShares(long now, int limit) => Many(
"SELECT id FROM shares WHERE expires_at<=@now LIMIT @n", r => IdAt(r, 0), ("@now", now), ("@n", limit));
public int CountShareVisitors(Id16 id) =>
(int)(long)Scalar("SELECT COUNT(*) FROM share_visitors WHERE share_id=@id", ("@id", id.ToArray()))!;
public bool HasShareVisitor(Id16 id, ReadOnlySpan<byte> endpointHash) => Scalar(
"SELECT 1 FROM share_visitors WHERE share_id=@id AND endpoint_hash=@ep",
("@id", id.ToArray()), ("@ep", endpointHash.ToArray())) is not null;
public void InsertShareVisitor(Id16 id, ReadOnlySpan<byte> endpointHash, long joinedAt) => Run(
"INSERT INTO share_visitors (share_id,endpoint_hash,joined_at) VALUES (@id,@ep,@at)",
("@id", id.ToArray()), ("@ep", endpointHash.ToArray()), ("@at", joinedAt));
public void DeleteShareRows(Id16 id)
{
byte[] key = id.ToArray();
Run("DELETE FROM share_frames WHERE share_id=@id", ("@id", key));
Run("DELETE FROM share_items WHERE share_id=@id", ("@id", key));
Run("DELETE FROM share_visitors WHERE share_id=@id", ("@id", key));
Run("DELETE FROM shares WHERE id=@id", ("@id", key));
}
private static ShareItem MapShareItem(SqliteDataReader r) => new(
IdAt(r, 0), IdAt(r, 1), (FrameFlags)r.GetInt32(2), r.GetInt64(3), r.GetInt64(4), r.GetInt32(5),
r.GetInt64(6), LongOrNull(r, 7), r.GetInt64(8), r.GetInt64(9), r.GetInt64(10), LongOrNull(r, 11));
public IReadOnlyList<ShareItem> ListShareItems(Id16 id) => Many(
$"SELECT {ShareItemCols} FROM share_items WHERE share_id=@id ORDER BY created_at,item_id", MapShareItem, ("@id", id.ToArray()));
public int CountShareItems(Id16 id) => (int)(long)Scalar("SELECT COUNT(*) FROM share_items WHERE share_id=@id", ("@id", id.ToArray()))!;
public void AddShareBytes(Id16 id, long delta) => Run("UPDATE shares SET bytes=bytes+@d WHERE id=@id", ("@d", delta), ("@id", id.ToArray()));
public IReadOnlyList<ShareItem> ListExpiredShareItems(long now, int limit) => Many(
$"SELECT {ShareItemCols} FROM share_items WHERE expires_at<=@now OR (close_deadline IS NOT NULL AND close_deadline<=@now) LIMIT @n",
MapShareItem, ("@now", now), ("@n", limit));
public ShareItem? GetShareItem(Id16 shareId, Id16 itemId) => One(
$"SELECT {ShareItemCols} FROM share_items WHERE share_id=@s AND item_id=@i", MapShareItem,
("@s", shareId.ToArray()), ("@i", itemId.ToArray()));
public void InsertShareItem(ShareItem i) => Run(
"INSERT INTO share_items (share_id,item_id,flags,max_downloads,downloads,frame_count,max_index,final_index,bytes,created_at,expires_at,close_deadline) VALUES (@s,@i,@f,@md,@d,@fc,@mi,@fi,@b,@cr,@ex,@cd)",
("@s", i.ShareId.ToArray()), ("@i", i.ItemId.ToArray()), ("@f", (int)i.Flags),
("@md", i.MaxDownloads), ("@d", i.Downloads), ("@fc", i.FrameCount), ("@mi", i.MaxIndex), ("@fi", i.FinalIndex),
("@b", i.Bytes), ("@cr", i.CreatedAt), ("@ex", i.ExpiresAt), ("@cd", i.CloseDeadline));
public void UpdateShareItem(ShareItem i) => Run(
"UPDATE share_items SET downloads=@d,frame_count=@fc,max_index=@mi,final_index=@fi,bytes=@b,expires_at=@ex,close_deadline=@cd WHERE share_id=@s AND item_id=@i",
("@d", i.Downloads), ("@fc", i.FrameCount), ("@mi", i.MaxIndex), ("@fi", i.FinalIndex), ("@b", i.Bytes),
("@ex", i.ExpiresAt), ("@cd", i.CloseDeadline), ("@s", i.ShareId.ToArray()), ("@i", i.ItemId.ToArray()));
public void DeleteShareItemRows(Id16 shareId, Id16 itemId)
{
Run("DELETE FROM share_frames WHERE share_id=@s AND item_id=@i", ("@s", shareId.ToArray()), ("@i", itemId.ToArray()));
Run("DELETE FROM share_items WHERE share_id=@s AND item_id=@i", ("@s", shareId.ToArray()), ("@i", itemId.ToArray()));
}
private static ShareFrame MapShareFrame(SqliteDataReader r) => new(
IdAt(r, 0), IdAt(r, 1), (uint)r.GetInt64(2), r.GetInt64(3), r.GetInt32(4), Blob(r, 5), BlobOrNull(r, 6),
r.IsDBNull(7) ? null : r.GetString(7));
public ShareFrame? GetShareFrame(Id16 s, Id16 i, uint n) => One(
$"SELECT {ShareFrameCols} FROM share_frames WHERE share_id=@s AND item_id=@i AND frame_index=@n", MapShareFrame,
("@s", s.ToArray()), ("@i", i.ToArray()), ("@n", (long)n));
public void InsertShareFrame(ShareFrame f) => Run(
$"INSERT INTO share_frames ({ShareFrameCols}) VALUES (@s,@i,@n,@seq,@len,@digest,@inline,@key)",
("@s", f.ShareId.ToArray()), ("@i", f.ItemId.ToArray()), ("@n", (long)f.Index), ("@seq", f.AckSeq),
("@len", f.Length), ("@digest", f.Digest), ("@inline", f.Inline), ("@key", f.BlobKey));
public IReadOnlyList<ShareFrame> ListShareFrames(Id16 s, Id16 i, uint from, int limit) => Many(
$"SELECT {ShareFrameCols} FROM share_frames WHERE share_id=@s AND item_id=@i AND frame_index>=@from ORDER BY frame_index LIMIT @n",
MapShareFrame, ("@s", s.ToArray()), ("@i", i.ToArray()), ("@from", (long)from), ("@n", limit));
// ───────── 频道 ─────────
private static Channel MapChannel(SqliteDataReader r) => new(
IdAt(r, 0),
(ChannelMode)r.GetInt32(1),
JsonSerializer.Deserialize(r.GetString(2), CoreJsonContext.Default.ChannelSettings)!,
BlobOrNull(r, 3),
BlobOrNull(r, 4),
Blob(r, 5),
r.GetInt64(6), r.GetInt64(7), r.GetInt64(8), r.GetInt64(9), r.GetInt64(10));
public Channel? GetChannel(Id16 id) =>
One($"SELECT {ChannelCols} FROM channels WHERE id=@id", MapChannel, ("@id", id.ToArray()));
public void InsertChannel(Channel c) => Run(
$"INSERT INTO channels ({ChannelCols}) VALUES (@id,@mode,@settings,@rh,@wh,@oh,@created,@expires,@seq,@comp,@bytes)",
("@id", c.Id.ToArray()), ("@mode", (int)c.Mode),
("@settings", JsonSerializer.Serialize(c.Settings, CoreJsonContext.Default.ChannelSettings)),
("@rh", c.ReadHash), ("@wh", c.WriteHash), ("@oh", c.OwnerHash),
("@created", c.CreatedAt), ("@expires", c.ExpiresAt),
("@seq", c.LastSeq), ("@comp", c.CompactedThrough), ("@bytes", c.Bytes));
public void UpdateChannel(Id16 id, ChannelSettings settings, long expiresAt) => Run(
"UPDATE channels SET settings=@s, expires_at=@e WHERE id=@id",
("@s", JsonSerializer.Serialize(settings, CoreJsonContext.Default.ChannelSettings)),
("@e", expiresAt), ("@id", id.ToArray()));
public void TouchChannel(Id16 id, long expiresAt) =>
Run("UPDATE channels SET expires_at=@e WHERE id=@id", ("@e", expiresAt), ("@id", id.ToArray()));
public void AddChannelBytes(Id16 id, long delta) =>
Run("UPDATE channels SET bytes=bytes+@d WHERE id=@id", ("@d", delta), ("@id", id.ToArray()));
public void SetCompactedThrough(Id16 id, long seq) =>
Run("UPDATE channels SET compacted_through=@s WHERE id=@id", ("@s", seq), ("@id", id.ToArray()));
public long NextSeq(Id16 id) =>
(long)Scalar("UPDATE channels SET last_seq=last_seq+1 WHERE id=@id RETURNING last_seq", ("@id", id.ToArray()))!;
public void DeleteChannelRows(Id16 id)
{
byte[] key = id.ToArray();
Run("DELETE FROM frames WHERE channel_id=@c", ("@c", key));
Run("DELETE FROM log WHERE channel_id=@c", ("@c", key));
Run("DELETE FROM items WHERE channel_id=@c", ("@c", key));
Run("DELETE FROM seats WHERE channel_id=@c", ("@c", key));
Run("DELETE FROM channels WHERE id=@c", ("@c", key));
}
public IReadOnlyList<Id16> ListExpiredChannels(long now, int limit) =>
Many("SELECT id FROM channels WHERE expires_at<=@now LIMIT @n", r => IdAt(r, 0), ("@now", now), ("@n", limit));
// ───────── 席位 ─────────
private static Seat MapSeat(SqliteDataReader r) => new(Blob(r, 0), r.GetInt64(1), BlobOrNull(r, 2));
public Seat? GetSeat(Id16 channelId, ReadOnlySpan<byte> endpointHash) => One(
"SELECT endpoint_hash, joined_at, label FROM seats WHERE channel_id=@c AND endpoint_hash=@e",
MapSeat, ("@c", channelId.ToArray()), ("@e", endpointHash.ToArray()));
public int CountSeats(Id16 channelId) =>
(int)(long)Scalar("SELECT COUNT(*) FROM seats WHERE channel_id=@c", ("@c", channelId.ToArray()))!;
public void InsertSeat(Id16 channelId, Seat seat) => Run(
"INSERT INTO seats (channel_id, endpoint_hash, label, joined_at) VALUES (@c,@e,@l,@j)",
("@c", channelId.ToArray()), ("@e", seat.EndpointHash), ("@l", seat.Label), ("@j", seat.JoinedAt));
public bool DeleteSeat(Id16 channelId, ReadOnlySpan<byte> endpointHash) => Run(
"DELETE FROM seats WHERE channel_id=@c AND endpoint_hash=@e",
("@c", channelId.ToArray()), ("@e", endpointHash.ToArray())) > 0;
public void SetSeatLabel(Id16 channelId, ReadOnlySpan<byte> endpointHash, byte[] label) => Run(
"UPDATE seats SET label=@l WHERE channel_id=@c AND endpoint_hash=@e",
("@l", label), ("@c", channelId.ToArray()), ("@e", endpointHash.ToArray()));
public IReadOnlyList<Seat> ListSeats(Id16 channelId) => Many(
"SELECT endpoint_hash, joined_at, label FROM seats WHERE channel_id=@c ORDER BY joined_at, endpoint_hash",
MapSeat, ("@c", channelId.ToArray()));
// ───────── 条目 ─────────
private static Item MapItem(SqliteDataReader r) => new(
IdAt(r, 0), IdAt(r, 1), Blob(r, 2), (FrameFlags)r.GetInt32(3),
r.GetInt64(4), r.GetInt64(5), r.GetInt32(6), r.GetInt64(7), LongOrNull(r, 8),
r.GetInt64(9), r.GetInt64(10), r.GetInt64(11), LongOrNull(r, 12),
r.IsDBNull(13) ? null : r.GetString(13));
public Item? GetItem(Id16 channelId, Id16 itemId) => One(
$"SELECT {ItemCols} FROM items WHERE channel_id=@c AND item_id=@i",
MapItem, ("@c", channelId.ToArray()), ("@i", itemId.ToArray()));
public void InsertItem(Item i) => Run(
$"INSERT INTO items ({ItemCols}) VALUES (@c,@i,@a,@f,@md,@d,@fc,@mi,@fi,@b,@cr,@ex,@cd,@l)",
("@c", i.ChannelId.ToArray()), ("@i", i.ItemId.ToArray()), ("@a", i.AuthorHash),
("@f", (int)i.Flags), ("@md", i.MaxDownloads), ("@d", i.Downloads),
("@fc", i.FrameCount), ("@mi", i.MaxIndex), ("@fi", i.FinalIndex),
("@b", i.Bytes), ("@cr", i.CreatedAt), ("@ex", i.ExpiresAt), ("@cd", i.CloseDeadline), ("@l", i.Label));
public void UpdateItem(Item i) => Run(
"UPDATE items SET downloads=@d, frame_count=@fc, max_index=@mi, final_index=@fi, bytes=@b, expires_at=@ex, close_deadline=@cd WHERE channel_id=@c AND item_id=@i",
("@d", i.Downloads), ("@fc", i.FrameCount), ("@mi", i.MaxIndex), ("@fi", i.FinalIndex),
("@b", i.Bytes), ("@ex", i.ExpiresAt), ("@cd", i.CloseDeadline),
("@c", i.ChannelId.ToArray()), ("@i", i.ItemId.ToArray()));
public void DeleteItemRows(Id16 channelId, Id16 itemId)
{
byte[] c = channelId.ToArray(), i = itemId.ToArray();
Run("DELETE FROM frames WHERE channel_id=@c AND item_id=@i", ("@c", c), ("@i", i));
// 墓碑保留,其余记录(帧、announce)随条目一起删除
Run("DELETE FROM log WHERE channel_id=@c AND item_id=@i AND kind<>@t", ("@c", c), ("@i", i), ("@t", (int)LogKind.Tombstone));
Run("DELETE FROM items WHERE channel_id=@c AND item_id=@i", ("@c", c), ("@i", i));
}
public int CountItems(Id16 channelId) =>
(int)(long)Scalar("SELECT COUNT(*) FROM items WHERE channel_id=@c", ("@c", channelId.ToArray()))!;
public IReadOnlyList<Item> ListOldestItems(Id16 channelId, int limit) => Many(
$"SELECT {ItemCols} FROM items WHERE channel_id=@c ORDER BY created_at, rowid LIMIT @n",
MapItem, ("@c", channelId.ToArray()), ("@n", limit));
public Item? FindSnapshotItem(Id16 channelId, Id16 excludeItemId) => One(
$"SELECT {ItemCols} FROM items WHERE channel_id=@c AND item_id<>@x AND (flags & @s)<>0 ORDER BY created_at DESC, rowid DESC LIMIT 1",
MapItem, ("@c", channelId.ToArray()), ("@x", excludeItemId.ToArray()), ("@s", (int)FrameFlags.Snapshot));
public IReadOnlyList<Item> ListExpiredItems(long now, int limit) => Many(
$"SELECT {ItemCols} FROM items WHERE expires_at<=@now OR (close_deadline IS NOT NULL AND close_deadline<=@now) LIMIT @n",
MapItem, ("@now", now), ("@n", limit));
// ───────── 帧 ─────────
private static StoredFrame MapFrame(SqliteDataReader r) => new(
IdAt(r, 0), IdAt(r, 1), (uint)r.GetInt64(2), r.GetInt64(3), r.GetInt32(4),
Blob(r, 5), BlobOrNull(r, 6), r.IsDBNull(7) ? null : r.GetString(7));
public StoredFrame? GetFrame(Id16 channelId, Id16 itemId, uint index) => One(
$"SELECT {FrameCols} FROM frames WHERE channel_id=@c AND item_id=@i AND frame_index=@x",
MapFrame, ("@c", channelId.ToArray()), ("@i", itemId.ToArray()), ("@x", (long)index));
public void InsertFrame(StoredFrame f) => Run(
$"INSERT INTO frames ({FrameCols}) VALUES (@c,@i,@x,@s,@len,@dg,@inl,@bk)",
("@c", f.ChannelId.ToArray()), ("@i", f.ItemId.ToArray()), ("@x", (long)f.Index),
("@s", f.Seq), ("@len", f.Length), ("@dg", f.Digest), ("@inl", f.Inline), ("@bk", f.BlobKey));
public IReadOnlyList<StoredFrame> ListFrames(Id16 channelId, Id16 itemId, uint fromIndex, int limit) => Many(
$"SELECT {FrameCols} FROM frames WHERE channel_id=@c AND item_id=@i AND frame_index>=@x ORDER BY frame_index LIMIT @n",
MapFrame, ("@c", channelId.ToArray()), ("@i", itemId.ToArray()), ("@x", (long)fromIndex), ("@n", limit));
public bool BlobKeyExists(string blobKey) =>
Scalar("SELECT 1 FROM frames WHERE blob_key=@k UNION ALL SELECT 1 FROM share_frames WHERE blob_key=@k LIMIT 1", ("@k", blobKey)) is not null;
// ───────── 日志 ─────────
public void InsertLog(Id16 channelId, LogEntry e) => Run(
"INSERT INTO log (channel_id, seq, kind, item_id, frame_index, aux, created_at) VALUES (@c,@s,@k,@i,@x,@a,@t)",
("@c", channelId.ToArray()), ("@s", e.Seq), ("@k", (int)e.Kind), ("@i", e.ItemId.ToArray()),
("@x", (long)e.FrameIndex), ("@a", (int)e.Aux), ("@t", e.CreatedAt));
public IReadOnlyList<LogEntry> ReadLog(Id16 channelId, long fromSeq, int limit) => Many(
"SELECT seq, kind, item_id, frame_index, aux, created_at FROM log WHERE channel_id=@c AND seq>=@s ORDER BY seq LIMIT @n",
r => new LogEntry(r.GetInt64(0), (LogKind)r.GetInt32(1), IdAt(r, 2), (uint)r.GetInt64(3), (byte)r.GetInt32(4), r.GetInt64(5)),
("@c", channelId.ToArray()), ("@s", fromSeq), ("@n", limit));
public long PruneTombstones(Id16 channelId, long olderThanMs)
{
byte[] c = channelId.ToArray();
var max = Scalar("SELECT MAX(seq) FROM log WHERE channel_id=@c AND kind=@k AND created_at<@t",
("@c", c), ("@k", (int)LogKind.Tombstone), ("@t", olderThanMs));
if (max is not long seq) return 0;
Run("DELETE FROM log WHERE channel_id=@c AND kind=@k AND seq<=@s AND created_at<@t",
("@c", c), ("@k", (int)LogKind.Tombstone), ("@s", seq), ("@t", olderThanMs));
return seq;
}
}
}
@@ -0,0 +1,17 @@
<Project Sdk="Microsoft.NET.Sdk">
<ItemGroup>
<ProjectReference Include="..\WebDrop.Core\WebDrop.Core.csproj" />
</ItemGroup>
<ItemGroup>
<PackageReference Include="Microsoft.Data.Sqlite" Version="10.0.12" />
</ItemGroup>
<PropertyGroup>
<TargetFramework>net10.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
</PropertyGroup>
</Project>