source: code/trunk/msgstore_fs.go@ 792

Last change on this file since 792 was 787, checked in by contact, 3 years ago

msgstore_fs: fix direct message targets

When fetching messages via draft/chathistory from a conversation
with another user, soju would send the following:

:sender PRIVMSG sender :hey

instead of

:sender PRIVMSG recipient :hey

because the file-system message store format doesn't contain the
original PRIVMSG target.

Fix this by doing some guesswork.
File size: 18.2 KB
Line 
1package soju
2
3import (
4 "bufio"
5 "context"
6 "fmt"
7 "io"
8 "os"
9 "path/filepath"
10 "sort"
11 "strings"
12 "time"
13
14 "git.sr.ht/~sircmpwn/go-bare"
15 "gopkg.in/irc.v3"
16)
17
18const (
19 fsMessageStoreMaxFiles = 20
20 fsMessageStoreMaxTries = 100
21)
22
23func escapeFilename(unsafe string) (safe string) {
24 if unsafe == "." {
25 return "-"
26 } else if unsafe == ".." {
27 return "--"
28 } else {
29 return strings.NewReplacer("/", "-", "\\", "-").Replace(unsafe)
30 }
31}
32
33type date struct {
34 Year, Month, Day int
35}
36
37func newDate(t time.Time) date {
38 year, month, day := t.Date()
39 return date{year, int(month), day}
40}
41
42func (d date) Time() time.Time {
43 return time.Date(d.Year, time.Month(d.Month), d.Day, 0, 0, 0, 0, time.Local)
44}
45
46type fsMsgID struct {
47 Date date
48 Offset bare.Int
49}
50
51func (fsMsgID) msgIDType() msgIDType {
52 return msgIDFS
53}
54
55func parseFSMsgID(s string) (netID int64, entity string, t time.Time, offset int64, err error) {
56 var id fsMsgID
57 netID, entity, err = parseMsgID(s, &id)
58 if err != nil {
59 return 0, "", time.Time{}, 0, err
60 }
61 return netID, entity, id.Date.Time(), int64(id.Offset), nil
62}
63
64func formatFSMsgID(netID int64, entity string, t time.Time, offset int64) string {
65 id := fsMsgID{
66 Date: newDate(t),
67 Offset: bare.Int(offset),
68 }
69 return formatMsgID(netID, entity, &id)
70}
71
72type fsMessageStoreFile struct {
73 *os.File
74 lastUse time.Time
75}
76
77// fsMessageStore is a per-user on-disk store for IRC messages.
78//
79// It mimicks the ZNC log layout and format. See the ZNC source:
80// https://github.com/znc/znc/blob/master/modules/log.cpp
81type fsMessageStore struct {
82 root string
83 user *User
84
85 // Write-only files used by Append
86 files map[string]*fsMessageStoreFile // indexed by entity
87}
88
89var _ messageStore = (*fsMessageStore)(nil)
90var _ chatHistoryMessageStore = (*fsMessageStore)(nil)
91
92func newFSMessageStore(root string, user *User) *fsMessageStore {
93 return &fsMessageStore{
94 root: filepath.Join(root, escapeFilename(user.Username)),
95 user: user,
96 files: make(map[string]*fsMessageStoreFile),
97 }
98}
99
100func (ms *fsMessageStore) logPath(network *Network, entity string, t time.Time) string {
101 year, month, day := t.Date()
102 filename := fmt.Sprintf("%04d-%02d-%02d.log", year, month, day)
103 return filepath.Join(ms.root, escapeFilename(network.GetName()), escapeFilename(entity), filename)
104}
105
106// nextMsgID queries the message ID for the next message to be written to f.
107func nextFSMsgID(network *Network, entity string, t time.Time, f *os.File) (string, error) {
108 offset, err := f.Seek(0, io.SeekEnd)
109 if err != nil {
110 return "", fmt.Errorf("failed to query next FS message ID: %v", err)
111 }
112 return formatFSMsgID(network.ID, entity, t, offset), nil
113}
114
115func (ms *fsMessageStore) LastMsgID(network *Network, entity string, t time.Time) (string, error) {
116 p := ms.logPath(network, entity, t)
117 fi, err := os.Stat(p)
118 if os.IsNotExist(err) {
119 return formatFSMsgID(network.ID, entity, t, -1), nil
120 } else if err != nil {
121 return "", fmt.Errorf("failed to query last FS message ID: %v", err)
122 }
123 return formatFSMsgID(network.ID, entity, t, fi.Size()-1), nil
124}
125
126func (ms *fsMessageStore) Append(network *Network, entity string, msg *irc.Message) (string, error) {
127 s := formatMessage(msg)
128 if s == "" {
129 return "", nil
130 }
131
132 var t time.Time
133 if tag, ok := msg.Tags["time"]; ok {
134 var err error
135 t, err = time.Parse(serverTimeLayout, string(tag))
136 if err != nil {
137 return "", fmt.Errorf("failed to parse message time tag: %v", err)
138 }
139 t = t.In(time.Local)
140 } else {
141 t = time.Now()
142 }
143
144 f := ms.files[entity]
145
146 // TODO: handle non-monotonic clock behaviour
147 path := ms.logPath(network, entity, t)
148 if f == nil || f.Name() != path {
149 dir := filepath.Dir(path)
150 if err := os.MkdirAll(dir, 0750); err != nil {
151 return "", fmt.Errorf("failed to create message logs directory %q: %v", dir, err)
152 }
153
154 ff, err := os.OpenFile(path, os.O_RDWR|os.O_CREATE|os.O_APPEND, 0640)
155 if err != nil {
156 return "", fmt.Errorf("failed to open message log file %q: %v", path, err)
157 }
158
159 if f != nil {
160 f.Close()
161 }
162 f = &fsMessageStoreFile{File: ff}
163 ms.files[entity] = f
164 }
165
166 f.lastUse = time.Now()
167
168 if len(ms.files) > fsMessageStoreMaxFiles {
169 entities := make([]string, 0, len(ms.files))
170 for name := range ms.files {
171 entities = append(entities, name)
172 }
173 sort.Slice(entities, func(i, j int) bool {
174 a, b := entities[i], entities[j]
175 return ms.files[a].lastUse.Before(ms.files[b].lastUse)
176 })
177 entities = entities[0 : len(entities)-fsMessageStoreMaxFiles]
178 for _, name := range entities {
179 ms.files[name].Close()
180 delete(ms.files, name)
181 }
182 }
183
184 msgID, err := nextFSMsgID(network, entity, t, f.File)
185 if err != nil {
186 return "", fmt.Errorf("failed to generate message ID: %v", err)
187 }
188
189 _, err = fmt.Fprintf(f, "[%02d:%02d:%02d] %s\n", t.Hour(), t.Minute(), t.Second(), s)
190 if err != nil {
191 return "", fmt.Errorf("failed to log message to %q: %v", f.Name(), err)
192 }
193
194 return msgID, nil
195}
196
197func (ms *fsMessageStore) Close() error {
198 var closeErr error
199 for _, f := range ms.files {
200 if err := f.Close(); err != nil {
201 closeErr = fmt.Errorf("failed to close message store: %v", err)
202 }
203 }
204 return closeErr
205}
206
207// formatMessage formats a message log line. It assumes a well-formed IRC
208// message.
209func formatMessage(msg *irc.Message) string {
210 switch strings.ToUpper(msg.Command) {
211 case "NICK":
212 return fmt.Sprintf("*** %s is now known as %s", msg.Prefix.Name, msg.Params[0])
213 case "JOIN":
214 return fmt.Sprintf("*** Joins: %s (%s@%s)", msg.Prefix.Name, msg.Prefix.User, msg.Prefix.Host)
215 case "PART":
216 var reason string
217 if len(msg.Params) > 1 {
218 reason = msg.Params[1]
219 }
220 return fmt.Sprintf("*** Parts: %s (%s@%s) (%s)", msg.Prefix.Name, msg.Prefix.User, msg.Prefix.Host, reason)
221 case "KICK":
222 nick := msg.Params[1]
223 var reason string
224 if len(msg.Params) > 2 {
225 reason = msg.Params[2]
226 }
227 return fmt.Sprintf("*** %s was kicked by %s (%s)", nick, msg.Prefix.Name, reason)
228 case "QUIT":
229 var reason string
230 if len(msg.Params) > 0 {
231 reason = msg.Params[0]
232 }
233 return fmt.Sprintf("*** Quits: %s (%s@%s) (%s)", msg.Prefix.Name, msg.Prefix.User, msg.Prefix.Host, reason)
234 case "TOPIC":
235 var topic string
236 if len(msg.Params) > 1 {
237 topic = msg.Params[1]
238 }
239 return fmt.Sprintf("*** %s changes topic to '%s'", msg.Prefix.Name, topic)
240 case "MODE":
241 return fmt.Sprintf("*** %s sets mode: %s", msg.Prefix.Name, strings.Join(msg.Params[1:], " "))
242 case "NOTICE":
243 return fmt.Sprintf("-%s- %s", msg.Prefix.Name, msg.Params[1])
244 case "PRIVMSG":
245 if cmd, params, ok := parseCTCPMessage(msg); ok && cmd == "ACTION" {
246 return fmt.Sprintf("* %s %s", msg.Prefix.Name, params)
247 } else {
248 return fmt.Sprintf("<%s> %s", msg.Prefix.Name, msg.Params[1])
249 }
250 default:
251 return ""
252 }
253}
254
255func (ms *fsMessageStore) parseMessage(line string, network *Network, entity string, ref time.Time, events bool) (*irc.Message, time.Time, error) {
256 var hour, minute, second int
257 _, err := fmt.Sscanf(line, "[%02d:%02d:%02d] ", &hour, &minute, &second)
258 if err != nil {
259 return nil, time.Time{}, fmt.Errorf("malformed timestamp prefix: %v", err)
260 }
261 line = line[11:]
262
263 var cmd string
264 var prefix *irc.Prefix
265 var params []string
266 if events && strings.HasPrefix(line, "*** ") {
267 parts := strings.SplitN(line[4:], " ", 2)
268 if len(parts) != 2 {
269 return nil, time.Time{}, nil
270 }
271 switch parts[0] {
272 case "Joins:", "Parts:", "Quits:":
273 args := strings.SplitN(parts[1], " ", 3)
274 if len(args) < 2 {
275 return nil, time.Time{}, nil
276 }
277 nick := args[0]
278 mask := strings.TrimSuffix(strings.TrimPrefix(args[1], "("), ")")
279 maskParts := strings.SplitN(mask, "@", 2)
280 if len(maskParts) != 2 {
281 return nil, time.Time{}, nil
282 }
283 prefix = &irc.Prefix{
284 Name: nick,
285 User: maskParts[0],
286 Host: maskParts[1],
287 }
288 var reason string
289 if len(args) > 2 {
290 reason = strings.TrimSuffix(strings.TrimPrefix(args[2], "("), ")")
291 }
292 switch parts[0] {
293 case "Joins:":
294 cmd = "JOIN"
295 params = []string{entity}
296 case "Parts:":
297 cmd = "PART"
298 if reason != "" {
299 params = []string{entity, reason}
300 } else {
301 params = []string{entity}
302 }
303 case "Quits:":
304 cmd = "QUIT"
305 if reason != "" {
306 params = []string{reason}
307 }
308 }
309 default:
310 nick := parts[0]
311 rem := parts[1]
312 if r := strings.TrimPrefix(rem, "is now known as "); r != rem {
313 cmd = "NICK"
314 prefix = &irc.Prefix{
315 Name: nick,
316 }
317 params = []string{r}
318 } else if r := strings.TrimPrefix(rem, "was kicked by "); r != rem {
319 args := strings.SplitN(r, " ", 2)
320 if len(args) != 2 {
321 return nil, time.Time{}, nil
322 }
323 cmd = "KICK"
324 prefix = &irc.Prefix{
325 Name: args[0],
326 }
327 reason := strings.TrimSuffix(strings.TrimPrefix(args[1], "("), ")")
328 params = []string{entity, nick}
329 if reason != "" {
330 params = append(params, reason)
331 }
332 } else if r := strings.TrimPrefix(rem, "changes topic to "); r != rem {
333 cmd = "TOPIC"
334 prefix = &irc.Prefix{
335 Name: nick,
336 }
337 topic := strings.TrimSuffix(strings.TrimPrefix(r, "'"), "'")
338 params = []string{entity, topic}
339 } else if r := strings.TrimPrefix(rem, "sets mode: "); r != rem {
340 cmd = "MODE"
341 prefix = &irc.Prefix{
342 Name: nick,
343 }
344 params = append([]string{entity}, strings.Split(r, " ")...)
345 } else {
346 return nil, time.Time{}, nil
347 }
348 }
349 } else {
350 var sender, text string
351 if strings.HasPrefix(line, "<") {
352 cmd = "PRIVMSG"
353 parts := strings.SplitN(line[1:], "> ", 2)
354 if len(parts) != 2 {
355 return nil, time.Time{}, nil
356 }
357 sender, text = parts[0], parts[1]
358 } else if strings.HasPrefix(line, "-") {
359 cmd = "NOTICE"
360 parts := strings.SplitN(line[1:], "- ", 2)
361 if len(parts) != 2 {
362 return nil, time.Time{}, nil
363 }
364 sender, text = parts[0], parts[1]
365 } else if strings.HasPrefix(line, "* ") {
366 cmd = "PRIVMSG"
367 parts := strings.SplitN(line[2:], " ", 2)
368 if len(parts) != 2 {
369 return nil, time.Time{}, nil
370 }
371 sender, text = parts[0], "\x01ACTION "+parts[1]+"\x01"
372 } else {
373 return nil, time.Time{}, nil
374 }
375
376 prefix = &irc.Prefix{Name: sender}
377 if entity == sender {
378 // This is a direct message from a user to us. We don't store own
379 // our nickname in the logs, so grab it from the network settings.
380 // Not very accurate since this may not match our nick at the time
381 // the message was received, but we can't do a lot better.
382 entity = GetNick(ms.user, network)
383 }
384 params = []string{entity, text}
385 }
386
387 year, month, day := ref.Date()
388 t := time.Date(year, month, day, hour, minute, second, 0, time.Local)
389
390 msg := &irc.Message{
391 Tags: map[string]irc.TagValue{
392 "time": irc.TagValue(formatServerTime(t)),
393 },
394 Prefix: prefix,
395 Command: cmd,
396 Params: params,
397 }
398 return msg, t, nil
399}
400
401func (ms *fsMessageStore) parseMessagesBefore(network *Network, entity string, ref time.Time, end time.Time, events bool, limit int, afterOffset int64) ([]*irc.Message, error) {
402 path := ms.logPath(network, entity, ref)
403 f, err := os.Open(path)
404 if err != nil {
405 if os.IsNotExist(err) {
406 return nil, nil
407 }
408 return nil, fmt.Errorf("failed to parse messages before ref: %v", err)
409 }
410 defer f.Close()
411
412 historyRing := make([]*irc.Message, limit)
413 cur := 0
414
415 sc := bufio.NewScanner(f)
416
417 if afterOffset >= 0 {
418 if _, err := f.Seek(afterOffset, io.SeekStart); err != nil {
419 return nil, nil
420 }
421 sc.Scan() // skip till next newline
422 }
423
424 for sc.Scan() {
425 msg, t, err := ms.parseMessage(sc.Text(), network, entity, ref, events)
426 if err != nil {
427 return nil, err
428 } else if msg == nil || !t.After(end) {
429 continue
430 } else if !t.Before(ref) {
431 break
432 }
433
434 historyRing[cur%limit] = msg
435 cur++
436 }
437 if sc.Err() != nil {
438 return nil, fmt.Errorf("failed to parse messages before ref: scanner error: %v", sc.Err())
439 }
440
441 n := limit
442 if cur < limit {
443 n = cur
444 }
445 start := (cur - n + limit) % limit
446
447 if start+n <= limit { // ring doesnt wrap
448 return historyRing[start : start+n], nil
449 } else { // ring wraps
450 history := make([]*irc.Message, n)
451 r := copy(history, historyRing[start:])
452 copy(history[r:], historyRing[:n-r])
453 return history, nil
454 }
455}
456
457func (ms *fsMessageStore) parseMessagesAfter(network *Network, entity string, ref time.Time, end time.Time, events bool, limit int) ([]*irc.Message, error) {
458 path := ms.logPath(network, entity, ref)
459 f, err := os.Open(path)
460 if err != nil {
461 if os.IsNotExist(err) {
462 return nil, nil
463 }
464 return nil, fmt.Errorf("failed to parse messages after ref: %v", err)
465 }
466 defer f.Close()
467
468 var history []*irc.Message
469 sc := bufio.NewScanner(f)
470 for sc.Scan() && len(history) < limit {
471 msg, t, err := ms.parseMessage(sc.Text(), network, entity, ref, events)
472 if err != nil {
473 return nil, err
474 } else if msg == nil || !t.After(ref) {
475 continue
476 } else if !t.Before(end) {
477 break
478 }
479
480 history = append(history, msg)
481 }
482 if sc.Err() != nil {
483 return nil, fmt.Errorf("failed to parse messages after ref: scanner error: %v", sc.Err())
484 }
485
486 return history, nil
487}
488
489func (ms *fsMessageStore) LoadBeforeTime(ctx context.Context, network *Network, entity string, start time.Time, end time.Time, limit int, events bool) ([]*irc.Message, error) {
490 start = start.In(time.Local)
491 end = end.In(time.Local)
492 history := make([]*irc.Message, limit)
493 remaining := limit
494 tries := 0
495 for remaining > 0 && tries < fsMessageStoreMaxTries && end.Before(start) {
496 buf, err := ms.parseMessagesBefore(network, entity, start, end, events, remaining, -1)
497 if err != nil {
498 return nil, err
499 }
500 if len(buf) == 0 {
501 tries++
502 } else {
503 tries = 0
504 }
505 copy(history[remaining-len(buf):], buf)
506 remaining -= len(buf)
507 year, month, day := start.Date()
508 start = time.Date(year, month, day, 0, 0, 0, 0, start.Location()).Add(-1)
509
510 if err := ctx.Err(); err != nil {
511 return nil, err
512 }
513 }
514
515 return history[remaining:], nil
516}
517
518func (ms *fsMessageStore) LoadAfterTime(ctx context.Context, network *Network, entity string, start time.Time, end time.Time, limit int, events bool) ([]*irc.Message, error) {
519 start = start.In(time.Local)
520 end = end.In(time.Local)
521 var history []*irc.Message
522 remaining := limit
523 tries := 0
524 for remaining > 0 && tries < fsMessageStoreMaxTries && start.Before(end) {
525 buf, err := ms.parseMessagesAfter(network, entity, start, end, events, remaining)
526 if err != nil {
527 return nil, err
528 }
529 if len(buf) == 0 {
530 tries++
531 } else {
532 tries = 0
533 }
534 history = append(history, buf...)
535 remaining -= len(buf)
536 year, month, day := start.Date()
537 start = time.Date(year, month, day+1, 0, 0, 0, 0, start.Location())
538
539 if err := ctx.Err(); err != nil {
540 return nil, err
541 }
542 }
543 return history, nil
544}
545
546func (ms *fsMessageStore) LoadLatestID(ctx context.Context, network *Network, entity, id string, limit int) ([]*irc.Message, error) {
547 var afterTime time.Time
548 var afterOffset int64
549 if id != "" {
550 var idNet int64
551 var idEntity string
552 var err error
553 idNet, idEntity, afterTime, afterOffset, err = parseFSMsgID(id)
554 if err != nil {
555 return nil, err
556 }
557 if idNet != network.ID || idEntity != entity {
558 return nil, fmt.Errorf("cannot find message ID: message ID doesn't match network/entity")
559 }
560 }
561
562 history := make([]*irc.Message, limit)
563 t := time.Now()
564 remaining := limit
565 tries := 0
566 for remaining > 0 && tries < fsMessageStoreMaxTries && !truncateDay(t).Before(afterTime) {
567 var offset int64 = -1
568 if afterOffset >= 0 && truncateDay(t).Equal(afterTime) {
569 offset = afterOffset
570 }
571
572 buf, err := ms.parseMessagesBefore(network, entity, t, time.Time{}, false, remaining, offset)
573 if err != nil {
574 return nil, err
575 }
576 if len(buf) == 0 {
577 tries++
578 } else {
579 tries = 0
580 }
581 copy(history[remaining-len(buf):], buf)
582 remaining -= len(buf)
583 year, month, day := t.Date()
584 t = time.Date(year, month, day, 0, 0, 0, 0, t.Location()).Add(-1)
585
586 if err := ctx.Err(); err != nil {
587 return nil, err
588 }
589 }
590
591 return history[remaining:], nil
592}
593
594func (ms *fsMessageStore) ListTargets(ctx context.Context, network *Network, start, end time.Time, limit int, events bool) ([]chatHistoryTarget, error) {
595 start = start.In(time.Local)
596 end = end.In(time.Local)
597 rootPath := filepath.Join(ms.root, escapeFilename(network.GetName()))
598 root, err := os.Open(rootPath)
599 if os.IsNotExist(err) {
600 return nil, nil
601 } else if err != nil {
602 return nil, err
603 }
604
605 // The returned targets are escaped, and there is no way to un-escape
606 // TODO: switch to ReadDir (Go 1.16+)
607 targetNames, err := root.Readdirnames(0)
608 root.Close()
609 if err != nil {
610 return nil, err
611 }
612
613 var targets []chatHistoryTarget
614 for _, target := range targetNames {
615 // target is already escaped here
616 targetPath := filepath.Join(rootPath, target)
617 targetDir, err := os.Open(targetPath)
618 if err != nil {
619 return nil, err
620 }
621
622 entries, err := targetDir.Readdir(0)
623 targetDir.Close()
624 if err != nil {
625 return nil, err
626 }
627
628 // We use mtime here, which may give imprecise or incorrect results
629 var t time.Time
630 for _, entry := range entries {
631 if entry.ModTime().After(t) {
632 t = entry.ModTime()
633 }
634 }
635
636 // The timestamps we get from logs have second granularity
637 t = truncateSecond(t)
638
639 // Filter out targets that don't fullfil the time bounds
640 if !isTimeBetween(t, start, end) {
641 continue
642 }
643
644 targets = append(targets, chatHistoryTarget{
645 Name: target,
646 LatestMessage: t,
647 })
648
649 if err := ctx.Err(); err != nil {
650 return nil, err
651 }
652 }
653
654 // Sort targets by latest message time, backwards or forwards depending on
655 // the order of the time bounds
656 sort.Slice(targets, func(i, j int) bool {
657 t1, t2 := targets[i].LatestMessage, targets[j].LatestMessage
658 if start.Before(end) {
659 return t1.Before(t2)
660 } else {
661 return !t1.Before(t2)
662 }
663 })
664
665 // Truncate the result if necessary
666 if len(targets) > limit {
667 targets = targets[:limit]
668 }
669
670 return targets, nil
671}
672
673func (ms *fsMessageStore) RenameNetwork(oldNet, newNet *Network) error {
674 oldDir := filepath.Join(ms.root, escapeFilename(oldNet.GetName()))
675 newDir := filepath.Join(ms.root, escapeFilename(newNet.GetName()))
676 // Avoid loosing data by overwriting an existing directory
677 if _, err := os.Stat(newDir); err == nil {
678 return fmt.Errorf("destination %q already exists", newDir)
679 }
680 return os.Rename(oldDir, newDir)
681}
682
683func truncateDay(t time.Time) time.Time {
684 year, month, day := t.Date()
685 return time.Date(year, month, day, 0, 0, 0, 0, t.Location())
686}
687
688func truncateSecond(t time.Time) time.Time {
689 year, month, day := t.Date()
690 return time.Date(year, month, day, t.Hour(), t.Minute(), t.Second(), 0, t.Location())
691}
692
693func isTimeBetween(t, start, end time.Time) bool {
694 if end.Before(start) {
695 end, start = start, end
696 }
697 return start.Before(t) && t.Before(end)
698}
Note: See TracBrowser for help on using the repository browser.