user.go

package soju

import (
	"context"
	"crypto/sha256"
	"encoding/binary"
	"encoding/hex"
	"errors"
	"fmt"
	"math/big"
	"net"
	"sort"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/SherClockHolmes/webpush-go"
	"gopkg.in/irc.v4"

	"codeberg.org/emersion/soju/database"
	"codeberg.org/emersion/soju/msgstore"
	"codeberg.org/emersion/soju/xirc"
)

type UserUpdateFunc func(record *database.User) error

type event interface{}

type eventUpstreamMessage struct {
	msg *irc.Message
	uc  *upstreamConn
}

type eventUpstreamConnectionError struct {
	net *network
	err error
}

type eventUpstreamConnected struct {
	uc *upstreamConn
}

type eventUpstreamDisconnected struct {
	uc *upstreamConn
}

type eventUpstreamError struct {
	uc  *upstreamConn
	err error
}

type eventDownstreamMessage struct {
	msg *irc.Message
	dc  *downstreamConn
}

type eventDownstreamConnected struct {
	dc *downstreamConn
}

type eventDownstreamDisconnected struct {
	dc *downstreamConn
}

type eventChannelDetach struct {
	uc   *upstreamConn
	name string
}

type eventBroadcast struct {
	msg *irc.Message
}

type eventStop struct{}

type eventUserUpdate struct {
	password    *string
	admin       *bool
	enabled     *bool
	maxNetworks *int
	done        chan error
}

type eventTryRegainNick struct {
	uc   *upstreamConn
	nick string
}

type eventUserRun struct {
	params []string
	ch     chan userRunMsg
}

type deliveredClientMap map[string]string // client name -> msg ID

type deliveredStore struct {
	m xirc.CaseMappingMap[deliveredClientMap]
}

func newDeliveredStore(cm xirc.CaseMapping) deliveredStore {
	return deliveredStore{xirc.NewCaseMappingMap[deliveredClientMap](cm)}
}

func (ds deliveredStore) Empty() bool {
	return ds.m.Len() == 0
}

func (ds deliveredStore) HasTarget(target string) bool {
	return ds.m.Get(target) != nil
}

func (ds deliveredStore) LoadID(target, clientName string) string {
	clients := ds.m.Get(target)
	if clients == nil {
		return ""
	}
	return clients[clientName]
}

func (ds deliveredStore) StoreID(target, clientName, msgID string) {
	clients := ds.m.Get(target)
	if clients == nil {
		clients = make(deliveredClientMap)
		ds.m.Set(target, clients)
	}
	clients[clientName] = msgID
}

func (ds deliveredStore) ForEachTarget(f func(target string)) {
	ds.m.ForEach(func(name string, _ deliveredClientMap) {
		f(name)
	})
}

func (ds deliveredStore) ForEachClient(f func(clientName string)) {
	clients := make(map[string]struct{})
	ds.m.ForEach(func(name string, delivered deliveredClientMap) {
		for clientName := range delivered {
			clients[clientName] = struct{}{}
		}
	})

	for clientName := range clients {
		f(clientName)
	}
}

type network struct {
	database.Network
	user    *user
	logger  Logger
	stopped chan struct{}

	conn      *upstreamConn
	channels  xirc.CaseMappingMap[*database.Channel]
	delivered deliveredStore
	lastError error
	casemap   xirc.CaseMapping

	pushTargetsMutex sync.Mutex
	pushTargets      xirc.CaseMappingMap[time.Time]
}

func newNetwork(user *user, record *database.Network, channels []database.Channel) *network {
	logger := &prefixLogger{user.logger, fmt.Sprintf("network %q: ", record.GetName())}

	// Initialize maps with the most strict case-mapping to avoid collisions:
	// we don't know which case-mapping will be used by the upstream server yet
	cm := xirc.CaseMappingASCII

	m := xirc.NewCaseMappingMap[*database.Channel](cm)
	for _, ch := range channels {
		ch := ch
		m.Set(ch.Name, &ch)
	}

	return &network{
		Network:     *record,
		user:        user,
		logger:      logger,
		stopped:     make(chan struct{}),
		channels:    m,
		delivered:   newDeliveredStore(cm),
		pushTargets: xirc.NewCaseMappingMap[time.Time](cm),
		casemap:     stdCaseMapping,
	}
}

func (net *network) forEachDownstream(f func(*downstreamConn)) {
	for _, dc := range net.user.downstreamConns {
		if dc.network != net {
			continue
		}
		f(dc)
	}
}

func (net *network) isStopped() bool {
	select {
	case <-net.stopped:
		return true
	default:
		return false
	}
}

func (net *network) equalCasemap(a, b string) bool {
	return net.casemap(a) == net.casemap(b)
}

func userIdent(u *database.User) string {
	// The ident is a string we will send to upstream servers in clear-text.
	// For privacy reasons, make sure it doesn't expose any meaningful user
	// metadata. We just use the base64-encoded hashed ID, so that people don't
	// start relying on the string being an integer or following a pattern.
	var b [64]byte
	binary.LittleEndian.PutUint64(b[:], uint64(u.ID))
	h := sha256.Sum256(b[:])
	return hex.EncodeToString(h[:16])
}

