package backend

import (
	"fmt"
	"strings"
	"time"

	"github.com/mem9-ai/drive9/pkg/embedding"
	"github.com/mem9-ai/drive9/pkg/logger"
	"github.com/mem9-ai/drive9/pkg/meta"
	"github.com/mem9-ai/drive9/pkg/s3client"
	"go.uber.org/zap"
)

const (
	// DefaultObjectGCGracePeriod delays deletion long enough for recent
	// references to become visible before the object GC worker rechecks them.
	DefaultObjectGCGracePeriod = 24 * time.Hour
	defaultImageExtractMaxSize = int64(8 << 20) // 8 MiB
	// DefaultImageExtractMaxSourceBytes bounds one image staged on local disk.
	DefaultImageExtractMaxSourceBytes = int64(1 << 30) // 1 GiB
	defaultImageExtractTimeout        = 120 * time.Second
	// DefaultImageExtractMaxTextBytes is the default stored semantic text cap
	// for image extraction. The image writeback path also enforces an
	// estimated-token cap before this byte cap is applied.
	DefaultImageExtractMaxTextBytes = 32 << 10 // 32 KiB
	defaultMaxExtractedTextBytes    = DefaultImageExtractMaxTextBytes
	defaultAudioExtractMaxSize      = int64(32 << 20) // 32 MiB
	// DefaultAudioExtractMaxSourceBytes bounds one audio source staged on disk.
	DefaultAudioExtractMaxSourceBytes = int64(5 << 30) // 5 GiB
	defaultAudioExtractTimeout        = 2 * time.Minute
	defaultMaxAudioExtractedTextBytes = 8 << 10 // 8 KiB
	// DefaultVideoExtractMaxSourceBytes is the process-wide staging ceiling for
	// video extraction. Timeline duration and provider payloads have separate
	// bounds; this limit protects the disk-backed task workspace.
	DefaultVideoExtractMaxSourceBytes = int64(10 << 30) // 10 GiB
	// DefaultVideoExtractMaxProviderPayloadBytes bounds one vision request.
	DefaultVideoExtractMaxProviderPayloadBytes = int64(8 << 20) // 8 MiB
	defaultVideoExtractTimeout                 = 5 * time.Minute
	defaultMaxVideoExtractedTextBytes          = 32 << 10              // 32 KiB
	defaultMaxUploadBytes                      = int64(10 * (1 << 30)) // 10 GiB
	defaultMaxTenantStorageBytes               = int64(50 * (1 << 30)) // 50 GiB
	// defaultMaxMonthlyLLMCostMillicents is the per-tenant monthly LLM spend
	// cap applied when no explicit budget is configured. $10.00 USD, expressed
	// in millicents (0.001 cents; $10 = 1000 cents = 1_000_000 millicents).
	// This is a defense-in-depth floor, not a product pricing tier: high
	// enough for a reasonable trial workload (hundreds of images or a few
	// hours of audio) and low enough that a runaway tenant is noticed before
	// meaningful financial impact. Operators who need a higher baseline raise
	// this constant; operators who want to disable it globally pass a
	// negative MaxMonthlyMillicents in Options.LLMCostBudget. Per-tenant
	// overrides via meta.QuotaConfig.MaxMonthlyCostMC continue to win.
	defaultMaxMonthlyLLMCostMillicents = int64(1_000_000) // $10.00
)

