package tushare_test

import (
	"context"
	"crypto/tls"
	"encoding/json"
	"io"
	"net/http"
	"net/http/httptest"
	"testing"

	"github.com/stretchr/testify/require"

	"github.com/tidbcloud/AgentTeam/internal/extension/providers/tushare"
	"github.com/tidbcloud/AgentTeam/internal/extension/registry"
	"github.com/tidbcloud/AgentTeam/internal/extension/runner"
	"github.com/tidbcloud/AgentTeam/internal/service/extensionflow"
)

// wireServer records every request and answers with the scripted status and
// body for every call.
type wireServer struct {
	server   *httptest.Server
	requests []wireRequest
	status   int
	body     string
}

type wireRequest struct {
	method string
	path   string
	query  string
	body   string
	header http.Header
}

func newWireServer(t *testing.T, status int, body string) *wireServer {
	t.Helper()
	wire := &wireServer{status: status, body: body}
	wire.server = httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		bodyBytes, err := io.ReadAll(r.Body)
		require.NoError(t, err)
		wire.requests = append(wire.requests, wireRequest{
			method: r.Method,
			path:   r.URL.Path,
			query:  r.URL.RawQuery,
			body:   string(bodyBytes),
			header: r.Header.Clone(),
		})
		w.Header().Set("Content-Type", "application/json")
		w.WriteHeader(status)
		_, _ = w.Write([]byte(body))
	}))
	t.Cleanup(wire.server.Close)
	return wire
}

func (wire *wireServer) last() wireRequest {
	if len(wire.requests) == 0 {
		panic("no recorded request")
	}
	return wire.requests[len(wire.requests)-1]
}

type staticCredentialSource struct {
	credentials runner.Credentials
}

func (source staticCredentialSource) ResolveProviderCredentials(
	context.Context, string, string,
) (runner.Credentials, error) {
	return source.credentials, nil
}

func insecureTransport() *http.Transport {
	return &http.Transport{TLSClientConfig: &tls.Config{InsecureSkipVerify: true}}
}

func invoke(t *testing.T, server *wireServer, action string, input string) (json.RawMessage, error) {
	t.Helper()
	client, err := runner.NewRESTClient(runner.RESTOptions{
		BaseURL:          server.server.URL,
		Auth:             tushare.AuthSpec(),
		StatusClassifier: tushare.ClassifyStatus,
		Transport:        insecureTransport(),
	})
	require.NoError(t, err)
	executor := runner.NewRESTExecutor(
		tushare.Domain,
		client,
		staticCredentialSource{credentials: runner.Credentials{tushare.APIKeyField: "tok"}},
		tushare.Handlers(),
	)
	result, err := executor.Invoke(context.Background(), registry.BuiltinInvocation{
		WorkspaceID:           "ws",
		ServiceInstallationID: "inst",
		ExtensionName:         tushare.Domain,
		Action:                action,
		Input:                 json.RawMessage(input),
	})
	if err != nil {
		return nil, err
	}
	return result.Data, nil
}

// classifiedCode asserts the Agent-visible closed error code the executor
// boundary produces.
func classifiedCode(t *testing.T, err error) extensionflow.InvocationErrorCode {
	t.Helper()
	code, ok := extensionflow.InvocationErrorCodeOf(err)
	require.True(t, ok, "error was not classified: %v", err)
	return code
}

func successBody(fields string, items string) string {
	return `{"code":0,"msg":null,"request_id":"req-1","data":{"fields":` + fields + `,"items":` + items + `}}`
}

func decodeBody(t *testing.T, raw string) map[string]any {
	t.Helper()
	var body map[string]any
	require.NoError(t, json.Unmarshal([]byte(raw), &body))
	return body
}

func TestWireListStocksProjectsCamelCaseRows(t *testing.T) {
	wire := newWireServer(t, http.StatusOK, successBody(
		`["ts_code","symbol","name","area","industry","market","exchange","list_status","list_date","delist_date","is_hs"]`,
		`[["000001.SZ","000001","平安银行","深圳","银行","主板","SZSE","L","19910403",null,"S"]]`))
	output, err := invoke(t, wire, "list-stocks", `{"exchange":"SZSE","listStatus":"L"}`)
	require.NoError(t, err)

	body := decodeBody(t, wire.last().body)
	require.Equal(t, "stock_basic", body["api_name"])
	require.Equal(t, map[string]any{"exchange": "SZSE", "list_status": "L"}, body["params"])
	require.Equal(t, "ts_code,symbol,name,area,industry,market,exchange,list_status,list_date,delist_date,is_hs", body["fields"])

	var result map[string]any
	require.NoError(t, json.Unmarshal(output, &result))
	stocks := result["stocks"].([]any)
	require.Len(t, stocks, 1)
	stock := stocks[0].(map[string]any)
	require.Equal(t, "000001.SZ", stock["tsCode"])
	require.Equal(t, "平安银行", stock["name"])
	require.Equal(t, "19910403", stock["listDate"])
	require.Nil(t, stock["delistDate"])
	require.Equal(t, "S", stock["isHs"])
}

