package service

import (
	"context"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"sort"
	"strings"
	"sync"
	"time"

	"github.com/sirupsen/logrus"

	"github.com/peterqiu0516/sub-store/internal/database"
	"github.com/peterqiu0516/sub-store/internal/filter"
	"github.com/peterqiu0516/sub-store/internal/model"
	"github.com/peterqiu0516/sub-store/internal/proxy"
	"github.com/peterqiu0516/sub-store/internal/render"
	"github.com/peterqiu0516/sub-store/internal/util"
)

// BuildOptions holds parameters for building a subscription result.
type BuildOptions struct {
	Source           *model.SourceRecord
	Collection       *model.CollectionRecord
	Sources          []model.SourceRecord
	Target           string
	TemplateConfig   map[string]any
	Settings         map[string]any
	RequestUserAgent string
	ForceRefresh     bool
	CacheRepo        *database.CacheRepo
	ProxyURL         string
}

// BuildResult holds the output of a subscription build.
type BuildResult struct {
	Body          string
	Metadata      model.SubscriptionResponseMetadata
	Nodes         int
	OriginalNodes int
}

// BuildSubscriptionResult runs the full subscription pipeline.
func BuildSubscriptionResult(ctx context.Context, opts BuildOptions) (*BuildResult, error) {
	proxies, originalNodes, metadata, err := loadProxyNodes(ctx, opts)
	if err != nil {
		return nil, err
	}
	if len(proxies) == 0 {
		return nil, fmt.Errorf("No available nodes found")
	}

	body, err := render.RenderBuildTarget(proxies, opts.Target, "", opts.TemplateConfig)
	if err != nil {
		return nil, err
	}

	return &BuildResult{
		Body:          body,
		Metadata:      metadata,
		Nodes:         len(proxies),
		OriginalNodes: originalNodes,
	}, nil
}

func loadProxyNodes(ctx context.Context, opts BuildOptions) ([]model.ProxyNode, int, model.SubscriptionResponseMetadata, error) {
	sources := getSources(opts)
	enabledSources := make([]model.SourceRecord, 0, len(sources))
	for _, s := range sources {
		if s.Enabled {
			enabledSources = append(enabledSources, s)
		}
	}
	if len(enabledSources) == 0 {
		return nil, 0, model.SubscriptionResponseMetadata{}, nil
	}

	metadataByIndex := make([]model.SubscriptionResponseMetadata, len(enabledSources))
	originalCounts := make([]int, len(enabledSources))

	tasks := make([]func() ([]model.ProxyNode, error), len(enabledSources))
	for i, sub := range enabledSources {
		i := i
		sub := sub
		tasks[i] = func() ([]model.ProxyNode, error) {
			raw, meta, err := loadSubscriptionRaw(ctx, sub, opts)
			if err != nil {
				return nil, err
			}
			metadataByIndex[i] = meta
			nodes := proxy.ParseProxies(raw)
			originalCounts[i] = len(nodes)
			filters := sub.Filters
			if filters == nil {
				filters = []model.FilterRule{}
			}
			processed := filter.ApplyFilters(nodes, filters, opts.Settings, filter.FilterContext{
				TargetPlatform: opts.Target,
				SourceId:       sub.ID,
			})
			// Tag each node with the source's alias for collection-level renaming
			if sub.Alias != "" {
				for j := range processed {
					if processed[j] == nil {
						continue
					}
					tagged := make(map[string]any, len(processed[j])+1)
					for k, v := range processed[j] {
						tagged[k] = v
					}
					tagged["_sourceAlias"] = sub.Alias
					processed[j] = tagged
				}
			}
			return processed, nil
		}
	}

	var proxyLists [][]model.ProxyNode
	var taskErr error
	if opts.Collection != nil && opts.Collection.IgnoreFailed {
		results := RunSettledWithConcurrency(tasks, getConcurrency(opts.Settings), getConcurrencyWait(opts.Settings))
		for _, r := range results {
			if r.Status == "fulfilled" {
				proxyLists = append(proxyLists, r.Value)
			}
		}
	} else {
		results, err := RunWithConcurrency(tasks, getConcurrency(opts.Settings), getConcurrencyWait(opts.Settings))
		if err != nil {
			taskErr = err
		}
		proxyLists = results
	}
	if taskErr != nil {
		return nil, 0, model.SubscriptionResponseMetadata{}, taskErr
	}

	var allProxies []model.ProxyNode
	originalNodes := 0
	for _, list := range proxyLists {
		allProxies = append(allProxies, list...)
	}
	for _, count := range originalCounts {
		originalNodes += count
	}

	// Apply collection-level rename if enabled. Cached egress geo fields are
	// merged in first (pure cache reads, HH-773) and stripped after renaming so
	// they don't leak into rendered output.
	if opts.Collection != nil && opts.Collection.RenameEnabled {
		allProxies = MergeCachedEgressGeo(opts.CacheRepo, allProxies)
		allProxies = filter.RenameCollectionNodes(allProxies, opts.Collection.RenameOptions)
		StripEgressGeoFields(allProxies)
	}

	// Ensure unique names
	allProxies = proxy.EnsureUniqueProxyNames(allProxies)

	// Sort by node name for consistent output
	sort.Slice(allProxies, func(i, j int) bool {
		return fmt.Sprintf("%v", allProxies[i]["name"]) < fmt.Sprintf("%v", allProxies[j]["name"])
	})

	// Select response metadata
	metadataMap := make(map[string]model.SubscriptionResponseMetadata, len(enabledSources))
	for i, s := range enabledSources {
		if s.ID != "" {
			metadataMap[s.ID] = metadataByIndex[i]
		}
	}
	metadata := selectResponseMetadata(enabledSources, metadataMap)
	return allProxies, originalNodes, metadata, nil
}

