From fa08fede9d4e9836ffa6855a6071a5b0e0c46278 Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Sun, 11 Oct 2026 13:46:59 +0100 Subject: [PATCH 01/15] feat(sync): add graph-sync, the one sync of a source into a store The command line and the web app each had their own loop that read a source's pages into the store, and neither passed the relations a page carries (sub-issues, assignees, repository membership) to items.upsert, so the store's hierarchy queries never had anything to answer from. graph-sync is a core package both can use: it groups scopes by source, writes each page as it is read, relations included, holds a transition or relation back until the model has both of its ends, and reports progress with a checkpoint that records the scopes and the filter its cursor was issued under, so an interrupted sync resumes where it stopped. It is bundled into process-graph as the process-graph/sync subpath. --- README.md | 1 + packages/core/graph/sync/AGENTS.md | 1 + packages/core/graph/sync/CLAUDE.md | 1 + packages/core/graph/sync/README.md | 59 +++ packages/core/graph/sync/eslint.config.ts | 9 + packages/core/graph/sync/package.json | 72 +++ packages/core/graph/sync/src/checkpoint.ts | 17 + packages/core/graph/sync/src/index.ts | 17 + packages/core/graph/sync/src/scopes.ts | 48 ++ packages/core/graph/sync/src/sync.ts | 292 ++++++++++++ .../core/graph/sync/src/sync.unit.test.ts | 429 ++++++++++++++++++ .../graph/sync/src/test-support/fixtures.ts | 155 +++++++ packages/core/graph/sync/stryker.config.ts | 3 + .../graph/sync/test/package.smoke.test.ts | 73 +++ packages/core/graph/sync/tsconfig.build.json | 20 + packages/core/graph/sync/tsconfig.json | 9 + packages/core/graph/sync/tsconfig.node.json | 8 + packages/core/graph/sync/vitest.config.ts | 16 + packages/targets/process-graph/README.md | 2 + .../targets/process-graph/eslint.config.ts | 1 + packages/targets/process-graph/package.json | 5 + .../src/entrypoints.unit.test.ts | 7 + packages/targets/process-graph/src/sync.ts | 2 + .../process-graph/test/package.smoke.test.ts | 29 ++ .../targets/process-graph/tsdown.config.ts | 1 + pnpm-lock.yaml | 52 +++ tsconfig.build.json | 3 + 27 files changed, 1332 insertions(+) create mode 120000 packages/core/graph/sync/AGENTS.md create mode 120000 packages/core/graph/sync/CLAUDE.md create mode 100644 packages/core/graph/sync/README.md create mode 100644 packages/core/graph/sync/eslint.config.ts create mode 100644 packages/core/graph/sync/package.json create mode 100644 packages/core/graph/sync/src/checkpoint.ts create mode 100644 packages/core/graph/sync/src/index.ts create mode 100644 packages/core/graph/sync/src/scopes.ts create mode 100644 packages/core/graph/sync/src/sync.ts create mode 100644 packages/core/graph/sync/src/sync.unit.test.ts create mode 100644 packages/core/graph/sync/src/test-support/fixtures.ts create mode 100644 packages/core/graph/sync/stryker.config.ts create mode 100644 packages/core/graph/sync/test/package.smoke.test.ts create mode 100644 packages/core/graph/sync/tsconfig.build.json create mode 100644 packages/core/graph/sync/tsconfig.json create mode 100644 packages/core/graph/sync/tsconfig.node.json create mode 100644 packages/core/graph/sync/vitest.config.ts create mode 100644 packages/targets/process-graph/src/sync.ts diff --git a/README.md b/README.md index dcbcc74..ce25fa1 100644 --- a/README.md +++ b/README.md @@ -71,6 +71,7 @@ The internal packages, private and bundled into `process-graph`: | [`graph-source-contract`](packages/core/graph/source-contract/README.md) | The `GraphSource` port as an oRPC contract: versioned, paged and streamed sync of bound activities, participants, transitions and run state, resumable by cursor, typed filters and scope listings, with an in-memory fake and a conformance kit. | | [`graph-source-github-issues`](packages/adapters/graph/source-github-issues/README.md) | The GitHub issues adapter for the `GraphSource` port: issues, `blockedBy` and `blocking`, sub-issues and assignees across organisations, repositories and Projects v2 boards, with server-side filters, a scope listing and per-owner tokens. | | [`graph-store-contract`](packages/core/graph/store-contract/README.md) | The persistence port as an oRPC contract: save, load and delete models, set run state and the facts a run reads (provided artefacts, condition values), upsert items and relations from a source, ready, blocked-chain, impact and hierarchy queries, a change feed; with an in-memory fake and a conformance kit. | +| [`graph-sync`](packages/core/graph/sync/README.md) | Syncs `GraphSource`s into a `GraphStore` page by page: participants, bound activities, transitions, relations and run state, resumable from a checkpoint and narrowed by a typed filter; the one sync the command line and the web app share. | | [`graph-store-sqlite`](packages/adapters/graph/store-sqlite/README.md) | The store port on SQLite through Drizzle: one table definition, committed drizzle-kit migrations applied the same way in every runtime, recursive SQL for the structural queries, against Drizzle's async SQLite database so any driver plugs in. | | [`graph-sqlite-node`](packages/adapters/graph/sqlite-node/README.md) | The SQLite store on Node's built-in `node:sqlite` through `drizzle-orm/node-sqlite`, for the command line and servers. | | [`graph-sqlite-wasm`](packages/adapters/graph/sqlite-wasm/README.md) | The SQLite store in the browser: SQLite wasm in a dedicated worker on OPFS `opfs-sahpool`, shared between tabs under a Web Lock with no special headers, served to the page over oRPC. | diff --git a/packages/core/graph/sync/AGENTS.md b/packages/core/graph/sync/AGENTS.md new file mode 120000 index 0000000..42061c0 --- /dev/null +++ b/packages/core/graph/sync/AGENTS.md @@ -0,0 +1 @@ +README.md \ No newline at end of file diff --git a/packages/core/graph/sync/CLAUDE.md b/packages/core/graph/sync/CLAUDE.md new file mode 120000 index 0000000..42061c0 --- /dev/null +++ b/packages/core/graph/sync/CLAUDE.md @@ -0,0 +1 @@ +README.md \ No newline at end of file diff --git a/packages/core/graph/sync/README.md b/packages/core/graph/sync/README.md new file mode 100644 index 0000000..4cc2e74 --- /dev/null +++ b/packages/core/graph/sync/README.md @@ -0,0 +1,59 @@ +# @exadev/process-graph-sync + +[![GitHub](https://img.shields.io/badge/GitHub-181717?logo=github&logoColor=white)](https://github.com/ExaDev/process-graph) [![CI](https://img.shields.io/github/actions/workflow/status/ExaDev/process-graph/ci.yml?branch=main)](https://github.com/ExaDev/process-graph/actions) + +A private workspace package, not published on its own: it is bundled into [`process-graph`](../../../targets/process-graph/README.md), which exposes it as `process-graph/sync`. + +Syncs work from `GraphSource`s ([`graph-source-contract`](../source-contract/README.md)) into a `GraphStore` ([`graph-store-contract`](../store-contract/README.md)). The command line's `sync` and the web app's sync both call it, so a sync is written the same way wherever it runs. `src/` imports nothing Node-only. + +```ts +import { syncScopes } from "@exadev/process-graph-sync"; + +const result = await syncScopes({ + store, + sources: (prefix) => (prefix === "github" ? github : undefined), + modelId: "acme", + scopes: ["github:acme", "github:partner/sdk"], + filter: { states: ["open"], labels: ["bug"] }, + signal, + onProgress: (progress) => { + record(progress.checkpoint); + }, +}); +``` + +## What a sync writes + +Scopes are grouped by their source prefix and read one source after another, in the order each source first appears. Each page is merged into the stored model through `items.upsert` as soon as it is read: participants, bound activities, transitions, relations (`child-of`, `assigned-to`, `part-of`) and run state. A model the store does not have is created by the first write. + +A store rejects a write that leaves a transition or relation naming an element or participant the model lacks, and a source may report a dependency or a sub-issue whose other end a later page brings. A transition or relation is therefore held back until the model has both of its ends, and written with the page that completes it. One whose end never arrives is written with the last page, where the store's rejection names the missing reference. + +## Progress and resuming + +After every page, `onProgress` receives the pages written, the revision of the last write, the distinct participants, activities, transitions and relations written, the gaps met and a `SyncCheckpoint`. A checkpoint names the model, the scopes, the filter, the source to read next and the cursor after the last page written within it; passing it back as `resume` reads only what is left. A source's cursor is valid only with the scopes and filter it was issued for, which is why the checkpoint carries both. `syncCheckpointSchema` is its Zod schema, for a consumer that stores it. The final progress has no checkpoint. + +A failure part way leaves every page before it written, and is passed on as the source or store threw it; the last progress's checkpoint says where to carry on. An abort rejects with the signal's reason. + +## Filters + +`filter` is the source contract's `SyncFilter`: run states, labels and an order. It is passed to every source the sync reads and recorded in every checkpoint. A source that cannot apply a filter to a scope refuses the sync with `BAD_REQUEST`; the GitHub issues adapter does so for a Projects v2 board. + +## Functions + +| Export | What it is | +| ---------------------- | ----------------------------------------------------------------------------------- | +| `syncScopes(request)` | Syncs the scopes into the store and resolves with the final progress. | +| `scopesBySource` | Groups source-qualified scopes by source prefix, in the order a sync reads them. | +| `syncCheckpointSchema` | The Zod schema of a `SyncCheckpoint`. | +| `UnknownSourceError` | Thrown for a scope whose source prefix no reader serves; carries the `prefix`. | +| `InvalidScopeError` | Thrown for a scope that names no source; carries the `scope`. | +| `SyncStore` | The part of the `GraphStore` client a sync calls: `models.load` and `items.upsert`. | + +## Commands + +```sh +pnpm build # tsc -b -> dist/ (ESM + .d.ts) +pnpm test # unit tests against the source and store contracts' in-memory fakes +pnpm test:smoke # the built dist/ exercised as a real artefact +pnpm test:mutation # Stryker +``` diff --git a/packages/core/graph/sync/eslint.config.ts b/packages/core/graph/sync/eslint.config.ts new file mode 100644 index 0000000..4c43201 --- /dev/null +++ b/packages/core/graph/sync/eslint.config.ts @@ -0,0 +1,9 @@ +import { defineConfig } from "eslint/config"; +import { packageLintConfig } from "../../../../eslint.shared.ts"; + +export default defineConfig( + packageLintConfig({ + tsconfigRootDir: import.meta.dirname, + isomorphic: true, + }), +); diff --git a/packages/core/graph/sync/package.json b/packages/core/graph/sync/package.json new file mode 100644 index 0000000..b292253 --- /dev/null +++ b/packages/core/graph/sync/package.json @@ -0,0 +1,72 @@ +{ + "name": "@exadev/process-graph-sync", + "version": "0.0.0", + "description": "Syncs a GraphSource into a GraphStore page by page: participants, bound activities, transitions, relations and run state, resumable from a checkpoint, with an optional typed filter.", + "private": true, + "type": "module", + "repository": { + "type": "git", + "url": "git+https://github.com/ExaDev/process-graph.git", + "directory": "packages/core/graph/sync" + }, + "homepage": "https://github.com/ExaDev/process-graph/tree/main/packages/core/graph/sync#readme", + "bugs": { + "url": "https://github.com/ExaDev/process-graph/issues" + }, + "exports": { + ".": { + "types": "./dist/index.d.ts", + "import": "./dist/index.js" + } + }, + "types": "./dist/index.d.ts", + "files": [ + "dist" + ], + "sideEffects": false, + "license": "MIT", + "scripts": { + "build": "turbo run _build", + "_build": "tsc -b tsconfig.build.json", + "lint": "turbo run _lint", + "_lint": "eslint . --cache --max-warnings 0", + "_lint:fix": "eslint . --fix --cache --max-warnings 0", + "typecheck": "turbo run _typecheck _typecheck:attw", + "_typecheck": "tsc -p tsconfig.json --noEmit && tsc -p tsconfig.node.json --noEmit", + "_typecheck:attw": "attw --pack --profile esm-only", + "test": "turbo run _test", + "_test": "vitest run --project unit", + "test:coverage": "turbo run _test:coverage", + "_test:coverage": "vitest run --project unit --coverage", + "test:smoke": "turbo run _test:smoke", + "_test:smoke": "vitest run --project smoke", + "test:mutation": "turbo run _test:mutation", + "_test:mutation": "stryker run stryker.config.ts", + "prepublishOnly": "pnpm run lint && pnpm run typecheck && pnpm run test && pnpm run build && publint && attw --pack --profile esm-only" + }, + "keywords": [ + "process-model", + "sync", + "github-issues", + "resumable" + ], + "devDependencies": { + "@arethetypeswrong/cli": "catalog:", + "@orpc/server": "catalog:", + "@stryker-mutator/core": "catalog:", + "@stryker-mutator/typescript-checker": "catalog:", + "@stryker-mutator/vitest-runner": "catalog:", + "@types/node": "catalog:", + "@vitest/coverage-v8": "catalog:", + "eslint": "catalog:", + "publint": "catalog:", + "typescript": "catalog:", + "vitest": "catalog:" + }, + "dependencies": { + "@exadev/process-graph-source-contract": "workspace:*", + "@exadev/process-graph-store-contract": "workspace:*", + "@orpc/client": "catalog:", + "zod": "catalog:" + } +} diff --git a/packages/core/graph/sync/src/checkpoint.ts b/packages/core/graph/sync/src/checkpoint.ts new file mode 100644 index 0000000..7343db1 --- /dev/null +++ b/packages/core/graph/sync/src/checkpoint.ts @@ -0,0 +1,17 @@ +import { syncFilterSchema } from "@exadev/process-graph-source-contract"; +import { z } from "zod"; + +/** + * Where a sync stopped: the model, every scope it was asked for in order, the filter it read them with, the prefix of the source to read next, and the cursor after the last page it wrote within that source's scopes (absent to read them from the start). Everything before it is in the store, so a sync started from it reads only what is left. A source's cursor is valid only with the scopes and filter it was issued for, which is why both are kept with it. + */ +export const syncCheckpointSchema = z + .strictObject({ + modelId: z.string().min(1), + scopes: z.array(z.string().min(1)).readonly(), + filter: syncFilterSchema.optional(), + source: z.string().min(1), + after: z.string().min(1).optional(), + }) + .readonly(); + +export type SyncCheckpoint = z.infer; diff --git a/packages/core/graph/sync/src/index.ts b/packages/core/graph/sync/src/index.ts new file mode 100644 index 0000000..831c600 --- /dev/null +++ b/packages/core/graph/sync/src/index.ts @@ -0,0 +1,17 @@ +/** + * Syncs work from `GraphSource`s into a `GraphStore`: page by page, resumable from a checkpoint, with an optional typed filter, writing participants, bound activities, transitions, relations and run state. One function serves every consumer, so the command line and the web app write a sync the same way. + */ +export { + syncScopes, + UnknownSourceError, + type SourceLookup, + type SyncProgress, + type SyncRequest, + type SyncStore, +} from "./sync.ts"; +export { syncCheckpointSchema, type SyncCheckpoint } from "./checkpoint.ts"; +export { + InvalidScopeError, + scopesBySource, + type SourceScopes, +} from "./scopes.ts"; diff --git a/packages/core/graph/sync/src/scopes.ts b/packages/core/graph/sync/src/scopes.ts new file mode 100644 index 0000000..7a9c203 --- /dev/null +++ b/packages/core/graph/sync/src/scopes.ts @@ -0,0 +1,48 @@ +/** The separator between a scope's source prefix and the rest, as in `github:owner/repo`. */ +const sourceSeparator = ":"; + +/** The scopes of one source, in the order they were given. */ +export interface SourceScopes { + /** The source's identifier prefix, such as `github`. */ + readonly prefix: string; + readonly scopes: readonly string[]; +} + +/** Thrown for a scope that names no source: it has no prefix before a colon. */ +export class InvalidScopeError extends Error { + /** The scope as it was given. */ + readonly scope: string; + + /** + * @param scope - The scope that names no source. + */ + constructor(scope: string) { + super( + `"${scope}" is not a scope: write :, such as github:owner/repo`, + ); + this.name = "InvalidScopeError"; + this.scope = scope; + } +} + +/** + * Groups source-qualified scopes by their source prefix, keeping each prefix where it first appears and each source's scopes in the order given. A sync reads the groups in this order, one source after another. + * @param scopes - Source-qualified scopes, such as `github:owner` or `github:owner/repo`. + * @returns One group per source prefix. + * @throws InvalidScopeError for a scope with nothing before its first colon, or no colon at all. + */ +export const scopesBySource = ( + scopes: readonly string[], +): readonly SourceScopes[] => { + const groups = new Map(); + for (const scope of scopes) { + const separator = scope.indexOf(sourceSeparator); + if (separator <= 0) { + throw new InvalidScopeError(scope); + } + const prefix = scope.slice(0, separator); + groups.set(prefix, [...(groups.get(prefix) ?? []), scope]); + } + + return [...groups].map(([prefix, grouped]) => ({ prefix, scopes: grouped })); +}; diff --git a/packages/core/graph/sync/src/sync.ts b/packages/core/graph/sync/src/sync.ts new file mode 100644 index 0000000..c606c53 --- /dev/null +++ b/packages/core/graph/sync/src/sync.ts @@ -0,0 +1,292 @@ +import type { + Gap, + GraphSourceClient, + Relation, + SourceTransition, + SyncFilter, + SyncPage, +} from "@exadev/process-graph-source-contract"; +import type { GraphStoreClient } from "@exadev/process-graph-store-contract"; +import { ORPCError } from "@orpc/client"; +import type { SyncCheckpoint } from "./checkpoint.ts"; +import { scopesBySource } from "./scopes.ts"; + +/** Finds the source that reads scopes with an identifier prefix, such as `github`, or nothing when no source does. */ +export type SourceLookup = (prefix: string) => GraphSourceClient | undefined; + +/** The part of the `GraphStore` client a sync calls: loading the model it merges into and upserting each page. */ +export interface SyncStore { + readonly v1: { + readonly models: Pick; + readonly items: Pick; + }; +} + +/** How far a sync has got. Each count is of distinct identifiers (relations by kind and both ends) written by this sync. */ +export interface SyncProgress { + /** Pages read and written. */ + readonly pages: number; + /** The revision the last write took. */ + readonly revision: number; + readonly participants: number; + /** Bound activities. */ + readonly activities: number; + readonly transitions: number; + /** Relations: sub-issues to their parents, assignees, repositories to their owners. */ + readonly relations: number; + /** The parts of the scope the sources could not read, in the order met. */ + readonly gaps: readonly Gap[]; + /** Where to resume; absent once every scope has been read. */ + readonly checkpoint: SyncCheckpoint | undefined; +} + +/** What to sync and where to. */ +export interface SyncRequest { + readonly store: SyncStore; + readonly sources: SourceLookup; + readonly modelId: string; + /** Source-qualified scopes, read in order, one source after another. */ + readonly scopes: readonly string[]; + /** Narrows and orders what each scope lists. It is recorded in every checkpoint, since a cursor is valid only with the filter it was issued under. */ + readonly filter?: SyncFilter | undefined; + /** A checkpoint from an interrupted sync of the same model, scopes and filter: the sync carries on from it. */ + readonly resume?: SyncCheckpoint | undefined; + /** Ends the sync between or during reads; what was written stays written. */ + readonly signal?: AbortSignal | undefined; + /** Called after every page is written, with the progress so far. */ + readonly onProgress?: ((progress: SyncProgress) => void) | undefined; +} + +/** Thrown when a scope names a source no reader serves. */ +export class UnknownSourceError extends Error { + /** The source prefix no reader serves. */ + readonly prefix: string; + + /** + * @param prefix - The source prefix no reader serves. + */ + constructor(prefix: string) { + super(`No source reads "${prefix}" scopes.`); + this.name = "UnknownSourceError"; + this.prefix = prefix; + } +} + +/** The oRPC client options carrying a signal, leaving the key out when there is none, since the options type does not accept an explicit `undefined`. */ +const callOptions = ( + signal: AbortSignal | undefined, +): { readonly signal?: AbortSignal } => + signal === undefined ? {} : { signal }; + +/** The identifiers of the elements and participants a model in the store already has, or none when the store has no such model. */ +const storedIdentifiers = async ( + store: SyncStore, + modelId: string, + signal: AbortSignal | undefined, +): Promise> => { + try { + const stored = await store.v1.models.load({ modelId }, callOptions(signal)); + + return new Set([ + ...stored.model.elements.map((element) => element.id), + ...stored.model.participants.map((participant) => participant.id), + ]); + } catch (error) { + if (error instanceof ORPCError && error.code === "NOT_FOUND") { + return new Set(); + } + throw error; + } +}; + +const relationKey = (relation: Readonly): string => + JSON.stringify([relation.kind, relation.from, relation.to]); + +/** Splits what is waiting to be written into what can be written now, whose ends are all known, and what must wait for a later page. */ +const settle = ( + pending: readonly T[], + endsOf: (item: T) => readonly string[], + known: ReadonlySet, + last: boolean, +): { readonly ready: readonly T[]; readonly held: readonly T[] } => { + const ready = pending.filter( + (item) => last || endsOf(item).every((id) => known.has(id)), + ); + + return { ready, held: pending.filter((item) => !ready.includes(item)) }; +}; + +/** + * Reads the scopes from their sources page by page and merges each page into the stored model as soon as it is read: participants, bound activities, transitions, relations and run state. Progress is kept as it is made, so a sync that is cancelled, rate limited or interrupted leaves everything it read in the store, and a checkpoint to carry on from. + * + * A store rejects a write that leaves a transition or a relation naming an element or participant the model lacks, and a source may report a dependency or a sub-issue whose other end a later page brings. A transition or relation is therefore held back until the model has both of its ends, and written with the page that completes it. One whose end never arrives is written with the last page, where the store's rejection says which reference is missing. + * @param request - The store, sources, model, scopes, an optional filter and checkpoint, the signal and a progress listener. + * @returns The final progress, with no checkpoint. + * @throws InvalidScopeError for a scope that names no source; UnknownSourceError for a scope whose source no reader serves; an Error for a checkpoint naming a source none of the scopes use, and for a sync of no scopes; and whatever a source or the store throws, after the pages before it are written. An abort rejects with the signal's reason. + */ +export const syncScopes = async ( + request: SyncRequest, +): Promise => { + const { store, sources, modelId, scopes, filter, resume, signal } = request; + const groups = scopesBySource(scopes).map((group) => { + const client = sources(group.prefix); + if (client === undefined) { + throw new UnknownSourceError(group.prefix); + } + + return { ...group, client }; + }); + if (groups.length === 0) { + throw new Error("A sync needs at least one scope."); + } + const startGroup = + resume === undefined + ? 0 + : groups.findIndex((group) => group.prefix === resume.source); + if (startGroup === -1) { + throw new Error( + "The checkpoint names a source that none of the scopes use; start the sync again.", + ); + } + + const known = new Set(await storedIdentifiers(store, modelId, signal)); + const written = { + participants: new Set(), + activities: new Set(), + transitions: new Set(), + relations: new Set(), + }; + const gaps: Gap[] = []; + let heldTransitions: readonly SourceTransition[] = []; + let heldRelations: readonly Relation[] = []; + let pages = 0; + let latest: SyncProgress | undefined; + + const checkpointAt = ( + source: string, + after: string | undefined, + ): SyncCheckpoint => ({ + modelId, + scopes, + ...(filter === undefined ? {} : { filter }), + source, + ...(after === undefined ? {} : { after }), + }); + + const write = async ( + page: SyncPage, + checkpoint: SyncCheckpoint | undefined, + ): Promise => { + for (const record of [...page.participants, ...page.activities]) { + known.add(record.id); + } + const last = checkpoint === undefined; + const transitions = settle( + [...heldTransitions, ...page.transitions], + (transition) => [transition.source, transition.target], + known, + last, + ); + const relations = settle( + [...heldRelations, ...page.relations], + (relation) => [relation.from, relation.to], + known, + last, + ); + heldTransitions = transitions.held; + heldRelations = relations.held; + const { revision } = await store.v1.items.upsert( + { + modelId, + items: { + participants: page.participants, + activities: page.activities, + transitions: transitions.ready, + relations: relations.ready, + overlay: page.overlay, + }, + }, + callOptions(signal), + ); + pages += 1; + for (const participant of page.participants) { + written.participants.add(participant.id); + } + for (const activity of page.activities) { + written.activities.add(activity.id); + } + for (const transition of transitions.ready) { + written.transitions.add(transition.id); + } + for (const relation of relations.ready) { + written.relations.add(relationKey(relation)); + } + gaps.push(...page.gaps); + latest = { + pages, + revision, + participants: written.participants.size, + activities: written.activities.size, + transitions: written.transitions.size, + relations: written.relations.size, + gaps: [...gaps], + checkpoint, + }; + request.onProgress?.(latest); + }; + + /** Writes each page an iterator yields, in order, each only after the one before is written. */ + const drain = async ( + pagesOf: Readonly>, + checkpointAfter: (page: SyncPage) => SyncCheckpoint | undefined, + ): Promise => { + const next = await pagesOf.next(); + if (next.done === true) { + return; + } + await write(next.value, checkpointAfter(next.value)); + await drain(pagesOf, checkpointAfter); + }; + + /** Reads the sources in order, each only after the one before has finished, since a source's failure must stop the ones after it. */ + const readFrom = async (index: number): Promise => { + const group = groups[index]; + if (group === undefined) { + return; + } + const after = index === startGroup ? resume?.after : undefined; + const following = groups[index + 1]; + const stream = await group.client.v1.sync.stream( + { + scope: group.scopes, + ...(after === undefined ? {} : { after }), + ...(filter === undefined ? {} : { filter }), + }, + callOptions(signal), + ); + const iterator = stream[Symbol.asyncIterator](); + try { + await drain(iterator, (page) => + page.next !== undefined + ? checkpointAt(group.prefix, page.next) + : following === undefined + ? undefined + : checkpointAt(following.prefix, undefined), + ); + } catch (error) { + // Closing the stream stops the source reading pages nobody will write. + await iterator.return(); + throw error; + } + await readFrom(index + 1); + }; + await readFrom(startGroup); + + if (latest === undefined) { + throw new Error( + "unreachable: every source answers a sync with at least one page", + ); + } + + return latest; +}; diff --git a/packages/core/graph/sync/src/sync.unit.test.ts b/packages/core/graph/sync/src/sync.unit.test.ts new file mode 100644 index 0000000..6aa3577 --- /dev/null +++ b/packages/core/graph/sync/src/sync.unit.test.ts @@ -0,0 +1,429 @@ +import { + createFakeGraphSource, + GraphSourceError, + type GraphSourcePager, + type SyncFilter, +} from "@exadev/process-graph-source-contract"; +import { describe, expect, test } from "vitest"; +import { InvalidScopeError, scopesBySource } from "./scopes.ts"; +import { syncScopes, UnknownSourceError, type SyncProgress } from "./sync.ts"; +import { + apiPage, + fakeScopes, + lookup, + newStore, + pageOf, +} from "./test-support/fixtures.ts"; + +const fakeGitHub = createFakeGraphSource({ + source: "github", + scopes: fakeScopes, +}); + +const modelId = "github:acme"; + +describe("syncScopes", () => { + test("writes every page as it is read, reports progress after each, and ends with no checkpoint", async () => { + const store = newStore(); + const progress: SyncProgress[] = []; + + const result = await syncScopes({ + store, + sources: lookup(fakeGitHub), + modelId, + scopes: ["github:acme"], + onProgress: (step) => { + progress.push(step); + }, + }); + + expect( + progress.map((step) => [step.pages, step.checkpoint?.after]), + ).toEqual([ + [1, "github%3Aacme@0"], + [2, undefined], + ]); + expect(result).toEqual({ + pages: 2, + revision: progress.at(-1)?.revision, + participants: 4, + activities: 4, + transitions: 2, + relations: 5, + gaps: [], + checkpoint: undefined, + }); + const stored = await store.v1.models.load({ modelId }); + expect( + stored.model.elements.map((element) => element.id).toSorted(), + ).toEqual([ + "github:acme/api#1", + "github:acme/api#2", + "github:acme/web#1", + "github:acme/web#9", + ]); + expect(stored.overlay).toMatchObject({ + "github:acme/api#1": "done", + "github:acme/web#9": "open", + }); + }); + + test("keeps the relations a source reports, so the store answers hierarchy queries", async () => { + const store = newStore(); + + await syncScopes({ + store, + sources: lookup(fakeGitHub), + modelId, + scopes: ["github:acme"], + }); + + const { relations } = await store.v1.relations.list({ modelId }); + expect(relations).toEqual([ + { kind: "part-of", from: "github:acme/api", to: "github:@acme" }, + { kind: "assigned-to", from: "github:acme/api#1", to: "github:@alice" }, + { kind: "child-of", from: "github:acme/api#2", to: "github:acme/web#9" }, + { kind: "part-of", from: "github:acme/web", to: "github:@acme" }, + { kind: "child-of", from: "github:acme/web#1", to: "github:acme/web#9" }, + ]); + expect( + await store.v1.query.children({ modelId, element: "github:acme/web#9" }), + ).toEqual({ children: ["github:acme/api#2", "github:acme/web#1"] }); + expect( + await store.v1.query.ancestors({ modelId, element: "github:acme/api#2" }), + ).toEqual({ + ancestors: [{ element: "github:acme/web#9", distance: 1 }], + }); + }); + + test("holds a transition and a relation back until the page that brings their other end", async () => { + const store = newStore(); + const early = pageOf([ + { + id: "github:acme/web#1", + label: "Build the page", + repository: "github:acme/web", + state: "open", + blockedBy: ["github:acme/api#2"], + parent: "github:acme/api#2", + }, + ]); + const late = pageOf([ + { + id: "github:acme/api#2", + label: "Build the endpoint", + repository: "github:acme/api", + state: "open", + }, + ]); + const progress: SyncProgress[] = []; + + await syncScopes({ + store, + sources: lookup( + createFakeGraphSource({ + source: "github", + scopes: { "github:acme": [early, late] }, + }), + ), + modelId, + scopes: ["github:acme"], + onProgress: (step) => { + progress.push(step); + }, + }); + + expect(progress.map((step) => step.transitions)).toEqual([0, 1]); + // The first page's own repository membership is written at once; its sub-issue waits for the parent the second page brings. + expect(progress.map((step) => step.relations)).toEqual([ + early.relations.length - 1, + early.relations.length + late.relations.length, + ]); + const stored = await store.v1.models.load({ modelId }); + expect(stored.model.transitions).toEqual([ + { + id: "github:acme/api#2->github:acme/web#1", + source: "github:acme/api#2", + target: "github:acme/web#1", + }, + ]); + expect( + await store.v1.query.children({ modelId, element: "github:acme/api#2" }), + ).toEqual({ children: ["github:acme/web#1"] }); + }); + + test("writes a relation whose end never arrives with the last page, where the store names the missing reference", async () => { + const orphan = pageOf([ + { + id: "github:acme/web#1", + label: "Build the page", + repository: "github:acme/web", + state: "open", + parent: "github:acme/web#404", + }, + ]); + + await expect( + syncScopes({ + store: newStore(), + sources: lookup( + createFakeGraphSource({ + source: "github", + scopes: { "github:acme": [orphan] }, + }), + ), + modelId, + scopes: ["github:acme"], + }), + ).rejects.toMatchObject({ + code: "UNPROCESSABLE_CONTENT", + data: { + issues: [expect.objectContaining({ code: "unknown-reference" })], + }, + }); + }); + + test("resumes from a checkpoint with only the pages after it", async () => { + const store = newStore(); + const read: string[] = []; + const counting: GraphSourcePager = { + source: "github", + page: async (request) => { + read.push(request.cursor ?? "start"); + + return fakeGitHub.page(request); + }, + }; + await store.v1.items.upsert({ + modelId, + items: { + participants: apiPage.participants, + activities: apiPage.activities, + transitions: apiPage.transitions, + overlay: apiPage.overlay, + }, + }); + + const result = await syncScopes({ + store, + sources: lookup(counting), + modelId, + scopes: ["github:acme"], + resume: { + modelId, + scopes: ["github:acme"], + source: "github", + after: "github%3Aacme@0", + }, + }); + + expect(read).toEqual(["github%3Aacme@0"]); + expect(result.pages).toBe(1); + const stored = await store.v1.models.load({ modelId }); + expect( + stored.model.transitions.map((transition) => transition.id), + ).toContain("github:acme/api#2->github:acme/web#1"); + }); + + test("reads with the filter it is given and records it in every checkpoint", async () => { + const filters: (SyncFilter | undefined)[] = []; + const recording: GraphSourcePager = { + source: "github", + page: async (request) => { + filters.push(request.filter); + + return fakeGitHub.page(request); + }, + }; + const filter: SyncFilter = { + states: ["open"], + labels: ["bug"], + order: { field: "updated", direction: "descending" }, + }; + const progress: SyncProgress[] = []; + + await syncScopes({ + store: newStore(), + sources: lookup(recording), + modelId, + scopes: ["github:acme"], + filter, + onProgress: (step) => { + progress.push(step); + }, + }); + + expect(filters).toEqual([filter, filter]); + // The fake source's cursor carries the filter it was issued under, as the GitHub adapter's does. + const checkpoint = progress[0]?.checkpoint; + expect(checkpoint?.filter).toEqual(filter); + expect(checkpoint?.after?.startsWith("github%3Aacme@0;")).toBe(true); + }); + + test("keeps the pages before a failure, reports where to resume, and passes the failure on", async () => { + const store = newStore(); + const progress: SyncProgress[] = []; + const limited: GraphSourcePager = { + source: "github", + page: async (request) => { + if (request.cursor !== undefined) { + throw new GraphSourceError({ + kind: "rate-limited", + message: "rate limited", + resetAt: "2100-01-01T00:00:00.000Z", + }); + } + + return fakeGitHub.page(request); + }, + }; + + await expect( + syncScopes({ + store, + sources: lookup(limited), + modelId, + scopes: ["github:acme"], + onProgress: (step) => { + progress.push(step); + }, + }), + ).rejects.toMatchObject({ code: "TOO_MANY_REQUESTS" }); + expect(progress.at(-1)?.checkpoint).toEqual({ + modelId, + scopes: ["github:acme"], + source: "github", + after: "github%3Aacme@0", + }); + const stored = await store.v1.models.load({ modelId }); + expect(stored.model.elements.map((element) => element.id)).toEqual([ + "github:acme/api#1", + "github:acme/api#2", + ]); + }); + + test("reads the sources in order and checkpoints the start of the next source between them", async () => { + const progress: SyncProgress[] = []; + const jira = createFakeGraphSource({ + source: "jira", + scopes: { + "jira:ACME": [ + pageOf([ + { + id: "jira:ACME-1", + label: "Plan", + repository: "jira:ACME/board", + state: "open", + }, + ]), + ], + }, + }); + + await syncScopes({ + store: newStore(), + sources: lookup(fakeGitHub, jira), + modelId, + scopes: ["github:acme", "jira:ACME"], + onProgress: (step) => { + progress.push(step); + }, + }); + + expect(progress.map((step) => step.checkpoint)).toEqual([ + { + modelId, + scopes: ["github:acme", "jira:ACME"], + source: "github", + after: "github%3Aacme@0", + }, + { modelId, scopes: ["github:acme", "jira:ACME"], source: "jira" }, + undefined, + ]); + }); + + test("reports the parts of the scope the source could not read", async () => { + const result = await syncScopes({ + store: newStore(), + sources: lookup(fakeGitHub), + modelId: "github:acme github:ghost", + scopes: ["github:acme", "github:ghost"], + }); + + expect(result.gaps).toEqual([ + { + target: "github:ghost", + reason: "not-found", + message: "github:ghost does not exist", + }, + ]); + }); + + test("stops when its signal is aborted, keeping what it wrote", async () => { + const store = newStore(); + const controller = new AbortController(); + + const outcome = syncScopes({ + store, + sources: lookup(fakeGitHub), + modelId, + scopes: ["github:acme"], + signal: controller.signal, + onProgress: () => { + controller.abort(); + }, + }); + + await expect(outcome).rejects.toThrow(); + const stored = await store.v1.models.load({ modelId }); + expect(stored.model.elements).toHaveLength(2); + }); + + test("refuses a scope whose source no reader serves, and one that names no source", async () => { + await expect( + syncScopes({ + store: newStore(), + sources: lookup(fakeGitHub), + modelId: "linear:team", + scopes: ["linear:team"], + }), + ).rejects.toThrow(UnknownSourceError); + await expect( + syncScopes({ + store: newStore(), + sources: lookup(fakeGitHub), + modelId, + scopes: ["acme/app"], + }), + ).rejects.toThrow(InvalidScopeError); + }); + + test("refuses a checkpoint for a source none of the scopes use", async () => { + await expect( + syncScopes({ + store: newStore(), + sources: lookup(fakeGitHub), + modelId, + scopes: ["github:acme"], + resume: { modelId, scopes: ["github:acme"], source: "jira" }, + }), + ).rejects.toThrow("The checkpoint names a source"); + }); +}); + +describe("scopesBySource", () => { + test("groups scopes by source, each source where it first appears, its scopes in order", () => { + expect( + scopesBySource(["github:a", "jira:X", "github:b/c", "jira:Y"]), + ).toEqual([ + { prefix: "github", scopes: ["github:a", "github:b/c"] }, + { prefix: "jira", scopes: ["jira:X", "jira:Y"] }, + ]); + }); + + test("names a scope with nothing before its colon", () => { + expect(() => scopesBySource([":a"])).toThrow( + '":a" is not a scope: write :, such as github:owner/repo', + ); + }); +}); diff --git a/packages/core/graph/sync/src/test-support/fixtures.ts b/packages/core/graph/sync/src/test-support/fixtures.ts new file mode 100644 index 0000000..2b37ffd --- /dev/null +++ b/packages/core/graph/sync/src/test-support/fixtures.ts @@ -0,0 +1,155 @@ +import { + createGraphSourceRouter, + type FakePage, + type GraphSourcePager, + type Relation, +} from "@exadev/process-graph-source-contract"; +import { + createFakeGraphStore, + createGraphStoreRouter, + type GraphStoreClient, +} from "@exadev/process-graph-store-contract"; +import { createRouterClient } from "@orpc/server"; +import type { SourceLookup } from "../sync.ts"; + +/** One issue of the fake GitHub source. */ +export interface Issue { + readonly id: string; + readonly label: string; + readonly repository: string; + readonly state: "done" | "open"; + readonly blockedBy?: readonly string[]; + /** The parent issue, for a sub-issue. */ + readonly parent?: string; + /** The people assigned to it, as account participants. */ + readonly assignees?: readonly string[]; +} + +const ownerOf = (repository: string): string => + `github:@${repository.slice(repository.indexOf(":") + 1, repository.indexOf("/"))}`; + +/** + * A page of the fake GitHub source, built as the real adapter builds one: each issue a bound activity whose actor is its repository, each repository `part-of` its owner, each blocker a transition to the issue it blocks, each sub-issue `child-of` its parent, each assignee a role participant the issue is `assigned-to`, and the state in the overlay. + * @param issues - The page's issues. + * @returns The page. + */ +export const pageOf = (issues: readonly Issue[]): FakePage => { + const repositories = [...new Set(issues.map((issue) => issue.repository))]; + const assignees = [ + ...new Set(issues.flatMap((issue) => issue.assignees ?? [])), + ]; + const relations: Relation[] = [ + ...repositories.map((repository): Relation => ({ + kind: "part-of", + from: repository, + to: ownerOf(repository), + })), + ...issues.flatMap((issue): Relation[] => [ + ...(issue.parent === undefined + ? [] + : [{ kind: "child-of", from: issue.id, to: issue.parent } as const]), + ...(issue.assignees ?? []).map((person): Relation => ({ + kind: "assigned-to", + from: issue.id, + to: person, + })), + ]), + ]; + + return { + participants: [ + ...[...new Set(repositories.map(ownerOf))].map((id) => ({ + id, + label: id.slice(id.indexOf("@") + 1), + kind: "system" as const, + })), + ...repositories.map((id) => ({ + id, + label: id.slice(id.indexOf(":") + 1), + kind: "system" as const, + })), + ...assignees.map((id) => ({ + id, + label: id.slice(id.indexOf("@") + 1), + kind: "role" as const, + })), + ], + activities: issues.map((issue) => ({ + kind: "activity", + id: issue.id, + label: issue.label, + source: issue.id, + actor: issue.repository, + })), + transitions: issues.flatMap((issue) => + (issue.blockedBy ?? []).map((blocker) => ({ + id: `${blocker}->${issue.id}`, + source: blocker, + target: issue.id, + })), + ), + relations, + overlay: Object.fromEntries(issues.map((issue) => [issue.id, issue.state])), + gaps: [], + }; +}; + +/** The first page of the fake `github:acme` scope: the API repository's issues, the second a sub-issue of an epic the next page brings. */ +export const apiPage: FakePage = pageOf([ + { + id: "github:acme/api#1", + label: "Design the schema", + repository: "github:acme/api", + state: "done", + assignees: ["github:@alice"], + }, + { + id: "github:acme/api#2", + label: "Build the endpoint", + repository: "github:acme/api", + state: "open", + blockedBy: ["github:acme/api#1"], + parent: "github:acme/web#9", + }, +]); + +/** The second page of the fake `github:acme` scope: the web repository's issues, the epic among them, with a blocker in the API repository. */ +export const webPage: FakePage = pageOf([ + { + id: "github:acme/web#1", + label: "Build the page", + repository: "github:acme/web", + state: "open", + blockedBy: ["github:acme/api#2"], + parent: "github:acme/web#9", + }, + { + id: "github:acme/web#9", + label: "Launch", + repository: "github:acme/web", + state: "open", + }, +]); + +/** The fake GitHub source's scopes. */ +export const fakeScopes: Readonly> = { + "github:acme": [apiPage, webPage], +}; + +/** A fresh in-memory store, through its oRPC client. */ +export const newStore = (): GraphStoreClient => + createRouterClient(createGraphStoreRouter(createFakeGraphStore())); + +/** + * The source lookup that serves each pager's scopes, by its source prefix, through its oRPC client. + * @param pagers - The pagers. + */ +export const lookup = + (...pagers: readonly GraphSourcePager[]): SourceLookup => + (prefix) => { + const pager = pagers.find((candidate) => candidate.source === prefix); + + return pager === undefined + ? undefined + : createRouterClient(createGraphSourceRouter(pager)); + }; diff --git a/packages/core/graph/sync/stryker.config.ts b/packages/core/graph/sync/stryker.config.ts new file mode 100644 index 0000000..6429b27 --- /dev/null +++ b/packages/core/graph/sync/stryker.config.ts @@ -0,0 +1,3 @@ +import { packageStrykerConfig } from "../../../../stryker.shared.ts"; + +export default packageStrykerConfig(); diff --git a/packages/core/graph/sync/test/package.smoke.test.ts b/packages/core/graph/sync/test/package.smoke.test.ts new file mode 100644 index 0000000..735274f --- /dev/null +++ b/packages/core/graph/sync/test/package.smoke.test.ts @@ -0,0 +1,73 @@ +import { + createFakeGraphSource, + createGraphSourceRouter, +} from "@exadev/process-graph-source-contract"; +import { + createFakeGraphStore, + createGraphStoreRouter, +} from "@exadev/process-graph-store-contract"; +import { createRouterClient } from "@orpc/server"; +import { expect, test } from "vitest"; +// Deliberately imports the built dist/, not src/: this is the check that catches a wrong exports map or a build misconfiguration that a unit test importing straight from src/ never touches. +import { syncCheckpointSchema, syncScopes } from "../dist/index.js"; + +test("the built package syncs a source's pages and relations into a store", async () => { + const store = createRouterClient( + createGraphStoreRouter(createFakeGraphStore()), + ); + const source = createRouterClient( + createGraphSourceRouter( + createFakeGraphSource({ + source: "fake", + scopes: { + "fake:team": [ + { + participants: [ + { id: "fake:team", label: "team", kind: "system" }, + ], + activities: [ + { + kind: "activity", + id: "fake:1", + source: "fake:1", + label: "Parent", + actor: "fake:team", + }, + { + kind: "activity", + id: "fake:2", + source: "fake:2", + label: "Child", + actor: "fake:team", + }, + ], + transitions: [], + relations: [{ kind: "child-of", from: "fake:2", to: "fake:1" }], + overlay: { "fake:1": "open", "fake:2": "open" }, + gaps: [], + }, + ], + }, + }), + ), + ); + + const result = await syncScopes({ + store, + sources: (prefix) => (prefix === "fake" ? source : undefined), + modelId: "team", + scopes: ["fake:team"], + }); + + expect(result.relations).toBe(1); + expect( + await store.v1.query.children({ modelId: "team", element: "fake:1" }), + ).toEqual({ children: ["fake:2"] }); + expect( + syncCheckpointSchema.parse({ + modelId: "team", + scopes: ["fake:team"], + source: "fake", + }), + ).toEqual({ modelId: "team", scopes: ["fake:team"], source: "fake" }); +}); diff --git a/packages/core/graph/sync/tsconfig.build.json b/packages/core/graph/sync/tsconfig.build.json new file mode 100644 index 0000000..ba95c1f --- /dev/null +++ b/packages/core/graph/sync/tsconfig.build.json @@ -0,0 +1,20 @@ +{ + "extends": "./tsconfig.json", + "compilerOptions": { + "composite": true, + "noEmit": false, + "declaration": true, + "declarationMap": false, + "rootDir": "src", + "outDir": "dist", + "rewriteRelativeImportExtensions": true + }, + "references": [ + { + "path": "../source-contract/tsconfig.build.json" + }, + { + "path": "../store-contract/tsconfig.build.json" + } + ] +} diff --git a/packages/core/graph/sync/tsconfig.json b/packages/core/graph/sync/tsconfig.json new file mode 100644 index 0000000..8895dd1 --- /dev/null +++ b/packages/core/graph/sync/tsconfig.json @@ -0,0 +1,9 @@ +{ + "extends": "../../../../tsconfig.base.json", + "compilerOptions": { + "lib": ["ES2024", "WebWorker"], + "types": [] + }, + "include": ["src/**/*.ts"], + "exclude": ["src/**/*.test.ts", "src/test-support/**"] +} diff --git a/packages/core/graph/sync/tsconfig.node.json b/packages/core/graph/sync/tsconfig.node.json new file mode 100644 index 0000000..8a460f4 --- /dev/null +++ b/packages/core/graph/sync/tsconfig.node.json @@ -0,0 +1,8 @@ +{ + "extends": "../../../../tsconfig.base.json", + "compilerOptions": { + "lib": ["ES2024"], + "types": ["node"] + }, + "include": ["src/**/*.ts", "test/**/*.ts", "scripts/**/*.ts", "*.config.ts"] +} diff --git a/packages/core/graph/sync/vitest.config.ts b/packages/core/graph/sync/vitest.config.ts new file mode 100644 index 0000000..01b9391 --- /dev/null +++ b/packages/core/graph/sync/vitest.config.ts @@ -0,0 +1,16 @@ +import { defineConfig } from "vitest/config"; + +export default defineConfig({ + test: { + projects: [ + { + extends: true, + test: { name: "unit", include: ["src/**/*.test.ts"] }, + }, + { + extends: true, + test: { name: "smoke", include: ["test/package.smoke.test.ts"] }, + }, + ], + }, +}); diff --git a/packages/targets/process-graph/README.md b/packages/targets/process-graph/README.md index 4986dc1..999f6c5 100644 --- a/packages/targets/process-graph/README.md +++ b/packages/targets/process-graph/README.md @@ -50,6 +50,7 @@ The package is built from the repository's internal packages, which are bundled | `process-graph/store/node` | `openNodeGraphStore`: the SQLite store on Node's built-in `node:sqlite`; `prepareDataFile`: makes its data file and directory private before it opens. | Node | | `process-graph/store/wasm` | `connectGraphStore`: connects a page to the store a dedicated worker serves. | Browser | | `process-graph/store/wasm/worker` | `serveGraphStoreInWorker`: serves the store from a dedicated worker on SQLite wasm and OPFS, shared between tabs with no special headers. | Dedicated worker | +| `process-graph/sync` | `syncScopes`: syncs sources into a store page by page, relations included, resumable from a checkpoint and narrowed by a typed filter. | Browser, Worker, Node | | `process-graph/react` | Headless React hooks and unstyled components: follow a stored model, derive its run, project, filter and select. | Browser | | `process-graph/ui` | Styled components over `process-graph/react`, the SVG sequence diagram and the theme contract with light and dark themes. | Browser | | `process-graph/ui/theme` | The theme contract `vars`, `lightTheme`, `darkTheme`, `lightValues` and `darkValues` without the components, for a Vanilla Extract stylesheet that themes `process-graph/ui` without loading React. | Any | @@ -69,6 +70,7 @@ The package is built from the repository's internal packages, which are bundled | `store/node` | `drizzle-orm` | | `store/wasm` | `@orpc/client` | | `store/wasm/worker` | `@orpc/contract`, `@orpc/server`, `drizzle-orm`, `@sqlite.org/sqlite-wasm` | +| `sync` | `@orpc/client` | | `react` | `react` | | `ui` | `react`, `@vanilla-extract/recipes`, `@vanilla-extract/css` (types) | | `ui/theme` | `@vanilla-extract/css` (types) | diff --git a/packages/targets/process-graph/eslint.config.ts b/packages/targets/process-graph/eslint.config.ts index 0c4d89d..7f782d9 100644 --- a/packages/targets/process-graph/eslint.config.ts +++ b/packages/targets/process-graph/eslint.config.ts @@ -10,6 +10,7 @@ const subpathEntries = [ "src/store/node.ts", "src/store/wasm.ts", "src/store/wasm/worker.ts", + "src/sync.ts", "src/react.ts", "src/ui.ts", "src/ui/theme.ts", diff --git a/packages/targets/process-graph/package.json b/packages/targets/process-graph/package.json index 98195d2..3891191 100644 --- a/packages/targets/process-graph/package.json +++ b/packages/targets/process-graph/package.json @@ -48,6 +48,10 @@ "types": "./dist/store/wasm/worker.d.ts", "import": "./dist/store/wasm/worker.js" }, + "./sync": { + "types": "./dist/sync.d.ts", + "import": "./dist/sync.js" + }, "./react": { "types": "./dist/react.d.ts", "import": "./dist/react.js" @@ -131,6 +135,7 @@ "@exadev/process-graph-sqlite-wasm": "1.0.6", "@exadev/process-graph-store-contract": "1.2.1", "@exadev/process-graph-store-sqlite": "1.2.1", + "@exadev/process-graph-sync": "0.0.0", "@exadev/process-graph-ui": "1.1.0", "@orpc/client": "2.0.0-beta.41", "@orpc/contract": "2.0.0-beta.41", diff --git a/packages/targets/process-graph/src/entrypoints.unit.test.ts b/packages/targets/process-graph/src/entrypoints.unit.test.ts index 28068e0..331d3a6 100644 --- a/packages/targets/process-graph/src/entrypoints.unit.test.ts +++ b/packages/targets/process-graph/src/entrypoints.unit.test.ts @@ -8,6 +8,7 @@ import * as githubIssuesNodePackage from "@exadev/process-graph-source-github-is import * as sqliteNodePackage from "@exadev/process-graph-sqlite-node"; import * as storeContractPackage from "@exadev/process-graph-store-contract"; import * as storeSqlitePackage from "@exadev/process-graph-store-sqlite"; +import * as syncPackage from "@exadev/process-graph-sync"; import * as uiPackage from "@exadev/process-graph-ui"; import * as uiThemePackage from "@exadev/process-graph-ui/theme"; import { describe, expect, test } from "vitest"; @@ -19,6 +20,7 @@ import * as githubEntry from "./source/github.ts"; import * as githubNodeEntry from "./source/github/node.ts"; import * as storeEntry from "./store.ts"; import * as storeNodeEntry from "./store/node.ts"; +import * as syncEntry from "./sync.ts"; import * as uiEntry from "./ui.ts"; import * as uiThemeEntry from "./ui/theme.ts"; import * as versionModule from "./version.ts"; @@ -63,6 +65,11 @@ const subpaths: readonly Subpath[] = [ entry: storeNodeEntry, layers: [["sqlite-node", sqliteNodePackage]], }, + { + name: "sync", + entry: syncEntry, + layers: [["sync", syncPackage]], + }, { name: "react", entry: reactEntry, diff --git a/packages/targets/process-graph/src/sync.ts b/packages/targets/process-graph/src/sync.ts new file mode 100644 index 0000000..be9f9b8 --- /dev/null +++ b/packages/targets/process-graph/src/sync.ts @@ -0,0 +1,2 @@ +// The `process-graph/sync` entry point: `syncScopes`, which syncs `GraphSource`s into a `GraphStore` page by page, relations included, resumable from a checkpoint and narrowed by a typed filter. The command line and the web app both sync through it. It takes the source and store clients it is given, so it runs unchanged in a browser, a Worker or Node. +export * from "@exadev/process-graph-sync"; diff --git a/packages/targets/process-graph/test/package.smoke.test.ts b/packages/targets/process-graph/test/package.smoke.test.ts index 5596d41..d581f85 100644 --- a/packages/targets/process-graph/test/package.smoke.test.ts +++ b/packages/targets/process-graph/test/package.smoke.test.ts @@ -181,6 +181,35 @@ describe("each subpath, imported from the installed package", () => { ).toEqual(["function", "function"]); }); + test("process-graph/sync syncs a source into a store, its relations included", () => { + expect( + evaluate(` + import { createRouterClient } from "@orpc/server"; + import { createFakeGraphSource, createGraphSourceRouter } from "process-graph/source/github"; + import { createFakeGraphStore, createGraphStoreRouter } from "process-graph/store"; + import { syncScopes } from "process-graph/sync"; + const activity = (id, label) => ({ kind: "activity", id, source: id, label, actor: "fake:team" }); + const source = createRouterClient(createGraphSourceRouter(createFakeGraphSource({ + source: "fake", + scopes: { + "fake:team": [{ + participants: [{ id: "fake:team", label: "team", kind: "system" }], + activities: [activity("fake:1", "Parent"), activity("fake:2", "Child")], + transitions: [], + relations: [{ kind: "child-of", from: "fake:2", to: "fake:1" }], + overlay: {}, + gaps: [], + }], + }, + }))); + const store = createRouterClient(createGraphStoreRouter(createFakeGraphStore())); + const result = await syncScopes({ store, sources: () => source, modelId: "team", scopes: ["fake:team"] }); + const { children } = await store.v1.query.children({ modelId: "team", element: "fake:1" }); + console.log(JSON.stringify({ relations: result.relations, children })); + `), + ).toEqual({ relations: 1, children: ["fake:2"] }); + }); + test("process-graph/react, ui and react-flow are separate subpaths, the styled components apart from the headless ones", () => { expect( evaluate(` diff --git a/packages/targets/process-graph/tsdown.config.ts b/packages/targets/process-graph/tsdown.config.ts index f8bad10..60cb19e 100644 --- a/packages/targets/process-graph/tsdown.config.ts +++ b/packages/targets/process-graph/tsdown.config.ts @@ -25,6 +25,7 @@ export default defineConfig({ "store/node": "src/store/node.ts", "store/wasm": "src/store/wasm.ts", "store/wasm/worker": "src/store/wasm/worker.ts", + sync: "src/sync.ts", react: "src/react.ts", ui: "src/ui.ts", "ui/theme": "src/ui/theme.ts", diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index b300456..455b8b3 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -1005,6 +1005,55 @@ importers: specifier: 'catalog:' version: 4.1.11(@types/node@26.6.4)(@vitest/browser-playwright@4.1.11)(@vitest/coverage-v8@4.1.11)(happy-dom@20.14.5)(vite@8.3.2(@types/node@26.6.4)(esbuild@0.28.2)(jiti@2.7.0)(terser@5.51.2)(yaml@2.9.1)) + packages/core/graph/sync: + dependencies: + '@exadev/process-graph-source-contract': + specifier: workspace:* + version: link:../source-contract + '@exadev/process-graph-store-contract': + specifier: workspace:* + version: link:../store-contract + '@orpc/client': + specifier: 'catalog:' + version: 2.0.0-beta.41 + zod: + specifier: 'catalog:' + version: 4.6.5 + devDependencies: + '@arethetypeswrong/cli': + specifier: 'catalog:' + version: 0.18.5 + '@orpc/server': + specifier: 'catalog:' + version: 2.0.0-beta.41 + '@stryker-mutator/core': + specifier: 'catalog:' + version: 10.0.0(@types/node@26.6.4) + '@stryker-mutator/typescript-checker': + specifier: 'catalog:' + version: 10.0.0(@stryker-mutator/core@10.0.0(@types/node@26.6.4))(typescript@6.0.3) + '@stryker-mutator/vitest-runner': + specifier: 'catalog:' + version: 10.0.0(@stryker-mutator/core@10.0.0(@types/node@26.6.4))(vitest@4.1.11) + '@types/node': + specifier: 'catalog:' + version: 26.6.4 + '@vitest/coverage-v8': + specifier: 'catalog:' + version: 4.1.11(@vitest/browser@4.1.11)(vitest@4.1.11) + eslint: + specifier: 'catalog:' + version: 10.12.0(jiti@2.7.0)(supports-color@7.2.0) + publint: + specifier: 'catalog:' + version: 0.3.25 + typescript: + specifier: 'catalog:' + version: 6.0.3 + vitest: + specifier: 'catalog:' + version: 4.1.11(@types/node@26.6.4)(@vitest/browser-playwright@4.1.11)(@vitest/coverage-v8@4.1.11)(happy-dom@20.14.5)(vite@8.3.2(@types/node@26.6.4)(esbuild@0.28.2)(jiti@2.7.0)(terser@5.51.2)(yaml@2.9.1)) + packages/targets/graph-cli: dependencies: '@exadev/process-graph-dag': @@ -1226,6 +1275,9 @@ importers: '@exadev/process-graph-store-sqlite': specifier: 1.2.1 version: link:../../adapters/graph/store-sqlite + '@exadev/process-graph-sync': + specifier: 0.0.0 + version: link:../../core/graph/sync '@exadev/process-graph-ui': specifier: 1.1.0 version: link:../../ui/graph/ui diff --git a/tsconfig.build.json b/tsconfig.build.json index 952697f..da53964 100644 --- a/tsconfig.build.json +++ b/tsconfig.build.json @@ -25,6 +25,9 @@ { "path": "packages/core/graph/store-contract/tsconfig.build.json" }, + { + "path": "packages/core/graph/sync/tsconfig.build.json" + }, { "path": "packages/adapters/graph/store-sqlite/tsconfig.build.json" }, From c59ff208ff994be5b77f0443c9816c0ffce747a8 Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Sun, 11 Oct 2026 19:36:36 +0100 Subject: [PATCH 02/15] feat(sync): report a source's interruption with where to carry on A source that fails while it is read now ends syncScopes with an InterruptedSyncError: the source's failure as its message and cause, the checkpoint that continues the failing source after the last page read from it, the scopes of the sources not reached, and whether any page was written. A host such as the command line turns that into its own resume hint without re-deriving it from progress. A store's failure is still thrown as the store threw it. --- packages/core/graph/sync/README.md | 19 ++-- packages/core/graph/sync/src/index.ts | 2 + packages/core/graph/sync/src/sync.ts | 99 ++++++++++++++----- .../core/graph/sync/src/sync.unit.test.ts | 94 +++++++++++++++++- 4 files changed, 178 insertions(+), 36 deletions(-) diff --git a/packages/core/graph/sync/README.md b/packages/core/graph/sync/README.md index 4cc2e74..0c8d043 100644 --- a/packages/core/graph/sync/README.md +++ b/packages/core/graph/sync/README.md @@ -32,7 +32,7 @@ A store rejects a write that leaves a transition or relation naming an element o After every page, `onProgress` receives the pages written, the revision of the last write, the distinct participants, activities, transitions and relations written, the gaps met and a `SyncCheckpoint`. A checkpoint names the model, the scopes, the filter, the source to read next and the cursor after the last page written within it; passing it back as `resume` reads only what is left. A source's cursor is valid only with the scopes and filter it was issued for, which is why the checkpoint carries both. `syncCheckpointSchema` is its Zod schema, for a consumer that stores it. The final progress has no checkpoint. -A failure part way leaves every page before it written, and is passed on as the source or store threw it; the last progress's checkpoint says where to carry on. An abort rejects with the signal's reason. +A failure part way leaves every page before it written. A source that fails while it is read, an abort included, ends the sync with an `InterruptedSyncError`: its message and cause are the source's failure, `resume` is the checkpoint that continues the failing source after the last page read from it, `remaining` the scopes of the sources after it that were not reached, and `saved` whether any page was written. A store's failure is thrown as the store threw it. ## Filters @@ -40,14 +40,15 @@ A failure part way leaves every page before it written, and is passed on as the ## Functions -| Export | What it is | -| ---------------------- | ----------------------------------------------------------------------------------- | -| `syncScopes(request)` | Syncs the scopes into the store and resolves with the final progress. | -| `scopesBySource` | Groups source-qualified scopes by source prefix, in the order a sync reads them. | -| `syncCheckpointSchema` | The Zod schema of a `SyncCheckpoint`. | -| `UnknownSourceError` | Thrown for a scope whose source prefix no reader serves; carries the `prefix`. | -| `InvalidScopeError` | Thrown for a scope that names no source; carries the `scope`. | -| `SyncStore` | The part of the `GraphStore` client a sync calls: `models.load` and `items.upsert`. | +| Export | What it is | +| ---------------------- | -------------------------------------------------------------------------------------------------------------------------- | +| `syncScopes(request)` | Syncs the scopes into the store and resolves with the final progress. | +| `scopesBySource` | Groups source-qualified scopes by source prefix, in the order a sync reads them. | +| `syncCheckpointSchema` | The Zod schema of a `SyncCheckpoint`. | +| `InterruptedSyncError` | Thrown when a source fails while it is read; carries `resume`, `remaining`, `saved` and the source's failure as its cause. | +| `UnknownSourceError` | Thrown for a scope whose source prefix no reader serves; carries the `prefix`. | +| `InvalidScopeError` | Thrown for a scope that names no source; carries the `scope`. | +| `SyncStore` | The part of the `GraphStore` client a sync calls: `models.load` and `items.upsert`. | ## Commands diff --git a/packages/core/graph/sync/src/index.ts b/packages/core/graph/sync/src/index.ts index 831c600..6cce107 100644 --- a/packages/core/graph/sync/src/index.ts +++ b/packages/core/graph/sync/src/index.ts @@ -2,8 +2,10 @@ * Syncs work from `GraphSource`s into a `GraphStore`: page by page, resumable from a checkpoint, with an optional typed filter, writing participants, bound activities, transitions, relations and run state. One function serves every consumer, so the command line and the web app write a sync the same way. */ export { + InterruptedSyncError, syncScopes, UnknownSourceError, + type SyncInterruption, type SourceLookup, type SyncProgress, type SyncRequest, diff --git a/packages/core/graph/sync/src/sync.ts b/packages/core/graph/sync/src/sync.ts index c606c53..3ec7104 100644 --- a/packages/core/graph/sync/src/sync.ts +++ b/packages/core/graph/sync/src/sync.ts @@ -72,6 +72,39 @@ export class UnknownSourceError extends Error { } } +/** Where a sync a source interrupted carries on, as data. */ +export interface SyncInterruption { + /** The checkpoint that continues the failing source: its prefix, after the cursor of the last page read from it, or from the start of its scopes when none was read. */ + readonly resume: SyncCheckpoint; + /** The scopes of the sources after the failing one, which were not read. */ + readonly remaining: readonly string[]; + /** Whether this sync wrote any pages before the failure. */ + readonly saved: boolean; +} + +/** + * Thrown by `syncScopes` when a source fails while it is being read: opening its stream or reading a page. Its message is the source's, its cause is what the source threw, and where to carry on is data in `resume` and `remaining`. A failure of the store, or of the request itself (an unknown source, a malformed scope), is thrown as it is. + */ +export class InterruptedSyncError extends Error implements SyncInterruption { + readonly resume: SyncCheckpoint; + + readonly remaining: readonly string[]; + + readonly saved: boolean; + + /** + * @param interruption - Where the sync carries on. + * @param cause - What the source threw. + */ + constructor(interruption: SyncInterruption, cause: unknown) { + super(cause instanceof Error ? cause.message : String(cause), { cause }); + this.name = "InterruptedSyncError"; + this.resume = interruption.resume; + this.remaining = interruption.remaining; + this.saved = interruption.saved; + } +} + /** The oRPC client options carrying a signal, leaving the key out when there is none, since the options type does not accept an explicit `undefined`. */ const callOptions = ( signal: AbortSignal | undefined, @@ -122,7 +155,7 @@ const settle = ( * A store rejects a write that leaves a transition or a relation naming an element or participant the model lacks, and a source may report a dependency or a sub-issue whose other end a later page brings. A transition or relation is therefore held back until the model has both of its ends, and written with the page that completes it. One whose end never arrives is written with the last page, where the store's rejection says which reference is missing. * @param request - The store, sources, model, scopes, an optional filter and checkpoint, the signal and a progress listener. * @returns The final progress, with no checkpoint. - * @throws InvalidScopeError for a scope that names no source; UnknownSourceError for a scope whose source no reader serves; an Error for a checkpoint naming a source none of the scopes use, and for a sync of no scopes; and whatever a source or the store throws, after the pages before it are written. An abort rejects with the signal's reason. + * @throws InvalidScopeError for a scope that names no source; UnknownSourceError for a scope whose source no reader serves; an Error for a checkpoint naming a source none of the scopes use, and for a sync of no scopes; InterruptedSyncError, after the pages before it are written, when a source fails while it is read, an abort included; and whatever the store throws. */ export const syncScopes = async ( request: SyncRequest, @@ -235,44 +268,60 @@ export const syncScopes = async ( request.onProgress?.(latest); }; - /** Writes each page an iterator yields, in order, each only after the one before is written. */ - const drain = async ( - pagesOf: Readonly>, - checkpointAfter: (page: SyncPage) => SyncCheckpoint | undefined, - ): Promise => { - const next = await pagesOf.next(); - if (next.done === true) { - return; - } - await write(next.value, checkpointAfter(next.value)); - await drain(pagesOf, checkpointAfter); - }; - /** Reads the sources in order, each only after the one before has finished, since a source's failure must stop the ones after it. */ const readFrom = async (index: number): Promise => { const group = groups[index]; if (group === undefined) { return; } - const after = index === startGroup ? resume?.after : undefined; + // The cursor after the last page read from this source, which is where it carries on if it fails. + let after = index === startGroup ? resume?.after : undefined; const following = groups[index + 1]; - const stream = await group.client.v1.sync.stream( - { - scope: group.scopes, - ...(after === undefined ? {} : { after }), - ...(filter === undefined ? {} : { filter }), - }, - callOptions(signal), - ); + const interrupted = (cause: unknown): InterruptedSyncError => + new InterruptedSyncError( + { + resume: checkpointAt(group.prefix, after), + remaining: groups.slice(index + 1).flatMap((next) => next.scopes), + saved: pages > 0, + }, + cause, + ); + const stream = await group.client.v1.sync + .stream( + { + scope: group.scopes, + ...(after === undefined ? {} : { after }), + ...(filter === undefined ? {} : { filter }), + }, + callOptions(signal), + ) + .catch((cause: unknown) => { + throw interrupted(cause); + }); const iterator = stream[Symbol.asyncIterator](); - try { - await drain(iterator, (page) => + + /** Writes each page the source yields, in order, each only after the one before is written. */ + const drain = async (): Promise => { + const next = await iterator.next().catch((cause: unknown) => { + throw interrupted(cause); + }); + if (next.done === true) { + return; + } + const page = next.value; + await write( + page, page.next !== undefined ? checkpointAt(group.prefix, page.next) : following === undefined ? undefined : checkpointAt(following.prefix, undefined), ); + after = page.next; + await drain(); + }; + try { + await drain(); } catch (error) { // Closing the stream stops the source reading pages nobody will write. await iterator.return(); diff --git a/packages/core/graph/sync/src/sync.unit.test.ts b/packages/core/graph/sync/src/sync.unit.test.ts index 6aa3577..72b9dbc 100644 --- a/packages/core/graph/sync/src/sync.unit.test.ts +++ b/packages/core/graph/sync/src/sync.unit.test.ts @@ -6,7 +6,12 @@ import { } from "@exadev/process-graph-source-contract"; import { describe, expect, test } from "vitest"; import { InvalidScopeError, scopesBySource } from "./scopes.ts"; -import { syncScopes, UnknownSourceError, type SyncProgress } from "./sync.ts"; +import { + InterruptedSyncError, + syncScopes, + UnknownSourceError, + type SyncProgress, +} from "./sync.ts"; import { apiPage, fakeScopes, @@ -288,7 +293,19 @@ describe("syncScopes", () => { progress.push(step); }, }), - ).rejects.toMatchObject({ code: "TOO_MANY_REQUESTS" }); + ).rejects.toMatchObject({ + name: "InterruptedSyncError", + message: "rate limited", + resume: { + modelId, + scopes: ["github:acme"], + source: "github", + after: "github%3Aacme@0", + }, + remaining: [], + saved: true, + cause: { code: "TOO_MANY_REQUESTS" }, + }); expect(progress.at(-1)?.checkpoint).toEqual({ modelId, scopes: ["github:acme"], @@ -302,6 +319,79 @@ describe("syncScopes", () => { ]); }); + test("reports a source that fails before its first page with the cursor it was resumed from, and the sources it did not reach", async () => { + const refusing: GraphSourcePager = { + source: "github", + page: async () => + Promise.reject( + new GraphSourceError({ kind: "unauthorised", message: "bad token" }), + ), + }; + const jira = createFakeGraphSource({ + source: "jira", + scopes: { "jira:ACME": [] }, + }); + + const failure = await syncScopes({ + store: newStore(), + sources: lookup(refusing, jira), + modelId, + scopes: ["github:acme", "jira:ACME", "github:partner"], + resume: { + modelId, + scopes: ["github:acme", "jira:ACME", "github:partner"], + source: "github", + after: "github%3Aacme@0", + }, + }).then( + () => undefined, + (error: unknown) => error, + ); + + expect(failure).toBeInstanceOf(InterruptedSyncError); + expect(failure).toMatchObject({ + message: "bad token", + resume: { + source: "github", + scopes: ["github:acme", "jira:ACME", "github:partner"], + after: "github%3Aacme@0", + }, + remaining: ["jira:ACME"], + saved: false, + cause: { code: "UNAUTHORIZED" }, + }); + }); + + test("passes a store's refusal on as it is, not as an interruption of the source", async () => { + const orphan = pageOf([ + { + id: "github:acme/web#1", + label: "Build the page", + repository: "github:acme/web", + state: "open", + parent: "github:acme/web#404", + }, + ]); + + const failure = await syncScopes({ + store: newStore(), + sources: lookup( + createFakeGraphSource({ + source: "github", + scopes: { "github:acme": [orphan] }, + }), + ), + modelId, + scopes: ["github:acme"], + }).then( + () => undefined, + (error: unknown) => error, + ); + + expect(failure).not.toBeInstanceOf(InterruptedSyncError); + expect(failure).toMatchObject({ code: "UNPROCESSABLE_CONTENT" }); + }); + test("reads the sources in order and checkpoints the start of the next source between them", async () => { const progress: SyncProgress[] = []; const jira = createFakeGraphSource({ From 690945e2035c2b8f04a7ee5ebd5e1ffd94f6dd7e Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Sun, 11 Oct 2026 13:47:45 +0100 Subject: [PATCH 03/15] feat(cli): keep relations on sync and add children and ancestors sync now runs through graph-sync, so the relations a source reports reach the store, and each page is written as it is read rather than in one write at the end. The revision a sync reports is therefore the revision of its last page's write. children and ancestors read the stored child-of relations through the store's hierarchy queries: an element's direct sub-tasks, and what it is a sub-task of, nearest first with the distance. --- packages/targets/graph-cli/README.md | 8 +- packages/targets/graph-cli/package.json | 1 + .../targets/graph-cli/src/cli.unit.test.ts | 99 +++++++- packages/targets/graph-cli/src/index.ts | 7 +- packages/targets/graph-cli/src/program.ts | 41 +++- packages/targets/graph-cli/src/queries.ts | 70 ++++++ packages/targets/graph-cli/src/render.ts | 16 +- packages/targets/graph-cli/src/sync.ts | 214 ++++++++---------- .../targets/graph-cli/src/sync.unit.test.ts | 3 +- .../targets/graph-cli/tsconfig.build.json | 3 + pnpm-lock.yaml | 3 + 11 files changed, 326 insertions(+), 139 deletions(-) diff --git a/packages/targets/graph-cli/README.md b/packages/targets/graph-cli/README.md index 16e20a9..893d963 100644 --- a/packages/targets/graph-cli/README.md +++ b/packages/targets/graph-cli/README.md @@ -19,10 +19,12 @@ process-graph show --view swimlane --format mermaid | Command | What it does | | --------------------------------------------------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `sync [--after ]` | Reads the scopes from their sources and merges them into the model in one write; prints what it wrote and any gaps. | +| `sync [--after ]` | Reads the scopes from their sources and merges each page into the model as it is read, relations included; prints what it wrote and any gaps. | | `ready` | The work that can start now. | | `blocked ` | The open work an element is waiting on: the ends of its blocked-by chain. | | `impact ` | The live work that depends on an element. | +| `children ` | The element's direct sub-tasks: the work a source reports as its children, such as GitHub sub-issues. | +| `ancestors ` | What the element is a sub-task of, its parent first and then each level up, with how many levels up each is. | | `path` | The critical path: the longest chain of work still ahead, each element counting one step. | | `cycles` | The cycles no decision closes. | | `handoffs` | The transitions from one actor's work to another's, such as a dependency across repositories or organisations. | @@ -81,7 +83,7 @@ The Mermaid emitters name nodes by their position in the view, since Mermaid doe Scopes are source-qualified: `github:owner` reads every repository an account owns and `github:owner/repo` one repository. The GitHub token comes from `GITHUB_TOKEN`, or from the GitHub CLI (`gh auth token`) when that is not set; it is sent only to GitHub and never printed. -A sync merges everything it reads into the model in one write, so a record synced again is replaced by identifier and the rest of the model is kept. When a source fails part way, as on a rate limit, the pages read before the failure are still written and the error prints the command that resumes the sync with `--after`. Relations a source reports (sub-issues, assignees, repository membership) have no place in the store and are not kept. +A sync merges each page into the model as it reads it, so a record synced again is replaced by identifier and the rest of the model is kept. It writes through `graph-sync`, the same sync the web app runs: participants, issues, dependencies, run state and the relations a source reports (sub-issues, assignees, repository membership), which `children` and `ancestors` read. When a source fails part way, as on a rate limit, the pages read before the failure are already written and the error prints the command that resumes the sync with `--after`. ## Library @@ -99,7 +101,7 @@ process.exitCode = await runCli(args, nodePorts(), { }); ``` -The operation behind each command is exported on its own and returns plain data rather than text, for hosts that want results, such as tools served over MCP: `sync`, `ready`, `blocked`, `impactOf`, `path`, `cycles`, `handoffsOf`, `why`, `show`, `add`, `setState`, `setArtefact`, `setFact`, `factsOf`, `validateStored` and `validateDocument`, each taking a `GraphSession` (a store client, a model id, the host's facts and a signal). The emitters `flowchartToMermaid`, `swimlaneToMermaid`, `stateMachineToMermaid`, `sequenceToMermaid` and their `...ToDot` counterparts are pure functions of a projection. +The operation behind each command is exported on its own and returns plain data rather than text, for hosts that want results, such as tools served over MCP: `sync`, `ready`, `blocked`, `impactOf`, `children`, `ancestors`, `path`, `cycles`, `handoffsOf`, `why`, `show`, `add`, `setState`, `setArtefact`, `setFact`, `factsOf`, `validateStored` and `validateDocument`, each taking a `GraphSession` (a store client, a model id, the host's facts and a signal). The emitters `flowchartToMermaid`, `swimlaneToMermaid`, `stateMachineToMermaid`, `sequenceToMermaid` and their `...ToDot` counterparts are pure functions of a projection. When a source fails part way, `sync` throws a `SyncInterruptedError`: a `CliError` with the source's failure as its message and exit status (a token port that cannot supply a token is a source failure, status 5), the error underneath it as its `cause`, and where to carry on as data rather than text: `resume`, the `SyncRequest` that continues the failing source after the cursor of its last page read, `remaining`, the scopes of the sources not reached, and `saved`, whether the pages read before the failure were written. The command line renders those as the `resume with:` and `then:` commands it prints. diff --git a/packages/targets/graph-cli/package.json b/packages/targets/graph-cli/package.json index 4f2a764..ccd08c1 100644 --- a/packages/targets/graph-cli/package.json +++ b/packages/targets/graph-cli/package.json @@ -75,6 +75,7 @@ "@exadev/process-graph-source-github-issues": "workspace:*", "@exadev/process-graph-sqlite-node": "workspace:*", "@exadev/process-graph-store-contract": "workspace:*", + "@exadev/process-graph-sync": "workspace:*", "@orpc/server": "catalog:", "commander": "catalog:", "trilean": "catalog:" diff --git a/packages/targets/graph-cli/src/cli.unit.test.ts b/packages/targets/graph-cli/src/cli.unit.test.ts index 4ea74cf..a27d698 100644 --- a/packages/targets/graph-cli/src/cli.unit.test.ts +++ b/packages/targets/graph-cli/src/cli.unit.test.ts @@ -1,6 +1,7 @@ import { createFakeGraphSource, GraphSourceError, + type FakePage, type GraphSourcePager, } from "@exadev/process-graph-source-contract"; import { afterEach, beforeEach, describe, expect, test } from "vitest"; @@ -35,7 +36,7 @@ describe("sync", () => { expect(result).toEqual({ code: exitCodes.success, stdout: - 'synced 4 activities, 4 transitions and 4 participants into "default" at revision 1\n' + + 'synced 4 activities, 4 transitions, 3 relations and 4 participants into "default" at revision 4\n' + "gap\tnot-found\tgithub:acme/missing\tgithub:acme/missing does not exist\n", stderr: "", }); @@ -46,10 +47,11 @@ describe("sync", () => { expect(JSON.parse(result.stdout)).toEqual({ modelId: "default", - revision: 1, + revision: 2, participants: 4, activities: 4, transitions: 4, + relations: 2, gaps: [], }); }); @@ -59,7 +61,7 @@ describe("sync", () => { const again = await harness.run("sync", "github:acme/web"); expect(again.stdout).toBe( - 'synced 1 activities, 0 transitions and 2 participants into "default" at revision 2\n', + 'synced 1 activities, 0 transitions, 1 relations and 2 participants into "default" at revision 4\n', ); expect((await harness.run("ready")).stdout).toBe(`${api}\tBuild the API\n`); }); @@ -150,6 +152,95 @@ describe("sync", () => { }); }); +describe("hierarchy", () => { + const epic = { + kind: "activity", + id: "github:acme/app#10", + source: "github:acme/app#10", + label: "Launch", + actor: "github:acme/app", + } as const; + const task = (number: string, label: string) => + ({ + kind: "activity", + id: `github:acme/app#${number}`, + source: `github:acme/app#${number}`, + label, + actor: "github:acme/app", + }) as const; + const story = task("11", "Sign-up flow"); + const subtask = task("12", "Validate the email"); + // The sub-task arrives a page before its parent, as a sub-issue in another repository can. + const pages: readonly FakePage[] = [ + { + participants: [ + { id: "github:acme/app", label: "acme/app", kind: "system" }, + ], + activities: [subtask], + transitions: [], + relations: [{ kind: "child-of", from: subtask.id, to: story.id }], + overlay: { [subtask.id]: "open" }, + gaps: [], + }, + { + participants: [ + { id: "github:acme/app", label: "acme/app", kind: "system" }, + ], + activities: [epic, story], + transitions: [], + relations: [{ kind: "child-of", from: story.id, to: epic.id }], + overlay: { [epic.id]: "open", [story.id]: "open" }, + gaps: [], + }, + ]; + + beforeEach(async () => { + harness.remove(); + harness = createHarness( + createFakeGraphSource({ + source: "github", + scopes: { "github:acme/app": pages }, + }), + ); + await harness.run("sync", "github:acme/app"); + }); + + test("children lists an element's sub-tasks from the relations sync kept", async () => { + expect(await harness.run("children", epic.id)).toEqual({ + code: exitCodes.success, + stdout: `${story.id}\tSign-up flow\n`, + stderr: "", + }); + expect((await harness.run("children", subtask.id)).stdout).toBe(""); + }); + + test("ancestors lists what an element is a sub-task of, nearest first, with the distance", async () => { + expect((await harness.run("ancestors", subtask.id)).stdout).toBe( + `${story.id}\tSign-up flow\t1\n${epic.id}\tLaunch\t2\n`, + ); + expect( + JSON.parse((await harness.run("ancestors", subtask.id, "--json")).stdout), + ).toEqual({ + element: { id: subtask.id, label: "Validate the email" }, + ancestors: [ + { id: story.id, label: "Sign-up flow", distance: 1 }, + { id: epic.id, label: "Launch", distance: 2 }, + ], + }); + }); + + test.each(["children", "ancestors"])( + "%s of an element the model lacks is not found", + async (command) => { + expect(await harness.run(command, "github:acme/app#99")).toEqual({ + code: exitCodes.notFound, + stdout: "", + stderr: 'no element "github:acme/app#99" in the model\n', + }); + }, + ); +}); + describe("reasoning over a synced model", () => { beforeEach(async () => { await syncFixture(); @@ -247,7 +338,7 @@ describe("reasoning over a synced model", () => { test("set changes an element's run state and the reasoning follows it", async () => { expect((await harness.run("set", api, "done")).stdout).toBe( - `set ${api} to done at revision 2\n`, + `set ${api} to done at revision 4\n`, ); expect((await harness.run("ready")).stdout).toBe(`${ui}\tBuild the UI\n`); }); diff --git a/packages/targets/graph-cli/src/index.ts b/packages/targets/graph-cli/src/index.ts index 083b425..48ddb44 100644 --- a/packages/targets/graph-cli/src/index.ts +++ b/packages/targets/graph-cli/src/index.ts @@ -21,17 +21,22 @@ export { type StoredRun, } from "./session.ts"; export { + ancestors, blocked, + children, cycles, handoffsOf, impactOf, path, ready, why, + type AncestorsResult, type BlockedResult, + type ChildrenResult, type CyclesResult, type ElementStatus, type HandoffsResult, + type HierarchyLink, type ImpactResult, type PathResult, type ReadyResult, @@ -74,11 +79,11 @@ export { export { sync, SyncInterruptedError, - type SourceLookup, type SyncInterruption, type SyncRequest, type SyncResult, } from "./sync.ts"; +export type { SourceLookup } from "@exadev/process-graph-sync"; export { flowchartToMermaid, sequenceToMermaid, diff --git a/packages/targets/graph-cli/src/program.ts b/packages/targets/graph-cli/src/program.ts index 05dd6eb..e0ac48c 100644 --- a/packages/targets/graph-cli/src/program.ts +++ b/packages/targets/graph-cli/src/program.ts @@ -20,7 +20,9 @@ import { CliError, failureOf } from "./errors.ts"; import { exitCodes, type ExitCode } from "./exit-codes.ts"; import type { CliPorts } from "./ports.ts"; import { + ancestors, blocked, + children, cycles, handoffsOf, impactOf, @@ -30,7 +32,9 @@ import { } from "./queries.ts"; import { renderAdded, + renderAncestors, renderBlocked, + renderChildren, renderFacts, renderCycles, renderHandoffs, @@ -248,11 +252,10 @@ export const runCli = async ( const { after } = options; await withSession(async (session) => { try { - const result = await sync( - session, - ports.source, - typeof after === "string" ? { scopes, after } : { scopes }, - ); + const result = await sync(session, ports.source, { + scopes, + ...(typeof after === "string" ? { after } : {}), + }); return { json: result, text: renderSync(result) }; } catch (error) { @@ -305,6 +308,34 @@ export const runCli = async ( })(command); }); + program + .command("children") + .description( + "list an element's sub-tasks: the work a source reports as its children", + ) + .argument("", "the element") + .action(async (id: string, _options: unknown, command: Command) => { + await withSession(async (session) => { + const result = await children(session, id); + + return { json: result, text: renderChildren(result) }; + })(command); + }); + + program + .command("ancestors") + .description( + "list what an element is a sub-task of, nearest first, with how many levels up each is", + ) + .argument("", "the element") + .action(async (id: string, _options: unknown, command: Command) => { + await withSession(async (session) => { + const result = await ancestors(session, id); + + return { json: result, text: renderAncestors(result) }; + })(command); + }); + program .command("path") .description( diff --git a/packages/targets/graph-cli/src/queries.ts b/packages/targets/graph-cli/src/queries.ts index 5c95101..6ccf62f 100644 --- a/packages/targets/graph-cli/src/queries.ts +++ b/packages/targets/graph-cli/src/queries.ts @@ -14,6 +14,7 @@ import { import { handoffs, type Handoff } from "@exadev/process-graph-projection"; import type { OpenIds } from "@exadev/process-graph-schema"; import { + callOptions, labelOf, loadRun, requireElement, @@ -85,6 +86,25 @@ export interface WhyResult { readonly chain: readonly WhyLink[]; } +/** An element of a hierarchy walk, with how many `child-of` relations away it is. */ +export interface HierarchyLink extends ElementSummary { + readonly distance: number; +} + +/** An element's sub-tasks: the elements related to it by `child-of`. */ +export interface ChildrenResult { + readonly element: ElementSummary; + /** Its direct children, in model order. */ + readonly children: readonly ElementSummary[]; +} + +/** What an element is a sub-task of, up to the top of its hierarchy. */ +export interface AncestorsResult { + readonly element: ElementSummary; + /** Its parent, its parent's parent and so on, nearest first, each once with the fewest relations it takes to reach it. */ + readonly ancestors: readonly HierarchyLink[]; +} + /** Each element counts as one step, so the critical path is the longest chain of elements. */ const stepWeight = 1; @@ -229,3 +249,53 @@ export const why = async ( chain, }; }; + +/** + * An element's direct sub-tasks, from the `child-of` relations a sync keeps, as the store's hierarchy query answers. + * @param session - The store and model. + * @param id - The element asked about. + * @throws CliError with `notFound` when the model or the element does not exist. + */ +export const children = async ( + session: GraphSession, + id: string, +): Promise => { + const { model } = await loadRun(session); + const element = requireElement(model, id); + const found = await session.store.v1.query.children( + { modelId: session.modelId, element: id }, + callOptions(session.signal), + ); + + return { + element: { id, label: labelOf(element) }, + children: summarise(model, found.children), + }; +}; + +/** + * The elements an element is a sub-task of, nearest first, from the `child-of` relations a sync keeps, as the store's hierarchy query answers. + * @param session - The store and model. + * @param id - The element asked about. + * @throws CliError with `notFound` when the model or the element does not exist. + */ +export const ancestors = async ( + session: GraphSession, + id: string, +): Promise => { + const { model } = await loadRun(session); + const element = requireElement(model, id); + const found = await session.store.v1.query.ancestors( + { modelId: session.modelId, element: id }, + callOptions(session.signal), + ); + + return { + element: { id, label: labelOf(element) }, + ancestors: found.ancestors.map((link) => ({ + id: link.element, + label: labelOf(requireElement(model, link.element)), + distance: link.distance, + })), + }; +}; diff --git a/packages/targets/graph-cli/src/render.ts b/packages/targets/graph-cli/src/render.ts index a7eab18..602d8e0 100644 --- a/packages/targets/graph-cli/src/render.ts +++ b/packages/targets/graph-cli/src/render.ts @@ -8,7 +8,9 @@ import type { Written, } from "./edit.ts"; import type { + AncestorsResult, BlockedResult, + ChildrenResult, CyclesResult, HandoffsResult, ImpactResult, @@ -41,6 +43,18 @@ export const renderBlocked = (result: BlockedResult): string => export const renderImpact = (result: ImpactResult): string => lines(elementRows(result.impacted)); +/** `children`: each direct sub-task's identifier and label. */ +export const renderChildren = (result: ChildrenResult): string => + lines(elementRows(result.children)); + +/** `ancestors`: each element above the one asked about, nearest first, with its distance. */ +export const renderAncestors = (result: AncestorsResult): string => + lines( + result.ancestors.map( + (link) => `${link.id}\t${link.label}\t${String(link.distance)}`, + ), + ); + /** `path`: the critical path, one element per line in dependency order. */ export const renderPath = (result: PathResult): string => lines(elementRows(result.path)); @@ -167,7 +181,7 @@ export const renderFacts = (result: FactsResult): string => /** `sync`: what was written, then each gap as its reason, target and message. */ export const renderSync = (result: SyncResult): string => lines([ - `synced ${String(result.activities)} activities, ${String(result.transitions)} transitions and ${String(result.participants)} participants into "${result.modelId}" at revision ${String(result.revision)}`, + `synced ${String(result.activities)} activities, ${String(result.transitions)} transitions, ${String(result.relations)} relations and ${String(result.participants)} participants into "${result.modelId}" at revision ${String(result.revision)}`, ...result.gaps.map( (gap) => `gap\t${gap.reason}\t${gap.target}\t${gap.message}`, ), diff --git a/packages/targets/graph-cli/src/sync.ts b/packages/targets/graph-cli/src/sync.ts index 4570539..8218d45 100644 --- a/packages/targets/graph-cli/src/sync.ts +++ b/packages/targets/graph-cli/src/sync.ts @@ -1,16 +1,17 @@ +import type { Gap } from "@exadev/process-graph-source-contract"; import { - assembleSync, - type Gap, - type GraphSourceClient, - type SyncPage, -} from "@exadev/process-graph-source-contract"; -import type { BoundActivity } from "@exadev/process-graph-store-contract"; + InterruptedSyncError, + InvalidScopeError, + scopesBySource, + syncScopes, + UnknownSourceError, + type SourceLookup, + type SourceScopes, + type SyncCheckpoint, +} from "@exadev/process-graph-sync"; import { CliError, failureOf, underlyingCause } from "./errors.ts"; import { exitCodes } from "./exit-codes.ts"; -import { callOptions, type GraphSession } from "./session.ts"; - -/** Finds the source that serves scopes with an identifier prefix, such as `github`. */ -export type SourceLookup = (prefix: string) => GraphSourceClient | undefined; +import type { GraphSession } from "./session.ts"; /** What to sync. */ export interface SyncRequest { @@ -63,66 +64,37 @@ export class SyncInterruptedError extends CliError implements SyncInterruption { /** What a sync wrote. */ export interface SyncResult { readonly modelId: string; - /** The revision the write took. */ + /** The revision the last write took. */ readonly revision: number; - /** The participants, activities and transitions written, each identifier counted once. */ + /** The participants, activities, transitions and relations written, each counted once. */ readonly participants: number; readonly activities: number; readonly transitions: number; + readonly relations: number; /** The parts of the scope the sources could not read, in the order they were met. */ readonly gaps: readonly Gap[]; } -/** The scopes of one source, in the order given. */ -interface SourceScopes { - readonly prefix: string; - readonly scopes: readonly string[]; -} - -const scopeSeparator = ":"; - -/** Groups scopes by their source prefix, keeping the order in which each prefix first appears. */ -const bySource = (scopes: readonly string[]): readonly SourceScopes[] => { - const groups = new Map(); - for (const scope of scopes) { - const separator = scope.indexOf(scopeSeparator); - if (separator <= 0) { - throw new CliError( - exitCodes.usage, - `"${scope}" is not a scope: write :, such as github:owner/repo`, - ); - } - const prefix = scope.slice(0, separator); - groups.set(prefix, [...(groups.get(prefix) ?? []), scope]); +/** A usage error for a scope that names no source or a source nothing reads, or nothing for any other error. */ +const usageError = (error: unknown): CliError | undefined => { + if (error instanceof InvalidScopeError) { + return new CliError(exitCodes.usage, error.message, { cause: error }); + } + if (error instanceof UnknownSourceError) { + return new CliError( + exitCodes.usage, + `no source reads "${error.prefix}" scopes`, + { cause: error }, + ); } - return [...groups].map(([prefix, grouped]) => ({ prefix, scopes: grouped })); -}; - -/** Bound activities are the only elements a sync assembles, since a page's activities are bound by schema. */ -const boundActivities = (pages: readonly SyncPage[]) => { - const { model, overlay, gaps } = assembleSync(pages); - const activities = model.elements.flatMap((element): BoundActivity[] => - element.kind === "activity" && element.source !== undefined - ? [{ ...element, source: element.source }] - : [], - ); - - return { - items: { - participants: model.participants, - activities, - transitions: model.transitions, - overlay, - }, - gaps, - }; + return undefined; }; /** - * Reads every page of the scopes from their sources and merges them into the session's model in one write, creating the model when the store has none. Records already in the model are replaced by identifier and the rest are kept, so syncing again refreshes what changed. + * Reads every page of the scopes from their sources and merges each into the session's model as it is read, creating the model when the store has none: participants, bound activities, transitions, relations (sub-issues, assignees, repository membership) and run state. Records already in the model are replaced by identifier and the rest are kept, so syncing again refreshes what changed. * - * When a source fails part way, the pages read before the failure are still written, and the error carries where to carry on: the request that resumes the failing source after the cursor of its last page read, and the scopes of the sources not reached. + * When a source fails part way, the pages read before the failure are already written, and the error carries where to carry on: the request that resumes the failing source after the cursor of its last page read, and the scopes of the sources not reached. * @param session - The store and model. * @param sources - The source for each identifier prefix. * @param request - The scopes, and a cursor to resume after. @@ -134,84 +106,78 @@ export const sync = async ( sources: SourceLookup, request: SyncRequest, ): Promise => { - const groups = bySource(request.scopes); - if (request.after !== undefined && groups.length > 1) { + const { scopes, after } = request; + let groups: readonly SourceScopes[]; + try { + groups = scopesBySource(scopes); + } catch (error) { + throw usageError(error) ?? error; + } + const [first] = groups; + if (after !== undefined && groups.length > 1) { throw new CliError( exitCodes.usage, "--after resumes one source's sync: give only that source's scopes", ); } - const clients = groups.map((group) => { - const client = sources(group.prefix); - if (client === undefined) { - throw new CliError( - exitCodes.usage, - `no source reads "${group.prefix}" scopes`, - ); - } - - return { ...group, client }; - }); - - const pages: SyncPage[] = []; - const write = async () => { - const { items, gaps } = boundActivities(pages); - const { revision } = await session.store.v1.items.upsert( - { modelId: session.modelId, items }, - callOptions(session.signal), - ); + const resume: SyncCheckpoint | undefined = + after === undefined || first === undefined + ? undefined + : { + modelId: session.modelId, + scopes, + source: first.prefix, + after, + }; + + try { + const result = await syncScopes({ + store: session.store, + sources, + modelId: session.modelId, + scopes, + resume, + signal: session.signal, + }); return { modelId: session.modelId, - revision, - participants: items.participants.length, - activities: items.activities.length, - transitions: items.transitions.length, - gaps, + revision: result.revision, + participants: result.participants, + activities: result.activities, + transitions: result.transitions, + relations: result.relations, + gaps: result.gaps, }; - }; - - /** Reads the sources in order, each only after the one before has finished, since a source's failure must stop the ones after it. */ - const readFrom = async (index: number): Promise => { - const group = clients.at(index); - if (group === undefined) { - return; + } catch (error) { + const usage = usageError(error); + if (usage !== undefined) { + throw usage; } - let cursor = index === 0 ? request.after : undefined; - try { - const stream = await group.client.v1.sync.stream( - cursor === undefined - ? { scope: group.scopes } - : { scope: group.scopes, after: cursor }, - callOptions(session.signal), - ); - for await (const page of stream) { - pages.push(page); - cursor = page.next; - } - } catch (error) { - const failure = failureOf(error); - const saved = pages.length > 0; - if (saved) { - await write(); - } - throw new SyncInterruptedError( - failure.exitCode, - failure.message, - { - resume: - cursor === undefined - ? { scopes: group.scopes } - : { scopes: group.scopes, after: cursor }, - remaining: clients.slice(index + 1).flatMap((next) => next.scopes), - saved, - }, - { cause: underlyingCause(error) }, + if (!(error instanceof InterruptedSyncError)) { + throw error; + } + const failure = failureOf(error.cause); + const { resume: from, remaining, saved } = error; + const failing = groups.find((group) => group.prefix === from.source); + if (failing === undefined) { + throw new Error( + `unreachable: the interrupted source "${from.source}" is one of the sync's own`, + { cause: error }, ); } - await readFrom(index + 1); - }; - await readFrom(0); - - return write(); + throw new SyncInterruptedError( + failure.exitCode, + failure.message, + { + resume: + from.after === undefined + ? { scopes: failing.scopes } + : { scopes: failing.scopes, after: from.after }, + remaining, + saved, + }, + { cause: underlyingCause(error.cause) }, + ); + } }; diff --git a/packages/targets/graph-cli/src/sync.unit.test.ts b/packages/targets/graph-cli/src/sync.unit.test.ts index 4a87f32..9edf235 100644 --- a/packages/targets/graph-cli/src/sync.unit.test.ts +++ b/packages/targets/graph-cli/src/sync.unit.test.ts @@ -16,7 +16,8 @@ import { createRouterClient } from "@orpc/server"; import { afterEach, describe, expect, test } from "vitest"; import { exitCodes } from "./exit-codes.ts"; import type { GraphSession } from "./session.ts"; -import { sync, SyncInterruptedError, type SourceLookup } from "./sync.ts"; +import type { SourceLookup } from "@exadev/process-graph-sync"; +import { sync, SyncInterruptedError } from "./sync.ts"; import { githubLikeSource } from "./test-support/fixtures.ts"; import { createHarness, type Harness } from "./test-support/harness.ts"; diff --git a/packages/targets/graph-cli/tsconfig.build.json b/packages/targets/graph-cli/tsconfig.build.json index 63a34ba..9a67854 100644 --- a/packages/targets/graph-cli/tsconfig.build.json +++ b/packages/targets/graph-cli/tsconfig.build.json @@ -27,6 +27,9 @@ { "path": "../../core/graph/store-contract/tsconfig.build.json" }, + { + "path": "../../core/graph/sync/tsconfig.build.json" + }, { "path": "../../adapters/graph/source-github-issues/tsconfig.build.json" }, diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 455b8b3..89000bc 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -1080,6 +1080,9 @@ importers: '@exadev/process-graph-store-contract': specifier: workspace:* version: link:../../core/graph/store-contract + '@exadev/process-graph-sync': + specifier: workspace:* + version: link:../../core/graph/sync '@orpc/server': specifier: 'catalog:' version: 2.0.0-beta.41 From e639f2b08807a1e57481e5a02e3eb06d725ce505 Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Sun, 11 Oct 2026 13:48:09 +0100 Subject: [PATCH 04/15] feat(web): sync through graph-sync and link an element's parent and sub-tasks The app's own sync loop is replaced by graph-sync, so a sync keeps the sub-issue, assignee and repository relations its pages carry, and the device's sync record stores graph-sync's checkpoint. The loop's tests moved with it. The details panel reads the selected element's parents and direct sub-tasks from the store's hierarchy queries, again on every revision, and lists each as a button that selects it. --- packages/targets/graph-web/README.md | 22 +- packages/targets/graph-web/package.json | 1 + .../targets/graph-web/src/element-details.tsx | 88 +++++- packages/targets/graph-web/src/hierarchy.ts | 88 ++++++ packages/targets/graph-web/src/ports.ts | 2 +- packages/targets/graph-web/src/sync-panel.tsx | 2 +- packages/targets/graph-web/src/sync-record.ts | 19 +- packages/targets/graph-web/src/sync.ts | 222 --------------- .../targets/graph-web/src/sync.unit.test.ts | 268 ------------------ .../graph-web/src/test-support/ports.ts | 11 +- packages/targets/graph-web/src/use-sync.ts | 10 +- packages/targets/graph-web/src/workspace.tsx | 4 + .../graph-web/src/workspace.unit.test.tsx | 35 +++ pnpm-lock.yaml | 3 + 14 files changed, 249 insertions(+), 526 deletions(-) create mode 100644 packages/targets/graph-web/src/hierarchy.ts delete mode 100644 packages/targets/graph-web/src/sync.ts delete mode 100644 packages/targets/graph-web/src/sync.unit.test.ts diff --git a/packages/targets/graph-web/README.md b/packages/targets/graph-web/README.md index e8e18cd..af5f027 100644 --- a/packages/targets/graph-web/README.md +++ b/packages/targets/graph-web/README.md @@ -8,17 +8,17 @@ React 19, Vite, Vanilla Extract over graph-ui's theme contract, vite-plugin-pwa ## Screens -| Screen | What it does | -| ------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| Token | Takes a GitHub token, says what it needs (read-only Issues and Metadata on a fine-grained token, or a classic token with no scopes for public repositories and `repo` for private ones) and where it goes. The field is write-only; a saved token is never shown again. | -| Scope | Any number of scopes: `owner`, `owner/repo`, `owner/projects/number`, a source-qualified scope such as `github:owner/repo`, or a pasted GitHub link to an owner, a repository or a Projects v2 board. With a token, suggestions list the viewer's account and organisations and, on request, an owner's repositories (the source's `scopes.roots` and `scopes.children`), read only when asked for. | -| Sync | Reads the scopes page by page and writes each page as it is read, announcing progress. Cancelling stops it between or during reads; a cancelled, rate-limited or interrupted sync resumes from its checkpoint, which is kept on the device so it survives a reload. | -| View | The model as a flowchart, swimlane, state machine or sequence diagram (`ProcessView` from graph-react-flow), filtered, with the critical path marked and a legend. A state machine follows an entity; a sequence follows the model's own scenario, else the critical path. | -| Lists | Ready, blocked (with what each waits on), the critical path, hand-offs between owners, cycles and structural problems: the command line's `ready`, `blocked`, `path`, `handoffs`, `cycles` and `validate`. | -| Details | The selected element's status and the rule that decided it, its blockers and what it holds up (`why`, `blocked`, `impact`), its issue on GitHub, buttons that set its run state (`set`), buttons that record each artefact it consumes or produces as provided, absent or unknown (`set artefact`) and each named fact its conditions read as true, false or unknown (`set fact`), so the statuses follow at once, and, for an issue in a repository outside the scope, one that adds that repository and syncs it. | -| Filters | Status, actor and source collection, as graph-ui's `FilterPanel`; the view and the lists follow them. | -| Add | Adds a participant, activity, decision, state, event, artefact or transition by hand, as the command line's `add` does; the store refuses a change that would leave the model invalid. | -| Notices | Offline; a new version ready to load; parts of the link that could not be used; a sync that stopped (a rate limit with its reset time, a token GitHub refused, anything else); and the parts of the scope the source could not read. | +| Screen | What it does | +| ------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| Token | Takes a GitHub token, says what it needs (read-only Issues and Metadata on a fine-grained token, or a classic token with no scopes for public repositories and `repo` for private ones) and where it goes. The field is write-only; a saved token is never shown again. | +| Scope | Any number of scopes: `owner`, `owner/repo`, `owner/projects/number`, a source-qualified scope such as `github:owner/repo`, or a pasted GitHub link to an owner, a repository or a Projects v2 board. With a token, suggestions list the viewer's account and organisations and, on request, an owner's repositories (the source's `scopes.roots` and `scopes.children`), read only when asked for. | +| Sync | Reads the scopes page by page through graph-sync, the sync the command line runs, and writes each page as it is read, sub-issue, assignee and repository relations included, announcing progress. Cancelling stops it between or during reads; a cancelled, rate-limited or interrupted sync resumes from its checkpoint, which is kept on the device so it survives a reload. | +| View | The model as a flowchart, swimlane, state machine or sequence diagram (`ProcessView` from graph-react-flow), filtered, with the critical path marked and a legend. A state machine follows an entity; a sequence follows the model's own scenario, else the critical path. | +| Lists | Ready, blocked (with what each waits on), the critical path, hand-offs between owners, cycles and structural problems: the command line's `ready`, `blocked`, `path`, `handoffs`, `cycles` and `validate`. | +| Details | The selected element's status and the rule that decided it, its blockers and what it holds up (`why`, `blocked`, `impact`), what it is a sub-task of and its own sub-tasks (`ancestors`, `children`), each a link that selects it, its issue on GitHub, buttons that set its run state (`set`), buttons that record each artefact it consumes or produces as provided, absent or unknown (`set artefact`) and each named fact its conditions read as true, false or unknown (`set fact`), so the statuses follow at once, and, for an issue in a repository outside the scope, one that adds that repository and syncs it. | +| Filters | Status, actor and source collection, as graph-ui's `FilterPanel`; the view and the lists follow them. | +| Add | Adds a participant, activity, decision, state, event, artefact or transition by hand, as the command line's `add` does; the store refuses a change that would leave the model invalid. | +| Notices | Offline; a new version ready to load; parts of the link that could not be used; a sync that stopped (a rate limit with its reset time, a token GitHub refused, anything else); and the parts of the scope the source could not read. | The theme follows the system's colour scheme until a person picks light or dark. Every list and the canvas move their selection with the arrow keys, Home, End and Escape, the view and list tabs with the arrow keys, Home and End, and a skip link leads past the header. diff --git a/packages/targets/graph-web/package.json b/packages/targets/graph-web/package.json index 4b569bf..afa44d4 100644 --- a/packages/targets/graph-web/package.json +++ b/packages/targets/graph-web/package.json @@ -50,6 +50,7 @@ "@exadev/process-graph-source-github-issues": "workspace:*", "@exadev/process-graph-sqlite-wasm": "workspace:*", "@exadev/process-graph-store-contract": "workspace:*", + "@exadev/process-graph-sync": "workspace:*", "@exadev/process-graph-ui": "workspace:*", "@orpc/client": "catalog:", "@orpc/server": "catalog:", diff --git a/packages/targets/graph-web/src/element-details.tsx b/packages/targets/graph-web/src/element-details.tsx index fbd1e55..f167c9d 100644 --- a/packages/targets/graph-web/src/element-details.tsx +++ b/packages/targets/graph-web/src/element-details.tsx @@ -12,9 +12,17 @@ import type { } from "@exadev/process-graph-schema"; import { DetailPanel } from "@exadev/process-graph-ui"; import { useId, useState, type ReactElement } from "react"; -import { button, muted, row, section, sectionHeading } from "./app.css.ts"; +import { + button, + list, + muted, + row, + section, + sectionHeading, +} from "./app.css.ts"; import { failureOf } from "./failure.ts"; import { githubIssueOf } from "./github-issue.ts"; +import { useHierarchy, type HierarchyClient } from "./hierarchy.ts"; import { RunFacts, type ArtefactState, type FactState } from "./run-facts.tsx"; /** The run states a person can set, and the text of each button. */ @@ -44,10 +52,55 @@ export interface ElementDetailsProps { readonly onSetState: (id: string, state: OverlayStatus) => Promise; /** Adds a scope and syncs it, so work outside the scope can be expanded on demand. */ readonly onAddScope: (scope: string) => void; + /** The store, for the element's place in the hierarchy its source reports. */ + readonly store: HierarchyClient; + readonly modelId: string; + /** The model's revision, so a write that changes the hierarchy is read again. */ + readonly revision: number; + /** Selects another element, as a parent or child link does. */ + readonly onSelect: (id: string) => void; } +interface RelativesProps { + readonly heading: string; + /** The relation each listed element has to the selected one, for `data-hierarchy`. */ + readonly relation: "parent" | "child"; + readonly ids: readonly string[]; + readonly labelOf: (id: string) => string; + readonly onSelect: (id: string) => void; +} + +/** One side of the hierarchy: each element a button that selects it. */ +const Relatives = (props: RelativesProps): ReactElement => { + const headingId = useId(); + + return ( +
+

