feat(browser-fetch): add lightpanda-backed fetch service and knowledge URL fallback
Adds a new browser-fetch microservice that wraps Lightpanda for rendering JS-heavy pages, and wires it into the knowledge URL parser as a fallback when the upstream link reader returns 403/429/5xx for an allow-listed domain (e.g. zhuanlan.zhihu.com). Includes config/env plumbing, hot-reload diff support, and full deploy assets (Dockerfile target, docker-compose, k3s manifests, image-build/package scripts).
This commit is contained in:
@@ -14,6 +14,7 @@ JWT_SECRET=geo-web-Aa123456-change-me-in-production
|
||||
OPS_JWT_SECRET=geo-ops-Aa123456-change-me-in-production
|
||||
OPS_DEFAULT_ADMIN_PASSWORD=Aa123456
|
||||
SCHEDULER_INTERNAL_METRICS_TOKEN=geo-scheduler-Aa123456-change-me-in-production
|
||||
BROWSER_FETCH_TOKEN=geo-browser-fetch-Aa123456-change-me-in-production
|
||||
POSTGRES_PASSWORD=geo_dev
|
||||
|
||||
# ─── Optional: MinIO credentials ─────────────────────────────────────────────
|
||||
|
||||
+19
-1
@@ -215,4 +215,22 @@ generation:
|
||||
task_recovery_batch_size: 100
|
||||
task_queued_stale_after: 2m
|
||||
task_max_attempts: 3
|
||||
|
||||
|
||||
browser_fetch:
|
||||
enabled: true
|
||||
provider: lightpanda
|
||||
service_url: http://browser-fetch:8082
|
||||
queue_size: 64
|
||||
worker_concurrency: 2
|
||||
request_timeout: 45s
|
||||
cache_ttl: 6h
|
||||
allowed_domains:
|
||||
- zhuanlan.zhihu.com
|
||||
- zhihu.com
|
||||
trigger_status:
|
||||
- 403
|
||||
- 429
|
||||
- 500
|
||||
- 502
|
||||
- 503
|
||||
- 504
|
||||
|
||||
@@ -18,6 +18,11 @@ services:
|
||||
pull_policy: never
|
||||
build: !reset null
|
||||
|
||||
browser-fetch:
|
||||
image: geo-rankly/browser-fetch:${IMAGE_TAG:-latest}
|
||||
pull_policy: never
|
||||
build: !reset null
|
||||
|
||||
worker-generate:
|
||||
image: geo-rankly/worker-generate:${IMAGE_TAG:-latest}
|
||||
pull_policy: never
|
||||
|
||||
@@ -16,6 +16,7 @@ x-app-env: &app-env
|
||||
JWT_ACCESS_TTL: ${JWT_ACCESS_TTL:-15m}
|
||||
JWT_REFRESH_TTL: ${JWT_REFRESH_TTL:-720h}
|
||||
SCHEDULER_INTERNAL_METRICS_TOKEN: ${SCHEDULER_INTERNAL_METRICS_TOKEN:-}
|
||||
BROWSER_FETCH_TOKEN: ${BROWSER_FETCH_TOKEN:-}
|
||||
OBJECT_STORAGE_ACCESS_KEY: ${MINIO_ROOT_USER:-minioadmin}
|
||||
OBJECT_STORAGE_SECRET_KEY: ${MINIO_ROOT_PASSWORD:-minioadmin}
|
||||
|
||||
@@ -261,6 +262,33 @@ services:
|
||||
retries: 5
|
||||
start_period: 10s
|
||||
|
||||
browser-fetch:
|
||||
image: ${IMAGE_REGISTRY:-ghcr.io/your-org/geo-rankly}/browser-fetch:${IMAGE_TAG:-latest}
|
||||
container_name: geo-browser-fetch
|
||||
build:
|
||||
context: ../server
|
||||
dockerfile: Dockerfile
|
||||
target: browser-fetch
|
||||
args:
|
||||
SERVICE: browser-fetch
|
||||
restart: unless-stopped
|
||||
environment:
|
||||
BROWSER_FETCH_TOKEN: ${BROWSER_FETCH_TOKEN:-geo-browser-fetch-change-me}
|
||||
BROWSER_FETCH_ALLOWED_DOMAINS: zhuanlan.zhihu.com,zhihu.com
|
||||
BROWSER_FETCH_QUEUE_SIZE: 128
|
||||
BROWSER_FETCH_WORKER_CONCURRENCY: 2
|
||||
BROWSER_FETCH_PER_DOMAIN_CONCURRENCY: 1
|
||||
BROWSER_FETCH_REQUEST_TIMEOUT: 45s
|
||||
BROWSER_FETCH_CACHE_TTL: 6h
|
||||
LIGHTPANDA_DRIVER: fetch
|
||||
LIGHTPANDA_FETCH_ARGS: --dump markdown
|
||||
healthcheck:
|
||||
test: ["/app/service", "-healthcheck"]
|
||||
interval: 10s
|
||||
timeout: 5s
|
||||
retries: 5
|
||||
start_period: 10s
|
||||
|
||||
worker-generate:
|
||||
image: ${IMAGE_REGISTRY:-ghcr.io/your-org/geo-rankly}/worker-generate:${IMAGE_TAG:-latest}
|
||||
container_name: geo-worker-generate
|
||||
|
||||
@@ -4,7 +4,7 @@
|
||||
|
||||
- infra: `postgres`, `monitoring-postgres`, `pgbouncer`, `monitoring-pgbouncer`, `rabbitmq`, `redis`, `qdrant`, `minio`
|
||||
- jobs: `migrate`, `minio-init`
|
||||
- apps: `tenant-api`, `worker-generate`, `kol-assist-worker`, `scheduler`, `ops-api`, `frontend`, `ops-web`
|
||||
- apps: `tenant-api`, `browser-fetch`, `worker-generate`, `kol-assist-worker`, `scheduler`, `ops-api`, `frontend`, `ops-web`
|
||||
- ingress: `geo-rankly.local` -> `frontend`, `ops.geo-rankly.local` -> `ops-web`
|
||||
|
||||
## 1. 准备镜像
|
||||
@@ -34,6 +34,7 @@ sudo k3s ctr images import images.tar
|
||||
- `OPS_JWT_SECRET`
|
||||
- `OPS_DEFAULT_ADMIN_PASSWORD`
|
||||
- `SCHEDULER_INTERNAL_METRICS_TOKEN`
|
||||
- `BROWSER_FETCH_TOKEN`
|
||||
- `POSTGRES_PASSWORD` / `RABBITMQ_DEFAULT_PASS` / `MINIO_ROOT_*`
|
||||
- `config/config.yaml` 里的 LLM / SiliconFlow API key
|
||||
- `config.local.yaml` 里对应的数据库、RabbitMQ、MinIO、JWT 配置
|
||||
@@ -131,6 +132,7 @@ kubectl -n geo-rankly port-forward svc/minio 9001:9001
|
||||
kubectl -n geo-rankly get all
|
||||
kubectl -n geo-rankly describe pod <pod-name>
|
||||
kubectl -n geo-rankly logs deploy/tenant-api -f
|
||||
kubectl -n geo-rankly logs deploy/browser-fetch -f
|
||||
kubectl -n geo-rankly logs deploy/scheduler -f
|
||||
kubectl -n geo-rankly logs deploy/worker-generate -f
|
||||
kubectl -n geo-rankly logs deploy/ops-api -f
|
||||
|
||||
@@ -1,5 +1,89 @@
|
||||
apiVersion: v1
|
||||
kind: Service
|
||||
metadata:
|
||||
name: browser-fetch
|
||||
labels:
|
||||
app.kubernetes.io/name: browser-fetch
|
||||
spec:
|
||||
selector:
|
||||
app.kubernetes.io/name: browser-fetch
|
||||
ports:
|
||||
- name: http
|
||||
port: 8082
|
||||
targetPort: http
|
||||
---
|
||||
apiVersion: apps/v1
|
||||
kind: Deployment
|
||||
metadata:
|
||||
name: browser-fetch
|
||||
labels:
|
||||
app.kubernetes.io/name: browser-fetch
|
||||
spec:
|
||||
replicas: 1
|
||||
selector:
|
||||
matchLabels:
|
||||
app.kubernetes.io/name: browser-fetch
|
||||
template:
|
||||
metadata:
|
||||
labels:
|
||||
app.kubernetes.io/name: browser-fetch
|
||||
spec:
|
||||
containers:
|
||||
- name: browser-fetch
|
||||
image: geo-rankly/browser-fetch:latest
|
||||
imagePullPolicy: IfNotPresent
|
||||
ports:
|
||||
- name: http
|
||||
containerPort: 8082
|
||||
env:
|
||||
- name: TZ
|
||||
value: Asia/Shanghai
|
||||
- name: BROWSER_FETCH_TOKEN
|
||||
valueFrom:
|
||||
secretKeyRef:
|
||||
name: geo-rankly-app-secret
|
||||
key: BROWSER_FETCH_TOKEN
|
||||
- name: BROWSER_FETCH_ALLOWED_DOMAINS
|
||||
value: zhuanlan.zhihu.com,zhihu.com
|
||||
- name: BROWSER_FETCH_QUEUE_SIZE
|
||||
value: "128"
|
||||
- name: BROWSER_FETCH_WORKER_CONCURRENCY
|
||||
value: "2"
|
||||
- name: BROWSER_FETCH_PER_DOMAIN_CONCURRENCY
|
||||
value: "1"
|
||||
- name: BROWSER_FETCH_REQUEST_TIMEOUT
|
||||
value: 45s
|
||||
- name: BROWSER_FETCH_CACHE_TTL
|
||||
value: 6h
|
||||
- name: LIGHTPANDA_DRIVER
|
||||
value: fetch
|
||||
- name: LIGHTPANDA_FETCH_ARGS
|
||||
value: --dump markdown
|
||||
readinessProbe:
|
||||
exec:
|
||||
command:
|
||||
- /app/service
|
||||
- -healthcheck
|
||||
initialDelaySeconds: 5
|
||||
periodSeconds: 10
|
||||
timeoutSeconds: 3
|
||||
livenessProbe:
|
||||
httpGet:
|
||||
path: /health/live
|
||||
port: http
|
||||
initialDelaySeconds: 15
|
||||
periodSeconds: 20
|
||||
timeoutSeconds: 3
|
||||
resources:
|
||||
requests:
|
||||
cpu: 500m
|
||||
memory: 512Mi
|
||||
limits:
|
||||
cpu: "2"
|
||||
memory: 2Gi
|
||||
---
|
||||
apiVersion: v1
|
||||
kind: Service
|
||||
metadata:
|
||||
name: tenant-api
|
||||
labels:
|
||||
@@ -60,6 +144,11 @@ spec:
|
||||
secretKeyRef:
|
||||
name: geo-rankly-app-secret
|
||||
key: MINIO_ROOT_PASSWORD
|
||||
- name: BROWSER_FETCH_TOKEN
|
||||
valueFrom:
|
||||
secretKeyRef:
|
||||
name: geo-rankly-app-secret
|
||||
key: BROWSER_FETCH_TOKEN
|
||||
- name: SCHEDULER_INTERNAL_METRICS_TOKEN
|
||||
valueFrom:
|
||||
secretKeyRef:
|
||||
@@ -152,6 +241,11 @@ spec:
|
||||
secretKeyRef:
|
||||
name: geo-rankly-app-secret
|
||||
key: MINIO_ROOT_PASSWORD
|
||||
- name: BROWSER_FETCH_TOKEN
|
||||
valueFrom:
|
||||
secretKeyRef:
|
||||
name: geo-rankly-app-secret
|
||||
key: BROWSER_FETCH_TOKEN
|
||||
resources:
|
||||
requests:
|
||||
cpu: 200m
|
||||
|
||||
@@ -129,8 +129,8 @@ llm:
|
||||
provider: ark
|
||||
base_url: https://ark.cn-beijing.volces.com/api/v3
|
||||
api_key: "7fb6c66b-129c-4935-9617-709236c4205a"
|
||||
model: doubao-seed-2-0-lite-260215
|
||||
knowledge_url_model: doubao-seed-2-0-mini-260215
|
||||
model: doubao-seed-2-0-pro-260215 # doubao-seed-2-0-lite-260215 #
|
||||
knowledge_url_model: doubao-seed-2-0-mini-260215 #doubao-seed-2-0-mini-260215
|
||||
timeout: 2m
|
||||
max_output_tokens: 16000
|
||||
temperature: 0.7
|
||||
@@ -172,3 +172,22 @@ generation:
|
||||
task_recovery_batch_size: 100
|
||||
task_queued_stale_after: 2m
|
||||
task_max_attempts: 3
|
||||
|
||||
browser_fetch:
|
||||
enabled: true
|
||||
provider: lightpanda
|
||||
service_url: http://browser-fetch:8082
|
||||
queue_size: 64
|
||||
worker_concurrency: 2
|
||||
request_timeout: 45s
|
||||
cache_ttl: 6h
|
||||
allowed_domains:
|
||||
- zhuanlan.zhihu.com
|
||||
- zhihu.com
|
||||
trigger_status:
|
||||
- 403
|
||||
- 429
|
||||
- 500
|
||||
- 502
|
||||
- 503
|
||||
- 504
|
||||
|
||||
@@ -37,6 +37,9 @@ images:
|
||||
- name: geo-rankly/tenant-api
|
||||
newName: geo-rankly/tenant-api
|
||||
newTag: latest
|
||||
- name: geo-rankly/browser-fetch
|
||||
newName: geo-rankly/browser-fetch
|
||||
newTag: latest
|
||||
- name: geo-rankly/worker-generate
|
||||
newName: geo-rankly/worker-generate
|
||||
newTag: latest
|
||||
|
||||
@@ -12,6 +12,7 @@ stringData:
|
||||
OPS_JWT_SECRET: geo-ops-change-me-in-production
|
||||
OPS_DEFAULT_ADMIN_PASSWORD: Aa123456
|
||||
SCHEDULER_INTERNAL_METRICS_TOKEN: geo-metrics-token
|
||||
BROWSER_FETCH_TOKEN: geo-browser-fetch-change-me
|
||||
config.local.yaml: |
|
||||
database:
|
||||
password: ${POSTGRES_PASSWORD:geo_dev}
|
||||
@@ -31,3 +32,6 @@ stringData:
|
||||
|
||||
scheduler:
|
||||
internal_metrics_token: "geo-metrics-token"
|
||||
|
||||
browser_fetch:
|
||||
token: ${BROWSER_FETCH_TOKEN:geo-browser-fetch-change-me}
|
||||
|
||||
@@ -16,7 +16,7 @@ REGISTRY_PASSWORD="${REGISTRY_PASSWORD:-}"
|
||||
REGISTRY_PLAIN_HTTP="${REGISTRY_PLAIN_HTTP:-true}"
|
||||
REGISTRY_SCHEME="${REGISTRY_SCHEME:-}"
|
||||
IMAGE_TAG="${IMAGE_TAG:-}"
|
||||
IMAGE_LIST="${IMAGE_LIST:-tenant-api ops-api worker-generate kol-assist-worker scheduler frontend ops-web migrate}"
|
||||
IMAGE_LIST="${IMAGE_LIST:-tenant-api browser-fetch ops-api worker-generate kol-assist-worker scheduler frontend ops-web migrate}"
|
||||
DELETE_OLD_IMAGE_TAGS="${DELETE_OLD_IMAGE_TAGS:-true}"
|
||||
PUSH_LATEST="${PUSH_LATEST:-true}"
|
||||
FORCE_BUILD="${FORCE_BUILD:-false}"
|
||||
@@ -257,6 +257,15 @@ build_image() {
|
||||
-f "${REPO_ROOT}/server/Dockerfile" \
|
||||
"${REPO_ROOT}/server"
|
||||
;;
|
||||
browser-fetch)
|
||||
docker build "${platform_args[@]}" \
|
||||
--target browser-fetch \
|
||||
--build-arg "SERVICE=browser-fetch" \
|
||||
--tag "${repo}:${IMAGE_TAG}" \
|
||||
--tag "${repo}:latest" \
|
||||
-f "${REPO_ROOT}/server/Dockerfile" \
|
||||
"${REPO_ROOT}/server"
|
||||
;;
|
||||
tenant-api|worker-generate|kol-assist-worker|scheduler|ops-api)
|
||||
docker build "${platform_args[@]}" \
|
||||
--target runtime \
|
||||
|
||||
@@ -59,7 +59,7 @@ cd "${REPO_ROOT}"
|
||||
|
||||
if [[ -n "${SOURCE_IMAGE_REGISTRY}" ]]; then
|
||||
echo " pulling from ${SOURCE_IMAGE_REGISTRY}"
|
||||
for svc in tenant-api worker-generate kol-assist-worker ops-api scheduler migrate frontend ops-web; do
|
||||
for svc in tenant-api browser-fetch worker-generate kol-assist-worker ops-api scheduler migrate frontend ops-web; do
|
||||
echo " pulling ${SOURCE_IMAGE_REGISTRY}/${svc}:${IMAGE_TAG}"
|
||||
docker pull --platform "${PLATFORM}" "${SOURCE_IMAGE_REGISTRY}/${svc}:${IMAGE_TAG}"
|
||||
docker tag "${SOURCE_IMAGE_REGISTRY}/${svc}:${IMAGE_TAG}" "${LOCAL_PREFIX}/${svc}:${IMAGE_TAG}"
|
||||
@@ -81,6 +81,7 @@ else
|
||||
}
|
||||
|
||||
_build runtime tenant-api --build-arg SERVICE=tenant-api
|
||||
_build browser-fetch browser-fetch --build-arg SERVICE=browser-fetch
|
||||
_build runtime worker-generate --build-arg SERVICE=worker-generate
|
||||
_build runtime kol-assist-worker --build-arg SERVICE=kol-assist-worker
|
||||
_build runtime ops-api --build-arg SERVICE=ops-api
|
||||
@@ -127,6 +128,7 @@ mkdir -p "${WORK_DIR}"
|
||||
|
||||
ALL_IMAGES=(
|
||||
"${LOCAL_PREFIX}/tenant-api:${IMAGE_TAG}"
|
||||
"${LOCAL_PREFIX}/browser-fetch:${IMAGE_TAG}"
|
||||
"${LOCAL_PREFIX}/worker-generate:${IMAGE_TAG}"
|
||||
"${LOCAL_PREFIX}/kol-assist-worker:${IMAGE_TAG}"
|
||||
"${LOCAL_PREFIX}/scheduler:${IMAGE_TAG}"
|
||||
|
||||
@@ -23,6 +23,12 @@ RUN --mount=type=cache,target=/go/pkg/mod \
|
||||
go build -trimpath -ldflags="-s -w" \
|
||||
-o /bin/service ./cmd/${SERVICE}
|
||||
|
||||
RUN --mount=type=cache,target=/go/pkg/mod \
|
||||
--mount=type=cache,target=/root/.cache/go-build \
|
||||
CGO_ENABLED=0 GOOS="${TARGETOS:-$(go env GOOS)}" GOARCH="${TARGETARCH:-$(go env GOARCH)}" \
|
||||
go build -trimpath -ldflags="-s -w" \
|
||||
-o /bin/browser-fetch ./cmd/browser-fetch
|
||||
|
||||
RUN --mount=type=cache,target=/go/pkg/mod \
|
||||
--mount=type=cache,target=/root/.cache/go-build \
|
||||
CGO_ENABLED=1 GOOS="${TARGETOS:-$(go env GOOS)}" GOARCH="${TARGETARCH:-$(go env GOARCH)}" \
|
||||
@@ -53,3 +59,13 @@ COPY configs/ /app/configs/
|
||||
|
||||
EXPOSE 8080 8081
|
||||
ENTRYPOINT ["/app/service"]
|
||||
|
||||
# ─── Stage 4: Browser fetch runtime (bundles Lightpanda) ──────────────────────
|
||||
FROM lightpanda/browser:nightly AS browser-fetch
|
||||
|
||||
WORKDIR /app
|
||||
|
||||
COPY --from=builder /bin/browser-fetch /app/service
|
||||
|
||||
EXPOSE 8082
|
||||
ENTRYPOINT ["/app/service"]
|
||||
|
||||
@@ -0,0 +1,877 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"crypto/subtle"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"log"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"os"
|
||||
"os/exec"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
)
|
||||
|
||||
const (
|
||||
defaultPort = 8082
|
||||
defaultQueueSize = 128
|
||||
defaultWorkerConcurrency = 2
|
||||
defaultPerDomainConcurrency = 1
|
||||
defaultRequestTimeout = 45 * time.Second
|
||||
defaultCacheTTL = 6 * time.Hour
|
||||
defaultBreakerThreshold = 5
|
||||
defaultBreakerCooldown = 2 * time.Minute
|
||||
)
|
||||
|
||||
type fetchConfig struct {
|
||||
Port int
|
||||
Token string
|
||||
QueueSize int
|
||||
WorkerConcurrency int
|
||||
PerDomainConcurrency int
|
||||
RequestTimeout time.Duration
|
||||
CacheTTL time.Duration
|
||||
AllowedDomains []string
|
||||
LightpandaDriver string
|
||||
LightpandaBin string
|
||||
LightpandaArgs []string
|
||||
LightpandaFetchArgs []string
|
||||
WorkerMaxRequests int
|
||||
BreakerThreshold int
|
||||
BreakerCooldown time.Duration
|
||||
}
|
||||
|
||||
type fetchService struct {
|
||||
cfg fetchConfig
|
||||
jobs chan fetchJob
|
||||
cache *memoryCache
|
||||
limiters *domainLimiters
|
||||
breakers *domainBreakers
|
||||
metrics *fetchMetrics
|
||||
shuttingDown atomic.Bool
|
||||
}
|
||||
|
||||
type fetchJob struct {
|
||||
rawURL string
|
||||
timeout time.Duration
|
||||
host string
|
||||
release func()
|
||||
resp chan fetchJobResult
|
||||
}
|
||||
|
||||
type fetchJobResult struct {
|
||||
result fetchResponse
|
||||
err error
|
||||
}
|
||||
|
||||
type fetchRequest struct {
|
||||
URL string `json:"url"`
|
||||
TimeoutMS int64 `json:"timeout_ms,omitempty"`
|
||||
Options map[string]any `json:"options,omitempty"`
|
||||
}
|
||||
|
||||
type fetchResponse struct {
|
||||
URL string `json:"url,omitempty"`
|
||||
Title string `json:"title,omitempty"`
|
||||
Text string `json:"text,omitempty"`
|
||||
Markdown string `json:"markdown,omitempty"`
|
||||
}
|
||||
|
||||
type memoryCacheEntry struct {
|
||||
result fetchResponse
|
||||
expiresAt time.Time
|
||||
}
|
||||
|
||||
type memoryCache struct {
|
||||
mu sync.Mutex
|
||||
entries map[string]memoryCacheEntry
|
||||
}
|
||||
|
||||
type domainLimiters struct {
|
||||
mu sync.Mutex
|
||||
capacity int
|
||||
items map[string]chan struct{}
|
||||
}
|
||||
|
||||
type domainBreakerState struct {
|
||||
failures int
|
||||
openedAt time.Time
|
||||
}
|
||||
|
||||
type domainBreakers struct {
|
||||
mu sync.Mutex
|
||||
threshold int
|
||||
cooldown time.Duration
|
||||
items map[string]domainBreakerState
|
||||
}
|
||||
|
||||
type fetchMetrics struct {
|
||||
accepted atomic.Int64
|
||||
cacheHits atomic.Int64
|
||||
queueFull atomic.Int64
|
||||
domainLimited atomic.Int64
|
||||
authFailed atomic.Int64
|
||||
success atomic.Int64
|
||||
failed atomic.Int64
|
||||
breakerOpen atomic.Int64
|
||||
}
|
||||
|
||||
type lightpandaWorker struct {
|
||||
id int
|
||||
cfg fetchConfig
|
||||
client *mcpClient
|
||||
handled int
|
||||
}
|
||||
|
||||
type mcpClient struct {
|
||||
cfg fetchConfig
|
||||
cmd *exec.Cmd
|
||||
stdin io.WriteCloser
|
||||
stdout *bufio.Scanner
|
||||
stderr bytes.Buffer
|
||||
nextID int64
|
||||
started bool
|
||||
}
|
||||
|
||||
type mcpResponse struct {
|
||||
JSONRPC string `json:"jsonrpc"`
|
||||
ID int64 `json:"id,omitempty"`
|
||||
Result json.RawMessage `json:"result,omitempty"`
|
||||
Error *mcpError `json:"error,omitempty"`
|
||||
}
|
||||
|
||||
type mcpError struct {
|
||||
Code int `json:"code"`
|
||||
Message string `json:"message"`
|
||||
}
|
||||
|
||||
type mcpToolResult struct {
|
||||
Content []mcpContentItem `json:"content"`
|
||||
IsError bool `json:"isError"`
|
||||
}
|
||||
|
||||
type mcpContentItem struct {
|
||||
Type string `json:"type"`
|
||||
Text string `json:"text"`
|
||||
}
|
||||
|
||||
func main() {
|
||||
cfg := loadFetchConfig()
|
||||
if len(os.Args) > 1 && os.Args[1] == "-healthcheck" {
|
||||
runHealthcheck(cfg)
|
||||
return
|
||||
}
|
||||
service := newFetchService(cfg)
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
service.Start(ctx)
|
||||
|
||||
mux := http.NewServeMux()
|
||||
mux.HandleFunc("/health/live", service.handleLive)
|
||||
mux.HandleFunc("/health/ready", service.handleReady)
|
||||
mux.HandleFunc("/metrics", service.handleMetrics)
|
||||
mux.HandleFunc("/fetch", service.handleFetch)
|
||||
|
||||
server := &http.Server{
|
||||
Addr: fmt.Sprintf(":%d", cfg.Port),
|
||||
Handler: mux,
|
||||
ReadHeaderTimeout: 5 * time.Second,
|
||||
}
|
||||
|
||||
log.Printf("browser-fetch listening on :%d workers=%d queue=%d domains=%v", cfg.Port, cfg.WorkerConcurrency, cfg.QueueSize, cfg.AllowedDomains)
|
||||
if err := server.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
|
||||
log.Fatalf("browser-fetch server stopped: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func runHealthcheck(cfg fetchConfig) {
|
||||
client := &http.Client{Timeout: 2 * time.Second}
|
||||
resp, err := client.Get(fmt.Sprintf("http://127.0.0.1:%d/health/ready", cfg.Port))
|
||||
if err != nil {
|
||||
log.Fatalf("healthcheck failed: %v", err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
if resp.StatusCode >= http.StatusBadRequest {
|
||||
log.Fatalf("healthcheck failed: status=%d", resp.StatusCode)
|
||||
}
|
||||
}
|
||||
|
||||
func loadFetchConfig() fetchConfig {
|
||||
cfg := fetchConfig{
|
||||
Port: envInt("BROWSER_FETCH_PORT", defaultPort),
|
||||
Token: strings.TrimSpace(os.Getenv("BROWSER_FETCH_TOKEN")),
|
||||
QueueSize: envInt("BROWSER_FETCH_QUEUE_SIZE", defaultQueueSize),
|
||||
WorkerConcurrency: envInt("BROWSER_FETCH_WORKER_CONCURRENCY", defaultWorkerConcurrency),
|
||||
PerDomainConcurrency: envInt("BROWSER_FETCH_PER_DOMAIN_CONCURRENCY", defaultPerDomainConcurrency),
|
||||
RequestTimeout: envDuration("BROWSER_FETCH_REQUEST_TIMEOUT", defaultRequestTimeout),
|
||||
CacheTTL: envDuration("BROWSER_FETCH_CACHE_TTL", defaultCacheTTL),
|
||||
AllowedDomains: splitCSV(os.Getenv("BROWSER_FETCH_ALLOWED_DOMAINS")),
|
||||
LightpandaDriver: strings.ToLower(strings.TrimSpace(envString("LIGHTPANDA_DRIVER", "fetch"))),
|
||||
LightpandaBin: strings.TrimSpace(envString("LIGHTPANDA_BIN", "lightpanda")),
|
||||
LightpandaArgs: splitCommandArgs(envString("LIGHTPANDA_ARGS", "mcp")),
|
||||
LightpandaFetchArgs: splitCommandArgs(envString("LIGHTPANDA_FETCH_ARGS", "--dump markdown")),
|
||||
WorkerMaxRequests: envInt("LIGHTPANDA_WORKER_MAX_REQUESTS", 100),
|
||||
BreakerThreshold: envInt("BROWSER_FETCH_BREAKER_THRESHOLD", defaultBreakerThreshold),
|
||||
BreakerCooldown: envDuration("BROWSER_FETCH_BREAKER_COOLDOWN", defaultBreakerCooldown),
|
||||
}
|
||||
if cfg.QueueSize <= 0 {
|
||||
cfg.QueueSize = defaultQueueSize
|
||||
}
|
||||
if cfg.WorkerConcurrency <= 0 {
|
||||
cfg.WorkerConcurrency = defaultWorkerConcurrency
|
||||
}
|
||||
if cfg.PerDomainConcurrency <= 0 {
|
||||
cfg.PerDomainConcurrency = defaultPerDomainConcurrency
|
||||
}
|
||||
if cfg.RequestTimeout <= 0 {
|
||||
cfg.RequestTimeout = defaultRequestTimeout
|
||||
}
|
||||
if cfg.CacheTTL < 0 {
|
||||
cfg.CacheTTL = 0
|
||||
}
|
||||
if cfg.BreakerThreshold <= 0 {
|
||||
cfg.BreakerThreshold = defaultBreakerThreshold
|
||||
}
|
||||
if cfg.BreakerCooldown <= 0 {
|
||||
cfg.BreakerCooldown = defaultBreakerCooldown
|
||||
}
|
||||
if cfg.LightpandaDriver == "" {
|
||||
cfg.LightpandaDriver = "fetch"
|
||||
}
|
||||
return cfg
|
||||
}
|
||||
|
||||
func newFetchService(cfg fetchConfig) *fetchService {
|
||||
return &fetchService{
|
||||
cfg: cfg,
|
||||
jobs: make(chan fetchJob, cfg.QueueSize),
|
||||
cache: &memoryCache{entries: make(map[string]memoryCacheEntry)},
|
||||
limiters: &domainLimiters{capacity: cfg.PerDomainConcurrency, items: make(map[string]chan struct{})},
|
||||
breakers: &domainBreakers{threshold: cfg.BreakerThreshold, cooldown: cfg.BreakerCooldown, items: make(map[string]domainBreakerState)},
|
||||
metrics: &fetchMetrics{},
|
||||
}
|
||||
}
|
||||
|
||||
func (s *fetchService) Start(ctx context.Context) {
|
||||
for i := 0; i < s.cfg.WorkerConcurrency; i++ {
|
||||
worker := &lightpandaWorker{id: i, cfg: s.cfg}
|
||||
go worker.Run(ctx, s.jobs, s.metrics, s.breakers)
|
||||
}
|
||||
}
|
||||
|
||||
func (s *fetchService) handleLive(w http.ResponseWriter, _ *http.Request) {
|
||||
writeJSON(w, http.StatusOK, map[string]string{"status": "alive"})
|
||||
}
|
||||
|
||||
func (s *fetchService) handleReady(w http.ResponseWriter, _ *http.Request) {
|
||||
if s.cfg.Token == "" {
|
||||
writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "BROWSER_FETCH_TOKEN is required"})
|
||||
return
|
||||
}
|
||||
if len(s.cfg.AllowedDomains) == 0 {
|
||||
writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "BROWSER_FETCH_ALLOWED_DOMAINS is required"})
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, map[string]string{"status": "ready"})
|
||||
}
|
||||
|
||||
func (s *fetchService) handleMetrics(w http.ResponseWriter, r *http.Request) {
|
||||
if !s.authorized(r) {
|
||||
s.metrics.authFailed.Add(1)
|
||||
writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "unauthorized"})
|
||||
return
|
||||
}
|
||||
w.Header().Set("Content-Type", "text/plain; charset=utf-8")
|
||||
_, _ = fmt.Fprintf(w, "browser_fetch_requests_accepted_total %d\n", s.metrics.accepted.Load())
|
||||
_, _ = fmt.Fprintf(w, "browser_fetch_cache_hits_total %d\n", s.metrics.cacheHits.Load())
|
||||
_, _ = fmt.Fprintf(w, "browser_fetch_queue_full_total %d\n", s.metrics.queueFull.Load())
|
||||
_, _ = fmt.Fprintf(w, "browser_fetch_domain_limited_total %d\n", s.metrics.domainLimited.Load())
|
||||
_, _ = fmt.Fprintf(w, "browser_fetch_auth_failed_total %d\n", s.metrics.authFailed.Load())
|
||||
_, _ = fmt.Fprintf(w, "browser_fetch_success_total %d\n", s.metrics.success.Load())
|
||||
_, _ = fmt.Fprintf(w, "browser_fetch_failed_total %d\n", s.metrics.failed.Load())
|
||||
_, _ = fmt.Fprintf(w, "browser_fetch_breaker_open_total %d\n", s.metrics.breakerOpen.Load())
|
||||
_, _ = fmt.Fprintf(w, "browser_fetch_queue_depth %d\n", len(s.jobs))
|
||||
}
|
||||
|
||||
func (s *fetchService) handleFetch(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Method != http.MethodPost {
|
||||
writeJSON(w, http.StatusMethodNotAllowed, map[string]string{"error": "method not allowed"})
|
||||
return
|
||||
}
|
||||
if !s.authorized(r) {
|
||||
s.metrics.authFailed.Add(1)
|
||||
writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "unauthorized"})
|
||||
return
|
||||
}
|
||||
|
||||
var req fetchRequest
|
||||
if err := json.NewDecoder(io.LimitReader(r.Body, 1<<20)).Decode(&req); err != nil {
|
||||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid json"})
|
||||
return
|
||||
}
|
||||
rawURL := strings.TrimSpace(req.URL)
|
||||
host, err := validateFetchURL(rawURL, s.cfg.AllowedDomains)
|
||||
if err != nil {
|
||||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": err.Error()})
|
||||
return
|
||||
}
|
||||
if s.breakers.IsOpen(host) {
|
||||
s.metrics.breakerOpen.Add(1)
|
||||
writeJSON(w, http.StatusTooManyRequests, map[string]string{"error": "domain circuit breaker is open"})
|
||||
return
|
||||
}
|
||||
|
||||
cacheKey := fetchCacheKey(rawURL)
|
||||
if s.cfg.CacheTTL > 0 {
|
||||
if cached, ok := s.cache.Get(cacheKey); ok {
|
||||
s.metrics.cacheHits.Add(1)
|
||||
writeJSON(w, http.StatusOK, cached)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
release, ok := s.limiters.TryAcquire(host)
|
||||
if !ok {
|
||||
s.metrics.domainLimited.Add(1)
|
||||
writeJSON(w, http.StatusTooManyRequests, map[string]string{"error": "domain concurrency limit reached"})
|
||||
return
|
||||
}
|
||||
|
||||
timeout := s.cfg.RequestTimeout
|
||||
if req.TimeoutMS > 0 {
|
||||
timeout = time.Duration(req.TimeoutMS) * time.Millisecond
|
||||
if timeout > s.cfg.RequestTimeout {
|
||||
timeout = s.cfg.RequestTimeout
|
||||
}
|
||||
}
|
||||
job := fetchJob{
|
||||
rawURL: rawURL,
|
||||
timeout: timeout,
|
||||
host: host,
|
||||
release: release,
|
||||
resp: make(chan fetchJobResult, 1),
|
||||
}
|
||||
|
||||
select {
|
||||
case s.jobs <- job:
|
||||
s.metrics.accepted.Add(1)
|
||||
default:
|
||||
release()
|
||||
s.metrics.queueFull.Add(1)
|
||||
writeJSON(w, http.StatusTooManyRequests, map[string]string{"error": "browser fetch queue is full"})
|
||||
return
|
||||
}
|
||||
|
||||
select {
|
||||
case result := <-job.resp:
|
||||
if result.err != nil {
|
||||
writeJSON(w, http.StatusBadGateway, map[string]string{"error": result.err.Error()})
|
||||
return
|
||||
}
|
||||
if s.cfg.CacheTTL > 0 {
|
||||
s.cache.Set(cacheKey, result.result, s.cfg.CacheTTL)
|
||||
}
|
||||
writeJSON(w, http.StatusOK, result.result)
|
||||
case <-r.Context().Done():
|
||||
writeJSON(w, http.StatusGatewayTimeout, map[string]string{"error": "request cancelled"})
|
||||
}
|
||||
}
|
||||
|
||||
func (s *fetchService) authorized(r *http.Request) bool {
|
||||
expected := s.cfg.Token
|
||||
if expected == "" {
|
||||
return false
|
||||
}
|
||||
value := strings.TrimSpace(r.Header.Get("Authorization"))
|
||||
value = strings.TrimPrefix(value, "Bearer ")
|
||||
if value == "" {
|
||||
value = strings.TrimSpace(r.URL.Query().Get("token"))
|
||||
}
|
||||
return subtle.ConstantTimeCompare([]byte(value), []byte(expected)) == 1
|
||||
}
|
||||
|
||||
func (w *lightpandaWorker) Run(ctx context.Context, jobs <-chan fetchJob, metrics *fetchMetrics, breakers *domainBreakers) {
|
||||
defer w.closeClient()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case job := <-jobs:
|
||||
result, err := w.Fetch(ctx, job.rawURL, job.timeout)
|
||||
job.release()
|
||||
if err != nil {
|
||||
metrics.failed.Add(1)
|
||||
breakers.RecordFailure(job.host)
|
||||
} else {
|
||||
metrics.success.Add(1)
|
||||
breakers.RecordSuccess(job.host)
|
||||
}
|
||||
job.resp <- fetchJobResult{result: result, err: err}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (w *lightpandaWorker) Fetch(parent context.Context, rawURL string, timeout time.Duration) (fetchResponse, error) {
|
||||
if timeout <= 0 {
|
||||
timeout = w.cfg.RequestTimeout
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(parent, timeout)
|
||||
defer cancel()
|
||||
|
||||
var markdown string
|
||||
var err error
|
||||
if w.cfg.LightpandaDriver == "mcp" {
|
||||
markdown, err = w.fetchViaMCP(ctx, rawURL)
|
||||
if err != nil {
|
||||
w.closeClient()
|
||||
return fetchResponse{}, err
|
||||
}
|
||||
} else {
|
||||
markdown, err = w.fetchViaCommand(ctx, rawURL)
|
||||
if err != nil {
|
||||
return fetchResponse{}, err
|
||||
}
|
||||
}
|
||||
w.handled++
|
||||
|
||||
markdown = strings.TrimSpace(markdown)
|
||||
text := normalizeText(markdown)
|
||||
if text == "" {
|
||||
return fetchResponse{}, errors.New("lightpanda returned empty markdown")
|
||||
}
|
||||
return fetchResponse{
|
||||
URL: rawURL,
|
||||
Title: extractMarkdownTitle(markdown),
|
||||
Text: text,
|
||||
Markdown: markdown,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (w *lightpandaWorker) fetchViaCommand(ctx context.Context, rawURL string) (string, error) {
|
||||
args := append([]string{"fetch"}, w.cfg.LightpandaFetchArgs...)
|
||||
args = append(args, rawURL)
|
||||
cmd := exec.CommandContext(ctx, w.cfg.LightpandaBin, args...)
|
||||
var stderr bytes.Buffer
|
||||
cmd.Stderr = &stderr
|
||||
out, err := cmd.Output()
|
||||
if err != nil {
|
||||
msg := strings.TrimSpace(stderr.String())
|
||||
if msg != "" {
|
||||
return "", fmt.Errorf("lightpanda fetch failed: %w: %s", err, truncateString(msg, 2048))
|
||||
}
|
||||
return "", fmt.Errorf("lightpanda fetch failed: %w", err)
|
||||
}
|
||||
return string(out), nil
|
||||
}
|
||||
|
||||
func (w *lightpandaWorker) fetchViaMCP(ctx context.Context, rawURL string) (string, error) {
|
||||
if w.client == nil || (w.cfg.WorkerMaxRequests > 0 && w.handled >= w.cfg.WorkerMaxRequests) {
|
||||
w.closeClient()
|
||||
w.client = newMCPClient(w.cfg)
|
||||
w.handled = 0
|
||||
}
|
||||
return w.client.Markdown(ctx, rawURL)
|
||||
}
|
||||
|
||||
func (w *lightpandaWorker) closeClient() {
|
||||
if w.client != nil {
|
||||
w.client.Close()
|
||||
w.client = nil
|
||||
}
|
||||
}
|
||||
|
||||
func newMCPClient(cfg fetchConfig) *mcpClient {
|
||||
return &mcpClient{cfg: cfg}
|
||||
}
|
||||
|
||||
func (c *mcpClient) Markdown(ctx context.Context, rawURL string) (string, error) {
|
||||
if err := c.ensureStarted(ctx); err != nil {
|
||||
return "", err
|
||||
}
|
||||
id := atomic.AddInt64(&c.nextID, 1)
|
||||
req := map[string]any{
|
||||
"jsonrpc": "2.0",
|
||||
"id": id,
|
||||
"method": "tools/call",
|
||||
"params": map[string]any{
|
||||
"name": "markdown",
|
||||
"arguments": map[string]any{
|
||||
"url": rawURL,
|
||||
},
|
||||
},
|
||||
}
|
||||
resp, err := c.call(ctx, id, req)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
var toolResult mcpToolResult
|
||||
if err := json.Unmarshal(resp.Result, &toolResult); err != nil {
|
||||
return "", fmt.Errorf("decode lightpanda markdown result: %w", err)
|
||||
}
|
||||
texts := make([]string, 0, len(toolResult.Content))
|
||||
for _, item := range toolResult.Content {
|
||||
if item.Type == "text" && strings.TrimSpace(item.Text) != "" {
|
||||
texts = append(texts, item.Text)
|
||||
}
|
||||
}
|
||||
text := strings.TrimSpace(strings.Join(texts, "\n\n"))
|
||||
if toolResult.IsError {
|
||||
if text == "" {
|
||||
text = "lightpanda tool returned error"
|
||||
}
|
||||
return "", errors.New(text)
|
||||
}
|
||||
return text, nil
|
||||
}
|
||||
|
||||
func (c *mcpClient) ensureStarted(ctx context.Context) error {
|
||||
if c.started {
|
||||
return nil
|
||||
}
|
||||
args := append([]string{}, c.cfg.LightpandaArgs...)
|
||||
cmd := exec.Command(c.cfg.LightpandaBin, args...)
|
||||
stdin, err := cmd.StdinPipe()
|
||||
if err != nil {
|
||||
return fmt.Errorf("open lightpanda stdin: %w", err)
|
||||
}
|
||||
stdout, err := cmd.StdoutPipe()
|
||||
if err != nil {
|
||||
return fmt.Errorf("open lightpanda stdout: %w", err)
|
||||
}
|
||||
cmd.Stderr = &c.stderr
|
||||
if err := cmd.Start(); err != nil {
|
||||
return fmt.Errorf("start lightpanda mcp: %w", err)
|
||||
}
|
||||
scanner := bufio.NewScanner(stdout)
|
||||
scanner.Buffer(make([]byte, 64*1024), 32*1024*1024)
|
||||
c.cmd = cmd
|
||||
c.stdin = stdin
|
||||
c.stdout = scanner
|
||||
c.started = true
|
||||
|
||||
id := atomic.AddInt64(&c.nextID, 1)
|
||||
req := map[string]any{
|
||||
"jsonrpc": "2.0",
|
||||
"id": id,
|
||||
"method": "initialize",
|
||||
"params": map[string]any{
|
||||
"protocolVersion": "2024-11-05",
|
||||
"capabilities": map[string]any{},
|
||||
"clientInfo": map[string]string{
|
||||
"name": "geo-browser-fetch",
|
||||
"version": "1.0.0",
|
||||
},
|
||||
},
|
||||
}
|
||||
if _, err := c.call(ctx, id, req); err != nil {
|
||||
c.Close()
|
||||
return fmt.Errorf("initialize lightpanda mcp: %w", err)
|
||||
}
|
||||
initialized := map[string]any{
|
||||
"jsonrpc": "2.0",
|
||||
"method": "notifications/initialized",
|
||||
"params": map[string]any{},
|
||||
}
|
||||
return c.writeJSON(initialized)
|
||||
}
|
||||
|
||||
func (c *mcpClient) call(ctx context.Context, id int64, payload map[string]any) (mcpResponse, error) {
|
||||
if err := c.writeJSON(payload); err != nil {
|
||||
return mcpResponse{}, err
|
||||
}
|
||||
type responseResult struct {
|
||||
resp mcpResponse
|
||||
err error
|
||||
}
|
||||
done := make(chan responseResult, 1)
|
||||
go func() {
|
||||
if c.stdout == nil {
|
||||
done <- responseResult{err: errors.New("lightpanda mcp stdout is not open")}
|
||||
return
|
||||
}
|
||||
for c.stdout.Scan() {
|
||||
line := strings.TrimSpace(c.stdout.Text())
|
||||
if line == "" {
|
||||
continue
|
||||
}
|
||||
var resp mcpResponse
|
||||
if err := json.Unmarshal([]byte(line), &resp); err != nil {
|
||||
continue
|
||||
}
|
||||
if resp.ID != id {
|
||||
continue
|
||||
}
|
||||
if resp.Error != nil {
|
||||
done <- responseResult{err: fmt.Errorf("lightpanda mcp error %d: %s", resp.Error.Code, resp.Error.Message)}
|
||||
return
|
||||
}
|
||||
done <- responseResult{resp: resp}
|
||||
return
|
||||
}
|
||||
err := c.stdout.Err()
|
||||
if err == nil {
|
||||
err = errors.New("lightpanda mcp stdout closed")
|
||||
}
|
||||
stderr := strings.TrimSpace(c.stderr.String())
|
||||
if stderr != "" {
|
||||
err = fmt.Errorf("%w: %s", err, truncateString(stderr, 2048))
|
||||
}
|
||||
done <- responseResult{err: err}
|
||||
}()
|
||||
|
||||
select {
|
||||
case result := <-done:
|
||||
return result.resp, result.err
|
||||
case <-ctx.Done():
|
||||
return mcpResponse{}, ctx.Err()
|
||||
}
|
||||
}
|
||||
|
||||
func (c *mcpClient) writeJSON(payload map[string]any) error {
|
||||
data, err := json.Marshal(payload)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
data = append(data, '\n')
|
||||
_, err = c.stdin.Write(data)
|
||||
return err
|
||||
}
|
||||
|
||||
func (c *mcpClient) Close() {
|
||||
if c.stdin != nil {
|
||||
_ = c.stdin.Close()
|
||||
}
|
||||
if c.cmd != nil && c.cmd.Process != nil {
|
||||
_ = c.cmd.Process.Kill()
|
||||
_, _ = c.cmd.Process.Wait()
|
||||
}
|
||||
c.started = false
|
||||
c.cmd = nil
|
||||
c.stdin = nil
|
||||
c.stdout = nil
|
||||
c.stderr.Reset()
|
||||
}
|
||||
|
||||
func (c *memoryCache) Get(key string) (fetchResponse, bool) {
|
||||
now := time.Now()
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
entry, ok := c.entries[key]
|
||||
if !ok {
|
||||
return fetchResponse{}, false
|
||||
}
|
||||
if now.After(entry.expiresAt) {
|
||||
delete(c.entries, key)
|
||||
return fetchResponse{}, false
|
||||
}
|
||||
return entry.result, true
|
||||
}
|
||||
|
||||
func (c *memoryCache) Set(key string, result fetchResponse, ttl time.Duration) {
|
||||
if ttl <= 0 {
|
||||
return
|
||||
}
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
c.entries[key] = memoryCacheEntry{result: result, expiresAt: time.Now().Add(ttl)}
|
||||
}
|
||||
|
||||
func (l *domainLimiters) TryAcquire(host string) (func(), bool) {
|
||||
l.mu.Lock()
|
||||
ch := l.items[host]
|
||||
if ch == nil {
|
||||
ch = make(chan struct{}, l.capacity)
|
||||
l.items[host] = ch
|
||||
}
|
||||
l.mu.Unlock()
|
||||
select {
|
||||
case ch <- struct{}{}:
|
||||
return func() { <-ch }, true
|
||||
default:
|
||||
return func() {}, false
|
||||
}
|
||||
}
|
||||
|
||||
func (b *domainBreakers) IsOpen(host string) bool {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
state := b.items[host]
|
||||
if state.openedAt.IsZero() {
|
||||
return false
|
||||
}
|
||||
if time.Since(state.openedAt) >= b.cooldown {
|
||||
delete(b.items, host)
|
||||
return false
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
func (b *domainBreakers) RecordSuccess(host string) {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
delete(b.items, host)
|
||||
}
|
||||
|
||||
func (b *domainBreakers) RecordFailure(host string) {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
state := b.items[host]
|
||||
if !state.openedAt.IsZero() && time.Since(state.openedAt) < b.cooldown {
|
||||
b.items[host] = state
|
||||
return
|
||||
}
|
||||
state.failures++
|
||||
if state.failures >= b.threshold {
|
||||
state.openedAt = time.Now()
|
||||
}
|
||||
b.items[host] = state
|
||||
}
|
||||
|
||||
func validateFetchURL(rawURL string, allowedDomains []string) (string, error) {
|
||||
parsed, err := url.Parse(rawURL)
|
||||
if err != nil || parsed.Scheme == "" || parsed.Host == "" {
|
||||
return "", errors.New("invalid url")
|
||||
}
|
||||
if parsed.Scheme != "http" && parsed.Scheme != "https" {
|
||||
return "", errors.New("only http and https urls are allowed")
|
||||
}
|
||||
host := strings.ToLower(strings.TrimSpace(parsed.Hostname()))
|
||||
if host == "" {
|
||||
return "", errors.New("url host is required")
|
||||
}
|
||||
if ip := net.ParseIP(host); ip != nil {
|
||||
return "", errors.New("ip literal urls are not allowed")
|
||||
}
|
||||
if !domainAllowed(host, allowedDomains) {
|
||||
return "", errors.New("domain is not allowed")
|
||||
}
|
||||
return host, nil
|
||||
}
|
||||
|
||||
func domainAllowed(host string, allowedDomains []string) bool {
|
||||
if len(allowedDomains) == 0 {
|
||||
return false
|
||||
}
|
||||
for _, domain := range allowedDomains {
|
||||
domain = strings.ToLower(strings.TrimPrefix(strings.TrimSpace(domain), "."))
|
||||
if domain == "" {
|
||||
continue
|
||||
}
|
||||
if host == domain || strings.HasSuffix(host, "."+domain) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func fetchCacheKey(rawURL string) string {
|
||||
normalized := strings.TrimSpace(rawURL)
|
||||
if parsed, err := url.Parse(normalized); err == nil && parsed.Host != "" {
|
||||
parsed.Fragment = ""
|
||||
normalized = parsed.String()
|
||||
}
|
||||
sum := sha256.Sum256([]byte(normalized))
|
||||
return hex.EncodeToString(sum[:])
|
||||
}
|
||||
|
||||
func extractMarkdownTitle(markdown string) string {
|
||||
for _, line := range strings.Split(markdown, "\n") {
|
||||
line = strings.TrimSpace(line)
|
||||
if strings.HasPrefix(line, "# ") {
|
||||
return strings.TrimSpace(strings.TrimPrefix(line, "# "))
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func normalizeText(value string) string {
|
||||
lines := strings.Split(value, "\n")
|
||||
out := make([]string, 0, len(lines))
|
||||
for _, line := range lines {
|
||||
line = strings.TrimSpace(line)
|
||||
if line != "" {
|
||||
out = append(out, line)
|
||||
}
|
||||
}
|
||||
return strings.Join(out, "\n")
|
||||
}
|
||||
|
||||
func writeJSON(w http.ResponseWriter, status int, payload any) {
|
||||
w.Header().Set("Content-Type", "application/json; charset=utf-8")
|
||||
w.WriteHeader(status)
|
||||
_ = json.NewEncoder(w).Encode(payload)
|
||||
}
|
||||
|
||||
func splitCSV(value string) []string {
|
||||
parts := strings.Split(value, ",")
|
||||
out := make([]string, 0, len(parts))
|
||||
for _, part := range parts {
|
||||
part = strings.TrimSpace(part)
|
||||
if part != "" {
|
||||
out = append(out, part)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func splitCommandArgs(value string) []string {
|
||||
parts := strings.Fields(value)
|
||||
if len(parts) == 0 {
|
||||
return nil
|
||||
}
|
||||
return parts
|
||||
}
|
||||
|
||||
func envString(key, fallback string) string {
|
||||
if value := strings.TrimSpace(os.Getenv(key)); value != "" {
|
||||
return value
|
||||
}
|
||||
return fallback
|
||||
}
|
||||
|
||||
func envInt(key string, fallback int) int {
|
||||
value := strings.TrimSpace(os.Getenv(key))
|
||||
if value == "" {
|
||||
return fallback
|
||||
}
|
||||
parsed, err := strconv.Atoi(value)
|
||||
if err != nil {
|
||||
return fallback
|
||||
}
|
||||
return parsed
|
||||
}
|
||||
|
||||
func envDuration(key string, fallback time.Duration) time.Duration {
|
||||
value := strings.TrimSpace(os.Getenv(key))
|
||||
if value == "" {
|
||||
return fallback
|
||||
}
|
||||
parsed, err := time.ParseDuration(value)
|
||||
if err == nil {
|
||||
return parsed
|
||||
}
|
||||
ms, err := strconv.Atoi(value)
|
||||
if err != nil {
|
||||
return fallback
|
||||
}
|
||||
return time.Duration(ms) * time.Millisecond
|
||||
}
|
||||
|
||||
func truncateString(value string, max int) string {
|
||||
if len(value) <= max {
|
||||
return value
|
||||
}
|
||||
return value[:max]
|
||||
}
|
||||
@@ -204,3 +204,23 @@ generation:
|
||||
worker_concurrency: 8
|
||||
stream_enabled: false
|
||||
article_timeout: 8m
|
||||
|
||||
browser_fetch:
|
||||
enabled: false
|
||||
provider: lightpanda
|
||||
service_url: http://localhost:8082
|
||||
token: dev-browser-fetch-token
|
||||
queue_size: 64
|
||||
worker_concurrency: 2
|
||||
request_timeout: 45s
|
||||
cache_ttl: 6h
|
||||
allowed_domains:
|
||||
- zhuanlan.zhihu.com
|
||||
- zhihu.com
|
||||
trigger_status:
|
||||
- 403
|
||||
- 429
|
||||
- 500
|
||||
- 502
|
||||
- 503
|
||||
- 504
|
||||
|
||||
@@ -216,4 +216,24 @@ generation:
|
||||
task_recovery_batch_size: 100
|
||||
task_queued_stale_after: 2m
|
||||
task_max_attempts: 3
|
||||
|
||||
browser_fetch:
|
||||
enabled: false
|
||||
provider: lightpanda
|
||||
service_url: http://localhost:8082
|
||||
token: dev-browser-fetch-token
|
||||
queue_size: 64
|
||||
worker_concurrency: 2
|
||||
request_timeout: 45s
|
||||
cache_ttl: 6h
|
||||
allowed_domains:
|
||||
- zhuanlan.zhihu.com
|
||||
- zhihu.com
|
||||
trigger_status:
|
||||
- 403
|
||||
- 429
|
||||
- 500
|
||||
- 502
|
||||
- 503
|
||||
- 504
|
||||
|
||||
|
||||
@@ -38,6 +38,7 @@ type Config struct {
|
||||
Compliance ComplianceConfig `mapstructure:"compliance"`
|
||||
Retrieval RetrievalConfig `mapstructure:"retrieval"`
|
||||
Generation GenerationConfig `mapstructure:"generation"`
|
||||
BrowserFetch BrowserFetchConfig `mapstructure:"browser_fetch"`
|
||||
}
|
||||
|
||||
type ServerConfig struct {
|
||||
@@ -348,6 +349,19 @@ type GenerationConfig struct {
|
||||
TaskStateCheckLookback time.Duration `mapstructure:"task_state_check_lookback"`
|
||||
}
|
||||
|
||||
type BrowserFetchConfig struct {
|
||||
Enabled bool `mapstructure:"enabled"`
|
||||
Provider string `mapstructure:"provider"`
|
||||
ServiceURL string `mapstructure:"service_url"`
|
||||
Token string `mapstructure:"token"`
|
||||
QueueSize int `mapstructure:"queue_size"`
|
||||
WorkerConcurrency int `mapstructure:"worker_concurrency"`
|
||||
RequestTimeout time.Duration `mapstructure:"request_timeout"`
|
||||
CacheTTL time.Duration `mapstructure:"cache_ttl"`
|
||||
AllowedDomains []string `mapstructure:"allowed_domains"`
|
||||
TriggerStatus []int `mapstructure:"trigger_status"`
|
||||
}
|
||||
|
||||
func Load(configPath string) (*Config, error) {
|
||||
cfg, _, err := loadWithFiles(configPath)
|
||||
return cfg, err
|
||||
@@ -382,6 +396,7 @@ func loadWithFiles(configPath string) (*Config, []string, error) {
|
||||
normalizeBrandLibraryConfig(&cfg.BrandLibrary)
|
||||
NormalizeComplianceConfig(&cfg.Compliance)
|
||||
NormalizeGenerationConfig(&cfg.Generation)
|
||||
NormalizeBrowserFetchConfig(&cfg.BrowserFetch)
|
||||
|
||||
files := []string{configFile}
|
||||
if localConfigFile != "" {
|
||||
@@ -511,6 +526,42 @@ func compactStringSlice(values []string) []string {
|
||||
return out
|
||||
}
|
||||
|
||||
func compactIntSlice(values []int) []int {
|
||||
if values == nil {
|
||||
return nil
|
||||
}
|
||||
out := make([]int, 0, len(values))
|
||||
seen := make(map[int]struct{}, len(values))
|
||||
for _, value := range values {
|
||||
if value <= 0 {
|
||||
continue
|
||||
}
|
||||
if _, exists := seen[value]; exists {
|
||||
continue
|
||||
}
|
||||
seen[value] = struct{}{}
|
||||
out = append(out, value)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func parseIntList(value string) []int {
|
||||
parts := strings.Split(value, ",")
|
||||
out := make([]int, 0, len(parts))
|
||||
for _, part := range parts {
|
||||
part = strings.TrimSpace(part)
|
||||
if part == "" {
|
||||
continue
|
||||
}
|
||||
n, err := strconv.Atoi(part)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
out = append(out, n)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func normalizeMonitoringDispatchConfig(cfg *MonitoringDispatchConfig) {
|
||||
if cfg == nil {
|
||||
return
|
||||
@@ -643,6 +694,35 @@ func NormalizeGenerationConfig(cfg *GenerationConfig) {
|
||||
}
|
||||
}
|
||||
|
||||
func NormalizeBrowserFetchConfig(cfg *BrowserFetchConfig) {
|
||||
if cfg == nil {
|
||||
return
|
||||
}
|
||||
cfg.Provider = strings.ToLower(strings.TrimSpace(cfg.Provider))
|
||||
if cfg.Provider == "" {
|
||||
cfg.Provider = "lightpanda"
|
||||
}
|
||||
cfg.ServiceURL = strings.TrimRight(strings.TrimSpace(cfg.ServiceURL), "/")
|
||||
cfg.Token = strings.TrimSpace(cfg.Token)
|
||||
if cfg.QueueSize <= 0 {
|
||||
cfg.QueueSize = 64
|
||||
}
|
||||
if cfg.WorkerConcurrency <= 0 {
|
||||
cfg.WorkerConcurrency = 1
|
||||
}
|
||||
if cfg.RequestTimeout <= 0 {
|
||||
cfg.RequestTimeout = 45 * time.Second
|
||||
}
|
||||
if cfg.CacheTTL <= 0 {
|
||||
cfg.CacheTTL = 6 * time.Hour
|
||||
}
|
||||
cfg.AllowedDomains = compactStringSlice(cfg.AllowedDomains)
|
||||
cfg.TriggerStatus = compactIntSlice(cfg.TriggerStatus)
|
||||
if len(cfg.TriggerStatus) == 0 {
|
||||
cfg.TriggerStatus = []int{403, 429, 500, 502, 503, 504}
|
||||
}
|
||||
}
|
||||
|
||||
func NormalizeComplianceConfig(cfg *ComplianceConfig) {
|
||||
if cfg == nil {
|
||||
return
|
||||
@@ -759,6 +839,36 @@ func applyEnvOverrides(cfg *Config) {
|
||||
if d, ok := lookupDurationEnv("GENERATION_TASK_STATE_CHECK_LOOKBACK"); ok {
|
||||
cfg.Generation.TaskStateCheckLookback = d
|
||||
}
|
||||
if enabled, ok := lookupBoolEnv("BROWSER_FETCH_ENABLED"); ok {
|
||||
cfg.BrowserFetch.Enabled = enabled
|
||||
}
|
||||
if provider, ok := lookupNonEmptyEnv("BROWSER_FETCH_PROVIDER"); ok {
|
||||
cfg.BrowserFetch.Provider = provider
|
||||
}
|
||||
if serviceURL, ok := lookupFirstNonEmptyEnv("BROWSER_FETCH_SERVICE_URL", "LIGHTPANDA_BROWSER_FETCH_URL"); ok {
|
||||
cfg.BrowserFetch.ServiceURL = serviceURL
|
||||
}
|
||||
if token, ok := lookupNonEmptyEnv("BROWSER_FETCH_TOKEN"); ok {
|
||||
cfg.BrowserFetch.Token = token
|
||||
}
|
||||
if n, ok := lookupIntEnv("BROWSER_FETCH_QUEUE_SIZE"); ok {
|
||||
cfg.BrowserFetch.QueueSize = n
|
||||
}
|
||||
if n, ok := lookupIntEnv("BROWSER_FETCH_WORKER_CONCURRENCY"); ok {
|
||||
cfg.BrowserFetch.WorkerConcurrency = n
|
||||
}
|
||||
if d, ok := lookupDurationEnv("BROWSER_FETCH_REQUEST_TIMEOUT"); ok {
|
||||
cfg.BrowserFetch.RequestTimeout = d
|
||||
}
|
||||
if d, ok := lookupDurationEnv("BROWSER_FETCH_CACHE_TTL"); ok {
|
||||
cfg.BrowserFetch.CacheTTL = d
|
||||
}
|
||||
if domains, ok := lookupNonEmptyEnv("BROWSER_FETCH_ALLOWED_DOMAINS"); ok {
|
||||
cfg.BrowserFetch.AllowedDomains = strings.Split(domains, ",")
|
||||
}
|
||||
if statuses, ok := lookupNonEmptyEnv("BROWSER_FETCH_TRIGGER_STATUS"); ok {
|
||||
cfg.BrowserFetch.TriggerStatus = parseIntList(statuses)
|
||||
}
|
||||
if apiKey, ok := lookupNonEmptyEnv("QDRANT_API_KEY"); ok {
|
||||
cfg.Qdrant.APIKey = apiKey
|
||||
}
|
||||
|
||||
@@ -557,6 +557,93 @@ generation: {}
|
||||
}
|
||||
}
|
||||
|
||||
func TestLoadAppliesBrowserFetchDefaultsAndOverrides(t *testing.T) {
|
||||
configPath := writeTestConfig(t, `
|
||||
browser_fetch:
|
||||
enabled: true
|
||||
service_url: " http://browser-fetch:8082/ "
|
||||
token: " token "
|
||||
allowed_domains:
|
||||
- " zhuanlan.zhihu.com "
|
||||
- ""
|
||||
trigger_status:
|
||||
- 403
|
||||
- 403
|
||||
- 500
|
||||
`)
|
||||
|
||||
cfg, err := Load(configPath)
|
||||
if err != nil {
|
||||
t.Fatalf("Load() error = %v", err)
|
||||
}
|
||||
|
||||
if !cfg.BrowserFetch.Enabled {
|
||||
t.Fatalf("expected browser fetch enabled")
|
||||
}
|
||||
if cfg.BrowserFetch.Provider != "lightpanda" {
|
||||
t.Fatalf("expected default provider lightpanda, got %q", cfg.BrowserFetch.Provider)
|
||||
}
|
||||
if cfg.BrowserFetch.ServiceURL != "http://browser-fetch:8082" {
|
||||
t.Fatalf("expected normalized service url, got %q", cfg.BrowserFetch.ServiceURL)
|
||||
}
|
||||
if cfg.BrowserFetch.Token != "token" {
|
||||
t.Fatalf("expected normalized token, got %q", cfg.BrowserFetch.Token)
|
||||
}
|
||||
if cfg.BrowserFetch.QueueSize != 64 ||
|
||||
cfg.BrowserFetch.WorkerConcurrency != 1 ||
|
||||
cfg.BrowserFetch.RequestTimeout != 45*time.Second ||
|
||||
cfg.BrowserFetch.CacheTTL != 6*time.Hour {
|
||||
t.Fatalf("unexpected browser fetch defaults: %#v", cfg.BrowserFetch)
|
||||
}
|
||||
if len(cfg.BrowserFetch.AllowedDomains) != 1 || cfg.BrowserFetch.AllowedDomains[0] != "zhuanlan.zhihu.com" {
|
||||
t.Fatalf("unexpected allowed domains: %#v", cfg.BrowserFetch.AllowedDomains)
|
||||
}
|
||||
if len(cfg.BrowserFetch.TriggerStatus) != 2 ||
|
||||
cfg.BrowserFetch.TriggerStatus[0] != 403 ||
|
||||
cfg.BrowserFetch.TriggerStatus[1] != 500 {
|
||||
t.Fatalf("unexpected trigger status: %#v", cfg.BrowserFetch.TriggerStatus)
|
||||
}
|
||||
}
|
||||
|
||||
func TestLoadAppliesBrowserFetchEnvOverrides(t *testing.T) {
|
||||
t.Setenv("BROWSER_FETCH_ENABLED", "true")
|
||||
t.Setenv("BROWSER_FETCH_PROVIDER", "lightpanda_service")
|
||||
t.Setenv("BROWSER_FETCH_SERVICE_URL", "http://env-browser-fetch:8082/")
|
||||
t.Setenv("BROWSER_FETCH_TOKEN", "env-token")
|
||||
t.Setenv("BROWSER_FETCH_QUEUE_SIZE", "11")
|
||||
t.Setenv("BROWSER_FETCH_WORKER_CONCURRENCY", "3")
|
||||
t.Setenv("BROWSER_FETCH_REQUEST_TIMEOUT", "12s")
|
||||
t.Setenv("BROWSER_FETCH_CACHE_TTL", "30m")
|
||||
t.Setenv("BROWSER_FETCH_ALLOWED_DOMAINS", "zhihu.com,example.com")
|
||||
t.Setenv("BROWSER_FETCH_TRIGGER_STATUS", "403,429,broken,503")
|
||||
|
||||
configPath := writeTestConfig(t, `
|
||||
browser_fetch: {}
|
||||
`)
|
||||
|
||||
cfg, err := Load(configPath)
|
||||
if err != nil {
|
||||
t.Fatalf("Load() error = %v", err)
|
||||
}
|
||||
|
||||
if !cfg.BrowserFetch.Enabled ||
|
||||
cfg.BrowserFetch.Provider != "lightpanda_service" ||
|
||||
cfg.BrowserFetch.ServiceURL != "http://env-browser-fetch:8082" ||
|
||||
cfg.BrowserFetch.Token != "env-token" ||
|
||||
cfg.BrowserFetch.QueueSize != 11 ||
|
||||
cfg.BrowserFetch.WorkerConcurrency != 3 ||
|
||||
cfg.BrowserFetch.RequestTimeout != 12*time.Second ||
|
||||
cfg.BrowserFetch.CacheTTL != 30*time.Minute {
|
||||
t.Fatalf("unexpected browser fetch env config: %#v", cfg.BrowserFetch)
|
||||
}
|
||||
if got := cfg.BrowserFetch.AllowedDomains; len(got) != 2 || got[0] != "zhihu.com" || got[1] != "example.com" {
|
||||
t.Fatalf("unexpected env allowed domains: %#v", got)
|
||||
}
|
||||
if got := cfg.BrowserFetch.TriggerStatus; len(got) != 3 || got[0] != 403 || got[1] != 429 || got[2] != 503 {
|
||||
t.Fatalf("unexpected env trigger status: %#v", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestStoreReloadsConfigAndKeepsPreviousOnInvalidYAML(t *testing.T) {
|
||||
configPath := writeTestConfig(t, `
|
||||
jwt:
|
||||
|
||||
@@ -80,6 +80,12 @@ func Diff(previous, current *Config) []FieldChange {
|
||||
if previous.Retrieval != current.Retrieval {
|
||||
addChange("retrieval", true)
|
||||
}
|
||||
if previous.BrowserFetch.QueueSize != current.BrowserFetch.QueueSize ||
|
||||
previous.BrowserFetch.WorkerConcurrency != current.BrowserFetch.WorkerConcurrency {
|
||||
addChange("browser_fetch.worker", false)
|
||||
} else if !reflect.DeepEqual(previous.BrowserFetch, current.BrowserFetch) {
|
||||
addChange("browser_fetch", true)
|
||||
}
|
||||
if previous.Generation.QueueSize != current.Generation.QueueSize ||
|
||||
previous.Generation.WorkerConcurrency != current.Generation.WorkerConcurrency {
|
||||
addChange("generation.worker", false)
|
||||
|
||||
@@ -0,0 +1,236 @@
|
||||
package app
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/geo-platform/tenant-api/internal/shared/config"
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
var errBrowserFetchQueueFull = errors.New("browser fetch queue is full")
|
||||
|
||||
type browserFetchResult struct {
|
||||
URL string `json:"url,omitempty"`
|
||||
Title string `json:"title,omitempty"`
|
||||
Text string `json:"text,omitempty"`
|
||||
Markdown string `json:"markdown,omitempty"`
|
||||
}
|
||||
|
||||
type browserFetchRequest struct {
|
||||
URL string `json:"url"`
|
||||
TimeoutMS int64 `json:"timeout_ms,omitempty"`
|
||||
Options map[string]interface{} `json:"options,omitempty"`
|
||||
ctx context.Context `json:"-"`
|
||||
cfg config.BrowserFetchConfig `json:"-"`
|
||||
resp chan browserFetchRequestResult `json:"-"`
|
||||
}
|
||||
|
||||
type browserFetchRequestResult struct {
|
||||
result browserFetchResult
|
||||
err error
|
||||
}
|
||||
|
||||
type browserFetchQueue struct {
|
||||
logger *zap.Logger
|
||||
client *http.Client
|
||||
jobs chan browserFetchRequest
|
||||
cache *browserFetchMemoryCache
|
||||
}
|
||||
|
||||
type browserFetchCacheEntry struct {
|
||||
result browserFetchResult
|
||||
expiresAt time.Time
|
||||
}
|
||||
|
||||
type browserFetchMemoryCache struct {
|
||||
mu sync.Mutex
|
||||
entries map[string]browserFetchCacheEntry
|
||||
}
|
||||
|
||||
func (s *KnowledgeService) browserFetchQueue(cfg config.BrowserFetchConfig) *browserFetchQueue {
|
||||
s.browserFetchMu.Lock()
|
||||
defer s.browserFetchMu.Unlock()
|
||||
if s.browserFetchQ != nil {
|
||||
return s.browserFetchQ
|
||||
}
|
||||
|
||||
config.NormalizeBrowserFetchConfig(&cfg)
|
||||
queue := &browserFetchQueue{
|
||||
logger: s.logger,
|
||||
client: &http.Client{},
|
||||
jobs: make(chan browserFetchRequest, cfg.QueueSize),
|
||||
cache: &browserFetchMemoryCache{
|
||||
entries: make(map[string]browserFetchCacheEntry),
|
||||
},
|
||||
}
|
||||
|
||||
for i := 0; i < cfg.WorkerConcurrency; i++ {
|
||||
go queue.runWorker(i)
|
||||
}
|
||||
s.browserFetchQ = queue
|
||||
return queue
|
||||
}
|
||||
|
||||
func (q *browserFetchQueue) Fetch(ctx context.Context, rawURL string, cfg config.BrowserFetchConfig) (browserFetchResult, error) {
|
||||
if q == nil {
|
||||
return browserFetchResult{}, errBrowserFetchNotAvailable
|
||||
}
|
||||
config.NormalizeBrowserFetchConfig(&cfg)
|
||||
cacheKey := browserFetchCacheKey(rawURL)
|
||||
if cfg.CacheTTL > 0 {
|
||||
if result, ok := q.cache.Get(cacheKey); ok {
|
||||
return result, nil
|
||||
}
|
||||
}
|
||||
|
||||
request := browserFetchRequest{
|
||||
URL: rawURL,
|
||||
TimeoutMS: int64(cfg.RequestTimeout / time.Millisecond),
|
||||
Options: map[string]interface{}{
|
||||
"provider": cfg.Provider,
|
||||
},
|
||||
ctx: ctx,
|
||||
cfg: cfg,
|
||||
resp: make(chan browserFetchRequestResult, 1),
|
||||
}
|
||||
|
||||
select {
|
||||
case q.jobs <- request:
|
||||
case <-ctx.Done():
|
||||
return browserFetchResult{}, ctx.Err()
|
||||
default:
|
||||
return browserFetchResult{}, errBrowserFetchQueueFull
|
||||
}
|
||||
|
||||
select {
|
||||
case result := <-request.resp:
|
||||
if result.err != nil {
|
||||
return browserFetchResult{}, result.err
|
||||
}
|
||||
if cfg.CacheTTL > 0 {
|
||||
q.cache.Set(cacheKey, result.result, cfg.CacheTTL)
|
||||
}
|
||||
return result.result, nil
|
||||
case <-ctx.Done():
|
||||
return browserFetchResult{}, ctx.Err()
|
||||
}
|
||||
}
|
||||
|
||||
func (q *browserFetchQueue) runWorker(workerID int) {
|
||||
for request := range q.jobs {
|
||||
result, err := q.execute(context.Background(), request)
|
||||
if err != nil && q.logger != nil {
|
||||
q.logger.Warn("browser fetch request failed",
|
||||
zap.Int("worker_id", workerID),
|
||||
zap.String("url", request.URL),
|
||||
zap.Error(err),
|
||||
)
|
||||
}
|
||||
request.resp <- browserFetchRequestResult{result: result, err: err}
|
||||
}
|
||||
}
|
||||
|
||||
func (q *browserFetchQueue) execute(parent context.Context, request browserFetchRequest) (browserFetchResult, error) {
|
||||
if request.ctx != nil {
|
||||
parent = request.ctx
|
||||
}
|
||||
timeout := time.Duration(request.TimeoutMS) * time.Millisecond
|
||||
if timeout <= 0 {
|
||||
timeout = 45 * time.Second
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(parent, timeout)
|
||||
defer cancel()
|
||||
|
||||
return q.callService(ctx, request.URL, request.cfg)
|
||||
}
|
||||
|
||||
func (q *browserFetchQueue) callService(ctx context.Context, rawURL string, cfg config.BrowserFetchConfig) (browserFetchResult, error) {
|
||||
config.NormalizeBrowserFetchConfig(&cfg)
|
||||
endpoint := strings.TrimRight(cfg.ServiceURL, "/") + "/fetch"
|
||||
payload := browserFetchRequest{
|
||||
URL: rawURL,
|
||||
TimeoutMS: int64(cfg.RequestTimeout / time.Millisecond),
|
||||
}
|
||||
body, err := json.Marshal(payload)
|
||||
if err != nil {
|
||||
return browserFetchResult{}, fmt.Errorf("marshal browser fetch request: %w", err)
|
||||
}
|
||||
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(body))
|
||||
if err != nil {
|
||||
return browserFetchResult{}, fmt.Errorf("create browser fetch request: %w", err)
|
||||
}
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
if cfg.Token != "" {
|
||||
req.Header.Set("Authorization", "Bearer "+cfg.Token)
|
||||
}
|
||||
|
||||
resp, err := q.client.Do(req)
|
||||
if err != nil {
|
||||
return browserFetchResult{}, fmt.Errorf("execute browser fetch request: %w", err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
if resp.StatusCode >= http.StatusBadRequest {
|
||||
data, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
|
||||
return browserFetchResult{}, fmt.Errorf("browser fetch service failed: status=%d body=%s", resp.StatusCode, strings.TrimSpace(string(data)))
|
||||
}
|
||||
|
||||
var result browserFetchResult
|
||||
if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
|
||||
return browserFetchResult{}, fmt.Errorf("decode browser fetch response: %w", err)
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func (c *browserFetchMemoryCache) Get(key string) (browserFetchResult, bool) {
|
||||
if c == nil || key == "" {
|
||||
return browserFetchResult{}, false
|
||||
}
|
||||
now := time.Now()
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
entry, ok := c.entries[key]
|
||||
if !ok {
|
||||
return browserFetchResult{}, false
|
||||
}
|
||||
if now.After(entry.expiresAt) {
|
||||
delete(c.entries, key)
|
||||
return browserFetchResult{}, false
|
||||
}
|
||||
return entry.result, true
|
||||
}
|
||||
|
||||
func (c *browserFetchMemoryCache) Set(key string, result browserFetchResult, ttl time.Duration) {
|
||||
if c == nil || key == "" || ttl <= 0 {
|
||||
return
|
||||
}
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
c.entries[key] = browserFetchCacheEntry{
|
||||
result: result,
|
||||
expiresAt: time.Now().Add(ttl),
|
||||
}
|
||||
}
|
||||
|
||||
func browserFetchCacheKey(rawURL string) string {
|
||||
normalized := strings.TrimSpace(rawURL)
|
||||
if parsed, err := url.Parse(normalized); err == nil && parsed.Host != "" {
|
||||
parsed.Fragment = ""
|
||||
normalized = parsed.String()
|
||||
}
|
||||
sum := sha256.Sum256([]byte(normalized))
|
||||
return hex.EncodeToString(sum[:])
|
||||
}
|
||||
@@ -49,6 +49,8 @@ type KnowledgeService struct {
|
||||
jobs chan knowledgeParseJob
|
||||
cleanupMu sync.Mutex
|
||||
cleanupInFlight map[string]struct{}
|
||||
browserFetchMu sync.Mutex
|
||||
browserFetchQ *browserFetchQueue
|
||||
}
|
||||
|
||||
type knowledgeParseJob struct {
|
||||
@@ -199,7 +201,7 @@ func (s *KnowledgeService) WithConfigProvider(provider runtimeConfigProvider) *K
|
||||
func (s *KnowledgeService) runtimeConfig() knowledgeRuntimeConfig {
|
||||
if s != nil && s.configProvider != nil {
|
||||
if cfg := s.configProvider.Current(); cfg != nil {
|
||||
return knowledgeRuntimeFromConfig(cfg.Retrieval, cfg.LLM)
|
||||
return knowledgeRuntimeFromFullConfig(cfg)
|
||||
}
|
||||
}
|
||||
if s == nil {
|
||||
|
||||
@@ -4,12 +4,16 @@ import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/geo-platform/tenant-api/internal/shared/config"
|
||||
"github.com/geo-platform/tenant-api/internal/shared/llm"
|
||||
"github.com/geo-platform/tenant-api/internal/tenant/prompts"
|
||||
"go.uber.org/zap"
|
||||
@@ -23,6 +27,37 @@ const (
|
||||
defaultKnowledgeURLMarkdownOutputToken = 12000
|
||||
)
|
||||
|
||||
var errBrowserFetchNotAvailable = fmt.Errorf("browser fetch fallback is not available")
|
||||
|
||||
type sourceFetchRestrictedError struct {
|
||||
URL string
|
||||
Status int
|
||||
Cause error
|
||||
}
|
||||
|
||||
func (e sourceFetchRestrictedError) Error() string {
|
||||
cause := ""
|
||||
if e.Cause != nil {
|
||||
cause = strings.TrimSpace(e.Cause.Error())
|
||||
}
|
||||
if cause == "" {
|
||||
cause = "source website requires browser rendering or blocks server-side fetch"
|
||||
}
|
||||
if e.Status > 0 {
|
||||
return fmt.Sprintf("source website fetch restricted status=%d: %s", e.Status, cause)
|
||||
}
|
||||
return "source website fetch restricted: " + cause
|
||||
}
|
||||
|
||||
func (e sourceFetchRestrictedError) Unwrap() error {
|
||||
return e.Cause
|
||||
}
|
||||
|
||||
func isSourceFetchRestrictedError(err error) bool {
|
||||
var restricted sourceFetchRestrictedError
|
||||
return errors.As(err, &restricted)
|
||||
}
|
||||
|
||||
type arkToolExecuteRequest struct {
|
||||
ActionName string `json:"action_name"`
|
||||
ToolName string `json:"tool_name"`
|
||||
@@ -50,7 +85,20 @@ type arkWebDataItem struct {
|
||||
func (s *KnowledgeService) parseWebsiteKnowledge(ctx context.Context, rawURL string) (*knowledgeParsedContent, error) {
|
||||
item, err := s.executeLinkReader(ctx, rawURL)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
if !s.shouldTryBrowserFetchFallback(rawURL, err) {
|
||||
return nil, classifySourceFetchError(rawURL, err)
|
||||
}
|
||||
fallback, fallbackErr := s.parseWebsiteKnowledgeWithBrowserFetch(ctx, rawURL)
|
||||
if fallbackErr == nil {
|
||||
return fallback, nil
|
||||
}
|
||||
if s.logger != nil {
|
||||
s.logger.Warn("knowledge browser fetch fallback failed",
|
||||
zap.String("url", rawURL),
|
||||
zap.Error(fallbackErr),
|
||||
)
|
||||
}
|
||||
return nil, classifySourceFetchError(rawURL, err)
|
||||
}
|
||||
|
||||
text := normalizeKnowledgeText(item.Content)
|
||||
@@ -76,6 +124,65 @@ func (s *KnowledgeService) parseWebsiteKnowledge(ctx context.Context, rawURL str
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (s *KnowledgeService) parseWebsiteKnowledgeWithBrowserFetch(ctx context.Context, rawURL string) (*knowledgeParsedContent, error) {
|
||||
runtimeCfg := s.runtimeConfig()
|
||||
browserCfg := runtimeCfg.BrowserFetch
|
||||
config.NormalizeBrowserFetchConfig(&browserCfg)
|
||||
if !browserCfg.Enabled {
|
||||
return nil, errBrowserFetchNotAvailable
|
||||
}
|
||||
|
||||
provider := strings.ToLower(strings.TrimSpace(browserCfg.Provider))
|
||||
if provider != "lightpanda" && provider != "lightpanda_service" {
|
||||
return nil, fmt.Errorf("unsupported browser fetch provider %q", browserCfg.Provider)
|
||||
}
|
||||
|
||||
if strings.TrimSpace(browserCfg.ServiceURL) == "" {
|
||||
return nil, fmt.Errorf("%w: browser_fetch.service_url is empty", errBrowserFetchNotAvailable)
|
||||
}
|
||||
|
||||
callCtx := ctx
|
||||
if browserCfg.RequestTimeout > 0 {
|
||||
var cancel context.CancelFunc
|
||||
callCtx, cancel = context.WithTimeout(ctx, browserCfg.RequestTimeout)
|
||||
defer cancel()
|
||||
}
|
||||
|
||||
result, err := s.browserFetchQueue(browserCfg).Fetch(callCtx, rawURL, browserCfg)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
text := normalizeKnowledgeText(result.Text)
|
||||
if text == "" {
|
||||
text = normalizeKnowledgeText(result.Markdown)
|
||||
}
|
||||
if text == "" {
|
||||
return nil, fmt.Errorf("browser fetch returned empty content")
|
||||
}
|
||||
|
||||
title := strings.TrimSpace(result.Title)
|
||||
markdown := strings.TrimSpace(result.Markdown)
|
||||
if markdown == "" {
|
||||
var err error
|
||||
markdown, err = s.formatWebsiteMarkdown(ctx, title, text, rawURL)
|
||||
if err != nil {
|
||||
if s.logger != nil {
|
||||
s.logger.Warn("knowledge browser fetch markdown formatting failed, fallback to local markdown",
|
||||
zap.String("url", rawURL),
|
||||
zap.Error(err),
|
||||
)
|
||||
}
|
||||
markdown = buildKnowledgeMarkdownFromText(title, text)
|
||||
}
|
||||
}
|
||||
|
||||
return &knowledgeParsedContent{
|
||||
Text: text,
|
||||
Markdown: markdown,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (s *KnowledgeService) executeLinkReader(ctx context.Context, rawURL string) (*arkWebDataItem, error) {
|
||||
runtimeCfg := s.runtimeConfig()
|
||||
baseURL := strings.TrimRight(strings.TrimSpace(runtimeCfg.ArkBaseURL), "/")
|
||||
@@ -135,6 +242,90 @@ func (s *KnowledgeService) executeLinkReader(ctx context.Context, rawURL string)
|
||||
return &item, nil
|
||||
}
|
||||
|
||||
func classifySourceFetchError(rawURL string, err error) error {
|
||||
if err == nil {
|
||||
return nil
|
||||
}
|
||||
status := extractHTTPStatusFromError(err)
|
||||
if status == http.StatusForbidden || status == http.StatusTooManyRequests || status >= http.StatusInternalServerError {
|
||||
return sourceFetchRestrictedError{URL: rawURL, Status: status, Cause: err}
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
func extractHTTPStatusFromError(err error) int {
|
||||
if err == nil {
|
||||
return 0
|
||||
}
|
||||
message := err.Error()
|
||||
idx := strings.Index(message, "status=")
|
||||
if idx < 0 {
|
||||
return 0
|
||||
}
|
||||
value := message[idx+len("status="):]
|
||||
end := 0
|
||||
for end < len(value) && value[end] >= '0' && value[end] <= '9' {
|
||||
end++
|
||||
}
|
||||
if end == 0 {
|
||||
return 0
|
||||
}
|
||||
status, parseErr := strconv.Atoi(value[:end])
|
||||
if parseErr != nil {
|
||||
return 0
|
||||
}
|
||||
return status
|
||||
}
|
||||
|
||||
func (s *KnowledgeService) shouldTryBrowserFetchFallback(rawURL string, err error) bool {
|
||||
if s == nil {
|
||||
return false
|
||||
}
|
||||
runtimeCfg := s.runtimeConfig()
|
||||
browserCfg := runtimeCfg.BrowserFetch
|
||||
config.NormalizeBrowserFetchConfig(&browserCfg)
|
||||
if !browserCfg.Enabled {
|
||||
return false
|
||||
}
|
||||
if !browserFetchDomainAllowed(rawURL, browserCfg.AllowedDomains) {
|
||||
return false
|
||||
}
|
||||
status := extractHTTPStatusFromError(err)
|
||||
if status <= 0 {
|
||||
return false
|
||||
}
|
||||
for _, trigger := range browserCfg.TriggerStatus {
|
||||
if status == trigger {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func browserFetchDomainAllowed(rawURL string, allowedDomains []string) bool {
|
||||
if len(allowedDomains) == 0 {
|
||||
return false
|
||||
}
|
||||
parsed, err := url.Parse(rawURL)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
host := strings.ToLower(strings.TrimSpace(parsed.Hostname()))
|
||||
if host == "" {
|
||||
return false
|
||||
}
|
||||
for _, domain := range allowedDomains {
|
||||
domain = strings.ToLower(strings.TrimPrefix(strings.TrimSpace(domain), "."))
|
||||
if domain == "" {
|
||||
continue
|
||||
}
|
||||
if host == domain || strings.HasSuffix(host, "."+domain) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func (s *KnowledgeService) formatWebsiteMarkdown(ctx context.Context, title, content, rawURL string) (string, error) {
|
||||
content = strings.TrimSpace(content)
|
||||
if content == "" {
|
||||
|
||||
@@ -26,6 +26,7 @@ type knowledgeRuntimeConfig struct {
|
||||
ArkBaseURL string
|
||||
ArkAPIKey string
|
||||
URLMarkdownModel string
|
||||
BrowserFetch config.BrowserFetchConfig
|
||||
}
|
||||
|
||||
type generationStreamDisabledConfigProvider struct {
|
||||
@@ -103,3 +104,14 @@ func knowledgeRuntimeFromConfig(retrievalCfg config.RetrievalConfig, llmCfg conf
|
||||
URLMarkdownModel: model,
|
||||
}
|
||||
}
|
||||
|
||||
func knowledgeRuntimeFromFullConfig(cfg *config.Config) knowledgeRuntimeConfig {
|
||||
if cfg == nil {
|
||||
return knowledgeRuntimeFromConfig(config.RetrievalConfig{}, config.LLMConfig{})
|
||||
}
|
||||
runtime := knowledgeRuntimeFromConfig(cfg.Retrieval, cfg.LLM)
|
||||
browserFetch := cfg.BrowserFetch
|
||||
config.NormalizeBrowserFetchConfig(&browserFetch)
|
||||
runtime.BrowserFetch = browserFetch
|
||||
return runtime
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user