WebDrop/server/src/WebDrop.Api/LogStreamer.cs

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, Func<bool>? validateAccount = null)
    {
        using var sub = hub.Subscribe(id);
        ILogSink? sink = null;
        long cursor = from;
        var (settingsV, seatsV, _) = sub.State;

        try
        {
            while (!ct.IsCancellationRequested)
            {
                if (validateAccount is not null && !validateAccount()) throw new WebDropException(ErrorCode.AccountRequired);
                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);
    }
}