source: code/trunk/user.go@ 420

Last change on this file since 420 was 416, checked in by hubert, 5 years ago

Make DB.DeleteChannel take the channel ID

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