source: code/trunk/user.go@ 680

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

Add context support to user and network mutations

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