fix(realtime): scope sse streams by authorized route (#786)

* fix(realtime): scope sse streams by authorized route

* fix(realtime): preserve scoped and broadcast streams

Author: cssbruno

internal/appgen/appgen_test.go

@@ -1153,8 +1153,10 @@ func TestGenerateWritesRealtimeFanoutForSubscriptions(t *testing.T) {
 		`var realtimeFanout gowdkrealtime.PresentationFanout = gowdkrealtime.NewSSE()`,
 		`func RegisterRealtimeFanout(fanout gowdkrealtime.PresentationFanout)`,
 		`"example.com/app/contracts/patients.PatientNotice": true`,
-		`event.Category == gowdkcontracts.PresentationEvent`,
-		`gowdkcontracts.PresentationFanoutCommandEventSink(realtimeSubscriptionFanout{inner: fanout})`,
+		`var realtimeSubscriptionBroadcasts map[string]bool = map[string]bool{"example.com/app/contracts/patients.PatientNotice": true}`,
+		`realtimeAudienceScopedEvents(event)...`,
+		`scopedFanout := realtimeSubscriptionFanout{inner: fanout}`,
+		`gowdkcontracts.PresentationFanoutCommandEventSink(scopedFanout)`,
 		`gowdkcontracts.CompositeCommandEventSink(gowdkcontracts.InProcessCommandEventSink(), fanoutSink)`,
 	} {
 		if !strings.Contains(source, expected) {
@@ -1210,8 +1212,11 @@ func TestGenerateWritesRealtimeQueryInvalidationFanout(t *testing.T) {
 	for _, expected := range []string{
 		`gowdkcontracts.QueryInvalidationPresentationEventType: true`,
 		`var realtimeQueryInvalidations []gowdkcontracts.QueryInvalidation = []gowdkcontracts.QueryInvalidation{gowdkcontracts.QueryInvalidation{EventCategory: gowdkcontracts.DomainEvent, EventType: "example.com/app/contracts/patients.PatientCreated", QueryType: "example.com/app/contracts/patients.GetPatientPage"}}`,
-		`gowdkcontracts.QueryInvalidationCommandEventSink(fanout, realtimeQueryInvalidations)`,
-		`gowdkcontracts.CompositeCommandEventSink(gowdkcontracts.InProcessCommandEventSink(), gowdkcontracts.QueryInvalidationCommandEventSink(fanout, realtimeQueryInvalidations), fanoutSink)`,
+		`scopedFanout := realtimeSubscriptionFanout{inner: fanout}`,
+		`gowdkcontracts.QueryInvalidationCommandEventSink(scopedFanout, realtimeQueryInvalidations)`,
+		`gowdkcontracts.CompositeCommandEventSink(gowdkcontracts.InProcessCommandEventSink(), gowdkcontracts.QueryInvalidationCommandEventSink(scopedFanout, realtimeQueryInvalidations), fanoutSink)`,
+		`var realtimeQueryBroadcasts map[string]bool = map[string]bool{"example.com/app/contracts/patients.GetPatientPage": true}`,
+		`unscopedQueries := []string{}`,
 		// Single-flight write path: the command adapter tells the submitting
 		// client which g:query regions to refresh via the X-GOWDK-Queries header.
 		`invalidatedQueries := gowdkcontracts.InvalidatedQueryTypes(realtimeQueryInvalidations, events)`,
@@ -1234,6 +1239,63 @@ func TestGenerateWritesRealtimeQueryInvalidationFanout(t *testing.T) {
 	}
 }
 
+func TestRealtimeBroadcastsTrackMixedScopedAndUnscopedOwners(t *testing.T) {
+	options := Options{IR: &gwdkir.Program{
+		Pages: []gwdkir.Page{{ID: "patients", Route: "/patients"}},
+		RealtimeSubscriptions: []gwdkir.RealtimeSubscription{{
+			Query:           "patients.GetPatientPage",
+			Event:           "patients.PatientNotice",
+			EventImportPath: "example.com/app/contracts/patients",
+			EventType:       "PatientNotice",
+			Status:          gwdkir.ContractBindingBound,
+			OwnerKind:       gwdkir.SourcePage,
+			OwnerID:         "patients",
+		}, {
+			Query:           "cards.GetPatientCard",
+			Event:           "patients.PatientNotice",
+			EventImportPath: "example.com/app/contracts/patients",
+			EventType:       "PatientNotice",
+			Status:          gwdkir.ContractBindingBound,
+			OwnerKind:       gwdkir.SourceComponent,
+			OwnerID:         "PatientCard",
+		}},
+		QueryInvalidations: []gwdkir.QueryInvalidation{{
+			Query:         "patients.GetPatientPage",
+			QueryType:     "example.com/app/contracts/patients.GetPatientPage",
+			Event:         "example.com/app/contracts/patients.PatientCreated",
+			EventType:     "example.com/app/contracts/patients.PatientCreated",
+			EventCategory: "domain",
+			Status:        gwdkir.ContractBindingBound,
+			OwnerKind:     gwdkir.SourcePage,
+			OwnerID:       "patients",
+		}, {
+			Query:         "cards.GetPatientCard",
+			QueryType:     "example.com/app/contracts/patients.GetPatientCard",
+			Event:         "example.com/app/contracts/patients.PatientCreated",
+			EventType:     "example.com/app/contracts/patients.PatientCreated",
+			EventCategory: "domain",
+			Status:        gwdkir.ContractBindingBound,
+			OwnerKind:     gwdkir.SourceComponent,
+			OwnerID:       "PatientCard",
+		}},
+	}}
+
+	subscriptionAudiences := realtimeSubscriptionAudiences(options)
+	if got := subscriptionAudiences["example.com/app/contracts/patients.PatientNotice"]; len(got) != 1 || got[0] != "gowdk.route.0" {
+		t.Fatalf("subscription audiences = %#v, want route audience", subscriptionAudiences)
+	}
+	if !realtimeSubscriptionBroadcasts(options)["example.com/app/contracts/patients.PatientNotice"] {
+		t.Fatalf("expected mixed subscription event type to retain broadcast delivery")
+	}
+	queryAudiences := realtimeQueryAudiences(options)
+	if got := queryAudiences["example.com/app/contracts/patients.GetPatientPage"]; len(got) != 1 || got[0] != "gowdk.route.0" {
+		t.Fatalf("query audiences = %#v, want route audience", queryAudiences)
+	}
+	if !realtimeQueryBroadcasts(options)["example.com/app/contracts/patients.GetPatientCard"] {
+		t.Fatalf("expected component-owned query invalidation to retain broadcast delivery")
+	}
+}
+
 func TestGenerateRegistersSingleFlightRegionRenderers(t *testing.T) {
 	root := t.TempDir()
 	outputDir := filepath.Join(root, "dist")
@@ -1553,11 +1615,16 @@ func TestGenerateGuardsRealtimeStreamForSubscribedPages(t *testing.T) {
 	for _, expected := range []string{
 		`neturl "net/url"`,
 		`gowdkroute "github.com/cssbruno/gowdk/runtime/route"`,
+		`gowdkrealtime.NewSSE(gowdkrealtime.WithSSEAudienceFromRequest(realtimeStreamAudience))`,
+		`var realtimeSubscriptionAudiences map[string][]string = map[string][]string{"example.com/app/contracts/patients.PatientNotice": []string{"gowdk.route.0"}}`,
+		`func realtimeStreamAudience(request *http.Request) []string`,
 		`func realtimeStreamGuards(request *http.Request) []string`,
-		`request.URL.Query().Get("path")`,
+		`func realtimeStreamPath(request *http.Request) string`,
 		`referer := request.Referer()`,
 		`neturl.Parse(referer)`,
+		`return refererURL.Path`,
 		`gowdkroute.Match("/dashboard", requestPath)`,
+		`return []string{"gowdk.route.0"}`,
 		`return []string{"auth.required"}`,
 		`if !runGuards(response, request, realtimeStreamGuards(request))`,
 		`RegisterGuards(GOWDKGuardRegistry())`,
@@ -1566,6 +1633,9 @@ func TestGenerateGuardsRealtimeStreamForSubscribedPages(t *testing.T) {
 			t.Fatalf("expected generated guarded realtime source to contain %q:\n%s", expected, source)
 		}
 	}
+	if strings.Contains(source, `Query().Get("path")`) {
+		t.Fatalf("generated realtime stream must not authorize from client query path:\n%s", source)
+	}
 }
 
 func TestBoundActionFieldDecodePanicsOnUnsupportedFieldType(t *testing.T) {
@@ -7152,6 +7222,10 @@ func TestGeneratedBinaryCommandSetsInvalidatedQueriesHeader(t *testing.T) {
 	writeTestFile(t, filepath.Join(outputDir, "patients", "index.html"), "<main>Patients page</main>")
 
 	program := &gwdkir.Program{
+		Pages: []gwdkir.Page{
+			{ID: "dashboard", Route: "/dashboard"},
+			{ID: "patients", Route: "/patients"},
+		},
 		ContractRefs: []gwdkir.ContractReference{{
 			Kind:        gwdkir.ContractCommand,
 			Name:        "patients.CreatePatient",
@@ -7479,8 +7553,13 @@ func TestGeneratedBinaryRealtimeFanoutStreamsSubscribedPresentationEvents(t *tes
 	appDir := filepath.Join(root, "generated-app")
 	binaryPath := filepath.Join(root, "site")
 	writeTestFile(t, filepath.Join(outputDir, "patients", "index.html"), "<main>Patients page</main>")
+	writeTestFile(t, filepath.Join(outputDir, "dashboard", "index.html"), "<main>Dashboard page</main>")
 
 	program := &gwdkir.Program{
+		Pages: []gwdkir.Page{
+			{ID: "dashboard", Route: "/dashboard"},
+			{ID: "patients", Route: "/patients"},
+		},
 		ContractRefs: []gwdkir.ContractReference{{
 			Kind:        gwdkir.ContractCommand,
 			Name:        "patients.CreatePatient",
@@ -7506,6 +7585,14 @@ func TestGeneratedBinaryRealtimeFanoutStreamsSubscribedPresentationEvents(t *tes
 			Status:          gwdkir.ContractBindingBound,
 			OwnerKind:       gwdkir.SourcePage,
 			OwnerID:         "patients",
+		}, {
+			Query:           "patients.GetDashboard",
+			Event:           "patients.OtherNotice",
+			EventImportPath: "gowdk-generated-app/patients",
+			EventType:       "OtherNotice",
+			Status:          gwdkir.ContractBindingBound,
+			OwnerKind:       gwdkir.SourcePage,
+			OwnerID:         "dashboard",
 		}},
 	}
 	if _, err := GenerateWithOptions(outputDir, appDir, Options{Config: csrfDisabledConfig(), IR: program}); err != nil {
@@ -7578,10 +7665,11 @@ func HandleCreatePatient(ctx context.Context, command CreatePatient) (CreatePati
 
 	streamCtx, cancelStream := context.WithCancel(context.Background())
 	defer cancelStream()
-	streamRequest, err := http.NewRequestWithContext(streamCtx, http.MethodGet, "http://"+addr+"/_gowdk/realtime/events", nil)
+	streamRequest, err := http.NewRequestWithContext(streamCtx, http.MethodGet, "http://"+addr+"/_gowdk/realtime/events?path=/dashboard", nil)
 	if err != nil {
 		t.Fatal(err)
 	}
+	streamRequest.Header.Set("Referer", "http://"+addr+"/patients")
 	streamResponse, err := http.DefaultClient.Do(streamRequest)
 	if err != nil {
 		t.Fatal(err)
@@ -7643,6 +7731,19 @@ func HandleCreatePatient(ctx context.Context, command CreatePatient) (CreatePati
 			t.Fatalf("realtime stream included unsubscribed event %q in %s", unexpected, dataLine)
 		}
 	}
+	noLeakDeadline := time.After(300 * time.Millisecond)
+	for {
+		select {
+		case line := <-lines:
+			if strings.HasPrefix(line, "data: ") && strings.Contains(line, "OtherNotice") {
+				t.Fatalf("route-scoped realtime stream leaked dashboard event: %s", line)
+			}
+		case err := <-readErrs:
+			t.Fatalf("read realtime stream after first event: %v", err)
+		case <-noLeakDeadline:
+			return
+		}
+	}
 }
 
 func TestGeneratedBinaryRealtimeStreamGuardDenialClosesStream(t *testing.T) {

internal/appgen/source.go

@@ -115,7 +115,7 @@ func runtimeImportMap(options Options) map[string]string {
 		imports["gowdkseo"] = "github.com/cssbruno/gowdk/runtime/seo"
 		imports["gowdkseositemap"] = strings.TrimSpace(options.Sitemap.Dynamic.ImportPath)
 	}
-	if generatedRealtimeStreamUsesRouteMatching(options) {
+	if generatedRealtimeStreamUsesRouteMatching(options) || generatedRealtimeStreamUsesAudience(options) {
 		imports["gowdkroute"] = "github.com/cssbruno/gowdk/runtime/route"
 		imports["neturl"] = "net/url"
 	}

internal/appgen/source_contracts.go

@@ -554,15 +554,13 @@ func currentContractEventSinkDecl(realtime bool, queryInvalidations bool) ast.De
 		&ast.IfStmt{
 			Cond: &ast.BinaryExpr{X: id("fanout"), Op: token.NEQ, Y: id("nil")},
 			Body: block(
-				define([]ast.Expr{id("fanoutSink")}, call(sel("gowdkcontracts", "PresentationFanoutCommandEventSink"), &ast.CompositeLit{
-					Type: id("realtimeSubscriptionFanout"),
-					Elts: []ast.Expr{keyValue("inner", id("fanout"))},
-				})),
+				define([]ast.Expr{id("scopedFanout")}, realtimeSubscriptionFanoutExpr(id("fanout"))),
+				define([]ast.Expr{id("fanoutSink")}, call(sel("gowdkcontracts", "PresentationFanoutCommandEventSink"), id("scopedFanout"))),
 				&ast.IfStmt{
 					Cond: &ast.BinaryExpr{X: id("sink"), Op: token.NEQ, Y: id("nil")},
-					Body: block(&ast.ReturnStmt{Results: []ast.Expr{realtimeCompositeSinkExpr(queryInvalidations, id("sink"), id("fanoutSink"), id("fanout"))}}),
+					Body: block(&ast.ReturnStmt{Results: []ast.Expr{realtimeCompositeSinkExpr(queryInvalidations, id("sink"), id("fanoutSink"), id("scopedFanout"))}}),
 				},
-				&ast.ReturnStmt{Results: []ast.Expr{realtimeCompositeSinkExpr(queryInvalidations, call(sel("gowdkcontracts", "InProcessCommandEventSink")), id("fanoutSink"), id("fanout"))}},
+				&ast.ReturnStmt{Results: []ast.Expr{realtimeCompositeSinkExpr(queryInvalidations, call(sel("gowdkcontracts", "InProcessCommandEventSink")), id("fanoutSink"), id("scopedFanout"))}},
 			),
 		},
 	)
@@ -572,6 +570,13 @@ func currentContractEventSinkDecl(realtime bool, queryInvalidations bool) ast.De
 	}, stmts)
 }
 
+func realtimeSubscriptionFanoutExpr(fanout ast.Expr) ast.Expr {
+	return &ast.CompositeLit{
+		Type: id("realtimeSubscriptionFanout"),
+		Elts: []ast.Expr{keyValue("inner", fanout)},
+	}
+}
+
 func realtimeCompositeSinkExpr(queryInvalidations bool, base ast.Expr, fanoutSink ast.Expr, fanout ast.Expr) ast.Expr {
 	args := []ast.Expr{base}
 	if queryInvalidations {