neoirc/internal/ircserver/relay_test.go

package ircserver_test

import (
	"context"
	"encoding/json"
	"slices"
	"strings"
	"testing"
	"time"

	"sneak.berlin/go/neoirc/internal/ircserver"
	"sneak.berlin/go/neoirc/internal/service"
)

// queueMessage queues a message for the user with nick
// recipient the way the HTTP API sends one: stored once, then
// queued to the recipient's session.
func (env *testEnv) queueMessage(
	t *testing.T,
	recipient, command, from, to string,
	body []string,
) {
	t.Helper()

	ctx := t.Context()

	bodyJSON, err := json.Marshal(body)
	if err != nil {
		t.Fatalf("marshal body: %v", err)
	}

	msgID, _, err := env.database.InsertMessage(
		ctx, command, from, to, 0, 0, nil, bodyJSON, nil,
	)
	if err != nil {
		t.Fatalf("insert message: %v", err)
	}

	sessionID, err := env.database.GetSessionByNick(ctx, recipient)
	if err != nil {
		t.Fatalf("get session of %s: %v", recipient, err)
	}

	err = env.database.EnqueueToSession(ctx, sessionID, msgID, 0)
	if err != nil {
		t.Fatalf("enqueue: %v", err)
	}

	env.brk.Notify(sessionID)
}

// TestRelayKeepsOneWaitChannel checks that an IRC listener
// connection holds at most one broker wait channel across
// several relay ticks, both while idle and after the relay
// has delivered it a message, and none once it closes.
func TestRelayKeepsOneWaitChannel(t *testing.T) {
	t.Parallel()

	env := newTestEnv(t)

	client := env.dial(t)
	client.register("idle")

	sessionID, err := env.database.GetSessionByNick(t.Context(), "idle")
	if err != nil {
		t.Fatalf("get session of idle: %v", err)
	}

	env.checkOneWaitChannel(t, sessionID)

	env.queueMessage(
		t, "idle", "PRIVMSG", "carol", "idle", []string{"hello"},
	)

	client.readUntil(func(l string) bool {
		return strings.Contains(l, "hello")
	})

	env.checkOneWaitChannel(t, sessionID)

	err = client.conn.Close()
	if err != nil {
		t.Fatalf("close: %v", err)
	}

	deadline := time.Now().Add(testTimeout)

	for env.brk.ListenerCount(sessionID) != 0 {
		if time.Now().After(deadline) {
			t.Fatalf("after close: got %d wait channels, want 0",
				env.brk.ListenerCount(sessionID))
		}

		time.Sleep(relayCheckStep)
	}
}

// checkOneWaitChannel fails the test if the session
// sessionID holds more than one broker wait channel at any of
// several relay ticks.
func (env *testEnv) checkOneWaitChannel(
	t *testing.T,
	sessionID int64,
) {
	t.Helper()

	const ticks = 5

	for tick := range ticks {
		time.Sleep(ircserver.PollInterval)

		got := env.brk.ListenerCount(sessionID)
		if got > 1 {
			t.Fatalf("tick %d: got %d wait channels, want at most 1",
				tick+1, got)
		}
	}
}

// TestRelayMultiLineBody checks that each line of a PRIVMSG
// or NOTICE body sent over the HTTP API reaches an IRC
// listener user as its own line.
func TestRelayMultiLineBody(t *testing.T) {
	t.Parallel()

	env := newTestEnv(t)

	alice := env.dial(t)
	alice.register("alice")

	for _, command := range []string{"PRIVMSG", "NOTICE"} {
		env.queueMessage(
			t, "alice", command, "carol", "alice",
			[]string{"first line", "second line"},
		)

		lines := alice.readUntil(func(l string) bool {
			return strings.Contains(l, "second line")
		})

		want := []string{
			":carol!carol@* " + command + " alice :first line",
			":carol!carol@* " + command + " alice :second line",
		}

		got := lines[max(0, len(lines)-len(want)):]
		if !slices.Equal(got, want) {
			t.Errorf("%s: got %q, want %q", command, got, want)
		}
	}
}

