103 lines
3.7 KiB
TypeScript
103 lines
3.7 KiB
TypeScript
import { EventV2Bridge } from "@/event-v2-bridge"
|
|
import { InstanceState } from "@/effect/instance-state"
|
|
import { GlobalBus } from "@/bus/global"
|
|
import { EventV2 } from "@opencode-ai/core/event"
|
|
import * as Log from "@opencode-ai/core/util/log"
|
|
import { Effect, Queue } from "effect"
|
|
import * as Stream from "effect/Stream"
|
|
import { HttpServerResponse } from "effect/unstable/http"
|
|
import { HttpApiBuilder } from "effect/unstable/httpapi"
|
|
import * as Sse from "effect/unstable/encoding/Sse"
|
|
import { EventApi } from "../groups/event"
|
|
|
|
const log = Log.create({ service: "server" })
|
|
|
|
function eventData(data: unknown): Sse.Event {
|
|
return {
|
|
_tag: "Event",
|
|
event: "message",
|
|
id: undefined,
|
|
data: JSON.stringify(data),
|
|
}
|
|
}
|
|
|
|
function eventID() {
|
|
return EventV2.ID.create()
|
|
}
|
|
|
|
function eventResponse(events: EventV2.Interface) {
|
|
return Effect.gen(function* () {
|
|
const instance = yield* InstanceState.context
|
|
const workspaceID = yield* InstanceState.workspaceID
|
|
// Listener registration is eager, so events published after this point cannot
|
|
// be lost while the HTTP body fiber is starting or emitting server.connected.
|
|
const queue = yield* Queue.unbounded<EventV2.Payload>()
|
|
const unsubscribe = yield* events.listen((event) => Effect.sync(() => Queue.offerUnsafe(queue, event)))
|
|
yield* Effect.addFinalizer(() => unsubscribe)
|
|
const stream = Stream.fromQueue(queue).pipe(
|
|
Stream.filter(
|
|
(event) =>
|
|
event.location?.directory === instance.directory &&
|
|
(event.location.workspaceID === undefined || event.location.workspaceID === workspaceID),
|
|
),
|
|
Stream.map((event) => ({ id: event.id, type: event.type, properties: event.data })),
|
|
)
|
|
const disposed = Stream.callback<{ id: string; type: string; properties: unknown }>((queue) => {
|
|
const listener = (event: {
|
|
directory?: string
|
|
payload: { id?: string; type?: string; properties?: unknown }
|
|
}) => {
|
|
if (event.directory !== instance.directory || event.payload.type !== "server.instance.disposed") return
|
|
Queue.offerUnsafe(queue, {
|
|
id: event.payload.id ?? eventID(),
|
|
type: "server.instance.disposed",
|
|
properties: event.payload.properties ?? {},
|
|
})
|
|
}
|
|
return Effect.acquireRelease(
|
|
Effect.sync(() => GlobalBus.on("event", listener)),
|
|
() => Effect.sync(() => GlobalBus.off("event", listener)),
|
|
)
|
|
})
|
|
const output = stream.pipe(
|
|
Stream.merge(disposed, { haltStrategy: "left" }),
|
|
Stream.takeUntil((event) => event.type === "server.instance.disposed"),
|
|
)
|
|
const heartbeat = Stream.tick("10 seconds").pipe(
|
|
Stream.drop(1),
|
|
Stream.map(() => ({ id: eventID(), type: "server.heartbeat", properties: {} })),
|
|
)
|
|
|
|
log.info("event connected")
|
|
return HttpServerResponse.stream(
|
|
Stream.make({ id: eventID(), type: "server.connected", properties: {} }).pipe(
|
|
Stream.concat(output.pipe(Stream.merge(heartbeat, { haltStrategy: "left" }))),
|
|
Stream.map(eventData),
|
|
Stream.pipeThroughChannel(Sse.encode()),
|
|
Stream.encodeText,
|
|
Stream.ensuring(Effect.sync(() => log.info("event disconnected"))),
|
|
),
|
|
{
|
|
contentType: "text/event-stream",
|
|
headers: {
|
|
"Cache-Control": "no-cache, no-transform",
|
|
"X-Accel-Buffering": "no",
|
|
"X-Content-Type-Options": "nosniff",
|
|
},
|
|
},
|
|
)
|
|
})
|
|
}
|
|
|
|
export const eventHandlers = HttpApiBuilder.group(EventApi, "event", (handlers) =>
|
|
Effect.gen(function* () {
|
|
const events = yield* EventV2Bridge.Service
|
|
return handlers.handleRaw(
|
|
"subscribe",
|
|
Effect.fn("EventHttpApi.subscribe")(function* () {
|
|
return yield* eventResponse(events)
|
|
}),
|
|
)
|
|
}),
|
|
)
|