feat(notes): operation log, block leases and sync API for collaborative editing

Phase 1 of the collaboration plan (server side):
- internal/collab: block tree with parent IDs and fractional order keys,
  operations with conflict rules (version check, block leases, conflict
  copies instead of lost text, restore on late edits, cycle checks),
  in-memory leases/sessions and long-poll notifier.
- Storage: per-note lock + transaction, append-only effect log (doc_ops),
  idempotent results per opId (doc_op_results), snapshot, log paging,
  central DocRole. Old flat block data is converted to the tree.
- The existing whole-document save (PUT/PATCH) runs as an adapter through
  the same engine, so the current editor keeps working and writes to the log.
- API: GET /docs/{id}/snapshot, POST /docs/{id}/sync, GET /docs/{id}/changes
  (long poll), POST /docs/{id}/leave; DOC_SEQ poke for WebSocket/SSE rooms.
- Backend accepts editor (write) shares for notes and folders; the forced
  read-only migration is gone. The share dialog still offers read only.
- Tests: conflict matrix, order keys, leases, storage adapter round trip,
  idempotency, API protocol incl. long poll, and a randomized simulation
  (4 clients, lost/duplicated requests and responses, offline phases,
  lease expiry) checking convergence, no lost text and exactly-once results.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

internal/api/docsync.go

