Changeset 319 in code for trunk


Ignore:
Timestamp:
Jun 5, 2020, 9:50:31 PM (5 years ago)
Author:
delthas
Message:

Add support for downstream CHATHISTORY

This adds support for the WIP (at the time of this commit)
draft/chathistory extension, based on the draft at [1] and the
additional comments at [2].

This gets the history by parsing the chat logs, and is therefore only
enabled when the logs are enabled and the log path is configured.

Getting the history only from the logs adds some restrictions:

  • we cannot get history by msgid (those are not logged)
  • we cannot get the users masks (maybe they could be inferred from the JOIN etc, but it is not worth the effort and would not work every time)

The regular soju network history is not sent to clients that support
draft/chathistory, so that they can fetch what they need by manually
calling CHATHISTORY.

The only supported command is BEFORE for now, because that is the only
required command for an app that offers an "infinite history scrollback"
feature.

Regarding implementation, rather than reading the file from the end in
reverse, we simply start from the beginning of each log file, store each
PRIVMSG into a ring, then add the last lines of that ring into the
history we'll return later. The message parsing implementation must be
kept somewhat fast because an app could potentially request thousands of
messages in several files. Here we are using simple sscanf and indexOf
rather than regexps.

In case some log files do not contain any message (for example because
the user had not joined a channel at that time), we try up to a 100 days
of empty log files before giving up.

[1]: ​https://github.com/prawnsalad/ircv3-specifications/pull/3/files
[2]: ​https://github.com/ircv3/ircv3-specifications/pull/393/files#r350210018

Location:
trunk
Files:
3 edited

Legend:

