WIP: feat: online catalog with CRDT sync, GossipSub, and Ed25519 signing

.gitignore

@@ -24,3 +24,17 @@ vite.config.js.timestamp-*
 **/cert.pem
 **/key.pem
 **/*.orig
+data-*/
+.node-*/
+
+# Editor/IDE local configs
+.[a-z]*/
+!.git/
+!.github/
+/[A-Z]*.md
+!/README.md
+
+# Test artifacts
+**/test-results/
+**/.playwright-real-data/
+**/playwright-report/

.peer-pids

@@ -0,0 +1,4 @@
+50372
+231896
+215452
+207536

backend/src/api/api.ts

@@ -10,6 +10,8 @@ import { initFsHandlers } from './fs.ts';
 import { initLISHsHandlers } from './lishs.ts';
 import { initTransferHandlers } from './transfer.ts';
 import { initEventsHandlers } from './events.ts';
+import { initCatalogHandlers } from './catalog.ts';
+import { type CatalogManager } from '../catalog/catalog-manager.ts';
 import { initSystemHandlers } from './system.ts';
 interface ClientData {
 	subscribedEvents: Set<string>;
@@ -41,7 +43,7 @@ export class APIServer {
 	private readonly dataServer: DataServer;
 	private readonly networks: Networks;
 
-	constructor(dataDir: string, dataServer: DataServer, networks: Networks, settings: Settings, options: APIServerOptions) {
+	constructor(dataDir: string, dataServer: DataServer, networks: Networks, settings: Settings, options: APIServerOptions, catalogManager?: CatalogManager | undefined) {
 		this.dataDir = dataDir;
 		this.dataServer = dataServer;
 		this.networks = networks;
@@ -60,6 +62,13 @@ export class APIServer {
 		const _fs = initFsHandlers();
 		const _lishs = initLISHsHandlers(this.dataServer, emitTo, broadcastFn);
 		const _transfer = initTransferHandlers(this.networks, this.dataServer, this.dataDir, emitTo, broadcastFn);
+		const _catalog = catalogManager ? initCatalogHandlers(catalogManager, {
+			networks: this.networks,
+			dataServer: this.dataServer,
+			dataDir: this.dataDir,
+			emit: emitTo,
+			broadcast: broadcastFn,
+		}) : null;
 		const hasSubscribers = (event: string): boolean => {
 			for (const client of this.clients) {
 				if (client.data.subscribedEvents.has(event) || client.data.subscribedEvents.has('*')) return true;
@@ -144,6 +153,22 @@ export class APIServer {
 			'fs.exists': _fs.exists,
 			'fs.writeText': _fs.writeText,
 			'fs.writeCompressed': _fs.writeCompressed,
+			// Catalog (optional — requires CatalogManager)
+			...(_catalog ? {
+				'catalog.list': _catalog.list,
+				'catalog.get': _catalog.get,
+				'catalog.search': _catalog.search,
+				'catalog.publish': _catalog.publish,
+				'catalog.update': _catalog.update,
+				'catalog.remove': _catalog.remove,
+				'catalog.getAccess': _catalog.getAccess,
+				'catalog.grantRole': _catalog.grantRole,
+				'catalog.revokeRole': _catalog.revokeRole,
+				'catalog.getSyncStatus': _catalog.getSyncStatus,
+				'catalog.startDownload': _catalog.startDownload,
+				'catalog.pauseDownload': _catalog.pauseDownload,
+				'catalog.resumeDownload': _catalog.resumeDownload,
+			} : {}),
 			// System
 			'system.ram': _system.ram,
 			'system.storage': _system.storage,

backend/src/api/catalog.ts

@@ -0,0 +1,190 @@
+import { type CatalogManager } from '../catalog/catalog-manager.ts';
+import { type Networks } from '../lishnet/lishnets.ts';
+import { type DataServer } from '../lish/data-server.ts';
+import { Downloader } from '../protocol/downloader.ts';
+import { type IStoredLISH, type HashAlgorithm, CodedError, ErrorCodes } from '@shared';
+import { Utils } from '../utils.ts';
+import { join } from 'path';
+import type { CatalogEntryRow, CatalogACLRow } from '../db/catalog.ts';
+
+const assert = Utils.assertParams;
+
+export interface StartDownloadResult {
+	status: 'downloading' | 'not_available';
+	message: string;
+	downloadDir?: string;
+}
+
+export interface CatalogHandlers {
+	list: (p: { networkID: string; limit?: number }) => CatalogEntryRow[];
+	get: (p: { networkID: string; lishID: string }) => CatalogEntryRow | null;
+	search: (p: { networkID: string; query: string; limit?: number }) => CatalogEntryRow[];
+	publish: (p: {
+		networkID: string; lishID: string; name?: string; description?: string;
+		chunkSize: number; checksumAlgo: string; totalSize: number; fileCount: number;
+		manifestHash: string; contentType?: string; tags?: string[];
+	}) => Promise<void>;
+	update: (p: { networkID: string; lishID: string; name?: string; description?: string; contentType?: string; tags?: string[] }) => Promise<void>;
+	remove: (p: { networkID: string; lishID: string }) => Promise<void>;
+	getAccess: (p: { networkID: string }) => CatalogACLRow | null;
+	grantRole: (p: { networkID: string; delegatee: string; role: 'admin' | 'moderator' }) => Promise<void>;
+	revokeRole: (p: { networkID: string; delegatee: string; role: 'admin' | 'moderator' }) => Promise<void>;
+	getSyncStatus: (p: { networkID: string }) => { entryCount: number; tombstoneCount: number; lastSyncAt: string | null };
+	startDownload: (p: { networkID: string; lishID: string }, client: any) => Promise<StartDownloadResult>;
+	pauseDownload: (p: { lishID: string }) => { success: boolean };
+	resumeDownload: (p: { lishID: string }) => { success: boolean };
+}
+
+type EmitFn = (client: any, event: string, data: any) => void;
+
+type BroadcastFn = (event: string, data: any) => void;
+
+export interface CatalogHandlerDeps {
+	networks: Networks;
+	dataServer: DataServer;
+	dataDir: string;
+	emit: EmitFn;
+	broadcast: BroadcastFn;
+}
+
+// Track active downloaders for pause/resume
+const activeDownloaders = new Map<string, Downloader>();
+
+export function initCatalogHandlers(catalogManager: CatalogManager, deps?: CatalogHandlerDeps): CatalogHandlers {
+	return {
+		list(p) {
+			assert(p, ['networkID']);
+			return catalogManager.list(p.networkID, p.limit);
+		},
+		get(p) {
+			assert(p, ['networkID', 'lishID']);
+			return catalogManager.get(p.networkID, p.lishID);
+		},
+		search(p) {
+			assert(p, ['networkID', 'query']);
+			return catalogManager.search(p.networkID, p.query, p.limit);
+		},
+		async publish(p) {
+			assert(p, ['networkID', 'lishID', 'chunkSize', 'checksumAlgo', 'totalSize', 'fileCount', 'manifestHash']);
+			await catalogManager.publish(p.networkID, p);
+		},
+		async update(p) {
+			assert(p, ['networkID', 'lishID']);
+			const fields: { name?: string; description?: string; contentType?: string; tags?: string[] } = {};
+			if (p.name !== undefined) fields.name = p.name;
+			if (p.description !== undefined) fields.description = p.description;
+			if (p.contentType !== undefined) fields.contentType = p.contentType;
+			if (p.tags !== undefined) fields.tags = p.tags;
+			await catalogManager.update(p.networkID, p.lishID, fields);
+		},
+		async remove(p) {
+			assert(p, ['networkID', 'lishID']);
+			await catalogManager.remove(p.networkID, p.lishID);
+		},
+		getAccess(p) {
+			assert(p, ['networkID']);
+			return catalogManager.getAccess(p.networkID);
+		},
+		async grantRole(p) {
+			assert(p, ['networkID', 'delegatee', 'role']);
+			await catalogManager.grantRole(p.networkID, p.delegatee, p.role);
+		},
+		async revokeRole(p) {
+			assert(p, ['networkID', 'delegatee', 'role']);
+			await catalogManager.revokeRole(p.networkID, p.delegatee, p.role);
+		},
+		getSyncStatus(p) {
+			assert(p, ['networkID']);
+			return catalogManager.getSyncStatus(p.networkID);
+		},
+		async startDownload(p, client): Promise<StartDownloadResult> {
+			assert(p, ['networkID', 'lishID']);
+			const entry = catalogManager.get(p.networkID, p.lishID);
+			if (!entry) {
+				return { status: 'not_available', message: 'Entry not found in catalog' };
+			}
+			if (!deps) {
+				return { status: 'not_available', message: 'Download infrastructure not available' };
+			}
+
+			// Build a stub LISH manifest from catalog entry metadata.
+			// The downloader will broadcast "want" on GossipSub and peers with the actual
+			// chunks will respond with "have" — the manifest only needs id, name, chunkSize, checksumAlgo.
+			const stubManifest: IStoredLISH = {
+				id: entry.lish_id,
+				name: entry.name ?? entry.lish_id,
+				description: entry.description ?? undefined,
+				created: entry.published_at ?? new Date().toISOString(),
+				chunkSize: entry.chunk_size,
+				checksumAlgo: (entry.checksum_algo as HashAlgorithm) ?? 'sha256',
+			};
+
+			try {
+				const network = deps.networks.getRunningNetwork();
+				const downloadDir = join(deps.dataDir, 'downloads', Date.now().toString());
+				const downloader = new Downloader(downloadDir, network, deps.dataServer, p.networkID);
+				await downloader.initFromManifest(stubManifest);
+
+				// Notify ALL clients when manifest is imported (LISH appears in downloads)
+				downloader.setManifestImportedCallback(lishID => {
+					const detail = deps.dataServer.getDetail(lishID);
+					if (detail) deps.broadcast('lishs:add', detail);
+				});
+
+				// Broadcast progress to ALL connected clients
+				downloader.setProgressCallback(info => {
+					deps.broadcast('transfer.download:progress', {
+						lishID: entry.lish_id,
+						downloadedChunks: info.downloadedChunks,
+						totalChunks: info.totalChunks,
+						peers: info.peers,
+						bytesPerSecond: info.bytesPerSecond,
+					});
+				});
+
+				// Track downloader for pause/resume
+				activeDownloaders.set(entry.lish_id, downloader);
+
+				// Start async download — broadcast completion/error to ALL clients
+				downloader
+					.download()
+					.then(() => {
+						activeDownloaders.delete(entry.lish_id);
+						deps.broadcast('transfer.download:complete', { downloadDir, lishID: entry.lish_id, name: entry.name });
+					})
+					.catch(err => {
+						activeDownloaders.delete(entry.lish_id);
+						if (err instanceof CodedError) deps.broadcast('transfer.download:error', { error: err.code, errorDetail: err.detail, lishID: entry.lish_id });
+						else deps.broadcast('transfer.download:error', { error: ErrorCodes.DOWNLOAD_ERROR, errorDetail: err.message, lishID: entry.lish_id });
+					});
+
+				return {
+					status: 'downloading',
+					message: `Download started for "${entry.name}". Looking for peers with the file...`,
+					downloadDir,
+				};
+			} catch (err: any) {
+				return {
+					status: 'not_available',
+					message: `Cannot start download: ${err.message}`,
+				};
+			}
+		},
+		pauseDownload(p) {
+			assert(p, ['lishID']);
+			const dl = activeDownloaders.get(p.lishID);
+			if (!dl) return { success: false };
+			dl.pause();
+			deps?.broadcast('transfer.download:paused', { lishID: p.lishID });
+			return { success: true };
+		},
+		resumeDownload(p) {
+			assert(p, ['lishID']);
+			const dl = activeDownloaders.get(p.lishID);
+			if (!dl) return { success: false };
+			dl.resume();
+			deps?.broadcast('transfer.download:resumed', { lishID: p.lishID });
+			return { success: true };
+		},
+	};
+}

backend/src/app.ts

@@ -6,6 +6,7 @@ import { DataServer } from './lish/data-server.ts';
 import { openDatabase } from './db/database.ts';
 import { APIServer } from './api/api.ts';
 import { Settings } from './settings.ts';
+import { CatalogManager } from './catalog/catalog-manager.ts';
 import { setWorkerUrl } from './lish/lish.ts';
 
 // Parse command line arguments
@@ -64,6 +65,24 @@ const dataServer = new DataServer(db);
 const networks = new Networks(db, dataDir, dataServer, settings, enablePink);
 networks.init();
 
+const catalogManager = new CatalogManager({
+	db,
+	getPrivateKey: () => networks.getRunningNetwork().getPrivateKey() as any,
+	getLocalPeerID: () => {
+		try { return networks.getRunningNetwork().getNodeInfo()?.peerID ?? 'local'; } catch { return 'local'; }
+	},
+	broadcast: (networkID, op) => {
+		try {
+			const net = networks.getRunningNetwork();
+			net.broadcast(`lish/${networkID}`, { type: 'catalog_op', ...op });
+		} catch { /* network not running — skip broadcast */ }
+	},
+	emitEvent: (event, data) => {
+		try { apiServer.broadcastEvent(event, data); } catch { /* server not started */ }
+	},
+});
+networks.setCatalogManager(catalogManager);
+
 // Apply speed limits from settings
 import { Downloader } from './protocol/downloader.ts';
 import { setMaxUploadSpeed, setUploadBroadcast, initUploadState } from './protocol/lish-protocol.ts';
@@ -81,7 +100,7 @@ const apiServer = new APIServer(dataDir, dataServer, networks, settings, {
 	secure: apiSecure,
 	keyFile: apiKeyFile,
 	certFile: apiCertFile,
-});
+}, catalogManager);
 
 // Wire upload progress broadcast (after apiServer is created)
 setUploadBroadcast((event, data) => apiServer.broadcastEvent(event, data));