WebDrop/server/src/WebDrop.Api/ChannelHub.cs

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 = [];

    // 在线状态:频道 -> 账号 -> 账号设备 -> 活动订阅数。只记账号,访客没有可跨设备识别的身份,不参与。
    private readonly Dictionary<Id16, Dictionary<string, Dictionary<string, int>>> _presence = [];

    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);
        }
    }

    /// <summary>登记一个活动订阅,释放返回值即下线。只供成员接口读取,不向订阅流推送,以免改变流的线格式。</summary>
    public IDisposable Connected(Id16 id, string? account, string deviceKey)
    {
        if (account is null) return NoPresence.Instance;
        string endpoint = deviceKey;
        lock (_gate)
        {
            if (!_presence.TryGetValue(id, out var accounts)) _presence[id] = accounts = [];
            if (!accounts.TryGetValue(account, out var endpoints)) accounts[account] = endpoints = [];
            endpoints[endpoint] = endpoints.GetValueOrDefault(endpoint) + 1;
        }
        return new Connection(() => Disconnected(id, account, endpoint));
    }

    private void Disconnected(Id16 id, string account, string endpoint)
    {
        lock (_gate)
        {
            if (_presence.TryGetValue(id, out var accounts) && accounts.TryGetValue(account, out var endpoints) && endpoints.TryGetValue(endpoint, out int count))
            {
                if (count <= 1) endpoints.Remove(endpoint); else endpoints[endpoint] = count - 1;
                if (endpoints.Count == 0) accounts.Remove(account);
                if (accounts.Count == 0) _presence.Remove(id);
            }
        }
    }

    public (bool Online, int Devices) Presence(Id16 id, string account)
    {
        lock (_gate)
        {
            return _presence.TryGetValue(id, out var accounts) && accounts.TryGetValue(account, out var endpoints)
                ? (endpoints.Count > 0, endpoints.Count) : (false, 0);
        }
    }

    private sealed class Connection(Action release) : IDisposable
    {
        private Action? _release = release;
        public void Dispose() => Interlocked.Exchange(ref _release, null)?.Invoke();
    }

    private sealed class NoPresence : IDisposable
    {
        public static readonly NoPresence Instance = new();
        public void Dispose() { }
    }

    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 MembersChanged(Id16 channelId) => Publish(channelId, s => s.MembersVersion++);
    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 MembersVersion;
        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 MaxMembers, bool Deleted) State
        {
            get { lock (_hub._gate) return (_signal.SettingsVersion, _signal.MembersVersion, _signal.Deleted); }
        }

        public void Dispose()
        {
            if (Interlocked.Exchange(ref _disposed, 1) == 0) _hub.Release(_id, _signal);
        }
    }
}