func (net *network) runConn(ctx context.Context) error {
	net.user.srv.metrics.upstreams.Add(1)
	defer net.user.srv.metrics.upstreams.Add(-1)

	ctx, cancel := context.WithCancel(ctx)
	defer cancel()
	done := ctx.Done() // This must not be subject to timeout
	ctx, cancelTimeout := context.WithTimeout(ctx, time.Minute)
	defer cancelTimeout()

	uc, err := connectToUpstream(ctx, net)
	if err != nil {
		return fmt.Errorf("failed to connect: %w", err)
	}
	defer uc.Close()

	// The context is cancelled by the caller when the network is stopped.
	go func() {
		<-done
		uc.Close()
	}()

	if net.user.srv.Identd != nil {
		net.user.srv.Identd.Store(uc.RemoteAddr().String(), uc.LocalAddr().String(), userIdent(&net.user.User))
		defer net.user.srv.Identd.Delete(uc.RemoteAddr().String(), uc.LocalAddr().String())
	}

	// TODO: this is racy, we're not running in the user goroutine yet
	// uc.register accesses user/network DB records
	uc.register(ctx)
	if err := uc.runUntilRegistered(ctx); err != nil {
		return fmt.Errorf("failed to register: %w", err)
	}

	for _, command := range uc.network.ConnectCommands {
		m, err := irc.ParseMessage(command)
		if err != nil {
			uc.logger.Printf("failed to parse connect command %q: %v", command, err)
		} else {
			uc.SendMessage(ctx, m)
		}
	}

	// Some channels require the user to be logged in to be joined
	// which may need to happen asynchronously either via connect
	// commands or CertFP without SASL EXTERNAL on networks that
	// don't support this. Work around these old networks by
	// leaving 3 seconds for the authentication to be processed.
	var delay bool

	if len(uc.network.ConnectCommands) > 0 {
		uc.logger.Debugf("sent %d connect command(s)", len(uc.network.ConnectCommands))
		// Connect commands usually send a message to NickServ or similar, which
		// asynchronously informs the IRC server that the user is logged in.
		delay = true
	}

	if uc.network.SASL.Mechanism == "EXTERNAL" && !uc.supportsSASL("EXTERNAL") {
		uc.logger.Debugf("client certificate configured but the upstream doesn't advertise SASL EXTERNAL, delaying joining channels")
		delay = true
	}

	if delay {
		select {
		case <-ctx.Done():
		case <-time.After(3 * time.Second):
		}
	}

	if uc.network.channels.Len() > 0 {
		var channels, keys []string
		uc.network.channels.ForEach(func(_ string, ch *database.Channel) {
			channels = append(channels, ch.Name)
			keys = append(keys, ch.Key)
		})

		for _, msg := range xirc.GenerateJoin(channels, keys) {
			uc.SendMessage(ctx, msg)
		}
	}

	net.user.events <- eventUpstreamConnected{uc}
	defer func() {
		net.user.events <- eventUpstreamDisconnected{uc}
	}()

	if err := uc.readMessages(net.user.events); err != nil {
		return fmt.Errorf("failed to handle messages: %w", err)
	}

	return nil
}

func (net *network) run() {
	if !net.user.Enabled || !net.Enabled {
		return
	}

	ctx, cancel := context.WithCancel(context.TODO())
	go func() {
		<-net.stopped
		cancel()
	}()

	var lastTry time.Time
	backoff := newBackoffer(retryConnectMinDelay, retryConnectMaxDelay, retryConnectJitter)
	for {
		if net.isStopped() {
			return
		}

		delay := backoff.Next() - time.Now().Sub(lastTry)
		if delay > 0 {
			net.logger.Printf("waiting %v before trying to reconnect to %q", delay.Truncate(time.Second), net.Addr)
			time.Sleep(delay)
		}
		lastTry = time.Now()

		if err := net.runConn(ctx); err != nil {
			text := err.Error()
			temp := true
			var regErr registrationError
			if errors.As(err, &regErr) {
				text = "failed to register: " + regErr.Reason()
				temp = regErr.Temporary()
			}

			net.logger.Printf("connection error to %q: %v", net.Addr, text)
			net.user.events <- eventUpstreamConnectionError{net, fmt.Errorf("connection error: %v", err)}
			net.user.srv.metrics.upstreamConnectErrorsTotal.Inc()

			if !temp {
				return
			}
		} else {
			backoff.Reset()
		}
	}
}

func (net *network) stop() {
	if !net.isStopped() {
		close(net.stopped)
	}
}

