Cleaned up request acks. Added internal service bus for internal messaging.

Bus/Data/RedemptionInitiation.cs

@@ -0,0 +1,10 @@
+using HermesSocketLibrary.Requests.Messages;
+
+namespace TwitchChatTTS.Bus.Data
+{
+    public class RedemptionInitiation
+    {
+        public required IEnumerable<Redemption> Redemptions { get; set; }
+        public required IDictionary<string, RedeemableAction> Actions { get; set; }
+    }
+}

Bus/ServiceBusCentral.cs

@@ -0,0 +1,83 @@
+using System.Collections.Immutable;
+using Serilog;
+
+namespace TwitchChatTTS.Bus
+{
+    public class ServiceBusCentral
+    {
+        private readonly IDictionary<string, ServiceBusObservable> _topics;
+        private readonly IDictionary<string, ISet<IObserver<ServiceBusData>>> _receivers;
+        private readonly ILogger _logger;
+        private readonly object _lock;
+
+        public ServiceBusCentral(ILogger logger)
+        {
+            _topics = new Dictionary<string, ServiceBusObservable>();
+            _receivers = new Dictionary<string, ISet<IObserver<ServiceBusData>>>();
+            _logger = logger;
+            _lock = new object();
+        }
+
+        public void Add(string topic, IObserver<ServiceBusData> observer)
+        {
+            lock (_lock)
+            {
+                if (!_receivers.TryGetValue(topic, out var observers))
+                {
+                    observers = new HashSet<IObserver<ServiceBusData>>();
+                    _receivers.Add(topic, observers);
+                }
+                observers.Add(observer);
+            }
+        }
+
+        public ServiceBusObservable GetTopic(string topic)
+        {
+            lock (_lock)
+            {
+                if (!_topics.TryGetValue(topic, out var bus))
+                {
+                    bus = new ServiceBusObservable(topic, this);
+                    _topics.Add(topic, bus);
+                }
+                return bus;
+            }
+        }
+
+        public IEnumerable<IObserver<ServiceBusData>> GetObservers(string topic)
+        {
+            lock (_lock)
+            {
+                if (_receivers.TryGetValue(topic, out var observers))
+                    return observers.ToImmutableArray();
+            }
+            return [];
+        }
+
+        public bool RemoveObserver(string topic, IObserver<ServiceBusData> observer)
+        {
+            lock (_lock)
+            {
+                if (_receivers.TryGetValue(topic, out var observers))
+                    return observers.Remove(observer);
+            }
+            return false;
+        }
+
+        public void Send(object sender, string topic, object value)
+        {
+            var observers = GetObservers(topic);
+            foreach (var consumer in observers)
+            {
+                try
+                {
+                    consumer.OnNext(new ServiceBusData(sender, topic, value));
+                }
+                catch (Exception ex)
+                {
+                    _logger.Error(ex, "Failed to execute observer on send.");
+                }
+            }
+        }
+    }
+}

Bus/ServiceBusData.cs

@@ -0,0 +1,18 @@
+namespace TwitchChatTTS.Bus
+{
+    public class ServiceBusData
+    {
+        public string Topic { get; }
+        public object? Sender { get; }
+        public object? Value { get; }
+        public DateTime Timestamp { get; }
+
+        public ServiceBusData(object sender, string topic, object value)
+        {
+            Topic = topic;
+            Sender = sender;
+            Value = value;
+            Timestamp = DateTime.UtcNow;
+        }
+    }
+}

Bus/ServiceBusObservable.cs

@@ -0,0 +1,41 @@
+using System.Reactive;
+
+namespace TwitchChatTTS.Bus
+{
+    public class ServiceBusObservable : ObservableBase<ServiceBusData>
+    {
+        private readonly string _topic;
+        private readonly ServiceBusCentral _central;
+
+        public ServiceBusObservable(string topic, ServiceBusCentral central)
+        {
+            _topic = topic;
+            _central = central;
+        }
+
+        protected override IDisposable SubscribeCore(IObserver<ServiceBusData> observer)
+        {
+            _central.Add(_topic, observer);
+            return new ServiceBusUnsubscriber(_topic, _central, observer);
+        }
+
+        private sealed class ServiceBusUnsubscriber : IDisposable
+        {
+            private readonly string _topic;
+            private readonly ServiceBusCentral _central;
+            private readonly IObserver<ServiceBusData> _receiver;
+
+            public ServiceBusUnsubscriber(string topic, ServiceBusCentral central, IObserver<ServiceBusData> receiver)
+            {
+                _topic = topic;
+                _central = central;
+                _receiver = receiver;
+            }
+
+            public void Dispose()
+            {
+                _central.RemoveObserver(_topic, _receiver);
+            }
+        }
+    }
+}

Bus/ServiceBusObserver.cs

@@ -0,0 +1,31 @@
+using System.Reactive;
+using Serilog;
+
+namespace TwitchChatTTS.Bus
+{
+    public class ServiceBusObserver : ObserverBase<ServiceBusData>
+    {
+        private readonly Action<ServiceBusData> _action;
+        private readonly ILogger _logger;
+
+        public ServiceBusObserver(Action<ServiceBusData> action, ILogger logger)
+        {
+            _action = action;
+            _logger = logger;
+        }
+
+        protected override void OnCompletedCore()
+        {
+        }
+
+        protected override void OnErrorCore(Exception error)
+        {
+            _logger.Error(error, "Error occurred.");
+        }
+
+        protected override void OnNextCore(ServiceBusData value)
+        {
+            _action.Invoke(value);
+        }
+    }
+}

Chat/Observers/TTSPublisher.cs

@@ -5,24 +5,17 @@ namespace TwitchChatTTS.Chat.Observers
     public class TTSPublisher : IObservable<TTSGroupedMessage>
     {
         private readonly HashSet<IObserver<TTSGroupedMessage>> _observers;
-        private readonly HashSet<TTSGroupedMessage> _messages;
 
 
         public TTSPublisher()
         {
             _observers = new();
-            _messages = new();
         }
 
 
         public IDisposable Subscribe(IObserver<TTSGroupedMessage> observer)
         {
-            if (_observers.Add(observer))
-            {
-                foreach (var item in _messages)
-                    observer.OnNext(item);
-            }
-
+            _observers.Add(observer);
             return new Unsubscriber<TTSGroupedMessage>(_observers, observer);
         }
     }

Hermes/Socket/Handlers/RequestAckHandler.cs

@@ -1,59 +1,25 @@
-using System.Collections.Concurrent;
-using System.Text.Json;
-using System.Text.RegularExpressions;
 using CommonSocketLibrary.Abstract;
 using CommonSocketLibrary.Common;
-using HermesSocketLibrary.Requests.Callbacks;
-using HermesSocketLibrary.Requests.Messages;
 using HermesSocketLibrary.Socket.Data;
-using HermesSocketServer.Models;
-using Microsoft.Extensions.DependencyInjection;
 using Serilog;
