Add support for automatic stream compression (defaults off)

[jhaynie]

const stream = await ctx.stream.create('my-stream', {
      contentType: 'text/markdown',
      compress: true,
    });

Added Features:

• Stream.write() method for direct writing without calling getWriter()
• Stream.bytesWritten (readonly) - tracks total bytes written to stream
• Stream.compressed (readonly) - indicates if stream uses compression
• Both write() and getWriter() patterns work correctly with proper locking

Implementation Details:

• StreamImpl now maintains activeWriter for write() method
• Added bytesWritten getter that references underlying sink's total counter
• Added compressed property set during stream creation from props
• Fixed locking mechanism to support both direct write() and getWriter() usage

Bug Fixes:

• Fixed test timeout issues caused by backpressure in stream mocks
• Updated test mocks to continuously drain ReadableStream to prevent blocking
• Resolved WritableStream locking conflicts between write() and getWriter()

Test Coverage:

• Added 8 new test cases for direct write() functionality
• Tests verify write() with multiple data types (string, Uint8Array, ArrayBuffer, objects)
• Tests verify bytesWritten tracking with write() and getWriter()
• Tests verify bytesWritten accuracy across different data types
• Tests verify compressed property for both compressed and uncompressed streams
• All 44 stream tests passing (up from 36)

Summary by CodeRabbit

• New Features

  • Optional gzip compression for stream uploads (Content-Encoding when enabled).
  • Direct write capability on streams (stream.write) alongside writer flow.
  • Stream metadata: bytesWritten and compressed accessors.
  • New create-stream option to enable compression (compress?: true).

• Tests

  • Expanded coverage for write(), close(), multi-type writes, compression, headers, and lifecycle.

• Chores

  • Tightened generic type for vector search params (TypeScript).

[coderabbitai]

Caution

Review failed

The pull request is closed.

Walkthrough

Adds optional gzip compression for stream uploads, a direct write API on Stream, and exposes Stream.bytesWritten and Stream.compressed; updates StreamImpl constructor and request headers to support Content‑Encoding: gzip, plus tests and a patch changeset.

Changes

