Add database upgrade file 31.

Resolve "Add line change time log pseudoevent"

Closes #136

See merge request wgp/dougal/software!45

etc/db/upgrades/upgrade31-v0.4.4-add-event_log_changes-function.sql

@@ -0,0 +1,104 @@
+-- Add event_log_changes function
+--
+-- New schema version: 0.4.4
+--
+-- ATTENTION:
+--
+-- ENSURE YOU HAVE BACKED UP THE DATABASE BEFORE RUNNING THIS SCRIPT.
+--
+--
+-- NOTE: This upgrade affects all schemas in the database.
+-- NOTE: Each application starts a transaction, which must be committed
+-- or rolled back.
+--
+-- This adds a function event_log_changes which returns the subset of
+-- events from event_log_full which have been modified on or after a
+-- given timestamp.
+--
+-- To apply, run as the dougal user:
+--
+-- psql <<EOF
+-- \i $THIS_FILE
+-- COMMIT;
+-- EOF
+--
+-- NOTE: It can be applied multiple times without ill effect.
+--
+
+BEGIN;
+
+CREATE OR REPLACE PROCEDURE pg_temp.show_notice (notice text) AS $$
+BEGIN
+	RAISE NOTICE '%', notice;
+END;
+$$ LANGUAGE plpgsql;
+
+CREATE OR REPLACE PROCEDURE pg_temp.upgrade_survey_schema (schema_name text) AS $outer$
+BEGIN
+
+	RAISE NOTICE 'Updating schema %', schema_name;
+	-- We need to set the search path because some of the trigger
+	-- functions reference other tables in survey schemas assuming
+	-- they are in the search path.
+	EXECUTE format('SET search_path TO %I,public', schema_name);
+
+	CREATE OR REPLACE FUNCTION event_log_changes(ts0 timestamptz)
+		RETURNS SETOF event_log_full
+		LANGUAGE sql
+		AS $$
+		SELECT *
+		FROM event_log_full
+		WHERE lower(validity) > ts0 OR upper(validity) IS NOT NULL AND upper(validity) > ts0
+		ORDER BY lower(validity);
+	$$;
+
+END;
+$outer$ LANGUAGE plpgsql;
+
+CREATE OR REPLACE PROCEDURE pg_temp.upgrade () AS $outer$
+DECLARE
+	row RECORD;
+	current_db_version TEXT;
+BEGIN
+
+	SELECT value->>'db_schema' INTO current_db_version FROM public.info WHERE key = 'version';
+
+	IF current_db_version >= '0.4.4' THEN
+		RAISE EXCEPTION
+			USING MESSAGE='Patch already applied';
+	END IF;
+
+	IF current_db_version != '0.4.3' THEN
+		RAISE EXCEPTION
+			USING MESSAGE='Invalid database version: ' || current_db_version,
+				HINT='Ensure all previous patches have been applied.';
+	END IF;
+
+	FOR row IN
+		SELECT schema_name FROM information_schema.schemata
+		WHERE schema_name LIKE 'survey_%'
+		ORDER BY schema_name
+	LOOP
+		CALL pg_temp.upgrade_survey_schema(row.schema_name);
+	END LOOP;
+END;
+$outer$ LANGUAGE plpgsql;
+
+CALL pg_temp.upgrade();
+
+CALL pg_temp.show_notice('Cleaning up');
+DROP PROCEDURE pg_temp.upgrade_survey_schema (schema_name text);
+DROP PROCEDURE pg_temp.upgrade ();
+
+CALL pg_temp.show_notice('Updating db_schema version');
+INSERT INTO public.info VALUES ('version', '{"db_schema": "0.4.4"}')
+ON CONFLICT (key) DO UPDATE
+	SET value = public.info.value || '{"db_schema": "0.4.4"}' WHERE public.info.key = 'version';
+
+
+CALL pg_temp.show_notice('All done. You may now run "COMMIT;" to persist the changes');
+DROP PROCEDURE pg_temp.show_notice (notice text);
+
+--
+--NOTE Run `COMMIT;` now if all went well
+--

lib/www/server/api/index.js

