mirror of https://git.sr.ht/~ashkeel/strimertul
318 lines
10 KiB
Go
318 lines
10 KiB
Go
package eventsub
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"log/slog"
|
|
"time"
|
|
|
|
"git.sr.ht/~ashkeel/strimertul/log"
|
|
|
|
"git.sr.ht/~ashkeel/strimertul/database"
|
|
"git.sr.ht/~ashkeel/strimertul/utils"
|
|
"github.com/gorilla/websocket"
|
|
lru "github.com/hashicorp/golang-lru/v2"
|
|
"github.com/nicklaw5/helix/v2"
|
|
)
|
|
|
|
const websocketEndpoint = "wss://eventsub.wss.twitch.tv/ws"
|
|
|
|
type Client struct {
|
|
ctx context.Context
|
|
twitchAPI *helix.Client
|
|
db database.Database
|
|
logger *slog.Logger
|
|
user helix.User
|
|
|
|
eventCache *lru.Cache[string, time.Time]
|
|
savedSubscriptions map[string]bool
|
|
}
|
|
|
|
func Setup(ctx context.Context, twitchAPI *helix.Client, user helix.User, db database.Database, logger *slog.Logger) (*Client, error) {
|
|
eventCache, err := lru.New[string, time.Time](128)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("could not create LRU cache for events: %w", err)
|
|
}
|
|
|
|
client := &Client{
|
|
ctx: ctx,
|
|
twitchAPI: twitchAPI,
|
|
db: db,
|
|
logger: logger,
|
|
user: user,
|
|
eventCache: eventCache,
|
|
savedSubscriptions: make(map[string]bool),
|
|
}
|
|
|
|
go client.eventSubLoop()
|
|
|
|
return client, nil
|
|
}
|
|
|
|
func (c *Client) eventSubLoop() {
|
|
endpoint := websocketEndpoint
|
|
var err error
|
|
var connection *websocket.Conn
|
|
for endpoint != "" {
|
|
endpoint, connection, err = c.connectWebsocket(endpoint, connection)
|
|
if err != nil {
|
|
c.logger.Error("EventSub websocket read error", log.Error(err))
|
|
}
|
|
}
|
|
if connection != nil {
|
|
utils.Close(connection)
|
|
}
|
|
}
|
|
|
|
func readLoop(connection *websocket.Conn, recv chan<- []byte, wsErr chan<- error) {
|
|
for {
|
|
messageType, messageData, err := connection.ReadMessage()
|
|
if err != nil {
|
|
wsErr <- err
|
|
close(recv)
|
|
close(wsErr)
|
|
return
|
|
}
|
|
if messageType != websocket.TextMessage {
|
|
continue
|
|
}
|
|
|
|
recv <- messageData
|
|
}
|
|
}
|
|
|
|
func (c *Client) connectWebsocket(url string, oldConnection *websocket.Conn) (string, *websocket.Conn, error) {
|
|
connection, _, err := websocket.DefaultDialer.Dial(url, nil)
|
|
if err != nil {
|
|
c.logger.Error("Could not establish a connection to the EventSub websocket", log.Error(err))
|
|
return "", nil, err
|
|
}
|
|
|
|
received := make(chan []byte, 10)
|
|
wsErr := make(chan error, 1)
|
|
|
|
go readLoop(connection, received, wsErr)
|
|
|
|
for {
|
|
// Wait for next message or closing/error
|
|
var messageData []byte
|
|
select {
|
|
case <-c.ctx.Done():
|
|
return "", nil, nil
|
|
case err = <-wsErr:
|
|
return url, nil, err // Return the endpoint so we can reconnect
|
|
case messageData = <-received:
|
|
}
|
|
|
|
var wsMessage WebsocketMessage
|
|
err = json.Unmarshal(messageData, &wsMessage)
|
|
if err != nil {
|
|
c.logger.Error("Error decoding EventSub message", log.Error(err))
|
|
continue
|
|
}
|
|
|
|
reconnectURL, done, err := c.processMessage(wsMessage, oldConnection)
|
|
if done {
|
|
return reconnectURL, connection, err
|
|
}
|
|
}
|
|
}
|
|
|
|
func (c *Client) processMessage(wsMessage WebsocketMessage, oldConnection *websocket.Conn) (string, bool, error) {
|
|
switch wsMessage.Metadata.MessageType {
|
|
case "session_keepalive":
|
|
// Nothing to do
|
|
case "session_welcome":
|
|
var welcomeData WelcomeMessagePayload
|
|
err := json.Unmarshal(wsMessage.Payload, &welcomeData)
|
|
if err != nil {
|
|
c.logger.Error("Error decoding EventSub welcome message", slog.String("message-type", wsMessage.Metadata.MessageType), log.Error(err))
|
|
break
|
|
}
|
|
c.logger.Info("Connection to EventSub websocket established", slog.String("session-id", welcomeData.Session.ID))
|
|
|
|
// We can only close the old connection once the new one has been established
|
|
if oldConnection != nil {
|
|
utils.Close(oldConnection)
|
|
}
|
|
|
|
// Add subscription to websocket session
|
|
err = c.addSubscriptionsForSession(welcomeData.Session.ID)
|
|
if err != nil {
|
|
c.logger.Error("Could not add subscriptions", log.Error(err))
|
|
break
|
|
}
|
|
case "session_reconnect":
|
|
var reconnectData WelcomeMessagePayload
|
|
err := json.Unmarshal(wsMessage.Payload, &reconnectData)
|
|
if err != nil {
|
|
c.logger.Error("Error decoding EventSub session reconnect parameters", slog.String("message-type", wsMessage.Metadata.MessageType), log.Error(err))
|
|
break
|
|
}
|
|
c.logger.Info("EventSub websocket requested a reconnection", slog.String("session-id", reconnectData.Session.ID), slog.String("reconnect-url", reconnectData.Session.ReconnectURL))
|
|
|
|
return reconnectData.Session.ReconnectURL, true, nil
|
|
case "notification":
|
|
go c.processEvent(wsMessage)
|
|
case "revocation":
|
|
// TODO idk what to do here
|
|
}
|
|
return "", false, nil
|
|
}
|
|
|
|
func (c *Client) processEvent(message WebsocketMessage) {
|
|
// Check if we processed this already
|
|
if message.Metadata.MessageID != "" {
|
|
if c.eventCache.Contains(message.Metadata.MessageID) {
|
|
c.logger.Debug("Received duplicate event, ignoring", slog.String("message-id", message.Metadata.MessageID))
|
|
return
|
|
}
|
|
}
|
|
defer c.eventCache.Add(message.Metadata.MessageID, message.Metadata.MessageTimestamp)
|
|
|
|
// Decode data
|
|
var notificationData NotificationMessagePayload
|
|
err := json.Unmarshal(message.Payload, ¬ificationData)
|
|
if err != nil {
|
|
c.logger.Error("Error decoding EventSub notification payload", slog.String("message-type", message.Metadata.MessageType), log.Error(err))
|
|
}
|
|
notificationData.Date = time.Now()
|
|
|
|
eventKey := fmt.Sprintf("%s%s", EventKeyPrefix, notificationData.Subscription.Type)
|
|
historyKey := fmt.Sprintf("%s%s", HistoryKeyPrefix, notificationData.Subscription.Type)
|
|
err = c.db.PutJSON(eventKey, notificationData)
|
|
if err != nil {
|
|
c.logger.Error("Error storing event to database", slog.String("key", eventKey), log.Error(err))
|
|
}
|
|
|
|
var archive []NotificationMessagePayload
|
|
err = c.db.GetJSON(historyKey, &archive)
|
|
if err != nil {
|
|
archive = []NotificationMessagePayload{}
|
|
}
|
|
archive = append(archive, notificationData)
|
|
if len(archive) > HistorySize {
|
|
archive = archive[len(archive)-HistorySize:]
|
|
}
|
|
err = c.db.PutJSON(historyKey, archive)
|
|
if err != nil {
|
|
c.logger.Error("Error storing event to database", slog.String("key", historyKey), log.Error(err))
|
|
}
|
|
}
|
|
|
|
func (c *Client) addSubscriptionsForSession(session string) error {
|
|
if c.savedSubscriptions[session] {
|
|
// Already subscribed
|
|
return nil
|
|
}
|
|
|
|
transport := helix.EventSubTransport{
|
|
Method: "websocket",
|
|
SessionID: session,
|
|
}
|
|
for topic, version := range subscriptionVersions {
|
|
sub, err := c.twitchAPI.CreateEventSubSubscription(&helix.EventSubSubscription{
|
|
Type: topic,
|
|
Version: version,
|
|
Status: "enabled",
|
|
Transport: transport,
|
|
Condition: topicCondition(topic, c.user.ID),
|
|
})
|
|
if sub.Error != "" || sub.ErrorMessage != "" {
|
|
c.logger.Error("EventSub Subscription error", slog.String("topic", topic), slog.String("topic-version", version), slog.String("err", sub.Error), slog.String("message", sub.ErrorMessage))
|
|
return fmt.Errorf("%s: %s", sub.Error, sub.ErrorMessage)
|
|
}
|
|
if err != nil {
|
|
return fmt.Errorf("error subscribing to %s: %w", topic, err)
|
|
}
|
|
}
|
|
c.savedSubscriptions[session] = true
|
|
return nil
|
|
}
|
|
|
|
func topicCondition(topic string, id string) helix.EventSubCondition {
|
|
switch topic {
|
|
case helix.EventSubTypeChannelRaid:
|
|
return helix.EventSubCondition{
|
|
ToBroadcasterUserID: id,
|
|
}
|
|
case helix.EventSubTypeChannelFollow:
|
|
return helix.EventSubCondition{
|
|
BroadcasterUserID: id,
|
|
ModeratorUserID: id,
|
|
}
|
|
case
|
|
helix.EventSubTypeChannelChatMessage,
|
|
helix.EventSubTypeChannelChatNotification:
|
|
{
|
|
return helix.EventSubCondition{
|
|
BroadcasterUserID: id,
|
|
UserID: id,
|
|
}
|
|
}
|
|
default:
|
|
return helix.EventSubCondition{
|
|
BroadcasterUserID: id,
|
|
}
|
|
}
|
|
}
|
|
|
|
type WebsocketMessage struct {
|
|
Metadata Metadata `json:"metadata"`
|
|
Payload json.RawMessage `json:"payload"`
|
|
}
|
|
|
|
type WelcomeMessagePayload struct {
|
|
Session struct {
|
|
ID string `json:"id"`
|
|
Status string `json:"status"`
|
|
ConnectedAt time.Time `json:"connected_at"`
|
|
KeepaliveTimeoutSeconds int `json:"keepalive_timeout_seconds"`
|
|
ReconnectURL string `json:"reconnect_url,omitempty"`
|
|
} `json:"session"`
|
|
}
|
|
|
|
type NotificationMessagePayload struct {
|
|
Subscription helix.EventSubSubscription `json:"subscription"`
|
|
Event json.RawMessage `json:"event"`
|
|
Date time.Time `json:"date,omitempty"`
|
|
}
|
|
|
|
type Metadata struct {
|
|
MessageID string `json:"message_id"`
|
|
MessageType string `json:"message_type"`
|
|
MessageTimestamp time.Time `json:"message_timestamp"`
|
|
SubscriptionType string `json:"subscription_type"`
|
|
SubscriptionVersion string `json:"subscription_version"`
|
|
}
|
|
|
|
var subscriptionVersions = map[string]string{
|
|
helix.EventSubTypeChannelUpdate: "1",
|
|
helix.EventSubTypeChannelFollow: "2",
|
|
helix.EventSubTypeChannelSubscription: "1",
|
|
helix.EventSubTypeChannelSubscriptionGift: "1",
|
|
helix.EventSubTypeChannelSubscriptionMessage: "1",
|
|
helix.EventSubTypeChannelCheer: "1",
|
|
helix.EventSubTypeChannelRaid: "1",
|
|
helix.EventSubTypeChannelChatMessage: "1",
|
|
helix.EventSubTypeChannelChatNotification: "1",
|
|
helix.EventSubTypeChannelPollBegin: "1",
|
|
helix.EventSubTypeChannelPollProgress: "1",
|
|
helix.EventSubTypeChannelPollEnd: "1",
|
|
helix.EventSubTypeChannelPredictionBegin: "1",
|
|
helix.EventSubTypeChannelPredictionProgress: "1",
|
|
helix.EventSubTypeChannelPredictionLock: "1",
|
|
helix.EventSubTypeChannelPredictionEnd: "1",
|
|
helix.EventSubTypeHypeTrainBegin: "1",
|
|
helix.EventSubTypeHypeTrainProgress: "1",
|
|
helix.EventSubTypeHypeTrainEnd: "1",
|
|
helix.EventSubTypeChannelPointsCustomRewardAdd: "1",
|
|
helix.EventSubTypeChannelPointsCustomRewardUpdate: "1",
|
|
helix.EventSubTypeChannelPointsCustomRewardRemove: "1",
|
|
helix.EventSubTypeChannelPointsCustomRewardRedemptionAdd: "1",
|
|
helix.EventSubTypeChannelPointsCustomRewardRedemptionUpdate: "1",
|
|
helix.EventSubTypeStreamOnline: "1",
|
|
helix.EventSubTypeStreamOffline: "1",
|
|
}
|