Temporal GraphRAG for TypeScript

const schema = defineGraphSchema({
	nodes: {
		site: z.object({ name: z.string(), region: z.enum(['us', 'eu', 'apac']) }),
		gateway: z.object({ name: z.string(), firmware: z.string() }),
		alert: z.object({ severity: z.enum(['low', 'high']) }),
	},
	edges: {
		// `from`/`to` constrain the endpoints; `single` makes the rel single-valued per source,
		// so each addEdge closes the previous live one. `data` validates edge payloads.
		deployedAt: { from: 'gateway', to: 'site', single: true },
		raised: { from: 'gateway', to: 'alert', data: z.object({ at: z.number() }) },
	},
	// Optional per-type embedding policy: what text a node is embedded from (default: its `body`),
	// and whether long inputs are split into chunks. A data-only type stays searchable this way.
	embedding: {
		site: { text: (d) => `${d.name} (${d.region})` },
		alert: { chunk: { size: 1200, overlap: 120 } },
	},
});

type Schema = typeof schema;

const gw = await g.addNode({
	type: 'gateway',
	data: { name: 'gw-1', firmware: '2.1.0' },
	body: '…',
});

await g.updateNode(gw.id, { data: { firmware: '2.2.0' } }); // shallow merge, opens a new version
await g.updateNode(gw.id, { body: 'edited' }); // the embedding input changed ⇒ re-embedded
await g.addEdge({ rel: 'raised', src: gw.id, dst: alert.id, data: { at: Date.now() } });
await g.deleteEdge(edgeId);
await g.deleteNode(gw.id); // closes the version and drops its vectors; the history is kept

import { bulkLoad, bulkEdges } from 'graphx';

await bulkLoad(db, schema, rows, { embedder, chunkSize: 500 });
await bulkEdges(db, schema, edgeRows);

await g.write(async (graph) => {
	await graph.addNode({ type: 'site', data: { name: 'eu-west-1', region: 'eu' } });
	await graph.addNode({ type: 'gateway', data: { name: 'gw-2', firmware: '2.2.0' } });
});

const note = await g.atomic((scope) => scope.addNode({ type: 'note', data: { path: 'A.md' } }));

import { history, diff, changeFeed, timeline } from 'graphx';

await history(db, id); // every version of a node, oldest first
await diff(db, t1, t2); // nodes and edges added, changed and removed between two instants
await changeFeed(db, cursor, { limit: 500 }); // CDC: keyset stream of node + edge versions
await timeline(db, { buckets: 120 }); // change-point extent + density histogram + snap ticks

// Vector seeds, then a time-respecting walk out from them
await g.retrieve({ query: 'overheating sensor', k: 10, maxDepth: 2, rels: ['raised'] });

// Vector + full-text fused with reciprocal rank fusion, then the same walk
await g.hybridRetrieve({
	query: 'overheating sensor',
	k: 10,
	rrfK: 60,
	rerank: async (query, candidates) => …, // optional — jevRerank() below, or your own
	mmr: { k: 10, lambda: 0.7 }, // optional diversification
});

await g.getNode(id); // AnyNode<S> | null
await g.getNodeVersion(id); // the version row: data, body, uri, revision, valid_from/valid_to
await g.getNodeContent(id); // body + content_type + hash
await g.neighbors(id, { rels: ['deployedAt'], direction: 'forward' }); // AnyNode[]
await g.neighborsPage(id, { rels: ['raised'], limit: 100 }); // keyset-paginated
await g.listNodes({ type: 'alert', q: 'overheating', limit: 50 }); // { nodes, nextCursor }
await g.listNodeVersions({ type: 'alert', limit: 50 }); // the same page, each row with body + revision
await g.listEdges({ rel: 'raised', source: 'jev' }); // { edges: [{ id, src, dst, weight, data, … }] }
await g.graphSlice({ type: 'gateway' }); // canvas projection: ids, labels, links

import { match } from 'graphx';

const q = await match(schema, db)
	.node('d', 'device')
	.in('raised') // .out(), .in(), .both(); rel options take weight, data and direction filters
	.node('a', 'alert')
	.select('d', 'a');

const rows = await q.run(); // rows[0].d.data and rows[0].a.data are typed per alias
const page = await q.page({ limit: 100 }); // keyset pagination over the same pattern

