Big refactor, fixes both wasm and podman

Now the scripts are fetched when the message is received and each scripts
gets called with their respective executor. This simplifies greatly the
executors as they no longer have to fetch the scripts themselves. They
retain the store if they need to fetch more things (like libraries for
the lua executor).

Author: numkem

.dir-locals.el

@@ -0,0 +1,2 @@
+((go-mode . ((eglot-workspace-configuration . 
+              (:gopls (:buildFlags ["-tags=podman,wasmtime"]))))))

cmd/server/http.go

@@ -50,7 +50,7 @@ func NewHttpNatsProxy(port int, natsURL string) (*httpNatsProxy, error) {
 func (p *httpNatsProxy) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	// Extract context from incoming request headers
 	ctx := otel.GetTextMapPropagator().Extract(r.Context(), propagation.HeaderCarrier(r.Header))
-	
+
 	// Start a new span for the HTTP request
 	ctx, span := tracer.Start(ctx, "http.request",
 		trace.WithSpanKind(trace.SpanKindServer),
@@ -112,8 +112,9 @@ func (p *httpNatsProxy) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	ctx, cancel := context.WithTimeout(ctx, timeout)
 	defer cancel()
 
+	// Change the url passed to the fuction to remove the subject
 	url := strings.ReplaceAll(r.URL.String(), "/"+subject, "")
-	log.Debug(url)
+	log.Debugf("URL: %s", url)
 	body, err := json.Marshal(&executor.Message{
 		Payload: payload,
 		Method:  r.Method,
@@ -141,13 +142,16 @@ func (p *httpNatsProxy) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	msg.Data = body
 	otel.GetTextMapPropagator().Inject(ctx, natsHeaderCarrier(msg.Header))
 
+	// Send the message and wait for the response
 	response, err := p.nc.RequestMsgWithContext(ctx, msg)
 	if err != nil {
 		natsSpan.RecordError(err)
 		natsSpan.SetStatus(codes.Error, "NATS request failed")
 		natsSpan.End()
+
 		span.SetStatus(codes.Error, "Service unavailable")
 		span.SetAttributes(attribute.Int("http.status_code", http.StatusServiceUnavailable))
+
 		w.WriteHeader(http.StatusServiceUnavailable)
 		w.Write([]byte(err.Error()))
 		return
@@ -156,7 +160,7 @@ func (p *httpNatsProxy) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	natsSpan.SetStatus(codes.Ok, "")
 	natsSpan.End()
 
-	rep := new(executor.Reply)
+	rep := new(Reply)
 	err = json.Unmarshal(response.Data, rep)
 	if err != nil {
 		span.RecordError(err)
@@ -188,38 +192,41 @@ func (p *httpNatsProxy) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	}
 
 	// Go through all the scripts to see if one is HTML
-	if t, sr := hasHTMLResult(rep.AllResults); t {
-		span.SetAttributes(attribute.Bool("response.is_html", true))
-		var hasContentType bool
-		for k, v := range sr.Headers {
-			if k == "Content-Type" {
-				hasContentType = true
+	for _, scrRes := range rep.Results {
+		if scrRes.IsHTML {
+
+			span.SetAttributes(attribute.Bool("response.is_html", true))
+			var hasContentType bool
+			for k, v := range scrRes.Headers {
+				if k == "Content-Type" {
+					hasContentType = true
+				}
+				w.Header().Add(k, v)
+			}
+			if !hasContentType {
+				w.Header().Add("Content-Type", "text/html")
+			}
+			span.SetAttributes(
+				attribute.Int("http.status_code", scrRes.Code),
+				attribute.Int("http.response.body_size", len(scrRes.Payload)),
+			)
+			w.WriteHeader(scrRes.Code)
+
+			_, err = w.Write(scrRes.Payload)
+			if err != nil {
+				span.RecordError(err)
+				log.WithFields(fields).Errorf("failed to write reply back to HTTP response: %v", err)
 			}
-			w.Header().Add(k, v)
-		}
-		if !hasContentType {
-			w.Header().Add("Content-Type", "text/html")
-		}
-		span.SetAttributes(
-			attribute.Int("http.status_code", sr.Code),
-			attribute.Int("http.response.body_size", len(sr.Payload)),
-		)
-		w.WriteHeader(sr.Code)
 
-		_, err = w.Write(sr.Payload)
-		if err != nil {
-			span.RecordError(err)
-			log.WithFields(fields).Errorf("failed to write reply back to HTTP response: %v", err)
+			span.SetStatus(codes.Ok, "")
+			// Since only the HTML page reply can "win" we ignore the rest
+			return
 		}
-
-		span.SetStatus(codes.Ok, "")
-		// Since only the HTML page reply can "win" we ignore the rest
-		return
 	}
 
 	// Convert the results to bytes
 	span.SetAttributes(attribute.Bool("response.is_html", false))
-	rr, err := json.Marshal(rep.AllResults)
+	rr, err := json.Marshal(rep.Results)
 	if err != nil {
 		span.RecordError(err)
 		log.WithFields(fields).Errorf("failed to serialize all results to JSON: %v", err)

cmd/server/main.go

@@ -4,9 +4,11 @@ import (
 	"encoding/json"
 	"errors"
 	"flag"
+	"fmt"
 	"os"
 	"os/signal"
 	"strings"
+	"sync"
 	"time"
 
 	natsserver "github.com/nats-io/nats-server/v2/server"
@@ -22,6 +24,7 @@ import (
 	"github.com/numkem/msgscript"
 	"github.com/numkem/msgscript/executor"
 	msgplugin "github.com/numkem/msgscript/plugins"
+	"github.com/numkem/msgscript/script"
 	msgstore "github.com/numkem/msgscript/store"
 )
 
@@ -171,7 +174,6 @@ func main() {
 		span.SetAttributes(
 			attribute.Bool("message.raw", m.Raw),
 			attribute.Bool("message.async", m.Async),
-			attribute.String("message.executor", m.Executor),
 		)
 
 		// The above unmarshalling only applies to the structure of the JSON.
@@ -184,33 +186,90 @@ func main() {
 			}
 		}
 
-		replier := &replier{nc: nc}
-		var messageReply executor.ReplyFunc
 		if m.Async {
 			span.SetAttributes(attribute.String("reply.mode", "async"))
-			messageReply = replier.AsyncReply(m, msg)
 			err = nc.Publish(msg.Reply, []byte("{}"))
 			if err != nil {
 				span.RecordError(err)
 				span.SetStatus(codes.Error, "Failed to publish async reply")
 				log.WithFields(fields).Errorf("failed to reply to message: %v", err)
+
+				replyWithError(nc, fmt.Errorf("failed to reply to message: %v", err), msg.Reply)
 				return
 			}
 		} else {
 			span.SetAttributes(attribute.String("reply.mode", "sync"))
-			messageReply = replier.SyncReply(m, msg)
 		}
 
-		exec, err := executor.ExecutorByName(m.Executor, executors)
+		cctx, getScriptsSpan := mainTracer.Start(ctx, "nats.handle_message.get_scripts", trace.WithAttributes(
+			attribute.String("script.Name", m.Subject),
+			attribute.String("script.URL", m.URL),
+		))
+
+		scripts, err := scriptStore.GetScripts(cctx, m.Subject)
 		if err != nil {
+			log.WithError(err).WithField("subject", m.Subject).Error("failed to get scripts for subject")
 			span.RecordError(err)
-			span.SetStatus(codes.Error, "Failed to get executor")
-			log.WithError(err).Error("failed to get executor for message")
+			span.SetStatus(codes.Error, "Failed to get scripts")
+
+			replyWithError(nc, fmt.Errorf("failed to get scripts for subject"), msg.Reply)
+			return
+		}
+		getScriptsSpan.SetStatus(codes.Ok, fmt.Sprintf("found %d scripts", len(scripts)))
+		getScriptsSpan.End()
+
+		_, executeScriptsSpan := mainTracer.Start(ctx, "nats.handle_message.run_scripts")
+		defer executeScriptsSpan.End()
+
+		var wg sync.WaitGroup
+		allResults := make(chan *executor.ScriptResult, len(scripts))
+		for _, scr := range scripts {
+			wg.Add(1)
+			go func(ctx context.Context, msg *executor.Message, script *script.Script) {
+				defer wg.Done()
+
+				// Pass the context with trace info to the executor
+				exec, err := executor.ExecutorByName(scr.Executor, executors)
+				if err != nil {
+					executeScriptsSpan.RecordError(err)
+					executeScriptsSpan.SetStatus(codes.Error, "Failed to get executor")
+					log.WithError(err).Error("failed to get executor for script")
+
+					allResults <- &executor.ScriptResult{Error: fmt.Sprintf("failed to get executor for script: %v", err)}
+					return
+				}
+
+				rep := exec.HandleMessage(ctx, m, scr)
+				allResults <- rep
+			}(ctx, m, scr)
+		}
+		wg.Wait()
+
+		close(allResults)
+
+		_, parseReplySpan := mainTracer.Start(ctx, "nats.handle_message.parse_replies")
+		msgRep := new(Reply)
+		for res := range allResults {
+			if res.IsHTML {
+				msgRep.HTML = true
+			}
+
+			msgRep.Results = append(msgRep.Results, res)
+		}
+		parseReplySpan.SetStatus(codes.Ok, "responses parsed")
+		parseReplySpan.End()
+
+		_, natsReplaySpan := mainTracer.Start(ctx, "nats.handle_message.send_reply")
+		err = replyMessage(nc, m, msg.Reply, msgRep)
+		if err != nil {
+			natsReplaySpan.RecordError(err)
+			natsReplaySpan.SetStatus(codes.Error, "Failed to send reply through NATS")
+
+			log.WithError(err).Errorf("failed to send reply through NATS")
 			return
 		}
 
-		// Pass the context with trace info to the executor
-		exec.HandleMessage(ctx, m, messageReply)
+		log.WithField("subject", msg.Subject).Debugf("finished running %d scripts", len(scripts))
 		span.SetStatus(codes.Ok, "Message handled")
 	})
 	if err != nil {

cmd/server/reply.go

@@ -1,68 +1,58 @@
 package main
 
 import (
+	"encoding/json"
+	"fmt"
+
 	"github.com/nats-io/nats.go"
 	log "github.com/sirupsen/logrus"
 
 	"github.com/numkem/msgscript/executor"
 )
 
-type replier struct {
-	nc *nats.Conn
+type Reply struct {
+	Results []*executor.ScriptResult `json:"script_result"`
+	HTML    bool                     `json:"is_html"`
+	Error   string                   `json:"error,omitempty"`
 }
 
-func (rep *replier) SyncReply(m *executor.Message, msg *nats.Msg) executor.ReplyFunc {
-	return func(r *executor.Reply) {
-		fields := log.Fields{
-			"Subject": m.Subject,
-			"URL":     m.URL,
-			"Method":  m.Method,
-		}
-
-		if r.Error != "" {
-			log.WithFields(fields).Errorf("error while running script: %s", r.Error)
-		}
+func replyMessage(nc *nats.Conn, msg *executor.Message, replySubject string, rep *Reply) error {
+	fields := log.Fields{
+		"Subject": msg.Subject,
+		"URL":     msg.URL,
+		"Method":  msg.Method,
+	}
 
-		// Send a reply if the message has a reply subject
-		if msg.Reply == "" {
-			return
-		}
+	// Send a reply if the message has a reply subject
+	if replySubject == "" {
+		return nil
+	}
 
-		var reply []byte
-		var err error
-		if !m.Raw {
-			reply, err = r.JSON()
-			if err != nil {
-				log.WithFields(fields).Errorf("failed to serialize script reply to JSON: %v", err)
-			}
-		} else {
-			if r.Error != "" {
-				reply = []byte(r.Error)
-			} else {
-				reply = r.Bytes()
-			}
-		}
+	var payload []byte
+	var err error
+	payload, err = json.Marshal(rep)
+	if err != nil {
+		log.WithFields(fields).Errorf("failed to serialize script reply to JSON: %v", err)
+		return fmt.Errorf("failed to serialize script reply to JSON: %v", err)
+	}
 
-		log.WithFields(fields).Debugf("sent reply: %s", reply)
-		err = rep.nc.Publish(msg.Reply, reply)
-		if err != nil {
-			log.WithFields(fields).Errorf("failed to publish reply after running script: %v", err)
-		}
+	log.WithFields(fields).Debugf("sent reply: %s", string(payload))
+	err = nc.Publish(replySubject, payload)
+	if err != nil {
+		log.WithFields(fields).Errorf("failed to publish reply after running script: %v", err)
 	}
-}
 
-func (re *replier) AsyncReply(m *executor.Message, msg *nats.Msg) executor.ReplyFunc {
-	return func(r *executor.Reply) {
-		fields := log.Fields{
-			"Subject": m.Subject,
-			"URL":     m.URL,
-			"Method":  m.Method,
-		}
+	return nil
+}
 
-		if r.Error != "" {
-			log.WithFields(fields).Errorf("error while running script: %s", r.Error)
-		}
+func replyWithError(nc *nats.Conn, resErr error, replySubject string) {
+	payload, err := json.Marshal(&Reply{Error: resErr.Error()})
+	if err != nil {
+		log.Errorf("failed to serialize script reply to JSON: %v", err)
+	}
 
-		log.WithFields(fields).Debug("async reply received")
+	err = nc.Publish(replySubject, payload)
+	if err != nil {
+		log.Errorf("failed to reply with error: %v", err)
 	}
 }