package server

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/mem9-ai/drive9/pkg/backend"
	"github.com/mem9-ai/drive9/pkg/meta"
)

// fakeTenantExtractConfigReader is an in-memory tenantExtractConfigReader
// with query counting for cache assertions.
type fakeTenantExtractConfigReader struct {
	mu    sync.Mutex
	rows  map[string]meta.TenantExtractConfig
	err   error
	calls int
}

func (f *fakeTenantExtractConfigReader) GetTenantExtractConfig(_ context.Context, tenantID, mediaType string) (*meta.TenantExtractConfig, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.calls++
	if f.err != nil {
		return nil, f.err
	}
	row, ok := f.rows[tenantID+"\x00"+mediaType]
	if !ok {
		row, ok = f.rows[tenantID]
	}
	if !ok {
		return nil, meta.ErrNotFound
	}
	cp := row
	return &cp, nil
}

func (f *fakeTenantExtractConfigReader) callCount() int {
	f.mu.Lock()
	defer f.mu.Unlock()
	return f.calls
}

func (f *fakeTenantExtractConfigReader) setErr(err error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.err = err
}

// fakeExtractDecryptor returns a fixed plaintext or error.
type fakeExtractDecryptor struct {
	plaintext string
	err       error
	calls     int
}

func (f *fakeExtractDecryptor) Decrypt(_ context.Context, _ []byte) ([]byte, error) {
	f.calls++
	if f.err != nil {
		return nil, f.err
	}
	return []byte(f.plaintext), nil
}

type stubImageTextExtractor struct{}

func (stubImageTextExtractor) ExtractImageText(context.Context, backend.ImageExtractRequest) (string, backend.ImageExtractUsage, error) {
	return "stub", backend.ImageExtractUsage{}, nil
}

type stubAudioTextExtractor struct{}

func (stubAudioTextExtractor) ExtractAudioText(context.Context, backend.AudioExtractRequest) (string, backend.AudioExtractUsage, error) {
	return "stub", backend.AudioExtractUsage{}, nil
}

type stubVideoTextExtractor struct{}

func (stubVideoTextExtractor) ExtractVideoText(context.Context, backend.VideoExtractRequest) (string, backend.VideoExtractUsage, error) {
	return "stub", backend.VideoExtractUsage{}, nil
}

func defaultTestRuntime() *backend.ImageExtractRuntime {
	return &backend.ImageExtractRuntime{
		Extractor:           stubImageTextExtractor{},
		Timeout:             7 * time.Second,
		MaxImageBytes:       1234,
		MaxExtractTextBytes: 567,
	}
}

func enabledImageRow(tenantID string) meta.TenantExtractConfig {
	return meta.TenantExtractConfig{
		TenantID:     tenantID,
		MediaType:    meta.ExtractMediaTypeImage,
		Enabled:      true,
		APIBase:      "https://extractor.example.com",
		APIKeyCipher: []byte("cipher-bytes"),
		Model:        "vision-model-1",
		Prompt:       "describe this image for search",
	}
}

func enabledMediaRow(tenantID, mediaType string) meta.TenantExtractConfig {
	return meta.TenantExtractConfig{
		TenantID:     tenantID,
		MediaType:    mediaType,
		Enabled:      true,
		APIBase:      "https://extractor.example.com",
		APIKeyCipher: []byte("cipher-bytes"),
		Model:        "media-model-1",
		Prompt:       "extract searchable media text",
		Protocol:     backend.MediaProviderProtocolOpenAI,
	}
}

func TestExtractConfigResolverComposesVideoAndAudioRuntime(t *testing.T) {
	workspace, err := backend.NewMediaExtractWorkspace(backend.MediaExtractWorkspaceOptions{RootDir: t.TempDir()})
	if err != nil {
		t.Fatal(err)
	}
	store := &fakeTenantExtractConfigReader{rows: map[string]meta.TenantExtractConfig{
		"t1\x00" + meta.ExtractMediaTypeVideo: enabledMediaRow("t1", meta.ExtractMediaTypeVideo),
		"t1\x00" + meta.ExtractMediaTypeAudio: enabledMediaRow("t1", meta.ExtractMediaTypeAudio),
	}}
	r := newExtractConfigResolver(store, &fakeExtractDecryptor{plaintext: "sk-tenant"}, nil, time.Minute,
		ExtractResolverDefaults{Workspace: workspace})
	t.Cleanup(r.Close)

	res, err := r.ResolveVideo(context.Background(), "t1")
	if err != nil {
		t.Fatalf("ResolveVideo: %v", err)
	}
	if res.state != extractResolveReady || res.runtime.Extractor == nil {
		t.Fatalf("video resolution = %+v, want ready runtime", res)
	}
	if res.runtime.AudioRuntime == nil || res.runtime.AudioRuntime.Extractor == nil {
		t.Fatalf("audio runtime = %+v, want tenant audio runtime", res.runtime.AudioRuntime)
	}
	if res.runtime.Workspace != workspace || res.runtime.AudioRuntime.Workspace != workspace {
		t.Fatalf("workspaces = video:%p audio:%p, want shared %p", res.runtime.Workspace, res.runtime.AudioRuntime.Workspace, workspace)
	}
}

