soju/user.go
Simon Ser d9186e994d
Add support for detached channels
Channels can now be detached by leaving them with the reason "detach",
and re-attached by joining them again. Upon detaching the channel is
no longer forwarded to downstream connections. Upon re-attaching the
history buffer is sent.
2020-05-01 15:18:14 +02:00

465 lines
10 KiB
Go

package soju
import (
"fmt"
"time"
"gopkg.in/irc.v3"
)
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 networkHistory struct {
offlineClients map[string]uint64 // indexed by client name
ring *Ring // can be nil if there are no offline clients
}
type network struct {
Network
user *user
stopped chan struct{}
conn *upstreamConn
channels map[string]*Channel
history map[string]*networkHistory // indexed by entity
offlineClients map[string]struct{} // indexed by client name
lastError error
}
func newNetwork(user *user, record *Network, channels []Channel) *network {
m := make(map[string]*Channel, len(channels))
for _, ch := range channels {
ch := ch
m[ch.Name] = &ch
}
return &network{
Network: *record,
user: user,
stopped: make(chan struct{}),
channels: m,
history: make(map[string]*networkHistory),
offlineClients: make(map[string]struct{}),
}
}
func (net *network) forEachDownstream(f func(*downstreamConn)) {
net.user.forEachDownstream(func(dc *downstreamConn) {
if dc.network != nil && dc.network != net {
return
}
f(dc)
})
}
func (net *network) run() {
var lastTry time.Time
for {
select {
case <-net.stopped:
return
default:
// This space is intentionally left blank
}
if dur := time.Now().Sub(lastTry); dur < retryConnectMinDelay {
delay := retryConnectMinDelay - dur
net.user.srv.Logger.Printf("waiting %v before trying to reconnect to %q", delay.Truncate(time.Second), net.Addr)
time.Sleep(delay)
}
lastTry = time.Now()
uc, err := connectToUpstream(net)
if err != nil {
net.user.srv.Logger.Printf("failed to connect to upstream server %q: %v", net.Addr, err)
net.user.events <- eventUpstreamConnectionError{net, fmt.Errorf("failed to connect: %v", err)}
continue
}
uc.register()
if err := uc.runUntilRegistered(); err != nil {
uc.logger.Printf("failed to register: %v", err)
net.user.events <- eventUpstreamConnectionError{net, fmt.Errorf("failed to register: %v", err)}
uc.Close()
continue
}
net.user.events <- eventUpstreamConnected{uc}
if err := uc.readMessages(net.user.events); err != nil {
uc.logger.Printf("failed to handle messages: %v", err)
net.user.events <- eventUpstreamError{uc, fmt.Errorf("failed to handle messages: %v", err)}
}
uc.Close()
net.user.events <- eventUpstreamDisconnected{uc}
}
}
func (net *network) Stop() {
select {
case <-net.stopped:
return
default:
close(net.stopped)
}
if net.conn != nil {
net.conn.Close()
}
}
func (net *network) createUpdateChannel(ch *Channel) error {
if current, ok := net.channels[ch.Name]; ok {
ch.ID = current.ID // update channel if it already exists
}
if err := net.user.srv.db.StoreChannel(net.ID, ch); err != nil {
return err
}
prev := net.channels[ch.Name]
net.channels[ch.Name] = ch
if prev != nil && prev.Detached != ch.Detached {
history := net.history[ch.Name]
if ch.Detached {
net.user.srv.Logger.Printf("network %q: detaching channel %q", net.GetName(), ch.Name)
net.forEachDownstream(func(dc *downstreamConn) {
net.offlineClients[dc.clientName] = struct{}{}
if history != nil {
history.offlineClients[dc.clientName] = history.ring.Cur()
}
dc.SendMessage(&irc.Message{
Prefix: dc.prefix(),
Command: "PART",
Params: []string{dc.marshalEntity(net, ch.Name), "Detach"},
})
})
} else {
net.user.srv.Logger.Printf("network %q: attaching channel %q", net.GetName(), ch.Name)
var uch *upstreamChannel
if net.conn != nil {
uch = net.conn.channels[ch.Name]
}
net.forEachDownstream(func(dc *downstreamConn) {
dc.SendMessage(&irc.Message{
Prefix: dc.prefix(),
Command: "JOIN",
Params: []string{dc.marshalEntity(net, ch.Name)},
})
if uch != nil {
forwardChannel(dc, uch)
}
if history != nil {
dc.sendNetworkHistory(net)
}
})
}
}
return nil
}
func (net *network) deleteChannel(name string) error {
if err := net.user.srv.db.DeleteChannel(net.ID, name); err != nil {
return err
}
delete(net.channels, name)
return nil
}
type user struct {
User
srv *Server
events chan event
networks []*network
downstreamConns []*downstreamConn
// LIST commands in progress
pendingLISTs []pendingLIST
}
type pendingLIST struct {
downstreamID uint64
// list of per-upstream LIST commands not yet sent or completed
pendingCommands map[int64]*irc.Message
}
func newUser(srv *Server, record *User) *user {
return &user{
User: *record,
srv: srv,
events: make(chan event, 64),
}
}
func (u *user) forEachNetwork(f func(*network)) {
for _, network := range u.networks {
f(network)
}
}
func (u *user) forEachUpstream(f func(uc *upstreamConn)) {
for _, network := range u.networks {
if network.conn == nil {
continue
}
f(network.conn)
}
}
func (u *user) forEachDownstream(f func(dc *downstreamConn)) {
for _, dc := range u.downstreamConns {
f(dc)
}
}
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) run() {
networks, err := u.srv.db.ListNetworks(u.Username)
if err != nil {
u.srv.Logger.Printf("failed to list networks for user %q: %v", u.Username, err)
return
}
for _, record := range networks {
record := record
channels, err := u.srv.db.ListChannels(record.ID)
if err != nil {
u.srv.Logger.Printf("failed to list channels for user %q, network %q: %v", u.Username, record.GetName(), err)
}
network := newNetwork(u, &record, channels)
u.networks = append(u.networks, network)
go network.run()
}
for e := range u.events {
switch e := e.(type) {
case eventUpstreamConnected:
uc := e.uc
uc.network.conn = uc
uc.updateAway()
uc.forEachDownstream(func(dc *downstreamConn) {
dc.updateSupportedCaps()
sendServiceNOTICE(dc, fmt.Sprintf("connected to %s", uc.network.GetName()))
})
uc.network.lastError = nil
case eventUpstreamDisconnected:
uc := e.uc
uc.network.conn = nil
for _, ml := range uc.messageLoggers {
if err := ml.Close(); err != nil {
uc.logger.Printf("failed to close message logger: %v", err)
}
}
uc.endPendingLISTs(true)
uc.forEachDownstream(func(dc *downstreamConn) {
dc.updateSupportedCaps()
})
if uc.network.lastError == nil {
uc.forEachDownstream(func(dc *downstreamConn) {
sendServiceNOTICE(dc, fmt.Sprintf("disconnected from %s", uc.network.GetName()))
})
}
case eventUpstreamConnectionError:
net := e.net
if net.lastError == nil || net.lastError.Error() != e.err.Error() {
net.forEachDownstream(func(dc *downstreamConn) {
sendServiceNOTICE(dc, fmt.Sprintf("failed connecting/registering to %s: %v", net.GetName(), e.err))
})
}
net.lastError = e.err
case eventUpstreamError:
uc := e.uc
uc.forEachDownstream(func(dc *downstreamConn) {
sendServiceNOTICE(dc, fmt.Sprintf("disconnected from %s: %v", uc.network.GetName(), e.err))
})
uc.network.lastError = e.err
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(msg); err != nil {
uc.logger.Printf("failed to handle message %q: %v", msg, err)
}
case eventDownstreamConnected:
dc := e.dc
if err := dc.welcome(); err != nil {
dc.logger.Printf("failed to handle new registered connection: %v", err)
break
}
u.downstreamConns = append(u.downstreamConns, dc)
u.forEachUpstream(func(uc *upstreamConn) {
uc.updateAway()
})
dc.updateSupportedCaps()
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:]...)
break
}
}
// Save history if we're the last client with this name
skipHistory := make(map[*network]bool)
u.forEachDownstream(func(conn *downstreamConn) {
if dc.clientName == conn.clientName {
skipHistory[conn.network] = true
}
})
dc.forEachNetwork(func(net *network) {
if skipHistory[net] || skipHistory[nil] {
return
}
net.offlineClients[dc.clientName] = struct{}{}
for target, history := range net.history {
if ch, ok := net.channels[target]; ok && ch.Detached {
continue
}
history.offlineClients[dc.clientName] = history.ring.Cur()
}
})
u.forEachUpstream(func(uc *upstreamConn) {
uc.updateAway()
})
case eventDownstreamMessage:
msg, dc := e.msg, e.dc
if dc.isClosed() {
dc.logger.Printf("ignoring message on closed connection: %v", msg)
break
}
err := dc.handleMessage(msg)
if ircErr, ok := err.(ircError); ok {
ircErr.Message.Prefix = dc.srv.prefix()
dc.SendMessage(ircErr.Message)
} else if err != nil {
dc.logger.Printf("failed to handle message %q: %v", msg, err)
dc.Close()
}
default:
u.srv.Logger.Printf("received unknown event type: %T", e)
}
}
}
func (u *user) createNetwork(net *Network) (*network, error) {
if net.ID != 0 {
panic("tried creating an already-existing network")
}
network := newNetwork(u, net, nil)
err := u.srv.db.StoreNetwork(u.Username, &network.Network)
if err != nil {
return nil, err
}
u.networks = append(u.networks, network)
go network.run()
return network, nil
}
func (u *user) deleteNetwork(id int64) error {
for i, net := range u.networks {
if net.ID != id {
continue
}
if err := u.srv.db.DeleteNetwork(net.ID); err != nil {
return err
}
u.forEachDownstream(func(dc *downstreamConn) {
if dc.network != nil && dc.network == net {
dc.Close()
}
})
net.Stop()
u.networks = append(u.networks[:i], u.networks[i+1:]...)
return nil
}
panic("tried deleting a non-existing network")
}
func (u *user) updatePassword(hashed string) error {
u.User.Password = hashed
return u.srv.db.UpdatePassword(&u.User)
}