func getSources(opts BuildOptions) []model.SourceRecord {
	if opts.Collection == nil {
		if opts.Source != nil {
			return []model.SourceRecord{*opts.Source}
		}
		return []model.SourceRecord{}
	}
	sourceIds := opts.Collection.SourceIds
	if len(sourceIds) == 0 {
		return opts.Sources
	}
	var result []model.SourceRecord
	for _, id := range sourceIds {
		for _, s := range opts.Sources {
			if s.ID == id || s.Name == id {
				result = append(result, s)
				break
			}
		}
	}
	return result
}

func loadSubscriptionRaw(ctx context.Context, sub model.SourceRecord, opts BuildOptions) (string, model.SubscriptionResponseMetadata, error) {
	if sub.Type == "local" || sub.Content != "" {
		return sub.Content, metadataFromSource(sub), nil
	}

	urls := splitSourceUrls(sub.URL)
	if len(urls) == 0 {
		return "", metadataFromSource(sub), nil
	}
	if len(urls) > util.MaxRemoteSourceUrls {
		urls = urls[:util.MaxRemoteSourceUrls]
	}

	tasks := make([]func() (fetchResult, error), len(urls))
	for i, u := range urls {
		u := u
		tasks[i] = func() (fetchResult, error) {
			content, meta, err := fetchSubscriptionUrl(ctx, u, sub, opts)
			return fetchResult{content: content, meta: meta}, err
		}
	}

	results, err := RunWithConcurrency(tasks, getConcurrency(opts.Settings), getConcurrencyWait(opts.Settings))
	if err != nil {
		return "", model.SubscriptionResponseMetadata{}, err
	}

	var contents []string
	var metadata model.SubscriptionResponseMetadata
	for i, r := range results {
		contents = append(contents, proxy.DecodeMaybeBase64(r.content))
		if i == 0 {
			metadata = r.meta
		}
	}

	return strings.Join(contents, "\n"), metadata, nil
}

type fetchResult struct {
	content string
	meta    model.SubscriptionResponseMetadata
}

func splitSourceUrls(raw string) []string {
	var result []string
	for _, line := range strings.Split(raw, "\n") {
		line = strings.TrimSpace(line)
		if strings.HasPrefix(line, "http://") || strings.HasPrefix(line, "https://") {
			result = append(result, line)
		}
	}
	return result
}