func (net *network) detach(ctx context.Context, ch *database.Channel) {
	if ch.Detached {
		return
	}

	net.logger.Printf("detaching channel %q", ch.Name)

	ch.Detached = true

	if net.user.msgStore != nil {
		nameCM := net.casemap(ch.Name)
		lastID, err := net.user.msgStore.LastMsgID(ctx, &net.Network, nameCM, time.Now())
		if err != nil {
			net.logger.Printf("failed to get last message ID for channel %q: %v", ch.Name, err)
		}
		ch.DetachedInternalMsgID = lastID
	}

	if net.conn != nil {
		uch := net.conn.channels.Get(ch.Name)
		if uch != nil {
			uch.updateAutoDetach(0)
		}
	}

	net.forEachDownstream(func(dc *downstreamConn) {
		dc.SendMessage(ctx, &irc.Message{
			Prefix:  dc.prefix(),
			Command: "PART",
			Params:  []string{ch.Name, "Detach"},
		})
	})
}

func (net *network) attach(ctx context.Context, ch *database.Channel) {
	if !ch.Detached {
		return
	}

	net.logger.Printf("attaching channel %q", ch.Name)

	detachedMsgID := ch.DetachedInternalMsgID
	ch.Detached = false
	ch.DetachedInternalMsgID = ""

	var uch *upstreamChannel
	if net.conn != nil {
		uch = net.conn.channels.Get(ch.Name)

		net.conn.updateChannelAutoDetach(ch.Name)
	}

	net.forEachDownstream(func(dc *downstreamConn) {
		dc.SendMessage(ctx, &irc.Message{
			Prefix:  dc.prefix(),
			Command: "JOIN",
			Params:  []string{ch.Name},
		})

		if uch != nil {
			forwardChannel(ctx, dc, uch)
		}

		if detachedMsgID != "" {
			dc.sendTargetBacklog(ctx, net, ch.Name, detachedMsgID)
		}
	})
}

func (net *network) deleteChannel(ctx context.Context, name string) error {
	ch := net.channels.Get(name)
	if ch == nil {
		return fmt.Errorf("unknown channel %q", name)
	}
	if net.conn != nil {
		uch := net.conn.channels.Get(ch.Name)
		if uch != nil {
			uch.updateAutoDetach(0)
		}
	}

	if err := net.user.srv.db.DeleteChannel(ctx, ch.ID); err != nil {
		return err
	}
	net.channels.Del(name)
	return nil
}

func (net *network) updateCasemapping(newCasemap xirc.CaseMapping) {
	net.casemap = newCasemap
	net.channels.SetCaseMapping(newCasemap)
	net.delivered.m.SetCaseMapping(newCasemap)
	net.pushTargetsMutex.Lock()
	net.pushTargets.SetCaseMapping(newCasemap)
	net.pushTargetsMutex.Unlock()
	if uc := net.conn; uc != nil {
		uc.channels.SetCaseMapping(newCasemap)
		uc.channels.ForEach(func(_ string, uch *upstreamChannel) {
			uch.Members.SetCaseMapping(newCasemap)
		})
		uc.users.SetCaseMapping(newCasemap)
		uc.monitored.SetCaseMapping(newCasemap)
	}
	net.forEachDownstream(func(dc *downstreamConn) {
		dc.updateCasemapping()
	})
}

func (net *network) storeClientDeliveryReceipts(ctx context.Context, clientName string) {
	if !net.user.hasPersistentMsgStore() {
		return
	}

	var receipts []database.DeliveryReceipt
	net.delivered.ForEachTarget(func(target string) {
		msgID := net.delivered.LoadID(target, clientName)
		if msgID == "" {
			return
		}
		receipts = append(receipts, database.DeliveryReceipt{
			Target:        target,
			InternalMsgID: msgID,
		})
	})

	if err := net.user.srv.db.StoreClientDeliveryReceipts(ctx, net.ID, clientName, receipts); err != nil {
		net.logger.Printf("failed to store delivery receipts for client %q: %v", clientName, err)
	}
}

func (net *network) isHighlight(msg *irc.Message) bool {
	if msg.Command != "PRIVMSG" && msg.Command != "NOTICE" {
		return false
	}

	text := msg.Params[1]

	nick := database.GetNick(&net.user.User, &net.Network)
	if net.conn != nil {
		nick = net.conn.nick
	}

	// TODO: use case-mapping aware comparison here
	return msg.Prefix.Name != nick && isHighlight(text, nick)
}

func (net *network) detachedMessageNeedsRelay(ch *database.Channel, msg *irc.Message) bool {
	highlight := net.isHighlight(msg)
	return ch.RelayDetached == database.FilterMessage || ((ch.RelayDetached == database.FilterHighlight || ch.RelayDetached == database.FilterDefault) && highlight)
}

func (net *network) autoSaveSASLPlain(ctx context.Context, username, password string) {
	// User may have e.g. EXTERNAL mechanism configured. We do not want to
	// automatically erase the key pair or any other credentials.
	if net.SASL.Mechanism != "" && net.SASL.Mechanism != "PLAIN" {
		return
	}

	net.logger.Printf("auto-saving SASL PLAIN credentials with username %q", username)
	net.SASL.Mechanism = "PLAIN"
	net.SASL.Plain.Username = username
	net.SASL.Plain.Password = password
	if err := net.user.srv.db.StoreNetwork(ctx, net.user.ID, &net.Network); err != nil {
		net.logger.Printf("failed to save SASL PLAIN credentials: %v", err)
	}
}

