mast/server/main.go
aynish d46d67830a Sync implementation using cr-sqlite + websockets, across JS and CLI clients (#1)
- Implements CRDT syncing using cr-slqlite.
- Initializes a simple vite application from the cr-sqlite repos that uses Svelte
- implements a websocket go server backend for communicating syncs between applications

Needs lots of refactoring, everything is hardcoded

Co-authored-by: Anish Lakhwara <anish+git@lakhwara.com>
Reviewed-on: #1
2024-08-31 18:06:10 -07:00

201 lines
4.5 KiB
Go

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))
}