// The trade-calendar contract requires startDate and endDate at the schema
// layer (upstream declares them required); the executor-level wire harness
// asserts the catalog carries the requirement.
func TestWireTradeCalendarRequiresDates(t *testing.T) {
	for _, action := range tushare.Definition().Actions {
		if action.Name != "get-trade-calendar" {
			continue
		}
		require.Equal(t, []any{"startDate", "endDate"}, action.InputSchema["required"])
	}
}

func TestWireTradeCalendarSendsParams(t *testing.T) {
	wire := newWireServer(t, http.StatusOK,
		successBody(`["exchange","cal_date","is_open","pretrade_date"]`,
			`[["SSE","20180903",1,"20180903"]]`))
	output, err := invoke(t, wire, "get-trade-calendar",
		`{"startDate":"20180901","endDate":"20180902","isOpen":1}`)
	require.NoError(t, err)
	body := decodeBody(t, wire.last().body)
	require.Equal(t, "trade_cal", body["api_name"])
	require.Empty(t, wire.last().header.Get("Authorization"),
		"Tushare token must travel only in the request body")
	require.Equal(t, map[string]any{
		"start_date": "20180901", "end_date": "20180902", "is_open": float64(1),
	}, body["params"])

	var result map[string]any
	require.NoError(t, json.Unmarshal(output, &result))
	row := result["calendar"].([]any)[0].(map[string]any)
	require.Equal(t, "20180903", row["calDate"])
	require.Equal(t, float64(1), row["isOpen"])
}

func TestWireTypedIntegerLexicalsFailClosed(t *testing.T) {
	wire := newWireServer(t, http.StatusOK,
		successBody(`["exchange","cal_date","is_open"]`, `[]`))
	for _, lexical := range []string{"1e1", "1.0", "1.5"} {
		_, err := invoke(t, wire, "get-trade-calendar",
			`{"startDate":"20180901","endDate":"20180902","isOpen":`+lexical+`}`)
		require.Error(t, err, lexical)
		require.Equal(t, extensionflow.InvocationValidation, classifiedCode(t, err), lexical)
	}
	require.Empty(t, wire.requests,
		"non-plain integer lexicals must fail before the Tushare POST")
}

func TestWireDatedActionsRequireBound(t *testing.T) {
	wire := newWireServer(t, http.StatusOK, successBody(`[]`, `[]`))
	for _, action := range []string{"get-daily-quotes", "get-daily-basic", "get-adjustment-factors"} {
		_, err := invoke(t, wire, action, `{"startDate":"20180901"}`)
		require.Error(t, err, action)
		require.Equal(t, extensionflow.InvocationValidation, classifiedCode(t, err), action)
	}
	require.Empty(t, wire.requests)
}

func TestWireDailyQuotesProjectsNullableNumbers(t *testing.T) {
	wire := newWireServer(t, http.StatusOK, successBody(
		`["ts_code","trade_date","open","high","low","close","pre_close","change","pct_chg","vol","amount"]`,
		`[["000001.SZ","20180903",9.19,9.25,9.1,9.15,9.19,-0.04,-0.44,1316069.35,1203679.035]]`))
	output, err := invoke(t, wire, "get-daily-quotes", `{"tsCode":"000001.SZ"}`)
	require.NoError(t, err)
	body := decodeBody(t, wire.last().body)
	require.Equal(t, "daily", body["api_name"])
	require.Equal(t, map[string]any{"ts_code": "000001.SZ"}, body["params"])

	var result map[string]any
	require.NoError(t, json.Unmarshal(output, &result))
	row := result["quotes"].([]any)[0].(map[string]any)
	require.Equal(t, "000001.SZ", row["tsCode"])
	require.Equal(t, 9.19, row["open"])
	require.Equal(t, -0.44, row["pctChg"])
}