// TestRelayModeParameters checks that a MODE change reaches
// IRC listener members with its values as separate
// parameters, both for the bodies the HTTP API stores and
// for a change made on the IRC listener.
func TestRelayModeParameters(t *testing.T) {
	t.Parallel()

	env := newTestEnv(t)

	alice := env.dial(t)
	alice.register("alice")

	for _, step := range []struct {
		body []string
		want string
	}{
		{[]string{"+k", "secret"}, "+k secret"},
		{[]string{"-k", "*"}, "-k *"},
		{[]string{"+l", "50"}, "+l 50"},
		{[]string{"-l"}, "-l"},
		{[]string{"+b", "*!*@*.example.com"}, "+b *!*@*.example.com"},
		{[]string{"-b", "*!*@*.example.com"}, "-b *!*@*.example.com"},
		{[]string{"+o bob"}, "+o bob"},
		{[]string{"+m"}, "+m"},
	} {
		env.queueMessage(t, "alice", "MODE", "op", "#modes", step.body)

		lines := alice.readUntil(func(l string) bool {
			return strings.Contains(l, " MODE ")
		})

		got := lines[len(lines)-1]
		want := ":op!op@* MODE #modes " + step.want

		if got != want {
			t.Errorf("body %q: got %q, want %q", step.body, got, want)
		}
	}

	bob := env.dial(t)
	bob.register("bob")

	alice.joinAndDrain("#modes")
	bob.joinAndDrain("#modes")

	alice.send("MODE #modes +v bob")

	const change = " MODE #modes +v bob"

	for name, client := range map[string]*testClient{
		"alice": alice, "bob": bob,
	} {
		lines := client.readUntil(func(l string) bool {
			return strings.Contains(l, " MODE ")
		})

		got := lines[len(lines)-1]
		if !strings.HasSuffix(got, change) {
			t.Errorf("%s: got %q, want it to end in %q", name, got, change)
		}
	}
}

// TestRelayInviteWithoutChannel checks that an INVITE whose
// body names no channel sends the invited IRC listener user
// no INVITE line.
func TestRelayInviteWithoutChannel(t *testing.T) {
	t.Parallel()

	env := newTestEnv(t)

	bob := env.dial(t)
	bob.register("bob")

	// A nil body is stored as JSON null, which is not an array.
	for _, body := range [][]string{nil, {}, {""}} {
		env.queueMessage(t, "bob", "INVITE", "carol", "bob", body)
	}

	// The relay sends a user's queue in order, so this arrives
	// after anything sent for the INVITEs above.
	env.queueMessage(t, "bob", "PRIVMSG", "carol", "bob",
		[]string{"the end"})

	lines := bob.readUntil(func(l string) bool {
		return strings.HasSuffix(l, " :the end")
	})

	for _, line := range lines {
		if strings.Contains(line, " INVITE ") {
			t.Errorf("got %q, want no INVITE line", line)
		}
	}
}

// TestRelayLineBreaks checks that a CR, LF or NUL in a stored
// body line, key or mask never reaches an IRC listener user
// as a line break, so no extra line arrives, and that a body
// line keeps its parameter and leading ':' once they are
// removed.
func TestRelayLineBreaks(t *testing.T) {
	t.Parallel()

	env := newTestEnv(t)

	bob := env.dial(t)
	bob.register("bob")

	env.queueMessage(t, "bob", "PRIVMSG", "carol", "bob", []string{
		"one\r\n:admin!a@b PRIVMSG bob :hi",
		"two\n:admin!a@b NOTICE bob :hi\x00",
	})
	env.queueMessage(t, "bob", "MODE", "op", "#breaks",
		[]string{"+k", "key\r\nERROR :Closing link"})
	env.queueMessage(t, "bob", "MODE", "op", "#breaks",
		[]string{"+b", "*!*@*\r\n:admin!a@b KICK #breaks bob"})
	env.queueMessage(t, "bob", "PRIVMSG", "carol", "bob",
		[]string{"\r\n", "\x00:hi"})
	env.queueMessage(t, "bob", "PRIVMSG", "carol", "bob",
		[]string{"the end"})

	lines := bob.readUntil(func(l string) bool {
		return strings.HasSuffix(l, " :the end")
	})

	want := []string{
		":carol!carol@* PRIVMSG bob :one:admin!a@b PRIVMSG bob :hi",
		":carol!carol@* PRIVMSG bob :two:admin!a@b NOTICE bob :hi",
		":op!op@* MODE #breaks +k :keyERROR :Closing link",
		":op!op@* MODE #breaks +b :*!*@*:admin!a@b KICK #breaks bob",
		":carol!carol@* PRIVMSG bob :",
		":carol!carol@* PRIVMSG bob ::hi",
		":carol!carol@* PRIVMSG bob :the end",
	}

	got := lines[max(0, len(lines)-len(want)):]
	if !slices.Equal(got, want) {
		t.Errorf("got %q, want %q", got, want)
	}
}

