source: code/trunk/user.go@ 360

Last change on this file since 360 was 359, checked in by contact, 5 years ago

Prevent error handler from falling through in user.run

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