mast/server/main.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))
}