nandi.uk
This repository has no description
12 kB
369 lines
1/**
2 * rdf:ingest
3 *
4 * Ingest the SciOS Resilient Data Futures discourse graph from
5 * https://github.com/jring-o/rdf into network.cosmik.connection records
6 * on the PDS.
7 *
8 * Each node becomes a vertex at https://rdf.scios.tech/node/{ID}
9 * Each edge from frontmatter becomes a network.cosmik.connection record.
10 *
11 * Edge type mapping:
12 * addresses → ADDRESSES (Claim → Question)
13 * supports → SUPPORTS (Evidence→Claim, Claim→Claim)
14 * opposes → OPPOSES (Claim→Claim, Evidence→Claim)
15 * derivedFrom → DERIVED_FROM (Evidence → Source)
16 * informs → INFORMS (Method → Claim)
17 * usesMethod → USES_METHOD (Claim → Method)
18 *
19 * Usage:
20 * bun rdf:ingest # dry run (prints plan)
21 * bun rdf:ingest --write # write to PDS
22 * bun rdf:ingest --write --json # write + JSON output
23 */
24
25import { resolve, dirname, join } from 'node:path';
26import { fileURLToPath } from 'node:url';
27import { readdirSync, readFileSync, mkdirSync, writeFileSync } from 'node:fs';
28import { AtpAgent } from '@atproto/api';
29import { loadDotEnv } from './cli-utils.js';
30
31const __dirname = dirname(fileURLToPath(import.meta.url));
32await loadDotEnv(resolve(__dirname, '..', '.env'));
33await loadDotEnv('/home/nandi/code/swarm/.env');
34
35const RDF_BASE = 'https://rdf.scios.tech/node/';
36const PDS_IDENTIFIER = process.env.PDS_IDENTIFIER ?? process.env.BLUESKY_IDENTIFIER ?? '';
37const PDS_APP_PASSWORD = process.env.PDS_APP_PASSWORD ?? process.env.BLUESKY_APP_PASSWORD ?? '';
38const PDS_SERVICE = process.env.PDS_SERVICE ?? 'https://amanita.us-east.host.bsky.network';
39const RDF_REPO = process.env.RDF_REPO ?? '/home/nandi/code/rdf';
40const PROVENANCE_PATH = process.env.RDF_INGEST_PROVENANCE ?? resolve(RDF_REPO, '.rdf-ingest.json');
41
42// --- Types ---
43
44interface NodeData {
45 id: string;
46 type: string;
47 title: string;
48 status: string;
49 sourceSection: string;
50 created: string;
51 edges: Record<string, string[]>;
52 body: string;
53 url: string;
54}
55
56interface EdgeToWrite {
57 source: string; // https://rdf.scios.tech/node/{from-id}
58 target: string; // https://rdf.scios.tech/node/{to-id}
59 connectionType: string;
60 note: string;
61 fromId: string;
62 toId: string;
63 edgeType: string; // original RDF edge type
64}
65
66const EDGE_TYPE_MAP: Record<string, string> = {
67 addresses: 'ADDRESSES',
68 supports: 'SUPPORTS',
69 opposes: 'OPPOSES',
70 derivedFrom: 'DERIVED_FROM',
71 informs: 'INFORMS',
72 usesMethod: 'USES_METHOD',
73};
74
75const TYPE_DIRS: Record<string, string> = {
76 claim: 'claims',
77 evidence: 'evidence',
78 question: 'questions',
79 source: 'sources',
80 method: 'methods',
81};
82
83// --- Parse ---
84
85function parseFrontmatter(content: string): { fm: Record<string, unknown>; body: string } {
86 const match = content.match(/^---\n([\s\S]*?)\n---\n([\s\S]*)$/);
87 if (!match) return { fm: {}, body: content };
88
89 const fmText = match[1];
90 const body = match[2].trim();
91
92 // Two-pass YAML parser:
93 // Pass 1: extract top-level keys and nested dicts (like edges:)
94 // Pass 2: for each nested dict, parse sub-keys with inline arrays
95
96 const fm: Record<string, unknown> = {};
97 const lines = fmText.split('\n');
98
99 let i = 0;
100 while (i < lines.length) {
101 const line = lines[i];
102
103 // Top-level key: value
104 const kvMatch = line.match(/^(\w+):\s*(.*)$/);
105 if (kvMatch) {
106 const key = kvMatch[1];
107 const val = kvMatch[2].trim();
108
109 if (val === '') {
110 // Could be a nested dict or empty — check next lines for indentation
111 const subDict: Record<string, string[]> = {};
112 let j = i + 1;
113 while (j < lines.length && lines[j].match(/^\s{2,}/)) {
114 const subMatch = lines[j].match(/^\s+(\w+):\s*(.*)$/);
115 if (subMatch) {
116 const subKey = subMatch[1];
117 const subVal = subMatch[2].trim();
118 if (subVal.startsWith('[') && subVal.endsWith(']')) {
119 subDict[subKey] = subVal.slice(1, -1).split(',').map(s => s.trim().replace(/['"]/g, '')).filter(Boolean);
120 } else if (subVal === '' || subVal === '[]') {
121 subDict[subKey] = [];
122 } else {
123 subDict[subKey] = [subVal.replace(/^['"]|['"]$/g, '')];
124 }
125 }
126 j++;
127 }
128 if (Object.keys(subDict).length > 0) {
129 fm[key] = subDict;
130 i = j;
131 continue;
132 }
133 // Empty dict
134 fm[key] = {};
135 i++;
136 continue;
137 }
138
139 if (val.startsWith('[') && val.endsWith(']')) {
140 fm[key] = val.slice(1, -1).split(',').map(s => s.trim().replace(/['"]/g, '')).filter(Boolean);
141 } else if (val === '[]') {
142 fm[key] = [];
143 } else {
144 fm[key] = val.replace(/^['"]|['"]$/g, '');
145 }
146 }
147 i++;
148 }
149
150 return { fm, body };
151}
152
153function loadNodes(): NodeData[] {
154 const nodes: NodeData[] = [];
155
156 for (const [type, dir] of Object.entries(TYPE_DIRS)) {
157 const dirPath = join(RDF_REPO, 'graph', dir);
158 let files: string[];
159 try {
160 files = readdirSync(dirPath).filter(f => f.endsWith('.md')).sort();
161 } catch {
162 console.error(` Skipping ${dirPath}: not found`);
163 continue;
164 }
165
166 for (const file of files) {
167 const content = readFileSync(join(dirPath, file), 'utf-8');
168 const { fm, body } = parseFrontmatter(content);
169
170 const id = (fm.id as string) ?? file.replace('.md', '');
171 const edges: Record<string, string[]> = {};
172 // fm.edges is the nested dict from YAML: { addresses: [Q-0001], supports: [...], ... }
173 const fmEdges = fm.edges as Record<string, string[] | unknown> | undefined;
174 if (fmEdges && typeof fmEdges === 'object') {
175 for (const [key, val] of Object.entries(fmEdges)) {
176 if (EDGE_TYPE_MAP[key]) {
177 edges[key] = Array.isArray(val) ? val : [val as string];
178 }
179 }
180 }
181 // Also check top-level edge keys (flat frontmatter style)
182 for (const [key, val] of Object.entries(fm)) {
183 if (key === 'edges') continue;
184 if (EDGE_TYPE_MAP[key] && !edges[key]) {
185 edges[key] = Array.isArray(val) ? val : [val as string];
186 }
187 }
188
189 nodes.push({
190 id,
191 type: (fm.type as string) ?? type,
192 title: (fm.title as string) ?? id,
193 status: (fm.status as string) ?? 'draft',
194 sourceSection: (fm.source_section as string) ?? '',
195 created: (fm.created as string) ?? '',
196 edges,
197 body,
198 url: `${RDF_BASE}${id}`,
199 });
200 }
201 }
202
203 return nodes;
204}
205
206function extractEdges(nodes: NodeData[]): EdgeToWrite[] {
207 const nodeMap = new Map(nodes.map(n => [n.id, n]));
208 const edges: EdgeToWrite[] = [];
209 const seen = new Set<string>();
210
211 for (const node of nodes) {
212 for (const [edgeType, targets] of Object.entries(node.edges)) {
213 const connectionType = EDGE_TYPE_MAP[edgeType];
214 if (!connectionType) continue;
215
216 for (const toId of targets) {
217 const key = `${node.id}:${edgeType}:${toId}`;
218 if (seen.has(key)) continue;
219 seen.add(key);
220
221 if (!nodeMap.has(toId)) {
222 console.error(` Warning: ${node.id} references missing node ${toId}`);
223 }
224
225 edges.push({
226 source: node.url,
227 target: `${RDF_BASE}${toId}`,
228 connectionType,
229 note: `${node.id} --${edgeType}--> ${toId}`,
230 fromId: node.id,
231 toId,
232 edgeType,
233 });
234 }
235 }
236 }
237
238 return edges;
239}
240
241// --- PDS writes ---
242
243async function writeEdges(edges: EdgeToWrite[]): Promise<{ wrote: number; uris: string[] }> {
244 if (!PDS_IDENTIFIER || !PDS_APP_PASSWORD) {
245 throw new Error('PDS_IDENTIFIER and PDS_APP_PASSWORD must be set');
246 }
247
248 const agent = new AtpAgent({ service: PDS_SERVICE });
249 await agent.login({ identifier: PDS_IDENTIFIER, password: PDS_APP_PASSWORD });
250 const did = agent.session!.did;
251
252 let wrote = 0;
253 const uris: string[] = [];
254
255 // Batch in groups of 10 to avoid rate limits
256 for (let i = 0; i < edges.length; i += 10) {
257 const batch = edges.slice(i, i + 10);
258 const results = await Promise.allSettled(
259 batch.map(async (edge) => {
260 const record = {
261 $type: 'network.cosmik.connection',
262 source: edge.source,
263 target: edge.target,
264 connectionType: edge.connectionType,
265 note: edge.note,
266 createdAt: new Date().toISOString(),
267 };
268
269 // Use createRecord to get auto-generated TID rkeys (putRecord with custom rkeys fails validation)
270 const res = await agent.api.com.atproto.repo.createRecord({
271 repo: did,
272 collection: 'network.cosmik.connection',
273 record,
274 });
275 return res.data.uri;
276 }),
277 );
278
279 for (const result of results) {
280 if (result.status === 'fulfilled') {
281 wrote++;
282 uris.push(result.value);
283 } else {
284 console.error(` Failed: ${result.reason}`);
285 }
286 }
287
288 // Small delay between batches
289 if (i + 10 < edges.length) {
290 await new Promise(r => setTimeout(r, 500));
291 }
292 }
293
294 return { wrote, uris };
295}
296
297// --- Main ---
298
299async function main(): Promise<void> {
300 const args = process.argv.slice(2);
301 const wantWrite = args.includes('--write');
302 const wantJson = args.includes('--json');
303
304 console.error('Loading nodes from RDF repo...');
305 const nodes = loadNodes();
306
307 const typeCounts: Record<string, number> = {};
308 for (const n of nodes) typeCounts[n.type] = (typeCounts[n.type] ?? 0) + 1;
309 console.error(` ${nodes.length} nodes: ${Object.entries(typeCounts).map(([t, c]) => `${t}: ${c}`).join(', ')}`);
310
311 console.error('Extracting edges...');
312 const edges = extractEdges(nodes);
313
314 const edgeTypeCounts: Record<string, number> = {};
315 for (const e of edges) edgeTypeCounts[e.edgeType] = (edgeTypeCounts[e.edgeType] ?? 0) + 1;
316 console.error(` ${edges.length} edges: ${Object.entries(edgeTypeCounts).map(([t, c]) => `${t}: ${c}`).join(', ')}`);
317
318 if (wantWrite) {
319 console.error(`Writing ${edges.length} edges to PDS...`);
320 const { wrote, uris } = await writeEdges(edges);
321 console.error(` Wrote ${wrote}/${edges.length} edges`);
322
323 saveJson(PROVENANCE_PATH, {
324 ingestedAt: new Date().toISOString(),
325 nodes: nodes.length,
326 edges: edges.length,
327 wrote,
328 uris: uris.slice(0, 100),
329 repo: 'https://github.com/jring-o/rdf',
330 });
331 }
332
333 if (wantJson) {
334 console.log(JSON.stringify({ nodes, edges, wrote: wantWrite ? edges.length : 0 }, null, 2));
335 return;
336 }
337
338 // Summary output
339 console.log(`\nRDF Discourse Graph: ${nodes.length} nodes, ${edges.length} edges`);
340 if (!wantWrite) {
341 console.log('(dry run — use --write to publish to PDS)');
342 }
343
344 // Sample edges
345 console.log('\nEdge sample:');
346 for (const e of edges.slice(0, 20)) {
347 console.log(` ${e.fromId} --${e.edgeType}--> ${e.toId} [${e.connectionType}]`);
348 }
349 if (edges.length > 20) {
350 console.log(` ... and ${edges.length - 20} more`);
351 }
352
353 // Node sample
354 console.log('\nNode sample:');
355 for (const n of nodes.slice(0, 10)) {
356 const edgeCount = Object.values(n.edges).flat().length;
357 console.log(` ${n.id} [${n.type}] "${n.title.slice(0, 60)}" (${edgeCount} edges)`);
358 }
359}
360
361function saveJson(path: string, value: unknown): void {
362 mkdirSync(dirname(path), { recursive: true });
363 writeFileSync(path, JSON.stringify(value, null, 2) + '\n', 'utf-8');
364}
365
366main().catch((err) => {
367 console.error('Fatal:', err instanceof Error ? err.message : String(err));
368 process.exit(1);
369});