/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package http

import (
	"bytes"
	"context"
	"fmt"
	"net"
	"net/http"
	"net/http/httptest"
	"strings"
	"sync"
	"testing"
	"time"
)

import (
	"github.com/stretchr/testify/assert"
)

import (
	clienthttp "github.com/apache/dubbo-go-pixiu/pkg/client/http"
	"github.com/apache/dubbo-go-pixiu/pkg/common/constant"
	"github.com/apache/dubbo-go-pixiu/pkg/common/extension/filter"
	commonmock "github.com/apache/dubbo-go-pixiu/pkg/common/mock"
	"github.com/apache/dubbo-go-pixiu/pkg/common/router/trie"
	contexthttp "github.com/apache/dubbo-go-pixiu/pkg/context/http"
	"github.com/apache/dubbo-go-pixiu/pkg/context/mock"
	"github.com/apache/dubbo-go-pixiu/pkg/logger"
	"github.com/apache/dubbo-go-pixiu/pkg/model"
)

const (
	DEMO = "dgp.filters.http.demo"
	// Kind is the kind of plugin.
	Kind = DEMO
)

var (
	eventCh = make(chan string, 3)
)

type (
	// Plugin is http filter plugin.
	Plugin struct {
	}
	// HeaderFilter is http filter instance
	DemoFilterFactory struct {
		conf *Config
	}
	DemoFilter struct {
		str string
	}
	// Config describe the config of ResponseFilter
	Config struct {
		Foo string `json:"foo,omitempty" yaml:"foo,omitempty"`
		Bar string `json:"bar,omitempty" yaml:"bar,omitempty"`
	}
)

func (p *Plugin) Kind() string {
	return Kind
}

func (p *Plugin) CreateFilterFactory() (filter.HttpFilterFactory, error) {
	return &DemoFilterFactory{conf: &Config{Foo: "default foo", Bar: "default bar"}}, nil
}

func (f *DemoFilter) Decode(ctx *contexthttp.HttpContext) filter.FilterStatus {
	logger.Info("decode phase: ", f.str)

	runes := []rune(f.str)
	for i := 0; i < len(runes)/2; i += 1 {
		runes[i], runes[len(runes)-1-i] = runes[len(runes)-1-i], runes[i]
	}
	f.str = string(runes)

	return filter.Continue
}

func (f *DemoFilter) Encode(ctx *contexthttp.HttpContext) filter.FilterStatus {
	logger.Info("encode phase: ", f.str)
	return filter.Continue
}

func (f *DemoFilterFactory) PrepareFilterChain(ctx *contexthttp.HttpContext, chain filter.FilterChain) error {
	c := f.conf
	str := fmt.Sprintf("%s is drinking in the %s", c.Foo, c.Bar)
	demoFilter := &DemoFilter{str: str}
	chain.AppendDecodeFilters(demoFilter)
	chain.AppendEncodeFilters(demoFilter)
	return nil
}

func (f *DemoFilterFactory) Config() any {
	return f.conf
}

func (f *DemoFilterFactory) Apply() error {
	return nil
}

func TestCreateHttpConnectionManager(t *testing.T) {
	filter.RegisterHttpFilter(&Plugin{})

	hcmc := model.HttpConnectionManagerConfig{
		RouteConfig: model.RouteConfiguration{
			RouteTrie: trie.NewTrieWithDefault("POST/api/v1/**", model.RouteAction{
				Cluster:                     "test_dubbo",
				ClusterNotFoundResponseCode: 505,
			}),
			Dynamic: false,
		},
		HTTPFilters: []*model.HTTPFilter{
			{
				Name:   DEMO,
				Config: nil,
			},
		},
		ServerName:        "test_http_dubbo",
		GenerateRequestID: false,
		IdleTimeoutStr:    "100",
	}

	hcm := CreateHttpConnectionManager(&hcmc)
	assert.Equal(t, len(hcm.filterManager.GetFactory()), 1)
	request, err := http.NewRequest("POST", "http://www.dubbogopixiu.com/api/v1?name=tc", bytes.NewReader([]byte("{\"id\":\"12345\"}")))
	assert.NoError(t, err)
	request.Header = map[string][]string{
		"X-Dgp-Way": {"Dubbo"},
	}
	assert.NoError(t, err)
	c := mock.GetMockHTTPContext(request)
	err = hcm.findRoute(c)
	assert.NoError(t, err)
	err = hcm.Handle(c)
	assert.NoError(t, err)
}

