feat: cross-network calls [OpenAgriNet/network-adapter#78]

core/module/handler/stdHandler.go

@@ -339,6 +339,7 @@ func (h *stdHandler) stepCtx(r *http.Request, rh http.Header) (*model.StepContex
 		Body:                 body,
 		Role:                 h.role,
 		SubID:                subID,
+		OwnSubscriberID:      h.SubscriberID,
 		RespHeader:           rh,
 		ProtocolVersion:      protocolVersion,
 		MessageID:            messageID,

pkg/model/admittednetwork.go

@@ -1,52 +1,99 @@
 package model
 
-import "time"
+import (
+	"strings"
+	"time"
+)
+
+// A schema pack names its attributes document; a message is interpreted
+// against the context document beside it.
+const (
+	packFile    = "attributes.yaml"
+	contextFile = "context.jsonld"
+)
+
+// Agreement is what we agreed with another network. A direction is agreed when
+// it is present; our own entry, and every provider and upstream, has neither.
+type Agreement struct {
+	Source    *FederationDirection `json:"source,omitempty"`
+	Requester *FederationDirection `json:"requester,omitempty"`
+}
 
 // AdmittedNetwork is another network we admitted, as our registry holds it.
-// Unlike Subscription, it carries what we agreed with that network.
 type AdmittedNetwork struct {
 	ID           string
 	DiscoveryURL string
 
 	// How long this network lets us keep what we crawled. Zero forbids caching.
 	ProjectionTtl time.Duration
 
-	// A direction is agreed when it is present. Our own entry, and every
-	// provider and upstream, carries neither.
-	Source    *FederationDirection
-	Requester *FederationDirection
+	Agreement
 }
 
 // FederationDirection is one agreed direction. SchemaPacks is what the
 // agreement covers: for a source what we take, for a requester what they may ask.
 type FederationDirection struct {
-	SchemaPacks []string
+	SchemaPacks []string `json:"schema_packs,omitempty"`
 }
 
+// Agreed reports whether this is a network we admitted at all. False is an
+// ordinary participant, which no federation rule applies to.
+func (a Agreement) Agreed() bool { return a.Source != nil || a.Requester != nil }
+
 // IsSource reports whether we agreed to read from this network.
-func (n AdmittedNetwork) IsSource() bool { return n.Source != nil }
+func (a Agreement) IsSource() bool { return a.Source != nil }
 
 // IsRequester reports whether we agreed this network may read from us.
-func (n AdmittedNetwork) IsRequester() bool { return n.Requester != nil }
+func (a Agreement) IsRequester() bool { return a.Requester != nil }
 
 // SourcePacks is what we take from this network, empty if we read nothing.
-func (n AdmittedNetwork) SourcePacks() []string { return n.Source.packs() }
+func (a Agreement) SourcePacks() []string { return a.Source.packs() }
 
 // RequesterPacks is what this network may ask us for, empty if it may not ask.
-func (n AdmittedNetwork) RequesterPacks() []string { return n.Requester.packs() }
+func (a Agreement) RequesterPacks() []string { return a.Requester.packs() }
 
 // Directions names what we agreed, for a log line. Empty means nothing.
-func (n AdmittedNetwork) Directions() []string {
+func (a Agreement) Directions() []string {
 	var agreed []string
-	if n.IsSource() {
+	if a.IsSource() {
 		agreed = append(agreed, "source")
 	}
-	if n.IsRequester() {
+	if a.IsRequester() {
 		agreed = append(agreed, "requester")
 	}
 	return agreed
 }
 
+// SourceContexts is what we take, as the context documents a discover names.
+func (a Agreement) SourceContexts() []string { return contextsOf(a.SourcePacks()) }
+
+// RequesterContexts is what this network may ask us for, named the same way.
+func (a Agreement) RequesterContexts() []string { return contextsOf(a.RequesterPacks()) }
+
+func contextsOf(packs []string) []string {
+	contexts := make([]string, 0, len(packs))
+	for _, pack := range packs {
+		if trimmed := strings.TrimSpace(pack); trimmed != "" {
+			contexts = append(contexts, contextOf(trimmed))
+		}
+	}
+	return contexts
+}
+
+// contextOf is the context document beside a pack. A pack URL names the
+// attributes document, but an agreement recorded with the context document
+// already is taken as it stands rather than having a second one appended.
+func contextOf(pack string) string {
+	if strings.HasSuffix(pack, contextFile) {
+		return pack
+	}
+	base := strings.TrimSuffix(pack, packFile)
+	if !strings.HasSuffix(base, "/") {
+		base += "/"
+	}
+	return base + contextFile
+}
+
 func (d *FederationDirection) packs() []string {
 	if d == nil {
 		return nil

pkg/model/admittednetwork_test.go

@@ -1,6 +1,9 @@
 package model
 
-import "testing"
+import (
+	"slices"
+	"testing"
+)
 
 func TestADirectionIsAgreedWhenItIsPresent(t *testing.T) {
 	t.Parallel()
@@ -12,9 +15,9 @@ func TestADirectionIsAgreedWhenItIsPresent(t *testing.T) {
 		isSource    bool
 		isRequester bool
 	}{
-		"source only":    {AdmittedNetwork{Source: packs}, true, false},
-		"requester only": {AdmittedNetwork{Requester: packs}, false, true},
-		"both":           {AdmittedNetwork{Source: packs, Requester: packs}, true, true},
+		"source only":    {AdmittedNetwork{Agreement: Agreement{Source: packs}}, true, false},
+		"requester only": {AdmittedNetwork{Agreement: Agreement{Requester: packs}}, false, true},
+		"both":           {AdmittedNetwork{Agreement: Agreement{Source: packs, Requester: packs}}, true, true},
 		"neither":        {AdmittedNetwork{}, false, false},
 	} {
 		t.Run(name, func(t *testing.T) {
@@ -31,10 +34,10 @@ func TestADirectionIsAgreedWhenItIsPresent(t *testing.T) {
 func TestThePacksWeTakeComeFromTheSourceDirection(t *testing.T) {
 	t.Parallel()
 
-	network := AdmittedNetwork{
+	network := AdmittedNetwork{Agreement: Agreement{
 		Source:    &FederationDirection{SchemaPacks: []string{"mandi"}},
 		Requester: &FederationDirection{SchemaPacks: []string{"weather"}},
-	}
+	}}
 
 	if got := network.SourcePacks(); len(got) != 1 || got[0] != "mandi" {
 		t.Errorf("SourcePacks() = %v, want what we take, not what we give", got)
@@ -47,9 +50,90 @@ func TestThePacksWeTakeComeFromTheSourceDirection(t *testing.T) {
 func TestAnUnagreedDirectionHasNoPacks(t *testing.T) {
 	t.Parallel()
 
-	network := AdmittedNetwork{Source: &FederationDirection{SchemaPacks: []string{"mandi"}}}
+	network := AdmittedNetwork{Agreement: Agreement{Source: &FederationDirection{SchemaPacks: []string{"mandi"}}}}
 
 	if got := network.RequesterPacks(); got != nil {
 		t.Errorf("RequesterPacks() = %v, want none -- they may not read from us at all", got)
 	}
 }
+
+func TestAnAgreementReadsAsTheContextsADiscoverNames(t *testing.T) {
+	t.Parallel()
+
+	a := Agreement{
+		Source:    &FederationDirection{SchemaPacks: []string{"https://x/MandiPrice/v0.1/attributes.yaml"}},
+		Requester: &FederationDirection{SchemaPacks: []string{"https://x/Weather/v0.1/attributes.yaml", "  "}},
+	}
+
+	if got := a.SourceContexts(); len(got) != 1 || got[0] != "https://x/MandiPrice/v0.1/context.jsonld" {
+		t.Errorf("SourceContexts() = %v, want the context document beside the pack", got)
+	}
+	if got := a.RequesterContexts(); len(got) != 1 || got[0] != "https://x/Weather/v0.1/context.jsonld" {
+		t.Errorf("RequesterContexts() = %v, want the blank entry dropped", got)
+	}
+}
+
+func TestAnUnagreedDirectionNamesNoContexts(t *testing.T) {
+	t.Parallel()
+
+	if got := (Agreement{}).RequesterContexts(); len(got) != 0 {
+		t.Errorf("RequesterContexts() = %v, want none", got)
+	}
+}
+
+func TestOnlyAnAdmittedNetworkHasAnAgreement(t *testing.T) {
+	t.Parallel()
+
+	packs := &FederationDirection{SchemaPacks: []string{"mandi"}}
+
+	for name, tc := range map[string]struct {
+		agreement  Agreement
+		agreed     bool
+		directions []string
+	}{
+		"source only":    {Agreement{Source: packs}, true, []string{"source"}},
+		"requester only": {Agreement{Requester: packs}, true, []string{"requester"}},
+		"both":           {Agreement{Source: packs, Requester: packs}, true, []string{"source", "requester"}},
+		"a provider":     {Agreement{}, false, nil},
+	} {
+		t.Run(name, func(t *testing.T) {
+			if got := tc.agreement.Agreed(); got != tc.agreed {
+				t.Errorf("Agreed() = %v, want %v", got, tc.agreed)
+			}
+			if got := tc.agreement.Directions(); !slices.Equal(got, tc.directions) {
+				t.Errorf("Directions() = %v, want %v", got, tc.directions)
+			}
+		})
+	}
+}
+
+func TestAPackAlreadyNamingItsContextIsLeftAlone(t *testing.T) {
+	t.Parallel()
+
+	for name, tc := range map[string]struct{ pack, want string }{
+		"attributes": {
+			"https://x/MandiPrice/v0.1/attributes.yaml",
+			"https://x/MandiPrice/v0.1/context.jsonld",
+		},
+		"already the context": {
+			"https://x/MandiPrice/v0.1/context.jsonld",
+			"https://x/MandiPrice/v0.1/context.jsonld",
+		},
+		"trailing slash": {
+			"https://x/MandiPrice/v0.1/",
+			"https://x/MandiPrice/v0.1/context.jsonld",
+		},
+		"no filename": {
+			"https://x/MandiPrice/v0.1",
+			"https://x/MandiPrice/v0.1/context.jsonld",
+		},
+	} {
+		t.Run(name, func(t *testing.T) {
+			a := Agreement{Source: &FederationDirection{SchemaPacks: []string{tc.pack}}}
+			got := a.SourceContexts()
+			if len(got) != 1 || got[0] != tc.want {
+				t.Errorf("SourceContexts(%q) = %v, want [%s]", tc.pack, got, tc.want)
+			}
+		})
+	}
+}

pkg/model/model.go

@@ -33,6 +33,10 @@ type Subscription struct {
 	Updated            time.Time `json:"updated,omitzero" format:"date-time"`
 	Nonce              string    `json:"nonce,omitzero"`
 	NetworkMemberships []string  `json:"network_memberships,omitempty"`
+
+	// Agreement is set only on a network we admitted. It says which way round we
+	// agreed and what the agreement covers; every other participant has none.
+	Agreement Agreement `json:"agreement,omitzero"`
 }
 
 // nonUsableKeySubscriptionStatuses are the Subscription.Status values that mean a
@@ -390,10 +394,14 @@ type Keyset struct {
 // StepContext holds context information for a request processing step.
 type StepContext struct {
 	context.Context
-	Request              *http.Request
-	Body                 []byte
-	Route                *Route
-	SubID                string
+	Request *http.Request
+	Body    []byte
+	Route   *Route
+	SubID   string
+
+	// OwnSubscriberID is who THIS module is, from its own config. SubID comes
+	// from the body, so on a cross-network request SubID is THEM.
+	OwnSubscriberID      string
 	Role                 Role
 	RespHeader           http.Header
 	ProtocolVersion      string // Protocol version parsed from context.version (e.g. "2.0.0")

pkg/plugin/implementation/AgricultureFacility/live_test.go

@@ -208,7 +208,7 @@ func TestLiveAgainstPocra(t *testing.T) {
 	for _, facilityType := range governedTypes {
 		t.Run(facilityType, func(t *testing.T) {
 			captured = nil
-			stepCtx := &model.StepContext{Context: t.Context(), Body: []byte(liveRequest(t, facilityType))}
+			stepCtx := &model.StepContext{Context: t.Context(), OwnSubscriberID: "oan.test", Body: []byte(liveRequest(t, facilityType))}
 			started := time.Now()
 			if err := step.Run(stepCtx); err != nil {
 				t.Fatalf("Run() returned an unexpected error: %v", err)
@@ -360,7 +360,7 @@ func TestLiveCategoryLeakIsFiltered(t *testing.T) {
 		}
 
 		captured = nil
-		stepCtx := &model.StepContext{Context: t.Context(), Body: body}
+		stepCtx := &model.StepContext{Context: t.Context(), OwnSubscriberID: "oan.test", Body: body}
 		if err := step.Run(stepCtx); err != nil {
 			t.Fatalf("%s: Run() returned an unexpected error: %v", facilityType, err)
 		}

pkg/plugin/implementation/AgricultureFacility/mappings_test.go

@@ -78,7 +78,7 @@ const declaredContext = "https://raw.githubusercontent.com/OpenAgriNet/network-s
 const callerMessageID = "a1b2c3d4-e5f6-4789-abcd-ef1234567890"
 
 const selectRequest = `{
-  "context": { "version": "2.0.0", "action": "select",
+  "context": { "version": "2.0.0", "action": "select", "receiverId": "oan.test",
     "networkId": "oan-dev",
     "transactionId": "f9e8d7c6-b5a4-4321-9876-543210fedcba",
     "messageId": "a1b2c3d4-e5f6-4789-abcd-ef1234567890",
@@ -268,7 +268,7 @@ func runSelect(t *testing.T, request string) (sent map[string]any, answer map[st
 	}
 	defer closeStep()
 
-	stepCtx := &model.StepContext{Context: t.Context(), Body: []byte(request)}
+	stepCtx := &model.StepContext{Context: t.Context(), OwnSubscriberID: "oan.test", Body: []byte(request)}
 	runErr = step.Run(stepCtx)
 	if runErr != nil {
 		return sent, nil, runErr
@@ -922,7 +922,7 @@ func runAgainst(t *testing.T, request, providerBody string) map[string]any {
 	}
 	defer closeStep()
 
-	stepCtx := &model.StepContext{Context: t.Context(), Body: []byte(request)}
+	stepCtx := &model.StepContext{Context: t.Context(), OwnSubscriberID: "oan.test", Body: []byte(request)}
 	if err := step.Run(stepCtx); err != nil {
 		t.Fatalf("Run() returned an unexpected error: %v", err)
 	}
@@ -1405,7 +1405,7 @@ func runSplitSearch(t *testing.T, types []string, byCode map[string]string) (map
 		t.Fatalf("could not rebuild the payload: %v", err)
 	}
 
-	stepCtx := &model.StepContext{Context: t.Context(), Body: body}
+	stepCtx := &model.StepContext{Context: t.Context(), OwnSubscriberID: "oan.test", Body: body}
 	if err := step.Run(stepCtx); err != nil {
 		t.Fatalf("Run() returned an unexpected error: %v", err)
 	}

pkg/plugin/implementation/AgricultureFacility/search_test.go

@@ -77,7 +77,7 @@ func runSearchTimed(t *testing.T, types []string, cfg *AgricultureFacility.Confi
 		t.Fatalf("could not rebuild the payload: %v", err)
 	}
 
-	return step.Run(&model.StepContext{Context: t.Context(), Body: body})
+	return step.Run(&model.StepContext{Context: t.Context(), OwnSubscriberID: "oan.test", Body: body})
 }
 
 // A search's calls are sequential unless the deployment raised the limit.

pkg/plugin/implementation/KnowledgeAdvisory/live_test.go

@@ -120,7 +120,7 @@ func TestLiveEndToEnd(t *testing.T) {
 
 	// Two requests: the second must reuse the token rather than exchange again.
 	for i := 1; i <= 2; i++ {
-		ctx := &model.StepContext{Context: t.Context(), Body: []byte(selectRequest)}
+		ctx := &model.StepContext{Context: t.Context(), OwnSubscriberID: "oan.test", Body: []byte(selectRequest)}
 		if err := step.Run(ctx); err != nil {
 			t.Fatalf("request %d: %v", i, err)
 		}

pkg/plugin/implementation/KnowledgeAdvisory/mappings_test.go

@@ -43,7 +43,7 @@ const shippedBindingKey = "knowledge-provider|" + shippedCapability
 // topics to search for and the languages it can read, and no advisory of its
 // own -- the pack's OnDemand branch requires exactly those two.
 const selectRequest = `{
-  "context": { "version": "2.0.0", "action": "select",
+  "context": { "version": "2.0.0", "action": "select", "receiverId": "oan.test",
     "networkId": "oan-dev",
     "transactionId": "9f2c1a8e-4b70-4d31-9c55-6f2e0b1d7a44",
     "messageId": "7d41b9e0-52a6-4c18-8b73-1e9f0a4c6d22",
@@ -226,7 +226,7 @@ func runShippedWith(t *testing.T, request, providerBody string) (map[string]any,
 	}
 	defer closeStep()
 
-	stepCtx := &model.StepContext{Context: t.Context(), Body: []byte(request)}
+	stepCtx := &model.StepContext{Context: t.Context(), OwnSubscriberID: "oan.test", Body: []byte(request)}
 	if err := step.Run(stepCtx); err != nil {
 		t.Fatalf("Run() returned an unexpected error: %v", err)
 	}