package server

import (
	"context"
	"errors"
	"fmt"
	"net/http"
	"sync"
	"time"

	"github.com/mem9-ai/drive9/pkg/backend"
	"github.com/mem9-ai/drive9/pkg/meta"
	"golang.org/x/sync/singleflight"
)

// defaultExtractConfigRetentionTTL is the default worst-case physical
// retention budget for a resolved per-tenant extract config (including its
// plaintext provider key). The effective cache TTL and the janitor sweep
// interval are both half of it, so an entry is physically reaped at most
// one retention budget after it was cached. Config changes take effect
// within one effective TTL window.
const defaultExtractConfigRetentionTTL = time.Minute

// extractConfigFailureBackoff bounds how often one pod retries a failed
// resolution for the same tenant. It covers the semantic worker's initial
// retry burst without delaying recovery for a normal cache-TTL window.
const extractConfigFailureBackoff = time.Second

const (
	// Built-in task limits for tenant runtimes when no process default
	// exists to inherit them from. They mirror the backend's env defaults.
	// The fallback timeout stays configurable per process: ConfigureExtractResolver
	// overrides it from DRIVE9_IMAGE_EXTRACT_TIMEOUT_SECONDS.
	defaultTenantImageExtractTimeout      = 120 * time.Second
	defaultTenantImageExtractMaxImageByte = int64(8 << 20) // 8 MiB
	defaultTenantAudioExtractTimeout      = 2 * time.Minute
	defaultTenantAudioExtractMaxBytes     = int64(32 << 20)
	defaultTenantQwenASRMaxAudioBytes     = int64(7 << 20)
	defaultTenantVideoExtractTimeout      = 2 * time.Minute

	// defaultTenantImageExtractValidationTimeout bounds the small provider
	// probe sent on admin extract-config PUT. It is deliberately shorter than
	// the extraction timeout: the validation image is tiny and a config write
	// must not hang for the full extraction budget.
	defaultTenantImageExtractValidationTimeout = 20 * time.Second
)

// extractResolveState is the outcome of resolving a tenant's extract config.
type extractResolveState int

const (
	extractResolveUnknown extractResolveState = iota
	extractResolveReady
	extractResolveDisabled      // tenant off-marker
	extractResolveNotConfigured // row absent AND no process default
)

// resolvedImageExtract is the effective image extract configuration for one
// tenant. runtime is valid only when state == extractResolveReady.
type resolvedImageExtract struct {
	state   extractResolveState
	runtime backend.ImageExtractRuntime
}

type resolvedAudioExtract struct {
	state   extractResolveState
	runtime backend.AudioExtractRuntime
}

type resolvedVideoExtract struct {
	state   extractResolveState
	runtime backend.VideoExtractRuntime
}

// tenantExtractConfigReader is the slice of the meta store the resolver reads.
// *meta.Store satisfies it; tests use an in-memory fake.
type tenantExtractConfigReader interface {
	GetTenantExtractConfig(ctx context.Context, tenantID, mediaType string) (*meta.TenantExtractConfig, error)
}

// extractKeyDecryptor decrypts persisted tenant extractor API keys.
// The tenant pool satisfies it.
type extractKeyDecryptor interface {
	Decrypt(ctx context.Context, cipher []byte) ([]byte, error)
}

// errExtractConfigConcurrentUpdate is returned by ResolveImage when Evict
// ran while the resolution was in flight: the just-resolved config may be
// stale, so it is discarded — neither cached nor returned to the caller.
// The error is retryable: the next call re-resolves against the new
// generation. (Chosen over transparently re-resolving once, so an Evict
// storm cannot multiply meta reads and the caller decides the retry
// cadence; task workers already treat resolve errors as retryable.)
var errExtractConfigConcurrentUpdate = errors.New("extract config changed during resolution")

type extractConfigCacheEntry struct {
	resolved  any
	err       error
	expiresAt time.Time
}

