feat: refactor HTTPServer to use pipeline

cmd/oastrix/server.go

@@ -13,6 +13,9 @@ import (
 	"github.com/rsclarke/oastrix/internal/auth"
 	"github.com/rsclarke/oastrix/internal/db"
 	"github.com/rsclarke/oastrix/internal/logging"
+	"github.com/rsclarke/oastrix/internal/plugins"
+	"github.com/rsclarke/oastrix/internal/plugins/core/defaultresponse"
+	"github.com/rsclarke/oastrix/internal/plugins/core/storage"
 	"github.com/rsclarke/oastrix/internal/server"
 	"github.com/spf13/cobra"
 	"go.uber.org/zap"
@@ -109,8 +112,23 @@ func runServer(cmd *cobra.Command, args []string) error {
 		acme.SetLogger(logger.Named("certmagic"))
 	}
 
+	pipeline := plugins.NewPipeline(logger.Named("pipeline"))
+
+	storagePlugin := storage.New(database)
+	if err := storagePlugin.Init(plugins.InitContext{Logger: logger.Named("storage")}); err != nil {
+		return fmt.Errorf("init storage plugin: %w", err)
+	}
+	pipeline.SetStore(storagePlugin)
+	pipeline.Register(storagePlugin)
+
+	defaultResp := defaultresponse.New(serverFlags.publicIP)
+	if err := defaultResp.Init(plugins.InitContext{Logger: logger.Named("defaultresponse")}); err != nil {
+		return fmt.Errorf("init defaultresponse plugin: %w", err)
+	}
+	pipeline.Register(defaultResp)
+
 	httpSrv := &server.HTTPServer{
-		DB:       database,
+		Pipeline: pipeline,
 		Domain:   serverFlags.domain,
 		PublicIP: serverFlags.publicIP,
 		Logger:   logger.Named("http"),

internal/plugins/core/storage/plugin.go

@@ -53,9 +53,24 @@ func (p *Plugin) OnPreStore(ctx context.Context, e *events.Event) error {
 	return nil
 }
 
-// Store persists an interaction draft to the database and returns the interaction ID.
-// It creates the base interaction, the kind-specific record (HTTP or DNS), and any attributes.
-func (p *Plugin) Store(ctx context.Context, draft *events.InteractionDraft) (int64, error) {
+// ResolveTokenID looks up a token by its value and returns the ID.
+func (p *Plugin) ResolveTokenID(_ context.Context, tokenValue string) (int64, bool, error) {
+	token, err := db.GetTokenByValue(p.db, tokenValue)
+	if err != nil {
+		return 0, false, err
+	}
+	if token == nil {
+		return 0, false, nil
+	}
+	return token.ID, true, nil
+}
+
+// CreateInteraction persists an interaction draft to the database and returns the interaction ID.
+func (p *Plugin) CreateInteraction(_ context.Context, draft *events.InteractionDraft) (int64, error) {
+	if draft.TokenID == 0 {
+		return 0, nil
+	}
+
 	id, err := db.CreateInteraction(
 		p.db,
 		draft.TokenID,
@@ -111,11 +126,10 @@ func (p *Plugin) Store(ctx context.Context, draft *events.InteractionDraft) (int
 		}
 	}
 
-	if len(draft.Attributes) > 0 {
-		if err := db.SaveAttributes(p.db, id, draft.Attributes); err != nil {
-			return 0, fmt.Errorf("save attributes: %w", err)
-		}
-	}
-
 	return id, nil
 }
+
+// SaveAttributes persists plugin attributes for an interaction.
+func (p *Plugin) SaveAttributes(_ context.Context, interactionID int64, attrs map[string]any) error {
+	return db.SaveAttributes(p.db, interactionID, attrs)
+}

internal/plugins/core/storage/plugin_test.go

@@ -131,6 +131,62 @@ func TestOnPreStoreUnknownToken(t *testing.T) {
 	}
 }
 
+func TestResolveTokenID(t *testing.T) {
+	database := setupTestDB(t)
+	p := New(database)
+	_ = p.Init(plugins.InitContext{Logger: zap.NewNop()})
+
+	tokenID, err := db.CreateToken(database, "test-token", nil, nil)
+	if err != nil {
+		t.Fatalf("CreateToken failed: %v", err)
+	}
+
+	id, found, err := p.ResolveTokenID(context.Background(), "test-token")
+	if err != nil {
+		t.Fatalf("ResolveTokenID failed: %v", err)
+	}
+	if !found {
+		t.Error("expected token to be found")
+	}
+	if id != tokenID {
+		t.Errorf("ID = %d, want %d", id, tokenID)
+	}
+
+	id, found, err = p.ResolveTokenID(context.Background(), "nonexistent")
+	if err != nil {
+		t.Fatalf("ResolveTokenID failed: %v", err)
+	}
+	if found {
+		t.Error("expected token not to be found")
+	}
+	if id != 0 {
+		t.Errorf("ID = %d, want 0", id)
+	}
+}
+
+func TestCreateInteractionSkipsZeroTokenID(t *testing.T) {
+	database := setupTestDB(t)
+	p := New(database)
+	_ = p.Init(plugins.InitContext{Logger: zap.NewNop()})
+
+	draft := &events.InteractionDraft{
+		TokenID:    0,
+		Kind:       events.KindHTTP,
+		RemoteIP:   "192.168.1.1",
+		RemotePort: 54321,
+		TLS:        false,
+		Summary:    "GET /",
+	}
+
+	id, err := p.CreateInteraction(context.Background(), draft)
+	if err != nil {
+		t.Fatalf("CreateInteraction failed: %v", err)
+	}
+	if id != 0 {
+		t.Errorf("expected 0 ID for zero TokenID, got %d", id)
+	}
+}
+
 func TestStoreHTTPInteraction(t *testing.T) {
 	database := setupTestDB(t)
 	p := New(database)
@@ -160,7 +216,7 @@ func TestStoreHTTPInteraction(t *testing.T) {
 		},
 	}
 
-	id, err := p.Store(context.Background(), draft)
+	id, err := p.CreateInteraction(context.Background(), draft)
 	if err != nil {
 		t.Fatalf("Store failed: %v", err)
 	}
@@ -220,7 +276,7 @@ func TestStoreDNSInteraction(t *testing.T) {
 		},
 	}
 
-	id, err := p.Store(context.Background(), draft)
+	id, err := p.CreateInteraction(context.Background(), draft)
 	if err != nil {
 		t.Fatalf("Store failed: %v", err)
 	}
@@ -244,7 +300,7 @@ func TestStoreDNSInteraction(t *testing.T) {
 	}
 }
 
