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..c057d39 100644 --- a/core/server_extract.go +++ b/core/server_extract.go @@ -418,3 +418,133 @@ func SanitizeExtractError(err error) string { } return msg } + +const maxBatchExtractURLs = 20 + +type batchExtractPayload struct { + URLs []string `json:"urls"` + Mode string `json:"mode"` +} + +// batchExtractItem is the WebUI-compatible response item per URL. +type batchExtractItem struct { + PageContent string `json:"page_content"` + Metadata map[string]string `json:"metadata"` +} + +func (s *Server) handleBatchExtract(c *fiber.Ctx) error { + startedAt := time.Now() + 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 &APIError{HTTPStatus: fiber.StatusBadRequest, ErrorCode: "invalid_request", Message: "request body is required"} + } + if err := c.BodyParser(&body); err != nil { + return &APIError{HTTPStatus: fiber.StatusBadRequest, ErrorCode: "invalid_request", Message: "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 &APIError{HTTPStatus: fiber.StatusBadRequest, ErrorCode: "invalid_request", Message: "urls array is required and must contain at least one valid URL"} + } + if len(urls) > maxBatchExtractURLs { + return &APIError{ + HTTPStatus: fiber.StatusBadRequest, + ErrorCode: "invalid_request", + Message: 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, + }, + } + }(i, u) + } + wg.Wait() + + _ = startedAt + return c.JSON(results) +} diff --git a/core/server_extract_test.go b/core/server_extract_test.go index 3ac26f0..d8eaaa3 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 TestBatchExtractReturnsWebUIFormat(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) + } +}