source: code/trunk/user.go@ 483

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

Move network.clients to user

No need to have this list per-network.
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 lastError error
69 casemap casemapping
70}
71
72func newNetwork(user *user, record *Network, channels []Channel) *network {
73 m := channelCasemapMap{newCasemapMap(0)}
74 for _, ch := range channels {
75 ch := ch
76 m.SetValue(ch.Name, &ch)
77 }
78
79 return &network{
80 Network: *record,
81 user: user,
82 stopped: make(chan struct{}),
83 channels: m,
84 delivered: deliveredCasemapMap{newCasemapMap(0)},
85 casemap: casemapRFC1459,
86 }
87}
88
89func (net *network) forEachDownstream(f func(*downstreamConn)) {
90 net.user.forEachDownstream(func(dc *downstreamConn) {
91 if dc.network != nil && dc.network != net {
92 return
93 }
94 f(dc)
95 })
96}
97
98func (net *network) isStopped() bool {
99 select {
100 case <-net.stopped:
101 return true
102 default:
103 return false
104 }
105}
106
107func userIdent(u *User) string {
108 // The ident is a string we will send to upstream servers in clear-text.
109 // For privacy reasons, make sure it doesn't expose any meaningful user
110 // metadata. We just use the base64-encoded hashed ID, so that people don't
111 // start relying on the string being an integer or following a pattern.
112 var b [64]byte
113 binary.LittleEndian.PutUint64(b[:], uint64(u.ID))
114 h := sha256.Sum256(b[:])
115 return hex.EncodeToString(h[:16])
116}
117
118func (net *network) run() {
119 var lastTry time.Time
120 for {
121 if net.isStopped() {
122 return
123 }
124
125 if dur := time.Now().Sub(lastTry); dur < retryConnectDelay {
126 delay := retryConnectDelay - dur
127 net.user.srv.Logger.Printf("waiting %v before trying to reconnect to %q", delay.Truncate(time.Second), net.Addr)
128 time.Sleep(delay)
129 }
130 lastTry = time.Now()
131
132 uc, err := connectToUpstream(net)
133 if err != nil {
134 net.user.srv.Logger.Printf("failed to connect to upstream server %q: %v", net.Addr, err)
135 net.user.events <- eventUpstreamConnectionError{net, fmt.Errorf("failed to connect: %v", err)}
136 continue
137 }
138
139 if net.user.srv.Identd != nil {
140 net.user.srv.Identd.Store(uc.RemoteAddr().String(), uc.LocalAddr().String(), userIdent(&net.user.User))
141 }
142
143 uc.register()
144 if err := uc.runUntilRegistered(); err != nil {
145 text := err.Error()
146 if regErr, ok := err.(registrationError); ok {
147 text = string(regErr)
148 }
149 uc.logger.Printf("failed to register: %v", text)
150 net.user.events <- eventUpstreamConnectionError{net, fmt.Errorf("failed to register: %v", text)}
151 uc.Close()
152 continue
153 }
154
155 // TODO: this is racy with net.stopped. If the network is stopped
156 // before the user goroutine receives eventUpstreamConnected, the
157 // connection won't be closed.
158 net.user.events <- eventUpstreamConnected{uc}
159 if err := uc.readMessages(net.user.events); err != nil {
160 uc.logger.Printf("failed to handle messages: %v", err)
161 net.user.events <- eventUpstreamError{uc, fmt.Errorf("failed to handle messages: %v", err)}
162 }
163 uc.Close()
164 net.user.events <- eventUpstreamDisconnected{uc}
165
166 if net.user.srv.Identd != nil {
167 net.user.srv.Identd.Delete(uc.RemoteAddr().String(), uc.LocalAddr().String())
168 }
169 }
170}
171
172func (net *network) stop() {
173 if !net.isStopped() {
174 close(net.stopped)
175 }
176
177 if net.conn != nil {
178 net.conn.Close()
179 }
180}
181
182func (net *network) detach(ch *Channel) {
183 if ch.Detached {
184 return
185 }
186 ch.Detached = true
187 net.user.srv.Logger.Printf("network %q: detaching channel %q", net.GetName(), ch.Name)
188
189 if net.conn != nil {
190 uch := net.conn.channels.Value(ch.Name)
191 if uch != nil {
192 uch.updateAutoDetach(0)
193 }
194 }
195
196 net.forEachDownstream(func(dc *downstreamConn) {
197 dc.SendMessage(&irc.Message{
198 Prefix: dc.prefix(),
199 Command: "PART",
200 Params: []string{dc.marshalEntity(net, ch.Name), "Detach"},
201 })
202 })
203}
204
205func (net *network) attach(ch *Channel) {
206 if !ch.Detached {
207 return
208 }
209 ch.Detached = false
210 net.user.srv.Logger.Printf("network %q: attaching channel %q", net.GetName(), ch.Name)
211
212 var uch *upstreamChannel
213 if net.conn != nil {
214 uch = net.conn.channels.Value(ch.Name)
215
216 net.conn.updateChannelAutoDetach(ch.Name)
217 }
218
219 net.forEachDownstream(func(dc *downstreamConn) {
220 dc.SendMessage(&irc.Message{
221 Prefix: dc.prefix(),
222 Command: "JOIN",
223 Params: []string{dc.marshalEntity(net, ch.Name)},
224 })
225
226 if uch != nil {
227 forwardChannel(dc, uch)
228 }
229
230 dc.sendTargetBacklog(net, ch.Name)
231 })
232}
233
234func (net *network) deleteChannel(name string) error {
235 ch := net.channels.Value(name)
236 if ch == nil {
237 return fmt.Errorf("unknown channel %q", name)
238 }
239 if net.conn != nil {
240 uch := net.conn.channels.Value(ch.Name)
241 if uch != nil {
242 uch.updateAutoDetach(0)
243 }
244 }
245
246 if err := net.user.srv.db.DeleteChannel(ch.ID); err != nil {
247 return err
248 }
249 net.channels.Delete(name)
250 return nil
251}
252
253func (net *network) updateCasemapping(newCasemap casemapping) {
254 net.casemap = newCasemap
255 net.channels.SetCasemapping(newCasemap)
256 net.delivered.SetCasemapping(newCasemap)
257 if net.conn != nil {
258 net.conn.channels.SetCasemapping(newCasemap)
259 for _, entry := range net.conn.channels.innerMap {
260 uch := entry.value.(*upstreamChannel)
261 uch.Members.SetCasemapping(newCasemap)
262 }
263 }
264}
265
266type user struct {
267 User
268 srv *Server
269
270 events chan event
271 done chan struct{}
272
273 networks []*network
274 downstreamConns []*downstreamConn
275 msgStore messageStore
276 clients map[string]struct{} // indexed by client name
277
278 // LIST commands in progress
279 pendingLISTs []pendingLIST
280}
281
282type pendingLIST struct {
283 downstreamID uint64
284 // list of per-upstream LIST commands not yet sent or completed
285 pendingCommands map[int64]*irc.Message
286}
287
288func newUser(srv *Server, record *User) *user {
289 var msgStore messageStore
290 if srv.LogPath != "" {
291 msgStore = newFSMessageStore(srv.LogPath, record.Username)
292 } else {
293 msgStore = newMemoryMessageStore()
294 }
295
296 return &user{
297 User: *record,
298 srv: srv,
299 events: make(chan event, 64),
300 done: make(chan struct{}),
301 msgStore: msgStore,
302 clients: make(map[string]struct{}),
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 if network.lastError != nil {
450 sendServiceNOTICE(dc, fmt.Sprintf("disconnected from %s: %v", network.GetName(), network.lastError))
451 }
452 })
453
454 u.forEachUpstream(func(uc *upstreamConn) {
455 uc.updateAway()
456 })
457
458 u.clients[dc.clientName] = struct{}{}
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 u.forEachUpstream(func(uc *upstreamConn) {
470 uc.updateAway()
471 })
472 case eventDownstreamMessage:
473 msg, dc := e.msg, e.dc
474 if dc.isClosed() {
475 dc.logger.Printf("ignoring message on closed connection: %v", msg)
476 break
477 }
478 err := dc.handleMessage(msg)
479 if ircErr, ok := err.(ircError); ok {
480 ircErr.Message.Prefix = dc.srv.prefix()
481 dc.SendMessage(ircErr.Message)
482 } else if err != nil {
483 dc.logger.Printf("failed to handle message %q: %v", msg, err)
484 dc.Close()
485 }
486 case eventStop:
487 u.forEachDownstream(func(dc *downstreamConn) {
488 dc.Close()
489 })
490 for _, n := range u.networks {
491 n.stop()
492 }
493 return
494 default:
495 u.srv.Logger.Printf("received unknown event type: %T", e)
496 }
497 }
498}
499
500func (u *user) handleUpstreamDisconnected(uc *upstreamConn) {
501 uc.network.conn = nil
502
503 uc.endPendingLISTs(true)
504
505 for _, entry := range uc.channels.innerMap {
506 uch := entry.value.(*upstreamChannel)
507 uch.updateAutoDetach(0)
508 }
509
510 uc.forEachDownstream(func(dc *downstreamConn) {
511 dc.updateSupportedCaps()
512 })
513
514 if uc.network.lastError == nil {
515 uc.forEachDownstream(func(dc *downstreamConn) {
516 sendServiceNOTICE(dc, fmt.Sprintf("disconnected from %s", uc.network.GetName()))
517 })
518 }
519}
520
521func (u *user) addNetwork(network *network) {
522 u.networks = append(u.networks, network)
523 go network.run()
524}
525
526func (u *user) removeNetwork(network *network) {
527 network.stop()
528
529 u.forEachDownstream(func(dc *downstreamConn) {
530 if dc.network != nil && dc.network == network {
531 dc.Close()
532 }
533 })
534
535 for i, net := range u.networks {
536 if net == network {
537 u.networks = append(u.networks[:i], u.networks[i+1:]...)
538 return
539 }
540 }
541
542 panic("tried to remove a non-existing network")
543}
544
545func (u *user) createNetwork(record *Network) (*network, error) {
546 if record.ID != 0 {
547 panic("tried creating an already-existing network")
548 }
549
550 network := newNetwork(u, record, nil)
551 err := u.srv.db.StoreNetwork(u.ID, &network.Network)
552 if err != nil {
553 return nil, err
554 }
555
556 u.addNetwork(network)
557
558 return network, nil
559}
560
561func (u *user) updateNetwork(record *Network) (*network, error) {
562 if record.ID == 0 {
563 panic("tried updating a new network")
564 }
565
566 network := u.getNetworkByID(record.ID)
567 if network == nil {
568 panic("tried updating a non-existing network")
569 }
570
571 if err := u.srv.db.StoreNetwork(u.ID, record); err != nil {
572 return nil, err
573 }
574
575 // Most network changes require us to re-connect to the upstream server
576
577 channels := make([]Channel, 0, network.channels.Len())
578 for _, entry := range network.channels.innerMap {
579 ch := entry.value.(*Channel)
580 channels = append(channels, *ch)
581 }
582
583 updatedNetwork := newNetwork(u, record, channels)
584
585 // If we're currently connected, disconnect and perform the necessary
586 // bookkeeping
587 if network.conn != nil {
588 network.stop()
589 // Note: this will set network.conn to nil
590 u.handleUpstreamDisconnected(network.conn)
591 }
592
593 // Patch downstream connections to use our fresh updated network
594 u.forEachDownstream(func(dc *downstreamConn) {
595 if dc.network != nil && dc.network == network {
596 dc.network = updatedNetwork
597 }
598 })
599
600 // We need to remove the network after patching downstream connections,
601 // otherwise they'll get closed
602 u.removeNetwork(network)
603
604 // This will re-connect to the upstream server
605 u.addNetwork(updatedNetwork)
606
607 return updatedNetwork, nil
608}
609
610func (u *user) deleteNetwork(id int64) error {
611 network := u.getNetworkByID(id)
612 if network == nil {
613 panic("tried deleting a non-existing network")
614 }
615
616 if err := u.srv.db.DeleteNetwork(network.ID); err != nil {
617 return err
618 }
619
620 u.removeNetwork(network)
621 return nil
622}
623
624func (u *user) updatePassword(hashed string) error {
625 u.User.Password = hashed
626 return u.srv.db.StoreUser(&u.User)
627}
628
629func (u *user) stop() {
630 u.events <- eventStop{}
631 <-u.done
632}
Note: See TracBrowser for help on using the repository browser.