// transcription_test.go drives the full Transcribe chain against fake OSS +
// fake DashScope servers (httptest, ZERO real cloud calls): upload signature
// shape, async header, channel_id parameterization, PENDING->SUCCEEDED
// polling, channel-to-speaker mapping with ms->s conversion, interleaved
// merge ordering, FAILED tasks, validation errors, and the best-effort OSS
// delete.
//
// transcription_test.go 用 fake OSS + fake DashScope (httptest, 零真云调用)
// 驱动完整 Transcribe 链: 上传签名形状 / 异步头 / channel_id 参数 / 轮询 /
// 声道映射毫秒转秒 / 交叉合并排序 / FAILED / 入参校验 / 尽力删除.

package aliyunasr

import (
	"context"
	"encoding/json"
	"io"
	"net/http"
	"net/http/httptest"
	"strings"
	"sync/atomic"
	"testing"
	"time"

	"git.flytoex.net/yuanwei/flyto-agent/core/pkg/flyto"
)

// fakeCloud bundles one httptest server playing BOTH roles (OSS bucket +
// DashScope API + result document host), recording what it saw.
//
// fakeCloud 用一个 httptest 服务器同时扮演 OSS 桶 + DashScope API + 结果文档
// 宿主, 记录收到的一切.
type fakeCloud struct {
	srv *httptest.Server

	putBody    []byte
	putAuth    string
	putDeleted atomic.Bool

	submitBody   []byte
	submitAsync  string
	submitAuth   string
	pollCount    atomic.Int32
	pendingPolls int32 // how many polls answer PENDING before SUCCEEDED. ZH: 前几次轮询答 PENDING
	failTask     bool

	resultDoc string
}

func newFakeCloud(t *testing.T) *fakeCloud {
	t.Helper()
	fc := &fakeCloud{pendingPolls: 1}
	mux := http.NewServeMux()

	// OSS: PUT/DELETE /transcribe/... (object keys carry the prefix).
	mux.HandleFunc("/transcribe/", func(w http.ResponseWriter, r *http.Request) {
		switch r.Method {
		case http.MethodPut:
			fc.putBody, _ = io.ReadAll(r.Body)
			fc.putAuth = r.Header.Get("Authorization")
			w.WriteHeader(http.StatusOK)
		case http.MethodDelete:
			fc.putDeleted.Store(true)
			w.WriteHeader(http.StatusNoContent)
		case http.MethodGet: // DashScope would fetch the signed URL; not exercised
			w.WriteHeader(http.StatusOK)
		default:
			w.WriteHeader(http.StatusMethodNotAllowed)
		}
	})

	// DashScope: submit + poll + result document.
	mux.HandleFunc("/api/v1/services/audio/asr/transcription", func(w http.ResponseWriter, r *http.Request) {
		fc.submitBody, _ = io.ReadAll(r.Body)
		fc.submitAsync = r.Header.Get("X-DashScope-Async")
		fc.submitAuth = r.Header.Get("Authorization")
		_ = json.NewEncoder(w).Encode(map[string]any{
			"output": map[string]any{"task_id": "task-123", "task_status": "PENDING"},
		})
	})
	mux.HandleFunc("/api/v1/tasks/task-123", func(w http.ResponseWriter, r *http.Request) {
		n := fc.pollCount.Add(1)
		if fc.failTask {
			_ = json.NewEncoder(w).Encode(map[string]any{
				"output": map[string]any{"task_status": "FAILED", "message": "file format not supported"},
			})
			return
		}
		if n <= fc.pendingPolls {
			_ = json.NewEncoder(w).Encode(map[string]any{
				"output": map[string]any{"task_status": "RUNNING"},
			})
			return
		}
		_ = json.NewEncoder(w).Encode(map[string]any{
			"output": map[string]any{
				"task_status": "SUCCEEDED",
				"results": []map[string]any{
					{"transcription_url": fc.srv.URL + "/result.json", "subtask_status": "SUCCEEDED"},
				},
			},
		})
	})
	mux.HandleFunc("/result.json", func(w http.ResponseWriter, r *http.Request) {
		_, _ = w.Write([]byte(fc.resultDoc))
	})

	fc.srv = httptest.NewServer(mux)
	t.Cleanup(fc.srv.Close)
	return fc
}