-using TwitchChatTTS.Chat.Commands.Limits;
-using TwitchChatTTS.Chat.Emotes;
-using TwitchChatTTS.Chat.Groups;
-using TwitchChatTTS.Chat.Groups.Permissions;
-using TwitchChatTTS.Twitch.Redemptions;
+using TwitchChatTTS.Hermes.Socket.Requests;
 
 namespace TwitchChatTTS.Hermes.Socket.Handlers
 {
     public class RequestAckHandler : IWebSocketHandler
     {
-        private User _user;
-        private readonly ICallbackManager<HermesRequestData> _callbackManager;
-        private readonly IChatterGroupManager _groups;
-        private readonly IUsagePolicy<long> _policies;
-        private readonly TwitchApiClient _twitch;
-        private readonly NightbotApiClient _nightbot;
-        private readonly IServiceProvider _serviceProvider;
-        private readonly JsonSerializerOptions _options;
+        private readonly RequestAckManager _manager;
         private readonly ILogger _logger;
 
-        private readonly object _voicesAvailableLock = new object();
-
         public int OperationCode { get; } = 4;
 
 
         public RequestAckHandler(
-            ICallbackManager<HermesRequestData> callbackManager,
-            IChatterGroupManager groups,
-            IUsagePolicy<long> policies,
-            TwitchApiClient twitch,
-            NightbotApiClient nightbot,
-            IServiceProvider serviceProvider,
-            User user,
-            JsonSerializerOptions options,
+            RequestAckManager manager,
             ILogger logger
         )
         {
-            _callbackManager = callbackManager;
-            _groups = groups;
-            _policies = policies;
-            _twitch = twitch;
-            _nightbot = nightbot;
-            _serviceProvider = serviceProvider;
-            _user = user;
-            _options = options;
+            _manager = manager;
             _logger = logger;
         }
 
@@ -67,384 +33,13 @@ namespace TwitchChatTTS.Hermes.Socket.Handlers
                 return;
             }
 
-            HermesRequestData? hermesRequestData = null;
-            if (!string.IsNullOrEmpty(message.Request.RequestId))
-            {
-                hermesRequestData = _callbackManager.Take(message.Request.RequestId);
-                if (hermesRequestData == null)
-                    _logger.Warning($"Could not find callback for request [request id: {message.Request.RequestId}][type: {message.Request.Type}]");
-                else if (hermesRequestData.Data == null)
-                    hermesRequestData.Data = new Dictionary<string, object>();
-            }
-
             _logger.Debug($"Received a Hermes request message [type: {message.Request.Type}][data: {string.Join(',', message.Request.Data?.Select(entry => entry.Key + '=' + entry.Value) ?? Array.Empty<string>())}]");
-            if (message.Request.Type == "get_tts_voices")
+            var json = message.Data?.ToString();
+            if (message.Request.Type == null || json == null)
             {
-                var voices = JsonSerializer.Deserialize<IEnumerable<VoiceDetails>>(message.Data.ToString(), _options);
-                if (voices == null)
-                    return;
-
-                lock (_voicesAvailableLock)
-                {
-                    _user.VoicesAvailable = voices.ToDictionary(e => e.Id, e => e.Name);
-                }
-                _logger.Information("Updated all available voices for TTS.");
-            }
-            else if (message.Request.Type == "create_tts_user")
-            {
-                if (!long.TryParse(message.Request.Data["chatter"].ToString(), out long chatterId))
-                {
-                    _logger.Warning($"Failed to parse chatter id [chatter id: {message.Request.Data["chatter"]}]");
-                    return;
-                }
-                string userId = message.Request.Data["user"].ToString();
-                string voiceId = message.Request.Data["voice"].ToString();
-
-                _user.VoicesSelected.Add(chatterId, voiceId);
-                _logger.Information($"Added new TTS voice [voice: {voiceId}] for user [user id: {userId}]");
-            }
-            else if (message.Request.Type == "update_tts_user")
-            {
-                if (!long.TryParse(message.Request.Data["chatter"].ToString(), out long chatterId))
-                {
-                    _logger.Warning($"Failed to parse chatter id [chatter id: {message.Request.Data["chatter"]}]");
-                    return;
-                }
-                string userId = message.Request.Data["user"].ToString();
-                string voiceId = message.Request.Data["voice"].ToString();
-
-                _user.VoicesSelected[chatterId] = voiceId;
-                _logger.Information($"Updated TTS voice [voice: {voiceId}] for user [user id: {userId}]");
-            }
-            else if (message.Request.Type == "create_tts_voice")
-            {
-                string? voice = message.Request.Data["voice"].ToString();
-                string? voiceId = message.Data.ToString();
-                if (voice == null || voiceId == null)
-                    return;
-
-                lock (_voicesAvailableLock)
-                {
-                    var list = _user.VoicesAvailable.ToDictionary(k => k.Key, v => v.Value);
-                    list.Add(voiceId, voice);
-                    _user.VoicesAvailable = list;
-                }
-                _logger.Information($"Created new tts voice [voice: {voice}][id: {voiceId}].");
-            }
-            else if (message.Request.Type == "delete_tts_voice")
-            {
-                var voice = message.Request.Data["voice"].ToString();
-                if (!_user.VoicesAvailable.TryGetValue(voice, out string? voiceName) || voiceName == null)
-                    return;
-
-                lock (_voicesAvailableLock)
-                {
-                    var dict = _user.VoicesAvailable.ToDictionary(k => k.Key, v => v.Value);
-                    dict.Remove(voice);
-                    _user.VoicesAvailable.Remove(voice);
-                }
-                _logger.Information($"Deleted a voice [voice: {voiceName}]");
-            }
-            else if (message.Request.Type == "update_tts_voice")
-            {
-                string voiceId = message.Request.Data["idd"].ToString();
-                string voice = message.Request.Data["voice"].ToString();
-
-                if (!_user.VoicesAvailable.TryGetValue(voiceId, out string? voiceName) || voiceName == null)
-                    return;
-
-                _user.VoicesAvailable[voiceId] = voice;
-                _logger.Information($"Updated TTS voice [voice: {voice}][id: {voiceId}]");
-            }
-            else if (message.Request.Type == "get_connections")
-            {
-                var connections = JsonSerializer.Deserialize<IEnumerable<Connection>>(message.Data?.ToString(), _options);
-                if (connections == null)
-                {
-                    _logger.Error("Null value was given when attempting to fetch connections.");
-                    _logger.Debug(message.Data?.ToString());
-                    return;
-                }
-
-                _user.TwitchConnection = connections.FirstOrDefault(c => c.Type == "twitch" && c.Default);
-                _user.NightbotConnection = connections.FirstOrDefault(c => c.Type == "nightbot" && c.Default);
-
-                if (_user.TwitchConnection != null)
-                    _twitch.Initialize(_user.TwitchConnection.ClientId, _user.TwitchConnection.AccessToken);
-                if (_user.NightbotConnection != null)
-                    _nightbot.Initialize(_user.NightbotConnection.ClientId, _user.NightbotConnection.AccessToken);
-
-                _logger.Information($"Fetched connections from TTS account [count: {connections.Count()}][twitch: {_user.TwitchConnection != null}][nightbot: {_user.NightbotConnection != null}]");
-            }
-            else if (message.Request.Type == "get_tts_users")
-            {
-                var users = JsonSerializer.Deserialize<IDictionary<long, string>>(message.Data.ToString(), _options);
-                if (users == null)
-                    return;
-
-                var temp = new ConcurrentDictionary<long, string>();
-                foreach (var entry in users)
-                    temp.TryAdd(entry.Key, entry.Value);
-                _user.VoicesSelected = temp;
-                _logger.Information($"Updated {temp.Count()} chatters' selected voice.");
-            }
-            else if (message.Request.Type == "get_chatter_ids")
-            {
-                var chatters = JsonSerializer.Deserialize<IEnumerable<long>>(message.Data.ToString(), _options);
-                if (chatters == null)
-                    return;
-
-                _user.Chatters = [.. chatters];
-                _logger.Information($"Fetched {chatters.Count()} chatters' id.");
-            }
-            else if (message.Request.Type == "get_emotes")
-            {
-                var emotes = JsonSerializer.Deserialize<IEnumerable<EmoteInfo>>(message.Data.ToString(), _options);
-                if (emotes == null)
-                    return;
-
-                var emoteDb = _serviceProvider.GetRequiredService<IEmoteDatabase>();
-                var count = 0;
-                var duplicateNames = 0;
-                foreach (var emote in emotes)
-                {
-                    if (emoteDb.Get(emote.Name) == null)
-                    {
-                        emoteDb.Add(emote.Name, emote.Id);
-                        count++;
-                    }
-                    else
-                        duplicateNames++;
-                }
-                _logger.Information($"Fetched {count} emotes from various sources.");
-                if (duplicateNames > 0)
-                    _logger.Warning($"Found {duplicateNames} emotes with duplicate names.");
-            }
-            else if (message.Request.Type == "get_enabled_tts_voices")
-            {
-                var enabledTTSVoices = JsonSerializer.Deserialize<IEnumerable<string>>(message.Data.ToString(), _options);
-                if (enabledTTSVoices == null)
-                {
-                    _logger.Error("Failed to load enabled tts voices.");
-                    return;
-                }
-
-                if (_user.VoicesEnabled == null)
-                    _user.VoicesEnabled = enabledTTSVoices.ToHashSet();
-                else
-                    _user.VoicesEnabled.Clear();
-                foreach (var voice in enabledTTSVoices)
-                    _user.VoicesEnabled.Add(voice);
-                _logger.Information($"TTS voices [count: {_user.VoicesEnabled.Count}] have been enabled.");
-            }
-            else if (message.Request.Type == "get_permissions")
-            {
-                var groupInfo = JsonSerializer.Deserialize<GroupInfo>(message.Data.ToString(), _options);
-                if (groupInfo == null)
-                {
-                    _logger.Error("Failed to load groups & permissions.");
-                    return;
-                }
-
-                var chatterGroupManager = _serviceProvider.GetRequiredService<IChatterGroupManager>();
-                var permissionManager = _serviceProvider.GetRequiredService<IGroupPermissionManager>();
-
-                permissionManager.Clear();
-                chatterGroupManager.Clear();
-
-                var groupsById = groupInfo.Groups.ToDictionary(g => g.Id, g => g);
-                foreach (var group in groupInfo.Groups)
-                    chatterGroupManager.Add(group);
-
-                foreach (var permission in groupInfo.GroupPermissions)
-                {
-                    _logger.Debug($"Adding group permission [permission id: {permission.Id}][group id: {permission.GroupId}][path: {permission.Path}][allow: {permission.Allow?.ToString() ?? "null"}]");
-                    if (!groupsById.TryGetValue(permission.GroupId, out var group))
-                    {
-                        _logger.Warning($"Failed to find group by id [permission id: {permission.Id}][group id: {permission.GroupId}][path: {permission.Path}]");
-                        continue;
-                    }
-
-
-                    var path = $"{group.Name}.{permission.Path}";
-                    permissionManager.Set(path, permission.Allow);
-                    _logger.Debug($"Added group permission [id: {permission.Id}][group id: {permission.GroupId}][path: {permission.Path}]");
-                }
-
-                _logger.Information($"Groups [count: {groupInfo.Groups.Count()}] & Permissions [count: {groupInfo.GroupPermissions.Count()}] have been loaded.");
-
-                foreach (var chatter in groupInfo.GroupChatters)
-                    if (groupsById.TryGetValue(chatter.GroupId, out var group))
-                        chatterGroupManager.Add(chatter.ChatterId, group.Name);
-                _logger.Information($"Users in each group [count: {groupInfo.GroupChatters.Count()}] have been loaded.");
-            }
-            else if (message.Request.Type == "get_tts_word_filters")
-            {
-                var wordFilters = JsonSerializer.Deserialize<IEnumerable<TTSWordFilter>>(message.Data.ToString(), _options);
-                if (wordFilters == null)
-                {
-                    _logger.Error("Failed to load word filters.");
-                    return;
-                }
-
-                var filters = wordFilters.Where(f => f.Search != null && f.Replace != null).ToArray();
-                foreach (var filter in filters)
-                {
-                    try
-                    {
-                        var re = new Regex(filter.Search!, RegexOptions.Compiled);
-                        re.Match(string.Empty);
-                        filter.Regex = re;
-                    }
-                    catch (Exception) { }
-                }
-                _user.RegexFilters = filters;
-                _logger.Information($"TTS word filters [count: {_user.RegexFilters.Count()}] have been refreshed.");
-            }
-            else if (message.Request.Type == "update_tts_voice_state")
-            {
-                string voiceId = message.Request.Data?["voice"].ToString()!;
-                bool state = message.Request.Data?["state"].ToString()!.ToLower() == "true";
-
-                if (!_user.VoicesAvailable.TryGetValue(voiceId, out string? voiceName) || voiceName == null)
-                {
-                    _logger.Warning($"Failed to find voice by id [id: {voiceId}]");
-                    return;
-                }
-
-                if (state)
-                    _user.VoicesEnabled.Add(voiceId);
-                else
-                    _user.VoicesEnabled.Remove(voiceId);
-                _logger.Information($"Updated voice state [voice: {voiceName}][new state: {(state ? "enabled" : "disabled")}]");
-            }
-            else if (message.Request.Type == "get_redemptions")
-            {
-                IEnumerable<Redemption>? redemptions = JsonSerializer.Deserialize<IEnumerable<Redemption>>(message.Data!.ToString()!, _options);
-                if (redemptions != null)
-                {
-                    _logger.Information($"Redemptions [count: {redemptions.Count()}] loaded.");
-                    if (hermesRequestData != null)
-                        hermesRequestData.Data!.Add("redemptions", redemptions);
-                }
-                else
-                    _logger.Information(message.Data.GetType().ToString());
-            }
-            else if (message.Request.Type == "get_redeemable_actions")
-            {
-                IEnumerable<RedeemableAction>? actions = JsonSerializer.Deserialize<IEnumerable<RedeemableAction>>(message.Data!.ToString()!, _options);
-                if (actions == null)
-                {
-                    _logger.Warning("Failed to read the redeemable actions for redemptions.");
-                    return;
-                }
-                if (hermesRequestData?.Data == null || hermesRequestData.Data["redemptions"] is not IEnumerable<Redemption> redemptions)
-                {
-                    _logger.Warning("Failed to read the redemptions while updating redemption actions.");
-                    return;
-                }
-
-                _logger.Information($"Redeemable actions [count: {actions.Count()}] loaded.");
-                var redemptionManager = _serviceProvider.GetRequiredService<IRedemptionManager>();
-                redemptionManager.Initialize(redemptions, actions.ToDictionary(a => a.Name, a => a));
-            }
-            else if (message.Request.Type == "get_default_tts_voice")
-            {
-                string? defaultVoice = message.Data?.ToString();
-                if (defaultVoice != null)
-                {
-                    _user.DefaultTTSVoice = defaultVoice;
-                    _logger.Information($"Default TTS voice was changed to '{defaultVoice}'.");
-                }
-            }
-            else if (message.Request.Type == "update_default_tts_voice")
-            {
-                if (message.Request.Data?.TryGetValue("voice", out object? voice) == true && voice is string v)
-                {
-                    _user.DefaultTTSVoice = v;
-                    _logger.Information($"Default TTS voice was changed to '{v}'.");
-                }
-                else
-                    _logger.Warning("Failed to update default TTS voice via request.");
-            }
-            else if (message.Request.Type == "get_policies")
-            {
-                var policies = JsonSerializer.Deserialize<IEnumerable<PolicyMessage>>(message.Data!.ToString()!, _options);
-                if (policies == null || !policies.Any())
-                {
-                    _logger.Information($"Policies have been set to default.");
-                    _policies.Set("everyone", "tts", 100, TimeSpan.FromSeconds(15));
-                    return;
-                }
-
-                foreach (var policy in policies)
-                {
-                    var group = _groups.Get(policy.GroupId.ToString());
-                    if (policy == null)
-                    {
-                        _logger.Debug($"Policy data failed");
-                        continue;
-                    }
-                    _logger.Debug($"Policy data [policy id: {policy.Id}][path: {policy.Path}][group id: {policy.GroupId}][group name: {group?.Name}]");
-                    _policies.Set(group?.Name ?? string.Empty, policy.Path, policy.Usage, TimeSpan.FromMilliseconds(policy.Span));
-                }
-                _logger.Information($"Policies have been loaded, a total of {policies.Count()} policies.");
-            }
-            else if (message.Request.Type == "update_policy")
-            {
-                var policy = JsonSerializer.Deserialize<PolicyMessage>(message.Data!.ToString()!, _options);
-                var group = _groups.Get(policy.GroupId.ToString());
-                if (policy == null || group == null)
-                {
-                    _logger.Debug($"Policy data failed");
-                    return;
-                }
-                _logger.Debug($"Policy data [policy id: {policy.Id}][path: {policy.Path}][group id: {policy.GroupId}][group name: {group?.Name}]");
-                _policies.Set(group?.Name ?? string.Empty, policy.Path, policy.Usage, TimeSpan.FromMilliseconds(policy.Span));
-                _logger.Information($"Policy has been updated [policy id: {policy.Id}]");
-            }
-            else if (message.Request.Type == "create_policy")
-            {
-                var policy = JsonSerializer.Deserialize<PolicyMessage>(message.Data!.ToString()!, _options);
-
-                if (policy == null)
-                {
-                    _logger.Debug($"Policy data failed");
-                    return;
-                }
-                var group = _groups.Get(policy.GroupId.ToString());
-                if (group == null)
-                {
-                    _logger.Debug($"Group data failed");
-                    return;
-                }
-                _logger.Debug($"Policy data [policy id: {policy.Id}][path: {policy.Path}][group id: {policy.GroupId}][group name: {group?.Name}]");
-                _policies.Set(group?.Name, policy.Path, policy.Usage, TimeSpan.FromMilliseconds(policy.Span));
-                _logger.Information($"Policy has been updated [policy id: {policy.Id}]");
-            }
-            else if (message.Request.Type == "update_policies")
-            {
-                var policy = JsonSerializer.Deserialize<PolicyMessage>(message.Data!.ToString()!, _options);
-                var group = _groups.Get(policy.GroupId.ToString());
-                if (policy == null)
-                {
-                    _logger.Debug($"Policy data failed");
-                    return;
-                }
-                _logger.Debug($"Policy data [policy id: {policy.Id}][path: {policy.Path}][group id: {policy.GroupId}][group name: {group?.Name}]");
-                _policies.Set(group?.Name ?? string.Empty, policy.Path, policy.Usage, TimeSpan.FromMilliseconds(policy.Span));
-                _logger.Information($"Policy has been updated [policy id: {policy.Id}]");
-            }
-            else
-            {
-                _logger.Warning($"Found unknown request type when acknowledging [type: {message.Request.Type}]");
-            }
-
-            if (hermesRequestData != null)
-            {
-                _logger.Debug($"Callback was found for request [request id: {message.Request.RequestId}][type: {message.Request.Type}]");
-                hermesRequestData.Callback?.Invoke(hermesRequestData.Data);
+                return;
             }
+            _manager.Fulfill(message.Request.Type, message.Request.RequestId, json, message.Request.Data);
         }
     }
 

