kenn-io--agentsview/internal/server/huma_routes_push_internal_test.go

package server

import (
	"bytes"
	"context"
	"encoding/json"
	"log"
	"net/http"
	"net/http/httptest"
	"os"
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	"go.kenn.io/agentsview/internal/config"
	duckdbsync "go.kenn.io/agentsview/internal/duckdb"
	"go.kenn.io/agentsview/internal/postgres"
)

// stubVectorPushSource is a no-op postgres.VectorPushSource: the gating test
// only needs identity, never a method call.
type stubVectorPushSource struct{}

func (stubVectorPushSource) Generation(
	context.Context,
) (postgres.VectorGenerationInfo, bool, error) {
	return postgres.VectorGenerationInfo{}, false, nil
}

func (stubVectorPushSource) SessionDocHashes(
	context.Context,
) (map[string]string, error) {
	return nil, nil
}

func (stubVectorPushSource) SessionDocs(
	context.Context, string,
) ([]postgres.VectorPushDoc, string, error) {
	return nil, "", nil
}

// TestPGPushProgressLoggerThrottlesAndReportsPhases pins the daemon-side
// push heartbeat: reports inside the throttle window log nothing, and the
// vector phase logs its own counters instead of the session line.
func TestPGPushProgressLoggerThrottlesAndReportsPhases(t *testing.T) {
	origInterval := pushProgressLogInterval
	pushProgressLogInterval = time.Hour
	t.Cleanup(func() { pushProgressLogInterval = origInterval })

	var buf bytes.Buffer
	origOut := log.Writer()
	log.SetOutput(&buf)
	t.Cleanup(func() { log.SetOutput(origOut) })

	logProgress := newPGPushProgressLogger()
	logProgress(postgres.PushProgress{SessionsDone: 1, SessionsTotal: 10, MessagesDone: 5})
	logProgress(postgres.PushProgress{SessionsDone: 2, SessionsTotal: 10, MessagesDone: 9})
	assert.Contains(t, buf.String(), "pg push: 1/10 session(s), 5 messages")
	assert.NotContains(t, buf.String(), "2/10",
		"second report inside the throttle window must not log")

	pushProgressLogInterval = 0
	logProgress(postgres.PushProgress{
		Phase:               "vectors",
		VectorSessionsDone:  3,
		VectorSessionsTotal: 7,
		VectorChunksPushed:  42,
	})
	assert.Contains(t, buf.String(),
		"pg push: vectors 3/7 session(s) scanned, 42 chunks")

	logProgress(postgres.PushProgress{
		Phase:         "preparing",
		SessionsDone:  500,
		SessionsTotal: 46000,
	})
	assert.Contains(t, buf.String(), "pg push: preparing 500/46000 session(s)")

	logProgress(postgres.PushProgress{Phase: "preparing"})
	assert.Contains(t, buf.String(),
		"pg push: preparing (sync state, metadata, fingerprints)",
		"zero-total preparing report renders the setup-stage line")
}

func TestPGPushVectorSourceGating(t *testing.T) {
	disabled := false
	tests := []struct {
		name      string
		wired     bool
		pushFlag  *bool
		noVectors bool
		wantSrc   bool
	}{
		{name: "wired and enabled", wired: true, wantSrc: true},
		{name: "no source wired", wired: false, wantSrc: false},
		{name: "target opts out", wired: true, pushFlag: &disabled, wantSrc: false},
		{name: "caller passed --no-vectors", wired: true, noVectors: true, wantSrc: false},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			s := &Server{}
			if tt.wired {
				s.vectorPushSource = stubVectorPushSource{}
			}
			got := s.pgPushVectorSource(
				config.PGConfig{PushVectors: tt.pushFlag}, tt.noVectors,
			)
			if tt.wantSrc {
				assert.NotNil(t, got)
			} else {
				assert.Nil(t, got)
			}
		})
	}
}