// Options configures Dat9Backend behavior.
type Options struct {
	// ExtractTmpDir controls the disk-backed workspace used by video/audio
	// extraction and oversized image staging. Empty uses os.TempDir().
	ExtractTmpDir string
	// ExtractMaxConcurrentTasks bounds process-local media staging and
	// preprocessing. A non-positive value uses
	// DefaultMediaExtractMaxConcurrentTasks.
	ExtractMaxConcurrentTasks int
	AsyncImageExtract         AsyncImageExtractOptions
	// AsyncAudioExtract configures durable audio transcript extraction for the
	// TiDB auto-embedding path. Unlike async image extract, there is no in-process
	// queue; work is delivered only via semantic_tasks when runtime is wired.
	AsyncAudioExtract AsyncAudioExtractOptions
	AsyncVideoExtract AsyncVideoExtractOptions
	QueryEmbedding    QueryEmbeddingOptions
	MaxUploadBytes    int64
	// ObjectGCGracePeriod controls how long newly enqueued object GC candidates
	// wait before becoming eligible for reachability checks and deletion.
	// A non-positive value uses DefaultObjectGCGracePeriod.
	ObjectGCGracePeriod time.Duration
	// MaxTenantStorageBytes caps the total logical storage a single tenant may
	// occupy across confirmed files plus in-flight upload reservations.
	MaxTenantStorageBytes int64
	// DatabaseAutoEmbedding controls whether semantic text is embedded by the
	// database itself rather than by the app-managed embed worker. When enabled,
	// runtime write/query paths rely on database-side embedding behavior.
	DatabaseAutoEmbedding bool
	// AppSemanticTasksEnabled controls whether the app-managed embed worker
	// path may enqueue semantic_tasks for text and description embedding.
	// When false, shouldEnqueueEmbedForRevision short-circuits to false,
	// preventing orphaned task rows when no DRIVE9_EMBED_* worker is
	// configured. This does not affect the DatabaseAutoEmbedding (TiDB
	// auto) path or image/audio extract tasks.
	AppSemanticTasksEnabled bool
	// AppEmbeddingGeneration, AppEmbeddingConfigRevision, and
	// AppEmbeddingTenantManaged identify the effective app-managed provider
	// wired into QueryEmbedding.Client.
	AppEmbeddingGeneration     uint64
	AppEmbeddingConfigRevision uint64
	AppEmbeddingTenantManaged  bool
	// MaxMediaLLMFiles caps the number of confirmed image+audio files per tenant
	// that trigger LLM extraction tasks (img_extract_text, audio_extract_text).
	// Files beyond this limit are still stored but their LLM tasks are not enqueued.
	// Zero or negative means use the default (100).
	MaxMediaLLMFiles int64
	// LLMCostBudget configures the monthly LLM cost budget for this tenant.
	LLMCostBudget LLMCostBudgetOptions
	// TenantID is used for per-write S3 encryption context and audit metadata.
	TenantID string
	// TiDBCloudOrgID is used only as the tidbcloud_org_id label on tenant
	// metrics. Empty means "guest".
	TiDBCloudOrgID string
	// StorageNamespaceID is the control-plane namespace for S3 object lifecycle.
	StorageNamespaceID string
	// S3EncryptionPolicy is the already-resolved policy for this backend.
	// The zero value is normalized to the global default of explicit no encryption.
	S3EncryptionPolicy meta.ResolvedS3EncryptionPolicy
	// AppendLogS3 is the optional Directory Bucket client. It is separate from
	// S3Client so ordinary S3 routing and presigning remain unchanged.
	AppendLogS3 s3client.AppendLogClient
	// AppendLogWriteDisabled keeps inherited append-log objects readable while
	// refusing append-log mutations for a tenant that cannot preserve their COW
	// semantics (currently native tenant forks).
	AppendLogWriteDisabled bool
	// InlineThreshold overrides the DB-inline vs S3 storage cutoff. When 0,
	// DefaultInlineThreshold is used. The same value is surfaced via
	// /v1/status so clients can pick a matching upload strategy.
	InlineThreshold int64
	// TextExtractMaxBytes overrides the synchronous text extraction cap. When
	// 0, DefaultTextExtractMaxBytes is used. Independent of InlineThreshold.
	TextExtractMaxBytes int64
}

