OwnCord/Server/ws/ringbuffer.go

package ws

import "github.com/owncord/server/syncutil"

// eventEntry stores a broadcast event for potential replay.
type eventEntry struct {
	seq       uint64
	channelID int64 // 0 = global broadcast, >0 = channel-scoped
	data      []byte
}

// EventRingBuffer is a bounded, thread-safe ring buffer for recent broadcast events.
type EventRingBuffer struct {
	mu      syncutil.RWMutex
	entries []eventEntry
	size    int
	pos     int // next write position
	count   int // total entries stored (up to size)
}

// NewEventRingBuffer creates a ring buffer with the given capacity.
func NewEventRingBuffer(size int) *EventRingBuffer {
	return &EventRingBuffer{
		entries: make([]eventEntry, size),
		size:    size,
	}
}

// Push adds an event to the ring buffer.
// channelID identifies the channel scope (0 = global broadcast).
func (rb *EventRingBuffer) Push(seq uint64, channelID int64, data []byte) {
	rb.mu.Lock()
	defer rb.mu.Unlock()
	rb.entries[rb.pos] = eventEntry{seq: seq, channelID: channelID, data: data}
	rb.pos = (rb.pos + 1) % rb.size
	if rb.count < rb.size {
		rb.count++
	}
}

// EventsSince returns all events with seq > afterSeq, in order.
// Returns nil if afterSeq is too old (no longer in the buffer).
func (rb *EventRingBuffer) EventsSince(afterSeq uint64) [][]byte {
	rb.mu.RLock()
	defer rb.mu.RUnlock()

	if rb.count == 0 {
		return nil
	}

	// Find the oldest entry in the buffer.
	oldestIdx := (rb.pos - rb.count + rb.size) % rb.size
	oldestSeq := rb.entries[oldestIdx].seq

	// If the requested seq is at or older than our oldest, we can't guarantee
	// full coverage — return nil to trigger a full ready payload.
	if afterSeq <= oldestSeq {
		return nil
	}

	// Likewise if the client claims events newer than anything we ever held:
	// its counter and ours disagree (a restart can reseed seq below a client's
	// remembered lastSeq), so an empty slice here would be read as "caught up"
	// and freeze that client. afterSeq == newestSeq is the legitimate caught-up
	// case and still returns an empty replay.
	if afterSeq > rb.newestSeqLocked() {
		return nil
	}

	result := make([][]byte, 0)
	for i := 0; i < rb.count; i++ {
		idx := (oldestIdx + i) % rb.size
		e := rb.entries[idx]
		if e.seq > afterSeq {
			result = append(result, e.data)
		}
	}
	return result
}

// newestSeqLocked returns the highest sequence number in the buffer. Callers
// must hold rb.mu and must have checked rb.count > 0.
func (rb *EventRingBuffer) newestSeqLocked() uint64 {
	return rb.entries[(rb.pos-1+rb.size)%rb.size].seq
}

// EventsSinceFiltered returns events with seq > afterSeq whose channelID is
// in allowedChannelIDs or whose channelID is 0 (global broadcasts).
// Returns nil if afterSeq is too old (same semantics as EventsSince).
func (rb *EventRingBuffer) EventsSinceFiltered(afterSeq uint64, allowedChannelIDs map[int64]bool) [][]byte {
	rb.mu.RLock()
	defer rb.mu.RUnlock()

	if rb.count == 0 {
		return nil
	}

	oldestIdx := (rb.pos - rb.count + rb.size) % rb.size
	oldestSeq := rb.entries[oldestIdx].seq

	if afterSeq <= oldestSeq {
		return nil
	}

	// See EventsSince: a client ahead of everything we ever buffered must get a
	// full ready, not a silent "caught up".
	if afterSeq > rb.newestSeqLocked() {
		return nil
	}

	result := make([][]byte, 0)
	for i := 0; i < rb.count; i++ {
		idx := (oldestIdx + i) % rb.size
		e := rb.entries[idx]
		if e.seq > afterSeq {
			// channelID 0 = global broadcast, always include.
			// channelID > 0 = channel-scoped, include only if allowed.
			if e.channelID == 0 || allowedChannelIDs[e.channelID] {
				result = append(result, e.data)
			}
		}
	}
	return result
}

// OldestSeq returns the oldest sequence number in the buffer, or 0 if empty.
func (rb *EventRingBuffer) OldestSeq() uint64 {
	rb.mu.RLock()
	defer rb.mu.RUnlock()
	if rb.count == 0 {
		return 0
	}
	oldestIdx := (rb.pos - rb.count + rb.size) % rb.size
	return rb.entries[oldestIdx].seq
}