Hermes/Socket/Requests/CreatePolicyAck.cs

@@ -0,0 +1,52 @@
+using System.Text.Json;
+using HermesSocketServer.Models;
+using Serilog;
+using TwitchChatTTS.Chat.Commands.Limits;
+using TwitchChatTTS.Chat.Groups;
+
+namespace TwitchChatTTS.Hermes.Socket.Requests
+{
+    public class CreatePolicyAck : IRequestAck
+    {
+        public string Name => "create_policy";
+        private readonly IChatterGroupManager _groups;
+        private readonly IUsagePolicy<long> _policies;
+        private readonly JsonSerializerOptions _options;
+        private readonly ILogger _logger;
+
+        public CreatePolicyAck(IChatterGroupManager groups, IUsagePolicy<long> policies, JsonSerializerOptions options, ILogger logger)
+        {
+            _groups = groups;
+            _policies = policies;
+            _options = options;
+            _logger = logger;
+        }
+
+        public void Acknowledge(string requestId, string json, IDictionary<string, object>? requestData)
+        {
+            if (requestData == null)
+            {
+                _logger.Warning("Request data is null.");
+                return;
+            }
+
+            var policy = JsonSerializer.Deserialize<PolicyMessage>(json, _options);
+            if (policy == null)
+            {
+                _logger.Warning($"Policy data failed: null");
+                return;
+            }
+
+            var group = _groups.Get(policy.GroupId.ToString());
+            if (group == null)
+            {
+                _logger.Warning($"Policy data failed: group id not found [group id: {policy.GroupId}][policy id: {policy.Id}]");
+                return;
+            }
+
+            _logger.Debug($"Policy data [policy id: {policy.Id}][path: {policy.Path}][group id: {policy.GroupId}][group name: {group.Name}]");
+            _policies.Set(group.Name, policy.Path, policy.Usage, TimeSpan.FromMilliseconds(policy.Span));
+            _logger.Information($"Policy has been updated [policy id: {policy.Id}]");
+        }
+    }
+}