// LLMCostBudgetOptions configures the monthly LLM cost budget.
type LLMCostBudgetOptions struct {
	// MaxMonthlyMillicents is the monthly cost cap in millicents (0.001 cents).
	//
	// Tri-state:
	//   > 0  — explicit per-tenant cap in millicents.
	//   == 0 — unset; the default defense-in-depth cap
	//          (defaultMaxMonthlyLLMCostMillicents, currently $10) is applied.
	//   < 0  — explicit opt-out; disables the monthly cost budget gate.
	//
	// The zero-value meaning changed intentionally: leaving this field
	// unset no longer yields "unlimited". Operators that truly need no
	// monthly cap must pass a negative value. Per-tenant overrides via
	// meta.QuotaConfig.MaxMonthlyCostMC still win over this value.
	MaxMonthlyMillicents int64
	// VisionCostPerKTokenMillicents is the cost per 1K tokens for Vision API calls.
	VisionCostPerKTokenMillicents int64
	// AudioLLMCostPerKTokenMillicents is the cost per 1K tokens for token-based
	// audio models (e.g. gpt-4o-transcribe).
	AudioLLMCostPerKTokenMillicents int64
	// WhisperCostPerMinuteMillicents is the cost per minute for duration-based
	// audio models (e.g. whisper-1).
	WhisperCostPerMinuteMillicents int64
	// FallbackImageCostMillicents is used when the Vision API does not return
	// token usage. Must be > 0 for cost tracking to work with such providers.
	FallbackImageCostMillicents int64
	// FallbackAudioCostMillicents is used when the audio API returns neither
	// duration nor token usage. Must be > 0 for cost tracking to work.
	FallbackAudioCostMillicents int64
}

// AsyncImageExtractOptions controls async image->text extraction. Delivery is
// fully durable via semantic_tasks; the semantic worker claims and processes
// img_extract_text tasks using this backend's extractor and S3 client.
type AsyncImageExtractOptions struct {
	Enabled bool
	// Capability reports that a resolver-backed multi-tenant server can
	// execute img_extract_text tasks with an explicitly resolved runtime
	// (ProcessImageExtractTaskWithRuntime) even when no process-default
	// extractor is configured. It gates task enqueue and worker task-type
	// registration, not SupportsAsyncImageExtract.
	Capability bool
	// QueueSize and Workers are deprecated: image extraction no longer uses an
	// in-process channel. They remain for config compatibility but are ignored.
	QueueSize           int
	Workers             int
	MaxSourceBytes      int64
	MaxImageBytes       int64
	TaskTimeout         time.Duration
	MaxExtractTextBytes int
	Extractor           ImageTextExtractor
}

// AsyncImageExtractWillWireRuntime reports whether async image extraction will be
// wired on a Dat9Backend built from opts. When Enabled is true, configureOptions
// always assigns a concrete ImageTextExtractor (NewBasicImageTextExtractor when nil),
// so this matches effective SupportsAsyncImageExtract after backend construction.
// Capability alone also wires the runtime: the server resolves a per-tenant
// extractor at task-processing time instead of using a process default.
func AsyncImageExtractWillWireRuntime(opts AsyncImageExtractOptions) bool {
	return opts.Enabled || opts.Capability
}

// AsyncAudioExtractOptions configures audio transcript extraction for the database
// auto-embedding path. Delivery uses semantic_tasks only; no local worker queue.
type AsyncAudioExtractOptions struct {
	Enabled bool
	// Capability reports that a resolver-backed multi-tenant server can
	// resolve an audio runtime after claiming a durable task even when no
	// process-default extractor is configured.
	Capability          bool
	Protocol            string
	MaxSourceBytes      int64
	MaxAudioBytes       int64
	TaskTimeout         time.Duration
	MaxExtractTextBytes int
	Extractor           AudioTextExtractor
}

// AsyncAudioExtractWillWireRuntime reports whether async audio extraction can
// execute either through a process-default extractor or a tenant resolver.
func AsyncAudioExtractWillWireRuntime(opts AsyncAudioExtractOptions) bool {
	return (opts.Enabled && opts.Extractor != nil) || opts.Capability
}

// AsyncVideoExtractOptions configures video visual extraction via FFmpeg +
// Vision model. Delivery uses semantic_tasks only; no local worker queue.
type AsyncVideoExtractOptions struct {
	Enabled bool
	// Capability reports that a resolver-backed multi-tenant server can
	// resolve a video runtime after claiming a durable task.
	Capability              bool
	AllTenants              bool // true when allowlist is "*"
	MaxSourceBytes          int64
	MaxProviderPayloadBytes int64
	MaxVideoLLMFiles        int64 // per-tenant video-file extraction admission limit (default 30)
	TaskTimeout             time.Duration
	MaxExtractTextBytes     int
	Extractor               VideoTextExtractor
	// TenantAllowlist restricts video extraction to only the listed tenant
	// IDs. Ignored when AllTenants is true. An empty or nil map with
	// AllTenants false means no tenant is allowed (fail-closed).
	TenantAllowlist map[string]struct{}
}

