source: code/trunk/user.go@ 722

Last change on this file since 722 was 722, checked in by contact, 4 years ago

Use RPL_LOGGEDIN/OUT to mirror upstream status

This will allow clients to properly show/hide UI to login and
register.
File size: 23.4 KB
Line 
1package soju
2
3import (
4 "context"
5 "crypto/sha256"
6 "encoding/binary"
7 "encoding/hex"
8 "fmt"
9 "math/big"
10 "net"
11 "time"
12
13 "gopkg.in/irc.v3"
14)
15
16type event interface{}
17
18type eventUpstreamMessage struct {
19 msg *irc.Message
20 uc *upstreamConn
21}
22
23type eventUpstreamConnectionError struct {
24 net *network
25 err error
26}
27
28type eventUpstreamConnected struct {
29 uc *upstreamConn
30}
31
32type eventUpstreamDisconnected struct {
33 uc *upstreamConn
34}
35
36type eventUpstreamError struct {
37 uc *upstreamConn
38 err error
39}
40
41type eventDownstreamMessage struct {
42 msg *irc.Message
43 dc *downstreamConn
44}
45
46type eventDownstreamConnected struct {
47 dc *downstreamConn
48}
49
50type eventDownstreamDisconnected struct {
51 dc *downstreamConn
52}
53
54type eventChannelDetach struct {
55 uc *upstreamConn
56 name string
57}
58
59type eventBroadcast struct {
60 msg *irc.Message
61}
62
63type eventStop struct{}
64
65type eventUserUpdate struct {
66 password *string
67 admin *bool
68 done chan error
69}
70
71type deliveredClientMap map[string]string // client name -> msg ID
72
73type deliveredStore struct {
74 m deliveredCasemapMap
75}
76
77func newDeliveredStore() deliveredStore {
78 return deliveredStore{deliveredCasemapMap{newCasemapMap(0)}}
79}
80
81func (ds deliveredStore) HasTarget(target string) bool {
82 return ds.m.Value(target) != nil
83}
84
85func (ds deliveredStore) LoadID(target, clientName string) string {
86 clients := ds.m.Value(target)
87 if clients == nil {
88 return ""
89 }
90 return clients[clientName]
91}
92
93func (ds deliveredStore) StoreID(target, clientName, msgID string) {
94 clients := ds.m.Value(target)
95 if clients == nil {
96 clients = make(deliveredClientMap)
97 ds.m.SetValue(target, clients)
98 }
99 clients[clientName] = msgID
100}
101
102func (ds deliveredStore) ForEachTarget(f func(target string)) {
103 for _, entry := range ds.m.innerMap {
104 f(entry.originalKey)
105 }
106}
107
108func (ds deliveredStore) ForEachClient(f func(clientName string)) {
109 clients := make(map[string]struct{})
110 for _, entry := range ds.m.innerMap {
111 delivered := entry.value.(deliveredClientMap)
112 for clientName := range delivered {
113 clients[clientName] = struct{}{}
114 }
115 }
116
117 for clientName := range clients {
118 f(clientName)
119 }
120}
121
122type network struct {
123 Network
124 user *user
125 logger Logger
126 stopped chan struct{}
127
128 conn *upstreamConn
129 channels channelCasemapMap
130 delivered deliveredStore
131 lastError error
132 casemap casemapping
133}
134
135func newNetwork(user *user, record *Network, channels []Channel) *network {
136 logger := &prefixLogger{user.logger, fmt.Sprintf("network %q: ", record.GetName())}
137
138 m := channelCasemapMap{newCasemapMap(0)}
139 for _, ch := range channels {
140 ch := ch
141 m.SetValue(ch.Name, &ch)
142 }
143
144 return &network{
145 Network: *record,
146 user: user,
147 logger: logger,
148 stopped: make(chan struct{}),
149 channels: m,
150 delivered: newDeliveredStore(),
151 casemap: casemapRFC1459,
152 }
153}
154
155func (net *network) forEachDownstream(f func(*downstreamConn)) {
156 net.user.forEachDownstream(func(dc *downstreamConn) {
157 if dc.network == nil && !dc.isMultiUpstream {
158 return
159 }
160 if dc.network != nil && dc.network != net {
161 return
162 }
163 f(dc)
164 })
165}
166
167func (net *network) isStopped() bool {
168 select {
169 case <-net.stopped:
170 return true
171 default:
172 return false
173 }
174}
175
176func userIdent(u *User) string {
177 // The ident is a string we will send to upstream servers in clear-text.
178 // For privacy reasons, make sure it doesn't expose any meaningful user
179 // metadata. We just use the base64-encoded hashed ID, so that people don't
180 // start relying on the string being an integer or following a pattern.
181 var b [64]byte
182 binary.LittleEndian.PutUint64(b[:], uint64(u.ID))
183 h := sha256.Sum256(b[:])
184 return hex.EncodeToString(h[:16])
185}
186
187func (net *network) run() {
188 if !net.Enabled {
189 return
190 }
191
192 var lastTry time.Time
193 for {
194 if net.isStopped() {
195 return
196 }
197
198 if dur := time.Now().Sub(lastTry); dur < retryConnectDelay {
199 delay := retryConnectDelay - dur
200 net.logger.Printf("waiting %v before trying to reconnect to %q", delay.Truncate(time.Second), net.Addr)
201 time.Sleep(delay)
202 }
203 lastTry = time.Now()
204
205 uc, err := connectToUpstream(net)
206 if err != nil {
207 net.logger.Printf("failed to connect to upstream server %q: %v", net.Addr, err)
208 net.user.events <- eventUpstreamConnectionError{net, fmt.Errorf("failed to connect: %v", err)}
209 continue
210 }
211
212 if net.user.srv.Identd != nil {
213 net.user.srv.Identd.Store(uc.RemoteAddr().String(), uc.LocalAddr().String(), userIdent(&net.user.User))
214 }
215
216 net.user.srv.metrics.upstreams.Add(1)
217
218 uc.register()
219 if err := uc.runUntilRegistered(); err != nil {
220 text := err.Error()
221 if regErr, ok := err.(registrationError); ok {
222 text = string(regErr)
223 }
224 uc.logger.Printf("failed to register: %v", text)
225 net.user.events <- eventUpstreamConnectionError{net, fmt.Errorf("failed to register: %v", text)}
226 uc.Close()
227 continue
228 }
229
230 // TODO: this is racy with net.stopped. If the network is stopped
231 // before the user goroutine receives eventUpstreamConnected, the
232 // connection won't be closed.
233 net.user.events <- eventUpstreamConnected{uc}
234 if err := uc.readMessages(net.user.events); err != nil {
235 uc.logger.Printf("failed to handle messages: %v", err)
236 net.user.events <- eventUpstreamError{uc, fmt.Errorf("failed to handle messages: %v", err)}
237 }
238 uc.Close()
239 net.user.events <- eventUpstreamDisconnected{uc}
240
241 if net.user.srv.Identd != nil {
242 net.user.srv.Identd.Delete(uc.RemoteAddr().String(), uc.LocalAddr().String())
243 }
244
245 net.user.srv.metrics.upstreams.Add(-1)
246 }
247}
248
249func (net *network) stop() {
250 if !net.isStopped() {
251 close(net.stopped)
252 }
253
254 if net.conn != nil {
255 net.conn.Close()
256 }
257}
258
259func (net *network) detach(ch *Channel) {
260 if ch.Detached {
261 return
262 }
263
264 net.logger.Printf("detaching channel %q", ch.Name)
265
266 ch.Detached = true
267
268 if net.user.msgStore != nil {
269 nameCM := net.casemap(ch.Name)
270 lastID, err := net.user.msgStore.LastMsgID(&net.Network, nameCM, time.Now())
271 if err != nil {
272 net.logger.Printf("failed to get last message ID for channel %q: %v", ch.Name, err)
273 }
274 ch.DetachedInternalMsgID = lastID
275 }
276
277 if net.conn != nil {
278 uch := net.conn.channels.Value(ch.Name)
279 if uch != nil {
280 uch.updateAutoDetach(0)
281 }
282 }
283
284 net.forEachDownstream(func(dc *downstreamConn) {
285 dc.SendMessage(&irc.Message{
286 Prefix: dc.prefix(),
287 Command: "PART",
288 Params: []string{dc.marshalEntity(net, ch.Name), "Detach"},
289 })
290 })
291}
292
293func (net *network) attach(ch *Channel) {
294 if !ch.Detached {
295 return
296 }
297
298 net.logger.Printf("attaching channel %q", ch.Name)
299
300 detachedMsgID := ch.DetachedInternalMsgID
301 ch.Detached = false
302 ch.DetachedInternalMsgID = ""
303
304 var uch *upstreamChannel
305 if net.conn != nil {
306 uch = net.conn.channels.Value(ch.Name)
307
308 net.conn.updateChannelAutoDetach(ch.Name)
309 }
310
311 net.forEachDownstream(func(dc *downstreamConn) {
312 dc.SendMessage(&irc.Message{
313 Prefix: dc.prefix(),
314 Command: "JOIN",
315 Params: []string{dc.marshalEntity(net, ch.Name)},
316 })
317
318 if uch != nil {
319 forwardChannel(dc, uch)
320 }
321
322 if detachedMsgID != "" {
323 dc.sendTargetBacklog(context.TODO(), net, ch.Name, detachedMsgID)
324 }
325 })
326}
327
328func (net *network) deleteChannel(ctx context.Context, name string) error {
329 ch := net.channels.Value(name)
330 if ch == nil {
331 return fmt.Errorf("unknown channel %q", name)
332 }
333 if net.conn != nil {
334 uch := net.conn.channels.Value(ch.Name)
335 if uch != nil {
336 uch.updateAutoDetach(0)
337 }
338 }
339
340 if err := net.user.srv.db.DeleteChannel(ctx, ch.ID); err != nil {
341 return err
342 }
343 net.channels.Delete(name)
344 return nil
345}
346
347func (net *network) updateCasemapping(newCasemap casemapping) {
348 net.casemap = newCasemap
349 net.channels.SetCasemapping(newCasemap)
350 net.delivered.m.SetCasemapping(newCasemap)
351 if uc := net.conn; uc != nil {
352 uc.channels.SetCasemapping(newCasemap)
353 for _, entry := range uc.channels.innerMap {
354 uch := entry.value.(*upstreamChannel)
355 uch.Members.SetCasemapping(newCasemap)
356 }
357 uc.monitored.SetCasemapping(newCasemap)
358 }
359 net.forEachDownstream(func(dc *downstreamConn) {
360 dc.monitored.SetCasemapping(newCasemap)
361 })
362}
363
364func (net *network) storeClientDeliveryReceipts(clientName string) {
365 if !net.user.hasPersistentMsgStore() {
366 return
367 }
368
369 var receipts []DeliveryReceipt
370 net.delivered.ForEachTarget(func(target string) {
371 msgID := net.delivered.LoadID(target, clientName)
372 if msgID == "" {
373 return
374 }
375 receipts = append(receipts, DeliveryReceipt{
376 Target: target,
377 InternalMsgID: msgID,
378 })
379 })
380
381 if err := net.user.srv.db.StoreClientDeliveryReceipts(context.TODO(), net.ID, clientName, receipts); err != nil {
382 net.logger.Printf("failed to store delivery receipts for client %q: %v", clientName, err)
383 }
384}
385
386func (net *network) isHighlight(msg *irc.Message) bool {
387 if msg.Command != "PRIVMSG" && msg.Command != "NOTICE" {
388 return false
389 }
390
391 text := msg.Params[1]
392
393 nick := net.Nick
394 if net.conn != nil {
395 nick = net.conn.nick
396 }
397
398 // TODO: use case-mapping aware comparison here
399 return msg.Prefix.Name != nick && isHighlight(text, nick)
400}
401
402func (net *network) detachedMessageNeedsRelay(ch *Channel, msg *irc.Message) bool {
403 highlight := net.isHighlight(msg)
404 return ch.RelayDetached == FilterMessage || ((ch.RelayDetached == FilterHighlight || ch.RelayDetached == FilterDefault) && highlight)
405}
406
407type user struct {
408 User
409 srv *Server
410 logger Logger
411
412 events chan event
413 done chan struct{}
414
415 networks []*network
416 downstreamConns []*downstreamConn
417 msgStore messageStore
418}
419
420func newUser(srv *Server, record *User) *user {
421 logger := &prefixLogger{srv.Logger, fmt.Sprintf("user %q: ", record.Username)}
422
423 var msgStore messageStore
424 if logPath := srv.Config().LogPath; logPath != "" {
425 msgStore = newFSMessageStore(logPath, record.Username)
426 } else {
427 msgStore = newMemoryMessageStore()
428 }
429
430 return &user{
431 User: *record,
432 srv: srv,
433 logger: logger,
434 events: make(chan event, 64),
435 done: make(chan struct{}),
436 msgStore: msgStore,
437 }
438}
439
440func (u *user) forEachNetwork(f func(*network)) {
441 for _, network := range u.networks {
442 f(network)
443 }
444}
445
446func (u *user) forEachUpstream(f func(uc *upstreamConn)) {
447 for _, network := range u.networks {
448 if network.conn == nil {
449 continue
450 }
451 f(network.conn)
452 }
453}
454
455func (u *user) forEachDownstream(f func(dc *downstreamConn)) {
456 for _, dc := range u.downstreamConns {
457 f(dc)
458 }
459}
460
461func (u *user) getNetwork(name string) *network {
462 for _, network := range u.networks {
463 if network.Addr == name {
464 return network
465 }
466 if network.Name != "" && network.Name == name {
467 return network
468 }
469 }
470 return nil
471}
472
473func (u *user) getNetworkByID(id int64) *network {
474 for _, net := range u.networks {
475 if net.ID == id {
476 return net
477 }
478 }
479 return nil
480}
481
482func (u *user) run() {
483 defer func() {
484 if u.msgStore != nil {
485 if err := u.msgStore.Close(); err != nil {
486 u.logger.Printf("failed to close message store for user %q: %v", u.Username, err)
487 }
488 }
489 close(u.done)
490 }()
491
492 networks, err := u.srv.db.ListNetworks(context.TODO(), u.ID)
493 if err != nil {
494 u.logger.Printf("failed to list networks for user %q: %v", u.Username, err)
495 return
496 }
497
498 for _, record := range networks {
499 record := record
500 channels, err := u.srv.db.ListChannels(context.TODO(), record.ID)
501 if err != nil {
502 u.logger.Printf("failed to list channels for user %q, network %q: %v", u.Username, record.GetName(), err)
503 continue
504 }
505
506 network := newNetwork(u, &record, channels)
507 u.networks = append(u.networks, network)
508
509 if u.hasPersistentMsgStore() {
510 receipts, err := u.srv.db.ListDeliveryReceipts(context.TODO(), record.ID)
511 if err != nil {
512 u.logger.Printf("failed to load delivery receipts for user %q, network %q: %v", u.Username, network.GetName(), err)
513 return
514 }
515
516 for _, rcpt := range receipts {
517 network.delivered.StoreID(rcpt.Target, rcpt.Client, rcpt.InternalMsgID)
518 }
519 }
520
521 go network.run()
522 }
523
524 for e := range u.events {
525 switch e := e.(type) {
526 case eventUpstreamConnected:
527 uc := e.uc
528
529 uc.network.conn = uc
530
531 uc.updateAway()
532 uc.updateMonitor()
533
534 netIDStr := fmt.Sprintf("%v", uc.network.ID)
535 uc.forEachDownstream(func(dc *downstreamConn) {
536 dc.updateSupportedCaps()
537
538 if !dc.caps["soju.im/bouncer-networks"] {
539 sendServiceNOTICE(dc, fmt.Sprintf("connected to %s", uc.network.GetName()))
540 }
541
542 dc.updateNick()
543 dc.updateRealname()
544 dc.updateAccount()
545 })
546 u.forEachDownstream(func(dc *downstreamConn) {
547 if dc.caps["soju.im/bouncer-networks-notify"] {
548 dc.SendMessage(&irc.Message{
549 Prefix: dc.srv.prefix(),
550 Command: "BOUNCER",
551 Params: []string{"NETWORK", netIDStr, "state=connected"},
552 })
553 }
554 })
555 uc.network.lastError = nil
556 case eventUpstreamDisconnected:
557 u.handleUpstreamDisconnected(e.uc)
558 case eventUpstreamConnectionError:
559 net := e.net
560
561 stopped := false
562 select {
563 case <-net.stopped:
564 stopped = true
565 default:
566 }
567
568 if !stopped && (net.lastError == nil || net.lastError.Error() != e.err.Error()) {
569 net.forEachDownstream(func(dc *downstreamConn) {
570 sendServiceNOTICE(dc, fmt.Sprintf("failed connecting/registering to %s: %v", net.GetName(), e.err))
571 })
572 }
573 net.lastError = e.err
574 case eventUpstreamError:
575 uc := e.uc
576
577 uc.forEachDownstream(func(dc *downstreamConn) {
578 sendServiceNOTICE(dc, fmt.Sprintf("disconnected from %s: %v", uc.network.GetName(), e.err))
579 })
580 uc.network.lastError = e.err
581 case eventUpstreamMessage:
582 msg, uc := e.msg, e.uc
583 if uc.isClosed() {
584 uc.logger.Printf("ignoring message on closed connection: %v", msg)
585 break
586 }
587 if err := uc.handleMessage(msg); err != nil {
588 uc.logger.Printf("failed to handle message %q: %v", msg, err)
589 }
590 case eventChannelDetach:
591 uc, name := e.uc, e.name
592 c := uc.network.channels.Value(name)
593 if c == nil || c.Detached {
594 continue
595 }
596 uc.network.detach(c)
597 if err := uc.srv.db.StoreChannel(context.TODO(), uc.network.ID, c); err != nil {
598 u.logger.Printf("failed to store updated detached channel %q: %v", c.Name, err)
599 }
600 case eventDownstreamConnected:
601 dc := e.dc
602
603 if dc.network != nil {
604 dc.monitored.SetCasemapping(dc.network.casemap)
605 }
606
607 if err := dc.welcome(context.TODO()); err != nil {
608 dc.logger.Printf("failed to handle new registered connection: %v", err)
609 break
610 }
611
612 u.downstreamConns = append(u.downstreamConns, dc)
613
614 dc.forEachNetwork(func(network *network) {
615 if network.lastError != nil {
616 sendServiceNOTICE(dc, fmt.Sprintf("disconnected from %s: %v", network.GetName(), network.lastError))
617 }
618 })
619
620 u.forEachUpstream(func(uc *upstreamConn) {
621 uc.updateAway()
622 })
623 case eventDownstreamDisconnected:
624 dc := e.dc
625
626 for i := range u.downstreamConns {
627 if u.downstreamConns[i] == dc {
628 u.downstreamConns = append(u.downstreamConns[:i], u.downstreamConns[i+1:]...)
629 break
630 }
631 }
632
633 dc.forEachNetwork(func(net *network) {
634 net.storeClientDeliveryReceipts(dc.clientName)
635 })
636
637 u.forEachUpstream(func(uc *upstreamConn) {
638 uc.updateAway()
639 uc.updateMonitor()
640 })
641 case eventDownstreamMessage:
642 msg, dc := e.msg, e.dc
643 if dc.isClosed() {
644 dc.logger.Printf("ignoring message on closed connection: %v", msg)
645 break
646 }
647 err := dc.handleMessage(context.TODO(), msg)
648 if ircErr, ok := err.(ircError); ok {
649 ircErr.Message.Prefix = dc.srv.prefix()
650 dc.SendMessage(ircErr.Message)
651 } else if err != nil {
652 dc.logger.Printf("failed to handle message %q: %v", msg, err)
653 dc.Close()
654 }
655 case eventBroadcast:
656 msg := e.msg
657 u.forEachDownstream(func(dc *downstreamConn) {
658 dc.SendMessage(msg)
659 })
660 case eventUserUpdate:
661 // copy the user record because we'll mutate it
662 record := u.User
663
664 if e.password != nil {
665 record.Password = *e.password
666 }
667 if e.admin != nil {
668 record.Admin = *e.admin
669 }
670
671 e.done <- u.updateUser(context.TODO(), &record)
672
673 // If the password was updated, kill all downstream connections to
674 // force them to re-authenticate with the new credentials.
675 if e.password != nil {
676 u.forEachDownstream(func(dc *downstreamConn) {
677 dc.Close()
678 })
679 }
680 case eventStop:
681 u.forEachDownstream(func(dc *downstreamConn) {
682 dc.Close()
683 })
684 for _, n := range u.networks {
685 n.stop()
686
687 n.delivered.ForEachClient(func(clientName string) {
688 n.storeClientDeliveryReceipts(clientName)
689 })
690 }
691 return
692 default:
693 panic(fmt.Sprintf("received unknown event type: %T", e))
694 }
695 }
696}
697
698func (u *user) handleUpstreamDisconnected(uc *upstreamConn) {
699 uc.network.conn = nil
700
701 uc.endPendingCommands()
702
703 for _, entry := range uc.channels.innerMap {
704 uch := entry.value.(*upstreamChannel)
705 uch.updateAutoDetach(0)
706 }
707
708 netIDStr := fmt.Sprintf("%v", uc.network.ID)
709 uc.forEachDownstream(func(dc *downstreamConn) {
710 dc.updateSupportedCaps()
711 })
712
713 // If the network has been removed, don't send a state change notification
714 found := false
715 for _, net := range u.networks {
716 if net == uc.network {
717 found = true
718 break
719 }
720 }
721 if !found {
722 return
723 }
724
725 u.forEachDownstream(func(dc *downstreamConn) {
726 if dc.caps["soju.im/bouncer-networks-notify"] {
727 dc.SendMessage(&irc.Message{
728 Prefix: dc.srv.prefix(),
729 Command: "BOUNCER",
730 Params: []string{"NETWORK", netIDStr, "state=disconnected"},
731 })
732 }
733 })
734
735 if uc.network.lastError == nil {
736 uc.forEachDownstream(func(dc *downstreamConn) {
737 if !dc.caps["soju.im/bouncer-networks"] {
738 sendServiceNOTICE(dc, fmt.Sprintf("disconnected from %s", uc.network.GetName()))
739 }
740 })
741 }
742}
743
744func (u *user) addNetwork(network *network) {
745 u.networks = append(u.networks, network)
746 go network.run()
747}
748
749func (u *user) removeNetwork(network *network) {
750 network.stop()
751
752 u.forEachDownstream(func(dc *downstreamConn) {
753 if dc.network != nil && dc.network == network {
754 dc.Close()
755 }
756 })
757
758 for i, net := range u.networks {
759 if net == network {
760 u.networks = append(u.networks[:i], u.networks[i+1:]...)
761 return
762 }
763 }
764
765 panic("tried to remove a non-existing network")
766}
767
768func (u *user) checkNetwork(record *Network) error {
769 for _, net := range u.networks {
770 if net.GetName() == record.GetName() && net.ID != record.ID {
771 return fmt.Errorf("a network with the name %q already exists", record.GetName())
772 }
773 }
774 return nil
775}
776
777func (u *user) createNetwork(ctx context.Context, record *Network) (*network, error) {
778 if record.ID != 0 {
779 panic("tried creating an already-existing network")
780 }
781
782 if err := u.checkNetwork(record); err != nil {
783 return nil, err
784 }
785
786 if max := u.srv.Config().MaxUserNetworks; max >= 0 && len(u.networks) >= max {
787 return nil, fmt.Errorf("maximum number of networks reached")
788 }
789
790 network := newNetwork(u, record, nil)
791 err := u.srv.db.StoreNetwork(ctx, u.ID, &network.Network)
792 if err != nil {
793 return nil, err
794 }
795
796 u.addNetwork(network)
797
798 idStr := fmt.Sprintf("%v", network.ID)
799 attrs := getNetworkAttrs(network)
800 u.forEachDownstream(func(dc *downstreamConn) {
801 if dc.caps["soju.im/bouncer-networks-notify"] {
802 dc.SendMessage(&irc.Message{
803 Prefix: dc.srv.prefix(),
804 Command: "BOUNCER",
805 Params: []string{"NETWORK", idStr, attrs.String()},
806 })
807 }
808 })
809
810 return network, nil
811}
812
813func (u *user) updateNetwork(ctx context.Context, record *Network) (*network, error) {
814 if record.ID == 0 {
815 panic("tried updating a new network")
816 }
817
818 // If the realname is reset to the default, just wipe the per-network
819 // setting
820 if record.Realname == u.Realname {
821 record.Realname = ""
822 }
823
824 if err := u.checkNetwork(record); err != nil {
825 return nil, err
826 }
827
828 network := u.getNetworkByID(record.ID)
829 if network == nil {
830 panic("tried updating a non-existing network")
831 }
832
833 if err := u.srv.db.StoreNetwork(ctx, u.ID, record); err != nil {
834 return nil, err
835 }
836
837 // Most network changes require us to re-connect to the upstream server
838
839 channels := make([]Channel, 0, network.channels.Len())
840 for _, entry := range network.channels.innerMap {
841 ch := entry.value.(*Channel)
842 channels = append(channels, *ch)
843 }
844
845 updatedNetwork := newNetwork(u, record, channels)
846
847 // If we're currently connected, disconnect and perform the necessary
848 // bookkeeping
849 if network.conn != nil {
850 network.stop()
851 // Note: this will set network.conn to nil
852 u.handleUpstreamDisconnected(network.conn)
853 }
854
855 // Patch downstream connections to use our fresh updated network
856 u.forEachDownstream(func(dc *downstreamConn) {
857 if dc.network != nil && dc.network == network {
858 dc.network = updatedNetwork
859 }
860 })
861
862 // We need to remove the network after patching downstream connections,
863 // otherwise they'll get closed
864 u.removeNetwork(network)
865
866 // The filesystem message store needs to be notified whenever the network
867 // is renamed
868 fsMsgStore, isFS := u.msgStore.(*fsMessageStore)
869 if isFS && updatedNetwork.GetName() != network.GetName() {
870 if err := fsMsgStore.RenameNetwork(&network.Network, &updatedNetwork.Network); err != nil {
871 network.logger.Printf("failed to update FS message store network name to %q: %v", updatedNetwork.GetName(), err)
872 }
873 }
874
875 // This will re-connect to the upstream server
876 u.addNetwork(updatedNetwork)
877
878 // TODO: only broadcast attributes that have changed
879 idStr := fmt.Sprintf("%v", updatedNetwork.ID)
880 attrs := getNetworkAttrs(updatedNetwork)
881 u.forEachDownstream(func(dc *downstreamConn) {
882 if dc.caps["soju.im/bouncer-networks-notify"] {
883 dc.SendMessage(&irc.Message{
884 Prefix: dc.srv.prefix(),
885 Command: "BOUNCER",
886 Params: []string{"NETWORK", idStr, attrs.String()},
887 })
888 }
889 })
890
891 return updatedNetwork, nil
892}
893
894func (u *user) deleteNetwork(ctx context.Context, id int64) error {
895 network := u.getNetworkByID(id)
896 if network == nil {
897 panic("tried deleting a non-existing network")
898 }
899
900 if err := u.srv.db.DeleteNetwork(ctx, network.ID); err != nil {
901 return err
902 }
903
904 u.removeNetwork(network)
905
906 idStr := fmt.Sprintf("%v", network.ID)
907 u.forEachDownstream(func(dc *downstreamConn) {
908 if dc.caps["soju.im/bouncer-networks-notify"] {
909 dc.SendMessage(&irc.Message{
910 Prefix: dc.srv.prefix(),
911 Command: "BOUNCER",
912 Params: []string{"NETWORK", idStr, "*"},
913 })
914 }
915 })
916
917 return nil
918}
919
920func (u *user) updateUser(ctx context.Context, record *User) error {
921 if u.ID != record.ID {
922 panic("ID mismatch when updating user")
923 }
924
925 realnameUpdated := u.Realname != record.Realname
926 if err := u.srv.db.StoreUser(ctx, record); err != nil {
927 return fmt.Errorf("failed to update user %q: %v", u.Username, err)
928 }
929 u.User = *record
930
931 if realnameUpdated {
932 // Re-connect to networks which use the default realname
933 var needUpdate []Network
934 u.forEachNetwork(func(net *network) {
935 if net.Realname == "" {
936 needUpdate = append(needUpdate, net.Network)
937 }
938 })
939
940 var netErr error
941 for _, net := range needUpdate {
942 if _, err := u.updateNetwork(ctx, &net); err != nil {
943 netErr = err
944 }
945 }
946 if netErr != nil {
947 return netErr
948 }
949 }
950
951 return nil
952}
953
954func (u *user) stop() {
955 u.events <- eventStop{}
956 <-u.done
957}
958
959func (u *user) hasPersistentMsgStore() bool {
960 if u.msgStore == nil {
961 return false
962 }
963 _, isMem := u.msgStore.(*memoryMessageStore)
964 return !isMem
965}
966
967// localAddrForHost returns the local address to use when connecting to host.
968// A nil address is returned when the OS should automatically pick one.
969func (u *user) localTCPAddrForHost(host string) (*net.TCPAddr, error) {
970 upstreamUserIPs := u.srv.Config().UpstreamUserIPs
971 if len(upstreamUserIPs) == 0 {
972 return nil, nil
973 }
974
975 ips, err := net.LookupIP(host)
976 if err != nil {
977 return nil, err
978 }
979
980 wantIPv6 := false
981 for _, ip := range ips {
982 if ip.To4() == nil {
983 wantIPv6 = true
984 break
985 }
986 }
987
988 var ipNet *net.IPNet
989 for _, in := range upstreamUserIPs {
990 if wantIPv6 == (in.IP.To4() == nil) {
991 ipNet = in
992 break
993 }
994 }
995 if ipNet == nil {
996 return nil, nil
997 }
998
999 var ipInt big.Int
1000 ipInt.SetBytes(ipNet.IP)
1001 ipInt.Add(&ipInt, big.NewInt(u.ID+1))
1002 ip := net.IP(ipInt.Bytes())
1003 if !ipNet.Contains(ip) {
1004 return nil, fmt.Errorf("IP network %v too small", ipNet)
1005 }
1006
1007 return &net.TCPAddr{IP: ip}, nil
1008}
Note: See TracBrowser for help on using the repository browser.