feat(chat): keep the chat in the database
@@ -19,6 +19,7 @@ WallCooldown string `yaml:"wallCooldown"` // between posts of one key (10m) ChatMaxLen int `yaml:"chatMaxLen"` // cells per message (280) ChatPerMinute *int `yaml:"chatPerMinute"` // messages per key; 0 = unlimited (20) ChatBurst int `yaml:"chatBurst"` // (5) + ChatKeep *int `yaml:"chatKeep"` // lines kept in the chat log; 0 = all (2000) } // communitySettings is CommunityConfig with defaults applied. @@ -30,6 +31,7 @@ wallCooldown time.Duration chatMaxLen int chatPerMinute int chatBurst int + chatKeep int } func (c *Config) community() communitySettings { @@ -41,6 +43,7 @@ wallCooldown: 10 * time.Minute, chatMaxLen: 280, chatPerMinute: 20, chatBurst: 5, + chatKeep: 2000, } if c == nil || c.Community == nil { return s @@ -66,6 +69,9 @@ s.chatPerMinute = max(*cc.ChatPerMinute, 0) } if cc.ChatBurst > 0 { s.chatBurst = cc.ChatBurst + } + if cc.ChatKeep != nil { + s.chatKeep = max(*cc.ChatKeep, 0) } return s
@@ -118,6 +118,7 @@ } cc := s.communityCfg s.wall = community.NewWall(cst, cc.wallMaxLen, cc.wallCooldown) s.chat = community.NewHub(cst, cc.chatMaxLen, cc.chatPerMinute, cc.chatBurst) + s.chat.SetKeep(cc.chatKeep) s.setGames(resolveGames(cfg.Games))
@@ -12,8 +12,9 @@ "github.com/runawaydevil/hammer/internal/ratelimit" ) const ( - historySize = 100 // messages replayed to someone joining - subBuffer = 64 // messages queued per member before drops + historySize = 100 // messages replayed to someone joining + subBuffer = 64 // messages queued per member before drops + chatKeep = 2000 // lines kept in the chat log by default ) // Kind is what a chat message is. @@ -77,6 +78,7 @@ rate *ratelimit.Limiter maxCells int now func() time.Time + keep int // lines kept in the chat log; 0 = all next int seq uint64 subs map[int]*Sub @@ -84,17 +86,34 @@ history []Message mutes map[string]time.Time // identity -> muted until } -// NewHub returns an empty room. Messages are cut to maxCells cells, and each -// identity may send perMinute messages a minute with bursts of burst. +// NewHub returns the room. Messages are cut to maxCells cells, and each +// identity may send perMinute messages a minute with bursts of burst. With a +// store, what people say is kept in the database and the recent history +// survives a restart. func NewHub(st *Store, maxCells, perMinute, burst int) *Hub { - return &Hub{ + h := &Hub{ st: st, rate: ratelimit.New(perMinute, burst), maxCells: maxCells, now: time.Now, + keep: chatKeep, subs: map[int]*Sub{}, mutes: map[string]time.Time{}, } + if st != nil { + if history, top, err := st.recentChat(historySize); err == nil { + h.history, h.seq = history, top + } + } + + return h +} + +// SetKeep sets how many lines the chat log keeps (0 = all). +func (h *Hub) SetKeep(n int) { + h.mu.Lock() + h.keep = max(n, 0) + h.mu.Unlock() } // Sub is one connection in the room. @@ -191,7 +210,11 @@ if ok, wait := h.rate.Allow(s.member.Identity); !ok { return &WaitError{Wait: wait} } - h.broadcastLocked(Message{Kind: kind, Nick: s.member.Nick, Text: text}) + msg := h.broadcastLocked(Message{Kind: kind, Nick: s.member.Nick, Text: text}) + if h.st != nil { + // A log that cannot be written must not silence the room. + _ = h.st.logChat(msg, s.member.Identity, h.keep) + } return nil } @@ -303,8 +326,9 @@ } } // broadcastLocked stamps msg, keeps it in the history and fans it out. A full -// member queue drops the message for that member only. -func (h *Hub) broadcastLocked(msg Message) { +// member queue drops the message for that member only. It returns msg as +// stamped. +func (h *Hub) broadcastLocked(msg Message) Message { h.seq++ msg.ID = h.seq msg.At = h.now() @@ -323,4 +347,6 @@ case s.c <- msg: default: } } + + return msg }
@@ -381,3 +381,55 @@ if m := recv(t, a); m.Kind != KindNick || m.Nick != "ana" || m.Extra != "aninha" || a.Nick() != "aninha" { t.Fatalf("rename: %+v nick=%s", m, a.Nick()) } } + +func TestChatHistorySurvivesARestart(t *testing.T) { + st := testStore(t) + + h := NewHub(st, 280, 0, 0) + a, _, _ := h.Join(Member{Identity: "fp:a", Nick: "ana"}) + if err := a.Say("oi pessoal"); err != nil { + t.Fatal(err) + } + if err := a.Me("acena"); err != nil { + t.Fatal(err) + } + a.Leave() + + // A new hub over the same database: the server restarted. + h2 := NewHub(st, 280, 0, 0) + _, hist, _ := h2.Join(Member{Identity: "fp:b", Nick: "bia"}) + if len(hist) != 2 || hist[0].Text != "oi pessoal" || hist[1].Kind != KindMe || hist[1].Nick != "ana" { + t.Fatalf("history after restart = %+v", hist) + } + + // New messages carry ids after the old ones. + b, _, _ := h2.Join(Member{Identity: "fp:c", Nick: "caio"}) + _ = b.Say("cheguei") + _, hist, _ = h2.Join(Member{Identity: "fp:d", Nick: "duda"}) + if last := hist[len(hist)-1]; last.Text != "cheguei" || last.ID <= hist[0].ID { + t.Fatalf("ids did not move on: %+v", hist) + } +} + +func TestChatLogIsPruned(t *testing.T) { + st := testStore(t) + h := NewHub(st, 280, 0, 0) + h.SetKeep(5) + a, _, _ := h.Join(Member{Identity: "fp:a", Nick: "ana"}) + for i := 0; i < 12; i++ { + if err := a.Say(fmt.Sprint(i)); err != nil { + t.Fatal(err) + } + } + var n int + if err := st.db.QueryRow(`select count(*) from chat_log`).Scan(&n); err != nil { + t.Fatal(err) + } + if n != 5 { + t.Fatalf("%d lines kept, want 5", n) + } + _, hist, _ := NewHub(st, 280, 0, 0).Join(Member{Identity: "fp:b", Nick: "bia"}) + if len(hist) != 5 || hist[0].Text != "7" || hist[4].Text != "11" { + t.Fatalf("history = %+v", hist) + } +}
@@ -18,6 +18,14 @@ text text not null, created integer not null ); create index if not exists oneliners_identity on oneliners(identity, created); +create table if not exists chat_log ( + id integer primary key autoincrement, + at integer not null, + kind text not null, + identity text not null, + nick text not null, + text text not null +); create table if not exists bans ( identity text not null primary key, nick text not null default '', @@ -76,3 +84,54 @@ } return until == 0 || s.now().Unix() < until } + +// logChat keeps one line of the chat and prunes the log to its newest keep +// lines. Only what people said is kept; joins and leaves are not. +func (s *Store) logChat(m Message, identity string, keep int) error { + res, err := s.db.Exec(`insert into chat_log(at, kind, identity, nick, text) values(?,?,?,?,?)`, + m.At.Unix(), string(m.Kind), identity, m.Nick, m.Text) + if err != nil { + return err + } + if keep <= 0 { + return nil + } + if id, err := res.LastInsertId(); err == nil && id > int64(keep) { + _, err = s.db.Exec(`delete from chat_log where id <= ?`, id-int64(keep)) + + return err + } + + return nil +} + +// recentChat returns the newest n lines of the chat, oldest first, and the +// highest id in the log. +func (s *Store) recentChat(n int) ([]Message, uint64, error) { + rows, err := s.db.Query(`select id, at, kind, nick, text from chat_log order by id desc limit ?`, n) + if err != nil { + return nil, 0, err + } + defer rows.Close() + + var out []Message + var top uint64 + for rows.Next() { + var ( + m Message + at int64 + kind string + ) + if err := rows.Scan(&m.ID, &at, &kind, &m.Nick, &m.Text); err != nil { + return nil, 0, err + } + m.At, m.Kind = time.Unix(at, 0), Kind(kind) + top = max(top, m.ID) + out = append(out, m) + } + for i, j := 0, len(out)-1; i < j; i, j = i+1, j-1 { + out[i], out[j] = out[j], out[i] + } + + return out, top, rows.Err() +}