func TestExtractConfigResolverUsesQwenRawPayloadLimit(t *testing.T) {
	row := enabledMediaRow("t1", meta.ExtractMediaTypeAudio)
	row.Protocol = backend.MediaProviderProtocolQwenASR
	store := &fakeTenantExtractConfigReader{rows: map[string]meta.TenantExtractConfig{
		"t1\x00" + meta.ExtractMediaTypeAudio: row,
	}}
	r := newExtractConfigResolver(store, &fakeExtractDecryptor{plaintext: "sk-tenant"}, nil, time.Minute)
	t.Cleanup(r.Close)

	res, err := r.ResolveAudio(context.Background(), "t1")
	if err != nil {
		t.Fatal(err)
	}
	if res.runtime.MaxAudioBytes != defaultTenantQwenASRMaxAudioBytes {
		t.Fatalf("MaxAudioBytes=%d, want qwen limit %d", res.runtime.MaxAudioBytes, defaultTenantQwenASRMaxAudioBytes)
	}
}

func TestExtractConfigResolverAudioOffMarkerDisablesVideoAudio(t *testing.T) {
	store := &fakeTenantExtractConfigReader{rows: map[string]meta.TenantExtractConfig{
		"t1\x00" + meta.ExtractMediaTypeVideo: enabledMediaRow("t1", meta.ExtractMediaTypeVideo),
		"t1\x00" + meta.ExtractMediaTypeAudio: {TenantID: "t1", MediaType: meta.ExtractMediaTypeAudio, Enabled: false},
	}}
	r := newExtractConfigResolver(store, &fakeExtractDecryptor{plaintext: "sk-tenant"}, nil, time.Minute)
	t.Cleanup(r.Close)

	res, err := r.ResolveVideo(context.Background(), "t1")
	if err != nil {
		t.Fatalf("ResolveVideo: %v", err)
	}
	if res.state != extractResolveReady {
		t.Fatalf("state = %v, want ready", res.state)
	}
	if res.runtime.AudioRuntime != nil {
		t.Fatalf("audio runtime = %+v, want nil for off-marker", res.runtime.AudioRuntime)
	}
}

func TestExtractConfigResolverVideoUsesDefaultAudioWhenTenantRowAbsent(t *testing.T) {
	defAudio := &backend.AudioExtractRuntime{Extractor: stubAudioTextExtractor{}, Timeout: time.Minute, MaxAudioBytes: 123}
	defVideo := &backend.VideoExtractRuntime{Extractor: stubVideoTextExtractor{}, Timeout: 2 * time.Minute, MaxSourceBytes: 456}
	store := &fakeTenantExtractConfigReader{rows: map[string]meta.TenantExtractConfig{}}
	r := newExtractConfigResolver(store, &fakeExtractDecryptor{}, nil, time.Minute,
		ExtractResolverDefaults{Audio: defAudio, Video: defVideo})
	t.Cleanup(r.Close)

	res, err := r.ResolveVideo(context.Background(), "t1")
	if err != nil {
		t.Fatalf("ResolveVideo: %v", err)
	}
	if res.state != extractResolveReady || res.runtime.Extractor != defVideo.Extractor {
		t.Fatalf("video resolution = %+v, want process default", res)
	}
	if res.runtime.AudioRuntime == nil || res.runtime.AudioRuntime.Extractor != defAudio.Extractor {
		t.Fatalf("audio runtime = %+v, want process default", res.runtime.AudioRuntime)
	}
	if defVideo.AudioRuntime != nil {
		t.Fatal("resolving a tenant must not mutate the shared default video runtime")
	}
}

func TestExtractConfigResolverTenantMediaInheritsProcessLimits(t *testing.T) {
	const (
		imageSourceLimit  = int64(1 << 30)
		audioSourceLimit  = int64(5 << 30)
		videoSourceLimit  = int64(10 << 30)
		videoPayloadLimit = int64(9 << 20)
		audioTimeout      = 6 * time.Minute
	)
	store := &fakeTenantExtractConfigReader{rows: map[string]meta.TenantExtractConfig{
		"t1\x00" + meta.ExtractMediaTypeImage: enabledMediaRow("t1", meta.ExtractMediaTypeImage),
		"t1\x00" + meta.ExtractMediaTypeAudio: enabledMediaRow("t1", meta.ExtractMediaTypeAudio),
		"t1\x00" + meta.ExtractMediaTypeVideo: enabledMediaRow("t1", meta.ExtractMediaTypeVideo),
	}}
	r := newExtractConfigResolver(store, &fakeExtractDecryptor{plaintext: "sk-tenant"}, nil, time.Minute,
		ExtractResolverDefaults{
			ImageMaxSourceBytes:          imageSourceLimit,
			AudioMaxSourceBytes:          audioSourceLimit,
			VideoMaxSourceBytes:          videoSourceLimit,
			VideoMaxProviderPayloadBytes: videoPayloadLimit,
			AudioTenantFallbackTimeout:   audioTimeout,
			VideoTenantFallbackTimeout:   7 * time.Minute,
		})
	t.Cleanup(r.Close)

	imageRes, err := r.ResolveImage(context.Background(), "t1")
	if err != nil {
		t.Fatal(err)
	}
	if imageRes.runtime.MaxSourceBytes != imageSourceLimit {
		t.Fatalf("image MaxSourceBytes=%d, want %d", imageRes.runtime.MaxSourceBytes, imageSourceLimit)
	}
	audioRes, err := r.ResolveAudio(context.Background(), "t1")
	if err != nil {
		t.Fatal(err)
	}
	if audioRes.runtime.MaxSourceBytes != audioSourceLimit {
		t.Fatalf("audio MaxSourceBytes=%d, want %d", audioRes.runtime.MaxSourceBytes, audioSourceLimit)
	}
	if audioRes.runtime.Timeout != audioTimeout {
		t.Fatalf("audio timeout=%v, want %v", audioRes.runtime.Timeout, audioTimeout)
	}
	videoRes, err := r.ResolveVideo(context.Background(), "t1")
	if err != nil {
		t.Fatal(err)
	}
	if videoRes.runtime.MaxSourceBytes != videoSourceLimit {
		t.Fatalf("video MaxSourceBytes=%d, want %d", videoRes.runtime.MaxSourceBytes, videoSourceLimit)
	}
	if videoRes.runtime.Timeout != 7*time.Minute {
		t.Fatalf("video timeout=%v, want 7m", videoRes.runtime.Timeout)
	}
}

