package main import ( "database/sql" "encoding/base64" "encoding/json" "fmt" "log" "net/http" "strconv" "strings" "mast/db" "github.com/gorilla/websocket" sqlite3 "github.com/mattn/go-sqlite3" ) var upgrader = websocket.Upgrader{ 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() 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 } 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) } } func printChanges(changes []interface{}) { 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 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)) // 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) } // Convert bytes to string decodedString := string(decodedBytes) // 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) } 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() } func broadcastChanges(sender *websocket.Conn, changes interface{}) { for client := range clients { if client != sender { client.WriteJSON(map[string]interface{}{"type": "changes", "data": changes}) } } } func main() { 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) } err = mast.RunMigrations(db) defer db.Close() // Initialize database and tables... http.HandleFunc("/sync", handleWebSocket) log.Fatal(http.ListenAndServe(":8080", nil)) }