77
88 "github.com/adfinis-sygroup/mopsos/app/models"
99 otelObs "github.com/cloudevents/sdk-go/observability/opentelemetry/v2/client"
10- cloudevents "github.com/cloudevents/sdk-go/v2"
1110 "github.com/cloudevents/sdk-go/v2/binding"
12- "github.com/cloudevents/sdk-go/v2/protocol "
11+ "github.com/cloudevents/sdk-go/v2/event "
1312 httproto "github.com/cloudevents/sdk-go/v2/protocol/http"
1413 http_logrus "github.com/improbable-eng/go-httpwares/logging/logrus"
1514 "github.com/sirupsen/logrus"
@@ -20,9 +19,15 @@ import (
2019type Server struct {
2120 config * Config
2221
23- EventChan chan <- cloudevents. Event
22+ EventChan chan <- models. EventData
2423}
2524
25+ type eventContext string
26+
27+ var ContextUsername eventContext = "mopsos.username"
28+ var ContextEvent eventContext = "mopsos.event"
29+ var ContextRecord eventContext = "mopsos.record"
30+
2631// NewServer creates a server that receives CloudEvents from the network
2732func NewServer (cfg * Config ) * Server {
2833 return & Server {
@@ -33,91 +38,116 @@ func NewServer(cfg *Config) *Server {
3338// Start starts the server and listens for incoming events
3439func (s * Server ) Start () {
3540 mux := http .NewServeMux ()
36- mux .HandleFunc ("/health" , func (w http.ResponseWriter , r * http.Request ) {
37- // an example API handler
38- err := json .NewEncoder (w ).Encode (map [string ]bool {"ok" : true })
39- if err != nil {
40- logrus .WithError (err ).Error ("error encoding response" )
41- }
42- })
43- mux .Handle ("/webhook" , otelhttp .NewHandler (http .HandlerFunc (func (w http.ResponseWriter , r * http.Request ) {
44- ctx := r .Context ()
41+ mux .HandleFunc ("/health" , s .HandleHealthCheck )
42+ mux .Handle ("/webhook" , otelhttp .NewHandler (
43+ s .Authenticate (
44+ s .LoadEvent (
45+ s .Validate (
46+ http .HandlerFunc (s .HandleWebhook ),
47+ ),
48+ ),
49+ ),
50+ "webhook-receiver" ),
51+ )
52+
53+ logrus .WithField ("listener" , s .config .HttpListener ).Info ("Starting server" )
54+ loggingMiddleware := http_logrus .Middleware (
55+ logrus .WithFields (logrus.Fields {}),
56+ )(mux )
57+ logrus .Fatal (http .ListenAndServe (s .config .HttpListener , loggingMiddleware ))
58+ }
4559
60+ // WithEventChannel sets the event channel for the server
61+ func (s * Server ) WithEventChannel (eventChan chan <- models.EventData ) * Server {
62+ s .EventChan = eventChan
63+ return s
64+ }
65+
66+ // Authenticate middleware handles checking credentials
67+ func (s * Server ) Authenticate (next http.Handler ) http.Handler {
68+ return http .HandlerFunc (func (w http.ResponseWriter , r * http.Request ) {
4669 // get basic auth credentials
4770 username , password , ok := r .BasicAuth ()
4871 if ! ok {
4972 http .Error (w , "missing Authorization header" , http .StatusUnauthorized )
5073 return
5174 }
52- if ! s .checkAuth (username , password ) {
75+
76+ logrus .WithFields (logrus.Fields {
77+ "username" : username ,
78+ }).Debug ("checking credentials" )
79+ if s .config .BasicAuthUsers [username ] != password {
5380 http .Error (w , "invalid credentials" , http .StatusUnauthorized )
5481 return
5582 }
83+ ctx := context .WithValue (r .Context (), ContextUsername , username )
5684
85+ next .ServeHTTP (w , r .WithContext (ctx ))
86+ })
87+ }
88+
89+ // LoadEvent middlerware loads event from the request
90+ func (s * Server ) LoadEvent (next http.Handler ) http.Handler {
91+ return http .HandlerFunc (func (w http.ResponseWriter , r * http.Request ) {
5792 // get event
5893 message := httproto .NewMessageFromHttpRequest (r )
59- event , err := binding .ToEvent (context . TODO (), message )
94+ event , err := binding .ToEvent (r . Context (), message )
6095 if err != nil {
6196 logrus .WithError (err ).Error ("failed to decode event" )
6297 return
6398 }
99+ if s .config .EnableTracing {
100+ // inject the span context into the event so it can be use i.e. while inserting to the database
101+ otelObs .InjectDistributedTracingExtension (r .Context (), * event )
102+ }
103+ logrus .Debugf ("received event: %v" , event )
64104
65- // TODO consider how to harmonise this with what the handler does later on
105+ ctx := context .WithValue (r .Context (), ContextEvent , event )
106+ next .ServeHTTP (w , r .WithContext (ctx ))
107+ })
108+ }
109+
110+ // Validate middleware handles checking received events for validity
111+ func (s * Server ) Validate (next http.Handler ) http.Handler {
112+ return http .HandlerFunc (func (w http.ResponseWriter , r * http.Request ) {
113+ event := r .Context ().Value (ContextEvent ).(* event.Event )
66114 record := & models.Record {}
115+
67116 if err := event .DataAs (record ); err != nil {
68117 logrus .WithError (err ).Errorf ("failed to unmarshal event data" )
69118 http .Error (w , "failed to unmarshal event data" , http .StatusInternalServerError )
70119 return
71120 }
72121
73122 // reject record that have not been sent from the right auth
74- if record .ClusterName != username {
123+ if record .ClusterName != r . Context (). Value ( ContextUsername ).( string ) {
75124 http .Error (w , "event data does not match username" , http .StatusUnauthorized )
76125 return
77126 }
78127
79- err = s .HandleReceivedEvent (ctx , * event )
80- if err != nil {
81- logrus .WithError (err ).Error ("failed to handle event" )
82- return
83- }
84- // return 202 accepted once the event is on the queue
85- w .WriteHeader (http .StatusAccepted )
86- }), "webhook-receiver" ))
87-
88- logrus .WithField ("listener" , s .config .HttpListener ).Info ("Starting server" )
89- loggingMiddleware := http_logrus .Middleware (
90- logrus .WithFields (logrus.Fields {}),
91- )(mux )
92- logrus .Fatal (http .ListenAndServe (s .config .HttpListener , loggingMiddleware ))
128+ ctx := context .WithValue (r .Context (), ContextRecord , record )
129+ next .ServeHTTP (w , r .WithContext (ctx ))
130+ })
93131}
94132
95- // WithEventChannel sets the event channel for the server
96- func (s * Server ) WithEventChannel (eventChan chan <- cloudevents.Event ) * Server {
97- s .EventChan = eventChan
98- return s
133+ func (s * Server ) HandleHealthCheck (w http.ResponseWriter , r * http.Request ) {
134+ // an example API handler
135+ err := json .NewEncoder (w ).Encode (map [string ]bool {"ok" : true })
136+ if err != nil {
137+ logrus .WithError (err ).Error ("error encoding response" )
138+ }
99139}
100140
101- // HandleReceivedEvent is the handler for the cloudevents receiver, public for testing
102- func (s * Server ) HandleReceivedEvent (ctx context.Context , event cloudevents.Event ) protocol.Result {
103-
104- if s .config .EnableTracing {
105- // inject the span context into the event so it can be use i.e. while inserting to the database
106- otelObs .InjectDistributedTracingExtension (ctx , event )
107- }
141+ func (s * Server ) HandleWebhook (w http.ResponseWriter , r * http.Request ) {
142+ // get middleware data from context
143+ event := r .Context ().Value (ContextEvent ).(* event.Event )
144+ record := r .Context ().Value (ContextRecord ).(* models.Record )
108145
109146 // send the event to the main app via the async channel
110- s .EventChan <- event
111-
112- logrus .Debugf ("received event: %v" , event )
113-
114- return nil
115- }
116-
117- // checkAuth checks if the username and password are correct
118- func (s * Server ) checkAuth (username , password string ) bool {
119- logrus .WithFields (logrus.Fields {
120- "username" : username ,
121- }).Debug ("checking credentials" )
122- return s .config .BasicAuthUsers [username ] == password
147+ s .EventChan <- models.EventData {
148+ Event : * event ,
149+ Record : * record ,
150+ }
151+ // return 202 accepted once the event is on the queue
152+ w .WriteHeader (http .StatusAccepted )
123153}
0 commit comments