Multiple updates

Author: xaionaro

cmd/streamd/main.go

@@ -2,8 +2,8 @@ package main
 
 import (
 	"context"
+	"crypto/tls"
 	"log"
-	"net"
 	"net/http"
 	_ "net/http/pprof"
 	"os"
@@ -19,11 +19,13 @@ import (
 	"github.com/getsentry/sentry-go"
 	grpc_recovery "github.com/grpc-ecosystem/go-grpc-middleware/recovery"
 	"github.com/spf13/pflag"
+	"github.com/xaionaro-go/eventbus"
 	"github.com/xaionaro-go/grpcproxy/grpcproxyserver"
 	"github.com/xaionaro-go/grpcproxy/protobuf/go/proxy_grpc"
 	"github.com/xaionaro-go/obs-grpc-proxy/protobuf/go/obs_grpc"
 	"github.com/xaionaro-go/observability"
 	"github.com/xaionaro-go/streamctl/cmd/streamd/ui"
+	"github.com/xaionaro-go/streamctl/pkg/cert"
 	"github.com/xaionaro-go/streamctl/pkg/streamcontrol"
 	"github.com/xaionaro-go/streamctl/pkg/streamd"
 	"github.com/xaionaro-go/streamctl/pkg/streamd/config"
@@ -127,6 +129,8 @@ func main() {
 	}
 	defer belt.Flush(ctx)
 
+	eventbus.LoggingEnabled = true
+
 	configPathExpanded, err := xpath.Expand(*configPath)
 	if err != nil {
 		l.Fatalf("unable to get the path to the data file: %v", err)
@@ -181,7 +185,16 @@ func main() {
 			}
 		})
 
-		listener, err := net.Listen("tcp", *listenAddr)
+		cert, err := cert.GenerateSelfSignedForServer()
+		if err != nil {
+			logger.Panicf(ctx, "unable to generate the certificate: %v", err)
+		}
+
+		listener, err := tls.Listen("tcp", *listenAddr, &tls.Config{
+			Certificates: []tls.Certificate{cert},
+			NextProtos:   []string{"h2"},
+		})
+		//listener, err := net.Listen("tcp", *listenAddr)
 		if err != nil {
 			log.Fatalf("failed to listen: %v", err)
 		}

cmd/streampanel/FyneApp.toml

@@ -5,4 +5,4 @@ Website = "https://github.com/xaionaro/streamctl"
   Name = "streampanel"
   ID = "center.dx.streampanel"
   Version = "0.1.0"
-  Build = 432
+  Build = 433

cmd/streampanel/runtime.go

@@ -16,6 +16,7 @@ import (
 	"github.com/facebookincubator/go-belt"
 	"github.com/facebookincubator/go-belt/tool/logger"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
+	"github.com/xaionaro-go/eventbus"
 	"github.com/xaionaro-go/observability"
 )
 
@@ -115,6 +116,8 @@ func initRuntime(
 
 	seppukuIfMemHugeLeak(ctx)
 
+	eventbus.LoggingEnabled = true
+
 	ctx, cancelFn := context.WithCancel(ctx)
 	return ctx, func() {
 		defer belt.Flush(ctx)

cmd/streampanel/streamd.go

@@ -3,6 +3,7 @@ package main
 import (
 	"bytes"
 	"context"
+	"crypto/tls"
 	"encoding/gob"
 	"fmt"
 	"net"
@@ -20,6 +21,7 @@ import (
 	"github.com/xaionaro-go/obs-grpc-proxy/protobuf/go/obs_grpc"
 	"github.com/xaionaro-go/observability"
 	"github.com/xaionaro-go/streamctl/cmd/streamd/ui"
+	"github.com/xaionaro-go/streamctl/pkg/cert"
 	"github.com/xaionaro-go/streamctl/pkg/mainprocess"
 	"github.com/xaionaro-go/streamctl/pkg/streamcontrol"
 	"github.com/xaionaro-go/streamctl/pkg/streamd"
@@ -257,7 +259,17 @@ func initGRPCServers(
 ) (net.Listener, *grpc.Server, *server.GRPCServer, obs_grpc.OBSServer, proxy_grpc.NetworkProxyServer) {
 	logger.Debugf(ctx, "initGRPCServers")
 	defer logger.Debugf(ctx, "/initGRPCServers")
-	listener, err := net.Listen("tcp", listenAddr)
+
+	cert, err := cert.GenerateSelfSignedForServer()
+	if err != nil {
+		logger.Panicf(ctx, "unable to generate the certificate: %v", err)
+	}
+
+	logger.Debugf(ctx, "generated certificate %#+v", cert)
+	listener, err := tls.Listen("tcp", listenAddr, &tls.Config{
+		Certificates: []tls.Certificate{cert},
+		NextProtos:   []string{"h2"},
+	})
 	if err != nil {
 		logger.Panicf(ctx, "failed to listen: %v", err)
 	}

go.mod

@@ -27,15 +27,14 @@ replace github.com/asticode/go-astiav v0.36.0 => github.com/xaionaro-go/astiav v
 
 replace github.com/bluenviron/mediacommon/v2 v2.0.1-0.20250324151931-b8ce69d15d3d => github.com/xaionaro-go/mediacommon/v2 v2.0.0-20250420012906-03d6d69ac3b7
 
-replace github.com/asaskevich/EventBus v0.0.0-20200907212545-49d423059eef => github.com/mysteriumnetwork/EventBus v0.0.0-20220414214953-84469ec2b111
-
 require (
 	github.com/facebookincubator/go-belt v0.0.0-20250308011339-62fb7027b11f
 	github.com/go-git/go-billy/v5 v5.6.2
 	github.com/goccy/go-yaml v1.17.1
 	github.com/hashicorp/go-multierror v1.1.1
 	github.com/nicklaw5/helix/v2 v2.30.1-0.20240715193454-0151ccccf980
 	github.com/spf13/cobra v1.8.1
+	github.com/xaionaro-go/eventbus v0.0.0-20250712220612-3cb9e775fb98
 	github.com/xaionaro-go/logrustash v0.0.0-20240804141650-d48034780a5f // indirect
 	golang.org/x/oauth2 v0.30.0
 	google.golang.org/api v0.239.0
@@ -256,7 +255,6 @@ require (
 	github.com/adeithe/go-twitch v0.3.1
 	github.com/andreykaipov/goobs v1.4.1
 	github.com/anthonynsimon/bild v0.14.0
-	github.com/asaskevich/EventBus v0.0.0-20200907212545-49d423059eef
 	github.com/asticode/go-astiav v0.36.0
 	github.com/bamiaux/rez v0.0.0-20170731184118-29f4463c688b
 	github.com/bluenviron/gortsplib/v4 v4.12.4-0.20250324174248-61372cfa6800
@@ -326,6 +324,8 @@ require (
 require (
 	github.com/BurntSushi/xgbutil v0.0.0-20190907113008-ad855c713046
 	github.com/cloudwego/eino-ext/components/model/openai v0.0.0-20250424061409-ccd60fbc7c1c
+	github.com/coder/websocket v1.8.13
+	github.com/joeyak/go-twitch-eventsub/v3 v3.0.0
 	github.com/phuslu/goid v1.0.2 // indirect
 	github.com/pion/datachannel v1.5.10 // indirect
 	github.com/pion/dtls/v2 v2.2.12 // indirect

go.sum

@@ -216,6 +216,8 @@ github.com/cncf/xds/go v0.0.0-20210922020428-25de7278fc84/go.mod h1:eXthEFrGJvWH
 github.com/cncf/xds/go v0.0.0-20211001041855-01bcc9b48dfe/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
 github.com/cncf/xds/go v0.0.0-20211011173535-cb28da3451f1/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
 github.com/cockroachdb/apd v1.1.0/go.mod h1:8Sl8LxpKi29FqWXR16WEFZRNSz3SoPzUzeMeY4+DwBQ=
+github.com/coder/websocket v1.8.13 h1:f3QZdXy7uGVz+4uCJy2nTZyM0yTBj8yANEHhqlXZ9FE=
+github.com/coder/websocket v1.8.13/go.mod h1:LNVeNrXQZfe5qhS9ALED3uA+l5pPqvwXg3CKoDBB2gs=
 github.com/coreos/go-oidc/v3 v3.11.0 h1:Ia3MxdwpSw702YW0xgfmP1GVCMA9aEFWu12XUZ3/OtI=
 github.com/coreos/go-oidc/v3 v3.11.0/go.mod h1:gE3LgjOgFoHi9a4ce4/tJczr0Ai2/BoDhf0r5lltWI0=
 github.com/coreos/go-semver v0.3.0/go.mod h1:nnelYz7RCh+5ahJtPPxZlU+153eP4D4r3EedlOD2RNk=
@@ -613,6 +615,8 @@ github.com/jfreymuth/oggvorbis v1.0.5 h1:u+Ck+R0eLSRhgq8WTmffYnrVtSztJcYrl588DM4
 github.com/jfreymuth/oggvorbis v1.0.5/go.mod h1:1U4pqWmghcoVsCJJ4fRBKv9peUJMBHixthRlBeD6uII=
 github.com/jfreymuth/vorbis v1.0.2 h1:m1xH6+ZI4thH927pgKD8JOH4eaGRm18rEE9/0WKjvNE=
 github.com/jfreymuth/vorbis v1.0.2/go.mod h1:DoftRo4AznKnShRl1GxiTFCseHr4zR9BN3TWXyuzrqQ=
+github.com/joeyak/go-twitch-eventsub/v3 v3.0.0 h1:6BDgmYJynNDyCP7P+wM9jPQnE3leJAi58nohDnzliJ4=
+github.com/joeyak/go-twitch-eventsub/v3 v3.0.0/go.mod h1:rpqOjYP1ftWDj3H4D8fA58AdOpkvK9YvODoduDpPCQU=
 github.com/josharian/intern v1.0.0 h1:vlS4z54oSdjm0bgjRigI+G1HpF+tI+9rE5LLzOg8HmY=
 github.com/josharian/intern v1.0.0/go.mod h1:5DoeVV0s6jJacbCEi61lwdGj/aVlrQvzHFFd8Hwg//Y=
 github.com/jpillora/backoff v1.0.0/go.mod h1:J/6gKK9jxlEcS3zixgDgUAsiuZ7yrSoa/FX5e0EB2j4=
@@ -741,8 +745,6 @@ github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq
 github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
 github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U=
 github.com/mwitkow/go-conntrack v0.0.0-20190716064945-2f068394615f/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U=
-github.com/mysteriumnetwork/EventBus v0.0.0-20220414214953-84469ec2b111 h1:7s+VqlctjdVjy1z0slV2giUawTnv1A6vWj9oKKfgPhI=
-github.com/mysteriumnetwork/EventBus v0.0.0-20220414214953-84469ec2b111/go.mod h1:ef8wV5ITJhXSTG1sUkcHPAQF7lh83c7l875IvrYU7H0=
 github.com/neelance/astrewrite v0.0.0-20160511093645-99348263ae86/go.mod h1:kHJEU3ofeGjhHklVoIGuVj85JJwZ6kWPaJwCIxgnFmo=
 github.com/neelance/sourcemap v0.0.0-20151028013722-8c68805598ab/go.mod h1:Qr6/a/Q4r9LP1IltGz7tA7iOK1WonHEYhu1HRBA7ZiM=
 github.com/nfnt/resize v0.0.0-20180221191011-83c6a9932646 h1:zYyBkD/k9seD2A7fsi6Oo2LfFZAehjjQMERAvZLEDnQ=
@@ -1088,6 +1090,8 @@ github.com/xaionaro-go/avpipeline v0.0.0-20250525204026-17104bc4baca h1:Cls4rEim
 github.com/xaionaro-go/avpipeline v0.0.0-20250525204026-17104bc4baca/go.mod h1:LMh5Qi7cuntcktUezfA9toVCUCCsx9pjyGDWe9GLt9A=
 github.com/xaionaro-go/datacounter v1.0.4 h1:+QMZLmu73R5WGkQfUPwlXF/JFN+Weo4iuDZkiL2wVm8=
 github.com/xaionaro-go/datacounter v1.0.4/go.mod h1:Sf9vBevuV6w5iE6K3qJ9pWVKcyS60clWBUSQLjt5++c=
+github.com/xaionaro-go/eventbus v0.0.0-20250712220612-3cb9e775fb98 h1:8Dtop2JCrZl9bRpIVPDIQWio1gXjDmcaVKjMeT8sZQ0=
+github.com/xaionaro-go/eventbus v0.0.0-20250712220612-3cb9e775fb98/go.mod h1:zSbWHZpDvsRhjD3Sr3bruqqsWotjXvsIKmx6/THwXFw=
 github.com/xaionaro-go/fyne/v2 v2.0.0-20250622004601-3a26ee69528a h1:awMQXlaweeiSZB4rSNfMmJGJriyn1ca/m/lglBi9uyA=
 github.com/xaionaro-go/fyne/v2 v2.0.0-20250622004601-3a26ee69528a/go.mod h1:0GOXKqyvNwk3DLmsFu9v0oYM0ZcD1ysGnlHCerKoAmo=
 github.com/xaionaro-go/go-rtmp v0.0.0-20241009130244-1e3160f27f42 h1:izCjREd+62HDF9FRYqUI7dgJNdUxAIysEuqed8lBcDY=

pkg/cert/generate_self_signed.go

@@ -0,0 +1,44 @@
+package cert
+
+import (
+	"crypto/ed25519"
+	"crypto/rand"
+	"crypto/tls"
+	"crypto/x509"
+	"crypto/x509/pkix"
+	"encoding/pem"
+	"math/big"
+	"time"
+)
+
+func GenerateSelfSignedForServer() (tls.Certificate, error) {
+	pub, priv, err := ed25519.GenerateKey(rand.Reader)
+	if err != nil {
+		return tls.Certificate{}, err
+	}
+
+	tmpl := x509.Certificate{
+		SerialNumber: big.NewInt(1),
+		Subject: pkix.Name{
+			Organization: []string{"DX.center"},
+		},
+		NotBefore: time.Now(),
+		NotAfter:  time.Now().Add(10 * 365 * 24 * time.Hour),
+
+		KeyUsage:    x509.KeyUsageDigitalSignature,
+		ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
+		DNSNames:    []string{"wingout.dx.center"},
+	}
+
+	certDER, err := x509.CreateCertificate(rand.Reader, &tmpl, &tmpl, pub, priv)
+	if err != nil {
+		return tls.Certificate{}, err
+	}
+
+	keyBytes, err := x509.MarshalPKCS8PrivateKey(priv)
+
+	certPEM := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: certDER})
+	keyPEM := pem.EncodeToMemory(&pem.Block{Type: "PRIVATE KEY", Bytes: keyBytes})
+
+	return tls.X509KeyPair(certPEM, keyPEM)
+}

pkg/chatmessagesstorage/load.go

@@ -33,6 +33,7 @@ func (s *ChatMessagesStorage) loadLocked(ctx context.Context) (_err error) {
 	if err != nil {
 		return fmt.Errorf("unable to parse file '%s': %w", s.FilePath, err)
 	}
+	logger.Debugf(ctx, "loaded %d messages", len(s.Messages))
 	s.sortAndDeduplicateAndTruncate(ctx)
 	return nil
 }

pkg/ringbuffer/ring_buffer.go

@@ -0,0 +1,39 @@
+package ringbuffer
+
+import (
+	"context"
+	"slices"
+
+	"github.com/xaionaro-go/xsync"
+)
+
+type RingBuffer[T comparable] struct {
+	Storage           []T
+	CurrentWriteIndex uint
+	Locker            xsync.Mutex
+}
+
+func New[T comparable](size uint) *RingBuffer[T] {
+	return &RingBuffer[T]{
+		Storage: make([]T, 0, size),
+	}
+}
+
+func (r *RingBuffer[T]) Add(item T) {
+	r.Locker.Do(context.TODO(), func() {
+		if r.CurrentWriteIndex >= uint(len(r.Storage)) {
+			r.Storage = r.Storage[:len(r.Storage)+1]
+		}
+		r.Storage[r.CurrentWriteIndex] = item
+		r.CurrentWriteIndex++
+		if r.CurrentWriteIndex >= uint(cap(r.Storage)) {
+			r.CurrentWriteIndex = 0
+		}
+	})
+}
+
+func (r *RingBuffer[T]) Contains(item T) bool {
+	return xsync.DoR1(context.TODO(), &r.Locker, func() bool {
+		return slices.Contains(r.Storage, item)
+	})
+}

pkg/streamcontrol/kick/kick.go

@@ -11,6 +11,7 @@ import (
 
 	http "github.com/Danny-Dasilva/fhttp"
 	"github.com/davecgh/go-spew/spew"
+	"github.com/facebookincubator/go-belt"
 	"github.com/facebookincubator/go-belt/tool/logger"
 	"github.com/google/uuid"
 	"github.com/scorfly/gokick"
@@ -64,6 +65,7 @@ func New(
 	cfg Config,
 	saveCfgFn func(Config) error,
 ) (*Kick, error) {
+	ctx = belt.WithField(ctx, "controller", ID)
 	if cfg.Config.Channel == "" {
 		return nil, fmt.Errorf("channel is not set")
 	}