Hermes/Socket/Requests/CreateTTSUserAck.cs

@@ -0,0 +1,58 @@
+using Serilog;
+
+namespace TwitchChatTTS.Hermes.Socket.Requests
+{
+    public class CreateTTSUserAck : IRequestAck
+    {
+        public string Name => "create_tts_user";
+        private readonly User _user;
+        private readonly ILogger _logger;
+
+        public CreateTTSUserAck(User user, ILogger logger)
+        {
+            _user = user;
+            _logger = logger;
+        }
+
+        public void Acknowledge(string requestId, string json, IDictionary<string, object>? requestData)
+        {
+            if (requestData == null)
+            {
+                _logger.Warning("Request data is null.");
+                return;
+            }
+            
+            if (!long.TryParse(requestData["chatter"].ToString(), out long chatterId))
+            {
+                _logger.Warning($"Failed to parse chatter id [chatter id: {requestData["chatter"]}]");
+                return;
+            }
+            if (chatterId <= 0)
+            {
+                _logger.Warning($"Chatter Id is invalid [chatter id: {chatterId}]");
+                return;
+            }
+
+            var userId = requestData["user"].ToString();
+            var voiceId = requestData["voice"].ToString();
+            if (string.IsNullOrEmpty(userId))
+            {
+                _logger.Warning("User Id is invalid.");
+                return;
+            }
+            if (string.IsNullOrEmpty(voiceId))
+            {
+                _logger.Warning("Voice Id is invalid.");
+                return;
+            }
+            if (!_user.VoicesAvailable.TryGetValue(voiceId, out var voiceName))
+            {
+                _logger.Warning($"Voice Id does not exist [voice id: {voiceId}]");
+                return;
+            }
+
+            _user.VoicesSelected.Add(chatterId, voiceId);
+            _logger.Information($"Created a new TTS user [user id: {userId}][voice id: {voiceId}][voice name: {voiceName}].");
+        }
+    }
+}

Hermes/Socket/Requests/CreateTTSVoiceAck.cs

@@ -0,0 +1,47 @@
+using Serilog;
+
+namespace TwitchChatTTS.Hermes.Socket.Requests
+{
+    public class CreateTTSVoiceAck : IRequestAck
+    {
+        public string Name => "create_tts_voice";
+        private readonly User _user;
+        private readonly ILogger _logger;
+
+        public CreateTTSVoiceAck(User user, ILogger logger)
+        {
+            _user = user;
+            _logger = logger;
+        }
+
+        public void Acknowledge(string requestId, string json, IDictionary<string, object>? requestData)
+        {
+            if (requestData == null)
+            {
+                _logger.Warning("Request data is null.");
+                return;
+            }
+            
+            var voice = requestData["voice"].ToString()!;
+            var voiceId = json;
+            if (string.IsNullOrEmpty(voice))
+            {
+                _logger.Warning("Voice name is invalid.");
+                return;
+            }
+            if (string.IsNullOrEmpty(voiceId))
+            {
+                _logger.Warning("Voice Id is invalid.");
+                return;
+            }
+            if (_user.VoicesAvailable.TryGetValue(voiceId, out var voiceName))
+            {
+                _logger.Warning($"Voice Id already exists [voice id: {voiceId}][voice name: {voiceName}]");
+                return;
+            }
+
+            _user.VoicesAvailable.Add(voiceId, voice);
+            _logger.Information($"Created a new tts voice [voice: {voice}][id: {voiceId}]");
+        }
+    }
+}

Hermes/Socket/Requests/DeletePolicyAck.cs

@@ -0,0 +1,43 @@
+using Serilog;
+using TwitchChatTTS.Chat.Commands.Limits;
+using TwitchChatTTS.Chat.Groups;
+
+namespace TwitchChatTTS.Hermes.Socket.Requests
+{
+    public class DeletePolicyAck : IRequestAck
+    {
+        public string Name => "delete_policy";
+        private readonly IChatterGroupManager _groups;
+        private readonly IUsagePolicy<long> _policies;
+        private readonly ILogger _logger;
+
+        public DeletePolicyAck(IChatterGroupManager groups, IUsagePolicy<long> policies, ILogger logger)
+        {
+            _groups = groups;
+            _policies = policies;
+            _logger = logger;
+        }
+
+        public void Acknowledge(string requestId, string json, IDictionary<string, object>? requestData)
+        {
+            var data = json.Split('/');
+            if (data.Length != 2)
+            {
+                _logger.Error("Deleting a policy failed: data received is invalid.");
+                return;
+            }
+
+            var groupId = data[0];
+            var path = data[1];
+            var group = _groups.Get(groupId);
+            if (group == null)
+            {
+                _logger.Warning($"Deleting a policy failed: group id does not exist [group id: {groupId}][path: {path}]");
+                return;
+            }
+
+            _policies.Remove(group.Name, path);
+            _logger.Information($"Policy has been deleted [group id: {groupId}][path: {path}]");
+        }
+    }
+}

Hermes/Socket/Requests/DeleteTTSVoiceAck.cs

@@ -0,0 +1,41 @@
+using Serilog;
+
+namespace TwitchChatTTS.Hermes.Socket.Requests
+{
+    public class DeleteTTSVoiceAck : IRequestAck
+    {
+        public string Name => "delete_tts_voice";
+        private readonly User _user;
+        private readonly ILogger _logger;
+
+        public DeleteTTSVoiceAck(User user, ILogger logger)
+        {
+            _user = user;
+            _logger = logger;
+        }
+
+        public void Acknowledge(string requestId, string json, IDictionary<string, object>? requestData)
+        {
+            if (requestData == null)
+            {
+                _logger.Warning("Request data is null.");
+                return;
+            }
+            
+            var voice = requestData["voice"].ToString();
+            if (string.IsNullOrEmpty(voice))
+            {
+                _logger.Warning($"Voice Id is invalid [voice id: {voice}]");
+                return;
+            }
+            if (!_user.VoicesAvailable.TryGetValue(voice, out string? voiceName))
+            {
+                _logger.Warning($"Voice Id does not exist [voice id: {voice}]");
+                return;
+            }
+
+            _user.VoicesAvailable.Remove(voice);
+            _logger.Information($"Deleted a voice [voice id: {voice}][voice name: {voiceName}]");
+        }
+    }
+}

Hermes/Socket/Requests/GetChatterIdsAck.cs

@@ -0,0 +1,38 @@
+using System.Text.Json;
+using Serilog;
+
+namespace TwitchChatTTS.Hermes.Socket.Requests
+{
+    public class GetChatterIdsAck : IRequestAck
+    {
+        public string Name => "get_chatter_ids";
+        private readonly User _user;
+        private readonly JsonSerializerOptions _options;
+        private readonly ILogger _logger;
+
+        public GetChatterIdsAck(User user, JsonSerializerOptions options, ILogger logger)
+        {
+            _user = user;
+            _options = options;
+            _logger = logger;
+        }
+
+        public void Acknowledge(string requestId, string json, IDictionary<string, object>? requestData)
+        {
+            var chatters = JsonSerializer.Deserialize<IEnumerable<long>>(json, _options);
+            if (chatters == null)
+            {
+                _logger.Warning("Chatters is null.");
+                return;
+            }
+            if (!chatters.Any())
+            {
+                _logger.Warning("Chatters is empty.");
+                return;
+            }
+
+            _user.Chatters = [.. chatters];
+            _logger.Information($"Fetched chatters [count: {chatters.Count()}]");
+        }
+    }
+}

Hermes/Socket/Requests/GetConnectionsAck.cs

