source: code/trunk/user.go@ 423

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

Add message store abstraction

Introduce a messageStore type, which will allow for multiple
implementations (e.g. in the DB or in-memory instead of on-disk).

The message store is per-user so that we don't need to deal with locking
and it's easier to implement per-user limits.
File size: 13.6 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 msgStore *messageStore
253
254 // LIST commands in progress
255 pendingLISTs []pendingLIST
256}
257
258type pendingLIST struct {
259 downstreamID uint64
260 // list of per-upstream LIST commands not yet sent or completed
261 pendingCommands map[int64]*irc.Message
262}
263
264func newUser(srv *Server, record *User) *user {
265 var msgStore *messageStore
266 if srv.LogPath != "" {
267 msgStore = newMessageStore(srv.LogPath, record.Username)
268 }
269
270 return &user{
271 User: *record,
272 srv: srv,
273 events: make(chan event, 64),
274 done: make(chan struct{}),
275 msgStore: msgStore,
276 }
277}
278
279func (u *user) forEachNetwork(f func(*network)) {
280 for _, network := range u.networks {
281 f(network)
282 }
283}
284
285func (u *user) forEachUpstream(f func(uc *upstreamConn)) {
286 for _, network := range u.networks {
287 if network.conn == nil {
288 continue
289 }
290 f(network.conn)
291 }
292}
293
294func (u *user) forEachDownstream(f func(dc *downstreamConn)) {
295 for _, dc := range u.downstreamConns {
296 f(dc)
297 }
298}
299
300func (u *user) getNetwork(name string) *network {
301 for _, network := range u.networks {
302 if network.Addr == name {
303 return network
304 }
305 if network.Name != "" && network.Name == name {
306 return network
307 }
308 }
309 return nil
310}
311
312func (u *user) getNetworkByID(id int64) *network {
313 for _, net := range u.networks {
314 if net.ID == id {
315 return net
316 }
317 }
318 return nil
319}
320
321func (u *user) run() {
322 defer func() {
323 if u.msgStore != nil {
324 if err := u.msgStore.Close(); err != nil {
325 u.srv.Logger.Printf("failed to close message store for user %q: %v", u.Username, err)
326 }
327 }
328 close(u.done)
329 }()
330
331 networks, err := u.srv.db.ListNetworks(u.ID)
332 if err != nil {
333 u.srv.Logger.Printf("failed to list networks for user %q: %v", u.Username, err)
334 return
335 }
336
337 for _, record := range networks {
338 record := record
339 channels, err := u.srv.db.ListChannels(record.ID)
340 if err != nil {
341 u.srv.Logger.Printf("failed to list channels for user %q, network %q: %v", u.Username, record.GetName(), err)
342 continue
343 }
344
345 network := newNetwork(u, &record, channels)
346 u.networks = append(u.networks, network)
347
348 go network.run()
349 }
350
351 for e := range u.events {
352 switch e := e.(type) {
353 case eventUpstreamConnected:
354 uc := e.uc
355
356 uc.network.conn = uc
357
358 uc.updateAway()
359
360 uc.forEachDownstream(func(dc *downstreamConn) {
361 dc.updateSupportedCaps()
362 sendServiceNOTICE(dc, fmt.Sprintf("connected to %s", uc.network.GetName()))
363
364 dc.updateNick()
365 })
366 uc.network.lastError = nil
367 case eventUpstreamDisconnected:
368 u.handleUpstreamDisconnected(e.uc)
369 case eventUpstreamConnectionError:
370 net := e.net
371
372 stopped := false
373 select {
374 case <-net.stopped:
375 stopped = true
376 default:
377 }
378
379 if !stopped && (net.lastError == nil || net.lastError.Error() != e.err.Error()) {
380 net.forEachDownstream(func(dc *downstreamConn) {
381 sendServiceNOTICE(dc, fmt.Sprintf("failed connecting/registering to %s: %v", net.GetName(), e.err))
382 })
383 }
384 net.lastError = e.err
385 case eventUpstreamError:
386 uc := e.uc
387
388 uc.forEachDownstream(func(dc *downstreamConn) {
389 sendServiceNOTICE(dc, fmt.Sprintf("disconnected from %s: %v", uc.network.GetName(), e.err))
390 })
391 uc.network.lastError = e.err
392 case eventUpstreamMessage:
393 msg, uc := e.msg, e.uc
394 if uc.isClosed() {
395 uc.logger.Printf("ignoring message on closed connection: %v", msg)
396 break
397 }
398 if err := uc.handleMessage(msg); err != nil {
399 uc.logger.Printf("failed to handle message %q: %v", msg, err)
400 }
401 case eventDownstreamConnected:
402 dc := e.dc
403
404 if err := dc.welcome(); err != nil {
405 dc.logger.Printf("failed to handle new registered connection: %v", err)
406 break
407 }
408
409 u.downstreamConns = append(u.downstreamConns, dc)
410
411 u.forEachUpstream(func(uc *upstreamConn) {
412 uc.updateAway()
413 })
414
415 dc.updateSupportedCaps()
416 case eventDownstreamDisconnected:
417 dc := e.dc
418
419 for i := range u.downstreamConns {
420 if u.downstreamConns[i] == dc {
421 u.downstreamConns = append(u.downstreamConns[:i], u.downstreamConns[i+1:]...)
422 break
423 }
424 }
425
426 // Save history if we're the last client with this name
427 skipHistory := make(map[*network]bool)
428 u.forEachDownstream(func(conn *downstreamConn) {
429 if dc.clientName == conn.clientName {
430 skipHistory[conn.network] = true
431 }
432 })
433
434 dc.forEachNetwork(func(net *network) {
435 if skipHistory[net] || skipHistory[nil] {
436 return
437 }
438
439 net.offlineClients[dc.clientName] = struct{}{}
440 })
441
442 u.forEachUpstream(func(uc *upstreamConn) {
443 uc.updateAway()
444 })
445 case eventDownstreamMessage:
446 msg, dc := e.msg, e.dc
447 if dc.isClosed() {
448 dc.logger.Printf("ignoring message on closed connection: %v", msg)
449 break
450 }
451 err := dc.handleMessage(msg)
452 if ircErr, ok := err.(ircError); ok {
453 ircErr.Message.Prefix = dc.srv.prefix()
454 dc.SendMessage(ircErr.Message)
455 } else if err != nil {
456 dc.logger.Printf("failed to handle message %q: %v", msg, err)
457 dc.Close()
458 }
459 case eventStop:
460 u.forEachDownstream(func(dc *downstreamConn) {
461 dc.Close()
462 })
463 for _, n := range u.networks {
464 n.stop()
465 }
466 return
467 default:
468 u.srv.Logger.Printf("received unknown event type: %T", e)
469 }
470 }
471}
472
473func (u *user) handleUpstreamDisconnected(uc *upstreamConn) {
474 uc.network.conn = nil
475
476 uc.endPendingLISTs(true)
477
478 uc.forEachDownstream(func(dc *downstreamConn) {
479 dc.updateSupportedCaps()
480 })
481
482 if uc.network.lastError == nil {
483 uc.forEachDownstream(func(dc *downstreamConn) {
484 sendServiceNOTICE(dc, fmt.Sprintf("disconnected from %s", uc.network.GetName()))
485 })
486 }
487}
488
489func (u *user) addNetwork(network *network) {
490 u.networks = append(u.networks, network)
491 go network.run()
492}
493
494func (u *user) removeNetwork(network *network) {
495 network.stop()
496
497 u.forEachDownstream(func(dc *downstreamConn) {
498 if dc.network != nil && dc.network == network {
499 dc.Close()
500 }
501 })
502
503 for i, net := range u.networks {
504 if net == network {
505 u.networks = append(u.networks[:i], u.networks[i+1:]...)
506 return
507 }
508 }
509
510 panic("tried to remove a non-existing network")
511}
512
513func (u *user) createNetwork(record *Network) (*network, error) {
514 if record.ID != 0 {
515 panic("tried creating an already-existing network")
516 }
517
518 network := newNetwork(u, record, nil)
519 err := u.srv.db.StoreNetwork(u.ID, &network.Network)
520 if err != nil {
521 return nil, err
522 }
523
524 u.addNetwork(network)
525
526 return network, nil
527}
528
529func (u *user) updateNetwork(record *Network) (*network, error) {
530 if record.ID == 0 {
531 panic("tried updating a new network")
532 }
533
534 network := u.getNetworkByID(record.ID)
535 if network == nil {
536 panic("tried updating a non-existing network")
537 }
538
539 if err := u.srv.db.StoreNetwork(u.ID, record); err != nil {
540 return nil, err
541 }
542
543 // Most network changes require us to re-connect to the upstream server
544
545 channels := make([]Channel, 0, len(network.channels))
546 for _, ch := range network.channels {
547 channels = append(channels, *ch)
548 }
549
550 updatedNetwork := newNetwork(u, record, channels)
551
552 // If we're currently connected, disconnect and perform the necessary
553 // bookkeeping
554 if network.conn != nil {
555 network.stop()
556 // Note: this will set network.conn to nil
557 u.handleUpstreamDisconnected(network.conn)
558 }
559
560 // Patch downstream connections to use our fresh updated network
561 u.forEachDownstream(func(dc *downstreamConn) {
562 if dc.network != nil && dc.network == network {
563 dc.network = updatedNetwork
564 }
565 })
566
567 // We need to remove the network after patching downstream connections,
568 // otherwise they'll get closed
569 u.removeNetwork(network)
570
571 // This will re-connect to the upstream server
572 u.addNetwork(updatedNetwork)
573
574 return updatedNetwork, nil
575}
576
577func (u *user) deleteNetwork(id int64) error {
578 network := u.getNetworkByID(id)
579 if network == nil {
580 panic("tried deleting a non-existing network")
581 }
582
583 if err := u.srv.db.DeleteNetwork(network.ID); err != nil {
584 return err
585 }
586
587 u.removeNetwork(network)
588 return nil
589}
590
591func (u *user) updatePassword(hashed string) error {
592 u.User.Password = hashed
593 return u.srv.db.StoreUser(&u.User)
594}
595
596func (u *user) stop() {
597 u.events <- eventStop{}
598 <-u.done
599}
Note: See TracBrowser for help on using the repository browser.