fix: marshaling of WithSubject events

If a struct is composed with an interface the go json marshaler
treats that as a field, thus we need to have a custom json marshaler
for the type.

openmeter/watermill/marshaler/source.go

@@ -1,6 +1,7 @@
 package marshaler
 
 import (
+	"encoding/json"
 	"errors"
 
 	"github.com/openmeterio/openmeter/internal/event/metadata"
@@ -42,3 +43,9 @@ func (e *eventWithSource) Validate() error {
 func (e *eventWithSource) EventName() string {
 	return e.Event.EventName()
 }
+
+// MarshalJSON marshals the event only, as JSON library embeds the Event name into the output,
+// if the composed object is a pointer to an interface. (e.g. we would get "Event": {} in the payload)
+func (e *eventWithSource) MarshalJSON() ([]byte, error) {
+	return json.Marshal(e.Event)
+}

openmeter/watermill/marshaler/source_test.go

@@ -0,0 +1,47 @@
+package marshaler
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/openmeterio/openmeter/internal/event/metadata"
+)
+
+type event struct {
+	Value string `json:"value"`
+}
+
+func (e *event) EventMetadata() metadata.EventMetadata {
+	return metadata.EventMetadata{}
+}
+
+func (e *event) Validate() error {
+	return nil
+}
+
+func (e *event) EventName() string {
+	return "event"
+}
+
+func TestWithSubject(t *testing.T) {
+	marshaler := New(nil)
+
+	ev := &event{
+		Value: "value",
+	}
+
+	evWithSource := WithSource("source", ev)
+	msg, err := marshaler.Marshal(evWithSource)
+
+	// Check if the source is set in the metadata
+	assert.NoError(t, err)
+	assert.Equal(t, "source", msg.Metadata.Get(CloudEventsHeaderSource))
+
+	// Check if the event can be unmarshaled
+	unmarshaledEvent := &event{}
+	err = marshaler.Unmarshal(msg, unmarshaledEvent)
+	assert.NoError(t, err)
+
+	assert.Equal(t, ev, unmarshaledEvent)
+}