// AsyncVideoExtractWillWireRuntime reports whether async video extraction can
// execute either through a process-default extractor or a tenant resolver.
func AsyncVideoExtractWillWireRuntime(opts AsyncVideoExtractOptions) bool {
	return (opts.Enabled && opts.Extractor != nil) || opts.Capability
}

// ParseVideoExtractTenantAllowlist parses a raw allowlist string into
// (allTenants, allowlist, error). Rules:
//   - empty → off (false, nil, nil)
//   - "*"   → all tenants (true, nil, nil)
//   - "a,b" → specific tenants (false, {"a":{},"b":{}}, nil)
//   - "*,a" or "a-*" → error (no mixing, no glob patterns)
func ParseVideoExtractTenantAllowlist(raw string) (allTenants bool, allowlist map[string]struct{}, err error) {
	raw = strings.TrimSpace(raw)
	if raw == "" {
		return false, nil, nil
	}
	allowlist = make(map[string]struct{})
	for _, tok := range strings.Split(raw, ",") {
		tok = strings.TrimSpace(tok)
		if tok == "" {
			continue
		}
		if tok == "*" {
			allTenants = true
		} else if strings.Contains(tok, "*") {
			return false, nil, fmt.Errorf("glob/prefix patterns not supported (got %q); use exact tenant IDs or \"*\" for all", tok)
		} else {
			allowlist[tok] = struct{}{}
		}
	}
	if allTenants && len(allowlist) > 0 {
		return false, nil, fmt.Errorf("\"*\" cannot be mixed with tenant IDs (got %q)", raw)
	}
	if allTenants {
		return true, nil, nil
	}
	// All tokens were empty (e.g. "," or ", ,") — treat as off.
	if len(allowlist) == 0 {
		return false, nil, nil
	}
	return false, allowlist, nil
}

// QueryEmbeddingOptions controls app-side query embedding for semantic search.
type QueryEmbeddingOptions struct {
	Client embedding.Client
}

// AppEmbeddingRuntime is the effective app-managed provider contract resolved
// once when a tenant backend is built. Generation gates vector visibility;
// ConfigRevision gates one durable backfill attempt for the current provider.
type AppEmbeddingRuntime struct {
	Client         embedding.Client
	Generation     uint64
	ConfigRevision uint64
	TenantManaged  bool
}

