source: code/trunk/user.go@ 482

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

Simplify network.offlineClients

Replace it with a list of all clients (online or offline).
File size: 14.2 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 clients 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 clients: 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 dc.SendMessage(&irc.Message{
200 Prefix: dc.prefix(),
201 Command: "PART",
202 Params: []string{dc.marshalEntity(net, ch.Name), "Detach"},
203 })
204 })
205}
206
207func (net *network) attach(ch *Channel) {
208 if !ch.Detached {
209 return
210 }
211 ch.Detached = false
212 net.user.srv.Logger.Printf("network %q: attaching channel %q", net.GetName(), ch.Name)
213
214 var uch *upstreamChannel
215 if net.conn != nil {
216 uch = net.conn.channels.Value(ch.Name)
217
218 net.conn.updateChannelAutoDetach(ch.Name)
219 }
220
221 net.forEachDownstream(func(dc *downstreamConn) {
222 dc.SendMessage(&irc.Message{
223 Prefix: dc.prefix(),
224 Command: "JOIN",
225 Params: []string{dc.marshalEntity(net, ch.Name)},
226 })
227
228 if uch != nil {
229 forwardChannel(dc, uch)
230 }
231
232 dc.sendTargetBacklog(net, ch.Name)
233 })
234}
235
236func (net *network) deleteChannel(name string) error {
237 ch := net.channels.Value(name)
238 if ch == nil {
239 return fmt.Errorf("unknown channel %q", name)
240 }
241 if net.conn != nil {
242 uch := net.conn.channels.Value(ch.Name)
243 if uch != nil {
244 uch.updateAutoDetach(0)
245 }
246 }
247
248 if err := net.user.srv.db.DeleteChannel(ch.ID); err != nil {
249 return err
250 }
251 net.channels.Delete(name)
252 return nil
253}
254
255func (net *network) updateCasemapping(newCasemap casemapping) {
256 net.casemap = newCasemap
257 net.channels.SetCasemapping(newCasemap)
258 net.delivered.SetCasemapping(newCasemap)
259 if net.conn != nil {
260 net.conn.channels.SetCasemapping(newCasemap)
261 for _, entry := range net.conn.channels.innerMap {
262 uch := entry.value.(*upstreamChannel)
263 uch.Members.SetCasemapping(newCasemap)
264 }
265 }
266}
267
268type user struct {
269 User
270 srv *Server
271
272 events chan event
273 done chan struct{}
274
275 networks []*network
276 downstreamConns []*downstreamConn
277 msgStore messageStore
278
279 // LIST commands in progress
280 pendingLISTs []pendingLIST
281}
282
283type pendingLIST struct {
284 downstreamID uint64
285 // list of per-upstream LIST commands not yet sent or completed
286 pendingCommands map[int64]*irc.Message
287}
288
289func newUser(srv *Server, record *User) *user {
290 var msgStore messageStore
291 if srv.LogPath != "" {
292 msgStore = newFSMessageStore(srv.LogPath, record.Username)
293 } else {
294 msgStore = newMemoryMessageStore()
295 }
296
297 return &user{
298 User: *record,
299 srv: srv,
300 events: make(chan event, 64),
301 done: make(chan struct{}),
302 msgStore: msgStore,
303 }
304}
305
306func (u *user) forEachNetwork(f func(*network)) {
307 for _, network := range u.networks {
308 f(network)
309 }
310}
311
312func (u *user) forEachUpstream(f func(uc *upstreamConn)) {
313 for _, network := range u.networks {
314 if network.conn == nil {
315 continue
316 }
317 f(network.conn)
318 }
319}
320
321func (u *user) forEachDownstream(f func(dc *downstreamConn)) {
322 for _, dc := range u.downstreamConns {
323 f(dc)
324 }
325}
326
327func (u *user) getNetwork(name string) *network {
328 for _, network := range u.networks {
329 if network.Addr == name {
330 return network
331 }
332 if network.Name != "" && network.Name == name {
333 return network
334 }
335 }
336 return nil
337}
338
339func (u *user) getNetworkByID(id int64) *network {
340 for _, net := range u.networks {
341 if net.ID == id {
342 return net
343 }
344 }
345 return nil
346}
347
348func (u *user) run() {
349 defer func() {
350 if u.msgStore != nil {
351 if err := u.msgStore.Close(); err != nil {
352 u.srv.Logger.Printf("failed to close message store for user %q: %v", u.Username, err)
353 }
354 }
355 close(u.done)
356 }()
357
358 networks, err := u.srv.db.ListNetworks(u.ID)
359 if err != nil {
360 u.srv.Logger.Printf("failed to list networks for user %q: %v", u.Username, err)
361 return
362 }
363
364 for _, record := range networks {
365 record := record
366 channels, err := u.srv.db.ListChannels(record.ID)
367 if err != nil {
368 u.srv.Logger.Printf("failed to list channels for user %q, network %q: %v", u.Username, record.GetName(), err)
369 continue
370 }
371
372 network := newNetwork(u, &record, channels)
373 u.networks = append(u.networks, network)
374
375 go network.run()
376 }
377
378 for e := range u.events {
379 switch e := e.(type) {
380 case eventUpstreamConnected:
381 uc := e.uc
382
383 uc.network.conn = uc
384
385 uc.updateAway()
386
387 uc.forEachDownstream(func(dc *downstreamConn) {
388 dc.updateSupportedCaps()
389 sendServiceNOTICE(dc, fmt.Sprintf("connected to %s", uc.network.GetName()))
390
391 dc.updateNick()
392 })
393 uc.network.lastError = nil
394 case eventUpstreamDisconnected:
395 u.handleUpstreamDisconnected(e.uc)
396 case eventUpstreamConnectionError:
397 net := e.net
398
399 stopped := false
400 select {
401 case <-net.stopped:
402 stopped = true
403 default:
404 }
405
406 if !stopped && (net.lastError == nil || net.lastError.Error() != e.err.Error()) {
407 net.forEachDownstream(func(dc *downstreamConn) {
408 sendServiceNOTICE(dc, fmt.Sprintf("failed connecting/registering to %s: %v", net.GetName(), e.err))
409 })
410 }
411 net.lastError = e.err
412 case eventUpstreamError:
413 uc := e.uc
414
415 uc.forEachDownstream(func(dc *downstreamConn) {
416 sendServiceNOTICE(dc, fmt.Sprintf("disconnected from %s: %v", uc.network.GetName(), e.err))
417 })
418 uc.network.lastError = e.err
419 case eventUpstreamMessage:
420 msg, uc := e.msg, e.uc
421 if uc.isClosed() {
422 uc.logger.Printf("ignoring message on closed connection: %v", msg)
423 break
424 }
425 if err := uc.handleMessage(msg); err != nil {
426 uc.logger.Printf("failed to handle message %q: %v", msg, err)
427 }
428 case eventChannelDetach:
429 uc, name := e.uc, e.name
430 c := uc.network.channels.Value(name)
431 if c == nil || c.Detached {
432 continue
433 }
434 uc.network.detach(c)
435 if err := uc.srv.db.StoreChannel(uc.network.ID, c); err != nil {
436 u.srv.Logger.Printf("failed to store updated detached channel %q: %v", c.Name, err)
437 }
438 case eventDownstreamConnected:
439 dc := e.dc
440
441 if err := dc.welcome(); err != nil {
442 dc.logger.Printf("failed to handle new registered connection: %v", err)
443 break
444 }
445
446 u.downstreamConns = append(u.downstreamConns, dc)
447
448 dc.forEachNetwork(func(network *network) {
449 network.clients[dc.clientName] = struct{}{}
450 if network.lastError != nil {
451 sendServiceNOTICE(dc, fmt.Sprintf("disconnected from %s: %v", network.GetName(), network.lastError))
452 }
453 })
454
455 u.forEachUpstream(func(uc *upstreamConn) {
456 uc.updateAway()
457 })
458 case eventDownstreamDisconnected:
459 dc := e.dc
460
461 for i := range u.downstreamConns {
462 if u.downstreamConns[i] == dc {
463 u.downstreamConns = append(u.downstreamConns[:i], u.downstreamConns[i+1:]...)
464 break
465 }
466 }
467
468 u.forEachUpstream(func(uc *upstreamConn) {
469 uc.updateAway()
470 })
471 case eventDownstreamMessage:
472 msg, dc := e.msg, e.dc
473 if dc.isClosed() {
474 dc.logger.Printf("ignoring message on closed connection: %v", msg)
475 break
476 }
477 err := dc.handleMessage(msg)
478 if ircErr, ok := err.(ircError); ok {
479 ircErr.Message.Prefix = dc.srv.prefix()
480 dc.SendMessage(ircErr.Message)
481 } else if err != nil {
482 dc.logger.Printf("failed to handle message %q: %v", msg, err)
483 dc.Close()
484 }
485 case eventStop:
486 u.forEachDownstream(func(dc *downstreamConn) {
487 dc.Close()
488 })
489 for _, n := range u.networks {
490 n.stop()
491 }
492 return
493 default:
494 u.srv.Logger.Printf("received unknown event type: %T", e)
495 }
496 }
497}
498
499func (u *user) handleUpstreamDisconnected(uc *upstreamConn) {
500 uc.network.conn = nil
501
502 uc.endPendingLISTs(true)
503
504 for _, entry := range uc.channels.innerMap {
505 uch := entry.value.(*upstreamChannel)
506 uch.updateAutoDetach(0)
507 }
508
509 uc.forEachDownstream(func(dc *downstreamConn) {
510 dc.updateSupportedCaps()
511 })
512
513 if uc.network.lastError == nil {
514 uc.forEachDownstream(func(dc *downstreamConn) {
515 sendServiceNOTICE(dc, fmt.Sprintf("disconnected from %s", uc.network.GetName()))
516 })
517 }
518}
519
520func (u *user) addNetwork(network *network) {
521 u.networks = append(u.networks, network)
522 go network.run()
523}
524
525func (u *user) removeNetwork(network *network) {
526 network.stop()
527
528 u.forEachDownstream(func(dc *downstreamConn) {
529 if dc.network != nil && dc.network == network {
530 dc.Close()
531 }
532 })
533
534 for i, net := range u.networks {
535 if net == network {
536 u.networks = append(u.networks[:i], u.networks[i+1:]...)
537 return
538 }
539 }
540
541 panic("tried to remove a non-existing network")
542}
543
544func (u *user) createNetwork(record *Network) (*network, error) {
545 if record.ID != 0 {
546 panic("tried creating an already-existing network")
547 }
548
549 network := newNetwork(u, record, nil)
550 err := u.srv.db.StoreNetwork(u.ID, &network.Network)
551 if err != nil {
552 return nil, err
553 }
554
555 u.addNetwork(network)
556
557 return network, nil
558}
559
560func (u *user) updateNetwork(record *Network) (*network, error) {
561 if record.ID == 0 {
562 panic("tried updating a new network")
563 }
564
565 network := u.getNetworkByID(record.ID)
566 if network == nil {
567 panic("tried updating a non-existing network")
568 }
569
570 if err := u.srv.db.StoreNetwork(u.ID, record); err != nil {
571 return nil, err
572 }
573
574 // Most network changes require us to re-connect to the upstream server
575
576 channels := make([]Channel, 0, network.channels.Len())
577 for _, entry := range network.channels.innerMap {
578 ch := entry.value.(*Channel)
579 channels = append(channels, *ch)
580 }
581
582 updatedNetwork := newNetwork(u, record, channels)
583
584 // If we're currently connected, disconnect and perform the necessary
585 // bookkeeping
586 if network.conn != nil {
587 network.stop()
588 // Note: this will set network.conn to nil
589 u.handleUpstreamDisconnected(network.conn)
590 }
591
592 // Patch downstream connections to use our fresh updated network
593 u.forEachDownstream(func(dc *downstreamConn) {
594 if dc.network != nil && dc.network == network {
595 dc.network = updatedNetwork
596 }
597 })
598
599 // We need to remove the network after patching downstream connections,
600 // otherwise they'll get closed
601 u.removeNetwork(network)
602
603 // This will re-connect to the upstream server
604 u.addNetwork(updatedNetwork)
605
606 return updatedNetwork, nil
607}
608
609func (u *user) deleteNetwork(id int64) error {
610 network := u.getNetworkByID(id)
611 if network == nil {
612 panic("tried deleting a non-existing network")
613 }
614
615 if err := u.srv.db.DeleteNetwork(network.ID); err != nil {
616 return err
617 }
618
619 u.removeNetwork(network)
620 return nil
621}
622
623func (u *user) updatePassword(hashed string) error {
624 u.User.Password = hashed
625 return u.srv.db.StoreUser(&u.User)
626}
627
628func (u *user) stop() {
629 u.events <- eventStop{}
630 <-u.done
631}
Note: See TracBrowser for help on using the repository browser.