feat: notify document updates over websocket

apps/server/go.mod

@@ -4,6 +4,7 @@ go 1.24.2
 
 require (
 	github.com/go-chi/chi/v5 v5.2.5
+	github.com/gorilla/websocket v1.5.3
 	github.com/tursodatabase/go-libsql v0.0.0-20260424063416-3051e37e6e04
 	github.com/yuin/goldmark v1.8.2
 )

apps/server/go.sum

@@ -4,6 +4,8 @@ github.com/go-chi/chi/v5 v5.2.5 h1:Eg4myHZBjyvJmAFjFvWgrqDTXFyOzjj7YIm3L3mu6Ug=
 github.com/go-chi/chi/v5 v5.2.5/go.mod h1:X7Gx4mteadT3eDOMTsXzmI4/rwUpOwBHLpAfupzFJP0=
 github.com/google/go-cmp v0.5.9 h1:O2Tfq5qg4qc4AmwVlvv0oLiVAGB7enBSJ2x2DqQFi38=
 github.com/google/go-cmp v0.5.9/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
+github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
+github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/libsql/sqlite-antlr4-parser v0.0.0-20240327125255-dbf53b6cbf06 h1:JLvn7D+wXjH9g4Jsjo+VqmzTUpl/LX7vfr6VOfSWTdM=
 github.com/libsql/sqlite-antlr4-parser v0.0.0-20240327125255-dbf53b6cbf06/go.mod h1:FUkZ5OHjlGPjnM2UyGJz9TypXQFgYqw6AFNO1UiROTM=
 github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=

apps/server/internal/app/app.go

@@ -14,6 +14,7 @@ import (
 	"github.com/tim/md-hub-secure/apps/server/internal/docs"
 	"github.com/tim/md-hub-secure/apps/server/internal/httpserver"
 	"github.com/tim/md-hub-secure/apps/server/internal/markdown"
+	"github.com/tim/md-hub-secure/apps/server/internal/realtime"
 	"github.com/tim/md-hub-secure/apps/server/internal/store"
 )
 
@@ -21,6 +22,8 @@ type App struct {
 	cfg    config.Config
 	logger *slog.Logger
 	db     database.DB
+	docs   *docs.Service
+	hub    *realtime.Hub
 	server *http.Server
 }
 