+ {props.heading} +

+
    + {props.ids.map((id) => ( +
  • + +
    • + ))} +
+
+ ); +}; + /** - * The selected element: why it has its status, what blocks it and what it holds up (the command line's `why`, `blocked` and `impact`), its issue on GitHub, controls to set its run state, controls to record the artefacts it consumes or produces and the facts its conditions read (`set artefact`, `set fact`), and, for an issue in a repository outside the scope, a control that adds that repository to the scope. + * The selected element: why it has its status, what blocks it and what it holds up (the command line's `why`, `blocked` and `impact`), what it is a sub-task of and its own sub-tasks (`ancestors` and `children`), each a link that selects it, its issue on GitHub, controls to set its run state, controls to record the artefacts it consumes or produces and the facts its conditions read (`set artefact`, `set fact`), and, for an issue in a repository outside the scope, a control that adds that repository to the scope. * @param props - The run, the element and the actions. * @returns The details. */ @@ -62,8 +115,13 @@ export const ElementDetails = (props: ElementDetailsProps): ReactElement => { onSetArtefact, onSetFact, onAddScope, + store, + modelId, + revision, + onSelect, } = props; const { id } = element; + const hierarchy = useHierarchy(store, modelId, id, revision); const explanation = useWhyStatus(run, id); const { blockers } = useBlockedChain(run, id); const impact = useImpact(run, id); @@ -92,6 +150,32 @@ export const ElementDetails = (props: ElementDetailsProps): ReactElement => { impact={impact} labelOf={labelOf} /> + {hierarchy.status === "failed" ? ( +

+ The sub-tasks could not be read: {hierarchy.message} +

+ ) : hierarchy.status === "loaded" ? ( + <> + {hierarchy.parents.length === 0 ? null : ( + + )} + {hierarchy.children.length === 0 ? null : ( + + )} + + ) : null}

Run state diff --git a/packages/targets/graph-web/src/hierarchy.ts b/packages/targets/graph-web/src/hierarchy.ts new file mode 100644 index 0000000..fb14d74 --- /dev/null +++ b/packages/targets/graph-web/src/hierarchy.ts @@ -0,0 +1,88 @@ +import type { GraphStoreClient } from "@exadev/process-graph-store-contract"; +import { useEffect, useState } from "react"; +import { failureOf } from "./failure.ts"; + +/** The part of the `GraphStore` client the hierarchy reads: the structural queries over `child-of` relations. */ +export interface HierarchyClient { + readonly v1: { + readonly query: Pick< + GraphStoreClient["v1"]["query"], + "children" | "ancestors" + >; + }; +} + +/** An element's place in the hierarchy a source reports, as the store's relations hold it. */ +export type Hierarchy = + | { readonly status: "loading" } + | { + readonly status: "loaded"; + /** What the element is a sub-task of: the ancestors one relation away, in model order. */ + readonly parents: readonly string[]; + /** Its direct sub-tasks, in model order. */ + readonly children: readonly string[]; + } + | { readonly status: "failed"; readonly message: string }; + +/** A parent is one `child-of` relation above its child. */ +const parentDistance = 1; + +const loading: Hierarchy = { status: "loading" }; + +/** + * Reads an element's parents and children from the store's hierarchy queries, again whenever the model's revision changes, so a sync that brings a sub-issue or its parent shows at once. + * @param store - The store client. Keep it stable between renders. + * @param modelId - The model. + * @param element - The element, which the model declares. + * @param revision - The model's revision, so a change to the model reads the hierarchy again. + * @returns `loading` until both queries answer, then the parents and children, or `failed` with what the store said. + */ +export const useHierarchy = ( + store: HierarchyClient, + modelId: string, + element: string, + revision: number, +): Hierarchy => { + const key = `${modelId}\n${element}\n${String(revision)}`; + const [state, setState] = useState<{ + readonly key: string; + readonly value: Hierarchy; + }>({ key, value: loading }); + + useEffect(() => { + const controller = new AbortController(); + const { signal } = controller; + Promise.all([ + store.v1.query.ancestors({ modelId, element }, { signal }), + store.v1.query.children({ modelId, element }, { signal }), + ]).then( + ([above, below]) => { + setState({ + key, + value: { + status: "loaded", + parents: above.ancestors + .filter((link) => link.distance === parentDistance) + .map((link) => link.element), + children: below.children, + }, + }); + }, + (error: unknown) => { + // An abort is this effect being cleaned up, not a failure to read. + if (!signal.aborted) { + setState({ + key, + value: { status: "failed", message: failureOf(error).message }, + }); + } + }, + ); + + return () => { + controller.abort(); + }; + }, [store, modelId, element, key]); + + return state.key === key ? state.value : loading; +}; diff --git a/packages/targets/graph-web/src/ports.ts b/packages/targets/graph-web/src/ports.ts index 2795c02..fa13b80 100644 --- a/packages/targets/graph-web/src/ports.ts +++ b/packages/targets/graph-web/src/ports.ts @@ -1,7 +1,7 @@ import type { FlowLayoutOptions } from "@exadev/process-graph-react-flow"; import type { GraphStoreClient } from "@exadev/process-graph-store-contract"; import type { SyncRecord } from "./sync-record.ts"; -import type { SourceLookup } from "./sync.ts"; +import type { SourceLookup } from "@exadev/process-graph-sync"; /** A value that changes over time and tells its listeners when it does, in the shape React's `useSyncExternalStore` reads. */ export interface Observable { diff --git a/packages/targets/graph-web/src/sync-panel.tsx b/packages/targets/graph-web/src/sync-panel.tsx index 09d022d..49a52b1 100644 --- a/packages/targets/graph-web/src/sync-panel.tsx +++ b/packages/targets/graph-web/src/sync-panel.tsx @@ -7,7 +7,7 @@ import { section, sectionHeading, } from "./app.css.ts"; -import type { SyncProgress } from "./sync.ts"; +import type { SyncProgress } from "@exadev/process-graph-sync"; import type { SyncHandle } from "./use-sync.ts"; export interface SyncPanelProps { diff --git a/packages/targets/graph-web/src/sync-record.ts b/packages/targets/graph-web/src/sync-record.ts index 9c98c60..5bc2868 100644 --- a/packages/targets/graph-web/src/sync-record.ts +++ b/packages/targets/graph-web/src/sync-record.ts @@ -1,28 +1,15 @@ import { gapSchema } from "@exadev/process-graph-source-contract"; +import { syncCheckpointSchema } from "@exadev/process-graph-sync"; import { z } from "zod"; /** - * Where a sync stopped: the model, every scope it was asked for in order, the prefix of the source to read next, and the cursor after the last page it wrote within that source's scopes (absent to read them from the start). Everything before it is in the store, so a sync started from it reads only what is left. - */ -const checkpointSchema = z - .strictObject({ - modelId: z.string().min(1), - scopes: z.array(z.string().min(1)).readonly(), - source: z.string().min(1), - after: z.string().min(1).optional(), - }) - .readonly(); - -export type SyncCheckpoint = z.infer; - -/** - * What the app remembers about a model's last sync: where it stopped, when it did not finish, and the parts of the scope the sources could not read. Kept on the device, so after a reload, or offline, the app still says the model is incomplete and offers to carry on. + * What the app remembers about a model's last sync: where it stopped, when it did not finish, and the parts of the scope the sources could not read. Kept on the device, so after a reload, or offline, the app still says the model is incomplete and offers to carry on. The checkpoint is graph-sync's, which records the scopes and filter its cursor was issued under. */ export const syncRecordSchema = z .strictObject({ modelId: z.string().min(1), /** Present while the sync has more to read. */ - checkpoint: checkpointSchema.optional(), + checkpoint: syncCheckpointSchema.optional(), gaps: z.array(gapSchema).readonly(), }) .readonly(); diff --git a/packages/targets/graph-web/src/sync.ts b/packages/targets/graph-web/src/sync.ts deleted file mode 100644 index 760c216..0000000 --- a/packages/targets/graph-web/src/sync.ts +++ /dev/null @@ -1,222 +0,0 @@ -import type { - Gap, - GraphSourceClient, - SourceTransition, - SyncPage, -} from "@exadev/process-graph-source-contract"; -import type { GraphStoreClient } from "@exadev/process-graph-store-contract"; -import { ORPCError } from "@orpc/client"; -import type { SyncCheckpoint } from "./sync-record.ts"; - -/** Finds the source that reads scopes with an identifier prefix, such as `github`. */ -export type SourceLookup = (prefix: string) => GraphSourceClient | undefined; - -/** How far a sync has got. */ -export interface SyncProgress { - /** Pages read and written. */ - readonly pages: number; - /** Distinct bound activities written. */ - readonly activities: number; - /** Distinct transitions written. */ - readonly transitions: number; - /** The parts of the scope the sources could not read, in the order met. */ - readonly gaps: readonly Gap[]; - /** Where to resume; absent once every scope has been read. */ - readonly checkpoint: SyncCheckpoint | undefined; -} - -/** What to sync and where to. */ -export interface SyncRequest { - readonly store: GraphStoreClient; - readonly sources: SourceLookup; - readonly modelId: string; - readonly scopes: readonly string[]; - /** A checkpoint from an interrupted sync of the same model and scopes: the sync carries on from it. */ - readonly resume?: SyncCheckpoint | undefined; - /** Ends the sync between or during reads; what was written stays written. */ - readonly signal: AbortSignal; - /** Called after every page is written, with the progress so far. */ - readonly onProgress: (progress: SyncProgress) => void; -} - -/** Thrown when a scope names a source the app has no reader for. */ -export class UnknownSourceError extends Error { - /** - * @param prefix - The source prefix no reader serves. - */ - constructor(prefix: string) { - super(`No source reads "${prefix}" scopes.`); - this.name = "UnknownSourceError"; - } -} - -const sourceSeparator = ":"; - -/** The scopes of each source, in the order each source first appears. */ -const bySource = ( - scopes: readonly string[], -): readonly { - readonly prefix: string; - readonly scopes: readonly string[]; -}[] => { - const groups = new Map(); - for (const scope of scopes) { - const prefix = scope.slice(0, scope.indexOf(sourceSeparator)); - groups.set(prefix, [...(groups.get(prefix) ?? []), scope]); - } - - return [...groups].map(([prefix, grouped]) => ({ prefix, scopes: grouped })); -}; - -/** The identifiers of the elements a model in the store already has, or none when the store has no such model. */ -const storedElements = async ( - store: GraphStoreClient, - modelId: string, - signal: AbortSignal, -): Promise> => { - try { - const stored = await store.v1.models.load({ modelId }, { signal }); - - return new Set(stored.model.elements.map((element) => element.id)); - } catch (error) { - if (error instanceof ORPCError && error.code === "NOT_FOUND") { - return new Set(); - } - throw error; - } -}; - -/** - * Reads the scopes from their sources page by page and writes each page into the model as soon as it is read, so progress is kept as it is made: a sync that is cancelled, rate limited or closed with the tab leaves everything it read in the store, and a checkpoint to carry on from. - * - * A store rejects a write that leaves a transition pointing at an element the model lacks, and a source may report a dependency on work a later page brings. A transition is therefore held back until the model has both of its ends, and written with the page that completes it. One whose end never arrives is written with the last page, where the store's rejection says which reference is missing. - * - * Relations a source reports (sub-issues, assignees, repository membership) have no place in the store and are not kept, as the command line's sync does. - * @param request - The store, sources, model, scopes, an optional checkpoint, the signal and a progress listener. - * @returns The final progress, with no checkpoint. - * @throws UnknownSourceError for a scope whose source the app cannot read; an Error for a checkpoint naming a source none of the scopes use; and whatever a source or the store throws, after the pages before it are written. An abort rejects with the signal's reason. - */ -export const runSync = async (request: SyncRequest): Promise => { - const { store, sources, modelId, scopes, resume, signal, onProgress } = - request; - const groups = bySource(scopes).map((group) => { - const client = sources(group.prefix); - if (client === undefined) { - throw new UnknownSourceError(group.prefix); - } - - return { ...group, client }; - }); - const startGroup = - resume === undefined - ? 0 - : groups.findIndex((group) => group.prefix === resume.source); - if (startGroup === -1) { - throw new Error( - "The checkpoint names a source that none of the scopes use; start the sync again.", - ); - } - - const known = new Set(await storedElements(store, modelId, signal)); - const activities = new Set(); - const transitions = new Set(); - const gaps: Gap[] = []; - let held: readonly SourceTransition[] = []; - let pages = 0; - - const write = async ( - page: SyncPage, - checkpoint: SyncCheckpoint | undefined, - ): Promise => { - for (const activity of page.activities) { - known.add(activity.id); - } - const last = checkpoint === undefined; - const candidates = [...held, ...page.transitions]; - const ready = candidates.filter( - (transition) => - last || (known.has(transition.source) && known.has(transition.target)), - ); - held = candidates.filter((transition) => !ready.includes(transition)); - await store.v1.items.upsert( - { - modelId, - items: { - participants: page.participants, - activities: page.activities, - transitions: ready, - overlay: page.overlay, - }, - }, - { signal }, - ); - pages += 1; - for (const activity of page.activities) { - activities.add(activity.id); - } - for (const transition of ready) { - transitions.add(transition.id); - } - gaps.push(...page.gaps); - onProgress({ - pages, - activities: activities.size, - transitions: transitions.size, - gaps: [...gaps], - checkpoint, - }); - }; - - /** Writes each page an iterator yields, in order, each only after the one before is written. */ - const drain = async ( - pagesOf: Readonly>, - checkpointAfter: (page: SyncPage) => SyncCheckpoint | undefined, - ): Promise => { - const next = await pagesOf.next(); - if (next.done === true) { - return; - } - await write(next.value, checkpointAfter(next.value)); - await drain(pagesOf, checkpointAfter); - }; - - /** Reads the sources in order, each only after the one before has finished, since a source's failure must stop the ones after it. */ - const readFrom = async (index: number): Promise => { - const group = groups[index]; - if (group === undefined) { - return; - } - const after = index === startGroup ? resume?.after : undefined; - const following = groups[index + 1]; - const stream = await group.client.v1.sync.stream( - after === undefined - ? { scope: group.scopes } - : { scope: group.scopes, after }, - { signal }, - ); - const iterator = stream[Symbol.asyncIterator](); - try { - await drain(iterator, (page) => - page.next !== undefined - ? { modelId, scopes, source: group.prefix, after: page.next } - : following === undefined - ? undefined - : { modelId, scopes, source: following.prefix }, - ); - } catch (error) { - // Closing the stream stops the source reading pages nobody will write. - await iterator.return(); - throw error; - } - await readFrom(index + 1); - }; - await readFrom(startGroup); - - return { - pages, - activities: activities.size, - transitions: transitions.size, - gaps, - checkpoint: undefined, - }; -}; diff --git a/packages/targets/graph-web/src/sync.unit.test.ts b/packages/targets/graph-web/src/sync.unit.test.ts deleted file mode 100644 index cb6828f..0000000 --- a/packages/targets/graph-web/src/sync.unit.test.ts +++ /dev/null @@ -1,268 +0,0 @@ -import { - createFakeGraphSource, - createGraphSourceRouter, - GraphSourceError, - type GraphSourcePager, -} from "@exadev/process-graph-source-contract"; -import { - createFakeGraphStore, - createGraphStoreRouter, -} from "@exadev/process-graph-store-contract"; -import { createRouterClient } from "@orpc/server"; -import { describe, expect, test } from "vitest"; -import { failureOf } from "./failure.ts"; -import { runSync, UnknownSourceError, type SyncProgress } from "./sync.ts"; -import { apiPage, fakeScopes, pageOf } from "./test-support/ports.ts"; - -const newStore = () => - createRouterClient(createGraphStoreRouter(createFakeGraphStore())); - -const fakeGitHub = createFakeGraphSource({ - source: "github", - scopes: fakeScopes, -}); - -const lookup = (pager: GraphSourcePager) => (prefix: string) => - prefix === "github" - ? createRouterClient(createGraphSourceRouter(pager)) - : undefined; - -const modelId = "github:acme"; - -describe("runSync", () => { - test("writes every page as it is read, reports progress after each, and ends with no checkpoint", async () => { - const store = newStore(); - const progress: SyncProgress[] = []; - - const result = await runSync({ - store, - sources: lookup(fakeGitHub), - modelId, - scopes: ["github:acme"], - signal: new AbortController().signal, - onProgress: (step) => { - progress.push(step); - }, - }); - - expect( - progress.map((step) => [step.pages, step.checkpoint?.after]), - ).toEqual([ - [1, "github%3Aacme@0"], - [2, undefined], - ]); - expect(result).toEqual({ - pages: 2, - activities: 5, - transitions: 4, - gaps: [], - checkpoint: undefined, - }); - const stored = await store.v1.models.load({ modelId }); - expect( - stored.model.elements.map((element) => element.id).toSorted(), - ).toEqual([ - "github:acme/api#1", - "github:acme/api#2", - "github:acme/web#1", - "github:acme/web#2", - "github:partner/sdk#1", - ]); - expect(stored.overlay).toMatchObject({ - "github:acme/api#1": "done", - "github:acme/web#2": "open", - }); - }); - - test("resumes from a checkpoint with only the pages after it", async () => { - const store = newStore(); - const read: string[] = []; - const counting: GraphSourcePager = { - source: "github", - page: async (request) => { - read.push(request.cursor ?? "start"); - - return fakeGitHub.page(request); - }, - }; - await store.v1.items.upsert({ - modelId, - items: { - participants: apiPage.participants, - activities: apiPage.activities, - transitions: apiPage.transitions, - overlay: apiPage.overlay, - }, - }); - - const result = await runSync({ - store, - sources: lookup(counting), - modelId, - scopes: ["github:acme"], - resume: { - modelId, - scopes: ["github:acme"], - source: "github", - after: "github%3Aacme@0", - }, - signal: new AbortController().signal, - onProgress: () => undefined, - }); - - expect(read).toEqual(["github%3Aacme@0"]); - expect(result.pages).toBe(1); - const stored = await store.v1.models.load({ modelId }); - expect( - stored.model.transitions.map((transition) => transition.id), - ).toContain("github:acme/api#1->github:acme/api#2"); - }); - - test("holds a transition back until the page that brings its other end", async () => { - const store = newStore(); - const early = pageOf([ - { - id: "github:acme/web#1", - label: "Build the page", - repository: "github:acme/web", - state: "open", - blockedBy: ["github:acme/api#2"], - }, - ]); - const late = pageOf([ - { - id: "github:acme/api#2", - label: "Build the endpoint", - repository: "github:acme/api", - state: "open", - blockedBy: [], - }, - ]); - const progress: SyncProgress[] = []; - - await runSync({ - store, - sources: lookup( - createFakeGraphSource({ - source: "github", - scopes: { "github:acme": [early, late] }, - }), - ), - modelId, - scopes: ["github:acme"], - signal: new AbortController().signal, - onProgress: (step) => { - progress.push(step); - }, - }); - - expect(progress.map((step) => step.transitions)).toEqual([0, 1]); - const stored = await store.v1.models.load({ modelId }); - expect(stored.model.transitions).toEqual([ - { - id: "github:acme/api#2->github:acme/web#1", - source: "github:acme/api#2", - target: "github:acme/web#1", - }, - ]); - }); - - test("keeps the pages before a failure, reports where to resume, and passes the failure on", async () => { - const store = newStore(); - const progress: SyncProgress[] = []; - const limited: GraphSourcePager = { - source: "github", - page: async (request) => { - if (request.cursor !== undefined) { - throw new GraphSourceError({ - kind: "rate-limited", - message: "rate limited", - resetAt: "2100-01-01T00:00:00.000Z", - }); - } - - return fakeGitHub.page(request); - }, - }; - - const failure = await runSync({ - store, - sources: lookup(limited), - modelId, - scopes: ["github:acme"], - signal: new AbortController().signal, - onProgress: (step) => { - progress.push(step); - }, - }).catch((error: unknown) => failureOf(error)); - - expect(failure).toEqual({ - kind: "rate-limited", - message: "rate limited", - resetAt: "2100-01-01T00:00:00.000Z", - }); - expect(progress.at(-1)?.checkpoint).toEqual({ - modelId, - scopes: ["github:acme"], - source: "github", - after: "github%3Aacme@0", - }); - const stored = await store.v1.models.load({ modelId }); - expect(stored.model.elements.map((element) => element.id)).toEqual([ - "github:acme/api#1", - "github:acme/api#2", - ]); - }); - - test("reports the parts of the scope the source could not read", async () => { - const result = await runSync({ - store: newStore(), - sources: lookup(fakeGitHub), - modelId: "github:acme github:ghost", - scopes: ["github:acme", "github:ghost"], - signal: new AbortController().signal, - onProgress: () => undefined, - }); - - expect(result.gaps).toEqual([ - { - target: "github:ghost", - reason: "not-found", - message: "github:ghost does not exist", - }, - ]); - }); - - test("stops when its signal is aborted, keeping what it wrote", async () => { - const store = newStore(); - const controller = new AbortController(); - - const outcome = runSync({ - store, - sources: lookup(fakeGitHub), - modelId, - scopes: ["github:acme"], - signal: controller.signal, - onProgress: () => { - controller.abort(); - }, - }); - - await expect(outcome).rejects.toThrow(); - const stored = await store.v1.models.load({ modelId }); - expect(stored.model.elements).toHaveLength(2); - }); - - test("refuses a scope whose source the app cannot read", async () => { - await expect( - runSync({ - store: newStore(), - sources: lookup(fakeGitHub), - modelId: "linear:team", - scopes: ["linear:team"], - signal: new AbortController().signal, - onProgress: () => undefined, - }), - ).rejects.toThrow(UnknownSourceError); - }); -}); diff --git a/packages/targets/graph-web/src/test-support/ports.ts b/packages/targets/graph-web/src/test-support/ports.ts index e0ca604..9d56522 100644 --- a/packages/targets/graph-web/src/test-support/ports.ts +++ b/packages/targets/graph-web/src/test-support/ports.ts @@ -103,6 +103,8 @@ interface Issue { readonly repository: string; readonly state: "done" | "open"; readonly blockedBy: readonly string[]; + /** The issue this one is a sub-issue of. */ + readonly parent?: string; } const repositoryParticipant = (repository: string) => ({ @@ -112,7 +114,7 @@ const repositoryParticipant = (repository: string) => ({ }); /** - * A page of the fake GitHub source, built as the real adapter builds one: each issue a bound activity whose actor is its repository, each blocker a transition from it to the issue it blocks, and the state in the overlay. + * A page of the fake GitHub source, built as the real adapter builds one: each issue a bound activity whose actor is its repository, each blocker a transition from it to the issue it blocks, each sub-issue `child-of` its parent, and the state in the overlay. * @param issues - The page's issues, with every blocker among them. */ export const pageOf = (issues: readonly Issue[]): FakePage => ({ @@ -133,7 +135,11 @@ export const pageOf = (issues: readonly Issue[]): FakePage => ({ target: issue.id, })), ), - relations: [], + relations: issues.flatMap((issue) => + issue.parent === undefined + ? [] + : [{ kind: "child-of" as const, from: issue.id, to: issue.parent }], + ), overlay: Object.fromEntries(issues.map((issue) => [issue.id, issue.state])), gaps: [], }); @@ -180,6 +186,7 @@ const webPage: FakePage = pageOf([ repository: "github:acme/web", state: "open", blockedBy: ["github:acme/api#2", "github:partner/sdk#1"], + parent: "github:acme/web#2", }, { id: "github:acme/web#2", diff --git a/packages/targets/graph-web/src/use-sync.ts b/packages/targets/graph-web/src/use-sync.ts index ff9b88d..d5059a0 100644 --- a/packages/targets/graph-web/src/use-sync.ts +++ b/packages/targets/graph-web/src/use-sync.ts @@ -2,8 +2,12 @@ import type { Gap } from "@exadev/process-graph-source-contract"; import { useCallback, useEffect, useRef, useState } from "react"; import { failureOf, type Failure } from "./failure.ts"; import type { AppPorts } from "./ports.ts"; -import type { SyncCheckpoint, SyncRecord } from "./sync-record.ts"; -import { runSync, type SyncProgress } from "./sync.ts"; +import { + syncScopes, + type SyncCheckpoint, + type SyncProgress, +} from "@exadev/process-graph-sync"; +import type { SyncRecord } from "./sync-record.ts"; /** Where the sync of the shown model stands in this page. */ type SyncState = @@ -92,7 +96,7 @@ export const useSync = ( gaps: [...earlierGaps, ...progress.gaps], }); settle({ phase: "running", progress: undefined }); - runSync({ + syncScopes({ store: ports.store, sources: ports.sources(token), modelId, diff --git a/packages/targets/graph-web/src/workspace.tsx b/packages/targets/graph-web/src/workspace.tsx index 31295f0..133f5f3 100644 --- a/packages/targets/graph-web/src/workspace.tsx +++ b/packages/targets/graph-web/src/workspace.tsx @@ -246,6 +246,10 @@ const Workspace = (props: WorkspaceProps): ReactElement => { onSetArtefact={setArtefact} onSetFact={setFact} onAddScope={onAddScope} + store={ports.store} + modelId={modelId} + revision={props.stored.revision} + onSelect={select} /> )} { }); }); + test("link an element's parent and sub-tasks, each selecting it", async () => { + const ports = await synced(`${acme}&select=github%3Aacme%2Fweb%231`); + + const parents = await waitFor(() => { + const found = document.querySelector('[data-hierarchy="parent"]'); + if (!(found instanceof HTMLElement)) { + throw new Error("no parent is listed yet"); + } + + return found; + }); + expect(within(parents).getByRole("heading").textContent).toBe("Part of"); + expect(document.querySelector('[data-hierarchy="child"]')).toBeNull(); + fireEvent.click( + within(parents).getByRole("button", { name: "Ship the release" }), + ); + + expect(addressOf(ports).selected).toBe("github:acme/web#2"); + const children = await waitFor(() => { + const found = document.querySelector('[data-hierarchy="child"]'); + if (!(found instanceof HTMLElement)) { + throw new Error("no sub-task is listed yet"); + } + + return found; + }); + expect(within(children).getByRole("heading").textContent).toBe("Sub-tasks"); + expect( + within(children) + .getAllByRole("button") + .map((button) => button.textContent), + ).toEqual(["Build the page"]); + expect(document.querySelector('[data-hierarchy="parent"]')).toBeNull(); + }); + test("say when a link selects an element the model does not have", async () => { await synced(`${acme}&select=github%3Aacme%2Fapi%2399`); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 89000bc..b02ef21 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -1153,6 +1153,9 @@ importers: '@exadev/process-graph-store-contract': specifier: workspace:* version: link:../../core/graph/store-contract + '@exadev/process-graph-sync': + specifier: workspace:* + version: link:../../core/graph/sync '@exadev/process-graph-ui': specifier: workspace:* version: link:../../ui/graph/ui From 32eaabced0e4a86dc105f3a5d0819738abd63ea9 Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Sun, 11 Oct 2026 13:53:52 +0100 Subject: [PATCH 05/15] feat(cli): filter a sync by state, label and order sync takes --state (open or done) and --label, each repeatable, and --order (created, updated or comments) with --direction (asc or desc), and passes them to the source as the contract's typed filter. The source's cursor carries the filter it was issued under, so the resume command printed after a failure repeats the filter's options. A source that cannot filter a scope refuses the sync; GitHub's refusal for a Projects v2 board is reported with its own message and the source failure exit status. --- packages/targets/graph-cli/README.md | 52 +++--- .../targets/graph-cli/src/cli.unit.test.ts | 154 ++++++++++++++++++ packages/targets/graph-cli/src/program.ts | 112 ++++++++++++- packages/targets/graph-cli/src/render.ts | 29 +++- packages/targets/graph-cli/src/sync.ts | 35 ++-- 5 files changed, 342 insertions(+), 40 deletions(-) diff --git a/packages/targets/graph-cli/README.md b/packages/targets/graph-cli/README.md index 893d963..4ae7ab4 100644 --- a/packages/targets/graph-cli/README.md +++ b/packages/targets/graph-cli/README.md @@ -17,31 +17,31 @@ process-graph show --view swimlane --format mermaid ## Commands -| Command | What it does | -| --------------------------------------------------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `sync [--after ]` | Reads the scopes from their sources and merges each page into the model as it is read, relations included; prints what it wrote and any gaps. | -| `ready` | The work that can start now. | -| `blocked ` | The open work an element is waiting on: the ends of its blocked-by chain. | -| `impact ` | The live work that depends on an element. | -| `children ` | The element's direct sub-tasks: the work a source reports as its children, such as GitHub sub-issues. | -| `ancestors ` | What the element is a sub-task of, its parent first and then each level up, with how many levels up each is. | -| `path` | The critical path: the longest chain of work still ahead, each element counting one step. | -| `cycles` | The cycles no decision closes. | -| `handoffs` | The transitions from one actor's work to another's, such as a dependency across repositories or organisations. | -| `why ` | The rule that decided an element's status, its needs and artefacts, and the blocked-by chain behind it. | -| `show [--view ] [--format ]` | Draws the model. Views: `flowchart` (default), `swimlane`, `state`, `sequence`. Formats: `mermaid` (default), `dot`, `json` (the projection itself). | -| `add participant