More sync

main.go

@@ -124,8 +124,8 @@ func main() { // {{{
 
 	http.HandleFunc("/user/authenticate", AuthManager.AuthenticationHandler)
 
-	http.HandleFunc("/sync/from_server/{sequence}/{offset}", authenticated(actionSyncNode))
+	http.HandleFunc("/sync/from_server/{sequence}/{offset}", authenticated(actionSyncFromServer))
-	http.HandleFunc("/sync/to_server/{client}", authenticated(actionSyncNode))
+	http.HandleFunc("/sync/to_server", authenticated(actionSyncToServer))
 
 	http.HandleFunc("/node/retrieve/{uuid}", authenticated(actionNodeRetrieve))
 
@@ -242,7 +242,7 @@ func pageSync(w http.ResponseWriter, r *http.Request) { // {{{
 	}
 } // }}}
 
-func actionSyncNode(w http.ResponseWriter, r *http.Request) { // {{{
+func actionSyncFromServer(w http.ResponseWriter, r *http.Request) { // {{{
 	// The purpose of the Client UUID is to avoid
 	// sending nodes back once again to a client that
 	// just created or modified it.
@@ -296,6 +296,27 @@ func actionNodeRetrieve(w http.ResponseWriter, r *http.Request) { // {{{
 		"Node": node,
 	})
 } // }}}