@@ -46,7 +49,12 @@ func New(ctx context.Context, cfg config.Config, logger *slog.Logger) (*App, err
 	renderer := markdown.NewRenderer()
 	repo := docs.NewRepository(db.SQL())
 	service := docs.NewService(cfg.Content.SourceDir, contentStore, renderer, repo, logger)
-	if err := service.SyncSourceDir(ctx); err != nil {
+	hub := realtime.NewHub(logger)
+	service.OnChange(func(change docs.DocumentChange) {
+		hub.Broadcast(realtime.Event{Type: "document_version", Data: change})
+	})
+
+	if _, err := service.SyncSourceDir(ctx); err != nil {
 		logger.Warn("initial content sync failed", "error", err)
 	}
 
@@ -56,6 +64,7 @@ func New(ctx context.Context, cfg config.Config, logger *slog.Logger) (*App, err
 		Documents:    service,
 		Repository:   repo,
 		ContentStore: contentStore,
+		Hub:          hub,
 	})
 	if err != nil {
 		return nil, fmt.Errorf("build http handler: %w", err)
@@ -74,11 +83,15 @@ func New(ctx context.Context, cfg config.Config, logger *slog.Logger) (*App, err
 		cfg:    cfg,
 		logger: logger,
 		db:     db,
+		docs:   service,
+		hub:    hub,
 		server: server,
 	}, nil
 }
 
 func (a *App) Run(ctx context.Context) error {
+	go a.watchDocuments(ctx)
+
 	go func() {
 		<-ctx.Done()
 		shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
@@ -93,6 +106,22 @@ func (a *App) Run(ctx context.Context) error {
 	return a.server.ListenAndServe()
 }
 
+func (a *App) watchDocuments(ctx context.Context) {
+	ticker := time.NewTicker(2 * time.Second)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			if _, err := a.docs.SyncSourceDir(ctx); err != nil {
+				a.logger.Warn("document sync poll failed", "error", err)
+			}
+		}
+	}
+}
+
 func (a *App) Close() error {
 	return a.db.Close()
 }

apps/server/internal/docs/service.go

@@ -10,6 +10,7 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
 
 	"github.com/tim/md-hub-secure/apps/server/internal/markdown"
 	"github.com/tim/md-hub-secure/apps/server/internal/store"
@@ -21,6 +22,15 @@ type Service struct {
 	renderer  *markdown.Renderer
 	repo      *Repository
 	logger    *slog.Logger
+	mu        sync.Mutex
+	onChange  func(DocumentChange)
+}
+
+type DocumentChange struct {
+	Path         string `json:"path"`
+	Title        string `json:"title"`
+	Hash         string `json:"hash"`
+	PreviousHash string `json:"previousHash,omitempty"`
 }
 
 type Page struct {
@@ -41,27 +51,56 @@ func NewService(sourceDir string, store *store.ContentStore, renderer *markdown.
 	}
 }
 
-func (s *Service) SyncSourceDir(ctx context.Context) error {
+func (s *Service) OnChange(callback func(DocumentChange)) {
-	return filepath.WalkDir(s.sourceDir, func(path string, entry fs.DirEntry, err error) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.onChange = callback
+}
+
+func (s *Service) SyncSourceDir(ctx context.Context) ([]DocumentChange, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	var changes []DocumentChange
+	err := filepath.WalkDir(s.sourceDir, func(path string, entry fs.DirEntry, err error) error {
 		if err != nil {
 			return err
 		}
 		if entry.IsDir() || filepath.Ext(path) != ".md" {
 			return nil
 		}
-		return s.syncFile(ctx, path)
+
+		change, err := s.syncFile(ctx, path)
+		if err != nil {
+			return err
+		}
+		if change != nil {
+			changes = append(changes, *change)
+		}
+		return nil
 	})
+	if err != nil {
+		return nil, err
+	}
+
+	if s.onChange != nil {
+		for _, change := range changes {
+			s.onChange(change)
+		}
+	}
+
+	return changes, nil
 }
 
 func (s *Service) ListDocuments(ctx context.Context) ([]DocumentRecord, error) {
-	if err := s.SyncSourceDir(ctx); err != nil {
+	if _, err := s.SyncSourceDir(ctx); err != nil {
 		return nil, err
 	}
 	return s.repo.ListDocuments(ctx)
 }
 
 func (s *Service) LoadPage(ctx context.Context, requestPath string) (*Page, error) {
-	if err := s.SyncSourceDir(ctx); err != nil {
+	if _, err := s.SyncSourceDir(ctx); err != nil {
 		return nil, err
 	}
 
@@ -101,31 +140,31 @@ func normalizeRequestPath(path string) string {
 	return path
 }
 
-func (s *Service) syncFile(ctx context.Context, path string) error {
+func (s *Service) syncFile(ctx context.Context, path string) (*DocumentChange, error) {
 	content, err := os.ReadFile(path)
 	if err != nil {
-		return fmt.Errorf("read content file %s: %w", path, err)
+		return nil, fmt.Errorf("read content file %s: %w", path, err)
 	}
 
 	rendered, err := s.renderer.Render(content)
 	if err != nil {
-		return fmt.Errorf("render content file %s: %w", path, err)
+		return nil, fmt.Errorf("render content file %s: %w", path, err)
 	}
 
 	record, err := s.store.PutBytes(content)
 	if err != nil {
-		return fmt.Errorf("store content file %s: %w", path, err)
+		return nil, fmt.Errorf("store content file %s: %w", path, err)
 	}
 
 	relative, err := filepath.Rel(s.sourceDir, path)
 	if err != nil {
-		return fmt.Errorf("compute relative path %s: %w", path, err)
+		return nil, fmt.Errorf("compute relative path %s: %w", path, err)
 	}
 	relative = filepath.ToSlash(relative)
 
 	existing, err := s.repo.GetDocumentByPath(ctx, relative)
 	if err != nil && !errors.Is(err, sql.ErrNoRows) {
-		return err
+		return nil, err
 	}
 
 	previousHash := ""
@@ -134,7 +173,7 @@ func (s *Service) syncFile(ctx context.Context, path string) error {
 		previousHash = existing.CurrentHash
 		documentID = existing.ID
 		if existing.CurrentHash == record.Hash {
-			return nil
+			return nil, nil
 		}
 	}
 
@@ -146,9 +185,14 @@ func (s *Service) syncFile(ctx context.Context, path string) error {
 		PreviousHash: previousHash,
 		Tags:         rendered.Tags,
 	}); err != nil {
-		return err
+		return nil, err
 	}
 
 	s.logger.Debug("synced document", "path", relative, "hash", record.Hash)
-	return nil
+	return &DocumentChange{
+		Path:         relative,
+		Title:        rendered.Title,
+		Hash:         record.Hash,
+		PreviousHash: previousHash,
+	}, nil
 }

apps/server/internal/httpserver/middleware.go

@@ -1,10 +1,28 @@
 package httpserver
 
 import (
+	"bufio"
+	"fmt"
+	"net"
 	"net/http"
 	"time"
+
+	"github.com/go-chi/chi/v5/middleware"
 )
 
+func (s *Server) timeoutExceptWebSocket(timeout time.Duration) func(http.Handler) http.Handler {
+	timeoutMiddleware := middleware.Timeout(timeout)
+	return func(next http.Handler) http.Handler {
+		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			if r.URL.Path == "/ws" {
+				next.ServeHTTP(w, r)
+				return
+			}
+			timeoutMiddleware(next).ServeHTTP(w, r)
+		})
+	}
+}
+
 func (s *Server) requestLogger(next http.Handler) http.Handler {
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		ww := &statusWriter{ResponseWriter: w, status: http.StatusOK}
@@ -21,7 +39,7 @@ func (s *Server) requestLogger(next http.Handler) http.Handler {
 
 func (s *Server) securityHeaders(next http.Handler) http.Handler {
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		w.Header().Set("Content-Security-Policy", "default-src 'self'; base-uri 'self'; frame-ancestors 'none'; img-src 'self' data:; style-src 'self' 'unsafe-inline'; script-src 'self'; connect-src 'self'; object-src 'none'")
+		w.Header().Set("Content-Security-Policy", "default-src 'self'; base-uri 'self'; frame-ancestors 'none'; img-src 'self' data:; style-src 'self' 'unsafe-inline'; script-src 'self'; connect-src 'self' ws: wss:; object-src 'none'")
 		w.Header().Set("Referrer-Policy", "strict-origin-when-cross-origin")
 		w.Header().Set("X-Content-Type-Options", "nosniff")
 		w.Header().Set("X-Frame-Options", "DENY")
@@ -40,3 +58,11 @@ func (w *statusWriter) WriteHeader(status int) {
 	w.status = status
 	w.ResponseWriter.WriteHeader(status)
 }
+
+func (w *statusWriter) Hijack() (net.Conn, *bufio.ReadWriter, error) {
+	hijacker, ok := w.ResponseWriter.(http.Hijacker)
+	if !ok {
+		return nil, nil, fmt.Errorf("response writer does not support hijacking")
+	}
+	return hijacker.Hijack()
+}

apps/server/internal/httpserver/server.go

@@ -14,6 +14,7 @@ import (
 
 	"github.com/tim/md-hub-secure/apps/server/internal/config"
 	"github.com/tim/md-hub-secure/apps/server/internal/docs"
+	"github.com/tim/md-hub-secure/apps/server/internal/realtime"
 	"github.com/tim/md-hub-secure/apps/server/internal/store"
 )
 
@@ -26,6 +27,7 @@ type Dependencies struct {
 	Documents    *docs.Service
 	Repository   *docs.Repository
 	ContentStore *store.ContentStore
+	Hub          *realtime.Hub
 }
 
 type Server struct {
@@ -34,6 +36,7 @@ type Server struct {
 	documents    *docs.Service
 	repository   *docs.Repository
 	contentStore *store.ContentStore
+	hub          *realtime.Hub
 	templates    *template.Template
 	webEnabled   bool
 }
@@ -54,6 +57,7 @@ func New(deps Dependencies) (http.Handler, error) {
 		documents:    deps.Documents,
 		repository:   deps.Repository,
 		contentStore: deps.ContentStore,
+		hub:          deps.Hub,
 		templates:    templates,
 	}
 
@@ -65,12 +69,13 @@ func New(deps Dependencies) (http.Handler, error) {
 	router.Use(middleware.RequestID)
 	router.Use(middleware.RealIP)
 	router.Use(middleware.Recoverer)
-	router.Use(middleware.Timeout(30 * time.Second))
+	router.Use(server.timeoutExceptWebSocket(30 * time.Second))
 	router.Use(server.requestLogger)
 	router.Use(server.securityHeaders)
 
 	router.Get("/", server.handleIndex)
 	router.Get("/health", server.handleHealth)
+	router.Get("/ws", server.handleWebSocket)
 	router.Get("/docs", server.handleDocsIndexRedirect)
 	router.Get("/docs/*", server.handleDocument)
 	router.Post("/api/uploads", server.handleUpload)

apps/server/internal/httpserver/static/realtime.js

@@ -0,0 +1,35 @@
+(function () {
+  const notice = document.querySelector("[data-version-notice]");
+  const reload = document.querySelector("[data-version-reload]");
+  const documentShell = document.querySelector("[data-document-path][data-document-hash]");
+
+  if (!notice || !reload || !documentShell || !window.WebSocket) {
+    return;
+  }
+
+  const currentPath = documentShell.getAttribute("data-document-path");
+  const currentHash = documentShell.getAttribute("data-document-hash");
+  const protocol = window.location.protocol === "https:" ? "wss:" : "ws:";
+  const socket = new WebSocket(protocol + "//" + window.location.host + "/ws");
+
+  socket.addEventListener("message", function (event) {
+    let payload;
+    try {
+      payload = JSON.parse(event.data);
+    } catch {
+      return;
+    }
+
+    if (payload.type !== "document_version" || !payload.data) {
+      return;
+    }
+
+    if (payload.data.path === currentPath && payload.data.hash !== currentHash) {
+      notice.hidden = false;
+    }
+  });
+
+  reload.addEventListener("click", function () {
+    window.location.reload();
+  });
+})();

apps/server/internal/httpserver/static/site.css

@@ -177,6 +177,41 @@ code {
   max-width: 100%;
 }
 
+.version-notice {
+  position: fixed;
+  right: 1rem;
+  bottom: 1rem;
+  display: flex;
+  align-items: center;
+  gap: 0.85rem;
+  max-width: min(28rem, calc(100vw - 2rem));
+  padding: 0.75rem 0.85rem;
+  border: 1px solid var(--border);
+  border-radius: var(--radius-sm);
+  background: var(--panel-strong);
+  box-shadow: var(--shadow);
+}
+
+.version-notice[hidden] {
+  display: none;
+}
+
+.version-notice p {
+  margin: 0;
+  color: var(--text);
+}
+
+.version-notice button {
+  min-height: 2.2rem;
+  padding: 0 0.8rem;
+  border: 0;
+  border-radius: var(--radius-sm);
+  color: white;
+  background: var(--accent);
+  font: 700 0.9rem/1 ui-monospace, SFMono-Regular, monospace;
+  cursor: pointer;
+}
+
 @media (max-width: 720px) {
   .site-header__inner {
     flex-direction: column;
@@ -191,4 +226,3 @@ code {
     padding: 1.25rem;
   }
 }
-

apps/server/internal/httpserver/templates/base.gohtml

@@ -27,6 +27,11 @@
         {{ template "error_content" .Data }}
       {{ end }}
     </main>
+    <div class="version-notice" data-version-notice hidden>
+      <p>A newer version is available.</p>
+      <button type="button" data-version-reload>Reload</button>
+    </div>
+    <script src="/static/realtime.js" defer></script>
   </body>
 </html>
 {{ end }}

apps/server/internal/httpserver/templates/document.gohtml

@@ -1,7 +1,7 @@
 {{ define "document.gohtml" }}{{ template "base" . }}{{ end }}
 
 {{ define "document_content" }}
-<article class="document-shell">
+<article class="document-shell" data-document-path="{{ .Path }}" data-document-hash="{{ .Hash }}">
   <div class="document-meta">
     <p class="eyebrow">{{ .Path }}</p>
     <h1>{{ .Title }}</h1>

apps/server/internal/httpserver/websocket.go

@@ -0,0 +1,88 @@
+package httpserver
+
+import (
+	"net/http"
+	"time"
+
+	"github.com/gorilla/websocket"
+
+	"github.com/tim/md-hub-secure/apps/server/internal/realtime"
+)
+
+const (
+	wsWriteWait  = 10 * time.Second
+	wsPongWait   = 60 * time.Second
+	wsPingPeriod = 45 * time.Second
+)
+
+var wsUpgrader = websocket.Upgrader{
+	ReadBufferSize:  1024,
+	WriteBufferSize: 1024,
+	CheckOrigin: func(r *http.Request) bool {
+		origin := r.Header.Get("Origin")
+		if origin == "" {
+			return true
+		}
+		return origin == "http://"+r.Host || origin == "https://"+r.Host
+	},
+}
+
+func (s *Server) handleWebSocket(w http.ResponseWriter, r *http.Request) {
+	conn, err := wsUpgrader.Upgrade(w, r, nil)
+	if err != nil {
+		s.logger.Warn("upgrade websocket", "error", err)
+		return
+	}
+
+	client := s.hub.NewClient(16)
+	s.hub.Register(client)
+
+	go s.readWebSocket(conn, client)
+	s.writeWebSocket(conn, client)
+}
+
+func (s *Server) readWebSocket(conn *websocket.Conn, client *realtime.Client) {
+	defer func() {
+		s.hub.Unregister(client)
+		_ = conn.Close()
+	}()
+	conn.SetReadLimit(1024)
+	_ = conn.SetReadDeadline(time.Now().Add(wsPongWait))
+	conn.SetPongHandler(func(string) error {
+		return conn.SetReadDeadline(time.Now().Add(wsPongWait))
+	})
+
+	for {
+		if _, _, err := conn.NextReader(); err != nil {
+			return
+		}
+	}
+}
+
+func (s *Server) writeWebSocket(conn *websocket.Conn, client *realtime.Client) {
+	ticker := time.NewTicker(wsPingPeriod)
+	defer func() {
+		ticker.Stop()
+		s.hub.Unregister(client)
+		_ = conn.Close()
+	}()
+
+	for {
+		select {
+		case message, ok := <-client.Send():
+			_ = conn.SetWriteDeadline(time.Now().Add(wsWriteWait))
+			if !ok {
+				_ = conn.WriteMessage(websocket.CloseMessage, []byte{})
+				return
+			}
+			if err := conn.WriteMessage(websocket.TextMessage, message); err != nil {
+				return
+			}
+		case <-ticker.C:
+			_ = conn.SetWriteDeadline(time.Now().Add(wsWriteWait))
+			if err := conn.WriteMessage(websocket.PingMessage, nil); err != nil {
+				return
+			}
+		}
+	}
+}

apps/server/internal/realtime/hub.go

@@ -0,0 +1,70 @@
+package realtime
+
+import (
+	"encoding/json"
+	"log/slog"
+	"sync"
+)
+
+type Event struct {
+	Type string `json:"type"`
+	Data any    `json:"data"`
+}
+
+type Client struct {
+	send chan []byte
+}
+
+type Hub struct {
+	logger  *slog.Logger
+	mu      sync.RWMutex
+	clients map[*Client]struct{}
+}
+
+func NewHub(logger *slog.Logger) *Hub {
+	return &Hub{
+		logger:  logger,
+		clients: make(map[*Client]struct{}),
+	}
+}
+
+func (h *Hub) NewClient(buffer int) *Client {
+	return &Client{send: make(chan []byte, buffer)}
+}
+
+func (h *Hub) Register(client *Client) {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	h.clients[client] = struct{}{}
+}
+
+func (h *Hub) Unregister(client *Client) {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	if _, ok := h.clients[client]; ok {
+		delete(h.clients, client)
+		close(client.send)
+	}
+}
+
+func (h *Hub) Broadcast(event Event) {
+	payload, err := json.Marshal(event)
+	if err != nil {
+		h.logger.Error("marshal realtime event", "error", err)
+		return
+	}
+
+	h.mu.RLock()
+	defer h.mu.RUnlock()
+	for client := range h.clients {
+		select {
+		case client.send <- payload:
+		default:
+			h.logger.Warn("dropping realtime event for slow client")
+		}
+	}
+}
+
+func (c *Client) Send() <-chan []byte {
+	return c.send
+}