@@ -0,0 +1,228 @@
+package api
+
+import (
+	"encoding/json"
+	"net/http"
+	"strconv"
+	"time"
+
+	"github.com/go-chi/chi/v5"
+	"github.com/sandrodach/mindsift/internal/auth"
+	"github.com/sandrodach/mindsift/internal/collab"
+	"github.com/sandrodach/mindsift/internal/storage"
+)
+
+const (
+	syncMaxOps     = 500
+	syncMaxLog     = 1000
+	syncMaxBody    = 8 << 20
+	longPollMax    = 25 * time.Second
+	maxClientIDLen = 64
+)
+
+type syncLeases struct {
+	Granted []string           `json:"granted"`
+	Denied  []string           `json:"denied"`
+	Others  []collab.LeaseInfo `json:"others"`
+}
+
+type syncResponse struct {
+	Results  []collab.Result       `json:"results"`
+	Ops      []storage.DocLogEntry `json:"ops"`
+	HeadSeq  int64                 `json:"headSeq"`
+	HasMore  bool                  `json:"hasMore"`
+	Resync   bool                  `json:"resync,omitempty"` // client state is ahead of the server: reload the snapshot
+	Role     string                `json:"role"`
+	Shared   bool                  `json:"shared"`
+	Sessions int                   `json:"sessions"`
+	Presence []collab.Presence     `json:"presence"`
+	Leases   syncLeases            `json:"leases"`
+}
+
+// docAccess resolves the caller's role; it writes 404 and returns "" without access.
+func (s *Server) docAccess(w http.ResponseWriter, r *http.Request) (userID, docID, role string) {
+	userID = auth.GetUserID(r.Context())
+	docID = chi.URLParam(r, "id")
+	role = s.store.DocRole(userID, docID)
+	if role == "" {
+		writeError(w, http.StatusNotFound, storage.ErrDocNotFound.Error())
+	}
+	return userID, docID, role
+}
+
+func (s *Server) handleDocSnapshot(w http.ResponseWriter, r *http.Request) {
+	_, docID, role := s.docAccess(w, r)
+	if role == "" {
+		return
+	}
+	snap, err := s.store.GetDocSnapshot(docID)
+	if err != nil {
+		writeError(w, http.StatusNotFound, err.Error())
+		return
+	}
+	writeJSON(w, http.StatusOK, map[string]interface{}{
+		"docId":        snap.DocID,
+		"title":        snap.Title,
+		"titleVersion": snap.TitleVersion,
+		"headSeq":      snap.HeadSeq,
+		"blocks":       snap.Blocks,
+		"role":         role,
+		"shared":       s.store.IsDocShared(docID),
+	})
+}
+
+// handleDocSync is the single round trip of the sync protocol: send own operations, hold or release
+// block leases, and receive every log entry after lastSeq.
+func (s *Server) handleDocSync(w http.ResponseWriter, r *http.Request) {
+	userID, docID, role := s.docAccess(w, r)
+	if role == "" {
+		return
+	}
+	var req struct {
+		ClientID string      `json:"clientId"`
+		LastSeq  int64       `json:"lastSeq"`
+		Ops      []collab.Op `json:"ops"`
+		Leases   *struct {
+			Hold    []string `json:"hold"`
+			Release []string `json:"release"`
+		} `json:"leases"`
+	}
+	if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, syncMaxBody)).Decode(&req); err != nil {
+		writeError(w, http.StatusBadRequest, "Ungültiges JSON")
+		return
+	}
+	if req.ClientID == "" || len(req.ClientID) > maxClientIDLen {
+		writeError(w, http.StatusBadRequest, "clientId fehlt oder ist zu lang")
+		return
+	}
+	if len(req.Ops) > syncMaxOps {
+		writeError(w, http.StatusRequestEntityTooLarge, "Zu viele Änderungen in einer Anfrage")
+		return
+	}
+	canEdit := storage.CanEditDoc(role)
+	if len(req.Ops) > 0 && !canEdit {
+		writeError(w, http.StatusForbidden, "Keine Schreibrechte für diese Notiz (nur Lesen)")
+		return
+	}
+
+	actor := collab.Actor{UserID: userID, ClientID: req.ClientID}
+	if u, err := s.store.GetUserByID(userID); err == nil {
+		actor.Name = u.DisplayName
+	}
+	s.leases.Touch(docID, actor)
+
+	resp := syncResponse{Results: []collab.Result{}, Role: role, Leases: syncLeases{Granted: []string{}, Denied: []string{}}}
+	changed := false
+	if req.Leases != nil && canEdit {
+		s.leases.ReleaseAllExcept(docID, req.ClientID, req.Leases.Hold)
+		s.leases.Release(docID, req.ClientID, req.Leases.Release)
+		resp.Leases.Granted, resp.Leases.Denied = s.leases.Hold(docID, actor, req.Leases.Hold)
+		changed = true
+	}
+
+	if len(req.Ops) > 0 {
+		results, _, err := s.store.ApplyDocOps(docID, actor, req.Ops, s.leases.View(docID))
+		if err != nil {
+			writeError(w, http.StatusInternalServerError, err.Error())
+			return
+		}
+		resp.Results = results
+		// New blocks (and conflict copies) belong to their author right away
+		var own []string
+		for i, res := range results {
+			if res.Status == collab.StatusApplied && req.Ops[i].Kind == collab.OpInsert {
+				own = append(own, req.Ops[i].BlockID)
+			}
+			if res.ConflictBlockID != "" {
+				own = append(own, res.ConflictBlockID)
+			}
+			if res.Seq > 0 {
+				changed = true
+			}
+		}
+		if len(own) > 0 {
+			granted, _ := s.leases.Hold(docID, actor, own)
+			resp.Leases.Granted = append(resp.Leases.Granted, granted...)
+		}
+	}
+
+	entries, head, err := s.store.GetDocOpsSince(docID, req.LastSeq, syncMaxLog)
+	if err != nil {
+		writeError(w, http.StatusInternalServerError, err.Error())
+		return
+	}
+	if req.LastSeq > head {
+		resp.Resync = true
+		entries = []storage.DocLogEntry{}
+	}
+	resp.Ops = entries
+	resp.HeadSeq = head
+	resp.HasMore = len(entries) == syncMaxLog && entries[len(entries)-1].Seq < head
+	resp.Shared = s.store.IsDocShared(docID)
+	resp.Sessions, resp.Presence = s.leases.Sessions(docID)
+	resp.Leases.Others = s.leases.Others(docID, req.ClientID)
+
+	if changed {
+		s.notifyDocChanged(docID, head)
+	}
+	writeJSON(w, http.StatusOK, resp)
+}
+
+// notifyDocChanged wakes long-poll requests and pokes WebSocket/SSE clients in the note's room.
+func (s *Server) notifyDocChanged(docID string, head int64) {
+	s.notifier.Notify(docID)
+	s.wsHub.BroadcastDocEvent(docID, "DOC_SEQ", map[string]interface{}{"docId": docID, "headSeq": head})
+}
+
+// handleDocChanges is the long-poll fallback: it returns as soon as the note changes (or leases/presence
+// change), at the latest after `wait` seconds. Clients then call /sync to fetch the data.
+func (s *Server) handleDocChanges(w http.ResponseWriter, r *http.Request) {
+	_, docID, role := s.docAccess(w, r)
+	if role == "" {
+		return
+	}
+	since, _ := strconv.ParseInt(r.URL.Query().Get("since"), 10, 64)
+	wait := longPollMax
+	if sec, err := strconv.Atoi(r.URL.Query().Get("wait")); err == nil && sec >= 0 && time.Duration(sec)*time.Second < wait {
+		wait = time.Duration(sec) * time.Second
+	}
+	_ = http.NewResponseController(w).SetWriteDeadline(time.Now().Add(wait + 10*time.Second))
+
+	woken := s.notifier.Changed(docID) // subscribe before reading, so no change can slip through
+	_, head, err := s.store.GetDocOpsSince(docID, since, 0)
+	if err != nil {
+		writeError(w, http.StatusInternalServerError, err.Error())
+		return
+	}
+	changed := head > since
+	if !changed {
+		select {
+		case <-woken:
+			changed = true
+		case <-time.After(wait):
+		case <-r.Context().Done():
+			return
+		}
+		_, head, _ = s.store.GetDocOpsSince(docID, since, 0)
+	}
+	writeJSON(w, http.StatusOK, map[string]interface{}{"headSeq": head, "changed": changed})
+}
+
+// handleDocLeave ends a client's session on a note and frees its leases (tab closed, app in background).
+func (s *Server) handleDocLeave(w http.ResponseWriter, r *http.Request) {
+	_, docID, role := s.docAccess(w, r)
+	if role == "" {
+		return
+	}
+	var req struct {
+		ClientID string `json:"clientId"`
+	}
+	if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 4096)).Decode(&req); err != nil || req.ClientID == "" {
+		writeError(w, http.StatusBadRequest, "clientId fehlt")
+		return
+	}
+	s.leases.Leave(docID, req.ClientID)
+	_, head, _ := s.store.GetDocOpsSince(docID, 0, 0)
+	s.notifyDocChanged(docID, head)
+	writeJSON(w, http.StatusOK, map[string]bool{"success": true})
+}