Unmodified
Added
Removed

  • trunk/downstream.go

    r315 r319  
    4343                        "Not enough parameters",
    4444                },
     45        }}
     46}
     47
     48func newChatHistoryError(subcommand string, target string) ircError {
     49        return ircError{&irc.Message{
     50                Command: "FAIL",
     51                Params:  []string{"CHATHISTORY", "MESSAGE_ERROR", subcommand, target, "Messages could not be retrieved"},
    4552        }}
    4653}
    … …  
    107114        for k, v := range permanentDownstreamCaps {
    108115                dc.supportedCaps[k] = v
     116        }
     117        if srv.LogPath != "" {
     118                dc.supportedCaps["draft/chathistory"] = ""
    109119        }
    110120        return dc
    … …  
    786796        })
    787797        // TODO: RPL_ISUPPORT
     798        // TODO: send CHATHISTORY in RPL_ISUPPORT when implemented
    788799        dc.SendMessage(&irc.Message{
    789800                Prefix:  dc.srv.prefix(),
    … …  
    826837
    827838func (dc *downstreamConn) sendNetworkHistory(net *network) {
     839        if dc.caps["draft/chathistory"] {
     840                return
     841        }
    828842        for target, history := range net.history {
    829843                if ch, ok := net.channels[target]; ok && ch.Detached {
    … …  
    15111525                        Params:  []string{upstreamUser, upstreamChannel},
    15121526                })
     1527        case "CHATHISTORY":
     1528                var subcommand string
     1529                if err := parseMessageParams(msg, &subcommand); err != nil {
     1530                        return err
     1531                }
     1532                var target, criteria, limitStr string
     1533                if err := parseMessageParams(msg, nil, &target, &criteria, &limitStr); err != nil {
     1534                        return ircError{&irc.Message{
     1535                                Command: "FAIL",
     1536                                Params:  []string{"CHATHISTORY", "NEED_MORE_PARAMS", subcommand, "Missing parameters"},
     1537                        }}
     1538                }
     1539
     1540                if dc.srv.LogPath == "" {
     1541                        return ircError{&irc.Message{
     1542                                Command: irc.ERR_UNKNOWNCOMMAND,
     1543                                Params:  []string{dc.nick, subcommand, "Unknown command"},
     1544                        }}
     1545                }
     1546
     1547                uc, entity, err := dc.unmarshalEntity(target)
     1548                if err != nil {
     1549                        return err
     1550                }
     1551
     1552                // TODO: support msgid criteria
     1553                criteriaParts := strings.SplitN(criteria, "=", 2)
     1554                if len(criteriaParts) != 2 || criteriaParts[0] != "timestamp" {
     1555                        return ircError{&irc.Message{
     1556                                Command: "FAIL",
     1557                                Params:  []string{"CHATHISTORY", "UNKNOWN_CRITERIA", criteria, "Unknown criteria"},
     1558                        }}
     1559                }
     1560
     1561                timestamp, err := time.Parse(serverTimeLayout, criteriaParts[1])
     1562                if err != nil {
     1563                        return ircError{&irc.Message{
     1564                                Command: "FAIL",
     1565                                Params:  []string{"CHATHISTORY", "INVALID_CRITERIA", criteria, "Invalid criteria"},
     1566                        }}
     1567                }
     1568
     1569                limit, err := strconv.Atoi(limitStr)
     1570                if err != nil || limit < 0 || limit > dc.srv.HistoryLimit {
     1571                        return ircError{&irc.Message{
     1572                                Command: "FAIL",
     1573                                Params:  []string{"CHATHISTORY", "INVALID_LIMIT", limitStr, "Invalid limit"},
     1574                        }}
     1575                }
     1576
     1577                switch subcommand {
     1578                case "BEFORE":
     1579                        batchRef := "history"
     1580                        dc.SendMessage(&irc.Message{
     1581                                Prefix:  dc.srv.prefix(),
     1582                                Command: "BATCH",
     1583                                Params:  []string{"+" + batchRef, "chathistory", target},
     1584                        })
     1585
     1586                        history := make([]*irc.Message, limit)
     1587                        remaining := limit
     1588
     1589                        tries := 0
     1590                        for remaining > 0 {
     1591                                buf, err := parseMessagesBefore(uc.network, entity, timestamp, remaining)
     1592                                if err != nil {
     1593                                        dc.logger.Printf("failed parsing log messages for chathistory: %v", err)
     1594                                        return newChatHistoryError(subcommand, target)
     1595                                }
     1596                                if len(buf) == 0 {
     1597                                        tries++
     1598                                        if tries >= 100 {
     1599                                                break
     1600                                        }
     1601                                } else {
     1602                                        tries = 0
     1603                                }
     1604                                copy(history[remaining-len(buf):], buf)
     1605                                remaining -= len(buf)
     1606                                year, month, day := timestamp.Date()
     1607                                timestamp = time.Date(year, month, day, 0, 0, 0, 0, timestamp.Location()).Add(-1)
     1608                        }
     1609
     1610                        for _, m := range history[remaining:] {
     1611                                m.Tags["batch"] = irc.TagValue(batchRef)
     1612                                dc.SendMessage(dc.marshalMessage(m, uc.network))
     1613                        }
     1614
     1615                        dc.SendMessage(&irc.Message{
     1616                                Prefix:  dc.srv.prefix(),
     1617                                Command: "BATCH",
     1618                                Params:  []string{"-" + batchRef},
     1619                        })
     1620                default:
     1621                        // TODO: support AFTER, LATEST, BETWEEN
     1622                        return ircError{&irc.Message{
     1623                                Command: "FAIL",
     1624                                Params:  []string{"CHATHISTORY", "UNKNOWN_COMMAND", subcommand, "Unknown command"},
     1625                        }}
     1626                }
    15131627        default:
    15141628                dc.logger.Printf("unhandled message: %v", msg)

  • trunk/logger.go

    r250 r319  
    22
    33import (
     4        "bufio"
    45        "fmt"
    56        "os"
    … …  
    133134        }
    134135}
     136
     137func parseMessagesBefore(network *network, entity string, timestamp time.Time, limit int) ([]*irc.Message, error) {
     138        year, month, day := timestamp.Date()
     139        path := logPath(network, entity, timestamp)
     140        f, err := os.Open(path)
     141        if err != nil {
     142                if os.IsNotExist(err) {
     143                        return nil, nil
     144                }
     145                return nil, err
     146        }
     147        defer f.Close()
     148
     149        historyRing := make([]*irc.Message, limit)
     150        cur := 0
     151
     152        sc := bufio.NewScanner(f)
     153        for sc.Scan() {
     154                line := sc.Text()
     155                var hour, minute, second int
     156                _, err := fmt.Sscanf(line, "[%02d:%02d:%02d] ", &hour, &minute, &second)
     157                if err != nil {
     158                        return nil, err
     159                }
     160                message := line[11:]
     161                // TODO: support NOTICE
     162                if !strings.HasPrefix(message, "<") {
     163                        continue
     164                }
     165                i := strings.Index(message, "> ")
     166                if i == -1 {
     167                        continue
     168                }
     169                t := time.Date(year, month, day, hour, minute, second, 0, time.Local)
     170                if !t.Before(timestamp) {
     171                        break
     172                }
     173
     174                sender := message[1:i]
     175                text := message[i+2:]
     176                historyRing[cur%limit] = &irc.Message{
     177                        Tags: map[string]irc.TagValue{
     178                                "time": irc.TagValue(t.UTC().Format(serverTimeLayout)),
     179                        },
     180                        Prefix: &irc.Prefix{
     181                                Name: sender,
     182                        },
     183                        Command: "PRIVMSG",
     184                        Params:  []string{entity, text},
     185                }
     186                cur++
     187        }
     188        if sc.Err() != nil {
     189                return nil, sc.Err()
     190        }
     191
     192        n := limit
     193        if cur < limit {
     194                n = cur
     195        }
     196        start := (cur - n + limit) % limit
     197
     198        if start+n <= limit { // ring doesnt wrap
     199                return historyRing[start : start+n], nil
     200        } else { // ring wraps
     201                history := make([]*irc.Message, n)
     202                r := copy(history, historyRing[start:])
     203                copy(history[r:], historyRing[:n-r])
     204                return history, nil
     205        }
     206}

  • trunk/server.go

    r316 r319  
    3939
    4040type Server struct {
    41         Hostname string
    42         Logger   Logger
    43         RingCap  int
    44         LogPath  string
    45         Debug    bool
     41        Hostname     string
     42        Logger       Logger
     43        RingCap      int
     44        HistoryLimit int
     45        LogPath      string
     46        Debug        bool
    4647
    4748        db *DB
    … …  
    5354func NewServer(db *DB) *Server {
    5455        return &Server{
    55                 Logger:  log.New(log.Writer(), "", log.LstdFlags),
    56                 RingCap: 4096,
    57                 users:   make(map[string]*user),
    58                 db:      db,
     56                Logger:       log.New(log.Writer(), "", log.LstdFlags),
     57                RingCap:      4096,
     58                HistoryLimit: 1000,
     59                users:        make(map[string]*user),
     60                db:           db,
    5961        }
    6062}
Note: See TracChangeset for help on using the changeset viewer.