package core import ( "bytes" "context" "encoding/json" "errors" "fmt" "html" "os" "path/filepath" "regexp" "runtime" "sort" "strconv" "strings" "sync/atomic" "time" "github.com/gofiber/fiber/v2" browserprofile "github.com/karust/openserp/core/browser" "github.com/karust/openserp/core/fpcheck" "github.com/karust/openserp/core/fpcheck/detectors" apidocs "github.com/karust/openserp/docs" "github.com/sirupsen/logrus" "golang.org/x/time/rate" ) var credentialedURLPattern = regexp.MustCompile(`(?i)\b[a-z][a-z0-9+.-]*://[^\s/@]+(?::[^\s/@]*)?@[^\s]+`) // DefaultFingerprintArtifactDir is the artifact directory used when none is // configured. It is relative to the server's working directory at start time. var DefaultFingerprintArtifactDir = filepath.Join("core", "testdata") // SearchEngine defines the contract required by the HTTP server and resilient // search pipeline. type SearchEngine interface { // Search runs a web search request and returns normalized results. // Implementations should return sentinel errors such as ErrCaptcha and // ErrSearchTimeout for policy-aware handling. Search(context.Context, Query) ([]SearchResult, error) // SearchImage runs an image search request and returns normalized results. SearchImage(context.Context, Query) ([]SearchResult, error) // IsInitialized reports whether the engine is ready to serve requests. IsInitialized() bool // Name returns a stable engine identifier used in routes and telemetry. Name() string // GetRateLimiter returns an engine-specific limiter used by resilient search. GetRateLimiter() *rate.Limiter } // Server exposes OpenSERP HTTP endpoints backed by one or more search engines. type Server struct { app *fiber.App addr string searchEngines []SearchEngine cache *ResponseCache resilient *ResilientSearcher startTime time.Time opts ServerOptions draining atomic.Bool } // ServerOptions configures HTTP server middleware and resilience behavior. type ServerOptions struct { // CacheTTL controls response cache entry lifetime. Zero disables caching. CacheTTL time.Duration // CacheMaxSize is the maximum number of cached entries. CacheMaxSize int // EnableCORS enables cross-origin headers with the CORS config below. EnableCORS bool // CORS contains allowed origins, methods, and headers when CORS is enabled. CORS CORSConfig // AllowEndpointFallback allows dedicated engine routes to fall back to other // healthy engines when the primary engine fails. AllowEndpointFallback bool // EnableDebugEndpoints enables debug-only routes such as fingerprint checks. EnableDebugEndpoints bool // FingerprintArtifactDir is where debug fingerprint screenshots are written. FingerprintArtifactDir string // FingerprintBrowserOpts are the defaults for debug fingerprint runs. FingerprintBrowserOpts BrowserOpts // Resilience defines retry/circuit-breaker/proxy strategy settings. Resilience ResilientConfig // MegaTimeout bounds total wait time for a /mega/* request. When a // mega request exceeds this deadline, engines that have already // responded contribute their results and slower engines are reported // as failed with a context-deadline error. Zero disables the bound // (legacy behavior — wait until the slowest engine finishes). MegaTimeout time.Duration } // DefaultServerOptions returns production-oriented defaults for cache, CORS, // and resilient search policies. func DefaultServerOptions() ServerOptions { return ServerOptions{ CacheTTL: 5 * time.Minute, CacheMaxSize: 1000, EnableCORS: true, CORS: DefaultCORSConfig(), AllowEndpointFallback: false, EnableDebugEndpoints: false, FingerprintArtifactDir: DefaultFingerprintArtifactDir, FingerprintBrowserOpts: BrowserOpts{ IsHeadless: true, Timeout: 30 * time.Second, }, Resilience: DefaultResilientConfig(), MegaTimeout: 90 * time.Second, } } // NewServer creates a Server with DefaultServerOptions and registers all // routes for the provided engines. func NewServer(host string, port int, searchEngines ...SearchEngine) *Server { return NewServerWithOptions(host, port, DefaultServerOptions(), searchEngines...) } // NewServerWithOptions builds a Server, installs middleware, and registers API // routes. The returned server is ready to Listen; call Shutdown for graceful // stop. func NewServerWithOptions(host string, port int, opts ServerOptions, searchEngines ...SearchEngine) *Server { addr := fmt.Sprintf("%s:%d", host, port) app := fiber.New(fiber.Config{ ErrorHandler: JSONErrorMiddleware(), BodyLimit: 10 * 1024 * 1024, }) serv := Server{ app: app, addr: addr, searchEngines: searchEngines, resilient: NewResilientSearcher(searchEngines, opts.Resilience), startTime: time.Now(), opts: opts, } serv.draining.Store(false) logrus.Info("Resilient search enabled: retry + circuit breaker") if opts.AllowEndpointFallback { logrus.Warn("Dedicated endpoint fallback is enabled") } if opts.CacheTTL > 0 && opts.CacheMaxSize > 0 { serv.cache = NewResponseCache(opts.CacheTTL, opts.CacheMaxSize) logrus.WithFields(logrus.Fields{ "cache_ttl": opts.CacheTTL.String(), "cache_max_size": opts.CacheMaxSize, }).Info("Response cache enabled") } app.Use(RequestContextMiddleware()) if opts.EnableCORS { app.Use(CORSMiddleware(opts.CORS)) } app.Use(RequestLoggerMiddleware()) app.Get("/openapi.yaml", serv.handleOpenAPISpec) app.Get("/docs", serv.handleSwaggerUI) app.Get("/docs/", serv.handleSwaggerUI) app.Get("/health", serv.handleHealthCheck) app.Get("/ready", serv.handleReadinessCheck) app.Get("/stats", serv.handleStats) app.Get("/stats/cache", serv.handleCacheStats) app.Get("/stats/proxy", serv.handleProxyStats) app.Get("/stats/cb", serv.handleCircuitBreakerStats) if opts.EnableDebugEndpoints { app.Get("/debug/fingerprint-check", serv.handleFingerprintCheck) } for _, engine := range searchEngines { locEngine := engine endpointName := strings.ToLower(locEngine.Name()) if endpointName == "duckduckgo" { endpointName = "duck" } serv.app.Get(fmt.Sprintf("/%s/search", endpointName), func(c *fiber.Ctx) error { return serv.handleDedicatedEndpoint(c, locEngine, false) }) serv.app.Get(fmt.Sprintf("/%s/image", endpointName), func(c *fiber.Ctx) error { return serv.handleDedicatedEndpoint(c, locEngine, true) }) } for _, engine := range searchEngines { parser, ok := engine.(HTMLParser) if !ok { continue } locParser := parser parserEndpointName := strings.ToLower(parser.Name()) serv.app.Post(fmt.Sprintf("/%s/parse", parserEndpointName), func(c *fiber.Ctx) error { return serv.handleParseEndpoint(c, locParser) }) if parserEndpointName == "duckduckgo" { serv.app.Post("/duck/parse", func(c *fiber.Ctx) error { return serv.handleParseEndpoint(c, locParser) }) } } serv.app.Get("/mega/search", serv.handleMegaSearch) serv.app.Get("/mega/image", serv.handleMegaImage) serv.app.Get("/mega/engines", serv.handleListEngines) return &serv } func (s *Server) handleDedicatedEndpoint(c *fiber.Ctx, engine SearchEngine, isImage bool) error { startedAt := time.Now() requestCtx := withRequestUsage(c.UserContext(), engine.Name()) c.SetUserContext(requestCtx) defer setNetworkBytesHeader(c, requestCtx) defer setBrowserProfileHeader(c, requestCtx) if profileID := strings.TrimSpace(c.Get(useProfileHeader)); profileID != "" { if _, ok := browserprofile.ProfileByID(profileID); !ok { return errInvalidParam(fmt.Sprintf("%s: unknown profile id %q", useProfileHeader, profileID)) } requestCtx = WithForcedProfileID(requestCtx, profileID) c.SetUserContext(requestCtx) } q := Query{} if err := q.InitFromContext(c); err != nil { WithRequest(c.UserContext()).WithError(err).Warn("Invalid query parameters") return err } if err := s.validateRequestProxyURL(&q); err != nil { WithRequest(c.UserContext()).WithError(err).Warn("Invalid request proxy URL") return err } format, err := resolveFormat(c) if err != nil { return err } requestCtx = WithQueryHash(c.UserContext(), QueryHashFromQuery(q)) c.SetUserContext(requestCtx) requestID := RequestIDFromContext(requestCtx) action := "search" if isImage { action = "image" } WithRequest(requestCtx). WithField("action", action). Debugf("Starting %s request for query: %s", action, q.Text) if format == "json" && !ShouldBypassCacheForProxyMarket(q) { if hit, err := s.tryServeCacheHit( c, startedAt, cacheHitCandidate{ key: BuildCacheKey(engine.Name(), action, q), logMessage: fmt.Sprintf("Cache hit for %s %s: %s", engine.Name(), action, q.Text), }, ); hit || err != nil { return err } } engineNames := []string{engine.Name()} if isImage { var ( res []SearchResult usedEngine string proxyMeta ProxyExecutionMeta searchErr error ) if s.opts.AllowEndpointFallback { res, usedEngine, proxyMeta, searchErr = s.resilient.SearchImageWithFallback(requestCtx, engine, q) } else { res, usedEngine, proxyMeta, searchErr = s.resilient.SearchImagePrimary(requestCtx, engine, q) } s.applyProxyHeaders(c, proxyMeta) if searchErr != nil { WithRequest(requestCtx).WithFields(logrus.Fields{"action": action}).WithError(searchErr).Error("Search failed") return searchAPIError(searchErr, usedEngine, q, proxyMeta) } env := NewImageEnvelope(q, requestID, startedAt, engineNames) if usedEngine != "" && usedEngine != engine.Name() { env.Meta.EnginesFailed = []string{engine.Name()} } ectx := EnrichContext{Engine: usedEngine, Query: q} for _, r := range res { env.Results = append(env.Results, EnrichImageResult(r, ectx)) } env.Finalize(startedAt, q) if format == "json" { cacheStatus := s.cacheEnvelopeIfEligible(engine.Name(), usedEngine, action, q, env) if cacheStatus != "" { c.Set("X-Cache", cacheStatus) } } if usedEngine != "" && usedEngine != engine.Name() { c.Set("X-Fallback-Engine", usedEngine) } WithRequest(requestCtx).WithFields(logrus.Fields{"action": action, "results_count": len(res)}).Info("Search completed") return sendImageEnvelope(c, format, env) } var ( res []SearchResult usedEngine string proxyMeta ProxyExecutionMeta searchErr error ) if s.opts.AllowEndpointFallback { res, usedEngine, proxyMeta, searchErr = s.resilient.SearchWithFallback(requestCtx, engine, q) } else { res, usedEngine, proxyMeta, searchErr = s.resilient.SearchPrimary(requestCtx, engine, q) } s.applyProxyHeaders(c, proxyMeta) if searchErr != nil { WithRequest(requestCtx).WithFields(logrus.Fields{"action": action}).WithError(searchErr).Error("Search failed") return searchAPIError(searchErr, usedEngine, q, proxyMeta) } env := NewEnvelope(q, requestID, startedAt, engineNames) if usedEngine != "" && usedEngine != engine.Name() { env.Meta.EnginesFailed = []string{engine.Name()} } ectx := EnrichContext{Engine: usedEngine, Query: q} for _, r := range res { env.Results = append(env.Results, EnrichResult(r, ectx)) } env.Finalize(startedAt, q) if format == "json" { cacheStatus := s.cacheEnvelopeIfEligible(engine.Name(), usedEngine, action, q, env) if cacheStatus != "" { c.Set("X-Cache", cacheStatus) } } if usedEngine != "" && usedEngine != engine.Name() { c.Set("X-Fallback-Engine", usedEngine) } completionCtx := requestCtx if usedEngine != "" { completionCtx = WithEngine(completionCtx, usedEngine) } WithRequest(completionCtx).WithFields(logrus.Fields{"action": action, "results_count": len(res)}).Info("Search completed") return sendEnvelope(c, format, env) } func (s *Server) handleParseEndpoint(c *fiber.Ctx, parser HTMLParser) error { startedAt := time.Now() requestCtx := withRequestUsage(c.UserContext(), parser.Name()) c.SetUserContext(requestCtx) body := c.Body() if len(body) == 0 { return errInvalidParam("request body is empty") } format, err := resolveFormat(c) if err != nil { return err } results, err := parser.ParseHTML(bytes.NewReader(body)) if err != nil { return &APIError{ HTTPStatus: fiber.StatusBadRequest, ErrorCode: "parser_failure", Message: fmt.Sprintf("failed to parse HTML: %v", err), } } requestID := RequestIDFromContext(requestCtx) q := Query{} env := NewEnvelope(q, requestID, startedAt, []string{parser.Name()}) ectx := EnrichContext{Engine: parser.Name(), Query: q} for _, r := range results { env.Results = append(env.Results, EnrichResult(r, ectx)) } env.Finalize(startedAt, q) return sendEnvelope(c, format, env) } type searchErrorSpec struct { status int code string message string } func mapSearchError(err error) searchErrorSpec { switch { case errors.Is(err, ErrCaptcha): return searchErrorSpec{status: fiber.StatusTooManyRequests, code: "captcha_detected", message: "captcha detected"} case errors.Is(err, ErrBlocked): return searchErrorSpec{status: fiber.StatusForbidden, code: "blocked", message: "search engine blocked the request"} case errors.Is(err, ErrRateLimited): return searchErrorSpec{status: fiber.StatusTooManyRequests, code: "rate_limited", message: "search engine rate limited the request"} case errors.Is(err, ErrSearchTimeout): return searchErrorSpec{status: fiber.StatusGatewayTimeout, code: "search_timeout", message: ErrSearchTimeout.Error()} case errors.Is(err, ErrProxyAuth): return searchErrorSpec{status: fiber.StatusServiceUnavailable, code: "proxy_auth", message: "proxy authentication failed"} case errors.Is(err, ErrProxyConnect): return searchErrorSpec{status: fiber.StatusServiceUnavailable, code: "proxy_connect", message: "proxy connection failed"} case errors.Is(err, ErrTimeout): return searchErrorSpec{status: fiber.StatusServiceUnavailable, code: "proxy_timeout", message: "proxy request timed out"} case errors.Is(err, ErrProxyUnavailable): return searchErrorSpec{status: fiber.StatusServiceUnavailable, code: "proxy_unavailable", message: "proxy unavailable"} case errors.Is(err, ErrParser): return searchErrorSpec{status: fiber.StatusBadGateway, code: "parser_failure", message: "parser failure"} case errors.Is(err, ErrEngineInternal): return searchErrorSpec{status: fiber.StatusBadGateway, code: "engine_internal", message: "engine internal error"} case errors.Is(err, ErrAllEnginesFailed): return searchErrorSpec{status: fiber.StatusBadGateway, code: "all_engines_failed", message: "all search engines failed"} case errors.Is(err, ErrCircuitOpen): return searchErrorSpec{status: fiber.StatusServiceUnavailable, code: "circuit_open", message: "engine circuit breaker is open"} case errors.Is(err, context.DeadlineExceeded): return searchErrorSpec{status: fiber.StatusGatewayTimeout, code: "request_timeout", message: "request timed out"} case errors.Is(err, context.Canceled): return searchErrorSpec{status: fiber.StatusServiceUnavailable, code: "request_canceled", message: "request canceled"} } return searchErrorSpec{status: fiber.StatusBadGateway, code: "engine_internal", message: err.Error()} } func searchAPIError(err error, engineName string, q Query, proxyMeta ProxyExecutionMeta) *APIError { spec := mapSearchError(err) meta := searchErrorMeta(engineName, q, proxyMeta) addErrorDetail(meta, err, spec.message, q) return &APIError{ HTTPStatus: spec.status, ErrorCode: spec.code, Message: spec.message, Meta: meta, } } func searchErrorMeta(engineName string, q Query, proxyMeta ProxyExecutionMeta) map[string]interface{} { meta := map[string]interface{}{} if strings.TrimSpace(engineName) != "" { meta["engine"] = engineName } proxyUsed := strings.TrimSpace(proxyMeta.Used) if proxyUsed == "" && strings.TrimSpace(q.ProxyURL) != "" { proxyUsed = MaskProxyURL(q.ProxyURL) } if proxyUsed != "" { meta["proxy_used"] = proxyUsed } if q.ProxyCountry != "" { meta["proxy_country"] = q.ProxyCountry } if q.ProxyClass != "" { meta["proxy_class"] = q.ProxyClass } if q.ProxyProvider != "" { meta["proxy_provider"] = q.ProxyProvider } if q.ProxySessionID != "" { meta["proxy_session_id"] = q.ProxySessionID } return meta } func addErrorDetail(meta map[string]interface{}, err error, message string, q Query) { if err == nil { return } detail := sanitizeErrorDetail(err.Error(), q) if detail == "" || detail == message { return } meta["error_detail"] = detail } func engineErrorDetail(engineName string, err error, q Query) EngineErrorDetail { spec := mapSearchError(err) return EngineErrorDetail{ Engine: engineName, Error: spec.code, Message: sanitizeErrorDetail(err.Error(), q), } } func sanitizeErrorDetail(detail string, q Query) string { detail = strings.TrimSpace(detail) if detail == "" { return "" } if q.ProxyURL != "" { detail = strings.ReplaceAll(detail, q.ProxyURL, MaskProxyURL(q.ProxyURL)) } return maskCredentialedURLs(detail) } func maskCredentialedURLs(detail string) string { return credentialedURLPattern.ReplaceAllStringFunc(detail, func(raw string) string { trimmed := strings.TrimRight(raw, `.,;)]}`) suffix := strings.TrimPrefix(raw, trimmed) return MaskProxyURL(trimmed) + suffix }) } // cacheEnvelopeIfEligible stores the envelope JSON and returns the cache status header value. func (s *Server) cacheEnvelopeIfEligible(engineName, usedEngine, action string, q Query, payload interface{}) string { if s.cache == nil { return "" } if ShouldBypassCacheForProxyMarket(q) { s.cache.RecordBypass() return "BYPASS" } // Don't cache fallback responses so the primary engine can recover. if usedEngine != engineName { s.cache.RecordBypass() return "BYPASS" } // Detect empty results via reflection-free type switch. switch v := payload.(type) { case *Envelope: if len(v.Results) == 0 { s.cache.RecordBypass() return "BYPASS" } case *ImageEnvelope: if len(v.Results) == 0 { s.cache.RecordBypass() return "BYPASS" } } cacheKey := BuildCacheKey(engineName, action, q) if s.cacheJSON(cacheKey, payload) { return "MISS" } return "BYPASS" } // HealthStatus is returned by /health and summarizes service state. type HealthStatus struct { Status string `json:"status"` Uptime string `json:"uptime"` Engines []EngineHealth `json:"engines"` System map[string]interface{} `json:"system"` } // ReadinessStatus is returned by /ready to indicate if this instance can // receive new traffic. type ReadinessStatus struct { Status string `json:"status"` } // EngineHealth describes availability of one configured engine. type EngineHealth struct { Name string `json:"name"` Initialized bool `json:"initialized"` Status string `json:"status"` } // handleHealthCheck returns current service and engine status. // Degraded state stays HTTP 200 to avoid unnecessary restarts in orchestrators. func (s *Server) handleHealthCheck(c *fiber.Ctx) error { engines := make([]EngineHealth, 0, len(s.searchEngines)) availableEngines := 0 for _, engine := range s.searchEngines { status := "ready" isAvailable := true if !engine.IsInitialized() { status = "not_initialized" isAvailable = false } for _, cbStat := range s.resilient.GetCircuitBreakerStats() { engineName, _ := cbStat["engine"].(string) if engineName != engine.Name() { continue } circuitState, _ := cbStat["state"].(string) if circuitState == "open" { status = "circuit_open" isAvailable = false } break } if isAvailable { availableEngines++ } engines = append(engines, EngineHealth{ Name: engine.Name(), Initialized: engine.IsInitialized(), Status: status, }) } overallStatus := "healthy" totalEngines := len(s.searchEngines) switch { case totalEngines == 0 || availableEngines == 0: overallStatus = "unhealthy" case availableEngines < totalEngines: overallStatus = "degraded" } var memStats runtime.MemStats runtime.ReadMemStats(&memStats) health := HealthStatus{ Status: overallStatus, Uptime: time.Since(s.startTime).Round(time.Second).String(), Engines: engines, System: map[string]interface{}{ "goroutines": runtime.NumGoroutine(), "memory_mb": memStats.Alloc / 1024 / 1024, "go_version": runtime.Version(), }, } if overallStatus == "unhealthy" { c.Status(fiber.StatusServiceUnavailable) } return c.JSON(health) } func (s *Server) handleReadinessCheck(c *fiber.Ctx) error { status := ReadinessStatus{Status: "ready"} if s.draining.Load() { status.Status = "draining" return c.Status(fiber.StatusServiceUnavailable).JSON(status) } return c.JSON(status) } func (s *Server) handleStats(c *fiber.Ctx) error { return c.JSON(map[string]interface{}{ "cache": s.cacheStatsPayload(), "proxy": s.resilient.GetProxyStats(), "circuit_breakers": s.resilient.GetCircuitBreakerStats(), "captcha": CaptchaSolverMetrics(), }) } func (s *Server) handleCacheStats(c *fiber.Ctx) error { return c.JSON(s.cacheStatsPayload()) } func (s *Server) handleProxyStats(c *fiber.Ctx) error { return c.JSON(s.resilient.GetProxyStats()) } func (s *Server) handleCircuitBreakerStats(c *fiber.Ctx) error { return c.JSON(map[string]interface{}{ "circuit_breakers": s.resilient.GetCircuitBreakerStats(), }) } func (s *Server) handleFingerprintCheck(c *fiber.Ctx) error { requestCtx := withRequestUsage(c.UserContext(), "fingerprint") c.SetUserContext(requestCtx) defer setNetworkBytesHeader(c, requestCtx) defer setBrowserProfileHeader(c, requestCtx) if profileID := strings.TrimSpace(c.Get(useProfileHeader)); profileID != "" { if _, ok := browserprofile.ProfileByID(profileID); !ok { return errInvalidParam(fmt.Sprintf("%s: unknown profile id %q", useProfileHeader, profileID)) } requestCtx = WithForcedProfileID(requestCtx, profileID) c.SetUserContext(requestCtx) } req, err := s.parseFingerprintCheckRequest(c) if err != nil { return err } runCtx, cancel := context.WithTimeout(requestCtx, time.Duration(req.timeoutMs)*time.Millisecond) defer cancel() browser, err := NewBrowser(req.browserOpts) if err != nil { return fiber.NewError(fiber.StatusServiceUnavailable, fmt.Sprintf("failed to create debug browser: %v", err)) } defer func() { if closeErr := browser.Close(); closeErr != nil { WithRequest(c.UserContext()).WithError(closeErr).Warn("failed to close debug fingerprint browser") } }() artifactDir := defaultFingerprintArtifactDir(s.opts.FingerprintArtifactDir) reports := make([]fpcheck.Report, 0, len(req.detectors)) for idx, detector := range req.detectors { runOpts := fpcheck.RunOptions{ ArtifactDir: artifactDir, } if req.waitMs > 0 && idx == len(req.detectors)-1 { runOpts.WaitBeforeExtract = time.Duration(req.waitMs) * time.Millisecond } report, runErr := fpcheck.RunWithOptions(runCtx, browser, detector, runOpts) if runErr != nil { if errors.Is(runCtx.Err(), context.DeadlineExceeded) { return fiber.NewError(fiber.StatusGatewayTimeout, fmt.Sprintf("fingerprint check timed out after %dms", req.timeoutMs)) } return fiber.NewError( fiber.StatusServiceUnavailable, fmt.Sprintf("detector %s failed: %v", detector.Name(), runErr), ) } reports = append(reports, report) } return c.JSON(reports) } type fingerprintCheckRequest struct { detectors []fpcheck.Detector timeoutMs int waitMs int browserOpts BrowserOpts } func (s *Server) parseFingerprintCheckRequest(c *fiber.Ctx) (fingerprintCheckRequest, error) { detectorName := strings.TrimSpace(c.Query("detector", "all")) customURL := strings.TrimSpace(c.Query("url", "")) customSelector := strings.TrimSpace(c.Query("selector", "")) selectedDetectors, err := detectors.SelectWithCustomSelector(detectorName, customURL, customSelector) if err != nil { return fingerprintCheckRequest{}, fiber.NewError(fiber.StatusBadRequest, err.Error()) } headless, err := parseOptionalBoolQuery(c.Query("headless", ""), true) if err != nil { return fingerprintCheckRequest{}, fiber.NewError(fiber.StatusBadRequest, fmt.Sprintf("invalid headless query value: %v", err)) } if !headless && strings.TrimSpace(os.Getenv("DISPLAY")) == "" { WithRequest(c.UserContext()).Warn("headless=false ignored because DISPLAY is not set; forcing headless mode") headless = true } timeoutMs, err := parsePositiveIntQuery(c.Query("timeout_ms", ""), 150000) if err != nil { return fingerprintCheckRequest{}, fiber.NewError(fiber.StatusBadRequest, fmt.Sprintf("invalid timeout_ms query value: %v", err)) } waitMs, err := parseNonNegativeIntQuery(c.Query("wait_ms", ""), 0) if err != nil { return fingerprintCheckRequest{}, fiber.NewError(fiber.StatusBadRequest, fmt.Sprintf("invalid wait_ms query value: %v", err)) } browserOpts := s.opts.FingerprintBrowserOpts browserOpts.IsHeadless = headless browserOpts.Timeout = time.Duration(timeoutMs) * time.Millisecond browserOpts.LeavePageOpen = false browserOpts.UserAgent = strings.TrimSpace(c.Query("user_agent", browserOpts.UserAgent)) browserOpts.ProxyURL = strings.TrimSpace(c.Query("proxy", browserOpts.ProxyURL)) browserOpts.LanguageCode = strings.TrimSpace(c.Query("language", browserOpts.LanguageCode)) if headerProxyURL := strings.TrimSpace(c.Get("X-Proxy-URL")); headerProxyURL != "" { if !s.opts.Resilience.Proxy.Proxies.AllowRequestProxyURL { return fingerprintCheckRequest{}, &APIError{ HTTPStatus: fiber.StatusBadRequest, ErrorCode: "bad_request", Reason: ReasonRequestProxyURLDisabled, Message: "X-Proxy-URL is disabled by server configuration", } } normalized, err := NormalizeProxyURL(headerProxyURL) if err != nil { return fingerprintCheckRequest{}, errInvalidParam(fmt.Sprintf("X-Proxy-URL: %v", err)) } if IsAuthenticatedSocksProxyURL(normalized) { return fingerprintCheckRequest{}, &APIError{ HTTPStatus: fiber.StatusBadRequest, ErrorCode: "bad_request", Reason: ReasonUnsupportedProxyScheme, Message: "authenticated SOCKS proxies are not supported in browser mode", } } browserOpts.ProxyURL = normalized } insecureDefault := browserOpts.Insecure if detectors.IsCustom(detectorName) { insecureDefault = true } insecure, err := parseOptionalBoolQuery(c.Query("insecure", ""), insecureDefault) if err != nil { return fingerprintCheckRequest{}, fiber.NewError(fiber.StatusBadRequest, fmt.Sprintf("invalid insecure query value: %v", err)) } browserOpts.Insecure = insecure return fingerprintCheckRequest{ detectors: selectedDetectors, timeoutMs: timeoutMs, waitMs: waitMs, browserOpts: browserOpts, }, nil } func defaultFingerprintArtifactDir(value string) string { trimmed := strings.TrimSpace(value) if trimmed == "" { return DefaultFingerprintArtifactDir } return trimmed } func parseOptionalBoolQuery(raw string, defaultValue bool) (bool, error) { raw = strings.TrimSpace(raw) if raw == "" { return defaultValue, nil } value, err := strconv.ParseBool(raw) if err != nil { return false, err } return value, nil } func parsePositiveIntQuery(raw string, defaultValue int) (int, error) { raw = strings.TrimSpace(raw) if raw == "" { return defaultValue, nil } value, err := strconv.Atoi(raw) if err != nil { return 0, err } if value <= 0 { return 0, fmt.Errorf("must be > 0") } return value, nil } func parseNonNegativeIntQuery(raw string, defaultValue int) (int, error) { raw = strings.TrimSpace(raw) if raw == "" { return defaultValue, nil } value, err := strconv.Atoi(raw) if err != nil { return 0, err } if value < 0 { return 0, fmt.Errorf("must be >= 0") } return value, nil } // MegaSearchResult extends SearchResult with the engine source name. type MegaSearchResult struct { SearchResult Engine string `json:"engine"` } const ( megaModeBalanced = "balanced" megaModeAny = "any" megaModeFast = "fast" ) type megaRunConfig struct { Mode string Dedupe bool Merge bool } func (s *Server) handleMegaSearch(c *fiber.Ctx) error { return s.handleMegaEndpoint(c, "search") } func (s *Server) handleMegaImage(c *fiber.Ctx) error { return s.handleMegaEndpoint(c, "image") } func (s *Server) handleMegaEndpoint(c *fiber.Ctx, action string) error { startedAt := time.Now() requestCtx := withRequestUsage(c.UserContext(), "mega") c.SetUserContext(requestCtx) defer setNetworkBytesHeader(c, requestCtx) defer setBrowserProfileHeader(c, requestCtx) q := Query{} if err := q.InitFromContext(c); err != nil { WithRequest(c.UserContext()).WithError(err).Warn("Invalid query parameters") return err } if err := s.validateRequestProxyURL(&q); err != nil { WithRequest(c.UserContext()).WithError(err).Warn("Invalid request proxy URL") return err } format, err := resolveFormat(c) if err != nil { return err } requestCtx = WithQueryHash(c.UserContext(), QueryHashFromQuery(q)) c.SetUserContext(requestCtx) requestID := RequestIDFromContext(requestCtx) runCfg, err := parseMegaRunConfig(c) if err != nil { return err } enginesToUse := s.resolveEngines(requestCtx, c.Query("engines", "")) if len(enginesToUse) == 0 { return &APIError{HTTPStatus: 400, Reason: ReasonNoEngines, Message: "no valid search engines specified"} } engineNames := make([]string, len(enginesToUse)) for i, engine := range enginesToUse { engineNames[i] = engine.Name() } engineNamesJoined := strings.Join(engineNames, ",") s.applyProxyHeaders(c, s.resilient.ResolveMegaProxyMeta(q, enginesToUse)) WithRequest(requestCtx).WithFields(logrus.Fields{ "action": action, "engines": engineNamesJoined, "mode": runCfg.Mode, }).Debugf("Starting mega %s request for query: %s", action, q.Text) if format == "json" && !ShouldBypassCacheForProxyMarket(q) && runCfg.Mode != megaModeFast { cacheHitCandidates := []cacheHitCandidate{ { key: s.buildMegaCacheKey(action, enginesToUse, q, runCfg), logMessage: fmt.Sprintf("Cache hit for mega %s: engines=%s query=%s mode=%s", action, engineNamesJoined, q.Text, runCfg.Mode), }, } if runCfg.Mode == megaModeBalanced { cacheableEngines := s.megaCacheableEngines(enginesToUse) if len(cacheableEngines) > 0 && len(cacheableEngines) < len(enginesToUse) { cacheHitCandidates = append(cacheHitCandidates, cacheHitCandidate{ key: s.buildMegaCacheKey(action, cacheableEngines, q, runCfg), logMessage: fmt.Sprintf("Cache hit for mega %s partial set: engines=%s query=%s mode=%s", action, engineNamesJoined, q.Text, runCfg.Mode), }) } } if hit, err := s.tryServeCacheHit(c, startedAt, cacheHitCandidates...); hit || err != nil { return err } } runCtx := requestCtx if s.opts.MegaTimeout > 0 { var cancel context.CancelFunc runCtx, cancel = context.WithTimeout(requestCtx, s.opts.MegaTimeout) defer cancel() } var ( rawResults []MegaSearchResult responded []string engineErrors []EngineErrorDetail ) switch runCfg.Mode { case megaModeAny: rawResults, responded, engineErrors = s.resilient.searchAnyDetailed(runCtx, q, enginesToUse, action == "image") case megaModeFast: rawResults, responded, engineErrors = s.resilient.searchFastestDetailed(runCtx, q, enginesToUse, action == "image") default: if action == "image" { rawResults, responded, engineErrors = s.resilient.searchAllImageParallelDetailed(runCtx, q, enginesToUse) } else { rawResults, responded, engineErrors = s.resilient.searchAllParallelDetailed(runCtx, q, enginesToUse) } } rawResults = s.applyMegaMergePolicy(rawResults, enginesToUse, runCfg) enginesFailed := engineErrorNames(engineErrors) if len(responded) == 0 { err := fmt.Errorf("%w: %s", ErrAllEnginesFailed, strings.Join(enginesFailed, ",")) apiErr := searchAPIError(err, "mega", q, ProxyExecutionMeta{}) apiErr.Meta["engine_errors"] = engineErrors WithRequest(requestCtx).WithFields(logrus.Fields{ "action": action, "engines": engineNamesJoined, }).WithError(err).Error("Mega search failed") return apiErr } if action == "image" { imageResults := rawResults if runCfg.Dedupe { imageResults = s.deduplicateMegaResults(imageResults) } env := NewImageEnvelope(q, requestID, startedAt, engineNames) env.Meta.EnginesResponded = responded env.Meta.EnginesFailed = enginesFailed env.Meta.EngineErrors = engineErrors for _, r := range imageResults { ectx := EnrichContext{Engine: r.Engine, Query: q} env.Results = append(env.Results, EnrichImageResult(r.SearchResult, ectx)) } env.Finalize(startedAt, q) if format == "json" && s.cache != nil && runCfg.Mode != megaModeFast { c.Set("X-Cache", s.cacheMegaImageResults(action, enginesToUse, q, env, runCfg)) } WithRequest(requestCtx).WithFields(logrus.Fields{ "action": action, "engines_count": len(enginesToUse), "results_count": len(env.Results), }).Info("Mega search completed") return sendImageEnvelope(c, format, env) } webResults := rawResults if runCfg.Dedupe { webResults = s.deduplicateMegaResults(webResults) } env := NewEnvelope(q, requestID, startedAt, engineNames) env.Meta.EnginesResponded = responded env.Meta.EnginesFailed = enginesFailed env.Meta.EngineErrors = engineErrors for _, r := range webResults { ectx := EnrichContext{Engine: r.Engine, Query: q} env.Results = append(env.Results, EnrichResult(r.SearchResult, ectx)) } env.Finalize(startedAt, q) if runCfg.Merge { allEnriched := make([]Result, 0, len(rawResults)) for _, r := range rawResults { ectx := EnrichContext{Engine: r.Engine, Query: q} allEnriched = append(allEnriched, EnrichResult(r.SearchResult, ectx)) } clusters := BuildClusters(allEnriched, len(enginesToUse)) if len(clusters) > 0 { env.Clusters = &clusters } } if format == "json" && s.cache != nil && runCfg.Mode != megaModeFast { c.Set("X-Cache", s.cacheMegaEnvelopeResults(action, enginesToUse, q, env, runCfg)) } WithRequest(requestCtx).WithFields(logrus.Fields{ "action": action, "engines_count": len(enginesToUse), "results_count": len(env.Results), }).Info("Mega search completed") return sendEnvelope(c, format, env) } func engineErrorNames(details []EngineErrorDetail) []string { names := make([]string, 0, len(details)) for _, d := range details { names = append(names, d.Engine) } return names } func parseMegaRunConfig(c *fiber.Ctx) (megaRunConfig, error) { cfg := megaRunConfig{ Mode: megaModeBalanced, Dedupe: true, Merge: true, } mode := strings.ToLower(strings.TrimSpace(c.Query("mode", megaModeBalanced))) switch mode { case "", megaModeBalanced: cfg.Mode = megaModeBalanced case megaModeAny: cfg.Mode = megaModeAny case megaModeFast: cfg.Mode = megaModeFast default: return megaRunConfig{}, errInvalidParam("mode: must be one of fast, any, balanced") } if raw := strings.TrimSpace(c.Query("dedupe", "")); raw != "" { value, err := strconv.ParseBool(raw) if err != nil { return megaRunConfig{}, errInvalidParam(fmt.Sprintf("dedupe: %v", err)) } cfg.Dedupe = value } if raw := strings.TrimSpace(c.Query("merge", "")); raw != "" { value, err := strconv.ParseBool(raw) if err != nil { return megaRunConfig{}, errInvalidParam(fmt.Sprintf("merge: %v", err)) } cfg.Merge = value } return cfg, nil } func (s *Server) applyMegaMergePolicy(results []MegaSearchResult, engines []SearchEngine, cfg megaRunConfig) []MegaSearchResult { if cfg.Merge || len(results) == 0 { return results } byEngine := make(map[string]struct{}, len(results)) for _, r := range results { byEngine[r.Engine] = struct{}{} } selected := "" for _, eng := range engines { if _, ok := byEngine[eng.Name()]; ok { selected = eng.Name() break } } if selected == "" { return []MegaSearchResult{} } filtered := make([]MegaSearchResult, 0, len(results)) for _, r := range results { if r.Engine == selected { filtered = append(filtered, r) } } return filtered } func (s *Server) handleListEngines(c *fiber.Ctx) error { var engines []map[string]interface{} for _, engine := range s.searchEngines { engineInfo := map[string]interface{}{ "name": engine.Name(), "initialized": engine.IsInitialized(), } for _, cbStat := range s.resilient.GetCircuitBreakerStats() { engineName, _ := cbStat["engine"].(string) if engineName == engine.Name() { engineInfo["circuit_state"] = cbStat["state"] break } } engines = append(engines, engineInfo) } return c.JSON(map[string]interface{}{ "engines": engines, "total": len(engines), }) } func (s *Server) resolveEngines(ctx context.Context, enginesParam string) []SearchEngine { if enginesParam == "" { return s.searchEngines } var enginesToUse []SearchEngine seen := make(map[string]bool) engineNames := strings.Split(enginesParam, ",") for _, name := range engineNames { name = strings.TrimSpace(strings.ToLower(name)) if name == "" || seen[name] { continue } name = resolveEngineAlias(name) matched := false for _, engine := range s.searchEngines { if strings.ToLower(engine.Name()) == name { enginesToUse = append(enginesToUse, engine) seen[name] = true matched = true break } } if !matched { WithRequest(ctx).Warnf("Unknown engine %q requested, skipping", name) } } return enginesToUse } func resolveEngineAlias(name string) string { switch name { case "duck", "ddg": return "duckduckgo" default: return name } } func (s *Server) deduplicateMegaResults(results []MegaSearchResult) []MegaSearchResult { urlMap := make(map[string]MegaSearchResult) order := []string{} for _, result := range results { if result.URL == "" { continue } normalizedURL := NormalizeURLForClustering(result.URL) if normalizedURL == "" { continue } key := resultDedupKey(SearchResult{URL: normalizedURL, Ad: result.Ad}) existing, exists := urlMap[key] if !exists { urlMap[key] = result order = append(order, key) continue } if betterMegaResult(result, existing) { urlMap[key] = result } } deduped := make([]MegaSearchResult, 0, len(urlMap)) for _, key := range order { deduped = append(deduped, urlMap[key]) } sort.Slice(deduped, func(i, j int) bool { if resultLess(deduped[i].SearchResult, deduped[j].SearchResult) { return true } if resultLess(deduped[j].SearchResult, deduped[i].SearchResult) { return false } if deduped[i].Engine != deduped[j].Engine { return deduped[i].Engine < deduped[j].Engine } return NormalizeURLForClustering(deduped[i].URL) < NormalizeURLForClustering(deduped[j].URL) }) return deduped } func betterMegaResult(candidate, current MegaSearchResult) bool { if candidate.Rank > 0 && (current.Rank <= 0 || candidate.Rank < current.Rank) { return true } if candidate.Rank == current.Rank && candidate.Engine < current.Engine { return true } return false } type cacheHitCandidate struct { key string logMessage string } func (s *Server) tryServeCacheHit(c *fiber.Ctx, startedAt time.Time, candidates ...cacheHitCandidate) (bool, error) { if s.cache == nil { return false, nil } for _, candidate := range candidates { cached, ok := s.cache.Get(candidate.key) if !ok { continue } cached = refreshCachedMeta(cached, RequestIDFromContext(c.UserContext()), startedAt) c.Set("Content-Type", "application/json") c.Set("X-Cache", "HIT") WithRequest(c.UserContext()).Debug(candidate.logMessage) return true, c.Send(cached) } return false, nil } func refreshCachedMeta(data []byte, requestID string, startedAt time.Time) []byte { var payload map[string]any if err := json.Unmarshal(data, &payload); err != nil { return data } meta, ok := payload["meta"].(map[string]any) if !ok { return data } meta["request_id"] = requestID meta["requested_at"] = startedAt.UTC().Format(time.RFC3339) delete(meta, "timestamp") meta["took_ms"] = time.Since(startedAt).Milliseconds() refreshed, err := json.Marshal(payload) if err != nil { return data } return refreshed } func (s *Server) cacheJSON(cacheKey string, payload interface{}) bool { if s.cache == nil { return false } data, err := json.Marshal(payload) if err != nil { s.cache.RecordBypass() return false } s.cache.Set(cacheKey, data) return true } func (s *Server) cacheMegaEnvelopeResults(action string, enginesToUse []SearchEngine, q Query, env *Envelope, cfg megaRunConfig) string { if s.cache == nil { return "" } if ShouldBypassCacheForProxyMarket(q) { s.cache.RecordBypass() return "BYPASS" } if len(env.Results) == 0 { s.cache.RecordBypass() return "BYPASS" } cacheEngines := s.megaCacheableEngines(enginesToUse) if len(cacheEngines) == 0 { s.cache.RecordBypass() return "BYPASS" } if s.cacheJSON(s.buildMegaCacheKey(action, cacheEngines, q, cfg), env) { return "MISS" } return "BYPASS" } func (s *Server) cacheMegaImageResults(action string, enginesToUse []SearchEngine, q Query, env *ImageEnvelope, cfg megaRunConfig) string { if s.cache == nil { return "" } if ShouldBypassCacheForProxyMarket(q) { s.cache.RecordBypass() return "BYPASS" } if len(env.Results) == 0 { s.cache.RecordBypass() return "BYPASS" } cacheEngines := s.megaCacheableEngines(enginesToUse) if len(cacheEngines) == 0 { s.cache.RecordBypass() return "BYPASS" } if s.cacheJSON(s.buildMegaCacheKey(action, cacheEngines, q, cfg), env) { return "MISS" } return "BYPASS" } func (s *Server) buildMegaCacheKey(action string, engines []SearchEngine, q Query, cfg megaRunConfig) string { names := make([]string, 0, len(engines)) for _, eng := range engines { names = append(names, strings.ToLower(strings.TrimSpace(eng.Name()))) } sort.Strings(names) // Deduplicate engine names in key to keep cache stable when order differs // or repeated names are passed in the engines query parameter. uniq := names[:0] last := "" for _, name := range names { if name == last { continue } uniq = append(uniq, name) last = name } prefix := fmt.Sprintf("mega:%s:%t:%t:%s", cfg.Mode, cfg.Merge, cfg.Dedupe, strings.Join(uniq, ",")) return BuildCacheKey(prefix, action, q) } func (s *Server) megaCacheableEngines(engines []SearchEngine) []SearchEngine { open := make(map[string]bool) for _, stat := range s.resilient.GetCircuitBreakerStats() { name, _ := stat["engine"].(string) state, _ := stat["state"].(string) if strings.EqualFold(state, "open") { open[strings.ToLower(strings.TrimSpace(name))] = true } } cacheable := make([]SearchEngine, 0, len(engines)) for _, eng := range engines { if open[strings.ToLower(strings.TrimSpace(eng.Name()))] { continue } cacheable = append(cacheable, eng) } return cacheable } func (s *Server) cacheStatsPayload() interface{} { if s.cache == nil { return map[string]interface{}{"status": false} } return s.cache.Stats() } func (s *Server) applyProxyHeaders(c *fiber.Ctx, meta ProxyExecutionMeta) { mode := meta.Mode if mode == "" { mode = ProxyModeOff } tag := meta.Tag used := meta.Used if mode == ProxyModeOff { tag = "" used = "direct" } c.Set("X-Proxy-Mode", mode) if tag != "" { c.Set("X-Proxy-Tag", tag) } c.Set("X-Proxy-Used", used) } func setNetworkBytesHeader(c *fiber.Ctx, ctx context.Context) { c.Set("X-Network-Bytes", strconv.FormatInt(NetworkBytesFromContext(ctx), 10)) } func setBrowserProfileHeader(c *fiber.Ctx, ctx context.Context) { profileIDs := BrowserProfileIDsFromContext(ctx) if len(profileIDs) == 0 { return } c.Set(browserProfileIDHeader, strings.Join(profileIDs, ",")) } func withRequestUsage(ctx context.Context, engine string) context.Context { return WithBrowserProfileUsage(WithNetworkUsage(WithEngine(ctx, engine))) } func (s *Server) validateRequestProxyURL(q *Query) error { if q == nil || strings.TrimSpace(q.ProxyURL) == "" || q.ProxyOverride == ProxyOverrideDirect { return nil } if !s.opts.Resilience.Proxy.Proxies.AllowRequestProxyURL { return &APIError{ HTTPStatus: fiber.StatusBadRequest, ErrorCode: "bad_request", Reason: ReasonRequestProxyURLDisabled, Message: "X-Proxy-URL is disabled by server configuration", } } if s.opts.Resilience.Proxy.Runtime == ProxyRuntimeBrowser && IsAuthenticatedSocksProxyURL(q.ProxyURL) { return &APIError{ HTTPStatus: fiber.StatusBadRequest, ErrorCode: "bad_request", Reason: ReasonUnsupportedProxyScheme, Message: "authenticated SOCKS proxies are not supported in browser mode", } } return nil } func (s *Server) handleOpenAPISpec(c *fiber.Ctx) error { c.Set("Content-Type", "application/yaml; charset=utf-8") return c.Send(apidocs.OpenAPIYAML) } func (s *Server) handleSwaggerUI(c *fiber.Ctx) error { const specPath = "/openapi.yaml" page := fmt.Sprintf(` OpenSERP API Docs
`, html.EscapeString(specPath)) c.Set("Content-Type", "text/html; charset=utf-8") return c.SendString(page) } // resolveFormat returns the output format from ?format= or Accept header. // Supported values: "json" (default), "markdown", "text", "ndjson". func resolveFormat(c *fiber.Ctx) (string, error) { raw := strings.ToLower(strings.TrimSpace(c.Query("format", ""))) if raw == "" { accept := strings.ToLower(c.Get("Accept")) switch { case strings.Contains(accept, "text/markdown"): raw = "markdown" case strings.Contains(accept, "text/plain"): raw = "text" case strings.Contains(accept, "application/x-ndjson"): raw = "ndjson" default: raw = "json" } } switch raw { case "json", "markdown", "text", "ndjson": return raw, nil } return "", &APIError{HTTPStatus: 400, Reason: ReasonUnknownFormat, Message: fmt.Sprintf("unknown format %q: accepted values are json, markdown, text, ndjson", raw)} } // sendEnvelope serialises env according to the requested format and writes the // response. For non-JSON formats the envelope is NOT cached because format // variants would pollute the JSON cache. func sendEnvelope(c *fiber.Ctx, format string, env *Envelope) error { switch format { case "markdown": c.Set("Content-Type", "text/markdown; charset=utf-8") return c.Send(RenderMarkdown(env)) case "text": c.Set("Content-Type", "text/plain; charset=utf-8") return c.Send(RenderText(env)) case "ndjson": c.Set("Content-Type", "application/x-ndjson; charset=utf-8") return c.Send(RenderNDJSON(env)) default: return c.JSON(env) } } // sendImageEnvelope is sendEnvelope for ImageEnvelope. func sendImageEnvelope(c *fiber.Ctx, format string, env *ImageEnvelope) error { switch format { case "markdown": c.Set("Content-Type", "text/markdown; charset=utf-8") return c.Send(RenderMarkdownImage(env)) case "text": c.Set("Content-Type", "text/plain; charset=utf-8") return c.Send(RenderTextImage(env)) case "ndjson": c.Set("Content-Type", "application/x-ndjson; charset=utf-8") return c.Send(RenderNDJSONImage(env)) default: return c.JSON(env) } } // SetDraining controls readiness state exposed by /ready. func (s *Server) SetDraining(draining bool) { s.draining.Store(draining) } const defaultShutdownTimeout = 30 * time.Second // Listen starts the Fiber HTTP server on the configured address. func (s *Server) Listen() error { s.SetDraining(false) return s.app.Listen(s.addr) } // Shutdown gracefully stops the Fiber HTTP server. func (s *Server) Shutdown() error { return s.ShutdownWithTimeout(defaultShutdownTimeout) } // ShutdownWithTimeout drains the server before force-closing active // connections when timeout is exceeded. func (s *Server) ShutdownWithTimeout(timeout time.Duration) error { if timeout <= 0 { timeout = defaultShutdownTimeout } s.SetDraining(true) return s.app.ShutdownWithTimeout(timeout) }