733 lines
37 KiB
C#
733 lines
37 KiB
C#
using System.Security.Cryptography;
|
|
|
|
namespace WebDrop.Core;
|
|
|
|
/// <summary>
|
|
/// 频道的全部业务逻辑。依赖抽象(存储、blob、时钟、事件),不含 HTTP、SQL、文件细节。
|
|
/// 约定:所有失败以 <see cref="WebDropException"/> 表达;提交成功后才发事件;blob 先写后提交。
|
|
/// </summary>
|
|
public sealed partial class ChannelService(
|
|
IResourceStore store,
|
|
IBlobStore blobs,
|
|
IClock clock,
|
|
ServerLimits limits,
|
|
IChannelEvents events)
|
|
{
|
|
private enum SeatRule { None, Required }
|
|
|
|
// ───────────────────────── 频道 ─────────────────────────
|
|
|
|
public ChannelInfo CreateChannel(CreateChannelRequest req, byte[] endpointHash, string creatorAccount, string? accountDeviceId = null)
|
|
{
|
|
if (string.IsNullOrWhiteSpace(creatorAccount)) throw new WebDropException(ErrorCode.AccountRequired);
|
|
if (req.KeyPackage is not null && req.KeyPackage.Length is < 1 or > 16_384) throw new WebDropException(ErrorCode.BadRequest);
|
|
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 ?? CreationDefaults.Channel(limits);
|
|
if (string.IsNullOrWhiteSpace(settings.Name)) throw new WebDropException(ErrorCode.SpaceNameRequired);
|
|
settings.Name = settings.Name.Trim();
|
|
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 (passwordHash, passwordSalt) = HashJoinPassword(req.JoinPassword);
|
|
var channel = new Channel(id, mode, settings, readHash, writeHash, ownerHash,
|
|
now, now + settings.TtlSeconds * 1000L, 0, 0, 0, creatorAccount, passwordHash, passwordSalt);
|
|
|
|
store.Write(tx =>
|
|
{
|
|
if (tx.HasResourceId(id)) throw new WebDropException(ErrorCode.Conflict);
|
|
tx.InsertChannel(channel);
|
|
tx.PutWorkspace(id, NewWorkspace(creatorAccount, now));
|
|
tx.PutChannelMember(id, new ChannelMember(creatorAccount, "owner", now));
|
|
if (req.KeyPackage is not null) tx.PutKeyPackage(id, creatorAccount, req.KeyPackage, now);
|
|
tx.InsertSeat(id, new Seat(endpointHash, now, null, creatorAccount, accountDeviceId));
|
|
return 0;
|
|
});
|
|
return ToInfo(channel, 1);
|
|
}
|
|
|
|
public ChannelInfo GetChannel(Id16 id, Credentials creds) => store.Read(tx =>
|
|
{
|
|
var (channel, principal) = Authenticate(tx, id, creds, SeatRule.None);
|
|
return ToInfo(channel, tx.CountMembers(id), principal);
|
|
});
|
|
|
|
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);
|
|
if (patch.JoinPassword is not null)
|
|
{
|
|
var (hash, salt) = HashJoinPassword(patch.JoinPassword);
|
|
tx.SetJoinPassword(id, hash, salt);
|
|
channel = channel with { JoinPasswordHash = hash, JoinPasswordSalt = salt };
|
|
}
|
|
return ToInfo(channel with { Settings = updated, ExpiresAt = expiresAt }, tx.CountMembers(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, string? password = null)
|
|
{
|
|
var (joined, isNew) = store.Write(tx =>
|
|
{
|
|
var (channel, principal) = Authenticate(tx, id, creds, SeatRule.None);
|
|
var existing = tx.GetSeat(id, principal.EndpointHash);
|
|
if (existing?.AccountName is not null && existing.AccountName != creds.AccountName) throw new WebDropException(ErrorCode.AccountRequired);
|
|
if (existing?.AccountDeviceId is not null && existing.AccountDeviceId != creds.AccountDeviceId) throw new WebDropException(ErrorCode.AccountRequired);
|
|
int membersUsed = tx.CountMembers(id);
|
|
bool accountAlreadyMember = creds.AccountName is not null && tx.GetChannelMember(id, creds.AccountName) is not null;
|
|
if (existing is not null)
|
|
{
|
|
if (existing.AccountName is not null && existing.AccountDeviceId is null && creds.AccountDeviceId is not null)
|
|
tx.BindSeatDevice(id, principal.EndpointHash, creds.AccountDeviceId);
|
|
return (new SeatJoined { EndpointHash = Codec.Encode(existing.EndpointHash), JoinedAt = existing.JoinedAt, SeatsUsed = membersUsed, SeatsMax = channel.Settings.Seats }, false);
|
|
}
|
|
|
|
CheckAdmission(tx, channel, creds, password);
|
|
|
|
bool newMember = creds.AccountName is null || !accountAlreadyMember;
|
|
if (newMember && membersUsed >= channel.Settings.Seats) throw new WebDropException(ErrorCode.SeatsFull);
|
|
long now = clock.NowMs;
|
|
tx.InsertSeat(id, new Seat(principal.EndpointHash, now, null, creds.AccountName, creds.AccountDeviceId));
|
|
if (creds.AccountName is not null) tx.PutChannelMember(id, new ChannelMember(creds.AccountName, "member", now));
|
|
return (new SeatJoined { EndpointHash = Codec.Encode(principal.EndpointHash), JoinedAt = now, SeatsUsed = membersUsed + (newMember ? 1 : 0), 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),
|
|
AccountName = s.AccountName,
|
|
}).ToList();
|
|
});
|
|
|
|
public void SetSeatLabel(Id16 id, Credentials creds, byte[] label)
|
|
{
|
|
ValidateSeatLabel(label);
|
|
store.Write(tx =>
|
|
{
|
|
var (_, principal) = Authenticate(tx, id, creds, SeatRule.Required);
|
|
tx.SetSeatLabel(id, principal.EndpointHash, label);
|
|
return 0;
|
|
});
|
|
events.SeatsChanged(id);
|
|
}
|
|
|
|
public void SetSeatLabelForEndpoint(Id16 id, Credentials creds, byte[] endpointHash, byte[] label)
|
|
{
|
|
ValidateSeatLabel(label);
|
|
store.Write(tx =>
|
|
{
|
|
var (_, principal) = Authenticate(tx, id, creds, SeatRule.None);
|
|
RequireOwner(principal);
|
|
if (tx.GetSeat(id, endpointHash) is null) throw new WebDropException(ErrorCode.NotFound);
|
|
tx.SetSeatLabel(id, endpointHash, label);
|
|
return 0;
|
|
});
|
|
events.SeatsChanged(id);
|
|
}
|
|
|
|
private void ValidateSeatLabel(byte[] label)
|
|
{
|
|
if (label.Length == 0) throw new WebDropException(ErrorCode.BadRequest);
|
|
if (label.Length > limits.MaxSeatLabelBytes) throw new WebDropException(ErrorCode.TooLarge);
|
|
}
|
|
|
|
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);
|
|
tx.DenyEndpoint(id, endpointHash);
|
|
return 0;
|
|
});
|
|
events.SeatsChanged(id);
|
|
}
|
|
|
|
// ───────────────────────── 追加帧 ─────────────────────────
|
|
|
|
public async Task<AppendResult> AppendFrameAsync(
|
|
Id16 channelId, Credentials creds, ReadOnlyMemory<byte> body,
|
|
long? maxDownloads, CancellationToken ct = default, SignedContentEventProof? signedEvent = null)
|
|
{
|
|
if (!Envelope.TryParse(body.Span, out var env)) throw new WebDropException(ErrorCode.BadRequest);
|
|
if (env.PayloadLength > limits.MaxPayloadBytes) throw new WebDropException(ErrorCode.TooLarge);
|
|
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);
|
|
CheckObjectWrite(tx, 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, signedEvent));
|
|
}
|
|
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, SignedContentEventProof? signedEvent)
|
|
{
|
|
var (channel, principal) = Authenticate(tx, channelId, creds, SeatRule.Required);
|
|
CheckWritable(channel, principal, env);
|
|
CheckObjectWrite(tx, 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);
|
|
var crypto = tx.GetWorkspaceCrypto(channelId);
|
|
var objectEnvelope = crypto.Objects.Find(o => o.ItemId == env.ItemId.ToBase64Url());
|
|
if (signedEvent is not null && (objectEnvelope is null || !env.IsFinal)) throw new WebDropException(ErrorCode.BadRequest);
|
|
if (objectEnvelope is not null && env.IsFinal)
|
|
{
|
|
if (signedEvent is null)
|
|
{
|
|
if (creds.RequireSignedContent) throw new WebDropException(ErrorCode.Forbidden);
|
|
}
|
|
else
|
|
{
|
|
if (objectEnvelope.AuthorAccount != principal.AccountName || signedEvent.Username != principal.AccountName || creds.AccountDeviceId != signedEvent.DeviceId
|
|
|| objectEnvelope.ChannelId != signedEvent.ChannelId || objectEnvelope.Domain != signedEvent.Domain
|
|
|| objectEnvelope.Epoch != signedEvent.Epoch || signedEvent.FrameCount != env.FrameIndex + 1
|
|
|| signedEvent.EventIdBytes.Length != 16 || signedEvent.ManifestHash.Length != 32 || signedEvent.Signature.Length != 64
|
|
|| objectEnvelope.Event is not null
|
|
|| signedEvent.Domain == CryptoDomain.Chat && signedEvent.EventKind is not ("chat.create" or "chat.edit" or "chat.snapshot")
|
|
|| signedEvent.Domain == CryptoDomain.Files && signedEvent.EventKind is not ("file.publish" or "file.catalog"))
|
|
throw new WebDropException(ErrorCode.Forbidden);
|
|
if (signedEvent.EventKind == "chat.edit")
|
|
{
|
|
if (signedEvent.TargetItemId is null || !Codec.TryDecodeExact(signedEvent.TargetItemId, 16, out var targetId)) throw new WebDropException(ErrorCode.BadRequest);
|
|
var targetItem = tx.GetItem(channelId, Id16.FromBytes(targetId));
|
|
var targetObject = crypto.Objects.Find(o => o.ItemId == signedEvent.TargetItemId);
|
|
if (targetItem is null || targetObject?.Domain != CryptoDomain.Chat || !ChannelPolicy.CanDeleteItem(channel, targetItem, principal))
|
|
throw new WebDropException(ErrorCode.Forbidden);
|
|
}
|
|
else if (signedEvent.TargetItemId is not null) throw new WebDropException(ErrorCode.BadRequest);
|
|
var manifestFrames = tx.ListFrames(channelId, env.ItemId, 0, limits.MaxFramesPerItem);
|
|
if (manifestFrames.Count != env.FrameIndex) throw new WebDropException(ErrorCode.Conflict);
|
|
var digests = new List<byte[]>(manifestFrames.Count + 1);
|
|
for (int i = 0; i < manifestFrames.Count; i++)
|
|
{
|
|
if (manifestFrames[i].Index != i) throw new WebDropException(ErrorCode.Conflict);
|
|
digests.Add(manifestFrames[i].Digest);
|
|
}
|
|
digests.Add(digest);
|
|
if (!CryptographicOperations.FixedTimeEquals(SignatureCodec.FrameManifestHash(digests), signedEvent.ManifestHash))
|
|
throw new WebDropException(ErrorCode.Conflict);
|
|
}
|
|
}
|
|
|
|
if (env.FrameIndex == 0)
|
|
{
|
|
if (env.IsControl && tx.CountControlItems(channelId) >= 4096) throw new WebDropException(ErrorCode.TooLarge);
|
|
if (item is not null) throw new WebDropException(ErrorCode.BadRequest);
|
|
long max = env.IsGated ? maxDownloads ?? channel.Settings.FileMaxDownloads : 0;
|
|
if (env.IsGated && !ServerSettingChoices.IsFileMaxDownloadsAllowed(max, channel.Settings.FileMaxDownloads)) throw new WebDropException(ErrorCode.BadRequest);
|
|
item = new Item(channelId, env.ItemId, principal.EndpointHash, env.ItemFlags, max, 0, 0, -1, null, 0,
|
|
now, env.IsControl ? long.MaxValue : now + ttlMs, null, principal.AccountName);
|
|
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 (signedEvent is not null && objectEnvelope is not null)
|
|
{
|
|
objectEnvelope.Event = ToSignedContentEventRecord(signedEvent);
|
|
crypto.Revision++;
|
|
tx.PutWorkspaceCrypto(channelId, crypto);
|
|
}
|
|
|
|
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)
|
|
{
|
|
if (env.IsControl && (env.IsGated || env.IsSnapshot || env.FrameIndex != 0 || !env.IsFinal || env.PayloadLength > 16_400)) throw new WebDropException(ErrorCode.BadRequest);
|
|
// 非 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<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);
|
|
|
|
long now = clock.NowMs;
|
|
var list = new List<(long Seq, byte[]? Bytes, StoredFrame? Frame)>();
|
|
foreach (var e in tx.ReadLog(id, from, limit))
|
|
{
|
|
if (e.ItemId != default && !CanReadObject(tx, channel, principal, e.ItemId)) continue;
|
|
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<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 (space, 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 (!CanReadObject(tx, space, principal, itemId)) throw new WebDropException(ErrorCode.Forbidden);
|
|
|
|
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;
|
|
});
|
|
}
|
|
|
|
/// <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 (space, 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 (!CanReadObject(tx, space, principal, itemId)) throw new WebDropException(ErrorCode.Forbidden);
|
|
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<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 ObjectEnvelope? ObjectSigningContext(Id16 id, Credentials creds, Id16 itemId) => store.Read(tx =>
|
|
{
|
|
_ = Authenticate(tx, id, creds, SeatRule.Required);
|
|
return tx.GetWorkspaceCrypto(id).Objects.Find(o => o.ItemId == itemId.ToBase64Url());
|
|
});
|
|
|
|
public void DeleteItem(Id16 id, Credentials creds, Id16 itemId, SignedContentEventProof? signedEvent = null)
|
|
{
|
|
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 crypto = tx.GetWorkspaceCrypto(id);
|
|
var obj = crypto.Objects.Find(o => o.ItemId == itemId.ToBase64Url());
|
|
if (creds.RequireSignedContent && obj is not null)
|
|
{
|
|
if (signedEvent is null || obj?.Event is null || signedEvent.Username != principal.AccountName
|
|
|| signedEvent.DeviceId != creds.AccountDeviceId || signedEvent.ChannelId != obj.ChannelId
|
|
|| signedEvent.Domain != obj.Domain || signedEvent.Epoch != obj.Epoch
|
|
|| signedEvent.FrameCount != obj.Event.FrameCount || signedEvent.ManifestHash.Length != 32
|
|
|| !Codec.TryDecodeExact(obj.Event.ManifestHash, 32, out var oldManifest)
|
|
|| !CryptographicOperations.FixedTimeEquals(signedEvent.ManifestHash, oldManifest)
|
|
|| signedEvent.EventKind != (obj.Domain == CryptoDomain.Chat ? "chat.delete" : "file.delete")
|
|
|| signedEvent.TargetItemId is not null || signedEvent.EventIdBytes.Length != 16 || signedEvent.Signature.Length != 64
|
|
|| obj.DeleteEvent is not null)
|
|
throw new WebDropException(ErrorCode.Forbidden);
|
|
obj.DeleteEvent = ToSignedContentEventRecord(signedEvent);
|
|
crypto.Revision++;
|
|
tx.PutWorkspaceCrypto(id, crypto);
|
|
}
|
|
else if (signedEvent is not null) throw new WebDropException(ErrorCode.BadRequest);
|
|
|
|
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;
|
|
}
|
|
|
|
private static SignedContentEventRecord ToSignedContentEventRecord(SignedContentEventProof signedEvent) => new()
|
|
{
|
|
EventId = Codec.Encode(signedEvent.EventIdBytes), Kind = signedEvent.EventKind, FrameCount = signedEvent.FrameCount,
|
|
ManifestHash = Convert.ToBase64String(signedEvent.ManifestHash).TrimEnd('=').Replace('+', '-').Replace('/', '_'),
|
|
SignerDeviceId = signedEvent.DeviceId,
|
|
Signature = Convert.ToBase64String(signedEvent.Signature).TrimEnd('=').Replace('+', '-').Replace('/', '_'),
|
|
TargetItemId = signedEvent.TargetItemId,
|
|
SignerRootPublicKey = signedEvent.RootPublicKey, SignerDevicePublicKey = signedEvent.DevicePublicKey,
|
|
DeviceCertificateSignature = Convert.ToBase64String(signedEvent.DeviceCertificateSignature).TrimEnd('=').Replace('+', '-').Replace('/', '_'),
|
|
};
|
|
|
|
// ───────────────────────── 后台清理 ─────────────────────────
|
|
|
|
/// <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);
|
|
var member = creds.AccountName is null ? null : tx.GetChannelMember(id, creds.AccountName);
|
|
if (role == Role.Owner && member?.Role != "owner") throw new WebDropException(ErrorCode.AccountRequired);
|
|
if (seat == SeatRule.Required)
|
|
{
|
|
var endpoint = tx.GetSeat(id, creds.EndpointHash);
|
|
if (endpoint is null) throw new WebDropException(ErrorCode.NotJoined);
|
|
if (endpoint.AccountName is not null && (endpoint.AccountName != creds.AccountName || member is null)) throw new WebDropException(ErrorCode.AccountRequired);
|
|
if (endpoint.AccountDeviceId is not null && endpoint.AccountDeviceId != creds.AccountDeviceId) throw new WebDropException(ErrorCode.AccountRequired);
|
|
}
|
|
return (channel, new Principal(role, creds.EndpointHash, creds.AccountName));
|
|
}
|
|
|
|
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 membersUsed, Principal? principal = null) => new()
|
|
{
|
|
CanUpload = ChannelPolicy.CanWrite(channel, principal ?? new Principal(Role.Owner, [], channel.CreatorAccount)),
|
|
CanManageStructure = principal is null || principal.Role == Role.Owner,
|
|
Id = channel.Id.ToBase64Url(),
|
|
CreatorAccount = channel.CreatorAccount,
|
|
PasswordRequired = channel.JoinPasswordHash is not null,
|
|
Mode = channel.Mode,
|
|
Settings = channel.Settings,
|
|
SeatsUsed = membersUsed,
|
|
OldestSeq = channel.CompactedThrough + 1,
|
|
LastSeq = channel.LastSeq,
|
|
CreatedAt = channel.CreatedAt,
|
|
ExpiresAt = channel.ExpiresAt,
|
|
};
|
|
}
|