// extractConfigResolver resolves the effective image extract runtime per
// tenant. It is fail-closed: the process default is used only when the meta
// store confirms the config row is absent (meta.ErrNotFound); any other
// failure — meta read, decrypt, or extractor build — returns an error and no
// runtime. Successful resolutions are cached per tenant for a short
// effective TTL (half the retention budget, 30s by default); expired
// entries are never served when re-resolution fails. A janitor goroutine
// sweeps expired entries on the same interval, so worst-case physical
// retention — plaintext provider keys included — stays within the retention
// budget (60s by default); Close stops it. Admin writes and
// local tenant lifecycle transitions evict the tenant's entry immediately.
// Concurrent misses for one tenant are coalesced, and a process-wide
// generation fence prevents an in-flight resolution that raced an eviction
// from re-caching stale state without retaining an unbounded tenant-ID map.
type extractConfigResolver struct {
	store                        tenantExtractConfigReader
	dec                          extractKeyDecryptor
	def                          *backend.ImageExtractRuntime
	defAudio                     *backend.AudioExtractRuntime
	defVideo                     *backend.VideoExtractRuntime
	imageMaxSourceBytes          int64
	audioMaxSourceBytes          int64
	videoMaxSourceBytes          int64
	videoMaxProviderPayloadBytes int64
	audioTenantFallbackTimeout   time.Duration
	videoTenantFallbackTimeout   time.Duration
	workspace                    *backend.MediaExtractWorkspace
	// ttl is the effective cache TTL: half the retention budget passed to
	// the constructor. The janitor sweeps on the same interval, so
	// worst-case physical retention is ttl (serve) + ttl (sweep) = the
	// retention budget.
	ttl time.Duration
	// tenantFallbackTimeout is the extraction timeout for tenant runtimes
	// when no process default exists to inherit from. ConfigureExtractResolver
	// overrides it from process configuration; it defaults to
	// defaultTenantImageExtractTimeout.
	tenantFallbackTimeout time.Duration

	// now and newOpenAIExtractor are replaceable in tests (fake clock and
	// composed-config capture without network access).
	now                func() time.Time
	newOpenAIExtractor func(cfg backend.OpenAIImageTextExtractorConfig) (backend.ImageTextExtractor, error)
	tenantHTTPClient   *http.Client

	mu         sync.Mutex
	cache      map[string]extractConfigCacheEntry
	audioCache map[string]extractConfigCacheEntry
	videoCache map[string]extractConfigCacheEntry
	// generation is process-wide: an eviction may make an unrelated in-flight
	// resolve retry, but config writes are rare and the fence uses constant
	// memory instead of retaining every evicted tenant ID.
	generation uint64
	flight     singleflight.Group

	janitorCancel context.CancelFunc
	janitorWG     sync.WaitGroup
	closeOnce     sync.Once
}

// extractConfigJanitorMinInterval floors the janitor sweep interval so
// pathological sub-second TTLs (tests) cannot spin the sweep loop.
const extractConfigJanitorMinInterval = time.Second

// newExtractConfigResolver builds a resolver. def is the process-default
// image extract runtime; nil means there is no process default and absent
// rows resolve to extractResolveNotConfigured. retentionTTL is the
// worst-case physical retention budget for cached entries (plaintext
// provider keys included); a non-positive value uses
// defaultExtractConfigRetentionTTL. The effective cache TTL and the janitor
// sweep interval are both derived as retentionTTL/2, so an expired entry is
// physically reaped within the budget. The cache janitor starts with the
// resolver; call Close to stop it.
func newExtractConfigResolver(store tenantExtractConfigReader, dec extractKeyDecryptor, def *backend.ImageExtractRuntime, retentionTTL time.Duration, mediaDefaults ...ExtractResolverDefaults) *extractConfigResolver {
	return newExtractConfigResolverWithFallback(store, dec, def, retentionTTL, 0, mediaDefaults...)
}