+func actionSyncToServer(w http.ResponseWriter, r *http.Request) { // {{{
+	user := getUser(r)
+
+	body, _ := r.GetBody()
+	data, _ := io.ReadAll(body)
+	var request struct {
+		ClientUUID string
+		NodeData   string
+	}
+	err := json.Unmarshal(data, &request)
+	if err != nil {
+		httpError(w, err)
+		return
+	}
+
+	db.Exec(`CALL add_nodes(%d, %s::jsonb)`, user.ID, request.ClientUUID, request.NodeData)
+
+	responseData(w, map[string]interface{}{
+		"OK": true,
+	})
+} // }}}
 
 func createNewUser(username string) { // {{{
 	reader := bufio.NewReader(os.Stdin)

sql/00006.sql

@@ -0,0 +1,16 @@
+CREATE TABLE public.node_history (
+	id serial4 NOT NULL,
+	user_id int4 NOT NULL,
+	uuid bpchar(36) NOT NULL,
+	parents varchar[] NULL,
+	created timestamptz NOT NULL,
+	updated timestamptz NOT NULL,
+	name varchar(256) NOT NULL,
+	"content" text NOT NULL,
+	content_encrypted text NOT NULL,
+	markdown bool DEFAULT false NOT NULL,
+	client bpchar(36) DEFAULT ''::bpchar NOT NULL,
+	CONSTRAINT node_history_pk PRIMARY KEY (id),
+	CONSTRAINT node_history_user_fk FOREIGN KEY (user_id) REFERENCES public."user"(id) ON DELETE RESTRICT ON UPDATE RESTRICT
+);
+CREATE INDEX node_history_uuid_idx ON public.node USING btree (uuid);

sql/00007.sql

@@ -0,0 +1,162 @@
+CREATE TYPE json_ancestor_array as ("Ancestors" varchar[]);
+
+
+CREATE OR REPLACE PROCEDURE add_nodes(p_user_id int4, p_client_uuid varchar, p_nodes jsonb)
+LANGUAGE PLPGSQL AS $$
+
+DECLARE
+	node_data     jsonb;
+	node_updated  timestamptz;
+	db_updated    timestamptz;
+	db_uuid       bpchar;
+	db_client     bpchar;
+	db_client_seq int;
+	node_uuid     bpchar;
+
+BEGIN
+	RAISE NOTICE '--------------------------';
+	FOR node_data IN SELECT * FROM jsonb_array_elements(p_nodes)
+  	LOOP
+		node_uuid = (node_data->>'UUID')::bpchar;
+		node_updated = (node_data->>'Updated')::timestamptz;
+
+		/* Retrieve the current modified timestamp for this node from the database. */
+		SELECT
+			uuid, updated, client, client_sequence
+		INTO
+			db_uuid, db_updated, db_client, db_client_seq
+		FROM public."node"
+		WHERE
+			user_id = p_user_id AND
+			uuid = node_uuid;
+
+		/* Is the node not in database? It needs to be created. */
+		IF db_uuid IS NULL THEN
+			RAISE NOTICE '01 New node %', node_uuid;
+			INSERT INTO public."node" (
+				user_id, "uuid", parent_uuid, created, updated,
+				"name", "content", markdown, "content_encrypted",
+				client, client_sequence
+			)
+			VALUES(
+				p_user_id,
+				node_uuid,
+				(node_data->>'ParentUUID')::bpchar,
+				(node_data->>'Created')::timestamptz,
+				(node_data->>'Updated')::timestamptz,
+				(node_data->>'Name')::varchar,
+				(node_data->>'Content')::text,
+				(node_data->>'Markdown')::bool,
+				'', /* content_encrypted */
+				p_client_uuid,
+				(node_data->>'ClientSequence')::int
+			);
+			CONTINUE;
+		END IF;
+
+		
+		/* The client could send a specific node again if it didn't receive the OK from this procedure before. */
+		IF db_updated = node_updated AND db_client = p_client_uuid AND db_client_seq = (node_data->>'ClientSequence')::int THEN
+			RAISE NOTICE '04, already recorded, %, %', db_client, db_client_seq;
+			CONTINUE;
+		END IF;
+
+		/* Determine if the incoming node data is to go into history or replace the current node. */
+		IF db_updated > node_updated THEN
+			RAISE NOTICE '02 DB newer, % > % (%))', db_updated, node_updated, node_uuid;
+			/* Incoming node is going straight to history since it is older than the current node. */
+			INSERT INTO node_history(
+				user_id, "uuid", parents, created, updated,
+				"name", "content", markdown, "content_encrypted",
+				client, client_sequence
+			)
+			VALUES(
+				p_user_id,
+				node_uuid,
+				(jsonb_populate_record(null::json_ancestor_array, node_data))."Ancestors",
+				(node_data->>'Created')::timestamptz,
+				(node_data->>'Updated')::timestamptz,
+				(node_data->>'Name')::varchar,
+				(node_data->>'Content')::text,
+				(node_data->>'Markdown')::bool,
+				'', /* content_encrypted */
+				p_client_uuid,
+				(node_data->>'ClientSequence')::int
+			)
+			ON CONFLICT (client, client_sequence)
+			DO NOTHING;
+		ELSE
+			RAISE NOTICE '03 Client newer, % > % (%, %)', node_updated, db_updated, node_uuid, (node_data->>'ClientSequence');
+			/* Incoming node is newer and will replace the current node.
+			 *
+			 * The current node is copied to the node_history table and then modified in place
+			 * with the incoming data. */
+			INSERT INTO node_history(
+				user_id, "uuid", parents,
+				created, updated, "name", "content", markdown, "content_encrypted",
+				client, client_sequence
+			)
+			SELECT
+				user_id,
+				"uuid",
+				(
+					WITH RECURSIVE nodes AS (
+						SELECT
+							uuid,
+							COALESCE(parent_uuid, '') AS parent_uuid,
+							name,
+							0 AS depth
+						FROM node
+						WHERE
+							uuid = node_uuid
+
+						UNION
+						
+						SELECT
+							n.uuid,
+							COALESCE(n.parent_uuid, '') AS parent_uuid,
+							n.name,
+							nr.depth+1 AS depth
+						FROM node n
+						INNER JOIN nodes nr ON n.uuid = nr.parent_uuid
+					)
+					SELECT ARRAY (
+						SELECT name
+						FROM nodes
+						ORDER BY depth DESC
+						OFFSET 1 /* discard itself */
+					)
+				),
+				created,
+				updated,
+				name,
+				content,
+				markdown,
+				content_encrypted,
+				client,
+				client_sequence
+			FROM public."node"
+			WHERE
+				user_id = p_user_id AND
+				uuid = node_uuid
+			ON CONFLICT (client, client_sequence)
+			DO NOTHING;
+
+			/* Current node in database is updated with incoming data. */
+			UPDATE public."node"
+			SET
+				updated = (node_data->>'Updated')::timestamptz,
+				updated_seq = nextval('node_updates'),
+				name = (node_data->>'Name')::varchar,
+				content = (node_data->>'Content')::text,
+				markdown = (node_data->>'Markdown')::bool,
+				client = p_client_uuid,
+				client_sequence = (node_data->>'ClientSequence')::int
+			WHERE
+				user_id = p_user_id AND
+				uuid = node_uuid;
+		END IF;
+
+  	END LOOP;
+END
+$$;

sql/00008.sql

@@ -0,0 +1,2 @@
+ALTER TABLE node         ADD COLUMN Client_sequence int NULL;
+ALTER TABLE node_history ADD COLUMN Client_sequence int NULL;

sql/00009.sql

@@ -0,0 +1 @@
+CREATE UNIQUE INDEX node_history_client_idx ON public.node_history (client,client_sequence);

static/js/node_store.mjs

@@ -50,7 +50,7 @@ export class NodeStore {
 							break
 
 						case 5:
-							sendQueue = db.createObjectStore('send_queue', { keyPath: ['UUID', 'Updated'] })
+							sendQueue = db.createObjectStore('send_queue', { keyPath: 'ClientSequence', autoIncrement: true })
 							sendQueue.createIndex('updated', 'Updated', { unique: false })
 							break
 

static/js/sync.mjs

@@ -92,10 +92,10 @@ export class Sync {
 			const nodesToSend = await nodeStore.sendQueue.retrieve(100)
 			const clientUUID = await nodeStore.getAppState('client_uuid')
 			const request = {
-				Nodes: nodesToSend,
+				NodeData: JSON.stringify(nodesToSend),
 				ClientUUID: clientUUID.value,
 			}
-			res = await API.query('POST', `/sync/from_server/${oldMax}/${offset}`, { ClientUUID: clientUUID.value })
+			res = await API.query('POST', `/sync/to_server/${oldMax}/${offset}`, request)
 			console.log(res)
 
 		} catch (e) {