Added Twitch PubSub support. Heavily ripped off code from GambaSesh though rewrote it to use Websocket.Client and will refactor the JSON parsing once I know exactly what the payload looks like. Presently untested. Also added a new config option to suppress messages, so one can run a local copy of the bot without disturbing anyone and made the Proxy a single config rather than having one for KF and one for Pusher.

2026-05-02 20:42:04 -04:00 · 2024-06-17 19:10:52 +08:00
parent f7278c6d3f
commit ec6c050e23
5 changed files with 209 additions and 37 deletions
--- a/KfChatDotNetKickBot/KickBot.cs
+++ b/KfChatDotNetKickBot/KickBot.cs
@@ -25,6 +25,7 @@ public class KickBot
    // Suppresses the command handler on initial start, so it doesn't pick up things already handled on restart
    private bool _initialStartCooldown = true;
    private readonly CancellationToken _cancellationToken = new();
+    private readonly TwitchWs _twitchWs;
    
    public KickBot()
    {
@@ -45,12 +46,12 @@ public class KickBot
            WsUri = _config.KfWsEndpoint,
            XfSessionToken = _xfSessionToken,
            CookieDomain = _config.KfWsEndpoint.Host,
-            Proxy = _config.KfProxy,
+            Proxy = _config.Proxy,
            ReconnectTimeout = _config.KfReconnectTimeout
        });

        _kickClient = new KickWsClient.KickWsClient(_config.PusherEndpoint.ToString(),
-            _config.PusherProxy, _config.PusherReconnectTimeout);
+            _config.Proxy, _config.PusherReconnectTimeout);

        _kfClient.OnMessages += OnKfChatMessage;
        _kfClient.OnUsersParted += OnUsersParted;
@@ -76,11 +77,34 @@ public class KickBot
        _logger.Debug("Creating ping thread and starting it");
        var pingThread = new Thread(PingThread);
        pingThread.Start();
-        
-        while (true)
+
+        if (_config.BossmanJackTwitchId != null)
        {
-            Console.ReadLine();
+            _logger.Debug("Creating Twitch live stream notification client");
+            _twitchWs = new TwitchWs([_config.BossmanJackTwitchId.Value], _config.Proxy, _cancellationToken);
+            _twitchWs.OnStreamStateUpdated += OnTwitchStreamStateUpdated;
+            _twitchWs.StartWsClient().Wait(_cancellationToken);
        }
+        else
+        {
+            _logger.Debug("Ignoring Twitch client as TwitchChannels is not defined");
+        }
+
+        _logger.Debug("Blocking the main thread");
+        var exitEvent = new ManualResetEvent(false);
+        exitEvent.WaitOne();
+    }
+
+    private void OnTwitchStreamStateUpdated(object sender, int channelId, bool isLive)
+    {
+        _logger.Info($"BossmanJack stream event came in. isLive => {isLive}");
+        if (isLive)
+        {
+            _sendChatMessage("BossmanJack just went live on Twitch! https://www.twitch.tv/thebossmanjack");
+            _sendChatMessage("Ad-free re-stream at https://kick.com/wheelfan courtesy of @Kees H");
+            return;
+        }
+        _sendChatMessage("BossmanJack is no longer live! :lossmanjack:");
    }

    private void OnFailedToJoinRoom(object sender, string message)
@@ -110,7 +134,7 @@ public class KickBot
    private async Task RefreshXfToken()
    {
        var cookie = await KfTokenService.FetchSessionTokenAsync(_config.KfDomain, _config.KfUsername, _config.KfPassword,
-            _config.ChromiumPath, _config.KfProxy);
+            _config.ChromiumPath, _config.Proxy);
        _logger.Debug($"FetchSessionTokenAsync returned {cookie}");
        _xfSessionToken = cookie;
    }
