erik/acdream
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 2

refactor(runtime): own communication and social state

Browse source 
Construct chat history, negotiated channels, friends, squelch, and reply targets in one presentation-independent Runtime owner. Make live routing, retained UI, devtools, and current-runtime projections borrow the exact instances, preserve reconnect reset semantics, and publish failure-isolated reentrant-safe chat commits.

Co-authored-by: Codex <codex@openai.com>
This commit is contained in:
Erik 2026-07-26 07:50:16 +02:00
parent e321410770
commit c9d25ade50
19 changed files with 684 additions and 117 deletions
Download patch file Download diff file Expand all files Collapse all files

263
src/AcDream.Runtime/Gameplay/RuntimeCommunicationState.cs Normal file
View file

@ -0,0 +1,263 @@
using AcDream.Core.Chat;
using AcDream.Core.Social;
namespace AcDream.Runtime.Gameplay;
public readonly record struct RuntimeCommunicationEvent(
ulong Sequence,
RuntimeChatEntry Entry);
public interface IRuntimeCommunicationObserver
{
void OnChat(in RuntimeCommunicationEvent delta);
}
public interface IRuntimeCommunicationEventSource
{
IDisposable Subscribe(IRuntimeCommunicationObserver observer);
}
/// <summary>
/// Canonical presentation-independent owner for communication and social
/// state. Graphical, headless, plugin, and bot hosts borrow these exact
/// instances.
/// </summary>
public sealed class RuntimeCommunicationState : IDisposable
{
private readonly RuntimeCommunicationEventStream _events;
private bool _disposed;
public RuntimeCommunicationState(int maximumChatEntries = 500)
{
Chat = new ChatLog(maximumChatEntries);
CommandTargets = new ChatCommandTargetState(Chat);
_events = new RuntimeCommunicationEventStream(Chat);
TurbineChat = new TurbineChatState();
Friends = new FriendsState();
Squelch = new SquelchState();
View = new CommunicationView(Chat);
}
public ChatLog Chat { get; }
public ChatCommandTargetState CommandTargets { get; }
public TurbineChatState TurbineChat { get; }
public FriendsState Friends { get; }
public SquelchState Squelch { get; }
public IRuntimeChatView View { get; }
public IRuntimeCommunicationEventSource Events => _events;
public bool IsDisposed => _disposed;
public ulong LastSequence => _events.LastSequence;
public int SubscriberCount => _events.SubscriberCount;
public int PendingDispatchCount => _events.PendingDispatchCount;
public bool IsDispatching => _events.IsDispatching;
public long DispatchFailureCount => _events.DispatchFailureCount;
public Exception? LastDispatchFailure => _events.LastDispatchFailure;
public void ResetCommandTargets() => CommandTargets.ResetSession();
public void ResetChatIdentity() => Chat.ResetSessionIdentity();
public void ResetNegotiatedChannels() => TurbineChat.Reset();
public void ResetFriends() => Friends.Clear();
public void ResetSquelch() => Squelch.Clear();
public void Dispose()
{
if (_disposed)
return;
_disposed = true;
_events.Dispose();
CommandTargets.ResetSession();
CommandTargets.Dispose();
TurbineChat.Reset();
Friends.Clear();
Squelch.Clear();
Chat.ResetSessionIdentity();
}
private sealed class CommunicationView(ChatLog chat) : IRuntimeChatView
{
public long Revision => chat.Revision;
public int Count => chat.Count;
}
}
internal sealed class RuntimeCommunicationEventStream
: IRuntimeCommunicationEventSource,
IDisposable
{
private readonly ChatLog _chat;
private readonly object _gate = new();
private readonly List<RuntimeCommunicationEvent> _pendingDispatch = [];
private IRuntimeCommunicationObserver[] _observers = [];
private long _sequence;
private long _dispatchFailureCount;
private bool _dispatching;
private bool _disposed;
public RuntimeCommunicationEventStream(ChatLog chat)
{
_chat = chat ?? throw new ArgumentNullException(nameof(chat));
_chat.EntryAppended += OnEntryAppended;
}
public int SubscriberCount => Volatile.Read(ref _observers).Length;
public ulong LastSequence
{
get
{
lock (_gate)
return unchecked((ulong)_sequence);
}
}
public int PendingDispatchCount
{
get
{
lock (_gate)
return _pendingDispatch.Count;
}
}
public bool IsDispatching
{
get
{
lock (_gate)
return _dispatching;
}
}
public long DispatchFailureCount => Interlocked.Read(ref _dispatchFailureCount);
public Exception? LastDispatchFailure { get; private set; }
public IDisposable Subscribe(IRuntimeCommunicationObserver observer)
{
ArgumentNullException.ThrowIfNull(observer);
lock (_gate)
{
ObjectDisposedException.ThrowIf(_disposed, this);
IRuntimeCommunicationObserver[] current = _observers;
if (Array.IndexOf(current, observer) >= 0)
throw new InvalidOperationException(
"The communication observer is already subscribed.");
var replacement =
new IRuntimeCommunicationObserver[current.Length + 1];
Array.Copy(current, replacement, current.Length);
replacement[^1] = observer;
Volatile.Write(ref _observers, replacement);
}
return new Subscription(this, observer);
}
public void Dispose()
{
lock (_gate)
{
if (_disposed)
return;
_disposed = true;
_chat.EntryAppended -= OnEntryAppended;
_pendingDispatch.Clear();
_dispatching = false;
Volatile.Write(ref _observers, []);
}
}
private void OnEntryAppended(ChatEntry entry)
{
lock (_gate)
{
if (_disposed)
return;
ulong sequence = unchecked((ulong)++_sequence);
_pendingDispatch.Add(new RuntimeCommunicationEvent(
sequence,
new RuntimeChatEntry(
_chat.Revision,
entry.SenderGuid,
(int)entry.Kind,
entry.Sender,
entry.Text,
entry.ChannelName)));
if (_dispatching)
return;
_dispatching = true;
}
int index = 0;
while (true)
{
RuntimeCommunicationEvent pending;
lock (_gate)
{
if (index >= _pendingDispatch.Count)
{
_pendingDispatch.Clear();
_dispatching = false;
return;
}
pending = _pendingDispatch[index++];
}
Dispatch(in pending);
}
}
private void Dispatch(in RuntimeCommunicationEvent delta)
{
IRuntimeCommunicationObserver[] observers =
Volatile.Read(ref _observers);
foreach (IRuntimeCommunicationObserver observer in observers)
{
try
{
observer.OnChat(in delta);
}
catch (Exception error)
{
Interlocked.Increment(ref _dispatchFailureCount);
LastDispatchFailure = error;
}
}
}
private void Unsubscribe(IRuntimeCommunicationObserver observer)
{
lock (_gate)
{
IRuntimeCommunicationObserver[] current = _observers;
int index = Array.IndexOf(current, observer);
if (index < 0)
return;
if (current.Length == 1)
{
Volatile.Write(ref _observers, []);
return;
}
var replacement =
new IRuntimeCommunicationObserver[current.Length - 1];
if (index > 0)
Array.Copy(current, 0, replacement, 0, index);
if (index < current.Length - 1)
{
Array.Copy(
current,
index + 1,
replacement,
index,
current.Length - index - 1);
}
Volatile.Write(ref _observers, replacement);
}
}
private sealed class Subscription(
RuntimeCommunicationEventStream owner,
IRuntimeCommunicationObserver observer)
: IDisposable
{
private RuntimeCommunicationEventStream? _owner = owner;
public void Dispose() =>
Interlocked.Exchange(ref _owner, null)?.Unsubscribe(observer);
}
}