source: code/trunk/user.go@ 467

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

Send NOTICE to downstream when upstream is disconnected

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