feat(tasks): the adapter reads a project tree (#1555, row 62)
With tracker.descendants on, the tracker project is the root of a tree.
Each reconcile reads the sync bot's project list (archived projects and
negative pseudo-project ids left out) and walks it by parent_project_id.
A child that passes the install checks is read through its board, and
one that doesn't is watch-only (bucket 0) and never refuses the
business. A tick reads each project in the tree. A child answering 403
or 404 leaves the tree only if GET /projects/<id> then answers 403 or
404 too, and its open tasks get the one-by-one read. Task refs carry
the project, so a task moved inside the tree shows under its new ref
and the old ref gets its tombstone. tracker.descendants reaches the adapter
through the business var and packages/cli/src/config.mjs (lead decision
87). Off, the adapter reads as v0 did.
Darkwing built it (agents/darkwing/work/queue-62), in three rounds.
Round 3 was approved by Filbert (27317) and Dewey (27338) on manifest
8b81e836, 13 paths. Their non-blocking notes and Darkwing's follow-ups
(two surviving mutants in test 18, the fake's view-404 code) go to a
follow-up row.
Sage's gate on 918a4a19, sequential and teed: node suites tasks 71,
business 60, bus 80, cli 110 and queue 150, 0 failed; webui 39 of 40,
then 40 of 40 on three reruns (row 62 touches no webui file; a timing
test under load 17 with Dewey's gate running beside it); test-auth 15,
test-conductor 17, test-config 24, test-discord 66,
test-extension-package 18, test-foundation 44, test-queue 27,
test-release 14, test-task 98, all 0 failed (Docker on). edc5de71..HEAD
touches none of the 13 paths.
Live read, 2026-10-11T21:28:03Z, tasks.woltje.com v2.7.0, root #19,
descendants on, sync key only, GETs only (the startup scope probes not
sent): reconcile 15 GETs, 42 snapshots; tick 8 GETs, 0 snapshots.
Children 18, 20 and 21 are watch-only. A missing view id on #18 answers
404 (code 3014) while #18 reads 200, as the README says. Record:
agents/sage/work/vikunja-setup/2026-10-11_tick-62.json.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
Claude Opus 5.5
@@ -5,7 +5,7 @@ import { startup } from './startup.mjs';
|
||||
import { tick, reconcile } from './sync.mjs';
|
||||
import { verbs } from './verbs.mjs';
|
||||
// S3's adapter for startBroker({tasks}). `trackers` is the host's plain-data boot config:
|
||||
// {<business>: {baseUrl, project, pollSeconds, reconcileMinutes}}. A business with no entry has
|
||||
// {<business>: {baseUrl, project, pollSeconds, reconcileMinutes, descendants}}. A business with no entry has
|
||||
// no tracker, and its task verbs refuse. Startup runs in the background so the broker boots
|
||||
// without waiting on Vikunja; verbs refuse with tracker-starting until it passes.
|
||||
export const TIMEOUT = 60000;
|
||||
@@ -18,8 +18,10 @@ function settings(name, t, b) {
|
||||
apiRoot(t.baseUrl);
|
||||
const pollSeconds = t.pollSeconds ?? 30,
|
||||
reconcileMinutes = t.reconcileMinutes ?? 60;
|
||||
const descendants = t.descendants ?? false;
|
||||
if (!positive(t.project) || !Number.isSafeInteger(pollSeconds) || pollSeconds < 10 || !positive(reconcileMinutes))
|
||||
throw new BusError('tracker-config');
|
||||
if (typeof descendants !== 'boolean') throw new BusError('tracker-config');
|
||||
if (!positive(b.tracker?.sync?.botId)) throw new BusError('tracker-config');
|
||||
const roles = {};
|
||||
for (const [role, r] of Object.entries(b.roles ?? {}))
|
||||
@@ -34,7 +36,7 @@ function settings(name, t, b) {
|
||||
if (!positive(id)) throw new BusError('tracker-config');
|
||||
labels.set(id, title);
|
||||
}
|
||||
return { business: name, baseUrl: t.baseUrl, project: t.project, pollSeconds, reconcileMinutes, roles, pm: pms[0], labels };
|
||||
return { business: name, baseUrl: t.baseUrl, project: t.project, pollSeconds, reconcileMinutes, descendants, roles, pm: pms[0], labels };
|
||||
}
|
||||
// Files behind the business's Vikunja references, to notice a token rotated on disk.
|
||||
function files(b) {
|
||||
@@ -74,6 +76,7 @@ export function tasksAdapter({
|
||||
call,
|
||||
per: 50,
|
||||
view: null,
|
||||
tree: null,
|
||||
bots: new Set([businesses[name].tracker.sync.botId, ...Object.values(s.roles).map((r) => r.botId)]),
|
||||
sync: use('@sync'),
|
||||
as: (role, method, path, opts) => use(role)(method, path, opts),
|
||||
|
||||
@@ -6,9 +6,13 @@ import { NULL_DATE } from './digest.mjs';
|
||||
// scope 401 before 404, bucket_id 0 outside the board and the move, `updated` reported to the
|
||||
// second but filtered on the stored value, moves between open buckets not bumping `updated`,
|
||||
// silent unassign, 403 on removing an absent label, and bot label visibility through the owner.
|
||||
// Project trees follow the v2.7.0 source (row 62 probes.md): a grant on a project reaches every
|
||||
// descendant, the nearest grant wins unless one on the path is admin, archiving a project archives
|
||||
// its descendants, and the project list leaves out archived projects unless asked for them.
|
||||
const ROUTES = [
|
||||
['GET', /^\/info$/, null, 'info'],
|
||||
['GET', /^\/labels$/, ['labels', 'read_all'], 'labels'],
|
||||
['GET', /^\/projects$/, ['projects', 'read_all'], 'projects'],
|
||||
['GET', /^\/projects\/(\d+)$/, ['projects', 'read_one'], 'project'],
|
||||
['GET', /^\/projects\/(\d+)\/views$/, ['projects_views', 'read_all'], 'views'],
|
||||
['GET', /^\/projects\/(\d+)\/views\/(\d+)\/buckets$/, ['projects', 'views_buckets'], 'buckets'],
|
||||
@@ -80,6 +84,7 @@ export class FakeVikunja {
|
||||
#tasks = new Map();
|
||||
#labels = new Map();
|
||||
#faults = [];
|
||||
#pseudo = [];
|
||||
constructor({ version = 'v2.7.0', now = () => Date.now() } = {}) {
|
||||
this.version = version;
|
||||
this.#now = now;
|
||||
@@ -98,9 +103,9 @@ export class FakeVikunja {
|
||||
revoke(token) {
|
||||
this.#tokens.delete(token);
|
||||
}
|
||||
project(title, owner) {
|
||||
project(title, owner, { parent = 0 } = {}) {
|
||||
const id = ++this.#seq.project;
|
||||
this.#projects.set(id, { id, title, owner, shares: new Map(), index: 0 });
|
||||
this.#projects.set(id, { id, title, owner, parent, archived: !!this.#projects.get(parent)?.archived, shares: new Map(), index: 0 });
|
||||
for (const [kind, position] of [
|
||||
['list', 100],
|
||||
['gantt', 200],
|
||||
@@ -115,6 +120,34 @@ export class FakeVikunja {
|
||||
share(project, user, permission = 0) {
|
||||
this.#projects.get(project).shares.set(user, permission);
|
||||
}
|
||||
unshare(project, user) {
|
||||
this.#projects.get(project).shares.delete(user);
|
||||
}
|
||||
// Moves a project under another parent, or to the top level with 0.
|
||||
reparent(project, parent) {
|
||||
this.#projects.get(project).parent = parent;
|
||||
}
|
||||
// Vikunja writes is_archived down the tree, so a descendant reads as archived too.
|
||||
archive(project, archived = true) {
|
||||
for (const p of this.#projects.values()) if (p.id === project || this.#ancestors(p.id).includes(project)) p.archived = archived;
|
||||
}
|
||||
// Vikunja deletes a project's descendants and every task in them with it.
|
||||
deleteProject(project) {
|
||||
const gone = [...this.#projects.keys()].filter((id) => id === project || this.#ancestors(id).includes(project));
|
||||
for (const t of [...this.#tasks.values()]) if (gone.includes(t.project)) this.#tasks.delete(t.id);
|
||||
for (const id of gone) this.#projects.delete(id);
|
||||
}
|
||||
// Drops a view, its buckets and the tasks' places in them. The project still reads, and a read
|
||||
// of the view answers 404 with code 13001.
|
||||
deleteView(view) {
|
||||
for (const [id, b] of this.#buckets) if (b.view === view) this.#buckets.delete(id);
|
||||
for (const t of this.#tasks.values()) t.buckets.delete(view);
|
||||
this.#views.delete(view);
|
||||
}
|
||||
// A saved filter or Favorites: a negative id in the project list, as Project.ReadAll appends.
|
||||
pseudoProject(user, id = -1, title = 'Favorites') {
|
||||
this.#pseudo.push({ id, title, user });
|
||||
}
|
||||
bucket(view, title) {
|
||||
const id = ++this.#seq.bucket;
|
||||
this.#buckets.set(id, { id, title, view, position: id * 100 });
|
||||
@@ -231,11 +264,39 @@ export class FakeVikunja {
|
||||
});
|
||||
return id;
|
||||
}
|
||||
// Parents, nearest first. A cycle can't be made through the setup methods the tests use.
|
||||
#ancestors(project) {
|
||||
const out = [];
|
||||
for (let p = this.#projects.get(project)?.parent; p && !out.includes(p); p = this.#projects.get(p)?.parent) out.push(p);
|
||||
return out;
|
||||
}
|
||||
// project_access.go: owner or a grant on the project or any ancestor; admin anywhere wins,
|
||||
// otherwise the nearest grant.
|
||||
#access(user, project) {
|
||||
const p = this.#projects.get(project);
|
||||
if (!p) return null;
|
||||
if (p.owner === user) return 2;
|
||||
return p.shares.has(user) ? p.shares.get(user) : null;
|
||||
if (!this.#projects.has(project)) return null;
|
||||
let nearest = null;
|
||||
for (const id of [project, ...this.#ancestors(project)]) {
|
||||
const p = this.#projects.get(id);
|
||||
if (!p) break;
|
||||
if (p.owner === user) return 2;
|
||||
const a = p.shares.has(user) ? p.shares.get(user) : null;
|
||||
if (a === 2) return 2;
|
||||
nearest ??= a;
|
||||
}
|
||||
return nearest;
|
||||
}
|
||||
#projectJson(p) {
|
||||
return {
|
||||
id: p.id,
|
||||
title: p.title,
|
||||
description: '',
|
||||
owner: this.#userJson(p.owner),
|
||||
parent_project_id: p.parent,
|
||||
is_archived: p.archived,
|
||||
position: p.id * 100,
|
||||
created: second(0),
|
||||
updated: second(0),
|
||||
};
|
||||
}
|
||||
#userJson(id) {
|
||||
const u = this.#users.get(id);
|
||||
@@ -378,7 +439,15 @@ export class FakeVikunja {
|
||||
_project(user, [p]) {
|
||||
const r = this.#projectFor(user, p);
|
||||
if (r.error) return r.error;
|
||||
return { status: 200, body: { id: p, title: r.p.title, description: '', owner: this.#userJson(r.p.owner), is_archived: false } };
|
||||
return { status: 200, body: this.#projectJson(r.p) };
|
||||
}
|
||||
_projects(user, _a, q) {
|
||||
const archived = q.get('is_archived') === 'true';
|
||||
const list = [...this.#projects.values()]
|
||||
.filter((p) => this.#access(user, p.id) !== null && (archived || !p.archived))
|
||||
.map((p) => this.#projectJson(p));
|
||||
for (const x of this.#pseudo) if (x.user === user) list.push({ id: x.id, title: x.title, description: '', owner: this.#userJson(user), parent_project_id: 0, is_archived: false });
|
||||
return { status: 200, body: page(list, q) };
|
||||
}
|
||||
_views(user, [p], q) {
|
||||
const r = this.#projectFor(user, p);
|
||||
|
||||
@@ -35,7 +35,8 @@ export async function startup(ctx) {
|
||||
if (mine.some((s) => s.state === 'expired')) refuse('credential-expired');
|
||||
// Sync identity: the control must pass and the write probe must be refused by scope.
|
||||
expect(await ctx.sync('GET', `/projects/${ctx.project}`), 200, 'tracker-project');
|
||||
await install(ctx);
|
||||
ctx.view = await inspect(ctx, ctx.project);
|
||||
ctx.tree = new Map([[ctx.project, ctx.view]]);
|
||||
const probe = await ctx.sync('PATCH', `/tasks/${MISSING}`, { body: {} });
|
||||
if (probe.status === 404 || probe.status === 403 || outcome(probe) === 'ok') refuse('scope-too-broad');
|
||||
expect(probe, 401, 'tracker-unexpected');
|
||||
@@ -62,13 +63,14 @@ export async function startup(ctx) {
|
||||
}
|
||||
}
|
||||
// Install checks: one manual kanban view, the five buckets once each, done and default wired.
|
||||
async function install(ctx) {
|
||||
const views = await all(ctx.syncCall, null, `/projects/${ctx.project}/views`, {}, ctx.per);
|
||||
// Returns the view, or refuses tracker-install. A failed read refuses tracker-unavailable.
|
||||
async function inspect(ctx, project) {
|
||||
const views = await all(ctx.syncCall, null, `/projects/${project}/views`, {}, ctx.per);
|
||||
if (!views.ok) refuse('tracker-unavailable');
|
||||
const kanban = views.items.filter((x) => x?.view_kind === 'kanban');
|
||||
if (kanban.length !== 1 || kanban[0].bucket_configuration_mode !== 'manual') refuse('tracker-install');
|
||||
const k = kanban[0];
|
||||
const buckets = await all(ctx.syncCall, null, `/projects/${ctx.project}/views/${k.id}/buckets`, {}, ctx.per);
|
||||
const buckets = await all(ctx.syncCall, null, `/projects/${project}/views/${k.id}/buckets`, {}, ctx.per);
|
||||
if (!buckets.ok) refuse('tracker-unavailable');
|
||||
const ids = {};
|
||||
for (const title of BUCKETS) {
|
||||
@@ -77,7 +79,7 @@ async function install(ctx) {
|
||||
ids[title] = hits[0].id;
|
||||
}
|
||||
if (k.done_bucket_id !== ids.done || k.default_bucket_id !== ids.todo) refuse('tracker-install');
|
||||
ctx.view = {
|
||||
return {
|
||||
id: k.id,
|
||||
done: ids.done,
|
||||
todo: ids.todo,
|
||||
@@ -85,3 +87,32 @@ async function install(ctx) {
|
||||
titles: Object.fromEntries(Object.entries(ids).map(([t, id]) => [id, t])),
|
||||
};
|
||||
}
|
||||
// With tracker.descendants on, the tracker project is the root of a tree: the root plus every
|
||||
// project the sync bot's project list shows under it by parent_project_id, archived ones left out.
|
||||
// The first reconcile and each one after it run this. A child that passes the install checks is
|
||||
// read through its board; one that fails is watch-only (null) and never refuses the business.
|
||||
// ctx.tree maps each project to its view; the root keeps the view startup checked.
|
||||
export async function discover(ctx) {
|
||||
const r = await all(ctx.syncCall, null, '/projects', {}, ctx.per);
|
||||
if (!r.ok) refuse(r.r.status === 401 ? 'tracker-unauthorized' : 'tracker-unavailable');
|
||||
const children = new Map();
|
||||
for (const p of r.items) {
|
||||
if (!p || typeof p !== 'object' || !Number.isSafeInteger(p.id)) refuse('tracker-shape');
|
||||
// Saved filters and Favorites come back with negative ids.
|
||||
if (p.id <= 0) continue;
|
||||
if (!Number.isSafeInteger(p.parent_project_id) || p.parent_project_id < 0 || typeof p.is_archived !== 'boolean') refuse('tracker-shape');
|
||||
if (p.is_archived || p.id === ctx.project) continue;
|
||||
if (!children.has(p.parent_project_id)) children.set(p.parent_project_id, []);
|
||||
children.get(p.parent_project_id).push(p.id);
|
||||
}
|
||||
const tree = new Map([[ctx.project, ctx.view]]);
|
||||
const next = [ctx.project];
|
||||
while (next.length)
|
||||
for (const id of (children.get(next.shift()) ?? []).sort((a, b) => a - b)) {
|
||||
if (tree.has(id)) continue;
|
||||
tree.set(id, await inspect(ctx, id).catch((e) => (e instanceof BusError && e.code === 'tracker-install' ? null : Promise.reject(e))));
|
||||
next.push(id);
|
||||
}
|
||||
ctx.tree = tree;
|
||||
return tree;
|
||||
}
|
||||
|
||||
+97
-50
@@ -1,10 +1,12 @@
|
||||
import { BusError } from '../../bus/src/broker.mjs';
|
||||
import { outcome, refusal, all } from './vikunja.mjs';
|
||||
import { ref, checkTask, taskFields, digest, tombstone } from './digest.mjs';
|
||||
import { ref, parseRef, checkTask, taskFields, digest, tombstone } from './digest.mjs';
|
||||
import { discover } from './startup.mjs';
|
||||
// The sync bot's reads (addendum B sections 3 and 5). A tick reads the whole open board, because a
|
||||
// move between open buckets doesn't bump `updated`, then the cursor: every task updated since the
|
||||
// previous tick's start minus WINDOW_MS. The window covers the async bump lag and the one-second
|
||||
// rounding (probes.md, U and Paging). Overlap is deduped by digest against task_current.
|
||||
// With tracker.descendants on, each tick and reconcile does this once per project in ctx.tree.
|
||||
export const WINDOW_MS = 60000;
|
||||
const iso = (ms) => new Date(ms).toISOString();
|
||||
const floorSecond = (ms) => iso(Math.floor(ms / 1000) * 1000);
|
||||
@@ -14,11 +16,16 @@ export const normal = (updated) => iso(Date.parse(updated));
|
||||
const shape = () => {
|
||||
throw new BusError('tracker-shape');
|
||||
};
|
||||
// A watch-only project has no board the adapter trusts, so its snapshots carry bucket 0: no
|
||||
// bucket state. Vikunja bucket ids start at 1, and nothing here reads 0 back as a bucket.
|
||||
export const WATCH_ONLY = 0;
|
||||
const fieldsFor = (task, bucket) => (bucket === WATCH_ONLY ? { ...taskFields(task, 1), bucket } : taskFields(task, bucket));
|
||||
const doneBucket = (view) => (view ? view.done : WATCH_ONLY);
|
||||
// The open board, every page. A bucket's tasks are null when it is empty; `count` is the bucket's total.
|
||||
export async function board(ctx) {
|
||||
export async function board(ctx, project = ctx.project, view = ctx.view) {
|
||||
const seen = new Map();
|
||||
for (let page = 1; page <= 1000; page++) {
|
||||
const r = await ctx.sync('GET', `/projects/${ctx.project}/views/${ctx.view.id}/buckets/tasks`, {
|
||||
const r = await ctx.sync('GET', `/projects/${project}/views/${view.id}/buckets/tasks`, {
|
||||
query: { filter: 'done = false', per_page: ctx.per, page },
|
||||
});
|
||||
if (outcome(r) !== 'ok') throw new BusError(refusal(r));
|
||||
@@ -37,12 +44,17 @@ export async function board(ctx) {
|
||||
}
|
||||
throw new BusError('tracker-paging');
|
||||
}
|
||||
// A watch-only project's open tasks, from its task list.
|
||||
async function openTasks(ctx, project, view) {
|
||||
if (view) return board(ctx, project, view);
|
||||
return new Map((await walk(ctx, 'done = false', {}, project)).map((t) => [t.id, { task: t, bucket: WATCH_ONLY }]));
|
||||
}
|
||||
// A list walk keyed by id, so a write landing mid-walk can't shift a page under it.
|
||||
export async function walk(ctx, filter, query = {}) {
|
||||
export async function walk(ctx, filter, query = {}, project = ctx.project) {
|
||||
const items = [];
|
||||
let last = 0;
|
||||
for (let i = 0; i < 100000; i++) {
|
||||
const r = await ctx.sync('GET', `/projects/${ctx.project}/tasks`, {
|
||||
const r = await ctx.sync('GET', `/projects/${project}/tasks`, {
|
||||
query: {
|
||||
...query,
|
||||
filter: [filter, `id > ${last}`].filter(Boolean).join(' && '),
|
||||
@@ -56,7 +68,7 @@ export async function walk(ctx, filter, query = {}) {
|
||||
if (!Array.isArray(r.json?.items)) shape();
|
||||
for (const t of r.json.items) {
|
||||
checkTask(t);
|
||||
if (t.id <= last || t.project_id !== ctx.project) shape();
|
||||
if (t.id <= last || t.project_id !== project) shape();
|
||||
last = t.id;
|
||||
items.push(t);
|
||||
}
|
||||
@@ -87,9 +99,9 @@ async function foreignComments(ctx, id, sinceMs) {
|
||||
}
|
||||
// One task read. Records nothing that a newer self snapshot already covers, and nothing that
|
||||
// matches task_current; a comment count alone is an event without a snapshot.
|
||||
function consider(ctx, current, { task, bucket, via, readAt, comments = 0 }, out) {
|
||||
const r = ref(ctx.project, task.id);
|
||||
const fields = taskFields(task, bucket);
|
||||
function consider(ctx, current, project, { task, bucket, via, readAt, comments = 0 }, out) {
|
||||
const r = ref(project, task.id);
|
||||
const fields = fieldsFor(task, bucket);
|
||||
const d = digest(fields);
|
||||
const c = current.get(r);
|
||||
const stale = c?.source === 'self' && readAt <= c.at;
|
||||
@@ -109,21 +121,27 @@ function consider(ctx, current, { task, bucket, via, readAt, comments = 0 }, out
|
||||
body: { via, digest: d, previous: c?.digest ?? null, changed: changed(c?.fields, fields), ...(comments ? { comments } : {}) },
|
||||
});
|
||||
}
|
||||
// Open tasks the board no longer shows: done, moved, deleted or no longer shared.
|
||||
// Open tasks the board no longer shows: done, moved, deleted or no longer shared. A task whose
|
||||
// project left the tree (archived, moved out from under the root, or no longer shared) is out of
|
||||
// the business's reach: no-access, or moved if the task itself changed project.
|
||||
async function missing(ctx, current, refs, out) {
|
||||
for (const r of refs) {
|
||||
const c = current.get(r);
|
||||
const id = Number(r.slice(r.indexOf('/') + 1));
|
||||
const { project, task: id } = parseRef(r);
|
||||
const readAt = iso(ctx.clock());
|
||||
const x = await ctx.sync('GET', `/tasks/${id}`);
|
||||
let fields, updated, event;
|
||||
if (x.status === 200) {
|
||||
checkTask(x.json);
|
||||
updated = normal(x.json.updated);
|
||||
if (x.json.project_id !== ctx.project) {
|
||||
if (x.json.project_id !== project) {
|
||||
fields = tombstone('moved', x.json.project_id);
|
||||
event = { reason: 'moved', project: x.json.project_id };
|
||||
} else if (x.json.done) fields = taskFields(x.json, ctx.view.done);
|
||||
} else if (!ctx.tree.has(project)) {
|
||||
updated = c.updated;
|
||||
fields = tombstone('no-access');
|
||||
event = { reason: 'no-access' };
|
||||
} else if (x.json.done) fields = fieldsFor(x.json, doneBucket(ctx.tree.get(project)));
|
||||
else continue; // still open in the project: the board read raced a move, the next tick sees it
|
||||
} else if (outcome(x) === 'not-found' || x.status === 403) {
|
||||
updated = c.updated;
|
||||
@@ -147,39 +165,63 @@ 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 });
|
||||
};
|
||||
// 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
|
||||
// view deleted while the project still reads, so the tick reads the project itself and drops the
|
||||
// child only on a 403 or 404 there. Then missing() reads its open tasks one by one, as after a
|
||||
// reconcile. Anything else fails the tick, and so does a 403 or 404 on the root.
|
||||
async function leftTree(ctx, project, e) {
|
||||
if (project === ctx.project || !(e instanceof BusError) || (e.code !== 'tracker-forbidden' && e.code !== 'tracker-not-found')) return false;
|
||||
const r = await ctx.sync('GET', `/projects/${project}`);
|
||||
return r.status === 403 || r.status === 404;
|
||||
}
|
||||
export async function tick(ctx) {
|
||||
const startMs = ctx.clock();
|
||||
const readAt = iso(startMs);
|
||||
const sinceMs = Math.floor(((ctx.lastStart ?? startMs) - WINDOW_MS) / 1000) * 1000;
|
||||
const open = await board(ctx);
|
||||
const hits = new Map((await walk(ctx, `updated > ${floorSecond(sinceMs)}`)).map((t) => [t.id, t]));
|
||||
const reads = [];
|
||||
for (const [project, view] of ctx.tree) {
|
||||
try {
|
||||
const open = await openTasks(ctx, project, view);
|
||||
const hits = new Map((await walk(ctx, `updated > ${floorSecond(sinceMs)}`, {}, project)).map((t) => [t.id, t]));
|
||||
reads.push({ project, view, open, hits });
|
||||
} catch (e) {
|
||||
if (!(await leftTree(ctx, project, e))) throw e;
|
||||
ctx.tree.delete(project);
|
||||
}
|
||||
}
|
||||
const current = new Map(ctx.broker.taskView(ctx.business, 'current').map((x) => [x.task_ref, x]));
|
||||
const out = { snapshots: [], events: [] };
|
||||
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);
|
||||
}
|
||||
for (const [id, { task, bucket }] of open) {
|
||||
const hit = hits.get(id);
|
||||
const use = hit && Date.parse(hit.updated) >= Date.parse(task.updated) ? hit : task;
|
||||
const b = use.done ? ctx.view.done : bucket;
|
||||
consider(ctx, current, { task: use, bucket: b, via: hit ? 'cursor' : 'board', readAt, comments: comments.get(id) }, out);
|
||||
}
|
||||
const explained = new Set();
|
||||
for (const [id, t] of hits) {
|
||||
if (open.has(id)) continue;
|
||||
// An open task the board didn't show raced a move; the next board read places it.
|
||||
if (!t.done) continue;
|
||||
explained.add(ref(ctx.project, id));
|
||||
consider(ctx, current, { task: t, bucket: ctx.view.done, via: 'cursor', readAt, comments: comments.get(id) }, out);
|
||||
// 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);
|
||||
}
|
||||
for (const [id, { task, bucket }] of open) {
|
||||
const hit = hits.get(id);
|
||||
const use = hit && Date.parse(hit.updated) >= Date.parse(task.updated) ? hit : task;
|
||||
const b = use.done ? doneBucket(view) : bucket;
|
||||
placed.add(ref(project, id));
|
||||
consider(ctx, current, project, { task: use, bucket: b, via: hit ? 'cursor' : 'board', readAt, comments: comments.get(id) }, out);
|
||||
}
|
||||
for (const [id, t] of hits) {
|
||||
if (open.has(id)) continue;
|
||||
// An open task the board didn't show raced a move; the next board read places it.
|
||||
if (!t.done) continue;
|
||||
placed.add(ref(project, id));
|
||||
consider(ctx, current, project, { task: t, bucket: doneBucket(view), via: 'cursor', readAt, comments: comments.get(id) }, out);
|
||||
}
|
||||
}
|
||||
const gone = ctx.broker
|
||||
.taskView(ctx.business, 'open')
|
||||
.map((x) => x.task_ref)
|
||||
.filter((r) => !explained.has(r) && !open.has(Number(r.slice(r.indexOf('/') + 1))));
|
||||
.filter((r) => !placed.has(r));
|
||||
await missing(ctx, current, gone, out);
|
||||
record(ctx, out);
|
||||
ctx.lastStart = startMs;
|
||||
@@ -187,27 +229,32 @@ export async function tick(ctx) {
|
||||
}
|
||||
// The hourly full read, and the first read at startup. comment_count catches comments on tasks the
|
||||
// cursor window missed; a changed count is checked against the comment list before it is reported.
|
||||
// With tracker.descendants on it reads the project list first, so a new child joins the tree here
|
||||
// and a child that left it has its tasks read one by one through missing().
|
||||
export async function reconcile(ctx) {
|
||||
const startMs = ctx.clock();
|
||||
const readAt = iso(startMs);
|
||||
const sinceMs = ctx.lastReconcile ?? startMs;
|
||||
const open = await board(ctx);
|
||||
const list = await walk(ctx, '', { expand: 'comment_count' });
|
||||
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 listed = new Set();
|
||||
for (const t of list) {
|
||||
listed.add(ref(ctx.project, t.id));
|
||||
const bucket = t.done ? ctx.view.done : open.get(t.id)?.bucket;
|
||||
if (!bucket) continue;
|
||||
let comments = 0;
|
||||
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);
|
||||
consider(ctx, current, { task: t, bucket, via: 'reconcile', readAt, comments }, out);
|
||||
}
|
||||
for (const { project, view, open, list } of reads)
|
||||
for (const t of list) {
|
||||
listed.add(ref(project, t.id));
|
||||
const bucket = t.done ? doneBucket(view) : open.get(t.id)?.bucket;
|
||||
if (bucket === undefined) continue;
|
||||
let comments = 0;
|
||||
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);
|
||||
consider(ctx, current, project, { task: t, bucket, via: 'reconcile', readAt, comments }, out);
|
||||
}
|
||||
const gone = [...current.values()]
|
||||
.filter((x) => x.fields.gone === undefined && !listed.has(x.task_ref))
|
||||
.map((x) => x.task_ref);
|
||||
|
||||
Reference in New Issue
Block a user