Cohort / File(s)	Summary
Release Notes / Changeset \.changeset/hip-actors-worry.md	Adds a patch changeset for @agentuity/sdk announcing automatic stream compression and related API additions.
Streaming Implementation src/apis/stream.ts	Adds gzip transform pipeline when props.compress is truthy (dynamic import of node:stream wrappers), centralizes request headers, conditionally adds Content-Encoding: gzip, updates StreamImpl constructor to accept compressed, adds active writer reuse, coercion for write inputs, bytesWritten tracking, and compressed/bytesWritten accessors.
Types / Public API src/types.ts	Adds compress?: true to CreateStreamProps; extends Stream interface with readonly bytesWritten: number, readonly compressed: boolean, and `write(chunk: string
API type formatting src/apis/api.ts	Reformats ServiceName union type from multiline to single-line representation (no behavioral change).
Router formatting src/router/context.ts	Whitespace/formatting adjustments (blank lines) with no logic change.
Tests — streaming behavior & compression test/stream.test.ts	Adds comprehensive tests for Stream.write(), bytesWritten tracking, multiple chunk types, compressed vs. uncompressed flows (Content‑Encoding header, size checks), writer/stream interactions, abort/error propagation, and API key header behavior.
Tests — api & utils formatting test/apis/api.test.ts, test/utils/stringify.test.ts	Formatting and expectation reflows; no functional changes to tests' intent or assertions.

Sequence Diagram(s)

sequenceDiagram
  autonumber
  participant U as User Code
  participant SDK as SDK Stream API
  participant G as Gzip Transform (optional)
  participant HTTP as HTTP PUT

  U->>SDK: createStream({ compress?: true, ... })
  SDK->>SDK: build underlying ReadableStream
  alt compress == true
    SDK->>G: create gzip Transform, pipe underlying -> gzip
    G-->>SDK: compressed ReadableStream
    SDK->>SDK: add header `Content-Encoding: gzip`
  else compress != true
    SDK->>SDK: use original ReadableStream
  end
  U->>SDK: stream.write(chunk) or getWriter() -> writer.write(chunk)
  SDK->>HTTP: PUT /stream with headers and body (ReadableStream)
  HTTP-->>SDK: response
  SDK-->>U: stream handle (bytesWritten, compressed)

Loading

Estimated code review effort

🎯 3 (Moderate) | ⏱️ ~25 minutes

Possibly related PRs

• Add support for automatic stream compression (defaults off) #186 — Modifies the same stream implementation and public API to add gzip compression and direct write support (overlaps on StreamImpl, bytesWritten/compressed, CreateStreamProps.compress, and tests).

Poem

I nibbled bytes and gave them zip,
Chunks went in on a gzip trip,
Headers set, the PUT took wing,
Counters ticked — a tiny thing.
Rabbit cheers — the stream can sing. 🐇✨

Pre-merge checks and finishing touches

✅ Passed checks (3 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title Check	✅ Passed	The title clearly and concisely summarizes the main feature introduced—automatic stream compression—and notes its default state, making it easy for reviewers to understand the primary change at a glance.
Docstring Coverage	✅ Passed	No functions found in the changes. Docstring coverage check skipped.

---

📜 Recent review details

Configuration used: CodeRabbit UI

Review profile: CHILL

Plan: Pro

📥 Commits

Reviewing files that changed from the base of the PR and between 6ace7b0 and 8816ae3.

📒 Files selected for processing (1)

• src/apis/stream.ts (8 hunks)

---

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

[coderabbitai]

Actionable comments posted: 2

🧹 Nitpick comments (2)

src/apis/stream.ts (2)

224-227: Wrap long declaration for 80-char limit

This destructuring declaration exceeds the 80-character limit in our guidelines. Break it across lines to stay within the style rules.

---

230-235: Also wrap long assignments

Both createGzip() assignments exceed 80 characters. Please fold them to comply with the line-length requirement.

📜 Review details

Configuration used: CodeRabbit UI

Review profile: CHILL

Plan: Pro

📥 Commits

Reviewing files that changed from the base of the PR and between 41e0e12 and cf98635.

📒 Files selected for processing (3)

• .changeset/hip-actors-worry.md (1 hunks)
• src/apis/stream.ts (3 hunks)
• src/types.ts (1 hunks)

🧰 Additional context used

📓 Path-based instructions (2)

{src,test}/**/!(*.d).ts

📄 CodeRabbit inference engine (AGENT.md)

{src,test}/**/!(*.d).ts: Use strict TypeScript and prefer unknown over any
Use ESM import/export syntax; avoid CommonJS require/module.exports
Use relative imports for internal modules
Keep imports organized (sorted, no unused imports)
Use tabs with a visual width of 2 spaces
Limit lines to a maximum of 80 characters
Use single quotes for strings
Use proper Error types; do not throw strings
Prefer template literals over string concatenation

Files:

• src/types.ts
• src/apis/stream.ts

src/apis/**

📄 CodeRabbit inference engine (AGENT.md)

Place core API implementations under src/apis/ (email, discord, keyvalue, vector, objectstore)

Files:

• src/apis/stream.ts

🧬 Code graph analysis (1)

src/apis/stream.ts (1)

src/apis/api.ts (1)

• getFetch (12-14)

[coderabbitai]

Actionable comments posted: 0

🧹 Nitpick comments (2)

.changeset/hip-actors-worry.md (1)

8-8: Minor wording improvement.

The phrase "has compression enabled" is clearer than "has compression" for describing boolean state.

Apply this diff:

-- Added property `compressed` to the Stream interface which represents if the stream has compression enabled
+- Added property `compressed` to the Stream interface which represents whether the stream has compression enabled

src/apis/stream.ts (1)

394-401: Consider alternative for bytesWritten exposure.

Using Object.defineProperty to dynamically bind _bytesWritten to the closure variable total works but is unconventional. Consider instead storing total in the instance (e.g., updating this._bytesWritten in the sink's write method) for clarity.

Example alternative approach (update the sink's write method):

async write(chunk: string | Uint8Array | ArrayBuffer | Buffer | object) {
	// ... existing coercion logic ...
	await writer.write(binaryChunk);
	total += binaryChunk.length;
	stream._bytesWritten = total;  // Keep instance in sync
}

Then remove the Object.defineProperty block and rely on the instance field directly. This avoids closure magic and makes the data flow more explicit.

📜 Review details

Configuration used: CodeRabbit UI

Review profile: CHILL

Plan: Pro

📥 Commits

Reviewing files that changed from the base of the PR and between cf98635 and 8b53b1b.

📒 Files selected for processing (4)

• .changeset/hip-actors-worry.md (1 hunks)
• src/apis/stream.ts (6 hunks)
• src/types.ts (2 hunks)
• test/stream.test.ts (1 hunks)

🧰 Additional context used

📓 Path-based instructions (3)

{src,test}/**/!(*.d).ts

📄 CodeRabbit inference engine (AGENT.md)

{src,test}/**/!(*.d).ts: Use strict TypeScript and prefer unknown over any
Use ESM import/export syntax; avoid CommonJS require/module.exports
Use relative imports for internal modules
Keep imports organized (sorted, no unused imports)
Use tabs with a visual width of 2 spaces
Limit lines to a maximum of 80 characters
Use single quotes for strings
Use proper Error types; do not throw strings
Prefer template literals over string concatenation

Files:

• src/types.ts
• test/stream.test.ts
• src/apis/stream.ts

test/**

📄 CodeRabbit inference engine (AGENT.md)

Tests must mirror the source structure under the test/ directory

Files:

• test/stream.test.ts

src/apis/**

📄 CodeRabbit inference engine (AGENT.md)

Place core API implementations under src/apis/ (email, discord, keyvalue, vector, objectstore)

Files:

• src/apis/stream.ts

🧬 Code graph analysis (2)

test/stream.test.ts (1)

src/apis/api.ts (1)

• setFetch (8-10)

src/apis/stream.ts (3)

src/types.ts (1)

• Stream (364-398)

src/utils/stringify.ts (1)

• safeStringify (6-40)

src/apis/api.ts (1)

• getFetch (12-14)

🔇 Additional comments (11)

src/types.ts (2)

346-351: LGTM!

The compress?: true type is intentional per maintainer feedback—it enforces explicit opt-in by preventing boolean variables. The optional ? handles the default-false case.

---

373-384: LGTM!

The new Stream interface members are well-typed and documented:

• bytesWritten and compressed are appropriately readonly
• write() accepts the same types as the underlying sink
• Signatures match the implementation

test/stream.test.ts (2)

1127-1390: LGTM!

Comprehensive test coverage for the new write() method and bytesWritten tracking:

• Verifies direct write() with multiple data types
• Tracks bytesWritten correctly across write() and getWriter() usage
• Tests compressed property for both enabled and disabled states
• Validates PUT request initiation behavior

---

1392-1596: LGTM!

Thorough compression test coverage:

• Verifies Content-Encoding: gzip header when enabled
• Confirms actual data compression (size reduction)
• Tests error propagation through the compression pipeline
• Good edge-case handling

src/apis/stream.ts (7)

7-7: LGTM!

Import of createGzip for compression support is appropriate.

---

15-24: LGTM!

New private fields and updated constructor correctly store compression state and provide writer tracking for the direct write() API.

---

26-32: LGTM!

Public getters for bytesWritten and compressed expose the internal state as specified in the Stream interface.

---

34-55: LGTM!

The write() method correctly:

• Coerces strings, ArrayBuffer, Uint8Array, and objects to Uint8Array
• Uses safeStringify for object serialization
• Reuses a single active writer across calls

---

61-104: LGTM!

The close() method gracefully handles:

• Active writer cleanup when using direct write() API
• Already-closed and locked stream states
• Proper error propagation for unexpected failures

---

262-294: LGTM!

The compression pipeline is correctly wired:

• Gzip stream converts to Web streams via node:stream adapters
• Error handlers abort the fetch and writer on compression failures (addressing past review comment)
• Base readable is replaced with compressed readable for the PUT body

---

308-318: LGTM!

Centralized headers object improves maintainability and correctly adds Content-Encoding: gzip when compression is enabled.

[coderabbitai]

Actionable comments posted: 0

🧹 Nitpick comments (1)

src/apis/stream.ts (1)

7-7: Consider conditional import of zlib.

The createGzip import at the top level is only used when props?.compress is true. Since you already conditionally import node:stream at line 274, consider doing the same for node:zlib to avoid loading it when compression is disabled.

-import { createGzip } from 'node:zlib';

Then at line 283:

+							const { createGzip } = await import('node:zlib');
 							const gzipStream = createGzip();

📜 Review details

Configuration used: CodeRabbit UI

Review profile: CHILL

Plan: Pro

📥 Commits

Reviewing files that changed from the base of the PR and between 8b53b1b and 6ace7b0.

📒 Files selected for processing (7)

• src/apis/api.ts (1 hunks)
• src/apis/stream.ts (8 hunks)
• src/router/context.ts (2 hunks)
• src/types.ts (2 hunks)
• test/apis/api.test.ts (7 hunks)
• test/stream.test.ts (21 hunks)
• test/utils/stringify.test.ts (5 hunks)

✅ Files skipped from review due to trivial changes (4)

• src/router/context.ts
• src/apis/api.ts
• test/apis/api.test.ts
• test/utils/stringify.test.ts

🧰 Additional context used

📓 Path-based instructions (3)

{src,test}/**/!(*.d).ts

📄 CodeRabbit inference engine (AGENT.md)

{src,test}/**/!(*.d).ts: Use strict TypeScript and prefer unknown over any
Use ESM import/export syntax; avoid CommonJS require/module.exports
Use relative imports for internal modules
Keep imports organized (sorted, no unused imports)
Use tabs with a visual width of 2 spaces
Limit lines to a maximum of 80 characters
Use single quotes for strings
Use proper Error types; do not throw strings
Prefer template literals over string concatenation

Files:

• src/types.ts
• src/apis/stream.ts
• test/stream.test.ts

src/apis/**

📄 CodeRabbit inference engine (AGENT.md)

Place core API implementations under src/apis/ (email, discord, keyvalue, vector, objectstore)

Files:

• src/apis/stream.ts

test/**

📄 CodeRabbit inference engine (AGENT.md)

Tests must mirror the source structure under the test/ directory

Files:

• test/stream.test.ts

🧬 Code graph analysis (2)

src/apis/stream.ts (3)

src/types.ts (1)

• Stream (364-400)

src/utils/stringify.ts (1)

• safeStringify (6-40)

src/apis/api.ts (1)

• getFetch (12-14)

test/stream.test.ts (2)

src/types.ts (1)

• CreateStreamProps (335-352)

src/apis/api.ts (1)

• setFetch (8-10)

🪛 GitHub Actions: Run Tests

src/apis/stream.ts

[error] 361-361: TS2341: Property '_bytesWritten' is private and only accessible within class 'StreamImpl'. While running 'tsc --emitDeclarationOnly --declaration'.

🔇 Additional comments (9)

src/types.ts (3)

346-351: Documentation is clear and accurate.

The JSDoc properly warns that clients must handle gzip decompression, and the literal true type enforces explicit opt-in as intended.

---

373-386: Public API additions look correct.

The new bytesWritten, compressed properties and write(chunk) method properly extend the Stream interface with appropriate types and documentation.

---

478-479: Type constraint tightened appropriately.

Constraining VectorSearchParams<T extends JsonObject = JsonObject> ensures metadata filtering aligns with the JsonObject shapes used throughout the API surface.

test/stream.test.ts (1)

1-1657: LGTM! Comprehensive test coverage for stream compression and direct write functionality.

The test suite thoroughly covers:

• Stream creation validation and props handling
• Streaming functionality with various data types
• Content-Type handling (custom and default)
• Direct write() and close() methods
• bytesWritten tracking across different write patterns
• Compression behavior including Content-Encoding header verification
• Error handling and edge cases

The tests are well-structured, use proper mocking patterns, and verify both interface correctness and data flow architecture.

src/apis/stream.ts (5)

273-303: Compression pipeline correctly propagates errors.

The gzip compression setup properly handles error propagation from both pipeline stages to abort the fetch and writer, addressing the concern from the previous review.

---

42-62: Direct write() method correctly reuses active writer.

The implementation properly coerces various chunk types to Uint8Array and maintains a single activeWriter across multiple write calls, which is efficient and correct.

---

68-108: close() method gracefully handles multiple scenarios.

The enhanced error handling correctly manages already-closed streams, locked streams with active writers, and other edge cases. The logic is sound and aligns with the test expectations.

---

317-326: Headers construction is clean and conditional.

The centralized headers object with conditional Content-Encoding: gzip is more maintainable than inline header construction.

---

361-361: Fix compilation error: Cannot access private property from closure.

The pipeline failure indicates that _bytesWritten is private but accessed from the underlyingSink.write() closure at line 361. This causes a TypeScript compilation error (TS2341).

Solution: Change _bytesWritten to public or add a setter method:

-	private _bytesWritten = 0;
+	public _bytesWritten = 0;

Or add a setter method:

+	setBytesWritten(value: number): void {
+		this._bytesWritten = value;
+	}

Then update line 361:

 					if (streamInstance) {
-						streamInstance._bytesWritten = total;
+						streamInstance.setBytesWritten(total);
 					}

Likely an incorrect or invalid review comment.