// TestRelayNickDiffersOnlyInCase checks that an IRC listener
// user alice receives a JOIN, PRIVMSG, PART, NICK and QUIT
// from Alice in a channel they share: nicks are
// case-sensitive, so Alice is someone else.
func TestRelayNickDiffersOnlyInCase(t *testing.T) {
	t.Parallel()

	env := newTestEnv(t)

	alice := env.dial(t)
	alice.register("alice")
	alice.joinAndDrain("#case")

	capitalAlice := env.dial(t)
	capitalAlice.register("Alice")

	for _, step := range []struct {
		line, want string
	}{
		{"JOIN #case", ":Alice!Alice@* JOIN #case"},
		{"PRIVMSG #case :hi there", ":Alice!Alice@* PRIVMSG #case :hi there"},
		{"PART #case :see you", ":Alice!Alice@* PART #case :see you"},
		{"JOIN #case", ":Alice!Alice@* JOIN #case"},
		{"NICK Alicia", ":Alice!Alice@* NICK Alicia"},
		{"NICK Alice", ":Alicia!Alicia@* NICK Alice"}, // back for QUIT
		{"QUIT", ":Alice!Alice@* QUIT :Quit: Connection closed"},
	} {
		capitalAlice.send(step.line)

		got := alice.readLine()
		if got != step.want {
			t.Errorf("%s: got %q, want %q", step.line, got, step.want)
		}
	}
}

// TestRelayOwnChangeOnce checks that an IRC listener user who
// sets a topic, changes a channel mode or kicks gets one TOPIC,
// MODE or KICK line for it, and so does the other member.
func TestRelayOwnChangeOnce(t *testing.T) {
	t.Parallel()

	for _, step := range []struct {
		command, line string
	}{
		{"TOPIC", "TOPIC #dup :new topic"},
		{"MODE", "MODE #dup +m"},
		{"KICK", "KICK #dup carl :bye"},
	} {
		t.Run(step.command, func(t *testing.T) {
			t.Parallel()

			env := newTestEnv(t)

			alice := env.dial(t)
			alice.register("alice")
			alice.joinAndDrain("#dup")

			carl := env.dial(t)
			carl.register("carl")
			carl.joinAndDrain("#dup")

			alice.send(step.line)

			// The reply to alice's next command comes after her
			// command has finished, so her own copy comes before it.
			beforePong := alice.sendAndExpect("PING :done", "PONG")

			// The relay sends a user's queue in order, so a message
			// queued now arrives after any relayed copy.
			for nick, client := range map[string]*testClient{
				"alice": alice, "carl": carl,
			} {
				env.queueMessage(
					t, nick, "PRIVMSG", "dave", nick,
					[]string{"after the change"},
				)

				lines := client.readUntil(func(l string) bool {
					return strings.HasSuffix(l, " :after the change")
				})

				if client == alice {
					lines = slices.Concat(beforePong, lines)
				}

				var changes []string

				for _, line := range lines {
					if strings.Contains(line, " "+step.command+" #dup ") {
						changes = append(changes, line)
					}
				}

				if len(changes) != 1 {
					t.Errorf("%s got %q, want one %s line",
						nick, changes, step.command)
				}
			}
		})
	}
}

