tia-love/internal/app/logstream.go

package app

import (
	"bytes"
	"context"
	"io"
	"strings"
	"time"

	tea "charm.land/bubbletea/v2"
	"github.com/docker/docker/pkg/stdcopy"

	"gitea.oxpoe.qzz.io/oxpoe/tia-love/internal/docker"
)

// --- pesan ---

// logBatchMsg — sekumpulan baris log (dikirim per batch agar hemat render).
type logBatchMsg struct {
	id    int64
	lines []string
	eof   bool
}

// logTickMsg — tidak ada data dalam window idle; re-arm tanpa state change.
type logTickMsg struct{ id int64 }

// statsMsg / statsTickMsg — snapshot metrik kontainer (tab Statistik).
type statsMsg struct {
	id  int64
	st  docker.Stats
	err error
}
type statsTickMsg struct{ id int64 }

const statsInterval = time.Second

func fetchStats(id int64, cl *docker.Client, cid string) tea.Cmd {
	return func() tea.Msg {
		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		st, err := cl.StatsOnce(ctx, cid)
		return statsMsg{id: id, st: st, err: err}
	}
}

func statsNext(id int64) tea.Cmd {
	return tea.Tick(statsInterval, func(time.Time) tea.Msg { return statsTickMsg{id: id} })
}

// inspectMsg / envMsg / procsMsg — data tab Konfigurasi / Env / Proses
// (one-shot saat tab dibuka atau `r`).
type inspectMsg struct {
	id   int64
	data []byte
	err  error
}
type envMsg struct {
	id  int64
	env []string
	err error
}
type procsMsg struct {
	id     int64
	titles []string
	rows   [][]string
	err    error
}

func fetchInspect(id int64, cl *docker.Client, cid string) tea.Cmd {
	return func() tea.Msg {
		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		b, err := cl.InspectJSON(ctx, cid)
		return inspectMsg{id: id, data: b, err: err}
	}
}

func fetchEnv(id int64, cl *docker.Client, cid string) tea.Cmd {
	return func() tea.Msg {
		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		env, err := cl.InspectEnv(ctx, cid)
		return envMsg{id: id, env: env, err: err}
	}
}

func fetchProcs(id int64, cl *docker.Client, cid string) tea.Cmd {
	return func() tea.Msg {
		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		titles, rows, err := cl.TopProcesses(ctx, cid)
		return procsMsg{id: id, titles: titles, rows: rows, err: err}
	}
}

const (
	logBatchMax  = 200                    // maksimum baris per batch
	logIdle      = 200 * time.Millisecond // tunggu baris pertama
	logDrainWait = 50 * time.Millisecond  // jendela serap baris siap berikutnya
)

// lineWriter — pecah byte stream jadi baris dan dorong ke channel.
// Bila konsumen lambat (buffer penuh), baris di-drop supaya UI tetap responsif
// (shortcut: log yang lewat di detik ekstrem bisa hilang — listener batch
// akan mengejar; upgrade bila perlu arsip log lengkap di file).
type lineWriter struct {
	ch  chan<- string
	buf []byte
}

func newLineWriter(ch chan<- string) *lineWriter { return &lineWriter{ch: ch} }

func (w *lineWriter) Write(p []byte) (int, error) {
	w.buf = append(w.buf, p...)
	for {
		i := bytes.IndexByte(w.buf, '\n')
		if i < 0 {
			break
		}
		line := strings.TrimRight(string(w.buf[:i]), "\r")
		w.buf = w.buf[i+1:]
		select {
		case w.ch <- line:
		default:
		}
	}
	if len(w.buf) > 1<<20 { // baris raksasa tanpa newline: jaga memori
		w.buf = nil
	}
	return len(p), nil
}

func (w *lineWriter) flush() {
	if len(w.buf) == 0 {
		return
	}
	select {
	case w.ch <- string(w.buf):
	default:
	}
	w.buf = nil
}

// streamContainerLogs — goroutine per stream: inspect (TTY) → docker logs →
// demux → lineWriter → channel. Menutup channel saat selesai/ctx dibatalkan.
func streamContainerLogs(ctx context.Context, cl *docker.Client, id string, tail int, timestamps bool, ch chan<- string) {
	defer close(ch)

	tty, err := cl.ContainerTty(ctx, id)
	if err != nil {
		ch <- "! " + err.Error()
		return
	}
	rc, err := cl.StreamLogs(ctx, id, tail, timestamps)
	if err != nil {
		ch <- "! " + err.Error()
		return
	}
	defer rc.Close()

	lw := newLineWriter(ch)
	if tty {
		_, _ = io.Copy(lw, rc)
	} else {
		_, _ = stdcopy.StdCopy(lw, lw, rc) // stdout & stderr dicampur
	}
	lw.flush()
}

// waitLogBatch — tea.Cmd: tunggu baris pertama (maks logIdle), lalu serap
// sebanyak mungkin dalam jendela pendek supaya 1 render menampung banyak baris.
func waitLogBatch(id int64, ch <-chan string) tea.Cmd {
	return func() tea.Msg {
		var lines []string
		select {
		case l, ok := <-ch:
			if !ok {
				return logBatchMsg{id: id, eof: true}
			}
			lines = append(lines, l)
		case <-time.After(logIdle):
			return logTickMsg{id: id}
		}

		timer := time.NewTimer(logDrainWait)
		defer timer.Stop()
		for len(lines) < logBatchMax {
			select {
			case l, ok := <-ch:
				if !ok {
					return logBatchMsg{id: id, lines: lines, eof: true}
				}
				lines = append(lines, l)
			case <-timer.C:
				return logBatchMsg{id: id, lines: lines}
			}
		}
		return logBatchMsg{id: id, lines: lines}
	}
}