Skip to content

Commit 4cd5884

Browse files
Implement durable-streams HTTP server with Node.js runtime
Complete durable-streams implementation with three layers: - Stream (persistence, append/subscribe) - StreamManager (multi-stream coordination) - HTTP routes (REST + SSE endpoints) Migrate from Bun to Node.js: - Use @effect/platform-node for HTTP server - Use tsx for TypeScript execution - Add native fetch e2e tests alongside Effect tests 38 tests passing (stream, manager, Effect HTTP, native fetch e2e). 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
1 parent b35ca01 commit 4cd5884

12 files changed

Lines changed: 1187 additions & 102 deletions

‎CLAUDE.md‎

Lines changed: 38 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -333,14 +333,51 @@ describe("MyService", () => {
333333
static readonly testLayer = Layer.sync(MyService, () => {
334334
// Mutable state is fine in tests - JS is single-threaded
335335
const store = new Map<string, Data>()
336-
336+
337337
return MyService.of({
338338
get: (key) => Effect.succeed(store.get(key)),
339339
set: (key, value) => Effect.sync(() => void store.set(key, value))
340340
})
341341
})
342342
```
343343

344+
## TestClock in vitest
345+
346+
`@effect/vitest` uses `TestClock` by default—time doesn't advance automatically. Use `TestClock.adjust` to advance time in tests with `Effect.sleep`:
347+
348+
```typescript
349+
import { Effect, Fiber, TestClock } from "effect"
350+
import { it } from "@effect/vitest"
351+
352+
it.effect("handles sleep with TestClock", () =>
353+
Effect.gen(function*() {
354+
const fiber = yield* Effect.fork(
355+
Effect.gen(function*() {
356+
yield* Effect.sleep("1 second")
357+
return "done"
358+
})
359+
)
360+
361+
yield* TestClock.adjust("1 second") // Advance virtual time
362+
363+
const result = yield* Fiber.join(fiber)
364+
expect(result).toBe("done")
365+
})
366+
)
367+
```
368+
369+
For real time instead of test clock, use `it.live`:
370+
371+
```typescript
372+
it.live("uses real clock", () =>
373+
Effect.gen(function*() {
374+
yield* Effect.sleep("10 millis") // Actually waits
375+
})
376+
)
377+
```
378+
379+
**Alternative:** For PubSub subscription tests, `PubSub.subscribe` guarantees subscription is established when effect completes—often you can publish immediately without sleep.
380+
344381
## Layer Memoization
345382

346383
Layers are memoized by reference. Functions returning layers defeat memoization—each call creates a new object, causing duplicate construction, resource leaks, and inconsistent state.

‎bun.lock‎

Lines changed: 98 additions & 34 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎package.json‎

Lines changed: 8 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -5,14 +5,14 @@
55
"private": true,
66
"scripts": {
77
"prepare": "effect-language-service patch",
8-
"mini-agent": "doppler run -- bun src/cli/main.ts",
9-
"start": "bun src/cli/main.ts",
10-
"dev": "bun --watch src/cli/main.ts",
8+
"mini-agent": "doppler run -- npx tsx src/cli/main.ts",
9+
"start": "npx tsx src/cli/main.ts",
10+
"dev": "npx tsx --watch src/cli/main.ts",
1111
"typecheck": "tsc --noEmit",
1212
"lint": "eslint \"src/**/*.ts\" \"src/**/*.tsx\" \"test/**/*.ts\"",
1313
"lint:fix": "eslint \"src/**/*.ts\" \"src/**/*.tsx\" \"test/**/*.ts\" --fix",
14-
"check": "bun run typecheck && bun run lint",
15-
"check:fix": "bun run typecheck && bun run lint:fix",
14+
"check": "npm run typecheck && npm run lint",
15+
"check:fix": "npm run typecheck && npm run lint:fix",
1616
"test": "doppler run -- vitest run",
1717
"test:watch": "doppler run -- vitest"
1818
},
@@ -23,8 +23,9 @@
2323
"@eslint/compat": "^1.1.1",
2424
"@eslint/eslintrc": "^3.1.0",
2525
"@eslint/js": "^9.10.0",
26-
"@types/bun": "latest",
26+
"@types/node": "^22",
2727
"@types/react": "19",
28+
"tsx": "^4.19.0",
2829
"@typescript-eslint/eslint-plugin": "^8.4.0",
2930
"@typescript-eslint/parser": "^8.4.0",
3031
"eslint": "^9.10.0",
@@ -48,6 +49,7 @@
4849
"@effect/opentelemetry": "^0.59.1",
4950
"@effect/platform": "^0.93.5",
5051
"@effect/platform-bun": "^0.85.0",
52+
"@effect/platform-node": "^0.104.0",
5153
"@effect/rpc": "^0.72.2",
5254
"@effect/rpc-http": "^0.52.4",
5355
"@opentelemetry/otlp-transformer": "^0.208.0",

‎src/durable-streams/cli.ts‎

Lines changed: 60 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,60 @@
1+
/**
2+
* Durable Streams CLI
3+
*
4+
* Commands:
5+
* - start: Start HTTP server for durable streams
6+
*/
7+
import { Command, Options } from "@effect/cli"
8+
import { HttpServer } from "@effect/platform"
9+
import { NodeHttpServer } from "@effect/platform-node"
10+
import { Console, Effect, Layer } from "effect"
11+
import { createServer } from "node:http"
12+
import { durableStreamsRouter } from "./http-routes.ts"
13+
import { StreamManagerService } from "./stream-manager.ts"
14+
15+
/** Port option */
16+
const portOption = Options.integer("port").pipe(
17+
Options.withAlias("p"),
18+
Options.withDescription("Port to listen on"),
19+
Options.withDefault(3000)
20+
)
21+
22+
/** Host option */
23+
const hostOption = Options.text("host").pipe(
24+
Options.withAlias("h"),
25+
Options.withDescription("Host to bind to"),
26+
Options.withDefault("0.0.0.0")
27+
)
28+
29+
/** Start command - launches HTTP server */
30+
const startCommand = Command.make(
31+
"start",
32+
{ host: hostOption, port: portOption },
33+
({ host, port }) =>
34+
Effect.gen(function*() {
35+
yield* Console.log(`Starting durable-streams server on ${host}:${port}`)
36+
37+
// Build service layers
38+
const serviceLayer = StreamManagerService.InMemory
39+
40+
// HTTP server layer
41+
const serverLayer = HttpServer.serve(durableStreamsRouter).pipe(
42+
Layer.provide(NodeHttpServer.layer(createServer, { port })),
43+
Layer.provide(serviceLayer)
44+
)
45+
46+
return yield* Layer.launch(serverLayer)
47+
})
48+
)
49+
50+
/** Main CLI */
51+
export const cli = Command.make("durable-streams").pipe(
52+
Command.withSubcommands([startCommand])
53+
)
54+
55+
/** Run CLI with args */
56+
export const run = (args: ReadonlyArray<string>) =>
57+
Command.run(cli, {
58+
name: "durable-streams",
59+
version: "0.1.0"
60+
})(args)

0 commit comments

Comments
 (0)