@@ -0,0 +1,52 @@
+using System.Text.Json;
+using HermesSocketLibrary.Socket.Data;
+using Serilog;
+
+namespace TwitchChatTTS.Hermes.Socket.Requests
+{
+    public class GetConnectionsAck : IRequestAck
+    {
+        public string Name => "get_connections";
+        private readonly TwitchApiClient _twitch;
+        private readonly NightbotApiClient _nightbot;
+        private readonly User _user;
+        private readonly JsonSerializerOptions _options;
+        private readonly ILogger _logger;
+
+        public GetConnectionsAck
+        (
+            TwitchApiClient twitch,
+            NightbotApiClient nightbot,
+            User user,
+            JsonSerializerOptions options,
+            ILogger logger
+        )
+        {
+            _twitch = twitch;
+            _nightbot = nightbot;
+            _user = user;
+            _options = options;
+            _logger = logger;
+        }
+
+        public void Acknowledge(string requestId, string json, IDictionary<string, object>? requestData)
+        {
+            var connections = JsonSerializer.Deserialize<IEnumerable<Connection>>(json, _options);
+            if (connections == null)
+            {
+                _logger.Error("Null value was given when attempting to fetch connections.");
+                return;
+            }
+
+            _user.TwitchConnection = connections.FirstOrDefault(c => c.Type == "twitch" && c.Default);
+            _user.NightbotConnection = connections.FirstOrDefault(c => c.Type == "nightbot" && c.Default);
+
+            _logger.Information($"Fetched connections from TTS account [count: {connections.Count()}][twitch: {_user.TwitchConnection != null}][nightbot: {_user.NightbotConnection != null}]");
+
+            if (_user.TwitchConnection != null)
+                _twitch.Initialize(_user.TwitchConnection.ClientId, _user.TwitchConnection.AccessToken);
+            if (_user.NightbotConnection != null)
+                _nightbot.Initialize(_user.NightbotConnection.ClientId, _user.NightbotConnection.AccessToken);
+        }
+    }
+}

Hermes/Socket/Requests/GetDefaultTTSVoiceAck.cs

@@ -0,0 +1,34 @@
+using System.Text.Json;
+using Serilog;
+
+namespace TwitchChatTTS.Hermes.Socket.Requests
+{
+    public class GetDefaultTTSVoiceAck : IRequestAck
+    {
+        public string Name => "get_default_tts_voice";
+        private readonly User _user;
+        private readonly JsonSerializerOptions _options;
+        private readonly ILogger _logger;
+
+        public GetDefaultTTSVoiceAck(User user, JsonSerializerOptions options, ILogger logger)
+        {
+            _user = user;
+            _options = options;
+            _logger = logger;
+        }
+
+        public void Acknowledge(string requestId, string json, IDictionary<string, object>? requestData)
+        {
+            string? defaultVoice = json;
+            if (defaultVoice != null)
+            {
+                _user.DefaultTTSVoice = defaultVoice;
+                _logger.Information($"Default TTS voice was changed [voice: {defaultVoice}]");
+            }
+            else
+            {
+                _logger.Error($"Failed to load default TTS voice.");
+            }
+        }
+    }
+}

Hermes/Socket/Requests/GetEmotesAck.cs

@@ -0,0 +1,48 @@
+using System.Text.Json;
+using HermesSocketLibrary.Requests.Messages;
+using Serilog;
+using TwitchChatTTS.Chat.Emotes;
+
+namespace TwitchChatTTS.Hermes.Socket.Requests
+{
+    public class GetEmotesAck : IRequestAck
+    {
+        public string Name => "get_emotes";
+        private readonly IEmoteDatabase _emotes;
+        private readonly JsonSerializerOptions _options;
+        private readonly ILogger _logger;
+
+        public GetEmotesAck(IEmoteDatabase emotes, JsonSerializerOptions options, ILogger logger)
+        {
+            _emotes = emotes;
+            _options = options;
+            _logger = logger;
+        }
+
+        public void Acknowledge(string requestId, string json, IDictionary<string, object>? requestData)
+        {
+            var data = JsonSerializer.Deserialize<IEnumerable<EmoteInfo>>(json, _options);
+            if (data == null)
+            {
+                _logger.Warning("Emotes is null.");
+                return;
+            }
+
+            var count = 0;
+            var duplicateNames = 0;
+            foreach (var emote in data)
+            {
+                if (_emotes.Get(emote.Name) == null)
+                {
+                    _emotes.Add(emote.Name, emote.Id);
+                    count++;
+                }
+                else
+                    duplicateNames++;
+            }
+            _logger.Information($"Fetched emotes of various sources [count: {count}]");
+            if (duplicateNames > 0)
+                _logger.Warning($"Found {duplicateNames} emotes with duplicate names.");
+        }
+    }
+}

Hermes/Socket/Requests/GetEnabledTTSVoices.cs

@@ -0,0 +1,41 @@
+using System.Text.Json;
+using Serilog;
+
+namespace TwitchChatTTS.Hermes.Socket.Requests
+{
+    public class GetEnabledTTSVoicesAck : IRequestAck
+    {
+        public string Name => "get_enabled_tts_voices";
+        private readonly User _user;
+        private readonly JsonSerializerOptions _options;
+        private readonly ILogger _logger;
+
+        public GetEnabledTTSVoicesAck(User user, JsonSerializerOptions options, ILogger logger)
+        {
+            _user = user;
+            _options = options;
+            _logger = logger;
+        }
+
+        public void Acknowledge(string requestId, string json, IDictionary<string, object>? requestData)
+        {
+            var enabledTTSVoices = JsonSerializer.Deserialize<IEnumerable<string>>(json, _options);
+            if (enabledTTSVoices == null)
+            {
+                _logger.Warning("Failed to load enabled tts voices.");
+                return;
+            }
+
+            if (_user.VoicesEnabled == null)
+                _user.VoicesEnabled = enabledTTSVoices.ToHashSet();
+            else
+            {
+                _user.VoicesEnabled.Clear();
+                foreach (var voice in enabledTTSVoices)
+                    _user.VoicesEnabled.Add(voice);
+            }
+
+            _logger.Information($"TTS voices [count: {_user.VoicesEnabled.Count}] have been enabled.");
+        }
+    }
+}

Hermes/Socket/Requests/GetPermissionsAck.cs

@@ -0,0 +1,67 @@
+using System.Text.Json;
+using HermesSocketLibrary.Requests.Messages;
+using Serilog;
+using TwitchChatTTS.Chat.Groups;
+using TwitchChatTTS.Chat.Groups.Permissions;
+
+namespace TwitchChatTTS.Hermes.Socket.Requests
+{
+    public class GetPermissionsAck : IRequestAck
+    {
+        public string Name => "get_permissions";
+        private readonly IGroupPermissionManager _permissions;
+        private readonly IChatterGroupManager _groups;
+        private readonly JsonSerializerOptions _options;
+        private readonly ILogger _logger;
+
+        public GetPermissionsAck(
+            IGroupPermissionManager permissions,
+            IChatterGroupManager groups,
+            JsonSerializerOptions options,
+            ILogger logger)
+        {
+            _permissions = permissions;
+            _groups = groups;
+            _options = options;
+            _logger = logger;
+        }
+
+        public void Acknowledge(string requestId, string json, IDictionary<string, object>? requestData)
+        {
+            var groupInfo = JsonSerializer.Deserialize<GroupInfo>(json, _options);
+            if (groupInfo == null)
+            {
+                _logger.Error("Failed to load groups & permissions.");
+                return;
+            }
+
+            _permissions.Clear();
+            _groups.Clear();
+
+            var groupsById = groupInfo.Groups.ToDictionary(g => g.Id, g => g);
+            foreach (var group in groupInfo.Groups)
+                _groups.Add(group);
+
+            foreach (var permission in groupInfo.GroupPermissions)
+            {
+                _logger.Debug($"Adding group permission [permission id: {permission.Id}][group id: {permission.GroupId}][path: {permission.Path}][allow: {permission.Allow?.ToString() ?? "null"}]");
+                if (!groupsById.TryGetValue(permission.GroupId, out var group))
+                {
+                    _logger.Warning($"Failed to find group by id [permission id: {permission.Id}][group id: {permission.GroupId}][path: {permission.Path}]");
+                    continue;
+                }
+
+                var path = $"{group.Name}.{permission.Path}";
+                _permissions.Set(path, permission.Allow);
+                _logger.Debug($"Added group permission [id: {permission.Id}][group id: {permission.GroupId}][path: {permission.Path}]");
+            }
+
+            _logger.Information($"Groups [count: {groupInfo.Groups.Count()}] & Permissions [count: {groupInfo.GroupPermissions.Count()}] have been loaded.");
+
+            foreach (var chatter in groupInfo.GroupChatters)
+                if (groupsById.TryGetValue(chatter.GroupId, out var group))
+                    _groups.Add(chatter.ChatterId, group.Name);
+            _logger.Information($"Users in each group [count: {groupInfo.GroupChatters.Count()}] have been loaded.");
+        }
+    }
+}