internal/api/docsync_test.go

@@ -0,0 +1,198 @@
+package api
+
+import (
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"github.com/sandrodach/mindsift/internal/collab"
+	"github.com/sandrodach/mindsift/internal/domain"
+)
+
+func syncDoc(t *testing.T, ts *httptest.Server, token, docID string, body map[string]interface{}) (int, syncResponse) {
+	t.Helper()
+	resp := doJSON(t, ts, http.MethodPost, "/api/v1/docs/"+docID+"/sync", token, body)
+	var out syncResponse
+	if resp.StatusCode == http.StatusOK {
+		if err := json.NewDecoder(resp.Body).Decode(&out); err != nil {
+			t.Fatalf("decode sync response: %v", err)
+		}
+	}
+	return resp.StatusCode, out
+}
+
+func strp(s string) *string { return &s }
+
+type collabFixture struct {
+	ts                             *httptest.Server
+	docID                          string
+	ownerTok, editorTok, viewerTok string
+	outsiderTok                    string
+}
+
+func newCollabFixture(t *testing.T) collabFixture {
+	t.Helper()
+	_, store, ts, owner, ownerTok := startRealtimeServer(t)
+	editor, editorTok := newTestUser(t, store, "editor@example.com")
+	viewer, viewerTok := newTestUser(t, store, "viewer@example.com")
+	_, outsiderTok := newTestUser(t, store, "outsider2@example.com")
+	doc, err := store.CreateDocument(owner.ID, "", "Gemeinsam", "📄", "", []domain.DocumentBlock{{ID: "b1", Type: "bullet", Content: "Erster Punkt"}}, nil, false)
+	if err != nil {
+		t.Fatalf("create doc: %v", err)
+	}
+	if _, err := store.CreateDocShare(owner.ID, doc.ID, editor.ID, "write"); err != nil {
+		t.Fatalf("share editor: %v", err)
+	}
+	if _, err := store.CreateDocShare(owner.ID, doc.ID, viewer.ID, "read"); err != nil {
+		t.Fatalf("share viewer: %v", err)
+	}
+	return collabFixture{ts: ts, docID: doc.ID, ownerTok: ownerTok, editorTok: editorTok, viewerTok: viewerTok, outsiderTok: outsiderTok}
+}
+
+func TestDocSyncBetweenTwoEditors(t *testing.T) {
+	f := newCollabFixture(t)
+
+	// Owner inserts a block and edits the existing one in one round trip
+	code, a := syncDoc(t, f.ts, f.ownerTok, f.docID, map[string]interface{}{
+		"clientId": "owner-tab", "lastSeq": 0,
+		"ops": []collab.Op{
+			{OpID: "o1", Kind: collab.OpInsert, BlockID: "b2", ParentID: strp(""), OrderKey: "a5", Content: strp("Zweiter Punkt")},
+			{OpID: "o2", Kind: collab.OpSetContent, BlockID: "b1", BaseVersion: 1, Content: strp("Erster Punkt (bearbeitet)")},
+		},
+	})
+	if code != http.StatusOK || a.Results[0].Status != collab.StatusApplied || a.Results[1].Status != collab.StatusApplied {
+		t.Fatalf("owner sync: %d %+v", code, a.Results)
+	}
+	if len(a.Leases.Granted) == 0 || a.Leases.Granted[0] != "b2" {
+		t.Fatalf("inserted block must be leased to its author: %+v", a.Leases)
+	}
+
+	// Editor pulls everything since 0 and replays it
+	code, b := syncDoc(t, f.ts, f.editorTok, f.docID, map[string]interface{}{"clientId": "editor-tab", "lastSeq": 0})
+	if code != http.StatusOK || b.HeadSeq != a.HeadSeq || b.Role != "editor" || !b.Shared {
+		t.Fatalf("editor pull: %d head=%d role=%s shared=%v", code, b.HeadSeq, b.Role, b.Shared)
+	}
+	replay := collab.NewDoc()
+	for _, e := range b.Ops {
+		replay.ApplyEffect(e.Effect)
+	}
+	flat := replay.Flatten()
+	if len(flat) != 2 || flat[0].Content != "Erster Punkt (bearbeitet)" || flat[1].Content != "Zweiter Punkt" {
+		t.Fatalf("replayed state wrong: %+v", flat)
+	}
+	if len(b.Leases.Others) == 0 {
+		t.Fatal("editor must see the owner's lease on b2")
+	}
+
+	// Editor writes into the leased block: owner keeps it, editor's text becomes a conflict copy
+	_, c := syncDoc(t, f.ts, f.editorTok, f.docID, map[string]interface{}{
+		"clientId": "editor-tab", "lastSeq": b.HeadSeq,
+		"ops": []collab.Op{{OpID: "e1", Kind: collab.OpSetContent, BlockID: "b2", BaseVersion: 1, Content: strp("Editor war hier")}},
+	})
+	if c.Results[0].Status != collab.StatusConflict || c.Results[0].ConflictBlockID == "" {
+		t.Fatalf("write into foreign lease must conflict: %+v", c.Results)
+	}
+}
+
+func TestDocSyncRetryIsIdempotent(t *testing.T) {
+	f := newCollabFixture(t)
+	body := map[string]interface{}{
+		"clientId": "tab", "lastSeq": 0,
+		"ops": []collab.Op{{OpID: "same-op", Kind: collab.OpSetContent, BlockID: "b1", BaseVersion: 1, Content: strp("Einmal")}},
+	}
+	_, first := syncDoc(t, f.ts, f.ownerTok, f.docID, body)
+	_, second := syncDoc(t, f.ts, f.ownerTok, f.docID, body) // response of the first was "lost"
+	if first.Results[0] != second.Results[0] || first.HeadSeq != second.HeadSeq {
+		t.Fatalf("retry applied twice: %+v / %+v", first, second)
+	}
+}
+
+func TestDocSyncPermissions(t *testing.T) {
+	f := newCollabFixture(t)
+	op := map[string]interface{}{
+		"clientId": "tab", "lastSeq": 0,
+		"ops": []collab.Op{{OpID: "v1", Kind: collab.OpSetContent, BlockID: "b1", BaseVersion: 1, Content: strp("Leser schreibt")}},
+	}
+	if code, _ := syncDoc(t, f.ts, f.viewerTok, f.docID, op); code != http.StatusForbidden {
+		t.Errorf("viewer sending ops: want 403, got %d", code)
+	}
+	code, pull := syncDoc(t, f.ts, f.viewerTok, f.docID, map[string]interface{}{"clientId": "tab", "lastSeq": 0, "leases": map[string]interface{}{"hold": []string{"b1"}}})
+	if code != http.StatusOK || pull.Role != "viewer" || len(pull.Leases.Granted) != 0 {
+		t.Errorf("viewer pull: %d role=%s granted=%v", code, pull.Role, pull.Leases.Granted)
+	}
+	if code, _ := syncDoc(t, f.ts, f.outsiderTok, f.docID, map[string]interface{}{"clientId": "tab"}); code != http.StatusNotFound {
+		t.Errorf("outsider: want 404, got %d", code)
+	}
+	for _, path := range []string{"/snapshot", "/changes?wait=0"} {
+		if resp := doJSON(t, f.ts, http.MethodGet, "/api/v1/docs/"+f.docID+path, f.outsiderTok, nil); resp.StatusCode != http.StatusNotFound {
+			t.Errorf("outsider %s: want 404, got %d", path, resp.StatusCode)
+		}
+	}
+}
+
+func TestDocSnapshotMatchesLog(t *testing.T) {
+	f := newCollabFixture(t)
+	syncDoc(t, f.ts, f.ownerTok, f.docID, map[string]interface{}{
+		"clientId": "tab", "lastSeq": 0,
+		"ops": []collab.Op{{OpID: "t1", Kind: collab.OpSetTitle, BaseVersion: 0, Title: strp("Neuer Titel")}},
+	})
+	resp := doJSON(t, f.ts, http.MethodGet, "/api/v1/docs/"+f.docID+"/snapshot", f.editorTok, nil)
+	var snap struct {
+		Title   string         `json:"title"`
+		HeadSeq int64          `json:"headSeq"`
+		Blocks  []collab.Block `json:"blocks"`
+		Role    string         `json:"role"`
+	}
+	if err := json.NewDecoder(resp.Body).Decode(&snap); err != nil {
+		t.Fatal(err)
+	}
+	if snap.Title != "Neuer Titel" || snap.HeadSeq == 0 || len(snap.Blocks) != 1 || snap.Role != "editor" {
+		t.Fatalf("snapshot: %+v", snap)
+	}
+}
+
+func TestDocChangesLongPollWakesOnChange(t *testing.T) {
+	f := newCollabFixture(t)
+	_, base := syncDoc(t, f.ts, f.editorTok, f.docID, map[string]interface{}{"clientId": "editor-tab", "lastSeq": 0})
+
+	type pollResult struct {
+		Changed bool  `json:"changed"`
+		HeadSeq int64 `json:"headSeq"`
+	}
+	done := make(chan pollResult, 1)
+	start := time.Now()
+	go func() {
+		req, _ := http.NewRequest(http.MethodGet, f.ts.URL+"/api/v1/docs/"+f.docID+"/changes?wait=20&since="+itoa(base.HeadSeq), nil)
+		req.Header.Set("Authorization", "Bearer "+f.editorTok)
+		resp, err := http.DefaultClient.Do(req)
+		if err != nil {
+			done <- pollResult{}
+			return
+		}
+		defer resp.Body.Close()
+		var r pollResult
+		_ = json.NewDecoder(resp.Body).Decode(&r)
+		done <- r
+	}()
+
+	time.Sleep(300 * time.Millisecond)
+	syncDoc(t, f.ts, f.ownerTok, f.docID, map[string]interface{}{
+		"clientId": "owner-tab", "lastSeq": 0,
+		"ops": []collab.Op{{OpID: "lp1", Kind: collab.OpSetContent, BlockID: "b1", BaseVersion: 1, Content: strp("Weckruf")}},
+	})
+	select {
+	case r := <-done:
+		if !r.Changed || r.HeadSeq <= base.HeadSeq || time.Since(start) > 5*time.Second {
+			t.Fatalf("long poll result: %+v after %v", r, time.Since(start))
+		}
+	case <-time.After(10 * time.Second):
+		t.Fatal("long poll did not wake up")
+	}
+}
+
+func itoa(n int64) string {
+	b, _ := json.Marshal(n)
+	return string(b)
+}