// broadcastWebPush broadcasts a Web Push message for the given IRC message.
//
// Broadcasting the message to all Web Push endpoints might take a while, so
// callers should call this function in a new goroutine.
func (net *network) broadcastWebPush(msg *irc.Message) {
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
	defer cancel()

	subs, err := net.user.srv.db.ListWebPushSubscriptions(ctx, net.user.ID, net.ID)
	if err != nil {
		net.logger.Printf("failed to list Web push subscriptions: %v", err)
		return
	}

	for _, sub := range subs {
		err := net.user.srv.sendWebPush(ctx, &webpush.Subscription{
			Endpoint: sub.Endpoint,
			Keys: webpush.Keys{
				Auth:   sub.Keys.Auth,
				P256dh: sub.Keys.P256DH,
			},
		}, sub.Keys.VAPID, msg)
		if err == errWebPushSubscriptionExpired {
			if err := net.user.srv.db.DeleteWebPushSubscription(ctx, sub.ID); err != nil {
				net.logger.Printf("failed to delete expired Web Push subscription %q: %v", sub.Endpoint, err)
			} else {
				net.logger.Debugf("deleted expired Web Push subscription %q", sub.Endpoint)
			}
		} else if err != nil {
			net.logger.Printf("failed to send Web push notification to endpoint %q: %v", sub.Endpoint, err)
			// If it failed for any reason and is old, delete it
			if time.Since(sub.UpdatedAt) > webpushPruneSubscriptionDelay {
				if err := net.user.srv.db.DeleteWebPushSubscription(ctx, sub.ID); err != nil {
					net.logger.Printf("failed to delete pruned Web Push subscription %q: %v", sub.Endpoint, err)
				} else {
					net.logger.Printf("deleted pruned Web Push subscription %q", sub.Endpoint)
				}
			}
		}
	}
}

type user struct {
	database.User
	srv    *Server
	logger Logger

	events chan event
	done   chan struct{}

	numDownstreamConns atomic.Int64

	networks        []*network
	downstreamConns []*downstreamConn
	msgStore        msgstore.Store
}

func newUser(srv *Server, record *database.User) *user {
	logger := &prefixLogger{srv.Logger, fmt.Sprintf("user %q: ", record.Username)}

	var msgStore msgstore.Store
	switch srv.Config().MsgStore.Driver {
	case msgstore.DriverFS:
		msgStore = msgstore.NewFSStore(srv.Config().MsgStore.Source, record)
	case msgstore.DriverDB:
		msgStore = msgstore.NewDBStore(srv.db)
	case msgstore.DriverMemory:
		msgStore = msgstore.NewMemoryStore()
	}

	return &user{
		User:     *record,
		srv:      srv,
		logger:   logger,
		events:   make(chan event, 64),
		done:     make(chan struct{}),
		msgStore: msgStore,
	}
}

func (u *user) forEachUpstream(f func(uc *upstreamConn)) {
	for _, network := range u.networks {
		if network.conn == nil {
			continue
		}
		f(network.conn)
	}
}

func (u *user) getNetwork(name string) *network {
	for _, network := range u.networks {
		if network.Addr == name {
			return network
		}
		if network.Name != "" && network.Name == name {
			return network
		}
	}
	return nil
}

func (u *user) getNetworkByID(id int64) *network {
	for _, net := range u.networks {
		if net.ID == id {
			return net
		}
	}
	return nil
}