@@ -163,6 +187,12 @@ public class KickBot

    private void _sendChatMessage(string message, bool bypassSeshDetect = false)
    {
+        if (_config.SuppressChatMessages)
+        {
+            _logger.Info("Not sending message as SuppressChatMessages is enabled");
+            _logger.Info($"Message was: {message}");
+            return;
+        }
        if (_gambaSeshPresent && _config.EnableGambaSeshDetect && !bypassSeshDetect)
        {
            _logger.Info($"Not sending message '{message}' as GambaSesh is present");
--- a/KfChatDotNetKickBot/Models.cs
+++ b/KfChatDotNetKickBot/Models.cs
@@ -1,29 +0,0 @@
-namespace KfChatDotNetKickBot;
-
-public class Models
-{
-    public class ConfigModel
-    {
-        public Uri PusherEndpoint { get; set; } =
-            new("wss://ws-us2.pusher.com/app/eb1d5f283081a78b932c?protocol=7&client=js&version=7.6.0&flash=false");
-
-        public Uri KfWsEndpoint { get; set; } = new("wss://kiwifarms.st:9443/chat.ws");
-
-        public List<string> PusherChannels { get; set; } = [];
-        public int KfChatRoomId { get; set; }
-        // Proxy to use for connecting to Sneedchat
-        public string? KfProxy { get; set; }
-        // Proxy to use for the Pusher websocket
-        // e.g. socks5://blahblah:1080
-        public string? PusherProxy { get; set; }
-        public int KfReconnectTimeout { get; set; } = 30;
-        public int PusherReconnectTimeout { get; set; } = 30;
-        public bool EnableGambaSeshDetect { get; set; } = true;
-        public int GambaSeshUserId { get; set; } = 168162;
-        public string KickIcon { get; set; } = "https://i.ibb.co/0cqwscx/kick.png";
-        public string KfDomain { get; set; } = "kiwifarms.st";
-        public required string KfUsername { get; set; }
-        public required string KfPassword { get; set; }
-        public string ChromiumPath { get; set; } = "chromium_install";
-    }
-}
--- a/KfChatDotNetKickBot/Models/ConfigModel.cs
+++ b/KfChatDotNetKickBot/Models/ConfigModel.cs
@@ -0,0 +1,26 @@
+namespace KfChatDotNetKickBot.Models;
+
+public class ConfigModel
+{
+    public Uri PusherEndpoint { get; set; } =
+        new("wss://ws-us2.pusher.com/app/eb1d5f283081a78b932c?protocol=7&client=js&version=7.6.0&flash=false");
+
+    public Uri KfWsEndpoint { get; set; } = new("wss://kiwifarms.st:9443/chat.ws");
+
+    public List<string> PusherChannels { get; set; } = [];
+    public int KfChatRoomId { get; set; }
+    // Proxy to use for everything
+    public string? Proxy { get; set; }
+    public int KfReconnectTimeout { get; set; } = 30;
+    public int PusherReconnectTimeout { get; set; } = 30;
+    public bool EnableGambaSeshDetect { get; set; } = true;
+    public int GambaSeshUserId { get; set; } = 168162;
+    public string KickIcon { get; set; } = "https://i.ibb.co/0cqwscx/kick.png";
+    public string KfDomain { get; set; } = "kiwifarms.st";
+    public required string KfUsername { get; set; }
+    public required string KfPassword { get; set; }
+    public string ChromiumPath { get; set; } = "chromium_install";
+    public int? BossmanJackTwitchId { get; set; } = null;
+    // Used for testing
+    public bool SuppressChatMessages { get; set; } = false;
+}
--- a/KfChatDotNetKickBot/Services/TwitchWs.cs
+++ b/KfChatDotNetKickBot/Services/TwitchWs.cs
@@ -0,0 +1,145 @@
+using System.Net;
+using System.Net.WebSockets;
+using System.Text.Json;
+using NLog;
+using Websocket.Client;
+
+namespace KfChatDotNetKickBot.Services;
+
+public class TwitchWs
+{
+    private Logger _logger = LogManager.GetCurrentClassLogger();
+    private WebsocketClient _wsClient;
+    private Uri _wsUri = new("wss://pubsub-edge.twitch.tv/v1");
+    private int _reconnectTimeout = 300;
+    private string? _proxy;
+    private List<int> _channels;
+    public delegate void OnStreamStateUpdateEventHandler(object sender, int channelId, bool isLive);
+    public event OnStreamStateUpdateEventHandler OnStreamStateUpdated;
+    private CancellationToken _cancellationToken = CancellationToken.None;
+
+    public TwitchWs(List<int> channels,  string? proxy = null, CancellationToken? cancellationToken = null)
+    {
+        _proxy = proxy;
+        _channels = channels;
+        if (cancellationToken != null) _cancellationToken = cancellationToken.Value;
+    }
+    
+    public async Task StartWsClient()
+    {
+        _logger.Debug("StartWsClient() called, creating client");
+        await CreateWsClient();
+    }
+
+    private async Task CreateWsClient()
+    {
+        var factory = new Func<ClientWebSocket>(() =>
+        {
+            var clientWs = new ClientWebSocket();
+            if (_proxy == null) return clientWs;
+            _logger.Debug($"Using proxy address {_proxy}");
+            clientWs.Options.Proxy = new WebProxy(_proxy);
+            return clientWs;
+        });
+        
+        var client = new WebsocketClient(_wsUri, factory)
+        {
+            ReconnectTimeout = TimeSpan.FromSeconds(_reconnectTimeout)
+        };
+        
+        client.ReconnectionHappened.Subscribe(WsReconnection);
+        client.MessageReceived.Subscribe(WsMessageReceived);
+        client.DisconnectionHappened.Subscribe(WsDisconnection);
+
+        _wsClient = client;
+        
+        _logger.Debug("Websocket client has been built, about to start");
+        await client.Start();
+        _logger.Debug("Websocket client started!");
+        SendPing();
+        _ = PeriodicPing();
+    }
+    
+    public bool IsConnected()
+    {
+        return _wsClient is { IsRunning: true };
+    }
+
+    private void SendPing()
+    {
+        _logger.Debug("Sending ping to Twitch");
+        _wsClient.Send("{\"type\":\"PING\"}");
+    }
+
+    private async Task PeriodicPing()
+    {
+        using var timer = new PeriodicTimer(TimeSpan.FromMinutes(1));
+        while (await timer.WaitForNextTickAsync(_cancellationToken))
+        {
+            SendPing();
+        }
+    }
+    
+    private void WsDisconnection(DisconnectionInfo disconnectionInfo)
+    {
+        _logger.Error($"Client disconnected from the chat (or never successfully connected). Type is {disconnectionInfo.Type}");
+        _logger.Error(disconnectionInfo.Exception);
+    }
+    
+    private void WsReconnection(ReconnectionInfo reconnectionInfo)
+    {
+        _logger.Error($"Websocket connection dropped and reconnected. Reconnection type is {reconnectionInfo.Type}");
+        _logger.Info("Sending subscription requests");
+        foreach (var channel in _channels)
+        {
+            _logger.Info($"Subscribing to {channel}");
+            _wsClient.Send("{\"data\":{\"topics\":[\"video-playback-by-id." + _channels + "\"]},\"nonce\":\"" + Guid.NewGuid() + "\",\"type\":\"LISTEN\"}");
+        }
+    }
+
+    private void WsMessageReceived(ResponseMessage message)
+    {
+        if (message.Text == null)
+        {
+            _logger.Info("Twitch sent a null message");
+            return;
+        }
+        _logger.Debug($"Received event from Twitch: {message.Text}");
+
+        try
+        {
+            var packet = JsonSerializer.Deserialize<JsonElement>(message.Text);
+            if (packet.GetProperty("type").GetString() != "MESSAGE")
+                return;
+            var data = packet.GetProperty("data")!;
+            var topicString = data.GetProperty("topic")!.GetString()!;
+            if (!topicString.StartsWith("video-playback-by-id."))
+                return;
+            var topicParts = topicString.Split('.');
+            var channelId = int.Parse(topicParts[^1]);
+            var twitchMessage = data.GetProperty("message")!.GetString()!;
+
+            if (twitchMessage.Contains("\"type\":\"stream-up\""))
+            {
+                OnStreamStateUpdated?.Invoke(this, channelId, true);
+                return;
+            }
+
+            if (twitchMessage.Contains("\"type\":\"stream-down\""))
+            {
+                OnStreamStateUpdated?.Invoke(this, channelId, false);
+                return;
+            }
+            _logger.Info("Message from Twitch was unhandled");
+            _logger.Info(message.Text);
+        }
+        catch (Exception e)
+        {
+            _logger.Error("Failed to handle message from Twitch");
+            _logger.Error(e);
+            _logger.Error("--- JSON Payload ---");
+            _logger.Error(message.Text);
+            _logger.Error("--- End of JSON Payload ---");
+        }
+    }
+}
--- a/KfChatDotNetKickBot/config.json
+++ b/KfChatDotNetKickBot/config.json
@@ -1,6 +1,6 @@
 {
  "PusherChannels": ["chatrooms.2507974.v2", "channel.2515504"],
-  "KfProxy": "socks5://us-lax-wg-socks5-203.relays.mullvad.net:1080",
+  "Proxy": "socks5://us-lax-wg-socks5-203.relays.mullvad.net:1080",
  "KfChatRoomId": 15,
-  "XfTokenValue": "fill this in with the value from xf_session"
+  "BossmanJackTwitchId": 114122847
 }