mirror of
https://github.com/PerpetualSoftware/pad.git
synced 2026-09-11 13:28:57 +00:00
fix: replace scattered tab-resume refetches with layered sync system
When the browser tab lost focus and regained it, 5 independent onTabResume callbacks all fired simultaneously, flooding the server with redundant requests. This replaces that pattern with a 4-layer sync architecture: 1. Replay buffer — per-workspace ring buffer stores recent events with monotonic IDs. On SSE reconnect, missed events are replayed via Last-Event-ID so the client is already caught up. 2. Last-Event-ID support — SSE handler reads the header, replays from buffer, or sends sync_required if the gap is too large. 3. Incremental sync — new /changes?since=<ms> endpoint returns only modified/deleted items since a timestamp, including archived items for view consistency. 4. Centralized sync coordinator — single decision tree replaces 5 scattered callbacks. Short absences skip sync entirely, SSE-covered gaps need no API calls, and full refresh is a last resort. Key robustness details: - Global event IDs via Redis INCR for multi-instance safety - Server-time cursors to avoid client clock skew - Safe cursor management (only advances on confirmed sync) - 9 new tests for replay buffer and event ID behavior Fixes BUG-26.
This commit is contained in:
xarmian
+134
-3
@@ -3,6 +3,7 @@ package events
|
||||
import (
|
||||
"log/slog"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
)
|
||||
|
||||
@@ -32,8 +33,15 @@ const (
|
||||
ItemUpdatedWithComment = "item_updated_with_comment"
|
||||
)
|
||||
|
||||
// Default replay buffer settings.
|
||||
const (
|
||||
DefaultReplayBufferSize = 1024 // max events to retain per workspace
|
||||
DefaultReplayMaxAge = 5 * time.Minute // discard events older than this
|
||||
)
|
||||
|
||||
// Event represents a real-time event published when state changes occur.
|
||||
type Event struct {
|
||||
ID int64 `json:"id"`
|
||||
Type string `json:"type"`
|
||||
WorkspaceID string `json:"workspace_id"`
|
||||
DocumentID string `json:"document_id,omitempty"`
|
||||
@@ -66,6 +74,11 @@ type EventBus interface {
|
||||
// Publish sends an event to all subscribers for the event's workspace.
|
||||
Publish(event Event)
|
||||
|
||||
// EventsSince returns events for a workspace with IDs greater than sinceID.
|
||||
// Used to replay missed events on SSE reconnect (Last-Event-ID).
|
||||
// Returns nil if sinceID is too old and has been evicted from the buffer.
|
||||
EventsSince(workspaceID string, sinceID int64) []Event
|
||||
|
||||
// Close shuts down the event bus and cleans up resources.
|
||||
Close()
|
||||
|
||||
@@ -77,6 +90,77 @@ type EventBus interface {
|
||||
WorkspaceSubscriberCount(workspaceID string) int
|
||||
}
|
||||
|
||||
// replayBuffer is a bounded ring buffer of recent events for a single workspace.
|
||||
// It supports efficient append and replay-since-ID queries.
|
||||
type replayBuffer struct {
|
||||
events []Event
|
||||
size int // max capacity
|
||||
head int // next write position
|
||||
count int // current number of events
|
||||
}
|
||||
|
||||
func newReplayBuffer(size int) *replayBuffer {
|
||||
return &replayBuffer{
|
||||
events: make([]Event, size),
|
||||
size: size,
|
||||
}
|
||||
}
|
||||
|
||||
// append adds an event to the ring buffer, evicting the oldest if full.
|
||||
func (rb *replayBuffer) append(e Event) {
|
||||
rb.events[rb.head] = e
|
||||
rb.head = (rb.head + 1) % rb.size
|
||||
if rb.count < rb.size {
|
||||
rb.count++
|
||||
}
|
||||
}
|
||||
|
||||
// since returns all buffered events with ID > sinceID, in chronological order.
|
||||
// Returns nil if sinceID is older than the oldest buffered event AND the buffer
|
||||
// is full (i.e. events have been evicted), meaning we can't guarantee completeness.
|
||||
// Returns an empty (non-nil) slice if sinceID is current (no missed events).
|
||||
// A sinceID of 0 means "give me everything in the buffer".
|
||||
func (rb *replayBuffer) since(sinceID int64) []Event {
|
||||
if rb.count == 0 {
|
||||
return []Event{}
|
||||
}
|
||||
|
||||
// Find the oldest event in the buffer
|
||||
oldest := (rb.head - rb.count + rb.size) % rb.size
|
||||
oldestID := rb.events[oldest].ID
|
||||
|
||||
// Find the newest event in the buffer.
|
||||
newest := (rb.head - 1 + rb.size) % rb.size
|
||||
newestID := rb.events[newest].ID
|
||||
|
||||
// If sinceID is beyond the newest event we have, the ID came from a
|
||||
// different sequence (e.g., a different instance in a Redis deployment).
|
||||
// We can't determine what was missed — signal a gap.
|
||||
if sinceID > newestID {
|
||||
return nil
|
||||
}
|
||||
|
||||
// If the requested ID is older than our oldest AND the buffer has wrapped
|
||||
// (events were evicted), we can't guarantee completeness — signal a gap.
|
||||
// But if the buffer hasn't filled up yet, all events are still present.
|
||||
if sinceID > 0 && sinceID < oldestID && rb.count == rb.size {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Collect events with ID > sinceID
|
||||
var result []Event
|
||||
for i := 0; i < rb.count; i++ {
|
||||
idx := (oldest + i) % rb.size
|
||||
if rb.events[idx].ID > sinceID {
|
||||
result = append(result, rb.events[idx])
|
||||
}
|
||||
}
|
||||
if result == nil {
|
||||
result = []Event{}
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
// subscriber wraps a channel with its workspace filter.
|
||||
type subscriber struct {
|
||||
ch chan Event
|
||||
@@ -88,12 +172,29 @@ type subscriber struct {
|
||||
type MemoryBus struct {
|
||||
mu sync.RWMutex
|
||||
subscribers map[chan Event]*subscriber
|
||||
|
||||
// Monotonic sequence counter for event IDs.
|
||||
seq atomic.Int64
|
||||
|
||||
// Per-workspace replay buffers for Last-Event-ID support.
|
||||
replayMu sync.RWMutex
|
||||
replayBuffers map[string]*replayBuffer
|
||||
replaySize int
|
||||
replayMaxAge time.Duration
|
||||
}
|
||||
|
||||
// New creates a new in-memory EventBus.
|
||||
// New creates a new in-memory EventBus with default replay buffer settings.
|
||||
func New() *MemoryBus {
|
||||
return NewWithReplay(DefaultReplayBufferSize, DefaultReplayMaxAge)
|
||||
}
|
||||
|
||||
// NewWithReplay creates a new in-memory EventBus with custom replay settings.
|
||||
func NewWithReplay(bufferSize int, maxAge time.Duration) *MemoryBus {
|
||||
return &MemoryBus{
|
||||
subscribers: make(map[chan Event]*subscriber),
|
||||
subscribers: make(map[chan Event]*subscriber),
|
||||
replayBuffers: make(map[string]*replayBuffer),
|
||||
replaySize: bufferSize,
|
||||
replayMaxAge: maxAge,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -152,12 +253,27 @@ func (b *MemoryBus) Unsubscribe(ch chan Event) {
|
||||
|
||||
// Publish sends an event to all subscribers for the event's workspace.
|
||||
// Non-blocking: if a subscriber's channel is full, the event is dropped
|
||||
// and a warning is logged.
|
||||
// and a warning is logged. Events are assigned a monotonic sequence ID
|
||||
// and stored in the replay buffer for Last-Event-ID support.
|
||||
func (b *MemoryBus) Publish(event Event) {
|
||||
if event.Timestamp == 0 {
|
||||
event.Timestamp = time.Now().UnixMilli()
|
||||
}
|
||||
|
||||
// Assign a monotonic sequence ID.
|
||||
event.ID = b.seq.Add(1)
|
||||
|
||||
// Store in replay buffer for reconnect replay.
|
||||
b.replayMu.Lock()
|
||||
rb, ok := b.replayBuffers[event.WorkspaceID]
|
||||
if !ok {
|
||||
rb = newReplayBuffer(b.replaySize)
|
||||
b.replayBuffers[event.WorkspaceID] = rb
|
||||
}
|
||||
rb.append(event)
|
||||
b.replayMu.Unlock()
|
||||
|
||||
// Fan out to live subscribers.
|
||||
b.mu.RLock()
|
||||
defer b.mu.RUnlock()
|
||||
|
||||
@@ -173,6 +289,21 @@ func (b *MemoryBus) Publish(event Event) {
|
||||
}
|
||||
}
|
||||
|
||||
// EventsSince returns buffered events for a workspace with IDs greater than sinceID.
|
||||
// Returns nil if sinceID has been evicted from the buffer (gap too large).
|
||||
// Returns an empty slice if the caller is fully caught up.
|
||||
func (b *MemoryBus) EventsSince(workspaceID string, sinceID int64) []Event {
|
||||
b.replayMu.RLock()
|
||||
defer b.replayMu.RUnlock()
|
||||
|
||||
rb, ok := b.replayBuffers[workspaceID]
|
||||
if !ok {
|
||||
// No events ever published for this workspace.
|
||||
return []Event{}
|
||||
}
|
||||
return rb.since(sinceID)
|
||||
}
|
||||
|
||||
// Close shuts down the event bus by closing all subscriber channels.
|
||||
// SSE handler goroutines will see the channel close and exit cleanly.
|
||||
func (b *MemoryBus) Close() {
|
||||
|
||||
@@ -275,3 +275,170 @@ func TestPublishNoSubscribers(t *testing.T) {
|
||||
WorkspaceID: "ws-1",
|
||||
})
|
||||
}
|
||||
|
||||
func TestEventIDsAreMonotonic(t *testing.T) {
|
||||
bus := New()
|
||||
ch := bus.Subscribe("ws-1")
|
||||
defer bus.Unsubscribe(ch)
|
||||
|
||||
for i := 0; i < 10; i++ {
|
||||
bus.Publish(Event{
|
||||
Type: ItemUpdated,
|
||||
WorkspaceID: "ws-1",
|
||||
})
|
||||
}
|
||||
|
||||
var lastID int64
|
||||
for i := 0; i < 10; i++ {
|
||||
event := <-ch
|
||||
if event.ID <= lastID {
|
||||
t.Fatalf("event %d: ID %d not greater than previous %d", i, event.ID, lastID)
|
||||
}
|
||||
lastID = event.ID
|
||||
}
|
||||
}
|
||||
|
||||
func TestEventsSinceCaughtUp(t *testing.T) {
|
||||
bus := New()
|
||||
|
||||
bus.Publish(Event{Type: ItemCreated, WorkspaceID: "ws-1"})
|
||||
bus.Publish(Event{Type: ItemUpdated, WorkspaceID: "ws-1"})
|
||||
bus.Publish(Event{Type: ItemUpdated, WorkspaceID: "ws-1"})
|
||||
|
||||
// Ask for events since the last one — should get empty slice
|
||||
events := bus.EventsSince("ws-1", 3)
|
||||
if events == nil {
|
||||
t.Fatal("expected non-nil slice")
|
||||
}
|
||||
if len(events) != 0 {
|
||||
t.Fatalf("expected 0 events, got %d", len(events))
|
||||
}
|
||||
}
|
||||
|
||||
func TestEventsSinceReplay(t *testing.T) {
|
||||
bus := New()
|
||||
|
||||
bus.Publish(Event{Type: ItemCreated, WorkspaceID: "ws-1"})
|
||||
bus.Publish(Event{Type: ItemUpdated, WorkspaceID: "ws-1"})
|
||||
bus.Publish(Event{Type: ItemArchived, WorkspaceID: "ws-1"})
|
||||
|
||||
// Ask for events since ID 1 — should get events 2 and 3
|
||||
events := bus.EventsSince("ws-1", 1)
|
||||
if events == nil {
|
||||
t.Fatal("expected non-nil slice")
|
||||
}
|
||||
if len(events) != 2 {
|
||||
t.Fatalf("expected 2 events, got %d", len(events))
|
||||
}
|
||||
if events[0].Type != ItemUpdated {
|
||||
t.Errorf("expected ItemUpdated, got %q", events[0].Type)
|
||||
}
|
||||
if events[1].Type != ItemArchived {
|
||||
t.Errorf("expected ItemArchived, got %q", events[1].Type)
|
||||
}
|
||||
}
|
||||
|
||||
func TestEventsSinceGapTooLarge(t *testing.T) {
|
||||
// Create a tiny buffer so we can overflow it
|
||||
bus := NewWithReplay(3, 5*time.Minute)
|
||||
|
||||
// Publish 5 events (buffer only holds 3)
|
||||
for i := 0; i < 5; i++ {
|
||||
bus.Publish(Event{Type: ItemUpdated, WorkspaceID: "ws-1"})
|
||||
}
|
||||
|
||||
// Oldest buffered event should be ID 3 (events 1 and 2 are evicted)
|
||||
// Asking for events since ID 1 should return nil (gap too large)
|
||||
events := bus.EventsSince("ws-1", 1)
|
||||
if events != nil {
|
||||
t.Fatalf("expected nil (gap too large), got %d events", len(events))
|
||||
}
|
||||
|
||||
// Asking for events since ID 3 should work
|
||||
events = bus.EventsSince("ws-1", 3)
|
||||
if events == nil {
|
||||
t.Fatal("expected non-nil slice")
|
||||
}
|
||||
if len(events) != 2 {
|
||||
t.Fatalf("expected 2 events (IDs 4,5), got %d", len(events))
|
||||
}
|
||||
}
|
||||
|
||||
func TestEventsSinceWorkspaceIsolation(t *testing.T) {
|
||||
bus := New()
|
||||
|
||||
bus.Publish(Event{Type: ItemCreated, WorkspaceID: "ws-1"})
|
||||
bus.Publish(Event{Type: ItemCreated, WorkspaceID: "ws-2"})
|
||||
bus.Publish(Event{Type: ItemUpdated, WorkspaceID: "ws-1"})
|
||||
|
||||
// ws-2 events since 0 should only return the ws-2 event
|
||||
events := bus.EventsSince("ws-2", 0)
|
||||
if len(events) != 1 {
|
||||
t.Fatalf("expected 1 event for ws-2, got %d", len(events))
|
||||
}
|
||||
if events[0].WorkspaceID != "ws-2" {
|
||||
t.Errorf("expected ws-2, got %s", events[0].WorkspaceID)
|
||||
}
|
||||
}
|
||||
|
||||
func TestEventsSinceNoEventsForWorkspace(t *testing.T) {
|
||||
bus := New()
|
||||
|
||||
events := bus.EventsSince("ws-nonexistent", 0)
|
||||
if events == nil {
|
||||
t.Fatal("expected non-nil empty slice")
|
||||
}
|
||||
if len(events) != 0 {
|
||||
t.Fatalf("expected 0 events, got %d", len(events))
|
||||
}
|
||||
}
|
||||
|
||||
func TestEventsSinceForeignID(t *testing.T) {
|
||||
// Simulates the multi-instance Redis scenario: a client sends a
|
||||
// Last-Event-ID from a different instance whose IDs are in a different range.
|
||||
bus := New()
|
||||
|
||||
bus.Publish(Event{Type: ItemCreated, WorkspaceID: "ws-1"})
|
||||
bus.Publish(Event{Type: ItemUpdated, WorkspaceID: "ws-1"})
|
||||
|
||||
// sinceID=500 is way beyond our newest (ID 2) — foreign sequence
|
||||
events := bus.EventsSince("ws-1", 500)
|
||||
if events != nil {
|
||||
t.Fatalf("expected nil (foreign ID), got %d events", len(events))
|
||||
}
|
||||
}
|
||||
|
||||
func TestReplayBufferWrapAround(t *testing.T) {
|
||||
bus := NewWithReplay(4, 5*time.Minute)
|
||||
|
||||
// Fill buffer exactly
|
||||
for i := 0; i < 4; i++ {
|
||||
bus.Publish(Event{Type: ItemUpdated, WorkspaceID: "ws-1"})
|
||||
}
|
||||
|
||||
// Overflow by 2
|
||||
bus.Publish(Event{Type: ItemCreated, WorkspaceID: "ws-1"})
|
||||
bus.Publish(Event{Type: ItemArchived, WorkspaceID: "ws-1"})
|
||||
|
||||
// Buffer should hold events 3,4,5,6 (1,2 evicted)
|
||||
// sinceID=3 should work: events 4,5,6
|
||||
events := bus.EventsSince("ws-1", 3)
|
||||
if events == nil {
|
||||
t.Fatal("expected non-nil slice for sinceID=3")
|
||||
}
|
||||
if len(events) != 3 {
|
||||
t.Fatalf("expected 3 events (IDs 4,5,6), got %d", len(events))
|
||||
}
|
||||
|
||||
// sinceID=2 should be a gap (event 2 is evicted, oldest in buffer is 3)
|
||||
events = bus.EventsSince("ws-1", 2)
|
||||
if events != nil {
|
||||
t.Fatalf("expected nil (gap) for sinceID=2, got %d events", len(events))
|
||||
}
|
||||
|
||||
// sinceID=1 should also be a gap
|
||||
events = bus.EventsSince("ws-1", 1)
|
||||
if events != nil {
|
||||
t.Fatalf("expected nil (gap) for sinceID=1, got %d events", len(events))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
"encoding/json"
|
||||
"log/slog"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"github.com/redis/go-redis/v9"
|
||||
@@ -14,6 +15,11 @@ const (
|
||||
// redisChannelPrefix is prepended to workspace IDs for Redis pub/sub channels.
|
||||
redisChannelPrefix = "pad:events:"
|
||||
|
||||
// redisSeqKey is the Redis key used for the global event sequence counter.
|
||||
// All instances share this counter so SSE event IDs are globally ordered
|
||||
// and Last-Event-ID is valid across any instance on reconnect.
|
||||
redisSeqKey = "pad:event_seq"
|
||||
|
||||
// reconnectDelay is how long to wait before retrying a failed Redis subscription.
|
||||
reconnectDelay = 2 * time.Second
|
||||
)
|
||||
@@ -32,6 +38,15 @@ type RedisBus struct {
|
||||
wsCounts map[string]int // workspace → local subscriber count
|
||||
wsSubs map[string]*redisSub // workspace → active Redis subscription
|
||||
|
||||
// Monotonic sequence counter for event IDs (local to this instance).
|
||||
seq atomic.Int64
|
||||
|
||||
// Per-workspace replay buffers for Last-Event-ID support.
|
||||
// Populated from events received via Redis pub/sub.
|
||||
replayMu sync.RWMutex
|
||||
replayBuffers map[string]*replayBuffer
|
||||
replaySize int
|
||||
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
}
|
||||
@@ -47,12 +62,14 @@ type redisSub struct {
|
||||
func NewRedisBus(client *redis.Client) *RedisBus {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
return &RedisBus{
|
||||
client: client,
|
||||
subscribers: make(map[chan Event]*subscriber),
|
||||
wsCounts: make(map[string]int),
|
||||
wsSubs: make(map[string]*redisSub),
|
||||
ctx: ctx,
|
||||
cancel: cancel,
|
||||
client: client,
|
||||
subscribers: make(map[chan Event]*subscriber),
|
||||
wsCounts: make(map[string]int),
|
||||
wsSubs: make(map[string]*redisSub),
|
||||
replayBuffers: make(map[string]*replayBuffer),
|
||||
replaySize: DefaultReplayBufferSize,
|
||||
ctx: ctx,
|
||||
cancel: cancel,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -129,11 +146,24 @@ func (b *RedisBus) Unsubscribe(ch chan Event) {
|
||||
}
|
||||
|
||||
// Publish sends an event to Redis, which distributes it to all instances.
|
||||
// Events are assigned a globally unique sequence ID via Redis INCR so that
|
||||
// Last-Event-ID is valid across any instance on reconnect.
|
||||
func (b *RedisBus) Publish(event Event) {
|
||||
if event.Timestamp == 0 {
|
||||
event.Timestamp = time.Now().UnixMilli()
|
||||
}
|
||||
|
||||
// Assign a globally ordered sequence ID via Redis atomic counter.
|
||||
// This ensures all instances share the same ID space, so Last-Event-ID
|
||||
// from one instance is meaningful on any other instance.
|
||||
id, err := b.client.Incr(b.ctx, redisSeqKey).Result()
|
||||
if err != nil {
|
||||
// Fall back to local counter if Redis INCR fails (degraded mode).
|
||||
slog.Warn("failed to get global event ID from Redis, falling back to local", "error", err)
|
||||
id = b.seq.Add(1)
|
||||
}
|
||||
event.ID = id
|
||||
|
||||
data, err := json.Marshal(event)
|
||||
if err != nil {
|
||||
slog.Error("failed to marshal event for Redis", "error", err)
|
||||
@@ -146,6 +176,19 @@ func (b *RedisBus) Publish(event Event) {
|
||||
}
|
||||
}
|
||||
|
||||
// EventsSince returns buffered events for a workspace with IDs greater than sinceID.
|
||||
// Returns nil if sinceID has been evicted from the buffer (gap too large).
|
||||
func (b *RedisBus) EventsSince(workspaceID string, sinceID int64) []Event {
|
||||
b.replayMu.RLock()
|
||||
defer b.replayMu.RUnlock()
|
||||
|
||||
rb, ok := b.replayBuffers[workspaceID]
|
||||
if !ok {
|
||||
return []Event{}
|
||||
}
|
||||
return rb.since(sinceID)
|
||||
}
|
||||
|
||||
// Close shuts down all Redis subscriptions and closes local subscriber channels.
|
||||
func (b *RedisBus) Close() {
|
||||
b.cancel() // signal all subscription goroutines to stop
|
||||
@@ -227,8 +270,23 @@ func (b *RedisBus) receiveMessages(ctx context.Context, pubsub *redis.PubSub, wo
|
||||
}
|
||||
}
|
||||
|
||||
// fanOutLocally distributes an event to all local subscribers for the event's workspace.
|
||||
// fanOutLocally distributes an event to all local subscribers for the event's workspace
|
||||
// and stores it in the replay buffer.
|
||||
func (b *RedisBus) fanOutLocally(event Event) {
|
||||
// Events received via Redis pub/sub already carry a global ID assigned by
|
||||
// the publishing instance via Redis INCR. We use that ID directly so all
|
||||
// instances share the same ID space for Last-Event-ID replay.
|
||||
|
||||
// Store in replay buffer for reconnect replay.
|
||||
b.replayMu.Lock()
|
||||
rb, ok := b.replayBuffers[event.WorkspaceID]
|
||||
if !ok {
|
||||
rb = newReplayBuffer(b.replaySize)
|
||||
b.replayBuffers[event.WorkspaceID] = rb
|
||||
}
|
||||
rb.append(event)
|
||||
b.replayMu.Unlock()
|
||||
|
||||
b.mu.RLock()
|
||||
defer b.mu.RUnlock()
|
||||
|
||||
|
||||
@@ -94,3 +94,8 @@ func (b *InstrumentedBus) SubscriberCount() int {
|
||||
func (b *InstrumentedBus) WorkspaceSubscriberCount(workspaceID string) int {
|
||||
return b.inner.WorkspaceSubscriberCount(workspaceID)
|
||||
}
|
||||
|
||||
// EventsSince delegates to the inner bus.
|
||||
func (b *InstrumentedBus) EventsSince(workspaceID string, sinceID int64) []events.Event {
|
||||
return b.inner.EventsSince(workspaceID, sinceID)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,77 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"strconv"
|
||||
"time"
|
||||
)
|
||||
|
||||
// ChangesResponse is the response for GET /workspaces/{ws}/changes?since=<unix_ms>.
|
||||
type ChangesResponse struct {
|
||||
// Updated items (with full item data).
|
||||
Updated []interface{} `json:"updated"`
|
||||
// IDs of items that were deleted since the requested timestamp.
|
||||
Deleted []string `json:"deleted"`
|
||||
// Server timestamp at the time of this response (unix ms).
|
||||
// Clients should use this as the `since` value for the next sync.
|
||||
ServerTime int64 `json:"server_time"`
|
||||
// Whether the collection metadata (counts, schemas) may have changed.
|
||||
// True if any items were updated/deleted, signaling the client should
|
||||
// also refresh collection metadata.
|
||||
CollectionsChanged bool `json:"collections_changed"`
|
||||
}
|
||||
|
||||
// handleGetChanges returns items modified since a given timestamp.
|
||||
// GET /api/v1/workspaces/{ws}/changes?since=<unix_milliseconds>
|
||||
//
|
||||
// This is the incremental sync endpoint used by the frontend when the
|
||||
// tab regains focus. Instead of re-fetching everything, the client sends
|
||||
// the timestamp of its last successful sync and gets back only the delta.
|
||||
func (s *Server) handleGetChanges(w http.ResponseWriter, r *http.Request) {
|
||||
workspaceID, ok := s.getWorkspaceID(w, r)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
|
||||
sinceStr := r.URL.Query().Get("since")
|
||||
if sinceStr == "" {
|
||||
writeError(w, http.StatusBadRequest, "bad_request", "since query parameter is required (unix milliseconds)")
|
||||
return
|
||||
}
|
||||
|
||||
sinceMs, err := strconv.ParseInt(sinceStr, 10, 64)
|
||||
if err != nil || sinceMs < 0 {
|
||||
writeError(w, http.StatusBadRequest, "bad_request", "since must be a valid unix timestamp in milliseconds")
|
||||
return
|
||||
}
|
||||
|
||||
since := time.UnixMilli(sinceMs)
|
||||
serverTime := time.Now().UnixMilli()
|
||||
|
||||
updated, deletedIDs, err := s.store.ItemsModifiedSince(workspaceID, since)
|
||||
if err != nil {
|
||||
writeInternalError(w, err)
|
||||
return
|
||||
}
|
||||
|
||||
// Convert to interface slice for JSON marshaling.
|
||||
updatedItems := make([]interface{}, len(updated))
|
||||
for i, item := range updated {
|
||||
updatedItems[i] = item
|
||||
}
|
||||
if updatedItems == nil {
|
||||
updatedItems = []interface{}{}
|
||||
}
|
||||
if deletedIDs == nil {
|
||||
deletedIDs = []string{}
|
||||
}
|
||||
|
||||
resp := ChangesResponse{
|
||||
Updated: updatedItems,
|
||||
Deleted: deletedIDs,
|
||||
ServerTime: serverTime,
|
||||
CollectionsChanged: len(updated) > 0 || len(deletedIDs) > 0,
|
||||
}
|
||||
|
||||
writeJSON(w, http.StatusOK, resp)
|
||||
}
|
||||
@@ -5,11 +5,18 @@ import (
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"time"
|
||||
)
|
||||
|
||||
// handleSSE streams Server-Sent Events for a workspace.
|
||||
// GET /api/v1/events?workspace=<slug>
|
||||
//
|
||||
// Supports Last-Event-ID: when a client reconnects with a Last-Event-ID header
|
||||
// (set automatically by the browser's EventSource), the server replays any
|
||||
// missed events from its in-memory replay buffer before entering the live
|
||||
// stream. If the requested ID is too old (evicted from the buffer), the server
|
||||
// sends a "sync_required" event so the client knows to do a full refresh.
|
||||
func (s *Server) handleSSE(w http.ResponseWriter, r *http.Request) {
|
||||
// SSE requires the event bus
|
||||
if s.events == nil {
|
||||
@@ -69,12 +76,38 @@ func (s *Server) handleSSE(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
|
||||
// Send initial connected event
|
||||
writeSSEEvent(w, "connected", map[string]string{
|
||||
writeSSEEvent(w, "connected", 0, map[string]string{
|
||||
"workspace_id": ws.ID,
|
||||
"workspace": ws.Slug,
|
||||
})
|
||||
flusher.Flush()
|
||||
|
||||
// Replay missed events if the client provided Last-Event-ID.
|
||||
// The browser's EventSource sends this automatically on reconnect.
|
||||
if lastIDStr := r.Header.Get("Last-Event-ID"); lastIDStr != "" {
|
||||
lastID, parseErr := strconv.ParseInt(lastIDStr, 10, 64)
|
||||
if parseErr == nil && lastID > 0 {
|
||||
missed := s.events.EventsSince(ws.ID, lastID)
|
||||
if missed == nil {
|
||||
// Gap too large — buffer evicted. Tell client to do a full sync.
|
||||
slog.Info("SSE replay gap too large, sending sync_required",
|
||||
"workspace", ws.Slug, "last_event_id", lastID)
|
||||
writeSSEEvent(w, "sync_required", 0, map[string]string{
|
||||
"reason": "Event buffer exceeded. Full sync required.",
|
||||
})
|
||||
flusher.Flush()
|
||||
} else if len(missed) > 0 {
|
||||
slog.Info("SSE replaying missed events",
|
||||
"workspace", ws.Slug, "last_event_id", lastID, "count", len(missed))
|
||||
for _, event := range missed {
|
||||
writeSSEEvent(w, event.Type, event.ID, event)
|
||||
flusher.Flush()
|
||||
}
|
||||
}
|
||||
// If len(missed) == 0: client is caught up, nothing to replay.
|
||||
}
|
||||
}
|
||||
|
||||
// Keepalive ticker
|
||||
keepalive := time.NewTicker(30 * time.Second)
|
||||
defer keepalive.Stop()
|
||||
@@ -91,7 +124,7 @@ func (s *Server) handleSSE(w http.ResponseWriter, r *http.Request) {
|
||||
// Channel closed (unsubscribed)
|
||||
return
|
||||
}
|
||||
writeSSEEvent(w, event.Type, event)
|
||||
writeSSEEvent(w, event.Type, event.ID, event)
|
||||
flusher.Flush()
|
||||
|
||||
case <-keepalive.C:
|
||||
@@ -103,11 +136,16 @@ func (s *Server) handleSSE(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
|
||||
// writeSSEEvent writes a single SSE event to the response writer.
|
||||
func writeSSEEvent(w http.ResponseWriter, eventType string, data interface{}) {
|
||||
// If eventID > 0, an "id:" field is included for Last-Event-ID support.
|
||||
func writeSSEEvent(w http.ResponseWriter, eventType string, eventID int64, data interface{}) {
|
||||
jsonData, err := json.Marshal(data)
|
||||
if err != nil {
|
||||
slog.Error("failed to marshal SSE event", "error", err)
|
||||
return
|
||||
}
|
||||
fmt.Fprintf(w, "event: %s\ndata: %s\n\n", eventType, jsonData)
|
||||
if eventID > 0 {
|
||||
fmt.Fprintf(w, "id: %d\nevent: %s\ndata: %s\n\n", eventID, eventType, jsonData)
|
||||
} else {
|
||||
fmt.Fprintf(w, "event: %s\ndata: %s\n\n", eventType, jsonData)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -415,6 +415,9 @@ func (s *Server) setupRouter() {
|
||||
|
||||
// Dashboard (v2)
|
||||
r.Get("/dashboard", s.handleGetDashboard)
|
||||
|
||||
// Incremental sync — returns items changed since a timestamp
|
||||
r.Get("/changes", s.handleGetChanges)
|
||||
})
|
||||
})
|
||||
|
||||
|
||||
@@ -1605,6 +1605,70 @@ func scanItems(rows *sql.Rows) ([]models.Item, error) {
|
||||
return items, rows.Err()
|
||||
}
|
||||
|
||||
// ItemsModifiedSince returns items in a workspace that were updated after the
|
||||
// given timestamp. Used for incremental sync on tab resume. Also returns IDs of
|
||||
// items that were deleted (hard-deleted or archived) since the timestamp.
|
||||
//
|
||||
// The updated list includes both active AND recently archived items (those with
|
||||
// deleted_at > since). This lets the frontend update archived views correctly —
|
||||
// an item that was just archived needs its full data to appear in archived views,
|
||||
// not just its ID in the deleted list.
|
||||
func (s *Store) ItemsModifiedSince(workspaceID string, since time.Time) (updated []models.Item, deletedIDs []string, err error) {
|
||||
sinceStr := since.UTC().Format(time.RFC3339)
|
||||
|
||||
// Fetch updated items: active items modified since the timestamp,
|
||||
// PLUS items archived since the timestamp (so archived views can update).
|
||||
query := s.q(`
|
||||
SELECT i.id, i.workspace_id, i.collection_id, i.title, i.slug, i.content, i.fields, i.tags,
|
||||
i.pinned, i.sort_order, i.parent_id, i.assigned_user_id, i.agent_role_id, i.role_sort_order,
|
||||
i.created_by, i.last_modified_by, i.source,
|
||||
i.item_number, i.created_at, i.updated_at,
|
||||
c.slug, c.name, c.icon, c.prefix,
|
||||
COALESCE(au.name, ''), COALESCE(au.email, ''),
|
||||
COALESCE(ar.name, ''), COALESCE(ar.slug, ''), COALESCE(ar.icon, '')
|
||||
FROM items i
|
||||
JOIN collections c ON c.id = i.collection_id
|
||||
LEFT JOIN users au ON au.id = i.assigned_user_id
|
||||
LEFT JOIN agent_roles ar ON ar.id = i.agent_role_id
|
||||
WHERE i.workspace_id = ?
|
||||
AND i.updated_at > ?
|
||||
AND (i.deleted_at IS NULL OR i.deleted_at > ?)
|
||||
ORDER BY i.updated_at ASC
|
||||
`)
|
||||
|
||||
rows, err := s.db.Query(query, workspaceID, sinceStr, sinceStr)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
updated, err = scanItems(rows)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
|
||||
// Fetch IDs of items deleted since the timestamp.
|
||||
delQuery := s.q(`
|
||||
SELECT id FROM items
|
||||
WHERE workspace_id = ?
|
||||
AND deleted_at IS NOT NULL
|
||||
AND deleted_at > ?
|
||||
`)
|
||||
delRows, err := s.db.Query(delQuery, workspaceID, sinceStr)
|
||||
if err != nil {
|
||||
return updated, nil, err
|
||||
}
|
||||
defer delRows.Close()
|
||||
for delRows.Next() {
|
||||
var id string
|
||||
if err := delRows.Scan(&id); err != nil {
|
||||
return updated, nil, err
|
||||
}
|
||||
deletedIDs = append(deletedIDs, id)
|
||||
}
|
||||
|
||||
return updated, deletedIDs, delRows.Err()
|
||||
}
|
||||
|
||||
func hydrateItemComputedMetadata(item *models.Item) {
|
||||
if item == nil {
|
||||
return
|
||||
|
||||
@@ -32,7 +32,8 @@ import type {
|
||||
AgentRole,
|
||||
AgentRoleCreate,
|
||||
AgentRoleUpdate,
|
||||
RoleBoardLane
|
||||
RoleBoardLane,
|
||||
ChangesResponse
|
||||
} from '$lib/types';
|
||||
|
||||
const BASE = '/api/v1';
|
||||
@@ -402,6 +403,14 @@ export const api = {
|
||||
request<DashboardResponse>(`/workspaces/${ws}/dashboard`)
|
||||
},
|
||||
|
||||
// ── Incremental Sync ─────────────────────────────────────────────────────
|
||||
|
||||
changes: {
|
||||
/** Fetch items modified since the given timestamp (unix ms). */
|
||||
since: (ws: string, sinceMs: number) =>
|
||||
request<ChangesResponse>(`/workspaces/${ws}/changes?since=${sinceMs}`)
|
||||
},
|
||||
|
||||
// ── Search ────────────────────────────────────────────────────────────────
|
||||
|
||||
search: (query: string, workspace?: string) => {
|
||||
|
||||
@@ -3,7 +3,7 @@
|
||||
import { onDestroy, onMount } from 'svelte';
|
||||
import { api } from '$lib/api/client';
|
||||
import { sseService } from '$lib/services/sse.svelte';
|
||||
import { visibility } from '$lib/services/visibility.svelte';
|
||||
import { syncService } from '$lib/services/sync.svelte';
|
||||
import type { Item } from '$lib/types';
|
||||
import { parseFields, formatItemRef } from '$lib/types';
|
||||
import { dndzone, TRIGGERS, SHADOW_ITEM_MARKER_PROPERTY_NAME } from 'svelte-dnd-action';
|
||||
@@ -29,7 +29,7 @@
|
||||
let loading = $state(true);
|
||||
let error = $state('');
|
||||
let unsubscribeSSE: (() => void) | null = null;
|
||||
let unsubscribeVisibility: (() => void) | null = null;
|
||||
let unsubscribeSync: (() => void) | null = null;
|
||||
|
||||
let expandedIds = $state<Set<string>>(new Set());
|
||||
|
||||
@@ -138,9 +138,12 @@
|
||||
});
|
||||
|
||||
onMount(() => {
|
||||
unsubscribeVisibility = visibility.onTabResume(() => {
|
||||
unsubscribeSync = syncService.onSync((result) => {
|
||||
if (!wsSlug || !itemSlug) return;
|
||||
loadChildren();
|
||||
// Only reload children on actual changes, not when caught up
|
||||
if (result.type !== 'caught_up') {
|
||||
loadChildren();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
@@ -158,7 +161,7 @@
|
||||
|
||||
onDestroy(() => {
|
||||
unsubscribeSSE?.();
|
||||
unsubscribeVisibility?.();
|
||||
unsubscribeSync?.();
|
||||
});
|
||||
|
||||
function formatLabel(value: string): string {
|
||||
|
||||
@@ -4,6 +4,7 @@ export type SSEStatus = 'disconnected' | 'connected' | 'reconnecting';
|
||||
|
||||
export interface ItemEvent {
|
||||
type: string;
|
||||
id?: number;
|
||||
workspace_id: string;
|
||||
item_id: string;
|
||||
title: string;
|
||||
@@ -31,14 +32,29 @@ const ITEM_EVENTS = [
|
||||
function createSSEService() {
|
||||
let status = $state<SSEStatus>('disconnected');
|
||||
let lastEventTime = $state<number>(0);
|
||||
let needsSync = $state<boolean>(false);
|
||||
let eventSource: EventSource | null = null;
|
||||
let currentWorkspace: string = '';
|
||||
const callbacks = new SvelteSet<ItemEventCallback>();
|
||||
|
||||
function connect(workspaceSlug: string) {
|
||||
// If already connected to the same workspace, don't reconnect.
|
||||
// The browser's EventSource handles reconnection automatically
|
||||
// with Last-Event-ID, so destroying it would lose that state.
|
||||
if (eventSource && currentWorkspace === workspaceSlug) {
|
||||
// EventSource is already connected (or auto-reconnecting).
|
||||
// readyState: 0=CONNECTING, 1=OPEN, 2=CLOSED
|
||||
if (eventSource.readyState !== EventSource.CLOSED) {
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
// Different workspace or closed connection — create new EventSource
|
||||
if (eventSource) {
|
||||
disconnect();
|
||||
}
|
||||
|
||||
currentWorkspace = workspaceSlug;
|
||||
const url = `/api/v1/events?workspace=${encodeURIComponent(workspaceSlug)}`;
|
||||
eventSource = new EventSource(url);
|
||||
|
||||
@@ -48,12 +64,25 @@ function createSSEService() {
|
||||
|
||||
eventSource.onerror = () => {
|
||||
status = 'reconnecting';
|
||||
// EventSource auto-reconnects and sends Last-Event-ID.
|
||||
// The server replays missed events from its buffer.
|
||||
};
|
||||
|
||||
eventSource.addEventListener('connected', () => {
|
||||
status = 'connected';
|
||||
});
|
||||
|
||||
// Handle sync_required: server's replay buffer couldn't cover the gap.
|
||||
// Trigger an immediate sync rather than waiting for a visibility change,
|
||||
// so the UI stays fresh even when the tab is actively visible.
|
||||
eventSource.addEventListener('sync_required', () => {
|
||||
needsSync = true;
|
||||
// Dynamic import to avoid circular dependency
|
||||
import('./sync.svelte').then(({ syncService }) => {
|
||||
syncService.triggerSync();
|
||||
});
|
||||
});
|
||||
|
||||
for (const eventType of ITEM_EVENTS) {
|
||||
eventSource.addEventListener(eventType, (e: MessageEvent) => {
|
||||
const data: ItemEvent = JSON.parse(e.data);
|
||||
@@ -70,9 +99,19 @@ function createSSEService() {
|
||||
eventSource.close();
|
||||
eventSource = null;
|
||||
}
|
||||
currentWorkspace = '';
|
||||
status = 'disconnected';
|
||||
}
|
||||
|
||||
/** Force reconnect (e.g., after auth change). */
|
||||
function reconnect() {
|
||||
const ws = currentWorkspace;
|
||||
if (ws) {
|
||||
disconnect();
|
||||
connect(ws);
|
||||
}
|
||||
}
|
||||
|
||||
function onItemEvent(callback: ItemEventCallback): () => void {
|
||||
callbacks.add(callback);
|
||||
return () => {
|
||||
@@ -80,6 +119,10 @@ function createSSEService() {
|
||||
};
|
||||
}
|
||||
|
||||
function clearSyncFlag() {
|
||||
needsSync = false;
|
||||
}
|
||||
|
||||
return {
|
||||
get status() {
|
||||
return status;
|
||||
@@ -87,9 +130,14 @@ function createSSEService() {
|
||||
get lastEventTime() {
|
||||
return lastEventTime;
|
||||
},
|
||||
get needsSync() {
|
||||
return needsSync;
|
||||
},
|
||||
connect,
|
||||
disconnect,
|
||||
onItemEvent
|
||||
reconnect,
|
||||
onItemEvent,
|
||||
clearSyncFlag
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,214 @@
|
||||
/**
|
||||
* Sync coordinator — centralizes tab-resume data synchronization.
|
||||
*
|
||||
* Instead of every page/component independently refetching everything on
|
||||
* visibilitychange, this service:
|
||||
*
|
||||
* 1. Tracks the last successful sync timestamp
|
||||
* 2. On tab resume, checks SSE health first
|
||||
* 3. If SSE replayed missed events: no action needed (already caught up)
|
||||
* 4. If SSE signals sync_required: uses the /changes endpoint for a delta sync
|
||||
* 5. Notifies registered page-level callbacks with the sync result
|
||||
* 6. Only does a full refetch as a last resort (long absence, errors)
|
||||
*
|
||||
* Pages register lightweight callbacks that receive the sync result and can
|
||||
* update their local state accordingly — no more blind full refetches.
|
||||
*/
|
||||
|
||||
import { api } from '$lib/api/client';
|
||||
import { sseService } from '$lib/services/sse.svelte';
|
||||
import type { Item, ChangesResponse } from '$lib/types';
|
||||
|
||||
export type SyncResult = {
|
||||
type: 'caught_up'; // SSE was healthy, nothing missed
|
||||
} | {
|
||||
type: 'incremental'; // Delta sync via /changes
|
||||
changes: ChangesResponse;
|
||||
} | {
|
||||
type: 'full_refresh'; // Gap too large or error — caller should reload everything
|
||||
};
|
||||
|
||||
type SyncCallback = (result: SyncResult) => void;
|
||||
|
||||
/**
|
||||
* How long the tab must have been hidden before we bother syncing at all.
|
||||
* Short absences (< 2s) almost certainly had no changes.
|
||||
*/
|
||||
const MIN_ABSENCE_MS = 2000;
|
||||
|
||||
/**
|
||||
* If the tab has been hidden longer than this, skip incremental sync
|
||||
* and go straight to full refresh. The /changes endpoint may return
|
||||
* too much data for very long absences.
|
||||
*/
|
||||
const MAX_INCREMENTAL_MS = 10 * 60 * 1000; // 10 minutes
|
||||
|
||||
function createSyncService() {
|
||||
let lastSyncTime = $state<number>(Date.now());
|
||||
let hiddenSince = $state<number>(0);
|
||||
let syncing = $state<boolean>(false);
|
||||
let wsSlug = $state<string>('');
|
||||
let initialized = false;
|
||||
|
||||
const callbacks = new Set<SyncCallback>();
|
||||
|
||||
// Track when the tab was hidden/shown
|
||||
function init() {
|
||||
if (initialized || typeof document === 'undefined') return;
|
||||
initialized = true;
|
||||
|
||||
document.addEventListener('visibilitychange', () => {
|
||||
if (document.hidden) {
|
||||
hiddenSince = Date.now();
|
||||
} else {
|
||||
onTabResume();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
async function setWorkspace(slug: string) {
|
||||
wsSlug = slug;
|
||||
// Seed the sync cursor from the server's clock, not the client's.
|
||||
// This avoids clock-skew issues where Date.now() on the client
|
||||
// is ahead/behind the server, causing missed or duplicate changes.
|
||||
try {
|
||||
const changes = await api.changes.since(slug, Date.now());
|
||||
lastSyncTime = changes.server_time;
|
||||
} catch {
|
||||
// Fallback to client time if the server call fails.
|
||||
// Not ideal, but better than leaving the cursor at 0.
|
||||
lastSyncTime = Date.now();
|
||||
}
|
||||
}
|
||||
|
||||
/** Called when the tab becomes visible again. */
|
||||
async function onTabResume() {
|
||||
if (syncing || !wsSlug) return;
|
||||
|
||||
const absence = hiddenSince > 0 ? Date.now() - hiddenSince : 0;
|
||||
hiddenSince = 0;
|
||||
|
||||
// Very short absence — SSE almost certainly kept up, skip sync
|
||||
if (absence < MIN_ABSENCE_MS) return;
|
||||
|
||||
syncing = true;
|
||||
try {
|
||||
const result = await determineSync(absence);
|
||||
// Only advance the cursor for incremental syncs (we know exactly
|
||||
// what the server returned). For full_refresh, DON'T advance here —
|
||||
// the cursor stays put until a page callback successfully reloads
|
||||
// and calls markSynced(). This prevents data loss if the reload fails.
|
||||
if (result.type === 'incremental') {
|
||||
lastSyncTime = result.changes.server_time;
|
||||
}
|
||||
// For 'caught_up': cursor stays as-is (nothing was missed).
|
||||
// For 'full_refresh': cursor stays as-is until markSynced() is called.
|
||||
notify(result);
|
||||
} catch {
|
||||
// On error, tell pages to do a full refresh as a safe fallback.
|
||||
// Don't advance cursor — retry on next tab resume.
|
||||
notify({ type: 'full_refresh' });
|
||||
} finally {
|
||||
syncing = false;
|
||||
}
|
||||
}
|
||||
|
||||
async function determineSync(absenceMs: number): Promise<SyncResult> {
|
||||
// If SSE says it needs a full sync (buffer overflow), respect that
|
||||
if (sseService.needsSync) {
|
||||
sseService.clearSyncFlag();
|
||||
return doIncrementalOrFull(absenceMs);
|
||||
}
|
||||
|
||||
// If SSE is connected and the absence was short enough that the
|
||||
// replay buffer should have covered it, we're caught up.
|
||||
if (sseService.status === 'connected' && absenceMs < MAX_INCREMENTAL_MS) {
|
||||
// SSE EventSource auto-reconnects with Last-Event-ID.
|
||||
// If the server replayed events, the SSE callbacks already
|
||||
// updated the store. Check if SSE received events recently.
|
||||
const timeSinceLastEvent = Date.now() - sseService.lastEventTime;
|
||||
|
||||
// If SSE got events recently (within the absence window), it
|
||||
// likely replayed everything we missed.
|
||||
if (sseService.lastEventTime > 0 && timeSinceLastEvent < absenceMs + 5000) {
|
||||
return { type: 'caught_up' };
|
||||
}
|
||||
}
|
||||
|
||||
return doIncrementalOrFull(absenceMs);
|
||||
}
|
||||
|
||||
async function doIncrementalOrFull(absenceMs: number): Promise<SyncResult> {
|
||||
// Very long absence — skip incremental, do full refresh
|
||||
if (absenceMs > MAX_INCREMENTAL_MS) {
|
||||
return { type: 'full_refresh' };
|
||||
}
|
||||
|
||||
// Try incremental sync via /changes endpoint
|
||||
try {
|
||||
const changes = await api.changes.since(wsSlug, lastSyncTime);
|
||||
if (changes.updated.length === 0 && changes.deleted.length === 0) {
|
||||
return { type: 'caught_up' };
|
||||
}
|
||||
return { type: 'incremental', changes };
|
||||
} catch {
|
||||
// /changes failed — fall back to full refresh
|
||||
return { type: 'full_refresh' };
|
||||
}
|
||||
}
|
||||
|
||||
function onSync(cb: SyncCallback): () => void {
|
||||
callbacks.add(cb);
|
||||
return () => { callbacks.delete(cb); };
|
||||
}
|
||||
|
||||
function notify(result: SyncResult) {
|
||||
for (const cb of callbacks) {
|
||||
try {
|
||||
cb(result);
|
||||
} catch {
|
||||
// Don't let one failing callback break others
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** Mark a successful data load (updates the sync timestamp). */
|
||||
function markSynced() {
|
||||
lastSyncTime = Date.now();
|
||||
}
|
||||
|
||||
/**
|
||||
* Trigger a sync immediately (e.g., when SSE sends sync_required
|
||||
* while the tab is still visible). This bypasses the visibility
|
||||
* change listener and runs the sync directly.
|
||||
*/
|
||||
async function triggerSync() {
|
||||
if (syncing || !wsSlug) return;
|
||||
syncing = true;
|
||||
try {
|
||||
// SSE told us there's a gap — try incremental, fall back to full
|
||||
const result = await doIncrementalOrFull(MAX_INCREMENTAL_MS);
|
||||
if (result.type === 'incremental') {
|
||||
lastSyncTime = result.changes.server_time;
|
||||
}
|
||||
// For full_refresh: don't advance cursor until pages confirm success.
|
||||
notify(result);
|
||||
} catch {
|
||||
notify({ type: 'full_refresh' });
|
||||
} finally {
|
||||
syncing = false;
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
get syncing() { return syncing; },
|
||||
get lastSyncTime() { return lastSyncTime; },
|
||||
init,
|
||||
setWorkspace,
|
||||
onSync,
|
||||
markSynced,
|
||||
triggerSync
|
||||
};
|
||||
}
|
||||
|
||||
export const syncService = createSyncService();
|
||||
@@ -547,6 +547,15 @@ export interface DashboardResponse {
|
||||
}[];
|
||||
}
|
||||
|
||||
// ─── Incremental Sync ────────────────────────────────────────────────────────
|
||||
|
||||
export interface ChangesResponse {
|
||||
updated: Item[];
|
||||
deleted: string[];
|
||||
server_time: number;
|
||||
collections_changed: boolean;
|
||||
}
|
||||
|
||||
// ─── Search ──────────────────────────────────────────────────────────────────
|
||||
|
||||
export interface SearchResult {
|
||||
|
||||
@@ -5,7 +5,7 @@
|
||||
import { collectionStore } from '$lib/stores/collections.svelte';
|
||||
import { editorStore } from '$lib/stores/editor.svelte';
|
||||
import { sseService } from '$lib/services/sse.svelte';
|
||||
import { visibility } from '$lib/services/visibility.svelte';
|
||||
import { syncService } from '$lib/services/sync.svelte';
|
||||
import { api } from '$lib/api/client';
|
||||
import { toastStore } from '$lib/stores/toast.svelte';
|
||||
|
||||
@@ -13,23 +13,26 @@
|
||||
|
||||
let wsSlug = $derived(page.params.workspace ?? '');
|
||||
let unsubscribeSSE: (() => void) | null = null;
|
||||
let unsubscribeVisibility: (() => void) | null = null;
|
||||
let unsubscribeSync: (() => void) | null = null;
|
||||
|
||||
onMount(() => {
|
||||
visibility.init();
|
||||
unsubscribeVisibility = visibility.onTabResume(() => {
|
||||
// Initialize the sync coordinator (sets up visibilitychange listener once)
|
||||
syncService.init();
|
||||
|
||||
// Listen for sync results to refresh collection metadata
|
||||
unsubscribeSync = syncService.onSync((result) => {
|
||||
if (!wsSlug) return;
|
||||
// Reconnect SSE — events may have been lost while the tab was hidden
|
||||
sseService.connect(wsSlug);
|
||||
// Refresh collection metadata (counts, etc.)
|
||||
collectionStore.loadCollections(wsSlug);
|
||||
if (result.type === 'full_refresh' || (result.type === 'incremental' && result.changes.collections_changed)) {
|
||||
collectionStore.loadCollections(wsSlug);
|
||||
}
|
||||
});
|
||||
|
||||
connectSSE();
|
||||
});
|
||||
|
||||
onDestroy(() => {
|
||||
unsubscribeSSE?.();
|
||||
unsubscribeVisibility?.();
|
||||
unsubscribeSync?.();
|
||||
sseService.disconnect();
|
||||
});
|
||||
|
||||
@@ -38,6 +41,7 @@
|
||||
if (wsSlug) {
|
||||
workspaceStore.setCurrent(wsSlug);
|
||||
collectionStore.loadCollections(wsSlug);
|
||||
syncService.setWorkspace(wsSlug);
|
||||
connectSSE();
|
||||
}
|
||||
});
|
||||
|
||||
@@ -5,7 +5,7 @@
|
||||
import { browser } from '$app/environment';
|
||||
import { api } from '$lib/api/client';
|
||||
import { workspaceStore } from '$lib/stores/workspace.svelte';
|
||||
import { visibility } from '$lib/services/visibility.svelte';
|
||||
import { syncService } from '$lib/services/sync.svelte';
|
||||
import { relativeTime } from '$lib/utils/markdown';
|
||||
import { itemUrlId } from '$lib/types';
|
||||
import OnboardingChecklist from '$lib/components/OnboardingChecklist.svelte';
|
||||
@@ -40,20 +40,24 @@
|
||||
if (wsSlug) load(wsSlug);
|
||||
});
|
||||
|
||||
let unsubscribeVisibility: (() => void) | null = null;
|
||||
let unsubscribeSync: (() => void) | null = null;
|
||||
|
||||
onMount(() => {
|
||||
pollTimer = setInterval(() => {
|
||||
if (wsSlug) load(wsSlug, true);
|
||||
}, 30000);
|
||||
unsubscribeVisibility = visibility.onTabResume(() => {
|
||||
if (wsSlug) load(wsSlug, true);
|
||||
// Dashboard always does a full reload on any sync signal since it's
|
||||
// an aggregated view (counts, activity, suggestions change with any item update)
|
||||
unsubscribeSync = syncService.onSync((result) => {
|
||||
if (result.type !== 'caught_up' && wsSlug) {
|
||||
load(wsSlug, true);
|
||||
}
|
||||
});
|
||||
return () => clearInterval(pollTimer);
|
||||
});
|
||||
|
||||
onDestroy(() => {
|
||||
unsubscribeVisibility?.();
|
||||
unsubscribeSync?.();
|
||||
});
|
||||
|
||||
async function load(slug: string, silent = false) {
|
||||
|
||||
@@ -11,7 +11,7 @@
|
||||
import QuickActionsMenu from '$lib/components/common/QuickActionsMenu.svelte';
|
||||
import { onDestroy, onMount } from 'svelte';
|
||||
import { sseService } from '$lib/services/sse.svelte';
|
||||
import { visibility } from '$lib/services/visibility.svelte';
|
||||
import { syncService } from '$lib/services/sync.svelte';
|
||||
import { toastStore } from '$lib/stores/toast.svelte';
|
||||
|
||||
type ViewMode = 'list' | 'board' | 'table';
|
||||
@@ -129,36 +129,60 @@
|
||||
});
|
||||
});
|
||||
|
||||
// Silently refresh items when the tab regains focus (SSE events may have been lost)
|
||||
let unsubscribeVisibility: (() => void) | null = null;
|
||||
// Sync coordinator — handle tab-resume data refresh efficiently
|
||||
let unsubscribeSync: (() => void) | null = null;
|
||||
|
||||
onMount(() => {
|
||||
unsubscribeVisibility = visibility.onTabResume(async () => {
|
||||
unsubscribeSync = syncService.onSync(async (result) => {
|
||||
if (!wsSlug || !collSlug) return;
|
||||
|
||||
if (result.type === 'caught_up') return;
|
||||
|
||||
if (result.type === 'incremental') {
|
||||
// Apply incremental changes to this collection's item list
|
||||
const changes = result.changes;
|
||||
let changed = false;
|
||||
|
||||
for (const updated of changes.updated) {
|
||||
const existingIdx = items.findIndex(i => i.id === updated.id);
|
||||
|
||||
if (updated.collection_slug === collSlug) {
|
||||
// Item belongs to this collection — update or add
|
||||
changed = true;
|
||||
if (existingIdx >= 0) {
|
||||
items[existingIdx] = updated;
|
||||
} else {
|
||||
items = [...items, updated];
|
||||
}
|
||||
} else if (existingIdx >= 0) {
|
||||
// Item was moved OUT of this collection — remove it
|
||||
changed = true;
|
||||
items = items.filter(i => i.id !== updated.id);
|
||||
}
|
||||
}
|
||||
|
||||
for (const deletedId of changes.deleted) {
|
||||
const idx = items.findIndex(i => i.id === deletedId);
|
||||
if (idx >= 0) {
|
||||
changed = true;
|
||||
items = items.filter(i => i.id !== deletedId);
|
||||
}
|
||||
}
|
||||
|
||||
// Refresh progress data if anything changed
|
||||
if (changed) {
|
||||
await refreshProgress(wsSlug, collSlug, items);
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
// Full refresh fallback
|
||||
try {
|
||||
const listParams = showArchived ? { include_archived: true } : undefined;
|
||||
const freshItems = await api.items.listByCollection(wsSlug, collSlug, listParams);
|
||||
items = freshItems;
|
||||
|
||||
// Update progress data without resetting view state
|
||||
if (collSlug === 'plans') {
|
||||
const progress = await api.items.plansProgress(wsSlug).catch(() => []);
|
||||
const map: Record<string, { total: number; done: number }> = {};
|
||||
for (const p of progress) {
|
||||
map[p.item_id] = { total: p.total, done: p.done };
|
||||
}
|
||||
itemProgress = map;
|
||||
} else {
|
||||
const map: Record<string, { total: number; done: number }> = {};
|
||||
for (const it of freshItems) {
|
||||
if (!it.content) continue;
|
||||
const total = (it.content.match(/- \[[ x]\]/g) ?? []).length;
|
||||
if (total === 0) continue;
|
||||
const done = (it.content.match(/- \[x\]/g) ?? []).length;
|
||||
map[it.id] = { total, done };
|
||||
}
|
||||
itemProgress = map;
|
||||
}
|
||||
await refreshProgress(wsSlug, collSlug, freshItems);
|
||||
syncService.markSynced(); // Advance cursor now that reload succeeded
|
||||
} catch {
|
||||
// Ignore — will catch up on next SSE event
|
||||
}
|
||||
@@ -167,9 +191,30 @@
|
||||
|
||||
onDestroy(() => {
|
||||
unsubscribeSSE?.();
|
||||
unsubscribeVisibility?.();
|
||||
unsubscribeSync?.();
|
||||
});
|
||||
|
||||
async function refreshProgress(ws: string, coll: string, itemList: typeof items) {
|
||||
if (coll === 'plans') {
|
||||
const progress = await api.items.plansProgress(ws).catch(() => []);
|
||||
const map: Record<string, { total: number; done: number }> = {};
|
||||
for (const p of progress) {
|
||||
map[p.item_id] = { total: p.total, done: p.done };
|
||||
}
|
||||
itemProgress = map;
|
||||
} else {
|
||||
const map: Record<string, { total: number; done: number }> = {};
|
||||
for (const it of itemList) {
|
||||
if (!it.content) continue;
|
||||
const total = (it.content.match(/- \[[ x]\]/g) ?? []).length;
|
||||
if (total === 0) continue;
|
||||
const done = (it.content.match(/- \[x\]/g) ?? []).length;
|
||||
map[it.id] = { total, done };
|
||||
}
|
||||
itemProgress = map;
|
||||
}
|
||||
}
|
||||
|
||||
async function loadCollection(ws: string, coll: string, includeArchived = false) {
|
||||
loading = true;
|
||||
try {
|
||||
|
||||
@@ -3,7 +3,7 @@
|
||||
import { tick, onMount, onDestroy } from 'svelte';
|
||||
import { api } from '$lib/api/client';
|
||||
import { collectionStore } from '$lib/stores/collections.svelte';
|
||||
import { visibility } from '$lib/services/visibility.svelte';
|
||||
import { syncService } from '$lib/services/sync.svelte';
|
||||
import Editor from '$lib/components/editor/Editor.svelte';
|
||||
import EditorBubbleMenu from '$lib/components/editor/EditorBubbleMenu.svelte';
|
||||
import EditorLinkPopover from '$lib/components/editor/EditorLinkPopover.svelte';
|
||||
@@ -87,24 +87,45 @@
|
||||
}
|
||||
});
|
||||
|
||||
// Refresh item when the tab regains focus (SSE events may have been lost)
|
||||
let unsubscribeVisibility: (() => void) | null = null;
|
||||
// Sync coordinator — refresh item data on tab resume
|
||||
let unsubscribeSync: (() => void) | null = null;
|
||||
|
||||
onMount(() => {
|
||||
unsubscribeVisibility = visibility.onTabResume(async () => {
|
||||
unsubscribeSync = syncService.onSync(async (result) => {
|
||||
if (!wsSlug || !itemSlug || !item) return;
|
||||
// Don't refresh if the user is actively editing
|
||||
if (saveStatus === 'saving' || editingTitle) return;
|
||||
|
||||
if (result.type === 'caught_up') return;
|
||||
|
||||
if (result.type === 'incremental') {
|
||||
// Check if our item is in the changed set
|
||||
const updated = result.changes.updated.find(i => i.id === item!.id);
|
||||
if (updated) {
|
||||
// Merge server state without disrupting the editor
|
||||
item = {
|
||||
...updated,
|
||||
content: item!.content
|
||||
};
|
||||
itemLinks = await api.links.list(wsSlug, updated.slug).catch(() => []);
|
||||
}
|
||||
// Check if our item was deleted
|
||||
if (result.changes.deleted.includes(item!.id)) {
|
||||
// Item was deleted — navigate back to collection
|
||||
goto(`/${wsSlug}/${collSlug}`);
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
// Full refresh fallback
|
||||
try {
|
||||
const updated = await api.items.get(wsSlug, itemSlug);
|
||||
// Merge server state without disrupting the editor:
|
||||
// update fields/metadata but preserve local content to avoid resetting the editor
|
||||
item = {
|
||||
...updated,
|
||||
content: item!.content
|
||||
};
|
||||
// Refresh links too
|
||||
itemLinks = await api.links.list(wsSlug, updated.slug).catch(() => []);
|
||||
syncService.markSynced(); // Advance cursor now that reload succeeded
|
||||
} catch {
|
||||
// Ignore — will catch up on next event
|
||||
}
|
||||
@@ -112,7 +133,7 @@
|
||||
});
|
||||
|
||||
onDestroy(() => {
|
||||
unsubscribeVisibility?.();
|
||||
unsubscribeSync?.();
|
||||
});
|
||||
|
||||
async function loadData() {
|
||||
|
||||
Reference in New Issue
Block a user