Refactor apps logs log streaming (#4086)

## Changes
- Show body details on dial error only if it's a JSON
- Extract backoff, tail buffer, formatter and consume state for better
readability

## Why
To increase code readability for easier maintenance.

## Tests

Unit tests are in place. I also did manual tests for the command usage
as before.

## Screenshots

### Before
We printed HTML which weren't usefull really:
<img width="1007" height="772" alt="Screenshot 2025-12-04 at 12 17 05"
src="https://github.com/user-attachments/assets/d0281c44-dbed-4db5-8a46-656460b317c3"
/>
### After

We don't print HTML response. Only if JSON is returned then it would be
printed.
<img width="800" height="108" alt="image"
src="https://github.com/user-attachments/assets/78bcf52c-8a45-48a9-a5f2-d83031e4538b"
/>

Author: pkosiec

libs/apps/logstream/backoff.go

@@ -0,0 +1,56 @@
+package logstream
+
+import (
+	"context"
+	"time"
+)
+
+const (
+	initialReconnectBackoff = 200 * time.Millisecond
+	maxReconnectBackoff     = 5 * time.Second
+)
+
+// backoffStrategy manages exponential backoff for reconnection attempts.
+type backoffStrategy struct {
+	initial time.Duration
+	max     time.Duration
+	current time.Duration
+}
+
+// newBackoffStrategy creates a new backoffStrategy with the given initial and max durations.
+func newBackoffStrategy(initial, max time.Duration) *backoffStrategy {
+	return &backoffStrategy{
+		initial: initial,
+		max:     max,
+		current: initial,
+	}
+}
+
+// Wait blocks until the current backoff duration has elapsed or the context is canceled.
+func (b *backoffStrategy) Wait(ctx context.Context) error {
+	if b.current <= 0 {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		default:
+			return nil
+		}
+	}
+
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-time.After(b.current):
+		return nil
+	}
+}
+
+// Next increases the backoff duration exponentially, capped at the max duration.
+func (b *backoffStrategy) Next() {
+	b.current = min(b.current*2, b.max)
+}
+
+// Reset returns the backoff duration to the initial value.
+func (b *backoffStrategy) Reset() {
+	b.current = b.initial
+}

libs/apps/logstream/backoff_test.go

@@ -0,0 +1,70 @@
+package logstream
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestBackoffStrategy_Next(t *testing.T) {
+	b := newBackoffStrategy(100*time.Millisecond, 500*time.Millisecond)
+	assert.Equal(t, 100*time.Millisecond, b.current)
+
+	b.Next()
+	assert.Equal(t, 200*time.Millisecond, b.current)
+
+	b.Next()
+	assert.Equal(t, 400*time.Millisecond, b.current)
+
+	assertMsg := "should be capped at max"
+	b.Next()
+	assert.Equal(t, 500*time.Millisecond, b.current, assertMsg)
+	b.Next()
+	assert.Equal(t, 500*time.Millisecond, b.current, assertMsg)
+}
+
+func TestBackoffStrategy_Reset(t *testing.T) {
+	b := newBackoffStrategy(100*time.Millisecond, 1*time.Second)
+	assert.Equal(t, 100*time.Millisecond, b.current)
+
+	b.Next()
+	b.Next()
+	assert.Equal(t, 400*time.Millisecond, b.current)
+
+	b.Reset()
+	assert.Equal(t, 100*time.Millisecond, b.current)
+}
+
+func TestBackoffStrategy_Wait(t *testing.T) {
+	t.Run("blocks for duration", func(t *testing.T) {
+		b := newBackoffStrategy(50*time.Millisecond, 100*time.Millisecond)
+
+		start := time.Now()
+		err := b.Wait(context.Background())
+		elapsed := time.Since(start)
+
+		require.NoError(t, err)
+		assert.GreaterOrEqual(t, elapsed, 50*time.Millisecond)
+		assert.Less(t, elapsed, 100*time.Millisecond)
+	})
+
+	t.Run("returns early on cancel", func(t *testing.T) {
+		b := newBackoffStrategy(1*time.Second, 5*time.Second)
+
+		ctx, cancel := context.WithCancel(context.Background())
+		go func() {
+			time.Sleep(20 * time.Millisecond)
+			cancel()
+		}()
+
+		start := time.Now()
+		err := b.Wait(ctx)
+		elapsed := time.Since(start)
+
+		assert.ErrorIs(t, err, context.Canceled)
+		assert.Less(t, elapsed, 100*time.Millisecond, "should return early on cancel")
+	})
+}

libs/apps/logstream/consume_state.go

