iotlcm/worker-svr/internal/service/data/async_manual_sync_test.go

package data

import (
	"context"
	"errors"
	"testing"
	"time"

	"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/model"
)

// reviewBatchFake 复用 durableFake 的持久化语义，只补手动同步需要的仓储能力。
type reviewBatchFake struct {
	*durableFake
	candidates []string
	finalized  []string
}

func newReviewBatchFake(candidates ...string) *reviewBatchFake {
	return &reviewBatchFake{durableFake: newDurableFake(), candidates: candidates}
}

func (f *reviewBatchFake) ListReviewSubBatches(_ context.Context, _ string) ([]string, error) {
	return append([]string(nil), f.candidates...), nil
}

// 手动同步不触碰校验/终态写入路径，这里只补齐 Repository 契约。
func (f *reviewBatchFake) Load(context.Context, string) (*Task, error)        { return &Task{}, nil }
func (f *reviewBatchFake) SaveValidationFailure(context.Context, *Task) error { return nil }
func (f *reviewBatchFake) SaveResult(context.Context, *Task, *Detail, Outcome) (Outcome, error) {
	return Outcome{}, nil
}
func (f *reviewBatchFake) Finalize(context.Context, *Task) error { return nil }

// manualSyncUpstream 补上同步办理入口，确保手动同步一旦误走同步路径立即暴露。
type manualSyncUpstream struct {
	*asyncFake
}

func (u *manualSyncUpstream) ChangeSimApnFunction(context.Context, *model.SimCardApnChangeInput) ([]model.SimBaseRes[model.SimCardApnChangeOutputItem], error) {
	return nil, errors.New("手动同步不得调用同步办理入口")
}

func (f *reviewBatchFake) FinalizeReadyBatch(_ context.Context, batchNo string) error {
	f.finalized = append(f.finalized, batchNo)
	return nil
}

func (f *reviewBatchFake) ClaimReviewSubBatch(_ context.Context, operationID, token string, now time.Time) (*SubBatch, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	sub, ok := f.batches[operationID]
	if !ok || (sub.Status != "review" && sub.Status != "review_query") {
		return nil, nil
	}
	sub.Status = "review_query"
	sub.LeaseToken = token
	sub.LeaseUntil = now.Add(subBatchLease)
	f.batches[operationID] = sub
	out := sub
	return &out, nil
}

// 手动同步只用既有受理号只读补查：不重发办理、不清零自动补查计数。
func TestManualSyncQueriesFrozenIntentsWithoutResubmitting(t *testing.T) {
	ctx := context.Background()
	sub := fixtureSub()
	repo := newReviewBatchFake(sub.OperationID)
	sub.Status = "review"
	sub.JobID = "job-1"
	sub.PollCount = 4
	repo.batches[sub.OperationID] = sub

	upstream := &manualSyncUpstream{asyncFake: &asyncFake{
		submit: func(*model.SimCardApnChangeInput) ([]model.SimBaseRes[model.SimCardApnChangeOutputItem], error) {
			t.Fatal("手动同步不得重发办理")
			return nil, nil
		},
		query: func(in *model.SimCardApnChangeInput) ([]model.SimBaseRes[model.SimCardApnChangeOutputItem], error) {
			if in.ReqData.QueryJobID != "job-1" {
				t.Fatalf("手动同步必须按既有 jobId 查询: %+v", in.ReqData)
			}
			out := make([]model.SimBaseRes[model.SimCardApnChangeOutputItem], 0, len(in.SimKeys))
			for _, key := range in.SimKeys {
				out = append(out, model.SimBaseRes[model.SimCardApnChangeOutputItem]{SimKey: key, Code: 0, ResData: model.SimCardApnChangeOutputItem{OrderNum: "job-1"}})
			}
			return out, nil
		},
	}}
	executor := &executor{repository: repo, upstream: upstream}

	queried, err := executor.QueryReviewBatch(ctx, " P-1 ")
	if err != nil {
		t.Fatalf("QueryReviewBatch() error = %v", err)
	}
	if queried != 1 {
		t.Fatalf("queried = %d, want 1", queried)
	}
	if upstream.submitCalls != 0 {
		t.Fatalf("submitCalls = %d, want 0", upstream.submitCalls)
	}
	got := repo.batches[sub.OperationID]
	if got.Status != "done" {
		t.Fatalf("状态 = %s, want done", got.Status)
	}
	if got.PollCount != 4 {
		t.Fatalf("pollCount = %d, want 4（手动同步不得清零自动补查计数）", got.PollCount)
	}
	// 子批收敛与批量入口都会尝试汇总父批，重复汇总是幂等的。
	if len(repo.finalized) == 0 || repo.finalized[0] != "P-1" {
		t.Fatalf("父批未汇总: %+v", repo.finalized)
	}
	if len(repo.outcomes[sub.OperationID]) != len(sub.DetailIDs) {
		t.Fatalf("逐卡结果未写回: %+v", repo.outcomes[sub.OperationID])
	}
}