import { journey, shortestPath, pagerank, community, centrality, topNodes, buildCSR } from 'graphx';

// A time-respecting walk: only edges valid at each step are followed. `from` (epoch ms) is required.
await journey(db, { start: id, from: 0, maxDepth: 6, direction: 'forward' });

await shortestPath(db, srcId, dstId, { weighted: true, rels: ['deployedAt'] });
await pagerank(db, { damping: 0.85 }); // Map<id, score>
await community(db); // label propagation → Map<id, community>
await centrality(db, 'degree'); // 'degree' | 'in' | 'out'
await topNodes(db, { by: 'pagerank', type: 'gateway', limit: 10 }); // reads persisted analytics
await topNodes(db, { by: 'score:risk', type: 'alert' }); // or a persisted score — see scoreNodes
await buildCSR(db); // the compressed mirror the analytics run over, if you want it directly

import { openai, voyage, ollama } from 'graphx/embedders';
import { hashEmbed, fixtureEmbed, defineEmbedder } from 'graphx';

openai('text-embedding-3-small'); // OPENAI_API_KEY, optional `dim`, `baseUrl`, `batchSize`
voyage('voyage-3'); // VOYAGE_API_KEY
ollama('nomic-embed-text'); // local, no key
hashEmbed(); // deterministic and model-free — tests, demos, offline
fixtureEmbed({ path: './fixtures/emb.json' }); // record real vectors once, replay them offline
defineEmbedder({ id: 'acme:v1', embed: async (texts) => … }); // anything else

import { TriggerRunner, embedTrigger, webhookAction, deadLetters } from 'graphx';

const runner = new TriggerRunner(g, {
	name: 'alerts',
	triggers: [
		{
			name: 'notify',
			match: { op: 'node.create', label: 'alert' }, // an absent field matches anything
			action: webhookAction({ url: 'https://example.com/hook', secret: process.env.HOOK }),
			retries: 5,
		},
		embedTrigger(), // the other half of `embedding: 'lazy'`
	],
});
await runner.runOnce(); // or .start() to poll
await deadLetters(db, { subscription: 'alerts' }); // what exhausted its retries, and why

import { createApp, hashEmbed } from 'graphx';
import { schema } from './schema.ts';

const { app } = await createApp({
	schema,
	embedder: hashEmbed(), // every write embeds through it; no dimension to configure
	db: 'iot_demo',
	openapi: { title: 'iot-fleet' },
});

// Typed routes + GET /openapi.json + an interactive reference at /docs
Bun.serve({ port: 8899, fetch: app.fetch });

import { GraphProvider, createGraphHooks } from 'graphx/react';
import type { Schema } from './schema.ts';

const g = createGraphHooks<Schema>();

// <GraphProvider bootstrap="/demo"> fetches the tenant/project/user ids itself
g.useNode(id, 'gateway'); // NodeOf<Schema,'gateway'> | null
g.useNeighbors(id, { rel: 'deployedAt' }); // site[]
g.useListNodes({ type: 'alert' }); // alert[]
g.useMatch((q) => q.node('d', 'device').in('raised').node('a', 'alert').select('d', 'a'));
g.useRetrieve({ query: 'overheating sensor', k: 10 });
g.useAddNode(); // mutations: add/update/delete node, add/delete edge, bulk load
g.useChangeFeedSync(); // tails /changes and invalidates exact keys
g.useGraphEvents(); // subscribes to the SSE stream

{
	"mcpServers": {
		"graphx": { "command": "bunx", "args": ["graphx", "mcp", "-c", "./graphx.config.ts"] }
	}
}

import { Auth, defineAuthModel, rel, tupleToUserset } from 'graphx/auth';

const model = defineAuthModel({
	user: {},
	folder: { parent: rel(), editor: rel(), viewer: rel().or('editor') },
	doc: {
		parent: rel(),
		editor: rel(),
		banned: rel(),
		// (direct ∪ editor ∪ viewer-of-the-parent-folder) − banned
		viewer: rel().self().or('editor').or(tupleToUserset('parent', 'viewer')).minus('banned'),
	},
});