// (a) absent row + default present → ready with the default runtime.
func TestExtractConfigResolverAbsentRowUsesDefault(t *testing.T) {
	store := &fakeTenantExtractConfigReader{rows: map[string]meta.TenantExtractConfig{}}
	dec := &fakeExtractDecryptor{plaintext: "unused"}
	def := defaultTestRuntime()
	r := newExtractConfigResolver(store, dec, def, time.Minute)
	t.Cleanup(r.Close)

	res, err := r.ResolveImage(context.Background(), "t1")
	if err != nil {
		t.Fatalf("ResolveImage: %v", err)
	}
	if res.state != extractResolveReady {
		t.Fatalf("state = %v, want ready", res.state)
	}
	if res.runtime.Extractor != def.Extractor ||
		res.runtime.Timeout != def.Timeout ||
		res.runtime.MaxImageBytes != def.MaxImageBytes ||
		res.runtime.MaxExtractTextBytes != def.MaxExtractTextBytes {
		t.Fatalf("runtime = %+v, want default %+v", res.runtime, *def)
	}
	if dec.calls != 0 {
		t.Fatalf("decrypt calls = %d, want 0 for absent row", dec.calls)
	}
}

// (b) absent row + no default → not_configured.
func TestExtractConfigResolverAbsentRowNoDefault(t *testing.T) {
	store := &fakeTenantExtractConfigReader{rows: map[string]meta.TenantExtractConfig{}}
	r := newExtractConfigResolver(store, &fakeExtractDecryptor{}, nil, time.Minute)
	t.Cleanup(r.Close)

	res, err := r.ResolveImage(context.Background(), "t1")
	if err != nil {
		t.Fatalf("ResolveImage: %v", err)
	}
	if res.state != extractResolveNotConfigured {
		t.Fatalf("state = %v, want not_configured", res.state)
	}
	if res.runtime.Extractor != nil {
		t.Fatalf("runtime extractor = %v, want nil", res.runtime.Extractor)
	}
}

// (c) off-marker row → disabled; no decrypt, no extractor built.
func TestExtractConfigResolverOffMarkerDisables(t *testing.T) {
	store := &fakeTenantExtractConfigReader{rows: map[string]meta.TenantExtractConfig{
		"t1": {TenantID: "t1", MediaType: meta.ExtractMediaTypeImage, Enabled: false},
	}}
	dec := &fakeExtractDecryptor{plaintext: "sk-tenant"}
	r := newExtractConfigResolver(store, dec, defaultTestRuntime(), time.Minute)
	t.Cleanup(r.Close)

	res, err := r.ResolveImage(context.Background(), "t1")
	if err != nil {
		t.Fatalf("ResolveImage: %v", err)
	}
	if res.state != extractResolveDisabled {
		t.Fatalf("state = %v, want disabled", res.state)
	}
	if res.runtime.Extractor != nil {
		t.Fatalf("runtime extractor = %v, want nil", res.runtime.Extractor)
	}
	if dec.calls != 0 {
		t.Fatalf("decrypt calls = %d, want 0 for off-marker", dec.calls)
	}
}

// captureOpenAIExtractor swaps the resolver's OpenAI extractor constructor
// with one that records the composed config and returns a stub extractor.
func captureOpenAIExtractor(r *extractConfigResolver, captured *backend.OpenAIImageTextExtractorConfig) {
	r.newOpenAIExtractor = func(cfg backend.OpenAIImageTextExtractorConfig) (backend.ImageTextExtractor, error) {
		*captured = cfg
		return stubImageTextExtractor{}, nil
	}
}

