types.go
1package sync
2
3import (
4 "context"
5 "fmt"
6 "sync"
7 "time"
8
9 "github.com/coder/websocket"
10)
11
12// DisconnectReason indicates why the sync client disconnected.
13type DisconnectReason int
14
15const (
16 // DisconnectReasonClientInitiated indicates the context was cancelled
17 // or the client was explicitly closed by the user.
18 DisconnectReasonClientInitiated DisconnectReason = iota
19
20 // DisconnectReasonYSweetInitiated indicates the y-sweet server closed
21 // the connection from its end.
22 DisconnectReasonYSweetInitiated
23
24 // DisconnectReasonDocumentIdle indicates the client disconnected because
25 // no other clients were connected and the idle timeout expired.
26 DisconnectReasonDocumentIdle
27)
28
29// String returns a human-readable description of the disconnect reason.
30func (r DisconnectReason) String() string {
31 switch r {
32 case DisconnectReasonClientInitiated:
33 return "ClientInitiated"
34 case DisconnectReasonYSweetInitiated:
35 return "YSweetInitiated"
36 case DisconnectReasonDocumentIdle:
37 return "DocumentIdle"
38 default:
39 return "Unknown"
40 }
41}
42
43// Status represents the connection state of the sync client.
44type Status int
45
46const (
47 // StatusDisconnected means no active connection to the server.
48 StatusDisconnected Status = iota
49 // StatusConnecting means the client is attempting to connect.
50 StatusConnecting
51 // StatusConnected means an active connection is established.
52 StatusConnected
53 // StatusDisconnecting means the client is shutting down gracefully.
54 StatusDisconnecting
55)
56
57// UpdateData holds raw binary update data from the Yjs sync protocol.
58type UpdateData []byte
59
60// Transaction defines the methods a transaction must provide for sync.
61type Transaction interface {
62 ApplyUpdate(data []byte) error
63 GetStateVector() []byte
64}
65
66// DocInterface defines the methods a document must provide for sync.
67// This allows the sync client to work with any implementation that matches this interface.
68type DocInterface interface {
69 // WithReadTransaction executes a callback within a read-only transaction.
70 WithReadTransaction(fn func(Transaction) error) error
71 // WithWriteTransaction executes a callback within a read-write transaction.
72 // The transaction parameter can apply updates using raw byte data.
73 WithWriteTransaction(fn func(Transaction) error) error
74 // GetStateDiff returns an update containing all changes since the given state vector.
75 GetStateDiff(stateVector []byte) []byte
76}
77
78// SyncClient manages a WebSocket connection to a y-sweet server.
79// It handles document synchronization, reconnection, and awareness tracking.
80type SyncClient struct {
81 endpoint string
82 authToken string
83 doc DocInterface
84 conn *websocket.Conn
85 status Status
86 onUpdate func(doc DocInterface) error
87 ackedVersion int64
88 localVersion int64
89 retries int
90 awareness *AwarenessClient
91 closeOnce sync.Once
92 done chan struct{}
93 disconnect func()
94 mu sync.RWMutex // protects status and syncConn
95
96 // Internal sync connection for actual WebSocket handling
97 syncConn *SyncConn
98}
99
100// Done returns a channel that is closed when the client is stopped.
101func (c *SyncClient) Done() <-chan struct{} {
102 return c.done
103}
104
105// Status returns the current connection state of the client.
106func (c *SyncClient) Status() Status {
107 c.mu.RLock()
108 defer c.mu.RUnlock()
109 return c.status
110}
111
112// NewSyncClient creates a new sync client for the document.
113// The client must be started with Start() to begin synchronization.
114func NewSyncClient(doc DocInterface, opts ...Option) *SyncClient {
115 options := defaultOptions()
116 for _, opt := range opts {
117 opt(&options)
118 }
119 disconnect := func() {}
120
121 client := &SyncClient{
122 endpoint: options.Endpoint,
123 authToken: options.AuthToken,
124 doc: doc,
125 status: StatusDisconnected,
126 onUpdate: options.OnUpdate,
127 ackedVersion: -1,
128 localVersion: 0,
129 retries: 0,
130 done: make(chan struct{}),
131 disconnect: disconnect,
132 }
133
134 // Create internal SyncConn for WebSocket handling
135 syncOpts := Options{
136 Endpoint: options.Endpoint,
137 AuthToken: options.AuthToken,
138 OnUpdate: options.OnUpdate,
139 AwarenessTimeout: options.AwarenessTimeout,
140 AwarenessState: options.AwarenessState,
141 OnConnect: options.OnConnect,
142 OnDisconnect: options.OnDisconnect,
143 }
144 client.syncConn = NewSyncConn(doc, syncOpts)
145
146 // Default to 5 minutes if not set
147 timeout := options.AwarenessTimeout
148 if timeout <= 0 {
149 timeout = 5 * time.Minute
150 }
151
152 // Set up the disconnect function to trigger disconnect with DocumentIdle reason
153 client.disconnect = func() {
154 if client.syncConn != nil {
155 client.syncConn.SetDisconnectReason(DisconnectReasonDocumentIdle)
156 // Trigger the disconnect callback directly with DocumentIdle reason
157 // Note: onDisconnect is the field, not opts.OnDisconnect
158 client.syncConn.statsMu.RLock()
159 onDisconnect := client.syncConn.onDisconnect
160 client.syncConn.statsMu.RUnlock()
161 if onDisconnect != nil {
162 onDisconnect(DisconnectReasonDocumentIdle)
163 }
164 client.syncConn.Close()
165 }
166 }
167
168 client.awareness = NewAwarenessClient(0, timeout, client.disconnect)
169
170 // Wire up awareness updates from SyncConn to the awareness client
171 client.syncConn.SetAwarenessUpdate(func(data []byte) error {
172 return client.awareness.HandleUpdate(data)
173 })
174
175 return client
176}
177
178// Option configures a SyncClient when passed to NewSyncClient.
179type Option func(*Options)
180
181// Options holds the configuration for a SyncClient.
182type Options struct {
183 Endpoint string
184 AuthToken string
185 OnUpdate func(doc DocInterface) error
186 AwarenessTimeout time.Duration
187 AwarenessState []byte
188 OnConnect func() // called when connected
189 OnDisconnect func(DisconnectReason) // called when disconnected with reason
190}
191
192// defaultOptions returns the default configuration options.
193func defaultOptions() Options {
194 return Options{}
195}
196
197// DefaultOptions returns the default configuration options.
198// This is useful for testing and for creating a base options struct.
199func DefaultOptions() Options {
200 return Options{}
201}
202
203// WithEndpoint sets the WebSocket endpoint URL for the sync client.
204// This is the address of the y-sweet server to connect to.
205func WithEndpoint(url string) Option {
206 return func(o *Options) {
207 o.Endpoint = url
208 }
209}
210
211// WithAuthToken sets the authentication token for the sync client.
212// This token is sent as the Authorization header when connecting.
213func WithAuthToken(token string) Option {
214 return func(o *Options) {
215 o.AuthToken = token
216 }
217}
218
219// WithOnUpdate sets a callback function that is called when the document
220// receives updates from the server. The callback receives the document.
221func WithOnUpdate(fn func(doc DocInterface) error) Option {
222 return func(o *Options) {
223 o.OnUpdate = fn
224 }
225}
226
227// WithAwarenessTimeout sets the timeout for disconnecting when no other
228// clients are visible via awareness. Default is 5 minutes.
229func WithAwarenessTimeout(timeout time.Duration) Option {
230 return func(o *Options) {
231 o.AwarenessTimeout = timeout
232 }
233}
234
235// WithAwarenessState sets the initial awareness state to broadcast.
236func WithAwarenessState(state []byte) Option {
237 return func(o *Options) {
238 o.AwarenessState = state
239 }
240}
241
242// WithOnConnect sets a callback function that is called when the client
243// successfully connects to the server.
244func WithOnConnect(fn func()) Option {
245 return func(o *Options) {
246 o.OnConnect = fn
247 }
248}
249
250// WithOnDisconnect sets a callback function that is called when the client
251// disconnects from the server. The callback receives the disconnect reason.
252func WithOnDisconnect(fn func(DisconnectReason)) Option {
253 return func(o *Options) {
254 o.OnDisconnect = fn
255 }
256}
257
258// SendUpdate sends a document update to the server.
259// It returns nil if not connected, so callers don't need to check status.
260func (c *SyncClient) SendUpdate(update UpdateData) error {
261 c.mu.RLock()
262 status := c.status
263 syncConn := c.syncConn
264 c.mu.RUnlock()
265
266 if status != StatusConnected || syncConn == nil {
267 return nil
268 }
269
270 // Access the connection through syncConn
271 conn := syncConn.GetConn()
272 if conn == nil {
273 return nil
274 }
275
276 msg := encodeSyncMessage(Update, update)
277 return conn.Write(context.Background(), websocket.MessageBinary, msg)
278}
279
280// Awareness returns the awareness client for this sync client.
281// The awareness client tracks other connected clients.
282func (c *SyncClient) Awareness() *AwarenessClient {
283 return c.awareness
284}
285
286func (c *SyncClient) SetDisconnectFn(fn func()) {
287 c.disconnect = fn
288 if c.awareness != nil {
289 c.awareness = NewAwarenessClient(0, 5*time.Minute, fn)
290 }
291}
292
293// SetAuthToken sets the authentication token after construction.
294func (c *SyncClient) SetAuthToken(token string) {
295 c.authToken = token
296}
297
298// SetOnUpdate sets the callback function that is called when the document
299// receives updates from the server.
300func (c *SyncClient) SetOnUpdate(fn func(DocInterface) error) {
301 c.mu.Lock()
302 c.onUpdate = fn
303 syncConn := c.syncConn
304 c.mu.Unlock()
305
306 if syncConn != nil {
307 syncConn.SetOnUpdate(fn)
308 }
309}
310
311// Start connects to the y-sweet server and begins synchronization.
312func (c *SyncClient) Start(ctx context.Context) error {
313 c.mu.RLock()
314 syncConn := c.syncConn
315 c.mu.RUnlock()
316
317 if syncConn == nil {
318 return fmt.Errorf("sync connection not initialized")
319 }
320
321 // Start the sync connection
322 err := syncConn.Start(ctx)
323
324 c.mu.Lock()
325 if err != nil {
326 c.status = StatusDisconnected
327 } else {
328 c.status = StatusConnected
329 }
330 c.mu.Unlock()
331
332 return err
333}
334
335// Close gracefully shuts down the sync connection.
336func (c *SyncClient) Close() error {
337 c.mu.Lock()
338 syncConn := c.syncConn
339 c.status = StatusDisconnected
340 c.mu.Unlock()
341
342 if syncConn != nil {
343 return syncConn.Close()
344 }
345 return nil
346}
347
348// SendPendingAndClose sends any pending updates then closes the connection.
349// This is safe to call from within the OnUpdate callback.
350func (c *SyncClient) SendPendingAndClose() error {
351 c.mu.RLock()
352 syncConn := c.syncConn
353 c.mu.RUnlock()
354
355 if syncConn == nil {
356 return nil
357 }
358 return syncConn.SendPendingAndClose()
359}
360
361// GetStats returns current connection statistics from the underlying sync connection.
362func (c *SyncClient) GetStats() Stats {
363 c.mu.RLock()
364 syncConn := c.syncConn
365 status := c.status
366 c.mu.RUnlock()
367
368 if syncConn == nil {
369 return Stats{
370 Status: status,
371 }
372 }
373 return syncConn.GetStats()
374}
375
376// Flush performs a best-effort sync of pending updates.
377func (c *SyncClient) Flush(ctx context.Context) error {
378 c.mu.RLock()
379 syncConn := c.syncConn
380 c.mu.RUnlock()
381
382 if syncConn == nil {
383 return fmt.Errorf("not connected")
384 }
385 return syncConn.Flush(ctx)
386}
387
388// GetSyncConn returns the underlying SyncConn for advanced use.
389// This allows access to lower-level connection methods.
390func (c *SyncClient) GetSyncConn() *SyncConn {
391 c.mu.RLock()
392 defer c.mu.RUnlock()
393 return c.syncConn
394}
395
396// SetDisconnectCallback sets a callback that is called when a disconnect
397// occurs with the reason for the disconnect.
398func (c *SyncClient) SetDisconnectCallback(fn func(DisconnectReason)) {
399 c.mu.Lock()
400 defer c.mu.Unlock()
401
402 // Wrap the callback to set the reason before calling
403 c.syncConn.statsMu.Lock()
404 originalOnDisconnect := c.syncConn.onDisconnect
405 c.syncConn.onDisconnect = func(reason DisconnectReason) {
406 if fn != nil {
407 fn(reason)
408 }
409 if originalOnDisconnect != nil {
410 originalOnDisconnect(reason)
411 }
412 }
413 c.syncConn.statsMu.Unlock()
414}