// newExtractConfigResolverWithFallback is newExtractConfigResolver with an
// explicit tenant fallback timeout (extraction timeout for tenant runtimes
// when no process default exists); non-positive values use
// defaultTenantImageExtractTimeout.
func newExtractConfigResolverWithFallback(store tenantExtractConfigReader, dec extractKeyDecryptor, def *backend.ImageExtractRuntime, retentionTTL time.Duration, tenantFallbackTimeout time.Duration, mediaDefaults ...ExtractResolverDefaults) *extractConfigResolver {
	if retentionTTL <= 0 {
		retentionTTL = defaultExtractConfigRetentionTTL
	}
	if tenantFallbackTimeout <= 0 {
		tenantFallbackTimeout = defaultTenantImageExtractTimeout
	}
	r := &extractConfigResolver{
		store: store,
		dec:   dec,
		def:   def,
		ttl:   retentionTTL / 2,
		now:   time.Now,
		newOpenAIExtractor: func(cfg backend.OpenAIImageTextExtractorConfig) (backend.ImageTextExtractor, error) {
			return backend.NewOpenAIImageTextExtractor(cfg)
		},
		tenantFallbackTimeout: tenantFallbackTimeout,
		cache:                 make(map[string]extractConfigCacheEntry),
		audioCache:            make(map[string]extractConfigCacheEntry),
		videoCache:            make(map[string]extractConfigCacheEntry),
	}
	if len(mediaDefaults) > 0 {
		r.defAudio = mediaDefaults[0].Audio
		r.defVideo = mediaDefaults[0].Video
		r.workspace = mediaDefaults[0].Workspace
		r.imageMaxSourceBytes = mediaDefaults[0].ImageMaxSourceBytes
		r.audioMaxSourceBytes = mediaDefaults[0].AudioMaxSourceBytes
		r.videoMaxSourceBytes = mediaDefaults[0].VideoMaxSourceBytes
		r.videoMaxProviderPayloadBytes = mediaDefaults[0].VideoMaxProviderPayloadBytes
		r.audioTenantFallbackTimeout = mediaDefaults[0].AudioTenantFallbackTimeout
		r.videoTenantFallbackTimeout = mediaDefaults[0].VideoTenantFallbackTimeout
	}
	if r.imageMaxSourceBytes <= 0 {
		r.imageMaxSourceBytes = backend.DefaultImageExtractMaxSourceBytes
	}
	if r.audioMaxSourceBytes <= 0 {
		r.audioMaxSourceBytes = backend.DefaultAudioExtractMaxSourceBytes
	}
	if r.videoMaxSourceBytes <= 0 {
		r.videoMaxSourceBytes = backend.DefaultVideoExtractMaxSourceBytes
	}
	if r.videoMaxProviderPayloadBytes <= 0 {
		r.videoMaxProviderPayloadBytes = backend.DefaultVideoExtractMaxProviderPayloadBytes
	}
	if r.audioTenantFallbackTimeout <= 0 {
		r.audioTenantFallbackTimeout = defaultTenantAudioExtractTimeout
	}
	if r.videoTenantFallbackTimeout <= 0 {
		r.videoTenantFallbackTimeout = defaultTenantVideoExtractTimeout
	}
	// Request contexts retain their media-specific deadlines. The shared safe
	// client uses the longest built-in tenant timeout so audio/video are not
	// accidentally capped by the shorter image default.
	clientTimeout := max(r.audioTenantFallbackTimeout, r.videoTenantFallbackTimeout, tenantFallbackTimeout)
	if r.def != nil {
		clientTimeout = max(clientTimeout, r.def.Timeout)
	}
	if r.defAudio != nil {
		clientTimeout = max(clientTimeout, r.defAudio.Timeout)
	}
	if r.defVideo != nil {
		clientTimeout = max(clientTimeout, r.defVideo.Timeout)
	}
	r.tenantHTTPClient = backend.NewTenantExtractorHTTPClient(clientTimeout)
	janitorCtx, cancel := context.WithCancel(context.Background())
	r.janitorCancel = cancel
	r.janitorWG.Add(1)
	go r.janitorLoop(janitorCtx)
	return r
}

// janitorLoop sweeps expired cache entries on the effective-TTL interval
// (half the retention budget) until Close.
func (r *extractConfigResolver) janitorLoop(ctx context.Context) {
	defer r.janitorWG.Done()
	ticker := time.NewTicker(max(r.ttl, extractConfigJanitorMinInterval))
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			r.sweepExpired()
		}
	}
}

