feat: add stdin/pipe log source for command composability (#10)

Implement StdinSource conforming to the Source interface, enabling piped
input from kubectl, docker, cat, and other commands.

Features:
- Pipe vs terminal detection (IsPipe)
- Buffered channel with configurable backpressure (block or drop-oldest)
- Support for long log lines (up to 1 MB)
- Graceful shutdown via context cancellation
- Comprehensive tests with mock stdin readers

Fixes #8

Author: clarabennettdev

internal/source/stdin.go

@@ -0,0 +1,158 @@
+package source
+
+import (
+	"bufio"
+	"context"
+	"fmt"
+	"io"
+	"os"
+	"sync"
+)
+
+const (
+	// DefaultBufferSize is the default capacity for the lines channel.
+	DefaultBufferSize = 1000
+
+	// DropOldest discards the oldest unread line when the buffer is full.
+	DropOldest BackpressureStrategy = iota
+	// Block waits until a reader consumes a line before accepting more.
+	Block
+)
+
+// BackpressureStrategy controls behaviour when the lines channel is full.
+type BackpressureStrategy int
+
+// StdinOption configures a StdinSource.
+type StdinOption func(*StdinSource)
+
+// WithBufferSize sets the capacity of the lines channel.
+func WithBufferSize(n int) StdinOption {
+	return func(s *StdinSource) { s.bufSize = n }
+}
+
+// WithBackpressure sets the backpressure strategy.
+func WithBackpressure(bp BackpressureStrategy) StdinOption {
+	return func(s *StdinSource) { s.backpressure = bp }
+}
+
+// WithReader overrides the default stdin reader (useful for testing).
+func WithReader(r io.Reader) StdinOption {
+	return func(s *StdinSource) { s.reader = r }
+}
+
+// StdinSource reads log lines from standard input. It is designed to work
+// with piped input such as:
+//
+//	kubectl logs -f pod | logpilot
+//	cat app.log | logpilot
+//	docker logs -f container | logpilot
+type StdinSource struct {
+	reader       io.Reader
+	lines        chan LogEntry
+	errs         chan error
+	bufSize      int
+	backpressure BackpressureStrategy
+	cancel       context.CancelFunc
+	once         sync.Once
+	done         chan struct{}
+}
+
+// NewStdinSource creates a new StdinSource with the given options.
+func NewStdinSource(opts ...StdinOption) *StdinSource {
+	s := &StdinSource{
+		reader:       os.Stdin,
+		bufSize:      DefaultBufferSize,
+		backpressure: Block,
+		done:         make(chan struct{}),
+	}
+	for _, o := range opts {
+		o(s)
+	}
+	s.lines = make(chan LogEntry, s.bufSize)
+	s.errs = make(chan error, 1)
+	return s
+}
+
+// IsPipe reports whether stdin appears to be a pipe (not a terminal).
+func IsPipe() bool {
+	fi, err := os.Stdin.Stat()
+	if err != nil {
+		return false
+	}
+	return (fi.Mode() & os.ModeCharDevice) == 0
+}
+
+// Lines returns the channel of log entries.
+func (s *StdinSource) Lines() <-chan LogEntry { return s.lines }
+
+// Errors returns the channel of errors.
+func (s *StdinSource) Errors() <-chan error { return s.errs }
+
+// Start reads lines from stdin until ctx is cancelled or EOF is reached.
+func (s *StdinSource) Start(ctx context.Context) error {
+	ctx, s.cancel = context.WithCancel(ctx)
+	defer close(s.lines)
+	defer close(s.errs)
+	defer close(s.done)
+
+	scanner := bufio.NewScanner(s.reader)
+	// Support very long log lines (up to 1 MB).
+	scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024)
+
+	for scanner.Scan() {
+		entry := LogEntry{
+			Line:   scanner.Text(),
+			Source: "stdin",
+		}
+		if !s.emit(ctx, entry) {
+			return ctx.Err()
+		}
+	}
+	if err := scanner.Err(); err != nil {
+		select {
+		case s.errs <- fmt.Errorf("stdin read error: %w", err):
+		default:
+		}
+		return err
+	}
+	return nil
+}
+
+// emit sends an entry to the lines channel, respecting backpressure strategy.
+func (s *StdinSource) emit(ctx context.Context, entry LogEntry) bool {
+	switch s.backpressure {
+	case DropOldest:
+		select {
+		case s.lines <- entry:
+		default:
+			// Channel full — drop oldest.
+			select {
+			case <-s.lines:
+			default:
+			}
+			select {
+			case s.lines <- entry:
+			case <-ctx.Done():
+				return false
+			}
+		}
+	default: // Block
+		select {
+		case s.lines <- entry:
+		case <-ctx.Done():
+			return false
+		}
+	}
+	return true
+}
+
+// Stop cancels reading and waits for the reader goroutine to finish.
+func (s *StdinSource) Stop() error {
+	s.once.Do(func() {
+		if s.cancel != nil {
+			s.cancel()
+		}
+	})
+	<-s.done
+	return nil
+}

