mirror of
https://gitlab.com/wgp/dougal/software.git
synced 2025-12-06 11:37:08 +00:00
319 lines
9.2 KiB
JavaScript
319 lines
9.2 KiB
JavaScript
const { schema2pid } = require('../../lib/db/connection');
|
||
const { event, project } = require('../../lib/db');
|
||
const { withinValidity } = require('../../lib/utils/ranges');
|
||
const unique = require('../../lib/utils/unique');
|
||
const { ALERT, ERROR, WARNING, NOTICE, INFO, DEBUG } = require('DOUGAL_ROOT/debug')(__filename);
|
||
|
||
class ReportLineChangeTime {
|
||
/* Data may come much faster than we can process it, so we put it
|
||
* in a queue and process it at our own pace.
|
||
*
|
||
* The run() method fills the queue with the necessary data and then
|
||
* calls processQueue().
|
||
*
|
||
* The processQueue() method looks takes the first two elements in
|
||
* the queue and processes them if they are not already being taken
|
||
* care of by a previous processQueue() call – this will happen when
|
||
* data is coming in faster than it can be processed.
|
||
*
|
||
* If the processQueue() call is the first to see the two bottommost
|
||
* two elements, it will process them and, when finished, it will set
|
||
* the `isPending` flag of the bottommost element to `false`, thus
|
||
* letting the next call know that it has work to do.
|
||
*
|
||
* If the queue was empty, run() will set the `isPending` flag of its
|
||
* first element to a falsy value, thus bootstrapping the process.
|
||
*/
|
||
static MAX_QUEUE_SIZE = 125000;
|
||
|
||
queue = [];
|
||
|
||
author = `*${this.constructor.name}*`;
|
||
|
||
async processQueue () {
|
||
DEBUG("Queue length", this.queue.length)
|
||
while (this.queue.length > 0) {
|
||
if (this.queue[0].isPending) {
|
||
DEBUG("Queue busy");
|
||
setImmediate(() => this.processQueue());
|
||
return;
|
||
}
|
||
|
||
const cur = this.queue.shift();
|
||
const next = this.queue[0];
|
||
const projectId = cur.pid;
|
||
// Are we being called because of a LGSP or because of a FGSP?
|
||
const forward = (cur.old?.labels?.includes("LGSP") || cur.new?.labels?.includes("LGSP"));
|
||
|
||
if (!projectId) {
|
||
WARNING("No projectID found in event", cur);
|
||
return;
|
||
}
|
||
|
||
async function getConfiguration (projectId) {
|
||
return await project.configuration.get(projectId);
|
||
}
|
||
|
||
|
||
async function getLineChangeTime (data, forward = false) {
|
||
if (forward) {
|
||
const ospEvents = await event.list(projectId, {label: "FGSP"});
|
||
// DEBUG("ospEvents", ospEvents);
|
||
const osp = ospEvents.filter(i => i.tstamp > data.tstamp).pop();
|
||
DEBUG("fsp", osp);
|
||
// DEBUG("data", data);
|
||
|
||
if (osp) {
|
||
DEBUG("lineChangeTime", osp.tstamp - data.tstamp);
|
||
return { lineChangeTime: osp.tstamp - data.tstamp, osp };
|
||
}
|
||
} else {
|
||
const ospEvents = await event.list(projectId, {label: "LGSP"});
|
||
// DEBUG("ospEvents", ospEvents);
|
||
const osp = ospEvents.filter(i => i.tstamp < data.tstamp).shift();
|
||
DEBUG("lsp", osp);
|
||
// DEBUG("data", data);
|
||
|
||
if (osp) {
|
||
DEBUG("lineChangeTime", data.tstamp - osp.tstamp);
|
||
return { lineChangeTime: data.tstamp - osp.tstamp, osp };
|
||
}
|
||
}
|
||
}
|
||
|
||
function parseInterval (dt) {
|
||
const daySeconds = (dt/1000) % 86400;
|
||
const d = Math.floor((dt/1000) / 86400);
|
||
const dateObject = new Date(null);
|
||
dateObject.setSeconds(daySeconds);
|
||
const [ h, m, s ] = dateObject.toISOString().slice(11, 19).split(":").map(Number);
|
||
return {d, h, m, s};
|
||
}
|
||
|
||
function formatInterval (i) {
|
||
let str = "";
|
||
for (let [k, v] of Object.entries(i)) {
|
||
if (v) {
|
||
str += " " + v + " " + k;
|
||
}
|
||
}
|
||
return str.trim();
|
||
}
|
||
|
||
// const deleteStaleEvents = async (id) => {
|
||
// if (id) {
|
||
// DEBUG("Will delete lct events related to record(s)", id);
|
||
//
|
||
// const jpq = Array.isArray(id)
|
||
// ? `$."${this.author}".parents ? (${[ ...new Set(id)].map(i => "@ == "+i).join(" || ")})`
|
||
// : `$."${this.author}".parents ? (@ == ${id})`;
|
||
// DEBUG("jpq", jpq);
|
||
// const staleEvents = await event.list(projectId, {jpq});
|
||
// DEBUG(staleEvents.length ?? 0, "events to delete");
|
||
// for (let staleEvent of staleEvents) {
|
||
// DEBUG("Deleting event", staleEvent.id);
|
||
// await event.del(projectId, staleEvent.id);
|
||
// }
|
||
// }
|
||
// }
|
||
|
||
const deleteStaleEvents = async (seq) => {
|
||
if (seq) {
|
||
DEBUG("Will delete lct events related to sequence(s)", seq);
|
||
|
||
const jpq = `$."${this.author}"`;
|
||
|
||
const opts = {jpq};
|
||
|
||
if (Array.isArray(seq)) {
|
||
opts.sequences = unique(seq).filter(i => !!i);
|
||
} else {
|
||
opts.sequence = seq;
|
||
}
|
||
|
||
const staleEvents = await event.list(projectId, opts);
|
||
DEBUG(staleEvents.length ?? 0, "events to delete");
|
||
for (let staleEvent of staleEvents) {
|
||
DEBUG(`Deleting event id ${staleEvent.id} (seq = ${staleEvent.sequence}, point = ${staleEvent.point})`);
|
||
await event.del(projectId, staleEvent.id);
|
||
}
|
||
}
|
||
}
|
||
|
||
const createLineChangeTimeEvents = async (lineChangeTime, data, osp) => {
|
||
|
||
const events = [];
|
||
const cfg = (await project.configuration.get(projectId));
|
||
const nlcd = cfg?.production?.nominalLineChangeDuration * 60*1000; // m → ms
|
||
DEBUG("nlcd", nlcd);
|
||
if (nlcd && lineChangeTime > nlcd) {
|
||
const excess = lineChangeTime-nlcd;
|
||
const excessString = formatInterval(parseInterval(excess));
|
||
DEBUG("excess", excess, excessString);
|
||
|
||
// ref: The later of the two events
|
||
const ref = forward ? osp : data;
|
||
const payload = {
|
||
// tstamp: new Date(ref.tstamp-1),
|
||
sequence: ref.sequence,
|
||
point: ref.point,
|
||
remarks: `_Nominal line change duration exceeded by ${excessString}_`,
|
||
labels: [ "Nav", "Prod" ],
|
||
meta: {
|
||
auto: true,
|
||
author: this.author,
|
||
[this.author]: {
|
||
parents: [
|
||
data.id,
|
||
osp.id
|
||
],
|
||
type: "excess",
|
||
value: excess
|
||
}
|
||
}
|
||
}
|
||
|
||
events.push(payload);
|
||
DEBUG("Created line change duration exceeded event", projectId, payload);
|
||
}
|
||
|
||
|
||
const lctString = formatInterval(parseInterval(lineChangeTime));
|
||
|
||
// ref: The later of the two events
|
||
const ref = forward ? osp : data;
|
||
const payload = {
|
||
// tstamp: new Date(ref.tstamp-1),
|
||
sequence: ref.sequence,
|
||
point: ref.point,
|
||
remarks: `Line change time: ${lctString}`,
|
||
labels: [ "Nav", "Prod" ],
|
||
meta: {
|
||
auto: true,
|
||
author: this.author,
|
||
[this.author]: {
|
||
parents: [
|
||
data.id,
|
||
osp.id
|
||
],
|
||
type: "lineChangeTime",
|
||
value: lineChangeTime
|
||
}
|
||
}
|
||
};
|
||
|
||
events.push(payload);
|
||
DEBUG("Created line change duration event", projectId, payload);
|
||
|
||
return events;
|
||
}
|
||
|
||
const maybePostEvent = async (projectId, payload) => {
|
||
DEBUG("Posting event", projectId, payload);
|
||
await event.post(projectId, payload);
|
||
}
|
||
|
||
try {
|
||
// DEBUG("Previous", prev);
|
||
DEBUG("Current", cur);
|
||
DEBUG("Forward search", forward);
|
||
|
||
// We have these scenarios to consider:
|
||
// INSERT:
|
||
// `old` will be NULL
|
||
// Add event with line change time:
|
||
// - match validity with `new`
|
||
// - meta.ReportLineChangeTime.link refers to new.uid (or new.id?)
|
||
// UPDATE:
|
||
// `old` is not NULL
|
||
// `new` is not NULL
|
||
// Delete previous event from event_log (not event_log_full)
|
||
// Add event with line change time:
|
||
// - match validity with `new`
|
||
// - meta.ReportLineChangeTime.link refers to new.uid (or new.id?)
|
||
// DELETE:
|
||
// `old` is not NULL
|
||
// `new` will be NULL
|
||
// Delete previous event from event_log (not event_log_full)
|
||
|
||
await deleteStaleEvents([cur.old?.sequence, cur.new?.sequence]);
|
||
|
||
if (cur.operation == "INSERT") {
|
||
// NOTE: UPDATE on the event_log view translates to one UPDATE plus one INSERT
|
||
// on event_log_full, so we don't need to worry about UPDATE here.
|
||
const data = cur.new;
|
||
DEBUG("INSERT seen: will add lct events related to ", data.id);
|
||
|
||
if (withinValidity(data.validity)) {
|
||
DEBUG("Event within validity period", data.validity, new Date());
|
||
|
||
data.tstamp = new Date(data.tstamp);
|
||
const { lineChangeTime, osp } = await getLineChangeTime(data, forward);
|
||
|
||
if (lineChangeTime) {
|
||
|
||
const events = await createLineChangeTimeEvents(lineChangeTime, data, osp);
|
||
|
||
if (events?.length) {
|
||
DEBUG("Deleting other events for sequence", events[0].sequence);
|
||
await deleteStaleEvents(events[0].sequence);
|
||
}
|
||
|
||
for (let payload of events) {
|
||
await maybePostEvent(projectId, payload);
|
||
}
|
||
}
|
||
} else {
|
||
DEBUG("Event outside of validity range", data.validity, "lct events not inserted");
|
||
}
|
||
|
||
}
|
||
|
||
|
||
// Processing of this shot has already been completed.
|
||
// The queue can now move forward.
|
||
} catch (err) {
|
||
ERROR("ReportLineChangeTime Error")
|
||
ERROR(err);
|
||
} finally {
|
||
if (next) {
|
||
next.isPending = false;
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
async run (data) {
|
||
if (!data || data.channel !== "event") {
|
||
return;
|
||
}
|
||
|
||
if (!(data.payload?.new?.labels) && !(data.payload?.old?.labels)) {
|
||
return;
|
||
}
|
||
|
||
const n = data.payload.new;
|
||
const o = data.payload.old;
|
||
|
||
if (!n?.labels?.includes("FGSP") && !o?.labels?.includes("FGSP") &&
|
||
!n?.labels?.includes("LGSP") && !o?.labels?.includes("LGSP")) {
|
||
return;
|
||
}
|
||
|
||
if (this.queue.length < ReportLineChangeTime.MAX_QUEUE_SIZE) {
|
||
|
||
this.queue.push({
|
||
...data.payload,
|
||
isPending: this.queue.length,
|
||
});
|
||
|
||
} else {
|
||
ALERT("ReportLineChangeTime queue full at", this.queue.length);
|
||
}
|
||
|
||
await this.processQueue();
|
||
}
|
||
}
|
||
|
||
module.exports = ReportLineChangeTime;
|