// (d) enabled row → ready with a tenant extractor composed from the row:
// model/base from the row, decrypted key, prompt composed incl. suffix.
func TestExtractConfigResolverEnabledRowBuildsTenantExtractor(t *testing.T) {
	row := enabledImageRow("t1")
	store := &fakeTenantExtractConfigReader{rows: map[string]meta.TenantExtractConfig{"t1": row}}
	dec := &fakeExtractDecryptor{plaintext: "sk-tenant-key"}
	workspace, err := backend.NewMediaExtractWorkspace(backend.MediaExtractWorkspaceOptions{RootDir: t.TempDir()})
	if err != nil {
		t.Fatal(err)
	}
	r := newExtractConfigResolver(store, dec, defaultTestRuntime(), time.Minute, ExtractResolverDefaults{Workspace: workspace})
	t.Cleanup(r.Close)

	var captured backend.OpenAIImageTextExtractorConfig
	captureOpenAIExtractor(r, &captured)

	res, err := r.ResolveImage(context.Background(), "t1")
	if err != nil {
		t.Fatalf("ResolveImage: %v", err)
	}
	if res.state != extractResolveReady {
		t.Fatalf("state = %v, want ready", res.state)
	}
	if res.runtime.Extractor == nil {
		t.Fatal("runtime extractor is nil, want fallback-wrapped tenant extractor")
	}
	if res.runtime.Workspace != workspace {
		t.Fatalf("runtime workspace=%p, want %p", res.runtime.Workspace, workspace)
	}
	if captured.BaseURL != row.APIBase {
		t.Errorf("BaseURL = %q, want %q", captured.BaseURL, row.APIBase)
	}
	if captured.Model != row.Model {
		t.Errorf("Model = %q, want %q", captured.Model, row.Model)
	}
	if captured.APIKey != "sk-tenant-key" {
		t.Errorf("APIKey = %q, want decrypted %q", captured.APIKey, "sk-tenant-key")
	}
	wantPrompt := backend.ComposeImageExtractPrompt(row.Prompt)
	if captured.Prompt != wantPrompt {
		t.Errorf("Prompt = %q, want composed %q", captured.Prompt, wantPrompt)
	}
	if !strings.Contains(captured.Prompt, row.Prompt) {
		t.Errorf("Prompt %q does not include tenant prompt %q", captured.Prompt, row.Prompt)
	}
	if !strings.Contains(captured.Prompt, `"attributes"`) {
		t.Errorf("Prompt %q missing attributes suffix", captured.Prompt)
	}
	if captured.Client == nil {
		t.Error("Client is nil, want SSRF-safe tenant extractor client")
	}
	if dec.calls != 1 {
		t.Errorf("decrypt calls = %d, want 1", dec.calls)
	}
	// Limits are inherited from the process default when one exists.
	def := *defaultTestRuntime()
	if res.runtime.Timeout != def.Timeout ||
		res.runtime.MaxImageBytes != def.MaxImageBytes ||
		res.runtime.MaxExtractTextBytes != def.MaxExtractTextBytes {
		t.Errorf("runtime limits = %+v, want inherited from default %+v", res.runtime, def)
	}
}

func TestExtractConfigResolverSafeClientHonorsLongestConfiguredTimeout(t *testing.T) {
	def := defaultTestRuntime()
	def.Timeout = 5 * time.Minute
	r := newExtractConfigResolver(
		&fakeTenantExtractConfigReader{rows: map[string]meta.TenantExtractConfig{}},
		&fakeExtractDecryptor{}, def, time.Minute,
		ExtractResolverDefaults{Audio: &backend.AudioExtractRuntime{Timeout: 3 * time.Minute}},
	)
	t.Cleanup(r.Close)
	if r.tenantHTTPClient.Timeout != 5*time.Minute {
		t.Fatalf("client timeout=%v, want 5m", r.tenantHTTPClient.Timeout)
	}
}

func TestExtractConfigResolverReusesSafeClientWithoutShorteningMediaRequests(t *testing.T) {
	store := &fakeTenantExtractConfigReader{rows: map[string]meta.TenantExtractConfig{
		"t1": enabledImageRow("t1"),
		"t2": enabledImageRow("t2"),
	}}
	def := defaultTestRuntime()
	r := newExtractConfigResolver(store, &fakeExtractDecryptor{plaintext: "sk"}, def, time.Minute)
	t.Cleanup(r.Close)
	var captured []backend.OpenAIImageTextExtractorConfig
	r.newOpenAIExtractor = func(cfg backend.OpenAIImageTextExtractorConfig) (backend.ImageTextExtractor, error) {
		captured = append(captured, cfg)
		return stubImageTextExtractor{}, nil
	}

	for _, tenantID := range []string{"t1", "t2"} {
		if _, err := r.ResolveImage(context.Background(), tenantID); err != nil {
			t.Fatalf("ResolveImage(%s): %v", tenantID, err)
		}
	}
	if len(captured) != 2 {
		t.Fatalf("extractor configs = %d, want 2", len(captured))
	}
	if captured[0].Client == nil || captured[0].Client != captured[1].Client {
		t.Fatalf("tenant clients = %p/%p, want one resolver-level client", captured[0].Client, captured[1].Client)
	}
	if captured[0].Client.Timeout != defaultTenantVideoExtractTimeout {
		t.Fatalf("client timeout = %v, want longest media timeout %v", captured[0].Client.Timeout, defaultTenantVideoExtractTimeout)
	}
}

// (d2) enabled row with empty prompt still gets the composed default prompt
// plus attributes suffix, and built-in limits when no process default exists.
func TestExtractConfigResolverEnabledRowWithoutDefault(t *testing.T) {
	row := enabledImageRow("t1")
	row.Prompt = ""
	store := &fakeTenantExtractConfigReader{rows: map[string]meta.TenantExtractConfig{"t1": row}}
	r := newExtractConfigResolver(store, &fakeExtractDecryptor{plaintext: "sk"}, nil, time.Minute)
	t.Cleanup(r.Close)

	var captured backend.OpenAIImageTextExtractorConfig
	captureOpenAIExtractor(r, &captured)

	res, err := r.ResolveImage(context.Background(), "t1")
	if err != nil {
		t.Fatalf("ResolveImage: %v", err)
	}
	if res.state != extractResolveReady {
		t.Fatalf("state = %v, want ready", res.state)
	}
	if captured.Prompt != backend.ComposeImageExtractPrompt("") {
		t.Errorf("Prompt = %q, want default prompt with suffix", captured.Prompt)
	}
	if res.runtime.Timeout <= 0 || res.runtime.MaxImageBytes <= 0 || res.runtime.MaxExtractTextBytes <= 0 {
		t.Errorf("runtime limits = %+v, want positive built-in defaults", res.runtime)
	}
}

