addition of new dashboard
This commit is contained in:
69
backend/common/config.js
Normal file
69
backend/common/config.js
Normal file
@@ -0,0 +1,69 @@
|
||||
const DEFAULT_KEY_CACHE = { ttlMs: 30 * 60 * 1000, max: 10_000 };
|
||||
|
||||
const config = {
|
||||
debug: false,
|
||||
dashboard: { core: {}, audit: {}, keyCache: { ...DEFAULT_KEY_CACHE } },
|
||||
};
|
||||
|
||||
export async function loadConfig() {
|
||||
if (process.env.DEBUG) config.debug = process.env.DEBUG;
|
||||
if (process.env.POSTGRES_URL) config.postgresUrl = process.env.POSTGRES_URL;
|
||||
if (process.env.PUBLIC_CORE_URL) config.publicCoreUrl = process.env.PUBLIC_CORE_URL;
|
||||
if (process.env.PUBLIC_ARCHIVE_URL) config.publicArchiveUrl = process.env.PUBLIC_ARCHIVE_URL;
|
||||
if (process.env.PUBLIC_CALLBACK_URL) config.publicCallbackUrl = process.env.PUBLIC_CALLBACK_URL;
|
||||
if (process.env.SECURE_SECRET) config.secureSecret = process.env.SECURE_SECRET;
|
||||
if (process.env.DEFAULT_FEDID_URL) config.defaultFedidUrl = process.env.DEFAULT_FEDID_URL;
|
||||
if (process.env.AUTH_MODULES) {
|
||||
config.authModules = {};
|
||||
const authModules = process.env.AUTH_MODULES.split(',').map(a => a.trim());
|
||||
for (const type of authModules) {
|
||||
const authModulePath = `../modules/auth/${type}.js`;
|
||||
const { getModuleConfig } = await import(authModulePath);
|
||||
config.authModules[type] = getModuleConfig();
|
||||
}
|
||||
}
|
||||
config.appModules = {};
|
||||
let appModules = [
|
||||
'core',
|
||||
'archive',
|
||||
];
|
||||
if (process.env.APP_MODULES) {
|
||||
appModules = process.env.APP_MODULES.split(',').map(a => a.trim());
|
||||
}
|
||||
for (const type of appModules) {
|
||||
const appModulePath = `../modules/app/${type}.js`;
|
||||
const { getModuleConfig } = await import(appModulePath);
|
||||
config.appModules[type] = getModuleConfig();
|
||||
}
|
||||
if (process.env.PDP_TYPE) config.pdpType = process.env.PDP_TYPE;
|
||||
if (process.env.PDP_URL) config.pdpUrl = process.env.PDP_URL;
|
||||
if (process.env.INSTANT_QUEUE) config.instantQueue = process.env.INSTANT_QUEUE;
|
||||
|
||||
// Remote agreement-markdown caching: exponential backoff for failed fetches
|
||||
// so an invalid URL isn't retried on every referencing create. Defaults:
|
||||
// 1 minute → 1 day, factor 2, give up after 10 attempts. All overridable.
|
||||
config.agreementCache = {
|
||||
startingDelayMs: Number(process.env.AGREEMENT_CACHE_STARTING_DELAY_MS) || 60_000,
|
||||
maxDelayMs: Number(process.env.AGREEMENT_CACHE_MAX_DELAY_MS) || 86_400_000,
|
||||
maxRetries: Number(process.env.AGREEMENT_CACHE_MAX_RETRIES) || 10,
|
||||
factor: Number(process.env.AGREEMENT_CACHE_BACKOFF_FACTOR) || 2,
|
||||
};
|
||||
|
||||
// Optional federation: source dashboard core/audit data from remote servers
|
||||
// (via their /api/v1/data/dashboard/* API + an API key) instead of locally.
|
||||
// Unset => serve from the local DB (the single-operator default).
|
||||
const keyCacheTtl = Number(process.env.DASHBOARD_KEY_CACHE_TTL_MS);
|
||||
const keyCacheMax = Number(process.env.DASHBOARD_KEY_CACHE_MAX);
|
||||
config.dashboard = {
|
||||
core: { url: process.env.DASHBOARD_CORE_URL || null, key: process.env.DASHBOARD_CORE_KEY || null },
|
||||
audit: { url: process.env.DASHBOARD_AUDIT_URL || null, key: process.env.DASHBOARD_AUDIT_KEY || null },
|
||||
keyCache: {
|
||||
ttlMs: Number.isFinite(keyCacheTtl) ? keyCacheTtl : DEFAULT_KEY_CACHE.ttlMs,
|
||||
max: Number.isFinite(keyCacheMax) && keyCacheMax > 0 ? keyCacheMax : DEFAULT_KEY_CACHE.max,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
export function getConfig() {
|
||||
return config;
|
||||
}
|
||||
14
backend/common/http.js
Normal file
14
backend/common/http.js
Normal file
@@ -0,0 +1,14 @@
|
||||
export const fail = (res, status, code, message) =>
|
||||
res.status(status).json({ error: { code, message } });
|
||||
|
||||
// Requires a logged-in user (401) and turns a throw into a 500 envelope, so a
|
||||
// data-layer failure never masquerades as empty-but-200 data.
|
||||
export const sessionRoute = (label, fn) => async (req, res) => {
|
||||
if (!req.user) return fail(res, 401, "unauthorized", "Not logged in");
|
||||
try {
|
||||
await fn(req, res);
|
||||
} catch (e) {
|
||||
console.error(`${label} ${req.method} ${req.path}:`, e);
|
||||
fail(res, 500, "internal", "Internal error");
|
||||
}
|
||||
};
|
||||
197
backend/common/queue.js
Normal file
197
backend/common/queue.js
Normal file
@@ -0,0 +1,197 @@
|
||||
import { getConfig } from "../common/config.js";
|
||||
import { getPool } from "../db/index.js";
|
||||
import { sleep } from "./sleep.js";
|
||||
import axios from 'axios';
|
||||
|
||||
const backOffUrls = {};
|
||||
|
||||
async function getQueue(client, type, batchSize) {
|
||||
const now = new Date();
|
||||
const sql = `
|
||||
SELECT
|
||||
q.id,
|
||||
qu.id AS queue_url_id,
|
||||
qu.value AS url,
|
||||
q.headers,
|
||||
q.data,
|
||||
q.run_count
|
||||
FROM queue q
|
||||
INNER JOIN queue_type qt ON q.queue_type_id = qt.id
|
||||
INNER JOIN queue_url qu ON q.queue_url_id = qu.id
|
||||
WHERE qt.value = $1
|
||||
AND qu.next_run_ts <= $2
|
||||
ORDER BY qu.next_run_ts ASC, q.id ASC
|
||||
LIMIT $3;
|
||||
`;
|
||||
const data = [
|
||||
type,
|
||||
now,
|
||||
batchSize,
|
||||
];
|
||||
const res = await client.query(sql, data);
|
||||
return res.rows;
|
||||
}
|
||||
|
||||
export async function putQueue(client, type, url, headers, data) {
|
||||
await client.query(`
|
||||
INSERT INTO queue_url (
|
||||
value
|
||||
) VALUES (
|
||||
$1
|
||||
) ON CONFLICT DO NOTHING;
|
||||
`, [
|
||||
url,
|
||||
]);
|
||||
await client.query(`
|
||||
INSERT INTO queue (
|
||||
queue_type_id,
|
||||
queue_url_id,
|
||||
headers,
|
||||
data
|
||||
) VALUES (
|
||||
(
|
||||
SELECT id
|
||||
FROM queue_type
|
||||
WHERE value = $1
|
||||
),
|
||||
(
|
||||
SELECT id
|
||||
FROM queue_url
|
||||
WHERE value = $2
|
||||
),
|
||||
$3,
|
||||
$4
|
||||
);
|
||||
`, [
|
||||
type,
|
||||
url,
|
||||
headers,
|
||||
data
|
||||
]);
|
||||
const config = getConfig();
|
||||
if (config.instantQueue) {
|
||||
await processBatch(client, type);
|
||||
}
|
||||
}
|
||||
|
||||
async function updateQueue(client, item, lastFail) {
|
||||
// Queue back off
|
||||
// ==============
|
||||
// - Start with 30 seconds
|
||||
// - Double for every run
|
||||
// - Max 120 minutes
|
||||
const delay = 30;
|
||||
const maxDelay = 120 * 60;
|
||||
const delaySeconds = Math.min(delay * Math.pow(2, item.run_count - 1), maxDelay);
|
||||
const now = new Date();
|
||||
const nextRunTs = new Date(now.getTime() + delaySeconds * 1000);
|
||||
if (backOffUrls[item.queue_url_id]) {
|
||||
if (backOffUrls[item.queue_url_id] < nextRunTs) {
|
||||
backOffUrls[item.queue_url_id] = nextRunTs;
|
||||
} else {
|
||||
nextRunTs = backOffUrls[item.queue_url_id];
|
||||
}
|
||||
} else {
|
||||
backOffUrls[item.queue_url_id] = nextRunTs;
|
||||
}
|
||||
await client.query(`
|
||||
UPDATE queue_url SET
|
||||
next_run_ts = $1,
|
||||
updated_ts = $2
|
||||
WHERE id = $3
|
||||
`, [
|
||||
backOffUrls[item.queue_url_id].toISOString(),
|
||||
now.toISOString(),
|
||||
item.id,
|
||||
]);
|
||||
await client.query(`
|
||||
UPDATE queue SET
|
||||
run_count = $1,
|
||||
last_fail = $2,
|
||||
updated_ts = $3
|
||||
WHERE id = $4
|
||||
`, [
|
||||
parseInt(item.run_count) + 1,
|
||||
lastFail,
|
||||
now.toISOString(),
|
||||
item.id,
|
||||
]);
|
||||
}
|
||||
|
||||
async function deleteQueue(client, id) {
|
||||
await client.query(`
|
||||
DELETE FROM queue
|
||||
WHERE id = $1
|
||||
`, [
|
||||
id,
|
||||
]);
|
||||
}
|
||||
|
||||
async function processBatch(client, type) {
|
||||
const batchSize = 100;
|
||||
const queueList = await getQueue(client, type, batchSize);
|
||||
const now = new Date();
|
||||
for await (const item of queueList) {
|
||||
if (backOffUrls[item.queue_url_id] && backOffUrls[item.queue_url_id] > now)
|
||||
continue;
|
||||
try {
|
||||
let entry = 'unknown';
|
||||
if (item.data.audit) {
|
||||
const uuid = item.data.audit.eventId ?? item.data.audit.agreementId;
|
||||
const auditType = item.data.audit.eventId ? 'event' : 'agreement';
|
||||
entry = `${auditType}:${uuid}`
|
||||
}
|
||||
console.log(`${type.toUpperCase()} - ${entry} (${item.run_count})`);
|
||||
let result;
|
||||
try {
|
||||
result = await axios.post(
|
||||
item.url,
|
||||
item.data,
|
||||
{
|
||||
headers: item.headers,
|
||||
}
|
||||
)
|
||||
} catch (e) {
|
||||
result = e
|
||||
}
|
||||
if (result?.status === 200) {
|
||||
await deleteQueue(client, item.id);
|
||||
} else {
|
||||
let lastFail = `${result.status}`;
|
||||
if (result.response.data.error) {
|
||||
lastFail += ` - ${result.response.data.error}`
|
||||
}
|
||||
await updateQueue(client, item, lastFail);
|
||||
}
|
||||
} catch (e) {
|
||||
console.error(e);
|
||||
}
|
||||
}
|
||||
return queueList.length;
|
||||
}
|
||||
|
||||
async function processQueue(client, type) {
|
||||
while (true) {
|
||||
const count = await processBatch(client, type);
|
||||
if (count === 0) {
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async function watchQueue(client, type) {
|
||||
const repeat = 30 * 1000; // seconds
|
||||
while (true) {
|
||||
await processQueue(client, type);
|
||||
await sleep(repeat);
|
||||
}
|
||||
}
|
||||
|
||||
export async function watchAudits() {
|
||||
const client = await getPool();
|
||||
const config = getConfig();
|
||||
if (!config.instantQueue) {
|
||||
await watchQueue(client, 'audit');
|
||||
await client.release();
|
||||
}
|
||||
}
|
||||
1
backend/common/sleep.js
Normal file
1
backend/common/sleep.js
Normal file
@@ -0,0 +1 @@
|
||||
export const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
|
||||
Reference in New Issue
Block a user