source: code/trunk/msgstore_fs.go@ 566

Last change on this file since 566 was 558, checked in by greg, 4 years ago

Slightly relax new log file permissions

Make new log files group-readable by default. To retain the prior
behavior, soju can be started with a umask set to 0077.
File size: 13.7 KB
Line 
1package soju
2
3import (
4 "bufio"
5 "fmt"
6 "io"
7 "os"
8 "path/filepath"
9 "sort"
10 "strings"
11 "time"
12
13 "git.sr.ht/~sircmpwn/go-bare"
14 "gopkg.in/irc.v3"
15)
16
17const fsMessageStoreMaxTries = 100
18
19var escapeFilename = strings.NewReplacer("/", "-", "\\", "-")
20
21type date struct {
22 Year, Month, Day int
23}
24
25func newDate(t time.Time) date {
26 year, month, day := t.Date()
27 return date{year, int(month), day}
28}
29
30func (d date) Time() time.Time {
31 return time.Date(d.Year, time.Month(d.Month), d.Day, 0, 0, 0, 0, time.Local)
32}
33
34type fsMsgID struct {
35 Date date
36 Offset bare.Int
37}
38
39func (fsMsgID) msgIDType() msgIDType {
40 return msgIDFS
41}
42
43func parseFSMsgID(s string) (netID int64, entity string, t time.Time, offset int64, err error) {
44 var id fsMsgID
45 netID, entity, err = parseMsgID(s, &id)
46 if err != nil {
47 return 0, "", time.Time{}, 0, err
48 }
49 return netID, entity, id.Date.Time(), int64(id.Offset), nil
50}
51
52func formatFSMsgID(netID int64, entity string, t time.Time, offset int64) string {
53 id := fsMsgID{
54 Date: newDate(t),
55 Offset: bare.Int(offset),
56 }
57 return formatMsgID(netID, entity, &id)
58}
59
60// fsMessageStore is a per-user on-disk store for IRC messages.
61type fsMessageStore struct {
62 root string
63
64 files map[string]*os.File // indexed by entity
65}
66
67var _ messageStore = (*fsMessageStore)(nil)
68var _ chatHistoryMessageStore = (*fsMessageStore)(nil)
69
70func newFSMessageStore(root, username string) *fsMessageStore {
71 return &fsMessageStore{
72 root: filepath.Join(root, escapeFilename.Replace(username)),
73 files: make(map[string]*os.File),
74 }
75}
76
77func (ms *fsMessageStore) logPath(network *network, entity string, t time.Time) string {
78 year, month, day := t.Date()
79 filename := fmt.Sprintf("%04d-%02d-%02d.log", year, month, day)
80 return filepath.Join(ms.root, escapeFilename.Replace(network.GetName()), escapeFilename.Replace(entity), filename)
81}
82
83// nextMsgID queries the message ID for the next message to be written to f.
84func nextFSMsgID(network *network, entity string, t time.Time, f *os.File) (string, error) {
85 offset, err := f.Seek(0, io.SeekEnd)
86 if err != nil {
87 return "", fmt.Errorf("failed to query next FS message ID: %v", err)
88 }
89 return formatFSMsgID(network.ID, entity, t, offset), nil
90}
91
92func (ms *fsMessageStore) LastMsgID(network *network, entity string, t time.Time) (string, error) {
93 p := ms.logPath(network, entity, t)
94 fi, err := os.Stat(p)
95 if os.IsNotExist(err) {
96 return formatFSMsgID(network.ID, entity, t, -1), nil
97 } else if err != nil {
98 return "", fmt.Errorf("failed to query last FS message ID: %v", err)
99 }
100 return formatFSMsgID(network.ID, entity, t, fi.Size()-1), nil
101}
102
103func (ms *fsMessageStore) Append(network *network, entity string, msg *irc.Message) (string, error) {
104 s := formatMessage(msg)
105 if s == "" {
106 return "", nil
107 }
108
109 var t time.Time
110 if tag, ok := msg.Tags["time"]; ok {
111 var err error
112 t, err = time.Parse(serverTimeLayout, string(tag))
113 if err != nil {
114 return "", fmt.Errorf("failed to parse message time tag: %v", err)
115 }
116 t = t.In(time.Local)
117 } else {
118 t = time.Now()
119 }
120
121 // TODO: enforce maximum open file handles (LRU cache of file handles)
122 f := ms.files[entity]
123
124 // TODO: handle non-monotonic clock behaviour
125 path := ms.logPath(network, entity, t)
126 if f == nil || f.Name() != path {
127 if f != nil {
128 f.Close()
129 }
130
131 dir := filepath.Dir(path)
132 if err := os.MkdirAll(dir, 0750); err != nil {
133 return "", fmt.Errorf("failed to create message logs directory %q: %v", dir, err)
134 }
135
136 var err error
137 f, err = os.OpenFile(path, os.O_RDWR|os.O_CREATE|os.O_APPEND, 0640)
138 if err != nil {
139 return "", fmt.Errorf("failed to open message log file %q: %v", path, err)
140 }
141
142 ms.files[entity] = f
143 }
144
145 msgID, err := nextFSMsgID(network, entity, t, f)
146 if err != nil {
147 return "", fmt.Errorf("failed to generate message ID: %v", err)
148 }
149
150 _, err = fmt.Fprintf(f, "[%02d:%02d:%02d] %s\n", t.Hour(), t.Minute(), t.Second(), s)
151 if err != nil {
152 return "", fmt.Errorf("failed to log message to %q: %v", f.Name(), err)
153 }
154
155 return msgID, nil
156}
157
158func (ms *fsMessageStore) Close() error {
159 var closeErr error
160 for _, f := range ms.files {
161 if err := f.Close(); err != nil {
162 closeErr = fmt.Errorf("failed to close message store: %v", err)
163 }
164 }
165 return closeErr
166}
167
168// formatMessage formats a message log line. It assumes a well-formed IRC
169// message.
170func formatMessage(msg *irc.Message) string {
171 switch strings.ToUpper(msg.Command) {
172 case "NICK":
173 return fmt.Sprintf("*** %s is now known as %s", msg.Prefix.Name, msg.Params[0])
174 case "JOIN":
175 return fmt.Sprintf("*** Joins: %s (%s@%s)", msg.Prefix.Name, msg.Prefix.User, msg.Prefix.Host)
176 case "PART":
177 var reason string
178 if len(msg.Params) > 1 {
179 reason = msg.Params[1]
180 }
181 return fmt.Sprintf("*** Parts: %s (%s@%s) (%s)", msg.Prefix.Name, msg.Prefix.User, msg.Prefix.Host, reason)
182 case "KICK":
183 nick := msg.Params[1]
184 var reason string
185 if len(msg.Params) > 2 {
186 reason = msg.Params[2]
187 }
188 return fmt.Sprintf("*** %s was kicked by %s (%s)", nick, msg.Prefix.Name, reason)
189 case "QUIT":
190 var reason string
191 if len(msg.Params) > 0 {
192 reason = msg.Params[0]
193 }
194 return fmt.Sprintf("*** Quits: %s (%s@%s) (%s)", msg.Prefix.Name, msg.Prefix.User, msg.Prefix.Host, reason)
195 case "TOPIC":
196 var topic string
197 if len(msg.Params) > 1 {
198 topic = msg.Params[1]
199 }
200 return fmt.Sprintf("*** %s changes topic to '%s'", msg.Prefix.Name, topic)
201 case "MODE":
202 return fmt.Sprintf("*** %s sets mode: %s", msg.Prefix.Name, strings.Join(msg.Params[1:], " "))
203 case "NOTICE":
204 return fmt.Sprintf("-%s- %s", msg.Prefix.Name, msg.Params[1])
205 case "PRIVMSG":
206 if cmd, params, ok := parseCTCPMessage(msg); ok && cmd == "ACTION" {
207 return fmt.Sprintf("* %s %s", msg.Prefix.Name, params)
208 } else {
209 return fmt.Sprintf("<%s> %s", msg.Prefix.Name, msg.Params[1])
210 }
211 default:
212 return ""
213 }
214}
215
216func parseMessage(line, entity string, ref time.Time) (*irc.Message, time.Time, error) {
217 var hour, minute, second int
218 _, err := fmt.Sscanf(line, "[%02d:%02d:%02d] ", &hour, &minute, &second)
219 if err != nil {
220 return nil, time.Time{}, fmt.Errorf("malformed timestamp prefix: %v", err)
221 }
222 line = line[11:]
223
224 var cmd, sender, text string
225 if strings.HasPrefix(line, "<") {
226 cmd = "PRIVMSG"
227 parts := strings.SplitN(line[1:], "> ", 2)
228 if len(parts) != 2 {
229 return nil, time.Time{}, nil
230 }
231 sender, text = parts[0], parts[1]
232 } else if strings.HasPrefix(line, "-") {
233 cmd = "NOTICE"
234 parts := strings.SplitN(line[1:], "- ", 2)
235 if len(parts) != 2 {
236 return nil, time.Time{}, nil
237 }
238 sender, text = parts[0], parts[1]
239 } else if strings.HasPrefix(line, "* ") {
240 cmd = "PRIVMSG"
241 parts := strings.SplitN(line[2:], " ", 2)
242 if len(parts) != 2 {
243 return nil, time.Time{}, nil
244 }
245 sender, text = parts[0], "\x01ACTION "+parts[1]+"\x01"
246 } else {
247 return nil, time.Time{}, nil
248 }
249
250 year, month, day := ref.Date()
251 t := time.Date(year, month, day, hour, minute, second, 0, time.Local)
252
253 msg := &irc.Message{
254 Tags: map[string]irc.TagValue{
255 "time": irc.TagValue(t.UTC().Format(serverTimeLayout)),
256 },
257 Prefix: &irc.Prefix{Name: sender},
258 Command: cmd,
259 Params: []string{entity, text},
260 }
261 return msg, t, nil
262}
263
264func (ms *fsMessageStore) parseMessagesBefore(network *network, entity string, ref time.Time, end time.Time, limit int, afterOffset int64) ([]*irc.Message, error) {
265 path := ms.logPath(network, entity, ref)
266 f, err := os.Open(path)
267 if err != nil {
268 if os.IsNotExist(err) {
269 return nil, nil
270 }
271 return nil, fmt.Errorf("failed to parse messages before ref: %v", err)
272 }
273 defer f.Close()
274
275 historyRing := make([]*irc.Message, limit)
276 cur := 0
277
278 sc := bufio.NewScanner(f)
279
280 if afterOffset >= 0 {
281 if _, err := f.Seek(afterOffset, io.SeekStart); err != nil {
282 return nil, nil
283 }
284 sc.Scan() // skip till next newline
285 }
286
287 for sc.Scan() {
288 msg, t, err := parseMessage(sc.Text(), entity, ref)
289 if err != nil {
290 return nil, err
291 } else if msg == nil || !t.After(end) {
292 continue
293 } else if !t.Before(ref) {
294 break
295 }
296
297 historyRing[cur%limit] = msg
298 cur++
299 }
300 if sc.Err() != nil {
301 return nil, fmt.Errorf("failed to parse messages before ref: scanner error: %v", sc.Err())
302 }
303
304 n := limit
305 if cur < limit {
306 n = cur
307 }
308 start := (cur - n + limit) % limit
309
310 if start+n <= limit { // ring doesnt wrap
311 return historyRing[start : start+n], nil
312 } else { // ring wraps
313 history := make([]*irc.Message, n)
314 r := copy(history, historyRing[start:])
315 copy(history[r:], historyRing[:n-r])
316 return history, nil
317 }
318}
319
320func (ms *fsMessageStore) parseMessagesAfter(network *network, entity string, ref time.Time, end time.Time, limit int) ([]*irc.Message, error) {
321 path := ms.logPath(network, entity, ref)
322 f, err := os.Open(path)
323 if err != nil {
324 if os.IsNotExist(err) {
325 return nil, nil
326 }
327 return nil, fmt.Errorf("failed to parse messages after ref: %v", err)
328 }
329 defer f.Close()
330
331 var history []*irc.Message
332 sc := bufio.NewScanner(f)
333 for sc.Scan() && len(history) < limit {
334 msg, t, err := parseMessage(sc.Text(), entity, ref)
335 if err != nil {
336 return nil, err
337 } else if msg == nil || !t.After(ref) {
338 continue
339 } else if !t.Before(end) {
340 break
341 }
342
343 history = append(history, msg)
344 }
345 if sc.Err() != nil {
346 return nil, fmt.Errorf("failed to parse messages after ref: scanner error: %v", sc.Err())
347 }
348
349 return history, nil
350}
351
352func (ms *fsMessageStore) LoadBeforeTime(network *network, entity string, start time.Time, end time.Time, limit int) ([]*irc.Message, error) {
353 history := make([]*irc.Message, limit)
354 remaining := limit
355 tries := 0
356 for remaining > 0 && tries < fsMessageStoreMaxTries && end.Before(start) {
357 buf, err := ms.parseMessagesBefore(network, entity, start, end, remaining, -1)
358 if err != nil {
359 return nil, err
360 }
361 if len(buf) == 0 {
362 tries++
363 } else {
364 tries = 0
365 }
366 copy(history[remaining-len(buf):], buf)
367 remaining -= len(buf)
368 year, month, day := start.Date()
369 start = time.Date(year, month, day, 0, 0, 0, 0, start.Location()).Add(-1)
370 }
371
372 return history[remaining:], nil
373}
374
375func (ms *fsMessageStore) LoadAfterTime(network *network, entity string, start time.Time, end time.Time, limit int) ([]*irc.Message, error) {
376 var history []*irc.Message
377 remaining := limit
378 tries := 0
379 for remaining > 0 && tries < fsMessageStoreMaxTries && start.Before(end) {
380 buf, err := ms.parseMessagesAfter(network, entity, start, end, remaining)
381 if err != nil {
382 return nil, err
383 }
384 if len(buf) == 0 {
385 tries++
386 } else {
387 tries = 0
388 }
389 history = append(history, buf...)
390 remaining -= len(buf)
391 year, month, day := start.Date()
392 start = time.Date(year, month, day+1, 0, 0, 0, 0, start.Location())
393 }
394 return history, nil
395}
396
397func (ms *fsMessageStore) LoadLatestID(network *network, entity, id string, limit int) ([]*irc.Message, error) {
398 var afterTime time.Time
399 var afterOffset int64
400 if id != "" {
401 var idNet int64
402 var idEntity string
403 var err error
404 idNet, idEntity, afterTime, afterOffset, err = parseFSMsgID(id)
405 if err != nil {
406 return nil, err
407 }
408 if idNet != network.ID || idEntity != entity {
409 return nil, fmt.Errorf("cannot find message ID: message ID doesn't match network/entity")
410 }
411 }
412
413 history := make([]*irc.Message, limit)
414 t := time.Now()
415 remaining := limit
416 tries := 0
417 for remaining > 0 && tries < fsMessageStoreMaxTries && !truncateDay(t).Before(afterTime) {
418 var offset int64 = -1
419 if afterOffset >= 0 && truncateDay(t).Equal(afterTime) {
420 offset = afterOffset
421 }
422
423 buf, err := ms.parseMessagesBefore(network, entity, t, time.Time{}, remaining, offset)
424 if err != nil {
425 return nil, err
426 }
427 if len(buf) == 0 {
428 tries++
429 } else {
430 tries = 0
431 }
432 copy(history[remaining-len(buf):], buf)
433 remaining -= len(buf)
434 year, month, day := t.Date()
435 t = time.Date(year, month, day, 0, 0, 0, 0, t.Location()).Add(-1)
436 }
437
438 return history[remaining:], nil
439}
440
441func (ms *fsMessageStore) ListTargets(network *network, start, end time.Time, limit int) ([]chatHistoryTarget, error) {
442 rootPath := filepath.Join(ms.root, escapeFilename.Replace(network.GetName()))
443 root, err := os.Open(rootPath)
444 if err != nil {
445 return nil, err
446 }
447
448 // The returned targets are escaped, and there is no way to un-escape
449 // TODO: switch to ReadDir (Go 1.16+)
450 targetNames, err := root.Readdirnames(0)
451 root.Close()
452 if err != nil {
453 return nil, err
454 }
455
456 var targets []chatHistoryTarget
457 for _, target := range targetNames {
458 // target is already escaped here
459 targetPath := filepath.Join(rootPath, target)
460 targetDir, err := os.Open(targetPath)
461 if err != nil {
462 return nil, err
463 }
464
465 entries, err := targetDir.Readdir(0)
466 targetDir.Close()
467 if err != nil {
468 return nil, err
469 }
470
471 // We use mtime here, which may give imprecise or incorrect results
472 var t time.Time
473 for _, entry := range entries {
474 if entry.ModTime().After(t) {
475 t = entry.ModTime()
476 }
477 }
478
479 // The timestamps we get from logs have second granularity
480 t = truncateSecond(t)
481
482 // Filter out targets that don't fullfil the time bounds
483 if !isTimeBetween(t, start, end) {
484 continue
485 }
486
487 targets = append(targets, chatHistoryTarget{
488 Name: target,
489 LatestMessage: t,
490 })
491 }
492
493 // Sort targets by latest message time, backwards or forwards depending on
494 // the order of the time bounds
495 sort.Slice(targets, func(i, j int) bool {
496 t1, t2 := targets[i].LatestMessage, targets[j].LatestMessage
497 if start.Before(end) {
498 return t1.Before(t2)
499 } else {
500 return !t1.Before(t2)
501 }
502 })
503
504 // Truncate the result if necessary
505 if len(targets) > limit {
506 targets = targets[:limit]
507 }
508
509 return targets, nil
510}
511
512func truncateDay(t time.Time) time.Time {
513 year, month, day := t.Date()
514 return time.Date(year, month, day, 0, 0, 0, 0, t.Location())
515}
516
517func truncateSecond(t time.Time) time.Time {
518 year, month, day := t.Date()
519 return time.Date(year, month, day, t.Hour(), t.Minute(), t.Second(), 0, t.Location())
520}
521
522func isTimeBetween(t, start, end time.Time) bool {
523 if end.Before(start) {
524 end, start = start, end
525 }
526 return start.Before(t) && t.Before(end)
527}
Note: See TracBrowser for help on using the repository browser.