source: code/trunk/user.go@ 697

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

Set isMultiUpstream flag in downstreamConn.welcome()

This will make it easier to globally disable multi-upstream mode.

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