knowledge-governance/internal/httpapi/library_baseline_mutex_test.go

package httpapi

import (
	"context"
	"errors"
	"net/http"
	"net/http/httptest"
	"testing"
	"time"

	"knowledge-governance/internal/governance"
	"knowledge-governance/internal/jobs"
)

// nonEmptyTarget 模拟"目标整库非空且可读"的探测结果（真实实现直连目标库逐表计数）。
func nonEmptyTarget(context.Context, string) (bool, map[string]int64, error) {
	return false, map[string]int64{"knowledges": 1, "chunks": 2}, nil
}

// scopedProbe：internal 可读非空，external 没有目标 KB（与当前部署一致）。
func scopedProbe(_ context.Context, scope string) (bool, map[string]int64, error) {
	if scope == "external" {
		return false, nil, errors.New("该范围的目标知识库未配置")
	}
	return nonEmptyTarget(context.Background(), scope)
}

// baselineMutexStore 在 fake 之外**真的**实现"同一 scope 只允许一个在飞写作业"这条不变量，
// 用来验证接口的重试/双击行为（真实索引由 integration test
// TestLibraryWriteMutexCoversBaselineJobs 在 PostgreSQL 上证明）。
type baselineMutexStore struct {
	fakeLibraryStore
	activeByScope map[string]string
	enqueues      int
}

func (s *baselineMutexStore) CreateLibraryJob(ctx context.Context, job governance.LibraryJob, stages []string, now time.Time) (governance.LibraryJob, bool, error) {
	if s.activeByScope == nil {
		s.activeByScope = map[string]string{}
	}
	if owner := s.activeByScope[job.Scope]; owner != "" {
		return governance.LibraryJob{}, false, governance.ErrLibraryJobActive
	}
	s.activeByScope[job.Scope] = job.ID
	return s.fakeLibraryStore.CreateLibraryJob(ctx, job, stages, now)
}

func (s *baselineMutexStore) CreateLibraryDispatch(ctx context.Context, job governance.LibraryJob, stages []string, handoff bool, now time.Time) (governance.LibraryJob, bool, error) {
	if s.activeByScope == nil {
		s.activeByScope = map[string]string{}
	}
	if s.activeByScope[job.Scope] != "" {
		return governance.LibraryJob{}, false, governance.ErrLibraryJobActive
	}
	created, reused, err := s.fakeLibraryStore.CreateLibraryDispatch(ctx, job, stages, handoff, now)
	if err == nil {
		s.activeByScope[job.Scope] = created.ID
		s.enqueues++
	}
	return created, reused, err
}

func (s *baselineMutexStore) Enqueue(ctx context.Context, request jobs.EnqueueRequest, now time.Time) (*jobs.Job, error) {
	s.enqueues++
	return s.fakeLibraryStore.fakeJobStore.Enqueue(ctx, request, now)
}

// 双击/重试：第一次建作业成功，第二次被互斥拒绝，且**不会**再投递一个队列作业。
func TestLibraryBaselineDoubleSubmitCreatesOneJob(t *testing.T) {
	store := &baselineMutexStore{fakeLibraryStore: fakeLibraryStore{fakeJobStore: fakeJobStore{enqueueEchoRequestedID: true}}}
	config := Config{Store: store, LibraryScopeProbe: nonEmptyTarget}
	first := httptest.NewRecorder()
	libraryBaselineEnqueue(first, baselineRequest(`{"scope":"internal"}`, roleAdmin, "admin"), config)
	if first.Code != http.StatusAccepted {
		t.Fatalf("first status=%d body=%s", first.Code, first.Body.String())
	}
	if store.enqueues != 1 {
		t.Fatalf("first submit enqueues=%d, want 1", store.enqueues)
	}
	second := httptest.NewRecorder()
	libraryBaselineEnqueue(second, baselineRequest(`{"scope":"internal"}`, roleAdmin, "admin"), config)
	if second.Code != http.StatusConflict {
		t.Fatalf("second status=%d body=%s", second.Code, second.Body.String())
	}
	if body := decodeBody(t, second); body["code"] != "library_write_active" {
		t.Fatalf("second body=%v", body)
	}
	if store.enqueues != 1 {
		t.Fatalf("rejected submit must not enqueue again: enqueues=%d", store.enqueues)
	}
}

// 不同 scope 不会被同一把锁挡住（互斥粒度是 scope/实例，不是全局）。
func TestLibraryBaselineDifferentScopesAreNotBlocked(t *testing.T) {
	store := &baselineMutexStore{fakeLibraryStore: fakeLibraryStore{fakeJobStore: fakeJobStore{enqueueEchoRequestedID: true}}}
	config := Config{Store: store, LibraryScopeProbe: scopedProbe}
	internal := httptest.NewRecorder()
	libraryBaselineEnqueue(internal, baselineRequest(`{"scope":"internal"}`, roleAdmin, "admin"), config)
	if internal.Code != http.StatusAccepted {
		t.Fatalf("internal status=%d body=%s", internal.Code, internal.Body.String())
	}
	// external 在探针阶段就会被拒（该范围没有目标 KB），但绝不能是因为互斥。
	external := httptest.NewRecorder()
	libraryBaselineEnqueue(external, baselineRequest(`{"scope":"external"}`, roleAdmin, "admin"), config)
	if external.Code == http.StatusAccepted {
		t.Fatalf("external without a target KB must not be accepted: %s", external.Body.String())
	}
	body := decodeBody(t, external)
	if body["code"] == "library_write_active" {
		t.Fatalf("external must not be blocked by the internal scope's mutex: %v", body)
	}
}