func (u *user) run() {
	defer func() {
		if u.msgStore != nil {
			if err := u.msgStore.Close(); err != nil {
				u.logger.Printf("failed to close message store for user %q: %v", u.Username, err)
			}
		}
		close(u.done)
	}()

	ctx, cancel := context.WithCancel(context.TODO())
	go func() {
		<-u.done
		cancel()
	}()

	networks, err := u.srv.db.ListNetworks(ctx, u.ID)
	if err != nil {
		u.logger.Printf("failed to list networks for user %q: %v", u.Username, err)
		return
	}

	sort.Slice(networks, func(i, j int) bool {
		return networks[i].ID < networks[j].ID
	})

	for _, record := range networks {
		record := record
		channels, err := u.srv.db.ListChannels(ctx, record.ID)
		if err != nil {
			u.logger.Printf("failed to list channels for user %q, network %q: %v", u.Username, record.GetName(), err)
			continue
		}

		network := newNetwork(u, &record, channels)
		u.networks = append(u.networks, network)

		if u.hasPersistentMsgStore() {
			receipts, err := u.srv.db.ListDeliveryReceipts(ctx, record.ID)
			if err != nil {
				u.logger.Printf("failed to load delivery receipts for user %q, network %q: %v", u.Username, network.GetName(), err)
				return
			}

			for _, rcpt := range receipts {
				network.delivered.StoreID(rcpt.Target, rcpt.Client, rcpt.InternalMsgID)
			}
		}

		go network.run()
	}

	for e := range u.events {
		// TODO: set a timeout for event handling

		switch e := e.(type) {
		case eventUpstreamConnected:
			uc := e.uc

			uc.network.conn = uc

			uc.updateAway(ctx)
			uc.updateMonitor(ctx)

			uc.forEachDownstream(func(dc *downstreamConn) {
				dc.updateSupportedCaps(ctx)

				if !dc.caps.IsEnabled("soju.im/bouncer-networks") {
					sendServiceNOTICE(ctx, dc, fmt.Sprintf("connected to %s", uc.network.GetName()))
				}

				dc.updateNick(ctx)
				dc.updateHost(ctx)
				dc.updateRealname(ctx)
				dc.updateAccount(ctx)
				dc.updateCasemapping()
			})
			u.notifyBouncerNetworkState(ctx, uc.network.ID, irc.Tags{
				"state": "connected",
				"error": "",
			})
			uc.network.lastError = nil
		case eventUpstreamDisconnected:
			u.handleUpstreamDisconnected(ctx, e.uc)
		case eventUpstreamConnectionError:
			net := e.net

			stopped := false
			select {
			case <-net.stopped:
				stopped = true
			default:
			}

			if !stopped && (net.lastError == nil || net.lastError.Error() != e.err.Error()) {
				net.forEachDownstream(func(dc *downstreamConn) {
					sendServiceNOTICE(ctx, dc, fmt.Sprintf("failed connecting/registering to %s: %v", net.GetName(), e.err))
				})
			}
			net.lastError = e.err
			u.notifyBouncerNetworkState(ctx, net.ID, irc.Tags{
				"error": net.lastError.Error(),
			})
		case eventUpstreamError:
			uc := e.uc

			uc.forEachDownstream(func(dc *downstreamConn) {
				sendServiceNOTICE(ctx, dc, fmt.Sprintf("disconnected from %s: %v", uc.network.GetName(), e.err))
			})
			uc.network.lastError = e.err
			u.notifyBouncerNetworkState(ctx, uc.network.ID, irc.Tags{
				"error": uc.network.lastError.Error(),
			})
		case eventUpstreamMessage:
			msg, uc := e.msg, e.uc
			if uc.isClosed() {
				uc.logger.Printf("ignoring message on closed connection: %v", msg)
				break
			}
			if err := uc.handleMessage(ctx, msg); err != nil {
				uc.logger.Printf("failed to handle message %q: %v", msg, err)
			}
		case eventChannelDetach:
			uc, name := e.uc, e.name
			c := uc.network.channels.Get(name)
			if c == nil || c.Detached {
				continue
			}
			uc.network.detach(ctx, c)
			if err := uc.srv.db.StoreChannel(ctx, uc.network.ID, c); err != nil {
				u.logger.Printf("failed to store updated detached channel %q: %v", c.Name, err)
			}
		case eventDownstreamConnected:
			dc := e.dc

			if dc.network != nil {
				dc.monitored.SetCaseMapping(dc.network.casemap)
			}

			if !u.Enabled && u.srv.Config().EnableUsersOnAuth {
				err := u.updateUser(ctx, func(record *database.User) error {
					record.Enabled = true
					return nil
				})
				if err != nil {
					dc.logger.Printf("failed to enable user after successful authentication: %v", err)
				}
			}

			if !u.Enabled {
				dc.SendMessage(ctx, &irc.Message{
					Command: "ERROR",
					Params:  []string{"This bouncer account is disabled"},
				})
				// TODO: close dc after the error message is sent
				break
			}

			if err := dc.welcome(ctx, u); err != nil {
				if ircErr, ok := err.(ircError); ok {
					msg := ircErr.Message.Copy()
					msg.Prefix = dc.srv.prefix()
					dc.SendMessage(ctx, msg)
				} else {
					dc.SendMessage(ctx, &irc.Message{
						Command: "ERROR",
						Params:  []string{"Internal server error"},
					})
				}
				dc.logger.Printf("failed to handle new registered connection: %v", err)
				// TODO: close dc after the error message is sent
				break
			}

			u.downstreamConns = append(u.downstreamConns, dc)
			u.numDownstreamConns.Add(1)

			if network := dc.network; network != nil && network.lastError != nil {
				sendServiceNOTICE(ctx, dc, fmt.Sprintf("disconnected from %s: %v", network.GetName(), network.lastError))
			}

			u.forEachUpstream(func(uc *upstreamConn) {
				uc.updateAway(ctx)
			})

			if !dc.impersonating {
				u.bumpDownstreamInteractionTime(ctx)
			}
		case eventDownstreamDisconnected:
			dc := e.dc

			for i := range u.downstreamConns {
				if u.downstreamConns[i] == dc {
					u.downstreamConns = append(u.downstreamConns[:i], u.downstreamConns[i+1:]...)
					u.numDownstreamConns.Add(-1)
					break
				}
			}

			if dc.network != nil {
				dc.network.storeClientDeliveryReceipts(ctx, dc.clientName)
			}

			u.forEachUpstream(func(uc *upstreamConn) {
				uc.cancelPendingCommandsByDownstreamID(dc.id)
				uc.updateAway(ctx)
				uc.updateMonitor(ctx)
			})

			if !dc.impersonating {
				u.bumpDownstreamInteractionTime(ctx)
			}
		case eventDownstreamMessage:
			msg, dc := e.msg, e.dc
			if dc.isClosed() {
				dc.logger.Printf("ignoring message on closed connection: %v", msg)
				break
			}
			if err := dc.handleMessage(ctx, msg); err != nil {
				dc.logger.Printf("failed to handle message %q: %v", msg, err)
				dc.Close()
			}
		case eventBroadcast:
			msg := e.msg
			for _, dc := range u.downstreamConns {
				dc.SendMessage(ctx, msg)
			}
		case eventUserUpdate:
			e.done <- u.updateUser(ctx, func(record *database.User) error {
				if e.password != nil {
					record.Password = *e.password
				}
				if e.admin != nil {
					record.Admin = *e.admin
				}
				if e.enabled != nil {
					record.Enabled = *e.enabled
				}
				if e.maxNetworks != nil {
					record.MaxNetworks = *e.maxNetworks
				}
				return nil
			})

			// If the password was updated, kill all downstream connections to
			// force them to re-authenticate with the new credentials.
			if e.password != nil {
				for _, dc := range u.downstreamConns {
					dc.Close()
				}
			}
		case eventTryRegainNick:
			e.uc.tryRegainNick(ctx, e.nick)
		case eventUserRun:
			err := handleServiceCommand(&serviceContext{
				Context: ctx,
				user:    u,
				srv:     u.srv,
				// Here we are setting an admin context on any user run command
				// that is run.
				// As a reminder, user run can only be run by an admin.
				// This enables admins to run actions on a user with admin rights,
				// for example to add a network past the user limit.
				// Non-admin users cannot run user run, not even on themselves,
				// so this cannot be used to escalate privileges.
				admin: true,
				print: func(text string) {
					// Avoid blocking on e.print in case our context is canceled.
					select {
					case <-ctx.Done():
					case e.ch <- userRunMsg{message: text}:
					}
				},
			}, e.params)
			select {
			case <-ctx.Done():
			case e.ch <- userRunMsg{err: err}:
			}
		case eventStop:
			for _, dc := range u.downstreamConns {
				dc.Close()
			}
			for _, n := range u.networks {
				n.stop()

				n.delivered.ForEachClient(func(clientName string) {
					n.storeClientDeliveryReceipts(ctx, clientName)
				})
			}
			return
		default:
			panic(fmt.Sprintf("received unknown event type: %T", e))
		}
	}
}

