p2pds / src / start.ts

/**
 * Extracted server startup logic — importable and testable.
 *
 * startServer(config, opts?) does everything server.ts used to do at module scope:
 * creates DB, IPFS, replication, HTTP server, etc. and returns a ServerHandle
 * for programmatic control (tests, Tauri sidecar, etc.).
 */

import { getRequestListener } from "@hono/node-server";
import { createServer, type Server } from "node:http";
import { mkdirSync } from "node:fs";
import { resolve } from "node:path";
import Database from "better-sqlite3";
import { WebSocketServer } from "ws";
import pc from "picocolors";

import { loadPolicies } from "./config.js";
import type { Config } from "./config.js";
import { RepoManager } from "./repo-manager.js";
import { BlobStore } from "./blobs.js";
import { Firehose } from "./firehose.js";
import { IpfsService } from "./ipfs.js";
import { createApp } from "./index.js";
import { DidResolver } from "./did-resolver.js";
import { InMemoryDidCache } from "./did-cache.js";
import { ReplicationManager } from "./replication/replication-manager.js";
import { ReplicatedRepoReader } from "./replication/replicated-repo-reader.js";
import { PolicyEngine } from "./policy/engine.js";
import { PolicyStorage } from "./policy/storage.js";
import { migrateToNewPolicies } from "./policy/migrate.js";
import { HttpChallengeTransport } from "./replication/challenge-response/http-transport.js";
import { Libp2pChallengeTransport } from "./replication/challenge-response/libp2p-transport.js";
import { FailoverChallengeTransport } from "./replication/challenge-response/failover-transport.js";
import type { ChallengeTransport } from "./replication/challenge-response/transport.js";
import type { Libp2p } from "@libp2p/interface";
import { RateLimiter } from "./rate-limiter.js";
import { createOAuthClient, type OAuthClientManager } from "./oauth/client.js";
import { PdsClient } from "./oauth/pds-client.js";
import { type PdsClientRef, publishPeerRecord } from "./oauth/routes.js";
import { registerRepoSyncProtocol } from "./replication/libp2p-sync.js";

export interface StartServerOpts {
	/** Override DID resolver (e.g. mock resolver for tests). */
	didResolver?: DidResolver;
}

export interface ServerHandle {
	url: string;
	port: number;
	close: () => Promise<void>;
	replicationManager?: ReplicationManager;
	ipfsService?: IpfsService;
}

