Add thread-safe eventstream ring buffer

1 files changed, 59 insertions, 0 deletions

diff --git a/internal/tui/eventstream/ringbuffer.go b/internal/tui/eventstream/ringbuffer.go
new file mode 100644
index 0000000..a2ec1dc
--- /dev/null
+++ b/internal/tui/eventstream/ringbuffer.go
@@ -0,0 +1,59 @@
+package eventstream
+
+import "sync"
+
+const ringBufferCapacity = 10000
+
+type RingBuffer struct {
+	mu          sync.RWMutex
+	buf         []StreamEvent
+	start       int
+	size        int
+	totalPushed uint64
+}
+
+func NewRingBuffer() *RingBuffer {
+	return &RingBuffer{buf: make([]StreamEvent, ringBufferCapacity)}
+}
+
+func (r *RingBuffer) Push(ev StreamEvent) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	if r.size < ringBufferCapacity {
+		idx := (r.start + r.size) % ringBufferCapacity
+		r.buf[idx] = ev
+		r.size++
+	} else {
+		r.buf[r.start] = ev
+		r.start = (r.start + 1) % ringBufferCapacity
+	}
+	r.totalPushed++
+}
+
+func (r *RingBuffer) Snapshot() []StreamEvent {
+	r.mu.RLock()
+	defer r.mu.RUnlock()
+
+	if r.size == 0 {
+		return make([]StreamEvent, 0)
+	}
+
+	out := make([]StreamEvent, r.size)
+	for i := 0; i < r.size; i++ {
+		out[i] = r.buf[(r.start+i)%ringBufferCapacity]
+	}
+	return out
+}
+
+func (r *RingBuffer) Len() int {
+	r.mu.RLock()
+	defer r.mu.RUnlock()
+	return r.size
+}
+
+func (r *RingBuffer) TotalPushed() uint64 {
+	r.mu.RLock()
+	defer r.mu.RUnlock()
+	return r.totalPushed
+}