HyperSpaceOut/IRC-kosmi-relay
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add !votes command, fix vote tally timing, and improve Kosmi stability

Browse Source 
- Add !votes command (IRC + Kosmi) showing per-session and all-time vote
  breakdowns for the current game via new Jackbox API endpoints
  (GET sessions/{id}/games, sessions/{id}/votes, games/{id})
- Fix vote tally broadcasting: remove debounce timer, announce tallies
  only at game transitions or session end instead of after every vote
- Add !kreconnect IRC command to manually trigger Kosmi reconnection
- Add WebSocket ping/pong keepalive and write mutex to Kosmi client
  for connection stability
- Add watchConnection() auto-reconnect on unexpected Kosmi disconnects
- Remove old 2025-10-31 chat summaries; add votes command design doc

Made-with: Cursor
This commit is contained in:
cottongin
2026-03-16 20:56:18 -04:00
parent 1831b0e923
commit 88cc140087
15 changed files with 536 additions and 1398 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
bridge/config/config.go
View File

@@ -28,6 +28,8 @@ const (
EventUserTyping = "user_typing"
EventGetChannelMembers = "get_channel_members"
EventNoticeIRC = "notice_irc"
EventReconnectKosmi = "reconnect_kosmi"
EventVotesQuery = "votes_query"
)
const ParentIDNotFound = "msg-parent-not-found"

26
bridge/irc/handlers.go
View File

@@ -269,6 +269,32 @@ func (b *Birc) handlePrivMsg(client *girc.Client, event girc.Event) {
}
}
// Handle !kreconnect command: trigger Kosmi bridge reconnection
if strings.TrimSpace(rmsg.Text) == "!kreconnect" {
b.Log.Infof("!kreconnect command from %s on %s", event.Source.Name, rmsg.Channel)
b.Remote <- config.Message{
Username: "system",
Text: "kreconnect",
Channel: rmsg.Channel,
Account: b.Account,
Event: config.EventReconnectKosmi,
}
return
}
// Handle !votes command: query current game vote tally
if strings.TrimSpace(rmsg.Text) == "!votes" {
b.Log.Infof("!votes command from %s on %s", event.Source.Name, rmsg.Channel)
b.Remote <- config.Message{
Username: "system",
Text: "votes",
Channel: rmsg.Channel,
Account: b.Account,
Event: config.EventVotesQuery,
}
return
}
b.Log.Debugf("<= Sending message from %s on %s to gateway", event.Params[0], b.Account)
b.Remote <- rmsg
}

272
bridge/jackbox/client.go
View File