func (u *user) handleUpstreamDisconnected(ctx context.Context, uc *upstreamConn) {
	uc.network.conn = nil

	uc.stopRegainNickTimer()
	uc.abortPendingCommands()

	uc.channels.ForEach(func(_ string, uch *upstreamChannel) {
		uch.updateAutoDetach(0)
	})

	uc.forEachDownstream(func(dc *downstreamConn) {
		dc.updateSupportedCaps(ctx)
	})

	// If the network has been removed, don't send a state change notification
	found := false
	for _, net := range u.networks {
		if net == uc.network {
			found = true
			break
		}
	}
	if !found {
		return
	}

	u.notifyBouncerNetworkState(ctx, uc.network.ID, irc.Tags{"state": "disconnected"})

	if uc.network.lastError == nil {
		uc.forEachDownstream(func(dc *downstreamConn) {
			if !dc.caps.IsEnabled("soju.im/bouncer-networks") {
				sendServiceNOTICE(ctx, dc, fmt.Sprintf("disconnected from %s", uc.network.GetName()))
			}
		})
	}
}

func (u *user) notifyBouncerNetworkState(ctx context.Context, netID int64, attrs irc.Tags) {
	// Don't send state updates for removed networks
	found := false
	for _, net := range u.networks {
		if net.ID == netID {
			found = true
			break
		}
	}
	if !found {
		return
	}

	netIDStr := fmt.Sprintf("%v", netID)
	for _, dc := range u.downstreamConns {
		if dc.caps.IsEnabled("soju.im/bouncer-networks-notify") {
			dc.SendMessage(ctx, &irc.Message{
				Command: "BOUNCER",
				Params:  []string{"NETWORK", netIDStr, attrs.String()},
			})
		}
	}
}

func (u *user) addNetwork(network *network) {
	u.networks = append(u.networks, network)

	sort.Slice(u.networks, func(i, j int) bool {
		return u.networks[i].ID < u.networks[j].ID
	})

	go network.run()
}

func (u *user) removeNetwork(network *network) {
	network.stop()

	for _, dc := range u.downstreamConns {
		if dc.network != nil && dc.network == network {
			dc.Close()
		}
	}

	for i, net := range u.networks {
		if net == network {
			u.networks = append(u.networks[:i], u.networks[i+1:]...)
			return
		}
	}

	panic("tried to remove a non-existing network")
}

func (u *user) canEnableNewNetwork() error {
	max := u.MaxNetworks
	if max < 0 {
		max = u.srv.Config().MaxUserNetworks
	}
	if max < 0 {
		return nil
	}
	n := 0
	for _, network := range u.networks {
		if network.Enabled {
			n++
		}
	}
	if n >= max {
		return fmt.Errorf("maximum number of enabled networks reached")
	}
	return nil
}