// TestRelayOwnSessionOnce checks that a command a user makes
// from one client of their session reaches each of the
// session's clients once: an IRC listener connection and an
// HTTP API client. The connection that sent a command gets
// only its own echo, and none for PRIVMSG and NOTICE, which
// IRC does not echo.
//
//nolint:funlen // table-driven test
func TestRelayOwnSessionOnce(t *testing.T) {
	t.Parallel()

	hello := json.RawMessage(`["hello"]`)

	for _, step := range []struct {
		// line is the command as alice sends it on the IRC
		// listener.
		line string
		// api makes the same command for her session the way
		// the HTTP API does.
		api func(
			ctx context.Context, svc *service.Service, sessionID int64,
		) error
		// echoes is how many copies alice's connection gets of
		// line when it sends line itself.
		echoes int
	}{
		{"PRIVMSG #chan :hello", func(
			ctx context.Context, svc *service.Service, sessionID int64,
		) error {
			_, err := svc.SendChannelMessage(
				ctx, sessionID, 0, "alice", "PRIVMSG", "#chan", hello, nil,
			)

			return err
		}, 0},
		{"NOTICE bob :hello", func(
			ctx context.Context, svc *service.Service, sessionID int64,
		) error {
			_, err := svc.SendDirectMessage(
				ctx, sessionID, 0, "alice", "NOTICE", "bob", hello, nil,
			)

			return err
		}, 0},
		{"JOIN #other", func(
			ctx context.Context, svc *service.Service, sessionID int64,
		) error {
			_, err := svc.JoinChannel(ctx, sessionID, 0, "alice", "#other", "")

			return err
		}, 1},
		{"PART #chan :bye", func(
			ctx context.Context, svc *service.Service, sessionID int64,
		) error {
			return svc.PartChannel(ctx, sessionID, 0, "alice", "#chan", "bye")
		}, 1},
		{"NICK alice2", func(
			ctx context.Context, svc *service.Service, sessionID int64,
		) error {
			return svc.ChangeNick(ctx, sessionID, 0, "alice", "alice2")
		}, 1},
		{"TOPIC #chan :new topic", func(
			ctx context.Context, svc *service.Service, sessionID int64,
		) error {
			return svc.SetTopic(
				ctx, sessionID, 0, "alice", "#chan", "new topic",
			)
		}, 1},
		{"MODE #chan +m", func(
			ctx context.Context, svc *service.Service, sessionID int64,
		) error {
			chID, err := svc.ValidateChannelOp(ctx, sessionID, "#chan")
			if err != nil {
				return err
			}

			err = svc.SetChannelFlag(ctx, chID, 'm', true)
			if err != nil {
				return err
			}

			svc.BroadcastMode(ctx, 0, "alice", "#chan", chID, "+m")

			return nil
		}, 1},
		{"KICK #chan bob :bye", func(
			ctx context.Context, svc *service.Service, sessionID int64,
		) error {
			return svc.KickUser(
				ctx, sessionID, 0, "alice", "#chan", "bob", "bye",
			)
		}, 1},
	} {
		command := strings.Fields(step.line)[0]

		t.Run(command+" from IRC", func(t *testing.T) {
			t.Parallel()

			ircCopies, httpCopies := ownSessionCopies(t, step.line, nil)
			if ircCopies != step.echoes || httpCopies != 1 {
				t.Errorf("connection got %d, HTTP client got %d, want %d and 1",
					ircCopies, httpCopies, step.echoes)
			}
		})

		t.Run(command+" from the HTTP API", func(t *testing.T) {
			t.Parallel()

			ircCopies, httpCopies := ownSessionCopies(t, step.line, step.api)
			if ircCopies != 1 || httpCopies != 1 {
				t.Errorf("connection got %d, HTTP client got %d, want 1 and 1",
					ircCopies, httpCopies)
			}
		})
	}
}