Hermes/Socket/Requests/GetPoliciesAck.cs

@@ -0,0 +1,53 @@
+using System.Text.Json;
+using HermesSocketServer.Models;
+using Serilog;
+using TwitchChatTTS.Chat.Commands.Limits;
+using TwitchChatTTS.Chat.Groups;
+
+namespace TwitchChatTTS.Hermes.Socket.Requests
+{
+    public class GetPoliciesAck : IRequestAck
+    {
+        public string Name => "get_policies";
+        private readonly IChatterGroupManager _groups;
+        private readonly IUsagePolicy<long> _policies;
+        private readonly JsonSerializerOptions _options;
+        private readonly ILogger _logger;
+
+        public GetPoliciesAck(
+            IChatterGroupManager groups,
+            IUsagePolicy<long> policies,
+            JsonSerializerOptions options,
+            ILogger logger)
+        {
+            _groups = groups;
+            _policies = policies;
+            _options = options;
+            _logger = logger;
+        }
+
+        public void Acknowledge(string requestId, string json, IDictionary<string, object>? requestData)
+        {
+            var policies = JsonSerializer.Deserialize<IEnumerable<PolicyMessage>>(json, _options);
+            if (policies == null || !policies.Any())
+            {
+                _logger.Information($"No policies have been found. Policies have been set to default.");
+                _policies.Set("everyone", "tts", 25, TimeSpan.FromSeconds(15));
+                return;
+            }
+
+            foreach (var policy in policies)
+            {
+                var group = _groups.Get(policy.GroupId.ToString());
+                if (group == null)
+                {
+                    _logger.Debug($"Policy data failed: group id not found [group id: {policy.GroupId}][policy id: {policy.Id}]");
+                    continue;
+                }
+                _logger.Debug($"Policy data loaded [policy id: {policy.Id}][path: {policy.Path}][group id: {policy.GroupId}][group name: {group.Name}]");
+                _policies.Set(group.Name, policy.Path, policy.Usage, TimeSpan.FromMilliseconds(policy.Span));
+            }
+            _logger.Information($"Policies have been loaded [count: {policies.Count()}]");
+        }
+    }
+}

Hermes/Socket/Requests/GetRedeemableActionsAck.cs

@@ -0,0 +1,54 @@
+using System.Text.Json;
+using HermesSocketLibrary.Requests.Messages;
+using Serilog;
+using TwitchChatTTS.Bus;
+using TwitchChatTTS.Bus.Data;
+
+namespace TwitchChatTTS.Hermes.Socket.Requests
+{
+    public class GetRedeemableActionsAck : IRequestAck
+    {
+        public string Name => "get_redeemable_actions";
+        private readonly ServiceBusCentral _bus;
+        private readonly JsonSerializerOptions _options;
+        private readonly ILogger _logger;
+
+        public GetRedeemableActionsAck(
+            ServiceBusCentral bus,
+            JsonSerializerOptions options,
+            ILogger logger)
+        {
+            _bus = bus;
+            _options = options;
+            _logger = logger;
+        }
+
+        public void Acknowledge(string requestId, string json, IDictionary<string, object>? requestData)
+        {
+            if (requestData == null)
+            {
+                _logger.Warning("Request data is null.");
+                return;
+            }
+
+            if (requestData["redemptions"] is not IEnumerable<Redemption> redemptions)
+            {
+                _logger.Warning("Failed to read the redemptions while updating redemption actions.");
+                return;
+            }
+
+            IEnumerable<RedeemableAction>? actions = JsonSerializer.Deserialize<IEnumerable<RedeemableAction>>(json, _options);
+            if (actions == null)
+            {
+                _logger.Warning("Failed to read the redeemable actions for redemptions.");
+                return;
+            }
+
+            _logger.Information($"Redeemable actions loaded [count: {actions.Count()}]");
+            _bus.Send(this, "redemptions_initiation", new RedemptionInitiation() {
+                Redemptions = redemptions,
+                Actions = actions.ToDictionary(a => a.Name, a => a)
+            });
+        }
+    }
+}

Hermes/Socket/Requests/GetRedemptionsAck.cs

@@ -0,0 +1,52 @@
+using System.Text.Json;
+using HermesSocketLibrary.Requests.Callbacks;
+using HermesSocketLibrary.Requests.Messages;
+using Serilog;
+using TwitchChatTTS.Hermes.Socket.Handlers;
+
+namespace TwitchChatTTS.Hermes.Socket.Requests
+{
+    public class GetRedemptionsAck : IRequestAck
+    {
+        public string Name => "get_redemptions";
+        private readonly ICallbackManager<HermesRequestData> _callbacks;
+        private readonly JsonSerializerOptions _options;
+        private readonly ILogger _logger;
+
+        public GetRedemptionsAck(
+            ICallbackManager<HermesRequestData> callbacks,
+            JsonSerializerOptions options,
+            ILogger logger)
+        {
+            _callbacks = callbacks;
+            _options = options;
+            _logger = logger;
+        }
+
+        public void Acknowledge(string requestId, string json, IDictionary<string, object>? requestData)
+        {
+            HermesRequestData? hermesRequestData = null;
+            if (!string.IsNullOrEmpty(requestId))
+            {
+                hermesRequestData = _callbacks.Take(requestId);
+                if (hermesRequestData == null)
+                    _logger.Warning($"Could not find callback for request [request id: {requestId}][type: {GetType().Name}]");
+                else if (hermesRequestData.Data == null)
+                    hermesRequestData.Data = new Dictionary<string, object>();
+            }
+
+            IEnumerable<Redemption>? redemptions = JsonSerializer.Deserialize<IEnumerable<Redemption>>(json, _options);
+            if (redemptions != null)
+            {
+                _logger.Information($"Redemptions loaded [count: {redemptions.Count()}]");
+                if (hermesRequestData != null)
+                {
+                    hermesRequestData.Data!.Add("redemptions", redemptions);
+
+                    _logger.Debug($"Callback was found for request [request id: {requestId}][type: {GetType().Name}]");
+                    hermesRequestData.Callback?.Invoke(hermesRequestData.Data);
+                }
+            }
+        }
+    }
+}

Hermes/Socket/Requests/GetTTSUsersAck.cs

@@ -0,0 +1,35 @@
+using System.Collections.Concurrent;
+using System.Text.Json;
+using Serilog;
+
+namespace TwitchChatTTS.Hermes.Socket.Requests
+{
+    public class GetTTSUsersAck : IRequestAck
+    {
+        public string Name => "get_tts_users";
+        private readonly User _user;
+        private readonly JsonSerializerOptions _options;
+        private readonly ILogger _logger;
+
+        public GetTTSUsersAck(User user, JsonSerializerOptions options, ILogger logger)
+        {
+            _user = user;
+            _options = options;
+            _logger = logger;
+        }
+
+        public void Acknowledge(string requestId, string json, IDictionary<string, object>? requestData)
+        {
+            var users = JsonSerializer.Deserialize<IDictionary<long, string>>(json, _options);
+            if (users == null)
+                return;
+
+            var temp = new ConcurrentDictionary<long, string>();
+            foreach (var entry in users)
+                temp.TryAdd(entry.Key, entry.Value);
+            
+            _user.VoicesSelected = temp;
+            _logger.Information($"Updated chatters' selected voice [count: {temp.Count()}]");
+        }
+    }
+}

Hermes/Socket/Requests/GetTTSVoicesAck.cs

@@ -0,0 +1,40 @@
+using System.Collections.Concurrent;
+using System.Text.Json;
+using HermesSocketLibrary.Requests.Messages;
+using Serilog;
+
+namespace TwitchChatTTS.Hermes.Socket.Requests
+{
+    public class GetTTSVoicesAck : IRequestAck
+    {
+        public string Name => "get_tts_voices";
+        private readonly User _user;
+        private readonly JsonSerializerOptions _options;
+        private readonly ILogger _logger;
+
+        public GetTTSVoicesAck(User user, JsonSerializerOptions options, ILogger logger)
+        {
+            _user = user;
+            _options = options;
+            _logger = logger;
+        }
+
+        public void Acknowledge(string requestId, string json, IDictionary<string, object>? requestData)
+        {
+            var voices = JsonSerializer.Deserialize<IEnumerable<VoiceDetails>>(json, _options);
+            if (voices == null)
+            {
+                _logger.Warning("Voices received is null.");
+                return;
+            }
+            if (!voices.Any())
+            {
+                _logger.Warning("Voices received is empty.");
+                return;
+            }
+
+            _user.VoicesAvailable = new ConcurrentDictionary<string, string>(voices.ToDictionary(e => e.Id, e => e.Name));
+            _logger.Information($"Fetched all available voices for TTS [count: {_user.VoicesAvailable.Count}]");
+        }
+    }
+}

Hermes/Socket/Requests/GetTTSWordFiltersAck.cs

