[kick] Switch to chatwebhook for chat

Author: xaionaro

go.mod

@@ -376,5 +376,5 @@ require (
 	github.com/pion/turn/v2 v2.1.6 // indirect
 	github.com/pion/webrtc/v3 v3.3.0 // indirect
 	github.com/tiendc/go-deepcopy v1.5.2
-	github.com/xaionaro-go/chatwebhook v0.0.0-20251102204738-0b8b2966ba1d
+	github.com/xaionaro-go/chatwebhook v0.0.0-20251102210754-fc155de0d2a9
 )

go.sum

@@ -1121,6 +1121,8 @@ github.com/xaionaro-go/chatwebhook v0.0.0-20251102191426-33a5759f705c h1:fVqdR1E
 github.com/xaionaro-go/chatwebhook v0.0.0-20251102191426-33a5759f705c/go.mod h1:y6fX0jgyyfrjyj4QMRLAF/ieNHdWsccd2dcGlXNUPfI=
 github.com/xaionaro-go/chatwebhook v0.0.0-20251102204738-0b8b2966ba1d h1:Q8i1p+zyF+K5pF4uz0qI8BJCjcSmQ6HVozCAS5nhv9w=
 github.com/xaionaro-go/chatwebhook v0.0.0-20251102204738-0b8b2966ba1d/go.mod h1:y6fX0jgyyfrjyj4QMRLAF/ieNHdWsccd2dcGlXNUPfI=
+github.com/xaionaro-go/chatwebhook v0.0.0-20251102210754-fc155de0d2a9 h1:OjxCra9GSMYajfb+brJS4oFiPUTckxipJZeFBNlCKAQ=
+github.com/xaionaro-go/chatwebhook v0.0.0-20251102210754-fc155de0d2a9/go.mod h1:y6fX0jgyyfrjyj4QMRLAF/ieNHdWsccd2dcGlXNUPfI=
 github.com/xaionaro-go/cloudscraper v0.0.0-20251019213127-d3687042cb55 h1:rHvnuVmqa9M+id2Q8e0V/csp0jDhXIBAAuCC76mBW8k=
 github.com/xaionaro-go/cloudscraper v0.0.0-20251019213127-d3687042cb55/go.mod h1:mFODbc87KEjFfERrFjmAGajGZOBQq+kzMd+2wsAZSHo=
 github.com/xaionaro-go/datacounter v1.0.4 h1:+QMZLmu73R5WGkQfUPwlXF/JFN+Weo4iuDZkiL2wVm8=

pkg/streamcontrol/kick/chat_handler.go

@@ -39,26 +39,41 @@ func NewChatHandler(
 	}, nil
 }
 