func TestWireShareholderTradesValidation(t *testing.T) {
	wire := newWireServer(t, http.StatusOK, successBody(`[]`, `[]`))
	_, err := invoke(t, wire, "get-shareholder-trades", `{"startDate":"20180901"}`)
	require.Error(t, err)
	require.Equal(t, extensionflow.InvocationValidation, classifiedCode(t, err))
	_, err = invoke(t, wire, "get-shareholder-trades", `{}`)
	require.Error(t, err)
	require.Equal(t, extensionflow.InvocationValidation, classifiedCode(t, err))
	require.Empty(t, wire.requests)
}

func TestWireShareholderTradesMapsParamsAndRows(t *testing.T) {
	wire := newWireServer(t, http.StatusOK, successBody(
		`["ts_code","ann_date","holder_name","holder_type","in_de","change_vol","change_ratio","after_share","after_ratio","avg_price","total_share","begin_date","close_date"]`,
		`[["000001.SZ","20180901","某股东","C","IN",1000000.0,0.5,2000000.0,1.0,10.5,2000000.0,"20180801","20180901"]]`))
	output, err := invoke(t, wire, "get-shareholder-trades",
		`{"tsCode":"000001.SZ","tradeType":"IN","holderType":"C"}`)
	require.NoError(t, err)
	body := decodeBody(t, wire.last().body)
	require.Equal(t, "stk_holdertrade", body["api_name"])
	require.Equal(t, map[string]any{
		"ts_code": "000001.SZ", "trade_type": "IN", "holder_type": "C",
	}, body["params"])

	var result map[string]any
	require.NoError(t, json.Unmarshal(output, &result))
	row := result["shareholderTrades"].([]any)[0].(map[string]any)
	require.Equal(t, "20180901", row["announcementDate"])
	require.Equal(t, "IN", row["tradeType"])
	require.Equal(t, 0.5, row["changeRatio"])
}

func TestWireInputStringsAreTrimmed(t *testing.T) {
	wire := newWireServer(t, http.StatusOK, successBody(`[]`, `[]`))

	_, err := invoke(t, wire, "list-stocks",
		`{"exchange":" SZSE ","listStatus":" L ","isHs":" S "}`)
	require.NoError(t, err)
	require.Equal(t, map[string]any{
		"exchange": "SZSE", "list_status": "L", "is_hs": "S",
	}, decodeBody(t, wire.last().body)["params"])

	_, err = invoke(t, wire, "get-daily-quotes",
		`{"tsCode":" 000001.SZ ","tradeDate":" 20180903 ","startDate":" 20180901 ","endDate":" 20180902 "}`)
	require.NoError(t, err)
	require.Equal(t, map[string]any{
		"ts_code": "000001.SZ", "trade_date": "20180903",
		"start_date": "20180901", "end_date": "20180902",
	}, decodeBody(t, wire.last().body)["params"])

	_, err = invoke(t, wire, "get-shareholder-trades",
		`{"tsCode":" 000001.SZ ","announcementDate":" 20180903 ","startDate":" 20180901 ","endDate":" 20180902 ","tradeType":" IN ","holderType":" C "}`)
	require.NoError(t, err)
	require.Equal(t, map[string]any{
		"ts_code": "000001.SZ", "ann_date": "20180903",
		"start_date": "20180901", "end_date": "20180902",
		"trade_type": "IN", "holder_type": "C",
	}, decodeBody(t, wire.last().body)["params"])
}

func TestWireWhitespaceOnlyBoundsFailBeforeRequest(t *testing.T) {
	wire := newWireServer(t, http.StatusOK, successBody(`[]`, `[]`))
	for _, testCase := range []struct {
		action string
		input  string
	}{
		{
			action: "get-trade-calendar",
			input:  `{"startDate":"   ","endDate":" 20180902 "}`,
		},
		{action: "get-daily-quotes", input: `{"tsCode":"   "}`},
		{action: "get-shareholder-trades", input: `{"announcementDate":"   ","tradeType":" IN "}`},
	} {
		_, err := invoke(t, wire, testCase.action, testCase.input)
		require.Error(t, err, testCase.action)
		require.Equal(t, extensionflow.InvocationValidation, classifiedCode(t, err), testCase.action)
	}
	require.Empty(t, wire.requests, "whitespace-only bounds must fail before the Tushare POST")
}