@@ -0,0 +1,46 @@
+using System.Text.Json;
+using System.Text.RegularExpressions;
+using HermesSocketLibrary.Requests.Messages;
+using Serilog;
+
+namespace TwitchChatTTS.Hermes.Socket.Requests
+{
+    public class GetTTSWordFiltersAck : IRequestAck
+    {
+        public string Name => "get_tts_word_filters";
+        private readonly User _user;
+        private readonly JsonSerializerOptions _options;
+        private readonly ILogger _logger;
+
+        public GetTTSWordFiltersAck(User user, JsonSerializerOptions options, ILogger logger)
+        {
+            _user = user;
+            _options = options;
+            _logger = logger;
+        }
+
+        public void Acknowledge(string requestId, string json, IDictionary<string, object>? requestData)
+        {
+            var wordFilters = JsonSerializer.Deserialize<IEnumerable<TTSWordFilter>>(json, _options);
+            if (wordFilters == null)
+            {
+                _logger.Error("Failed to load word filters.");
+                return;
+            }
+
+            var filters = wordFilters.Where(f => f.Search != null && f.Replace != null).ToArray();
+            foreach (var filter in filters)
+            {
+                try
+                {
+                    var re = new Regex(filter.Search!, RegexOptions.Compiled);
+                    re.Match(string.Empty);
+                    filter.Regex = re;
+                }
+                catch (Exception) { }
+            }
+            _user.RegexFilters = filters;
+            _logger.Information($"TTS word filters [count: {_user.RegexFilters.Count()}] have been refreshed.");
+        }
+    }
+}

Hermes/Socket/Requests/IRequestAck.cs

@@ -0,0 +1,8 @@
+namespace TwitchChatTTS.Hermes.Socket.Requests
+{
+    public interface IRequestAck
+    {
+        string Name { get; }
+        void Acknowledge(string requestId, string json, IDictionary<string, object>? requestData);
+    }
+}

Hermes/Socket/Requests/RequestAckManager.cs

@@ -0,0 +1,38 @@
+using Serilog;
+
+namespace TwitchChatTTS.Hermes.Socket.Requests
+{
+    public class RequestAckManager
+    {
+        private readonly IDictionary<string, IRequestAck> _acknowledgements;
+        private readonly ILogger _logger;
+
+        public RequestAckManager(IEnumerable<IRequestAck> acks, ILogger logger)
+        {
+            _acknowledgements = acks.ToDictionary(a => a.Name, a => a);
+            _logger = logger;
+        }
+
+        public void Fulfill(string type, string requestId, string data, IDictionary<string, object>? requestData)
+        {
+            if (data == null)
+                return;
+            if (!_acknowledgements.TryGetValue(type, out var ack))
+            {
+                _logger.Warning($"Found unknown request type when acknowledging [type: {type}]");
+                return;
+            }
+
+            _logger.Debug($"Request acknowledgement found [type: {type}][data: {data}]");
+            try
+            {
+                ack.Acknowledge(requestId, data, requestData);
+                _logger.Debug($"Request acknowledged without error [type: {type}][data: {data}]");
+            }
+            catch (Exception ex)
+            {
+                _logger.Error(ex, "Failed to fulfill a request ackowledgement.");
+            }
+        }
+    }
+}

Hermes/Socket/Requests/UpdateDefaultTTSVoiceAck.cs

@@ -0,0 +1,37 @@
+using Serilog;
+
+namespace TwitchChatTTS.Hermes.Socket.Requests
+{
+    public class UpdateDefaultTTSVoiceAck : IRequestAck
+    {
+        public string Name => "update_default_tts_voice";
+        private readonly User _user;
+        private readonly ILogger _logger;
+
+        public UpdateDefaultTTSVoiceAck(User user, ILogger logger)
+        {
+            _user = user;
+            _logger = logger;
+        }
+
+        public void Acknowledge(string requestId, string json, IDictionary<string, object>? requestData)
+        {
+            if (requestData == null)
+            {
+                _logger.Warning("Request data is null.");
+                return;
+            }
+
+            if (requestData.TryGetValue("voice", out object? voice) == true && voice is string v)
+            {
+                if (_user.VoicesEnabled.Contains(v))
+                {
+                    _user.DefaultTTSVoice = v;
+                    _logger.Information($"Default TTS voice was changed to '{v}'.");
+                    return;
+                }
+            }
+            _logger.Warning("Failed to update default TTS voice via request.");
+        }
+    }
+}

Hermes/Socket/Requests/UpdatePolicyAck.cs

@@ -0,0 +1,45 @@
+using System.Text.Json;
+using HermesSocketServer.Models;
+using Serilog;
+using TwitchChatTTS.Chat.Commands.Limits;
+using TwitchChatTTS.Chat.Groups;
+
+namespace TwitchChatTTS.Hermes.Socket.Requests
+{
+    public class UpdatePolicyAck : IRequestAck
+    {
+        public string Name => "update_policy";
+        private readonly IChatterGroupManager _groups;
+        private readonly IUsagePolicy<long> _policies;
+        private readonly JsonSerializerOptions _options;
+        private readonly ILogger _logger;
+
+        public UpdatePolicyAck(IChatterGroupManager groups, IUsagePolicy<long> policies, JsonSerializerOptions options, ILogger logger)
+        {
+            _groups = groups;
+            _policies = policies;
+            _options = options;
+            _logger = logger;
+        }
+
+        public void Acknowledge(string requestId, string json, IDictionary<string, object>? requestData)
+        {
+            var policy = JsonSerializer.Deserialize<PolicyMessage>(json, _options);
+            if (policy == null)
+            {
+                _logger.Warning($"Policy data failed: null");
+                return;
+            }
+            var group = _groups.Get(policy.GroupId.ToString());
+            if (group == null)
+            {
+                _logger.Warning($"Policy data failed: group id not found [group id: {policy.GroupId}][policy id: {policy.Id}]");
+                return;
+            }
+
+            _logger.Debug($"Policy data loaded [policy id: {policy.Id}][path: {policy.Path}][group id: {policy.GroupId}][group name: {group.Name}]");
+            _policies.Set(group.Name, policy.Path, policy.Usage, TimeSpan.FromMilliseconds(policy.Span));
+            _logger.Information($"Policy has been updated [policy id: {policy.Id}]");
+        }
+    }
+}

Hermes/Socket/Requests/UpdateTTSUserAck.cs

@@ -0,0 +1,58 @@
+using Serilog;
+
+namespace TwitchChatTTS.Hermes.Socket.Requests
+{
+    public class UpdateTTSUserAck : IRequestAck
+    {
+        public string Name => "update_tts_user";
+        private readonly User _user;
+        private readonly ILogger _logger;
+
+        public UpdateTTSUserAck(User user, ILogger logger)
+        {
+            _user = user;
+            _logger = logger;
+        }
+
+        public void Acknowledge(string requestId, string json, IDictionary<string, object>? requestData)
+        {
+            if (requestData == null)
+            {
+                _logger.Warning("Request data is null.");
+                return;
+            }
+
+            if (!long.TryParse(requestData["chatter"].ToString(), out long chatterId))
+            {
+                _logger.Warning($"Failed to parse chatter id [chatter id: {requestData["chatter"]}]");
+                return;
+            }
+            if (chatterId <= 0)
+            {
+                _logger.Warning("Chatter Id is invalid.");
+                return;
+            }
+
+            var userId = requestData["user"].ToString();
+            var voiceId = requestData["voice"].ToString();
+            if (string.IsNullOrEmpty(userId))
+            {
+                _logger.Warning("User Id is invalid.");
+                return;
+            }
+            if (string.IsNullOrEmpty(voiceId))
+            {
+                _logger.Warning("Voice Id is invalid.");
+                return;
+            }
+            if (!_user.VoicesAvailable.TryGetValue(voiceId, out var voiceName))
+            {
+                _logger.Warning("Voice Id does not exist.");
+                return;
+            }
+
+            _user.VoicesSelected[chatterId] = voiceId;
+            _logger.Information($"Updated a TTS user's voice [user id: {userId}][voice: {voiceId}][voice name: {voiceName}]");
+        }
+    }
+}

Hermes/Socket/Requests/UpdateTTSVoiceAck.cs

@@ -0,0 +1,47 @@
+using Serilog;
+
+namespace TwitchChatTTS.Hermes.Socket.Requests
+{
+    public class UpdateTTSVoiceAck : IRequestAck
+    {
+        public string Name => "update_tts_voice";
+        private readonly User _user;
+        private readonly ILogger _logger;
+
+        public UpdateTTSVoiceAck(User user, ILogger logger)
+        {
+            _user = user;
+            _logger = logger;
+        }
+
+        public void Acknowledge(string requestId, string json, IDictionary<string, object>? requestData)
+        {
+            if (requestData == null)
+            {
+                _logger.Warning("Request data is null.");
+                return;
+            }
+            
+            var voice = requestData["voice"].ToString();
+            var voiceId = requestData["idd"].ToString();
+            if (string.IsNullOrEmpty(voice))
+            {
+                _logger.Warning("Voice name is invalid.");
+                return;
+            }
+            if (string.IsNullOrEmpty(voiceId))
+            {
+                _logger.Warning("Voice Id is invalid.");
+                return;
+            }
+            if (_user.VoicesAvailable.ContainsKey(voiceId))
+            {
+                _logger.Warning($"Voice Id already exists [voice id: {voiceId}]");
+                return;
+            }
+
+            _user.VoicesAvailable[voiceId] = voice;
+            _logger.Information($"Created a new tts voice [voice id: {voiceId}][voice name: {voice}]");
+        }
+    }
+}