// sweepExpired removes every cache entry whose TTL has passed so plaintext
// provider keys are reaped instead of living for the process lifetime.
func (r *extractConfigResolver) sweepExpired() {
	now := r.now()
	r.mu.Lock()
	defer r.mu.Unlock()
	for tenantID, ent := range r.cache {
		if !now.Before(ent.expiresAt) {
			delete(r.cache, tenantID)
		}
	}
	for tenantID, ent := range r.audioCache {
		if !now.Before(ent.expiresAt) {
			delete(r.audioCache, tenantID)
		}
	}
	for tenantID, ent := range r.videoCache {
		if !now.Before(ent.expiresAt) {
			delete(r.videoCache, tenantID)
		}
	}
}

// Evict drops tenantID's cached resolution on this pod, e.g. right after an
// admin PUT rewrote the row or a local tenant lifecycle transition ran. It
// also bumps the process generation under the same lock, so a resolve that
// was already in flight when the write committed discards its stale result
// instead of re-caching it. Nil-safe so callers need not check whether a
// resolver is wired.
func (r *extractConfigResolver) Evict(tenantID string) {
	if r == nil {
		return
	}
	r.mu.Lock()
	r.generation++
	delete(r.cache, tenantID)
	delete(r.audioCache, tenantID)
	delete(r.videoCache, tenantID)
	r.mu.Unlock()
}

// Close stops the cache janitor, waits for it to exit, then clears the cache
// and resets the generation fence so cached plaintext provider keys are
// released at shutdown. Workers are already stopped before the server calls
// this, so no further synchronization is needed. Nil- and repeat-safe.
func (r *extractConfigResolver) Close() {
	if r == nil {
		return
	}
	r.closeOnce.Do(func() {
		r.janitorCancel()
		r.janitorWG.Wait()
		r.tenantHTTPClient.CloseIdleConnections()
		r.mu.Lock()
		clear(r.cache)
		clear(r.audioCache)
		clear(r.videoCache)
		r.generation = 0
		r.mu.Unlock()
	})
}

// ResolveImage resolves the effective image extract runtime for tenantID.
// Fail-closed: only a confirmed meta.ErrNotFound falls back to the process
// default; any other error is returned with no runtime.
func (r *extractConfigResolver) ResolveImage(ctx context.Context, tenantID string) (resolvedImageExtract, error) {
	if res, err, ok := r.cached(tenantID); ok {
		return res, err
	}
	v, err, _ := r.flight.Do(tenantID, func() (any, error) {
		// A concurrent leader may have filled the cache before this caller
		// joined the flight.
		if res, err, ok := r.cached(tenantID); ok {
			return res, err
		}
		r.mu.Lock()
		generation := r.generation
		r.mu.Unlock()

		res, resolveErr := r.resolveImage(ctx, tenantID)
		r.mu.Lock()
		defer r.mu.Unlock()
		if r.generation != generation {
			return resolvedImageExtract{}, errExtractConfigConcurrentUpdate
		}
		expiresAt := r.now().Add(r.ttl)
		if resolveErr != nil {
			// Cancellation belongs to this execution context, not to the tenant
			// configuration, so it must not poison later callers.
			if errors.Is(resolveErr, context.Canceled) || errors.Is(resolveErr, context.DeadlineExceeded) {
				return resolvedImageExtract{}, resolveErr
			}
			expiresAt = r.now().Add(extractConfigFailureBackoff)
		}
		r.cache[tenantID] = extractConfigCacheEntry{resolved: res, err: resolveErr, expiresAt: expiresAt}
		return res, resolveErr
	})
	if err != nil {
		return resolvedImageExtract{}, err
	}
	return v.(resolvedImageExtract), nil
}

