Add support for the upstream echo-message capability

This adds support for upstream echo-message. This capability is
enabled when the upstream supports labeled-response.

When it is enabled, we don't echo downstream messages in the downstream
handler, but rather wait for the upstream to echo it, to produce it to
downstreams.

When it is disabled, we keep the same behaviour as before: produce the
message to all downstreams as soon as it is received from the
downstream.

In other words, the main functional difference is that when the upstream
supports labeled-response, the client will now receive an echo for its
messages when the server acknowledges them, rather than when soju acks
them.

Additionally, uc.produce was refactored to take an ID rather than a
downstream.

downstream.go

@@ -2527,27 +2527,32 @@ func (dc *downstreamConn) handleMessageRegistered(ctx context.Context, msg *irc.
 				Params:  upstreamParams,
 			})
 
-			echoParams := []string{upstreamName}
+			// If the upstream supports echo message, we'll produce the message
-			if msg.Command != "TAGMSG" {
+			// when it is echoed from the upstream.
-				echoParams = append(echoParams, text)
+			// Otherwise, produce/log it here because it's the last time we'll see it.
-			}
+			if !uc.caps.IsEnabled("echo-message") {
+				echoParams := []string{upstreamName}
+				if msg.Command != "TAGMSG" {
+					echoParams = append(echoParams, text)
+				}
 
-			echoTags := tags.Copy()
+				echoTags := tags.Copy()
-			echoTags["time"] = irc.TagValue(formatServerTime(time.Now()))
+				echoTags["time"] = irc.TagValue(formatServerTime(time.Now()))
-			if uc.account != "" {
+				if uc.account != "" {
-				echoTags["account"] = irc.TagValue(uc.account)
+					echoTags["account"] = irc.TagValue(uc.account)
+				}
+				echoMsg := &irc.Message{
+					Tags: echoTags,
+					Prefix: &irc.Prefix{
+						Name: uc.nick,
+						User: uc.username,
+						Host: uc.hostname,
+					},
+					Command: msg.Command,
+					Params:  echoParams,
+				}
+				uc.produce(upstreamName, echoMsg, dc.id)
 			}
-			echoMsg := &irc.Message{
-				Tags: echoTags,
-				Prefix: &irc.Prefix{
-					Name: uc.nick,
-					User: uc.username,
-					Host: uc.hostname,
-				},
-				Command: msg.Command,
-				Params:  echoParams,
-			}
-			uc.produce(upstreamName, echoMsg, dc)
 
 			uc.updateChannelAutoDetach(upstreamName)
 		}

upstream.go

@@ -484,15 +484,17 @@ func (uc *upstreamConn) handleMessage(ctx context.Context, msg *irc.Message) err
 		}
 
 		if msg.Prefix.User == "" && msg.Prefix.Host == "" { // server message
-			uc.produce("", msg, nil)
+			uc.produce("", msg, 0)
 		} else { // regular user message
 			target := entity
 			if uc.isOurNick(target) {
 				target = msg.Prefix.Name
 			}
 
+			self := uc.isOurNick(msg.Prefix.Name)
+
 			ch := uc.network.channels.Value(target)
-			if ch != nil && msg.Command != "TAGMSG" {
+			if ch != nil && msg.Command != "TAGMSG" && !self {
 				if ch.Detached {
 					uc.handleDetachedMessage(ctx, ch, msg)
 				}
@@ -503,7 +505,7 @@ func (uc *upstreamConn) handleMessage(ctx context.Context, msg *irc.Message) err
 				}
 			}
 
-			uc.produce(target, msg, nil)
+			uc.produce(target, msg, downstreamID)
 		}
 	case "CAP":
 		var subCmd string
@@ -526,7 +528,7 @@ func (uc *upstreamConn) handleMessage(ctx context.Context, msg *irc.Message) err
 				break // wait to receive all capabilities
 			}
 
-			uc.requestCaps(ctx)
+			uc.updateCaps(ctx)
 
 			if uc.requestSASL() {
 				break // we'll send CAP END after authentication is completed
@@ -563,7 +565,7 @@ func (uc *upstreamConn) handleMessage(ctx context.Context, msg *irc.Message) err
 				return newNeedMoreParamsError(msg.Command)
 			}
 			uc.handleSupportedCaps(subParams[0])