Hermes/Socket/Requests/UpdateTTSVoiceStateAck.cs

@@ -0,0 +1,58 @@
+using Serilog;
+
+namespace TwitchChatTTS.Hermes.Socket.Requests
+{
+    public class UpdateTTSVoiceStateAck : IRequestAck
+    {
+        public string Name => "update_tts_voice_state";
+        private readonly User _user;
+        private readonly ILogger _logger;
+
+        public UpdateTTSVoiceStateAck(User user, ILogger logger)
+        {
+            _user = user;
+            _logger = logger;
+        }
+
+        public void Acknowledge(string requestId, string json, IDictionary<string, object>? requestData)
+        {
+            if (requestData == null)
+            {
+                _logger.Warning("Request data is null.");
+                return;
+            }
+            
+            if (!long.TryParse(requestData["chatter"].ToString(), out long chatterId))
+            {
+                _logger.Warning($"Failed to parse chatter id [chatter id: {requestData["chatter"]}]");
+                return;
+            }
+            if (chatterId <= 0)
+            {
+                _logger.Warning("Chatter Id is invalid.");
+                return;
+            }
+
+            var userId = requestData["user"].ToString();
+            var voiceId = requestData["voice"].ToString();
+            if (string.IsNullOrEmpty(userId))
+            {
+                _logger.Warning("User Id is invalid.");
+                return;
+            }
+            if (string.IsNullOrEmpty(voiceId))
+            {
+                _logger.Warning("Voice Id is invalid.");
+                return;
+            }
+            if (!_user.VoicesAvailable.TryGetValue(voiceId, out var voiceName))
+            {
+                _logger.Warning("Voice Id does not exist.");
+                return;
+            }
+
+            _user.VoicesSelected[chatterId] = voiceId;
+            _logger.Information($"Updated a TTS user's voice [user id: {userId}][voice: {voiceId}][voice name: {voiceName}]");
+        }
+    }
+}

Startup.cs

@@ -33,6 +33,8 @@ using TwitchChatTTS.Chat.Speech;
 using TwitchChatTTS.Chat.Messaging;
 using TwitchChatTTS.Chat.Observers;
 using TwitchChatTTS.Chat.Commands.Limits;
+using TwitchChatTTS.Hermes.Socket.Requests;
+using TwitchChatTTS.Bus;
 
 // dotnet publish -r linux-x64 -p:PublishSingleFile=true --self-contained true
 // dotnet publish -r win-x64 -p:PublishSingleFile=true --self-contained true
@@ -78,11 +80,34 @@ s.AddSingleton<IChatCommand, OBSCommand>();
 s.AddSingleton<IChatCommand, TTSCommand>();
 s.AddSingleton<IChatCommand, VersionCommand>();
 s.AddSingleton<ICommandBuilder, CommandBuilder>();
-s.AddSingleton<IChatterGroupManager, ChatterGroupManager>();
-s.AddSingleton<IGroupPermissionManager, GroupPermissionManager>();
 s.AddSingleton<ICommandManager, CommandManager>();
 s.AddTransient<ICommandFactory, CommandFactory>();
 
+// Request acks
+s.AddSingleton<RequestAckManager>();
+s.AddTransient<IRequestAck, CreatePolicyAck>();
+s.AddTransient<IRequestAck, CreateTTSUserAck>();
+s.AddTransient<IRequestAck, CreateTTSVoiceAck>();
+s.AddTransient<IRequestAck, DeletePolicyAck>();
+s.AddTransient<IRequestAck, DeleteTTSVoiceAck>();
+s.AddTransient<IRequestAck, GetChatterIdsAck>();
+s.AddTransient<IRequestAck, GetConnectionsAck>();
+s.AddTransient<IRequestAck, GetDefaultTTSVoiceAck>();
+s.AddTransient<IRequestAck, GetEmotesAck>();
+s.AddTransient<IRequestAck, GetEnabledTTSVoicesAck>();
+s.AddTransient<IRequestAck, GetPermissionsAck>();
+s.AddTransient<IRequestAck, GetPoliciesAck>();
+s.AddTransient<IRequestAck, GetRedeemableActionsAck>();
+s.AddTransient<IRequestAck, GetRedemptionsAck>();
+s.AddTransient<IRequestAck, GetTTSUsersAck>();
+s.AddTransient<IRequestAck, GetTTSVoicesAck>();
+s.AddTransient<IRequestAck, GetTTSWordFiltersAck>();
+s.AddTransient<IRequestAck, UpdateDefaultTTSVoiceAck>();
+s.AddTransient<IRequestAck, UpdatePolicyAck>();
+s.AddTransient<IRequestAck, UpdateTTSUserAck>();
+s.AddTransient<IRequestAck, UpdateTTSVoiceAck>();
+s.AddTransient<IRequestAck, UpdateTTSVoiceStateAck>();
+
 s.AddSingleton<TTSPlayer>();
 s.AddSingleton<IRedemptionManager, RedemptionManager>();
 s.AddSingleton<HermesApiClient>();
@@ -91,11 +116,15 @@ s.AddSingleton<TwitchApiClient>();
 s.AddSingleton<SevenApiClient>();
 s.AddSingleton<IEmoteDatabase, EmoteDatabase>();
 
+s.AddSingleton<ServiceBusCentral>();
+
 s.AddSingleton<TTSConsumer>();
 s.AddSingleton<TTSPublisher>();
 
 s.AddSingleton<IChatMessageReader, ChatMessageReader>();
 s.AddSingleton<IUsagePolicy<long>, UsagePolicy<long>>();
+s.AddSingleton<IChatterGroupManager, ChatterGroupManager>();
+s.AddSingleton<IGroupPermissionManager, GroupPermissionManager>();
 
 // OBS websocket
 s.AddKeyedSingleton<IWebSocketHandler, HelloHandler>("obs");

Twitch/Redemptions/RedemptionManager.cs

@@ -5,6 +5,8 @@ using HermesSocketLibrary.Requests.Messages;
 using Microsoft.Extensions.DependencyInjection;
 using org.mariuszgromada.math.mxparser;
 using Serilog;
+using TwitchChatTTS.Bus;
+using TwitchChatTTS.Bus.Data;
 using TwitchChatTTS.Hermes.Socket;
 using TwitchChatTTS.OBS.Socket;
 using TwitchChatTTS.OBS.Socket.Data;
@@ -14,6 +16,7 @@ namespace TwitchChatTTS.Twitch.Redemptions
     public class RedemptionManager : IRedemptionManager
     {
         private readonly IDictionary<string, IList<RedeemableAction>> _store;
+        private readonly ServiceBusCentral _bus;
         private readonly User _user;
         private readonly OBSSocketClient _obs;
         private readonly HermesSocketClient _hermes;
@@ -25,6 +28,7 @@ namespace TwitchChatTTS.Twitch.Redemptions
 
 
         public RedemptionManager(
+            ServiceBusCentral bus,
             User user,
             [FromKeyedServices("obs")] SocketClient<WebSocketMessage> obs,
             [FromKeyedServices("hermes")] SocketClient<WebSocketMessage> hermes,
@@ -33,6 +37,7 @@ namespace TwitchChatTTS.Twitch.Redemptions
             ILogger logger)
         {
             _store = new Dictionary<string, IList<RedeemableAction>>();
+            _bus = bus;
             _user = user;
             _obs = (obs as OBSSocketClient)!;
             _hermes = (hermes as HermesSocketClient)!;
@@ -41,6 +46,12 @@ namespace TwitchChatTTS.Twitch.Redemptions
             _logger = logger;
             _random = new Random();
             _isReady = false;
+
+            var topic = _bus.GetTopic("redemptions_initiation");
+            topic.Subscribe(new ServiceBusObserver(data => {
+                if (data.Value is RedemptionInitiation obj)
+                    Initialize(obj.Redemptions, obj.Actions);
+            }, _logger));
         }
 
         private void Add(string twitchRedemptionId, RedeemableAction action)

TwitchChatTTS.csproj

@@ -19,6 +19,7 @@
     <PackageReference Include="Serilog.Sinks.Console" Version="6.0.0" />
     <PackageReference Include="Serilog.Sinks.Debug" Version="3.0.0" />
     <PackageReference Include="Serilog.Sinks.File" Version="6.0.0" />
+    <PackageReference Include="System.Reactive" Version="6.0.1" />
     <PackageReference Include="System.Text.Json" Version="8.0.4" />
     <PackageReference Include="NAudio.Core" Version="2.2.1" />
     <PackageReference Include="YamlDotNet" Version="16.0.0" />