// (d3) without a process default the tenant runtime timeout defaults to
// defaultTenantImageExtractTimeout, and the constructor fallback timeout
// overrides the image runtime limit. The shared tenant HTTP client retains
// the longest built-in media timeout so it does not truncate audio/video.
func TestExtractConfigResolverTenantFallbackTimeout(t *testing.T) {
	store := &fakeTenantExtractConfigReader{rows: map[string]meta.TenantExtractConfig{
		"t1": enabledImageRow("t1"),
		"t2": enabledImageRow("t2"),
	}}

	r := newExtractConfigResolver(store, &fakeExtractDecryptor{plaintext: "sk"}, nil, time.Minute)
	var captured []backend.OpenAIImageTextExtractorConfig
	r.newOpenAIExtractor = func(cfg backend.OpenAIImageTextExtractorConfig) (backend.ImageTextExtractor, error) {
		captured = append(captured, cfg)
		return stubImageTextExtractor{}, nil
	}
	res, err := r.ResolveImage(context.Background(), "t1")
	r.Close()
	if err != nil {
		t.Fatalf("ResolveImage(t1): %v", err)
	}
	if res.runtime.Timeout != defaultTenantImageExtractTimeout {
		t.Fatalf("runtime timeout = %v, want default %v", res.runtime.Timeout, defaultTenantImageExtractTimeout)
	}
	if captured[0].Client == nil || captured[0].Client.Timeout != defaultTenantVideoExtractTimeout {
		t.Fatalf("client timeout = %+v, want longest media timeout %v", captured[0].Client, defaultTenantVideoExtractTimeout)
	}

	r = newExtractConfigResolverWithFallback(store, &fakeExtractDecryptor{plaintext: "sk"}, nil, time.Minute, 90*time.Second)
	t.Cleanup(r.Close)
	captured = nil
	r.newOpenAIExtractor = func(cfg backend.OpenAIImageTextExtractorConfig) (backend.ImageTextExtractor, error) {
		captured = append(captured, cfg)
		return stubImageTextExtractor{}, nil
	}
	res, err = r.ResolveImage(context.Background(), "t2")
	if err != nil {
		t.Fatalf("ResolveImage(t2): %v", err)
	}
	if res.runtime.Timeout != 90*time.Second {
		t.Fatalf("runtime timeout after override = %v, want 90s", res.runtime.Timeout)
	}
	if captured[0].Client == nil || captured[0].Client.Timeout != defaultTenantVideoExtractTimeout {
		t.Fatalf("client timeout after override = %+v, want longest media timeout %v", captured[0].Client, defaultTenantVideoExtractTimeout)
	}

	// Non-positive fallback timeout keeps the built-in default.
	r2 := newExtractConfigResolverWithFallback(store, &fakeExtractDecryptor{plaintext: "sk"}, nil, time.Minute, 0)
	t.Cleanup(r2.Close)
	if timeout, _, _ := r2.runtimeLimits(); timeout != defaultTenantImageExtractTimeout {
		t.Fatalf("fallback timeout with zero override = %v, want %v", timeout, defaultTenantImageExtractTimeout)
	}
}

// (e) meta error other than ErrNotFound → error, never the default.
func TestExtractConfigResolverMetaErrorFailsClosed(t *testing.T) {
	store := &fakeTenantExtractConfigReader{rows: map[string]meta.TenantExtractConfig{}}
	store.setErr(errors.New("meta connection lost"))
	r := newExtractConfigResolver(store, &fakeExtractDecryptor{}, defaultTestRuntime(), time.Minute)
	t.Cleanup(r.Close)

	res, err := r.ResolveImage(context.Background(), "t1")
	if err == nil {
		t.Fatal("ResolveImage returned nil error for meta failure")
	}
	if res.state != extractResolveUnknown || res.runtime.Extractor != nil {
		t.Fatalf("res = %+v, want unknown state with no default fallback on error", res)
	}
}

// (f) decrypt error → error, no runtime.
func TestExtractConfigResolverDecryptErrorFailsClosed(t *testing.T) {
	store := &fakeTenantExtractConfigReader{rows: map[string]meta.TenantExtractConfig{"t1": enabledImageRow("t1")}}
	dec := &fakeExtractDecryptor{err: errors.New("kms unavailable")}
	r := newExtractConfigResolver(store, dec, defaultTestRuntime(), time.Minute)
	t.Cleanup(r.Close)

	res, err := r.ResolveImage(context.Background(), "t1")
	if err == nil {
		t.Fatal("ResolveImage returned nil error for decrypt failure")
	}
	if res.runtime.Extractor != nil {
		t.Fatalf("runtime extractor = %v, want nil", res.runtime.Extractor)
	}
}

// Invalid tenant api_base → error (fail-closed), even with a default present.
func TestExtractConfigResolverInvalidAPIBaseFailsClosed(t *testing.T) {
	row := enabledImageRow("t1")
	row.APIBase = "http://insecure.example.com"
	store := &fakeTenantExtractConfigReader{rows: map[string]meta.TenantExtractConfig{"t1": row}}
	r := newExtractConfigResolver(store, &fakeExtractDecryptor{plaintext: "sk"}, defaultTestRuntime(), time.Minute)
	t.Cleanup(r.Close)

	res, err := r.ResolveImage(context.Background(), "t1")
	if err == nil {
		t.Fatal("ResolveImage returned nil error for http api base")
	}
	if res.runtime.Extractor != nil {
		t.Fatalf("runtime extractor = %v, want nil", res.runtime.Extractor)
	}
}