func (u *user) checkNetwork(record *database.Network) error {
	url, err := record.URL()
	if err != nil {
		return err
	}
	if url.User != nil {
		return fmt.Errorf("%v:// URL must not have username and password information", url.Scheme)
	}
	if url.RawQuery != "" {
		return fmt.Errorf("%v:// URL must not have query values", url.Scheme)
	}
	if url.Fragment != "" {
		return fmt.Errorf("%v:// URL must not have a fragment", url.Scheme)
	}
	switch url.Scheme {
	case "ircs", "irc+insecure":
		if url.Host == "" {
			return fmt.Errorf("%v:// URL must have a host", url.Scheme)
		}
		if url.Path != "" {
			return fmt.Errorf("%v:// URL must not have a path", url.Scheme)
		}
	case "irc+unix", "unix":
		if !u.Admin {
			return fmt.Errorf("non-admin users cannot add networks with a unix address")
		}
		if url.Host != "" {
			return fmt.Errorf("%v:// URL must not have a host", url.Scheme)
		}
		if url.Path == "" {
			return fmt.Errorf("%v:// URL must have a path", url.Scheme)
		}
	default:
		return fmt.Errorf("unknown URL scheme %q", url.Scheme)
	}

	if record.GetName() == "" {
		return fmt.Errorf("network name cannot be empty")
	}
	if strings.HasPrefix(record.GetName(), "-") {
		// Can be mixed up with flags when sending commands to the service
		return fmt.Errorf("network name cannot start with a dash character")
	}

	for _, net := range u.networks {
		if net.GetName() == record.GetName() && net.ID != record.ID {
			return fmt.Errorf("a network with the name %q already exists", record.GetName())
		}
	}

	return nil
}

func (u *user) createNetwork(ctx context.Context, record *database.Network, enforceLimit bool) (*network, error) {
	if record.ID != 0 {
		panic("tried creating an already-existing network")
	}

	if enforceLimit && record.Enabled {
		if err := u.canEnableNewNetwork(); err != nil {
			return nil, err
		}
	}

	if err := u.checkNetwork(record); err != nil {
		return nil, err
	}

	network := newNetwork(u, record, nil)
	err := u.srv.db.StoreNetwork(ctx, u.ID, &network.Network)
	if err != nil {
		return nil, err
	}

	u.addNetwork(network)

	attrs := getNetworkAttrs(network)
	u.notifyBouncerNetworkState(ctx, network.ID, attrs)

	return network, nil
}

func (u *user) updateNetwork(ctx context.Context, record *database.Network, enforceLimit bool) (*network, error) {
	if record.ID == 0 {
		panic("tried updating a new network")
	}

	if enforceLimit && record.Enabled {
		if err := u.canEnableNewNetwork(); err != nil {
			return nil, err
		}
	}

	// If the nickname/realname is reset to the default, just wipe the
	// per-network setting
	if record.Nick == u.Nick {
		record.Nick = ""
	}
	if record.Realname == u.Realname {
		record.Realname = ""
	}

	if err := u.checkNetwork(record); err != nil {
		return nil, err
	}

	network := u.getNetworkByID(record.ID)
	if network == nil {
		panic("tried updating a non-existing network")
	}

	if err := u.srv.db.StoreNetwork(ctx, u.ID, record); err != nil {
		return nil, err
	}

	// Most network changes require us to re-connect to the upstream server

	channels := make([]database.Channel, 0, network.channels.Len())
	network.channels.ForEach(func(_ string, ch *database.Channel) {
		channels = append(channels, *ch)
	})

	updatedNetwork := newNetwork(u, record, channels)

	// If we're currently connected, disconnect and perform the necessary
	// bookkeeping
	network.stop()
	if network.conn != nil {
		// Note: this will set network.conn to nil
		u.handleUpstreamDisconnected(ctx, network.conn)
	}

	// Patch downstream connections to use our fresh updated network
	for _, dc := range u.downstreamConns {
		if dc.network != nil && dc.network == network {
			dc.network = updatedNetwork
		}
	}

	// We need to remove the network after patching downstream connections,
	// otherwise they'll get closed
	u.removeNetwork(network)

	// Transfer network attributes that aren't in the record
	updatedNetwork.delivered = network.delivered
	updatedNetwork.pushTargets = network.pushTargets
	if updatedNetwork.Enabled {
		updatedNetwork.lastError = network.lastError
	}
	updatedNetwork.updateCasemapping(network.casemap)

	// The filesystem message store needs to be notified whenever the network
	// is renamed
	renameNetMsgStore, ok := u.msgStore.(msgstore.RenameNetworkStore)
	if ok && updatedNetwork.GetName() != network.GetName() {
		if err := renameNetMsgStore.RenameNetwork(&network.Network, &updatedNetwork.Network); err != nil {
			network.logger.Printf("failed to update message store network name to %q: %v", updatedNetwork.GetName(), err)
		}
	}

	// This will re-connect to the upstream server
	u.addNetwork(updatedNetwork)

	// TODO: only broadcast attributes that have changed
	attrs := getNetworkAttrs(updatedNetwork)
	u.notifyBouncerNetworkState(ctx, updatedNetwork.ID, attrs)

	return updatedNetwork, nil
}

