10 files changed,
+392,
-31
+4,
-2
1@@ -1,7 +1,9 @@
2-module load_example
3+module sync_example
4
5-go 1.22
6+go 1.23
7
8 replace github.com/BTBurke/ygo => ../
9
10 require github.com/BTBurke/ygo v0.0.0
11+
12+require github.com/coder/websocket v1.8.14 // indirect
+2,
-0
1@@ -0,0 +1,2 @@
2+github.com/coder/websocket v1.8.14 h1:9L0p0iKiNOibykf283eHkKUHHrpG7f65OE3BhhO7v9g=
3+github.com/coder/websocket v1.8.14/go.mod h1:NX3SzP+inril6yawo5CQXx8+fk145lPDC6pumgx0mVg=
+123,
-0
1@@ -0,0 +1,123 @@
2+package main
3+
4+import (
5+ "context"
6+ "fmt"
7+ "log"
8+ "os"
9+ "os/signal"
10+ "syscall"
11+
12+ ygo "github.com/BTBurke/ygo"
13+ "github.com/BTBurke/ygo/ysweet"
14+)
15+
16+func main() {
17+ ctx, cancel := context.WithCancel(context.Background())
18+ defer cancel()
19+
20+ sigCh := make(chan os.Signal, 1)
21+ signal.Notify(sigCh, os.Interrupt, syscall.SIGINT)
22+ go func() {
23+ <-sigCh
24+ fmt.Println("\nShutting down...")
25+ cancel()
26+ }()
27+
28+ client, err := ysweet.NewClient("http://127.0.0.1:8080")
29+ if err != nil {
30+ log.Fatalf("Failed to create y-sweet client: %v", err)
31+ }
32+
33+ docID, err := client.NewDoc("")
34+ if err != nil {
35+ log.Fatalf("Failed to create document: %v", err)
36+ }
37+ fmt.Printf("Created document with ID: %s\n", docID)
38+
39+ auth, err := client.AuthDoc(docID)
40+ if err != nil {
41+ log.Fatalf("Failed to auth document: %v", err)
42+ }
43+ fmt.Printf("Auth response - URL: %s\n", auth.URL)
44+
45+ doc1, err := ygo.NewDoc()
46+ if err != nil {
47+ log.Fatalf("Failed to create doc1: %v", err)
48+ }
49+ defer doc1.Destroy()
50+
51+ doc2, err := ygo.NewDoc()
52+ if err != nil {
53+ log.Fatalf("Failed to create doc2: %v", err)
54+ }
55+ defer doc2.Destroy()
56+
57+ txt1, err := doc1.GetText("content")
58+ if err != nil {
59+ log.Fatalf("Failed to get text1: %v", err)
60+ }
61+ defer txt1.Destroy()
62+
63+ txt2, err := doc2.GetText("content")
64+ if err != nil {
65+ log.Fatalf("Failed to get text1: %v", err)
66+ }
67+ defer txt2.Destroy()
68+
69+ err = doc1.Sync(ctx, ygo.WithSyncEndpoint(auth.WebsocketURL()), ygo.WithOnUpdate(func(d *ygo.Doc) error {
70+ jsonBytes, err := d.MarshalJSON(ygo.WithText("content"))
71+ if err != nil {
72+ return err
73+ }
74+ fmt.Println("Doc1 received update:", string(jsonBytes))
75+ return nil
76+ }))
77+ if err != nil {
78+ log.Fatalf("Failed to sync doc1: %v", err)
79+ }
80+ fmt.Println("Doc1 connected")
81+
82+ var fired bool
83+ err = doc2.Sync(ctx, ygo.WithSyncEndpoint(auth.WebsocketURL()), ygo.WithOnUpdate(func(d *ygo.Doc) error {
84+ jsonBytes, err := d.MarshalJSON(ygo.WithText("content"))
85+ if err != nil {
86+ return err
87+ }
88+ fmt.Println("Doc2 received update:", string(jsonBytes))
89+ // send something back one time, without the fired flag it would repeatedly
90+ // add this text to the end
91+ if !fired {
92+ if err := d.WithWriteTransaction(func(txn *ygo.Transaction) error {
93+ length, err := txt2.Len(txn)
94+ if err != nil {
95+ return err
96+ }
97+ txt2.Insert(txn, length+1, " Right back at you from client 2")
98+ fired = true
99+ return nil
100+ }); err != nil {
101+ return err
102+ }
103+
104+ }
105+ return nil
106+ }))
107+ if err != nil {
108+ log.Fatalf("Failed to sync doc2: %v", err)
109+ }
110+ fmt.Println("Doc2 connected")
111+
112+ err = doc1.WithWriteTransaction(func(txn *ygo.Transaction) error {
113+ txt1.Insert(txn, 0, "Hello from client 1")
114+ return nil
115+ })
116+ if err != nil {
117+ log.Fatalf("Failed to insert text: %v", err)
118+ }
119+
120+ fmt.Println("Inserted text from doc1, waiting for updates...")
121+
122+ <-ctx.Done()
123+ fmt.Println("Done")
124+}
M
sync.go
+21,
-3
1@@ -15,21 +15,39 @@ type syncDocAdapter struct {
2
3 func (a *syncDocAdapter) WithReadTransaction(fn func(sync.Transaction) error) error {
4 return a.doc.WithReadTransaction(func(txn *Transaction) error {
5- adapter := &syncTransactionAdapter{txn: txn}
6+ adapter := &syncTransactionAdapter{txn: txn, doc: a.doc}
7 return fn(adapter)
8 })
9 }
10
11 func (a *syncDocAdapter) WithWriteTransaction(fn func(sync.Transaction) error) error {
12 return a.doc.WithWriteTransaction(func(txn *Transaction) error {
13- adapter := &syncTransactionAdapter{txn: txn}
14+ adapter := &syncTransactionAdapter{txn: txn, doc: a.doc}
15 return fn(adapter)
16 })
17 }
18
19+func (a *syncDocAdapter) GetStateDiff(stateVector []byte) []byte {
20+ var sv *StateVector
21+ if stateVector != nil {
22+ sv = &StateVector{data: stateVector}
23+ }
24+ var result []byte
25+ a.doc.WithReadTransaction(func(txn *Transaction) error {
26+ update := txn.GetStateDiff(sv)
27+ if update != nil {
28+ result = update.Data()
29+ }
30+ return nil
31+ })
32+ return result
33+}
34+
35 // syncTransactionAdapter adapts *Transaction to sync.Transaction
36 type syncTransactionAdapter struct {
37- txn *Transaction
38+ txn *Transaction
39+ doc *Doc
40+ postUpdateHook func()
41 }
42
43 func (a *syncTransactionAdapter) ApplyUpdate(data []byte) error {
+81,
-7
1@@ -90,7 +90,7 @@ func (s *SyncConn) sendSyncStep1(ctx context.Context) error {
2 return fmt.Errorf("failed to get state vector: %w", err)
3 }
4
5- msg := encodeMessage(SyncStep1, sv)
6+ msg := encodeSyncMessage(SyncStep1, sv)
7 err = s.conn.Write(ctx, websocket.MessageBinary, msg)
8 if err != nil {
9 return fmt.Errorf("failed to send SyncStep1: %w", err)
10@@ -99,6 +99,34 @@ func (s *SyncConn) sendSyncStep1(ctx context.Context) error {
11 return nil
12 }
13
14+func (s *SyncConn) sendSyncStep2(ctx context.Context) error {
15+ var sv []byte
16+ err := s.doc.WithReadTransaction(func(txn Transaction) error {
17+ stateVec := txn.GetStateVector()
18+ if stateVec == nil {
19+ return fmt.Errorf("failed to get state vector")
20+ }
21+ sv = stateVec
22+ return nil
23+ })
24+ if err != nil {
25+ return fmt.Errorf("failed to get state vector: %w", err)
26+ }
27+
28+ update := s.doc.GetStateDiff(sv)
29+ if update == nil {
30+ return fmt.Errorf("failed to get state diff")
31+ }
32+
33+ msg := encodeSyncMessage(SyncStep2, []byte(update))
34+ err = s.conn.Write(ctx, websocket.MessageBinary, msg)
35+ if err != nil {
36+ return fmt.Errorf("failed to send SyncStep2: %w", err)
37+ }
38+
39+ return nil
40+}
41+
42 func (s *SyncConn) handleMessage(ctx context.Context, data []byte) error {
43 if len(data) < 1 {
44 return fmt.Errorf("message too short")
45@@ -110,28 +138,69 @@ func (s *SyncConn) handleMessage(ctx context.Context, data []byte) error {
46 }
47
48 payload := data[n:]
49+
50 switch msgType {
51+ case uint64(MessageSync):
52+ return s.handleSyncMessage(payload)
53+ case uint64(MessageAwareness):
54+ return s.handleAwareness(payload)
55+ default:
56+ return fmt.Errorf("unknown message type: %d", msgType)
57+ }
58+}
59+
60+func (s *SyncConn) handleSyncMessage(data []byte) error {
61+ if len(data) < 1 {
62+ return fmt.Errorf("sync message payload too short")
63+ }
64+
65+ innerType, n, err := readVarUint(data)
66+ if err != nil {
67+ return fmt.Errorf("failed to read inner message type: %w", err)
68+ }
69+
70+ payload := data[n:]
71+
72+ switch innerType {
73+ case uint64(SyncStep1):
74+ return s.handleSyncStep1Response(payload)
75 case uint64(SyncStep2):
76 return s.handleSyncStep2(payload)
77 case uint64(Update):
78 return s.handleUpdate(payload)
79- case uint64(AwarenessUpdate):
80- return s.handleAwareness(payload)
81 default:
82- return fmt.Errorf("unknown message type: %d", msgType)
83+ return fmt.Errorf("unknown sync message type: %d", innerType)
84 }
85 }
86
87+func (s *SyncConn) handleSyncStep1Response(payload []byte) error {
88+ return s.sendSyncStep2(context.Background())
89+}
90+
91 func (s *SyncConn) handleSyncStep2(payload []byte) error {
92- update := UpdateData(payload)
93+ length, n, err := readVarUint(payload)
94+ if err != nil {
95+ return fmt.Errorf("failed to read update length: %w", err)
96+ }
97+ if len(payload) < n+int(length) {
98+ return fmt.Errorf("update payload too short: got %d bytes, expected %d", len(payload), n+int(length))
99+ }
100+ update := UpdateData(payload[n : n+int(length)])
101 return s.doc.WithWriteTransaction(func(txn Transaction) error {
102 return txn.ApplyUpdate(update)
103 })
104 }
105
106 func (s *SyncConn) handleUpdate(payload []byte) error {
107- update := UpdateData(payload)
108- err := s.doc.WithWriteTransaction(func(txn Transaction) error {
109+ length, n, err := readVarUint(payload)
110+ if err != nil {
111+ return fmt.Errorf("failed to read update length: %w", err)
112+ }
113+ if len(payload) < n+int(length) {
114+ return fmt.Errorf("update payload too short: got %d bytes, expected %d", len(payload), n+int(length))
115+ }
116+ update := UpdateData(payload[n : n+int(length)])
117+ err = s.doc.WithWriteTransaction(func(txn Transaction) error {
118 return txn.ApplyUpdate(update)
119 })
120 if err != nil {
121@@ -318,6 +387,11 @@ func (s *SyncConn) SetOnUpdate(fn func(DocInterface) error) {
122 s.onUpdate = fn
123 }
124
125+// GetConn returns the underlying WebSocket connection.
126+func (s *SyncConn) GetConn() *websocket.Conn {
127+ return s.conn
128+}
129+
130 // SetAwarenessUpdate sets a callback for handling awareness updates
131 // from the server.
132 func (s *SyncConn) SetAwarenessUpdate(fn func([]byte) error) {
+23,
-11
1@@ -2,20 +2,17 @@ package sync
2
3 import "errors"
4
5-// Message types for the Yjs sync protocol.
6 const (
7- // SyncStep1 is the first message in the sync handshake.
8- // The client sends its state vector to request missing updates.
9+ MessageSync uint8 = 0
10+ MessageAwareness uint8 = 1
11+ MessageAuth uint8 = 2
12+ MessageAwarenessQuery uint8 = 3
13+)
14+
15+const (
16 SyncStep1 uint8 = 0
17- // SyncStep2 is the response to SyncStep1.
18- // The server sends all updates the client is missing.
19 SyncStep2 uint8 = 1
20- // Update is an incremental document update.
21- // Sent after the initial sync to keep documents in sync.
22- Update uint8 = 2
23- // AwarenessUpdate carries awareness protocol data.
24- // Tracks presence and cursor information of clients.
25- AwarenessUpdate uint8 = 3
26+ Update uint8 = 2
27 )
28
29 func writeVarUint(b []byte, v uint64) int {
30@@ -72,6 +69,21 @@ func encodeMessage(msgType uint8, payload []byte) []byte {
31 return msg
32 }
33
34+func encodeSyncMessage(innerType uint8, payload []byte) []byte {
35+ msg := make([]byte, 0, 2+len(payload))
36+ msg = appendVarUint(msg, uint64(MessageSync))
37+ msg = appendVarUint(msg, uint64(innerType))
38+ msg = appendVarByteArray(msg, payload)
39+ return msg
40+}
41+
42+func encodeAwarenessMessage(payload []byte) []byte {
43+ msg := make([]byte, 0, 2+len(payload))
44+ msg = appendVarUint(msg, uint64(MessageAwareness))
45+ msg = appendVarByteArray(msg, payload)
46+ return msg
47+}
48+
49 func appendVarUint(b []byte, v uint64) []byte {
50 var buf [10]byte
51 n := writeVarUint(buf[:], v)
+0,
-3
1@@ -127,9 +127,6 @@ func TestMessageTypeConstants(t *testing.T) {
2 if Update != 2 {
3 t.Errorf("expected Update=2, got %d", Update)
4 }
5- if AwarenessUpdate != 3 {
6- t.Errorf("expected AwarenessUpdate=3, got %d", AwarenessUpdate)
7- }
8 }
9
10 func TestWriteReadVarUintRoundTrip(t *testing.T) {
+7,
-2
1@@ -6,8 +6,9 @@ import (
2 )
3
4 type mockDoc struct {
5- readTx func(func(Transaction) error) error
6- writeTx func(func(Transaction) error) error
7+ readTx func(func(Transaction) error) error
8+ writeTx func(func(Transaction) error) error
9+ stateDiff []byte
10 }
11
12 func (m *mockDoc) WithReadTransaction(fn func(Transaction) error) error {
13@@ -18,6 +19,10 @@ func (m *mockDoc) WithWriteTransaction(fn func(Transaction) error) error {
14 return m.writeTx(fn)
15 }
16
17+func (m *mockDoc) GetStateDiff(stateVector []byte) []byte {
18+ return m.stateDiff
19+}
20+
21 func TestSyncClientStruct(t *testing.T) {
22 doc := &mockDoc{}
23
+11,
-3
1@@ -40,6 +40,8 @@ type DocInterface interface {
2 // WithWriteTransaction executes a callback within a read-write transaction.
3 // The transaction parameter can apply updates using raw byte data.
4 WithWriteTransaction(fn func(Transaction) error) error
5+ // GetStateDiff returns an update containing all changes since the given state vector.
6+ GetStateDiff(stateVector []byte) []byte
7 }
8
9 // SyncClient manages a WebSocket connection to a y-sweet server.
10@@ -182,12 +184,18 @@ func WithAwarenessState(state []byte) Option {
11 // SendUpdate sends a document update to the server.
12 // It returns nil if not connected, so callers don't need to check status.
13 func (c *SyncClient) SendUpdate(update UpdateData) error {
14- if c.status != StatusConnected || c.conn == nil {
15+ if c.status != StatusConnected || c.syncConn == nil {
16 return nil
17 }
18
19- msg := encodeMessage(Update, update)
20- return c.conn.Write(context.Background(), websocket.MessageBinary, msg)
21+ // Access the connection through syncConn
22+ conn := c.syncConn.GetConn()
23+ if conn == nil {
24+ return nil
25+ }
26+
27+ msg := encodeSyncMessage(Update, update)
28+ return conn.Write(context.Background(), websocket.MessageBinary, msg)
29 }
30
31 // Awareness returns the awareness client for this sync client.
+120,
-0
1@@ -0,0 +1,120 @@
2+package ysweet
3+
4+import (
5+ "bytes"
6+ "encoding/json"
7+ "fmt"
8+ "net/http"
9+ "strings"
10+)
11+
12+type Client struct {
13+ baseURL string
14+}
15+
16+func NewClient(baseURL string) (*Client, error) {
17+ baseURL = strings.TrimSuffix(baseURL, "/")
18+ if baseURL == "" {
19+ return nil, fmt.Errorf("baseURL is required")
20+ }
21+ if !strings.HasPrefix(baseURL, "http://") && !strings.HasPrefix(baseURL, "https://") {
22+ return nil, fmt.Errorf("baseURL must start with http:// or https://")
23+ }
24+ return &Client{baseURL: baseURL}, nil
25+}
26+
27+type newDocRequest struct {
28+ DocID string `json:"docID"`
29+}
30+
31+type newDocResponse struct {
32+ DocID string `json:"docID"`
33+}
34+
35+type AuthDocResponse struct {
36+ URL string `json:"url"`
37+ BaseURL string `json:"baseUrl"`
38+ DocID string `json:"docId"`
39+ Token string `json:"token,omitempty"`
40+ Authorization string `json:"authorization,omitempty"`
41+}
42+
43+// Returns the actual WebsocketURL y-sweet expects a client to connect to, which is /d/{docID}/ws/{docID} for
44+// some unknown reason. Also includes token as query string, if defined.
45+func (a AuthDocResponse) WebsocketURL() string {
46+ if a.Token != "" {
47+ return fmt.Sprintf("%s/%s?token=%s", a.URL, a.DocID, a.Token)
48+ }
49+ return fmt.Sprintf("%s/%s", a.URL, a.DocID)
50+}
51+
52+func (c *Client) NewDoc(docID string) (string, error) {
53+ var body []byte
54+ var err error
55+ if docID != "" {
56+ body, err = json.Marshal(newDocRequest{DocID: docID})
57+ if err != nil {
58+ return "", fmt.Errorf("failed to marshal request: %w", err)
59+ }
60+ } else {
61+ body, err = json.Marshal(struct{}{})
62+ if err != nil {
63+ return "", fmt.Errorf("failed to marshal request: %w", err)
64+ }
65+ }
66+
67+ req, err := http.NewRequest(http.MethodPost, c.baseURL+"/doc/new", bytes.NewReader(body))
68+ if err != nil {
69+ return "", fmt.Errorf("failed to create request: %w", err)
70+ }
71+ req.Header.Set("Content-Type", "application/json")
72+
73+ resp, err := http.DefaultClient.Do(req)
74+ if err != nil {
75+ return "", fmt.Errorf("request failed: %w", err)
76+ }
77+ defer resp.Body.Close()
78+
79+ if resp.StatusCode < 200 || resp.StatusCode >= 300 {
80+ return "", fmt.Errorf("server returned status %d", resp.StatusCode)
81+ }
82+
83+ var result newDocResponse
84+ if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
85+ return "", fmt.Errorf("failed to decode response: %w", err)
86+ }
87+
88+ if docID != "" && result.DocID != docID {
89+ return "", fmt.Errorf("response docID %q does not match requested %q", result.DocID, docID)
90+ }
91+
92+ return result.DocID, nil
93+}
94+
95+func (c *Client) AuthDoc(docID string) (AuthDocResponse, error) {
96+ if docID == "" {
97+ return AuthDocResponse{}, fmt.Errorf("docID is required")
98+ }
99+
100+ req, err := http.NewRequest(http.MethodPost, c.baseURL+"/doc/"+docID+"/auth", nil)
101+ if err != nil {
102+ return AuthDocResponse{}, fmt.Errorf("failed to create request: %w", err)
103+ }
104+
105+ resp, err := http.DefaultClient.Do(req)
106+ if err != nil {
107+ return AuthDocResponse{}, fmt.Errorf("request failed: %w", err)
108+ }
109+ defer resp.Body.Close()
110+
111+ if resp.StatusCode < 200 || resp.StatusCode >= 300 {
112+ return AuthDocResponse{}, fmt.Errorf("server returned status %d", resp.StatusCode)
113+ }
114+
115+ var result AuthDocResponse
116+ if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
117+ return AuthDocResponse{}, fmt.Errorf("failed to decode response: %w", err)
118+ }
119+
120+ return result, nil
121+}