// Extractor construction failure → error (fail-closed).
func TestExtractConfigResolverExtractorBuildErrorFailsClosed(t *testing.T) {
	store := &fakeTenantExtractConfigReader{rows: map[string]meta.TenantExtractConfig{"t1": enabledImageRow("t1")}}
	r := newExtractConfigResolver(store, &fakeExtractDecryptor{plaintext: "sk"}, defaultTestRuntime(), time.Minute)
	t.Cleanup(r.Close)
	r.newOpenAIExtractor = func(backend.OpenAIImageTextExtractorConfig) (backend.ImageTextExtractor, error) {
		return nil, fmt.Errorf("bad config")
	}

	res, err := r.ResolveImage(context.Background(), "t1")
	if err == nil {
		t.Fatal("ResolveImage returned nil error for extractor build failure")
	}
	if res.runtime.Extractor != nil {
		t.Fatalf("runtime extractor = %v, want nil", res.runtime.Extractor)
	}
}

// (g) cache: a second call within TTL does not hit meta; past TTL with meta
// now failing → error, never a stale entry; recovery re-resolves.
func TestExtractConfigResolverCacheTTL(t *testing.T) {
	store := &fakeTenantExtractConfigReader{rows: map[string]meta.TenantExtractConfig{"t1": enabledImageRow("t1")}}
	dec := &fakeExtractDecryptor{plaintext: "sk"}
	r := newExtractConfigResolver(store, dec, nil, time.Minute)
	t.Cleanup(r.Close)

	now := time.Now()
	r.now = func() time.Time { return now }

	if _, err := r.ResolveImage(context.Background(), "t1"); err != nil {
		t.Fatalf("first ResolveImage: %v", err)
	}
	if got := store.callCount(); got != 1 {
		t.Fatalf("meta calls after first resolve = %d, want 1", got)
	}
	if _, err := r.ResolveImage(context.Background(), "t1"); err != nil {
		t.Fatalf("second ResolveImage: %v", err)
	}
	if got := store.callCount(); got != 1 {
		t.Fatalf("meta calls within TTL = %d, want 1 (cached)", got)
	}
	if dec.calls != 1 {
		t.Fatalf("decrypt calls within TTL = %d, want 1 (cached)", dec.calls)
	}

	// Past TTL with meta failing → error, no stale serve.
	now = now.Add(2 * time.Minute)
	store.setErr(errors.New("meta down"))
	res, err := r.ResolveImage(context.Background(), "t1")
	if err == nil {
		t.Fatalf("ResolveImage past TTL with meta failing returned nil error, res = %+v", res)
	}

	// Meta recovers → re-resolution succeeds and is cached again.
	store.setErr(nil)
	now = now.Add(extractConfigFailureBackoff + time.Millisecond)
	if _, err := r.ResolveImage(context.Background(), "t1"); err != nil {
		t.Fatalf("ResolveImage after recovery: %v", err)
	}
	if got := store.callCount(); got != 3 {
		t.Fatalf("meta calls = %d, want 3", got)
	}
	if _, err := r.ResolveImage(context.Background(), "t1"); err != nil {
		t.Fatalf("ResolveImage after re-cache: %v", err)
	}
	if got := store.callCount(); got != 3 {
		t.Fatalf("meta calls after re-cache = %d, want 3 (cached)", got)
	}
}

// (h) two resolver instances over the same store resolve the same row
// independently (no cross-instance coupling).
func TestExtractConfigResolverIndependentInstances(t *testing.T) {
	row := enabledImageRow("t1")
	store := &fakeTenantExtractConfigReader{rows: map[string]meta.TenantExtractConfig{"t1": row}}

	var captured1, captured2 backend.OpenAIImageTextExtractorConfig
	r1 := newExtractConfigResolver(store, &fakeExtractDecryptor{plaintext: "sk"}, nil, time.Minute)
	t.Cleanup(r1.Close)
	r2 := newExtractConfigResolver(store, &fakeExtractDecryptor{plaintext: "sk"}, nil, time.Minute)
	t.Cleanup(r2.Close)
	captureOpenAIExtractor(r1, &captured1)
	captureOpenAIExtractor(r2, &captured2)

	res1, err := r1.ResolveImage(context.Background(), "t1")
	if err != nil {
		t.Fatalf("r1.ResolveImage: %v", err)
	}
	res2, err := r2.ResolveImage(context.Background(), "t1")
	if err != nil {
		t.Fatalf("r2.ResolveImage: %v", err)
	}
	if res1.state != extractResolveReady || res2.state != extractResolveReady {
		t.Fatalf("states = %v, %v, want both ready", res1.state, res2.state)
	}
	if captured1.BaseURL != captured2.BaseURL ||
		captured1.Model != captured2.Model ||
		captured1.Prompt != captured2.Prompt {
		t.Fatalf("instances resolved different configs: %+v vs %+v", captured1, captured2)
	}
	if got := store.callCount(); got != 2 {
		t.Fatalf("meta calls = %d, want 2 (each instance resolves independently)", got)
	}
}