type openAPISpec struct {
	Paths map[string]map[string]openAPIOperation `json:"paths"`
}

type openAPIOperation struct {
	Parameters []openAPIParameter         `json:"parameters"`
	Responses  map[string]openAPIResponse `json:"responses"`
}

type openAPIParameter struct {
	In       string `json:"in"`
	Name     string `json:"name"`
	Required bool   `json:"required"`
}

type openAPIResponse struct {
	Content map[string]any `json:"content"`
}

func missingEnvRef(t testing.TB, name string) string {
	t.Helper()
	require.NoError(t, os.Unsetenv(name))
	return "${" + name + "}"
}

func testServerWithConfig(cfg config.Config) *Server {
	return &Server{cfg: cfg}
}

func readOpenAPISpec(t testing.TB, h http.Handler) openAPISpec {
	t.Helper()
	req := httptest.NewRequest(http.MethodGet, "/api/openapi.json", nil)
	req.Host = "127.0.0.1:0"
	w := httptest.NewRecorder()
	h.ServeHTTP(w, req)

	require.Equal(t, http.StatusOK, w.Code, "body: %s", w.Body.String())
	var spec openAPISpec
	require.NoError(t, json.Unmarshal(w.Body.Bytes(), &spec))
	return spec
}

func requireOpenAPIOperation(
	t testing.TB,
	spec openAPISpec,
	method string,
	path string,
) openAPIOperation {
	t.Helper()
	require.Contains(t, spec.Paths, path)
	require.Contains(t, spec.Paths[path], method)
	return spec.Paths[path][method]
}

func assertStreamingResponseContent(
	t testing.TB,
	content map[string]any,
) {
	t.Helper()
	assert.Contains(t, content, "text/event-stream")
	assert.Contains(t, content, "application/json")
}

func TestPGPushConfigRequestOverrideSkipsDaemonEnvResolution(t *testing.T) {
	const envName = "AGENTSVIEW_TEST_MISSING_PG_URL_25053"
	s := testServerWithConfig(config.Config{
		PG: config.PGConfig{URL: missingEnvRef(t, envName)},
	})
	req := daemonPushRequest{
		PG: &config.PGConfig{
			URL:         "postgres://user:pass@host/db",
			Schema:      "mirror",
			MachineName: "laptop",
		},
	}

	got, err := s.pgPushConfig(req)
	require.NoError(t, err)
	assert.Equal(t, "postgres://user:pass@host/db", got.URL)
	assert.Equal(t, "mirror", got.Schema)
	assert.Equal(t, "laptop", got.MachineName)
}

func TestPGPushRejectsIncludeAndExcludeProjects(t *testing.T) {
	s := testServerWithConfig(config.Config{})

	_, err := s.humaPGPush(context.Background(), &daemonPushInput{
		Body: daemonPushRequest{
			Projects:        []string{"alpha"},
			ExcludeProjects: []string{"beta"},
		},
	})
	require.Error(t, err)

	var statusErr interface{ GetStatus() int }
	require.ErrorAs(t, err, &statusErr)
	assert.Equal(t, http.StatusBadRequest, statusErr.GetStatus())
	assert.Contains(t, err.Error(),
		"projects and exclude_projects are mutually exclusive")
}

func TestDuckDBPushRejectsIncludeAndExcludeProjects(t *testing.T) {
	s := testServerWithConfig(config.Config{})

	_, err := s.humaDuckDBPush(context.Background(), &daemonPushInput{
		Body: daemonPushRequest{
			Projects:        []string{"alpha"},
			ExcludeProjects: []string{"beta"},
		},
	})
	require.Error(t, err)

	var statusErr interface{ GetStatus() int }
	require.ErrorAs(t, err, &statusErr)
	assert.Equal(t, http.StatusBadRequest, statusErr.GetStatus())
	assert.Contains(t, err.Error(),
		"projects and exclude_projects are mutually exclusive")
}

