MeBox/internal/service/strm_queue_test.go

package service

import (
	"context"
	"errors"
	"fmt"
	"net/http"
	"net/http/httptest"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/truewhile/MeBox/internal/model"
	"github.com/truewhile/MeBox/internal/repository"
	"github.com/truewhile/MeBox/internal/service/cloud"
	"github.com/truewhile/MeBox/internal/service/cloud115"
	"go.uber.org/zap"
)

func TestIs115Blocked(t *testing.T) {
	cases := []struct {
		err  error
		want bool
	}{
		{errors.New("115 接口返回 HTTP 405：<!doctypehtml>...访问被阻断"), true},
		{errors.New("115 接口返回 HTTP 405"), true},
		{errors.New("115 接口触发频控/安全拦截（HTTP 405）：阿里云 WAF 拦截页"), true},
		{errors.New("115 接口错误（770004）：访问频率过高"), true},
		{errors.New("115 接口错误（406）：达到访问上限"), true},
		{errors.New("下载失败：http 403"), false},
		{errors.New("解析下载地址失败：115 接口调用失败"), false},
		{errors.New("database is locked (517)"), false},
		{nil, false},
	}
	for _, c := range cases {
		if got := is115Blocked(c.err); got != c.want {
			t.Errorf("is115Blocked(%v) = %v, want %v", c.err, got, c.want)
		}
	}
}

func TestIsHTTPDownloadFailure(t *testing.T) {
	cases := []struct {
		err  error
		want bool
	}{
		{errors.New("http 403"), true},
		{errors.New("http 404"), true},
		{errors.New("http 410"), true},
		{errors.New("http 500"), true},
		{errors.New("Get \"https://x\": connection refused"), false},
		{nil, false},
	}
	for _, c := range cases {
		if got := isHTTPDownloadFailure(c.err); got != c.want {
			t.Errorf("isHTTPDownloadFailure(%v) = %v, want %v", c.err, got, c.want)
		}
	}
}

func TestStrmUploadTasksClearAndRetry(t *testing.T) {
	db := newServiceTestDB(t, &model.StrmUploadTask{})
	repos := repository.New(db)
	svc := NewStrmService(nil, zap.NewNop(), repos, nil)
	ctx := context.Background()

	tasks := []*model.StrmUploadTask{
		{Base: model.Base{ID: "task-pending"}, Status: model.StrmTaskPending, FileName: "1.nfo"},
		{Base: model.Base{ID: "task-running"}, Status: model.StrmTaskRunning, FileName: "2.nfo"},
		{Base: model.Base{ID: "task-done"}, Status: model.StrmTaskDone, FileName: "3.nfo"},
		{Base: model.Base{ID: "task-failed"}, Status: model.StrmTaskFailed, FileName: "4.nfo", Error: "some error", RetryCount: 3},
		{Base: model.Base{ID: "task-canceled"}, Status: model.StrmTaskCanceled, FileName: "5.nfo"},
	}
	for _, task := range tasks {
		if err := db.Create(task).Error; err != nil {
			t.Fatalf("failed to insert task: %v", err)
		}
	}

	// 1. RetryAllFailedUploadTasks
	retried, err := svc.RetryAllFailedUploadTasks(ctx)
	if err != nil {
		t.Fatalf("RetryAllFailedUploadTasks failed: %v", err)
	}
	if retried != 1 {
		t.Fatalf("expected 1 retried task, got %d", retried)
	}
	var failedTask model.StrmUploadTask
	if err := db.First(&failedTask, "id = ?", "task-failed").Error; err != nil {
		t.Fatalf("failed to get task-failed: %v", err)
	}
	if failedTask.Status != model.StrmTaskPending || failedTask.Error != "" || failedTask.RetryCount != 0 {
		t.Fatalf("task-failed was not reset properly: %+v", failedTask)
	}

	// 再次改为 failed 以便测试 ClearFinished
	db.Model(&model.StrmUploadTask{}).Where("id = ?", "task-failed").Updates(map[string]any{"status": model.StrmTaskFailed})

	// 2. ClearFinishedUploadTasks 应删除 done, failed, canceled 三条历史记录
	deleted, err := svc.ClearFinishedUploadTasks(ctx)
	if err != nil {
		t.Fatalf("ClearFinishedUploadTasks failed: %v", err)
	}
	if deleted != 3 {
		t.Fatalf("expected 3 deleted tasks (done, failed, canceled), got %d", deleted)
	}

	// 验证剩余的任务只有 pending 和 running
	var count int64
	db.Model(&model.StrmUploadTask{}).Count(&count)
	if count != 2 {
		t.Fatalf("expected 2 remaining tasks, got %d", count)
	}
}

