source: code/trunk/user.go@ 445

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

Add in-memory message store

Uses an in-memory ring buffer.

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