-			uc.requestCaps(ctx)
+			uc.updateCaps(ctx)
 		case "DEL":
 			if len(subParams) < 1 {
 				return newNeedMoreParamsError(msg.Command)
@@ -1011,7 +1013,7 @@ func (uc *upstreamConn) handleMessage(ctx context.Context, msg *irc.Message) err
 
 			chMsg := msg.Copy()
 			chMsg.Params[0] = ch
-			uc.produce(ch, chMsg, nil)
+			uc.produce(ch, chMsg, 0)
 		}
 	case "PART":
 		var channels string
@@ -1037,7 +1039,7 @@ func (uc *upstreamConn) handleMessage(ctx context.Context, msg *irc.Message) err
 
 			chMsg := msg.Copy()
 			chMsg.Params[0] = ch
-			uc.produce(ch, chMsg, nil)
+			uc.produce(ch, chMsg, 0)
 		}
 	case "KICK":
 		var channel, user string
@@ -1056,7 +1058,7 @@ func (uc *upstreamConn) handleMessage(ctx context.Context, msg *irc.Message) err
 			ch.Members.Delete(user)
 		}
 
-		uc.produce(channel, msg, nil)
+		uc.produce(channel, msg, 0)
 	case "QUIT":
 		if uc.isOurNick(msg.Prefix.Name) {
 			uc.logger.Printf("quit")
@@ -1106,7 +1108,7 @@ func (uc *upstreamConn) handleMessage(ctx context.Context, msg *irc.Message) err
 		} else {
 			ch.Topic = ""
 		}
-		uc.produce(ch.Name, msg, nil)
+		uc.produce(ch.Name, msg, 0)
 	case "MODE":
 		var name, modeStr string
 		if err := parseMessageParams(msg, &name, &modeStr); err != nil {
@@ -1851,7 +1853,7 @@ func (uc *upstreamConn) handleSupportedCaps(capsStr string) {
 	}
 }
 
-func (uc *upstreamConn) requestCaps(ctx context.Context) {
+func (uc *upstreamConn) updateCaps(ctx context.Context) {
 	var requestCaps []string
 	for c := range permanentUpstreamCaps {
 		if uc.caps.IsAvailable(c) && !uc.caps.IsEnabled(c) {
@@ -1859,6 +1861,13 @@ func (uc *upstreamConn) requestCaps(ctx context.Context) {
 		}
 	}
 
+	echoMessage := uc.caps.IsAvailable("labeled-response")
+	if !uc.caps.IsEnabled("echo-message") && echoMessage {
+		requestCaps = append(requestCaps, "echo-message")
+	} else if uc.caps.IsEnabled("echo-message") && !echoMessage {
+		requestCaps = append(requestCaps, "-echo-message")
+	}
+
 	if len(requestCaps) == 0 {
 		return
 	}
@@ -1924,6 +1933,7 @@ func (uc *upstreamConn) handleCapAck(ctx context.Context, name string, ok bool)
 			Command: "AUTHENTICATE",
 			Params:  []string{auth.Mechanism},
 		})
+	case "echo-message":
 	default:
 		if permanentUpstreamCaps[name] {
 			break
@@ -2089,9 +2099,10 @@ func (uc *upstreamConn) appendLog(entity string, msg *irc.Message) (msgID string
 // produce appends a message to the logs and forwards it to connected downstream
 // connections.
 //
-// If origin is not nil and origin doesn't support echo-message, the message is
+// originID is the id of the downstream (origin) that sent the message. If it is not 0
-// forwarded to all connections except origin.
+// and origin doesn't support echo-message, the message is forwarded to all
-func (uc *upstreamConn) produce(target string, msg *irc.Message, origin *downstreamConn) {
+// connections except origin.
+func (uc *upstreamConn) produce(target string, msg *irc.Message, originID uint64) {
 	var msgID string
 	if target != "" {
 		msgID = uc.appendLog(target, msg)
@@ -2102,7 +2113,7 @@ func (uc *upstreamConn) produce(target string, msg *irc.Message, origin *downstr
 	detached := ch != nil && ch.Detached
 
 	uc.forEachDownstream(func(dc *downstreamConn) {
-		if !detached && (dc != origin || dc.caps.IsEnabled("echo-message")) {
+		if !detached && (dc.id != originID || dc.caps.IsEnabled("echo-message")) {
 			dc.sendMessageWithID(dc.marshalMessage(msg, uc.network), msgID)
 		} else {
 			dc.advanceMessageWithID(msg, msgID)