// Copyright (c) 2018 Shivaram Lingamneni // released under the MIT license package history import ( "github.com/ergochat/ergo/irc/utils" "sync" "time" ) type ItemType uint const ( uninitializedItem ItemType = iota Privmsg Notice Join Part Kick Quit Mode Tagmsg Nick Topic Invite ) const ( initialAutoSize = 32 ) // Item represents an event (e.g., a PRIVMSG or a JOIN) and its associated data type Item struct { Type ItemType Nick string // this is the uncasefolded account name, if there's no account it should be set to "*" AccountName string // for non-privmsg items, we may stuff some other data in here Message utils.SplitMessage Tags map[string]string Params [1]string // for a DM, this is the casefolded nickname of the other party (whether this is // an incoming or outgoing message). this lets us emulate the "query buffer" functionality // required by CHATHISTORY: CfCorrespondent string `json:"CfCorrespondent,omitempty"` IsBot bool `json:"IsBot,omitempty"` } // HasMsgid tests whether a message has the message id `msgid`. func (item *Item) HasMsgid(msgid string) bool { return item.Message.Msgid == msgid } type Predicate func(item *Item) (matches bool) func Reverse(results []Item) { for i, j := 0, len(results)-1; i < j; i, j = i+1, j-1 { results[i], results[j] = results[j], results[i] } } // Buffer is a ring buffer holding message/event history for a channel or user type Buffer struct { sync.RWMutex // ring buffer, see irc/whowas.go for conventions buffer []Item start int end int maximumSize int window time.Duration lastDiscarded time.Time nowFunc func() time.Time } func NewHistoryBuffer(size int, window time.Duration) (result *Buffer) { result = new(Buffer) result.Initialize(size, window) return } func (hist *Buffer) Initialize(size int, window time.Duration) { hist.buffer = make([]Item, hist.initialSize(size, window)) hist.start = -1 hist.end = -1 hist.window = window hist.maximumSize = size hist.nowFunc = time.Now } // compute the initial size for the buffer, taking into account autoresize func (hist *Buffer) initialSize(size int, window time.Duration) (result int) { result = size if window != 0 { result = initialAutoSize if size < result { result = size // min(initialAutoSize, size) } } return } // Add adds a history item to the buffer func (list *Buffer) Add(item Item) { if item.Message.Time.IsZero() { item.Message.Time = time.Now().UTC() } list.Lock() defer list.Unlock() if len(list.buffer) == 0 { return } list.maybeExpand() var pos int if list.start == -1 { // empty pos = 0 list.start = 0 list.end = 1 % len(list.buffer) } else if list.start != list.end { // partially full pos = list.end list.end = (list.end + 1) % len(list.buffer) } else if list.start == list.end { // full pos = list.end list.end = (list.end + 1) % len(list.buffer) list.start = list.end // advance start as well, overwriting first entry // record the timestamp of the overwritten item if list.lastDiscarded.Before(list.buffer[pos].Message.Time) { list.lastDiscarded = list.buffer[pos].Message.Time } } list.buffer[pos] = item } func (list *Buffer) lookup(msgid string) (result Item, found bool) { predicate := func(item *Item) bool { return item.HasMsgid(msgid) } results := list.matchInternal(predicate, false, 1) if len(results) != 0 { return results[0], true } return } // Between returns all history items with a time `after` <= time <= `before`, // with an indication of whether the results are complete or are missing items // because some of that period was discarded. A zero value of `before` is considered // higher than all other times. func (list *Buffer) betweenHelper(start, end Selector, cutoff time.Time, pred Predicate, limit int) (results []Item, complete bool, err error) { var ascending bool defer func() { if !ascending { Reverse(results) } }() list.RLock() defer list.RUnlock() if len(list.buffer) == 0 { return } after := start.Time if start.Msgid != "" { item, found := list.lookup(start.Msgid) if !found { return } after = item.Message.Time } before := end.Time if end.Msgid != "" { item, found := list.lookup(end.Msgid) if !found { return } before = item.Message.Time } after, before, ascending = MinMaxAsc(after, before, cutoff) complete = after.Equal(list.lastDiscarded) || after.After(list.lastDiscarded) satisfies := func(item *Item) bool { return (after.IsZero() || item.Message.Time.After(after)) && (before.IsZero() || item.Message.Time.Before(before)) && (pred == nil || pred(item)) } return list.matchInternal(satisfies, ascending, limit), complete, nil } // returns all correspondents, in reverse time order func (list *Buffer) allCorrespondents() (results []TargetListing) { seen := make(utils.StringSet) list.RLock() defer list.RUnlock() if list.start == -1 || len(list.buffer) == 0 { return } // XXX traverse in reverse order, so we get the latest timestamp // of any message sent to/from the correspondent pos := list.prev(list.end) stop := list.start for { if !seen.Has(list.buffer[pos].CfCorrespondent) { seen.Add(list.buffer[pos].CfCorrespondent) results = append(results, TargetListing{ CfName: list.buffer[pos].CfCorrespondent, Time: list.buffer[pos].Message.Time, }) } if pos == stop { break } pos = list.prev(pos) } return } // list DM correspondents, as one input to CHATHISTORY TARGETS func (list *Buffer) listCorrespondents(start, end Selector, cutoff time.Time, limit int) (results []TargetListing, err error) { after := start.Time before := end.Time after, before, ascending := MinMaxAsc(after, before, cutoff) correspondents := list.allCorrespondents() if len(correspondents) == 0 { return } // XXX allCorrespondents returns results in reverse order, // so if we're ascending, we actually go backwards var i int if ascending { i = len(correspondents) - 1 } else { i = 0 } for 0 <= i && i < len(correspondents) && (limit == 0 || len(results) < limit) { if (after.IsZero() || correspondents[i].Time.After(after)) && (before.IsZero() || correspondents[i].Time.Before(before)) { results = append(results, correspondents[i]) } if ascending { i-- } else { i++ } } if !ascending { ReverseCorrespondents(results) } return } // implements history.Sequence, emulating a single history buffer (for a channel, // a single user's DMs, or a DM conversation) type bufferSequence struct { list *Buffer pred Predicate cutoff time.Time } func (list *Buffer) MakeSequence(correspondent string, cutoff time.Time) Sequence { var pred Predicate if correspondent != "" { pred = func(item *Item) bool { return item.CfCorrespondent == correspondent } } return &bufferSequence{ list: list, pred: pred, cutoff: cutoff, } } func (seq *bufferSequence) Between(start, end Selector, limit int) (results []Item, err error) { results, _, err = seq.list.betweenHelper(start, end, seq.cutoff, seq.pred, limit) return } func (seq *bufferSequence) Around(start Selector, limit int) (results []Item, err error) { return GenericAround(seq, start, limit) } func (seq *bufferSequence) ListCorrespondents(start, end Selector, limit int) (results []TargetListing, err error) { return seq.list.listCorrespondents(start, end, seq.cutoff, limit) } func (seq *bufferSequence) Cutoff() time.Time { return seq.cutoff } func (seq *bufferSequence) Ephemeral() bool { return true } // you must be holding the read lock to call this func (list *Buffer) matchInternal(predicate Predicate, ascending bool, limit int) (results []Item) { if list.start == -1 || len(list.buffer) == 0 { return } var pos, stop int if ascending { pos = list.start stop = list.prev(list.end) } else { pos = list.prev(list.end) stop = list.start } for { if predicate(&list.buffer[pos]) { results = append(results, list.buffer[pos]) } if pos == stop || (limit != 0 && len(results) == limit) { break } if ascending { pos = list.next(pos) } else { pos = list.prev(pos) } } return } // Delete deletes messages matching some predicate. func (list *Buffer) Delete(predicate Predicate) (count int) { list.Lock() defer list.Unlock() if list.start == -1 || len(list.buffer) == 0 { return } pos := list.start stop := list.prev(list.end) for { if predicate(&list.buffer[pos]) { list.buffer[pos] = Item{} count++ } if pos == stop { break } pos = list.next(pos) } return } // latest returns the items most recently added, up to `limit`. If `limit` is 0, // it returns all items. func (list *Buffer) latest(limit int) (results []Item) { results, _, _ = list.betweenHelper(Selector{}, Selector{}, time.Time{}, nil, limit) return } // LastDiscarded returns the latest time of any entry that was evicted // from the ring buffer. func (list *Buffer) LastDiscarded() time.Time { list.RLock() defer list.RUnlock() return list.lastDiscarded } func (list *Buffer) prev(index int) int { switch index { case 0: return len(list.buffer) - 1 default: return index - 1 } } func (list *Buffer) next(index int) int { switch index { case len(list.buffer) - 1: return 0 default: return index + 1 } } func (list *Buffer) maybeExpand() { if list.window == 0 { return // autoresize is disabled } length := list.length() if length < len(list.buffer) { return // we have spare capacity already } if len(list.buffer) == list.maximumSize { return // cannot expand any further } wouldDiscard := list.buffer[list.start].Message.Time if list.window < list.nowFunc().Sub(wouldDiscard) { return // oldest element is old enough to overwrite } newSize := utils.RoundUpToPowerOfTwo(length + 1) if list.maximumSize < newSize { newSize = list.maximumSize } list.resize(newSize) } // Resize shrinks or expands the buffer func (list *Buffer) Resize(maximumSize int, window time.Duration) { list.Lock() defer list.Unlock() if list.maximumSize == maximumSize && list.window == window { return // no-op } list.maximumSize = maximumSize list.window = window // three cases where we need to preemptively resize: // (1) we are not autoresizing // (2) the buffer is currently larger than maximumSize and needs to be shrunk // (3) the buffer is currently smaller than the recommended initial size // (including the case where it is currently disabled and needs to be enabled) // TODO make it possible to shrink the buffer so that it only contains `window` if window == 0 || maximumSize < len(list.buffer) { list.resize(maximumSize) } else { initialSize := list.initialSize(maximumSize, window) if len(list.buffer) < initialSize { list.resize(initialSize) } } } func (list *Buffer) resize(size int) { newbuffer := make([]Item, size) if list.start == -1 { // indices are already correct and nothing needs to be copied } else if size == 0 { // this is now the empty list list.start = -1 list.end = -1 } else { currentLength := list.length() start := list.start end := list.end // if we're truncating, keep the latest entries, not the earliest if size < currentLength { start = list.end - size if start < 0 { start += len(list.buffer) } // update lastDiscarded for discarded entries for i := list.start; i != start; i = (i + 1) % len(list.buffer) { if list.lastDiscarded.Before(list.buffer[i].Message.Time) { list.lastDiscarded = list.buffer[i].Message.Time } } } if start < end { copied := copy(newbuffer, list.buffer[start:end]) list.start = 0 list.end = copied % size } else { lenInitial := len(list.buffer) - start copied := copy(newbuffer, list.buffer[start:]) copied += copy(newbuffer[lenInitial:], list.buffer[:end]) list.start = 0 list.end = copied % size } } list.buffer = newbuffer } func (hist *Buffer) length() int { if hist.start == -1 { return 0 } else if hist.start < hist.end { return hist.end - hist.start } else { return len(hist.buffer) - (hist.start - hist.end) } }