func TestWireBusinessCodeClassification(t *testing.T) {
	cases := []struct {
		code    int
		message string
		want    extensionflow.InvocationErrorCode
	}{
		{40101, "token无效", extensionflow.InvocationProviderAuthUnavailable},
		{2002, "抱歉，您没有访问该接口的权限", extensionflow.InvocationProviderAuthUnavailable},
		{2003, "每小时最多访问该接口1次", extensionflow.InvocationRateLimited},
		{2004, "每天最多访问该接口10次", extensionflow.InvocationRateLimited},
		{9999, "其他错误", extensionflow.InvocationUpstreamFailure},
	}
	for _, testCase := range cases {
		body, err := json.Marshal(map[string]any{
			"code": testCase.code, "msg": testCase.message,
			"data": map[string]any{"fields": []string{}, "items": []any{}},
		})
		require.NoError(t, err)
		wire := newWireServer(t, http.StatusOK, string(body))
		_, err = invoke(t, wire, "list-stocks", `{}`)
		require.Error(t, err, testCase.message)
		require.Equal(t, testCase.want, classifiedCode(t, err), testCase.message)
	}
}

func TestWireHTTPStatusClassification(t *testing.T) {
	for _, status := range []int{http.StatusUnauthorized, http.StatusForbidden} {
		wire := newWireServer(t, status, `{"msg":"denied"}`)
		_, err := invoke(t, wire, "list-stocks", `{}`)
		require.Error(t, err)
		require.Equal(t, extensionflow.InvocationProviderAuthUnavailable, classifiedCode(t, err), "status %d", status)
	}
	wire := newWireServer(t, 429, `{"msg":"too many"}`)
	_, err := invoke(t, wire, "list-stocks", `{}`)
	require.Error(t, err)
	require.Equal(t, extensionflow.InvocationRateLimited, classifiedCode(t, err))
	wire = newWireServer(t, 500, `{"msg":"boom"}`)
	_, err = invoke(t, wire, "list-stocks", `{}`)
	require.Error(t, err)
	require.Equal(t, extensionflow.InvocationUpstreamFailure, classifiedCode(t, err))
}

func TestWireMalformedTableFailsClosed(t *testing.T) {
	for _, body := range []string{
		`{}`,
		`{"code":0}`,
		`{"code":0,"data":{"fields":"not-an-array","items":[]}}`,
		`{"code":0,"data":{"fields":["a"],"items":"not-an-array"}}`,
		`{"code":0,"data":{"fields":["a"],"items":["not-an-array"]}}`,
	} {
		wire := newWireServer(t, http.StatusOK, body)
		_, err := invoke(t, wire, "list-stocks", `{}`)
		require.Error(t, err, body)
		require.Equal(t, extensionflow.InvocationUpstreamFailure, classifiedCode(t, err), body)
	}
}

// The credential probe posts the upstream trade_cal validation request with
// the token in the body and rejects a credential whose probe returns no
// table fields.
func TestWireValidateCredentialsProbe(t *testing.T) {
	wire := newWireServer(t, http.StatusOK,
		successBody(`["exchange","cal_date","is_open"]`, `[["SSE","20180903",1]]`))
	client, err := runner.NewRESTClient(runner.RESTOptions{
		BaseURL:          wire.server.URL,
		Auth:             tushare.AuthSpec(),
		StatusClassifier: tushare.ClassifyStatus,
		Transport:        insecureTransport(),
	})
	require.NoError(t, err)
	require.NoError(t, tushare.ValidateCredentialsWithClient(
		context.Background(), client, runner.Credentials{tushare.APIKeyField: "tok"}))

	request := wire.last()
	require.Equal(t, "POST", request.method)
	require.Empty(t, request.header.Get("Authorization"),
		"Tushare probe token must travel only in the request body")
	body := decodeBody(t, request.body)
	require.Equal(t, "trade_cal", body["api_name"])
	require.Equal(t, "tok", body["token"])
	require.Equal(t, map[string]any{
		"exchange": "", "start_date": "20180901", "end_date": "20180902",
	}, body["params"])

	empty := newWireServer(t, http.StatusOK, successBody(`[]`, `[]`))
	emptyClient, err := runner.NewRESTClient(runner.RESTOptions{
		BaseURL:          empty.server.URL,
		Auth:             tushare.AuthSpec(),
		StatusClassifier: tushare.ClassifyStatus,
		Transport:        insecureTransport(),
	})
	require.NoError(t, err)
	err = tushare.ValidateCredentialsWithClient(
		context.Background(), emptyClient, runner.Credentials{tushare.APIKeyField: "tok"})
	require.Error(t, err)
	// ValidateCredentials returns the raw runtime error (the connection flow,
	// not the executor boundary, wraps it), so assert the runtime code.
	require.Equal(t, runner.ErrorProviderAuthUnavailable, runner.CodeOf(err))
}