@@ -24,10 +24,8 @@ type Client struct {
messageCallback func(string)
// Vote tracking
activeSessionID int
lastVoteResponse *VoteResponse
voteDebounceTimer *time.Timer
voteDebounceDelay time.Duration
activeSessionID int
lastVoteResponse *VoteResponse
}
// AuthResponse represents the authentication response from the API
@@ -67,6 +65,44 @@ type SessionResponse struct {
Session *Session `json:"session"`
}
// SessionGame represents a game within a session
type SessionGame struct {
ID int `json:"id"`
GameID int `json:"game_id"`
Title string `json:"title"`
PackName string `json:"pack_name"`
Status string `json:"status"`
RoomCode string `json:"room_code"`
}
// SessionVotesResponse represents the per-game vote breakdown for a session
type SessionVotesResponse struct {
SessionID int `json:"session_id"`
Votes []GameVoteSummary `json:"votes"`
}
// GameVoteSummary represents aggregated vote data for a single game in a session
type GameVoteSummary struct {
GameID int `json:"game_id"`
Title string `json:"title"`
PackName string `json:"pack_name"`
Upvotes int `json:"upvotes"`
Downvotes int `json:"downvotes"`
NetScore int `json:"net_score"`
TotalVotes int `json:"total_votes"`
}
// Game represents a game from the catalog
type Game struct {
ID int `json:"id"`
Title string `json:"title"`
PackName string `json:"pack_name"`
PopularityScore int `json:"popularity_score"`
Upvotes int `json:"upvotes"`
Downvotes int `json:"downvotes"`
PlayCount int `json:"play_count"`
}
// NewClient creates a new Jackbox API client
func NewClient(apiURL, adminPassword string, log *logrus.Entry) *Client {
return &Client{
@@ -76,7 +112,6 @@ func NewClient(apiURL, adminPassword string, log *logrus.Entry) *Client {
httpClient: &http.Client{
Timeout: 10 * time.Second,
},
voteDebounceDelay: 3 * time.Second, // Wait 3 seconds after last vote before broadcasting
}
}
@@ -102,13 +137,6 @@ func (c *Client) GetAndClearLastVoteResponse() *VoteResponse {
resp := c.lastVoteResponse
c.lastVoteResponse = nil
// Stop any pending debounce timer
if c.voteDebounceTimer != nil {
c.voteDebounceTimer.Stop()
c.voteDebounceTimer = nil
}
return resp
}
@@ -267,8 +295,21 @@ func (c *Client) SendVote(username, voteType string, timestamp time.Time) error
c.log.Debugf("Vote recorded for %s: %s - %d👍 %d👎",
voteResp.Game.Title, username, voteResp.Game.Upvotes, voteResp.Game.Downvotes)
// Debounce vote broadcasts - wait for activity to settle
c.debouncedVoteBroadcast(&voteResp)
// Accumulate vote; tally announced at game change or session end
c.storeVoteResponse(&voteResp)
// If local session tracking is stale, sync from the API.
// A successful vote means the API has an active session.
c.mu.RLock()
sessionID := c.activeSessionID
c.mu.RUnlock()
if sessionID == 0 {
go func() {
if session, err := c.GetActiveSession(); err == nil && session != nil {
c.SetActiveSession(session.ID)
}
}()
}
return nil
}
@@ -351,49 +392,168 @@ func (c *Client) GetActiveSession() (*Session, error) {
return &session, nil
}
// debouncedVoteBroadcast implements debouncing for vote broadcasts
// When there's an active session, it stores votes to be announced with the next game
// When there's no active session, it uses time-based debouncing (3 seconds)
func (c *Client) debouncedVoteBroadcast(voteResp *VoteResponse) {
// storeVoteResponse accumulates the latest vote response silently.
// The tally is announced later by handleGameAdded or AnnounceSessionEnd
// via GetAndClearLastVoteResponse.
func (c *Client) storeVoteResponse(voteResp *VoteResponse) {
c.mu.Lock()
defer c.mu.Unlock()
// Store the latest vote response
c.lastVoteResponse = voteResp
// If there's an active session, just accumulate votes silently
// They'll be announced when the next game is picked
if c.activeSessionID > 0 {
c.log.Debugf("Vote accumulated for %s (session active, will announce with next game)", voteResp.Game.Title)
// Cancel any existing timer since we're in session mode
if c.voteDebounceTimer != nil {
c.voteDebounceTimer.Stop()
c.voteDebounceTimer = nil
}
return
}
// No active session - use time-based debouncing
// If there's an existing timer, stop it
if c.voteDebounceTimer != nil {
c.voteDebounceTimer.Stop()
}
// Create a new timer that will fire after the debounce delay
c.voteDebounceTimer = time.AfterFunc(c.voteDebounceDelay, func() {
c.mu.Lock()
lastResp := c.lastVoteResponse
c.lastVoteResponse = nil
c.mu.Unlock()
if lastResp != nil {
// Broadcast the final vote result
message := fmt.Sprintf("🗳️ Voting complete for %s • %d👍 %d👎 (Score: %d)",
lastResp.Game.Title,
lastResp.Game.Upvotes, lastResp.Game.Downvotes, lastResp.Game.PopularityScore)
c.broadcastMessage(message)
c.log.Infof("Broadcast final vote result: %s - %d👍 %d👎",
lastResp.Game.Title, lastResp.Game.Upvotes, lastResp.Game.Downvotes)
}
})
c.log.Debugf("Vote accumulated for %s (will announce at game change or session end)", voteResp.Game.Title)
}
// GetSessionGames retrieves the list of games in a session
func (c *Client) GetSessionGames(sessionID int) ([]SessionGame, error) {
if err := c.ensureAuthenticated(); err != nil {
return nil, fmt.Errorf("authentication failed: %w", err)
}
c.mu.RLock()
token := c.token
c.mu.RUnlock()
url := fmt.Sprintf("%s/api/sessions/%d/games", c.apiURL, sessionID)
req, err := http.NewRequest("GET", url, nil)
if err != nil {
return nil, fmt.Errorf("failed to create request: %w", err)
}
req.Header.Set("Authorization", "Bearer "+token)
resp, err := c.httpClient.Do(req)
if err != nil {
return nil, fmt.Errorf("failed to send request: %w", err)
}
defer resp.Body.Close()
body, err := io.ReadAll(resp.Body)
if err != nil {
return nil, fmt.Errorf("failed to read response body: %w", err)
}
if resp.StatusCode == http.StatusUnauthorized {
c.log.Warn("Token expired, re-authenticating...")
if err := c.Authenticate(); err != nil {
return nil, fmt.Errorf("re-authentication failed: %w", err)
}
return c.GetSessionGames(sessionID)
}
if resp.StatusCode == http.StatusNotFound {
return nil, nil
}
if resp.StatusCode != http.StatusOK {
return nil, fmt.Errorf("unexpected status %d: %s", resp.StatusCode, string(body))
}
var games []SessionGame
if err := json.Unmarshal(body, &games); err != nil {
return nil, fmt.Errorf("failed to parse response: %w", err)
}
return games, nil
}
// GetSessionVotes retrieves the per-game vote breakdown for a session
func (c *Client) GetSessionVotes(sessionID int) (*SessionVotesResponse, error) {
if err := c.ensureAuthenticated(); err != nil {
return nil, fmt.Errorf("authentication failed: %w", err)
}
c.mu.RLock()
token := c.token
c.mu.RUnlock()
url := fmt.Sprintf("%s/api/sessions/%d/votes", c.apiURL, sessionID)
req, err := http.NewRequest("GET", url, nil)
if err != nil {
return nil, fmt.Errorf("failed to create request: %w", err)
}
req.Header.Set("Authorization", "Bearer "+token)
resp, err := c.httpClient.Do(req)
if err != nil {
return nil, fmt.Errorf("failed to send request: %w", err)
}
defer resp.Body.Close()
body, err := io.ReadAll(resp.Body)
if err != nil {
return nil, fmt.Errorf("failed to read response body: %w", err)
}
if resp.StatusCode == http.StatusUnauthorized {
c.log.Warn("Token expired, re-authenticating...")
if err := c.Authenticate(); err != nil {
return nil, fmt.Errorf("re-authentication failed: %w", err)
}
return c.GetSessionVotes(sessionID)
}
if resp.StatusCode == http.StatusNotFound {
return nil, nil
}
if resp.StatusCode != http.StatusOK {
return nil, fmt.Errorf("unexpected status %d: %s", resp.StatusCode, string(body))
}
var votesResp SessionVotesResponse
if err := json.Unmarshal(body, &votesResp); err != nil {
return nil, fmt.Errorf("failed to parse response: %w", err)
}
return &votesResp, nil
}
// GetGame retrieves a single game from the catalog by ID
func (c *Client) GetGame(gameID int) (*Game, error) {
if err := c.ensureAuthenticated(); err != nil {
return nil, fmt.Errorf("authentication failed: %w", err)
}
c.mu.RLock()
token := c.token
c.mu.RUnlock()
url := fmt.Sprintf("%s/api/games/%d", c.apiURL, gameID)
req, err := http.NewRequest("GET", url, nil)
if err != nil {
return nil, fmt.Errorf("failed to create request: %w", err)
}
req.Header.Set("Authorization", "Bearer "+token)
resp, err := c.httpClient.Do(req)
if err != nil {
return nil, fmt.Errorf("failed to send request: %w", err)
}
defer resp.Body.Close()
body, err := io.ReadAll(resp.Body)
if err != nil {
return nil, fmt.Errorf("failed to read response body: %w", err)
}
if resp.StatusCode == http.StatusUnauthorized {
c.log.Warn("Token expired, re-authenticating...")
if err := c.Authenticate(); err != nil {
return nil, fmt.Errorf("re-authentication failed: %w", err)
}
return c.GetGame(gameID)
}
if resp.StatusCode == http.StatusNotFound {
return nil, nil
}
if resp.StatusCode != http.StatusOK {
return nil, fmt.Errorf("unexpected status %d: %s", resp.StatusCode, string(body))
}
var game Game
if err := json.Unmarshal(body, &game); err != nil {
return nil, fmt.Errorf("failed to parse response: %w", err)
}
return &game, nil
}

56
bridge/kosmi/graphql_ws_client.go
View File

@@ -14,10 +14,14 @@ import (
)
const (
kosmiWSURL = "wss://engine.kosmi.io/gql-ws"
kosmiWSURL = "wss://engine.kosmi.io/gql-ws"
kosmiHTTPURL = "https://engine.kosmi.io/"
userAgent = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
appVersion = "4364"
userAgent = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
appVersion = "4364"
pingInterval = 30 * time.Second
pongTimeout = 90 * time.Second
writeWait = 10 * time.Second
)
// GraphQL-WS Protocol message types
@@ -40,6 +44,7 @@ type GraphQLWSClient struct {
messageCallback func(*NewMessagePayload)
connected bool
mu sync.RWMutex
writeMu sync.Mutex
done chan struct{}
}
@@ -208,9 +213,17 @@ func (c *GraphQLWSClient) Connect() error {
c.connected = true
c.mu.Unlock()
// Set up ping/pong keepalive
conn.SetReadDeadline(time.Now().Add(pongTimeout))
conn.SetPongHandler(func(string) error {
conn.SetReadDeadline(time.Now().Add(pongTimeout))
return nil
})
c.log.Info("Native WebSocket client connected and ready")
// Start message listener
// Start keepalive pinger and message listener
go c.startPing()
go c.listenForMessages()
return nil
@@ -359,7 +372,10 @@ func (c *GraphQLWSClient) SendMessage(text string) error {
},
}
if err := c.conn.WriteJSON(msg); err != nil {
c.writeMu.Lock()
err := c.conn.WriteJSON(msg)
c.writeMu.Unlock()
if err != nil {
return fmt.Errorf("failed to send message: %w", err)
}
@@ -396,3 +412,33 @@ func (c *GraphQLWSClient) IsConnected() bool {
return c.connected
}
// Done returns a channel that is closed when the client disconnects
func (c *GraphQLWSClient) Done() <-chan struct{} {
return c.done
}
// startPing sends WebSocket ping frames at a regular interval to keep the
// connection alive and detect stale connections early.
func (c *GraphQLWSClient) startPing() {
ticker := time.NewTicker(pingInterval)
defer ticker.Stop()
for {
select {
case <-ticker.C:
c.writeMu.Lock()
err := c.conn.WriteControl(
websocket.PingMessage, nil, time.Now().Add(writeWait),
)
c.writeMu.Unlock()
if err != nil {
c.log.Warnf("Ping failed, connection likely dead: %v", err)
c.conn.Close()
return
}
case <-c.done:
return
}
}
}

64
bridge/kosmi/kosmi.go
View File

@@ -22,25 +22,25 @@ type KosmiClient interface {
SendMessage(text string) error
OnMessage(callback func(*NewMessagePayload))
IsConnected() bool
Done() <-chan struct{}
}
// Bkosmi represents the Kosmi bridge
type Bkosmi struct {
*bridge.Config
client KosmiClient
roomID string
roomURL string
connected bool
authDone bool // Signals that authentication is complete (like IRC bridge)
msgChannel chan config.Message
jackboxClient *jackbox.Client
client KosmiClient
roomID string
roomURL string
connected bool
intentionalDisconnect bool
authDone bool // Signals that authentication is complete (like IRC bridge)
jackboxClient *jackbox.Client
}
// New creates a new Kosmi bridge instance
func New(cfg *bridge.Config) bridge.Bridger {
b := &Bkosmi{
Config: cfg,
msgChannel: make(chan config.Message, 100),
Config: cfg,
}
return b
@@ -110,9 +110,12 @@ func (b *Bkosmi) Connect() error {
}
b.connected = true
b.intentionalDisconnect = false
b.authDone = true // Signal that authentication is complete
b.Log.Info("Successfully connected to Kosmi")
go b.watchConnection()
return nil
}
@@ -120,15 +123,15 @@ func (b *Bkosmi) Connect() error {
func (b *Bkosmi) Disconnect() error {
b.Log.Info("Disconnecting from Kosmi")
b.intentionalDisconnect = true
b.connected = false
if b.client != nil {
if err := b.client.Disconnect(); err != nil {
b.Log.Errorf("Error closing Kosmi client: %v", err)
}
}
close(b.msgChannel)
b.connected = false
return nil
}
@@ -216,6 +219,19 @@ func (b *Bkosmi) handleIncomingMessage(payload *NewMessagePayload) {
}
}
// Handle !votes command: query current game vote tally
if strings.TrimSpace(body) == "!votes" {
b.Log.Infof("!votes command from %s", username)
b.Remote <- config.Message{
Username: "system",
Text: "votes",
Channel: "main",
Account: b.Account,
Event: config.EventVotesQuery,
}
return
}
// Create Matterbridge message
// Use "main" as the channel name for gateway matching
// Don't add prefix here - let the gateway's RemoteNickFormat handle it
@@ -240,6 +256,30 @@ func (b *Bkosmi) handleIncomingMessage(payload *NewMessagePayload) {
b.Remote <- rmsg
}
// watchConnection monitors the WebSocket client and sends EventFailure
// to the gateway when an unexpected disconnect occurs, triggering automatic
// reconnection via the gateway's reconnectBridge() mechanism.
func (b *Bkosmi) watchConnection() {
<-b.client.Done()
if b.intentionalDisconnect {
return
}
b.Log.Warn("Kosmi connection lost unexpectedly, requesting reconnection")
b.connected = false
if b.Remote != nil {
b.Remote <- config.Message{
Username: "system",
Text: "reconnect",
Channel: "",
Account: b.Account,
Event: config.EventFailure,
}
}
}
// extractRoomID extracts the room ID from a Kosmi room URL
// Supports formats:
// - https://app.kosmi.io/room/@roomname