add flush endpoint to node

e2e/tests/prod-trust-boundary.test.ts

@@ -104,6 +104,11 @@ const AUTH_REALTIME_INTERNAL_ENDPOINTS = [
 		method: "POST",
 		path: "/auth-realtime/api/internal/documents/doc-e2e/branches/branch-e2e/operations",
 	},
+	{
+		name: "a branch flush",
+		method: "POST",
+		path: "/auth-realtime/api/internal/documents/doc-e2e/branches/branch-e2e/flush",
+	},
 ] as const
 
 // spellings of a blocked path that a proxy is expected to still recognise.

server/auth-realtime/src/routes.test.ts

@@ -1457,6 +1457,40 @@ describe("createRoutes", () => {
 		})
 	})
 
+	describe("POST /internal/documents/:documentId/branches/:branchId/flush", () => {
+		it("flushes the branch and answers once core has it", async ({
+			expect,
+		}) => {
+			const { app, flushDocument } = build()
+
+			const res = await app.request(
+				"/internal/documents/doc1/branches/b1/flush",
+				{ method: "POST" },
+			)
+
+			expect(res.status).toBe(204)
+			expect(
+				flushDocument.mock.calls.map((call) => call[0]),
+			).toEqual(["doc1-b1"])
+		})
+
+		it("reports a flush that failed", async ({ expect }) => {
+			const flushDocument = vi
+				.fn()
+				.mockRejectedValue(
+					new Error("core unreachable"),
+				)
+			const { app } = build({ flushDocument })
+
+			const res = await app.request(
+				"/internal/documents/doc1/branches/b1/flush",
+				{ method: "POST" },
+			)
+
+			expect(res.status).toBe(500)
+		})
+	})
+
 	describe("POST /internal/documents/:documentId/branches/:branchId/operations", () => {
 		const path = "/internal/documents/doc1/branches/b1/operations"
 

server/auth-realtime/src/routes.ts

@@ -658,6 +658,25 @@ export function createRoutes({
 		})
 	})
 
+	// core calls this before it reads a branch it is about to change, so
+	// it reads what the editors hold.
+	app.post(
+		"/internal/documents/:documentId/branches/:branchId/flush",
+		async (c) => {
+			const branch = await flushed(c, () =>
+				flushBranch(
+					c.req.param("documentId"),
+					c.req.param("branchId"),
+				),
+			)
+			if (branch instanceof Response) {
+				return branch
+			}
+
+			return c.body(null, 204)
+		},
+	)
+
 	// Server-to-server endpoint called by the Go assistant to apply
 	// edit operations to a live Y.Doc. Each request opens (or reuses) a
 	// direct connection to the document, runs the batch inside a single

server/core/internal/assistant/edit/client.go

@@ -30,10 +30,10 @@ type Result struct {
 	Errors []OpError `json:"errors"`
 }
 
