source: code/trunk/user.go@ 480

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

Introduce deliveredClientMap

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