func (fc *fakeCloud) provider() *Provider {
	return New(Config{
		APIKey:  "sk-test",
		BaseURL: fc.srv.URL + "/api/v1",
		OSS: OSSConfig{
			Endpoint:        fc.srv.URL,
			Bucket:          "flytocall-audio",
			AccessKeyID:     "LTAItest",
			AccessKeySecret: "secret",
		},
		pollOverride: 5 * time.Millisecond,
	})
}

// twoChannelDoc is a canned stereo result: channel 0 and 1 sentences that
// must interleave by begin_time. 双声道结果样本, 须按 begin_time 交叉.
const twoChannelDoc = `{
  "properties": {"original_duration_in_milliseconds": 187000},
  "transcripts": [
    {"channel_id": 0, "text": "你好请问库存 好的明天答复", "sentences": [
      {"begin_time": 0, "end_time": 2100, "text": "你好请问库存"},
      {"begin_time": 9000, "end_time": 11500, "text": "好的明天答复"}
    ]},
    {"channel_id": 1, "text": "我查一下 现在缺货", "sentences": [
      {"begin_time": 3000, "end_time": 5000, "text": "我查一下"},
      {"begin_time": 6000, "end_time": 8000, "text": "现在缺货"}
    ]}
  ]
}`

func TestTranscribeFullChain(t *testing.T) {
	fc := newFakeCloud(t)
	fc.resultDoc = twoChannelDoc
	p := fc.provider()

	resp, err := p.Transcribe(context.Background(), &flyto.TranscriptionRequest{
		Audio:        strings.NewReader("RIFF-fake-wav-bytes"),
		Filename:     "call.wav",
		Model:        ModelParaformer8kV2,
		LeftSpeaker:  "客户",
		RightSpeaker: "坐席",
	})
	if err != nil {
		t.Fatalf("Transcribe: %v", err)
	}

	// Upload: body bytes reached OSS verbatim, V1 header signature shape.
	// 上传: 音频逐字节到 OSS, V1 头签名形状对.
	if string(fc.putBody) != "RIFF-fake-wav-bytes" {
		t.Errorf("OSS body = %q", fc.putBody)
	}
	if !strings.HasPrefix(fc.putAuth, "OSS LTAItest:") {
		t.Errorf("OSS auth = %q", fc.putAuth)
	}

	// Submit: bearer + async header + channel_id + signed URL with expiry.
	// 提交: bearer + 异步头 + channel_id + 带过期的签名 URL.
	if fc.submitAuth != "Bearer sk-test" {
		t.Errorf("submit auth = %q", fc.submitAuth)
	}
	if fc.submitAsync != "enable" {
		t.Errorf("X-DashScope-Async = %q", fc.submitAsync)
	}
	var submitted map[string]any
	if err := json.Unmarshal(fc.submitBody, &submitted); err != nil {
		t.Fatalf("submit body: %v", err)
	}
	if submitted["model"] != ModelParaformer8kV2 {
		t.Errorf("model = %v", submitted["model"])
	}
	params := submitted["parameters"].(map[string]any)
	if ch, _ := json.Marshal(params["channel_id"]); string(ch) != "[0,1]" {
		t.Errorf("channel_id = %s", ch)
	}
	fileURL := submitted["input"].(map[string]any)["file_urls"].([]any)[0].(string)
	if !strings.Contains(fileURL, "OSSAccessKeyId=LTAItest") || !strings.Contains(fileURL, "Expires=") || !strings.Contains(fileURL, "Signature=") {
		t.Errorf("file url not query-signed: %s", fileURL)
	}
	if !strings.Contains(fileURL, "call.wav") {
		t.Errorf("file url lost filename hint: %s", fileURL)
	}

	// Polling actually iterated (1 RUNNING + 1 SUCCEEDED).
	// 轮询真的迭代了 (1 次 RUNNING + 1 次 SUCCEEDED).
	if got := fc.pollCount.Load(); got < 2 {
		t.Errorf("poll count = %d, want >= 2", got)
	}

	// Mapping: 4 segments interleaved by begin_time, ms -> s, channel 0 ->
	// 客户, channel 1 -> 坐席. 映射: 4 段按 begin_time 交叉, 毫秒转秒.
	if len(resp.Segments) != 4 {
		t.Fatalf("segments = %d, want 4", len(resp.Segments))
	}
	wantOrder := []struct {
		speaker string
		start   float64
		text    string
	}{
		{"客户", 0, "你好请问库存"},
		{"坐席", 3, "我查一下"},
		{"坐席", 6, "现在缺货"},
		{"客户", 9, "好的明天答复"},
	}
	for i, w := range wantOrder {
		got := resp.Segments[i]
		if got.Speaker != w.speaker || got.Start != w.start || got.Text != w.text {
			t.Errorf("segment[%d] = {%s %.1f %q}, want {%s %.1f %q}",
				i, got.Speaker, got.Start, got.Text, w.speaker, w.start, w.text)
		}
	}
	if resp.Duration != 187 {
		t.Errorf("duration = %v, want 187", resp.Duration)
	}
	if !strings.Contains(resp.Text, "客户: 你好请问库存") || !strings.Contains(resp.Text, "坐席: 现在缺货") {
		t.Errorf("text missing speaker-prefixed lines: %q", resp.Text)
	}

	// Best-effort OSS delete fired on the happy path.
	// happy path 上尽力删除触发了.
	if !fc.putDeleted.Load() {
		t.Error("staged OSS object was not deleted")
	}
}