// test SSE case
func TestStreamingResponse(t *testing.T) {
	hcmc := model.HttpConnectionManagerConfig{
		RouteConfig: model.RouteConfiguration{
			RouteTrie: trie.NewTrieWithDefault("GET/api/sse", model.RouteAction{
				Cluster: "mock_stream_cluster",
			}),
		},
		HTTPFilters: []*model.HTTPFilter{
			{
				Name: commonmock.Kind,
			},
		},
	}

	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
	defer cancel()

	// mock server
	upstreamServer, _ := NewTestServerWithURL("localhost:8080", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set(constant.HeaderKeyContextType, constant.HeaderValueTextEventStream)
		w.Header().Set(constant.HeaderKeyCacheControl, constant.HeaderValueNoCache)
		flusher := w.(http.Flusher)

		for i := 1; i <= 3; i++ {
			select {
			case <-ctx.Done():
				return
			default:
				time.Sleep(10 * time.Millisecond)
				event := fmt.Sprintf("data: %d\nevent: %d\nid: %d\n\n", i, i, i)
				_, _ = w.Write([]byte(event))
				flusher.Flush()
				logger.Info("Upstream sent event ", i)
			}
		}
	}))
	defer upstreamServer.Close()

	req := httptest.NewRequest("GET", "http://localhost:8080/api/sse", nil).WithContext(ctx)

	done := make(chan struct{})

	httpCtx := &contexthttp.HttpContext{
		Request: req,
		Writer:  NewStreamRecorder(),
		Ctx:     ctx,
	}
	go func() {
		defer close(done)

		hcm := CreateHttpConnectionManager(&hcmc)

		if err := hcm.Handle(httpCtx); err != nil {
			t.Errorf("Handle failed: %v", err)
		}

		// test targetResp
		if httpCtx.TargetResp == nil {
			t.Error("TargetResp is nil")
			return
		}
	}()

	// event waiting test
	for {
		select {
		case event := <-eventCh:
			logger.Info("Received event: %s", strings.ReplaceAll(event, "\n", "\\n"))
		case <-done:
			receivedEvents := httpCtx.Writer.(*StreamRecorder).GetReceivedBuf()
			assert.Equal(t, 3, len(receivedEvents), "Should receive 3 events")
			return
		case <-time.After(5 * time.Second):
			t.Fatal("Test timeout")
			return
		}
	}
}

// mock recorder
type StreamRecorder struct {
	http.ResponseWriter
	http.Flusher
	mu          sync.Mutex
	receivedBuf []string
	headers     http.Header
	status      int
}

func NewStreamRecorder() *StreamRecorder {
	return &StreamRecorder{
		receivedBuf: make([]string, 0),
		headers:     make(http.Header),
	}
}

func (r *StreamRecorder) Header() http.Header {
	return r.headers
}

func (r *StreamRecorder) WriteHeader(statusCode int) {
	r.status = statusCode
}

func (r *StreamRecorder) Write(data []byte) (int, error) {
	eventCh <- string(data)
	r.mu.Lock()
	defer r.mu.Unlock()
	r.receivedBuf = append(r.receivedBuf, string(data))
	return len(data), nil
}

func (r *StreamRecorder) GetReceivedBuf() []string {
	bufCopy := make([]string, len(r.receivedBuf))
	r.mu.Lock()
	defer r.mu.Unlock()
	copy(bufCopy, r.receivedBuf)
	return bufCopy
}

func (r *StreamRecorder) Flush() {
}

func NewTestServerWithURL(URL string, handler http.Handler) (*httptest.Server, error) {
	ts := httptest.NewUnstartedServer(handler)
	if URL != "" {
		l, err := net.Listen("tcp", URL)
		if err != nil {
			return nil, err
		}
		err = ts.Listener.Close()
		if err != nil {
			return nil, err
		}
		ts.Listener = l
	}
	ts.Start()
	return ts, nil
}

// StreamHTTPRecorder Used to capture and test streaming HTTP responses over channels
type StreamHTTPRecorder struct {
	http.ResponseWriter
	mu          sync.Mutex
	receivedBuf []string
	headers     http.Header
	status      int
	flushCount  int
}

func NewStreamHTTPRecorder() *StreamHTTPRecorder {
	return &StreamHTTPRecorder{
		receivedBuf: make([]string, 0),
		headers:     make(http.Header),
		flushCount:  0,
	}
}

func (r *StreamHTTPRecorder) Header() http.Header {
	return r.headers
}

func (r *StreamHTTPRecorder) WriteHeader(statusCode int) {
	r.status = statusCode
}

func (r *StreamHTTPRecorder) Write(data []byte) (int, error) {
	eventCh <- string(data)
	r.mu.Lock()
	defer r.mu.Unlock()
	r.receivedBuf = append(r.receivedBuf, string(data))
	return len(data), nil
}

func (r *StreamHTTPRecorder) Flush() {
	r.flushCount++
}

func (r *StreamHTTPRecorder) GetReceivedBuf() []string {
	bufCopy := make([]string, len(r.receivedBuf))
	r.mu.Lock()
	defer r.mu.Unlock()
	copy(bufCopy, r.receivedBuf)
	return bufCopy
}

// Test a variety of common streaming HTTP response types
func TestStreamableHTTPResponse(t *testing.T) {
	// define the type of content you want to test
	contentTypes := []string{
		constant.HeaderValueTextPlain,
		constant.HeaderValueApplicationJson,
		constant.HeaderValueApplicationOctetStream,
		constant.HeaderValueApplicationNDJson,
	}

	for _, contentType := range contentTypes {
		t.Run(fmt.Sprintf("ContentType_%s", contentType), func(t *testing.T) {
			testStreamableResponse(t, contentType)
		})
	}
}