@@ -0,0 +1,100 @@
+package logstream
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"time"
+)
+
+// consumeState manages the tail buffer and flush timing during log consumption.
+type consumeState struct {
+	buffer        *tailBuffer
+	writer        io.Writer
+	flushed       bool
+	flushDeadline time.Time
+	tail          int
+	follow        bool
+}
+
+// newConsumeState creates a new consume state.
+func newConsumeState(tail int, follow bool, prefetch time.Duration, writer io.Writer, alreadyFlushed bool) *consumeState {
+	s := &consumeState{
+		buffer:  &tailBuffer{size: tail},
+		writer:  writer,
+		flushed: tail == 0 || alreadyFlushed,
+		tail:    tail,
+		follow:  follow,
+	}
+	if tail > 0 && prefetch > 0 && !alreadyFlushed {
+		s.flushDeadline = time.Now().Add(prefetch)
+	}
+	return s
+}
+
+// ReadDeadline returns the effective read deadline considering context and flush deadline.
+func (s *consumeState) ReadDeadline(ctx context.Context) time.Time {
+	ctxDeadline, hasCtxDeadline := ctx.Deadline()
+
+	if !s.flushDeadline.IsZero() {
+		if !hasCtxDeadline || s.flushDeadline.Before(ctxDeadline) {
+			return s.flushDeadline
+		}
+	}
+
+	if hasCtxDeadline {
+		return ctxDeadline
+	}
+	return time.Time{}
+}
+
+// HasPendingFlushDeadline returns true if a prefetch flush deadline is pending.
+func (s *consumeState) HasPendingFlushDeadline() bool {
+	return !s.flushDeadline.IsZero()
+}
+
+// HandleFlushTimeout handles a prefetch timeout by flushing the buffer.
+// Only call this when HasPendingFlushDeadline() returns true.
+// Returns true if reading should continue (following), false if done (not following).
+func (s *consumeState) HandleFlushTimeout() (shouldContinue bool, err error) {
+	s.flushDeadline = time.Time{}
+	if s.tail > 0 && !s.flushed {
+		if err := s.buffer.Flush(s.writer); err != nil {
+			return false, err
+		}
+		s.flushed = true
+	}
+	return s.follow, nil
+}
+
+// ProcessLine either buffers or writes the line depending on flush state.
+func (s *consumeState) ProcessLine(line string) error {
+	if s.tail > 0 && !s.flushed {
+		s.buffer.Add(line)
+		if s.flushDeadline.IsZero() && s.buffer.Len() >= s.tail && s.follow {
+			if err := s.buffer.Flush(s.writer); err != nil {
+				return err
+			}
+			s.flushed = true
+		}
+		return nil
+	}
+	_, err := fmt.Fprintln(s.writer, line)
+	return err
+}
+
+// FlushRemaining flushes any remaining buffered lines.
+func (s *consumeState) FlushRemaining() error {
+	if s.tail > 0 && !s.flushed {
+		if err := s.buffer.Flush(s.writer); err != nil {
+			return err
+		}
+		s.flushed = true
+	}
+	return nil
+}
+
+// IsFlushed returns whether the buffer has been flushed.
+func (s *consumeState) IsFlushed() bool {
+	return s.flushed
+}

libs/apps/logstream/formatter.go

@@ -0,0 +1,66 @@
+package logstream
+
+import (
+	"encoding/json"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/fatih/color"
+)
+
+// wsEntry represents a structured log entry from the websocket stream.
+type wsEntry struct {
+	Source    string  `json:"source"`
+	Timestamp float64 `json:"timestamp"`
+	Message   string  `json:"message"`
+}
+
+// parseLogEntry parses a raw log entry from the websocket stream.
+func parseLogEntry(raw []byte) (*wsEntry, error) {
+	var entry wsEntry
+	if err := json.Unmarshal(raw, &entry); err != nil {
+		return nil, err
+	}
+	return &entry, nil
+}
+
+// logFormatter formats log entries for output.
+type logFormatter struct {
+	colorize bool
+}
+
+// newLogFormatter creates a new log formatter.
+func newLogFormatter(colorize bool) *logFormatter {
+	return &logFormatter{colorize: colorize}
+}
+
+// FormatEntry formats a structured log entry for output.
+func (f *logFormatter) FormatEntry(entry *wsEntry) string {
+	timestamp := formatTimestamp(entry.Timestamp)
+	source := strings.ToUpper(entry.Source)
+	message := strings.TrimRight(entry.Message, "\r\n")
+
+	if f.colorize {
+		timestamp = color.HiBlackString(timestamp)
+		source = color.HiBlueString(source)
+	}
+
+	return fmt.Sprintf("%s [%s] %s", timestamp, source, message)
+}
+
+// FormatPlain formats a plain text message by trimming trailing newlines.
+func (f *logFormatter) FormatPlain(raw []byte) string {
+	return strings.TrimRight(string(raw), "\r\n")
+}
+
+// formatTimestamp formats a timestamp as a string.
+func formatTimestamp(ts float64) string {
+	if ts <= 0 {
+		return "----------"
+	}
+	sec := int64(ts)
+	nsec := int64((ts - float64(sec)) * 1e9)
+	t := time.Unix(sec, nsec).UTC()
+	return t.Format(time.RFC3339)
+}