-func TestStoreWithAttributes(t *testing.T) {
+func TestSaveAttributes(t *testing.T) {
 	database := setupTestDB(t)
 	p := New(database)
 	_ = p.Init(plugins.InitContext{Logger: zap.NewNop()})
@@ -271,27 +327,32 @@ func TestStoreWithAttributes(t *testing.T) {
 			Headers: map[string][]string{},
 			Body:    nil,
 		},
-		Attributes: map[string]any{
-			"plugin_data":  "test-value",
-			"numeric_data": 42,
-		},
 	}
 
-	id, err := p.Store(context.Background(), draft)
+	id, err := p.CreateInteraction(context.Background(), draft)
 	if err != nil {
-		t.Fatalf("Store failed: %v", err)
+		t.Fatalf("CreateInteraction failed: %v", err)
+	}
+
+	attrs := map[string]any{
+		"plugin_data":  "test-value",
+		"numeric_data": 42,
+	}
+	err = p.SaveAttributes(context.Background(), id, attrs)
+	if err != nil {
+		t.Fatalf("SaveAttributes failed: %v", err)
 	}
 
-	attrs, err := db.GetAttributes(database, id)
+	savedAttrs, err := db.GetAttributes(database, id)
 	if err != nil {
 		t.Fatalf("GetAttributes failed: %v", err)
 	}
 
-	if attrs["plugin_data"] != "test-value" {
-		t.Errorf("plugin_data = %v, want %q", attrs["plugin_data"], "test-value")
+	if savedAttrs["plugin_data"] != "test-value" {
+		t.Errorf("plugin_data = %v, want %q", savedAttrs["plugin_data"], "test-value")
 	}
-	if v, ok := attrs["numeric_data"].(float64); !ok || v != 42 {
-		t.Errorf("numeric_data = %v, want 42", attrs["numeric_data"])
+	if v, ok := savedAttrs["numeric_data"].(float64); !ok || v != 42 {
+		t.Errorf("numeric_data = %v, want 42", savedAttrs["numeric_data"])
 	}
 }
 
@@ -315,7 +376,7 @@ func TestStoreHTTPWithoutHTTPDraft(t *testing.T) {
 		HTTP:       nil,
 	}
 
-	id, err := p.Store(context.Background(), draft)
+	id, err := p.CreateInteraction(context.Background(), draft)
 	if err != nil {
 		t.Fatalf("Store failed: %v", err)
 	}
@@ -353,7 +414,7 @@ func TestStoreDNSWithoutDNSDraft(t *testing.T) {
 		DNS:        nil,
 	}
 
-	id, err := p.Store(context.Background(), draft)
+	id, err := p.CreateInteraction(context.Background(), draft)
 	if err != nil {
 		t.Fatalf("Store failed: %v", err)
 	}

internal/server/http.go

@@ -1,24 +1,23 @@
 package server
 
 import (
-	"database/sql"
-	"encoding/json"
 	"fmt"
 	"io"
 	"net"
 	"net/http"
 	"strconv"
 	"strings"
+	"time"
 
 	"github.com/caddyserver/certmagic"
-	"github.com/rsclarke/oastrix/internal/db"
-	"github.com/rsclarke/oastrix/internal/logging"
+	"github.com/rsclarke/oastrix/internal/events"
+	"github.com/rsclarke/oastrix/internal/plugins"
 	"go.uber.org/zap"
 )
 
 // HTTPServer handles HTTP requests and records interactions.
 type HTTPServer struct {
-	DB       *sql.DB
+	Pipeline *plugins.Pipeline
 	Domain   string
 	PublicIP string
 	Logger   *zap.Logger
@@ -97,20 +96,6 @@ func (s *HTTPServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	tok, err := db.GetTokenByValue(s.DB, token)
-	if err != nil {
-		s.Logger.Error("lookup token failed", logging.Token(token), zap.Error(err))
-		w.WriteHeader(http.StatusOK)
-		_, _ = w.Write([]byte("ok"))
-		return
-	}
-	if tok == nil {
-		s.Logger.Debug("unknown token", logging.Token(token))
-		w.WriteHeader(http.StatusOK)
-		_, _ = w.Write([]byte("ok"))
-		return
-	}
-
 	remoteIP, remotePortStr, err := net.SplitHostPort(r.RemoteAddr)
 	if err != nil {
 		remoteIP = r.RemoteAddr
@@ -127,19 +112,10 @@ func (s *HTTPServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 
 	summary := fmt.Sprintf("%s %s %s", r.Method, r.URL.Path, r.Proto)
 
-	interactionID, err := db.CreateInteraction(s.DB, tok.ID, "http", remoteIP, remotePort, tls, summary)
-	if err != nil {
-		s.Logger.Error("create interaction failed", zap.Error(err))
-		w.WriteHeader(http.StatusOK)
-		_, _ = w.Write([]byte("ok"))
-		return
-	}
-
 	headers := make(map[string][]string)
 	for k, v := range r.Header {
 		headers[k] = v
 	}
-	headersJSON, _ := json.Marshal(headers)
 
 	r.Body = http.MaxBytesReader(w, r.Body, 1<<20) // 1MB limit
 	body, err := io.ReadAll(r.Body)
@@ -148,11 +124,47 @@ func (s *HTTPServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		body = nil
 	}
 
-	err = db.CreateHTTPInteraction(s.DB, interactionID, r.Method, scheme, r.Host, r.URL.Path, r.URL.RawQuery, r.Proto, string(headersJSON), body)
-	if err != nil {
-		s.Logger.Error("create http interaction failed", zap.Error(err))
+	draft := &events.InteractionDraft{
+		TokenValue: token,
+		Kind:       events.KindHTTP,
+		OccurredAt: time.Now().Unix(),
+		RemoteIP:   remoteIP,
+		RemotePort: remotePort,
+		TLS:        tls,
+		Summary:    summary,
+		HTTP: &events.HTTPDraft{
+			Method:  r.Method,
+			Scheme:  scheme,
+			Host:    r.Host,
+			Path:    r.URL.Path,
+			Query:   r.URL.RawQuery,
+			Proto:   r.Proto,
+			Headers: headers,
+			Body:    body,
+		},
+		Attributes: make(map[string]any),
+	}
+
+	resp := &events.HTTPResponsePlan{
+		Status:  200,
+		Headers: make(map[string]string),
+		Body:    []byte("ok"),
+	}
+
+	e := &events.HTTPEvent{
+		Event:   events.Event{Draft: draft},
+		Req:     r,
+		Resp:    resp,
+		Scratch: make(map[string]any),
 	}
 
-	w.WriteHeader(http.StatusOK)
-	_, _ = w.Write([]byte("ok"))
+	if err := s.Pipeline.ProcessHTTP(r.Context(), e); err != nil {
+		s.Logger.Error("pipeline error", zap.Error(err))
+	}
+
+	for k, v := range e.Resp.Headers {
+		w.Header().Set(k, v)
+	}
+	w.WriteHeader(e.Resp.Status)
+	_, _ = w.Write(e.Resp.Body)
 }