// 上游仍无结论时明细保持待核查，且可以再次手动同步。
func TestManualSyncKeepsUnknownWhenUpstreamStillPending(t *testing.T) {
	ctx := context.Background()
	sub := fixtureSub()
	repo := newReviewBatchFake(sub.OperationID)
	sub.Status = "review"
	repo.batches[sub.OperationID] = sub

	upstream := &manualSyncUpstream{asyncFake: &asyncFake{
		submit: func(*model.SimCardApnChangeInput) ([]model.SimBaseRes[model.SimCardApnChangeOutputItem], error) {
			t.Fatal("手动同步不得重发办理")
			return nil, nil
		},
		query: func(in *model.SimCardApnChangeInput) ([]model.SimBaseRes[model.SimCardApnChangeOutputItem], error) {
			return pendingResponse(in, 14008), nil
		},
	}}
	executor := &executor{repository: repo, upstream: upstream}

	if _, err := executor.QueryReviewBatch(ctx, "P-1"); err != nil {
		t.Fatalf("QueryReviewBatch() error = %v", err)
	}
	got := repo.batches[sub.OperationID]
	if got.Status != "review" {
		t.Fatalf("状态 = %s, want review（未确认不得记失败）", got.Status)
	}
	if got.JobID != "job-1" {
		t.Fatalf("jobId = %s, want job-1", got.JobID)
	}
	if got.PollCount != 0 {
		t.Fatalf("pollCount = %d, want 0（手动同步不得占用自动补查次数）", got.PollCount)
	}
	// 再次同步仍可用，说明没有被自动窗口锁死。
	if _, err := executor.QueryReviewBatch(ctx, "P-1"); err != nil {
		t.Fatalf("第二次 QueryReviewBatch() error = %v", err)
	}
}

// 单个子批补查失败不影响其余子批，并把失败如实返回给调用方。
func TestManualSyncReportsPerIntentFailureAndContinues(t *testing.T) {
	ctx := context.Background()
	first := fixtureSub()
	first.Status = "review"
	first.JobID = "job-1"
	second := fixtureSub()
	second.OperationID = "intent-2"
	second.Status = "review"
	repo := newReviewBatchFake(first.OperationID, second.OperationID)
	repo.batches[first.OperationID] = first
	repo.batches[second.OperationID] = second

	upstream := &manualSyncUpstream{asyncFake: &asyncFake{
		submit: func(*model.SimCardApnChangeInput) ([]model.SimBaseRes[model.SimCardApnChangeOutputItem], error) {
			t.Fatal("手动同步不得重发办理")
			return nil, nil
		},
		query: func(in *model.SimCardApnChangeInput) ([]model.SimBaseRes[model.SimCardApnChangeOutputItem], error) {
			if in.ReqData.RecoverOnly {
				return nil, errors.New("上游查询不可用")
			}
			return pendingResponse(in, 0), nil
		},
	}}
	executor := &executor{repository: repo, upstream: upstream}

	queried, err := executor.QueryReviewBatch(ctx, "P-1")
	if err == nil {
		t.Fatal("QueryReviewBatch() error = nil, want error")
	}
	if queried != 1 {
		t.Fatalf("queried = %d, want 1（成功一个，失败一个）", queried)
	}
	if upstream.submitCalls != 0 {
		t.Fatalf("submitCalls = %d, want 0", upstream.submitCalls)
	}
}

// 空批次号直接拒绝，避免误触发全量扫描。
func TestManualSyncRejectsBlankBatchNo(t *testing.T) {
	executor := &executor{repository: newReviewBatchFake(), upstream: &manualSyncUpstream{asyncFake: &asyncFake{}}}
	if _, err := executor.QueryReviewBatch(context.Background(), "  "); err == nil {
		t.Fatal("QueryReviewBatch() blank batchNo error = nil, want error")
	}
}