feat: add irc-backed streamer split view

2026-04-29 00:40:35 +02:00
parent 01b044e305
commit ad1f5ec3ff
8 changed files with 1500 additions and 42 deletions
@@ -0,0 +1,252 @@
+// Package irc manages Twitch IRC chat connections for watched streamers.
+package irc
+
+import (
+	"bufio"
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"net"
+	"strings"
+	"sync"
+	"time"
+)
+
+const DefaultAddr = "irc.chat.twitch.tv:6667"
+
+// ConnectionState describes the lifecycle state of one IRC connection.
+type ConnectionState string
+
+const (
+	StatePending      ConnectionState = "pending"
+	StateJoined       ConnectionState = "joined"
+	StateDisconnected ConnectionState = "disconnected"
+)
+
+// Event carries one IRC lifecycle change or log line for a streamer connection.
+type Event struct {
+	Streamer string
+	State    ConnectionState
+	Line     string
+}
+
+// EventSink consumes IRC events emitted by a client or manager.
+type EventSink func(Event)
+
+// Client maintains one Twitch IRC connection for a single streamer channel.
+type Client struct {
+	Addr        string
+	Login       string
+	Token       string
+	Streamer    string
+	Once        bool
+	Debug       bool
+	Out         io.Writer
+	Events      EventSink
+	DialContext func(context.Context, string, string) (net.Conn, error)
+}
+
+// Run connects, authenticates, joins the configured streamer channel, and stays connected.
+func (c *Client) Run(ctx context.Context) error {
+	if c.Login == "" {
+		return errors.New("missing Twitch login")
+	}
+	if c.Token == "" {
+		return errors.New("missing OAuth access token")
+	}
+	if c.Streamer == "" {
+		return errors.New("missing streamer")
+	}
+
+	addr := c.Addr
+	if addr == "" {
+		addr = DefaultAddr
+	}
+
+	c.status("Connecting to %s\n", addr)
+	conn, err := c.dial(ctx, addr)
+	if err != nil {
+		return fmt.Errorf("connect to Twitch IRC: %w", err)
+	}
+
+	session := &session{
+		conn:  conn,
+		out:   c.Out,
+		debug: c.Debug,
+	}
+	defer session.close()
+
+	done := make(chan struct{})
+	defer close(done)
+	go func() {
+		select {
+		case <-ctx.Done():
+			_ = session.send("QUIT :parasocial shutting down", false)
+			session.close()
+		case <-done:
+		}
+	}()
+
+	if err := session.send("PASS oauth:"+c.Token, true); err != nil {
+		return err
+	}
+	c.status("Sent PASS\n")
+	if err := session.send("NICK "+c.Login, false); err != nil {
+		return err
+	}
+	c.status("Sent NICK %s\n", c.Login)
+	if err := session.send("JOIN #"+c.Streamer, false); err != nil {
+		return err
+	}
+	c.status("Sent JOIN #%s\n", c.Streamer)
+
+	reader := bufio.NewReader(conn)
+	joined := false
+	for {
+		line, err := reader.ReadString('\n')
+		if err != nil {
+			if ctx.Err() != nil {
+				return nil
+			}
+			if joined {
+				return fmt.Errorf("network interruption after join: %w", err)
+			}
+			return fmt.Errorf("IRC connection closed before join confirmation: %w", err)
+		}
+
+		line = strings.TrimRight(line, "\r\n")
+		session.debugIn(line)
+
+		if payload, ok := pingPayload(line); ok {
+			if err := session.send("PONG "+payload, false); err != nil {
+				return err
+			}
+			c.status("Responded to PING\n")
+			continue
+		}
+
+		if isAuthFailure(line) {
+			return fmt.Errorf("IRC authentication failure: %s", line)
+		}
+		if isJoinDenied(line, c.Streamer) {
+			return fmt.Errorf("IRC join denied: %s", line)
+		}
+		if isWelcome(line) {
+			c.status("IRC authentication accepted\n")
+			continue
+		}
+		if isJoinConfirmation(line, c.Login, c.Streamer) {
+			joinedLine := fmt.Sprintf("Joined #%s as %s", c.Streamer, c.Login)
+			c.status("%s\n", joinedLine)
+			c.emit(StateJoined, joinedLine)
+			joined = true
+			if c.Once {
+				if err := session.send("QUIT :parasocial --once complete", false); err != nil {
+					return err
+				}
+				return nil
+			}
+			continue
+		}
+		if joined {
+			c.status("%s\n", line)
+			c.emit("", line)
+		}
+	}
+}
+
+type session struct {
+	conn  net.Conn
+	out   io.Writer
+	debug bool
+	mu    sync.Mutex
+	once  sync.Once
+}
+
+func (s *session) send(line string, redact bool) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	debugLine := line
+	if redact {
+		debugLine = "PASS oauth:<redacted>"
+	}
+	if s.debug && s.out != nil {
+		fmt.Fprintf(s.out, "> %s\n", debugLine)
+	}
+	_, err := fmt.Fprintf(s.conn, "%s\r\n", line)
+	return err
+}
+
+func (s *session) debugIn(line string) {
+	if s.debug && s.out != nil {
+		fmt.Fprintf(s.out, "< %s\n", line)
+	}
+}
+
+func (s *session) close() {
+	s.once.Do(func() {
+		_ = s.conn.SetDeadline(time.Now())
+		_ = s.conn.Close()
+	})
+}
+
+func (c *Client) dial(ctx context.Context, addr string) (net.Conn, error) {
+	if c.DialContext != nil {
+		return c.DialContext(ctx, "tcp", addr)
+	}
+	dialer := &net.Dialer{Timeout: 30 * time.Second}
+	return dialer.DialContext(ctx, "tcp", addr)
+}
+
+func (c *Client) status(format string, args ...any) {
+	if c.Out == nil {
+		return
+	}
+	fmt.Fprintf(c.Out, format, args...)
+}
+
+func (c *Client) emit(state ConnectionState, line string) {
+	if c.Events == nil {
+		return
+	}
+	c.Events(Event{
+		Streamer: c.Streamer,
+		State:    state,
+		Line:     line,
+	})
+}
+
+func pingPayload(line string) (string, bool) {
+	if strings.HasPrefix(line, "PING ") {
+		return strings.TrimSpace(strings.TrimPrefix(line, "PING ")), true
+	}
+	return "", false
+}
+
+func isWelcome(line string) bool {
+	return strings.Contains(line, " 001 ")
+}
+
+func isAuthFailure(line string) bool {
+	lower := strings.ToLower(line)
+	return strings.Contains(lower, "login authentication failed") ||
+		strings.Contains(lower, "improperly formatted auth") ||
+		strings.Contains(lower, "error logging in")
+}
+
+func isJoinDenied(line, streamer string) bool {
+	lower := strings.ToLower(line)
+	channel := "#" + strings.ToLower(streamer)
+	return (strings.Contains(lower, " 403 ") && strings.Contains(lower, channel)) ||
+		(strings.Contains(lower, "notice "+channel) && strings.Contains(lower, "banned")) ||
+		(strings.Contains(lower, "notice "+channel) && strings.Contains(lower, "not permitted"))
+}
+
+func isJoinConfirmation(line, login, streamer string) bool {
+	lower := strings.ToLower(line)
+	channel := "#" + strings.ToLower(streamer)
+	return strings.Contains(lower, " join "+channel) &&
+		(strings.HasPrefix(lower, ":"+strings.ToLower(login)+"!") || strings.Contains(lower, "!"))
+}
@@ -0,0 +1,223 @@
+package irc
+
+import (
+	"bufio"
+	"context"
+	"io"
+	"net"
+	"reflect"
+	"strings"
+	"testing"
+	"time"
+)
+
+func TestRunOnceSendsAuthJoinAndRespondsToPing(t *testing.T) {
+	t.Parallel()
+
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer listener.Close()
+
+	lines := make(chan string, 10)
+	serverErr := make(chan error, 1)
+
+	go func() {
+		conn, err := listener.Accept()
+		if err != nil {
+			serverErr <- err
+			return
+		}
+		defer conn.Close()
+
+		reader := bufio.NewReader(conn)
+		for {
+			line, err := reader.ReadString('\n')
+			if err != nil {
+				if err == io.EOF {
+					serverErr <- nil
+					return
+				}
+				serverErr <- err
+				return
+			}
+			line = strings.TrimRight(line, "\r\n")
+			lines <- line
+			switch line {
+			case "NICK viewer":
+				_, _ = conn.Write([]byte(":tmi.twitch.tv 001 viewer :Welcome\r\n"))
+			case "JOIN #streamer":
+				_, _ = conn.Write([]byte("PING :tmi.twitch.tv\r\n"))
+			case "PONG :tmi.twitch.tv":
+				_, _ = conn.Write([]byte(":viewer!viewer@viewer.tmi.twitch.tv JOIN #streamer\r\n"))
+			case "QUIT :parasocial --once complete":
+				serverErr <- nil
+				return
+			}
+		}
+	}()
+
+	client := &Client{
+		Addr:     listener.Addr().String(),
+		Login:    "viewer",
+		Token:    "token",
+		Streamer: "streamer",
+		Once:     true,
+		Out:      io.Discard,
+	}
+	if err := client.Run(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+
+	want := []string{
+		"PASS oauth:token",
+		"NICK viewer",
+		"JOIN #streamer",
+		"PONG :tmi.twitch.tv",
+		"QUIT :parasocial --once complete",
+	}
+	for _, expected := range want {
+		select {
+		case got := <-lines:
+			if got != expected {
+				t.Fatalf("line = %q, want %q", got, expected)
+			}
+		case <-time.After(time.Second):
+			t.Fatalf("timed out waiting for %q", expected)
+		}
+	}
+
+	select {
+	case err := <-serverErr:
+		if err != nil {
+			t.Fatal(err)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("server did not finish")
+	}
+}
+
+func TestRunEmitsJoinAndPostJoinEvents(t *testing.T) {
+	t.Parallel()
+
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer listener.Close()
+
+	events := make(chan Event, 4)
+	serverErr := make(chan error, 1)
+
+	go func() {
+		conn, err := listener.Accept()
+		if err != nil {
+			serverErr <- err
+			return
+		}
+		defer conn.Close()
+
+		reader := bufio.NewReader(conn)
+		for {
+			line, err := reader.ReadString('\n')
+			if err != nil {
+				serverErr <- err
+				return
+			}
+			line = strings.TrimRight(line, "\r\n")
+			switch line {
+			case "NICK viewer":
+				_, _ = conn.Write([]byte(":tmi.twitch.tv 001 viewer :Welcome\r\n"))
+			case "JOIN #streamer":
+				_, _ = conn.Write([]byte(":viewer!viewer@viewer.tmi.twitch.tv JOIN #streamer\r\n"))
+				_, _ = conn.Write([]byte(":someone!someone@someone.tmi.twitch.tv PRIVMSG #streamer :hello\r\n"))
+				_, _ = conn.Write([]byte("PING :tmi.twitch.tv\r\n"))
+			case "PONG :tmi.twitch.tv":
+				serverErr <- nil
+				return
+			}
+		}
+	}()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	client := &Client{
+		Addr:     listener.Addr().String(),
+		Login:    "viewer",
+		Token:    "token",
+		Streamer: "streamer",
+		Events: func(event Event) {
+			events <- event
+		},
+		Out: io.Discard,
+	}
+
+	done := make(chan error, 1)
+	go func() {
+		done <- client.Run(ctx)
+	}()
+
+	var got []Event
+	for len(got) < 2 {
+		select {
+		case event := <-events:
+			got = append(got, event)
+			if event.Line == ":someone!someone@someone.tmi.twitch.tv PRIVMSG #streamer :hello" {
+				cancel()
+			}
+		case <-time.After(time.Second):
+			t.Fatal("timed out waiting for IRC events")
+		}
+	}
+
+	if err := <-done; err != nil {
+		t.Fatal(err)
+	}
+	if err := <-serverErr; err != nil && err != io.EOF {
+		t.Fatal(err)
+	}
+
+	want := []Event{
+		{Streamer: "streamer", State: StateJoined, Line: "Joined #streamer as viewer"},
+		{Streamer: "streamer", Line: ":someone!someone@someone.tmi.twitch.tv PRIVMSG #streamer :hello"},
+	}
+	if !reflect.DeepEqual(got[:2], want) {
+		t.Fatalf("events = %#v, want %#v", got[:2], want)
+	}
+}
+
+func TestAuthFailureReturnsError(t *testing.T) {
+	t.Parallel()
+
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer listener.Close()
+
+	go func() {
+		conn, err := listener.Accept()
+		if err != nil {
+			return
+		}
+		defer conn.Close()
+		_, _ = conn.Write([]byte(":tmi.twitch.tv NOTICE * :Login authentication failed\r\n"))
+	}()
+
+	client := &Client{
+		Addr:     listener.Addr().String(),
+		Login:    "viewer",
+		Token:    "bad-token",
+		Streamer: "streamer",
+		Out:      io.Discard,
+	}
+	err = client.Run(context.Background())
+	if err == nil {
+		t.Fatal("expected auth failure")
+	}
+	if !strings.Contains(err.Error(), "authentication failure") {
+		t.Fatalf("err = %v", err)
+	}
+}
@@ -0,0 +1,146 @@
+package irc
+
+import (
+	"context"
+	"io"
+	"net"
+	"strings"
+	"sync"
+)
+
+// Target identifies one streamer channel that should have an active IRC connection.
+type Target struct {
+	Login string
+}
+
+// Manager reconciles the active IRC connections against the desired watched set.
+type Manager struct {
+	Addr        string
+	Debug       bool
+	Out         io.Writer
+	Events      EventSink
+	DialContext func(context.Context, string, string) (net.Conn, error)
+	RunClient   func(context.Context, *Client) error
+
+	mu     sync.Mutex
+	active map[string]*managedClient
+	wg     sync.WaitGroup
+}
+
+type managedClient struct {
+	cancel context.CancelFunc
+}
+
+// Sync applies the desired IRC target set using the provided viewer credentials.
+func (m *Manager) Sync(ctx context.Context, viewerLogin, accessToken string, targets []Target) {
+	orderedTargets, desired := normalizeTargets(targets)
+
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	if m.active == nil {
+		m.active = make(map[string]*managedClient)
+	}
+
+	for login, client := range m.active {
+		if _, ok := desired[login]; ok {
+			continue
+		}
+		client.cancel()
+		delete(m.active, login)
+	}
+
+	if viewerLogin == "" || accessToken == "" {
+		return
+	}
+
+	for _, login := range orderedTargets {
+		if _, ok := m.active[login]; ok {
+			continue
+		}
+
+		runCtx, cancel := context.WithCancel(ctx)
+		client := &managedClient{cancel: cancel}
+		m.active[login] = client
+
+		m.wg.Add(1)
+		go func(streamer string, current *managedClient) {
+			defer m.wg.Done()
+			m.runClient(runCtx, viewerLogin, accessToken, streamer, current)
+		}(login, client)
+	}
+}
+
+func (m *Manager) runClient(ctx context.Context, viewerLogin, accessToken, streamer string, client *managedClient) {
+	run := m.RunClient
+	if run == nil {
+		run = func(ctx context.Context, conn *Client) error {
+			return conn.Run(ctx)
+		}
+	}
+
+	m.emit(Event{Streamer: streamer, State: StatePending})
+
+	_ = run(ctx, &Client{
+		Addr:        m.addr(),
+		Login:       viewerLogin,
+		Token:       accessToken,
+		Streamer:    streamer,
+		Debug:       m.Debug,
+		Out:         m.Out,
+		Events:      m.Events,
+		DialContext: m.DialContext,
+	})
+	m.emit(Event{Streamer: streamer, State: StateDisconnected})
+
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if current, ok := m.active[streamer]; ok && current == client {
+		delete(m.active, streamer)
+	}
+}
+
+// Close cancels every active connection and waits for their goroutines to exit.
+func (m *Manager) Close() {
+	m.mu.Lock()
+	for login, client := range m.active {
+		client.cancel()
+		delete(m.active, login)
+	}
+	m.mu.Unlock()
+	m.wg.Wait()
+}
+
+func (m *Manager) addr() string {
+	if m.Addr != "" {
+		return m.Addr
+	}
+	return DefaultAddr
+}
+
+func (m *Manager) emit(event Event) {
+	if m.Events == nil {
+		return
+	}
+	m.Events(event)
+}
+
+func normalizeTargets(targets []Target) ([]string, map[string]struct{}) {
+	ordered := make([]string, 0, 2)
+	normalized := make(map[string]struct{}, len(targets))
+	for _, target := range targets {
+		login := strings.ToLower(strings.TrimSpace(target.Login))
+		if login == "" {
+			continue
+		}
+		if _, ok := normalized[login]; ok {
+			continue
+		}
+		normalized[login] = struct{}{}
+		ordered = append(ordered, login)
+		if len(normalized) == 2 {
+			break
+		}
+	}
+	return ordered, normalized
+}
@@ -0,0 +1,223 @@
+package irc
+
+import (
+	"context"
+	"errors"
+	"reflect"
+	"sort"
+	"sync"
+	"testing"
+	"time"
+)
+
+func TestManagerSyncStartsAtMostTwoTargets(t *testing.T) {
+	t.Parallel()
+
+	started := make(chan string, 3)
+	manager := &Manager{
+		RunClient: func(ctx context.Context, client *Client) error {
+			started <- client.Streamer
+			<-ctx.Done()
+			return ctx.Err()
+		},
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	manager.Sync(ctx, "viewer", "token", []Target{
+		{Login: "alpha"},
+		{Login: "beta"},
+		{Login: "gamma"},
+	})
+
+	assertStartSet(t, started, "alpha", "beta")
+	assertNoStart(t, started)
+}
+
+func TestManagerSyncKeepsExistingTargetsConnected(t *testing.T) {
+	t.Parallel()
+
+	started := make(chan string, 3)
+	manager := &Manager{
+		RunClient: func(ctx context.Context, client *Client) error {
+			started <- client.Streamer
+			<-ctx.Done()
+			return ctx.Err()
+		},
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	manager.Sync(ctx, "viewer", "token", []Target{{Login: "alpha"}})
+	assertStartSet(t, started, "alpha")
+
+	manager.Sync(ctx, "viewer", "token", []Target{{Login: "alpha"}})
+	assertNoStart(t, started)
+}
+
+func TestManagerSyncCancelsRemovedTargets(t *testing.T) {
+	t.Parallel()
+
+	started := make(chan string, 4)
+	canceled := make(chan string, 4)
+	manager := &Manager{
+		RunClient: func(ctx context.Context, client *Client) error {
+			started <- client.Streamer
+			<-ctx.Done()
+			canceled <- client.Streamer
+			return ctx.Err()
+		},
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	manager.Sync(ctx, "viewer", "token", []Target{{Login: "alpha"}, {Login: "beta"}})
+	assertStartSet(t, started, "alpha", "beta")
+
+	manager.Sync(ctx, "viewer", "token", []Target{{Login: "beta"}, {Login: "gamma"}})
+	assertStartSet(t, started, "gamma")
+	assertCancels(t, canceled, "alpha")
+	assertNoStart(t, started)
+}
+
+func TestManagerSyncRestartsTargetAfterFailure(t *testing.T) {
+	t.Parallel()
+
+	started := make(chan string, 2)
+	var mu sync.Mutex
+	attempts := map[string]int{}
+
+	manager := &Manager{
+		RunClient: func(ctx context.Context, client *Client) error {
+			mu.Lock()
+			attempts[client.Streamer]++
+			attempt := attempts[client.Streamer]
+			mu.Unlock()
+
+			started <- client.Streamer
+			if attempt == 1 {
+				return errors.New("boom")
+			}
+			<-ctx.Done()
+			return ctx.Err()
+		},
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	manager.Sync(ctx, "viewer", "token", []Target{{Login: "alpha"}})
+	assertStartSet(t, started, "alpha")
+
+	waitFor(t, func() bool {
+		manager.mu.Lock()
+		defer manager.mu.Unlock()
+		_, ok := manager.active["alpha"]
+		return !ok
+	})
+
+	manager.Sync(ctx, "viewer", "token", []Target{{Login: "alpha"}})
+	assertStartSet(t, started, "alpha")
+}
+
+func TestManagerEmitsPendingAndDisconnectedEvents(t *testing.T) {
+	t.Parallel()
+
+	events := make(chan Event, 4)
+	manager := &Manager{
+		Events: func(event Event) {
+			events <- event
+		},
+		RunClient: func(ctx context.Context, client *Client) error {
+			<-ctx.Done()
+			return ctx.Err()
+		},
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	manager.Sync(ctx, "viewer", "token", []Target{{Login: "alpha"}})
+	cancel()
+	manager.Close()
+
+	var got []Event
+	for len(got) < 2 {
+		select {
+		case event := <-events:
+			got = append(got, event)
+		case <-time.After(time.Second):
+			t.Fatal("timed out waiting for manager events")
+		}
+	}
+
+	want := []Event{
+		{Streamer: "alpha", State: StatePending},
+		{Streamer: "alpha", State: StateDisconnected},
+	}
+	if !reflect.DeepEqual(got[:2], want) {
+		t.Fatalf("events = %#v, want %#v", got[:2], want)
+	}
+}
+
+func assertStartSet(t *testing.T, started <-chan string, want ...string) {
+	t.Helper()
+
+	got := make([]string, 0, len(want))
+	for range want {
+		select {
+		case login := <-started:
+			got = append(got, login)
+		case <-time.After(time.Second):
+			t.Fatalf("timed out waiting for %d starts", len(want))
+		}
+	}
+
+	sort.Strings(got)
+	sortedWant := append([]string(nil), want...)
+	sort.Strings(sortedWant)
+	for index := range sortedWant {
+		if got[index] != sortedWant[index] {
+			t.Fatalf("starts = %#v, want %#v", got, sortedWant)
+		}
+	}
+}
+
+func assertCancels(t *testing.T, canceled <-chan string, want ...string) {
+	t.Helper()
+
+	for _, expected := range want {
+		select {
+		case got := <-canceled:
+			if got != expected {
+				t.Fatalf("cancel = %q, want %q", got, expected)
+			}
+		case <-time.After(time.Second):
+			t.Fatalf("timed out waiting for cancel %q", expected)
+		}
+	}
+}
+
+func assertNoStart(t *testing.T, started <-chan string) {
+	t.Helper()
+
+	select {
+	case got := <-started:
+		t.Fatalf("unexpected start %q", got)
+	case <-time.After(100 * time.Millisecond):
+	}
+}
+
+func waitFor(t *testing.T, condition func() bool) {
+	t.Helper()
+
+	deadline := time.Now().Add(time.Second)
+	for time.Now().Before(deadline) {
+		if condition() {
+			return
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	t.Fatal("condition not met before timeout")
+}