func TestTranscribeNoDiarizeOmitsChannels(t *testing.T) {
	fc := newFakeCloud(t)
	fc.resultDoc = `{"properties":{"original_duration_in_milliseconds":1000},
		"transcripts":[{"channel_id":0,"text":"混轨全文","sentences":[
		{"begin_time":0,"end_time":1000,"text":"混轨全文"}]}]}`
	p := fc.provider()

	resp, err := p.Transcribe(context.Background(), &flyto.TranscriptionRequest{
		Audio: strings.NewReader("x"), Model: ModelParaformer8kV2,
	})
	if err != nil {
		t.Fatalf("Transcribe: %v", err)
	}
	var submitted map[string]any
	_ = json.Unmarshal(fc.submitBody, &submitted)
	if _, has := submitted["parameters"].(map[string]any)["channel_id"]; has {
		t.Error("channel_id must be omitted without both speaker labels")
	}
	// Single merged track: no speaker attribution. 单混轨不标说话人.
	if resp.Segments[0].Speaker != "" {
		t.Errorf("speaker = %q, want empty", resp.Segments[0].Speaker)
	}
}

func TestTranscribeTaskFailed(t *testing.T) {
	fc := newFakeCloud(t)
	fc.failTask = true
	p := fc.provider()
	_, err := p.Transcribe(context.Background(), &flyto.TranscriptionRequest{
		Audio: strings.NewReader("x"), Model: ModelParaformer8kV2,
	})
	if err == nil || !strings.Contains(err.Error(), "file format not supported") {
		t.Fatalf("want failed-task error with server reason, got %v", err)
	}
}

func TestTranscribeValidation(t *testing.T) {
	p := New(Config{APIKey: "k", OSS: OSSConfig{Endpoint: "http://x", Bucket: "b", AccessKeyID: "i", AccessKeySecret: "s"}})
	if _, err := p.Transcribe(context.Background(), &flyto.TranscriptionRequest{Model: "m"}); err == nil {
		t.Error("nil audio must error")
	}
	if _, err := p.Transcribe(context.Background(), &flyto.TranscriptionRequest{Audio: strings.NewReader("x")}); err == nil {
		t.Error("missing model must error")
	}
	noOSS := New(Config{APIKey: "k"})
	if _, err := noOSS.Transcribe(context.Background(), &flyto.TranscriptionRequest{Audio: strings.NewReader("x"), Model: "m"}); err == nil || !strings.Contains(err.Error(), "OSS") {
		t.Errorf("missing OSS config must fail loud naming OSS, got %v", err)
	}
	noKey := New(Config{OSS: OSSConfig{Endpoint: "http://x", Bucket: "b", AccessKeyID: "i", AccessKeySecret: "s"}})
	if _, err := noKey.Transcribe(context.Background(), &flyto.TranscriptionRequest{Audio: strings.NewReader("x"), Model: "m"}); err == nil || !strings.Contains(err.Error(), "api key") {
		t.Errorf("missing api key must fail loud, got %v", err)
	}
}

func TestModelsAndStream(t *testing.T) {
	p := New(Config{})
	infos, err := p.Models(context.Background())
	if err != nil || len(infos) != 1 || infos[0].ID != ModelParaformer8kV2 || !infos[0].SupportsTranscription {
		t.Fatalf("Models = %+v, %v", infos, err)
	}
	if _, err := p.Stream(context.Background(), &flyto.Request{}); err == nil {
		t.Error("Stream must fail loud on a transcription-only provider")
	}
	if p.Name() != "aliyun-asr" {
		t.Errorf("Name = %q", p.Name())
	}
}