using System.Security.Cryptography;
namespace WebDrop.Core;
///
/// 频道的全部业务逻辑。依赖抽象(存储、blob、时钟、事件),不含 HTTP、SQL、文件细节。
/// 约定:所有失败以 表达;提交成功后才发事件;blob 先写后提交。
///
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 ListSeats(Id16 id, Credentials creds) => store.Read(tx =>
{
var (_, principal) = Authenticate(tx, id, creds, SeatRule.None);
RequireOwner(principal);
return (IReadOnlyList)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 AppendFrameAsync(
Id16 channelId, Credentials creds, ReadOnlyMemory 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 prefixes)
{
public AppendResult Result { get; } = result;
public long LastSeq { get; } = lastSeq;
public List BlobPrefixesToDelete { get; } = prefixes;
}
private AppendOutcome AppendInTx(
IResourceTx tx, Id16 channelId, Credentials creds, Envelope env, ReadOnlyMemory 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();
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);
}
private static bool IsItemReadable(Item item, long now) =>
item.ExpiresAt > now && (item.CloseDeadline is null || item.CloseDeadline > now);
// ───────────────────────── 读取 ─────────────────────────
public async Task 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);
long now = clock.NowMs;
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.GetItem(id, e.ItemId) is { } frameItem && IsItemReadable(frameItem, now)
&& 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 && IsItemReadable(it, now))
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(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);
}
/// 开始拉取条目:鉴权,并按规则计一次下载。 为 0 才计数,大于 0 视为续传。
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.ExpiresAt <= now) throw new WebDropException(ErrorCode.Gone);
if (item.CloseDeadline is { } deadline && (fromIndex == 0 || now >= deadline))
throw new WebDropException(ErrorCode.Gone);
if (item.IsGated && fromIndex > 0 && !tx.HasDownloadGrant(id, itemId, principal.EndpointHash))
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 });
tx.AddDownloadGrant(id, itemId, principal.EndpointHash);
}
return 0;
});
}
/// 读取自 起连续的帧。条目未完成时可能返回空批次,调用方等待后重试。
public async Task 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.ExpiresAt <= clock.NowMs) throw new WebDropException(ErrorCode.Gone);
if (item.IsGated && !tx.HasDownloadGrant(id, itemId, principal.EndpointHash))
throw new WebDropException(ErrorCode.Gone);
if (item.CloseDeadline is { } d && clock.NowMs >= d) throw new WebDropException(ErrorCode.Gone);
var taken = new List();
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(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 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();
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 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;
}
// ───────────────────────── 后台清理 ─────────────────────────
/// 清理过期条目、下载达上限的条目、过期频道;可选清理孤立 blob。返回删除的条目和频道总数。
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();
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;
}
/// 删除没有任何帧引用、且超过 10 分钟的 blob(崩溃发生在写 blob 与提交事务之间会留下它们)。
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,
};
}