func testStreamableResponse(t *testing.T, contentType string) {
	hcmc := model.HttpConnectionManagerConfig{
		RouteConfig: model.RouteConfiguration{
			RouteTrie: trie.NewTrieWithDefault("GET/api/stream", model.RouteAction{
				Cluster: "mock_stream_cluster",
			}),
		},
		HTTPFilters: []*model.HTTPFilter{
			{
				Name: commonmock.Kind,
			},
		},
	}

	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
	defer cancel()

	// Clear any data that may have been left over from the previous test
	for len(eventCh) > 0 {
		<-eventCh
	}

	// mock server
	upstreamServer, _ := NewTestServerWithURL("localhost:8080", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set(constant.HeaderKeyContextType, contentType)
		flusher := w.(http.Flusher)

		// Generate appropriate test data based on content type
		var data []byte
		for i := 1; i <= 5; i++ {
			select {
			case <-ctx.Done():
				return
			default:
				time.Sleep(10 * time.Millisecond)

				switch contentType {
				case constant.HeaderValueApplicationJson:
					data = []byte(fmt.Sprintf(`{"id": %d, "message": "test chunk %d"}\n`, i, i))
				case constant.HeaderValueApplicationNDJson:
					data = []byte(fmt.Sprintf(`{"id": %d, "message": "test chunk %d"}\n`, i, i))
				case constant.HeaderValueApplicationOctetStream:
					data = []byte(fmt.Sprintf("CHUNK-%d", i))
				default: // text/plain
					data = []byte(fmt.Sprintf("Chunk %d\n", i))
				}

				_, _ = w.Write(data)
				flusher.Flush()
				logger.Info("Upstream sent chunk ", i)
			}
		}
	}))
	defer upstreamServer.Close()

	req := httptest.NewRequest("GET", "http://localhost:8080/api/stream", nil).WithContext(ctx)
	done := make(chan struct{})

	httpCtx := &contexthttp.HttpContext{
		Request: req,
		Writer:  NewStreamHTTPRecorder(),
		Ctx:     ctx,
	}

	go func() {
		defer close(done)

		hcm := CreateHttpConnectionManager(&hcmc)

		if err := hcm.Handle(httpCtx); err != nil {
			t.Errorf("Handle failed: %v", err)
		}

		// verify that targetResp exists
		if httpCtx.TargetResp == nil {
			t.Error("TargetResp is nil")
			return
		}
	}()

	// collect and validate responses
	receivedChunks := 0
	for {
		select {
		case event := <-eventCh:
			logger.Info("Received chunk: %s", event)
			receivedChunks++
		case <-done:
			receivedEvents := httpCtx.Writer.(*StreamHTTPRecorder).GetReceivedBuf()
			assert.Equal(t, 5, len(receivedEvents), "Should receive 5 chunks")
			return
		case <-time.After(5 * time.Second):
			t.Fatal("Test timeout")
			return
		}
	}
}

// TestIsStreamableResponse Test whether it is a function that can be streamed and responded
func TestIsStreamableResponse(t *testing.T) {
	tests := []struct {
		name     string
		headers  map[string]string
		expected bool
	}{
		{
			name: "sseResponse",
			headers: map[string]string{
				constant.HeaderKeyContextType: constant.HeaderValueTextEventStream,
			},
			expected: true,
		},
		{
			name: "chunkedEncodingResponses",
			headers: map[string]string{
				constant.HeaderKeyContextType:      constant.HeaderValueApplicationJson,
				constant.HeaderKeyTransferEncoding: constant.HeaderValueChunked,
			},
			expected: true,
		},
		{
			name: "JsonResponseWithoutContent-Length",
			headers: map[string]string{
				constant.HeaderKeyContextType: constant.HeaderValueApplicationJson,
			},
			expected: true,
		},
		{
			name: "The text response is large Content-Length",
			headers: map[string]string{
				constant.HeaderKeyContextType:   constant.HeaderValueTextPlain,
				constant.HeaderKeyContentLength: "2097152", // 2MB
			},
			expected: true,
		},
		{
			name: "JSON response Content-Length",
			headers: map[string]string{
				constant.HeaderKeyContextType:   constant.HeaderValueApplicationJson,
				constant.HeaderKeyContentLength: "1024", // 1KB
			},
			expected: false,
		},
		{
			name: "no stream Content-Type",
			headers: map[string]string{
				constant.HeaderKeyContextType: constant.HeaderValueImageJpeg,
			},
			expected: false,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			resp := &http.Response{
				Header: make(http.Header),
			}
			for k, v := range tt.headers {
				resp.Header.Set(k, v)
			}
			result := clienthttp.IsStreamableResponse(resp)
			assert.Equal(t, tt.expected, result, "IsStreamableResponse() return err")
		})
	}
}
