From a593d23f23e89d747490a06f7d68cc7a669139ff Mon Sep 17 00:00:00 2001 From: Anish Lakhwara Date: Tue, 6 Aug 2024 20:06:49 -0700 Subject: [PATCH] correctly working sync --- js/src/App.svelte | 7 +- main.go | 109 +++++++++--------- server/main.go | 281 ++++++++++++++++++++++------------------------ 3 files changed, 190 insertions(+), 207 deletions(-) diff --git a/js/src/App.svelte b/js/src/App.svelte index e8731b3..46b1cc6 100644 --- a/js/src/App.svelte +++ b/js/src/App.svelte @@ -14,9 +14,6 @@ const sqlite = await initWasm(() => wasmUrl); db = await sqlite.open('todo.db'); - await db.exec(` - DROP TABLE todos; - `); await db.exec(` CREATE TABLE IF NOT EXISTS todos ( id BLOB PRIMARY KEY NOT NULL, @@ -54,9 +51,9 @@ function sendChangesToServer(changes) { const formattedChanges = changes.map(change => ({ TableName: change[0], - PK: Array.isArray(change[1]) ? btoa(change[1]) : btoa(change[1]), + PK: Array.isArray(change[1]) ? btoa(String.fromCharCode.apply(null, change[1])) : btoa(change[1]), ColumnName: change[2], - Value: Array.isArray(change[3]) ? btoa(String.fromCharCode.apply(null, change[3])) : btoa(change[3]), + Value: Array.isArray(change[3]) ? btoa(String.fromCharCode.apply(null, change[3])) : change[3], ColVersion: Number(change[4]), DBVersion: Number(change[5]), SiteID: Array.isArray(change[6]) ? btoa(new TextDecoder('utf8').decode(change[6])) : btoa(change[6]), diff --git a/main.go b/main.go index 7576bfb..aa977f9 100644 --- a/main.go +++ b/main.go @@ -48,8 +48,8 @@ func initialModel() model { conn, _, err := websocket.DefaultDialer.Dial("ws://localhost:8080/sync", nil) if err != nil { - log.Println("Error connecting to WebSocket:", err) - panic(err) + log.Println("Error connecting to WebSocket:", err) + panic(err) } m := model{ @@ -65,18 +65,17 @@ func initialModel() model { } type CRSQLChange struct { - TableName string - PK string - ColumnName string - Value string - ColVersion int64 - DBVersion int64 - SiteID string - CL int64 - Seq int64 + TableName string + PK string + ColumnName string + Value string + ColVersion int64 + DBVersion int64 + SiteID string + CL int64 + Seq int64 } - func (m *model) addTodo(todo string) { id, err := uuid.NewV7() if err != nil { @@ -90,59 +89,57 @@ func (m *model) addTodo(todo string) { panic(err) } - rows, err := m.db.Query("SELECT * FROM crsql_changes") - if err != nil { - log.Println("Error querying crsql_changes:", err) - return - } - defer rows.Close() + rows, err := m.db.Query("SELECT * FROM crsql_changes") + if err != nil { + log.Println("Error querying crsql_changes:", err) + return + } + defer rows.Close() - var changes []CRSQLChange - for rows.Next() { - var change CRSQLChange - var pkRaw []byte - var valueRaw []byte - var siteIDRaw []byte - err := rows.Scan(&change.TableName, &pkRaw, &change.ColumnName, &valueRaw, - &change.ColVersion, &change.DBVersion, &siteIDRaw, &change.CL, &change.Seq) + var changes []CRSQLChange + for rows.Next() { + var change CRSQLChange + var pkRaw []byte + var valueRaw []byte + var siteIDRaw []byte + err := rows.Scan(&change.TableName, &pkRaw, &change.ColumnName, &valueRaw, + &change.ColVersion, &change.DBVersion, &siteIDRaw, &change.CL, &change.Seq) - if err != nil { - log.Println("Error scanning change:", err) - return - } - change.PK = base64.StdEncoding.EncodeToString(pkRaw) - change.Value = base64.StdEncoding.EncodeToString(valueRaw) - change.SiteID = base64.StdEncoding.EncodeToString(siteIDRaw) + if err != nil { + log.Println("Error scanning change:", err) + return + } + change.PK = base64.StdEncoding.EncodeToString(pkRaw) + change.Value = base64.StdEncoding.EncodeToString(valueRaw) + change.SiteID = base64.StdEncoding.EncodeToString(siteIDRaw) + changes = append(changes, change) + fmt.Println(change) + } - changes = append(changes, change) - fmt.Println(change) - } - - if len(changes) > 0 { - m.sendToWebSocket(changes) - } + if len(changes) > 0 { + m.sendToWebSocket(changes) + } m.loadTodos() } func (m *model) sendToWebSocket(changes []CRSQLChange) { - message := struct { - Type string `json:"type"` - Data []CRSQLChange `json:"data"` - }{ - Type: "changes", - Data: changes, - } + message := struct { + Type string `json:"type"` + Data []CRSQLChange `json:"data"` + }{ + Type: "changes", + Data: changes, + } - err := m.conn.WriteJSON(message) - if err != nil { - log.Println("Error sending changes to WebSocket:", err) - panic(err) - } + err := m.conn.WriteJSON(message) + if err != nil { + log.Println("Error sending changes to WebSocket:", err) + panic(err) + } } - func (m *model) loadTodos() { rows, err := m.db.Query("SELECT content FROM todos") if err != nil { @@ -199,9 +196,9 @@ func (m model) Update(msg tea.Msg) (tea.Model, tea.Cmd) { } func (m model) quit() tea.Cmd { - m.db.Close() - m.conn.Close() - return tea.Quit + m.db.Close() + m.conn.Close() + return tea.Quit } func (m model) View() string { diff --git a/server/main.go b/server/main.go index 1ad0341..7d30fbb 100644 --- a/server/main.go +++ b/server/main.go @@ -17,195 +17,184 @@ import ( ) var upgrader = websocket.Upgrader{ - CheckOrigin: func(r *http.Request) bool { - return true - }, + CheckOrigin: func(r *http.Request) bool { + return true + }, } var db *sql.DB var clients = make(map[*websocket.Conn]bool) func handleWebSocket(w http.ResponseWriter, r *http.Request) { - conn, err := upgrader.Upgrade(w, r, nil) - if err != nil { - log.Println(err) - return - } - defer conn.Close() + conn, err := upgrader.Upgrade(w, r, nil) + if err != nil { + log.Println(err) + return + } + defer conn.Close() - for { - _, message, err := conn.ReadMessage() - if err != nil { - log.Println(err) - return - } + for { + _, message, err := conn.ReadMessage() + if err != nil { + log.Println(err) + return + } - var msg struct { - Type string `json:"type"` - Data []interface{} `json:"data"` - } - if err := json.Unmarshal(message, &msg); err != nil { - log.Println("Error unmarshaling message:", err) - continue - } + var msg struct { + Type string `json:"type"` + Data []interface{} `json:"data"` + } + if err := json.Unmarshal(message, &msg); err != nil { + log.Println("Error unmarshaling message:", err) + continue + } - if msg.Type == "changes" { - fmt.Println("We're applying changes!!") - applyChanges(msg.Data) - } - } + if msg.Type == "changes" { + fmt.Println("We're applying changes!!") + applyChanges(msg.Data) + } + } } func printChange(change interface{}) { - // Attempt to type assert and print more details - if c, ok := change.(map[string]interface{}); ok { - for key, value := range c { - fmt.Printf(" %s: %v\n", key, value) - } - } else { - fmt.Printf(" Not a map: %T\n", change) - } + // Attempt to type assert and print more details + if c, ok := change.(map[string]interface{}); ok { + for key, value := range c { + fmt.Printf(" %s: %v\n", key, value) + } + } else { + fmt.Printf(" Not a map: %T\n", change) + } } func printChanges(changes []interface{}) { - for i, change := range changes { - fmt.Printf("Change %d: %+v\n", i, change) - printChange(change) - } + for i, change := range changes { + fmt.Printf("Change %d: %+v\n", i, change) + printChange(change) + } } type CRSQLChange struct { - TableName string - PK []byte - ColumnName string - Value []byte - ColVersion int64 - DBVersion int64 - SiteID []byte - CL int64 - Seq int64 + TableName string + PK []byte + ColumnName string + Value any + ColVersion int64 + DBVersion int64 + SiteID []byte + CL int64 + Seq int64 } func mapToCRSQLChange(m map[string]interface{}) (CRSQLChange, error) { var err error - change := CRSQLChange{} - - change.TableName = m["TableName"].(string) - change.ColumnName = m["ColumnName"].(string) - change.ColVersion = int64(m["ColVersion"].(float64)) - change.DBVersion = int64(m["DBVersion"].(float64)) - change.CL = int64(m["CL"].(float64)) - change.Seq = int64(m["Seq"].(float64)) - - // Handle []byte fields - change.PK, err = decodeBase64ToByteArray(m["PK"].(string)) - if err != nil { - return change, err - } - //fmt.Println(m["PK"]) - //fmt.Println(m["PK"].(string)) - //fmt.Println(base64.URLEncoding.DecodeString(m["PK"].(string))) - change.Value, err = decodeBase64ToByteArray(m["Value"].(string)) - if err != nil { - return change, err - } - change.SiteID, err = decodeBase64ToByteArray(m["SiteID"].(string)) - fmt.Println(change.SiteID) - if err != nil { - return change, err - } + change := CRSQLChange{} - return change, nil + change.TableName = m["TableName"].(string) + change.ColumnName = m["ColumnName"].(string) + change.ColVersion = int64(m["ColVersion"].(float64)) + change.DBVersion = int64(m["DBVersion"].(float64)) + change.CL = int64(m["CL"].(float64)) + change.Seq = int64(m["Seq"].(float64)) + + // Still unsure about this, it can encode anything + if m["Value"] != nil { + change.Value = m["Value"] + } + // Handle []byte fields + change.PK, err = decodeBase64ToByteArray(m["PK"].(string)) + if err != nil { + return change, err + } + change.SiteID, err = decodeBase64ToByteArray(m["SiteID"].(string)) + fmt.Println(change.SiteID) + if err != nil { + return change, err + } + + return change, nil } - func decodeBase64ToByteArray(encodedString string) ([]uint8, error) { - // Decode the base64 string - decodedBytes, err := base64.StdEncoding.DecodeString(encodedString) - if err != nil { - return nil, fmt.Errorf("error decoding base64: %w", err) - } + // Decode the base64 string + decodedBytes, err := base64.StdEncoding.DecodeString(encodedString) + if err != nil { + return nil, fmt.Errorf("error decoding base64: %w", err) + } - // Convert bytes to string - decodedString := string(decodedBytes) + // Convert bytes to string + decodedString := string(decodedBytes) - // Split the string by commas - stringSlice := strings.Split(decodedString, ",") + // Split the string by commas + stringSlice := strings.Split(decodedString, ",") - // Convert each string to an integer - intSlice := make([]uint8, len(stringSlice)) - for i, s := range stringSlice { - num, err := strconv.Atoi(s) - if err != nil { - return nil, fmt.Errorf("error converting string to int: %w", err) - } - intSlice[i] = byte(num) - } + // Convert each string to an integer + intSlice := make([]uint8, len(stringSlice)) + for i, s := range stringSlice { + num, err := strconv.Atoi(s) + if err != nil { + return nil, fmt.Errorf("error converting string to int: %w", err) + } + intSlice[i] = byte(num) + } - return intSlice, nil + return intSlice, nil } - func applyChanges(changes []interface{}) { - tx, _ := db.Begin() - for i, change := range changes { - fmt.Println("Change number: ", i) - fmt.Println(change) - if changeMap, ok := change.(map[string]interface{}); ok { - c, err := mapToCRSQLChange(changeMap) - if err != nil { - fmt.Println("Error converting to CRSQLChange:", err) - continue - } - _, err = tx.Exec(`INSERT INTO crsql_changes VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`, - c.TableName, c.PK, c.ColumnName, c.Value, c.ColVersion, c.DBVersion, c.SiteID, c.CL, c.Seq) - if err != nil { - log.Println("Error inserting change:", err) - } - log.Println("We inserted data successfully??") - fmt.Println(c) - } else { - fmt.Println("Not OK to convert change") - fmt.Println("Change look like:") - printChange(change) - continue - } - } - tx.Commit() + tx, _ := db.Begin() + for i, change := range changes { + fmt.Println("Change number: ", i) + fmt.Println(change) + if changeMap, ok := change.(map[string]interface{}); ok { + c, err := mapToCRSQLChange(changeMap) + if err != nil { + fmt.Println("Error converting to CRSQLChange:", err) + continue + } + _, err = tx.Exec(`INSERT INTO crsql_changes VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`, + c.TableName, c.PK, c.ColumnName, c.Value, c.ColVersion, c.DBVersion, c.SiteID, c.CL, c.Seq) + if err != nil { + log.Println("Error inserting change:", err) + } + log.Println("We inserted data successfully??") + fmt.Println(c) + } else { + fmt.Println("Not OK to convert change") + fmt.Println("Change look like:") + printChange(change) + continue + } + } + tx.Commit() } - func broadcastChanges(sender *websocket.Conn, changes interface{}) { - for client := range clients { - if client != sender { - client.WriteJSON(map[string]interface{}{"type": "changes", "data": changes}) - } - } + for client := range clients { + if client != sender { + client.WriteJSON(map[string]interface{}{"type": "changes", "data": changes}) + } + } } func main() { - var err error - val, err := base64.StdEncoding.DecodeString("MTY3LDU0LDQ5LDY3LDExMyw2MCw3NiwxMjAsMTU2LDExMCwyMjcsMTc2LDE4NCwyMjksMjQxLDE0Mw==") - decoded, _ := base64.StdEncoding.DecodeString("MTY3LDU0LDQ5LDY3LDExMyw2MCw3NiwxMjAsMTU2LDExMCwyMjcsMTc2LDE4NCwyMjksMjQxLDE0Mw==") - if err != nil { - log.Fatal("error:", err) - } - fmt.Printf("%q\n", val) - fmt.Printf("%q\n", decoded) - sql.Register("sqlite3_with_extensions", + var err error + if err != nil { + log.Fatal("error:", err) + } + sql.Register("sqlite3_with_extensions", &sqlite3.SQLiteDriver{ Extensions: []string{ "../db/crsqlite.so", }, }) - db, err = sql.Open("sqlite3_with_extensions", "./todo.sqlite") - if err != nil { - log.Fatal(err) - } + db, err = sql.Open("sqlite3_with_extensions", "./todo.sqlite") + if err != nil { + log.Fatal(err) + } err = mast.RunMigrations(db) - defer db.Close() + defer db.Close() - // Initialize database and tables... - http.HandleFunc("/sync", handleWebSocket) - log.Fatal(http.ListenAndServe(":8080", nil)) + // Initialize database and tables... + http.HandleFunc("/sync", handleWebSocket) + log.Fatal(http.ListenAndServe(":8080", nil)) } -