const auth = new Auth(g, model);
await auth.write([{ object: 'doc:readme', relation: 'editor', subject: 'user:tim' }]);
await auth.check('doc:readme', 'viewer', 'user:tim'); // true
await auth.check('doc:readme', 'viewer', 'user:tim', { asOf: lastWeek }); // as it stood then
await auth.expand('doc:readme', 'viewer'); // the userset tree
await auth.listObjects('user:tim', 'viewer', 'doc', { limit: 100 }); // keyset-paginated

import { ingestDir, watchDir } from 'graphx/ingest';

await ingestDir({
	dir: './vault',
	graph: g,
	source: 'notes', // namespaces the node `uri`, so two vaults never reconcile each other
	idField: 'id', // frontmatter key giving stable identity across renames
	edgeFields: { author: 'written_by' }, // frontmatter field → typed edge
	assets: { type: 'asset' }, // ![[embeds]] become nodes
	dangling: { type: 'stub' }, // links to unwritten notes become stubs
	tags: { type: 'tag' }, // #tags become shared nodes (off by default — they make hubs)
	prune: true, // retract nodes whose files vanished, scoped to this source
});

import { createBlobStore } from 'graphx/blob';

const blobs = createBlobStore({ client: s3, bucket: 'graphx', inlineLimit: 32_768 });
const ref = await blobs.put(bytes, 'application/pdf');
await g.addNode({ type: 'doc', data: { title }, uri: ref.uri, content_hash: ref.hash });
await blobs.presign(ref.uri, 900); // a short-lived download URL
await blobs.gc(liveHashes); // drop what no live node references

import { choice, createJev, noul, score } from 'graphx/jev';

const jev = createJev(); // model 'jev-latest'; retries 429, 529 and 5xx with backoff
const { answers } = await jev.ask(
	{ alert: 'gw-7 probe read 96C, fan failed' },
	{
		team: choice('Which team owns this?', { hardware: null, firmware: null, network: null }),
		urgent: noul('Does this need action today?'),
		severity: score('How severe is it?', ['cosmetic', 'degraded', 'down']),
	},
);
answers.team.choice; // 'hardware' | 'firmware' | 'network', plus probabilities and confidence
answers.urgent.noul; // probability of yes
answers.severity.score; // 0–2, landing between levels

import { jevRerank } from 'graphx/jev';

// Reads TYPESAFE_API_KEY
await g.hybridRetrieve({ query: 'overheating sensor', k: 10, rerank: jevRerank() });

jevRerank({
	minScore: 0.2, // drop candidates Jev judges unlikely to be relevant
	concurrency: 8, // requests in flight
	onError: 'keep', // an outage returns the fused order instead of failing the search
	guard: { onFlagged: (c, p) => console.warn('injection', c.id, p) }, // see below
});

import { judgePairs, resolveEntities } from 'graphx/jev';

const report = await resolveEntities(g, {
	type: 'deity',
	fields: ['name', 'pantheon'], // what Jev sees and compares — leave `source` out
	candidates: 5, // nearest same-type neighbours judged per node
	rels: { same: 'sameAs', review: 'maybeSameAs' }, // omit to report without writing
});
report.pairs.filter((p) => p.outcome === 'review'); // the curator's queue, with per-field agreement

// Or judge pairs you already have — an audit of an earlier matcher's links, say.
await judgePairs(g, [[srcId, dstId]], { fields: ['name', 'pantheon'] });

graphx new      <dir>                     Scaffold a starter project
graphx serve    [-c config] [-p 8899]     Typed HTTP routes + /openapi.json + /docs
graphx ingest   <dir> [options]           Ingest a vault into the graph
graphx triggers [-c config]               Run declarative triggers over the event outbox
graphx mcp      [-c config] [--read-only] Serve the graph to an MCP client over stdio
graphx reembed  [-c config] [--dry-run]   Re-embed every live node (also switches models)
graphx doctor   [-c config]               Embedding model, width and health of the namespace
graphx fork     <namespace> [-c config]   Branch the namespace into an empty one (--as-of <ms|ISO>)
graphx dedupe   <type> [-c config] [...]  Find duplicate nodes of a type and judge them with Jev
graphx ask      "<question>" [-c config]  Plan a plain-language question as a graph call, and run it