acdream/src/AcDream.Core/Plugins/WorldEvents.cs

// src/AcDream.Core/Plugins/WorldEvents.cs
using AcDream.Plugin.Abstractions;

namespace AcDream.Core.Plugins;

public sealed class WorldEvents : IEvents
{
    private readonly object _lock = new();
    // Late subscribers replay the current projected world, not every entity
    // encountered since process start. Spawn notifications themselves remain
    // ephemeral; the public plugin API does not expose a despawn event yet.
    private readonly Dictionary<uint, WorldEntitySnapshot> _current = new();
    private readonly List<Subscription> _subscriptions = new();
    private Subscription[] _liveSnapshot = Array.Empty<Subscription>();

    private sealed class Subscription(Action<WorldEntitySnapshot> handler)
    {
        public Action<WorldEntitySnapshot> Handler { get; } = handler;
        public Queue<WorldEntitySnapshot> Pending { get; } = new();
        public bool Replaying { get; set; } = true;
        public bool Active { get; set; } = true;
    }

    /// <summary>
    /// Called by the host as each entity is hydrated into the world. Records the
    /// snapshot for later replay and dispatches to current subscribers.
    /// </summary>
    public void FireEntitySpawned(WorldEntitySnapshot snapshot)
    {
        Subscription[] toNotify;
        lock (_lock)
        {
            _current[snapshot.Id] = snapshot;
            for (int i = 0; i < _subscriptions.Count; i++)
            {
                Subscription subscription = _subscriptions[i];
                if (subscription.Active && subscription.Replaying)
                    subscription.Pending.Enqueue(snapshot);
            }
            toNotify = _liveSnapshot;
        }

        for (int i = 0; i < toNotify.Length; i++)
        {
            try { toNotify[i].Handler(snapshot); }
            catch { /* plugin errors don't propagate out of event dispatch */ }
        }
    }

    /// <summary>
    /// Restore current replay membership without emitting another logical
    /// spawn notification. Used when a still-live object re-enters the world.
    /// </summary>
    public void UpsertCurrent(WorldEntitySnapshot snapshot)
    {
        lock (_lock)
            _current[snapshot.Id] = snapshot;
    }

    public bool ForgetEntity(uint id)
    {
        lock (_lock)
            return _current.Remove(id);
    }

    public void ClearCurrent()
    {
        lock (_lock)
        {
            _current.Clear();
            for (int i = 0; i < _subscriptions.Count; i++)
            {
                Subscription subscription = _subscriptions[i];
                if (subscription.Replaying)
                    subscription.Pending.Clear();
            }
        }
    }

    public event Action<WorldEntitySnapshot> EntitySpawned
    {
        add
        {
            ArgumentNullException.ThrowIfNull(value);
            var subscription = new Subscription(value);
            WorldEntitySnapshot[] replay;
            lock (_lock)
            {
                _subscriptions.Add(subscription);
                replay = _current.Values.ToArray();
            }

            // Replay outside the lock. Live events that arrive meanwhile queue
            // behind this snapshot and drain before the subscription joins the
            // direct multicast, preserving one monotonic delivery order.
            foreach (var s in replay)
            {
                lock (_lock)
                {
                    if (!subscription.Active)
                        return;
                    if (!_current.TryGetValue(s.Id, out WorldEntitySnapshot current)
                        || current != s)
                    {
                        continue;
                    }
                }

                try { subscription.Handler(s); }
                catch { /* plugin errors don't propagate out of += */ }
            }

            while (true)
            {
                WorldEntitySnapshot pending;
                lock (_lock)
                {
                    if (!subscription.Active)
                        return;
                    if (!subscription.Pending.TryDequeue(out pending))
                    {
                        subscription.Replaying = false;
                        RebuildLiveSnapshotLocked();
                        return;
                    }
                }

                try { subscription.Handler(pending); }
                catch { /* plugin errors don't propagate out of += */ }
            }
        }
        remove
        {
            if (value is null)
                return;
            lock (_lock)
            {
                for (int i = _subscriptions.Count - 1; i >= 0; i--)
                {
                    Subscription subscription = _subscriptions[i];
                    if (subscription.Handler != value)
                        continue;

                    subscription.Active = false;
                    subscription.Pending.Clear();
                    _subscriptions.RemoveAt(i);
                    if (!subscription.Replaying)
                        RebuildLiveSnapshotLocked();
                    break;
                }
            }
        }
    }

    private void RebuildLiveSnapshotLocked()
    {
        int count = 0;
        for (int i = 0; i < _subscriptions.Count; i++)
        {
            Subscription subscription = _subscriptions[i];
            if (subscription.Active && !subscription.Replaying)
                count++;
        }

        if (count == 0)
        {
            _liveSnapshot = Array.Empty<Subscription>();
            return;
        }

        var rebuilt = new Subscription[count];
        int write = 0;
        for (int i = 0; i < _subscriptions.Count; i++)
        {
            Subscription subscription = _subscriptions[i];
            if (subscription.Active && !subscription.Replaying)
                rebuilt[write++] = subscription;
        }
        _liveSnapshot = rebuilt;
    }
}