feat: add the Mandi provider plugin

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()
+                  }
+                }
+              })]
+            }
+          ]
+        }
+      }
+    }
+  )

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 "<participantId>|<capabilityCode>" to a call plan
 #   2. one mapping file per action, published at the URL that row names
@@ -233,14 +238,33 @@ 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
       # problem and is not.
       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
 
 # ----------------------------------------------------------------------------

install/build-plugins.sh

@@ -33,6 +33,7 @@ plugins=(
     "oanregistry"
     "jsonmapper"
     "weather"
+    "mandi"
     "manifestloader"
     "reqpreprocessor"
     "otelsetup"

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

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)
+}