func TestDuckDBPushRejectsMissingQuackTokenAsBadRequest(t *testing.T) {
	s := testServer(t, 30)

	_, err := s.humaDuckDBPush(context.Background(), &daemonPushInput{
		Body: daemonPushRequest{
			DuckDB: &config.DuckDBConfig{
				URL:         "quack:https://duck.example.test",
				MachineName: "workstation",
			},
		},
	})
	require.Error(t, err)

	var statusErr interface{ GetStatus() int }
	require.ErrorAs(t, err, &statusErr)
	assert.Equal(t, http.StatusBadRequest, statusErr.GetStatus())
	assert.Contains(t, err.Error(), "duckdb quack token is required")
}

func TestDuckDBPushConfigRequestOverrideSkipsDaemonEnvResolution(t *testing.T) {
	const envName = "AGENTSVIEW_TEST_MISSING_DUCKDB_PATH_25053"
	s := testServerWithConfig(config.Config{
		DuckDB: config.DuckDBConfig{Path: missingEnvRef(t, envName)},
	})
	req := daemonPushRequest{
		DuckDB: &config.DuckDBConfig{
			Path:        "/tmp/agentsview.duckdb",
			MachineName: "workstation",
		},
	}

	got, err := s.duckDBPushConfig(req)
	require.NoError(t, err)
	assert.Equal(t, "/tmp/agentsview.duckdb", got.Path)
	assert.Equal(t, "workstation", got.MachineName)
}

func TestDuckDBPushSyncOptionsDerivesRemoteTargetScope(t *testing.T) {
	duckCfg := config.DuckDBConfig{
		URL:   "quack:https://duck.example.test?token=secret",
		Token: "other-secret",
	}

	got := duckDBPushSyncOptions(daemonPushRequest{
		Projects: []string{"alpha"},
	}, duckCfg)

	assert.Equal(t, []string{"alpha"}, got.Projects)
	assert.Equal(t, duckdbsync.SyncStateTargetForConfig(duckCfg), got.SyncStateTarget)
	assert.NotContains(t, got.SyncStateTarget, "secret")
}

func TestSyncRemotesRouteIsStreaming(t *testing.T) {
	s := testServer(t, 30)
	spec := readOpenAPISpec(t, s.Handler())
	op := requireOpenAPIOperation(t, spec, "post", "/api/v1/sync/remotes")
	require.Contains(t, op.Responses, "200")
	assertStreamingResponseContent(t, op.Responses["200"].Content)
}

// TestPushRoutesAreStreaming pins that both push routes negotiate SSE (the
// CLI's daemon-delegated push renders the streamed progress) while still
// declaring a plain JSON response for non-streaming clients.
func TestPushRoutesAreStreaming(t *testing.T) {
	s := testServer(t, 30)
	spec := readOpenAPISpec(t, s.Handler())
	for _, path := range []string{"/api/v1/push/pg", "/api/v1/push/duckdb"} {
		op := requireOpenAPIOperation(t, spec, "post", path)
		require.Contains(t, op.Responses, "200", path)
		assertStreamingResponseContent(t, op.Responses["200"].Content)
	}
}

// TestNewPushProgressStreamSenderThrottles pins the SSE fan-out throttle: the
// session loop reports per session, and forwarding every report would emit
// one SSE event per row.
func TestNewPushProgressStreamSenderThrottles(t *testing.T) {
	origInterval := pushProgressStreamInterval
	pushProgressStreamInterval = time.Hour
	t.Cleanup(func() { pushProgressStreamInterval = origInterval })

	var got []int
	send := newPushProgressStreamSender(func(v int) { got = append(got, v) })
	send(1)
	send(2)
	assert.Equal(t, []int{1}, got,
		"second report inside the throttle window must not send")

	pushProgressStreamInterval = 0
	send(3)
	assert.Equal(t, []int{1, 3}, got)
}