internal/source/stdin_test.go

@@ -0,0 +1,156 @@
+package source
+
+import (
+	"context"
+	"io"
+	"strings"
+	"testing"
+	"time"
+)
+
+func stdinCollectLines(t *testing.T, s *StdinSource, timeout time.Duration) []LogEntry {
+	t.Helper()
+	var entries []LogEntry
+	deadline := time.After(timeout)
+	for {
+		select {
+		case entry, ok := <-s.Lines():
+			if !ok {
+				return entries
+			}
+			entries = append(entries, entry)
+		case <-deadline:
+			t.Fatal("timed out waiting for lines")
+			return nil
+		}
+	}
+}
+
+func TestStdinSource_BasicRead(t *testing.T) {
+	input := "line one\nline two\nline three\n"
+	src := NewStdinSource(WithReader(strings.NewReader(input)))
+
+	go src.Start(context.Background())
+	entries := stdinCollectLines(t, src, 2*time.Second)
+
+	want := []string{"line one", "line two", "line three"}
+	if len(entries) != len(want) {
+		t.Fatalf("got %d lines, want %d", len(entries), len(want))
+	}
+	for i, e := range entries {
+		if e.Line != want[i] {
+			t.Errorf("line %d: got %q, want %q", i, e.Line, want[i])
+		}
+		if e.Source != "stdin" {
+			t.Errorf("line %d: source = %q, want \"stdin\"", i, e.Source)
+		}
+	}
+}
+
+func TestStdinSource_EmptyInput(t *testing.T) {
+	src := NewStdinSource(WithReader(strings.NewReader("")))
+
+	go src.Start(context.Background())
+	entries := stdinCollectLines(t, src, 2*time.Second)
+
+	if len(entries) != 0 {
+		t.Fatalf("expected 0 lines, got %d", len(entries))
+	}
+}
+
+func TestStdinSource_ContextCancellation(t *testing.T) {
+	pr, pw := io.Pipe()
+
+	src := NewStdinSource(WithReader(pr))
+	ctx, cancel := context.WithCancel(context.Background())
+
+	done := make(chan struct{})
+	go func() {
+		src.Start(ctx)
+		close(done)
+	}()
+
+	pw.Write([]byte("hello\n"))
+	<-src.Lines()
+	cancel()
+	pw.Close() // unblock scanner
+
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("Start did not return after context cancellation")
+	}
+}
+
+func TestStdinSource_Stop(t *testing.T) {
+	pr, pw := io.Pipe()
+
+	src := NewStdinSource(WithReader(pr))
+
+	go src.Start(context.Background())
+
+	pw.Write([]byte("line\n"))
+	<-src.Lines()
+
+	go pw.Close()
+	src.Stop()
+
+	_, ok := <-src.Lines()
+	if ok {
+		t.Fatal("expected lines channel to be closed")
+	}
+}
+
+func TestStdinSource_DropOldest(t *testing.T) {
+	input := "a\nb\nc\nd\n"
+	src := NewStdinSource(
+		WithReader(strings.NewReader(input)),
+		WithBufferSize(2),
+		WithBackpressure(DropOldest),
+	)
+
+	go src.Start(context.Background())
+	entries := stdinCollectLines(t, src, 2*time.Second)
+
+	if len(entries) < 2 {
+		t.Fatalf("expected at least 2 lines, got %d", len(entries))
+	}
+	if entries[len(entries)-1].Line != "d" {
+		t.Errorf("last line should be 'd', got %q", entries[len(entries)-1].Line)
+	}
+}
+
+func TestStdinSource_LongLines(t *testing.T) {
+	long := strings.Repeat("x", 500_000)
+	src := NewStdinSource(WithReader(strings.NewReader(long + "\n")))
+
+	go src.Start(context.Background())
+	entries := stdinCollectLines(t, src, 2*time.Second)
+
+	if len(entries) != 1 || len(entries[0].Line) != 500_000 {
+		t.Fatalf("expected 1 line of 500000 chars, got %d lines", len(entries))
+	}
+}
+
+func TestStdinSource_Errors(t *testing.T) {
+	src := NewStdinSource(WithReader(strings.NewReader("")))
+
+	go src.Start(context.Background())
+	stdinCollectLines(t, src, 2*time.Second)
+
+	select {
+	case err := <-src.Errors():
+		if err != nil {
+			t.Fatalf("unexpected error: %v", err)
+		}
+	default:
+	}
+}
+
+func TestStdinSource_ImplementsSource(t *testing.T) {
+	var _ Source = (*StdinSource)(nil)
+}
+
+func TestIsPipe(t *testing.T) {
+	_ = IsPipe()
+}