From 8d4e8910e8f873b32f1cfe8d7f35fef18e6f468e Mon Sep 17 00:00:00 2001 From: Nas Kavian Date: Fri, 18 Sep 2026 18:05:24 -0700 Subject: [PATCH] feat(directory)!: add mixed service and collection discovery --- README.md | 16 +- agent/README.md | 5 + agent/agent.go | 2 +- directory/README.md | 115 ++++++- directory/client.go | 119 +++---- directory/client_test.go | 30 +- directory/conformance_test.go | 76 ++--- directory/search.go | 188 +++++++++++ directory/search_test.go | 299 ++++++++++++++++++ directory/shape_test.go | 12 +- directory/transport_test.go | 16 +- directory/types.go | 54 +++- directory/validation.go | 40 ++- examples/odp-agent-discovery/README.md | 13 +- examples/odp-agent-discovery/main.go | 17 + .../odp-agent-discovery/mock_directory.go | 50 ++- testdata/consumer/main.go | 8 +- 17 files changed, 870 insertions(+), 190 deletions(-) create mode 100644 directory/search.go create mode 100644 directory/search_test.go diff --git a/README.md b/README.md index db2489b..28d6d44 100644 --- a/README.md +++ b/README.md @@ -109,9 +109,9 @@ children := odp.CollectionSearchRequest{ODPVersion: odp.Version, ParentID: odp.S ## Directory discovery -Package `directory` searches candidate Services through the canonical production directory or its -fixed sandbox environment. It validates cached Service summaries, follows opaque same-origin -continuations, exposes structured facets, and provides keyword suggestions. +Package `directory` searches indexed Services and Collections through the canonical production +directory or its fixed sandbox environment. It validates results, follows opaque same-origin +continuations when offered, exposes structured facets, and provides search suggestions. ```go directoryClient, err := directory.New(directory.Options{}) @@ -127,7 +127,7 @@ for candidate, err := range directoryClient.SearchServices(ctx, directory.Search Options: []odp.PaymentOption{odp.PaymentOptionInflow, odp.PaymentOptionSolana}, }}, }, -}, directory.IterationOptions{MaxItems: 20}) { +}, directory.IterationOptions{MaxItems: 20}).Items { if err != nil { return err } @@ -135,7 +135,13 @@ for candidate, err := range directoryClient.SearchServices(ctx, directory.Search } ``` -See the [directory package guide](./directory/README.md) for page traversal, suggestions, and +Use `Search` and `ContinueSearch` for mixed Service/Collection results, or `SearchServices` and +`ContinueSearchServices` for Service-only discovery. Each returns independent `Items` and +`Responses` iterators. `Suggest` returns matching target names; `SuggestServices` provides +Service-only keyword suggestions. Mixed search is capped at 100 results without continuation; +refine the query rather than assuming every match was returned. + +See the [directory package guide](./directory/README.md) for response traversal, migration, suggestions, and sandbox usage. ## Agent integration diff --git a/agent/README.md b/agent/README.md index 515db37..29c806a 100644 --- a/agent/README.md +++ b/agent/README.md @@ -201,6 +201,11 @@ Agent processing. Recognized descriptors remain subject to current-version valid See the [runnable Agent example](../examples/odp-agent-discovery/README.md), which clearly labels and isolates its mock directory while querying live ODP Services. +For discovery that includes submitted Collections, call the Directory client's `Search` and +iterate its `Items`. A Collection result identifies its owning Service and remote Collection ID. +Inspect that Service, then call `GetCollection` with the ID. `SearchOfferingsAcrossServices` +continues to select Services only; it does not treat Collections as separate Services. + ## Related documentation - [Directory integration](../directory/README.md) diff --git a/agent/agent.go b/agent/agent.go index f2a0aa4..84d8764 100644 --- a/agent/agent.go +++ b/agent/agent.go @@ -44,7 +44,7 @@ func (agent *Agent) searchOfferingsAcrossServices(ctx context.Context, request F return } services := make([]directory.Service, 0, maxServices) - for service, err := range agent.directory.SearchServices(ctx, request.Services, directory.IterationOptions{MaxItems: maxServices}) { + for service, err := range agent.directory.SearchServices(ctx, request.Services, directory.IterationOptions{MaxItems: maxServices}).Items { if err != nil { yield(DiscoveryEvent{}, err) return diff --git a/directory/README.md b/directory/README.md index cab7b1c..383fe2d 100644 --- a/directory/README.md +++ b/directory/README.md @@ -1,14 +1,62 @@ # ODP directory package -Package `directory` searches the one canonical ODP directory for candidate Services. It does not -search Service catalogs. After discovery, an Agent inspects each result's live ODP document and +Package `directory` searches indexed Services and submitted Collections. It does not crawl +catalogs or index Offerings. After discovery, an Agent inspects each result's live ODP document and queries that Service's Collections and Offerings. The production origin is fixed at `https://api.inflowpay.ai`. Select `Sandbox` to use the fixed `https://sandbox.inflowpay.ai` environment. Callers cannot configure another origin. -## Search Services +## Search Services and Collections + +```go +directoryClient, err := directory.New(directory.Options{}) +if err != nil { + return err +} + +search := directoryClient.Search(ctx, directory.DirectorySearchRequest{ + SearchRequest: directory.SearchRequest{Query: "weather forecast", Limit: 25}, +}, directory.IterationOptions{MaxItems: 25}) + +for result, err := range search.Items { + if err != nil { + return err + } + switch result.Type { + case "service": + fmt.Printf("Service: %s (%s)\n", result.Service.Name, result.Service.ServiceOrigin) + case "collection": + fmt.Printf("Collection: %s, ID %s, through %s\n", + result.Collection.Name, result.Collection.ID, result.Service.ServiceOrigin) + default: + fmt.Printf("Unsupported result type: %s\n", result.Type) + } +} +``` + +Omit `Types` to select both types, or provide `Types: []string{"collection"}` or `[]string{"service"}`. +The list must be nonempty and distinct. Filters apply to the owning Service for either type. + +A Collection is identified by its owning Service origin and case-sensitive `Collection.ID`. +Inspect that Service, then call the Agent client's `GetCollection` with the ID. +`Result.IndexedAt` reports Collection freshness; `Result.Service.IndexedAt` reports its parent's +freshness. A Service may have `AvailableThrough` platform attribution. A Collection's attribution +is its owning `Service`. + +Unknown types retain the wire type in `Type` and complete JSON in `Raw`; their `Service` and +`Collection` pointers are nil. Do not treat them as Services. Known types are validated and +retain additive metadata in `Additional`. Nested Service parsing omits unverified execution +metadata such as endpoint paths, just as Service-only search does. + +The mixed endpoint returns at most 100 results (also its default limit), without continuation. +An absent `Next` does not promise that all matches were returned; refine the query or filters. +Collection search eligibility does not depend on permission to show a Directory landing card. +Mixed facets count all matching targets, not just returned items: a Service and two Collections +count as three. Descriptor facets use the owning Service's metadata. + +## Search only Services `SearchServices` lazily traverses directory pages and yields validated Service summaries. Filters are structured and work without natural-language interpretation. @@ -31,7 +79,7 @@ request := directory.SearchRequest{ Limit: 25, } -for candidate, err := range directoryClient.SearchServices(ctx, request, directory.IterationOptions{}) { +for candidate, err := range directoryClient.SearchServices(ctx, request, directory.IterationOptions{}).Items { if err != nil { return err } @@ -44,14 +92,16 @@ InFlow or Solana through MPP. A protocol-only `PaymentFilter{Name: odp.ProtocolM Service that advertises MPP. `Facets.Payments` reports protocol counts, `Facets.PaymentOptions` reports each protocol-option count independently, and `Facets.Trust` reports trust protocol counts. -Use `SearchPages` when facet counts or page-level additive members are needed: +Every search returns a `SearchSequence` with independent lazy `Items` and `Responses` iterators. +Iterating both performs two searches. Creating a sequence performs no network requests and +captures the request values. Use `Responses` for facets, issues, additive fields or continuation: ```go -for page, err := range directoryClient.SearchPages(ctx, request, directory.IterationOptions{}) { +for page, err := range directoryClient.SearchServices(ctx, request, directory.IterationOptions{}).Responses { if err != nil { return err } - consume(page.Items, page.Facets) + fmt.Printf("Results: %d, facets: %+v\n", len(page.Items), page.Facets) } ``` @@ -60,34 +110,47 @@ that leaves that origin, repeats a page already visited, or cannot be resolved e with an error. Stopping iteration stops network activity. `IterationOptions.MaxPages` is the caller's page budget and defaults to 16. Reaching it ends the -sequence without an error, and the last page keeps the `Next` that `ContinueSearchPages` or -`ContinueSearchServices` resumes from: +sequence without an error, and the last response keeps `Next`. Resume mixed search using +`ContinueSearch`, or Service-only search using `ContinueSearchServices`: ```go var resume string -for page, err := range directoryClient.SearchPages(ctx, request, directory.IterationOptions{MaxPages: 4}) { +for page, err := range directoryClient.SearchServices(ctx, request, directory.IterationOptions{MaxPages: 4}).Responses { if err != nil { return err } - consume(page.Items, page.Facets) + fmt.Printf("Results: %d, facets: %+v\n", len(page.Items), page.Facets) resume = page.Next } if resume != "" { - for page, err := range directoryClient.ContinueSearchPages(ctx, resume, directory.IterationOptions{}) { - // ... + for page, err := range directoryClient.ContinueSearchServices(ctx, resume, directory.IterationOptions{}).Responses { + if err != nil { + return err + } + fmt.Printf("Results: %d\n", len(page.Items)) } } ``` A budget above 10,000 pages is rejected, and a directory that offers a continuation for 10,000 consecutive pages ends the traversal with an error rather than quietly appearing exhausted. -`MaxItems` bounds `SearchServices` at up to 10,000 results. +`MaxItems` bounds item iteration at up to 10,000 results; zero leaves it unbounded within the +request budget. Response iteration does not truncate responses to that item limit. Pass a +`context.Context` to cancel requests. Both search families share the same transport policies. + +### Migrating existing Go callers + +This is a breaking Go API change. Replace `SearchPages(...)` with +`SearchServices(...).Responses`, and `ContinueSearchPages(...)` with +`ContinueSearchServices(...).Responses`. Append `.Items` to existing `SearchServices(...)` and +`ContinueSearchServices(...)` iteration. Replace `SearchPage` with `SearchResponse[Service]`. +There are no deprecated aliases. The Agent's cross-Service Offering discovery remains Service-only. ## Malformed results One unusable record does not discard the page it arrived on. A record that fails validation is -omitted from `Items` and reported in `SearchPage.Issues` with its index and the reason; -`IterationOptions.OnIssue` receives the same reports while iterating Services: +omitted from `Items` and reported in `SearchResponse.Issues` with its original index and the reason; +`IterationOptions.OnIssue` receives the same reports while iterating `Items`: ```go options := directory.IterationOptions{OnIssue: func(issue directory.Issue) { @@ -99,7 +162,25 @@ A malformed page envelope, by contrast, still fails the traversal. ## Suggestions -Suggestions help an Agent discover keyword vocabulary without downloading a global keyword list. +`Suggest` matches indexed names, descriptions and keywords, returning **names of matching +Services and Collections**, not the matching text itself. Despite `Prefix`, matching uses +substrings and whitespace-separated alternative terms. Names are deduplicated, with a maximum +and default of 25. These are candidate queries, not resource identifiers. Pass a selected string +to `Search`. Collection surfacing permission does not restrict suggestions. + +```go +names, err := directoryClient.Suggest(ctx, directory.SuggestionRequest{ + Prefix: "we", Limit: 10, + Filters: &directory.ServiceFilters{Keywords: []string{"weather"}}, +}) +``` + +`Suggest` sends POST `/v1/directory/suggestions`. Its optional `Filters` use the same +structure as search, including AEP, keywords, ODP operations, payments and trust. +Collection filters apply to the owning Service; the output remains names only. + +`SuggestServices` uses GET and retains keyword-prefix suggestions for Service-only discovery. +It does not accept filters: ```go suggestions, err := directoryClient.SuggestServices(ctx, directory.SuggestionRequest{ diff --git a/directory/client.go b/directory/client.go index e8c1fae..6e655f4 100644 --- a/directory/client.go +++ b/directory/client.go @@ -7,7 +7,6 @@ import ( "errors" "fmt" "io" - "iter" "mime" "net" "net/http" @@ -58,61 +57,48 @@ func New(options Options) (*Client, error) { return &Client{environment: environment, httpClient: &httpClient, originURL: originURL}, nil } -func (client *Client) SearchPages(ctx context.Context, request SearchRequest, options IterationOptions) iter.Seq2[SearchPage, error] { +func (client *Client) SearchServices(ctx context.Context, request SearchRequest, options IterationOptions) SearchSequence[Service] { body, validationError := validateSearchRequest(request) maxPages, budgetError := pageBudget(options.MaxPages) if validationError == nil { validationError = budgetError } - return func(yield func(SearchPage, error) bool) { + responses := func(yield func(SearchResponse[Service], error) bool) { if validationError != nil { - yield(SearchPage{}, validationError) + yield(SearchResponse[Service]{}, validationError) return } - client.traverse(ctx, client.originURL.JoinPath("v1", "services", "search"), http.MethodPost, body, maxPages, yield) + traverse(client, ctx, client.originURL.JoinPath("v1", "services", "search"), http.MethodPost, body, maxPages, parseSearchPage, yield) } + return sequence(responses, options) } -// ContinueSearchPages resumes a traversal from the Next of a page a previous search yielded, so a -// caller that stopped at its own page budget can pick the sequence up rather than start again. -func (client *Client) ContinueSearchPages(ctx context.Context, next string, options IterationOptions) iter.Seq2[SearchPage, error] { +func (client *Client) ContinueSearchServices(ctx context.Context, next string, options IterationOptions) SearchSequence[Service] { maxPages, validationError := pageBudget(options.MaxPages) - return func(yield func(SearchPage, error) bool) { + responses := func(yield func(SearchResponse[Service], error) bool) { if validationError != nil { - yield(SearchPage{}, validationError) + yield(SearchResponse[Service]{}, validationError) return } target, err := client.continuationURL(next) if err != nil { - yield(SearchPage{}, err) + yield(SearchResponse[Service]{}, err) return } - client.traverse(ctx, target, http.MethodGet, nil, maxPages, yield) + traverse(client, ctx, target, http.MethodGet, nil, maxPages, parseSearchPage, yield) } + return sequence(responses, options) } -func (client *Client) SearchServices(ctx context.Context, request SearchRequest, options IterationOptions) iter.Seq2[Service, error] { - return func(yield func(Service, error) bool) { - if options.MaxItems < 0 || options.MaxItems > maximumItems { - yield(Service{}, fmt.Errorf("maxItems must be an integer from 1 through %d", maximumItems)) - return - } - client.services(client.SearchPages(ctx, request, options), options, yield) - } +func (client *Client) SuggestServices(ctx context.Context, request SuggestionRequest) ([]string, error) { + return client.suggest(ctx, "services", request) } -// ContinueSearchServices resumes an item traversal from a page's Next. -func (client *Client) ContinueSearchServices(ctx context.Context, next string, options IterationOptions) iter.Seq2[Service, error] { - return func(yield func(Service, error) bool) { - if options.MaxItems < 0 || options.MaxItems > maximumItems { - yield(Service{}, fmt.Errorf("maxItems must be an integer from 1 through %d", maximumItems)) - return - } - client.services(client.ContinueSearchPages(ctx, next, options), options, yield) - } +func (client *Client) Suggest(ctx context.Context, request SuggestionRequest) ([]string, error) { + return client.suggest(ctx, "directory", request) } -func (client *Client) SuggestServices(ctx context.Context, request SuggestionRequest) ([]string, error) { +func (client *Client) suggest(ctx context.Context, resource string, request SuggestionRequest) ([]string, error) { prefix, err := requireText(request.Prefix, "prefix", 1, 128) if err != nil { return nil, err @@ -120,13 +106,34 @@ func (client *Client) SuggestServices(ctx context.Context, request SuggestionReq if request.Limit < 0 || request.Limit > 25 { return nil, errors.New("limit must be an integer from 1 through 25") } - target := client.originURL.JoinPath("v1", "services", "suggestions") - query := url.Values{"prefix": []string{prefix}} - if request.Limit != 0 { - query.Set("limit", strconv.Itoa(request.Limit)) + target := client.originURL.JoinPath("v1", resource, "suggestions") + method := http.MethodGet + var body []byte + if resource == "directory" { + request.Prefix = prefix + if request.Filters != nil { + filters, err := validateFilters(*request.Filters) + if err != nil { + return nil, err + } + request.Filters = &filters + } + body, err = json.Marshal(request) + if err != nil { + return nil, err + } + method = http.MethodPost + } else { + if request.Filters != nil { + return nil, errors.New("Service-only suggestions do not support filters") + } + query := url.Values{"prefix": []string{prefix}} + if request.Limit != 0 { + query.Set("limit", strconv.Itoa(request.Limit)) + } + target.RawQuery = query.Encode() } - target.RawQuery = query.Encode() - data, err := client.requestJSON(ctx, http.MethodGet, target, nil) + data, err := client.requestJSON(ctx, method, target, body) if err != nil { return nil, err } @@ -145,19 +152,19 @@ func pageBudget(requested int) (int, error) { // traverse walks the continuation chain, yielding each page until the caller stops, the directory // stops offering one, or a budget runs out. -func (client *Client) traverse(ctx context.Context, start *url.URL, method string, body []byte, maxPages int, yield func(SearchPage, error) bool) { +func traverse[Item any](client *Client, ctx context.Context, start *url.URL, method string, body []byte, maxPages int, parse func([]byte) (SearchResponse[Item], error), yield func(SearchResponse[Item], error) bool) { current := start requestBody := body visited := map[string]struct{}{} for pageNumber := 0; pageNumber < maxPages; pageNumber++ { data, err := client.requestJSON(ctx, method, current, requestBody) if err != nil { - yield(SearchPage{}, err) + yield(SearchResponse[Item]{}, err) return } - page, err := parseSearchPage(data) + page, err := parse(data) if err != nil { - yield(SearchPage{}, err) + yield(SearchResponse[Item]{}, err) return } if !yield(page, nil) || page.Next == "" { @@ -168,20 +175,20 @@ func (client *Client) traverse(ctx context.Context, start *url.URL, method strin // never going to be requested. if pageNumber+1 >= maxPages { if maxPages == maximumPages { - yield(SearchPage{}, fmt.Errorf("Directory pagination exceeded its %d-page traversal limit", maximumPages)) + yield(SearchResponse[Item]{}, fmt.Errorf("Directory pagination exceeded its %d-page traversal limit", maximumPages)) } return } next, err := client.continuationURL(page.Next) if err != nil { - yield(SearchPage{}, err) + yield(SearchResponse[Item]{}, err) return } // A continuation has to advance. Without this a directory that repeats one link keeps the // caller reading the same page until the budget runs out. key := traversalKey(next) if _, seen := visited[key]; seen { - yield(SearchPage{}, errors.New("Directory pagination loop detected")) + yield(SearchResponse[Item]{}, errors.New("Directory pagination loop detected")) return } visited[key] = struct{}{} @@ -191,32 +198,6 @@ func (client *Client) traverse(ctx context.Context, start *url.URL, method strin } } -func (client *Client) services(pages iter.Seq2[SearchPage, error], options IterationOptions, yield func(Service, error) bool) { - count := 0 - for page, err := range pages { - if err != nil { - yield(Service{}, err) - return - } - for _, issue := range page.Issues { - if options.OnIssue != nil { - options.OnIssue(issue) - } - } - for _, service := range page.Items { - count++ - if !yield(service, nil) { - return - } - // Checked after the yield: checking before it lets the enclosing loop pull another - // page whenever the budget falls exactly on a page boundary. - if options.MaxItems != 0 && count >= options.MaxItems { - return - } - } - } -} - func (client *Client) requestJSON(ctx context.Context, method string, target *url.URL, body []byte) ([]byte, error) { current := target for redirects := 0; ; redirects++ { diff --git a/directory/client_test.go b/directory/client_test.go index b5ddc58..cb5f25c 100644 --- a/directory/client_test.go +++ b/directory/client_test.go @@ -75,7 +75,7 @@ func TestSearchPagesUsesCanonicalOriginAndStructuredFilters(t *testing.T) { return response(http.StatusOK, `{"items":[`+serviceResult+`],"facets":{"enrollment":[{"value":{"name":"aep"},"count":1}],"keywords":[{"value":"gpu","count":1}],"operations":[{"value":{"authentication":"required","name":"get-offering"},"count":1}],"payment_options":[{"value":{"name":"mpp","option":"inflow"},"count":1},{"value":{"name":"mpp","option":"solana"},"count":1}],"payments":[{"value":{"authentication":"not-required","name":"mpp","options":["inflow","solana"]},"count":1}],"trust":[{"value":{"name":"tap"},"count":1}]}}`, nil), nil }) - pages := client(t, transport, directory.Production).SearchPages(t.Context(), directory.SearchRequest{ + pages := client(t, transport, directory.Production).SearchServices(t.Context(), directory.SearchRequest{ Query: "compute", Filters: &directory.ServiceFilters{ Enrollment: []odp.EnrollmentProtocol{{Name: odp.ProtocolAEP}}, @@ -87,7 +87,7 @@ func TestSearchPagesUsesCanonicalOriginAndStructuredFilters(t *testing.T) { Trust: []odp.TrustProtocol{{Name: odp.ProtocolTAP}}, }, Limit: 25, - }, directory.IterationOptions{}) + }, directory.IterationOptions{}).Responses page, err := first(pages) if err != nil { t.Fatal(err) @@ -127,7 +127,7 @@ func TestSearchPagesUsesSandboxOnlyWhenSelected(t *testing.T) { if value.Environment() != directory.Sandbox { t.Fatalf("environment = %q", value.Environment()) } - if _, err := first(value.SearchPages(t.Context(), directory.SearchRequest{}, directory.IterationOptions{})); err != nil { + if _, err := first(value.SearchServices(t.Context(), directory.SearchRequest{}, directory.IterationOptions{}).Responses); err != nil { t.Fatal(err) } if target != "https://sandbox.inflowpay.ai/v1/services/search" { @@ -140,7 +140,7 @@ func TestSearchPagesFiltersUnknownProtocols(t *testing.T) { transport := roundTripFunc(func(_ *http.Request) (*http.Response, error) { return response(http.StatusOK, `{"items":[`+result+`]}`, nil), nil }) - page, err := first(client(t, transport, directory.Production).SearchPages(t.Context(), directory.SearchRequest{}, directory.IterationOptions{})) + page, err := first(client(t, transport, directory.Production).SearchServices(t.Context(), directory.SearchRequest{}, directory.IterationOptions{}).Responses) if err != nil { t.Fatal(err) } @@ -155,7 +155,7 @@ func TestSearchPagesReportsMalformedKnownProtocol(t *testing.T) { transport := roundTripFunc(func(_ *http.Request) (*http.Response, error) { return response(http.StatusOK, `{"items":[`+result+`,`+serviceResult+`]}`, nil), nil }) - page, err := first(client(t, transport, directory.Production).SearchPages(t.Context(), directory.SearchRequest{}, directory.IterationOptions{})) + page, err := first(client(t, transport, directory.Production).SearchServices(t.Context(), directory.SearchRequest{}, directory.IterationOptions{}).Responses) if err != nil { t.Fatal(err) } @@ -179,7 +179,7 @@ func TestSearchServicesFollowsOpaqueContinuationWithGet(t *testing.T) { return response(http.StatusOK, `{"items":[`+serviceResult+`],"next":"/v1/services/search?cursor=opaque"}`, nil), nil }) var origins []string - for service, err := range client(t, transport, directory.Production).SearchServices(t.Context(), directory.SearchRequest{}, directory.IterationOptions{}) { + for service, err := range client(t, transport, directory.Production).SearchServices(t.Context(), directory.SearchRequest{}, directory.IterationOptions{}).Items { if err != nil { t.Fatal(err) } @@ -229,7 +229,7 @@ func TestSearchRejectsCrossOriginContinuation(t *testing.T) { return response(http.StatusOK, `{"items":[],"next":"https://other.example/search"}`, nil), nil }) var received error - for _, err := range client(t, transport, directory.Production).SearchPages(t.Context(), directory.SearchRequest{}, directory.IterationOptions{}) { + for _, err := range client(t, transport, directory.Production).SearchServices(t.Context(), directory.SearchRequest{}, directory.IterationOptions{}).Responses { if err != nil { received = err } @@ -246,7 +246,7 @@ func TestRequestErrorPreservesResponseDetails(t *testing.T) { transport := roundTripFunc(func(*http.Request) (*http.Response, error) { return response(http.StatusServiceUnavailable, `{"title":"Unavailable","detail":"try again shortly"}`, headers), nil }) - _, err := first(client(t, transport, directory.Production).SearchPages(t.Context(), directory.SearchRequest{}, directory.IterationOptions{})) + _, err := first(client(t, transport, directory.Production).SearchServices(t.Context(), directory.SearchRequest{}, directory.IterationOptions{}).Responses) var requestError *directory.RequestError if !errors.As(err, &requestError) { t.Fatalf("error = %v", err) @@ -264,7 +264,7 @@ func TestSearchStopsReadingOversizedResponse(t *testing.T) { transport := roundTripFunc(func(*http.Request) (*http.Response, error) { return response(http.StatusOK, strings.Repeat("x", 524_289), nil), nil }) - _, err := first(client(t, transport, directory.Production).SearchPages(t.Context(), directory.SearchRequest{}, directory.IterationOptions{})) + _, err := first(client(t, transport, directory.Production).SearchServices(t.Context(), directory.SearchRequest{}, directory.IterationOptions{}).Responses) if err == nil || !strings.Contains(err.Error(), "byte limit") { t.Fatalf("error = %v", err) } @@ -282,7 +282,7 @@ func TestRedirectPolicy(t *testing.T) { } return response(http.StatusOK, `{"items":[]}`, nil), nil }) - if _, err := first(client(t, transport, directory.Production).SearchPages(t.Context(), directory.SearchRequest{}, directory.IterationOptions{})); err != nil { + if _, err := first(client(t, transport, directory.Production).SearchServices(t.Context(), directory.SearchRequest{}, directory.IterationOptions{}).Responses); err != nil { t.Fatal(err) } if strings.Join(methods, ",") != "POST,GET" { @@ -296,7 +296,7 @@ func TestRedirectPolicy(t *testing.T) { headers.Set("Location", "https://other.example/search") return response(http.StatusTemporaryRedirect, "", headers), nil }) - _, err := first(client(t, transport, directory.Production).SearchPages(t.Context(), directory.SearchRequest{}, directory.IterationOptions{})) + _, err := first(client(t, transport, directory.Production).SearchServices(t.Context(), directory.SearchRequest{}, directory.IterationOptions{}).Responses) if err == nil || !strings.Contains(err.Error(), "changed origin") { t.Fatalf("error = %v", err) } @@ -308,7 +308,7 @@ func TestRedirectPolicy(t *testing.T) { headers.Set("Location", "/again") return response(http.StatusTemporaryRedirect, "", headers), nil }) - _, err := first(client(t, transport, directory.Production).SearchPages(t.Context(), directory.SearchRequest{}, directory.IterationOptions{})) + _, err := first(client(t, transport, directory.Production).SearchServices(t.Context(), directory.SearchRequest{}, directory.IterationOptions{}).Responses) if err == nil || !strings.Contains(err.Error(), "redirect limit") { t.Fatalf("error = %v", err) } @@ -337,7 +337,7 @@ func TestSearchValidation(t *testing.T) { t.Fatal("transport called for invalid input") return nil, nil }) - _, err := first(client(t, transport, directory.Production).SearchServices(t.Context(), test.request, test.options)) + _, err := first(client(t, transport, directory.Production).SearchServices(t.Context(), test.request, test.options).Items) if err == nil { t.Fatal("expected validation error") } @@ -352,7 +352,7 @@ func TestSearchPreservesAdditiveMembersAndHonorsItemLimit(t *testing.T) { return response(http.StatusOK, `{"trace":"abc","items":[`+firstService+`,`+secondService+`]}`, nil), nil }) var services []directory.Service - for service, err := range client(t, transport, directory.Production).SearchServices(t.Context(), directory.SearchRequest{}, directory.IterationOptions{MaxItems: 1}) { + for service, err := range client(t, transport, directory.Production).SearchServices(t.Context(), directory.SearchRequest{}, directory.IterationOptions{MaxItems: 1}).Items { if err != nil { t.Fatal(err) } @@ -370,7 +370,7 @@ func TestSearchUsesContext(t *testing.T) { }) ctx, cancel := context.WithCancel(t.Context()) cancel() - _, err := first(client(t, transport, directory.Production).SearchPages(ctx, directory.SearchRequest{}, directory.IterationOptions{})) + _, err := first(client(t, transport, directory.Production).SearchServices(ctx, directory.SearchRequest{}, directory.IterationOptions{}).Responses) if !errors.Is(err, context.Canceled) { t.Fatalf("error = %v", err) } diff --git a/directory/conformance_test.go b/directory/conformance_test.go index 32c2114..0c5db6c 100644 --- a/directory/conformance_test.go +++ b/directory/conformance_test.go @@ -39,10 +39,10 @@ func always(body string) func(*http.Request) (*http.Response, error) { } } -func collectPages(sequence func(func(directory.SearchPage, error) bool)) ([]directory.SearchPage, error) { - var pages []directory.SearchPage +func collectPages(sequence func(func(directory.SearchResponse[directory.Service], error) bool)) ([]directory.SearchResponse[directory.Service], error) { + var pages []directory.SearchResponse[directory.Service] var failure error - sequence(func(value directory.SearchPage, err error) bool { + sequence(func(value directory.SearchResponse[directory.Service], err error) bool { if err != nil { failure = err return false @@ -78,7 +78,7 @@ func TestServiceOriginsMustBePublicAndSecure(t *testing.T) { } for _, origin := range refused { body := page(strings.Replace(record, "https://compute.example", origin, 1)) - pages, err := collectPages(serve(t, always(body)).SearchPages(t.Context(), directory.SearchRequest{}, directory.IterationOptions{})) + pages, err := collectPages(serve(t, always(body)).SearchServices(t.Context(), directory.SearchRequest{}, directory.IterationOptions{}).Responses) if err != nil { t.Fatalf("%s: %v", origin, err) } @@ -88,7 +88,7 @@ func TestServiceOriginsMustBePublicAndSecure(t *testing.T) { } for _, origin := range []string{"https://compute.example", "https://8.8.8.8"} { body := page(strings.Replace(record, "https://compute.example", origin, 1)) - pages, err := collectPages(serve(t, always(body)).SearchPages(t.Context(), directory.SearchRequest{}, directory.IterationOptions{})) + pages, err := collectPages(serve(t, always(body)).SearchServices(t.Context(), directory.SearchRequest{}, directory.IterationOptions{}).Responses) if err != nil || len(pages[0].Items) != 1 { t.Errorf("%s: page = %#v, %v", origin, pages, err) } @@ -100,7 +100,7 @@ func TestUnverifiedServiceDocumentMembersDoNotReachTheCaller(t *testing.T) { // assert must not arrive looking like something the caller may act on. members := `"http":{"endpoint_base":"/evil"},"odp_version":"9.9","mcp":[{"name":"x","type":"streamable-http","url":"/mcp"}],` + `"payment_origins":["https://evil.example"],"search_capabilities":{"filters":{"inline":[]}},"branding":{},"rank":3` - pages, err := collectPages(serve(t, always(page(recordWith(members)))).SearchPages(t.Context(), directory.SearchRequest{}, directory.IterationOptions{})) + pages, err := collectPages(serve(t, always(page(recordWith(members)))).SearchServices(t.Context(), directory.SearchRequest{}, directory.IterationOptions{}).Responses) if err != nil || len(pages[0].Items) != 1 { t.Fatalf("pages = %#v, %v", pages, err) } @@ -116,7 +116,7 @@ func TestUnverifiedServiceDocumentMembersDoNotReachTheCaller(t *testing.T) { func TestOneUnusableRecordDoesNotDiscardItsPage(t *testing.T) { broken := strings.Replace(record, `"name":"Compute"`, `"name":""`, 1) body := page(broken + "," + record + "," + strings.Replace(record, "compute.example", "storage.example", 1)) - pages, err := collectPages(serve(t, always(body)).SearchPages(t.Context(), directory.SearchRequest{}, directory.IterationOptions{})) + pages, err := collectPages(serve(t, always(body)).SearchServices(t.Context(), directory.SearchRequest{}, directory.IterationOptions{}).Responses) if err != nil { t.Fatal(err) } @@ -126,7 +126,7 @@ func TestOneUnusableRecordDoesNotDiscardItsPage(t *testing.T) { // An item traversal cannot see the page, so the issues reach it through the callback. var seen []directory.Issue options := directory.IterationOptions{OnIssue: func(issue directory.Issue) { seen = append(seen, issue) }} - services, err := collectServices(serve(t, always(body)).SearchServices(t.Context(), directory.SearchRequest{}, options)) + services, err := collectServices(serve(t, always(body)).SearchServices(t.Context(), directory.SearchRequest{}, options).Items) if err != nil || len(services) != 2 || len(seen) != 1 { t.Fatalf("services = %d, issues = %#v, %v", len(services), seen, err) } @@ -145,30 +145,30 @@ func TestPaginationBudgetsAndResumption(t *testing.T) { // A caller's own page budget stops the traversal without an error, and the last page keeps a // continuation to resume from. client := serve(t, handler) - pages, err := collectPages(client.SearchPages(t.Context(), directory.SearchRequest{}, directory.IterationOptions{MaxPages: 3})) + pages, err := collectPages(client.SearchServices(t.Context(), directory.SearchRequest{}, directory.IterationOptions{MaxPages: 3}).Responses) if err != nil || len(pages) != 3 || pages[2].Next == "" { t.Fatalf("pages = %d, err = %v", len(pages), err) } - resumed, err := collectPages(client.ContinueSearchPages(t.Context(), pages[2].Next, directory.IterationOptions{MaxPages: 2})) + resumed, err := collectPages(client.ContinueSearchServices(t.Context(), pages[2].Next, directory.IterationOptions{MaxPages: 2}).Responses) if err != nil || len(resumed) != 2 { t.Fatalf("resumed = %d, err = %v", len(resumed), err) } - services, err := collectServices(client.ContinueSearchServices(t.Context(), pages[2].Next, directory.IterationOptions{MaxItems: 2})) + services, err := collectServices(client.ContinueSearchServices(t.Context(), pages[2].Next, directory.IterationOptions{MaxItems: 2}).Items) if err != nil || len(services) != 2 { t.Fatalf("resumed services = %d, err = %v", len(services), err) } // A page budget above the old sixteen is honoured rather than refused. - many, err := collectPages(client.SearchPages(t.Context(), directory.SearchRequest{}, directory.IterationOptions{MaxPages: 40})) + many, err := collectPages(client.SearchServices(t.Context(), directory.SearchRequest{}, directory.IterationOptions{MaxPages: 40}).Responses) if err != nil || len(many) != 40 { t.Fatalf("pages = %d, err = %v", len(many), err) } - if _, err := collectPages(client.ContinueSearchPages(t.Context(), "https://evil.example/p", directory.IterationOptions{})); err == nil { + if _, err := collectPages(client.ContinueSearchServices(t.Context(), "https://evil.example/p", directory.IterationOptions{}).Responses); err == nil { t.Fatal("off-origin continuation resumed") } - if _, err := collectServices(client.ContinueSearchServices(t.Context(), "x", directory.IterationOptions{MaxItems: -1})); err == nil { + if _, err := collectServices(client.ContinueSearchServices(t.Context(), "x", directory.IterationOptions{MaxItems: -1}).Items); err == nil { t.Fatal("invalid item budget accepted") } - if _, err := collectPages(client.ContinueSearchPages(t.Context(), "x", directory.IterationOptions{MaxPages: -1})); err == nil { + if _, err := collectPages(client.ContinueSearchServices(t.Context(), "x", directory.IterationOptions{MaxPages: -1}).Responses); err == nil { t.Fatal("invalid page budget accepted") } } @@ -179,7 +179,7 @@ func TestItemBudgetOnAPageBoundaryDoesNotFetchAnotherPage(t *testing.T) { requests++ return response(http.StatusOK, `{"items":[`+record+`,`+strings.Replace(record, "compute.example", "storage.example", 1)+`],"next":"/v1/services/search?cursor=next"}`, nil), nil }) - services, err := collectServices(client.SearchServices(t.Context(), directory.SearchRequest{}, directory.IterationOptions{MaxItems: 2})) + services, err := collectServices(client.SearchServices(t.Context(), directory.SearchRequest{}, directory.IterationOptions{MaxItems: 2}).Items) if err != nil || len(services) != 2 { t.Fatalf("services = %d, err = %v", len(services), err) } @@ -204,7 +204,7 @@ func TestContinuationsMustAdvanceAndStayOnOrigin(t *testing.T) { client := serve(t, func(*http.Request) (*http.Response, error) { return response(http.StatusOK, `{"items":[`+record+`],"next":"`+test.next+`"}`, nil), nil }) - _, err := collectPages(client.SearchPages(t.Context(), directory.SearchRequest{}, directory.IterationOptions{})) + _, err := collectPages(client.SearchServices(t.Context(), directory.SearchRequest{}, directory.IterationOptions{}).Responses) if err == nil || !strings.Contains(err.Error(), test.wantErr) { t.Errorf("%s: error = %v, want %q", name, err, test.wantErr) } @@ -219,7 +219,7 @@ func TestContinuationsMustAdvanceAndStayOnOrigin(t *testing.T) { } return response(http.StatusOK, `{"items":[`+record+`]}`, nil), nil }) - pages, err := collectPages(client.SearchPages(t.Context(), directory.SearchRequest{}, directory.IterationOptions{})) + pages, err := collectPages(client.SearchServices(t.Context(), directory.SearchRequest{}, directory.IterationOptions{}).Responses) if err != nil || len(pages) != 2 { t.Fatalf("pages = %d, err = %v", len(pages), err) } @@ -272,7 +272,7 @@ func TestFailureMessagesAreSafeToLog(t *testing.T) { client := serve(t, func(*http.Request) (*http.Response, error) { return response(test.status, test.body, headers), nil }) - _, err := collectPages(client.SearchPages(t.Context(), directory.SearchRequest{}, directory.IterationOptions{})) + _, err := collectPages(client.SearchServices(t.Context(), directory.SearchRequest{}, directory.IterationOptions{}).Responses) var failure *directory.RequestError if !errors.As(err, &failure) { t.Errorf("%s: error = %v", name, err) @@ -290,7 +290,7 @@ func TestOverlongFailureDetailIsTruncated(t *testing.T) { client := serve(t, func(*http.Request) (*http.Response, error) { return response(http.StatusBadRequest, `{"detail":"`+strings.Repeat("x", 4000)+`"}`, headers), nil }) - _, err := collectPages(client.SearchPages(t.Context(), directory.SearchRequest{}, directory.IterationOptions{})) + _, err := collectPages(client.SearchServices(t.Context(), directory.SearchRequest{}, directory.IterationOptions{}).Responses) if err == nil || !strings.HasSuffix(err.Error(), "…") || len([]rune(err.Error())) > 2_100 { t.Fatalf("message length = %d", len([]rune(err.Error()))) } @@ -342,11 +342,11 @@ func TestFacetMembersAreCheckedByName(t *testing.T) { // encoding/json matches member names case-insensitively, so counting members is not the same // as knowing which members arrived. body := `{"items":[],"facets":{"payment_options":[{"count":1,"value":{"NAME":"mpp","OPTION":"base"}}]}}` - if _, err := collectPages(serve(t, always(body)).SearchPages(t.Context(), directory.SearchRequest{}, directory.IterationOptions{})); err == nil { + if _, err := collectPages(serve(t, always(body)).SearchServices(t.Context(), directory.SearchRequest{}, directory.IterationOptions{}).Responses); err == nil { t.Fatal("mis-cased payment option facet accepted") } valid := `{"items":[],"facets":{"payment_options":[{"count":1,"value":{"name":"mpp","option":"base"}}]}}` - pages, err := collectPages(serve(t, always(valid)).SearchPages(t.Context(), directory.SearchRequest{}, directory.IterationOptions{})) + pages, err := collectPages(serve(t, always(valid)).SearchServices(t.Context(), directory.SearchRequest{}, directory.IterationOptions{}).Responses) if err != nil || pages[0].Facets == nil || len(pages[0].Facets.PaymentOptions) != 1 { t.Fatalf("facets = %#v, %v", pages, err) } @@ -371,13 +371,13 @@ func TestFacetShapesAreValidated(t *testing.T) { "facet not an array": `{"items":[],"facets":{"keywords":{}}}`, } for name, body := range cases { - if _, err := collectPages(serve(t, always(body)).SearchPages(t.Context(), directory.SearchRequest{}, directory.IterationOptions{})); err == nil { + if _, err := collectPages(serve(t, always(body)).SearchServices(t.Context(), directory.SearchRequest{}, directory.IterationOptions{}).Responses); err == nil { t.Errorf("%s: accepted", name) } } // A count written with an exponent is still a whole number. body := `{"items":[],"facets":{"keywords":[{"count":1e3,"value":"gpu"}]}}` - pages, err := collectPages(serve(t, always(body)).SearchPages(t.Context(), directory.SearchRequest{}, directory.IterationOptions{})) + pages, err := collectPages(serve(t, always(body)).SearchServices(t.Context(), directory.SearchRequest{}, directory.IterationOptions{}).Responses) if err != nil || pages[0].Facets.Keywords[0].Count != 1000 { t.Fatalf("facets = %#v, %v", pages, err) } @@ -385,7 +385,7 @@ func TestFacetShapesAreValidated(t *testing.T) { `"enrollment":[{"count":1,"value":{"name":"aep"}}],` + `"operations":[{"count":1,"value":{"authentication":"not-required","name":"list-offerings"}}],` + `"payments":[{"count":1,"value":{"authentication":"required","name":"x402"}}]}}` - pages, err = collectPages(serve(t, always(full)).SearchPages(t.Context(), directory.SearchRequest{}, directory.IterationOptions{})) + pages, err = collectPages(serve(t, always(full)).SearchServices(t.Context(), directory.SearchRequest{}, directory.IterationOptions{}).Responses) if err != nil { t.Fatal(err) } @@ -414,7 +414,7 @@ func TestServiceRecordsAreValidatedMemberByMember(t *testing.T) { "not an object": `[]`, } for name, item := range cases { - pages, err := collectPages(serve(t, always(page(item))).SearchPages(t.Context(), directory.SearchRequest{}, directory.IterationOptions{})) + pages, err := collectPages(serve(t, always(page(item))).SearchServices(t.Context(), directory.SearchRequest{}, directory.IterationOptions{}).Responses) if err != nil { t.Errorf("%s: %v", name, err) continue @@ -426,7 +426,7 @@ func TestServiceRecordsAreValidatedMemberByMember(t *testing.T) { // An unrecognised protocol is filtered out rather than making the record unusable, so the // Service survives with the protocols this version understands. unknown := recordWith(`"protocols":{"trust":[{"name":"tap"}],"payments":[{"authentication":"required","name":"other"}]}`) - filtered, err := collectPages(serve(t, always(page(unknown))).SearchPages(t.Context(), directory.SearchRequest{}, directory.IterationOptions{})) + filtered, err := collectPages(serve(t, always(page(unknown))).SearchServices(t.Context(), directory.SearchRequest{}, directory.IterationOptions{}).Responses) if err != nil || len(filtered[0].Items) != 1 { t.Fatalf("unknown protocol = %#v, %v", filtered, err) } @@ -436,7 +436,7 @@ func TestServiceRecordsAreValidatedMemberByMember(t *testing.T) { // RFC 3339 permits a lowercase separator and zone designator. lowercase := strings.Replace(record, "2026-01-01T00:00:00Z", "2026-01-01t00:00:00z", 1) - pages, err := collectPages(serve(t, always(page(lowercase))).SearchPages(t.Context(), directory.SearchRequest{}, directory.IterationOptions{})) + pages, err := collectPages(serve(t, always(page(lowercase))).SearchServices(t.Context(), directory.SearchRequest{}, directory.IterationOptions{}).Responses) if err != nil || len(pages[0].Items) != 1 { t.Fatalf("lowercase date-time = %#v, %v", pages, err) } @@ -444,7 +444,7 @@ func TestServiceRecordsAreValidatedMemberByMember(t *testing.T) { complete := recordWith(`"documentation_url":"https://compute.example/docs","keywords":["gpu"],` + `"status_url":"https://compute.example/status","support_url":"https://compute.example/support",` + `"website_url":"https://compute.example","protocols":{"trust":[{"name":"tap"}]}`) - pages, err = collectPages(serve(t, always(page(complete))).SearchPages(t.Context(), directory.SearchRequest{}, directory.IterationOptions{})) + pages, err = collectPages(serve(t, always(page(complete))).SearchServices(t.Context(), directory.SearchRequest{}, directory.IterationOptions{}).Responses) if err != nil || len(pages[0].Items) != 1 { t.Fatalf("complete record = %#v, %v", pages, err) } @@ -472,12 +472,12 @@ func TestPageEnvelopeIsValidated(t *testing.T) { "next empty": `{"items":[],"next":""}`, } for name, body := range cases { - if _, err := collectPages(serve(t, always(body)).SearchPages(t.Context(), directory.SearchRequest{}, directory.IterationOptions{})); err == nil { + if _, err := collectPages(serve(t, always(body)).SearchServices(t.Context(), directory.SearchRequest{}, directory.IterationOptions{}).Responses); err == nil { t.Errorf("%s: accepted", name) } } // A final page is commonly spelled with an explicit null continuation. - pages, err := collectPages(serve(t, always(`{"items":[],"next":null,"total":7}`)).SearchPages(t.Context(), directory.SearchRequest{}, directory.IterationOptions{})) + pages, err := collectPages(serve(t, always(`{"items":[],"next":null,"total":7}`)).SearchServices(t.Context(), directory.SearchRequest{}, directory.IterationOptions{}).Responses) if err != nil || len(pages) != 1 || pages[0].Next != "" { t.Fatalf("pages = %#v, %v", pages, err) } @@ -502,15 +502,15 @@ func TestSearchRequestFiltersAreValidated(t *testing.T) { // Every distinct operation-and-authentication pair is expressible, so the bound is on the pair // rather than on the seven operation names. accepted := directory.SearchRequest{Filters: &directory.ServiceFilters{Operations: operationFilters}} - if _, err := collectPages(serve(t, always(`{"items":[]}`)).SearchPages(t.Context(), accepted, directory.IterationOptions{})); err != nil { + if _, err := collectPages(serve(t, always(`{"items":[]}`)).SearchServices(t.Context(), accepted, directory.IterationOptions{}).Responses); err != nil { t.Fatalf("21 operation filters rejected: %v", err) } payments := make([]directory.PaymentFilter, 0, 3) for _, option := range []odp.PaymentOption{"base", "solana", "ethereum"} { payments = append(payments, directory.PaymentFilter{Authentication: odp.AuthenticationRequired, Name: odp.ProtocolMPP, Options: []odp.PaymentOption{option}}) } - if _, err := collectPages(serve(t, always(`{"items":[]}`)).SearchPages(t.Context(), - directory.SearchRequest{Filters: &directory.ServiceFilters{Payments: payments}}, directory.IterationOptions{})); err != nil { + if _, err := collectPages(serve(t, always(`{"items":[]}`)).SearchServices(t.Context(), + directory.SearchRequest{Filters: &directory.ServiceFilters{Payments: payments}}, directory.IterationOptions{}).Responses); err != nil { t.Fatalf("three payment filters rejected: %v", err) } @@ -539,12 +539,12 @@ func TestSearchRequestFiltersAreValidated(t *testing.T) { } for name, filters := range refused { request := directory.SearchRequest{Filters: &filters} - if _, err := collectPages(serve(t, always(`{"items":[]}`)).SearchPages(t.Context(), request, directory.IterationOptions{})); err == nil { + if _, err := collectPages(serve(t, always(`{"items":[]}`)).SearchServices(t.Context(), request, directory.IterationOptions{}).Responses); err == nil { t.Errorf("%s: accepted", name) } } - if _, err := collectPages(serve(t, always(`{"items":[]}`)).SearchPages(t.Context(), - directory.SearchRequest{Query: strings.Repeat("q", 513)}, directory.IterationOptions{})); err == nil { + if _, err := collectPages(serve(t, always(`{"items":[]}`)).SearchServices(t.Context(), + directory.SearchRequest{Query: strings.Repeat("q", 513)}, directory.IterationOptions{}).Responses); err == nil { t.Error("overlong query accepted") } } @@ -552,7 +552,7 @@ func TestSearchRequestFiltersAreValidated(t *testing.T) { func TestRequestValidationReportsTheRequestBeforeTheBudget(t *testing.T) { // A caller who fixes the budget should not then discover a second complaint about the body. request := directory.SearchRequest{Limit: 9999} - _, err := collectPages(serve(t, always(`{"items":[]}`)).SearchPages(t.Context(), request, directory.IterationOptions{MaxPages: 99_999})) + _, err := collectPages(serve(t, always(`{"items":[]}`)).SearchServices(t.Context(), request, directory.IterationOptions{MaxPages: 99_999}).Responses) if err == nil || !strings.Contains(err.Error(), "limit must be an integer from 1 through 100") { t.Fatalf("error = %v", err) } diff --git a/directory/search.go b/directory/search.go new file mode 100644 index 0000000..968a272 --- /dev/null +++ b/directory/search.go @@ -0,0 +1,188 @@ +package directory + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "iter" + "net/http" + "strings" + "time" + "unicode/utf8" + + odp "github.com/offering-protocol/odp-go" +) + +func (client *Client) Search(ctx context.Context, request DirectorySearchRequest, options IterationOptions) SearchSequence[Result] { + body, validationError := validateDirectorySearchRequest(request) + maxPages, budgetError := pageBudget(options.MaxPages) + if validationError == nil { + validationError = budgetError + } + responses := func(yield func(SearchResponse[Result], error) bool) { + if validationError != nil { + yield(SearchResponse[Result]{}, validationError) + return + } + traverse(client, ctx, client.originURL.JoinPath("v1", "directory", "search"), http.MethodPost, body, maxPages, parseDirectorySearchPage, yield) + } + return sequence(responses, options) +} + +func (client *Client) ContinueSearch(ctx context.Context, next string, options IterationOptions) SearchSequence[Result] { + maxPages, validationError := pageBudget(options.MaxPages) + responses := func(yield func(SearchResponse[Result], error) bool) { + if validationError != nil { + yield(SearchResponse[Result]{}, validationError) + return + } + target, err := client.continuationURL(next) + if err != nil { + yield(SearchResponse[Result]{}, err) + return + } + traverse(client, ctx, target, http.MethodGet, nil, maxPages, parseDirectorySearchPage, yield) + } + return sequence(responses, options) +} + +func sequence[Item any](responses iter.Seq2[SearchResponse[Item], error], options IterationOptions) SearchSequence[Item] { + items := func(yield func(Item, error) bool) { + var zero Item + if options.MaxItems < 0 || options.MaxItems > maximumItems { + yield(zero, fmt.Errorf("maxItems must be an integer from 1 through %d", maximumItems)) + return + } + count := 0 + for page, err := range responses { + if err != nil { + yield(zero, err) + return + } + for _, issue := range page.Issues { + if options.OnIssue != nil { + options.OnIssue(issue) + } + } + for _, item := range page.Items { + count++ + if !yield(item, nil) || (options.MaxItems != 0 && count >= options.MaxItems) { + return + } + } + } + } + return SearchSequence[Item]{Items: items, Responses: responses} +} + +func validateDirectorySearchRequest(request DirectorySearchRequest) ([]byte, error) { + base, err := normalizedSearchRequest(request.SearchRequest) + if err != nil { + return nil, err + } + if request.Types != nil && (len(request.Types) == 0 || len(request.Types) > 2 || !unique(request.Types)) { + return nil, errors.New("types must contain distinct service or collection values") + } + for _, value := range request.Types { + if value != "service" && value != "collection" { + return nil, errors.New("types must contain distinct service or collection values") + } + } + return json.Marshal(DirectorySearchRequest{SearchRequest: base, Types: request.Types}) +} + +func parseDirectorySearchPage(data []byte) (SearchResponse[Result], error) { + return parsePage(data, parseResult, IssueResult) +} + +func parseResult(data []byte) (Result, error) { + var object map[string]json.RawMessage + if err := json.Unmarshal(data, &object); err != nil { + return Result{}, err + } + kind, err := requiredText(object["type"], "type", 1, 128) + if err != nil { + return Result{}, err + } + if kind != "service" && kind != "collection" { + return Result{Type: kind, Raw: append(json.RawMessage(nil), data...)}, nil + } + service, err := parseService(object["service"]) + if err != nil { + return Result{}, err + } + serviceID, err := requiredText(service.Additional["service_id"], "service_id", 1, 128) + if err != nil { + return Result{}, err + } + delete(service.Additional, "service_id") + stamp, err := requiredText(object["indexed_at"], "indexed_at", 1, 64) + if err != nil { + return Result{}, err + } + indexedAt, err := time.Parse(time.RFC3339Nano, strings.ToUpper(stamp)) + if err != nil { + return Result{}, errors.New("indexed_at must be a date-time") + } + result := Result{Type: kind, Service: &IndexedService{Service: service, ServiceID: serviceID}, IndexedAt: indexedAt} + if kind == "service" { + result.Additional = cloneAdditional(object, "type", "service", "indexed_at", "available_through") + if raw, present := object["available_through"]; present { + reference, err := parseServiceReference(raw) + if err != nil { + return Result{}, err + } + result.AvailableThrough = &reference + } + return result, nil + } + var collection map[string]json.RawMessage + if err := json.Unmarshal(object["collection"], &collection); err != nil { + return Result{}, err + } + id, err := requiredText(collection["id"], "collection.id", 1, 128) + if err != nil || !odp.IsLocalResourceIdentifier(id) { + return Result{}, errors.New("collection.id must be a local resource identifier") + } + name, err := requiredText(collection["name"], "collection.name", 1, 128) + if err != nil { + return Result{}, err + } + var description string + if raw, present := collection["description"]; present { + if string(raw) == "null" || json.Unmarshal(raw, &description) != nil || utf8.RuneCountInString(description) > 1024 { + return Result{}, errors.New("collection.description is invalid") + } + } + result.Collection = &CollectionSummary{ID: id, Name: name, Description: description, Additional: cloneAdditional(collection, "id", "name", "description")} + result.Additional = cloneAdditional(object, "type", "service", "indexed_at", "collection") + return result, nil +} + +func parseServiceReference(data []byte) (ServiceReference, error) { + var object map[string]json.RawMessage + if err := json.Unmarshal(data, &object); err != nil { + return ServiceReference{}, err + } + id, err := requiredText(object["service_id"], "service_id", 1, 128) + if err != nil { + return ServiceReference{}, err + } + origin, err := requiredText(object["service_origin"], "service_origin", 1, 2048) + if err != nil { + return ServiceReference{}, err + } + canonical, err := odp.DeriveServiceOrigin(origin) + if err != nil || canonical != origin || !publicHTTPSOrigin(origin) { + return ServiceReference{}, errors.New("Attribution origin must be a canonical public HTTPS origin") + } + var name string + if raw, present := object["name"]; present { + name, err = requiredText(raw, "name", 1, 128) + if err != nil { + return ServiceReference{}, err + } + } + return ServiceReference{ServiceID: id, ServiceOrigin: origin, Name: name, Additional: cloneAdditional(object, "service_id", "service_origin", "name")}, nil +} diff --git a/directory/search_test.go b/directory/search_test.go new file mode 100644 index 0000000..05a05c1 --- /dev/null +++ b/directory/search_test.go @@ -0,0 +1,299 @@ +package directory_test + +import ( + "context" + "encoding/json" + "errors" + "io" + "net/http" + "strings" + "testing" + + "github.com/offering-protocol/odp-go/directory" +) + +func mixedResult(kind string) map[string]any { + var service map[string]any + if err := json.Unmarshal([]byte(serviceResult), &service); err != nil { + panic(err) + } + service["service_id"] = "ca0304cc-ab28-43e5-af94-7bdf11b40c6e" + result := map[string]any{"type": kind, "service": service, "indexed_at": "2026-09-18T12:00:00Z"} + if kind == "collection" { + result["collection"] = map[string]any{"id": "Weather", "name": "Weather forecasts", "description": "Forecasts and conditions."} + } + return result +} + +func mixedBody(t *testing.T, items ...any) string { + t.Helper() + encoded, err := json.Marshal(map[string]any{"items": items}) + if err != nil { + t.Fatal(err) + } + return string(encoded) +} + +func TestMixedSearchSequence(t *testing.T) { + service := mixedResult("service") + service["available_through"] = map[string]any{"service_id": "platform", "service_origin": "https://platform.example", "name": "Platform", "extra": true} + service["extra"] = "retained" + collection := mixedResult("collection") + collection["collection"].(map[string]any)["extra"] = true + unknown := map[string]any{"type": "future", "nested": map[string]any{"untouched": true}} + var calls int + value := client(t, func(request *http.Request) (*http.Response, error) { + calls++ + if request.Method != "POST" || request.URL.String() != "https://sandbox.inflowpay.ai/v1/directory/search" { + t.Fatalf("request = %s %s", request.Method, request.URL) + } + body, err := io.ReadAll(request.Body) + if err != nil { + t.Fatal(err) + } + if string(body) != `{"filters":{"keywords":["weather"]},"limit":10,"query":"forecast","types":["service","collection"]}` { + t.Fatalf("body = %s", body) + } + return response(200, mixedBody(t, service, collection, unknown, false), nil), nil + }, directory.Sandbox) + types := []string{"service", "collection"} + keywords := []string{"weather"} + issues := []directory.Issue{} + search := value.Search(t.Context(), directory.DirectorySearchRequest{ + SearchRequest: directory.SearchRequest{Query: "forecast", Limit: 10, Filters: &directory.ServiceFilters{Keywords: keywords}}, Types: types, + }, directory.IterationOptions{MaxItems: 2, OnIssue: func(issue directory.Issue) { issues = append(issues, issue) }}) + types[0], keywords[0] = "mutated", "mutated" + if calls != 0 { + t.Fatal("search was not lazy") + } + for result, err := range search.Responses { + if err != nil || len(result.Items) != 3 || len(result.Issues) != 1 || result.Issues[0].Index != 3 || result.Issues[0].Scope != directory.IssueResult { + t.Fatalf("result = %#v, %v", result, err) + } + if result.Items[0].AvailableThrough.Name != "Platform" || string(result.Items[0].Additional["extra"]) != `"retained"` || string(result.Items[0].AvailableThrough.Additional["extra"]) != "true" { + t.Fatalf("Service attribution or additional fields = %#v", result.Items[0]) + } + if result.Items[1].Collection.ID != "Weather" || result.Items[1].Service.ServiceID == "" || result.Items[1].IndexedAt.Equal(result.Items[1].Service.IndexedAt) || string(result.Items[1].Collection.Additional["extra"]) != "true" { + t.Fatalf("Collection = %#v", result.Items[1]) + } + if result.Items[2].Type != "future" || result.Items[2].Service != nil || !strings.Contains(string(result.Items[2].Raw), `"untouched":true`) { + t.Fatalf("unknown = %#v", result.Items[2]) + } + } + count := 0 + for _, err := range search.Items { + if err != nil { + t.Fatal(err) + } + count++ + } + if calls != 2 || count != 2 || len(issues) != 1 { + t.Fatalf("calls=%d, count=%d, issues=%v", calls, count, issues) + } +} + +func TestMixedSearchRejectsMalformedKnownEntries(t *testing.T) { + mutations := []func(map[string]any){ + func(v map[string]any) { v["type"] = "" }, + func(v map[string]any) { v["service"] = nil }, + func(v map[string]any) { delete(v["service"].(map[string]any), "service_id") }, + func(v map[string]any) { v["indexed_at"] = "yesterday" }, + func(v map[string]any) { delete(v, "indexed_at") }, + func(v map[string]any) { v["collection"] = false }, + func(v map[string]any) { v["collection"] = nil }, + func(v map[string]any) { v["collection"].(map[string]any)["id"] = "../outside" }, + func(v map[string]any) { v["collection"].(map[string]any)["name"] = "" }, + func(v map[string]any) { v["collection"].(map[string]any)["description"] = nil }, + func(v map[string]any) { v["collection"].(map[string]any)["description"] = 1 }, + func(v map[string]any) { v["collection"].(map[string]any)["description"] = strings.Repeat("x", 1025) }, + } + for index, mutate := range mutations { + item := mixedResult("collection") + mutate(item) + value := client(t, func(*http.Request) (*http.Response, error) { + return response(200, mixedBody(t, item, mixedResult("collection")), nil), nil + }, directory.Production) + for result, err := range value.Search(t.Context(), directory.DirectorySearchRequest{}, directory.IterationOptions{}).Responses { + if err != nil || len(result.Items) != 1 || len(result.Issues) != 1 || result.Issues[0].Index != 0 { + t.Fatalf("case %d: %#v, %v", index, result, err) + } + } + } + for _, reference := range []any{nil, false, map[string]any{}, map[string]any{"service_id": "x"}, map[string]any{"service_id": "x", "service_origin": "http://localhost"}, map[string]any{"service_id": "x", "service_origin": "https://platform.example", "name": nil}} { + item := mixedResult("service") + item["available_through"] = reference + value := client(t, func(*http.Request) (*http.Response, error) { + return response(200, mixedBody(t, item), nil), nil + }, directory.Production) + for result, err := range value.Search(t.Context(), directory.DirectorySearchRequest{}, directory.IterationOptions{}).Responses { + if err != nil || len(result.Items) != 0 || len(result.Issues) != 1 { + t.Fatalf("reference %v: %#v, %v", reference, result, err) + } + } + } +} + +func TestMixedSearchOptionalFields(t *testing.T) { + for _, description := range []any{"", nil} { + item := mixedResult("collection") + if description == nil { + delete(item["collection"].(map[string]any), "description") + } else { + item["collection"].(map[string]any)["description"] = description + } + service := mixedResult("service") + service["available_through"] = map[string]any{"service_id": "x", "service_origin": "https://platform.example"} + value := client(t, func(*http.Request) (*http.Response, error) { + return response(200, mixedBody(t, item, service), nil), nil + }, directory.Production) + for result, err := range value.Search(t.Context(), directory.DirectorySearchRequest{}, directory.IterationOptions{}).Responses { + if err != nil || len(result.Items) != 2 || len(result.Issues) != 0 { + t.Fatalf("result = %#v, %v", result, err) + } + } + } +} + +func TestMixedSearchBoundsAndContinuation(t *testing.T) { + var methods []string + value := client(t, func(request *http.Request) (*http.Response, error) { + methods = append(methods, request.Method) + return response(200, `{"items":[],"next":"/v1/directory/search?cursor=opaque","facets":{"keywords":[{"value":"weather","count":12}]}}`, nil), nil + }, directory.Production) + search := value.Search(t.Context(), directory.DirectorySearchRequest{}, directory.IterationOptions{MaxPages: 1}) + var next string + for result, err := range search.Responses { + if err != nil || result.Facets.Keywords[0].Count != 12 { + t.Fatalf("result=%#v, err=%v", result, err) + } + next = result.Next + } + for _, err := range value.ContinueSearch(t.Context(), next, directory.IterationOptions{MaxPages: 1}).Responses { + if err != nil { + t.Fatal(err) + } + } + if strings.Join(methods, ",") != "POST,GET" { + t.Fatalf("methods=%v", methods) + } + foundLoop := false + for _, err := range value.ContinueSearch(t.Context(), next, directory.IterationOptions{}).Items { + if err != nil { + foundLoop = strings.Contains(err.Error(), "loop") + } + } + if !foundLoop { + t.Fatal("loop was not reported") + } + before := len(methods) + for _, err := range value.ContinueSearch(t.Context(), "https://elsewhere.example", directory.IterationOptions{}).Items { + if err == nil { + t.Fatal("cross-origin continuation accepted") + } + } + if len(methods) != before { + t.Fatal("cross-origin request sent") + } +} + +func TestMixedSearchValidationBeforeTransport(t *testing.T) { + value := client(t, func(*http.Request) (*http.Response, error) { + t.Fatal("invalid request reached transport") + return nil, nil + }, directory.Production) + for _, request := range []directory.DirectorySearchRequest{ + {Types: []string{}}, {Types: []string{"service", "service"}}, {Types: []string{"future"}}, + {Types: []string{"service", "collection", "service"}}, {SearchRequest: directory.SearchRequest{Limit: -1}}, + } { + for _, err := range value.Search(t.Context(), request, directory.IterationOptions{}).Items { + if err == nil { + t.Fatal("invalid search accepted") + } + } + } + for _, options := range []directory.IterationOptions{{MaxItems: -1}, {MaxItems: 10_001}, {MaxPages: -1}} { + for _, search := range []directory.SearchSequence[directory.Result]{value.Search(t.Context(), directory.DirectorySearchRequest{}, options), value.ContinueSearch(t.Context(), "/next", options)} { + for _, err := range search.Items { + if err == nil { + t.Fatal("invalid options accepted") + } + } + } + } +} + +func TestMixedSearchTransportAndCancellation(t *testing.T) { + for _, body := range []string{`{}`, `{"items":null}`, `{"items":{}}`, `{"items":[],"facets":false}`} { + value := client(t, func(*http.Request) (*http.Response, error) { + return response(200, body, nil), nil + }, directory.Production) + for _, err := range value.Search(t.Context(), directory.DirectorySearchRequest{}, directory.IterationOptions{}).Items { + if err == nil { + t.Fatal("invalid envelope accepted") + } + } + } + ctx, cancel := context.WithCancel(t.Context()) + cancel() + value := client(t, func(request *http.Request) (*http.Response, error) { + return nil, request.Context().Err() + }, directory.Production) + for _, err := range value.Search(ctx, directory.DirectorySearchRequest{}, directory.IterationOptions{}).Items { + if !errors.Is(err, context.Canceled) { + t.Fatalf("cancellation = %v", err) + } + } +} + +func TestMixedSuggestionsFilters(t *testing.T) { + requests := 0 + value := client(t, func(request *http.Request) (*http.Response, error) { + requests++ + var payload directory.SuggestionRequest + if err := json.NewDecoder(request.Body).Decode(&payload); err != nil { + t.Fatal(err) + } + if request.Method != "POST" || request.Header.Get("Content-Type") != "application/json" || payload.Filters == nil || strings.Join(payload.Filters.Keywords, ",") != "weather" { + t.Fatalf("request=%s payload=%+v", request.Method, payload) + } + return response(200, `{"items":["Weather"]}`, nil), nil + }, directory.Production) + filters := &directory.ServiceFilters{Keywords: []string{"weather"}} + if _, err := value.Suggest(t.Context(), directory.SuggestionRequest{Prefix: "we", Filters: filters}); err != nil { + t.Fatal(err) + } + if _, err := value.Suggest(t.Context(), directory.SuggestionRequest{Prefix: "we", Filters: &directory.ServiceFilters{Keywords: []string{""}}}); err == nil { + t.Fatal("invalid filters accepted") + } + if _, err := value.SuggestServices(t.Context(), directory.SuggestionRequest{Prefix: "we", Filters: filters}); err == nil { + t.Fatal("Service-only filters accepted") + } + if requests != 1 || filters.Keywords[0] != "weather" { + t.Fatalf("requests=%d filters=%+v", requests, filters) + } +} + +func TestMixedSuggestionsUseSeparateEndpoint(t *testing.T) { + var paths []string + value := client(t, func(request *http.Request) (*http.Response, error) { + paths = append(paths, request.URL.Path) + if request.URL.Path == "/v1/directory/suggestions" { + var payload directory.SuggestionRequest + if err := json.NewDecoder(request.Body).Decode(&payload); err != nil || payload.Prefix != "we" || payload.Limit != 10 || request.Method != "POST" { + t.Fatalf("request=%s %s payload=%+v err=%v", request.Method, request.URL, payload, err) + } + } else if request.Method != "GET" || request.URL.Query().Get("prefix") != "we" || request.URL.Query().Get("limit") != "10" { + t.Fatalf("request=%s %s", request.Method, request.URL) + } + return response(200, `{"items":["AccuWeather","Atlas","Atlas"]}`, nil), nil + }, directory.Production) + for _, suggest := range []func(context.Context, directory.SuggestionRequest) ([]string, error){value.Suggest, value.SuggestServices} { + items, err := suggest(t.Context(), directory.SuggestionRequest{Prefix: "we", Limit: 10}) + if err != nil || strings.Join(items, ",") != "AccuWeather,Atlas" { + t.Fatalf("items=%v, err=%v", items, err) + } + } + if strings.Join(paths, ",") != "/v1/directory/suggestions,/v1/services/suggestions" { + t.Fatalf("paths=%v", paths) + } +} diff --git a/directory/shape_test.go b/directory/shape_test.go index cd846e2..1ff44ad 100644 --- a/directory/shape_test.go +++ b/directory/shape_test.go @@ -21,7 +21,7 @@ func TestUnreadableBodiesSurfaceTheirError(t *testing.T) { headers.Set("Content-Type", "application/json") return &http.Response{Body: failingBody{}, Header: headers, StatusCode: http.StatusOK}, nil }) - _, err := collectPages(client.SearchPages(t.Context(), directory.SearchRequest{}, directory.IterationOptions{})) + _, err := collectPages(client.SearchServices(t.Context(), directory.SearchRequest{}, directory.IterationOptions{}).Responses) if err == nil || !strings.Contains(err.Error(), "connection reset") { t.Fatalf("error = %v", err) } @@ -33,7 +33,7 @@ func TestFailureMessagesStayWithinTheirBudget(t *testing.T) { return withHeaders(http.StatusBadRequest, `{"detail":"`+strings.Repeat("d", 4_096)+`"}`, map[string]string{"Content-Type": "application/problem+json"}), nil }) - _, err := collectPages(long.SearchPages(t.Context(), directory.SearchRequest{}, directory.IterationOptions{})) + _, err := collectPages(long.SearchServices(t.Context(), directory.SearchRequest{}, directory.IterationOptions{}).Responses) if err == nil { t.Fatal("oversized detail accepted") } @@ -51,7 +51,7 @@ func TestFailureMessagesStayWithinTheirBudget(t *testing.T) { return withHeaders(http.StatusServiceUnavailable, `{"detail":"`+strings.Repeat("h", 32_768)+`"}`, map[string]string{"Content-Type": "application/json"}), nil }) - _, err = collectPages(huge.SearchPages(t.Context(), directory.SearchRequest{}, directory.IterationOptions{})) + _, err = collectPages(huge.SearchServices(t.Context(), directory.SearchRequest{}, directory.IterationOptions{}).Responses) if err == nil || err.Error() != "Directory request failed with HTTP 503" { t.Fatalf("message = %v", err) } @@ -79,7 +79,7 @@ func TestContinuationsAreCheckedBeforeTheyAreFollowed(t *testing.T) { } return response(http.StatusOK, `{"items":[],"next":"`+test.next+`"}`, nil), nil }) - _, err := collectPages(client.SearchPages(t.Context(), directory.SearchRequest{}, directory.IterationOptions{})) + _, err := collectPages(client.SearchServices(t.Context(), directory.SearchRequest{}, directory.IterationOptions{}).Responses) switch { case test.wantErr == "" && err != nil: t.Errorf("%s: %v", name, err) @@ -99,7 +99,7 @@ func TestOptionalServiceMembersAreCheckedForTheirType(t *testing.T) { "website_url": `"website_url":7`, } { client := serve(t, always(page(recordWith(members)))) - pages, err := collectPages(client.SearchPages(t.Context(), directory.SearchRequest{}, directory.IterationOptions{})) + pages, err := collectPages(client.SearchServices(t.Context(), directory.SearchRequest{}, directory.IterationOptions{}).Responses) if err != nil { t.Fatalf("%s: %v", name, err) } @@ -126,7 +126,7 @@ func TestFacetShapesAreRejectedMemberByMember(t *testing.T) { "an option naming no scheme": `"payment_options":[{"value":{"name":"tap","option":"inflow"},"count":1}]`, } { client := serve(t, always(`{"items":[],"facets":{`+facets+`}}`)) - if _, err := collectPages(client.SearchPages(t.Context(), directory.SearchRequest{}, directory.IterationOptions{})); err == nil { + if _, err := collectPages(client.SearchServices(t.Context(), directory.SearchRequest{}, directory.IterationOptions{}).Responses); err == nil { t.Errorf("%s accepted", name) } } diff --git a/directory/transport_test.go b/directory/transport_test.go index edebb55..74789c5 100644 --- a/directory/transport_test.go +++ b/directory/transport_test.go @@ -57,7 +57,7 @@ func TestResponsesAreRejectedOnTheirWireShape(t *testing.T) { } for name, test := range cases { client := serve(t, func(*http.Request) (*http.Response, error) { return test.build(), nil }) - _, err := collectPages(client.SearchPages(t.Context(), directory.SearchRequest{}, directory.IterationOptions{})) + _, err := collectPages(client.SearchServices(t.Context(), directory.SearchRequest{}, directory.IterationOptions{}).Responses) if err == nil || !strings.Contains(err.Error(), test.wantErr) { t.Errorf("%s: error = %v, want %q", name, err, test.wantErr) } @@ -89,7 +89,7 @@ func TestRedirectsRewriteTheMethodTheirStatusRequires(t *testing.T) { } return response(http.StatusOK, `{"items":[]}`, nil), nil }) - if _, err := collectPages(client.SearchPages(t.Context(), directory.SearchRequest{Query: "gpu"}, directory.IterationOptions{})); err != nil { + if _, err := collectPages(client.SearchServices(t.Context(), directory.SearchRequest{Query: "gpu"}, directory.IterationOptions{}).Responses); err != nil { t.Fatalf("status %d: %v", test.status, err) } if followed != test.want { @@ -121,7 +121,7 @@ func TestRedirectsAreBoundedAndOriginLocked(t *testing.T) { } return withHeaders(http.StatusFound, "", values), nil }) - _, err := collectPages(client.SearchPages(t.Context(), directory.SearchRequest{}, directory.IterationOptions{})) + _, err := collectPages(client.SearchServices(t.Context(), directory.SearchRequest{}, directory.IterationOptions{}).Responses) if err == nil || !strings.Contains(err.Error(), test.wantErr) { t.Errorf("%s: error = %v, want %q", name, err, test.wantErr) } @@ -135,7 +135,7 @@ func TestRedirectsAreBoundedAndOriginLocked(t *testing.T) { } return response(http.StatusOK, `{"items":[]}`, nil), nil }) - if _, err := collectPages(client.SearchPages(t.Context(), directory.SearchRequest{}, directory.IterationOptions{})); err != nil { + if _, err := collectPages(client.SearchServices(t.Context(), directory.SearchRequest{}, directory.IterationOptions{}).Responses); err != nil { t.Fatalf("five redirects: %v", err) } if hops != 6 { @@ -151,7 +151,7 @@ func TestRunawayTraversalIsReportedRatherThanTruncated(t *testing.T) { }) // A directory that never stops offering a continuation is a runaway, not an exhausted search, // so the guard reports rather than ending the sequence as though the results ran out. - _, err := collectPages(client.SearchPages(t.Context(), directory.SearchRequest{}, directory.IterationOptions{MaxPages: 10_000})) + _, err := collectPages(client.SearchServices(t.Context(), directory.SearchRequest{}, directory.IterationOptions{MaxPages: 10_000}).Responses) if err == nil || !strings.Contains(err.Error(), "traversal limit") { t.Fatalf("error = %v after %d requests", err, requests) } @@ -166,7 +166,7 @@ func TestCallerPageBudgetEndsTheSequenceQuietly(t *testing.T) { requests++ return response(http.StatusOK, fmt.Sprintf(`{"items":[],"next":"/v1/services/search?cursor=%d"}`, requests), nil), nil }) - pages, err := collectPages(client.SearchPages(t.Context(), directory.SearchRequest{}, directory.IterationOptions{MaxPages: 2})) + pages, err := collectPages(client.SearchServices(t.Context(), directory.SearchRequest{}, directory.IterationOptions{MaxPages: 2}).Responses) if err != nil || len(pages) != 2 || requests != 2 { t.Fatalf("pages = %d, requests = %d, err = %v", len(pages), requests, err) } @@ -179,7 +179,7 @@ func TestCallersCanStopEarly(t *testing.T) { return response(http.StatusOK, `{"items":[`+record+`],"next":"/v1/services/search?cursor=c"}`, nil), nil }) count := 0 - for range client.SearchPages(t.Context(), directory.SearchRequest{}, directory.IterationOptions{}) { + for range client.SearchServices(t.Context(), directory.SearchRequest{}, directory.IterationOptions{}).Responses { count++ break } @@ -187,7 +187,7 @@ func TestCallersCanStopEarly(t *testing.T) { t.Fatalf("pages = %d, requests = %d", count, requests) } services := 0 - for range client.SearchServices(t.Context(), directory.SearchRequest{}, directory.IterationOptions{}) { + for range client.SearchServices(t.Context(), directory.SearchRequest{}, directory.IterationOptions{}).Items { services++ break } diff --git a/directory/types.go b/directory/types.go index b9bcb06..606f983 100644 --- a/directory/types.go +++ b/directory/types.go @@ -1,8 +1,9 @@ -// Package directory provides the canonical ODP Service directory client. +// Package directory searches indexed Services and Collections in the canonical Directory. package directory import ( "encoding/json" + "iter" "net/http" "net/url" "time" @@ -52,6 +53,41 @@ type SearchRequest struct { Query string `json:"query,omitempty"` } +type DirectorySearchRequest struct { + SearchRequest + Types []string `json:"types,omitempty"` +} + +type IndexedService struct { + Service + ServiceID string +} + +type ServiceReference struct { + Additional odp.AdditionalMembers + ServiceID string + ServiceOrigin string + Name string +} + +type CollectionSummary struct { + Additional odp.AdditionalMembers + ID string + Name string + Description string +} + +// Result retains unknown resource types in Raw without interpreting them as Services. +type Result struct { + Additional odp.AdditionalMembers + Type string + Service *IndexedService + Collection *CollectionSummary + AvailableThrough *ServiceReference + IndexedAt time.Time + Raw json.RawMessage +} + type IterationOptions struct { MaxItems int MaxPages int @@ -96,20 +132,27 @@ type PaymentOptionFacetValue struct { Option odp.PaymentOption `json:"option"` } -type SearchPage struct { +type SearchResponse[Item any] struct { Additional odp.AdditionalMembers Facets *Facets // Issues reports records this page carried that could not be used. They are reported rather // than raised because a directory is not an ODP protocol role: one unusable record says // nothing about the rest of the page. Issues []Issue - Items []Service + Items []Item Next string } +// SearchSequence exposes independent, lazy traversals of the same search. +type SearchSequence[Item any] struct { + Items iter.Seq2[Item, error] + Responses iter.Seq2[SearchResponse[Item], error] +} + type IssueScope string const IssueService IssueScope = "service" +const IssueResult IssueScope = "result" // Issue describes one Directory record this client discarded, and why. type Issue struct { @@ -119,8 +162,9 @@ type Issue struct { } type SuggestionRequest struct { - Limit int - Prefix string + Filters *ServiceFilters `json:"filters,omitempty"` + Limit int `json:"limit,omitempty"` + Prefix string `json:"prefix"` } type RequestError struct { diff --git a/directory/validation.go b/directory/validation.go index 6d5609f..a065e68 100644 --- a/directory/validation.go +++ b/directory/validation.go @@ -37,26 +37,34 @@ var operations = []odp.Operation{ } func validateSearchRequest(request SearchRequest) ([]byte, error) { + validated, err := normalizedSearchRequest(request) + if err != nil { + return nil, err + } + return json.Marshal(validated) +} + +func normalizedSearchRequest(request SearchRequest) (SearchRequest, error) { validated := SearchRequest{} if request.Query != "" { query, err := requireText(request.Query, "query", 1, 512) if err != nil { - return nil, err + return SearchRequest{}, err } validated.Query = query } if request.Limit < 0 || request.Limit > 100 { - return nil, errors.New("limit must be an integer from 1 through 100") + return SearchRequest{}, errors.New("limit must be an integer from 1 through 100") } validated.Limit = request.Limit if request.Filters != nil { filters, err := validateFilters(*request.Filters) if err != nil { - return nil, err + return SearchRequest{}, err } validated.Filters = &filters } - return json.Marshal(validated) + return validated, nil } func validateFilters(filters ServiceFilters) (ServiceFilters, error) { @@ -83,42 +91,46 @@ func validateFilters(filters ServiceFilters) (ServiceFilters, error) { return ServiceFilters{Enrollment: enrollment, Keywords: keywords, Operations: operationFilters, Payments: paymentFilters, Trust: trust}, nil } -func parseSearchPage(data []byte) (SearchPage, error) { +func parseSearchPage(data []byte) (SearchResponse[Service], error) { + return parsePage(data, parseService, IssueService) +} + +func parsePage[Item any](data []byte, parse func([]byte) (Item, error), scope IssueScope) (SearchResponse[Item], error) { var object map[string]json.RawMessage if err := json.Unmarshal(data, &object); err != nil { - return SearchPage{}, err + return SearchResponse[Item]{}, err } itemsData, present := object["items"] if !present || string(itemsData) == "null" { - return SearchPage{}, errors.New("Directory search page items are invalid") + return SearchResponse[Item]{}, errors.New("Directory search page items are invalid") } var itemValues []json.RawMessage if err := json.Unmarshal(itemsData, &itemValues); err != nil || len(itemValues) > 100 { - return SearchPage{}, errors.New("Directory search page items are invalid") + return SearchResponse[Item]{}, errors.New("Directory search page items are invalid") } - items := make([]Service, 0, len(itemValues)) + items := make([]Item, 0, len(itemValues)) var issues []Issue for index, item := range itemValues { - parsed, err := parseService(item) + parsed, err := parse(item) if err != nil { - issues = append(issues, Issue{Index: index, Message: err.Error(), Scope: IssueService}) + issues = append(issues, Issue{Index: index, Message: err.Error(), Scope: scope}) continue } items = append(items, parsed) } next, err := optionalText(object["next"], "next", 2048) if err != nil { - return SearchPage{}, err + return SearchResponse[Item]{}, err } var facets *Facets if raw, ok := object["facets"]; ok { parsed, err := parseFacets(raw) if err != nil { - return SearchPage{}, err + return SearchResponse[Item]{}, err } facets = &parsed } - return SearchPage{ + return SearchResponse[Item]{ Additional: cloneAdditional(object, "items", "next", "facets"), Facets: facets, Issues: issues, Items: items, Next: next, }, nil diff --git a/examples/odp-agent-discovery/README.md b/examples/odp-agent-discovery/README.md index 1eea7ad..ce5831c 100644 --- a/examples/odp-agent-discovery/README.md +++ b/examples/odp-agent-discovery/README.md @@ -1,11 +1,16 @@ # ODP Agent discovery This example uses the Go Agent package to inspect live ODP Services, list terse Offerings, and fetch -each Offering's full details. +each Offering's full details. It also searches mixed Directory results and retrieves live details +for Collection results through their owning Service. The example composes a **mock directory** from the reachable Service origins supplied on the command line. The mock is isolated in `mock_directory.go`; it is not the canonical ODP directory and does -not imitate directory ranking, filtering, facets, or suggestions. +not imitate directory ranking, filtering, facets, or suggestions. It samples at most two Collections +from one response when both listing and detail operations permit anonymous access. This is local +example setup; the production Directory indexes explicitly submitted Collections without crawling +catalogs. Synthetic HTTPS origins in the mock map to the supplied local Service clients, so no +requests are sent to those example domains. Start the small Service in one terminal: @@ -30,4 +35,6 @@ go run ./examples/odp-agent-discovery \ Unreachable candidates are omitted before the mock directory is created. The output identifies the Service, prints its validated ODP Service Document, prints each terse Offering returned by catalog -navigation, and then prints the corresponding full Offering response. +navigation, and then prints the corresponding full Offering response. Mixed discovery prints each +Service name and the full details of sampled Collections. Collection sampling errors are reported +instead of silently presenting the sample as complete. diff --git a/examples/odp-agent-discovery/main.go b/examples/odp-agent-discovery/main.go index 3b0a675..03a8e1f 100644 --- a/examples/odp-agent-discovery/main.go +++ b/examples/odp-agent-discovery/main.go @@ -46,6 +46,23 @@ func main() { } printJSON("ODP Service Document", inspection.Document) } + for result, err := range mock.client.Search(ctx, directory.DirectorySearchRequest{}, directory.IterationOptions{}).Items { + if err != nil { + log.Fatal(err) + } + switch result.Type { + case "service": + fmt.Printf("Directory Service: %s\n", result.Service.Name) + case "collection": + full, err := mock.serviceClients[result.Service.ServiceOrigin].GetCollection(ctx, result.Collection.ID, "full") + if err != nil { + log.Fatal(err) + } + printJSON("Full Collection from "+result.Service.Name, full) + default: + fmt.Printf("Unsupported Directory result type: %s\n", result.Type) + } + } odpAgent, err := agent.New(agent.AgentOptions{ Directory: mock.client, diff --git a/examples/odp-agent-discovery/mock_directory.go b/examples/odp-agent-discovery/mock_directory.go index d5fa813..ef8637b 100644 --- a/examples/odp-agent-discovery/mock_directory.go +++ b/examples/odp-agent-discovery/mock_directory.go @@ -5,6 +5,7 @@ import ( "context" "encoding/json" "errors" + "fmt" "io" "net/http" "time" @@ -21,6 +22,7 @@ type mockDirectory struct { } type directoryServiceJSON struct { + ServiceID string `json:"service_id"` Description string `json:"description"` IndexedAt string `json:"indexed_at"` Keywords []string `json:"keywords,omitempty"` @@ -36,7 +38,8 @@ func createMockDirectory(ctx context.Context, candidates []string) (*mockDirecto serviceClients := make(map[string]*agent.ServiceClient) services := make([]directory.Service, 0, len(candidates)) wireServices := make([]directoryServiceJSON, 0, len(candidates)) - for _, candidate := range candidates { + wireResults := make([]any, 0, len(candidates)) + for index, candidate := range candidates { client, err := agent.NewServiceClient(agent.ServiceClientOptions{AllowLocalNetwork: true, ServiceURL: candidate}) if err != nil { continue @@ -46,18 +49,37 @@ func createMockDirectory(ctx context.Context, candidates []string) (*mockDirecto continue } document := inspection.Document + origin := fmt.Sprintf("https://service-%d.mock-directory.example", index+1) service := directory.Service{ Description: document.Description, IndexedAt: time.Now().UTC(), Keywords: document.Keywords, Language: document.Language, Localizations: document.Localizations, Name: document.Name, - Operations: document.Operations, Protocols: document.Protocols, ServiceOrigin: inspection.ServiceOrigin, + Operations: document.Operations, Protocols: document.Protocols, ServiceOrigin: origin, } services = append(services, service) - serviceClients[inspection.ServiceOrigin] = client - wireServices = append(wireServices, directoryServiceJSON{ + serviceClients[origin] = client + wireService := directoryServiceJSON{ + ServiceID: fmt.Sprintf("mock-%d", index+1), Description: document.Description, IndexedAt: service.IndexedAt.Format(time.RFC3339), Keywords: document.Keywords, Language: document.Language, Localizations: document.Localizations, Name: document.Name, - Operations: document.Operations, Protocols: document.Protocols, ServiceOrigin: inspection.ServiceOrigin, - }) + Operations: document.Operations, Protocols: document.Protocols, ServiceOrigin: origin, + } + wireServices = append(wireServices, wireService) + wireResults = append(wireResults, map[string]any{"type": "service", "service": wireService, "indexed_at": wireService.IndexedAt}) + anonymous := map[odp.Operation]bool{} + for _, operation := range document.Operations { + anonymous[operation.Name] = operation.Authentication != odp.AuthenticationRequired + } + if anonymous[odp.OperationListCollections] && anonymous[odp.OperationGetCollection] { + for collection, err := range client.ListCollections(ctx, agent.ListOptions{MaxItems: 2, MaxPages: 1}) { + if err != nil { + return nil, fmt.Errorf("sample Collections from %s: %w", candidate, err) + } + wireResults = append(wireResults, map[string]any{ + "type": "collection", "service": wireService, "indexed_at": wireService.IndexedAt, + "collection": map[string]any{"id": collection.ID, "name": collection.Name, "description": collection.Description}, + }) + } + } } if len(services) == 0 { return nil, errors.New("no configured ODP Services are reachable") @@ -68,12 +90,24 @@ func createMockDirectory(ctx context.Context, candidates []string) (*mockDirecto if err != nil { return nil, err } + mixedBody, err := json.Marshal(map[string]any{"items": wireResults}) + if err != nil { + return nil, err + } transport := roundTripFunc(func(request *http.Request) (*http.Response, error) { - if request.Method != http.MethodPost || request.URL.Path != "/v1/services/search" { + if request.Method != http.MethodPost || request.URL.Host != "api.inflowpay.ai" || request.URL.Scheme != "https" { return nil, errors.New("mock directory received an unsupported request") } + responseBody := body + switch request.URL.Path { + case "/v1/services/search": + case "/v1/directory/search": + responseBody = mixedBody + default: + return nil, errors.New("mock directory received an unsupported path") + } return &http.Response{ - Body: io.NopCloser(bytes.NewReader(body)), Header: http.Header{"Content-Type": []string{"application/json"}}, StatusCode: http.StatusOK, + Body: io.NopCloser(bytes.NewReader(responseBody)), Header: http.Header{"Content-Type": []string{"application/json"}}, StatusCode: http.StatusOK, }, nil }) directoryClient, err := directory.New(directory.Options{HTTPClient: &http.Client{Transport: transport}}) diff --git a/testdata/consumer/main.go b/testdata/consumer/main.go index 330ddd0..a411f68 100644 --- a/testdata/consumer/main.go +++ b/testdata/consumer/main.go @@ -1,6 +1,7 @@ package main import ( + "context" "fmt" odp "github.com/offering-protocol/odp-go" @@ -11,7 +12,12 @@ import ( func main() { _, _ = agent.NewServiceClient(agent.ServiceClientOptions{ServiceURL: "https://service.example"}) - _, _ = directory.New(directory.Options{}) + directoryClient, _ := directory.New(directory.Options{}) + var mixed directory.SearchSequence[directory.Result] = directoryClient.Search(context.Background(), directory.DirectorySearchRequest{}, directory.IterationOptions{}) + var services directory.SearchSequence[directory.Service] = directoryClient.SearchServices(context.Background(), directory.SearchRequest{}, directory.IterationOptions{}) + if mixed.Items == nil || mixed.Responses == nil || services.Items == nil || services.Responses == nil { + panic("missing search iterator") + } _, _ = service.NewStaticCatalog(service.StaticCatalogOptions{}) fmt.Println(odp.Version) }