func (b *Dat9Backend) configureOptions(opts Options) {
	workspace, workspaceErr := NewMediaExtractWorkspace(MediaExtractWorkspaceOptions{
		RootDir:            opts.ExtractTmpDir,
		MaxConcurrentTasks: opts.ExtractMaxConcurrentTasks,
	})
	if workspaceErr != nil {
		logger.Warn(backgroundWithTrace(), "backend_media_extract_workspace_unavailable", zap.Error(workspaceErr))
	} else {
		b.mediaExtractWorkspace = workspace
	}
	if opts.TenantID != "" {
		b.tenantID = opts.TenantID
	}
	b.tidbCloudOrgID = normalizeTenantMetricTiDBCloudOrgID(opts.TiDBCloudOrgID)
	if opts.StorageNamespaceID != "" {
		b.storageNamespaceID = opts.StorageNamespaceID
	}
	b.databaseAutoEmbedding = opts.DatabaseAutoEmbedding
	b.appSemanticTasksEnabled = opts.AppSemanticTasksEnabled
	b.appEmbeddingGeneration = opts.AppEmbeddingGeneration
	b.appEmbeddingConfigRevision = opts.AppEmbeddingConfigRevision
	b.appEmbeddingTenantManaged = opts.AppEmbeddingTenantManaged
	b.s3EncryptionPolicy = opts.S3EncryptionPolicy
	b.appendLogS3 = opts.AppendLogS3
	b.appendLogWriteDisabled = opts.AppendLogWriteDisabled
	if b.s3EncryptionPolicy.Mode == "" {
		resolved, err := meta.ResolveS3EncryptionPolicy(meta.DefaultS3EncryptionPolicy(), meta.S3EncryptionPolicy{Mode: meta.S3EncryptionModeInherit})
		if err == nil {
			b.s3EncryptionPolicy = resolved
		}
	}
	if opts.MaxUploadBytes > 0 {
		b.maxUploadBytes = opts.MaxUploadBytes
	} else {
		b.maxUploadBytes = defaultMaxUploadBytes
	}
	if opts.ObjectGCGracePeriod > 0 {
		b.objectGCGracePeriod = opts.ObjectGCGracePeriod
	} else {
		b.objectGCGracePeriod = DefaultObjectGCGracePeriod
	}
	if opts.MaxMediaLLMFiles > 0 {
		b.maxMediaLLMFiles = opts.MaxMediaLLMFiles
	} else {
		b.maxMediaLLMFiles = meta.DefaultMaxMediaLLMFiles()
	}
	if opts.MaxTenantStorageBytes > 0 {
		b.maxTenantStorageBytes = opts.MaxTenantStorageBytes
	} else {
		b.maxTenantStorageBytes = defaultMaxTenantStorageBytes
	}
	if opts.InlineThreshold > 0 {
		b.inlineThreshold = opts.InlineThreshold
	} else {
		b.inlineThreshold = DefaultInlineThreshold
	}
	if opts.TextExtractMaxBytes > 0 {
		b.textExtractMaxBytes = opts.TextExtractMaxBytes
	} else {
		b.textExtractMaxBytes = DefaultTextExtractMaxBytes
	}

	cb := opts.LLMCostBudget
	switch {
	case cb.MaxMonthlyMillicents > 0:
		b.maxMonthlyLLMCostMillicents = cb.MaxMonthlyMillicents
	case cb.MaxMonthlyMillicents < 0:
		b.maxMonthlyLLMCostMillicents = 0 // explicit opt-out
	default:
		b.maxMonthlyLLMCostMillicents = defaultMaxMonthlyLLMCostMillicents
	}
	b.visionCostPerKTokenMillicents = cb.VisionCostPerKTokenMillicents
	b.audioLLMCostPerKTokenMillicents = cb.AudioLLMCostPerKTokenMillicents
	b.whisperCostPerMinuteMillicents = cb.WhisperCostPerMinuteMillicents
	b.fallbackImageCostMillicents = cb.FallbackImageCostMillicents
	b.fallbackAudioCostMillicents = cb.FallbackAudioCostMillicents

	if opts.QueryEmbedding.Client != nil {
		b.queryEmbedder = opts.QueryEmbedding.Client
	} else {
		b.queryEmbedder = embedding.NopClient{}
	}

	cfg := opts.AsyncImageExtract
	b.imageExtractCapability = cfg.Capability
	if cfg.Enabled {
		if cfg.MaxSourceBytes <= 0 {
			cfg.MaxSourceBytes = DefaultImageExtractMaxSourceBytes
		}
		if cfg.MaxImageBytes <= 0 {
			cfg.MaxImageBytes = defaultImageExtractMaxSize
		}
		if cfg.TaskTimeout <= 0 {
			cfg.TaskTimeout = defaultImageExtractTimeout
		}
		if cfg.MaxExtractTextBytes <= 0 {
			cfg.MaxExtractTextBytes = defaultMaxExtractedTextBytes
		}
		if cfg.Extractor == nil {
			cfg.Extractor = NewBasicImageTextExtractor()
		}

		b.imageExtractEnabled = true
		b.imageExtractor = cfg.Extractor
		b.imageExtractTimeout = cfg.TaskTimeout
		b.imageExtractMaxSourceSize = cfg.MaxSourceBytes
		b.imageExtractMaxSize = cfg.MaxImageBytes
		b.maxExtractTextBytes = cfg.MaxExtractTextBytes
		globalBackendRuntimeMetrics.activateImage(b.runtimeMetricsID, 0, 0)

		logger.Info(backgroundWithTrace(), "backend_image_extract_runtime_configured",
			zap.Duration("task_timeout", cfg.TaskTimeout),
			zap.Int64("max_source_bytes", cfg.MaxSourceBytes),
			zap.Int64("max_image_bytes", cfg.MaxImageBytes),
			zap.Int("max_extract_text_bytes", cfg.MaxExtractTextBytes),
			zap.String("extractor_type", fmt.Sprintf("%T", cfg.Extractor)))
	}

	a := opts.AsyncAudioExtract
	b.audioExtractCapability = a.Capability
	if a.MaxSourceBytes <= 0 {
		a.MaxSourceBytes = DefaultAudioExtractMaxSourceBytes
	}
	b.audioExtractMaxSourceSize = a.MaxSourceBytes
	if a.Enabled && a.Extractor != nil {
		if a.MaxAudioBytes <= 0 {
			a.MaxAudioBytes = defaultAudioExtractMaxSize
		}
		if a.TaskTimeout <= 0 {
			a.TaskTimeout = defaultAudioExtractTimeout
		}
		if a.MaxExtractTextBytes <= 0 {
			a.MaxExtractTextBytes = defaultMaxAudioExtractedTextBytes
		}
		b.audioExtractEnabled = true
		b.audioExtractor = a.Extractor
		b.audioExtractTimeout = a.TaskTimeout
		b.audioExtractMaxSize = a.MaxAudioBytes
		b.maxAudioExtractTextBytes = a.MaxExtractTextBytes
		globalBackendRuntimeMetrics.activateAudio(b.runtimeMetricsID, a.MaxAudioBytes, a.MaxExtractTextBytes, a.TaskTimeout)
		logger.Info(backgroundWithTrace(), "backend_audio_extract_runtime_configured",
			zap.Duration("task_timeout", a.TaskTimeout),
			zap.Int64("max_source_bytes", a.MaxSourceBytes),
			zap.Int64("max_audio_bytes", a.MaxAudioBytes),
			zap.Int("max_extract_text_bytes", a.MaxExtractTextBytes),
			zap.String("extractor_type", fmt.Sprintf("%T", a.Extractor)))
	}

	v := opts.AsyncVideoExtract
	b.videoExtractCapability = v.Capability
	if v.MaxSourceBytes <= 0 {
		v.MaxSourceBytes = DefaultVideoExtractMaxSourceBytes
	}
	if v.MaxProviderPayloadBytes <= 0 {
		v.MaxProviderPayloadBytes = DefaultVideoExtractMaxProviderPayloadBytes
	}
	b.videoExtractMaxSourceSize = v.MaxSourceBytes
	if v.Enabled && v.Extractor != nil {
		if v.TaskTimeout <= 0 {
			v.TaskTimeout = defaultVideoExtractTimeout
		}
		if v.MaxExtractTextBytes <= 0 {
			v.MaxExtractTextBytes = defaultMaxVideoExtractedTextBytes
		}
		b.videoExtractEnabled = true
		b.videoExtractAllTenants = v.AllTenants
		b.videoExtractor = v.Extractor
		b.videoExtractTimeout = v.TaskTimeout
		b.maxVideoExtractTextBytes = v.MaxExtractTextBytes
		b.videoExtractTenantAllowlist = v.TenantAllowlist
		if v.MaxVideoLLMFiles > 0 {
			b.maxVideoLLMFiles = v.MaxVideoLLMFiles
		} else {
			b.maxVideoLLMFiles = meta.DefaultMaxVideoLLMFiles()
		}
		logger.Info(backgroundWithTrace(), "backend_video_extract_runtime_configured",
			zap.Duration("task_timeout", v.TaskTimeout),
			zap.Int64("max_source_bytes", v.MaxSourceBytes),
			zap.Int64("max_provider_payload_bytes", v.MaxProviderPayloadBytes),
			zap.Int("max_extract_text_bytes", v.MaxExtractTextBytes),
			zap.String("extractor_type", fmt.Sprintf("%T", v.Extractor)))
	}
}

// Close stops background workers owned by this backend instance. FileGC and
// quota outbox processing are now kick-driven through the unified tenant
// worker, so there are no per-backend goroutines to stop for them.
func (b *Dat9Backend) Close() {
	if b.imageExtractEnabled {
		globalBackendRuntimeMetrics.deactivateImage(b.runtimeMetricsID)
		b.imageExtractEnabled = false
	}
	if b.audioExtractEnabled {
		globalBackendRuntimeMetrics.deactivateAudio(b.runtimeMetricsID)
		b.audioExtractEnabled = false
	}
	if b.videoExtractEnabled {
		b.videoExtractEnabled = false
	}
	b.stopMutationWorker()
	if b.quotaConfigCache != nil {
		b.quotaConfigCache = nil
	}
}