Show an IRC listener user what their session sends from other clients (closes #179)

The relay dropped every PRIVMSG, NOTICE, JOIN, PART, NICK, QUIT, TOPIC,
MODE and KICK sent under the connection's own nick, so an IRC listener
connection never saw what its session did from another client. The
relay now drops nothing by nick. Instead, the service methods that fan
out a user's own command take the client whose queue to skip: the IRC
listener passes its own client, which has already written the echo, and
the HTTP API passes 0. Channel messages and PARTs now also go to the
sender's own session, so the HTTP handler no longer queues the
channel-message echo itself and SendChannelMessage no longer returns
the message's row ID.

Model: opus-5-5

internal/ircserver/commands.go

@@ -84,7 +84,7 @@ func (c *Conn) handleNick(
 	oldNick := c.nick
 
 	err := c.svc.ChangeNick(
-		ctx, c.sessionID, oldNick, newNick,
+		ctx, c.sessionID, c.clientID, oldNick, newNick,
 	)
 	if err != nil {
 		c.sendIRCError(err)
@@ -128,8 +128,8 @@ func (c *Conn) handlePrivmsg(
 	body, _ := json.Marshal([]string{text}) //nolint:errchkjson
 
 	if strings.HasPrefix(target, "#") {
-		_, _, err := c.svc.SendChannelMessage(
-			ctx, c.sessionID, c.nick,
+		_, err := c.svc.SendChannelMessage(
+			ctx, c.sessionID, c.clientID, c.nick,
 			msg.Command, target, body, nil,
 		)
 		if err != nil {
@@ -137,7 +137,7 @@ func (c *Conn) handlePrivmsg(
 		}
 	} else {
 		result, err := c.svc.SendDirectMessage(
-			ctx, c.sessionID, c.nick,
+			ctx, c.sessionID, c.clientID, c.nick,
 			msg.Command, target, body, nil,
 		)
 		if err != nil {
@@ -199,7 +199,7 @@ func (c *Conn) joinChannel(
 	channel, key string,
 ) {
 	result, err := c.svc.JoinChannel(
-		ctx, c.sessionID, c.nick, channel, key,
+		ctx, c.sessionID, c.clientID, c.nick, channel, key,
 	)
 	if err != nil {
 		c.sendIRCError(err)
@@ -337,7 +337,7 @@ func (c *Conn) partChannel(
 	channel, reason string,
 ) {
 	err := c.svc.PartChannel(
-		ctx, c.sessionID, c.nick, channel, reason,
+		ctx, c.sessionID, c.clientID, c.nick, channel, reason,
 	)
 	if err != nil {
 		c.sendIRCError(err)
@@ -401,7 +401,7 @@ func (c *Conn) handleTopic(
 	newTopic := msg.Params[1]
 
 	err := c.svc.SetTopic(
-		ctx, c.sessionID, c.nick, channel, newTopic,
+		ctx, c.sessionID, c.clientID, c.nick, channel, newTopic,
 	)
 	if err != nil {
 		c.sendIRCError(err)
@@ -708,7 +708,7 @@ func (c *Conn) applyChannelModes(
 		c.send(FormatMessage(c.hostmask(), "MODE", params...))
 
 		c.svc.BroadcastMode(
-			ctx, c.nick, channel, chID, modeReply,
+			ctx, c.clientID, c.nick, channel, chID, modeReply,
 		)
 	}
 }
@@ -1185,7 +1185,7 @@ func (c *Conn) handleKick(
 	}
 
 	err := c.svc.KickUser(
-		ctx, c.sessionID, c.nick,
+		ctx, c.sessionID, c.clientID, c.nick,
 		channel, targetNick, reason,
 	)
 	if err != nil {
@@ -1267,7 +1267,7 @@ func (c *Conn) handleInvite(
 
 	_, _, _ = c.svc.FanOut( //nolint:dogsled // fire-and-forget broadcast
 		ctx, "INVITE", c.nick, targetNick, targetID, 0,
-		nil, body, nil, []int64{targetID},
+		nil, body, nil, []int64{targetID}, 0,
 	)
 }
 

internal/ircserver/export_test.go

@@ -71,6 +71,11 @@ func (s *Server) Listener() net.Listener {
 	return s.listener
 }
 
+// Service returns the service the server's connections use.
+func (s *Server) Service() *service.Service {
+	return s.svc
+}
+
 // PingTimes returns how long the server's connections wait
 // with nothing read before they are sent a PING, and after it
 // before they are closed.

internal/ircserver/relay.go

@@ -198,15 +198,6 @@ func (c *Conn) deliverTextMessage(
 	from := msg.From
 	target := msg.To
 
-	c.mu.Lock()
-	nick := c.nick
-	c.mu.Unlock()
-
-	// Don't echo our own messages back.
-	if from == nick {
-		return
-	}
-
 	prefix := from
 	if !strings.Contains(prefix, "!") {
 		prefix = from + "!" + from + "@*"
@@ -219,16 +210,6 @@ func (c *Conn) deliverTextMessage(
 
 // deliverJoin sends a JOIN notification.
 func (c *Conn) deliverJoin(msg *db.IRCMessage) {
-	c.mu.Lock()
-	nick := c.nick
-	c.mu.Unlock()
-
-	// Don't echo our own JOINs (we already sent them
-	// during joinChannel).
-	if msg.From == nick {
-		return
-	}
-
 	prefix := msg.From + "!" + msg.From + "@*"
 	channel := msg.To
 
@@ -237,14 +218,6 @@ func (c *Conn) deliverJoin(msg *db.IRCMessage) {
 
 // deliverPart sends a PART notification.
 func (c *Conn) deliverPart(msg *db.IRCMessage, text string) {
-	c.mu.Lock()
-	nick := c.nick
-	c.mu.Unlock()
-
-	if msg.From == nick {
-		return
-	}
-
 	prefix := msg.From + "!" + msg.From + "@*"
 	channel := msg.To
 
@@ -262,14 +235,6 @@ func (c *Conn) deliverNickChange(
 	msg *db.IRCMessage,
 	newNick string,
 ) {
-	c.mu.Lock()
-	nick := c.nick
-	c.mu.Unlock()
-
-	if msg.From == nick {
-		return
-	}
-
 	prefix := msg.From + "!" + msg.From + "@*"
 
 	c.send(FormatMessage(prefix, "NICK", newNick))
@@ -280,14 +245,6 @@ func (c *Conn) deliverQuitMsg(
 	msg *db.IRCMessage,
 	text string,
 ) {
-	c.mu.Lock()
-	nick := c.nick
-	c.mu.Unlock()
-
-	if msg.From == nick {
-		return
-	}
-
 	prefix := msg.From + "!" + msg.From + "@*"
 
 	if text != "" {
@@ -304,15 +261,6 @@ func (c *Conn) deliverTopicChange(
 	msg *db.IRCMessage,
 	text string,
 ) {
-	c.mu.Lock()
-	nick := c.nick
-	c.mu.Unlock()
-
-	// handleTopic already echoed our own topic change.
-	if msg.From == nick {
-		return
-	}
-
 	prefix := msg.From + "!" + msg.From + "@*"
 	channel := msg.To
 
@@ -324,15 +272,6 @@ func (c *Conn) deliverKickMsg(
 	msg *db.IRCMessage,
 	text string,
 ) {
-	c.mu.Lock()
-	nick := c.nick
-	c.mu.Unlock()
-
-	// handleKick already echoed our own kick.
-	if msg.From == nick {
-		return
-	}
-
 	prefix := msg.From + "!" + msg.From + "@*"
 	channel := msg.To
 
@@ -403,15 +342,6 @@ func (c *Conn) deliverMode(
 		return
 	}
 
-	c.mu.Lock()
-	nick := c.nick
-	c.mu.Unlock()
-
-	// applyChannelModes already echoed our own mode change.
-	if msg.From == nick {
-		return
-	}
-
 	prefix := msg.From + "!" + msg.From + "@*"
 
 	params := slices.Concat(

internal/ircserver/relay_test.go

@@ -1,6 +1,7 @@
 package ircserver_test
 
 import (
+	"context"
 	"encoding/json"
 	"slices"
 	"strings"
@@ -8,6 +9,7 @@ import (
 	"time"
 
 	"sneak.berlin/go/neoirc/internal/ircserver"
+	"sneak.berlin/go/neoirc/internal/service"
 )
 
 // queueMessage queues a message for the user with nick
@@ -39,7 +41,7 @@ func (env *testEnv) queueMessage(
 		t.Fatalf("get session of %s: %v", recipient, err)
 	}
 
-	err = env.database.EnqueueToSession(ctx, sessionID, msgID)
+	err = env.database.EnqueueToSession(ctx, sessionID, msgID, 0)
 	if err != nil {
 		t.Fatalf("enqueue: %v", err)
 	}
@@ -391,3 +393,221 @@ func TestRelayOwnChangeOnce(t *testing.T) {
 		})
 	}
 }
+
+// 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 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 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()
+
+	const password = "alicepass"
+
+	ctx := t.Context()
+	env := newTestEnv(t)
+
+	alice := env.dial(t)
+	alice.send("PASS " + password)
+	alice.register("alice")
+	// Registration has stored the password by the time the
+	// JOIN is handled.
+	alice.joinAndDrain("#chan")
+
+	bob := env.dial(t)
+	bob.register("bob")
+	bob.joinAndDrain("#chan")
+
+	sessionID, httpClientID, _, err := env.database.LoginUser(
+		ctx, "alice", password, "127.0.0.1", "localhost",
+	)
+	if err != nil {
+		t.Fatalf("log in as alice: %v", err)
+	}
+
+	var lines []string
+
+	if api == nil {
+		alice.send(line)
+
+		// The reply to alice's next command comes after her
+		// command has finished.
+		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 a message queued now
+	// reaches alice's connection after any copy of the command.
+	// It is queued by session, as the command may change 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 := 0
+
+	for _, l := range lines {
+		if strings.Contains(l, start) {
+			ircCopies++
+		}
+	}
+
+	return ircCopies, httpCopies
+}