// ResolveAudio resolves the tenant audio runtime at worker execution time.
func (r *extractConfigResolver) ResolveAudio(ctx context.Context, tenantID string) (resolvedAudioExtract, error) {
	if ent, ok := r.cachedMedia(r.audioCache, tenantID); ok {
		if ent.err != nil {
			return resolvedAudioExtract{}, ent.err
		}
		return ent.resolved.(resolvedAudioExtract), nil
	}
	v, err, _ := r.flight.Do("audio\x00"+tenantID, func() (any, error) {
		if ent, ok := r.cachedMedia(r.audioCache, tenantID); ok {
			if ent.err != nil {
				return resolvedAudioExtract{}, ent.err
			}
			return ent.resolved, nil
		}
		r.mu.Lock()
		generation := r.generation
		r.mu.Unlock()
		res, resolveErr := r.resolveAudio(ctx, tenantID)
		r.mu.Lock()
		defer r.mu.Unlock()
		if r.generation != generation {
			return resolvedAudioExtract{}, errExtractConfigConcurrentUpdate
		}
		expires := r.now().Add(r.ttl)
		if resolveErr != nil && !errors.Is(resolveErr, context.Canceled) && !errors.Is(resolveErr, context.DeadlineExceeded) {
			expires = r.now().Add(extractConfigFailureBackoff)
		}
		if resolveErr == nil || (!errors.Is(resolveErr, context.Canceled) && !errors.Is(resolveErr, context.DeadlineExceeded)) {
			r.audioCache[tenantID] = extractConfigCacheEntry{resolved: res, err: resolveErr, expiresAt: expires}
		}
		return res, resolveErr
	})
	if err != nil {
		return resolvedAudioExtract{}, err
	}
	return v.(resolvedAudioExtract), nil
}

// ResolveVideo resolves the tenant video runtime at worker execution time.
func (r *extractConfigResolver) ResolveVideo(ctx context.Context, tenantID string) (resolvedVideoExtract, error) {
	if ent, ok := r.cachedMedia(r.videoCache, tenantID); ok {
		if ent.err != nil {
			return resolvedVideoExtract{}, ent.err
		}
		return ent.resolved.(resolvedVideoExtract), nil
	}
	v, err, _ := r.flight.Do("video\x00"+tenantID, func() (any, error) {
		if ent, ok := r.cachedMedia(r.videoCache, tenantID); ok {
			if ent.err != nil {
				return resolvedVideoExtract{}, ent.err
			}
			return ent.resolved, nil
		}
		r.mu.Lock()
		generation := r.generation
		r.mu.Unlock()
		res, resolveErr := r.resolveVideo(ctx, tenantID)
		r.mu.Lock()
		defer r.mu.Unlock()
		if r.generation != generation {
			return resolvedVideoExtract{}, errExtractConfigConcurrentUpdate
		}
		expires := r.now().Add(r.ttl)
		if resolveErr != nil && !errors.Is(resolveErr, context.Canceled) && !errors.Is(resolveErr, context.DeadlineExceeded) {
			expires = r.now().Add(extractConfigFailureBackoff)
		}
		if resolveErr == nil || (!errors.Is(resolveErr, context.Canceled) && !errors.Is(resolveErr, context.DeadlineExceeded)) {
			r.videoCache[tenantID] = extractConfigCacheEntry{resolved: res, err: resolveErr, expiresAt: expires}
		}
		return res, resolveErr
	})
	if err != nil {
		return resolvedVideoExtract{}, err
	}
	return v.(resolvedVideoExtract), nil
}

func (r *extractConfigResolver) cachedMedia(cache map[string]extractConfigCacheEntry, tenantID string) (extractConfigCacheEntry, bool) {
	r.mu.Lock()
	defer r.mu.Unlock()
	ent, ok := cache[tenantID]
	if !ok || !r.now().Before(ent.expiresAt) {
		return extractConfigCacheEntry{}, false
	}
	return ent, true
}

