File size: 1,895 Bytes
89a2873 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 | import { EventV2 } from "@opencode-ai/core/event"
import { OpenCodeEvent } from "@opencode-ai/protocol/groups/event"
import { Effect, Schema, Stream } from "effect"
import { HttpServerResponse } from "effect/unstable/http"
import { HttpApiBuilder } from "effect/unstable/httpapi"
import * as Sse from "effect/unstable/encoding/Sse"
import { Api } from "../api"
const subscriberCapacity = 256
function eventData(data: unknown): Sse.Event {
return {
_tag: "Event",
event: "message",
id: undefined,
data: JSON.stringify(Schema.encodeUnknownSync(OpenCodeEvent)(data)),
}
}
export const EventHandler = HttpApiBuilder.group(Api, "server.event", (handlers) =>
Effect.gen(function* () {
const events = yield* EventV2.Service
return handlers.handleRaw("event.subscribe", () =>
Effect.gen(function* () {
const connected = {
id: EventV2.ID.create(),
type: "server.connected",
data: {},
}
const output = Stream.unwrap(
Effect.gen(function* () {
// Acquiring the bounded stream installs its listener before readiness is observable.
const live = yield* EventV2.allBounded(events, subscriberCapacity)
return Stream.make(connected).pipe(Stream.concat(live))
}),
).pipe(Stream.map(eventData), Stream.pipeThroughChannel(Sse.encode()))
const heartbeat = Stream.tick("15 seconds").pipe(Stream.map(() => ": heartbeat\n\n"))
return HttpServerResponse.stream(
output.pipe(Stream.merge(heartbeat, { haltStrategy: "left" }), Stream.encodeText),
{
contentType: "text/event-stream",
headers: {
"Cache-Control": "no-cache, no-transform",
"X-Accel-Buffering": "no",
"X-Content-Type-Options": "nosniff",
},
},
)
}),
)
}),
)
|