feat(notes): push, long polling and fallback detection for note sync

Phase 4 of the collaboration plan: - The sync engine listens for DOC_SEQ in the note's realtime room and syncs right away. Without WebSocket/SSE it long-polls /docs/{id}/changes (25 s); after three failures it falls back to polling every 3 s and retries long polling when the browser comes back online. - With push or long polling the engine only syncs every 7 s while holding leases (renewal), otherwise every 30 s as a safety net. Focusing a block announces its lease immediately (otherwise it could take up to 30 s). - Realtime connection: SSE counts as connected only after its first event; a stream that stays silent for 5 s (buffering proxy) switches SSE off for the tab. App-level PING/PONG detects WebSockets that are accepted but forward nothing and switches to SSE after 35 s. - The editor shows the active channel for shared notes. - Tests: engine channel tests (push, heartbeat, lease on focus, long polling, interval fallback, switching), realtime fallback tests with fake WebSocket/EventSource, Go tests for PONG and DOC_SEQ delivery, and e2e/channels.e2e.mjs measuring latency behind six simulated proxy setups (0.2-0.3 s on all live channels, ~3 s with plain polling). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-28 14:10:29 +01:00
1 parent aa397c18cd
commit 8b7a53daab
@@ -687,7 +687,7 @@ export const DocumentEditor: React.FC<DocumentEditorProps> = ({
  }

  return (
-    <div className="flex-1 flex flex-col h-full min-h-0 bg-canvas overflow-y-auto">
+    <div className="flex-1 flex flex-col h-full min-h-0 bg-canvas overflow-y-auto" data-sync-channel={engineState?.channel}>
      {/* Top Document Action Bar */}
      <div className="sticky top-0 z-20 flex flex-wrap items-center justify-between gap-2 px-4 sm:px-6 py-2 bg-canvas border-b border-border">
        <div className="flex items-center gap-2 min-w-0">
@@ -766,6 +766,29 @@ export const DocumentEditor: React.FC<DocumentEditorProps> = ({

        {/* Right Action Buttons */}
        <div className="flex items-center gap-1.5">
+          {/* How changes of others arrive (only relevant for shared notes) */}
+          {engineState && (engineState.shared || engineState.sessions > 1) && (
+            <span
+              className="hidden sm:inline-flex items-center gap-1 px-2 py-0.5 rounded-btn text-[11px] text-muted bg-panel border border-border"
+              title={
+                engineState.channel === 'ws'
+                  ? 'Live-Verbindung über WebSocket'
+                  : engineState.channel === 'sse'
+                  ? 'Live-Verbindung über Server-Sent Events (WebSocket wird blockiert)'
+                  : engineState.channel === 'longpoll'
+                  ? 'Live über Long-Polling (WebSocket und SSE werden blockiert)'
+                  : 'Keine Live-Verbindung möglich – Änderungen werden alle 3 Sekunden abgefragt'
+              }
+            >
+              <span
+                className={`w-1.5 h-1.5 rounded-full ${
+                  engineState.channel === 'interval' ? 'bg-status-warn' : 'bg-status-ok'
+                }`}
+              />
+              {engineState.channel === 'longpoll' ? 'Live (Long-Polling)' : engineState.channel === 'interval' ? 'Abgleich alle 3 s' : 'Live'}
+            </span>
+          )}
+
          {/* Who else has this note open */}
          {otherPeople.length > 0 && (
            <span
@@ -0,0 +1,132 @@
+import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest';
+
+// Minimal stand-ins: tests decide when connections open and which messages arrive.
+class FakeWebSocket {
+  static CONNECTING = 0;
+  static OPEN = 1;
+  static CLOSING = 2;
+  static CLOSED = 3;
+  static instances: FakeWebSocket[] = [];
+  readyState = FakeWebSocket.CONNECTING;
+  sent: string[] = [];
+  onopen: (() => void) | null = null;
+  onmessage: ((e: { data: string }) => void) | null = null;
+  onclose: ((e: { wasClean: boolean; code: number }) => void) | null = null;
+  onerror: ((e: unknown) => void) | null = null;
+  constructor(public url: string) {
+    FakeWebSocket.instances.push(this);
+  }
+  open() {
+    this.readyState = FakeWebSocket.OPEN;
+    this.onopen?.();
+  }
+  send(data: string) {
+    this.sent.push(data);
+  }
+  close() {
+    this.readyState = FakeWebSocket.CLOSED;
+  }
+}
+
+class FakeEventSource {
+  static instances: FakeEventSource[] = [];
+  closed = false;
+  onopen: (() => void) | null = null;
+  onmessage: ((e: { data: string }) => void) | null = null;
+  onerror: ((e: unknown) => void) | null = null;
+  constructor(public url: string) {
+    FakeEventSource.instances.push(this);
+  }
+  open() {
+    this.onopen?.();
+  }
+  emit(event: string, payload: unknown = {}) {
+    this.onmessage?.({ data: JSON.stringify({ event, payload }) });
+  }
+  close() {
+    this.closed = true;
+  }
+}
+
+async function loadApi() {
+  vi.resetModules(); // fresh module state (transport flags) for every test
+  return import('../api');
+}
+
+describe('realtime fallback', () => {
+  beforeEach(() => {
+    vi.useFakeTimers();
+    FakeWebSocket.instances = [];
+    FakeEventSource.instances = [];
+    vi.stubGlobal('WebSocket', FakeWebSocket);
+    vi.stubGlobal('EventSource', FakeEventSource);
+    sessionStorage.clear();
+    localStorage.setItem('mindsift_jwt_token', 'test-token');
+    vi.stubGlobal('fetch', vi.fn(async () => new Response('{}', { status: 200 })));
+  });
+  afterEach(() => {
+    vi.useRealTimers();
+    vi.unstubAllGlobals();
+  });
+
+  it('a WebSocket that stays silent is replaced by SSE', async () => {
+    const api = await loadApi();
+    const states: string[] = [];
+    api.onRealtimeStatus((connected, transport) => states.push(`${connected}:${transport}`));
+    api.connectWebSocket();
+    FakeWebSocket.instances[0].open();
+    expect(api.getRealtimeTransport()).toBe('ws');
+
+    await vi.advanceTimersByTimeAsync(15_000);
+    expect(FakeWebSocket.instances[0].sent).toContain(JSON.stringify({ event: 'PING' }));
+    await vi.advanceTimersByTimeAsync(40_000); // no message at all, not even PONG
+    expect(FakeEventSource.instances).toHaveLength(1);
+    FakeEventSource.instances[0].open();
+    FakeEventSource.instances[0].emit('SSE_CONNECTED', { sseId: 'sse_1' });
+    expect(api.getRealtimeTransport()).toBe('sse');
+    api.disconnectWebSocket();
+  });
+
+  it('a WebSocket answering PONG stays in use', async () => {
+    const api = await loadApi();
+    api.connectWebSocket();
+    const ws = FakeWebSocket.instances[0];
+    ws.open();
+    for (let i = 0; i < 6; i++) {
+      await vi.advanceTimersByTimeAsync(15_000);
+      ws.onmessage?.({ data: JSON.stringify({ event: 'PONG' }) });
+    }
+    expect(FakeEventSource.instances).toHaveLength(0);
+    expect(api.getRealtimeTransport()).toBe('ws');
+    api.disconnectWebSocket();
+  });
+
+  it('SSE only counts as connected once the first event arrives', async () => {
+    sessionStorage.setItem('mindsift_ws_blocked', 'true');
+    const api = await loadApi();
+    api.connectWebSocket();
+    const es = FakeEventSource.instances[0];
+    es.open();
+    expect(api.isRealtimeConnected()).toBe(false);
+    es.emit('SSE_CONNECTED', { sseId: 'sse_1' });
+    expect(api.isRealtimeConnected()).toBe(true);
+    await vi.advanceTimersByTimeAsync(10_000);
+    expect(api.isRealtimeConnected()).toBe(true);
+    api.disconnectWebSocket();
+  });
+
+  it('a buffering proxy (stream opens, no events) switches SSE off for this tab', async () => {
+    sessionStorage.setItem('mindsift_ws_blocked', 'true');
+    const api = await loadApi();
+    api.connectWebSocket();
+    const es = FakeEventSource.instances[0];
+    es.open();
+    await vi.advanceTimersByTimeAsync(5_500);
+    expect(es.closed).toBe(true);
+    expect(api.isRealtimeConnected()).toBe(false);
+    expect(sessionStorage.getItem('mindsift_sse_buffered')).toBe('true');
+    await vi.advanceTimersByTimeAsync(60_000);
+    expect(FakeEventSource.instances).toHaveLength(1); // no reconnect loop: note sync uses long polling
+    api.disconnectWebSocket();
+  });
+});
@@ -1134,6 +1134,16 @@ export function onRealtimeStatus(callback: (connected: boolean, transport: Realt
 }

 let isWsBlockedByProxy = typeof window !== 'undefined' && sessionStorage.getItem('mindsift_ws_blocked') === 'true';
+// Some proxies buffer event streams: the connection opens, but events only arrive much later (or never).
+// Then SSE is useless and the note sync engines fall back to long polling.
+let isSseBufferedByProxy = typeof window !== 'undefined' && sessionStorage.getItem('mindsift_sse_buffered') === 'true';
+let sseFirstEventWatchdog: any = null;
+const SSE_FIRST_EVENT_TIMEOUT_MS = 5000;
+// Some proxies accept the WebSocket upgrade but then forward nothing. An app-level ping detects that.
+let wsHeartbeatTimer: any = null;
+let wsLastMessageAt = 0;
+const WS_PING_INTERVAL_MS = 15000;
+const WS_DEAD_AFTER_MS = 35000;
 let sseReconnectTimer: any = null;
 let sseReconnectDelay = 2000;
 const globalSubscribedRooms = new Set<string>();
@@ -1153,6 +1163,7 @@ function notifyRealtimeStatusChange(connected: boolean, transport: RealtimeTrans

 function switchToSSE(reason: string) {
  console.warn(`[MindSift Realtime] ${reason} -> Schalte sofort und dauerhaft auf Server-Sent Events (SSE) um.`);
+  clearInterval(wsHeartbeatTimer);
  isWsBlockedByProxy = true;
  try {
    sessionStorage.setItem('mindsift_ws_blocked', 'true');
@@ -1177,6 +1188,12 @@ export function connectSSE(rooms?: string[]): void {
  clearTimeout(sseReconnectTimer);
  clearTimeout(wsReconnectTimer);
  clearTimeout(wsConnectionWatchdog);
+  clearTimeout(sseFirstEventWatchdog);
+
+  if (isSseBufferedByProxy) {
+    notifyRealtimeStatusChange(false, 'none');
+    return;
+  }

  if (activeGlobalSSE) {
    try { activeGlobalSSE.close(); } catch {}
@@ -1190,18 +1207,28 @@ export function connectSSE(rooms?: string[]): void {

    es.onopen = () => {
      sseReconnectDelay = 2000;
-      notifyRealtimeStatusChange(true, 'sse');
-      if (activeSSEId && globalSubscribedRooms.size > 0) {
-        globalSubscribedRooms.forEach((room) => {
-          api.joinSSERoom(activeSSEId!, room, 'join').catch(() => {});
-        });
-      }
+      // Connected only counts once the first event really arrives (see SSE_CONNECTED below)
+      clearTimeout(sseFirstEventWatchdog);
+      sseFirstEventWatchdog = setTimeout(() => {
+        console.warn('[MindSift Realtime] SSE liefert keine Ereignisse (Proxy puffert) -> Long-Polling');
+        isSseBufferedByProxy = true;
+        try {
+          sessionStorage.setItem('mindsift_sse_buffered', 'true');
+        } catch {}
+        try {
+          es.close();
+        } catch {}
+        if (activeGlobalSSE === es) activeGlobalSSE = null;
+        notifyRealtimeStatusChange(false, 'none');
+      }, SSE_FIRST_EVENT_TIMEOUT_MS);
    };

    es.onmessage = (event) => {
      try {
        const msg = JSON.parse(event.data) as WSMessage;
        if (msg.event === 'SSE_CONNECTED') {
+          clearTimeout(sseFirstEventWatchdog);
+          notifyRealtimeStatusChange(true, 'sse');
          activeSSEId = msg.payload?.sseId || null;
          if (activeSSEId && globalSubscribedRooms.size > 0) {
            globalSubscribedRooms.forEach((room) => {
@@ -1223,6 +1250,7 @@ export function connectSSE(rooms?: string[]): void {
    };

    es.onerror = (e) => {
+      clearTimeout(sseFirstEventWatchdog);
      console.warn('[MindSift Realtime] SSE Verbindungsfehler:', e);
      if (activeTransport === 'sse') {
        notifyRealtimeStatusChange(false, 'none');
@@ -1279,6 +1307,17 @@ export function connectWebSocket(): void {

    ws.onopen = () => {
      clearTimeout(wsConnectionWatchdog);
+      wsLastMessageAt = Date.now();
+      clearInterval(wsHeartbeatTimer);
+      wsHeartbeatTimer = setInterval(() => {
+        if (Date.now() - wsLastMessageAt > WS_DEAD_AFTER_MS) {
+          switchToSSE('WebSocket liefert keine Antworten mehr (Proxy leitet nichts weiter)');
+          return;
+        }
+        try {
+          ws.send(JSON.stringify({ event: 'PING' }));
+        } catch {}
+      }, WS_PING_INTERVAL_MS);
      if (activeGlobalSSE) {
        try { activeGlobalSSE.close(); } catch {}
        activeGlobalSSE = null;
@@ -1295,8 +1334,10 @@ export function connectWebSocket(): void {
    };

    ws.onmessage = (event) => {
+      wsLastMessageAt = Date.now();
      try {
        const msg = JSON.parse(event.data) as WSMessage;
+        if (msg.event === 'PONG') return;
        wsListeners.forEach((fn) => {
          try {
            fn(msg);
@@ -1312,6 +1353,7 @@ export function connectWebSocket(): void {

    ws.onclose = (evt: CloseEvent) => {
      clearTimeout(wsConnectionWatchdog);
+      clearInterval(wsHeartbeatTimer);
      activeGlobalWS = null;

      // In corporate environments, blocked WebSockets close abruptly without clean handshake
@@ -1368,6 +1410,8 @@ if (typeof window !== 'undefined') {
 }

 export function disconnectWebSocket(): void {
+  clearInterval(wsHeartbeatTimer);
+  clearTimeout(sseFirstEventWatchdog);
  manualDisconnect = true;
  clearTimeout(wsReconnectTimer);
  clearTimeout(wsConnectionWatchdog);
@@ -4,6 +4,7 @@ import type { DocumentBlock } from '../../types';
 import { DocSyncEngine, formBatch } from '../engine';
 import { applyOp, cloneDoc, Doc, emptyDoc, Op, Result } from '../model';
 import { IndexedDBStorage, MemoryStorage } from '../persist';
+import { PushChannel, PushKind } from '../push';
 import { LogEntry, SnapshotResponse, SyncRequest, SyncResponse, Transport, TransportError } from '../transport';

 /** In-memory stand-in for the Go server: same operation rules, log and idempotency. */
@@ -13,6 +14,12 @@ class FakeServer {
  results = new Map<string, Result>();
  roles = new Map<string, string>(); // clientId -> role
  applyCount = new Map<string, number>();
+  private waiters: (() => void)[] = [];
+
+  /** Resolves on the next change (used by the long-poll stand-in). */
+  nextChange(): Promise<void> {
+    return new Promise((r) => this.waiters.push(r));
+  }

  constructor(initial: { id: string; content: string }[] = [{ id: 'b1', content: 'Start' }]) {
    let key = 'a0';
@@ -30,6 +37,11 @@ class FakeServer {
    for (const e of effects) this.log.push({ seq: this.log.length + 1, userId: clientId, clientId, opId: op.opId, effect: e });
    if (effects.length > 0) result.seq = this.log.length;
    this.results.set(op.opId, result);
+    if (effects.length > 0) {
+      const w = this.waiters;
+      this.waiters = [];
+      w.forEach((fn) => fn());
+    }
    return result;
  }

@@ -66,6 +78,8 @@ class FakeServer {
 /** Network between one client and the fake server, with switchable failures. */
 class FakeNet implements Transport {
  mode: 'ok' | 'down' | 'lose-response' = 'ok';
+  longPollDown = false;
+  changesCalls = 0;
  duringRequest: (() => void) | null = null; // runs while the request is "on the wire"

  constructor(private server: FakeServer, private role = 'owner') {}
@@ -84,6 +98,40 @@ class FakeNet implements Transport {
    return res;
  }
  async leave() {}
+  async changes(_docId: string, since: number, waitSec: number, signal?: AbortSignal) {
+    this.changesCalls++;
+    if (this.mode === 'down' || this.longPollDown) throw new TransportError('blocked', 0);
+    if (this.server.log.length === since) {
+      await Promise.race([
+        this.server.nextChange(),
+        new Promise((r) => setTimeout(r, waitSec * 1000)),
+        new Promise((_, reject) => signal?.addEventListener('abort', () => reject(new Error('aborted')))),
+      ]);
+    }
+    return { headSeq: this.server.log.length, changed: this.server.log.length !== since };
+  }
+}
+
+/** Push channel controlled by the test. */
+class FakePush implements PushChannel {
+  kind: PushKind = 'none';
+  private poke: (() => void) | null = null;
+  private status: ((k: PushKind) => void) | null = null;
+  subscribe(_docId: string, onPoke: () => void, onStatus: (k: PushKind) => void) {
+    this.poke = onPoke;
+    this.status = onStatus;
+    onStatus(this.kind);
+    return () => {
+      this.poke = null;
+    };
+  }
+  setKind(k: PushKind) {
+    this.kind = k;
+    this.status?.(k);
+  }
+  notify() {
+    this.poke?.();
+  }
 }

 function flatOf(engine: DocSyncEngine): DocumentBlock[] {
@@ -254,27 +302,23 @@ describe('DocSyncEngine', () => {
    expect(serverContents(server)).toEqual(['Start']);
  });

-  it('polls every 3 s right after reopening a shared note from local storage', async () => {
+  it('changes of others arrive within seconds right after reopening a shared note from local storage', async () => {
    vi.useFakeTimers();
    try {
      const server = new FakeServer();
      const net = new FakeNet(server);
-      let calls = 0;
-      const sync = net.sync.bind(net);
-      net.sync = async (...args: Parameters<typeof sync>) => {
-        calls++;
-        return sync(...args);
-      };
+      net.longPollDown = true; // worst case: only plain polling works
      // Opened before: the note is reopened from local storage (like after a reload)
      const storage = new MemoryStorage();
      const first = new DocSyncEngine('doc', { transport: net, storage, manual: true });
      await first.start();
      first.stop();
-      const engine = new DocSyncEngine('doc', { transport: net, storage });
+      const engine = new DocSyncEngine('doc', { transport: net, storage, push: new FakePush() });
      await engine.start();
-      const afterStart = calls;
+      await vi.advanceTimersByTimeAsync(5_000); // long polling gives up after three attempts
+      server.apply({ opId: 'remote', kind: 'set_content', blockId: 'b1', baseVersion: 1, content: 'Von anderen' }, 'other');
      await vi.advanceTimersByTimeAsync(3_500);
-      expect(calls).toBeGreaterThan(afterStart); // not only after 30 s
+      expect(contents(engine)).toEqual(['Von anderen']); // not only after 30 s
      engine.stop();
    } finally {
      vi.useRealTimers();
@@ -303,3 +347,141 @@ describe('DocSyncEngine', () => {
  });
 });

+
+describe('how changes of others arrive', () => {
+  async function live(push: FakePush, net: FakeNet) {
+    const engine = new DocSyncEngine('doc', { transport: net, storage: new MemoryStorage(), push });
+    await engine.start();
+    await vi.advanceTimersByTimeAsync(100); // let the start-up round trips settle
+    return engine;
+  }
+  const otherEdit = (server: FakeServer, text: string) =>
+    server.apply({ opId: `other-${text}`, kind: 'set_content', blockId: 'b1', baseVersion: server.doc.blocks.get('b1')!.version, content: text }, 'other');
+
+  it('push: syncs on a notification and otherwise only checks every 30 s', async () => {
+    vi.useFakeTimers();
+    try {
+      const server = new FakeServer();
+      const net = new FakeNet(server);
+      const push = new FakePush();
+      push.kind = 'ws';
+      const engine = await live(push, net);
+      expect(engine.getState().channel).toBe('ws');
+      otherEdit(server, 'Von anderen');
+      await vi.advanceTimersByTimeAsync(20_000);
+      expect(contents(engine)).toEqual(['Start']); // no notification, no poll yet
+      push.notify();
+      await vi.advanceTimersByTimeAsync(100);
+      expect(contents(engine)).toEqual(['Von anderen']);
+      expect(net.changesCalls).toBe(0);
+      engine.stop();
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
+  it('push: renews leases every 7 s while a block is held', async () => {
+    vi.useFakeTimers();
+    try {
+      const server = new FakeServer();
+      const net = new FakeNet(server);
+      const push = new FakePush();
+      push.kind = 'sse';
+      const engine = await live(push, net);
+      let syncs = 0;
+      const orig = net.sync.bind(net);
+      net.sync = async (...a: Parameters<typeof orig>) => {
+        syncs++;
+        return orig(...a);
+      };
+      engine.focusBlock('b1');
+      await vi.advanceTimersByTimeAsync(15_000);
+      expect(syncs).toBeGreaterThanOrEqual(2);
+      engine.stop();
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
+  it('without push: long polling delivers changes right away', async () => {
+    vi.useFakeTimers();
+    try {
+      const server = new FakeServer();
+      const net = new FakeNet(server);
+      const engine = await live(new FakePush(), net);
+      await vi.advanceTimersByTimeAsync(10);
+      expect(engine.getState().channel).toBe('longpoll');
+      otherEdit(server, 'Per Long-Polling');
+      await vi.advanceTimersByTimeAsync(200);
+      expect(contents(engine)).toEqual(['Per Long-Polling']);
+      engine.stop();
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
+  it('falls back to polling every 3 s when long polling is blocked too', async () => {
+    vi.useFakeTimers();
+    try {
+      const server = new FakeServer();
+      const net = new FakeNet(server);
+      net.longPollDown = true;
+      const engine = await live(new FakePush(), net);
+      await vi.advanceTimersByTimeAsync(5_000); // three failed attempts with back-off
+      expect(engine.getState().channel).toBe('interval');
+      otherEdit(server, 'Per Abfrage');
+      await vi.advanceTimersByTimeAsync(3_500);
+      expect(contents(engine)).toEqual(['Per Abfrage']);
+      engine.stop();
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
+  it('push: requests the lease right after focusing a block', async () => {
+    vi.useFakeTimers();
+    try {
+      const server = new FakeServer();
+      const net = new FakeNet(server);
+      const push = new FakePush();
+      push.kind = 'ws';
+      const engine = await live(push, net);
+      const holds: string[][] = [];
+      const orig = net.sync.bind(net);
+      net.sync = async (id: string, body: SyncRequest, o?: { keepalive?: boolean }) => {
+        holds.push(body.leases?.hold || []);
+        return orig(id, body, o);
+      };
+      engine.focusBlock('b1');
+      await vi.advanceTimersByTimeAsync(1_000);
+      expect(holds.some((h) => h.includes('b1'))).toBe(true);
+      engine.stop();
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
+  it('switches between push and long polling when the connection changes', async () => {
+    vi.useFakeTimers();
+    try {
+      const server = new FakeServer();
+      const net = new FakeNet(server);
+      const push = new FakePush();
+      push.kind = 'ws';
+      const engine = await live(push, net);
+      push.setKind('none');
+      await vi.advanceTimersByTimeAsync(10);
+      expect(engine.getState().channel).toBe('longpoll');
+      push.setKind('ws');
+      await vi.advanceTimersByTimeAsync(10);
+      expect(engine.getState().channel).toBe('ws');
+      otherEdit(server, 'Wieder per Push');
+      push.notify();
+      await vi.advanceTimersByTimeAsync(100);
+      expect(contents(engine)).toEqual(['Wieder per Push']);
+      engine.stop();
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+});
@@ -51,6 +51,7 @@ class Server {
        };
      },
      leave: async () => {},
+      changes: async () => ({ headSeq: this.log.length, changed: false }),
    };
  }
 }
@@ -8,9 +8,13 @@ import { applyEffect, applyOp, cloneDoc, Doc, docFromSnapshot, emptyDoc, Op, Res
 import { defaultStorage, SyncStorage } from './persist';
 import { httpTransport, LeaseInfo, Presence, Transport, TransportError } from './transport';
 import { buildStep, failedChecks, touchedBlocks, UndoStep } from './undo';
+import { PushChannel, PushKind, realtimePush } from './push';

 export type EngineStatus = 'loading' | 'saved' | 'saving' | 'offline' | 'error';

+/** How changes of others reach this client: push (WebSocket/SSE), long polling or plain polling. */
+export type SyncChannel = 'ws' | 'sse' | 'longpoll' | 'interval';
+
 export interface EngineState {
  ready: boolean;
  status: EngineStatus;
@@ -26,6 +30,7 @@ export interface EngineState {
  conflicts: Result[]; // conflicts of own operations since start
  canUndo: boolean;
  canRedo: boolean;
+  channel: SyncChannel;
 }

 export interface ActionResult {
@@ -37,15 +42,21 @@ export interface EngineOptions {
  transport?: Transport;
  storage?: SyncStorage;
  now?: () => number;
-  /** Disables timers and window listeners (tests drive the engine with syncNow()). */
+  /** Disables timers, window listeners, push and long polling (tests drive the engine with syncNow()). */
  manual?: boolean;
+  /** Push notifications; defaults to the app's realtime connection (not used in manual mode). */
+  push?: PushChannel;
 }

 const LEASE_GRACE_MS = 20_000; // keep a left block for this long
 const IDLE_RELEASE_MS = 60_000; // release even the focused block after this long without input
 const SYNC_DEBOUNCE_MS = 300;
-const POLL_COLLAB_MS = 3_000;
-const POLL_SOLO_MS = 30_000;
+const POLL_COLLAB_MS = 3_000; // shared note without push or long polling
+const POLL_SOLO_MS = 30_000; // private note, opened once
+const HEARTBEAT_MS = 7_000; // renews block leases while changes arrive by push/long polling
+const SAFETY_POLL_MS = 30_000; // in case a push notification got lost
+const LONG_POLL_WAIT_S = 25; // below typical proxy timeouts (nginx 60 s, Cloudflare 100 s)
+const LONG_POLL_MAX_FAILURES = 3;
 const BACKOFF_MAX_MS = 30_000;
 const UNDO_LIMIT = 200;
 const UNDO_GROUP_MS = 1_000; // keystrokes in the same block are one undo step until a pause this long
@@ -131,6 +142,14 @@ export class DocSyncEngine {
  private undoStack: UndoStep[] = [];
  private redoStack: UndoStep[] = [];

+  private push: PushChannel | null;
+  private pushKind: PushKind = 'none';
+  private unsubscribePush: (() => void) | null = null;
+  private pokeTimer: ReturnType<typeof setTimeout> | null = null;
+  private longPollRunning = false;
+  private longPollBroken = false;
+  private longPollAbort: AbortController | null = null;
+
  private syncing = false;
  private syncAgain = false;
  private stopped = false;
@@ -144,6 +163,7 @@ export class DocSyncEngine {
    this.storage = opts.storage || defaultStorage();
    this.now = opts.now || (() => Date.now());
    this.manual = !!opts.manual;
+    this.push = opts.push || (this.manual ? null : realtimePush);
  }

  // --- Public API ---
@@ -165,6 +185,13 @@ export class DocSyncEngine {
      // Storage unavailable: continue with the server state only
    }
    if (!this.manual) this.attachWindowListeners();
+    if (this.push) {
+      this.unsubscribePush = this.push.subscribe(
+        this.docId,
+        () => this.onPoke(),
+        (kind) => this.onPushStatus(kind)
+      );
+    }
    if (!this.ready) {
      try {
        await this.loadSnapshot();
@@ -185,6 +212,10 @@ export class DocSyncEngine {
    this.flushOnExit();
    this.stopped = true;
    if (this.timer) clearTimeout(this.timer);
+    if (this.pokeTimer) clearTimeout(this.pokeTimer);
+    this.longPollAbort?.abort();
+    this.unsubscribePush?.();
+    this.unsubscribePush = null;
    this.cleanup.forEach((fn) => fn());
    this.cleanup = [];
    this.listeners.clear();
@@ -212,6 +243,7 @@ export class DocSyncEngine {
      conflicts: this.conflicts,
      canUndo: this.undoStack.length > 0,
      canRedo: this.redoStack.length > 0,
+      channel: this.channel(),
    };
  }

@@ -314,9 +346,15 @@ export class DocSyncEngine {
  /** Tells the engine which block the cursor is in (null = none); used for block leases. */
  focusBlock(blockId: string | null): void {
    const t = this.now();
+    const changed = blockId !== this.focused;
    if (this.focused) this.activity.set(this.focused, t);
    this.focused = blockId;
    if (blockId) this.activity.set(blockId, t);
+    // Announce the lease right away: with push the next regular sync may be up to 30 s away
+    if (changed && blockId && this.canEdit && this.collaborative()) {
+      if (this.syncing) this.syncAgain = true;
+      else this.schedule(SYNC_DEBOUNCE_MS);
+    }
  }

  /** Runs one sync round trip now (or right after the running one). */
@@ -331,19 +369,26 @@ export class DocSyncEngine {
      this.timer = null;
    }
    this.syncing = true;
-    let delay: number;
+    let failed = false;
    try {
      await this.roundTrip();
      this.backoff = 0;
-      // Decided only now: the response tells whether the note is shared (3 s) or private (30 s)
-      delay = this.syncAgain || this.pending.length > 0 ? 0 : this.pollDelay();
    } catch (err) {
      this.handleError(err);
-      delay = this.status === 'error' ? this.pollDelay() : this.nextBackoff();
+      failed = true;
    } finally {
      this.syncing = false;
-      this.syncAgain = false;
      this.recompute();
+      if (!failed) this.ensureLongPoll();
+      // Decided only now: the response tells whether the note is shared and how changes arrive
+      const delay = failed
+        ? this.status === 'error'
+          ? this.pollDelay()
+          : this.nextBackoff()
+        : this.syncAgain || this.pending.length > 0
+        ? 0
+        : this.pollDelay();
+      this.syncAgain = false;
      this.emit();
      this.schedule(delay);
    }
@@ -548,8 +593,76 @@ export class DocSyncEngine {
    this.errorMessage = err instanceof Error ? err.message : String(err);
  }

+  private channel(): SyncChannel {
+    if (this.pushKind !== 'none') return this.pushKind;
+    if (this.longPollRunning && !this.longPollBroken) return 'longpoll';
+    return 'interval';
+  }
+
  private pollDelay(): number {
-    return this.collaborative() ? POLL_COLLAB_MS : POLL_SOLO_MS;
+    if (!this.collaborative()) return POLL_SOLO_MS;
+    if (this.channel() === 'interval') return POLL_COLLAB_MS;
+    return this.canEdit && this.holds().length > 0 ? HEARTBEAT_MS : SAFETY_POLL_MS;
+  }
+
+  private onPoke(): void {
+    if (this.pokeTimer || this.stopped) return;
+    // Several notifications in a row (e.g. typing of others) lead to one sync
+    this.pokeTimer = setTimeout(() => {
+      this.pokeTimer = null;
+      void this.syncNow();
+    }, 50);
+  }
+
+  private onPushStatus(kind: PushKind): void {
+    const wasLive = this.pushKind !== 'none';
+    this.pushKind = kind;
+    if (kind !== 'none') {
+      this.longPollAbort?.abort();
+      if (!wasLive) void this.syncNow(); // catch up on what was missed while disconnected
+    } else {
+      this.ensureLongPoll();
+      if (!this.syncing) this.schedule(this.pollDelay());
+    }
+    this.emit();
+  }
+
+  /** Starts long polling when there is no push channel (e.g. a proxy blocks WebSocket and SSE). */
+  private ensureLongPoll(): void {
+    if (this.manual || this.stopped || !this.ready || this.longPollRunning || this.longPollBroken) return;
+    if (this.pushKind !== 'none' || !this.collaborative() || this.status === 'error') return;
+    void this.longPollLoop();
+  }
+
+  private async longPollLoop(): Promise<void> {
+    this.longPollRunning = true;
+    this.emit();
+    let failures = 0;
+    try {
+      while (!this.stopped && this.pushKind === 'none' && this.collaborative() && this.status !== 'error') {
+        const ctrl = new AbortController();
+        this.longPollAbort = ctrl;
+        try {
+          const res = await this.transport.changes(this.docId, this.lastSeq, LONG_POLL_WAIT_S, ctrl.signal);
+          failures = 0;
+          if (res.changed || res.headSeq !== this.lastSeq) await this.syncNow();
+        } catch {
+          if (ctrl.signal.aborted || this.stopped) break;
+          failures++;
+          if (failures >= LONG_POLL_MAX_FAILURES) {
+            // Long polling does not get through either: plain polling every 3 s remains
+            this.longPollBroken = true;
+            break;
+          }
+          await new Promise((r) => setTimeout(r, 1000 * failures));
+        }
+      }
+    } finally {
+      this.longPollRunning = false;
+      this.longPollAbort = null;
+      this.emit();
+      if (!this.syncing && !this.stopped) this.schedule(this.pollDelay());
+    }
  }

  private nextBackoff(): number {
@@ -580,7 +693,10 @@ export class DocSyncEngine {
  }

  private attachWindowListeners(): void {
-    const onOnline = () => void this.syncNow();
+    const onOnline = () => {
+      this.longPollBroken = false; // the network changed: long polling may work again
+      void this.syncNow();
+    };
    const onVisibility = () => {
      if (document.visibilityState === 'hidden') this.flushOnExit();
      else void this.syncNow();
@@ -0,0 +1,25 @@
+// Push notifications for note sync: the server sends DOC_SEQ into the note's room whenever the note, its
+// leases or its presence change. Delivery uses the app-wide realtime connection (WebSocket, else SSE).
+import { createRealtimeConnection, getRealtimeTransport, isRealtimeConnected } from '../api';
+
+export type PushKind = 'ws' | 'sse' | 'none';
+
+export interface PushChannel {
+  /** Calls onPoke for changes of docId and onStatus whenever the channel goes up or down. */
+  subscribe(docId: string, onPoke: () => void, onStatus: (kind: PushKind) => void): () => void;
+}
+
+export const realtimePush: PushChannel = {
+  subscribe(docId, onPoke, onStatus) {
+    const current = (): PushKind => (isRealtimeConnected() ? (getRealtimeTransport() as PushKind) : 'none');
+    const conn = createRealtimeConnection({
+      docId,
+      onMessage: (msg: any) => {
+        if (msg?.event === 'DOC_SEQ' && msg.payload?.docId === docId) onPoke();
+      },
+      onStatusChange: () => onStatus(current()),
+    });
+    onStatus(current());
+    return () => conn.close();
+  },
+};
@@ -59,20 +59,28 @@ export class TransportError extends Error {
  }
 }

+export interface ChangesResponse {
+  headSeq: number;
+  changed: boolean;
+}
+
 export interface Transport {
  snapshot(docId: string): Promise<SnapshotResponse>;
+  /** Long poll: resolves when the note changes, at the latest after waitSec seconds. */
+  changes(docId: string, since: number, waitSec: number, signal?: AbortSignal): Promise<ChangesResponse>;
  sync(docId: string, body: SyncRequest, opts?: { keepalive?: boolean }): Promise<SyncResponse>;
  leave(docId: string, clientId: string): Promise<void>;
 }

 const TIMEOUT_MS = 15000;

-async function call<T>(path: string, init: RequestInit & { keepalive?: boolean } = {}): Promise<T> {
+async function call<T>(path: string, init: RequestInit & { keepalive?: boolean } = {}, timeoutMs = TIMEOUT_MS): Promise<T> {
  const headers: Record<string, string> = { 'Content-Type': 'application/json' };
  const token = getToken();
  if (token) headers.Authorization = `Bearer ${token}`;
  const controller = new AbortController();
-  const timer = setTimeout(() => controller.abort(), TIMEOUT_MS);
+  const timer = setTimeout(() => controller.abort(), timeoutMs);
+  init.signal?.addEventListener('abort', () => controller.abort());
  let res: Response;
  try {
    res = await fetch(path, { ...init, headers, cache: 'no-store', signal: init.keepalive ? undefined : controller.signal });
@@ -99,6 +107,8 @@ async function call<T>(path: string, init: RequestInit & { keepalive?: boolean }

 export const httpTransport: Transport = {
  snapshot: (docId) => call(`/api/v1/docs/${encodeURIComponent(docId)}/snapshot`),
+  changes: (docId, since, waitSec, signal) =>
+    call(`/api/v1/docs/${encodeURIComponent(docId)}/changes?since=${since}&wait=${waitSec}`, { signal }, (waitSec + 10) * 1000),
  sync: (docId, body, opts) =>
    call(`/api/v1/docs/${encodeURIComponent(docId)}/sync`, { method: 'POST', body: JSON.stringify(body), keepalive: opts?.keepalive }),
  leave: async (docId, clientId) => {