-// Client posts batched operations to the Node hocuspocus service's
-// internal operations endpoint. The base URL points at the Node
-// service (e.g. http://auth-realtime:8081); endpoint replaces its
-// path with the per-document operations path.
+// Client posts to the Node hocuspocus service's internal per-branch
+// endpoints: batched operations and flushes. The base URL points at the
+// Node service (e.g. http://auth-realtime:8081); endpoint replaces its
+// path with the per-branch action path.
 type Client struct {
 	// httpClient is the underlying HTTP client. Callers should
 	// pass one with a reasonable timeout configured.
@@ -44,8 +44,8 @@ type Client struct {
 }
 
 // NewClient constructs an edit-RPC client. The baseURL must include
-// the scheme and host of the Node service; the operations path is
-// appended per request.
+// the scheme and host of the Node service; the action path is appended
+// per request.
 func NewClient(httpClient *http.Client, baseURL string) *Client {
 	if httpClient == nil {
 		httpClient = http.DefaultClient
@@ -102,7 +102,7 @@ func (c *Client) Apply(
 		return Result{}, fmt.Errorf("marshaling operations: %w", err)
 	}
 
-	endpoint, err := c.endpoint(documentID, branchID)
+	endpoint, err := c.endpoint(documentID, branchID, "operations")
 	if err != nil {
 		return Result{}, err
 	}
@@ -142,16 +142,50 @@ func (c *Client) Apply(
 	return out, nil
 }
 
-// endpoint builds the per-document operations URL.
-func (c *Client) endpoint(documentID, branchID xid.ID) (string, error) {
+// Flush has the Node service store what the editors of the branch hold
+// and returns once core has it.
+func (c *Client) Flush(ctx context.Context, documentID, branchID xid.ID) error {
+	endpoint, err := c.endpoint(documentID, branchID, "flush")
+	if err != nil {
+		return err
+	}
+
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, http.NoBody)
+	if err != nil {
+		// NOCOV: the URL is pre-validated by endpoint, so the only
+		// remaining failure is a nil context.
+		return fmt.Errorf("building request: %w", err)
+	}
+
+	resp, err := c.httpClient.Do(req)
+	if err != nil {
+		return fmt.Errorf("posting flush: %w", err)
+	}
+	defer resp.Body.Close() //nolint:errcheck // error provides no meaningful info
+
+	if resp.StatusCode != http.StatusNoContent {
+		preview, err := io.ReadAll(io.LimitReader(resp.Body, _maxErrorPreviewBytes))
+		if err != nil {
+			return fmt.Errorf("node flush endpoint: status %d: reading response body: %w", resp.StatusCode, err)
+		}
+
+		return fmt.Errorf("node flush endpoint: status %d: %s", resp.StatusCode, string(preview))
+	}
+
+	return nil
+}
+
+// endpoint builds the URL of a per-branch action.
+func (c *Client) endpoint(documentID, branchID xid.ID, action string) (string, error) {
 	base, err := url.Parse(c.baseURL)
 	if err != nil {
 		return "", fmt.Errorf("parsing base url: %w", err)
 	}
 
-	base.Path = fmt.Sprintf("/api/internal/documents/%s/branches/%s/operations",
+	base.Path = fmt.Sprintf("/api/internal/documents/%s/branches/%s/%s",
 		documentID,
 		branchID,
+		action,
 	)
 
 	return base.String(), nil

server/core/internal/assistant/edit/client_test.go

@@ -235,6 +235,78 @@ func Test_Client_Apply(t *testing.T) {
 	}
 }
 
+func Test_Client_Flush(t *testing.T) {
+	t.Parallel()
+
+	docID, branchID := xid.New(), xid.New()
+	path := "/api/internal/documents/" + docID.String() + "/branches/" + branchID.String() + "/flush"
+
+	cc := map[string]struct {
+		BaseURL    string
+		CloseEarly bool
+		StatusCode int
+		Body       string
+		Err        error
+	}{
+		"Invalid base URL": {
+			BaseURL: "://bad",
+			Err:     assert.AnError,
+		},
+		"Transport error": {
+			CloseEarly: true,
+			Err:        assert.AnError,
+		},
+		"Non-204 status surfaces the body": {
+			StatusCode: http.StatusInternalServerError,
+			Body:       `{"error":"pending changes could not be stored"}`,
+			Err:        assert.AnError,
+		},
+		"Successful flush": {
+			StatusCode: http.StatusNoContent,
+		},
+	}
+
+	for cn, c := range cc {
+		t.Run(cn, func(t *testing.T) {
+			t.Parallel()
+
+			var gotPath, gotMethod string
+
+			srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				gotPath, gotMethod = r.URL.Path, r.Method
+
+				w.WriteHeader(c.StatusCode)
+				_, err := w.Write([]byte(c.Body))
+				assert.NoError(t, err)
+			}))
+			defer srv.Close()
+
+			base := srv.URL
+			if c.BaseURL != "" {
+				base = c.BaseURL
+			}
+
+			if c.CloseEarly {
+				srv.Close()
+			}
+
+			err := NewClient(srv.Client(), base).Flush(context.Background(), docID, branchID)
+			testutil.AssertEqualError(t, c.Err, err)
+
+			if c.StatusCode == 0 {
+				return
+			}
+
+			assert.Equal(t, path, gotPath)
+			assert.Equal(t, http.MethodPost, gotMethod)
+
+			if c.Body != "" {
+				assert.Contains(t, err.Error(), c.Body)
+			}
+		})
+	}
+}
+
 func Test_Client_endpoint(t *testing.T) {
 	t.Parallel()
 
@@ -259,7 +331,7 @@ func Test_Client_endpoint(t *testing.T) {
 		t.Run(cn, func(t *testing.T) {
 			t.Parallel()
 
-			got, err := NewClient(nil, c.BaseURL).endpoint(docID, branchID)
+			got, err := NewClient(nil, c.BaseURL).endpoint(docID, branchID, "operations")
 			testutil.AssertEqualError(t, c.Err, err)
 			assert.Equal(t, c.Expected, got)
 		})