func fetchSubscriptionUrl(ctx context.Context, url string, sub model.SourceRecord, opts BuildOptions) (string, model.SubscriptionResponseMetadata, error) {
	ua := getSourceUserAgent(sub, opts)
	cacheTtl := getCacheTtl(sub, opts)
	cacheKey := ""
	if cacheTtl > 0 && opts.CacheRepo != nil {
		cacheKey = util.SHA256Hex(url + "\n" + ua)
		if entry, ok := opts.CacheRepo.SafeGet(cacheKey); ok && !opts.ForceRefresh {
			return entry.Content, metadataFromCache(entry, "hit"), nil
		}
	}

	timeout := getTimeout(opts.Settings)
	httpClient := buildHTTPClient(timeout, opts.ProxyURL)

	req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
	if err != nil {
		return "", model.SubscriptionResponseMetadata{}, err
	}
	req.Header.Set("User-Agent", ua)

	resp, err := httpClient.Do(req)
	if err != nil {
		// Try stale cache
		if cacheKey != "" && opts.CacheRepo != nil {
			if entry, ok := opts.CacheRepo.SafeGet(cacheKey); ok {
				return entry.Content, metadataFromCache(entry, "stale"), nil
			}
		}
		return "", model.SubscriptionResponseMetadata{}, err
	}
	defer resp.Body.Close()

	if resp.StatusCode == 304 && cacheKey != "" && opts.CacheRepo != nil {
		if entry, ok := opts.CacheRepo.SafeGet(cacheKey); ok {
			return entry.Content, metadataFromCache(entry, "refresh"), nil
		}
	}
	if resp.StatusCode != 200 {
		return "", model.SubscriptionResponseMetadata{}, fmt.Errorf("Remote source %s failed: %d", sub.Name, resp.StatusCode)
	}

	// Read with limit
	body, err := io.ReadAll(io.LimitReader(resp.Body, int64(util.MaxRemoteSourceRespBytes)))
	if err != nil {
		return "", model.SubscriptionResponseMetadata{}, err
	}
	content := string(body)
	metadata := metadataFromResponse(resp, "miss")

	// Cache asynchronously
	if cacheKey != "" && cacheTtl > 0 && opts.CacheRepo != nil {
		metaMap := metadataToMap(metadata)
		go func() {
			opts.CacheRepo.SafePut(cacheKey, content, metaMap, cacheTtl)
		}()
	}

	return content, metadata, nil
}

func getSourceUserAgent(sub model.SourceRecord, opts BuildOptions) string {
	if sub.Meta != nil {
		if ua, ok := sub.Meta["ua"].(string); ok && ua != "" {
			return ua
		}
		if ua, ok := sub.Meta["userAgent"].(string); ok && ua != "" {
			return ua
		}
	}
	if opts.Settings != nil {
		if ua, ok := opts.Settings["defaultUserAgent"].(string); ok && ua != "" {
			return ua
		}
	}
	return "clash.meta/v1.19.24"
}

func getCacheTtl(sub model.SourceRecord, opts BuildOptions) int {
	if sub.Meta != nil {
		if ttl, ok := sub.Meta["cacheTtl"]; ok {
			if n := toInt(ttl); n > 0 {
				return clamp(n, 0, 3600)
			}
		}
	}
	if opts.Settings != nil {
		if ttl, ok := opts.Settings["remoteCacheTtl"]; ok {
			if n := toInt(ttl); n > 0 {
				return clamp(n, 0, 3600)
			}
		}
	}
	return 300
}

func getTimeout(settings map[string]any) time.Duration {
	if settings != nil {
		if t, ok := settings["defaultTimeout"]; ok {
			if n := toInt(t); n > 0 {
				return time.Duration(clamp(n, 1000, 120000)) * time.Millisecond
			}
		}
	}
	return 30 * time.Second
}