@@ -181,6 +181,9 @@ app.map({
 		post: [ mw.auth.access.write, mw.event.post ],
 		put: [ mw.auth.access.write, mw.event.put ],
 		delete: [ mw.auth.access.write, mw.event.delete ],
+		'changes/:since': {
+			get: [ mw.event.changes ]
+		},
 		// TODO Rename -/:sequence → sequence/:sequence
 		'-/:sequence/': { // NOTE: We need to avoid conflict with the next endpoint ☹
 			get: [ mw.event.sequence.get ],

lib/www/server/api/middleware/event/changes.js

@@ -0,0 +1,14 @@
+
+const { event } = require('../../../lib/db');
+
+const json = async function (req, res, next) {
+	try {
+		const response = await event.changes(req.params.project, req.params.since, req.query);
+		res.status(200).send(response);
+		next();
+	} catch (err) {
+		next(err);
+	}
+};
+
+module.exports = json;

lib/www/server/api/middleware/event/index.js

@@ -6,5 +6,6 @@ module.exports = {
 	post: require('./post'),
 	put: require('./put'),
 	patch: require('./patch'),
-	delete: require('./delete')
+	delete: require('./delete'),
+	changes: require('./changes')
 }

lib/www/server/events/handlers/report-line-change-time.js

@@ -35,7 +35,7 @@ class ReportLineChangeTime {
 		while (this.queue.length > 0) {
 			if (this.queue[0].isPending) {
 				DEBUG("Queue busy");
-				setImmediate(() => this.processQueue());
+				setTimeout(() => this.processQueue(), 1000); // We're not in a hurry
 				return;
 			}
 
@@ -46,7 +46,7 @@ class ReportLineChangeTime {
 			const forward = (cur.old?.labels?.includes("LGSP") || cur.new?.labels?.includes("LGSP"));
 
 			if (!projectId) {
-				WARNING("No projectID found in event", cur);
+				throw {message: "No projectID found in event", cur};
 				return;
 			}
 
@@ -267,6 +267,7 @@ class ReportLineChangeTime {
 	}
 
 	async run (data) {
+		DEBUG("Seen", data);
 		if (!data || data.channel !== "event") {
 			return;
 		}
@@ -285,10 +286,12 @@ class ReportLineChangeTime {
 
 		if (this.queue.length < ReportLineChangeTime.MAX_QUEUE_SIZE) {
 
-			this.queue.push({
-				...data.payload,
+			const item = {
+				...structuredClone(data.payload),
 				isPending: this.queue.length,
-			});
+			};
+			DEBUG("Queueing", item);
+			this.queue.push(item);
 
 		} else {
 			ALERT("ReportLineChangeTime queue full at", this.queue.length);

lib/www/server/lib/db/event/changes.js

@@ -0,0 +1,61 @@
+const { setSurvey } = require('../connection');
+const { replaceMarkers } = require('../../utils');
+
+function parseValidity (row) {
+	if (row.validity) {
+		const rx = /^(.)("([\d :.+-]+)")?,("([\d :.+-]+)")?([\]\)])$/;
+		const m = row.validity.match(rx);
+		row.validity = [ m[1], m[3], m[5], m[6] ];
+	}
+	return row;
+}
+
+function transform (row) {
+	if (row.validity[2]) {
+		return {
+			uid: row.uid,
+			id: row.id,
+			is_deleted: true
+		}
+	} else {
+		row.is_deleted = false;
+		row.has_edits = row.id != row.uid;
+		row.modified_on = row.validity[1];
+		delete row.uid;
+		delete row.validity;
+		return row;
+	}
+}
+
+function unique (rows) {
+	const o = {};
+	rows.forEach(row => o[row.id] = row);
+	return Object.values(o);
+}
+
+/**
+ * Get the event change history from a given epoch (ts0),
+ * for all events.
+ */
+async function changes (projectId, ts0, opts = {}) {
+
+	if (!projectId || !ts0) {
+		throw {status: 400, message: "Invalid request" };
+		return;
+	}
+
+	const client = await setSurvey(projectId);
+
+	const text = `
+		SELECT *
+		FROM event_log_changes($1);
+	`;
+
+	const res = await client.query(text, [ts0]);
+	client.release();
+	return opts.unique
+		? unique(res.rows.map(i => transform(replaceMarkers(parseValidity(i)))))
+		: res.rows.map(i => transform(replaceMarkers(parseValidity(i))));
+}
+
+module.exports = changes;

lib/www/server/lib/db/event/index.js

@@ -5,5 +5,6 @@ module.exports = {
 	post: require('./post'),
 	put: require('./put'),
 	patch: require('./patch'),
-	del: require('./delete')
+	del: require('./delete'),
+	changes: require('./changes')
 }

lib/www/server/spec/openapi.yaml

@@ -180,6 +180,16 @@ components:
             required: true
             example: 14707
 
+        Since:
+            description: Starting epoch
+            name: since
+            in: path
+            schema:
+                type: string
+                format: date-time
+            required: true
+            example: 1970-01-01T00:00:00Z
+
         QueryLimit:
             description: Maximum number of results to return
             name: limit
@@ -206,6 +216,16 @@ components:
                 pattern: "(([^\\s,;:]+)(\\s*[,;:\\s]\\s*)?)+"
             example: "line,point,tstamp"
 
+        Unique:
+            description: |
+                Return unique results. Any value at all represents `true`.
+            name: unique
+            in: query
+            schema:
+                type: string
+                pattern: ".+"
+            example: "t"
+
 
     schemas:
         Duration:
@@ -602,14 +622,26 @@ components:
                                 Flag to indicate that this event is read-only. It cannot be edited by the user or deleted. Typically this concerns system-generated events such as QC results or midnight shots.
                     additionalProperties: true
 
-        EventAbstract:
-            allOf:
-                -
+        EventIDAbstract:
             type: object
             properties:
                 id:
                     type: number
                     description: Event ID.
+
+
+        EventUIDAbstract:
+            type: object
+            properties:
+                uid:
+                    type: number
+                    description: Event instance unique ID. When an event is modified, the new entry acquires a different `uid` while keeping the same `id` as the original event.
+
+
+        EventAbstract:
+            allOf:
+                -
+                    $ref: "#/components/schemas/EventIDAbstract"
                 -
                     $ref: "#/components/schemas/EventNew"
 
@@ -659,6 +691,47 @@ components:
                                 * The third element is either an ISO-8601 timestamp or `null`. The latter indicates +∞. These are the events returned by endpoints that do not concern themselves with event history.
                                 * The fourth element is one of `]` or `)`. As before, it indicates either an open or closed interval.
 
+        EventChangesIsDeletedAbstract:
+            type: object
+            properties:
+                is_deleted:
+                    type: boolean
+                    description: >
+                        Flag to indicate whether this event or event instance (depending on the presence of a `uid` attribute) has been deleted.
+
+
+        EventChangesModified:
+            description: An event modification.
+            allOf:
+                -
+                    $ref: "#/components/schemas/EventAbstract"
+                -
+                    $ref: "#/components/schemas/EventChangesIsDeletedAbstract"
+
+        EventChangesDeleted:
+            description: |
+                Identification of a deleted event or event instance.
+
+                **Note:** the details of the deleted event are not included, only its `id` and `uid`.
+            allOf:
+                -
+                    $ref: "#/components/schemas/EventIDAbstract"
+                -
+                    $ref: "#/components/schemas/EventUIDAbstract"
+                -
+                    $ref: "#/components/schemas/EventChangesIsDeletedAbstract"
+
+        EventChanges:
+            description: List of event changes since the given epoch.
+            type: array
+            items:
+                anyOf:
+                    -
+                        $ref: "#/components/schemas/EventChangesDeleted"
+                    -
+                        $ref: "#/components/schemas/EventChangesModified"
+
+
         SeisExportEntryFSP:
             type: object
             properties:
@@ -1382,6 +1455,31 @@ paths:
                     $ref: "#/components/responses/401"
 
 
+    /project/{project}/changes/{since}:
+        get:
+            summary: Get event change history since epoch.
+            tags: [ "log" ]
+            security:
+                - BearerAuthGuest: []
+                - CookieAuthGuest: []
+            parameters:
+                - $ref: "#/components/parameters/Project"
+                - $ref: "#/components/parameters/Since"
+                - $ref: "#/components/parameters/Unique"
+            responses:
+                "200":
+                    description: List of project event changes. If `unique` is given, only the latest version of each event will be returned, otherwise the entire modification history is given, potentially including the same event `id` multiple times.
+                    content:
+                        application/json:
+                            schema:
+                                type: array
+                                items:
+                                    $ref: "#/components/schemas/EventChanges"
+
+                "401":
+                    $ref: "#/components/responses/401"
+
+
     /project/{project}/label:
         get:
             summary: Get project labels.