// merge.go — one room made of several (RFC 0018 §"Gathering chat").
//
// A team's room carries the lines from every repository under it, and
// nothing is copied to build it: every stored message already names the
// repository it was posted in, so a merged room is a READ across
// channels in one order. This file is that read, and it is here rather
// than in the ops layer because it is arithmetic over messages with no
// opinion about teams, repositories or authority.
package chat
import (
"sort"
"time"
)
// Backward reads one channel's page, newest-page-first: at most limit
// messages older than before (0 = start at the newest), returned
// OLDEST-FIRST, plus the seq to pass as before for the page after it —
// ChatPlane.History's contract exactly, so the ops layer passes its
// plane method straight in.
type Backward func(code string, before uint64, limit int) ([]Message, uint64, error)
// Cursor is where each channel has been read back to. A code absent
// from the map has not been read yet (start at its newest); a code
// mapped to 0 after a read is exhausted — the distinction matters,
// because "no cursor" and "no more" are opposite answers and a merged
// page must not restart a finished channel.
type Cursor struct {
Before map[string]uint64 `json:"before,omitempty"`
Done map[string]bool `json:"done,omitempty"`
}
// MergeBack returns the newest limit messages across codes, newest
// first, with the cursor for the page before it.
//
// It reads at most limit from each channel per page. That is
// proportional to the team's repositories, which is inherent: N streams
// cannot be merged without reading N of them. It is NOT proportional to
// the forge, or to the team's history.
func MergeBack(codes []string, cur Cursor, limit int, back Backward) ([]Message, Cursor, error) {
next := Cursor{Before: map[string]uint64{}, Done: map[string]bool{}}
for k, v := range cur.Before {
next.Before[k] = v
}
for k, v := range cur.Done {
next.Done[k] = v
}
if limit <= 0 || len(codes) == 0 {
return []Message{}, next, nil
}
// One buffer per channel, newest-first so the merge pops heads.
type buf struct {
code string
msgs []Message
more uint64 // History's next: 0 = this channel is exhausted
}
bufs := make([]*buf, 0, len(codes))
for _, code := range codes {
if cur.Done[code] {
continue
}
msgs, more, err := back(code, cur.Before[code], limit)
if err != nil {
return nil, cur, err
}
// Oldest-first in, newest-first out.
for i, j := 0, len(msgs)-1; i < j; i, j = i+1, j-1 {
msgs[i], msgs[j] = msgs[j], msgs[i]
}
bufs = append(bufs, &buf{code: code, msgs: msgs, more: more})
if more == 0 && len(msgs) == 0 {
next.Done[code] = true
}
}
// At is RFC3339Nano, whose fractional part has trailing zeros
// trimmed — so ".1" and ".05" do not compare correctly as strings.
// Parsed once per message rather than per comparison.
when := func(m Message) time.Time {
t, err := time.Parse(time.RFC3339Nano, m.At)
if err != nil {
return time.Time{}
}
return t
}
out := make([]Message, 0, limit)
taken := map[string]uint64{} // oldest seq consumed, per code
for len(out) < limit {
pick := -1
for i, b := range bufs {
if len(b.msgs) == 0 {
continue
}
if pick < 0 {
pick = i
continue
}
a, c := when(b.msgs[0]), when(bufs[pick].msgs[0])
// Newest first. A tie is broken by code then seq so the
// order is total: two channels stamped in the same
// nanosecond must not shuffle between renders.
if a.After(c) || (a.Equal(c) &&
(b.code < bufs[pick].code ||
(b.code == bufs[pick].code && b.msgs[0].Seq > bufs[pick].msgs[0].Seq))) {
pick = i
}
}
if pick < 0 {
break
}
b := bufs[pick]
m := b.msgs[0]
b.msgs = b.msgs[1:]
out = append(out, m)
taken[b.code] = m.Seq
}
// Where each channel has been read back to. A channel we consumed
// from resumes below its oldest consumed message; one we read and
// did not consume from keeps its old cursor, so its buffered
// messages are re-read rather than skipped.
for _, b := range bufs {
seq, consumed := taken[b.code]
switch {
case !consumed:
// nothing taken: leave the cursor alone.
case len(b.msgs) > 0:
// Still buffered messages older than the ones we emitted,
// so the next page starts at the oldest we consumed.
next.Before[b.code] = seq
case b.more == 0:
next.Before[b.code] = seq
next.Done[b.code] = true
default:
next.Before[b.code] = seq
}
}
return out, next, nil
}
// Codes is the channel list for a team's room: the team's own channel
// first, then its repositories'. Sorted so the merge's tie-break is
// stable across renders and nodes.
func Codes(teamCode string, repoCodes []string) []string {
out := append([]string{teamCode}, repoCodes...)
sort.Strings(out)
return out
}
// TeamCode is a team's own channel. Repository channels are keyed by
// the store's project code, which is 40 hex characters, so this prefix
// cannot collide with one.
func TeamCode(team string) string { return "team-" + team }