func (r *extractConfigResolver) resolveAudio(ctx context.Context, tenantID string) (resolvedAudioExtract, error) {
	row, err := r.store.GetTenantExtractConfig(ctx, tenantID, meta.ExtractMediaTypeAudio)
	if err != nil {
		if errors.Is(err, meta.ErrNotFound) {
			if r.defAudio != nil {
				return resolvedAudioExtract{state: extractResolveReady, runtime: *r.defAudio}, nil
			}
			return resolvedAudioExtract{state: extractResolveNotConfigured}, nil
		}
		return resolvedAudioExtract{}, fmt.Errorf("get audio extract config for tenant %s: %w", tenantID, err)
	}
	if !row.Enabled {
		return resolvedAudioExtract{state: extractResolveDisabled}, nil
	}
	key, err := r.dec.Decrypt(ctx, row.APIKeyCipher)
	if err != nil {
		return resolvedAudioExtract{}, fmt.Errorf("decrypt tenant %s audio api key: %w", tenantID, err)
	}
	if err := backend.ValidateTenantAPIBase(row.APIBase); err != nil {
		return resolvedAudioExtract{}, fmt.Errorf("validate tenant %s audio api base: %w", tenantID, err)
	}
	extractor, err := backend.NewAudioTextExtractorForProtocol(row.Protocol, backend.AudioTextExtractorConfig{
		BaseURL: row.APIBase, APIKey: string(key), Model: row.Model, Prompt: row.Prompt,
		Timeout: r.audioTenantFallbackTimeout, Client: r.tenantHTTPClient,
	})
	if err != nil {
		return resolvedAudioExtract{}, fmt.Errorf("build tenant %s audio extractor: %w", tenantID, err)
	}
	maxAudioBytes := defaultTenantAudioExtractMaxBytes
	if backend.NormalizeMediaProviderProtocol(row.Protocol) == backend.MediaProviderProtocolQwenASR {
		maxAudioBytes = defaultTenantQwenASRMaxAudioBytes
	}
	return resolvedAudioExtract{state: extractResolveReady, runtime: backend.AudioExtractRuntime{Extractor: extractor, Timeout: r.audioTenantFallbackTimeout, MaxSourceBytes: r.audioMaxSourceBytes, MaxAudioBytes: maxAudioBytes, MaxExtractTextBytes: backend.DefaultAudioExtractMaxTextBytes, Workspace: r.workspace}}, nil
}

func (r *extractConfigResolver) resolveVideo(ctx context.Context, tenantID string) (resolvedVideoExtract, error) {
	row, err := r.store.GetTenantExtractConfig(ctx, tenantID, meta.ExtractMediaTypeVideo)
	if err != nil {
		if errors.Is(err, meta.ErrNotFound) {
			if r.defVideo == nil {
				return resolvedVideoExtract{state: extractResolveNotConfigured}, nil
			}
			runtime := *r.defVideo
			audio, audioErr := r.ResolveAudio(ctx, tenantID)
			if audioErr != nil {
				return resolvedVideoExtract{}, audioErr
			}
			if audio.state == extractResolveReady {
				runtime.AudioRuntime = &audio.runtime
			} else {
				runtime.AudioRuntime = nil
			}
			return resolvedVideoExtract{state: extractResolveReady, runtime: runtime}, nil
		}
		return resolvedVideoExtract{}, fmt.Errorf("get video extract config for tenant %s: %w", tenantID, err)
	}
	if !row.Enabled {
		return resolvedVideoExtract{state: extractResolveDisabled}, nil
	}
	key, err := r.dec.Decrypt(ctx, row.APIKeyCipher)
	if err != nil {
		return resolvedVideoExtract{}, fmt.Errorf("decrypt tenant %s video api key: %w", tenantID, err)
	}
	if err := backend.ValidateTenantAPIBase(row.APIBase); err != nil {
		return resolvedVideoExtract{}, fmt.Errorf("validate tenant %s video api base: %w", tenantID, err)
	}
	videoExtractor, err := backend.NewVideoTextExtractorForProtocol(row.Protocol, backend.OpenAIVideoTextExtractorConfig{
		BaseURL: row.APIBase, APIKey: string(key), Model: row.Model, Prompt: row.Prompt,
		Timeout: r.videoTenantFallbackTimeout, Client: r.tenantHTTPClient,
		MaxProviderPayloadBytes: r.videoMaxProviderPayloadBytes,
	})
	if err != nil {
		return resolvedVideoExtract{}, fmt.Errorf("build tenant %s video extractor: %w", tenantID, err)
	}
	runtime := backend.VideoExtractRuntime{Extractor: videoExtractor, Timeout: r.videoTenantFallbackTimeout, MaxSourceBytes: r.videoMaxSourceBytes, MaxExtractTextBytes: backend.DefaultVideoExtractMaxTextBytes, Workspace: r.workspace}
	audio, err := r.ResolveAudio(ctx, tenantID)
	if err != nil {
		return resolvedVideoExtract{}, fmt.Errorf("resolve tenant %s video audio runtime: %w", tenantID, err)
	}
	if audio.state == extractResolveReady {
		runtime.AudioRuntime = &audio.runtime
	}
	return resolvedVideoExtract{state: extractResolveReady, runtime: runtime}, nil
}