internal/api/server.go

@@ -18,6 +18,7 @@ import (
 	"github.com/go-chi/chi/v5/middleware"
 	"github.com/go-chi/cors"
 	"github.com/sandrodach/mindsift/internal/auth"
+	"github.com/sandrodach/mindsift/internal/collab"
 	"github.com/sandrodach/mindsift/internal/config"
 	"github.com/sandrodach/mindsift/internal/domain"
 	"github.com/sandrodach/mindsift/internal/storage"
@@ -29,6 +30,9 @@ type Server struct {
 	store    *storage.Store
 	wsHub    *ws.Hub
 	staticFS fs.FS
+
+	leases   *collab.LeaseManager // block leases and open sessions of notes
+	notifier *collab.Notifier     // wakes long-poll requests on note changes
 }
 
 func NewServer(cfg *config.Config, store *storage.Store, wsHub *ws.Hub, staticFS fs.FS) *Server {
@@ -37,6 +41,8 @@ func NewServer(cfg *config.Config, store *storage.Store, wsHub *ws.Hub, staticFS
 		store:    store,
 		wsHub:    wsHub,
 		staticFS: staticFS,
+		leases:   collab.NewLeaseManager(nil),
+		notifier: collab.NewNotifier(),
 	}
 	wsHub.SetRoomAuthorizer(srv.canJoinRoom)
 	go srv.startAlarmTicker()
@@ -46,8 +52,7 @@ func NewServer(cfg *config.Config, store *storage.Store, wsHub *ws.Hub, staticFS
 // canJoinRoom allows realtime rooms only to users with read access: "doc:<id>" for notes, otherwise a shared list ID.
 func (s *Server) canJoinRoom(userID, room string) bool {
 	if docID, ok := strings.CutPrefix(room, "doc:"); ok {
-		_, err := s.store.GetDocumentByID(userID, docID)
-		return err == nil
+		return s.store.DocRole(userID, docID) != ""
 	}
 	_, err := s.store.GetListByID(userID, room)
 	return err == nil
@@ -159,6 +164,12 @@ func (s *Server) Router() http.Handler {
 		r.Delete("/api/v1/documents/{id}", s.handleDeleteDocument)
 		r.Post("/api/v1/documents/{id}/blocks/{blockId}/enhanced-task", s.handleCreateEnhancedTask)
 
+		// Collaborative note editing (operation log)
+		r.Get("/api/v1/docs/{id}/snapshot", s.handleDocSnapshot)
+		r.Post("/api/v1/docs/{id}/sync", s.handleDocSync)
+		r.Get("/api/v1/docs/{id}/changes", s.handleDocChanges)
+		r.Post("/api/v1/docs/{id}/leave", s.handleDocLeave)
+
 		// Document Sharing & Collaboration
 		r.Get("/api/v1/documents/{id}/shares", s.handleGetDocShares)
 		r.Post("/api/v1/documents/{id}/shares", s.handleCreateDocShare)