openvibers/openvibe.news Commits
Await what the transactions start (PostgreSQL pass follow-up): discussion, ingest and the API called async helpers inside sync callbacks without awaiting them. A sync callback that calls async code returns before the work is done, so writes could land outside or after their transaction. npm test 16/16.
Changes
@@@ -45,7 +45,7 @@ function createDiscussion({ store, publication, community, log = console }) {async function watch(storyId) {if (!community.enabled) return;const was = (await status(await storyById.get(storyId))).open;-- db.afterCommit(() => { flushing = flushing.then(() => follow(storyId, was)).catch((err) => log.warn(`[News] comment thread sync failed: ${err.message}`)); });++ db.afterCommit(() => { flushing = flushing.then(async () => await follow(storyId, was)).catch((err) => log.warn(`[News] comment thread sync failed: ${err.message}`)); });}return {@@@ -133,12 +133,12 @@ function createIngest({ store, config, clusters, outbox, sources, discussion = n* { outcome: created|updated|removed|unchanged|ignored|rejected, item?, reason? }.*/async function apply(item, { origin = 'pull' } = {}) {-- return db.tx(async () => {++ return await db.tx(async () => {if (!item || typeof item.id !== 'string') return { outcome: 'rejected', reason: 'item without id' };if (item.category && item.category !== config.sources.category) return { outcome: 'ignored', reason: `category ${item.category}` };const revision = Number.isInteger(item.revision) ? item.revision : 1;const cur = await q.bySourcesId.get(item.id);-- if (item.removed) return applyRemoval({ sourcesItemId: item.id, reason: item.removed.reason || 'removed upstream', revision, origin });++ if (item.removed) return await applyRemoval({ sourcesItemId: item.id, reason: item.removed.reason || 'removed upstream', revision, origin });if (cur && cur.sources_revision >= revision) return { outcome: 'unchanged', item: cur };const n = await normalise(item);if (!n.headline) {@@@ -180,7 +180,7 @@ function createIngest({ store, config, clusters, outbox, sources, discussion = n/** A Sources removal (takedown, licence, error): sticky; the stored summary is dropped. */async function applyRemoval({ sourcesItemId, reason, revision = null, origin = 'webhook' }) {-- return db.tx(async () => {++ return await db.tx(async () => {const cur = await q.bySourcesId.get(sourcesItemId);if (!cur) return { outcome: 'ignored', reason: 'never ingested' };if (cur.status === 'removed') return { outcome: 'unchanged', item: cur };@@@ -200,7 +200,7 @@ function createIngest({ store, config, clusters, outbox, sources, discussion = n/** sources.fetch.failed for a news source: recorded and relayed; nothing else changes. */async function upstreamFailure(payload) {-- return db.tx(async () => {++ return await db.tx(async () => {if (payload.category && payload.category !== config.sources.category) return { outcome: 'ignored' };await recordRun({ origin: 'sources', state: String(payload.state || 'failed').slice(0, 40), source_key: payload.source_key || null, error_code: payload.error_code || null,detail: `run ${payload.run_id || '?'}${payload.http_status ? ` HTTP ${payload.http_status}` : ''}${payload.consecutive_failures ? `, ${payload.consecutive_failures} consecutive failures` : ''}` });@@@ -178,7 +178,7 @@ function createApi(ctx) {}));// guards: the capability guard, then the route's per-actor budget (both before the body is read).-- const write = (method, path, guards, fn, status = 200) => router[method](path, guards, jsonBody, run(async (req) => fn(req, await mustStory(req), req.body || {}), status));++ const write = (method, path, guards, fn, status = 200) => router[method](path, guards, jsonBody, run(async (req) => await fn(req, await mustStory(req), req.body || {}), status));write('post', '/stories/:id/revisions', [guard('news.story.revise'), B('news.story.revise')], async (req, story, b) => {const out = await stories.revise(req.viewer, story, {@@@ -225,7 +225,7 @@ function createApi(ctx) {router.get('/clusters', guard('news.cluster.read'), run(async (req) => {editorRead(req);const limit = Math.min(Math.max(parseInt(req.query.limit, 10) || 50, 1), 200);-- return { clusters: await Promise.all((await clusters.recent({ limit })).map((c) => clusterDto(c))) };++ return { clusters: await Promise.all((await clusters.recent({ limit })).map(async (c) => await clusterDto(c))) };}));router.get('/clusters/:id', guard('news.cluster.read'), run(async (req) => {editorRead(req);@@@ -236,7 +236,7 @@ function createApi(ctx) {router.post('/clusters/audit/:auditId/reverse', guard('news.cluster.manage'), B('news.cluster.manage'), jsonBody, run(async (req) => {access.requireEditor(config, req.viewer);const out = await clusters.reverse(req.viewer, req.params.auditId, { reason: (req.body || {}).reason });-- return { audit: out.audit, clusters: await Promise.all(out.clusters.map((c) => clusterDto(c))) };++ return { audit: out.audit, clusters: await Promise.all(out.clusters.map(async (c) => await clusterDto(c))) };}));router.post('/clusters/:id/merge', guard('news.cluster.manage'), B('news.cluster.manage'), jsonBody, run(async (req) => {access.requireEditor(config, req.viewer);
@@ -45,7 +45,7 @@ function createDiscussion({ store, publication, community, log = console }) {
@@ -45,7 +45,7 @@ function createDiscussion({ store, publication, community, log = console }) {
async function watch(storyId) {
async function watch(storyId) {
if (!community.enabled) return;
if (!community.enabled) return;
const was = (await status(await storyById.get(storyId))).open;
const was = (await status(await storyById.get(storyId))).open;
- db.afterCommit(() => { flushing = flushing.then(() => follow(storyId, was)).catch((err) => log.warn(`[News] comment thread sync failed: ${err.message}`)); });
+ db.afterCommit(() => { flushing = flushing.then(async () => await follow(storyId, was)).catch((err) => log.warn(`[News] comment thread sync failed: ${err.message}`)); });
}
}
return {
return {
@@ -133,12 +133,12 @@ function createIngest({ store, config, clusters, outbox, sources, discussion = n
@@ -133,12 +133,12 @@ function createIngest({ store, config, clusters, outbox, sources, discussion = n
* { outcome: created|updated|removed|unchanged|ignored|rejected, item?, reason? }.
* { outcome: created|updated|removed|unchanged|ignored|rejected, item?, reason? }.
*/
*/
async function apply(item, { origin = 'pull' } = {}) {
async function apply(item, { origin = 'pull' } = {}) {
- return db.tx(async () => {
+ return await db.tx(async () => {
if (!item || typeof item.id !== 'string') return { outcome: 'rejected', reason: 'item without id' };
if (!item || typeof item.id !== 'string') return { outcome: 'rejected', reason: 'item without id' };
if (item.category && item.category !== config.sources.category) return { outcome: 'ignored', reason: `category ${item.category}` };
if (item.category && item.category !== config.sources.category) return { outcome: 'ignored', reason: `category ${item.category}` };
const revision = Number.isInteger(item.revision) ? item.revision : 1;
const revision = Number.isInteger(item.revision) ? item.revision : 1;
const cur = await q.bySourcesId.get(item.id);
const cur = await q.bySourcesId.get(item.id);
- if (item.removed) return applyRemoval({ sourcesItemId: item.id, reason: item.removed.reason || 'removed upstream', revision, origin });
+ if (item.removed) return await applyRemoval({ sourcesItemId: item.id, reason: item.removed.reason || 'removed upstream', revision, origin });
if (cur && cur.sources_revision >= revision) return { outcome: 'unchanged', item: cur };
if (cur && cur.sources_revision >= revision) return { outcome: 'unchanged', item: cur };
const n = await normalise(item);
const n = await normalise(item);
if (!n.headline) {
if (!n.headline) {
@@ -180,7 +180,7 @@ function createIngest({ store, config, clusters, outbox, sources, discussion = n
@@ -180,7 +180,7 @@ function createIngest({ store, config, clusters, outbox, sources, discussion = n
/** A Sources removal (takedown, licence, error): sticky; the stored summary is dropped. */
/** A Sources removal (takedown, licence, error): sticky; the stored summary is dropped. */
async function applyRemoval({ sourcesItemId, reason, revision = null, origin = 'webhook' }) {
async function applyRemoval({ sourcesItemId, reason, revision = null, origin = 'webhook' }) {
- return db.tx(async () => {
+ return await db.tx(async () => {
const cur = await q.bySourcesId.get(sourcesItemId);
const cur = await q.bySourcesId.get(sourcesItemId);
if (!cur) return { outcome: 'ignored', reason: 'never ingested' };
if (!cur) return { outcome: 'ignored', reason: 'never ingested' };
if (cur.status === 'removed') return { outcome: 'unchanged', item: cur };
if (cur.status === 'removed') return { outcome: 'unchanged', item: cur };
@@ -200,7 +200,7 @@ function createIngest({ store, config, clusters, outbox, sources, discussion = n
@@ -200,7 +200,7 @@ function createIngest({ store, config, clusters, outbox, sources, discussion = n
/** sources.fetch.failed for a news source: recorded and relayed; nothing else changes. */
/** sources.fetch.failed for a news source: recorded and relayed; nothing else changes. */
async function upstreamFailure(payload) {
async function upstreamFailure(payload) {
- return db.tx(async () => {
+ return await db.tx(async () => {
if (payload.category && payload.category !== config.sources.category) return { outcome: 'ignored' };
if (payload.category && payload.category !== config.sources.category) return { outcome: 'ignored' };
await recordRun({ origin: 'sources', state: String(payload.state || 'failed').slice(0, 40), source_key: payload.source_key || null, error_code: payload.error_code || null,
await recordRun({ origin: 'sources', state: String(payload.state || 'failed').slice(0, 40), source_key: payload.source_key || null, error_code: payload.error_code || null,
detail: `run ${payload.run_id || '?'}${payload.http_status ? ` HTTP ${payload.http_status}` : ''}${payload.consecutive_failures ? `, ${payload.consecutive_failures} consecutive failures` : ''}` });
detail: `run ${payload.run_id || '?'}${payload.http_status ? ` HTTP ${payload.http_status}` : ''}${payload.consecutive_failures ? `, ${payload.consecutive_failures} consecutive failures` : ''}` });
@@ -178,7 +178,7 @@ function createApi(ctx) {
@@ -178,7 +178,7 @@ function createApi(ctx) {
}));
}));
// guards: the capability guard, then the route's per-actor budget (both before the body is read).
// guards: the capability guard, then the route's per-actor budget (both before the body is read).
- const write = (method, path, guards, fn, status = 200) => router[method](path, guards, jsonBody, run(async (req) => fn(req, await mustStory(req), req.body || {}), status));
+ const write = (method, path, guards, fn, status = 200) => router[method](path, guards, jsonBody, run(async (req) => await fn(req, await mustStory(req), req.body || {}), status));
write('post', '/stories/:id/revisions', [guard('news.story.revise'), B('news.story.revise')], async (req, story, b) => {
write('post', '/stories/:id/revisions', [guard('news.story.revise'), B('news.story.revise')], async (req, story, b) => {
const out = await stories.revise(req.viewer, story, {
const out = await stories.revise(req.viewer, story, {
@@ -225,7 +225,7 @@ function createApi(ctx) {
@@ -225,7 +225,7 @@ function createApi(ctx) {
router.get('/clusters', guard('news.cluster.read'), run(async (req) => {
router.get('/clusters', guard('news.cluster.read'), run(async (req) => {
editorRead(req);
editorRead(req);
const limit = Math.min(Math.max(parseInt(req.query.limit, 10) || 50, 1), 200);
const limit = Math.min(Math.max(parseInt(req.query.limit, 10) || 50, 1), 200);
- return { clusters: await Promise.all((await clusters.recent({ limit })).map((c) => clusterDto(c))) };
+ return { clusters: await Promise.all((await clusters.recent({ limit })).map(async (c) => await clusterDto(c))) };
}));
}));
router.get('/clusters/:id', guard('news.cluster.read'), run(async (req) => {
router.get('/clusters/:id', guard('news.cluster.read'), run(async (req) => {
editorRead(req);
editorRead(req);
@@ -236,7 +236,7 @@ function createApi(ctx) {
@@ -236,7 +236,7 @@ function createApi(ctx) {
router.post('/clusters/audit/:auditId/reverse', guard('news.cluster.manage'), B('news.cluster.manage'), jsonBody, run(async (req) => {
router.post('/clusters/audit/:auditId/reverse', guard('news.cluster.manage'), B('news.cluster.manage'), jsonBody, run(async (req) => {
access.requireEditor(config, req.viewer);
access.requireEditor(config, req.viewer);
const out = await clusters.reverse(req.viewer, req.params.auditId, { reason: (req.body || {}).reason });
const out = await clusters.reverse(req.viewer, req.params.auditId, { reason: (req.body || {}).reason });
- return { audit: out.audit, clusters: await Promise.all(out.clusters.map((c) => clusterDto(c))) };
+ return { audit: out.audit, clusters: await Promise.all(out.clusters.map(async (c) => await clusterDto(c))) };
}));
}));
router.post('/clusters/:id/merge', guard('news.cluster.manage'), B('news.cluster.manage'), jsonBody, run(async (req) => {
router.post('/clusters/:id/merge', guard('news.cluster.manage'), B('news.cluster.manage'), jsonBody, run(async (req) => {
access.requireEditor(config, req.viewer);
access.requireEditor(config, req.viewer);