func (r *extractConfigResolver) cached(tenantID string) (resolvedImageExtract, error, bool) {
	r.mu.Lock()
	defer r.mu.Unlock()
	ent, ok := r.cache[tenantID]
	if !ok || !r.now().Before(ent.expiresAt) {
		return resolvedImageExtract{}, nil, false
	}
	res, ok := ent.resolved.(resolvedImageExtract)
	if !ok {
		return resolvedImageExtract{}, fmt.Errorf("cached image extract runtime has invalid type"), true
	}
	return res, ent.err, true
}

func (r *extractConfigResolver) resolveImage(ctx context.Context, tenantID string) (resolvedImageExtract, error) {
	row, err := r.store.GetTenantExtractConfig(ctx, tenantID, meta.ExtractMediaTypeImage)
	if err != nil {
		if errors.Is(err, meta.ErrNotFound) {
			return r.defaultResolution(), nil
		}
		return resolvedImageExtract{}, fmt.Errorf("get image extract config for tenant %s: %w", tenantID, err)
	}
	if !row.Enabled {
		return resolvedImageExtract{state: extractResolveDisabled}, nil
	}
	rt, err := r.buildTenantRuntime(ctx, row)
	if err != nil {
		return resolvedImageExtract{}, err
	}
	return resolvedImageExtract{state: extractResolveReady, runtime: rt}, nil
}

func (r *extractConfigResolver) defaultResolution() resolvedImageExtract {
	if r.def == nil {
		return resolvedImageExtract{state: extractResolveNotConfigured}
	}
	runtime := *r.def
	runtime.Workspace = r.workspace
	return resolvedImageExtract{state: extractResolveReady, runtime: runtime}
}

// buildTenantRuntime validates and builds the runtime for an enabled tenant
// row. Every failure (invalid api_base, decrypt error, extractor build
// error) is returned as an error; the caller must not fall back to the
// process default.
func (r *extractConfigResolver) buildTenantRuntime(ctx context.Context, row *meta.TenantExtractConfig) (backend.ImageExtractRuntime, error) {
	if err := backend.ValidateTenantAPIBase(row.APIBase); err != nil {
		return backend.ImageExtractRuntime{}, fmt.Errorf("validate tenant %s extractor api base: %w", row.TenantID, err)
	}
	key, err := r.dec.Decrypt(ctx, row.APIKeyCipher)
	if err != nil {
		return backend.ImageExtractRuntime{}, fmt.Errorf("decrypt tenant %s extractor api key: %w", row.TenantID, err)
	}
	extractor, err := r.newOpenAIExtractor(backend.OpenAIImageTextExtractorConfig{
		BaseURL: row.APIBase,
		APIKey:  string(key),
		Model:   row.Model,
		Prompt:  backend.ComposeImageExtractPrompt(row.Prompt),
		Client:  r.tenantHTTPClient,
	})
	if err != nil {
		return backend.ImageExtractRuntime{}, fmt.Errorf("build tenant %s image extractor: %w", row.TenantID, err)
	}
	timeout, maxImageBytes, maxExtractTextBytes := r.runtimeLimits()
	return backend.ImageExtractRuntime{
		// Tenant custom providers are OpenAI-compatible endpoints that accept
		// exactly jpeg/png/gif/webp; keep the magic sniff on so junk declared
		// as an image never reaches the provider.
		Extractor:               backend.NewFallbackImageTextExtractor(extractor, backend.NewBasicImageTextExtractor()),
		Timeout:                 timeout,
		MaxSourceBytes:          r.imageMaxSourceBytes,
		MaxImageBytes:           maxImageBytes,
		MaxExtractTextBytes:     maxExtractTextBytes,
		RequireRasterImageBytes: true,
		Workspace:               r.workspace,
	}, nil
}

// ExtractDefaultProvider carries the process default's OpenAI-compatible
// provider fields for admin GET rendering. APIKey is plaintext here; handlers
// mask it before rendering.
type ExtractDefaultProvider struct {
	APIBase string
	APIKey  string
	Model   string
	Prompt  string
}

