1 files changed, 136 insertions, 0 deletions
diff --git a/internal/ipc/server.go b/internal/ipc/server.go
new file mode 100644
index 0000000..11ba033
--- /dev/null
+++ b/internal/ipc/server.go
@@ -0,0 +1,136 @@
+package ipc
+
+import (
+	"encoding/json"
+	"log"
+	"net"
+	"sync"
+
+	"mitmux/internal/store"
+)
+
+// Hub fans out newly captured history entries to subscribed clients.
+type Hub struct {
+	mu   sync.Mutex
+	subs map[chan store.Summary]struct{}
+}
+
+// NewHub creates an empty Hub.
+func NewHub() *Hub {
+	return &Hub{subs: make(map[chan store.Summary]struct{})}
+}
+
+// Broadcast notifies all current subscribers of e. Slow subscribers
+// drop entries rather than blocking the proxy.
+func (h *Hub) Broadcast(e store.Summary) {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	for ch := range h.subs {
+		select {
+		case ch <- e:
+		default:
+		}
+	}
+}
+
+func (h *Hub) subscribe() chan store.Summary {
+	ch := make(chan store.Summary, 64)
+	h.mu.Lock()
+	h.subs[ch] = struct{}{}
+	h.mu.Unlock()
+	return ch
+}
+
+func (h *Hub) unsubscribe(ch chan store.Summary) {
+	h.mu.Lock()
+	delete(h.subs, ch)
+	h.mu.Unlock()
+	close(ch)
+}
+
+// Server serves the daemon side of the mitmux control protocol.
+type Server struct {
+	db  *store.Store
+	hub *Hub
+}
+
+// NewServer creates a control-protocol Server backed by db, broadcasting
+// through hub.
+func NewServer(db *store.Store, hub *Hub) *Server {
+	return &Server{db: db, hub: hub}
+}
+
+// Serve accepts connections on ln until it returns an error (e.g. the
+// listener is closed).
+func (s *Server) Serve(ln net.Listener) error {
+	for {
+		conn, err := ln.Accept()
+		if err != nil {
+			return err
+		}
+		go s.handleConn(conn)
+	}
+}
+
+func (s *Server) handleConn(conn net.Conn) {
+	defer conn.Close()
+	dec := json.NewDecoder(conn)
+	enc := json.NewEncoder(conn)
+
+	for {
+		var req Request
+		if err := dec.Decode(&req); err != nil {
+			return
+		}
+
+		switch req.Type {
+		case "list":
+			entries, err := s.db.List(req.Limit, req.BeforeID)
+			if err != nil {
+				enc.Encode(Response{Type: "error", Error: err.Error()})
+				continue
+			}
+			enc.Encode(Response{Type: "list", Entries: entries})
+
+		case "get":
+			e, err := s.db.Get(req.ID)
+			if err != nil {
+				enc.Encode(Response{Type: "error", Error: err.Error()})
+				continue
+			}
+			enc.Encode(Response{Type: "get", Detail: &EntryDetail{
+				Summary: store.Summary{
+					ID: e.ID, StartedAt: e.StartedAt, Duration: e.Duration,
+					Method: e.Method, Scheme: e.Scheme, Host: e.Host, Path: e.Path,
+					StatusCode: e.StatusCode, ReqSize: len(e.RequestRaw), RespSize: len(e.ResponseRaw),
+					Error: e.Error,
+				},
+				RequestRaw: e.RequestRaw, ResponseRaw: e.ResponseRaw,
+				RequestExact: e.RequestExact, ResponseExact: e.ResponseExact,
+			}})
+
+		case "subscribe":
+			sub := s.hub.subscribe()
+			defer s.hub.unsubscribe(sub)
+			for e := range sub {
+				e := e
+				if err := enc.Encode(Response{Type: "new", New: &e}); err != nil {
+					return
+				}
+			}
+			return
+
+		default:
+			enc.Encode(Response{Type: "error", Error: "unknown request type: " + req.Type})
+		}
+	}
+}
+
+// LogAndBroadcast is a convenience OnEntry callback: logs the entry and
+// broadcasts it through hub.
+func LogAndBroadcast(hub *Hub) func(store.Summary) {
+	return func(sum store.Summary) {
+		log.Printf("%s %s%s -> %d (%s)", sum.Method, sum.Host, sum.Path, sum.StatusCode, sum.Duration)
+		hub.Broadcast(sum)
+	}
+}