wip

pkg/loop/internal/core/services/capability/capabilities.go

@@ -215,7 +215,27 @@ func (t *triggerExecutableServer) RegisterTrigger(request *capabilitiespb.Trigge
 	}
 	responseCh, err := t.impl.RegisterTrigger(server.Context(), req)
 	if err != nil {
-		return fmt.Errorf("error registering trigger: %w", err)
+		// we send the error back
+		msg := &capabilitiespb.TriggerResponseMessage{
+			Message: &capabilitiespb.TriggerResponseMessage_Response{
+				Response: &capabilitiespb.TriggerResponse{
+					Error: err.Error(),
+				},
+			},
+		}
+		if err = server.Send(msg); err != nil {
+			return fmt.Errorf("error sending ERROR response for trigger %s: %w", request, err)
+		}
+		// return fmt.Errorf("error registering trigger server: %w", err)
+	}
+
+	msg := &capabilitiespb.TriggerResponseMessage{
+		Message: &capabilitiespb.TriggerResponseMessage_Response{
+			Response: &capabilitiespb.TriggerResponse{},
+		},
+	}
+	if err = server.Send(msg); err != nil {
+		return fmt.Errorf("error sending ACK response for trigger %s: %w", request, err)
 	}
 
 	defer func() {
@@ -268,6 +288,13 @@ func (t *triggerExecutableClient) RegisterTrigger(ctx context.Context, req capab
 	responseStream, err := t.grpc.RegisterTrigger(ctx, pb.TriggerRegistrationRequestToProto(req))
 	if err != nil {
 		return nil, fmt.Errorf("error registering trigger: %w", err)
+	ackMsg, err := responseStream.Recv()
+	if err != nil {
+		return nil, fmt.Errorf("failed to receive first registering trigger response: %w", err)
+	}
+
+	if ackMsg.GetResponse().GetError() != "" {
+		return nil, errors.New(fmt.Sprintf("failed registering trigger: %s", ackMsg.GetResponse().GetError()))
 	}
 
 	return forwardTriggerResponsesToChannel(ctx, t.Logger, req, responseStream.Recv)