agent-radar
2ed3415fb7
Publish Claude Code Binaries / Stage and publish binaries (push) Canceled after 0s
Publish Tutti App Runtime / Build darwin-arm64 (push) Canceled after 0s
Publish Tutti App Runtime / Build darwin-amd64 (push) Canceled after 0s
Publish Tutti App Runtime / Build linux-amd64 (push) Canceled after 0s
Publish Tutti App Runtime / Build windows-amd64 (push) Canceled after 0s
Sync Tutti Agent Skills / sync (push) Canceled after 0s
Windows Agent Adapters / Test Windows agent adapters (push) Canceled after 0s
Windows Daemon Adapters / Test Windows daemon adapters (push) Canceled after 0s
Publish Tutti App Runtime / Publish runtime catalog (push) Canceled after 0s
507 lines
15 KiB
Go
507 lines
15 KiB
Go
package api
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"log/slog"
|
|
"net/http"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/coder/websocket"
|
|
eventprotocol "github.com/tutti-os/tutti/services/tuttid/api/events/generated"
|
|
tuttigenerated "github.com/tutti-os/tutti/services/tuttid/api/generated"
|
|
"github.com/tutti-os/tutti/services/tuttid/apierrors"
|
|
eventstreamservice "github.com/tutti-os/tutti/services/tuttid/service/eventstream"
|
|
tuttitypes "github.com/tutti-os/tutti/services/tuttid/types"
|
|
)
|
|
|
|
type eventStreamClientKindFrame struct {
|
|
Kind string `json:"kind"`
|
|
RequestID string `json:"requestId,omitempty"`
|
|
}
|
|
|
|
type eventStreamClientPublishFrame struct {
|
|
Kind string `json:"kind"`
|
|
RequestID string `json:"requestId"`
|
|
Event json.RawMessage `json:"event"`
|
|
}
|
|
|
|
func eventStreamServiceUnavailableError() tuttigenerated.ServiceUnavailableErrorJSONResponse {
|
|
return serviceUnavailableError(
|
|
apierrors.EventStreamServiceUnavailable(
|
|
apierrors.WithDeveloperMessage("event stream service is unavailable"),
|
|
),
|
|
)
|
|
}
|
|
|
|
func (DaemonAPI) AttachEventStream(context.Context, tuttigenerated.AttachEventStreamRequestObject) (tuttigenerated.AttachEventStreamResponseObject, error) {
|
|
return tuttigenerated.AttachEventStream503JSONResponse{
|
|
ServiceUnavailableErrorJSONResponse: eventStreamServiceUnavailableError(),
|
|
}, nil
|
|
}
|
|
|
|
func (routes daemonRoutes) AttachEventStreamWebSocket(w http.ResponseWriter, r *http.Request) {
|
|
routes.api.attachEventStreamWebSocket(w, r)
|
|
}
|
|
|
|
func (api DaemonAPI) attachEventStreamWebSocket(w http.ResponseWriter, r *http.Request) {
|
|
if api.EventStreamService == nil {
|
|
writeEventStreamHTTPError(
|
|
w,
|
|
apierrors.EventStreamServiceUnavailable(
|
|
apierrors.WithDeveloperMessage("event stream service is unavailable"),
|
|
),
|
|
)
|
|
return
|
|
}
|
|
|
|
conn, err := websocket.Accept(w, r, &websocket.AcceptOptions{
|
|
InsecureSkipVerify: true,
|
|
})
|
|
if err != nil {
|
|
return
|
|
}
|
|
defer conn.Close(websocket.StatusNormalClosure, "event stream detached")
|
|
|
|
ctx, cancel := context.WithCancel(r.Context())
|
|
|
|
session := api.EventStreamService.OpenSession()
|
|
defer func() {
|
|
// Cancel first so the forwarder can distinguish ordinary handler
|
|
// teardown from a registry-initiated close after queue overflow.
|
|
cancel()
|
|
api.EventStreamService.CloseSession(session)
|
|
}()
|
|
|
|
outbound := make(chan any, 64)
|
|
writeErr := make(chan error, 1)
|
|
|
|
go writeEventStreamFrames(ctx, conn, outbound, writeErr)
|
|
go forwardEventStreamEvents(
|
|
ctx,
|
|
api.EventStreamService,
|
|
session,
|
|
outbound,
|
|
func() {
|
|
slog.Warn(
|
|
"event stream subscription closed after outbound queue overflow",
|
|
"event", "event_stream.subscription.overflow",
|
|
)
|
|
_ = conn.Close(
|
|
websocket.StatusTryAgainLater,
|
|
"event stream consumer fell behind; reconnect required",
|
|
)
|
|
},
|
|
)
|
|
|
|
if !enqueueEventStreamFrame(ctx, outbound, readyEventStreamFrame()) {
|
|
return
|
|
}
|
|
|
|
for {
|
|
select {
|
|
case err := <-writeErr:
|
|
if err != nil {
|
|
_ = conn.Close(websocket.StatusInternalError, err.Error())
|
|
}
|
|
return
|
|
default:
|
|
}
|
|
|
|
messageType, payload, err := conn.Read(ctx)
|
|
if err != nil {
|
|
return
|
|
}
|
|
if messageType != websocket.MessageText && messageType != websocket.MessageBinary {
|
|
continue
|
|
}
|
|
|
|
var frameKind eventStreamClientKindFrame
|
|
if err := json.Unmarshal(payload, &frameKind); err != nil {
|
|
if !enqueueEventStreamFrame(ctx, outbound, eventStreamErrorFrame("", &eventstreamservice.ValidationError{
|
|
Code: eventstreamservice.ValidationCodeInvalidPayload,
|
|
Message: "invalid event stream frame",
|
|
})) {
|
|
return
|
|
}
|
|
continue
|
|
}
|
|
|
|
switch strings.TrimSpace(frameKind.Kind) {
|
|
case "subscribe":
|
|
var frame eventprotocol.ClientSubscribeFrame
|
|
if err := strictDecodeEventStreamJSON(payload, &frame); err != nil {
|
|
if !enqueueEventStreamFrame(ctx, outbound, eventStreamErrorFrame(frameKind.RequestID, &eventstreamservice.ValidationError{
|
|
Code: eventstreamservice.ValidationCodeInvalidPayload,
|
|
Message: "invalid subscribe frame",
|
|
})) {
|
|
return
|
|
}
|
|
continue
|
|
}
|
|
|
|
if err := api.EventStreamService.Subscribe(
|
|
session,
|
|
topicsFromGenerated(frame.Topics),
|
|
eventScopeFromGenerated(frame.Scope),
|
|
); err != nil {
|
|
if !enqueueEventStreamFrame(ctx, outbound, eventStreamErrorFrame(frame.RequestID, err)) {
|
|
return
|
|
}
|
|
continue
|
|
}
|
|
if !enqueueEventStreamFrame(ctx, outbound, ackEventStreamFrame(frame.RequestID)) {
|
|
return
|
|
}
|
|
case "unsubscribe":
|
|
var frame eventprotocol.ClientUnsubscribeFrame
|
|
if err := strictDecodeEventStreamJSON(payload, &frame); err != nil {
|
|
if !enqueueEventStreamFrame(ctx, outbound, eventStreamErrorFrame(frameKind.RequestID, &eventstreamservice.ValidationError{
|
|
Code: eventstreamservice.ValidationCodeInvalidPayload,
|
|
Message: "invalid unsubscribe frame",
|
|
})) {
|
|
return
|
|
}
|
|
continue
|
|
}
|
|
|
|
if err := api.EventStreamService.Unsubscribe(
|
|
session,
|
|
topicsFromGenerated(frame.Topics),
|
|
eventScopeFromGenerated(frame.Scope),
|
|
); err != nil {
|
|
if !enqueueEventStreamFrame(ctx, outbound, eventStreamErrorFrame(frame.RequestID, err)) {
|
|
return
|
|
}
|
|
continue
|
|
}
|
|
if !enqueueEventStreamFrame(ctx, outbound, ackEventStreamFrame(frame.RequestID)) {
|
|
return
|
|
}
|
|
case "publish":
|
|
frame, event, err := parseEventStreamClientPublishFrame(payload)
|
|
if err != nil {
|
|
if !enqueueEventStreamFrame(ctx, outbound, eventStreamErrorFrame(frameKind.RequestID, &eventstreamservice.ValidationError{
|
|
Code: eventstreamservice.ValidationCodeInvalidPayload,
|
|
Message: "invalid publish frame",
|
|
})) {
|
|
return
|
|
}
|
|
continue
|
|
}
|
|
|
|
if err := api.EventStreamService.PublishFromClient(ctx, event); err != nil {
|
|
if !enqueueEventStreamFrame(ctx, outbound, eventStreamErrorFrame(frame.RequestID, err)) {
|
|
return
|
|
}
|
|
continue
|
|
}
|
|
if !enqueueEventStreamFrame(ctx, outbound, ackEventStreamFrame(frame.RequestID)) {
|
|
return
|
|
}
|
|
case "ping":
|
|
var frame eventprotocol.ClientPingFrame
|
|
if err := strictDecodeEventStreamJSON(payload, &frame); err != nil {
|
|
if !enqueueEventStreamFrame(ctx, outbound, eventStreamErrorFrame(frameKind.RequestID, &eventstreamservice.ValidationError{
|
|
Code: eventstreamservice.ValidationCodeInvalidPayload,
|
|
Message: "invalid ping frame",
|
|
})) {
|
|
return
|
|
}
|
|
continue
|
|
}
|
|
|
|
if !enqueueEventStreamFrame(ctx, outbound, eventprotocol.ServerPongFrame{
|
|
Kind: "pong",
|
|
RequestID: frame.RequestID,
|
|
SentAt: time.Now().UTC().Format(time.RFC3339Nano),
|
|
}) {
|
|
return
|
|
}
|
|
case "pong":
|
|
continue
|
|
default:
|
|
if !enqueueEventStreamFrame(ctx, outbound, eventStreamErrorFrame(frameKind.RequestID, &eventstreamservice.ValidationError{
|
|
Code: eventstreamservice.ValidationCodeInvalidPayload,
|
|
Message: "unknown event stream frame",
|
|
})) {
|
|
return
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func forwardEventStreamEvents(
|
|
ctx context.Context,
|
|
service EventStreamService,
|
|
session *eventstreamservice.Session,
|
|
outbound chan<- any,
|
|
onSubscriptionClosed func(),
|
|
) {
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case event, ok := <-service.Events(session):
|
|
if !ok {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
default:
|
|
}
|
|
if onSubscriptionClosed != nil {
|
|
onSubscriptionClosed()
|
|
}
|
|
return
|
|
}
|
|
if !enqueueEventStreamFrame(ctx, outbound, eventprotocol.ServerEventFrame{
|
|
Kind: "event",
|
|
Event: generatedEventEnvelope(event),
|
|
}) {
|
|
return
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func writeEventStreamFrames(
|
|
ctx context.Context,
|
|
conn *websocket.Conn,
|
|
outbound <-chan any,
|
|
writeErr chan<- error,
|
|
) {
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
writeErr <- nil
|
|
return
|
|
case frame := <-outbound:
|
|
payload, err := json.Marshal(frame)
|
|
if err != nil {
|
|
writeErr <- err
|
|
return
|
|
}
|
|
if err := conn.Write(ctx, websocket.MessageText, payload); err != nil {
|
|
writeErr <- err
|
|
return
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func enqueueEventStreamFrame(ctx context.Context, outbound chan<- any, frame any) bool {
|
|
select {
|
|
case <-ctx.Done():
|
|
return false
|
|
case outbound <- frame:
|
|
return true
|
|
}
|
|
}
|
|
|
|
func readyEventStreamFrame() eventprotocol.ServerReadyFrame {
|
|
return eventprotocol.ServerReadyFrame{
|
|
Kind: "ready",
|
|
ProtocolVersion: eventprotocol.BusinessEventProtocolVersion,
|
|
CatalogRevision: eventprotocol.BusinessEventCatalogRevision,
|
|
ServerTime: time.Now().UTC().Format(time.RFC3339Nano),
|
|
}
|
|
}
|
|
|
|
func ackEventStreamFrame(requestID string) eventprotocol.ServerAckFrame {
|
|
return eventprotocol.ServerAckFrame{
|
|
Kind: "ack",
|
|
RequestID: strings.TrimSpace(requestID),
|
|
AcceptedAt: time.Now().UTC().Format(time.RFC3339Nano),
|
|
}
|
|
}
|
|
|
|
func eventStreamErrorFrame(requestID string, err error) eventprotocol.ServerErrorFrame {
|
|
frame := eventprotocol.ServerErrorFrame{
|
|
Kind: "error",
|
|
Code: "internal_error",
|
|
Message: err.Error(),
|
|
}
|
|
|
|
if trimmedRequestID := strings.TrimSpace(requestID); trimmedRequestID != "" {
|
|
frame.RequestID = &trimmedRequestID
|
|
}
|
|
|
|
var validationErr *eventstreamservice.ValidationError
|
|
if errors.As(err, &validationErr) {
|
|
frame.Code = string(validationErr.Code)
|
|
frame.Message = validationErr.Message
|
|
}
|
|
|
|
return frame
|
|
}
|
|
|
|
func topicsFromGenerated(topics []eventprotocol.Topic) []string {
|
|
result := make([]string, 0, len(topics))
|
|
for _, topic := range topics {
|
|
result = append(result, string(topic))
|
|
}
|
|
return result
|
|
}
|
|
|
|
func generatedEventEnvelope(event eventstreamservice.PublishedEvent) eventprotocol.EventEnvelope {
|
|
envelope := eventprotocol.EventEnvelope{
|
|
ID: event.ID,
|
|
Topic: eventprotocol.Topic(event.Topic),
|
|
Version: event.Version,
|
|
EmittedAt: event.EmittedAt,
|
|
Payload: append(json.RawMessage(nil), event.Payload...),
|
|
}
|
|
if strings.TrimSpace(event.Scope.WorkspaceID) != "" {
|
|
workspaceID := strings.TrimSpace(event.Scope.WorkspaceID)
|
|
envelope.Scope = &eventprotocol.EventScope{
|
|
WorkspaceID: &workspaceID,
|
|
}
|
|
}
|
|
return envelope
|
|
}
|
|
|
|
func eventScopeFromGenerated(scope *eventprotocol.EventScope) eventstreamservice.EventScope {
|
|
if scope == nil || scope.WorkspaceID == nil {
|
|
return eventstreamservice.EventScope{}
|
|
}
|
|
return eventstreamservice.EventScope{WorkspaceID: *scope.WorkspaceID}
|
|
}
|
|
|
|
func writeEventStreamHTTPError(w http.ResponseWriter, err *apierrors.ProtocolError) {
|
|
if err == nil {
|
|
err = apierrors.ServiceUnavailable(
|
|
apierrors.ReasonEventStreamServiceUnavailable,
|
|
apierrors.WithDeveloperMessage("event stream service is unavailable"),
|
|
)
|
|
}
|
|
tuttitypes.WriteError(
|
|
w,
|
|
err.StatusCode,
|
|
string(err.Code),
|
|
err.Reason,
|
|
err.DeveloperMessage,
|
|
)
|
|
}
|
|
|
|
func strictDecodeEventStreamJSON(payload []byte, target any) error {
|
|
decoder := json.NewDecoder(bytes.NewReader(payload))
|
|
decoder.DisallowUnknownFields()
|
|
if err := decoder.Decode(target); err != nil {
|
|
return err
|
|
}
|
|
|
|
var extra json.RawMessage
|
|
if err := decoder.Decode(&extra); !errors.Is(err, io.EOF) {
|
|
if err == nil {
|
|
return fmt.Errorf("unexpected trailing JSON content")
|
|
}
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func parseEventStreamClientPublishFrame(payload []byte) (eventStreamClientPublishFrame, eventstreamservice.ClientEvent, error) {
|
|
var frame eventStreamClientPublishFrame
|
|
if err := strictDecodeEventStreamJSON(payload, &frame); err != nil {
|
|
return eventStreamClientPublishFrame{}, eventstreamservice.ClientEvent{}, err
|
|
}
|
|
|
|
var envelope eventprotocol.EventEnvelope
|
|
if err := strictDecodeEventStreamJSON(frame.Event, &envelope); err != nil {
|
|
return eventStreamClientPublishFrame{}, eventstreamservice.ClientEvent{}, err
|
|
}
|
|
if err := validateClientPublishEnvelope(envelope); err != nil {
|
|
return eventStreamClientPublishFrame{}, eventstreamservice.ClientEvent{}, err
|
|
}
|
|
|
|
definition, ok := eventprotocol.LookupEventDefinition(envelope.Topic)
|
|
if !ok {
|
|
return eventStreamClientPublishFrame{}, eventstreamservice.ClientEvent{}, &eventstreamservice.ValidationError{
|
|
Code: eventstreamservice.ValidationCodeInvalidTopic,
|
|
Message: fmt.Sprintf("unknown topic %q", strings.TrimSpace(string(envelope.Topic))),
|
|
Topic: strings.TrimSpace(string(envelope.Topic)),
|
|
}
|
|
}
|
|
if definition.Direction != eventprotocol.DirectionClientToServer {
|
|
return eventStreamClientPublishFrame{}, eventstreamservice.ClientEvent{}, &eventstreamservice.ValidationError{
|
|
Code: eventstreamservice.ValidationCodeInvalidDirection,
|
|
Message: fmt.Sprintf("topic %q does not allow %s", definition.Topic, eventstreamservice.DirectionClientToServer),
|
|
Topic: string(definition.Topic),
|
|
Direction: eventstreamservice.DirectionClientToServer,
|
|
}
|
|
}
|
|
if envelope.Version != definition.Version {
|
|
return eventStreamClientPublishFrame{}, eventstreamservice.ClientEvent{}, &eventstreamservice.ValidationError{
|
|
Code: eventstreamservice.ValidationCodeInvalidPayload,
|
|
Message: fmt.Sprintf("event version must be %d for topic %q", definition.Version, definition.Topic),
|
|
Topic: string(definition.Topic),
|
|
Direction: eventstreamservice.DirectionClientToServer,
|
|
}
|
|
}
|
|
|
|
prototype, ok := eventprotocol.EventPrototypeForTopic(envelope.Topic)
|
|
if !ok {
|
|
return eventStreamClientPublishFrame{}, eventstreamservice.ClientEvent{}, &eventstreamservice.ValidationError{
|
|
Code: eventstreamservice.ValidationCodeInvalidTopic,
|
|
Message: fmt.Sprintf("unknown topic %q", strings.TrimSpace(string(envelope.Topic))),
|
|
Topic: strings.TrimSpace(string(envelope.Topic)),
|
|
}
|
|
}
|
|
if err := strictDecodeEventStreamJSON(frame.Event, prototype); err != nil {
|
|
return eventStreamClientPublishFrame{}, eventstreamservice.ClientEvent{}, err
|
|
}
|
|
|
|
return frame, eventstreamservice.ClientEvent{
|
|
Topic: strings.TrimSpace(string(envelope.Topic)),
|
|
Payload: append([]byte(nil), envelope.Payload...),
|
|
}, nil
|
|
}
|
|
|
|
func validateClientPublishEnvelope(envelope eventprotocol.EventEnvelope) error {
|
|
if strings.TrimSpace(envelope.ID) == "" {
|
|
return &eventstreamservice.ValidationError{
|
|
Code: eventstreamservice.ValidationCodeInvalidPayload,
|
|
Message: "event id must not be empty",
|
|
Topic: strings.TrimSpace(string(envelope.Topic)),
|
|
}
|
|
}
|
|
if len(envelope.Payload) == 0 {
|
|
return &eventstreamservice.ValidationError{
|
|
Code: eventstreamservice.ValidationCodeInvalidPayload,
|
|
Message: "event payload is required",
|
|
Topic: strings.TrimSpace(string(envelope.Topic)),
|
|
}
|
|
}
|
|
|
|
emittedAt := strings.TrimSpace(envelope.EmittedAt)
|
|
if emittedAt == "" {
|
|
return &eventstreamservice.ValidationError{
|
|
Code: eventstreamservice.ValidationCodeInvalidPayload,
|
|
Message: "event emittedAt must not be empty",
|
|
Topic: strings.TrimSpace(string(envelope.Topic)),
|
|
}
|
|
}
|
|
if _, err := time.Parse(time.RFC3339Nano, emittedAt); err != nil {
|
|
return &eventstreamservice.ValidationError{
|
|
Code: eventstreamservice.ValidationCodeInvalidPayload,
|
|
Message: "event emittedAt must be a valid RFC3339 timestamp",
|
|
Topic: strings.TrimSpace(string(envelope.Topic)),
|
|
}
|
|
}
|
|
|
|
if envelope.Scope != nil && envelope.Scope.WorkspaceID != nil && strings.TrimSpace(*envelope.Scope.WorkspaceID) == "" {
|
|
return &eventstreamservice.ValidationError{
|
|
Code: eventstreamservice.ValidationCodeInvalidPayload,
|
|
Message: "event scope.workspaceId must not be empty",
|
|
Topic: strings.TrimSpace(string(envelope.Topic)),
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|