source: code/trunk/user.go@ 485

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

Introduce deliveredStore

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