From 00a5d4d8910745e57b10134829f1547c3122c6f4 Mon Sep 17 00:00:00 2001 From: ameersohel45 Date: Thu, 3 Sep 2026 15:12:16 +0530 Subject: [PATCH 1/3] feat: add the Mandi provider plugin [#8] Serves openagrinet:MandiPrice against Agmarknet's Vistaar select, as a sibling to weather: a domain package of 58 lines wrapping internal/upstream, which needed no change for it. That was the test of whether the machinery and domain split from 2b3cab1 actually held, and it did. The package has NO prerequisites, and the reason is the pack rather than luck. A MandiPrice select names the market it wants -- market.marketCode, market.district, market.state -- and carries a commodity code and a validity window, which is every parameter the upstream takes. There is no top-level location in the pack, so nothing has to turn a point into a market, which is the one thing the provider backend needs a spatial SQL query for and the one thing this adapter may not do. The mapping carries the whole contract. Three things in it are not obvious: the upstream's records use Title Case keys WITH SPACES -- `Modal Price` -- so they need backticks, and its prices are STRINGS, so they need $number before they satisfy the pack's numeric types. Both are pinned by a verbatim capture from the provider backend's own documentation. dates convert twice. The pack speaks ISO, the upstream speaks dd-MM-yyyy, so the request half converts out and the response half converts back. the pack requires none of the fields the upstream needs -- an OnDemand select requires only supportedCommodities and supportedPriceFields, leaving market and validity optional. So a spec-valid select can be unanswerable, and the mapping's required: block refuses those with its own message rather than earning a 400 or, worse, an empty result that reads as "no prices". Resource ids are built from codes rather than the names the upstream reports: "Kasdol APMC" and "Paddy(Common)" carry spaces and brackets, and an id a consumer may put in a URL should not. Verified: the shipped mapping run through the real mapper and the real step answers two records as two Direct resources with their prices converted, the offer's references rewritten to match, and absent min/max left absent rather than zeroed. Both directions validate -- the select against beckn.yaml and MandiPrice v0.1 in OnDemand mode, the on_select against beckn.yaml and the pack in Direct mode, with no errors. --- .../agmarknet/mandi-price.select.yaml | 200 ++++++++ install/build-plugins.sh | 1 + pkg/plugin/implementation/mandi/cmd/plugin.go | 104 ++++ pkg/plugin/implementation/mandi/mandi.go | 38 ++ .../implementation/mandi/mappings_test.go | 458 ++++++++++++++++++ .../implementation/mandi/prerequisites.go | 20 + 6 files changed, 821 insertions(+) create mode 100644 config/mappings/agmarknet/mandi-price.select.yaml create mode 100644 pkg/plugin/implementation/mandi/cmd/plugin.go create mode 100644 pkg/plugin/implementation/mandi/mandi.go create mode 100644 pkg/plugin/implementation/mandi/mappings_test.go create mode 100644 pkg/plugin/implementation/mandi/prerequisites.go diff --git a/config/mappings/agmarknet/mandi-price.select.yaml b/config/mappings/agmarknet/mandi-price.select.yaml new file mode 100644 index 00000000..2f0e70b3 --- /dev/null +++ b/config/mappings/agmarknet/mandi-price.select.yaml @@ -0,0 +1,200 @@ +# Agmarknet Vistaar, openagrinet:MandiPrice, select. Both directions, one file. +# +# One file per binding-action rather than one per direction, because both legs of +# an exchange are one contract: the response has to answer the request that was +# sent, and splitting them lets one change without the other. +# +# The upstream is Agmarknet's Vistaar select. It takes governed codes -- state, +# district, market, commodity -- plus a date range, and every one of them is in +# the payload, so nothing here needs resolving before the call. That is why the +# mandi plugin has no prerequisites: a MandiPrice select names the market it +# wants rather than a point to search from. +# +# NOTHING HERE IS OUTSIDE THE PACK. openagrinet:MandiPrice v0.1 carries every +# field this answer sets. Where the upstream reports something the pack has no +# home for, it is dropped rather than invented. + +# What this capability cannot serve, refused before the provider is called. +# +# The pack requires none of these: a MandiPrice select is OnDemand, and that +# branch requires only supportedCommodities and supportedPriceFields. It leaves +# market and validity optional, and defines market.district and market.state as +# "name or governed code". So a payload can be perfectly valid and still be +# unanswerable by this upstream, which wants codes and a date range. +# +# Refusing here names what is missing. Sending it anyway earns a 400 from +# Agmarknet, or worse an empty result that reads as "no prices". +required: + - check: | + ( + $ra := beckn.message.contract.commitments[0].resources[0].resourceAttributes; + $exists($ra.supportedCommodities[0].code) + ) + message: "this capability needs a commodity code in supportedCommodities[0].code" + - check: | + ( + $ra := beckn.message.contract.commitments[0].resources[0].resourceAttributes; + $exists($ra.market.state) and $exists($ra.market.district) + ) + message: "this capability needs governed state and district codes in market" + - check: | + ( + $ra := beckn.message.contract.commitments[0].resources[0].resourceAttributes; + $exists($ra.validity.startsAt) and $exists($ra.validity.endsAt) + ) + message: "this capability needs a validity window; it reports prices over a date range" + +# The upstream is a GET, so this object becomes the query string. The token is +# not here and must never be: it comes from the adapter's authScheme query, +# whose value is read from an environment variable. This file is published. +# +# marketcode is sent when the payload names one and omitted otherwise, which is +# what the upstream expects: without it the query widens from one market to the +# whole district. +request: | + ( + $ra := beckn.message.contract.commitments[0].resources[0].resourceAttributes; + + /* The upstream wants dd-MM-yyyy; the pack's validity is an ISO date. A + substring reorder rather than a date library, because these are dates + with no time and no zone, and $fromMillis would invent both. */ + $ddmmyyyy := function($iso) { + $substring($iso, 8, 2) & "-" & $substring($iso, 5, 2) & "-" & $substring($iso, 0, 4) + }; + + $base := { + "statecode": $ra.market.state, + "districtcode": $ra.market.district, + "commoditycode": $ra.supportedCommodities[0].code, + "from_date": $ddmmyyyy($ra.validity.startsAt), + "to_date": $ddmmyyyy($ra.validity.endsAt) + }; + + $exists($ra.market.marketCode) + ? $merge([$base, {"marketcode": $ra.market.marketCode}]) + : $base + ) + +# One resource per price record, in Direct mode. +# +# Direct is what the pack requires of an answer: the resource now CARRIES the +# observation rather than advertising that it could obtain one. It requires +# source, commodity, market, arrivalDate, prices and generatedAt, and all six +# are set below. +# +# The upstream's records use Title Case keys WITH SPACES -- `Modal Price`, not +# modalPrice -- so they need backticks, and its prices are STRINGS, so they need +# $number() before they satisfy the pack's numeric types. +response: | + ( + $records := $type(response) = "array" ? response + : $exists(response.data) ? response.data + : $exists(response.records) ? response.records + : []; + + $selected := beckn.message.contract.commitments[0]; + $ra := $selected.resources[0].resourceAttributes; + + /* Bound once because it is used twice -- for a resource's own id and for + the offer's reference to it. Two copies of one expression is how a + dangling reference gets reintroduced. */ + /* Built from CODES, not the names the upstream reports. A market name + carries spaces and a commodity name carries brackets -- "Kasdol APMC", + "Paddy(Common)" -- and an identifier that a consumer may put in a URL or + a filter should not. The codes are already in the payload, so they cost + nothing, and they are stable where a display name is not. + + The market code is optional: without it the query widened to the whole + district, so the district code is what identifies the scope. */ + $iso := function($ddmmyyyy) { + $substring($ddmmyyyy, 6, 4) & "-" & $substring($ddmmyyyy, 3, 2) & "-" & $substring($ddmmyyyy, 0, 2) + }; + + $scope := $exists($ra.market.marketCode) ? $ra.market.marketCode : $ra.market.district; + $resourceId := function($r) { + "res:agmarknet:" & $scope & ":" & $ra.supportedCommodities[0].code + & ":" & $iso($r.`Arrival Date`) + }; + + /* dd-MM-yyyy back to ISO, so the answer speaks the pack's date format + rather than the upstream's. */ + + /* Absent rather than present-and-empty: a consumer must be able to tell + "the market reported no minimum" from "the minimum was zero". */ + $priced := function($value) { $exists($value) ? $number($value) }; + + { + "context": { + "version": beckn.context.version, + "action": "on_select", + "networkId": beckn.context.networkId, + "transactionId": beckn.context.transactionId, + "messageId": beckn.context.messageId, + "timestamp": $now() + }, + "message": { + "contract": { + "commitments": [ + { + "status": { + "descriptor": { "code": "DRAFT", "name": "Draft" } + }, + /* The offer is echoed, but its references are not: the request + named an abstract price enquiry and the answer returns the + concrete observations. Leaving resourceIds as they arrived + would point the offer at an id appearing nowhere here. */ + "offer": $merge([ + $selected.offer, + { "resourceIds": [$map($records, function($r) { $resourceId($r) })] } + ]), + /* Wrapped: JSONata collapses a one-element sequence to a bare + value, so a single-record answer would return an object where + every other count returns a list. */ + "resources": [$map($records, function($r) { + { + "id": $resourceId($r), + /* Required by Commitment.resources in the Beckn v2 spec, + which defines no quantity property and carries no Quantity + schema at all -- a defect upstream. One resource is one + market's observation for one day, so one. */ + "quantity": 1, + "resourceAttributes": { + "@context": "https://schemas.openagrinet.global/schema/MandiPrice/v0.1/context.jsonld", + "@type": "openagrinet:MandiPrice", + "informationMode": "Direct", + "subjectCategories": $ra.subjectCategories, + "source": { + "sourceId": "agmarknet", + "sourceName": "Agmarknet Vistaar" + }, + "commodity": { + "code": $ra.supportedCommodities[0].code, + "name": $r.Commodity + }, + "commodityGroup": $r.Group, + "variety": $r.Variety, + "grade": $r.Grade, + "market": { + "marketName": $r.Market, + "marketCode": $ra.market.marketCode, + "district": $r.District, + "state": $r.State + }, + "arrivalDate": $iso($r.`Arrival Date`), + "prices": { + "minimum": $priced($r.`Min Price`), + "maximum": $priced($r.`Max Price`), + "modal": $number($r.`Modal Price`), + "currency": "INR", + "unit": $r.`Price Unit` + }, + "generatedAt": $now() + } + } + })] + } + ] + } + } + } + ) diff --git a/install/build-plugins.sh b/install/build-plugins.sh index 3bf07a01..8ac0d713 100755 --- a/install/build-plugins.sh +++ b/install/build-plugins.sh @@ -33,6 +33,7 @@ plugins=( "oanregistry" "jsonmapper" "weather" + "mandi" "manifestloader" "reqpreprocessor" "otelsetup" diff --git a/pkg/plugin/implementation/mandi/cmd/plugin.go b/pkg/plugin/implementation/mandi/cmd/plugin.go new file mode 100644 index 00000000..5b4fbb70 --- /dev/null +++ b/pkg/plugin/implementation/mandi/cmd/plugin.go @@ -0,0 +1,104 @@ +// Command plugin builds the mandi provider step as a loadable plugin. +// +// The filename of the built .so is the id a deployment names in providerSteps, +// so this package is mandi's whole public surface: a config map in, a step out. +package main + +import ( + "context" + "errors" + "fmt" + "strconv" + "strings" + + "github.com/beckn-one/beckn-onix/pkg/log" + "github.com/beckn-one/beckn-onix/pkg/plugin/definition" + "github.com/beckn-one/beckn-onix/pkg/plugin/implementation/mandi" +) + +// mandiProvider implements definition.ProviderStepProvider. +type mandiProvider struct{} + +// newStepFunc creates a new step. Indirected for tests. +var newStepFunc = mandi.New + +// parseConfig turns the plugin config map into a typed Config. Anything absent +// is left zero: mandi.New applies the defaults and validates the auth scheme, +// so those rules live in one place. +func (p mandiProvider) parseConfig(config map[string]string) (*mandi.Config, error) { + cfg := &mandi.Config{ + BindingKeys: splitList(config["bindingKeys"]), + // Absent means the Beckn v2 convention. See upstream.Config for why + // this is a default rather than something to set. + ProviderIDAt: config["providerIdAt"], + CapabilityCodeAt: config["capabilityCodeAt"], + AuthScheme: config["authScheme"], + UsernameEnv: config["usernameEnv"], + PasswordEnv: config["passwordEnv"], + HeaderName: config["headerName"], + HeaderValueEnv: config["headerValueEnv"], + QueryName: config["queryName"], + QueryValueEnv: config["queryValueEnv"], + } + + if raw, exists := config["maxResponseBytes"]; exists && raw != "" { + value, err := strconv.ParseInt(raw, 10, 64) + if err != nil { + return nil, fmt.Errorf("invalid maxResponseBytes value '%s': %w", raw, err) + } + if value <= 0 { + return nil, fmt.Errorf("maxResponseBytes must be positive, got %d", value) + } + cfg.MaxResponseBytes = value + } + + return cfg, nil +} + +// New creates a new mandi provider step instance. +func (p mandiProvider) New(ctx context.Context, registry definition.ProviderRecordLookup, mapper definition.Mapper, config map[string]string) (definition.Step, func() error, error) { + if ctx == nil { + return nil, nil, errors.New("context cannot be nil") + } + + cfg, err := p.parseConfig(config) + if err != nil { + log.Errorf(ctx, err, "Failed to parse mandi configuration") + return nil, nil, fmt.Errorf("failed to parse mandi configuration: %w", err) + } + + step, closer, err := newStepFunc(ctx, registry, mapper, cfg) + if err != nil { + log.Errorf(ctx, err, "Failed to create mandi step") + return nil, nil, err + } + + log.Infof(ctx, "Mandi step created successfully") + return step, closer, nil +} + +// splitList reads a comma-separated config value, which is how a list reaches a +// plugin -- the config is map[string]string. Blanks are dropped and spaces +// trimmed, so a trailing comma or a wrapped line is not a config error. +// +// A comma is unambiguous here: a binding key separates its own halves with a +// pipe. +func splitList(raw string) []string { + if strings.TrimSpace(raw) == "" { + return nil + } + var out []string + for _, part := range strings.Split(raw, ",") { + if trimmed := strings.TrimSpace(part); trimmed != "" { + out = append(out, trimmed) + } + } + return out +} + +// Provider is the exported plugin instance. +var Provider = mandiProvider{} + +// Compile-time proof the provider satisfies the interface the manager asserts +// against. A mismatch is otherwise a runtime cast failure at startup. +var _ definition.ProviderStepProvider = Provider diff --git a/pkg/plugin/implementation/mandi/mandi.go b/pkg/plugin/implementation/mandi/mandi.go new file mode 100644 index 00000000..26e8b94b --- /dev/null +++ b/pkg/plugin/implementation/mandi/mandi.go @@ -0,0 +1,38 @@ +// Package mandi serves the network's market price capabilities. +// +// One package per schema pack family, so which plugin owns a capability is +// readable from its binding key: openagrinet:MandiPrice is mandi's, +// openagrinet:WeatherObservation is weather's. +// +// Almost nothing lives here, and that is the point. Recognising a capability, +// resolving the call plan, authenticating, calling with the registry's budget +// and translating in both directions are all internal/upstream's, because none +// of them differ by domain. What this package owns is its name, and +// prerequisites -- the work a mapping cannot express, which is domain knowledge +// by definition. +// +// The upstream this was written against is Agmarknet's Vistaar API, whose +// select takes governed codes for state, district, market and commodity plus a +// date range, all of which a MandiPrice payload carries. So the package is a +// name and nothing else: see prerequisites.go for why that is worth stating. +package mandi + +import ( + "context" + + "github.com/beckn-one/beckn-onix/pkg/plugin/definition" + "github.com/beckn-one/beckn-onix/pkg/plugin/implementation/internal/upstream" +) + +// Config is upstream's, unchanged. Aliased here so a domain plugin's cmd package +// need not know where the machinery lives. +type Config = upstream.Config + +// New creates the mandi step. +// +// Which capabilities it answers to is configuration, with no default: a package +// serving a family cannot guess which of them a deployment has providers for. +func New(ctx context.Context, registry definition.ProviderRecordLookup, mapper definition.Mapper, + cfg *Config) (definition.Step, func() error, error) { + return upstream.New(ctx, registry, mapper, prerequisites, cfg) +} diff --git a/pkg/plugin/implementation/mandi/mappings_test.go b/pkg/plugin/implementation/mandi/mappings_test.go new file mode 100644 index 00000000..1eae729f --- /dev/null +++ b/pkg/plugin/implementation/mandi/mappings_test.go @@ -0,0 +1,458 @@ +package mandi_test + +// mappings_test.go runs the shipped mandi mapping through the real mapper and +// the real provider step. It is the only test that proves the three pieces fit: +// a mapping is JSONata inside YAML fetched over HTTP, and nothing but running +// it establishes that what is published actually produces valid Beckn. +// +// An external test package on purpose -- it uses the plugins exactly as the +// adapter does, through their exported surface and nothing else. + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "net/url" + "os" + "path/filepath" + "slices" + "strings" + "testing" + + "github.com/beckn-one/beckn-onix/pkg/model" + "github.com/beckn-one/beckn-onix/pkg/plugin/implementation/jsonmapper" + "github.com/beckn-one/beckn-onix/pkg/plugin/implementation/mandi" +) + +// mappingsDir is where the shipped mappings live, relative to this package. +const mappingsDir = "../../../../config/mappings/agmarknet" + +// shippedMapping is the file this binding-action publishes: one file, both +// directions. The action segment of the name must match the action the registry +// entry declares -- a mismatch would apply a correct mapping to the wrong call, +// silently. +const shippedMapping = "mandi-price.select.yaml" + +// shippedCapability is what the pack calls this capability, and the second half +// of the binding key the registry indexes the provider's record by. +const shippedCapability = "openagrinet:MandiPrice" + +const shippedBindingKey = "agmarknet|" + shippedCapability + +// selectRequest is a MandiPrice select in OnDemand mode: it names the market and +// commodity it wants prices for, and carries no prices of its own -- the pack +// forbids that combination. +const selectRequest = `{ + "context": { + "version": "2.0.0", + "action": "select", + "networkId": "oan-dev", + "transactionId": "9f2c1a8e-4b70-4d31-9c55-6f2e0b1d7a44", + "messageId": "7d41b9e0-52a6-4c18-8b73-1e9f0a4c6d22", + "timestamp": "2026-09-03T06:12:01.330Z" + }, + "message": { + "contract": { + "commitments": [ + { + "status": { "descriptor": { "code": "DRAFT", "name": "Draft" } }, + "resources": [ + { + "id": "res:agmarknet:price-enquiry", + "quantity": 1, + "resourceAttributes": { + "@context": "https://schemas.openagrinet.global/schema/MandiPrice/v0.1/context.jsonld", + "@type": "openagrinet:MandiPrice", + "informationMode": "OnDemand", + "subjectCategories": ["Market"], + "supportedCommodities": [{ "code": "2", "name": "Paddy(Common)" }], + "supportedPriceFields": ["Minimum", "Maximum", "Modal"], + "market": { + "marketName": "Kasdol APMC", + "marketCode": "2056", + "district": "96", + "state": "CG" + }, + "validity": { "startsAt": "2025-08-20", "endsAt": "2025-08-21" } + } + } + ], + "offer": { + "id": "offer:agmarknet:open-data", + "resourceIds": ["res:agmarknet:price-enquiry"], + "provider": { + "id": "agmarknet", + "descriptor": { "code": "AGMARKNET-01", "name": "Agmarknet Vistaar" } + } + } + } + ] + } + } +}` + +// providerResponse is a verbatim Agmarknet Vistaar answer, taken from the +// working example in the provider backend's own documentation. Two records, so +// the mapping is exercised on a list rather than a single object. +// +// Note what it is: Title Case keys WITH SPACES, and prices as STRINGS. Both are +// the reason the mapping needs backticks and $number, and pinning a real +// capture here is what keeps that honest. +const providerResponse = `[ + { + "Grade": "Non-FAQ", + "Group": "Cereals", + "State": "Chattisgarh", + "Market": "Kasdol APMC", + "Variety": "D.B.", + "District": "Balodabazar", + "Commodity": "Paddy(Common)", + "Max Price": "2100", + "Min Price": "1900", + "Price Unit": "Rs./Qtl", + "Modal Price": "2000", + "Arrival Date": "20-08-2025" + }, + { + "Grade": "FAQ", + "Group": "Cereals", + "State": "Chattisgarh", + "Market": "Kasdol APMC", + "Variety": "Common", + "District": "Balodabazar", + "Commodity": "Paddy(Common)", + "Price Unit": "Rs./Qtl", + "Modal Price": "2050", + "Arrival Date": "21-08-2025" + } +]` + +// serveMappings publishes the shipped mapping files over HTTP, which is how the +// mapper fetches them in production. +func serveMappings(t *testing.T) *httptest.Server { + t.Helper() + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, err := os.ReadFile(filepath.Join(mappingsDir, filepath.Base(r.URL.Path))) + if err != nil { + t.Errorf("could not read the mapping %q: %v", r.URL.Path, err) + w.WriteHeader(http.StatusNotFound) + return + } + fmt.Fprint(w, string(body)) + })) +} + +// stubRegistry answers with the call plan the live registry holds for this +// capability. +type stubRegistry struct{ plan *model.ProviderRecord } + +func (s *stubRegistry) ProviderRecord(context.Context, string) (*model.ProviderRecord, error) { + return s.plan, nil +} + +// runShipped drives the real step over the real mapping and returns the query +// the provider saw and the answer produced. +func runShipped(t *testing.T, request string) (url.Values, map[string]any) { + t.Helper() + + mappings := serveMappings(t) + defer mappings.Close() + + var gotQuery url.Values + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotQuery = r.URL.Query() + fmt.Fprint(w, providerResponse) + })) + defer upstream.Close() + + mapper, closeMapper, err := jsonmapper.New(context.Background(), &jsonmapper.Config{}) + if err != nil { + t.Fatalf("failed to build the mapper: %v", err) + } + defer closeMapper() + + registry := &stubRegistry{plan: &model.ProviderRecord{ + BindingKey: shippedBindingKey, + ParticipantID: "agmarknet", + CapabilityCode: shippedCapability, + BaseURL: upstream.URL, + Actions: map[string]model.ActionPlan{ + "select": {Method: http.MethodGet, Path: "/v1/fetch-agmarknet-vistaar", + Mappings: mappings.URL + "/" + shippedMapping, TimeoutMs: 30000, RetryMax: 3}, + }, + }} + + step, closeStep, err := mandi.New(context.Background(), registry, mapper, + &mandi.Config{BindingKeys: []string{shippedBindingKey}}) + if err != nil { + t.Fatalf("failed to build the step: %v", err) + } + defer closeStep() + + stepCtx := &model.StepContext{Context: t.Context(), Body: []byte(request)} + if err := step.Run(stepCtx); err != nil { + t.Fatalf("Run() returned an unexpected error: %v", err) + } + if len(stepCtx.ResponseBody) == 0 { + t.Fatal("the step produced no answer") + } + var answer map[string]any + if err := json.Unmarshal(stepCtx.ResponseBody, &answer); err != nil { + t.Fatalf("the answer is not JSON: %v\n%s", err, stepCtx.ResponseBody) + } + return gotQuery, answer +} + +func TestShippedMappingServesARealSelect(t *testing.T) { + gotQuery, answer := runShipped(t, selectRequest) + + // --- the request reached the provider as Agmarknet expects -------------- + // Every one of these comes off the payload. Nothing was resolved before the + // call, which is the whole claim of this plugin having no prerequisites. + for param, want := range map[string]string{ + "statecode": "CG", + "districtcode": "96", + "marketcode": "2056", + "commoditycode": "2", + // dd-MM-yyyy, not the ISO the payload carried. + "from_date": "20-08-2025", + "to_date": "21-08-2025", + } { + if got := gotQuery.Get(param); got != want { + t.Errorf("upstream query %s = %q, want %q", param, got, want) + } + } + // The credential is the adapter's business, never the mapping's. + if gotQuery.Has("token") { + t.Error("the mapping must not put a token in the query; authScheme does that") + } + + // --- the answer is Beckn ----------------------------------------------- + beckncontext, _ := answer["context"].(map[string]any) + if beckncontext["action"] != "on_select" { + t.Errorf("action = %v, want on_select", beckncontext["action"]) + } + if beckncontext["transactionId"] != "9f2c1a8e-4b70-4d31-9c55-6f2e0b1d7a44" { + t.Errorf("transactionId = %v, want the one from the request", beckncontext["transactionId"]) + } + // A mapping transforms a payload; it does not assert who anyone is. + for _, field := range []string{"bapId", "bapUri", "bppId", "bppUri"} { + if _, present := beckncontext[field]; present { + t.Errorf("response context carries %q; a mapping must not assert identity", field) + } + } + + // Written out so the answer can be validated against the Beckn v2 spec and + // the MandiPrice pack by tooling outside Go. Skipped unless asked for. + if path := os.Getenv("MANDI_DUMP_ANSWER"); path != "" { + raw, _ := json.MarshalIndent(answer, "", " ") + if err := os.WriteFile(path, raw, 0o600); err != nil { + t.Fatalf("could not write the answer: %v", err) + } + } + + commitment := firstCommitment(t, answer) + if status := commitment["status"].(map[string]any)["descriptor"].(map[string]any); status["code"] != "DRAFT" { + t.Errorf("status = %v, want DRAFT -- the spec's enum is DRAFT, ACTIVE, CLOSED", status["code"]) + } + + // --- one resource per price record -------------------------------------- + resources, _ := commitment["resources"].([]any) + if len(resources) != 2 { + t.Fatalf("got %d resources, want 2 -- one per record the provider answered with", len(resources)) + } + + returned := make([]string, 0, len(resources)) + for _, entry := range resources { + resource, _ := entry.(map[string]any) + id, _ := resource["id"].(string) + // Codes, not names: no spaces or brackets in an identifier. + if !strings.HasPrefix(id, "res:agmarknet:2056:2:") { + t.Errorf("resource id = %q, want one built from the market and commodity codes", id) + } + if strings.ContainsAny(id, " ()") { + t.Errorf("resource id %q contains a space or bracket; use codes, not display names", id) + } + // Required by Commitment.resources in the spec even though the spec + // defines no quantity property. + if _, present := resource["quantity"]; !present { + t.Errorf("resource %s carries no quantity", id) + } + returned = append(returned, id) + } + + // The offer must reference what was actually returned, not what was asked + // for. This is the assertion that fails the moment the offer is echoed. + offer, _ := commitment["offer"].(map[string]any) + referenced, _ := offer["resourceIds"].([]any) + if len(referenced) != len(returned) { + t.Fatalf("offer references %d resources, want %d", len(referenced), len(returned)) + } + for _, reference := range referenced { + if !slices.Contains(returned, reference.(string)) { + t.Errorf("offer references %v, which is not among the resources returned", reference) + } + } + if offer["id"] != "offer:agmarknet:open-data" { + t.Errorf("offer id = %v, want the one the request offered", offer["id"]) + } + + // --- the MandiPrice pack, Direct mode ----------------------------------- + first, _ := resources[0].(map[string]any) + attributes, _ := first["resourceAttributes"].(map[string]any) + for _, f := range []struct{ key, want string }{ + {"@type", "openagrinet:MandiPrice"}, + {"informationMode", "Direct"}, + } { + if attributes[f.key] != f.want { + t.Errorf("%s = %v, want %v", f.key, attributes[f.key], f.want) + } + } + // Direct requires all six of these. + for _, required := range []string{"source", "commodity", "market", "arrivalDate", "prices", "generatedAt"} { + if attributes[required] == nil { + t.Errorf("resourceAttributes carries no %q", required) + } + } + // OnDemand's fields must NOT appear: the pack forbids prices alongside + // them, and an answer advertising a capability is a category error. + for _, absent := range []string{"supportedCommodities", "supportedPriceFields"} { + if _, present := attributes[absent]; present { + t.Errorf("a Direct answer must not carry %q", absent) + } + } + + // --- the prices, converted from strings --------------------------------- + prices, _ := attributes["prices"].(map[string]any) + for field, want := range map[string]float64{"minimum": 1900, "maximum": 2100, "modal": 2000} { + got, ok := prices[field].(float64) + if !ok { + t.Errorf("prices.%s = %#v, want a number -- the upstream sends strings", field, prices[field]) + continue + } + if got != want { + t.Errorf("prices.%s = %v, want %v", field, got, want) + } + } + if prices["currency"] != "INR" || prices["unit"] != "Rs./Qtl" { + t.Errorf("prices currency/unit = %v/%v, want INR/Rs./Qtl", prices["currency"], prices["unit"]) + } + + // arrivalDate is ISO in the answer, though the upstream reported dd-MM-yyyy. + if attributes["arrivalDate"] != "2025-08-20" { + t.Errorf("arrivalDate = %v, want 2025-08-20 in ISO", attributes["arrivalDate"]) + } + + // The pack's enum is Crop, Livestock, Weather, Market, Scheme, Knowledge, + // Service -- so "Market", not "MarketPrice". Echoed from the request, which + // is why getting it wrong there would produce an invalid answer here. + categories, _ := attributes["subjectCategories"].([]any) + if len(categories) != 1 || categories[0] != "Market" { + t.Errorf("subjectCategories = %v, want [Market] from the pack's enum", categories) + } + + market, _ := attributes["market"].(map[string]any) + if market["marketName"] != "Kasdol APMC" || market["state"] != "Chattisgarh" { + t.Errorf("market = %v, want the names the provider reported", market) + } + + // --- a record the market reported only partially ------------------------- + // The second record has no Min or Max Price. Those must be absent, not zero: + // a consumer must be able to tell "not reported" from "reported as zero". + second, _ := resources[1].(map[string]any) + secondPrices, _ := second["resourceAttributes"].(map[string]any)["prices"].(map[string]any) + for _, absent := range []string{"minimum", "maximum"} { + if _, present := secondPrices[absent]; present { + t.Errorf("prices.%s is present for a record that did not report it", absent) + } + } + if secondPrices["modal"] != float64(2050) { + t.Errorf("the second record's modal price = %v, want 2050", secondPrices["modal"]) + } +} + +// The pack leaves every field this upstream needs optional, so a spec-valid +// select can still be unanswerable. The mapping refuses those before the +// provider is called, with its own message. +func TestShippedMappingRefusesWhatItCannotServe(t *testing.T) { + for _, tc := range []struct{ name, drop, expect string }{ + {"no commodity code", "supportedCommodities", "commodity code"}, + {"no market codes", "market", "state and district"}, + {"no validity window", "validity", "validity window"}, + } { + t.Run(tc.name, func(t *testing.T) { + // Built by deleting a key from the decoded fixture rather than by + // editing its text: removing the last member of an object leaves a + // trailing comma, and the resulting parse error would look like a + // mapping failure. + var payload map[string]any + if err := json.Unmarshal([]byte(selectRequest), &payload); err != nil { + t.Fatalf("the fixture is not JSON: %v", err) + } + attributes := payload["message"].(map[string]any)["contract"].(map[string]any)["commitments"].([]any)[0].(map[string]any)["resources"].([]any)[0].(map[string]any)["resourceAttributes"].(map[string]any) + if _, present := attributes[tc.drop]; !present { + t.Fatalf("the fixture has no %q, so this case tests nothing", tc.drop) + } + delete(attributes, tc.drop) + body, err := json.Marshal(payload) + if err != nil { + t.Fatalf("could not rebuild the payload: %v", err) + } + + mappings := serveMappings(t) + defer mappings.Close() + + called := false + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + called = true + fmt.Fprint(w, providerResponse) + })) + defer upstream.Close() + + mapper, closeMapper, err := jsonmapper.New(context.Background(), &jsonmapper.Config{}) + if err != nil { + t.Fatalf("failed to build the mapper: %v", err) + } + defer closeMapper() + + registry := &stubRegistry{plan: &model.ProviderRecord{ + BindingKey: shippedBindingKey, BaseURL: upstream.URL, + Actions: map[string]model.ActionPlan{ + "select": {Method: http.MethodGet, Path: "/v1/fetch-agmarknet-vistaar", + Mappings: mappings.URL + "/" + shippedMapping, TimeoutMs: 30000}, + }, + }} + step, closeStep, err := mandi.New(context.Background(), registry, mapper, + &mandi.Config{BindingKeys: []string{shippedBindingKey}}) + if err != nil { + t.Fatalf("failed to build the step: %v", err) + } + defer closeStep() + + stepCtx := &model.StepContext{Context: t.Context(), Body: body} + if err := step.Run(stepCtx); err == nil { + t.Fatal("expected an unserviceable payload to be refused") + } else if !strings.Contains(err.Error(), tc.expect) { + t.Errorf("error %q should carry the mapping's own message about %q", err, tc.expect) + } + if called { + t.Error("the provider was called for a payload the mapping refuses") + } + }) + } +} + +// firstCommitment reaches the one commitment an answer carries. +func firstCommitment(t *testing.T, answer map[string]any) map[string]any { + t.Helper() + message, _ := answer["message"].(map[string]any) + contract, _ := message["contract"].(map[string]any) + commitments, _ := contract["commitments"].([]any) + if len(commitments) != 1 { + t.Fatalf("got %d commitments, want 1", len(commitments)) + } + commitment, _ := commitments[0].(map[string]any) + return commitment +} diff --git a/pkg/plugin/implementation/mandi/prerequisites.go b/pkg/plugin/implementation/mandi/prerequisites.go new file mode 100644 index 00000000..2027248b --- /dev/null +++ b/pkg/plugin/implementation/mandi/prerequisites.go @@ -0,0 +1,20 @@ +package mandi + +import "github.com/beckn-one/beckn-onix/pkg/plugin/implementation/internal/upstream" + +// prerequisites is what a mandi capability needs that its payload does not +// carry, keyed by binding key. +// +// Empty, and for a better reason than weather's: a market price select names +// the market it wants. The MandiPrice pack has no top-level location -- only +// market.marketCode, market.district and market.state -- so there is nothing +// to resolve. Agmarknet's Vistaar select takes exactly those codes, and the +// mapping reads them straight off the payload. +// +// An entry would be needed only for real I/O: a commodity name to resolve to a +// code, a token to exchange, a point to turn into a market. Each of those is a +// different upstream than the one this was written against, and each would +// bring the question of where the provider-to-function binding belongs -- see +// the note in weather/prerequisites.go and prefer keeping the payload explicit +// over adding an entry here. +var prerequisites = upstream.Prerequisites{} From aaae9d52a76a785ab2a6c4e61627c9f96ba8b926 Mon Sep 17 00:00:00 2001 From: ameersohel45 Date: Sun, 6 Sep 2026 18:31:06 +0530 Subject: [PATCH 2/3] docs(config): serve both capabilities from the reference config [#8] Adds mandi alongside weather in config/oan-provider-adapter.yaml, so the reference shows the thing that is actually interesting about this design: two domain packages in one pipeline, sharing the module, the registry client and the mapper, and sharing nothing else. The whole cost of the second capability is one providerSteps entry and one line in steps. No routing table, no new module, no new port. Which one answers is decided by the payload -- each step builds a binding key from it, serves the request if the key is its own, and passes it through untouched if not -- so the order they appear in does not matter either. mandi uses authScheme query, because Agmarknet's Vistaar API takes its token as a query parameter. The adapter holds the parameter's name and the name of the environment variable carrying the value, never the value, and redacts it from the URL it logs -- so a token cannot reach the log by way of the request. Verified by booting this config in an image that has both plugins: both ProviderStep plugins load, the pipeline initialises as [validateSign validateSchema weather mandi signAck], and the module registers at /. Worth noting the published adapter image does NOT yet carry mandi.so, so against that image this config fails at startup with "plugin mandi not found" until it is rebuilt from this branch. --- config/oan-provider-adapter.yaml | 26 +++++++++++++++++++++++++- 1 file changed, 25 insertions(+), 1 deletion(-) diff --git a/config/oan-provider-adapter.yaml b/config/oan-provider-adapter.yaml index 6fdc9b13..18f13dea 100644 --- a/config/oan-provider-adapter.yaml +++ b/config/oan-provider-adapter.yaml @@ -4,6 +4,11 @@ # capability's call plan from the registry, calls the provider, and answers with # the mapped result. There is no callback -- the answer is the HTTP response. # +# Two capabilities are served here, weather and mandi, by two domain packages +# in one pipeline. They share this module, the registry client and the mapper, +# and share nothing else: which one answers is decided by the payload, not by +# the URL, the domain or the order they appear in. +# # Adding a provider is three things, and none of them is a Go change here: # 1. a registry row binding "|" to a call plan # 2. one mapping file per action, published at the URL that row names @@ -233,6 +238,24 @@ modules: # # maxResponseBytes: 4194304 # default: 4194304 (4 MiB) + # A second capability in the same pipeline, from a different domain + # package. Nothing about it is weather's business: a different + # upstream, a different mapping, a different set of prerequisites -- + # and the same two registry rows. This entry, plus "mandi" in steps + # below, is the entire cost of adding it. + - id: mandi + config: + bindingKeys: "agmarknet|openagrinet:MandiPrice" + + # Agmarknet's Vistaar API takes its token as a QUERY parameter, + # which is what authScheme query is for. The adapter holds the + # parameter's NAME and the name of the variable carrying the + # value -- never the value -- and redacts it from the URL it + # logs, so a token cannot reach the log by way of the request. + authScheme: query + queryName: token + queryValueEnv: MANDI_TOKEN + # Declaring a step above is not enough: THIS list is what runs. A step # that appears under providerSteps but not here never executes, and the # request falls through to the 404 above -- which looks like a registry @@ -240,7 +263,8 @@ modules: steps: - validateSign # the sender's key, from the registry - validateSchema # the pinned Beckn v2 spec - - weather # resolve, map out, call, map back + - weather # openagrinet:WeatherObservation, or pass through + - mandi # openagrinet:MandiPrice, or pass through - signAck # signs whatever the step answered with # ---------------------------------------------------------------------------- From 033807dd143807db487d9ff547757623518ab3a8 Mon Sep 17 00:00:00 2001 From: ameersohel45 Date: Mon, 7 Sep 2026 11:41:57 +0530 Subject: [PATCH 3/3] refactor(mappings): echo the caller's @context in the mandi mapping too [#8] Same change as the weather mapping: the response half read a hardcoded pack URL and now takes it off the incoming select, so the file never has to know which URL is current and cannot disagree with the caller. --- config/mappings/agmarknet/mandi-price.select.yaml | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/config/mappings/agmarknet/mandi-price.select.yaml b/config/mappings/agmarknet/mandi-price.select.yaml index 2f0e70b3..5492275d 100644 --- a/config/mappings/agmarknet/mandi-price.select.yaml +++ b/config/mappings/agmarknet/mandi-price.select.yaml @@ -93,6 +93,12 @@ response: | : []; $selected := beckn.message.contract.commitments[0]; + + /* The caller's own @context, echoed back rather than restated here. A + mapping that hardcodes it has to be reissued whenever the pack URL + moves, and it can disagree with what the request actually declared. + Backticks because @ is an operator in JSONata. */ + $ctx := beckn.message.contract.commitments[0].resources[0].resourceAttributes.`@context`; $ra := $selected.resources[0].resourceAttributes; /* Bound once because it is used twice -- for a resource's own id and for @@ -159,7 +165,7 @@ response: | market's observation for one day, so one. */ "quantity": 1, "resourceAttributes": { - "@context": "https://schemas.openagrinet.global/schema/MandiPrice/v0.1/context.jsonld", + "@context": $ctx, "@type": "openagrinet:MandiPrice", "informationMode": "Direct", "subjectCategories": $ra.subjectCategories,