func TestDownloadSemSeparateByProvider(t *testing.T) {
	svc := NewStrmService(nil, zap.NewNop(), nil, nil)
	svc.initDownloadSems(6)

	if cap(svc.downloadSem115) != strm115DownloadSemCap {
		t.Fatalf("115 sem cap = %d, want %d", cap(svc.downloadSem115), strm115DownloadSemCap)
	}
	if cap(svc.downloadSemDAV) != 6 {
		t.Fatalf("dav sem cap = %d, want 6", cap(svc.downloadSemDAV))
	}

	ctx := context.Background()
	for i := 0; i < strm115DownloadSemCap; i++ {
		if !svc.acquireDownloadSlot(ctx, model.StrmProvider115) {
			t.Fatalf("failed to acquire 115 slot %d", i)
		}
	}
	// 115 槽位占满后，OpenList 仍应能独立获取槽位。
	if !svc.acquireDownloadSlot(ctx, model.StrmProviderOpenList) {
		t.Fatal("openlist slot should remain available while 115 slots are full")
	}
	for i := 0; i < 5; i++ {
		if !svc.acquireDownloadSlot(ctx, model.StrmProviderOpenList) {
			t.Fatalf("failed to acquire extra openlist slot %d", i)
		}
	}
	timeoutCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond)
	defer cancel()
	if svc.acquireDownloadSlot(timeoutCtx, model.StrmProviderOpenList) {
		t.Fatal("expected openlist slots to be exhausted after 6 acquisitions")
	}
}

func TestRequeueDownloadTask(t *testing.T) {
	db := newServiceTestDB(t, &model.StrmDownloadTask{})
	repos := repository.New(db)
	svc := NewStrmService(nil, zap.NewNop(), repos, nil)
	now := time.Now()
	task := &model.StrmDownloadTask{
		Base:      model.Base{ID: "dl-running"},
		Status:    model.StrmTaskRunning,
		Provider:  model.StrmProviderOpenList,
		StartedAt: &now,
	}
	if err := db.Create(task).Error; err != nil {
		t.Fatalf("insert task: %v", err)
	}

	svc.requeueDownloadTask(task)

	var got model.StrmDownloadTask
	if err := db.First(&got, "id = ?", task.ID).Error; err != nil {
		t.Fatalf("load task: %v", err)
	}
	if got.Status != model.StrmTaskPending || got.StartedAt != nil {
		t.Fatalf("task not requeued: %+v", got)
	}
}