// ExtractResolverDefaults contains the process-default runtimes used only
// after the meta store confirms that the corresponding tenant row is absent.
type ExtractResolverDefaults struct {
	Image                        *backend.ImageExtractRuntime
	Audio                        *backend.AudioExtractRuntime
	AudioProtocol                string
	Video                        *backend.VideoExtractRuntime
	ImageMaxSourceBytes          int64
	AudioMaxSourceBytes          int64
	VideoMaxSourceBytes          int64
	VideoMaxProviderPayloadBytes int64
	AudioTenantFallbackTimeout   time.Duration
	VideoTenantFallbackTimeout   time.Duration
	Workspace                    *backend.MediaExtractWorkspace
	ImageTenantFallbackTimeout   time.Duration
}

// defaultImageExtractView is the admin-facing view of the process default.
// It belongs to Server presentation state rather than the runtime resolver.
type defaultImageExtractView struct {
	configured bool
	provider   bool
	apiBase    string
	apiKey     string // plaintext; handlers must mask before rendering
	model      string
	prompt     string
	protocol   string
}

// ConfigureExtractResolver builds the per-tenant extract config resolver from
// plain dependencies and wires it into the server and tenant worker. reader
// must implement GetTenantExtractConfig (satisfied by *meta.Store); dec must
// implement Decrypt (satisfied by *tenant.Pool). def is the process-default
// image extract runtime; nil means absent rows resolve to not_configured.
// defProvider describes the process default's provider fields for admin GET
// rendering; pass nil when the default is the basic extractor or when there is
// no default. tenantTimeout optionally overrides the extraction timeout used
// for tenant image runtimes when def is nil; non-positive values keep
// defaultTenantImageExtractTimeout. When def is set, tenant runtimes inherit
// its timeout instead.
func (s *Server) ConfigureExtractResolver(reader tenantExtractConfigReader, dec extractKeyDecryptor, def *backend.ImageExtractRuntime, defProvider *ExtractDefaultProvider, tenantTimeout ...time.Duration) {
	var fallbackTimeout time.Duration
	if len(tenantTimeout) > 0 {
		fallbackTimeout = tenantTimeout[0]
	}
	s.ConfigureExtractResolverDefaults(reader, dec, ExtractResolverDefaults{
		Image:                      def,
		ImageTenantFallbackTimeout: fallbackTimeout,
	}, defProvider)
}

// ConfigureExtractResolverDefaults wires all media defaults while retaining
// the existing image provider view contract.
func (s *Server) ConfigureExtractResolverDefaults(reader tenantExtractConfigReader, dec extractKeyDecryptor, defaults ExtractResolverDefaults, imageProvider *ExtractDefaultProvider) {
	r := newExtractConfigResolverWithFallback(reader, dec, defaults.Image, 0, defaults.ImageTenantFallbackTimeout, defaults)
	s.SetExtractConfigResolver(r)
	if defaults.Image != nil && imageProvider != nil {
		s.extractDefaultViews[meta.ExtractMediaTypeImage] = defaultImageExtractView{
			configured: true,
			provider:   true,
			apiBase:    imageProvider.APIBase,
			apiKey:     imageProvider.APIKey,
			model:      imageProvider.Model,
			prompt:     imageProvider.Prompt,
		}
	}
	if defaults.Audio != nil {
		s.extractDefaultViews[meta.ExtractMediaTypeAudio] = defaultImageExtractView{
			configured: true,
			protocol:   backend.NormalizeMediaProviderProtocol(defaults.AudioProtocol),
		}
	}
}

// runtimeLimits returns the task limits for a tenant runtime: inherited from
// the process default when one exists, otherwise the built-in defaults.
func (r *extractConfigResolver) runtimeLimits() (timeout time.Duration, maxImageBytes int64, maxExtractTextBytes int) {
	if r.def != nil {
		return r.def.Timeout, r.def.MaxImageBytes, r.def.MaxExtractTextBytes
	}
	return r.tenantFallbackTimeout, defaultTenantImageExtractMaxImageByte, backend.DefaultImageExtractMaxTextBytes
}