source: code/trunk/user.go@ 507

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

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