aynish/mast
1
0
Fork 0
Code Issues 2 Pull requests Packages Projects Releases Wiki Activity

Sync implementation using cr-sqlite + websockets, across JS and CLI clients #1

Merged
aynish merged 4 commits from testing-sync into main 2024-08-31 18:06:10 -07:00
Conversation 0 Commits 4 Files changed 22 +190 -207
3 changed files with 190 additions and 207 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files
Showing only changes of commit a593d23f23 - Show all commits

7
js/src/App.svelte
View file

@ -14,9 +14,6 @@
const sqlite = await initWasm(() => wasmUrl); const sqlite = await initWasm(() => wasmUrl);
db = await sqlite.open('todo.db'); db = await sqlite.open('todo.db');
await db.exec(`
DROP TABLE todos;
`);
await db.exec(` await db.exec(`
CREATE TABLE IF NOT EXISTS todos ( CREATE TABLE IF NOT EXISTS todos (
id BLOB PRIMARY KEY NOT NULL, id BLOB PRIMARY KEY NOT NULL,
@ -54,9 +51,9 @@
function sendChangesToServer(changes) { function sendChangesToServer(changes) {
const formattedChanges = changes.map(change => ({ const formattedChanges = changes.map(change => ({
TableName: change[0], 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], 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]), ColVersion: Number(change[4]),
DBVersion: Number(change[5]), DBVersion: Number(change[5]),
SiteID: Array.isArray(change[6]) ? btoa(new TextDecoder('utf8').decode(change[6])) : btoa(change[6]), SiteID: Array.isArray(change[6]) ? btoa(new TextDecoder('utf8').decode(change[6])) : btoa(change[6]),

109
main.go
View file

@ -48,8 +48,8 @@ func initialModel() model {
conn, _, err := websocket.DefaultDialer.Dial("ws://localhost:8080/sync", nil) conn, _, err := websocket.DefaultDialer.Dial("ws://localhost:8080/sync", nil)
if err != nil { if err != nil {
log.Println("Error connecting to WebSocket:", err) log.Println("Error connecting to WebSocket:", err)
panic(err) panic(err)
} }
m := model{ m := model{
@ -65,18 +65,17 @@ func initialModel() model {
} }
type CRSQLChange struct { type CRSQLChange struct {
TableName string TableName string
PK string PK string
ColumnName string ColumnName string
Value string Value string
ColVersion int64 ColVersion int64
DBVersion int64 DBVersion int64
SiteID string SiteID string
CL int64 CL int64
Seq int64 Seq int64
} }
func (m *model) addTodo(todo string) { func (m *model) addTodo(todo string) {
id, err := uuid.NewV7() id, err := uuid.NewV7()
if err != nil { if err != nil {
@ -90,59 +89,57 @@ func (m *model) addTodo(todo string) {
panic(err) panic(err)
} }
rows, err := m.db.Query("SELECT * FROM crsql_changes") rows, err := m.db.Query("SELECT * FROM crsql_changes")
if err != nil { if err != nil {
log.Println("Error querying crsql_changes:", err) log.Println("Error querying crsql_changes:", err)
return return
} }
defer rows.Close() defer rows.Close()
var changes []CRSQLChange var changes []CRSQLChange
for rows.Next() { for rows.Next() {
var change CRSQLChange var change CRSQLChange
var pkRaw []byte var pkRaw []byte
var valueRaw []byte var valueRaw []byte
var siteIDRaw []byte var siteIDRaw []byte
err := rows.Scan(&change.TableName, &pkRaw, &change.ColumnName, &valueRaw, err := rows.Scan(&change.TableName, &pkRaw, &change.ColumnName, &valueRaw,
&change.ColVersion, &change.DBVersion, &siteIDRaw, &change.CL, &change.Seq) &change.ColVersion, &change.DBVersion, &siteIDRaw, &change.CL, &change.Seq)
if err != nil { if err != nil {
log.Println("Error scanning change:", err) log.Println("Error scanning change:", err)
return return
} }
change.PK = base64.StdEncoding.EncodeToString(pkRaw) change.PK = base64.StdEncoding.EncodeToString(pkRaw)
change.Value = base64.StdEncoding.EncodeToString(valueRaw) change.Value = base64.StdEncoding.EncodeToString(valueRaw)
change.SiteID = base64.StdEncoding.EncodeToString(siteIDRaw) change.SiteID = base64.StdEncoding.EncodeToString(siteIDRaw)
changes = append(changes, change)
fmt.Println(change)
}
changes = append(changes, change) if len(changes) > 0 {
fmt.Println(change) m.sendToWebSocket(changes)
} }
if len(changes) > 0 {
m.sendToWebSocket(changes)
}
m.loadTodos() m.loadTodos()
} }
func (m *model) sendToWebSocket(changes []CRSQLChange) { func (m *model) sendToWebSocket(changes []CRSQLChange) {
message := struct { message := struct {
Type string `json:"type"` Type string `json:"type"`
Data []CRSQLChange `json:"data"` Data []CRSQLChange `json:"data"`
}{ }{
Type: "changes", Type: "changes",
Data: changes, Data: changes,
} }
err := m.conn.WriteJSON(message) err := m.conn.WriteJSON(message)
if err != nil { if err != nil {
log.Println("Error sending changes to WebSocket:", err) log.Println("Error sending changes to WebSocket:", err)
panic(err) panic(err)
} }
} }
func (m *model) loadTodos() { func (m *model) loadTodos() {
rows, err := m.db.Query("SELECT content FROM todos") rows, err := m.db.Query("SELECT content FROM todos")
if err != nil { if err != nil {
@ -199,9 +196,9 @@ func (m model) Update(msg tea.Msg) (tea.Model, tea.Cmd) {
} }
func (m model) quit() tea.Cmd { func (m model) quit() tea.Cmd {
m.db.Close() m.db.Close()
m.conn.Close() m.conn.Close()
return tea.Quit return tea.Quit
} }
func (m model) View() string { func (m model) View() string {

277
server/main.go
View file

@ -17,195 +17,184 @@ import (
) )
var upgrader = websocket.Upgrader{ var upgrader = websocket.Upgrader{
CheckOrigin: func(r *http.Request) bool { CheckOrigin: func(r *http.Request) bool {
return true return true
}, },
} }
var db *sql.DB var db *sql.DB
var clients = make(map[*websocket.Conn]bool) var clients = make(map[*websocket.Conn]bool)
func handleWebSocket(w http.ResponseWriter, r *http.Request) { func handleWebSocket(w http.ResponseWriter, r *http.Request) {
conn, err := upgrader.Upgrade(w, r, nil) conn, err := upgrader.Upgrade(w, r, nil)
if err != nil { if err != nil {
log.Println(err) log.Println(err)
return return
} }
defer conn.Close() defer conn.Close()
for { for {
_, message, err := conn.ReadMessage() _, message, err := conn.ReadMessage()
if err != nil { if err != nil {
log.Println(err) log.Println(err)
return return
} }
var msg struct { var msg struct {
Type string `json:"type"` Type string `json:"type"`
Data []interface{} `json:"data"` Data []interface{} `json:"data"`
} }
if err := json.Unmarshal(message, &msg); err != nil { if err := json.Unmarshal(message, &msg); err != nil {
log.Println("Error unmarshaling message:", err) log.Println("Error unmarshaling message:", err)
continue continue
} }
if msg.Type == "changes" { if msg.Type == "changes" {
fmt.Println("We're applying changes!!") fmt.Println("We're applying changes!!")
applyChanges(msg.Data) applyChanges(msg.Data)
} }
} }
} }
func printChange(change interface{}) { func printChange(change interface{}) {
// Attempt to type assert and print more details // Attempt to type assert and print more details
if c, ok := change.(map[string]interface{}); ok { if c, ok := change.(map[string]interface{}); ok {
for key, value := range c { for key, value := range c {
fmt.Printf(" %s: %v\n", key, value) fmt.Printf(" %s: %v\n", key, value)
} }
} else { } else {
fmt.Printf(" Not a map: %T\n", change) fmt.Printf(" Not a map: %T\n", change)
} }
} }
func printChanges(changes []interface{}) { func printChanges(changes []interface{}) {
for i, change := range changes { for i, change := range changes {
fmt.Printf("Change %d: %+v\n", i, change) fmt.Printf("Change %d: %+v\n", i, change)
printChange(change) printChange(change)
} }
} }
type CRSQLChange struct { type CRSQLChange struct {
TableName string TableName string
PK []byte PK []byte
ColumnName string ColumnName string
Value []byte Value any
ColVersion int64 ColVersion int64
DBVersion int64 DBVersion int64
SiteID []byte SiteID []byte
CL int64 CL int64
Seq int64 Seq int64
} }
func mapToCRSQLChange(m map[string]interface{}) (CRSQLChange, error) { func mapToCRSQLChange(m map[string]interface{}) (CRSQLChange, error) {
var err error var err error
change := CRSQLChange{} change := CRSQLChange{}
change.TableName = m["TableName"].(string) change.TableName = m["TableName"].(string)
change.ColumnName = m["ColumnName"].(string) change.ColumnName = m["ColumnName"].(string)
change.ColVersion = int64(m["ColVersion"].(float64)) change.ColVersion = int64(m["ColVersion"].(float64))
change.DBVersion = int64(m["DBVersion"].(float64)) change.DBVersion = int64(m["DBVersion"].(float64))
change.CL = int64(m["CL"].(float64)) change.CL = int64(m["CL"].(float64))
change.Seq = int64(m["Seq"].(float64)) change.Seq = int64(m["Seq"].(float64))
// Handle []byte fields // Still unsure about this, it can encode anything
change.PK, err = decodeBase64ToByteArray(m["PK"].(string)) if m["Value"] != nil {
if err != nil { change.Value = m["Value"]
return change, err }
} // Handle []byte fields
//fmt.Println(m["PK"]) change.PK, err = decodeBase64ToByteArray(m["PK"].(string))
//fmt.Println(m["PK"].(string)) if err != nil {
//fmt.Println(base64.URLEncoding.DecodeString(m["PK"].(string))) return change, err
change.Value, err = decodeBase64ToByteArray(m["Value"].(string)) }
if err != nil { change.SiteID, err = decodeBase64ToByteArray(m["SiteID"].(string))
return change, err fmt.Println(change.SiteID)
} if err != nil {
change.SiteID, err = decodeBase64ToByteArray(m["SiteID"].(string)) return change, err
fmt.Println(change.SiteID) }
if err != nil {
return change, err
}
return change, nil return change, nil
} }
func decodeBase64ToByteArray(encodedString string) ([]uint8, error) { func decodeBase64ToByteArray(encodedString string) ([]uint8, error) {
// Decode the base64 string // Decode the base64 string
decodedBytes, err := base64.StdEncoding.DecodeString(encodedString) decodedBytes, err := base64.StdEncoding.DecodeString(encodedString)
if err != nil { if err != nil {
return nil, fmt.Errorf("error decoding base64: %w", err) return nil, fmt.Errorf("error decoding base64: %w", err)
} }
// Convert bytes to string // Convert bytes to string
decodedString := string(decodedBytes) decodedString := string(decodedBytes)
// Split the string by commas // Split the string by commas
stringSlice := strings.Split(decodedString, ",") stringSlice := strings.Split(decodedString, ",")
// Convert each string to an integer // Convert each string to an integer
intSlice := make([]uint8, len(stringSlice)) intSlice := make([]uint8, len(stringSlice))
for i, s := range stringSlice { for i, s := range stringSlice {
num, err := strconv.Atoi(s) num, err := strconv.Atoi(s)
if err != nil { if err != nil {
return nil, fmt.Errorf("error converting string to int: %w", err) return nil, fmt.Errorf("error converting string to int: %w", err)
} }
intSlice[i] = byte(num) intSlice[i] = byte(num)
} }
return intSlice, nil return intSlice, nil
} }
func applyChanges(changes []interface{}) { func applyChanges(changes []interface{}) {
tx, _ := db.Begin() tx, _ := db.Begin()
for i, change := range changes { for i, change := range changes {
fmt.Println("Change number: ", i) fmt.Println("Change number: ", i)
fmt.Println(change) fmt.Println(change)
if changeMap, ok := change.(map[string]interface{}); ok { if changeMap, ok := change.(map[string]interface{}); ok {
c, err := mapToCRSQLChange(changeMap) c, err := mapToCRSQLChange(changeMap)
if err != nil { if err != nil {
fmt.Println("Error converting to CRSQLChange:", err) fmt.Println("Error converting to CRSQLChange:", err)
continue continue
} }
_, err = tx.Exec(`INSERT INTO crsql_changes VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`, _, 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) c.TableName, c.PK, c.ColumnName, c.Value, c.ColVersion, c.DBVersion, c.SiteID, c.CL, c.Seq)
if err != nil { if err != nil {
log.Println("Error inserting change:", err) log.Println("Error inserting change:", err)
} }
log.Println("We inserted data successfully??") log.Println("We inserted data successfully??")
fmt.Println(c) fmt.Println(c)
} else { } else {
fmt.Println("Not OK to convert change") fmt.Println("Not OK to convert change")
fmt.Println("Change look like:") fmt.Println("Change look like:")
printChange(change) printChange(change)
continue continue
} }
} }
tx.Commit() tx.Commit()
} }
func broadcastChanges(sender *websocket.Conn, changes interface{}) { func broadcastChanges(sender *websocket.Conn, changes interface{}) {
for client := range clients { for client := range clients {
if client != sender { if client != sender {
client.WriteJSON(map[string]interface{}{"type": "changes", "data": changes}) client.WriteJSON(map[string]interface{}{"type": "changes", "data": changes})
} }
} }
} }
func main() { func main() {
var err error var err error
val, err := base64.StdEncoding.DecodeString("MTY3LDU0LDQ5LDY3LDExMyw2MCw3NiwxMjAsMTU2LDExMCwyMjcsMTc2LDE4NCwyMjksMjQxLDE0Mw==") if err != nil {
decoded, _ := base64.StdEncoding.DecodeString("MTY3LDU0LDQ5LDY3LDExMyw2MCw3NiwxMjAsMTU2LDExMCwyMjcsMTc2LDE4NCwyMjksMjQxLDE0Mw==") log.Fatal("error:", err)
if err != nil { }
log.Fatal("error:", err) sql.Register("sqlite3_with_extensions",
}
fmt.Printf("%q\n", val)
fmt.Printf("%q\n", decoded)
sql.Register("sqlite3_with_extensions",
&sqlite3.SQLiteDriver{ &sqlite3.SQLiteDriver{
Extensions: []string{ Extensions: []string{
"../db/crsqlite.so", "../db/crsqlite.so",
}, },
}) })
db, err = sql.Open("sqlite3_with_extensions", "./todo.sqlite") db, err = sql.Open("sqlite3_with_extensions", "./todo.sqlite")
if err != nil { if err != nil {
log.Fatal(err) log.Fatal(err)
} }
err = mast.RunMigrations(db) err = mast.RunMigrations(db)
defer db.Close() defer db.Close()
// Initialize database and tables... // Initialize database and tables...
http.HandleFunc("/sync", handleWebSocket) http.HandleFunc("/sync", handleWebSocket)
log.Fatal(http.ListenAndServe(":8080", nil)) log.Fatal(http.ListenAndServe(":8080", nil))
} }