func (u *user) deleteNetwork(ctx context.Context, id int64) error {
	network := u.getNetworkByID(id)
	if network == nil {
		panic("tried deleting a non-existing network")
	}

	if err := u.srv.db.DeleteNetwork(ctx, network.ID); err != nil {
		return err
	}

	u.removeNetwork(network)

	idStr := fmt.Sprintf("%v", network.ID)
	for _, dc := range u.downstreamConns {
		if dc.caps.IsEnabled("soju.im/bouncer-networks-notify") {
			dc.SendMessage(ctx, &irc.Message{
				Command: "BOUNCER",
				Params:  []string{"NETWORK", idStr, "*"},
			})
		}
	}

	return nil
}

func (u *user) updateUser(ctx context.Context, update UserUpdateFunc) error {
	record := u.User // copy
	if err := update(&record); err != nil {
		return err
	}

	nickUpdated := u.Nick != record.Nick
	realnameUpdated := u.Realname != record.Realname
	enabledUpdated := u.Enabled != record.Enabled
	if err := u.srv.db.StoreUser(ctx, &record); err != nil {
		return fmt.Errorf("failed to update user %q: %v", u.Username, err)
	}
	u.User = record

	if nickUpdated {
		for _, net := range u.networks {
			if net.Nick != "" {
				continue
			}

			if uc := net.conn; uc != nil {
				uc.SendMessage(ctx, &irc.Message{
					Command: "NICK",
					Params:  []string{database.GetNick(&u.User, &net.Network)},
				})
			}
		}
	}

	if realnameUpdated || enabledUpdated {
		// Re-connect to networks which use the default realname
		var needUpdate []database.Network
		for _, net := range u.networks {
			// If only the realname was updated, maybe we can skip the
			// re-connect
			if realnameUpdated && !enabledUpdated {
				// If this network has a custom realname set, no need to
				// re-connect: the user-wide realname remains unused
				if net.Realname != "" {
					continue
				}

				// We only need to call updateNetwork for upstreams that don't
				// support setname
				if uc := net.conn; uc != nil && uc.caps.IsEnabled("setname") {
					uc.SendMessage(ctx, &irc.Message{
						Command: "SETNAME",
						Params:  []string{database.GetRealname(&u.User, &net.Network)},
					})
					continue
				}
			}

			needUpdate = append(needUpdate, net.Network)
		}

		var netErr error
		for _, net := range needUpdate {
			if _, err := u.updateNetwork(ctx, &net, false); err != nil {
				netErr = err
			}
		}
		if netErr != nil {
			return netErr
		}
	}

	if !u.Enabled {
		// TODO: send an error message before disconnecting
		for _, dc := range u.downstreamConns {
			dc.Close()
		}
	}

	return nil
}

func (u *user) stop(ctx context.Context) error {
	select {
	case <-u.done:
		return nil // already stopped
	case u.events <- eventStop{}:
		// we've requested to stop, let's wait for the user goroutine to exit
	case <-ctx.Done():
		return ctx.Err()
	}

	select {
	case <-u.done:
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}

func (u *user) hasPersistentMsgStore() bool {
	if u.msgStore == nil {
		return false
	}
	return !msgstore.IsMemoryStore(u.msgStore)
}

func (u *user) FormatServerTime(t time.Time) string {
	if u.msgStore != nil && msgstore.IsFSStore(u.msgStore) {
		// The FS message store truncates message timestamps to the second,
		// so truncate them here to get consistent timestamps.
		t = t.Truncate(time.Second)
	}
	return xirc.FormatServerTime(t)
}

// localTCPAddr returns the local address to use when connecting to a host.
// A nil address is returned when the OS should automatically pick one.
func (u *user) localTCPAddr(remoteIP net.IP) (*net.TCPAddr, error) {
	upstreamUserIPs := u.srv.Config().UpstreamUserIPs
	if len(upstreamUserIPs) == 0 {
		return nil, nil
	}

	wantIPv6 := remoteIP.To4() == nil
	var ipNet *net.IPNet
	for _, in := range upstreamUserIPs {
		if wantIPv6 == (in.IP.To4() == nil) {
			ipNet = in
			break
		}
	}
	if ipNet == nil {
		return nil, nil
	}

	var ipInt big.Int
	ipInt.SetBytes(ipNet.IP)
	ipInt.Add(&ipInt, big.NewInt(u.ID+1))
	ip := net.IP(ipInt.Bytes())
	if !ipNet.Contains(ip) {
		return nil, fmt.Errorf("IP network %v too small", ipNet)
	}

	return &net.TCPAddr{IP: ip}, nil
}

func (u *user) bumpDownstreamInteractionTime(ctx context.Context) {
	err := u.updateUser(ctx, func(record *database.User) error {
		record.DownstreamInteractedAt = time.Now()
		return nil
	})
	if err != nil {
		u.logger.Printf("failed to bump downstream interaction time: %v", err)
	}
}