// TestProcessUpload115DeletesStaleRemoteMetaFirst 验证 115 覆盖上传语义（以本地为准）：
// 探活未命中时，任务携带网盘旧文件 ID 必须先 /open/ufile/delete 再上传；
// 上传成功后 RemoteRef 回写为新 file_id。
func TestProcessUpload115DeletesStaleRemoteMetaFirst(t *testing.T) {
	svc := testStrmService(t)
	localDir := t.TempDir()
	localFile := filepath.Join(localDir, "movie.nfo")
	if err := os.WriteFile(localFile, []byte("local-nfo-data"), 0o644); err != nil {
		t.Fatal(err)
	}

	acct := &model.StrmAccount{Name: "fake115", Provider: "cloud115", Config: "{}", Enabled: true}
	if err := svc.repo.StrmAccount.Create(context.Background(), acct); err != nil {
		t.Fatal(err)
	}

	var mu sync.Mutex
	var calls []string
	deleteForm := map[string]string{}
	api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		_ = r.ParseForm()
		mu.Lock()
		defer mu.Unlock()
		switch r.URL.Path {
		case "/open/ufile/files":
			calls = append(calls, "probe")
			// 探活：目录为空 / 无同内容副本 → 继续删旧上传
			w.Write([]byte(`{"state":true,"data":[]}`))
		case "/open/ufile/delete":
			calls = append(calls, "delete")
			deleteForm["file_ids"] = r.FormValue("file_ids")
			deleteForm["parent_id"] = r.FormValue("parent_id")
			w.Write([]byte(`{"state":true,"data":[]}`))
		case "/open/upload/init":
			calls = append(calls, "upload")
			// 返回秒传成功，跳过 OSS 真实上传
			w.Write([]byte(`{"state":true,"data":{"status":2,"file_id":"new-1","pick_code":"new-pc-1","callback":null}}`))
		default:
			t.Errorf("unexpected 115 api path %s", r.URL.Path)
			w.Write([]byte(`{"state":false,"message":"unexpected path"}`))
		}
	}))
	defer api.Close()
	oldPro := cloud115.ProAPIBase
	cloud115.ProAPIBase = api.URL
	defer func() { cloud115.ProAPIBase = oldPro }()

	task := &model.StrmUploadTask{
		Base:       model.Base{ID: "up-del-1"},
		SyncPathID: "p1",
		AccountID:  acct.ID,
		Provider:   model.StrmProvider115,
		FileName:   "movie.nfo",
		LocalPath:  localFile,
		RemotePath: "777",
		RemoteRef:  "old-file-1",
		Status:     model.StrmTaskRunning,
	}
	svc.processUpload115(context.Background(), task)

	if task.Status != model.StrmTaskDone {
		t.Fatalf("upload task should succeed, status = %s, error = %s", task.Status, task.Error)
	}
	if task.RemoteRef != "new-1" {
		t.Fatalf("successful upload should record new file_id in RemoteRef, got %q", task.RemoteRef)
	}
	mu.Lock()
	defer mu.Unlock()
	if len(calls) != 3 || calls[0] != "probe" || calls[1] != "delete" || calls[2] != "upload" {
		t.Fatalf("expected probe→delete→upload, got calls = %v", calls)
	}
	if deleteForm["file_ids"] != "old-file-1" {
		t.Fatalf("delete file_ids = %q, want old-file-1", deleteForm["file_ids"])
	}
	if deleteForm["parent_id"] != "777" {
		t.Fatalf("delete parent_id = %q, want 777", deleteForm["parent_id"])
	}
}

