fix(composition): apply subscription filters on union and interface return types

[jensneuse]

Summary

• Composition silently dropped @openfed__subscriptionFilter whenever the subscription field returned a union or interface (federation-factory.ts:2839 gated on OBJECT_TYPE_DEFINITION and skipped without an error). Router then forwarded every event because no filter config was emitted — see Pylon #2913 / Linear ENG-9404.
• Walks every accessible concrete type via the existing concreteTypeNamesByAbstractTypeName index, validates the condition against each, and emits a single filter config when all walks succeed. @inaccessible members and implementers are skipped before any field-presence check; per-target failures produce a member-tagged composition error.
• Adds composition unit tests (success + per-target error + inaccessible-skip) and Kafka integration tests covering both members of a union, both implementers of an interface, and the case where an event with a non-member __typename produces an INVALID_GRAPHQL error frame.

Test plan

• pnpm --filter=@wundergraph/composition test — 1043 pass, 18 skipped
• cd router-tests && go test -run TestKafkaEvents -count=1 ./events/... — pass in 6.3s
• go test -race -run 'TestKafkaEvents/subscribe_async_with_filter$/(union|interface)_return_type' ./events/... — clean
• CI green on this PR

🤖 Generated with Claude Code

Summary by CodeRabbit

• New Features

  • Subscription filters now support union and interface return types; validation is performed per accessible concrete member/implementer and aggregated.

• Bug Fixes / Errors

  • Improved user-facing error messages for invalid subscription-filter conditions and for abstract types with no accessible concrete targets; runtime invalid-event errors surface as INVALID_GRAPHQL.

• Documentation

  • Directive docs updated with constraints, examples, @inaccessible handling, and runtime behavior for unions/interfaces.

• Demo

  • Added Kafka-backed, tag-filtered employee event subscription demo.

• Tests

  • Expanded tests for composition validation and runtime delivery of union/interface subscriptions.

[claude]

Claude Code Review

This repository is configured for manual code reviews. Comment @claude review to trigger a review and subscribe this PR to future pushes, or @claude review once for a one-time review.

