mosaicstack/stack
25
0 Fork 1
Code Issues 404 Pull Requests 13 Packages Projects Releases 15 Wiki Activity
Files
cli-parity
stack/agents/rocko/work/queue-49/r1/build.patch
T

1288 lines
70 KiB
Diff
Raw Permalink Blame History

diff --git a/packages/bus/README.md b/packages/bus/README.md
index 4694e08b5..2973590c5 100644
--- a/packages/bus/README.md
+++ b/packages/bus/README.md
@@ -447,15 +447,47 @@ scope probes and rotation/expiry notifications; S2 never probes real services.
## Schema and verification
-`schema.sql` is unchanged v3b, SHA-256
-`179ffe356d4ff19a49b5ebad39b6c6bfd7771deb8e1b7c55b5e746f039d69e65`.
-It has eight tables, **36** triggers and five views. The prototype's displayed
-35 is after deliberately dropping a guard. Source digest and a trusted full
-sqlite_master digest are checked at open, alongside schema version `3b`, WAL,
+`schema.sql` is v3c, SHA-256
+`7587ac1df813e5148fdb73dc4733ec8064e6a142e4647f0da71be009a73eebdb`.
+It has nine tables, 40 triggers and five views. Source digest and a trusted full
+sqlite_master digest are checked at open, alongside schema version `3c`, WAL,
quick_check and foreign-key integrity. Missing metadata refuses; it never
reinitializes an existing file. Full synchronous WAL transactions either commit
-or roll back. SQL UPDATE, DELETE and INSERT OR REPLACE refuse on every table.
-No migration or pruning command ships here.
+or roll back. SQL UPDATE, DELETE and INSERT OR REPLACE refuse on evidence tables.
+
+`task_comment_cursors` holds append-only `(business, task_id, last_comment)` rows.
+The latest sequence wins, and a database guard requires the ID to increase.
+Trusted sync commits a new cursor and its comment event in one transaction;
+agent snapshots and the public task digest/projection are unchanged. An empty
+poll writes no cursor. Growth is at most one row per observed new comment, and
+usually one per batch. IDs include bots, but bot comments never contribute to
+the external-comment count. Task IDs survive project moves within a business.
+
+Normal startup refuses a 3b database. The explicit offline command is:
+
+```sh
+node packages/bus/src/migrate-schema.mjs /absolute/dataRoot
+```
+
+Mos runs it only with the bus stopped, in an approved downtime/re-pin window,
+after a consistent backup. Development seats run it only against disposable
+fixtures. It takes the same exclusive writer lock, checks the pinned source
+schema and metadata, WAL, timestamps and integrity, then runs named migration
+steps in order. Each step atomically adds its DDL, changes only schema version
+and digest metadata, restores the metadata update guard, and verifies its target
+digest. The 3b→3c step also records an immutable `comment_cursor_epoch`. Prior
+steps stay pinned under `src/migrations/`; later versions append a step. A completed
+step remains valid if a later step refuses; rerunning resumes from that version.
+Already-current migration is a validated no-op. Missing stores refuse.
+
+Before a task has a cursor, comments at or after the epoch minus 60 seconds are
+eligible (rounded down to tracker seconds). This deliberately favors a possible
+cutover duplicate over dropping comments with up to one minute of tracker clock
+lag. Older historical comments are baselined without events; a tracker clock
+more than a minute behind may still lose initial comments, and a clock ahead can
+replay already reported comments once at migration. Once an ID is stored, only
+IDs govern freshness, including after a long outage. No other event semantics,
+tracker client, or pruning command changes here.
Run `node --test packages/bus/tests/*.test.mjs`. Tests use private disposable
directories, fixture token files, fake process tables, real local child brokers,
diff --git a/packages/bus/schema.sql b/packages/bus/schema.sql
index 6aef89004..e68424252 100644
--- a/packages/bus/schema.sql
+++ b/packages/bus/schema.sql
@@ -209,3 +209,21 @@ CREATE VIEW urgent_inbox AS
FROM decisions d
WHERE d.class = 'gated' AND d.blocking = 1
AND NOT EXISTS (SELECT 1 FROM decision_events x WHERE x.decision = d.id AND x.op IN ('resolved','withdrawn','expired'));
+CREATE TABLE task_comment_cursors (
+ seq INTEGER PRIMARY KEY AUTOINCREMENT,
+ business TEXT NOT NULL,
+ task_id INTEGER NOT NULL CHECK (task_id > 0),
+ last_comment INTEGER NOT NULL CHECK (last_comment > 0)
+) STRICT;
+CREATE INDEX task_comment_cursors_latest ON task_comment_cursors (business, task_id, seq);
+CREATE TRIGGER task_comment_cursors_no_update BEFORE UPDATE ON task_comment_cursors
+ BEGIN SELECT RAISE(ABORT, 'task_comment_cursors is append-only'); END;
+CREATE TRIGGER task_comment_cursors_no_delete BEFORE DELETE ON task_comment_cursors
+ BEGIN SELECT RAISE(ABORT, 'task_comment_cursors is append-only'); END;
+CREATE TRIGGER task_comment_cursors_no_replace BEFORE INSERT ON task_comment_cursors
+ WHEN EXISTS (SELECT 1 FROM task_comment_cursors WHERE seq = NEW.seq)
+ BEGIN SELECT RAISE(ABORT, 'task_comment_cursors is append-only'); END;
+CREATE TRIGGER task_comment_cursors_forward BEFORE INSERT ON task_comment_cursors
+ WHEN NEW.last_comment <= COALESCE((SELECT last_comment FROM task_comment_cursors
+ WHERE business = NEW.business AND task_id = NEW.task_id ORDER BY seq DESC LIMIT 1), 0)
+ BEGIN SELECT RAISE(ABORT, 'comment cursor must advance'); END;
diff --git a/packages/bus/src/broker.mjs b/packages/bus/src/broker.mjs
index f10e11b37..7c281d635 100644
--- a/packages/bus/src/broker.mjs
+++ b/packages/bus/src/broker.mjs
@@ -599,23 +599,35 @@ export class Broker {
}
// Trusted S3 adapter only. With a cap it records a role's own completed write: the external call
// already happened, so this records and does not re-authorize. Without one it records sync reads.
- recordTask({ cap = null, business, snapshots = [], events = [] }) {
+ recordTask({ cap = null, business, snapshots = [], events = [], commentCursors = [] }) {
const s = cap === null ? { business, role: null, run: null } : this.#session(cap);
this.#business(business);
if (cap !== null && (s.human || s.reader || s.business !== business)) fail('agent-required');
- if (!Array.isArray(snapshots) || !Array.isArray(events)) fail('invalid-request');
+ if (!Array.isArray(snapshots) || !Array.isArray(events) || !Array.isArray(commentCursors)) fail('invalid-request');
+ if (cap !== null && commentCursors.length) fail('agent-required');
const kinds = cap === null ? POLL_KINDS : SELF_KINDS;
- this.#secretCheck({ snapshots, events });
- return this.#store.transaction(() => ({
- snapshots: snapshots.map((x) => this.#snapshot(s, x)),
- events: events.map((e) => {
- keys(e, ['kind', 'body', 'subject'], ['kind', 'body']);
- if (!kinds.has(e.kind)) fail('reserved-event');
- if (!object(e.body)) fail('invalid-request');
- return this.#event(s, e.kind, e.body, e.subject ?? null);
- }),
- }));
+ this.#secretCheck({ snapshots, events, commentCursors });
+ return this.#store.transaction(() => {
+ for (const x of commentCursors) {
+ keys(x, ['task_id', 'last_comment'], ['task_id', 'last_comment']);
+ if (![x.task_id, x.last_comment].every((n) => Number.isSafeInteger(n) && n > 0)) fail('invalid-request');
+ this.#store.run(
+ 'INSERT INTO task_comment_cursors(business,task_id,last_comment) VALUES(?,?,?)',
+ business, x.task_id, x.last_comment,
+ );
+ }
+ return {
+ snapshots: snapshots.map((x) => this.#snapshot(s, x)),
+ events: events.map((e) => {
+ keys(e, ['kind', 'body', 'subject'], ['kind', 'body']);
+ if (!kinds.has(e.kind)) fail('reserved-event');
+ if (!object(e.body)) fail('invalid-request');
+ return this.#event(s, e.kind, e.body, e.subject ?? null);
+ }),
+ };
+ });
}
+
#snapshot(s, x) {
const poll = s.role === null;
keys(
@@ -649,6 +661,14 @@ export class Broker {
// priority and requirement (#taskRows).
taskView(business, view, arg = null) {
this.#business(business);
+ if (view === 'comments' && arg === null) return {
+ epoch: this.#store.get("SELECT value FROM meta WHERE key='comment_cursor_epoch'").value,
+ cursors: this.#store.all(
+ `SELECT task_id,last_comment FROM task_comment_cursors c WHERE business=? AND seq=(
+ SELECT max(seq) FROM task_comment_cursors WHERE business=c.business AND task_id=c.task_id
+ ) ORDER BY task_id`, business,
+ ),
+ };
const parse = (r) => ({ ...r, fields: JSON.parse(r.fields) });
if (view === 'current' && arg === null)
return this.#store.all('SELECT * FROM task_current WHERE business=? ORDER BY task_ref', business).map(parse);
diff --git a/packages/bus/src/store.mjs b/packages/bus/src/store.mjs
index 755e43041..0c4d14f2d 100644
--- a/packages/bus/src/store.mjs
+++ b/packages/bus/src/store.mjs
@@ -13,7 +13,19 @@ import { join, isAbsolute } from 'node:path';
import { createHash } from 'node:crypto';
const ddl = readFileSync(new URL('../schema.sql', import.meta.url), 'utf8');
-export const SCHEMA_SHA256 = '179ffe356d4ff19a49b5ebad39b6c6bfd7771deb8e1b7c55b5e746f039d69e65';
+export const SCHEMA_SHA256 = '7587ac1df813e5148fdb73dc4733ec8064e6a142e4647f0da71be009a73eebdb';
+export const SCHEMA_VERSION = '3c';
+// Ordered, named offline steps. New schema changes append a step and keep prior DDL pinned.
+const legacyDDL = readFileSync(new URL('./migrations/3b.sql', import.meta.url), 'utf8');
+const LEGACY_SHA256 = '179ffe356d4ff19a49b5ebad39b6c6bfd7771deb8e1b7c55b5e746f039d69e65';
+const migrations = [{
+ from: '3b',
+ to: '3c',
+ sql: readFileSync(new URL('./migrations/3b-3c.sql', import.meta.url), 'utf8'),
+ initialize: (store) => store.run(
+ 'INSERT INTO meta(key,value) VALUES (?,?)', 'comment_cursor_epoch', new Date().toISOString(),
+ ),
+}];
const timedTables = [
'events',
'role_claims',
@@ -37,6 +49,19 @@ const reference = new DatabaseSync(':memory:');
reference.exec(ddl);
const EXPECTED_DIGEST = schemaDigest(reference);
reference.close();
+const legacy = new DatabaseSync(':memory:');
+if (hex(legacyDDL) !== LEGACY_SHA256) throw Error('legacy-schema-source-mismatch');
+legacy.exec(legacyDDL);
+for (let i = 0; i < migrations.length; i++) {
+ const step = migrations[i];
+ if (i && migrations[i - 1].to !== step.from) throw Error('migration-chain-mismatch');
+ step.fromDigest = schemaDigest(legacy);
+ legacy.exec(step.sql);
+ step.toDigest = schemaDigest(legacy);
+}
+if (migrations.at(-1).to !== SCHEMA_VERSION || schemaDigest(legacy) !== EXPECTED_DIGEST)
+ throw Error('migration-target-mismatch');
+legacy.close();
function exists(path) {
try {
return lstatSync(path);
@@ -64,7 +89,7 @@ export class Store {
#lockStat;
#closed = false;
#inTransaction = false;
- constructor(dataRoot) {
+ constructor(dataRoot, { migrateSchema = false } = {}) {
if (typeof dataRoot !== 'string' || !isAbsolute(dataRoot)) throw Error('unsafe-path');
const rootStat = lstatSync(dataRoot);
if (!rootStat.isDirectory() || rootStat.isSymbolicLink() || rootStat.uid !== process.getuid())
@@ -94,6 +119,7 @@ export class Store {
}
this.path = join(this.directory, 'bus.sqlite');
const fresh = !exists(this.path);
+ if (fresh && migrateSchema) throw Error('migration-source-missing');
if (fresh) closeSync(openSync(this.path, 'wx', 0o600));
else safe(this.path);
for (const suffix of ['-wal', '-shm']) if (exists(this.path + suffix)) safe(this.path + suffix);
@@ -104,13 +130,15 @@ export class Store {
this.#db.exec('PRAGMA journal_mode=WAL');
this.transaction(() => {
this.#db.exec(ddl.replace('PRAGMA journal_mode = WAL;', ''));
- this.run('INSERT INTO meta(key,value) VALUES (?,?)', 'schema_version', '3b');
+ this.run('INSERT INTO meta(key,value) VALUES (?,?)', 'schema_version', SCHEMA_VERSION);
this.run('INSERT INTO meta(key,value) VALUES (?,?)', 'schema_digest', EXPECTED_DIGEST);
+ this.run('INSERT INTO meta(key,value) VALUES (?,?)', 'comment_cursor_epoch', new Date().toISOString());
});
}
+ if (!fresh && migrateSchema) this.#migrateSchema();
if (
schemaDigest(this.#db) !== EXPECTED_DIGEST ||
- this.get("SELECT value FROM meta WHERE key='schema_version'")?.value !== '3b' ||
+ this.get("SELECT value FROM meta WHERE key='schema_version'")?.value !== SCHEMA_VERSION ||
this.get("SELECT value FROM meta WHERE key='schema_digest'")?.value !== EXPECTED_DIGEST ||
this.get('PRAGMA journal_mode').journal_mode !== 'wal'
)
@@ -118,12 +146,41 @@ export class Store {
if (this.get('PRAGMA quick_check').quick_check !== 'ok' || this.all('PRAGMA foreign_key_check').length)
throw Error('integrity-failed');
this.#verifyTimes();
+ const epoch = this.get("SELECT value FROM meta WHERE key='comment_cursor_epoch'")?.value;
+ if (!epoch || !Number.isFinite(Date.parse(epoch)) || new Date(epoch).toISOString() !== epoch)
+ throw Error('invalid-comment-epoch');
} catch (e) {
if (fd !== undefined) closeSync(fd);
this.close();
throw e;
}
}
+ // Explicit offline operation only. Normal startup refuses old versions without changing them.
+ #migrateSchema() {
+ let version = this.get("SELECT value FROM meta WHERE key='schema_version'")?.value;
+ while (version !== SCHEMA_VERSION) {
+ const step = migrations.find((m) => m.from === version);
+ if (
+ !step || schemaDigest(this.#db) !== step.fromDigest ||
+ this.get("SELECT value FROM meta WHERE key='schema_digest'")?.value !== step.fromDigest ||
+ this.get('PRAGMA journal_mode').journal_mode !== 'wal'
+ ) throw Error('schema-mismatch');
+ if (this.get('PRAGMA quick_check').quick_check !== 'ok' || this.all('PRAGMA foreign_key_check').length)
+ throw Error('integrity-failed');
+ this.#verifyTimes();
+ this.transaction(() => {
+ this.#db.exec(step.sql);
+ // Only version metadata changes. The guard is restored before this step commits.
+ this.#db.exec('DROP TRIGGER meta_no_update');
+ this.run("UPDATE meta SET value=? WHERE key='schema_version'", step.to);
+ this.run("UPDATE meta SET value=? WHERE key='schema_digest'", step.toDigest);
+ step.initialize?.(this);
+ this.#db.exec("CREATE TRIGGER meta_no_update BEFORE UPDATE ON meta BEGIN SELECT RAISE(ABORT, 'meta is append-only'); END");
+ if (schemaDigest(this.#db) !== step.toDigest) throw Error('schema-mismatch');
+ });
+ version = step.to;
+ }
+ }
get(sql, ...args) {
return this.#db.prepare(sql).get(...args);
}
diff --git a/packages/bus/tests/schema.test.mjs b/packages/bus/tests/schema.test.mjs
index 4d00d56f4..5ef0c87dc 100644
--- a/packages/bus/tests/schema.test.mjs
+++ b/packages/bus/tests/schema.test.mjs
@@ -516,7 +516,7 @@ test('v3b prototype refusals, views and append-only mutations', (t) => {
const count = (type) => db.prepare('SELECT count(*) n FROM sqlite_master WHERE type = ?').get(type).n;
for (const label of refused) assert.ok(tested.has(label), label);
- assert.equal(count('table') - 1, 8);
- assert.equal(count('trigger'), 35);
+ assert.equal(count('table') - 1, 9);
+ assert.equal(count('trigger'), 39);
assert.equal(count('view'), 5);
});
diff --git a/packages/tasks/README.md b/packages/tasks/README.md
index f262ba339..0e19ebc91 100644
--- a/packages/tasks/README.md
+++ b/packages/tasks/README.md
@@ -427,3 +427,22 @@ run use a scratch container on 127.0.0.1, never a shared instance.
`deploy/vikunja/` has the compose file for installs without their own
Vikunja. See its README.
+
+### Durable comment cursor (row 49)
+
+The bus stores each business/task's highest observed comment ID separately from
+`task_current` and its digest. Adapter startup restores those IDs before its
+first reconcile. The first read of each task's comment list after startup catches
+outage comments, even outside the normal `updated` window. A reconcile also
+checks comment lists when the expanded count changes. Bot IDs advance the cursor
+but do not produce external-comment counts. The cursor and event commit together;
+in-memory ID, count and timestamp caches advance only after a successful commit.
+A crash after commit reloads the durable cursor and does not repeat the event.
+
+No cursor row is written until a new ID is observed. Empty tasks implicitly have
+cursor zero; initial timestamps use the database's creation/migration epoch minus
+a 60-second skew margin. This can duplicate comments at the first migration if
+the tracker clock is ahead, and clocks more than a minute behind remain outside
+the guarantee. Once a cursor exists, timestamps do not govern freshness.
+See the bus README for the explicit offline schema migration and Mos's separate
+approved downtime/re-pin procedure. No development seat migrates the live store.
diff --git a/packages/tasks/src/adapter.mjs b/packages/tasks/src/adapter.mjs
index f0f8cacfe..6a1f77e97 100644
--- a/packages/tasks/src/adapter.mjs
+++ b/packages/tasks/src/adapter.mjs
@@ -68,6 +68,7 @@ export function tasksAdapter({
const call = client({ baseUrl: s.baseUrl, fetch });
const use = (instance) => (method, path, opts) =>
credentials.use(`${name}/${instance}`, 'vikunja', (token) => call(token, method, path, opts));
+ const comments = broker.taskView(name, 'comments');
const ctx = {
...s,
broker,
@@ -86,7 +87,8 @@ export function tasksAdapter({
pending: 0,
lastStart: null,
lastReconcile: null,
- lastComment: new Map(),
+ lastComment: new Map(comments.cursors.map((c) => [c.task_id, c.last_comment])),
+ commentEpoch: Date.parse(comments.epoch),
commentsAt: new Map(),
commentCount: new Map(),
files: Object.fromEntries(Object.entries(files(businesses[name])).map(([i, f]) => [i, { file: f, stamp: stamp(f) }])),
diff --git a/packages/tasks/src/sync.mjs b/packages/tasks/src/sync.mjs
index 852d2b2cf..0f95d6741 100644
--- a/packages/tasks/src/sync.mjs
+++ b/packages/tasks/src/sync.mjs
@@ -79,12 +79,14 @@ export async function walk(ctx, filter, query = {}, project = ctx.project) {
const changed = (before, after) =>
Object.keys(after).filter((k) => JSON.stringify(before?.[k]) !== JSON.stringify(after[k]));
// Comments by anyone but this business's bots, newer than the last one seen. The first look at a
-// task counts only comments created inside the window, so a restart doesn't replay old ones.
+// task uses the database epoch minus the skew margin; later reads use durable IDs.
// `created` is to the second, so the bound is too.
-async function foreignComments(ctx, id, sinceMs) {
+async function foreignComments(ctx, id, out) {
const r = await all(ctx.syncCall, null, `/tasks/${id}/comments`, {}, ctx.per);
if (!r.ok) throw new BusError(refusal(r.r));
- const since = Math.floor(sinceMs / 1000) * 1000;
+ // No cursor yet: allow one minute of tracker clock skew around creation/migration.
+ // Once an ID is durable, tracker timestamps never affect freshness.
+ const since = Math.floor((ctx.commentEpoch - WINDOW_MS) / 1000) * 1000;
const last = ctx.lastComment.get(id);
let n = 0,
top = last ?? 0;
@@ -94,7 +96,7 @@ async function foreignComments(ctx, id, sinceMs) {
const fresh = last === undefined ? Date.parse(c.created) >= since : c.id > last;
if (fresh && !ctx.bots.has(c.author.id)) n++;
}
- ctx.lastComment.set(id, top);
+ if (top > (last ?? 0)) out.commentCursors.push({ task_id: id, last_comment: top });
return n;
}
// One task read. Records nothing that a newer self snapshot already covers, and nothing that
@@ -163,7 +165,9 @@ async function missing(ctx, current, refs, out) {
}
}
const record = (ctx, out) => {
- if (out.snapshots.length || out.events.length) ctx.broker.recordTask({ business: ctx.business, ...out });
+ if (out.snapshots.length || out.events.length || out.commentCursors.length)
+ ctx.broker.recordTask({ business: ctx.business, ...out });
+ for (const c of out.commentCursors) ctx.lastComment.set(c.task_id, c.last_comment);
};
// A child that answers 403 or 404 may have left the tree after the last reconcile: no longer
// shared, moved under a project the bot can't read, or deleted. A 404 can also be the child's kanban
@@ -191,17 +195,16 @@ export async function tick(ctx) {
}
}
const current = new Map(ctx.broker.taskView(ctx.business, 'current').map((x) => [x.task_ref, x]));
- const out = { snapshots: [], events: [] };
+ const out = { snapshots: [], events: [], commentCursors: [] };
// Refs some read this tick placed. Keyed by ref, not id: a task moved between two projects of the
// tree shows on the new one's board under the same id, and the old ref still needs its tombstone.
const placed = new Set();
for (const { project, view, open, hits } of reads) {
const comments = new Map();
for (const [id, t] of hits) {
- const key = normal(t.updated);
- if (ctx.commentsAt.get(id) === key) continue;
- comments.set(id, await foreignComments(ctx, id, sinceMs));
- ctx.commentsAt.set(id, key);
+ if (ctx.commentsAt.get(id) === normal(t.updated)) continue;
+ comments.set(id, await foreignComments(ctx, id, out));
+ // Cache advances only after the database transaction commits.
}
for (const [id, { task, bucket }] of open) {
const hit = hits.get(id);
@@ -224,6 +227,7 @@ export async function tick(ctx) {
.filter((r) => !placed.has(r));
await missing(ctx, current, gone, out);
record(ctx, out);
+ for (const { hits } of reads) for (const [id, t] of hits) ctx.commentsAt.set(id, normal(t.updated));
ctx.lastStart = startMs;
return { snapshots: out.snapshots.length, events: out.events.length };
}
@@ -234,13 +238,12 @@ export async function tick(ctx) {
export async function reconcile(ctx) {
const startMs = ctx.clock();
const readAt = iso(startMs);
- const sinceMs = ctx.lastReconcile ?? startMs;
if (ctx.descendants) await discover(ctx);
const reads = [];
for (const [project, view] of ctx.tree)
reads.push({ project, view, open: await openTasks(ctx, project, view), list: await walk(ctx, '', { expand: 'comment_count' }, project) });
const current = new Map(ctx.broker.taskView(ctx.business, 'current').map((x) => [x.task_ref, x]));
- const out = { snapshots: [], events: [] };
+ const out = { snapshots: [], events: [], commentCursors: [] };
const listed = new Set();
for (const { project, view, open, list } of reads)
for (const t of list) {
@@ -251,8 +254,7 @@ export async function reconcile(ctx) {
const count = t.comment_count;
if (count !== undefined && !Number.isSafeInteger(count)) shape();
const before = ctx.commentCount.get(t.id);
- if (count !== undefined) ctx.commentCount.set(t.id, count);
- if (before !== undefined && count !== before) comments = await foreignComments(ctx, t.id, sinceMs);
+ if (before === undefined || count !== before) comments = await foreignComments(ctx, t.id, out);
consider(ctx, current, project, { task: t, bucket, via: 'reconcile', readAt, comments }, out);
}
const gone = [...current.values()]
@@ -260,6 +262,8 @@ export async function reconcile(ctx) {
.map((x) => x.task_ref);
await missing(ctx, current, gone, out);
record(ctx, out);
+ for (const { list } of reads)
+ for (const t of list) if (t.comment_count !== undefined) ctx.commentCount.set(t.id, t.comment_count);
ctx.lastReconcile = startMs;
ctx.lastStart ??= startMs;
return { snapshots: out.snapshots.length, events: out.events.length };
diff --git a/packages/tasks/tests/world.mjs b/packages/tasks/tests/world.mjs
index 5085e7798..3fa6c68e3 100644
--- a/packages/tasks/tests/world.mjs
+++ b/packages/tasks/tests/world.mjs
@@ -104,8 +104,8 @@ export async function world(t, edit = {}) {
for (const [r, x] of Object.entries(roles)) if (x.credentials) references['demo/' + r] = x.credentials;
if (businesses.demo.tracker.sync.credentials) references['demo/@sync'] = businesses.demo.tracker.sync.credentials;
const credentials = new Credentials({ references, env, dataRoot: root });
- const store = new Store(root);
- const broker = new Broker({ store, businesses, secretCheck: (v) => credentials.assertClean(v) });
+ let store = new Store(root);
+ let broker = new Broker({ store, businesses, secretCheck: (v) => credentials.assertClean(v) });
const log = [];
const factory = tasksAdapter({
trackers: { demo: { baseUrl: 'http://vikunja.test', project, ...edit.tracker } },
@@ -114,7 +114,7 @@ export async function world(t, edit = {}) {
autostart: false,
...edit.adapter,
});
- const adapter = await factory({ broker, credentials, businesses });
+ let adapter = await factory({ broker, credentials, businesses });
t.after(async () => {
await adapter.close();
store.close();
@@ -145,6 +145,17 @@ export async function world(t, edit = {}) {
store.all('SELECT * FROM task_snapshots WHERE task_ref=? ORDER BY seq', taskRef).map((s) => ({ ...s, fields: JSON.parse(s.fields) }));
const call = (cap, verb, args) => broker.requestTask(cap, { verb, args }, adapter.handle);
return {
+ async stop() {
+ await adapter.close();
+ store.close();
+ },
+ async restart() {
+ store = new Store(root);
+ broker = new Broker({ store, businesses, secretCheck: (v) => credentials.assertClean(v) });
+ adapter = await factory({ broker, credentials, businesses });
+ await adapter.start('demo');
+ if (adapter.status()[0].state !== 'ready') throw Error(adapter.status()[0].refused);
+ },
root,
fake,
owner,
@@ -155,9 +166,9 @@ export async function world(t, edit = {}) {
tokens,
credentials,
businesses,
- store,
- broker,
- adapter,
+ get store() { return store; },
+ get broker() { return broker; },
+ get adapter() { return adapter; },
log,
caps,
agent,
diff --git a/scripts/land-work/read.mjs b/scripts/land-work/read.mjs
index 495a069eb..2402ed440 100644
--- a/scripts/land-work/read.mjs
+++ b/scripts/land-work/read.mjs
@@ -10,18 +10,14 @@
// "refuse<TAB>reason<TAB>detail" line and exit 2. A missing or unsafe
// database or business file exits 3.
//
-// Kept in step with packages/bus/src/store.mjs, which this imports from:
-// the schema_version literal '3b' below is a copy of the Store's. A schema
-// bump changes it there, and until it changes here every read refuses
-// bus-schema.
+// Schema version and digest follow the Store exports; a schema bump has one authority.
import { DatabaseSync } from 'node:sqlite';
import { readFileSync, lstatSync } from 'node:fs';
import { join } from 'node:path';
import { pathToFileURL } from 'node:url';
import { createHash } from 'node:crypto';
-import { schemaDigest, SCHEMA_SHA256 } from '../../packages/bus/src/store.mjs';
+import { schemaDigest, SCHEMA_SHA256, SCHEMA_VERSION } from '../../packages/bus/src/store.mjs';
-const SCHEMA_VERSION = '3b';
const MERGE = 'git.merge.protected';
const WORK_REF = /^work\/vikunja-([1-9][0-9]*)-([1-9][0-9]*)$/;
const ANY_WORK_REF = /\bwork\/[A-Za-z0-9._/-]*[A-Za-z0-9_]/g;
diff --git a/packages/bus/src/migrate-schema.mjs b/packages/bus/src/migrate-schema.mjs
new file mode 100644
index 000000000..b6b3893f4
--- /dev/null
+++ b/packages/bus/src/migrate-schema.mjs
@@ -0,0 +1,15 @@
+import { Store, SCHEMA_VERSION } from './store.mjs';
+// Offline, explicitly invoked in a human-approved downtime window. No config or secrets read.
+if (process.argv.length !== 3) {
+ process.stderr.write('usage: node packages/bus/src/migrate-schema.mjs /absolute/dataRoot\n');
+ process.exitCode = 2;
+} else {
+ try {
+ const store = new Store(process.argv[2], { migrateSchema: true });
+ store.close();
+ process.stdout.write(`bus schema ${SCHEMA_VERSION} ready\n`);
+ } catch (e) {
+ process.stderr.write(`migration refused: ${e.message}\n`);
+ process.exitCode = 1;
+ }
+}
diff --git a/packages/bus/src/migrations/3b-3c.sql b/packages/bus/src/migrations/3b-3c.sql
new file mode 100644
index 000000000..ba241af7c
--- /dev/null
+++ b/packages/bus/src/migrations/3b-3c.sql
@@ -0,0 +1,18 @@
+CREATE TABLE task_comment_cursors (
+ seq INTEGER PRIMARY KEY AUTOINCREMENT,
+ business TEXT NOT NULL,
+ task_id INTEGER NOT NULL CHECK (task_id > 0),
+ last_comment INTEGER NOT NULL CHECK (last_comment > 0)
+) STRICT;
+CREATE INDEX task_comment_cursors_latest ON task_comment_cursors (business, task_id, seq);
+CREATE TRIGGER task_comment_cursors_no_update BEFORE UPDATE ON task_comment_cursors
+ BEGIN SELECT RAISE(ABORT, 'task_comment_cursors is append-only'); END;
+CREATE TRIGGER task_comment_cursors_no_delete BEFORE DELETE ON task_comment_cursors
+ BEGIN SELECT RAISE(ABORT, 'task_comment_cursors is append-only'); END;
+CREATE TRIGGER task_comment_cursors_no_replace BEFORE INSERT ON task_comment_cursors
+ WHEN EXISTS (SELECT 1 FROM task_comment_cursors WHERE seq = NEW.seq)
+ BEGIN SELECT RAISE(ABORT, 'task_comment_cursors is append-only'); END;
+CREATE TRIGGER task_comment_cursors_forward BEFORE INSERT ON task_comment_cursors
+ WHEN NEW.last_comment <= COALESCE((SELECT last_comment FROM task_comment_cursors
+ WHERE business = NEW.business AND task_id = NEW.task_id ORDER BY seq DESC LIMIT 1), 0)
+ BEGIN SELECT RAISE(ABORT, 'comment cursor must advance'); END;
diff --git a/packages/bus/src/migrations/3b.sql b/packages/bus/src/migrations/3b.sql
new file mode 100644
index 000000000..6aef89004
--- /dev/null
+++ b/packages/bus/src/migrations/3b.sql
@@ -0,0 +1,211 @@
+PRAGMA journal_mode = WAL;
+PRAGMA foreign_keys = ON;
+CREATE TABLE meta (key TEXT PRIMARY KEY, value TEXT NOT NULL) STRICT;
+CREATE TABLE events (
+ seq INTEGER PRIMARY KEY AUTOINCREMENT,
+ id TEXT NOT NULL UNIQUE,
+ at TEXT NOT NULL,
+ business TEXT NOT NULL,
+ kind TEXT NOT NULL CHECK (kind IN (
+ 'session.launched',
+ 'session.ended',
+ 'action.allowed',
+ 'action.refused',
+ 'task.created',
+ 'task.assigned',
+ 'task.state',
+ 'task.closed',
+ 'task.changed.external',
+ 'task.conflict',
+ 'task.missing',
+ 'review.requested',
+ 'review.verdict',
+ 'human.input',
+ 'config.refused',
+ 'credential.expiring',
+ 'credential.expired',
+ 'credential.changed',
+ 'launch.revoked',
+ 'launch.restored',
+ 'digest.sent')),
+ actor_role TEXT, actor_run TEXT,
+ subject TEXT,
+ corrects TEXT REFERENCES events(id),
+ body TEXT NOT NULL CHECK (json_valid(body))
+) STRICT;
+CREATE TABLE role_claims (
+ seq INTEGER PRIMARY KEY AUTOINCREMENT,
+ at TEXT NOT NULL,
+ business TEXT NOT NULL, role TEXT NOT NULL,
+ op TEXT NOT NULL CHECK (op IN ('claim','release','revoke')),
+ holder_run TEXT NOT NULL,
+ harness TEXT NOT NULL, address TEXT,
+ by TEXT NOT NULL, reason TEXT,
+ decision TEXT
+) STRICT;
+CREATE TABLE decisions (
+ seq INTEGER PRIMARY KEY AUTOINCREMENT,
+ id TEXT NOT NULL UNIQUE,
+ at TEXT NOT NULL,
+ business TEXT NOT NULL, project TEXT,
+ raised_by_role TEXT NOT NULL, raised_by_run TEXT NOT NULL,
+ class TEXT NOT NULL CHECK (class IN ('routine','within-role','cross-role','gated')),
+ action TEXT NOT NULL,
+ route_to TEXT NOT NULL,
+ question TEXT NOT NULL,
+ options TEXT NOT NULL CHECK (json_valid(options) AND json_array_length(options) BETWEEN 2 AND 9),
+ recommendation TEXT NOT NULL,
+ task_ref TEXT CHECK (task_ref IS NULL OR (task_ref GLOB 'vikunja:[1-9]*/[1-9]*' AND NOT substr(task_ref, 9) GLOB '*[^0-9/]*' AND NOT substr(task_ref, 9) GLOB '*/*/*')), requirement_ref TEXT,
+ blocking INTEGER NOT NULL CHECK (blocking IN (0,1)),
+ supersedes TEXT REFERENCES decisions(id)
+) STRICT;
+CREATE TABLE decision_events (
+ seq INTEGER PRIMARY KEY AUTOINCREMENT,
+ decision TEXT NOT NULL REFERENCES decisions(id),
+ at TEXT NOT NULL,
+ op TEXT NOT NULL CHECK (op IN ('seen','resolved','withdrawn','expired')),
+ by TEXT NOT NULL,
+ choice TEXT, note TEXT, via TEXT
+) STRICT;
+CREATE TABLE messages (
+ seq INTEGER PRIMARY KEY AUTOINCREMENT,
+ id TEXT NOT NULL UNIQUE,
+ at TEXT NOT NULL,
+ business TEXT NOT NULL,
+ from_role TEXT NOT NULL, from_run TEXT NOT NULL,
+ to_role TEXT NOT NULL,
+ class TEXT NOT NULL,
+ in_reply_to TEXT REFERENCES messages(id),
+ decision TEXT REFERENCES decisions(id),
+ corrects TEXT REFERENCES messages(id),
+ body TEXT NOT NULL
+) STRICT;
+CREATE TABLE deliveries (
+ seq INTEGER PRIMARY KEY AUTOINCREMENT,
+ message TEXT NOT NULL REFERENCES messages(id),
+ at TEXT NOT NULL,
+ op TEXT NOT NULL CHECK (op IN ('routed','delivered','failed','read')),
+ holder_run TEXT, transport TEXT, address TEXT, detail TEXT
+) STRICT;
+CREATE TABLE task_snapshots (
+ seq INTEGER PRIMARY KEY AUTOINCREMENT,
+ at TEXT NOT NULL,
+ business TEXT NOT NULL,
+ task_ref TEXT NOT NULL CHECK (task_ref GLOB 'vikunja:[1-9]*/[1-9]*' AND NOT substr(task_ref, 9) GLOB '*[^0-9/]*' AND NOT substr(task_ref, 9) GLOB '*/*/*'),
+ updated TEXT NOT NULL,
+ etag TEXT,
+ digest TEXT NOT NULL CHECK (length(digest) = 64 AND NOT digest GLOB '*[^0-9a-f]*'),
+ fields TEXT NOT NULL CHECK (json_valid(fields)),
+ source TEXT NOT NULL CHECK (source IN ('self','poll')),
+ via TEXT CHECK (via IN ('board','cursor','task','reconcile')),
+ read_at TEXT,
+ role TEXT, run TEXT,
+ CHECK ((source = 'self') = (role IS NOT NULL AND run IS NOT NULL)),
+ CHECK ((source = 'poll') = (via IS NOT NULL AND read_at IS NOT NULL)),
+ CHECK (json_type(fields, '$.bucket') IS 'integer'
+ OR (source IS 'poll' AND via IS 'task' AND json_type(fields, '$.gone') IS 'text'))
+) STRICT;
+CREATE INDEX task_snapshots_ref ON task_snapshots (business, task_ref, seq);
+CREATE TRIGGER decisions_resolve_once BEFORE INSERT ON decision_events
+ WHEN NEW.op IN ('resolved','withdrawn','expired') AND EXISTS (
+ SELECT 1 FROM decision_events WHERE decision = NEW.decision AND op IN ('resolved','withdrawn','expired'))
+ BEGIN SELECT RAISE(ABORT, 'decision already closed'); END;
+CREATE TRIGGER decisions_resolved_choice BEFORE INSERT ON decision_events
+ WHEN NEW.op = 'resolved' AND (NEW.choice IS NULL OR NOT EXISTS (
+ SELECT 1 FROM decisions d, json_each(d.options) o WHERE d.id = NEW.decision AND json_extract(o.value,'$.key') = NEW.choice))
+ BEGIN SELECT RAISE(ABORT, 'resolution must name one of the options'); END;
+CREATE TRIGGER role_one_holder BEFORE INSERT ON role_claims
+ WHEN NEW.op = 'claim' AND (SELECT op FROM role_claims WHERE business = NEW.business AND role = NEW.role ORDER BY seq DESC LIMIT 1) = 'claim'
+ BEGIN SELECT RAISE(ABORT, 'role already held'); END;
+CREATE TRIGGER role_release_by_holder BEFORE INSERT ON role_claims
+ WHEN NEW.op IN ('release','revoke') AND COALESCE((SELECT op FROM role_claims WHERE business = NEW.business AND role = NEW.role ORDER BY seq DESC LIMIT 1),'') <> 'claim'
+ BEGIN SELECT RAISE(ABORT, 'role is not held'); END;
+CREATE TRIGGER role_release_same_run BEFORE INSERT ON role_claims
+ WHEN NEW.op = 'release' AND (SELECT holder_run FROM role_claims WHERE business = NEW.business AND role = NEW.role ORDER BY seq DESC LIMIT 1) <> NEW.holder_run
+ BEGIN SELECT RAISE(ABORT, 'only the holder releases; others revoke'); END;
+CREATE TRIGGER role_revoke_needs_decision BEFORE INSERT ON role_claims
+ WHEN NEW.op = 'revoke' AND NEW.decision IS NULL
+ BEGIN SELECT RAISE(ABORT, 'revoke needs a resolved decision'); END;
+CREATE TRIGGER meta_no_update BEFORE UPDATE ON meta BEGIN SELECT RAISE(ABORT, 'meta is append-only'); END;
+CREATE TRIGGER meta_no_delete BEFORE DELETE ON meta BEGIN SELECT RAISE(ABORT, 'meta is append-only'); END;
+CREATE TRIGGER events_no_update BEFORE UPDATE ON events BEGIN SELECT RAISE(ABORT, 'events is append-only'); END;
+CREATE TRIGGER events_no_delete BEFORE DELETE ON events BEGIN SELECT RAISE(ABORT, 'events is append-only'); END;
+CREATE TRIGGER role_claims_no_update BEFORE UPDATE ON role_claims BEGIN SELECT RAISE(ABORT, 'role_claims is append-only'); END;
+CREATE TRIGGER role_claims_no_delete BEFORE DELETE ON role_claims BEGIN SELECT RAISE(ABORT, 'role_claims is append-only'); END;
+CREATE TRIGGER decisions_no_update BEFORE UPDATE ON decisions BEGIN SELECT RAISE(ABORT, 'decisions is append-only'); END;
+CREATE TRIGGER decisions_no_delete BEFORE DELETE ON decisions BEGIN SELECT RAISE(ABORT, 'decisions is append-only'); END;
+CREATE TRIGGER decision_events_no_update BEFORE UPDATE ON decision_events BEGIN SELECT RAISE(ABORT, 'decision_events is append-only'); END;
+CREATE TRIGGER decision_events_no_delete BEFORE DELETE ON decision_events BEGIN SELECT RAISE(ABORT, 'decision_events is append-only'); END;
+CREATE TRIGGER messages_no_update BEFORE UPDATE ON messages BEGIN SELECT RAISE(ABORT, 'messages is append-only'); END;
+CREATE TRIGGER messages_no_delete BEFORE DELETE ON messages BEGIN SELECT RAISE(ABORT, 'messages is append-only'); END;
+CREATE TRIGGER deliveries_no_update BEFORE UPDATE ON deliveries BEGIN SELECT RAISE(ABORT, 'deliveries is append-only'); END;
+CREATE TRIGGER deliveries_no_delete BEFORE DELETE ON deliveries BEGIN SELECT RAISE(ABORT, 'deliveries is append-only'); END;
+CREATE TRIGGER meta_no_replace BEFORE INSERT ON meta WHEN EXISTS (SELECT 1 FROM meta WHERE key = NEW.key) BEGIN SELECT RAISE(ABORT, 'meta is append-only'); END;
+CREATE TRIGGER events_no_replace BEFORE INSERT ON events WHEN EXISTS (SELECT 1 FROM events WHERE seq = NEW.seq OR id = NEW.id) BEGIN SELECT RAISE(ABORT, 'events is append-only'); END;
+CREATE TRIGGER role_claims_no_replace BEFORE INSERT ON role_claims WHEN EXISTS (SELECT 1 FROM role_claims WHERE seq = NEW.seq) BEGIN SELECT RAISE(ABORT, 'role_claims is append-only'); END;
+CREATE TRIGGER decisions_no_replace BEFORE INSERT ON decisions WHEN EXISTS (SELECT 1 FROM decisions WHERE seq = NEW.seq OR id = NEW.id) BEGIN SELECT RAISE(ABORT, 'decisions is append-only'); END;
+CREATE TRIGGER decision_events_no_replace BEFORE INSERT ON decision_events WHEN EXISTS (SELECT 1 FROM decision_events WHERE seq = NEW.seq) BEGIN SELECT RAISE(ABORT, 'decision_events is append-only'); END;
+CREATE TRIGGER messages_no_replace BEFORE INSERT ON messages WHEN EXISTS (SELECT 1 FROM messages WHERE seq = NEW.seq OR id = NEW.id) BEGIN SELECT RAISE(ABORT, 'messages is append-only'); END;
+CREATE TRIGGER deliveries_no_replace BEFORE INSERT ON deliveries WHEN EXISTS (SELECT 1 FROM deliveries WHERE seq = NEW.seq) BEGIN SELECT RAISE(ABORT, 'deliveries is append-only'); END;
+CREATE TRIGGER task_snapshots_no_update BEFORE UPDATE ON task_snapshots BEGIN SELECT RAISE(ABORT, 'task_snapshots is append-only'); END;
+CREATE TRIGGER task_snapshots_no_delete BEFORE DELETE ON task_snapshots BEGIN SELECT RAISE(ABORT, 'task_snapshots is append-only'); END;
+CREATE TRIGGER task_snapshots_no_replace BEFORE INSERT ON task_snapshots WHEN EXISTS (SELECT 1 FROM task_snapshots WHERE seq = NEW.seq) BEGIN SELECT RAISE(ABORT, 'task_snapshots is append-only'); END;
+CREATE TRIGGER decisions_blocking_needs_task BEFORE INSERT ON decisions
+ WHEN NEW.blocking = 1 AND NEW.task_ref IS NULL
+ BEGIN SELECT RAISE(ABORT, 'a blocking decision cites the task it blocks'); END;
+CREATE TRIGGER events_launch_by_human BEFORE INSERT ON events
+ WHEN NEW.kind IN ('launch.revoked','launch.restored') AND (NEW.actor_role IS NOT NULL OR NEW.actor_run IS NOT NULL)
+ BEGIN SELECT RAISE(ABORT, 'only the human revokes or restores launching'); END;
+CREATE TRIGGER events_credential_body BEFORE INSERT ON events
+ WHEN NEW.kind GLOB 'credential.*' AND (
+ json_extract(NEW.body, '$.service') IS NULL OR json_extract(NEW.body, '$.service') NOT IN ('gitea','vikunja')
+ OR json_extract(NEW.body, '$.instance') IS NULL)
+ BEGIN SELECT RAISE(ABORT, 'credential events name a service and a role instance'); END;
+CREATE TRIGGER events_task_missing_body BEFORE INSERT ON events
+ WHEN NEW.kind = 'task.missing' AND (
+ json_extract(NEW.body, '$.reason') IS NULL OR json_extract(NEW.body, '$.reason') NOT IN ('moved','not-found','no-access')
+ OR (json_extract(NEW.body, '$.reason') = 'moved' AND json_type(NEW.body, '$.project') IS NOT 'integer'))
+ BEGIN SELECT RAISE(ABORT, 'task.missing carries a reason, and the new project when moved'); END;
+CREATE TRIGGER events_task_subject BEFORE INSERT ON events
+ WHEN NEW.kind GLOB 'task.*' AND (NEW.subject IS NULL OR NOT (NEW.subject GLOB 'vikunja:[1-9]*/[1-9]*' AND NOT substr(NEW.subject, 9) GLOB '*[^0-9/]*' AND NOT substr(NEW.subject, 9) GLOB '*/*/*'))
+ BEGIN SELECT RAISE(ABORT, 'a task event names its task in subject'); END;
+CREATE TRIGGER events_task_created_body BEFORE INSERT ON events
+ WHEN NEW.kind = 'task.created' AND (
+ json_type(NEW.body, '$.request') IS NOT 'text'
+ OR NOT EXISTS (SELECT 1 FROM events WHERE id = json_extract(NEW.body, '$.request')
+ AND kind = 'human.input' AND business = NEW.business)
+ OR json_type(NEW.body, '$.requirement') IS NOT 'text'
+ OR NOT json_extract(NEW.body, '$.requirement') GLOB 'REQ-[A-Z]*-[1-9]*'
+ OR json_extract(NEW.body, '$.requirement') GLOB 'REQ-*[^A-Z0-9-]*')
+ BEGIN SELECT RAISE(ABORT, 'task.created cites the human.input that asked for it and a requirement id'); END;
+CREATE VIEW launch_state AS
+ SELECT business, CASE kind WHEN 'launch.revoked' THEN 'revoked' ELSE 'allowed' END AS state, at
+ FROM events e WHERE kind IN ('launch.revoked','launch.restored')
+ AND seq = (SELECT max(seq) FROM events WHERE business = e.business AND kind IN ('launch.revoked','launch.restored'));
+CREATE VIEW task_external_changes AS
+ SELECT p.business, p.task_ref, p.seq, p.via, p.updated, p.digest, ls.digest AS self_digest
+ FROM task_snapshots p
+ LEFT JOIN task_snapshots ls ON ls.seq = (SELECT max(seq) FROM task_snapshots
+ WHERE business = p.business AND task_ref = p.task_ref AND source = 'self')
+ WHERE p.source = 'poll'
+ AND p.seq = (SELECT max(seq) FROM task_snapshots WHERE business = p.business AND task_ref = p.task_ref)
+ AND (ls.seq IS NULL OR (p.updated >= ls.updated AND p.read_at > ls.at AND p.digest <> ls.digest));
+CREATE VIEW task_current AS
+ SELECT s.seq, s.at, s.business, s.task_ref, s.updated, s.etag, s.digest, s.fields, s.source, s.via, s.read_at, s.role, s.run
+ FROM task_snapshots s
+ WHERE s.seq = (SELECT max(c.seq) FROM task_snapshots c
+ WHERE c.business = s.business AND c.task_ref = s.task_ref
+ AND (c.source = 'self'
+ OR c.read_at > coalesce((SELECT ls.at FROM task_snapshots ls
+ WHERE ls.business = c.business AND ls.task_ref = c.task_ref AND ls.source = 'self'
+ ORDER BY ls.seq DESC LIMIT 1), '')));
+CREATE VIEW tasks_open AS
+ SELECT s.business, s.task_ref, json_extract(s.fields, '$.bucket') AS bucket, s.seq
+ FROM task_current s
+ WHERE json_type(s.fields, '$.gone') IS NULL
+ AND json_extract(s.fields, '$.done') = 0;
+CREATE VIEW urgent_inbox AS
+ SELECT d.id, d.business, d.task_ref, d.question, d.at
+ FROM decisions d
+ WHERE d.class = 'gated' AND d.blocking = 1
+ AND NOT EXISTS (SELECT 1 FROM decision_events x WHERE x.decision = d.id AND x.op IN ('resolved','withdrawn','expired'));
diff --git a/packages/bus/tests/comment-cursors.test.mjs b/packages/bus/tests/comment-cursors.test.mjs
new file mode 100644
index 000000000..536dc1d1f
--- /dev/null
+++ b/packages/bus/tests/comment-cursors.test.mjs
@@ -0,0 +1,50 @@
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import { mkdtempSync, rmSync } from 'node:fs';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import { Store } from '../src/store.mjs';
+import { Broker } from '../src/broker.mjs';
+
+function fixture(t) {
+ const root = mkdtempSync(join(tmpdir(), 'comment-cursors-'));
+ const store = new Store(root);
+ const roles = { pm: { authority: { withinRole: ['message.send'], crossRole: [] } } };
+ const b = new Broker({ store, businesses: {
+ demo: { id: 'demo', human: 'jason', arbiters: { technical: 'pm', delivery: 'pm' }, roles },
+ other: { id: 'other', human: 'jason', arbiters: { technical: 'pm', delivery: 'pm' }, roles },
+ } });
+ t.after(() => { store.close(); rmSync(root, { recursive: true, force: true }); });
+ return { store, b, record: (commentCursors, extra = {}) => b.recordTask({ business: 'demo', commentCursors, ...extra }) };
+}
+
+test('comment cursors isolate businesses, move only forward and remain append-only', t => {
+ const { store, b, record } = fixture(t);
+ record([{ task_id: 7, last_comment: 11 }]);
+ record([{ task_id: 7, last_comment: 15 }]);
+ b.recordTask({ business: 'other', commentCursors: [{ task_id: 7, last_comment: 2 }] });
+ assert.deepEqual(b.taskView('demo', 'comments').cursors.map(c => ({ ...c })), [{ task_id: 7, last_comment: 15 }]);
+ assert.deepEqual(b.taskView('other', 'comments').cursors.map(c => ({ ...c })), [{ task_id: 7, last_comment: 2 }]);
+ for (const last_comment of [11, 15]) assert.throws(() => record([{ task_id: 7, last_comment }]), /must advance/);
+ for (const sql of [
+ 'UPDATE task_comment_cursors SET last_comment=20 WHERE seq=1',
+ 'DELETE FROM task_comment_cursors WHERE seq=1',
+ "INSERT OR REPLACE INTO task_comment_cursors VALUES(1,'demo',7,20)",
+ ]) assert.throws(() => store.run(sql), /append-only/);
+ assert.equal(store.get('SELECT count(*) n FROM task_comment_cursors').n, 3);
+});
+
+test('only trusted sync writes validated cursors; events and cursor share rollback', t => {
+ const { store, b, record } = fixture(t);
+ const cap = b.bindLaunch({ business: 'demo', role: 'pm', run: 'pm-run', harness: 'pi' });
+ for (const x of [null, {}, { task_id: 0, last_comment: 1 }, { task_id: 1, last_comment: 0 },
+ { task_id: 1, last_comment: 1.5 }, { task_id: 1, last_comment: 2 ** 53 },
+ { task_id: 1, last_comment: 1, extra: true }]) {
+ assert.throws(() => record([x]), e => e.code === 'invalid-request');
+ }
+ assert.throws(() => record([{ task_id: 1, last_comment: 1 }], { cap }), e => e.code === 'agent-required');
+ assert.throws(() => record([{ task_id: 1, last_comment: 1 }], {
+ events: [{ kind: 'human.input', body: {} }],
+ }), e => e.code === 'reserved-event');
+ assert.equal(store.get('SELECT count(*) n FROM task_comment_cursors').n, 0);
+});
diff --git a/packages/bus/tests/comment-migration.test.mjs b/packages/bus/tests/comment-migration.test.mjs
new file mode 100644
index 000000000..8d730e1eb
--- /dev/null
+++ b/packages/bus/tests/comment-migration.test.mjs
@@ -0,0 +1,117 @@
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import { mkdtempSync, mkdirSync, readFileSync, copyFileSync, chmodSync, rmSync } from 'node:fs';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import { DatabaseSync } from 'node:sqlite';
+import { createHash } from 'node:crypto';
+import { Store, schemaDigest } from '../src/store.mjs';
+
+function legacy(t) {
+ const root = mkdtempSync(join(tmpdir(), 'comment-migration-'));
+ t.after(() => rmSync(root, { recursive: true, force: true }));
+ const source = join(root, 'schema-current.sqlite');
+ const db = new DatabaseSync(source);
+ const ddl = readFileSync(new URL('./fixtures/schema-3b.sql', import.meta.url), 'utf8');
+ assert.equal(createHash('sha256').update(ddl).digest('hex'),
+ '179ffe356d4ff19a49b5ebad39b6c6bfd7771deb8e1b7c55b5e746f039d69e65');
+ db.exec(ddl);
+ db.prepare('INSERT INTO meta VALUES (?,?)').run('schema_version', '3b');
+ db.prepare('INSERT INTO meta VALUES (?,?)').run('schema_digest', schemaDigest(db));
+ db.prepare('INSERT INTO events(id,at,business,kind,body) VALUES(?,?,?,?,?)')
+ .run('preserved', '2026-10-15T00:00:00.000Z', 'demo', 'action.allowed', '{"fixture":true}');
+ db.close();
+ mkdirSync(join(root, 'bus'), { mode: 0o700 });
+ const path = join(root, 'bus/bus.sqlite');
+ copyFileSync(source, path);
+ chmodSync(path, 0o600);
+ return { root, source, path };
+}
+
+test('explicit 3b copy migration preserves evidence and opens as 3c; normal startup never migrates', t => {
+ const { root, source } = legacy(t);
+ const original = readFileSync(source);
+ assert.throws(() => new Store(root), /schema-mismatch/);
+ const store = new Store(root, { migrateSchema: true });
+ assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3c');
+ assert.equal(store.get("SELECT body FROM events WHERE id='preserved'").body, '{"fixture":true}');
+ assert.equal(store.get('SELECT count(*) n FROM task_comment_cursors').n, 0);
+ const epoch = store.get("SELECT value FROM meta WHERE key='comment_cursor_epoch'").value;
+ assert.equal(new Date(epoch).toISOString(), epoch);
+ store.close();
+ assert.deepEqual(readFileSync(source), original);
+ const again = new Store(root, { migrateSchema: true });
+ assert.equal(again.get("SELECT value FROM meta WHERE key='comment_cursor_epoch'").value, epoch);
+ again.close();
+ new Store(root).close();
+});
+
+test('migration refuses active writers and drifted legacy schemas, with no partial upgrade', t => {
+ const { root, path } = legacy(t);
+ const db = new DatabaseSync(path);
+ db.exec('DROP TRIGGER events_no_delete');
+ db.close();
+ assert.throws(() => new Store(root, { migrateSchema: true }), /schema-mismatch/);
+ const check = new DatabaseSync(path);
+ assert.equal(check.prepare("SELECT value FROM meta WHERE key='schema_version'").get().value, '3b');
+ assert.equal(check.prepare("SELECT count(*) n FROM sqlite_master WHERE name='task_comment_cursors'").get().n, 0);
+ check.close();
+ const other = legacy(t);
+ const current = new Store(other.root, { migrateSchema: true });
+ t.after(() => current.close());
+ assert.throws(() => new Store(other.root, { migrateSchema: true }), /writer-locked/);
+});
+
+test('named migration registry chains 3b through 3c to a fixture-only future version in order', async t => {
+ const { root } = legacy(t);
+ const { writeFileSync } = await import('node:fs');
+ const { pathToFileURL } = await import('node:url');
+ const moduleRoot = join(root, 'future-bus');
+ mkdirSync(join(moduleRoot, 'src'), { recursive: true });
+ mkdirSync(join(moduleRoot, 'src/migrations'));
+ const futureSQL = 'CREATE TABLE future_fixture (value TEXT) STRICT;\n';
+ const currentDDL = readFileSync(new URL('../schema.sql', import.meta.url), 'utf8');
+ const futureDDL = currentDDL + futureSQL;
+ const sha = createHash('sha256').update(futureDDL).digest('hex');
+ let source = readFileSync(new URL('../src/store.mjs', import.meta.url), 'utf8');
+ source = source.replace(/export const SCHEMA_SHA256 = '[a-f0-9]+';/, `export const SCHEMA_SHA256 = '${sha}';`)
+ .replace("export const SCHEMA_VERSION = '3c';", "export const SCHEMA_VERSION = '3d';")
+ .replace('}];\nconst timedTables', `}, { from: '3c', to: '3d', sql: ${JSON.stringify(futureSQL)} }];\nconst timedTables`);
+ writeFileSync(join(moduleRoot, 'src/store.mjs'), source);
+ writeFileSync(join(moduleRoot, 'schema.sql'), futureDDL);
+ for (const name of ['3b.sql', '3b-3c.sql'])
+ copyFileSync(new URL('../src/migrations/' + name, import.meta.url), join(moduleRoot, 'src/migrations', name));
+ const { Store: FutureStore } = await import(pathToFileURL(join(moduleRoot, 'src/store.mjs')));
+ const future = new FutureStore(root, { migrateSchema: true });
+ assert.equal(future.get("SELECT value FROM meta WHERE key='schema_version'").value, '3d');
+ assert.equal(future.get("SELECT name FROM sqlite_master WHERE name='future_fixture'").name, 'future_fixture');
+ assert.equal(future.get("SELECT count(*) n FROM task_comment_cursors").n, 0);
+ assert.equal(future.get("SELECT id FROM events").id, 'preserved');
+ future.close();
+ new FutureStore(root).close();
+});
+
+test('migration refuses mismatched source metadata without changing schema or evidence', t => {
+ const { root, path } = legacy(t);
+ const db = new DatabaseSync(path);
+ db.exec("DROP TRIGGER meta_no_update; UPDATE meta SET value='untrusted' WHERE key='schema_digest'; CREATE TRIGGER meta_no_update BEFORE UPDATE ON meta BEGIN SELECT RAISE(ABORT, 'meta is append-only'); END");
+ db.close();
+ assert.throws(() => new Store(root, { migrateSchema: true }), /schema-mismatch/);
+ const check = new DatabaseSync(path);
+ assert.equal(check.prepare("SELECT value FROM meta WHERE key='schema_version'").get().value, '3b');
+ assert.equal(check.prepare('SELECT id FROM events').get().id, 'preserved');
+ check.close();
+});
+
+test('offline command migrates only the supplied fixture root and reports usage errors', async t => {
+ const { spawnSync } = await import('node:child_process');
+ const { root } = legacy(t);
+ const command = new URL('../src/migrate-schema.mjs', import.meta.url).pathname;
+ const migrated = spawnSync(process.execPath, [command, root], { encoding: 'utf8' });
+ assert.equal(migrated.status, 0, migrated.stderr);
+ assert.match(migrated.stdout, /bus schema 3c ready/);
+ new Store(root).close();
+ const usage = spawnSync(process.execPath, [command], { encoding: 'utf8' });
+ assert.equal(usage.status, 2);
+ assert.match(usage.stderr, /usage:/);
+});
diff --git a/packages/bus/tests/fixtures/schema-3b.sql b/packages/bus/tests/fixtures/schema-3b.sql
new file mode 100644
index 000000000..6aef89004
--- /dev/null
+++ b/packages/bus/tests/fixtures/schema-3b.sql
@@ -0,0 +1,211 @@
+PRAGMA journal_mode = WAL;
+PRAGMA foreign_keys = ON;
+CREATE TABLE meta (key TEXT PRIMARY KEY, value TEXT NOT NULL) STRICT;
+CREATE TABLE events (
+ seq INTEGER PRIMARY KEY AUTOINCREMENT,
+ id TEXT NOT NULL UNIQUE,
+ at TEXT NOT NULL,
+ business TEXT NOT NULL,
+ kind TEXT NOT NULL CHECK (kind IN (
+ 'session.launched',
+ 'session.ended',
+ 'action.allowed',
+ 'action.refused',
+ 'task.created',
+ 'task.assigned',
+ 'task.state',
+ 'task.closed',
+ 'task.changed.external',
+ 'task.conflict',
+ 'task.missing',
+ 'review.requested',
+ 'review.verdict',
+ 'human.input',
+ 'config.refused',
+ 'credential.expiring',
+ 'credential.expired',
+ 'credential.changed',
+ 'launch.revoked',
+ 'launch.restored',
+ 'digest.sent')),
+ actor_role TEXT, actor_run TEXT,
+ subject TEXT,
+ corrects TEXT REFERENCES events(id),
+ body TEXT NOT NULL CHECK (json_valid(body))
+) STRICT;
+CREATE TABLE role_claims (
+ seq INTEGER PRIMARY KEY AUTOINCREMENT,
+ at TEXT NOT NULL,
+ business TEXT NOT NULL, role TEXT NOT NULL,
+ op TEXT NOT NULL CHECK (op IN ('claim','release','revoke')),
+ holder_run TEXT NOT NULL,
+ harness TEXT NOT NULL, address TEXT,
+ by TEXT NOT NULL, reason TEXT,
+ decision TEXT
+) STRICT;
+CREATE TABLE decisions (
+ seq INTEGER PRIMARY KEY AUTOINCREMENT,
+ id TEXT NOT NULL UNIQUE,
+ at TEXT NOT NULL,
+ business TEXT NOT NULL, project TEXT,
+ raised_by_role TEXT NOT NULL, raised_by_run TEXT NOT NULL,
+ class TEXT NOT NULL CHECK (class IN ('routine','within-role','cross-role','gated')),
+ action TEXT NOT NULL,
+ route_to TEXT NOT NULL,
+ question TEXT NOT NULL,
+ options TEXT NOT NULL CHECK (json_valid(options) AND json_array_length(options) BETWEEN 2 AND 9),
+ recommendation TEXT NOT NULL,
+ task_ref TEXT CHECK (task_ref IS NULL OR (task_ref GLOB 'vikunja:[1-9]*/[1-9]*' AND NOT substr(task_ref, 9) GLOB '*[^0-9/]*' AND NOT substr(task_ref, 9) GLOB '*/*/*')), requirement_ref TEXT,
+ blocking INTEGER NOT NULL CHECK (blocking IN (0,1)),
+ supersedes TEXT REFERENCES decisions(id)
+) STRICT;
+CREATE TABLE decision_events (
+ seq INTEGER PRIMARY KEY AUTOINCREMENT,
+ decision TEXT NOT NULL REFERENCES decisions(id),
+ at TEXT NOT NULL,
+ op TEXT NOT NULL CHECK (op IN ('seen','resolved','withdrawn','expired')),
+ by TEXT NOT NULL,
+ choice TEXT, note TEXT, via TEXT
+) STRICT;
+CREATE TABLE messages (
+ seq INTEGER PRIMARY KEY AUTOINCREMENT,
+ id TEXT NOT NULL UNIQUE,
+ at TEXT NOT NULL,
+ business TEXT NOT NULL,
+ from_role TEXT NOT NULL, from_run TEXT NOT NULL,
+ to_role TEXT NOT NULL,
+ class TEXT NOT NULL,
+ in_reply_to TEXT REFERENCES messages(id),
+ decision TEXT REFERENCES decisions(id),
+ corrects TEXT REFERENCES messages(id),
+ body TEXT NOT NULL
+) STRICT;
+CREATE TABLE deliveries (
+ seq INTEGER PRIMARY KEY AUTOINCREMENT,
+ message TEXT NOT NULL REFERENCES messages(id),
+ at TEXT NOT NULL,
+ op TEXT NOT NULL CHECK (op IN ('routed','delivered','failed','read')),
+ holder_run TEXT, transport TEXT, address TEXT, detail TEXT
+) STRICT;
+CREATE TABLE task_snapshots (
+ seq INTEGER PRIMARY KEY AUTOINCREMENT,
+ at TEXT NOT NULL,
+ business TEXT NOT NULL,
+ task_ref TEXT NOT NULL CHECK (task_ref GLOB 'vikunja:[1-9]*/[1-9]*' AND NOT substr(task_ref, 9) GLOB '*[^0-9/]*' AND NOT substr(task_ref, 9) GLOB '*/*/*'),
+ updated TEXT NOT NULL,
+ etag TEXT,
+ digest TEXT NOT NULL CHECK (length(digest) = 64 AND NOT digest GLOB '*[^0-9a-f]*'),
+ fields TEXT NOT NULL CHECK (json_valid(fields)),
+ source TEXT NOT NULL CHECK (source IN ('self','poll')),
+ via TEXT CHECK (via IN ('board','cursor','task','reconcile')),
+ read_at TEXT,
+ role TEXT, run TEXT,
+ CHECK ((source = 'self') = (role IS NOT NULL AND run IS NOT NULL)),
+ CHECK ((source = 'poll') = (via IS NOT NULL AND read_at IS NOT NULL)),
+ CHECK (json_type(fields, '$.bucket') IS 'integer'
+ OR (source IS 'poll' AND via IS 'task' AND json_type(fields, '$.gone') IS 'text'))
+) STRICT;
+CREATE INDEX task_snapshots_ref ON task_snapshots (business, task_ref, seq);
+CREATE TRIGGER decisions_resolve_once BEFORE INSERT ON decision_events
+ WHEN NEW.op IN ('resolved','withdrawn','expired') AND EXISTS (
+ SELECT 1 FROM decision_events WHERE decision = NEW.decision AND op IN ('resolved','withdrawn','expired'))
+ BEGIN SELECT RAISE(ABORT, 'decision already closed'); END;
+CREATE TRIGGER decisions_resolved_choice BEFORE INSERT ON decision_events
+ WHEN NEW.op = 'resolved' AND (NEW.choice IS NULL OR NOT EXISTS (
+ SELECT 1 FROM decisions d, json_each(d.options) o WHERE d.id = NEW.decision AND json_extract(o.value,'$.key') = NEW.choice))
+ BEGIN SELECT RAISE(ABORT, 'resolution must name one of the options'); END;
+CREATE TRIGGER role_one_holder BEFORE INSERT ON role_claims
+ WHEN NEW.op = 'claim' AND (SELECT op FROM role_claims WHERE business = NEW.business AND role = NEW.role ORDER BY seq DESC LIMIT 1) = 'claim'
+ BEGIN SELECT RAISE(ABORT, 'role already held'); END;
+CREATE TRIGGER role_release_by_holder BEFORE INSERT ON role_claims
+ WHEN NEW.op IN ('release','revoke') AND COALESCE((SELECT op FROM role_claims WHERE business = NEW.business AND role = NEW.role ORDER BY seq DESC LIMIT 1),'') <> 'claim'
+ BEGIN SELECT RAISE(ABORT, 'role is not held'); END;
+CREATE TRIGGER role_release_same_run BEFORE INSERT ON role_claims
+ WHEN NEW.op = 'release' AND (SELECT holder_run FROM role_claims WHERE business = NEW.business AND role = NEW.role ORDER BY seq DESC LIMIT 1) <> NEW.holder_run
+ BEGIN SELECT RAISE(ABORT, 'only the holder releases; others revoke'); END;
+CREATE TRIGGER role_revoke_needs_decision BEFORE INSERT ON role_claims
+ WHEN NEW.op = 'revoke' AND NEW.decision IS NULL
+ BEGIN SELECT RAISE(ABORT, 'revoke needs a resolved decision'); END;
+CREATE TRIGGER meta_no_update BEFORE UPDATE ON meta BEGIN SELECT RAISE(ABORT, 'meta is append-only'); END;
+CREATE TRIGGER meta_no_delete BEFORE DELETE ON meta BEGIN SELECT RAISE(ABORT, 'meta is append-only'); END;
+CREATE TRIGGER events_no_update BEFORE UPDATE ON events BEGIN SELECT RAISE(ABORT, 'events is append-only'); END;
+CREATE TRIGGER events_no_delete BEFORE DELETE ON events BEGIN SELECT RAISE(ABORT, 'events is append-only'); END;
+CREATE TRIGGER role_claims_no_update BEFORE UPDATE ON role_claims BEGIN SELECT RAISE(ABORT, 'role_claims is append-only'); END;
+CREATE TRIGGER role_claims_no_delete BEFORE DELETE ON role_claims BEGIN SELECT RAISE(ABORT, 'role_claims is append-only'); END;
+CREATE TRIGGER decisions_no_update BEFORE UPDATE ON decisions BEGIN SELECT RAISE(ABORT, 'decisions is append-only'); END;
+CREATE TRIGGER decisions_no_delete BEFORE DELETE ON decisions BEGIN SELECT RAISE(ABORT, 'decisions is append-only'); END;
+CREATE TRIGGER decision_events_no_update BEFORE UPDATE ON decision_events BEGIN SELECT RAISE(ABORT, 'decision_events is append-only'); END;
+CREATE TRIGGER decision_events_no_delete BEFORE DELETE ON decision_events BEGIN SELECT RAISE(ABORT, 'decision_events is append-only'); END;
+CREATE TRIGGER messages_no_update BEFORE UPDATE ON messages BEGIN SELECT RAISE(ABORT, 'messages is append-only'); END;
+CREATE TRIGGER messages_no_delete BEFORE DELETE ON messages BEGIN SELECT RAISE(ABORT, 'messages is append-only'); END;
+CREATE TRIGGER deliveries_no_update BEFORE UPDATE ON deliveries BEGIN SELECT RAISE(ABORT, 'deliveries is append-only'); END;
+CREATE TRIGGER deliveries_no_delete BEFORE DELETE ON deliveries BEGIN SELECT RAISE(ABORT, 'deliveries is append-only'); END;
+CREATE TRIGGER meta_no_replace BEFORE INSERT ON meta WHEN EXISTS (SELECT 1 FROM meta WHERE key = NEW.key) BEGIN SELECT RAISE(ABORT, 'meta is append-only'); END;
+CREATE TRIGGER events_no_replace BEFORE INSERT ON events WHEN EXISTS (SELECT 1 FROM events WHERE seq = NEW.seq OR id = NEW.id) BEGIN SELECT RAISE(ABORT, 'events is append-only'); END;
+CREATE TRIGGER role_claims_no_replace BEFORE INSERT ON role_claims WHEN EXISTS (SELECT 1 FROM role_claims WHERE seq = NEW.seq) BEGIN SELECT RAISE(ABORT, 'role_claims is append-only'); END;
+CREATE TRIGGER decisions_no_replace BEFORE INSERT ON decisions WHEN EXISTS (SELECT 1 FROM decisions WHERE seq = NEW.seq OR id = NEW.id) BEGIN SELECT RAISE(ABORT, 'decisions is append-only'); END;
+CREATE TRIGGER decision_events_no_replace BEFORE INSERT ON decision_events WHEN EXISTS (SELECT 1 FROM decision_events WHERE seq = NEW.seq) BEGIN SELECT RAISE(ABORT, 'decision_events is append-only'); END;
+CREATE TRIGGER messages_no_replace BEFORE INSERT ON messages WHEN EXISTS (SELECT 1 FROM messages WHERE seq = NEW.seq OR id = NEW.id) BEGIN SELECT RAISE(ABORT, 'messages is append-only'); END;
+CREATE TRIGGER deliveries_no_replace BEFORE INSERT ON deliveries WHEN EXISTS (SELECT 1 FROM deliveries WHERE seq = NEW.seq) BEGIN SELECT RAISE(ABORT, 'deliveries is append-only'); END;
+CREATE TRIGGER task_snapshots_no_update BEFORE UPDATE ON task_snapshots BEGIN SELECT RAISE(ABORT, 'task_snapshots is append-only'); END;
+CREATE TRIGGER task_snapshots_no_delete BEFORE DELETE ON task_snapshots BEGIN SELECT RAISE(ABORT, 'task_snapshots is append-only'); END;
+CREATE TRIGGER task_snapshots_no_replace BEFORE INSERT ON task_snapshots WHEN EXISTS (SELECT 1 FROM task_snapshots WHERE seq = NEW.seq) BEGIN SELECT RAISE(ABORT, 'task_snapshots is append-only'); END;
+CREATE TRIGGER decisions_blocking_needs_task BEFORE INSERT ON decisions
+ WHEN NEW.blocking = 1 AND NEW.task_ref IS NULL
+ BEGIN SELECT RAISE(ABORT, 'a blocking decision cites the task it blocks'); END;
+CREATE TRIGGER events_launch_by_human BEFORE INSERT ON events
+ WHEN NEW.kind IN ('launch.revoked','launch.restored') AND (NEW.actor_role IS NOT NULL OR NEW.actor_run IS NOT NULL)
+ BEGIN SELECT RAISE(ABORT, 'only the human revokes or restores launching'); END;
+CREATE TRIGGER events_credential_body BEFORE INSERT ON events
+ WHEN NEW.kind GLOB 'credential.*' AND (
+ json_extract(NEW.body, '$.service') IS NULL OR json_extract(NEW.body, '$.service') NOT IN ('gitea','vikunja')
+ OR json_extract(NEW.body, '$.instance') IS NULL)
+ BEGIN SELECT RAISE(ABORT, 'credential events name a service and a role instance'); END;
+CREATE TRIGGER events_task_missing_body BEFORE INSERT ON events
+ WHEN NEW.kind = 'task.missing' AND (
+ json_extract(NEW.body, '$.reason') IS NULL OR json_extract(NEW.body, '$.reason') NOT IN ('moved','not-found','no-access')
+ OR (json_extract(NEW.body, '$.reason') = 'moved' AND json_type(NEW.body, '$.project') IS NOT 'integer'))
+ BEGIN SELECT RAISE(ABORT, 'task.missing carries a reason, and the new project when moved'); END;
+CREATE TRIGGER events_task_subject BEFORE INSERT ON events
+ WHEN NEW.kind GLOB 'task.*' AND (NEW.subject IS NULL OR NOT (NEW.subject GLOB 'vikunja:[1-9]*/[1-9]*' AND NOT substr(NEW.subject, 9) GLOB '*[^0-9/]*' AND NOT substr(NEW.subject, 9) GLOB '*/*/*'))
+ BEGIN SELECT RAISE(ABORT, 'a task event names its task in subject'); END;
+CREATE TRIGGER events_task_created_body BEFORE INSERT ON events
+ WHEN NEW.kind = 'task.created' AND (
+ json_type(NEW.body, '$.request') IS NOT 'text'
+ OR NOT EXISTS (SELECT 1 FROM events WHERE id = json_extract(NEW.body, '$.request')
+ AND kind = 'human.input' AND business = NEW.business)
+ OR json_type(NEW.body, '$.requirement') IS NOT 'text'
+ OR NOT json_extract(NEW.body, '$.requirement') GLOB 'REQ-[A-Z]*-[1-9]*'
+ OR json_extract(NEW.body, '$.requirement') GLOB 'REQ-*[^A-Z0-9-]*')
+ BEGIN SELECT RAISE(ABORT, 'task.created cites the human.input that asked for it and a requirement id'); END;
+CREATE VIEW launch_state AS
+ SELECT business, CASE kind WHEN 'launch.revoked' THEN 'revoked' ELSE 'allowed' END AS state, at
+ FROM events e WHERE kind IN ('launch.revoked','launch.restored')
+ AND seq = (SELECT max(seq) FROM events WHERE business = e.business AND kind IN ('launch.revoked','launch.restored'));
+CREATE VIEW task_external_changes AS
+ SELECT p.business, p.task_ref, p.seq, p.via, p.updated, p.digest, ls.digest AS self_digest
+ FROM task_snapshots p
+ LEFT JOIN task_snapshots ls ON ls.seq = (SELECT max(seq) FROM task_snapshots
+ WHERE business = p.business AND task_ref = p.task_ref AND source = 'self')
+ WHERE p.source = 'poll'
+ AND p.seq = (SELECT max(seq) FROM task_snapshots WHERE business = p.business AND task_ref = p.task_ref)
+ AND (ls.seq IS NULL OR (p.updated >= ls.updated AND p.read_at > ls.at AND p.digest <> ls.digest));
+CREATE VIEW task_current AS
+ SELECT s.seq, s.at, s.business, s.task_ref, s.updated, s.etag, s.digest, s.fields, s.source, s.via, s.read_at, s.role, s.run
+ FROM task_snapshots s
+ WHERE s.seq = (SELECT max(c.seq) FROM task_snapshots c
+ WHERE c.business = s.business AND c.task_ref = s.task_ref
+ AND (c.source = 'self'
+ OR c.read_at > coalesce((SELECT ls.at FROM task_snapshots ls
+ WHERE ls.business = c.business AND ls.task_ref = c.task_ref AND ls.source = 'self'
+ ORDER BY ls.seq DESC LIMIT 1), '')));
+CREATE VIEW tasks_open AS
+ SELECT s.business, s.task_ref, json_extract(s.fields, '$.bucket') AS bucket, s.seq
+ FROM task_current s
+ WHERE json_type(s.fields, '$.gone') IS NULL
+ AND json_extract(s.fields, '$.done') = 0;
+CREATE VIEW urgent_inbox AS
+ SELECT d.id, d.business, d.task_ref, d.question, d.at
+ FROM decisions d
+ WHERE d.class = 'gated' AND d.blocking = 1
+ AND NOT EXISTS (SELECT 1 FROM decision_events x WHERE x.decision = d.id AND x.op IN ('resolved','withdrawn','expired'));
diff --git a/packages/tasks/tests/comment-restart.test.mjs b/packages/tasks/tests/comment-restart.test.mjs
new file mode 100644
index 000000000..c3dfa12fb
--- /dev/null
+++ b/packages/tasks/tests/comment-restart.test.mjs
@@ -0,0 +1,100 @@
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import { world } from './world.mjs';
+
+async function fixture(t) {
+ let now = Date.now();
+ const w = await world(t, { fake: { now: () => now }, adapter: { clock: () => now } });
+ await w.adapter.start('demo');
+ const owner = await w.ui('POST', `/projects/${w.project}/tasks`, { title: 'watched' });
+ assert.equal(owner.status, 201);
+ await w.adapter.reconcile('demo');
+ assert.equal(w.store.get('SELECT count(*) n FROM task_comment_cursors').n, 0);
+ const events = () => w.events('task.changed.external').filter(e => e.body.comments);
+ return { w, events, advance: () => { now += 3 * 3600000; } };
+}
+
+test('stopped bus resumes an empty task cursor after a long outage, exactly once', async t => {
+ const { w, events, advance } = await fixture(t);
+ await w.stop();
+ advance();
+ assert.equal((await w.ui('POST', '/tasks/1/comments', { comment: 'while bus stopped' })).status, 201);
+ advance();
+ await w.restart();
+ assert.equal(events().length, 1);
+ assert.equal(events()[0].body.comments, 1);
+ await w.adapter.tick('demo');
+ await w.adapter.reconcile('demo');
+ assert.equal(events().length, 1);
+});
+
+test('persisted IDs avoid replay, exclude bots and bound rows to observed new comments', async t => {
+ const { w, events, advance } = await fixture(t);
+ await w.ui('POST', '/tasks/1/comments', { comment: 'before restart' });
+ await w.adapter.reconcile('demo');
+ assert.equal(events().length, 1);
+ const rows = () => w.store.get('SELECT count(*) n FROM task_comment_cursors').n;
+ assert.equal(rows(), 1);
+ await w.adapter.reconcile('demo');
+ await w.adapter.tick('demo');
+ assert.equal(rows(), 1);
+ await w.stop();
+ advance();
+ const r = await w.fake.fetch('http://vikunja.test/api/v2/tasks/1/comments', {
+ method: 'POST', headers: { authorization: 'Bearer ' + w.tokens.coder, 'content-type': 'application/json' },
+ body: JSON.stringify({ comment: 'bot while stopped' }),
+ });
+ assert.equal(r.status, 201);
+ await w.ui('POST', '/tasks/1/comments', { comment: 'human while stopped' });
+ advance();
+ await w.restart();
+ assert.equal(events().length, 2);
+ assert.equal(events()[1].body.comments, 1);
+ assert.equal(rows(), 2);
+ await w.stop();
+ await w.restart();
+ await w.adapter.tick('demo');
+ assert.equal(events().length, 2);
+ assert.equal(rows(), 2);
+});
+
+test('commit followed by crash before cache advance neither loses nor duplicates a comment', async t => {
+ const { w, events } = await fixture(t);
+ await w.ui('POST', '/tasks/1/comments', { comment: 'committed before crash' });
+ const record = w.broker.recordTask.bind(w.broker);
+ w.broker.recordTask = args => { record(args); throw Error('simulated crash after commit'); };
+ await assert.rejects(w.adapter.reconcile('demo'), /simulated crash/);
+ assert.equal(events().length, 1);
+ await w.stop();
+ await w.restart();
+ assert.equal(events().length, 1);
+ await w.adapter.tick('demo');
+ assert.equal(events().length, 1);
+});
+
+test('failed event insert rolls back cursor and a retry still emits the comment once', async t => {
+ const { w, events } = await fixture(t);
+ await w.ui('POST', '/tasks/1/comments', { comment: 'retry me' });
+ w.store.run("CREATE TRIGGER fixture_fail BEFORE INSERT ON events WHEN NEW.kind='task.changed.external' BEGIN SELECT RAISE(ABORT,'fixture failure'); END");
+ await assert.rejects(w.adapter.reconcile('demo'), /fixture failure/);
+ w.store.run('DROP TRIGGER fixture_fail');
+ await w.adapter.reconcile('demo');
+ assert.equal(events().length, 1);
+ assert.equal(events()[0].body.comments, 1);
+});
+
+// Removing the epoch margin loses a tracker comment thirty seconds behind host time.
+test('initial cursor accepts a comment within the stated one-minute clock skew margin', async t => {
+ let trackerNow = Date.now() - 30000;
+ const skewed = await world(t, { fake: { now: () => trackerNow } });
+ await skewed.adapter.start('demo');
+ await skewed.ui('POST', `/projects/${skewed.project}/tasks`, { title: 'clock skew' });
+ await skewed.adapter.reconcile('demo');
+ await skewed.ui('POST', '/tasks/1/comments', { comment: 'thirty seconds behind' });
+ await skewed.adapter.reconcile('demo');
+ const found = skewed.events('task.changed.external').filter(e => e.body.comments);
+ assert.equal(found.length, 1);
+ assert.equal(found[0].body.comments, 1);
+ await skewed.adapter.reconcile('demo');
+ assert.equal(skewed.events('task.changed.external').filter(e => e.body.comments).length, 1);
+});
Reference in new issue View Git Blame Copy Permalink