// The constructor argument is the retention budget: the effective cache TTL
// is half of it, so entries stop being served once the effective TTL passes
// and worst-case physical retention (effective TTL + sweep interval) stays
// within the budget.
func TestExtractConfigResolverEffectiveTTLIsHalfRetention(t *testing.T) {
	store := &fakeTenantExtractConfigReader{rows: map[string]meta.TenantExtractConfig{"t1": enabledImageRow("t1")}}
	r := newExtractConfigResolver(store, &fakeExtractDecryptor{plaintext: "sk"}, nil, time.Minute)
	t.Cleanup(r.Close)

	now := time.Now()
	r.now = func() time.Time { return now }

	if _, err := r.ResolveImage(context.Background(), "t1"); err != nil {
		t.Fatalf("first ResolveImage: %v", err)
	}
	// Within the effective TTL (30s for a 60s retention budget): cached.
	now = now.Add(29 * time.Second)
	if _, err := r.ResolveImage(context.Background(), "t1"); err != nil {
		t.Fatalf("ResolveImage within effective TTL: %v", err)
	}
	if got := store.callCount(); got != 1 {
		t.Fatalf("meta calls within effective TTL = %d, want 1 (cached)", got)
	}
	// Past the effective TTL: re-resolved even though the retention budget
	// has not elapsed.
	now = now.Add(2 * time.Second) // t=31s
	if _, err := r.ResolveImage(context.Background(), "t1"); err != nil {
		t.Fatalf("ResolveImage past effective TTL: %v", err)
	}
	if got := store.callCount(); got != 2 {
		t.Fatalf("meta calls past effective TTL = %d, want 2 (re-resolved)", got)
	}
	// Physical retention bound: the entry cached at t=31s expires at t=61s
	// and is reaped by the next sweep, never outliving the 60s budget.
	now = now.Add(31 * time.Second) // t=62s
	r.sweepExpired()
	if cacheHas(r, "t1") {
		t.Fatal("entry physically retained past the 60s retention budget")
	}
}

// blockingExtractConfigReader pauses inside the first
// GetTenantExtractConfig call until released, letting tests interleave
// Evict with an in-flight resolve.
type blockingExtractConfigReader struct {
	row     meta.TenantExtractConfig
	entered chan struct{}
	release chan struct{}
	once    sync.Once
	mu      sync.Mutex
	calls   int
}

func (b *blockingExtractConfigReader) GetTenantExtractConfig(_ context.Context, _, _ string) (*meta.TenantExtractConfig, error) {
	b.mu.Lock()
	b.calls++
	b.mu.Unlock()
	b.once.Do(func() { close(b.entered) })
	<-b.release
	cp := b.row
	return &cp, nil
}

func (b *blockingExtractConfigReader) callCount() int {
	b.mu.Lock()
	defer b.mu.Unlock()
	return b.calls
}

// Concurrent cache misses for one tenant share one MetaDB resolution. This
// bounds a burst of worker retries to one control-plane read on each pod.
func TestExtractConfigResolverCoalescesConcurrentMisses(t *testing.T) {
	reader := &blockingExtractConfigReader{
		row:     enabledImageRow("t1"),
		entered: make(chan struct{}),
		release: make(chan struct{}),
	}
	r := newExtractConfigResolver(reader, &fakeExtractDecryptor{plaintext: "sk"}, nil, time.Minute)
	t.Cleanup(r.Close)

	const callers = 16
	start := make(chan struct{})
	results := make(chan error, callers)
	var ready sync.WaitGroup
	ready.Add(callers)
	for range callers {
		go func() {
			ready.Done()
			<-start
			_, err := r.ResolveImage(context.Background(), "t1")
			results <- err
		}()
	}
	ready.Wait()
	close(start)
	<-reader.entered
	close(reader.release)
	for range callers {
		if err := <-results; err != nil {
			t.Fatalf("ResolveImage: %v", err)
		}
	}
	if got := reader.callCount(); got != 1 {
		t.Fatalf("MetaDB reads = %d, want 1 for concurrent same-tenant misses", got)
	}
}

// Resolution errors are cached fail-closed for one second. The cached error
// never produces a runtime, and expiry retries MetaDB promptly.
func TestExtractConfigResolverFailureBackoff(t *testing.T) {
	store := &fakeTenantExtractConfigReader{rows: map[string]meta.TenantExtractConfig{}}
	store.setErr(errors.New("metadb unavailable"))
	r := newExtractConfigResolver(store, &fakeExtractDecryptor{}, defaultTestRuntime(), time.Minute)
	t.Cleanup(r.Close)

	now := time.Now()
	r.now = func() time.Time { return now }
	for i := 0; i < 2; i++ {
		res, err := r.ResolveImage(context.Background(), "t1")
		if err == nil || res.runtime.Extractor != nil {
			t.Fatalf("attempt %d: res=%+v err=%v, want fail-closed error", i, res, err)
		}
	}
	if got := store.callCount(); got != 1 {
		t.Fatalf("MetaDB reads during failure backoff = %d, want 1", got)
	}

	now = now.Add(time.Second + time.Millisecond)
	if _, err := r.ResolveImage(context.Background(), "t1"); err == nil {
		t.Fatal("ResolveImage after failure backoff returned nil error")
	}
	if got := store.callCount(); got != 2 {
		t.Fatalf("MetaDB reads after failure backoff = %d, want 2", got)
	}
}

func TestExtractConfigResolverEvictClearsFailureBackoff(t *testing.T) {
	store := &fakeTenantExtractConfigReader{rows: map[string]meta.TenantExtractConfig{}}
	store.setErr(errors.New("metadb unavailable"))
	r := newExtractConfigResolver(store, &fakeExtractDecryptor{}, defaultTestRuntime(), time.Minute)
	t.Cleanup(r.Close)

	if _, err := r.ResolveImage(context.Background(), "t1"); err == nil {
		t.Fatal("ResolveImage returned nil error")
	}
	store.setErr(nil)
	if _, err := r.ResolveImage(context.Background(), "t1"); err == nil {
		t.Fatal("ResolveImage before eviction ignored cached failure")
	}
	if got := store.callCount(); got != 1 {
		t.Fatalf("MetaDB reads before eviction = %d, want 1", got)
	}

	r.Evict("t1")
	if _, err := r.ResolveImage(context.Background(), "t1"); err != nil {
		t.Fatalf("ResolveImage after eviction: %v", err)
	}
	if got := store.callCount(); got != 2 {
		t.Fatalf("MetaDB reads after eviction = %d, want 2", got)
	}
}

