diff --git a/core/middleware.go b/core/middleware.go index dd407c2..0019ef1 100644 --- a/core/middleware.go +++ b/core/middleware.go @@ -65,10 +65,10 @@ func RequestContextMiddleware() fiber.Handler { // RequestTimeoutMiddleware bounds wall-clock time per request by attaching a // deadline to the user context, which fasthttp never cancels on client -// disconnect. /mega/* (MegaTimeout) and /extract (batch budget) are exempt. +// disconnect. /mega/* (MegaTimeout) and /extract* (batch budget) are exempt. func RequestTimeoutMiddleware(timeout time.Duration) fiber.Handler { return func(c *fiber.Ctx) error { - if strings.HasPrefix(c.Path(), "/mega/") || c.Path() == "/extract" { + if strings.HasPrefix(c.Path(), "/mega/") || c.Path() == "/extract" || c.Path() == "/extract/batch" { return c.Next() } ctx, cancel := context.WithTimeout(c.UserContext(), timeout) diff --git a/core/server.go b/core/server.go index 6bcd95e..9fa2c7c 100644 --- a/core/server.go +++ b/core/server.go @@ -235,6 +235,7 @@ func NewServerWithOptions(host string, port int, opts ServerOptions, searchEngin serv.app.Get("/mega/engines", serv.handleListEngines) serv.app.Get("/extract", serv.handleExtract) serv.app.Post("/extract", serv.handleExtract) + serv.app.Post("/extract/batch", serv.handleBatchExtract) return &serv } diff --git a/core/server_extract.go b/core/server_extract.go index a3ef971..ac30944 100644 --- a/core/server_extract.go +++ b/core/server_extract.go @@ -418,3 +418,129 @@ func SanitizeExtractError(err error) string { } return msg } + +const maxBatchExtractURLs = 20 + +type batchExtractPayload struct { + URLs []string `json:"urls"` + Mode string `json:"mode"` +} + +// batchExtractItem is a response item for a single extracted URL, using +// page_content/metadata keys +type batchExtractItem struct { + PageContent string `json:"page_content"` + Metadata map[string]string `json:"metadata"` +} + +func (s *Server) handleBatchExtract(c *fiber.Ctx) error { + requestCtx := withRequestUsage(c.UserContext(), "extract-batch") + c.SetUserContext(requestCtx) + defer setNetworkBytesHeader(c, requestCtx) + defer setBrowserProfileHeader(c, requestCtx) + + cfg := s.opts.Extract.Normalized() + if !cfg.Enabled { + return &APIError{HTTPStatus: fiber.StatusNotFound, ErrorCode: "not_found", Message: "Extraction is disabled"} + } + + var body batchExtractPayload + if len(c.Body()) == 0 { + return errInvalidParam("request body is required") + } + if err := c.BodyParser(&body); err != nil { + return errInvalidParam("invalid JSON body") + } + + // Deduplicate and normalize URLs. + seen := make(map[string]struct{}, len(body.URLs)) + var urls []string + for _, raw := range body.URLs { + u := extractpkg.NormalizeURL(strings.TrimSpace(raw)) + if u == "" { + continue + } + if _, dup := seen[u]; dup { + continue + } + seen[u] = struct{}{} + urls = append(urls, u) + } + if len(urls) == 0 { + return errInvalidParam("urls array is required and must contain at least one valid URL") + } + if len(urls) > maxBatchExtractURLs { + return errInvalidParam(fmt.Sprintf("urls array exceeds maximum of %d", maxBatchExtractURLs)) + } + + // Validate all URLs upfront. + for _, u := range urls { + if err := validateExtractTargetURL(c.UserContext(), u, cfg.AllowPrivateNetworks); err != nil { + return &APIError{HTTPStatus: fiber.StatusBadRequest, ErrorCode: "invalid_extract_url", Message: err.Error()} + } + } + + mode := firstNonEmpty(body.Mode, cfg.DefaultMode) + extractor := s.newExtractor() + results := make([]batchExtractItem, len(urls)) + + // Concurrent extraction with bounded parallelism (same pattern as + // EnrichEnvelopeWithExtraction). + ctx, cancel := context.WithTimeout(c.UserContext(), cfg.BatchTimeout(len(urls))) + defer cancel() + + sem := make(chan struct{}, cfg.MaxConcurrent) + var wg sync.WaitGroup + for i, u := range urls { + wg.Add(1) + sem <- struct{}{} + go func(idx int, url string) { + defer wg.Done() + defer func() { <-sem }() + + if err := ctx.Err(); err != nil { + results[idx] = batchExtractItem{ + PageContent: "", + Metadata: map[string]string{"source": url, "error": "batch timeout"}, + } + return + } + + req := extractpkg.ExtractRequest{ + URL: url, + Mode: extractpkg.Mode(mode), + ProxyURL: "", + LangCode: strings.TrimSpace(c.Query("lang")), + Timeout: cfg.Timeout, + MaxBytes: cfg.MaxBytes, + } + result, err := extractor.Extract(ctx, req) + if err != nil { + results[idx] = batchExtractItem{ + PageContent: "", + Metadata: map[string]string{ + "source": url, + "error": SanitizeExtractError(err), + }, + } + return + } + results[idx] = batchExtractItem{ + PageContent: result.Markdown, + Metadata: map[string]string{ + "source": url, + "title": result.Title, + "description": result.Description, + "lang": result.Lang, + "canonical": result.Canonical, + "mode_used": result.Meta.ModeUsed, + "fetched_at": result.Meta.FetchedAt, + "took_ms": fmt.Sprintf("%d", result.Meta.TookMs), + }, + } + }(i, u) + } + wg.Wait() + + return c.JSON(results) +} diff --git a/core/server_extract_test.go b/core/server_extract_test.go index 3ac26f0..8baad7f 100644 --- a/core/server_extract_test.go +++ b/core/server_extract_test.go @@ -2,8 +2,11 @@ package core import ( "context" + "encoding/json" + "fmt" "net/http" "net/http/httptest" + "reflect" "strings" "testing" "time" @@ -109,3 +112,148 @@ func TestValidateExtractTargetURLNormalizesBarePublicIP(t *testing.T) { t.Fatalf("expected bare public IP target to validate after scheme normalization: %v", err) } } + +func TestBatchExtractSingleURL(t *testing.T) { + target := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(`

Test Page

This is a test page with enough content to pass the minimum runes threshold for extraction in batch mode.

`)) + })) + defer target.Close() + + opts := DefaultServerOptions() + opts.Extract = extractpkg.Config{ + Enabled: true, + DefaultMode: string(extractpkg.ModeFast), + Timeout: time.Second, + MaxBytes: 256 * 1024, + MaxConcurrent: 2, + AllowPrivateNetworks: true, + } + s := NewServerWithOptions("127.0.0.1", 0, opts) + + body := fmt.Sprintf(`{"urls":["%s"]}`, target.URL) + req, err := http.NewRequest(http.MethodPost, "/extract/batch", strings.NewReader(body)) + if err != nil { + t.Fatal(err) + } + req.Header.Set("Content-Type", "application/json") + + resp, err := s.app.Test(req) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + t.Fatalf("status = %d, want %d", resp.StatusCode, http.StatusOK) + } + + var results []map[string]interface{} + if err := json.NewDecoder(resp.Body).Decode(&results); err != nil { + t.Fatalf("decode response: %v", err) + } + if len(results) != 1 { + t.Fatalf("results count = %d, want 1", len(results)) + } + if _, ok := results[0]["page_content"]; !ok { + t.Fatalf("expected page_content key, got keys: %v", reflect.ValueOf(results[0]).MapKeys()) + } + if _, ok := results[0]["metadata"]; !ok { + t.Fatalf("expected metadata key, got keys: %v", reflect.ValueOf(results[0]).MapKeys()) + } +} + +func TestBatchExtractHandlesMultipleURLs(t *testing.T) { + target := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(`

Multi

Page with sufficient content for batch extraction test that verifies concurrent processing works correctly.

`)) + })) + defer target.Close() + + opts := DefaultServerOptions() + opts.Extract = extractpkg.Config{ + Enabled: true, + DefaultMode: string(extractpkg.ModeFast), + Timeout: time.Second, + MaxBytes: 256 * 1024, + MaxConcurrent: 2, + AllowPrivateNetworks: true, + } + s := NewServerWithOptions("127.0.0.1", 0, opts) + + body := fmt.Sprintf(`{"urls":["%s/1","%s/2","%s/3"]}`, target.URL, target.URL, target.URL) + req, err := http.NewRequest(http.MethodPost, "/extract/batch", strings.NewReader(body)) + if err != nil { + t.Fatal(err) + } + req.Header.Set("Content-Type", "application/json") + + resp, err := s.app.Test(req) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + t.Fatalf("status = %d, want %d", resp.StatusCode, http.StatusOK) + } + + var results []map[string]interface{} + if err := json.NewDecoder(resp.Body).Decode(&results); err != nil { + t.Fatalf("decode response: %v", err) + } + if len(results) != 3 { + t.Fatalf("results count = %d, want 3", len(results)) + } +} + +func TestBatchExtractRejectsEmptyURLs(t *testing.T) { + opts := DefaultServerOptions() + opts.Extract = extractpkg.DefaultConfig() + s := NewServerWithOptions("127.0.0.1", 0, opts) + + req, err := http.NewRequest(http.MethodPost, "/extract/batch", strings.NewReader(`{"urls":[]}`)) + if err != nil { + t.Fatal(err) + } + req.Header.Set("Content-Type", "application/json") + + resp, err := s.app.Test(req) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusBadRequest { + t.Fatalf("status = %d, want %d", resp.StatusCode, http.StatusBadRequest) + } +} + +func TestBatchExtractRejectsURLsOverLimit(t *testing.T) { + opts := DefaultServerOptions() + opts.Extract = extractpkg.DefaultConfig() + s := NewServerWithOptions("127.0.0.1", 0, opts) + + // Build 21 URLs (limit is 20) + urls := make([]string, 21) + for i := 0; i < 21; i++ { + urls[i] = fmt.Sprintf("https://example.com/%d", i) + } + body, _ := json.Marshal(map[string][]string{"urls": urls}) + + req, err := http.NewRequest(http.MethodPost, "/extract/batch", strings.NewReader(string(body))) + if err != nil { + t.Fatal(err) + } + req.Header.Set("Content-Type", "application/json") + + resp, err := s.app.Test(req) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusBadRequest { + t.Fatalf("status = %d, want %d", resp.StatusCode, http.StatusBadRequest) + } +} diff --git a/core/server_timeout_test.go b/core/server_timeout_test.go index 709103c..a815768 100644 --- a/core/server_timeout_test.go +++ b/core/server_timeout_test.go @@ -12,8 +12,8 @@ import ( ) // FP-2: every endpoint that doesn't manage its own deadline budget must get -// one from RequestTimeoutMiddleware; /mega/* (MegaTimeout) and /extract -// (batch budget) keep theirs. +// one from RequestTimeoutMiddleware; /mega/* (MegaTimeout) and /extract, +// /extract/batch (batch budget) keep theirs. func TestRequestTimeoutMiddlewareSetsDeadlineExceptBudgetedPaths(t *testing.T) { app := fiber.New() app.Use(RequestTimeoutMiddleware(time.Minute)) @@ -32,12 +32,14 @@ func TestRequestTimeoutMiddlewareSetsDeadlineExceptBudgetedPaths(t *testing.T) { app.Post("/google/parse", record("/google/parse")) app.Get("/mega/search", record("/mega/search")) app.Get("/extract", record("/extract")) + app.Post("/extract/batch", record("/extract/batch")) for path, method := range map[string]string{ "/google/search": http.MethodGet, "/google/parse": http.MethodPost, "/mega/search": http.MethodGet, "/extract": http.MethodGet, + "/extract/batch": http.MethodPost, } { req := httptest.NewRequest(method, path, nil) resp, err := app.Test(req, -1) @@ -54,6 +56,7 @@ func TestRequestTimeoutMiddlewareSetsDeadlineExceptBudgetedPaths(t *testing.T) { "/google/parse": true, "/mega/search": false, "/extract": false, + "/extract/batch": false, } { if deadlines[path] != want { t.Errorf("%s: deadline attached = %v, want %v", path, deadlines[path], want) diff --git a/docs/openapi.yaml b/docs/openapi.yaml index 6fb6449..9e57353 100644 --- a/docs/openapi.yaml +++ b/docs/openapi.yaml @@ -547,6 +547,34 @@ paths: $ref: "#/components/responses/BadRequestError" "502": $ref: "#/components/responses/BadGatewayError" + + /extract/batch: + post: + tags: [Extract] + operationId: extractBatch + summary: Extract content from multiple URLs + description: > + Accepts an array of URLs and returns extracted page content for each + one. Each result contains `page_content` (markdown) and `metadata` + (title, source, lang, etc.). + requestBody: + required: true + content: + application/json: + schema: + $ref: "#/components/schemas/BatchExtractRequest" + responses: + "200": + description: Batch extraction results + content: + application/json: + schema: + type: array + items: + $ref: "#/components/schemas/BatchExtractItem" + "400": + $ref: "#/components/responses/BadRequestError" + /health: get: tags: [Health] @@ -2037,3 +2065,31 @@ components: $ref: "#/components/schemas/MegaEngineInfo" total: type: integer + + # ── Batch extract ───────────────────────────────────────────────── + BatchExtractRequest: + type: object + required: [urls] + properties: + urls: + type: array + items: + type: string + maxItems: 20 + description: URLs to extract content from (max 20) + mode: + type: string + enum: [auto, fast, rendered] + description: "Extraction mode (default: auto)" + + BatchExtractItem: + type: object + properties: + page_content: + type: string + description: Extracted markdown content + metadata: + type: object + additionalProperties: + type: string + description: Page metadata (title, source, lang, etc.)