+72 -137

mast-react-vite/src/App.tsx

reviewed

··· 1 1 - import { useState, useEffect, useMemo } from "react"; 1 1 + import { useState, useEffect, useMemo, useRef } from "react"; 2 2 import { useQuery } from "@vlcn.io/react"; 3 3 - import { ColumnDef } from "@tanstack/react-table"; 4 4 - import { DataTable } from "@/components/ui/data-table"; 5 3 import * as commandParser from "@/lib/command_js.js"; 6 4 import { Checkbox } from "@/components/ui/checkbox"; 7 5 import { ActionParser } from "@/components/ui/action-parser"; 8 6 import { useSelection } from "@/contexts/selection-context"; 9 7 import { SidebarProvider, SidebarTrigger } from "@/components/ui/sidebar"; 10 8 import { AppSidebar } from "@/components/ui/app-sidebar"; 9 9 + import { DataTable } from "@/components/ui/data-table"; 11 10 import { Badge } from "@/components/ui/badge"; 12 11 13 12 type Todo = { ··· 18 17 project: string; 19 18 }; 20 19 21 21 - export const columns: ColumnDef<Todo>[] = [ 22 22 - { 23 23 - id: "select", 24 24 - header: ({ table }) => ( 25 25 - <div className="flex items-center gap-2"> 26 26 - <p>0</p> 27 27 - <Checkbox 28 28 - checked={ 29 29 - table.getIsAllPageRowsSelected() || 30 30 - (table.getIsSomePageRowsSelected() && "indeterminate") 31 31 - } 32 32 - onCheckedChange={(value) => table.toggleAllPageRowsSelected(!!value)} 33 33 - aria-label="Select all" 34 34 - className="translate-y-[2px]" 35 35 - /> 36 36 - </div> 37 37 - ), 38 38 - cell: ({ row, table }) => ( 39 39 - <div className="flex items-center gap-2"> 40 40 - <p>{row.index + 1}</p> 41 41 - <Checkbox 42 42 - checked={row.getIsSelected()} 43 43 - onCheckedChange={(value) => { 44 44 - row.toggleSelected(!!value); 45 45 - // Get setNewText from table meta 46 46 - const setNewText = table.options.meta?.setNewText; 47 47 - const NewText = table.options.meta?.newText; 48 48 - if (setNewText && value === true) { 49 49 - const calculatedRowId = row.index + 1; 50 50 - setNewText(NewText + calculatedRowId + ","); 51 51 - } else { 52 52 - // TODO: 53 53 - // We need to _remove_ the value here 54 54 - // That probably means taking the NewText 55 55 - // Removing the first instance of "row.index," 56 56 - // And then setting that with setNewText 57 57 - } 58 58 - }} 59 59 - aria-label="Select row" 60 60 - className="translate-y-[2px]" 61 61 - /> 62 62 - </div> 63 63 - ), 64 64 - enableSorting: false, 65 65 - enableHiding: false, 66 66 - size: 20, 67 67 - }, 68 68 - { 69 69 - accessorKey: "description", 70 70 - header: "Description", 71 71 - }, 72 72 - { 73 73 - accessorKey: "project", 74 74 - header: "Project", 75 75 - }, 76 76 - { 77 77 - accessorKey: "tags", 78 78 - header: "Tags", 79 79 - cell: ({ row }) => { 80 80 - const formatted = JSON.parse(row.getValue("tags")); 81 81 - return <div className="font-medium">{formatted.join(", ")}</div>; 82 82 - }, 83 83 - }, 84 84 - ]; 85 85 - 86 86 - function App({ ctx }) { 20 20 + function App({ ctx, syncWorker, dbname }) { 87 21 const [newText, setNewText] = useState(""); 88 22 const { selectedItems, clearSelection, getSelectionString } = useSelection(); 89 23 const [currentAction, setCurrentAction] = useState("add"); 90 24 const [filterContext, setFilterContext] = useState({}); 91 91 - 92 25 const newConditions = []; 93 26 const newParams = []; 94 27 ··· 99 32 newConditions.push("description LIKE ?"); 100 33 newParams.push(`%${filterContext.filterDescription}%`); 101 34 } 35 35 + 102 36 if (filterContext.filterProject && filterContext.filterProject.length > 0) { 103 37 newConditions.push("project = ?"); 104 38 newParams.push(filterContext.filterProject); 105 39 } 40 40 + 106 41 if (filterContext.filterTags && filterContext.filterTags.length > 0) { 107 42 newConditions.push(`EXISTS ( 108 43 SELECT 1 FROM json_each(tags) ··· 126 61 `SELECT * FROM active_todos ${whereClause ? "WHERE " + whereClause : ""}`, 127 62 newParams, 128 63 ).data; 64 64 + 65 65 + // Trigger sync when todos change 66 66 + useEffect(() => { 67 67 + if (syncWorker && todos.length > 0) { 68 68 + syncWorker.postMessage({ 69 69 + type: 'SYNC_CHANGES', 70 70 + dbname 71 71 + }); 72 72 + } 73 73 + }, [todos, syncWorker]); 129 74 130 75 useEffect(() => { 131 76 if ( ··· 145 90 ); 146 91 } catch (error) { 147 92 console.log("Could not parse command"); 148 148 - console.log( getSelectionString() + " " + currentAction + " " + newText) 93 93 + console.log(getSelectionString() + " " + currentAction + " " + newText); 149 94 return {}; 150 95 } 151 96 }, [getSelectionString(), currentAction, newText]); ··· 164 109 working_id: 0, 165 110 }; 166 111 return [newTodo, ...todos]; 167 167 - 168 112 case "done": 169 113 return todos.map((todo) => { 170 114 if (selectedItems.has(todo.working_id)) { ··· 172 116 } 173 117 return todo; 174 118 }); 175 175 - 176 176 - case "modify": 177 177 - return todos.map((todo) => { 178 178 - if (selectedItems.has(todo.working_id)) { 179 179 - // Create a preview object that contains the original todo and the changes 180 180 - const previewChanges = {}; 181 181 - 182 182 - // Only add fields that are being changed to the preview 183 183 - if (parsedCommand.description) { 184 184 - previewChanges.description = parsedCommand.description; 185 185 - } 186 186 - 187 187 - if (parsedCommand.project) { 188 188 - previewChanges.project = parsedCommand.project; 189 189 - } 190 190 - 191 191 - if (parsedCommand.tags && parsedCommand.tags.length > 0) { 192 192 - // For tags, we want to show both existing and new tags in preview 193 193 - previewChanges.tags = JSON.stringify([ 194 194 - ...parsedCommand.tags 195 195 - ]); 196 196 - } 197 197 - 198 198 - // Return the original todo with preview changes attached 199 199 - return { 200 200 - ...todo, 201 201 - preview: previewChanges, 202 202 - previewMode: true 203 203 - }; 204 204 - } 205 205 - return todo; 119 119 + case "modify": 120 120 + return todos.map((todo) => { 121 121 + if (selectedItems.has(todo.working_id)) { 122 122 + // Create a preview object that contains the original todo and the changes 123 123 + const previewChanges = {}; 124 124 + 125 125 + // Only add fields that are being changed to the preview 126 126 + if (parsedCommand.description) { 127 127 + previewChanges.description = parsedCommand.description; 128 128 + } 129 129 + 130 130 + if (parsedCommand.project) { 131 131 + previewChanges.project = parsedCommand.project; 132 132 + } 133 133 + 134 134 + if (parsedCommand.tags && parsedCommand.tags.length > 0) { 135 135 + // For tags, we want to show both existing and new tags in preview 136 136 + previewChanges.tags = JSON.stringify([ 137 137 + ...parsedCommand.tags 138 138 + ]); 139 139 + } 140 140 + 141 141 + // Return the original todo with preview changes attached 142 142 + return { 143 143 + ...todo, 144 144 + preview: previewChanges, 145 145 + previewMode: true 146 146 + }; 147 147 + } 148 148 + return todo; 206 149 }); 207 207 - 208 208 - 209 150 case "filter": 210 151 // For filter, we don't need preview todos 211 152 return []; 212 212 - 213 153 default: 214 154 return []; 215 155 } ··· 259 199 selection.forEach((sel) => { 260 200 if (sel.type === "id") { 261 201 conditions.push(`id IN ( 262 262 - SELECT id 263 263 - FROM active_todos 264 264 - WHERE working_id IN (${sel.ids.map(() => "?").join(",")}) 265 265 - )`); 202 202 + SELECT id 203 203 + FROM active_todos 204 204 + WHERE working_id IN (${sel.ids.map(() => "?").join(",")}) 205 205 + )`); 266 206 params.push(...sel.ids); 267 207 } else if (sel.type === "tag") { 268 208 // TODO this is busted ··· 291 231 conditions: [], 292 232 params: [], 293 233 }); 294 294 - 295 234 if (conditions.length > 0) { 296 235 const sqlQuery = ` 297 297 - UPDATE todos 298 298 - SET completed = 1 299 299 - WHERE ${conditions.join(" OR ")} 300 300 - `; 301 301 - 236 236 + UPDATE todos 237 237 + SET completed = 1 238 238 + WHERE ${conditions.join(" OR ")} 239 239 + `; 302 240 ctx.db.exec(sqlQuery, params); 303 241 clearSelection(); 304 242 } ··· 313 251 case "modify": 314 252 const editParams = []; 315 253 const updates = []; 316 316 - 317 254 // Handle new values for SET clause 318 255 if (parsed.description && parsed.description.length > 0) { 319 256 updates.push("description = ?"); ··· 325 262 } 326 263 if (parsed.tags && parsed.tags.length > 0) { 327 264 updates.push(`tags = ( 328 328 - SELECT json_group_array(value) 329 329 - FROM ( 330 330 - SELECT DISTINCT value 331 331 - FROM ( 332 332 - SELECT value FROM json_each(tags) 333 333 - UNION 334 334 - SELECT value FROM json_each(json(?)) 335 335 - ) 336 336 - ) 337 337 - )`); 265 265 + SELECT json_group_array(value) 266 266 + FROM ( 267 267 + SELECT DISTINCT value 268 268 + FROM ( 269 269 + SELECT value FROM json_each(tags) 270 270 + UNION 271 271 + SELECT value FROM json_each(json(?)) 272 272 + ) 273 273 + ) 274 274 + )`); 338 275 editParams.push(JSON.stringify(parsed.tags)); 339 276 } 340 340 - 341 277 // Build WHERE clause using only selection conditions 342 342 - 343 278 const { conditions: editConditions, params: selectionParams } = 344 279 parseSelection({ 345 280 selection: parsed.selection, 346 281 conditions: [], 347 282 params: [], 348 283 }); 349 349 - 350 284 if (editConditions.length > 0 && updates.length > 0) { 351 285 const sqlQuery = ` 352 286 UPDATE todos ··· 371 305 break; 372 306 } 373 307 setNewText(""); 308 308 + 374 309 } catch (error) { 375 310 // TODO: 376 311 // This is actually bad ··· 386 321 <SidebarProvider defaultOpen={false} className="h-screen dark"> 387 322 <div className="hidden md:flex flex-col w-full h-svh"> 388 323 <AppSidebar ctx={ctx} 389 389 - filterContext={filterContext} 390 390 - setFilterContext={setFilterContext} /> 324 324 + filterContext={filterContext} 325 325 + setFilterContext={setFilterContext} /> 391 326 <div className="bg-muted h-14 w-full absolute top-0 " /> 392 327 <section className="flex-1 container py-12 h-[calc(100vh-theme(spacing.4))] overflow-hidden relative"> 393 328 <SidebarTrigger className="absolute top-4 left-8 border border-foreground" /> ··· 400 335 {todos.length} items found 401 336 <button 402 337 onClick={() => setFilterContext({})} 403 403 - className="hover:bg-muted rounded-full p-1" 404 404 - > 338 338 + className="hover:bg-muted rounded-full p-1"> 405 339 ✕ 406 340 </button> 407 341 </Badge> ··· 423 357 </div> 424 358 </section> 425 359 </div> 426 426 - 427 360 {/* Mobile view layout - shown only on mobile */} 428 361 <div className="md:hidden flex left-0 w-full h-full flex flex-col items-center pb-2"> 429 362 <AppSidebar ctx={ctx} 430 430 - filterContext={filterContext} 431 431 - setFilterContext={setFilterContext} /> 363 363 + filterContext={filterContext} 364 364 + setFilterContext={setFilterContext} /> 432 365 <section className="flex-1 py-12 h-[calc(100vh-theme(spacing.4))] w-full overflow-hidden relative"> 433 366 <div className="flex-1 bg-muted border-b border-foreground h-14 absolute inset-x-0 top-0 " /> 434 367 <SidebarTrigger className="fixed top-4 border border-foreground left-8" /> ··· 463 396 </div> 464 397 </section> 465 398 </div> 399 399 + 466 400 </SidebarProvider> 467 401 </> 468 402 ); 469 403 } 470 404 471 405 export default App; 406 406 +

+59

mast-react-vite/src/components/ui/sync-status.tsx

reviewed

··· 1 1 + type SyncStatusProps = { 2 2 + status: 'connecting' | 'connected' | 'disconnected' | 'error'; 3 3 + changesSent: number; 4 4 + changesReceived: number; 5 5 + lastError?: string; 6 6 + lastSyncTime?: Date; 7 7 + }; 8 8 + 9 9 + export function SyncStatus({ 10 10 + status, 11 11 + changesSent, 12 12 + changesReceived, 13 13 + lastError, 14 14 + lastSyncTime 15 15 + }: SyncStatusProps) { 16 16 + // Define status colors 17 17 + const statusColors = { 18 18 + connecting: 'bg-yellow-500', 19 19 + connected: 'bg-green-500', 20 20 + disconnected: 'bg-gray-500', 21 21 + error: 'bg-red-500' 22 22 + }; 23 23 + 24 24 + // Format last sync time 25 25 + const formattedTime = lastSyncTime 26 26 + ? new Intl.DateTimeFormat('en-US', { 27 27 + hour: '2-digit', 28 28 + minute: '2-digit', 29 29 + second: '2-digit' 30 30 + }).format(lastSyncTime) 31 31 + : 'Never'; 32 32 + 33 33 + return ( 34 34 + <div className="fixed bottom-4 right-4 p-3 bg-background border border-border rounded-lg shadow-lg z-50 text-sm"> 35 35 + <div className="flex items-center gap-2 mb-1"> 36 36 + <div className={`w-3 h-3 rounded-full ${statusColors[status]}`}></div> 37 37 + <span className="font-medium">Sync: {status}</span> 38 38 + </div> 39 39 + 40 40 + <div className="grid grid-cols-2 gap-x-4 gap-y-1 mt-2 text-xs text-muted-foreground"> 41 41 + <div>Changes sent:</div> 42 42 + <div>{changesSent}</div> 43 43 + 44 44 + <div>Changes received:</div> 45 45 + <div>{changesReceived}</div> 46 46 + 47 47 + <div>Last sync:</div> 48 48 + <div>{formattedTime}</div> 49 49 + 50 50 + {lastError && ( 51 51 + <> 52 52 + <div className="col-span-2 text-red-500 mt-1">Error: {lastError}</div> 53 53 + </> 54 54 + )} 55 55 + </div> 56 56 + </div> 57 57 + ); 58 58 + } 59 59 +

+96

mast-react-vite/src/hooks/use-sync.ts

reviewed

··· 1 1 + import { useState, useEffect, useCallback } from 'react'; 2 2 + 3 3 + interface SyncConfig { 4 4 + dbname: string; 5 5 + room: string; 6 6 + endpoint: string; 7 7 + worker: Worker; 8 8 + } 9 9 + 10 10 + export function useCustomSync({ 11 11 + dbname, 12 12 + room, 13 13 + endpoint, 14 14 + worker 15 15 + }: SyncConfig) { 16 16 + const [status, setStatus] = useState<string>('disconnected'); 17 17 + const [changesSent, setChangesSent] = useState<number>(0); 18 18 + const [changesReceived, setChangesReceived] = useState<number>(0); 19 19 + const [lastError, setLastError] = useState<string | null>(null); 20 20 + const [lastSyncTime, setLastSyncTime] = useState<Date | null>(null); 21 21 + 22 22 + // Initialize sync once on component mount 23 23 + useEffect(() => { 24 24 + if (!worker) return; 25 25 + 26 26 + // Start sync 27 27 + worker.postMessage({ 28 28 + type: 'START_SYNC', 29 29 + dbname, 30 30 + config: { 31 31 + room: room, 32 32 + url: endpoint 33 33 + } 34 34 + }); 35 35 + 36 36 + // Set up message handlers 37 37 + const handleMessage = (event: MessageEvent) => { 38 38 + const { type, dbname: eventDbname, count, error } = event.data; 39 39 + 40 40 + // Only process messages for our database 41 41 + if (eventDbname && eventDbname !== dbname) return; 42 42 + 43 43 + switch (type) { 44 44 + case 'SYNC_CONNECTED': 45 45 + setStatus('connected'); 46 46 + break; 47 47 + case 'SYNC_DISCONNECTED': 48 48 + setStatus('disconnected'); 49 49 + break; 50 50 + case 'SYNC_ERROR': 51 51 + setStatus('error'); 52 52 + setLastError(error); 53 53 + break; 54 54 + case 'CHANGES_SENT': 55 55 + setChangesSent(prev => prev + (count || 0)); 56 56 + setLastSyncTime(new Date()); 57 57 + break; 58 58 + case 'CHANGES_APPLIED': 59 59 + setChangesReceived(prev => prev + (count || 0)); 60 60 + setLastSyncTime(new Date()); 61 61 + break; 62 62 + } 63 63 + }; 64 64 + 65 65 + worker.addEventListener('message', handleMessage); 66 66 + 67 67 + // Clean up when component unmounts 68 68 + return () => { 69 69 + worker.removeEventListener('message', handleMessage); 70 70 + worker.postMessage({ 71 71 + type: 'STOP_SYNC', 72 72 + dbname 73 73 + }); 74 74 + }; 75 75 + }, [dbname, room, endpoint, worker]); 76 76 + 77 77 + // Function to trigger sync manually 78 78 + const syncChanges = useCallback(() => { 79 79 + if (worker) { 80 80 + worker.postMessage({ 81 81 + type: 'SYNC_CHANGES', 82 82 + dbname 83 83 + }); 84 84 + } 85 85 + }, [worker, dbname]); 86 86 + 87 87 + return { 88 88 + status, 89 89 + changesSent, 90 90 + changesReceived, 91 91 + lastError, 92 92 + lastSyncTime, 93 93 + syncChanges 94 94 + }; 95 95 + } 96 96 +

+91 -25

mast-react-vite/src/main.tsx

reviewed

··· 1 1 import { createRoot } from "react-dom/client"; 2 2 import "./index.css"; 3 3 import App from "./App.tsx"; 4 4 - 5 4 import { Helmet } from "react-helmet"; 6 5 import { DBAsync } from "@vlcn.io/xplat-api"; 7 6 import initWasm from "@vlcn.io/crsqlite-wasm"; 8 7 import wasmUrl from "@vlcn.io/crsqlite-wasm/crsqlite.wasm?url"; 9 8 import tblrx from "@vlcn.io/rx-tbl"; 10 10 - 11 9 import { SelectionProvider } from "@/contexts/selection-context"; 10 10 + import { useCustomSync } from "@/hooks/use-sync"; 11 11 + import { SyncStatus } from "@/components/ui/sync-status"; 12 12 + 13 13 + // Create a worker instance only once 14 14 + import SyncWorker from './worker/sync-worker.ts?worker'; 15 15 + const syncWorker = new SyncWorker(); 16 16 + 17 17 + // Function to get room ID from URL hash or generate a new one 18 18 + function getRoomId(): string { 19 19 + // Check URL hash for room parameter 20 20 + const hash = window.location.hash.substring(1); 21 21 + const params = new URLSearchParams(hash); 22 22 + const roomFromHash = params.get('room'); 23 23 + 24 24 + if (roomFromHash) { 25 25 + return roomFromHash; 26 26 + } 27 27 + 28 28 + // Check localStorage 29 29 + const storedRoom = localStorage.getItem("room"); 30 30 + if (storedRoom) { 31 31 + return storedRoom; 32 32 + } 33 33 + 34 34 + // Generate a new room ID if none exists 35 35 + const newRoomId = crypto.randomUUID().replaceAll("-", ""); 36 36 + localStorage.setItem("room", newRoomId); 37 37 + 38 38 + // Update URL with the new room ID 39 39 + updateUrlWithRoom(newRoomId); 40 40 + 41 41 + return newRoomId; 42 42 + } 43 43 + 44 44 + // Function to update URL with room ID 45 45 + function updateUrlWithRoom(roomId: string) { 46 46 + const hash = window.location.hash.substring(1); 47 47 + const params = new URLSearchParams(hash); 48 48 + params.set('room', roomId); 49 49 + window.location.hash = params.toString(); 50 50 + } 51 51 + 52 52 + // Custom sync component using your own implementation 53 53 + function CustomSyncComponent({ dbname, roomId }: { dbname: string, roomId: string }) { 54 54 + const endpoint = `ws://localhost:8080/sync`; 55 55 + 56 56 + const syncStats = useCustomSync({ 57 57 + dbname, 58 58 + room: roomId, 59 59 + endpoint, 60 60 + worker: syncWorker 61 61 + }); 62 62 + 63 63 + return <SyncStatus {...syncStats} />; 64 64 + } 12 65 13 66 const initDb = async () => { 14 67 const sqlite = await initWasm(() => wasmUrl); 15 15 - const db: DBAsync = await sqlite.open("todo.db"); 68 68 + 69 69 + // Get or create room ID 70 70 + const roomId = getRoomId(); 71 71 + 72 72 + // Use room ID in database name to isolate data per room 73 73 + const dbname = `todo-${roomId}.db`; 74 74 + const db: DBAsync = await sqlite.open(dbname); 16 75 await db.exec(` 17 17 - CREATE TABLE IF NOT EXISTS todos ( 18 18 - id BLOB PRIMARY KEY NOT NULL, 19 19 - description TEXT, 76 76 + CREATE TABLE IF NOT EXISTS todos ( 77 77 + id BLOB PRIMARY KEY NOT NULL, 78 78 + description TEXT, 20 79 project text, 21 80 tags text, 22 81 due text, 23 82 wait text, 24 83 priority text, 25 84 urgency real, 26 26 - completed INTEGER NOT NULL DEFAULT 0 27 27 - ); 28 28 - SELECT crsql_as_crr('todos'); 29 29 - CREATE VIEW IF NOT EXISTS active_todos AS 30 30 - SELECT 31 31 - id, 32 32 - ROW_NUMBER() OVER (ORDER BY id) as working_id, 33 33 - description, 34 34 - project, 35 35 - tags, 36 36 - due, 37 37 - wait, 38 38 - priority, 39 39 - urgency 40 40 - FROM todos 41 41 - WHERE completed = 0; 42 42 - `); 85 85 + completed INTEGER NOT NULL DEFAULT 0 86 86 + ); 87 87 + SELECT crsql_as_crr('todos'); 88 88 + CREATE VIEW IF NOT EXISTS active_todos AS 89 89 + SELECT 90 90 + id, 91 91 + ROW_NUMBER() OVER (ORDER BY id) as working_id, 92 92 + description, 93 93 + project, 94 94 + tags, 95 95 + due, 96 96 + wait, 97 97 + priority, 98 98 + urgency 99 99 + FROM todos 100 100 + WHERE completed = 0; 101 101 + `); 43 102 const rx = tblrx(db); 44 44 - return { db, rx }; 103 103 + return { db, rx, roomId, dbname }; 45 104 }; 46 105 47 106 const init = async () => { 48 107 const ctx = await initDb(); 108 108 + 109 109 + // Make the worker available globally for debugging 110 110 + (window as any).syncWorker = syncWorker; 111 111 + 49 112 createRoot(document.getElementById("root") as HTMLElement).render( 50 113 <> 51 114 <Helmet> ··· 56 119 </Helmet> 57 120 <main className="h-screen flex flex-col bg-background text-gray-200"> 58 121 <SelectionProvider> 59 59 - <App ctx={ctx} /> 122 122 + {/* Use the custom sync component with the database name and room ID */} 123 123 + <CustomSyncComponent dbname={ctx.dbname} roomId={ctx.roomId} /> 124 124 + <App ctx={ctx} syncWorker={syncWorker} dbname={ctx.dbname} roomId={ctx.roomId} /> 60 125 </SelectionProvider> 61 126 </main> 62 127 </>, ··· 65 130 66 131 document.getElementById("root")!.classList.add("dark"); 67 132 init(); 133 133 +

+361

mast-react-vite/src/worker/sync-worker.ts

reviewed

··· 1 1 + import initWasm from "@vlcn.io/crsqlite-wasm"; 2 2 + import wasmUrl from "@vlcn.io/crsqlite-wasm/crsqlite.wasm?url"; 3 3 + 4 4 + const DEBUG = true; 5 5 + 6 6 + function logDebug(message: string, ...data: any[]) { 7 7 + if (DEBUG) { 8 8 + console.log(`[SyncWorker] ${message}`, ...data); 9 9 + } 10 10 + } 11 11 + 12 12 + function logError(message: string, ...data: any[]) { 13 13 + console.error(`[SyncWorker] ${message}`, ...data); 14 14 + } 15 15 + 16 16 + // Store active connections - only one per database 17 17 + const connections: Record<string, { 18 18 + ws: WebSocket | null; 19 19 + db: any; 20 20 + lastSyncVersion: number; 21 21 + siteId?: Uint8Array; 22 22 + room: string; 23 23 + url: string; 24 24 + isConnecting: boolean; 25 25 + }> = {}; 26 26 + 27 27 + // Handle messages from the main thread 28 28 + self.onmessage = async (event) => { 29 29 + const { type, dbname, config } = event.data; 30 30 + logDebug(`Received message: ${type}`, { dbname, config }); 31 31 + 32 32 + switch (type) { 33 33 + case 'START_SYNC': 34 34 + await startSync(dbname, config); 35 35 + break; 36 36 + case 'STOP_SYNC': 37 37 + stopSync(dbname); 38 38 + break; 39 39 + case 'SYNC_CHANGES': 40 40 + // This is triggered when React tells us changes happened 41 41 + await sendChanges(dbname); 42 42 + break; 43 43 + } 44 44 + }; 45 45 + 46 46 + // Start syncing a database 47 47 + async function startSync(dbname: string, config: { room: string, url: string }) { 48 48 + try { 49 49 + // Check if we already have a connection for this database 50 50 + if (connections[dbname]) { 51 51 + logDebug(`Connection already exists for ${dbname}`); 52 52 + 53 53 + // If the connection is already established or in progress, do nothing 54 54 + if (connections[dbname].ws && 55 55 + (connections[dbname].ws.readyState === WebSocket.OPEN || 56 56 + connections[dbname].ws.readyState === WebSocket.CONNECTING) || 57 57 + connections[dbname].isConnecting) { 58 58 + logDebug(`WebSocket already connected or connecting for ${dbname}`); 59 59 + return; 60 60 + } 61 61 + 62 62 + // If the connection exists but WebSocket is closed, we'll reconnect below 63 63 + } 64 64 + 65 65 + logDebug(`Starting sync for ${dbname} with room ${config.room}`); 66 66 + 67 67 + // Initialize SQLite if needed 68 68 + const sqlite = await initWasm(() => wasmUrl); 69 69 + logDebug(`SQLite initialized`); 70 70 + 71 71 + // Open the database 72 72 + const db = await sqlite.open(dbname); 73 73 + logDebug(`Database opened: ${dbname}`); 74 74 + 75 75 + // Get site ID 76 76 + const siteIdResult = await db.execO<{site_id: Uint8Array}[]>("SELECT crsql_site_id() as site_id"); 77 77 + const siteId = siteIdResult[0].site_id; 78 78 + logDebug(`Site ID: ${Array.from(siteId)}`); 79 79 + 80 80 + // Store or update connection info 81 81 + if (!connections[dbname]) { 82 82 + connections[dbname] = { 83 83 + ws: null, 84 84 + db, 85 85 + lastSyncVersion: 0, 86 86 + siteId, 87 87 + room: config.room, 88 88 + url: config.url, 89 89 + isConnecting: true 90 90 + }; 91 91 + } else { 92 92 + connections[dbname].db = db; 93 93 + connections[dbname].siteId = siteId; 94 94 + connections[dbname].room = config.room; 95 95 + connections[dbname].url = config.url; 96 96 + connections[dbname].isConnecting = true; 97 97 + } 98 98 + 99 99 + // Create WebSocket connection with room in query parameter 100 100 + let wsUrl = config.url; 101 101 + if (!wsUrl.includes('?room=')) { 102 102 + const separator = wsUrl.includes('?') ? '&' : '?'; 103 103 + wsUrl = `${wsUrl}${separator}room=${config.room}`; 104 104 + } 105 105 + logDebug(`Connecting to WebSocket at ${wsUrl}`); 106 106 + 107 107 + const ws = new WebSocket(wsUrl); 108 108 + 109 109 + // Set up WebSocket event handlers 110 110 + ws.onopen = () => { 111 111 + logDebug(`WebSocket connected for ${dbname}`); 112 112 + connections[dbname].isConnecting = false; 113 113 + 114 114 + // Set up change listener using onUpdate 115 115 + db.onUpdate(async () => { 116 116 + logDebug(`Database update detected for ${dbname}`); 117 117 + await sendChanges(dbname); 118 118 + }); 119 119 + 120 120 + // Initial sync - request changes from server 121 121 + logDebug(`Sending initial pull request`); 122 122 + ws.send(JSON.stringify({ 123 123 + type: "pull", 124 124 + room: config.room 125 125 + })); 126 126 + 127 127 + // Notify main thread of connection 128 128 + self.postMessage({ type: 'SYNC_CONNECTED', dbname }); 129 129 + }; 130 130 + 131 131 + ws.onmessage = async (event) => { 132 132 + try { 133 133 + logDebug(`Received WebSocket message: ${event.data.substring(0, 100)}...`); 134 134 + const message = JSON.parse(event.data); 135 135 + 136 136 + if (message.type === "changes" && Array.isArray(message.data)) { 137 137 + logDebug(`Received ${message.data.length} changes from server`); 138 138 + await applyChanges(dbname, message.data); 139 139 + } 140 140 + } catch (error) { 141 141 + logError(`Error processing message:`, error); 142 142 + } 143 143 + }; 144 144 + 145 145 + ws.onerror = (error) => { 146 146 + logError(`WebSocket error:`, error); 147 147 + connections[dbname].isConnecting = false; 148 148 + // Notify main thread of error 149 149 + self.postMessage({ type: 'SYNC_ERROR', dbname, error: 'WebSocket error' }); 150 150 + }; 151 151 + 152 152 + ws.onclose = () => { 153 153 + logDebug(`WebSocket connection closed for ${dbname}`); 154 154 + connections[dbname].isConnecting = false; 155 155 + // Notify main thread of disconnection 156 156 + self.postMessage({ type: 'SYNC_DISCONNECTED', dbname }); 157 157 + }; 158 158 + 159 159 + // Store the WebSocket 160 160 + connections[dbname].ws = ws; 161 161 + 162 162 + } catch (error) { 163 163 + logError(`Error starting sync:`, error); 164 164 + if (connections[dbname]) { 165 165 + connections[dbname].isConnecting = false; 166 166 + } 167 167 + self.postMessage({ type: 'SYNC_ERROR', dbname, error: 'Failed to start sync' }); 168 168 + } 169 169 + } 170 170 + 171 171 + // Stop syncing a database 172 172 + function stopSync(dbname: string) { 173 173 + logDebug(`Stopping sync for ${dbname}`); 174 174 + const connection = connections[dbname]; 175 175 + if (!connection) { 176 176 + logDebug(`No connection found for ${dbname}`); 177 177 + return; 178 178 + } 179 179 + 180 180 + // Close WebSocket 181 181 + if (connection.ws && 182 182 + (connection.ws.readyState === WebSocket.OPEN || 183 183 + connection.ws.readyState === WebSocket.CONNECTING)) { 184 184 + logDebug(`Closing WebSocket for ${dbname}`); 185 185 + connection.ws.close(); 186 186 + connection.ws = null; 187 187 + } 188 188 + 189 189 + // Clean up 190 190 + delete connections[dbname]; 191 191 + logDebug(`Sync stopped for ${dbname}`); 192 192 + 193 193 + // Notify main thread 194 194 + self.postMessage({ type: 'SYNC_STOPPED', dbname }); 195 195 + } 196 196 + 197 197 + // Send changes to the server 198 198 + async function sendChanges(dbname: string) { 199 199 + const connection = connections[dbname]; 200 200 + if (!connection) { 201 201 + logDebug(`Cannot send changes - no connection for ${dbname}`); 202 202 + return; 203 203 + } 204 204 + 205 205 + if (!connection.ws || connection.ws.readyState !== WebSocket.OPEN) { 206 206 + logDebug(`Cannot send changes - WebSocket not open for ${dbname}`); 207 207 + return; 208 208 + } 209 209 + 210 210 + try { 211 211 + logDebug(`Querying for changes since version ${connection.lastSyncVersion}`); 212 212 + 213 213 + // Query for changes since last sync 214 214 + const changes = await connection.db.execA( 215 215 + `SELECT * FROM crsql_changes WHERE db_version > ? AND site_id = crsql_site_id()`, 216 216 + [connection.lastSyncVersion] 217 217 + ); 218 218 + 219 219 + if (changes.length === 0) { 220 220 + logDebug(`No changes to send for ${dbname}`); 221 221 + return; 222 222 + } 223 223 + 224 224 + logDebug(`Found ${changes.length} changes to send for ${dbname}`); 225 225 + 226 226 + // Format changes - with fixed site ID encoding 227 227 + const formattedChanges = changes.map(change => { 228 228 + // Properly encode binary data to base64 229 229 + const encodePK = change[1] instanceof Uint8Array 230 230 + ? btoa(String.fromCharCode.apply(null, change[1])) 231 231 + : btoa(String(change[1])); 232 232 + 233 233 + const encodeSiteID = change[6] instanceof Uint8Array 234 234 + ? btoa(String.fromCharCode.apply(null, change[6])) 235 235 + : btoa(String(change[6])); 236 236 + 237 237 + return { 238 238 + TableName: change[0], 239 239 + PK: encodePK, 240 240 + ColumnName: change[2], 241 241 + Value: change[3], 242 242 + ColVersion: Number(change[4]), 243 243 + DBVersion: Number(change[5]), 244 244 + SiteID: encodeSiteID, 245 245 + CL: Number(change[7]), 246 246 + Seq: Number(change[8]) 247 247 + }; 248 248 + }); 249 249 + 250 250 + // Send changes to server 251 251 + logDebug(`Sending ${changes.length} changes to server`); 252 252 + connection.ws.send(JSON.stringify({ 253 253 + type: "changes", 254 254 + data: formattedChanges 255 255 + })); 256 256 + 257 257 + // Update last sync version 258 258 + const maxVersion = Math.max(...changes.map(c => Number(c[5]))); 259 259 + connection.lastSyncVersion = maxVersion; 260 260 + logDebug(`Updated lastSyncVersion to ${maxVersion}`); 261 261 + 262 262 + // Notify main thread that changes were sent 263 263 + self.postMessage({ type: 'CHANGES_SENT', dbname, count: changes.length }); 264 264 + 265 265 + } catch (error) { 266 266 + logError(`Error sending changes:`, error); 267 267 + self.postMessage({ type: 'SYNC_ERROR', dbname, error: 'Failed to send changes' }); 268 268 + } 269 269 + } 270 270 + 271 271 + // Also update the applyChanges function to handle decoding correctly 272 272 + async function applyChanges(dbname: string, changes: any[]) { 273 273 + const connection = connections[dbname]; 274 274 + if (!connection) { 275 275 + logDebug(`Cannot apply changes - no connection for ${dbname}`); 276 276 + return; 277 277 + } 278 278 + 279 279 + if (changes.length === 0) { 280 280 + logDebug(`No changes to apply for ${dbname}`); 281 281 + return; 282 282 + } 283 283 + 284 284 + try { 285 285 + logDebug(`Applying ${changes.length} changes to ${dbname}`); 286 286 + 287 287 + // Use transaction for applying changes 288 288 + await connection.db.tx(async (tx: any) => { 289 289 + for (const change of changes) { 290 290 + logDebug(`Applying change: ${change.TableName}.${change.ColumnName}`); 291 291 + 292 292 + try { 293 293 + // Safely decode base64 strings to Uint8Array 294 294 + const decodePK = (str: string): Uint8Array => { 295 295 + try { 296 296 + return new Uint8Array( 297 297 + atob(str).split('').map(c => c.charCodeAt(0)) 298 298 + ); 299 299 + } catch (e) { 300 300 + logError(`Error decoding PK: ${e}`); 301 301 + return new Uint8Array(); 302 302 + } 303 303 + }; 304 304 + 305 305 + const decodeSiteID = (str: string): Uint8Array => { 306 306 + try { 307 307 + return new Uint8Array( 308 308 + atob(str).split('').map(c => c.charCodeAt(0)) 309 309 + ); 310 310 + } catch (e) { 311 311 + logError(`Error decoding SiteID: ${e}`); 312 312 + return new Uint8Array(); 313 313 + } 314 314 + }; 315 315 + 316 316 + // Convert the change format back to what CR-SQLite expects 317 317 + const decodedChange = [ 318 318 + change.TableName, 319 319 + decodePK(change.PK), 320 320 + change.ColumnName, 321 321 + change.Value, 322 322 + change.ColVersion, 323 323 + change.DBVersion, 324 324 + decodeSiteID(change.SiteID), 325 325 + change.CL, 326 326 + change.Seq 327 327 + ]; 328 328 + 329 329 + await tx.exec( 330 330 + `INSERT INTO crsql_changes VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`, 331 331 + decodedChange 332 332 + ); 333 333 + } catch (error) { 334 334 + logError(`Error applying individual change:`, error); 335 335 + // Continue with other changes 336 336 + } 337 337 + } 338 338 + }); 339 339 + 340 340 + logDebug(`Changes applied successfully`); 341 341 + 342 342 + // Update last sync version 343 343 + const maxVersion = Math.max(...changes.map(c => c.DBVersion)); 344 344 + if (maxVersion > connection.lastSyncVersion) { 345 345 + connection.lastSyncVersion = maxVersion; 346 346 + logDebug(`Updated lastSyncVersion to ${maxVersion}`); 347 347 + } 348 348 + 349 349 + // Notify main thread that changes were applied 350 350 + self.postMessage({ type: 'CHANGES_APPLIED', dbname, count: changes.length }); 351 351 + 352 352 + } catch (error) { 353 353 + logError(`Error applying changes:`, error); 354 354 + self.postMessage({ type: 'SYNC_ERROR', dbname, error: 'Failed to apply changes' }); 355 355 + } 356 356 + } 357 357 + 358 358 + 359 359 + // Export an empty default to satisfy TypeScript 360 360 + export default {}; 361 361 +

+75

mast-react-vite/src/worker/worker-interface.ts

reviewed

··· 1 1 + export class CustomWorkerInterface { 2 2 + private worker: Worker | null; 3 3 + private messageHandlers: Record<string, Array<(data: any) => void>> = {}; 4 4 + 5 5 + constructor(worker: Worker | null) { 6 6 + this.worker = worker; 7 7 + 8 8 + if (this.worker) { 9 9 + // Set up message handler 10 10 + this.worker.onmessage = (event) => { 11 11 + const { type, ...data } = event.data; 12 12 + 13 13 + // Call all registered handlers for this message type 14 14 + const handlers = this.messageHandlers[type] || []; 15 15 + for (const handler of handlers) { 16 16 + handler(data); 17 17 + } 18 18 + }; 19 19 + } 20 20 + } 21 21 + 22 22 + // Start syncing a database 23 23 + startSync(dbname: string, config: { room: string, url: string }) { 24 24 + if (!this.worker) { 25 25 + console.warn('Worker not initialized'); 26 26 + return; 27 27 + } 28 28 + 29 29 + this.worker.postMessage({ 30 30 + type: 'START_SYNC', 31 31 + dbname, 32 32 + config 33 33 + }); 34 34 + } 35 35 + 36 36 + // Stop syncing a database 37 37 + stopSync(dbname: string) { 38 38 + if (!this.worker) { 39 39 + console.warn('Worker not initialized'); 40 40 + return; 41 41 + } 42 42 + 43 43 + this.worker.postMessage({ 44 44 + type: 'STOP_SYNC', 45 45 + dbname 46 46 + }); 47 47 + } 48 48 + 49 49 + // Send changes to be synced 50 50 + syncChanges(dbname: string) { 51 51 + if (!this.worker) { 52 52 + console.warn('Worker not initialized'); 53 53 + return; 54 54 + } 55 55 + 56 56 + this.worker.postMessage({ 57 57 + type: 'SYNC_CHANGES', 58 58 + dbname 59 59 + }); 60 60 + } 61 61 + 62 62 + // Register a handler for a specific message type from the worker 63 63 + onMessage(type: string, handler: (data: any) => void) { 64 64 + if (!this.messageHandlers[type]) { 65 65 + this.messageHandlers[type] = []; 66 66 + } 67 67 + this.messageHandlers[type].push(handler); 68 68 + 69 69 + // Return unsubscribe function 70 70 + return () => { 71 71 + this.messageHandlers[type] = this.messageHandlers[type].filter(h => h !== handler); 72 72 + }; 73 73 + } 74 74 + } 75 75 +

+3

mast-react-vite/vite.config.ts

reviewed

··· 13 13 "@": path.resolve(__dirname, "./src"), 14 14 }, 15 15 }, 16 16 + worker: { 17 17 + format: 'es', // Use ES modules in workers 18 18 + }, 16 19 })

+299 -137

server/main.go

reviewed

··· 1 1 package main 2 2 3 3 import ( 4 4 - "database/sql" 5 4 "encoding/base64" 5 5 + "database/sql" 6 6 "encoding/json" 7 7 - "fmt" 8 7 "log" 9 8 "net/http" 10 10 - "strconv" 11 11 - "strings" 12 12 - 13 13 - "mast/db" 9 9 + "sync" 10 10 + "time" 14 11 15 12 "github.com/gorilla/websocket" 16 13 sqlite3 "github.com/mattn/go-sqlite3" ··· 22 19 }, 23 20 } 24 21 25 25 - var db *sql.DB 26 26 - var clients = make(map[*websocket.Conn]bool) 22 22 + // Room management 23 23 + type Room struct { 24 24 + clients map[*websocket.Conn]bool 25 25 + mu sync.Mutex 26 26 + } 27 27 + 28 28 + var rooms = make(map[string]*Room) 29 29 + var roomsMu sync.Mutex 30 30 + 31 31 + // Get or create a room 32 32 + func getRoom(roomID string) *Room { 33 33 + roomsMu.Lock() 34 34 + defer roomsMu.Unlock() 35 35 + 36 36 + if room, exists := rooms[roomID]; exists { 37 37 + return room 38 38 + } 39 39 + 40 40 + room := &Room{ 41 41 + clients: make(map[*websocket.Conn]bool), 42 42 + } 43 43 + rooms[roomID] = room 44 44 + return room 45 45 + } 46 46 + 47 47 + // Add client to room 48 48 + func addClientToRoom(roomID string, conn *websocket.Conn) { 49 49 + room := getRoom(roomID) 50 50 + 51 51 + room.mu.Lock() 52 52 + defer room.mu.Unlock() 53 53 + 54 54 + room.clients[conn] = true 55 55 + log.Printf("Client added to room %s. Total clients: %d", roomID, len(room.clients)) 56 56 + } 57 57 + 58 58 + // Remove client from room 59 59 + func removeClientFromRoom(roomID string, conn *websocket.Conn) { 60 60 + roomsMu.Lock() 61 61 + room, exists := rooms[roomID] 62 62 + roomsMu.Unlock() 63 63 + 64 64 + if !exists { 65 65 + return 66 66 + } 67 67 + 68 68 + room.mu.Lock() 69 69 + defer room.mu.Unlock() 70 70 + 71 71 + delete(room.clients, conn) 72 72 + log.Printf("Client removed from room %s. Remaining clients: %d", roomID, len(room.clients)) 73 73 + 74 74 + // Clean up empty rooms 75 75 + if len(room.clients) == 0 { 76 76 + roomsMu.Lock() 77 77 + delete(rooms, roomID) 78 78 + roomsMu.Unlock() 79 79 + log.Printf("Room %s removed (empty)", roomID) 80 80 + } 81 81 + } 82 82 + 83 83 + // Broadcast changes to all clients in a room except sender 84 84 + func broadcastToRoom(roomID string, sender *websocket.Conn, message []byte) { 85 85 + roomsMu.Lock() 86 86 + room, exists := rooms[roomID] 87 87 + roomsMu.Unlock() 88 88 + 89 89 + if !exists { 90 90 + return 91 91 + } 92 92 + 93 93 + room.mu.Lock() 94 94 + defer room.mu.Unlock() 95 95 + 96 96 + for client := range room.clients { 97 97 + if client != sender { 98 98 + client.WriteMessage(websocket.TextMessage, message) 99 99 + } 100 100 + } 101 101 + } 27 102 28 103 func handleWebSocket(w http.ResponseWriter, r *http.Request) { 104 104 + // Extract room ID from query parameters 105 105 + roomID := r.URL.Query().Get("room") 106 106 + if roomID == "" { 107 107 + http.Error(w, "Missing room parameter", http.StatusBadRequest) 108 108 + return 109 109 + } 110 110 + 29 111 conn, err := upgrader.Upgrade(w, r, nil) 30 112 if err != nil { 31 31 - log.Println(err) 113 113 + log.Println("Error upgrading connection:", err) 32 114 return 33 115 } 34 34 - defer conn.Close() 35 35 - 116 116 + 117 117 + // Add client to room 118 118 + addClientToRoom(roomID, conn) 119 119 + 120 120 + // Create database connection for this room 121 121 + dbPath := "./rooms/" + roomID + ".db" 122 122 + db, err := sql.Open("sqlite3_with_extensions", dbPath) 123 123 + if err != nil { 124 124 + log.Println("Error opening database:", err) 125 125 + conn.Close() 126 126 + removeClientFromRoom(roomID, conn) 127 127 + return 128 128 + } 129 129 + defer db.Close() 130 130 + 131 131 + // Ensure the database has the necessary tables 132 132 + setupDatabase(db) 133 133 + 134 134 + // Set up ping/pong to keep connection alive 135 135 + conn.SetPingHandler(func(string) error { 136 136 + conn.WriteControl(websocket.PongMessage, []byte{}, time.Now().Add(time.Second)) 137 137 + return nil 138 138 + }) 139 139 + 140 140 + // Clean up when connection closes 141 141 + defer func() { 142 142 + conn.Close() 143 143 + removeClientFromRoom(roomID, conn) 144 144 + }() 145 145 + 146 146 + // Send initial changes to client 147 147 + sendInitialChanges(conn, db) 148 148 + 149 149 + // Handle incoming messages 36 150 for { 37 151 _, message, err := conn.ReadMessage() 38 152 if err != nil { 39 39 - log.Println(err) 40 40 - return 41 41 - } 42 42 - 43 43 - var msg struct { 44 44 - Type string `json:"type"` 45 45 - Data []interface{} `json:"data"` 153 153 + log.Println("Read error:", err) 154 154 + break 46 155 } 156 156 + 157 157 + var msg map[string]interface{} 47 158 if err := json.Unmarshal(message, &msg); err != nil { 48 48 - log.Println("Error unmarshaling message:", err) 159 159 + log.Println("Error parsing message:", err) 49 160 continue 50 161 } 51 51 - 52 52 - if msg.Type == "changes" { 53 53 - fmt.Println("We're applying changes!!") 54 54 - applyChanges(msg.Data) 162 162 + 163 163 + msgType, ok := msg["type"].(string) 164 164 + if !ok { 165 165 + log.Println("Message missing 'type' field") 166 166 + continue 55 167 } 56 56 - } 57 57 - } 58 58 - 59 59 - func printChange(change interface{}) { 60 60 - // Attempt to type assert and print more details 61 61 - if c, ok := change.(map[string]interface{}); ok { 62 62 - for key, value := range c { 63 63 - fmt.Printf(" %s: %v\n", key, value) 168 168 + 169 169 + switch msgType { 170 170 + case "pull": 171 171 + // Client is requesting changes 172 172 + changes := getChangesFromDB(db, msg) 173 173 + response := map[string]interface{}{ 174 174 + "type": "changes", 175 175 + "data": changes, 176 176 + } 177 177 + responseJSON, _ := json.Marshal(response) 178 178 + conn.WriteMessage(websocket.TextMessage, responseJSON) 179 179 + 180 180 + case "changes": 181 181 + // Client is sending changes 182 182 + if data, ok := msg["data"].([]interface{}); ok { 183 183 + applyChangesToDB(db, data) 184 184 + 185 185 + // Broadcast changes to other clients in the same room 186 186 + broadcastToRoom(roomID, conn, message) 187 187 + } 64 188 } 65 65 - } else { 66 66 - fmt.Printf(" Not a map: %T\n", change) 67 189 } 68 190 } 69 191 70 70 - func printChanges(changes []interface{}) { 71 71 - for i, change := range changes { 72 72 - fmt.Printf("Change %d: %+v\n", i, change) 73 73 - printChange(change) 192 192 + func setupDatabase(db *sql.DB) { 193 193 + // Create todos table if it doesn't exist 194 194 + _, err := db.Exec(` 195 195 + CREATE TABLE IF NOT EXISTS todos ( 196 196 + id BLOB PRIMARY KEY NOT NULL, 197 197 + description TEXT, 198 198 + project text, 199 199 + tags text, 200 200 + due text, 201 201 + wait text, 202 202 + priority text, 203 203 + urgency real, 204 204 + completed INTEGER NOT NULL DEFAULT 0 205 205 + ); 206 206 + SELECT crsql_as_crr('todos'); 207 207 + `) 208 208 + if err != nil { 209 209 + log.Println("Error setting up database:", err) 74 210 } 75 211 } 76 212 77 77 - type CRSQLChange struct { 78 78 - TableName string 79 79 - PK []byte 80 80 - ColumnName string 81 81 - Value any 82 82 - ColVersion int64 83 83 - DBVersion int64 84 84 - SiteID []byte 85 85 - CL int64 86 86 - Seq int64 87 87 - } 88 88 - 89 89 - func mapToCRSQLChange(m map[string]interface{}) (CRSQLChange, error) { 90 90 - var err error 91 91 - change := CRSQLChange{} 92 92 - 93 93 - change.TableName = m["TableName"].(string) 94 94 - change.ColumnName = m["ColumnName"].(string) 95 95 - change.ColVersion = int64(m["ColVersion"].(float64)) 96 96 - change.DBVersion = int64(m["DBVersion"].(float64)) 97 97 - change.CL = int64(m["CL"].(float64)) 98 98 - change.Seq = int64(m["Seq"].(float64)) 99 99 - 100 100 - // Still unsure about this, it can encode anything 101 101 - if m["Value"] != nil { 102 102 - change.Value = m["Value"] 213 213 + func sendInitialChanges(conn *websocket.Conn, db *sql.DB) { 214 214 + // Query all changes 215 215 + rows, err := db.Query("SELECT * FROM crsql_changes") 216 216 + if err != nil { 217 217 + log.Println("Error querying changes:", err) 218 218 + return 103 219 } 104 104 - // Handle []byte fields 105 105 - change.PK, err = decodeBase64ToByteArray(m["PK"].(string)) 106 106 - if err != nil { 107 107 - return change, err 220 220 + defer rows.Close() 221 221 + 222 222 + var changes []map[string]interface{} 223 223 + 224 224 + for rows.Next() { 225 225 + var tableName string 226 226 + var pk []byte 227 227 + var columnName string 228 228 + var value interface{} 229 229 + var colVersion, dbVersion int64 230 230 + var siteID []byte 231 231 + var cl, seq int64 232 232 + 233 233 + if err := rows.Scan(&tableName, &pk, &columnName, &value, &colVersion, &dbVersion, &siteID, &cl, &seq); err != nil { 234 234 + log.Println("Error scanning row:", err) 235 235 + continue 236 236 + } 237 237 + 238 238 + change := map[string]interface{}{ 239 239 + "TableName": tableName, 240 240 + "PK": encodeToBase64(pk), 241 241 + "ColumnName": columnName, 242 242 + "Value": value, 243 243 + "ColVersion": colVersion, 244 244 + "DBVersion": dbVersion, 245 245 + "SiteID": encodeToBase64(siteID), 246 246 + "CL": cl, 247 247 + "Seq": seq, 248 248 + } 249 249 + 250 250 + changes = append(changes, change) 108 251 } 109 109 - change.SiteID, err = decodeBase64ToByteArray(m["SiteID"].(string)) 110 110 - fmt.Println(change.SiteID) 111 111 - if err != nil { 112 112 - return change, err 252 252 + 253 253 + response := map[string]interface{}{ 254 254 + "type": "changes", 255 255 + "data": changes, 113 256 } 114 114 - 115 115 - return change, nil 257 257 + 258 258 + responseJSON, _ := json.Marshal(response) 259 259 + conn.WriteMessage(websocket.TextMessage, responseJSON) 116 260 } 117 261 118 118 - func decodeBase64ToByteArray(encodedString string) ([]uint8, error) { 119 119 - // Decode the base64 string 120 120 - decodedBytes, err := base64.StdEncoding.DecodeString(encodedString) 262 262 + func getChangesFromDB(db *sql.DB, msg map[string]interface{}) []map[string]interface{} { 263 263 + // Implementation to get changes based on client's request 264 264 + // This would typically filter based on site_id and version 265 265 + 266 266 + rows, err := db.Query("SELECT * FROM crsql_changes") 121 267 if err != nil { 122 122 - return nil, fmt.Errorf("error decoding base64: %w", err) 268 268 + log.Println("Error querying changes:", err) 269 269 + return nil 123 270 } 124 124 - 125 125 - // Convert bytes to string 126 126 - decodedString := string(decodedBytes) 127 127 - 128 128 - // Split the string by commas 129 129 - stringSlice := strings.Split(decodedString, ",") 130 130 - 131 131 - // Convert each string to an integer 132 132 - intSlice := make([]uint8, len(stringSlice)) 133 133 - for i, s := range stringSlice { 134 134 - num, err := strconv.Atoi(s) 135 135 - if err != nil { 136 136 - return nil, fmt.Errorf("error converting string to int: %w", err) 271 271 + defer rows.Close() 272 272 + 273 273 + var changes []map[string]interface{} 274 274 + 275 275 + for rows.Next() { 276 276 + var tableName string 277 277 + var pk []byte 278 278 + var columnName string 279 279 + var value interface{} 280 280 + var colVersion, dbVersion int64 281 281 + var siteID []byte 282 282 + var cl, seq int64 283 283 + 284 284 + if err := rows.Scan(&tableName, &pk, &columnName, &value, &colVersion, &dbVersion, &siteID, &cl, &seq); err != nil { 285 285 + log.Println("Error scanning row:", err) 286 286 + continue 137 287 } 138 138 - intSlice[i] = byte(num) 288 288 + 289 289 + change := map[string]interface{}{ 290 290 + "TableName": tableName, 291 291 + "PK": encodeToBase64(pk), 292 292 + "ColumnName": columnName, 293 293 + "Value": value, 294 294 + "ColVersion": colVersion, 295 295 + "DBVersion": dbVersion, 296 296 + "SiteID": encodeToBase64(siteID), 297 297 + "CL": cl, 298 298 + "Seq": seq, 299 299 + } 300 300 + 301 301 + changes = append(changes, change) 139 302 } 140 140 - 141 141 - return intSlice, nil 303 303 + 304 304 + return changes 142 305 } 143 306 144 144 - func applyChanges(changes []interface{}) { 145 145 - tx, _ := db.Begin() 146 146 - for i, change := range changes { 147 147 - fmt.Println("Change number: ", i) 148 148 - fmt.Println(change) 307 307 + func applyChangesToDB(db *sql.DB, changes []interface{}) { 308 308 + tx, err := db.Begin() 309 309 + if err != nil { 310 310 + log.Println("Error starting transaction:", err) 311 311 + return 312 312 + } 313 313 + 314 314 + for _, change := range changes { 149 315 if changeMap, ok := change.(map[string]interface{}); ok { 150 150 - c, err := mapToCRSQLChange(changeMap) 151 151 - if err != nil { 152 152 - fmt.Println("Error converting to CRSQLChange:", err) 153 153 - continue 154 154 - } 155 155 - _, err = tx.Exec(`INSERT INTO crsql_changes VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`, 156 156 - c.TableName, c.PK, c.ColumnName, c.Value, c.ColVersion, c.DBVersion, c.SiteID, c.CL, c.Seq) 316 316 + tableName := changeMap["TableName"].(string) 317 317 + pk, _ := decodeBase64(changeMap["PK"].(string)) 318 318 + columnName := changeMap["ColumnName"].(string) 319 319 + value := changeMap["Value"] 320 320 + colVersion := int64(changeMap["ColVersion"].(float64)) 321 321 + dbVersion := int64(changeMap["DBVersion"].(float64)) 322 322 + siteID, _ := decodeBase64(changeMap["SiteID"].(string)) 323 323 + cl := int64(changeMap["CL"].(float64)) 324 324 + seq := int64(changeMap["Seq"].(float64)) 325 325 + 326 326 + _, err := tx.Exec( 327 327 + `INSERT INTO crsql_changes VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`, 328 328 + tableName, pk, columnName, value, colVersion, dbVersion, siteID, cl, seq, 329 329 + ) 157 330 if err != nil { 158 331 log.Println("Error inserting change:", err) 159 332 } 160 160 - log.Println("We inserted data successfully??") 161 161 - fmt.Println(c) 162 162 - } else { 163 163 - fmt.Println("Not OK to convert change") 164 164 - fmt.Println("Change look like:") 165 165 - printChange(change) 166 166 - continue 167 333 } 168 334 } 169 169 - tx.Commit() 335 335 + 336 336 + if err := tx.Commit(); err != nil { 337 337 + log.Println("Error committing transaction:", err) 338 338 + tx.Rollback() 339 339 + } 170 340 } 171 341 172 172 - func broadcastChanges(sender *websocket.Conn, changes interface{}) { 173 173 - for client := range clients { 174 174 - if client != sender { 175 175 - client.WriteJSON(map[string]interface{}{"type": "changes", "data": changes}) 176 176 - } 177 177 - } 342 342 + func encodeToBase64(data []byte) string { 343 343 + return base64.StdEncoding.EncodeToString(data) 178 344 } 179 179 - func main() { 180 180 - var err error 181 181 - if err != nil { 182 182 - log.Fatal("error:", err) 183 183 - } 184 184 - sql.Register("sqlite3_with_extensions", 185 185 - &sqlite3.SQLiteDriver{ 186 186 - Extensions: []string{ 187 187 - "../db/crsqlite.so", 188 188 - }, 189 189 - }) 190 190 - db, err = sql.Open("sqlite3_with_extensions", "./todo.sqlite") 191 191 - if err != nil { 192 192 - log.Fatal(err) 193 193 - } 194 194 - err = mast.RunMigrations(db) 195 195 - defer db.Close() 345 345 + 346 346 + func decodeBase64(encoded string) ([]byte, error) { 347 347 + return base64.StdEncoding.DecodeString(encoded) 348 348 + } 196 349 197 197 - // Initialize database and tables... 350 350 + func main() { 351 351 + // Register SQLite with CR-SQLite extension 352 352 + sql.Register("sqlite3_with_extensions", &sqlite3.SQLiteDriver{ 353 353 + Extensions: []string{"../db/crsqlite"}, 354 354 + }) 355 355 + 356 356 + // Create directory for room databases 198 357 http.HandleFunc("/sync", handleWebSocket) 358 358 + 359 359 + log.Println("WebSocket server started on :8080") 199 360 log.Fatal(http.ListenAndServe(":8080", nil)) 200 361 } 362 362 +