source: code/trunk/msgstore_fs.go@ 439

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

Turn messageStore into an interface

This allows for other implementations that aren't based on a filesystem.
File size: 10.8 KB
Line 
1package soju
2
3import (
4 "bufio"
5 "fmt"
6 "io"
7 "os"
8 "path/filepath"
9 "strings"
10 "time"
11
12 "gopkg.in/irc.v3"
13)
14
15const fsMessageStoreMaxTries = 100
16
17var escapeFilename = strings.NewReplacer("/", "-", "\\", "-")
18
19// fsMessageStore is a per-user on-disk store for IRC messages.
20type fsMessageStore struct {
21 root string
22
23 files map[string]*os.File // indexed by entity
24}
25
26func newFSMessageStore(root, username string) *fsMessageStore {
27 return &fsMessageStore{
28 root: filepath.Join(root, escapeFilename.Replace(username)),
29 files: make(map[string]*os.File),
30 }
31}
32
33func (ms *fsMessageStore) logPath(network *network, entity string, t time.Time) string {
34 year, month, day := t.Date()
35 filename := fmt.Sprintf("%04d-%02d-%02d.log", year, month, day)
36 return filepath.Join(ms.root, escapeFilename.Replace(network.GetName()), escapeFilename.Replace(entity), filename)
37}
38
39func parseMsgID(s string) (network, entity string, t time.Time, offset int64, err error) {
40 var year, month, day int
41 _, err = fmt.Sscanf(s, "%s %s %04d-%02d-%02d %d", &network, &entity, &year, &month, &day, &offset)
42 if err != nil {
43 return "", "", time.Time{}, 0, fmt.Errorf("invalid message ID: %v", err)
44 }
45 t = time.Date(year, time.Month(month), day, 0, 0, 0, 0, time.Local)
46 return network, entity, t, offset, nil
47}
48
49func formatMsgID(network, entity string, t time.Time, offset int64) string {
50 year, month, day := t.Date()
51 return fmt.Sprintf("%s %s %04d-%02d-%02d %d", network, entity, year, month, day, offset)
52}
53
54// nextMsgID queries the message ID for the next message to be written to f.
55func nextMsgID(network *network, entity string, t time.Time, f *os.File) (string, error) {
56 offset, err := f.Seek(0, io.SeekEnd)
57 if err != nil {
58 return "", err
59 }
60 return formatMsgID(network.GetName(), entity, t, offset), nil
61}
62
63func (ms *fsMessageStore) LastMsgID(network *network, entity string, t time.Time) (string, error) {
64 p := ms.logPath(network, entity, t)
65 fi, err := os.Stat(p)
66 if os.IsNotExist(err) {
67 return formatMsgID(network.GetName(), entity, t, -1), nil
68 } else if err != nil {
69 return "", err
70 }
71 return formatMsgID(network.GetName(), entity, t, fi.Size()-1), nil
72}
73
74func (ms *fsMessageStore) Append(network *network, entity string, msg *irc.Message) (string, error) {
75 s := formatMessage(msg)
76 if s == "" {
77 return "", nil
78 }
79
80 var t time.Time
81 if tag, ok := msg.Tags["time"]; ok {
82 var err error
83 t, err = time.Parse(serverTimeLayout, string(tag))
84 if err != nil {
85 return "", fmt.Errorf("failed to parse message time tag: %v", err)
86 }
87 t = t.In(time.Local)
88 } else {
89 t = time.Now()
90 }
91
92 // TODO: enforce maximum open file handles (LRU cache of file handles)
93 f := ms.files[entity]
94
95 // TODO: handle non-monotonic clock behaviour
96 path := ms.logPath(network, entity, t)
97 if f == nil || f.Name() != path {
98 if f != nil {
99 f.Close()
100 }
101
102 dir := filepath.Dir(path)
103 if err := os.MkdirAll(dir, 0700); err != nil {
104 return "", fmt.Errorf("failed to create message logs directory %q: %v", dir, err)
105 }
106
107 var err error
108 f, err = os.OpenFile(path, os.O_RDWR|os.O_CREATE|os.O_APPEND, 0600)
109 if err != nil {
110 return "", fmt.Errorf("failed to open message log file %q: %v", path, err)
111 }
112
113 ms.files[entity] = f
114 }
115
116 msgID, err := nextMsgID(network, entity, t, f)
117 if err != nil {
118 return "", fmt.Errorf("failed to generate message ID: %v", err)
119 }
120
121 _, err = fmt.Fprintf(f, "[%02d:%02d:%02d] %s\n", t.Hour(), t.Minute(), t.Second(), s)
122 if err != nil {
123 return "", fmt.Errorf("failed to log message to %q: %v", f.Name(), err)
124 }
125
126 return msgID, nil
127}
128
129func (ms *fsMessageStore) Close() error {
130 var closeErr error
131 for _, f := range ms.files {
132 if err := f.Close(); err != nil {
133 closeErr = fmt.Errorf("failed to close message store: %v", err)
134 }
135 }
136 return closeErr
137}
138
139// formatMessage formats a message log line. It assumes a well-formed IRC
140// message.
141func formatMessage(msg *irc.Message) string {
142 switch strings.ToUpper(msg.Command) {
143 case "NICK":
144 return fmt.Sprintf("*** %s is now known as %s", msg.Prefix.Name, msg.Params[0])
145 case "JOIN":
146 return fmt.Sprintf("*** Joins: %s (%s@%s)", msg.Prefix.Name, msg.Prefix.User, msg.Prefix.Host)
147 case "PART":
148 var reason string
149 if len(msg.Params) > 1 {
150 reason = msg.Params[1]
151 }
152 return fmt.Sprintf("*** Parts: %s (%s@%s) (%s)", msg.Prefix.Name, msg.Prefix.User, msg.Prefix.Host, reason)
153 case "KICK":
154 nick := msg.Params[1]
155 var reason string
156 if len(msg.Params) > 2 {
157 reason = msg.Params[2]
158 }
159 return fmt.Sprintf("*** %s was kicked by %s (%s)", nick, msg.Prefix.Name, reason)
160 case "QUIT":
161 var reason string
162 if len(msg.Params) > 0 {
163 reason = msg.Params[0]
164 }
165 return fmt.Sprintf("*** Quits: %s (%s@%s) (%s)", msg.Prefix.Name, msg.Prefix.User, msg.Prefix.Host, reason)
166 case "TOPIC":
167 var topic string
168 if len(msg.Params) > 1 {
169 topic = msg.Params[1]
170 }
171 return fmt.Sprintf("*** %s changes topic to '%s'", msg.Prefix.Name, topic)
172 case "MODE":
173 return fmt.Sprintf("*** %s sets mode: %s", msg.Prefix.Name, strings.Join(msg.Params[1:], " "))
174 case "NOTICE":
175 return fmt.Sprintf("-%s- %s", msg.Prefix.Name, msg.Params[1])
176 case "PRIVMSG":
177 if cmd, params, ok := parseCTCPMessage(msg); ok && cmd == "ACTION" {
178 return fmt.Sprintf("* %s %s", msg.Prefix.Name, params)
179 } else {
180 return fmt.Sprintf("<%s> %s", msg.Prefix.Name, msg.Params[1])
181 }
182 default:
183 return ""
184 }
185}
186
187func parseMessage(line, entity string, ref time.Time) (*irc.Message, time.Time, error) {
188 var hour, minute, second int
189 _, err := fmt.Sscanf(line, "[%02d:%02d:%02d] ", &hour, &minute, &second)
190 if err != nil {
191 return nil, time.Time{}, err
192 }
193 line = line[11:]
194
195 var cmd, sender, text string
196 if strings.HasPrefix(line, "<") {
197 cmd = "PRIVMSG"
198 parts := strings.SplitN(line[1:], "> ", 2)
199 if len(parts) != 2 {
200 return nil, time.Time{}, nil
201 }
202 sender, text = parts[0], parts[1]
203 } else if strings.HasPrefix(line, "-") {
204 cmd = "NOTICE"
205 parts := strings.SplitN(line[1:], "- ", 2)
206 if len(parts) != 2 {
207 return nil, time.Time{}, nil
208 }
209 sender, text = parts[0], parts[1]
210 } else if strings.HasPrefix(line, "* ") {
211 cmd = "PRIVMSG"
212 parts := strings.SplitN(line[2:], " ", 2)
213 if len(parts) != 2 {
214 return nil, time.Time{}, nil
215 }
216 sender, text = parts[0], "\x01ACTION "+parts[1]+"\x01"
217 } else {
218 return nil, time.Time{}, nil
219 }
220
221 year, month, day := ref.Date()
222 t := time.Date(year, month, day, hour, minute, second, 0, time.Local)
223
224 msg := &irc.Message{
225 Tags: map[string]irc.TagValue{
226 "time": irc.TagValue(t.UTC().Format(serverTimeLayout)),
227 },
228 Prefix: &irc.Prefix{Name: sender},
229 Command: cmd,
230 Params: []string{entity, text},
231 }
232 return msg, t, nil
233}
234
235func (ms *fsMessageStore) parseMessagesBefore(network *network, entity string, ref time.Time, limit int, afterOffset int64) ([]*irc.Message, error) {
236 path := ms.logPath(network, entity, ref)
237 f, err := os.Open(path)
238 if err != nil {
239 if os.IsNotExist(err) {
240 return nil, nil
241 }
242 return nil, err
243 }
244 defer f.Close()
245
246 historyRing := make([]*irc.Message, limit)
247 cur := 0
248
249 sc := bufio.NewScanner(f)
250
251 if afterOffset >= 0 {
252 if _, err := f.Seek(afterOffset, io.SeekStart); err != nil {
253 return nil, nil
254 }
255 sc.Scan() // skip till next newline
256 }
257
258 for sc.Scan() {
259 msg, t, err := parseMessage(sc.Text(), entity, ref)
260 if err != nil {
261 return nil, err
262 } else if msg == nil {
263 continue
264 } else if !t.Before(ref) {
265 break
266 }
267
268 historyRing[cur%limit] = msg
269 cur++
270 }
271 if sc.Err() != nil {
272 return nil, sc.Err()
273 }
274
275 n := limit
276 if cur < limit {
277 n = cur
278 }
279 start := (cur - n + limit) % limit
280
281 if start+n <= limit { // ring doesnt wrap
282 return historyRing[start : start+n], nil
283 } else { // ring wraps
284 history := make([]*irc.Message, n)
285 r := copy(history, historyRing[start:])
286 copy(history[r:], historyRing[:n-r])
287 return history, nil
288 }
289}
290
291func (ms *fsMessageStore) parseMessagesAfter(network *network, entity string, ref time.Time, limit int) ([]*irc.Message, error) {
292 path := ms.logPath(network, entity, ref)
293 f, err := os.Open(path)
294 if err != nil {
295 if os.IsNotExist(err) {
296 return nil, nil
297 }
298 return nil, err
299 }
300 defer f.Close()
301
302 var history []*irc.Message
303 sc := bufio.NewScanner(f)
304 for sc.Scan() && len(history) < limit {
305 msg, t, err := parseMessage(sc.Text(), entity, ref)
306 if err != nil {
307 return nil, err
308 } else if msg == nil || !t.After(ref) {
309 continue
310 }
311
312 history = append(history, msg)
313 }
314 if sc.Err() != nil {
315 return nil, sc.Err()
316 }
317
318 return history, nil
319}
320
321func (ms *fsMessageStore) LoadBeforeTime(network *network, entity string, t time.Time, limit int) ([]*irc.Message, error) {
322 history := make([]*irc.Message, limit)
323 remaining := limit
324 tries := 0
325 for remaining > 0 && tries < fsMessageStoreMaxTries {
326 buf, err := ms.parseMessagesBefore(network, entity, t, remaining, -1)
327 if err != nil {
328 return nil, err
329 }
330 if len(buf) == 0 {
331 tries++
332 } else {
333 tries = 0
334 }
335 copy(history[remaining-len(buf):], buf)
336 remaining -= len(buf)
337 year, month, day := t.Date()
338 t = time.Date(year, month, day, 0, 0, 0, 0, t.Location()).Add(-1)
339 }
340
341 return history[remaining:], nil
342}
343
344func (ms *fsMessageStore) LoadAfterTime(network *network, entity string, t time.Time, limit int) ([]*irc.Message, error) {
345 var history []*irc.Message
346 remaining := limit
347 tries := 0
348 now := time.Now()
349 for remaining > 0 && tries < fsMessageStoreMaxTries && t.Before(now) {
350 buf, err := ms.parseMessagesAfter(network, entity, t, remaining)
351 if err != nil {
352 return nil, err
353 }
354 if len(buf) == 0 {
355 tries++
356 } else {
357 tries = 0
358 }
359 history = append(history, buf...)
360 remaining -= len(buf)
361 year, month, day := t.Date()
362 t = time.Date(year, month, day+1, 0, 0, 0, 0, t.Location())
363 }
364 return history, nil
365}
366
367func truncateDay(t time.Time) time.Time {
368 year, month, day := t.Date()
369 return time.Date(year, month, day, 0, 0, 0, 0, t.Location())
370}
371
372func (ms *fsMessageStore) LoadLatestID(network *network, entity, id string, limit int) ([]*irc.Message, error) {
373 var afterTime time.Time
374 var afterOffset int64
375 if id != "" {
376 var idNet, idEntity string
377 var err error
378 idNet, idEntity, afterTime, afterOffset, err = parseMsgID(id)
379 if err != nil {
380 return nil, err
381 }
382 if idNet != network.GetName() || idEntity != entity {
383 return nil, fmt.Errorf("cannot find message ID: message ID doesn't match network/entity")
384 }
385 }
386
387 history := make([]*irc.Message, limit)
388 t := time.Now()
389 remaining := limit
390 tries := 0
391 for remaining > 0 && tries < fsMessageStoreMaxTries && !truncateDay(t).Before(afterTime) {
392 var offset int64 = -1
393 if afterOffset >= 0 && truncateDay(t).Equal(afterTime) {
394 offset = afterOffset
395 }
396
397 buf, err := ms.parseMessagesBefore(network, entity, t, remaining, offset)
398 if err != nil {
399 return nil, err
400 }
401 if len(buf) == 0 {
402 tries++
403 } else {
404 tries = 0
405 }
406 copy(history[remaining-len(buf):], buf)
407 remaining -= len(buf)
408 year, month, day := t.Date()
409 t = time.Date(year, month, day, 0, 0, 0, 0, t.Location()).Add(-1)
410 }
411
412 return history[remaining:], nil
413}
Note: See TracBrowser for help on using the repository browser.