+func (k *Kick) newChatHandler(
+	ctx context.Context,
+) (*ChatHandler, error) {
+	c, err := chatwebhookclient.New(ctx, chatwebhookclient.DefaultServerAddress)
+	if err != nil {
+		return nil, fmt.Errorf("kick: failed to create chat webhook client: %w", err)
+	}
+	return NewChatHandler(ctx, c)
+}
+
 func (h *ChatHandler) GetMessagesChan(
 	ctx context.Context,
 ) (<-chan streamcontrol.Event, error) {
 	ctx, cancelFn := context.WithCancel(ctx)
-	defer cancelFn()
 
 	inCh, err := h.Client.GetMessagesChan(ctx, kickcom.ID, "")
 	if err != nil {
+		cancelFn()
 		return nil, fmt.Errorf("kick: failed to get messages chan: %w", err)
 	}
 
 	outCh := make(chan streamcontrol.Event, 1)
 	observability.Go(ctx, func(ctx context.Context) {
 		defer close(outCh)
+		defer cancelFn()
+		logger.Debugf(ctx, "kick: started forwarding chat messages")
+		defer logger.Debugf(ctx, "kick: stopped forwarding chat messages")
 		for {
 			select {
 			case <-ctx.Done():
+				logger.Debugf(ctx, "kick: forwarding chat messages: context is closed; %v", ctx.Err())
 				return
 			case ev, ok := <-inCh:
 				if !ok {
+					logger.Debugf(ctx, "kick: forwarding chat messages: input channel is closed")
 					return
 				}
 				msg, err := convertKickEventToChatMessage(ev)

pkg/streamcontrol/kick/cmd/chatlistener/main.go

@@ -6,10 +6,12 @@ import (
 	"log"
 	"os"
 
+	"github.com/davecgh/go-spew/spew"
 	"github.com/facebookincubator/go-belt"
 	"github.com/facebookincubator/go-belt/tool/logger"
 	xlogrus "github.com/facebookincubator/go-belt/tool/logger/implementation/logrus"
 	"github.com/spf13/pflag"
+	chatwebhookclient "github.com/xaionaro-go/chatwebhook/pkg/grpc/client"
 	"github.com/xaionaro-go/streamctl/pkg/streamcontrol/kick"
 )
 
@@ -20,32 +22,36 @@ func assertNoError(err error) {
 	log.Panic(err)
 }
 
+func must[T any](v T, err error) T {
+	assertNoError(err)
+	return v
+}
+
 func main() {
 	logLevel := logger.LevelInfo
 	pflag.Var(&logLevel, "log-level", "")
+	chatwebhookAddr := pflag.String("chatwebhook-server", chatwebhookclient.DefaultServerAddress, "chat webhook gRPC server address")
 	pflag.Parse()
 
-	if pflag.NArg() != 1 {
-		fmt.Fprintf(os.Stderr, "expected 1 argument\n")
+	if pflag.NArg() != 0 {
+		fmt.Fprintf(os.Stderr, "expected 0 argument\n")
 		os.Exit(1)
 	}
 
-	channelSlug := pflag.Arg(0)
-
 	ctx := logger.CtxWithLogger(context.Background(), xlogrus.Default().WithLevel(logLevel))
 	logger.Default = func() logger.Logger {
 		return logger.FromCtx(ctx)
 	}
 	defer belt.Flush(ctx)
 
-	h, err := kick.NewChatHandlerOBSOLETE(ctx, channelSlug)
-	assertNoError(err)
-
-	msgCh, err := h.GetMessagesChan(ctx)
-	assertNoError(err)
+	logger.Debugf(ctx, "connecting to chat webhook server at %q", *chatwebhookAddr)
+	c := must(chatwebhookclient.New(ctx, *chatwebhookAddr))
+	h := must(kick.NewChatHandler(ctx, c))
+	msgCh := must(h.GetMessagesChan(ctx))
 
 	fmt.Println("started")
 	for ev := range msgCh {
-		fmt.Printf("%#+v\n", ev)
+		spew.Dump(ev)
 	}
+	fmt.Println("ended")
 }

pkg/streamcontrol/kick/cmd/chatlistenerobsolete/main.go

@@ -0,0 +1,54 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"os"
+
+	"github.com/davecgh/go-spew/spew"
+	"github.com/facebookincubator/go-belt"
+	"github.com/facebookincubator/go-belt/tool/logger"
+	xlogrus "github.com/facebookincubator/go-belt/tool/logger/implementation/logrus"
+	"github.com/spf13/pflag"
+	"github.com/xaionaro-go/streamctl/pkg/streamcontrol/kick"
+)
+
+func assertNoError(err error) {
+	if err == nil {
+		return
+	}
+	log.Panic(err)
+}
+
+func must[T any](v T, err error) T {
+	assertNoError(err)
+	return v
+}
+
+func main() {
+	logLevel := logger.LevelInfo
+	pflag.Var(&logLevel, "log-level", "")
+	pflag.Parse()
+
+	if pflag.NArg() != 1 {
+		fmt.Fprintf(os.Stderr, "expected 1 argument\n")
+		os.Exit(1)
+	}
+
+	channelSlug := pflag.Arg(0)
+
+	ctx := logger.CtxWithLogger(context.Background(), xlogrus.Default().WithLevel(logLevel))
+	logger.Default = func() logger.Logger {
+		return logger.FromCtx(ctx)
+	}
+	defer belt.Flush(ctx)
+
+	h := must(kick.NewChatHandlerOBSOLETE(ctx, channelSlug))
+	msgCh := must(h.GetMessagesChan(ctx))
+
+	fmt.Println("started")
+	for ev := range msgCh {
+		spew.Dump(ev)
+	}
+}

pkg/streamcontrol/kick/config.go

@@ -4,7 +4,6 @@ import (
 	"context"
 
 	"github.com/xaionaro-go/streamctl/pkg/streamcontrol"
-	streamctl "github.com/xaionaro-go/streamctl/pkg/streamcontrol"
 	kick "github.com/xaionaro-go/streamctl/pkg/streamcontrol/kick/types"
 )
 
@@ -17,10 +16,10 @@ type PlatformSpecificConfig = kick.PlatformSpecificConfig
 type OAuthHandler = kick.OAuthHandler
 
 func init() {
-	streamctl.RegisterPlatform[PlatformSpecificConfig, StreamProfile](ID)
+	streamcontrol.RegisterPlatform[PlatformSpecificConfig, StreamProfile](ID)
 }
 
-func InitConfig(cfg streamctl.Config) {
+func InitConfig(cfg streamcontrol.Config) {
 	kick.InitConfig(cfg)
 }
 

pkg/streamcontrol/kick/kick.go

@@ -158,7 +158,7 @@ func (k *Kick) keepAliveLoop(
 func (k *Kick) initChatHandlerNoLock(
 	ctx context.Context,
 ) error {
-	chatHandler, err := k.newChatHandlerOBSOLETE(ctx, k.CurrentConfig.Config.Channel)
+	chatHandler, err := k.newChatHandler(ctx)
 	if err == nil {
 		k.ChatHandler = chatHandler
 		return nil
@@ -176,7 +176,7 @@ func (k *Kick) initChatHandlerNoLock(
 			return fmt.Errorf("ctx is closed: %w", ctx.Err())
 		default:
 		}
-		chatHandler, err = k.newChatHandlerOBSOLETE(ctx, k.CurrentConfig.Config.Channel)
+		chatHandler, err = k.newChatHandler(ctx)
 		if err != nil {
 			logger.Debugf(ctx, "initChatHandler: unable to create a new chat handler: %v", err)
 			continue
@@ -602,6 +602,10 @@ func (k *Kick) GetChatMessagesChan(
 					k.resetChatHandler(ctx)
 					continue
 				}
+				if ev.TargetChannel.Slug != k.Channel.Slug {
+					logger.Warnf(ctx, "skipping a message for another channel: %q != %q", ev.TargetChannel.Slug, k.Channel.Slug)
+					continue
+				}
 				logger.Tracef(ctx, "GetChatMessagesChan: received a message")
 				outCh <- ev
 			}