// ownSessionCopies sets up alice with aliceWithHTTPClient. It
// makes the command line from her connection, or with api from
// the HTTP API client when api is not nil, and returns how many
// copies of it her connection and her HTTP API client got.
func ownSessionCopies(
	t *testing.T,
	line string,
	api func(ctx context.Context, svc *service.Service, sessionID int64) error,
) (int, int) {
	t.Helper()

	ctx := t.Context()
	env := newTestEnv(t)
	alice, _, sessionID, httpClientID := aliceWithHTTPClient(t, env)

	var lines []string

	if api == nil {
		alice.send(line)

		// Her echo comes before the reply to her next command.
		lines = alice.sendAndExpect("PING :done", "PONG")
	} else {
		err := api(ctx, env.srv.Service(), sessionID)
		if err != nil {
			t.Fatalf("%s over the HTTP API: %v", line, err)
		}
	}

	msgs, _, err := env.database.PollMessages(ctx, httpClientID, 0, 0)
	if err != nil {
		t.Fatalf("poll the HTTP API client: %v", err)
	}

	httpCopies := 0

	for _, msg := range msgs {
		if msg.Command == strings.Fields(line)[0] {
			httpCopies++
		}
	}

	// The relay sends a queue in order, so this arrives after any
	// copy. It is queued by session, as NICK changes her nick.
	endID, _, err := env.database.InsertMessage(
		ctx, "PRIVMSG", "dave", "alice", 0, 0,
		nil, json.RawMessage(`["the end"]`), nil,
	)
	if err != nil {
		t.Fatalf("insert message: %v", err)
	}

	err = env.database.EnqueueToSession(ctx, sessionID, endID, 0)
	if err != nil {
		t.Fatalf("enqueue: %v", err)
	}

	env.brk.Notify(sessionID)

	lines = append(lines, alice.readUntil(func(l string) bool {
		return strings.HasSuffix(l, " :the end")
	})...)

	// A copy starts with the command and its first parameter.
	start := " " + strings.Join(strings.Fields(line)[:2], " ")
	ircCopies := strings.Count(strings.Join(lines, "\n"), start)

	return ircCopies, httpCopies
}

// aliceWithHTTPClient connects alice to the IRC listener and
// has her join #chan, which bob is on too, then adds an HTTP API
// client to her session with the password her connection set,
// as POST /api/v1/login does. It returns her connection, bob's
// connection, her session and her HTTP API client.
func aliceWithHTTPClient(
	t *testing.T, env *testEnv,
) (*testClient, *testClient, int64, int64) {
	t.Helper()

	const password = "alicepass"

	alice := env.dial(t)
	alice.send("PASS " + password)
	alice.register("alice")
	// Registration stores the password before the JOIN runs.
	alice.joinAndDrain("#chan")

	bob := env.dial(t)
	bob.register("bob")
	bob.joinAndDrain("#chan")

	sessionID, httpClientID, _, err := env.database.LoginUser(
		t.Context(), "alice", password, "127.0.0.1", "localhost",
	)
	if err != nil {
		t.Fatalf("log in as alice: %v", err)
	}

	return alice, bob, sessionID, httpClientID
}

// TestRelayNickChangeFromHTTPAPI checks that once alice's nick
// is changed from her HTTP API client, her IRC listener
// connection uses the new nick: what she then sends from it
// reaches bob under the new nick.
func TestRelayNickChangeFromHTTPAPI(t *testing.T) {
	t.Parallel()

	env := newTestEnv(t)
	alice, bob, sessionID, _ := aliceWithHTTPClient(t, env)

	bob.joinAndDrain("#other")

	// As the HTTP API does.
	err := env.srv.Service().ChangeNick(
		t.Context(), sessionID, 0, "alice", "alice2",
	)
	if err != nil {
		t.Fatalf("change nick over the HTTP API: %v", err)
	}

	// Her connection has taken the new nick once it has sent her
	// the change.
	const change = ":alice!alice@* NICK alice2"

	alice.readUntil(func(l string) bool { return l == change })
	bob.readUntil(func(l string) bool { return l == change })

	for _, step := range []struct {
		line, want string
	}{
		{"PRIVMSG #chan :hi there", ":alice2!alice2@* PRIVMSG #chan :hi there"},
		{"JOIN #other", ":alice2!alice2@* JOIN #other"},
		{"NICK alice3", ":alice2!alice2@* NICK alice3"},
	} {
		alice.send(step.line)

		got := bob.readLine()
		if got != step.want {
			t.Errorf("%s: got %q, want %q", step.line, got, step.want)
		}
	}
}