// An Evict that lands while a resolve is in flight must fence the stale
// result. The fence is intentionally process-wide so it uses constant space:
// an unrelated config write may make this caller retry, but no per-tenant
// generation bookkeeping accumulates.
func TestExtractConfigResolverEvictFencesInFlightResolve(t *testing.T) {
	reader := &blockingExtractConfigReader{
		row:     enabledImageRow("t1"),
		entered: make(chan struct{}),
		release: make(chan struct{}),
	}
	r := newExtractConfigResolver(reader, &fakeExtractDecryptor{plaintext: "sk"}, nil, time.Minute)
	t.Cleanup(r.Close)

	type resolveOutcome struct {
		res resolvedImageExtract
		err error
	}
	done := make(chan resolveOutcome, 1)
	go func() {
		res, err := r.ResolveImage(context.Background(), "t1")
		done <- resolveOutcome{res, err}
	}()
	<-reader.entered // the resolve is now in flight
	r.Evict("t2")    // an unrelated admin PUT/delete commits mid-resolve
	close(reader.release)

	out := <-done
	if !errors.Is(out.err, errExtractConfigConcurrentUpdate) {
		t.Fatalf("paused caller err = %v, want errExtractConfigConcurrentUpdate", out.err)
	}
	if out.res.runtime.Extractor != nil {
		t.Fatal("paused caller returned the stale runtime")
	}
	if cacheHas(r, "t1") {
		t.Fatal("stale in-flight result was cached after Evict")
	}

	// The next resolve runs against the new generation and caches normally.
	res, err := r.ResolveImage(context.Background(), "t1")
	if err != nil {
		t.Fatalf("ResolveImage after fenced resolve: %v", err)
	}
	if res.state != extractResolveReady {
		t.Fatalf("state = %v, want ready", res.state)
	}
	if !cacheHas(r, "t1") {
		t.Fatal("fresh resolve was not cached")
	}
}

// cacheHas reports whether tenantID has an entry (expired or not) in the
// resolver cache map.
func cacheHas(r *extractConfigResolver, tenantID string) bool {
	r.mu.Lock()
	defer r.mu.Unlock()
	_, ok := r.cache[tenantID]
	return ok
}

// Expired entries must be physically removed from the cache map (not just
// not-served) so plaintext provider keys do not live for the process
// lifetime; fresh entries survive the sweep.
func TestExtractConfigResolverJanitorSweepsExpiredEntries(t *testing.T) {
	store := &fakeTenantExtractConfigReader{rows: map[string]meta.TenantExtractConfig{
		"t1": enabledImageRow("t1"),
		"t2": enabledImageRow("t2"),
	}}
	r := newExtractConfigResolver(store, &fakeExtractDecryptor{plaintext: "sk"}, nil, time.Minute)
	t.Cleanup(r.Close)

	now := time.Now()
	r.now = func() time.Time { return now }
	if _, err := r.ResolveImage(context.Background(), "t1"); err != nil {
		t.Fatalf("resolve t1: %v", err)
	}
	// Advance past t1's TTL, then cache a fresh entry for t2. The expired t1
	// entry is no longer served but still sits in the map until the sweep.
	now = now.Add(2 * time.Minute)
	if _, err := r.ResolveImage(context.Background(), "t2"); err != nil {
		t.Fatalf("resolve t2: %v", err)
	}
	if !cacheHas(r, "t1") || !cacheHas(r, "t2") {
		t.Fatal("both entries should be cached before the sweep")
	}

	r.sweepExpired()
	if cacheHas(r, "t1") {
		t.Fatal("expired t1 entry still in the cache map after the sweep")
	}
	if !cacheHas(r, "t2") {
		t.Fatal("fresh t2 entry was swept")
	}
}

// Close must stop the janitor goroutine (WaitGroup + cancel pattern) and be
// safe to call repeatedly.
func TestExtractConfigResolverCloseStopsJanitor(t *testing.T) {
	r := newExtractConfigResolver(&fakeTenantExtractConfigReader{rows: map[string]meta.TenantExtractConfig{}},
		&fakeExtractDecryptor{}, nil, time.Minute)
	t.Cleanup(r.Close)

	done := make(chan struct{})
	go func() {
		r.Close()
		close(done)
	}()
	select {
	case <-done:
	case <-time.After(5 * time.Second):
		t.Fatal("Close did not return; janitor goroutine still running")
	}
	// Repeat Close is a no-op and must not block or panic.
	r.Close()
}

// Close must drop every cached resolution and reset the constant-space
// generation fence so cached plaintext provider keys are released at
// shutdown instead of lingering until process exit.
func TestExtractConfigResolverCloseClearsCache(t *testing.T) {
	store := &fakeTenantExtractConfigReader{rows: map[string]meta.TenantExtractConfig{"t1": enabledImageRow("t1")}}
	r := newExtractConfigResolver(store, &fakeExtractDecryptor{plaintext: "sk"}, nil, time.Minute)

	if _, err := r.ResolveImage(context.Background(), "t1"); err != nil {
		t.Fatalf("ResolveImage: %v", err)
	}
	// Advance the process-wide generation fence.
	r.Evict("t2")
	if !cacheHas(r, "t1") {
		t.Fatal("entry was not cached before Close")
	}

	r.Close()

	r.mu.Lock()
	defer r.mu.Unlock()
	if len(r.cache) != 0 {
		t.Fatalf("cache has %d entries after Close, want 0", len(r.cache))
	}
	if r.generation != 0 {
		t.Fatalf("generation = %d after Close, want 0", r.generation)
	}
}