642 lines
17 KiB
Go
642 lines
17 KiB
Go
package service
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"io"
|
|
"math"
|
|
"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 := loadSubscriptionRaw(ctx, sub, opts)
|
|
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
|
|
if opts.Collection != nil && opts.Collection.RenameEnabled {
|
|
allProxies = filter.RenameCollectionNodes(allProxies, opts.Collection.RenameOptions)
|
|
}
|
|
|
|
// 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) {
|
|
if sub.Type == "local" || sub.Content != "" {
|
|
return sub.Content + sub.URL, metadataFromSource(sub)
|
|
}
|
|
|
|
urls := splitSourceUrls(sub.URL)
|
|
if len(urls) == 0 {
|
|
return "", metadataFromSource(sub)
|
|
}
|
|
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, _ := RunWithConcurrencyT(tasks, getConcurrency(opts.Settings), getConcurrencyWait(opts.Settings))
|
|
|
|
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
|
|
}
|
|
|
|
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))
|
|
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 // zero value
|
|
// In non-settled mode, we could cancel, but let's keep it simple
|
|
logrus.WithError(err).Warn("task failed")
|
|
continue
|
|
}
|
|
results[idx] = val
|
|
}
|
|
}()
|
|
}
|
|
wg.Wait()
|
|
return results, nil
|
|
}
|
|
|
|
// RunWithConcurrencyT is a variant for tasks returning a value + metadata.
|
|
func RunWithConcurrencyT[T any](tasks []func() (T, error), concurrency int, wait time.Duration) ([]T, error) {
|
|
return RunWithConcurrency(tasks, concurrency, wait)
|
|
}
|
|
|
|
// 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")
|
|
}
|
|
}
|
|
|
|
// Prevent unused import
|
|
var _ = math.MaxInt32
|
|
|
|
// 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),
|
|
},
|
|
}
|
|
}
|