// TestProcessUpload115SkipsWhenProbeFindsSameContent 验证上传/重试前探活命中同内容副本时跳过上传，
// 并清理其它脏副本，RemoteRef 回写为已存在的 file_id。
func TestProcessUpload115SkipsWhenProbeFindsSameContent(t *testing.T) {
	svc := testStrmService(t)
	localDir := t.TempDir()
	localFile := filepath.Join(localDir, "movie.nfo")
	content := []byte("already-on-115")
	if err := os.WriteFile(localFile, content, 0o644); err != nil {
		t.Fatal(err)
	}
	sha, err := cloud115.FileSHA1(localFile)
	if err != nil {
		t.Fatal(err)
	}

	acct := &model.StrmAccount{Name: "fake115", Provider: "cloud115", Config: "{}", Enabled: true}
	if err := svc.repo.StrmAccount.Create(context.Background(), acct); err != nil {
		t.Fatal(err)
	}

	var mu sync.Mutex
	var calls []string
	deleteForm := map[string]string{}
	api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		_ = r.ParseForm()
		mu.Lock()
		defer mu.Unlock()
		switch r.URL.Path {
		case "/open/ufile/files":
			calls = append(calls, "probe")
			w.Write([]byte(fmt.Sprintf(
				`{"state":true,"data":[
					{"fid":"keep-1","fc":"1","fn":"movie.nfo","fs":%d,"sha1":%q,"fta":"1"},
					{"fid":"dirty-2","fc":"1","fn":"movie.nfo","fs":9,"sha1":"DEADBEEF","fta":"1"}
				]}`, len(content), sha)))
		case "/open/ufile/delete":
			calls = append(calls, "delete")
			deleteForm["file_ids"] = r.FormValue("file_ids")
			w.Write([]byte(`{"state":true,"data":[]}`))
		case "/open/upload/init":
			calls = append(calls, "upload")
			w.Write([]byte(`{"state":true,"data":{"status":2,"file_id":"should-not","pick_code":"x","callback":null}}`))
		default:
			t.Errorf("unexpected 115 api path %s", r.URL.Path)
			w.Write([]byte(`{"state":false,"message":"unexpected path"}`))
		}
	}))
	defer api.Close()
	oldPro := cloud115.ProAPIBase
	cloud115.ProAPIBase = api.URL
	defer func() { cloud115.ProAPIBase = oldPro }()

	task := &model.StrmUploadTask{
		Base:       model.Base{ID: "up-skip-1"},
		SyncPathID: "p1",
		AccountID:  acct.ID,
		Provider:   model.StrmProvider115,
		FileName:   "movie.nfo",
		LocalPath:  localFile,
		RemotePath: "777",
		RemoteRef:  "old-ref",
		Size:       int64(len(content)),
		Status:     model.StrmTaskRunning,
	}
	svc.processUpload115(context.Background(), task)

	if task.Status != model.StrmTaskDone {
		t.Fatalf("probe hit should finish done, status=%s err=%s", task.Status, task.Error)
	}
	if task.RemoteRef != "keep-1" {
		t.Fatalf("RemoteRef should be matched file_id keep-1, got %q", task.RemoteRef)
	}
	mu.Lock()
	defer mu.Unlock()
	if len(calls) != 2 || calls[0] != "probe" || calls[1] != "delete" {
		t.Fatalf("expected probe→delete (no upload), got %v", calls)
	}
	// 脏副本 dirty-2 与任务旧 ref 都应被清理，keep-1 不得出现
	ids := strings.Split(deleteForm["file_ids"], ",")
	idSet := map[string]bool{}
	for _, id := range ids {
		idSet[strings.TrimSpace(id)] = true
	}
	if !idSet["dirty-2"] || !idSet["old-ref"] {
		t.Fatalf("delete should include dirty-2 and old-ref, got %q", deleteForm["file_ids"])
	}
	if idSet["keep-1"] {
		t.Fatalf("must not delete matched keep-1, got %q", deleteForm["file_ids"])
	}
}

