[TT-14794] fix issue where an invalid stream path results in 500 (#7047)

### **User description**
<details open>
<summary><a href="https://tyktech.atlassian.net/browse/TT-14794"
title="TT-14794" target="_blank">TT-14794</a></summary>
  <br />
  <table>
    <tr>
      <th>Summary</th>
<td>Calling a wrong path on streams API returns `There was a problem
proxying the request`</td>
    </tr>
    <tr>
      <th>Type</th>
      <td>
<img alt="Bug"
src="https://tyktech.atlassian.net/rest/api/2/universal_avatar/view/type/issuetype/avatar/10303?size=medium"
/>
        Bug
      </td>
    </tr>
    <tr>
      <th>Status</th>
      <td>In Dev</td>
    </tr>
    <tr>
      <th>Points</th>
      <td>N/A</td>
    </tr>
    <tr>
      <th>Labels</th>
      <td>-</td>
    </tr>
  </table>
</details>
<!--
  do not remove this marker as it will break jira-lint's functionality.
  added_by_jira_lint
-->

---

This PR fixes an issue, where an invalid path would result in a 500
Internal Server Error without giving any feedback to the user.

Example:

SSE Output path on `/get/stream`.

When calling `/get/steam` would result in `500`.

Now it will respond with `404`.


___

### **PR Type**
Bug fix, Tests


___

### **Description**
- Fix 500 error for invalid stream paths, return 404 instead

- Add tests for SSE output path error handling and success

- Implement test SSE client for streaming endpoint verification

- Improve debug logging for missing stream paths


___



### **Changes walkthrough** 📝
<table><thead><tr><th></th><th align="left">Relevant
files</th></tr></thead><tbody><tr><td><strong>Bug
fix</strong></td><td><table>
<tr>
  <td>
    <details>
<summary><strong>middleware.go</strong><dd><code>Return 404 and log when
stream path is missing</code>&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; </dd></summary>
<hr>

ee/middleware/streams/middleware.go

<li>Return 404 Not Found with error for missing stream paths<br> <li>
Add debug log for unmatched stream paths


</details>


  </td>
<td><a
href="https://github.com/TykTechnologies/tyk/pull/7047/files#diff-0ce428c0f09dca65e3df6e72d01fee63b6f237785e41e6ecf0ce34a8b65c74a5">+2/-1</a>&nbsp;
&nbsp; &nbsp; </td>

</tr>
</table></td></tr><tr><td><strong>Tests</strong></td><td><table>
<tr>
  <td>
    <details>
<summary><strong>mw_streaming_test.go</strong><dd><code>Add tests for
SSE output path error and success</code>&nbsp; &nbsp; &nbsp; &nbsp;
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; </dd></summary>
<hr>

gateway/mw_streaming_test.go

<li>Add test to verify 404 for invalid SSE output paths<br> <li> Add
test for successful SSE connection and message receipt<br> <li>
Implement helper SSE client for testing streaming endpoints


</details>


  </td>
<td><a
href="https://github.com/TykTechnologies/tyk/pull/7047/files#diff-a0d1bd0196a741537a3c850e340225c8993e49d709c838af0f1b48b9893af1da">+148/-0</a>&nbsp;
</td>

</tr>
</table></td></tr></tr></tbody></table>

___

> <details> <summary> Need help?</summary><li>Type <code>/help how to
...</code> in the comments thread for any questions about PR-Agent
usage.</li><li>Check out the <a
href="https://qodo-merge-docs.qodo.ai/usage-guide/">documentation</a>
for more information.</li></details>

Author: pvormste

ee/middleware/streams/middleware.go

@@ -227,7 +227,8 @@ func (s *Middleware) processStreamsConfig(r *http.Request, streams map[string]an
 func (s *Middleware) ProcessRequest(w http.ResponseWriter, r *http.Request, _ interface{}) (error, int) {
 	strippedPath := s.Spec.StripListenPath(r.URL.Path)
 	if !s.defaultManager.hasPath(strippedPath) {
-		return nil, http.StatusOK
+		s.Logger().Debugf("Path not found: %s", strippedPath)
+		return errors.New("not found"), http.StatusNotFound
 	}
 
 	s.Logger().Debugf("Processing request: %s, %s", r.URL.Path, strippedPath)

gateway/mw_streaming_test.go

@@ -3,6 +3,7 @@
 package gateway
 
 import (
+	"bufio"
 	"bytes"
 	"context"
 	"crypto/tls"
@@ -11,6 +12,7 @@ import (
 	"net/http"
 	"os"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -185,6 +187,7 @@ streams:
     output:
       http_server:
         ws_path: /subscribe
+        stream_path: /get/stream
 `
 
 func TestStreamingAPISingleClient(t *testing.T) {
@@ -1041,3 +1044,148 @@ func TestStreamingAPIGarbageCollection(t *testing.T) {
 	})
 	require.Equal(t, 0, streamManagersAfterGC)
 }
+
+func TestStreaming_HttpOutputPaths(t *testing.T) {
+	ts := StartTest(func(globalConf *config.Config) {
+		globalConf.Streaming.Enabled = true
+	})
+	t.Cleanup(ts.Close)
+
+	oasAPI, err := setupOASForStreamAPI(bentoHTTPServerTemplate)
+	require.NoError(t, err)
+
+	apiName := "output-paths-test-api"
+	ts.Gw.BuildAndLoadAPI(func(spec *APISpec) {
+		spec.Proxy.ListenPath = fmt.Sprintf("/%s", apiName)
+		spec.UseKeylessAccess = true
+		spec.IsOAS = true
+		spec.OAS = oasAPI
+		spec.OAS.Fill(*spec.APIDefinition)
+	})
+
+	apiUrl := fmt.Sprintf("%s/%s", ts.URL, apiName)
+
+	t.Run("should fail with 404 Not Found if the path is invalid", func(t *testing.T) {
+		pathWithTypo := "get/steam"
+		targetUrl := fmt.Sprintf("%s/%s", apiUrl, pathWithTypo)
+
+		sseClient := newTestStreamSSEClient(context.Background(), targetUrl)
+		statusCode, err := sseClient.Connect()
+
+		assert.NoError(t, err)
+		assert.Equal(t, http.StatusNotFound, statusCode)
+	})
+
+	t.Run("should connect to SSE endpoint and consume messages as expected", func(t *testing.T) {
+		correctPath := "get/stream"
+		targetUrl := fmt.Sprintf("%s/%s", apiUrl, correctPath)
+
+		sseContext, cancelFn := context.WithCancel(context.Background())
+		sseClient := newTestStreamSSEClient(sseContext, targetUrl)
+		go func() {
+			statusCode, err := sseClient.Connect()
+			require.NoError(t, err)
+			require.Equal(t, http.StatusOK, statusCode)
+		}()
+
+		defer func() {
+			err := sseClient.Close()
+			require.NoError(t, err)
+			cancelFn()
+		}()
+
+		receiveMessageChan := make(chan string)
+		go func() {
+			err := sseClient.ReadMessages(receiveMessageChan)
+			require.NoError(t, err)
+		}()
+
+		testMessage := `{"test": "message"}`
+		publishURL := fmt.Sprintf("%s/post", apiUrl)
+		publishResp, err := http.Post(publishURL, "application/json", bytes.NewReader([]byte(testMessage)))
+		require.NoError(t, err)
+		require.Equal(t, http.StatusOK, publishResp.StatusCode)
+
+		var message string
+		assert.Eventuallyf(t, func() bool {
+			message = <-receiveMessageChan
+			return true
+		}, 3*time.Second, 10*time.Millisecond, "SSE message not received")
+		assert.Equal(t, testMessage, message)
+	})
+}
+
+type testStreamSSEClient struct {
+	ctx    context.Context
+	url    string
+	client http.Client
+	resp   *http.Response
+	done   chan struct{}
+	wg     *sync.WaitGroup
+}
+
+func newTestStreamSSEClient(ctx context.Context, url string) *testStreamSSEClient {
+	wg := sync.WaitGroup{}
+	wg.Add(1)
+
+	return &testStreamSSEClient{
+		ctx:    ctx,
+		url:    url,
+		client: http.Client{},
+		done:   make(chan struct{}),
+		wg:     &wg,
+	}
+}
+
+func (sse *testStreamSSEClient) Connect() (statusCode int, err error) {
+	req, err := http.NewRequestWithContext(sse.ctx, http.MethodGet, sse.url, nil)
+	if err != nil {
+		return 0, fmt.Errorf("creating request: %w", err)
+	}
+
+	req.Header.Set("Accept", "text/event-stream")
+	req.Header.Set("Cache-Control", "no-cache")
+	req.Header.Set("Connection", "keep-alive")
+
+	sse.resp, err = sse.client.Do(req)
+	sse.wg.Done()
+	if err != nil {
+		close(sse.done)
+		if sse.resp != nil {
+			statusCode = sse.resp.StatusCode
+		}
+		return statusCode, fmt.Errorf("doing request: %w", err)
+	}
+
+	return sse.resp.StatusCode, nil
+}
+
+func (sse *testStreamSSEClient) Close() error {
+	close(sse.done)
+	return nil
+}
+
+func (sse *testStreamSSEClient) ReadMessages(messageChan chan<- string) (err error) {
+	sse.wg.Wait() // we need to make sure that the connection is ready
+	scanner := bufio.NewScanner(sse.resp.Body)
+
+	defer func() {
+		err = sse.resp.Body.Close()
+	}()
+
+	for scanner.Scan() {
+		select {
+		case <-sse.ctx.Done():
+			return nil
+		case <-sse.done:
+			return nil
+		default:
+			message := scanner.Text()
+			if message != "" {
+				messageChan <- message
+			}
+		}
+	}
+
+	return nil
+}