// Controlled peers can send malformed batches and data after overflow, which
// the daemon's agent bus cannot produce through the E2E provider.
package daemon

import (
	"errors"
	"io"
	"net"
	"net/http"
	"net/http/httptest"
	"testing"
)

func TestAgentStreamReadinessAndOverflowStopDelivery(t *testing.T) {
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "text/event-stream")
		_, _ = io.WriteString(w, "data: "+`{"events":[]}`+"\n\n"+
			"data: "+`{"events":[]}`+"\n\n"+
			"data: "+`{"events":[{"type":"overflow"}]}`+"\n\n"+
			"data: "+`{"events":[{"type":"text","session":"s","text":"must stay hidden"}]}`+"\n\n")
	}))
	defer server.Close()
	conn := NewConnection(ConnectionSnapshot{Port: server.Listener.Addr().(*net.TCPAddr).Port}, nil)
	var delivered [][]AgentEvent
	err := StreamAgents(t.Context(), conn, func(events []AgentEvent) error { delivered = append(delivered, events); return nil })
	if err != nil {
		t.Fatal(err)
	}
	if len(delivered) != 2 || len(delivered[0]) != 0 || len(delivered[1]) != 1 || delivered[1][0].Type != "overflow" {
		t.Fatalf("readiness or overflow delivery changed: %+v", delivered)
	}
}

func TestAgentStreamRejectsMalformedBatchBeforeDelivery(t *testing.T) {
	for name, batch := range map[string]string{
		"missing events": `{}`,
		"null events":    `{"events":null}`,
		"missing type":   `{"events":[{"session":"s"}]}`,
		"empty type":     `{"events":[{"type":""}]}`,
		"partial batch":  `{"events":[{"type":"text","session":"s","text":"must stay hidden"},{"type":"running","session":"s","running":"false"}]}`,
		"mixed overflow": `{"events":[{"type":"overflow"},{"type":"text","session":"s","text":"must stay hidden"}]}`,
	} {
		t.Run(name, func(t *testing.T) {
			server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
				w.Header().Set("Content-Type", "text/event-stream")
				_, _ = io.WriteString(w, "data: "+batch+"\n\n")
			}))
			defer server.Close()
			conn := NewConnection(ConnectionSnapshot{Port: server.Listener.Addr().(*net.TCPAddr).Port}, nil)
			err := StreamAgents(t.Context(), conn, func([]AgentEvent) error { t.Error("malformed batch partially delivered"); return nil })
			failure, ok := errors.AsType[*StreamError](err)
			if !ok || failure.Kind != StreamProtocol {
				t.Fatalf("malformed batch classified incorrectly: %v", err)
			}
		})
	}
}