_{Tip: disable this comment in your organization's Code Review settings.}

[coderabbitai]

Note

Reviews paused

It looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the reviews.auto_review.auto_pause_after_reviewed_commits setting.

Use the following commands to manage reviews:

• @coderabbitai resume to resume automatic reviews.
• @coderabbitai review to trigger a single review.

Use the checkboxes below for quick actions:

• ▶️ Resume reviews
• 🔍 Trigger review

Walkthrough

Adds target-aware validation for @openfed__subscriptionFilter on union and interface return types, three new subscription-filter error-message builders, composition and router tests for unions/interfaces and invalid __typename handling, plus demo subgraph/schema and config entries.

Changes

Subscription-filter target-aware validation and supporting artifacts

Layer / File(s)	Summary
Error messages / Data shape composition/src/errors/errors.ts	Adds three exported message-builder functions: subscriptionFilterUnionMemberInvalidErrorMessage, subscriptionFilterInterfaceImplementerInvalidErrorMessage, and subscriptionFilterNoAccessibleConcreteTypesErrorMessage.
Core validation helper composition/src/v1/federation/federation-factory.ts	Adds validateSubscriptionFilterForTarget(target: ObjectDefinitionData) to validate a directive's single-object argument against a concrete object, resetting per-target depth state and returning { condition?: SubscriptionCondition; errors: string[] }.
Concrete-target discovery composition/src/v1/federation/federation-factory.ts	Adds collectSubscriptionFilterConcreteTargets(abstractTypeName: string): ObjectDefinitionData[] to enumerate accessible concrete object types backing a union/interface, excluding @inaccessible members.
Validation orchestration / wiring composition/src/v1/federation/federation-factory.ts	Refactors validateSubscriptionFilterAndGenerateConfiguration and validateSubscriptionFiltersAndGenerateConfiguration to: fast-path object types; for unions/interfaces, collect concrete targets, error if none, validate per-target, wrap per-target failures with the new message builders (union-member vs implementer), aggregate errors, and only persist subscriptionFilterCondition when all targets validate.
Composition tests composition/tests/v1/directives/subscription-filter.test.ts	Extends tests for union/interface return types: success paths, per-target failure expectations with wrapped messages, and skipping validation for @inaccessible members.
Integration tests (router) router-tests/events/kafka_events_test.go	Adds WebSocket tests asserting tag-filtered delivery for union/interface subscriptions, behavior when event __typename is not a union member/implementer (router emits INVALID_GRAPHQL error naming the typename), and normalizes internal subgraph IDs in error assertions via subgraphIDPattern.
Demo schemas & config demo/graph.yaml, demo/pkg/subgraphs/employee-events/subgraph/schema.graphqls, demo/pkg/subgraphs/employeeupdated/subgraph/schema.graphqls	Adds employee-events subgraph entry and federated entity types (EmployeeUpdated, EmployeeDeleted, EmployeeChanged, EmployeeRemoved), a union EmployeeEvent, an interface EmployeeChange, and subscription fields using tag filtering.
Documentation docs-website/federation/directives/openfed__subscriptionfilter.mdx	Documents supported return kinds (object, union, interface), requirements to resolve fieldPath for each accessible concrete type, exclusion of @inaccessible members, runtime behavior for unknown __typename, and that composition emits one router configuration per concrete type validation.

Estimated code review effort

🎯 3 (Moderate) | ⏱️ ~25 minutes

Possibly related PRs

• wundergraph/cosmo#2764: Changes to FederationFactory and abstract-type handling that overlap with per-concrete-type validation and mapping logic used by subscription-filter validation.

🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 0.00% which is insufficient. The required threshold is 80.00%.	Write docstrings for the functions missing them to satisfy the coverage threshold.

✅ Passed checks (4 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title directly and clearly describes the main change: extending subscription filter support to union and interface return types, which was the core issue being fixed.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

Tip

💬 Introducing Slack Agent: The best way for teams to turn conversations into code.

Slack Agent is built on CodeRabbit's deep understanding of your code, so your team can collaborate across the entire SDLC without losing context.

• Generate code and open pull requests
• Plan features and break down work
• Investigate incidents and troubleshoot customer tickets together
• Automate recurring tasks and respond to alerts with triggers
• Summarize progress and report instantly

Built for teams:

• Shared memory across your entire org—no repeating context
• Per-thread sandboxes to safely plan and execute work
• Governance built-in—scoped access, auditability, and budget controls

One agent for your entire SDLC. Right inside Slack.

👉 Get started

---

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[codecov]

Codecov Report

❌ Patch coverage is 74.51737% with 66 lines in your changes missing coverage. Please review.
✅ Project coverage is 46.35%. Comparing base (893ff72) to head (f3de52e).
⚠️ Report is 1 commits behind head on main.

Files with missing lines	Patch %	Lines
...omposition/src/v1/federation/federation-factory.ts	77.83%	43 Missing ⚠️
composition/src/errors/errors.ts	64.06%	23 Missing ⚠️

❌ Your patch check has failed because the patch coverage (74.51%) is below the target coverage (90.00%). You can increase the patch coverage or adjust the target coverage.

Additional details and impacted files

@@             Coverage Diff             @@
##             main    #2797       +/-   ##
===========================================
- Coverage   65.01%   46.35%   -18.66%     
===========================================
  Files         573     1084      +511     
  Lines       71938   145677    +73739     
  Branches     4862     9335     +4473     
===========================================
+ Hits        46767    67535    +20768     
- Misses      23724    76396    +52672     
- Partials     1447     1746      +299     

Files with missing lines	Coverage Δ
composition/src/federation/types/results.ts	100.00% <100.00%> (ø)
composition/src/errors/errors.ts	81.90% <64.06%> (ø)
...omposition/src/v1/federation/federation-factory.ts	89.22% <77.83%> (ø)

... and 513 files with indirect coverage changes

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[github-actions]

Router-nonroot image scan passed

✅ No security vulnerabilities found in image:

ghcr.io/wundergraph/cosmo/router:sha-632617c115f65cc62ebddbc78b2f651d062d8d3b-nonroot

[coderabbitai]

🧹 Nitpick comments (4)

composition/tests/v1/directives/subscription-filter.test.ts (1)

459-500: Assert that only one matching config is emitted.

find(...) only proves the config exists once in the assertion path. If composition starts emitting the same subscription filter twice, these tests would still pass. Tighten the check to assert the filtered config list has exactly one entry.

Suggested assertion shape

- const subscriptionField = result.fieldConfigurations.find(
-   (fc) => fc.typeName === SUBSCRIPTION && fc.fieldName === 'onTaskEvent',
- );
- expect(subscriptionField).toStrictEqual({
+ const subscriptionFields = result.fieldConfigurations.filter(
+   (fc) => fc.typeName === SUBSCRIPTION && fc.fieldName === 'onTaskEvent',
+ );
+ expect(subscriptionFields).toStrictEqual([{
    argumentNames: ['phoneChannelId'],
    fieldName: 'onTaskEvent',
    typeName: SUBSCRIPTION,
    subscriptionFilterCondition: {
      in: {
        fieldPath: ['phoneChannelId'],
        values: ['{{ args.phoneChannelId }}'],
      },
    },
- });
+ }]);

Also applies to: 481-500

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@composition/tests/v1/directives/subscription-filter.test.ts` around lines 459
- 500, The test currently uses find(...) to locate the subscription field
(subscriptionField) which only proves existence; change the lookup to
filter(...) on result.fieldConfigurations for entries where fc.typeName ===
SUBSCRIPTION && fc.fieldName === 'onTaskEvent', then assert the filtered array
has length 1 and assert the single element equals the expected config; update
both tests referencing subscriptionField (the one for union and the one for
interface) to use this filtered-list + length === 1 check before the strict
equality assertion.

router-tests/events/kafka_events_test.go (1)

1087-1092: Use testenv.WSWriteJSON for the new subscription setup writes.

These new blocks still call conn.WriteJSON, which skips the retry/backoff behavior the router-test helpers provide. Switching to testenv.WSWriteJSON will make the union/interface subscription tests less timing-sensitive.

Suggested replacement

- require.NoError(t, conn.WriteJSON(&testenv.WebSocketMessage{
+ require.NoError(t, testenv.WSWriteJSON(t, conn, &testenv.WebSocketMessage{

Also applies to: 1161-1166, 1234-1239, 1294-1299

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@router-tests/events/kafka_events_test.go` around lines 1087 - 1092, Replace
direct conn.WriteJSON calls with the test helper testenv.WSWriteJSON so the
retry/backoff logic is used for subscription setup; specifically, in the blocks
that create the WebSocket via xEnv.InitGraphQLWebSocketConnection and then call
conn.WriteJSON(&testenv.WebSocketMessage{...}) (e.g., the instance at the shown
diff and the other occurrences at the ranges noted) switch those calls to
testenv.WSWriteJSON(conn, t, &testenv.WebSocketMessage{...}) (keeping the same
ID, Type and Payload fields) so the union/interface subscription tests use the
router-test helper's timing resilience.

composition/src/v1/federation/federation-factory.ts (2)

2809-2816: Add an explicit void return type to this helper.

This new method relies on inference even though the repo asks for explicit TypeScript return types.

Suggested change

   validateSubscriptionFilterAndGenerateConfiguration(
     directiveNode: ConstDirectiveNode,
     objectData: ObjectDefinitionData,
     fieldPath: string,
     fieldName: string,
     parentTypeName: string,
     directiveSubgraphName: string,
-  ) {
+  ): void {

As per coding guidelines "Use explicit type annotations for function parameters and return types in TypeScript".

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@composition/src/v1/federation/federation-factory.ts` around lines 2809 -
2816, The helper method validateSubscriptionFilterAndGenerateConfiguration
currently relies on inferred return type; update its signature to include an
explicit TypeScript return annotation of void (i.e., add ": void" after the
parameter list) so it complies with the project's rule for explicit return
types, and confirm there are no returned values inside the body (or
remove/adjust any return statements) to match the void annotation; reference the
function validateSubscriptionFilterAndGenerateConfiguration and its parameter
types (ConstDirectiveNode, ObjectDefinitionData) when making the change.

---

2836-2849: Consider returning concrete targets in a stable order.

concreteTypeNamesByAbstractTypeName is a Set, so the aggregated error order here depends on insertion order. Sorting out by target.name before returning would make member-tagged errors and the chosen first success deterministic across runs.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@composition/src/v1/federation/federation-factory.ts` around lines 2836 -
2849, collectSubscriptionFilterConcreteTargets currently iterates a Set
(concreteTypeNamesByAbstractTypeName) and returns results in insertion order;
make the return order deterministic by sorting the collected
ObjectDefinitionData array (out) by the concrete type name before returning.
Locate collectSubscriptionFilterConcreteTargets and, after populating out, sort
it by the type name field on each ObjectDefinitionData (use the name property on
the returned data) so member-tagged errors and first-success selection are
stable across runs.

🤖 Prompt for all review comments with AI agents

Verify each finding against the current code and only fix it if needed.

Nitpick comments:
In `@composition/src/v1/federation/federation-factory.ts`:
- Around line 2809-2816: The helper method
validateSubscriptionFilterAndGenerateConfiguration currently relies on inferred
return type; update its signature to include an explicit TypeScript return
annotation of void (i.e., add ": void" after the parameter list) so it complies
with the project's rule for explicit return types, and confirm there are no
returned values inside the body (or remove/adjust any return statements) to
match the void annotation; reference the function
validateSubscriptionFilterAndGenerateConfiguration and its parameter types
(ConstDirectiveNode, ObjectDefinitionData) when making the change.
- Around line 2836-2849: collectSubscriptionFilterConcreteTargets currently
iterates a Set (concreteTypeNamesByAbstractTypeName) and returns results in
insertion order; make the return order deterministic by sorting the collected
ObjectDefinitionData array (out) by the concrete type name before returning.
Locate collectSubscriptionFilterConcreteTargets and, after populating out, sort
it by the type name field on each ObjectDefinitionData (use the name property on
the returned data) so member-tagged errors and first-success selection are
stable across runs.

In `@composition/tests/v1/directives/subscription-filter.test.ts`:
- Around line 459-500: The test currently uses find(...) to locate the
subscription field (subscriptionField) which only proves existence; change the
lookup to filter(...) on result.fieldConfigurations for entries where
fc.typeName === SUBSCRIPTION && fc.fieldName === 'onTaskEvent', then assert the
filtered array has length 1 and assert the single element equals the expected
config; update both tests referencing subscriptionField (the one for union and
the one for interface) to use this filtered-list + length === 1 check before the
strict equality assertion.

In `@router-tests/events/kafka_events_test.go`:
- Around line 1087-1092: Replace direct conn.WriteJSON calls with the test
helper testenv.WSWriteJSON so the retry/backoff logic is used for subscription
setup; specifically, in the blocks that create the WebSocket via
xEnv.InitGraphQLWebSocketConnection and then call
conn.WriteJSON(&testenv.WebSocketMessage{...}) (e.g., the instance at the shown
diff and the other occurrences at the ranges noted) switch those calls to
testenv.WSWriteJSON(conn, t, &testenv.WebSocketMessage{...}) (keeping the same
ID, Type and Payload fields) so the union/interface subscription tests use the
router-test helper's timing resilience.

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Run ID: a0ec1613-f12a-4bb7-8737-408d89e66940

📥 Commits

Reviewing files that changed from the base of the PR and between a6a6956 and 609b86c.

📒 Files selected for processing (8)

• composition/src/errors/errors.ts
• composition/src/v1/federation/federation-factory.ts
• composition/tests/v1/directives/subscription-filter.test.ts
• demo/graph.yaml
• demo/pkg/subgraphs/employee-events/subgraph/schema.graphqls
• demo/pkg/subgraphs/employeeupdated/subgraph/schema.graphqls
• router-tests/events/kafka_events_test.go
• router-tests/testenv/testdata/configWithEdfs.json