// TestBatchResolve115Links 验证下载队列的批量换链：同账号多个 115 任务的
// pickcode 合并为一次 downurl 请求（官方接口支持逗号分隔多 pick_code），
// 重复引用去重、非 115 任务不参与、直链携带绑定 UA。
func TestBatchResolve115Links(t *testing.T) {
	svc := testStrmService(t)
	acct := &model.StrmAccount{Name: "fake115", Provider: model.StrmProvider115, Config: "{}", Enabled: true}
	if err := svc.repo.StrmAccount.Create(context.Background(), acct); err != nil {
		t.Fatal(err)
	}

	var mu sync.Mutex
	var requests []string
	api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		switch r.URL.Path {
		case "/open/ufile/downurl":
			_ = r.ParseForm()
			mu.Lock()
			requests = append(requests, r.PostFormValue("pick_code"))
			mu.Unlock()
			w.Write([]byte(`{"state":true,"data":{
				"111":{"pick_code":"q-pc-a","url":{"url":"http://cdn/a.mkv"}},
				"222":{"pick_code":"q-pc-b","url":{"url":"http://cdn/b.jpg"}}}}`))
		default:
			t.Errorf("unexpected 115 api path %s", r.URL.Path)
			w.Write([]byte(`{"state":false,"message":"unexpected path"}`))
		}
	}))
	defer api.Close()
	oldPro := cloud115.ProAPIBase
	cloud115.ProAPIBase = api.URL
	defer func() { cloud115.ProAPIBase = oldPro }()
	defer cloud115.ClearDownloadURLCache("q-pc-a")
	defer cloud115.ClearDownloadURLCache("q-pc-b")

	tasks := []*model.StrmDownloadTask{
		{SyncPathID: "p1", AccountID: acct.ID, Provider: model.StrmProvider115, RemoteRef: "q-pc-a"},
		{SyncPathID: "p1", AccountID: acct.ID, Provider: model.StrmProvider115, RemoteRef: "q-pc-a"}, // 重复引用
		{SyncPathID: "p1", AccountID: acct.ID, Provider: model.StrmProvider115, RemoteRef: "q-pc-b"},
		{SyncPathID: "p1", AccountID: acct.ID, Provider: model.StrmProviderOpenList, RemoteRef: "ol-ref"},
		{SyncPathID: "p1", AccountID: acct.ID, Provider: model.StrmProvider115, RemoteRef: "   "}, // 空引用
	}
	resolved := svc.batchResolve115Links(context.Background(), tasks)

	if got := resolved[dlResolveKey(acct.ID, "q-pc-a")]; got == nil || got.URL != "http://cdn/a.mkv" {
		t.Fatalf("missing/bad link for q-pc-a: %+v", resolved)
	}
	if got := resolved[dlResolveKey(acct.ID, "q-pc-b")]; got == nil || got.URL != "http://cdn/b.jpg" {
		t.Fatalf("missing/bad link for q-pc-b: %+v", resolved)
	}
	if got := resolved[dlResolveKey(acct.ID, "q-pc-a")].Headers["User-Agent"]; got != cloud115.DefaultUA {
		t.Fatalf("link UA = %q, want default bound UA", got)
	}
	if _, ok := resolved[dlResolveKey(acct.ID, "ol-ref")]; ok {
		t.Fatal("non-115 task should not be batch resolved")
	}
	mu.Lock()
	defer mu.Unlock()
	if len(requests) != 1 || requests[0] != "q-pc-a,q-pc-b" {
		t.Fatalf("expected single batched downurl request, got %v", requests)
	}
}

// TestProcessDownloadTaskUsesPreResolvedLink 验证任务执行时优先使用批量换链
// 预取的直链：downurl 接口保持失败，若任务仍走逐个换链则必然失败。
func TestProcessDownloadTaskUsesPreResolvedLink(t *testing.T) {
	svc := testStrmService(t)
	localDir := t.TempDir()
	acct := &model.StrmAccount{Name: "fake115", Provider: model.StrmProvider115, Config: "{}", Enabled: true}
	if err := svc.repo.StrmAccount.Create(context.Background(), acct); err != nil {
		t.Fatal(err)
	}

	api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.WriteHeader(http.StatusInternalServerError)
	}))
	defer api.Close()
	oldPro := cloud115.ProAPIBase
	cloud115.ProAPIBase = api.URL
	defer func() { cloud115.ProAPIBase = oldPro }()

	contentSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if r.Header.Get("User-Agent") != cloud115.DefaultUA {
			t.Errorf("download UA = %q, want bound default UA", r.Header.Get("User-Agent"))
		}
		_, _ = w.Write([]byte("nfo-content"))
	}))
	defer contentSrv.Close()

	task := &model.StrmDownloadTask{
		SyncPathID: "p1",
		AccountID:  acct.ID,
		Provider:   model.StrmProvider115,
		FileName:   "movie.nfo",
		RemoteRef:  "q-pc-pre",
		LocalPath:  filepath.Join(localDir, "movie.nfo"),
		Status:     model.StrmTaskRunning,
	}
	if err := svc.repo.StrmDownload.Create(context.Background(), task); err != nil {
		t.Fatal(err)
	}
	resolved := map[string]*cloud.DirectLink{
		dlResolveKey(acct.ID, "q-pc-pre"): {URL: contentSrv.URL, Headers: map[string]string{"User-Agent": cloud115.DefaultUA}},
	}
	svc.processDownloadTask(context.Background(), task, resolved)

	if task.Status != model.StrmTaskDone {
		t.Fatalf("task should be done via pre-resolved link, status = %s, error = %s", task.Status, task.Error)
	}
	data, err := os.ReadFile(task.LocalPath)
	if err != nil || string(data) != "nfo-content" {
		t.Fatalf("downloaded file mismatch: err = %v, data = %q", err, string(data))
	}
}