export async function startServer(
	config: Config,
	opts?: StartServerOpts,
): Promise<ServerHandle> {
	// Initialize rate limiter
	let rateLimiter: RateLimiter | undefined;
	if (config.RATE_LIMIT_ENABLED) {
		rateLimiter = new RateLimiter();
		rateLimiter.startCleanup(60_000);
	}

	// Ensure data directory exists
	const dataDir = resolve(config.DATA_DIR);
	mkdirSync(dataDir, { recursive: true });

	// Initialize SQLite database
	const dbPath = resolve(dataDir, "pds.db");
	const db = new Database(dbPath);
	db.pragma("journal_mode = WAL");
	db.pragma("synchronous = NORMAL");

	// Create node_identity table for OAuth-established identity
	db.exec(`
		CREATE TABLE IF NOT EXISTS node_identity (
			did TEXT PRIMARY KEY,
			handle TEXT,
			created_at TEXT NOT NULL DEFAULT (datetime('now'))
		)
	`);

	// Persist AUTH_TOKEN so the UI token survives restarts
	db.exec(`
		CREATE TABLE IF NOT EXISTS node_settings (
			key TEXT PRIMARY KEY,
			value TEXT NOT NULL
		)
	`);
	if (!process.env.AUTH_TOKEN) {
		const storedToken = db.prepare("SELECT value FROM node_settings WHERE key = 'auth_token'").get() as
			| { value: string }
			| undefined;
		if (storedToken) {
			config.AUTH_TOKEN = storedToken.value;
		} else {
			db.prepare("INSERT OR REPLACE INTO node_settings (key, value) VALUES ('auth_token', ?)").run(config.AUTH_TOKEN);
		}
	}

	// Load stored identity from previous OAuth login (if not overridden by env)
	const storedIdentity = db.prepare("SELECT did, handle FROM node_identity LIMIT 1").get() as
		| { did: string; handle: string | null }
		| undefined;
	if (storedIdentity && !config.DID) {
		config.DID = storedIdentity.did;
		if (storedIdentity.handle) {
			config.HANDLE = storedIdentity.handle;
		}
	}

	// Initialize IPFS service (if enabled)
	let ipfsService: IpfsService | undefined;
	if (config.IPFS_ENABLED) {
		ipfsService = new IpfsService({
			db,
			networking: config.IPFS_NETWORKING,
		});
	}

	// Initialize repo manager (requires DID + signing key)
	let repoManager: RepoManager | undefined;
	let blobStore: BlobStore | undefined;
	if (config.DID && config.SIGNING_KEY) {
		repoManager = new RepoManager(db, config as typeof config & { DID: string; SIGNING_KEY: string });
		blobStore = new BlobStore(dataDir, config.DID);
		repoManager.init(blobStore, ipfsService, ipfsService);
	}

	// Initialize firehose (repoManager is optional — without it, no local event backfill)
	const firehose = new Firehose(repoManager);

	// Initialize DID resolver (allow override for tests)
	const didResolver = opts?.didResolver ?? new DidResolver({
		didCache: new InMemoryDidCache(),
	});

	// Initialize policy storage and engine (always created — needed for offer management)
	const policyStorage = new PolicyStorage(db);
	policyStorage.initSchema();

	const policySet = config.POLICY_FILE ? loadPolicies(config) : null;
	const policyEngine = new PolicyEngine(
		policySet ?? { version: 1, policies: [] },
		policyStorage,
	);

	// Seed from POLICY_FILE only if DB is empty (first run with a policy file)
	if (policySet && policyStorage.count() === 0) {
		policyEngine.persistAll();
		console.log(pc.dim(`  Policies: seeded ${policySet.policies.length} from ${config.POLICY_FILE}`));
	} else if (policySet) {
		console.log(pc.dim(`  Policies: loaded ${policySet.policies.length} from ${config.POLICY_FILE}`));
	}

	// Load persisted policies from DB (merges with any file-loaded policies)
	policyEngine.loadFromDb();

	// Run one-time migration: config DIDs + admin_tracked_dids → policy engine
	const migrated = migrateToNewPolicies(db, policyStorage, policyEngine, config.REPLICATE_DIDS);
	if (migrated > 0) {
		console.log(pc.dim(`  Policies: migrated ${migrated} legacy DIDs`));
	}

	// Initialize OAuth client (if enabled)
	let oauthClientManager: OAuthClientManager | undefined;
	const pdsClientRef: PdsClientRef = { current: undefined };
	if (config.OAUTH_ENABLED) {
		oauthClientManager = await createOAuthClient(db, config);
		if (config.DID) {
			pdsClientRef.current = new PdsClient(oauthClientManager.client, config.DID);
		}
	}

	// Initialize replication manager and replicated repo reader (if IPFS enabled)
	// repoManager is optional — without it, manifest records are skipped but sync works
	let replicationManager: ReplicationManager | undefined;
	let replicatedRepoReader: ReplicatedRepoReader | undefined;
	if (ipfsService) {
		replicationManager = new ReplicationManager(
			db,
			config,
			repoManager,
			ipfsService,
			ipfsService,
			didResolver,
			undefined,
			undefined,
			policyEngine,
			pdsClientRef.current,
		);
		replicatedRepoReader = new ReplicatedRepoReader(
			ipfsService,
			replicationManager.getSyncStorage(),
		);
		replicationManager.setReplicatedRepoReader(replicatedRepoReader);
	}

	// Pass rate limiter to IPFS service for gossipsub rate limiting
	if (rateLimiter && ipfsService) {
		ipfsService.setRateLimiter(rateLimiter);
	}

	// --- Lazy IPFS startup: deferred until identity is established ---
	let ipfsStarted = false;

	async function startIpfsReplication(): Promise<void> {
		if (ipfsStarted || !ipfsService) return;
		ipfsStarted = true;

		try {
			console.log(pc.dim(`  IPFS:   starting...`));
			await ipfsService.start();
			const peerId = ipfsService.getPeerId();
			if (peerId) {
				console.log(pc.dim(`  PeerID: ${peerId}`));
			} else {
				console.log(pc.dim(`  IPFS:   local blockstore only (networking disabled)`));
			}
			await backfillIpfs();
			// Start replication after IPFS is ready
			if (replicationManager) {
				try {
					await replicationManager.init();

					// Auto-track the node's own DID (idempotent — no-op if already tracked)
					if (config.DID) {
						await replicationManager.addDid(config.DID).catch((err) => {
							console.warn("[start] Self-replication addDid failed:", err instanceof Error ? err.message : String(err));
						});
					}

					// Register libp2p repo sync protocol and set libp2p on ReplicationManager
					const libp2pForSync = ipfsService?.getLibp2p();
					if (libp2pForSync) {
						const syncStorage = replicationManager.getSyncStorage();
						registerRepoSyncProtocol(
							libp2pForSync as Libp2p,
							ipfsService!,
							syncStorage,
						);
						replicationManager.setLibp2p(libp2pForSync as Libp2p);
						console.log(pc.dim(`  Repo sync: libp2p protocol registered`));
					}

					replicationManager.startPeriodicSync();
					const trackedDids = replicationManager.getReplicateDids();
					console.log(pc.dim(`  Replication: tracking ${trackedDids.length} DIDs`));
					// Start firehose subscription for real-time updates
					if (config.FIREHOSE_ENABLED) {
						replicationManager.startFirehose();
					}
					// Start challenge scheduler if policy engine is available
					if (policyEngine) {
						const libp2pNode = ipfsService?.getLibp2p();
						let challengeTransport: ChallengeTransport;
						if (libp2pNode) {
							const libp2pTransport = new Libp2pChallengeTransport(libp2pNode as Libp2p);
							const httpTransport = new HttpChallengeTransport();
							const syncStorage = replicationManager.getSyncStorage();
							challengeTransport = new FailoverChallengeTransport(
								libp2pTransport,
								httpTransport,
								{
									resolveEndpoint: (httpEndpoint) => syncStorage.getMultiaddrForPdsEndpoint(httpEndpoint),
									onFallback: (endpoint, error) => {
										console.log(pc.dim(`  Challenge: libp2p failed for ${endpoint}, falling back to HTTP: ${error.message}`));
										replicationManager!.refreshPeerInfoForEndpoint(endpoint);
									},
								},
							);
						} else {
							challengeTransport = new HttpChallengeTransport();
						}
						replicationManager.startChallengeScheduler(challengeTransport);
						console.log(pc.dim(`  Challenges: scheduler started (${libp2pNode ? "libp2p+HTTP failover" : "HTTP"} transport)`));
					}

					// Set up auto-republish of peer record when multiaddrs change
					replicationManager.setPublishPeerRecordFn(async () => {
						if (pdsClientRef.current && ipfsService) {
							await publishPeerRecord(pdsClientRef.current, ipfsService, config.PUBLIC_URL);
						}
					});
				} catch (err) {
					console.error(pc.red(`  Replication startup failed:`), err);
				}
			}
		} catch (err) {
			ipfsStarted = false; // Allow retry on failure
			console.error(pc.red(`  IPFS startup failed:`), err);
		}
	}

	// Create Hono app
	const app = createApp(
		config,
		firehose,
		ipfsService,
		ipfsService,
		blobStore,
		replicationManager,
		replicatedRepoReader,
		repoManager,
		rateLimiter,
		oauthClientManager,
		pdsClientRef,
		db,
		startIpfsReplication,
		ipfsService,
	);

	// Create HTTP server using @hono/node-server's request listener
	const requestListener = getRequestListener(app.fetch);
	const httpServer = createServer(requestListener);

	// Set up WebSocket server for firehose with per-IP connection limits
	const wss = new WebSocketServer({ noServer: true });
	const firehoseConnections = new Map<string, number>();
	const maxFirehosePerIp = config.RATE_LIMIT_FIREHOSE_PER_IP;

	httpServer.on("upgrade", (request, socket, head) => {
		const url = new URL(request.url ?? "/", `http://localhost:${config.PORT}`);

		if (url.pathname === "/xrpc/com.atproto.sync.subscribeRepos") {
			const ip =
				(request.headers["x-forwarded-for"] as string)?.split(",")[0]?.trim() ??
				request.headers["x-real-ip"] as string ??
				"unknown";

			const current = firehoseConnections.get(ip) ?? 0;
			if (config.RATE_LIMIT_ENABLED && current >= maxFirehosePerIp) {
				socket.destroy();
				return;
			}

			wss.handleUpgrade(request, socket, head, (ws) => {
				firehoseConnections.set(ip, (firehoseConnections.get(ip) ?? 0) + 1);
				ws.on("close", () => {
					const count = (firehoseConnections.get(ip) ?? 1) - 1;
					if (count <= 0) {
						firehoseConnections.delete(ip);
					} else {
						firehoseConnections.set(ip, count);
					}
				});
				firehose.handleConnection(ws, request);
			});
		} else {
			socket.destroy();
		}
	});

	// Backfill existing blocks to IPFS (only when RepoManager created the blocks table)
	async function backfillIpfs(): Promise<void> {
		if (!ipfsService || !repoManager) return;

		const rows = db
			.prepare("SELECT cid, bytes FROM blocks")
			.all() as Array<{ cid: string; bytes: Buffer }>;

		if (rows.length === 0) return;

		let count = 0;
		for (const row of rows) {
			const hasIt = await ipfsService.hasBlock(row.cid);
			if (!hasIt) {
				await ipfsService.putBlock(row.cid, new Uint8Array(row.bytes));
				count++;
			}
		}

		if (count > 0) {
			console.log(pc.dim(`  IPFS:   backfilled ${count} blocks`));
		}
	}

	// Start the HTTP server and async initialization
	const { url, port } = await new Promise<{ url: string; port: number }>((resolveStart) => {
		httpServer.listen(config.PORT, async () => {
			const addr = httpServer.address() as { port: number };
			const actualPort = addr.port;
			const url = `http://localhost:${actualPort}`;

			console.log(
				pc.bold(`\nP2PDS running at `) +
					pc.cyan(url),
			);
			// Machine-readable line for sidecar consumers (Tauri, tests)
			console.log(`P2PDS_READY ${JSON.stringify({ port: actualPort, url })}`);
			if (config.DID) {
				console.log(pc.dim(`  DID:    ${config.DID}`));
			}
			if (config.HANDLE) {
				console.log(pc.dim(`  Handle: @${config.HANDLE}`));
			}
			console.log(pc.dim(`  Data:   ${dataDir}`));
			if (oauthClientManager) {
				if (pdsClientRef.current && await pdsClientRef.current.hasSession().catch(() => false)) {
					console.log(pc.dim(`  OAuth:  session active for ${config.DID}`));
				} else {
					console.log(pc.dim(`  OAuth:  enabled (no active session)`));
				}
			}

			// Start IPFS after HTTP server is listening (IPFS startup can be slow)
			if (ipfsService && config.DID) {
				await startIpfsReplication();
			} else if (ipfsService) {
				console.log(pc.dim(`  IPFS:   waiting for login`));
			}

			console.log();
			resolveStart({ url, port: actualPort });
		});
	});

	// Build close function
	const close = async (): Promise<void> => {
		if (rateLimiter) {
			rateLimiter.stop();
		}
		if (replicationManager) {
			replicationManager.stop();
		}
		wss.close();
		// Force-close keep-alive connections so httpServer.close() resolves promptly
		httpServer.closeAllConnections();
		await new Promise<void>((res) => httpServer.close(() => res()));
		if (ipfsService) {
			await ipfsService.stop();
		}
		db.close();
	};

	return {
		url,
		port,
		close,
		replicationManager,
		ipfsService,
	};
}