source: code/trunk/user.go@ 495

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

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