func getConcurrency(settings map[string]any) int {
	if settings != nil {
		if c, ok := settings["backendRequestConcurrency"]; ok {
			if n := toInt(c); n > 0 {
				return clamp(n, 1, 12)
			}
		}
	}
	return 3
}

func getConcurrencyWait(settings map[string]any) time.Duration {
	if settings != nil {
		if w, ok := settings["backendRequestConcurrencyWaitTime"]; ok {
			if n := toInt(w); n >= 0 {
				return time.Duration(n) * time.Millisecond
			}
		}
	}
	return 0
}

func metadataFromSource(sub model.SourceRecord) model.SubscriptionResponseMetadata {
	meta := sub.Meta
	if meta == nil {
		meta = map[string]any{}
	}
	return model.SubscriptionResponseMetadata{
		SubscriptionUserinfo:  getString(meta["subUserinfo"], getString(meta["subscriptionUserinfo"], "")),
		ProfileWebPageUrl:     getString(meta["profileWebPageUrl"], getString(meta["appUrl"], "")),
		ProfileUpdateInterval: getString(meta["profileUpdateInterval"], ""),
		CacheStatus:           "disabled",
	}
}

func metadataFromResponse(resp *http.Response, cacheStatus string) model.SubscriptionResponseMetadata {
	return model.SubscriptionResponseMetadata{
		SubscriptionUserinfo:  resp.Header.Get("subscription-userinfo"),
		ProfileWebPageUrl:     resp.Header.Get("profile-web-page-url"),
		ProfileUpdateInterval: resp.Header.Get("profile-update-interval"),
		ContentDisposition:    resp.Header.Get("content-disposition"),
		Etag:                  resp.Header.Get("etag"),
		LastModified:          resp.Header.Get("last-modified"),
		CacheStatus:           cacheStatus,
	}
}

func metadataFromCache(entry *database.CacheEntry, cacheStatus string) model.SubscriptionResponseMetadata {
	m := entry.Metadata
	return model.SubscriptionResponseMetadata{
		SubscriptionUserinfo:  getStringFromMap(m, "subscriptionUserinfo"),
		ProfileWebPageUrl:     getStringFromMap(m, "profileWebPageUrl"),
		ProfileUpdateInterval: getStringFromMap(m, "profileUpdateInterval"),
		ContentDisposition:    getStringFromMap(m, "contentDisposition"),
		Etag:                  getStringFromMap(m, "etag"),
		LastModified:          getStringFromMap(m, "lastModified"),
		CacheStatus:           cacheStatus,
	}
}

func metadataToMap(m model.SubscriptionResponseMetadata) map[string]any {
	return map[string]any{
		"subscriptionUserinfo":  m.SubscriptionUserinfo,
		"profileWebPageUrl":     m.ProfileWebPageUrl,
		"profileUpdateInterval": m.ProfileUpdateInterval,
		"contentDisposition":    m.ContentDisposition,
		"etag":                  m.Etag,
		"lastModified":          m.LastModified,
	}
}

func selectResponseMetadata(sources []model.SourceRecord, metadataMap map[string]model.SubscriptionResponseMetadata) model.SubscriptionResponseMetadata {
	for _, s := range sources {
		if meta, ok := metadataMap[s.ID]; ok {
			return meta
		}
	}
	return model.SubscriptionResponseMetadata{}
}

func toInt(v any) int {
	switch n := v.(type) {
	case int:
		return n
	case int64:
		return int(n)
	case float64:
		return int(n)
	case string:
		var i int
		fmt.Sscanf(n, "%d", &i)
		return i
	}
	return 0
}

func clamp(n, min, max int) int {
	if n < min {
		return min
	}
	if n > max {
		return max
	}
	return n
}

func getString(v any, def string) string {
	if s, ok := v.(string); ok && s != "" {
		return s
	}
	return def
}

func getStringFromMap(m map[string]any, key string) string {
	if v, ok := m[key]; ok {
		if s, ok := v.(string); ok {
			return s
		}
	}
	return ""
}

// --- Concurrency ---

type Result[T any] struct {
	Status string // "fulfilled" or "rejected"
	Value  T
	Err    error
}

// RunWithConcurrency runs tasks with a fixed worker pool, returning all results.
// Per review-resolution #36: preserves wait parameter.
func RunWithConcurrency[T any](tasks []func() (T, error), concurrency int, wait time.Duration) ([]T, error) {
	if concurrency < 1 {
		concurrency = 1
	}
	if concurrency > len(tasks) {
		concurrency = len(tasks)
	}
	results := make([]T, len(tasks))
	errs := make([]error, len(tasks))
	cursor := 0
	var mu sync.Mutex
	var wg sync.WaitGroup

	for w := 0; w < concurrency; w++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for {
				mu.Lock()
				idx := cursor
				cursor++
				mu.Unlock()
				if idx >= len(tasks) {
					return
				}
				if wait > 0 && idx > 0 {
					time.Sleep(wait)
				}
				val, err := tasks[idx]()
				if err != nil {
					results[idx] = val
					errs[idx] = err
					logrus.WithError(err).Warn("task failed")
					continue
				}
				results[idx] = val
			}
		}()
	}
	wg.Wait()
	return results, errors.Join(errs...)
}

// RunSettledWithConcurrency runs tasks with allSettled semantics.
// Per review-resolution #36: preserves wait parameter.
func RunSettledWithConcurrency[T any](tasks []func() (T, error), concurrency int, wait time.Duration) []Result[T] {
	if concurrency < 1 {
		concurrency = 1
	}
	if concurrency > len(tasks) {
		concurrency = len(tasks)
	}
	results := make([]Result[T], len(tasks))
	cursor := 0
	var mu sync.Mutex
	var wg sync.WaitGroup

	for w := 0; w < concurrency; w++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for {
				mu.Lock()
				idx := cursor
				cursor++
				mu.Unlock()
				if idx >= len(tasks) {
					return
				}
				if wait > 0 && idx > 0 {
					time.Sleep(wait)
				}
				val, err := tasks[idx]()
				if err != nil {
					results[idx] = Result[T]{Status: "rejected", Err: err}
				} else {
					results[idx] = Result[T]{Status: "fulfilled", Value: val}
				}
			}
		}()
	}
	wg.Wait()
	return results
}

// AsyncWriter manages goroutine lifecycle for async operations.
// Per review-resolution #34: sync.WaitGroup + context.
type AsyncWriter struct {
	wg     sync.WaitGroup
	ctx    context.Context
	cancel context.CancelFunc
}

func NewAsyncWriter() *AsyncWriter {
	ctx, cancel := context.WithCancel(context.Background())
	return &AsyncWriter{ctx: ctx, cancel: cancel}
}

func (aw *AsyncWriter) Write(fn func()) {
	aw.wg.Add(1)
	go func() {
		defer aw.wg.Done()
		select {
		case <-aw.ctx.Done():
			return
		default:
			fn()
		}
	}()
}

func (aw *AsyncWriter) Wait() {
	aw.cancel()
	aw.wg.Wait()
}

// WaitWithTimeout waits with a timeout for graceful shutdown.
func (aw *AsyncWriter) WaitWithTimeout(timeout time.Duration) {
	aw.cancel()
	done := make(chan struct{})
	go func() {
		aw.wg.Wait()
		close(done)
	}()
	select {
	case <-done:
	case <-time.After(timeout):
		logrus.Warn("AsyncWriter timed out waiting for goroutines")
	}
}

// buildHTTPClient creates an *http.Client with optional proxy support.
// If proxyURL is empty, a standard client is returned.
func buildHTTPClient(timeout time.Duration, proxyURL string) *http.Client {
	if proxyURL == "" {
		return &http.Client{Timeout: timeout}
	}
	pu, err := url.Parse(proxyURL)
	if err != nil {
		logrus.WithError(err).Warn("Invalid proxy_url, falling back to direct")
		return &http.Client{Timeout: timeout}
	}
	return &http.Client{
		Timeout: timeout,
		Transport: &http.Transport{
			Proxy: http.ProxyURL(pu),
		},
	}
}