p2pds / src / ipfs.test.ts

import { describe, it, expect, beforeEach, afterEach } from "vitest";
import { mkdtempSync, rmSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { IpfsService } from "./ipfs.js";
import {
	create as createCid,
	CODEC_RAW,
	toString as cidToString,
} from "@atcute/cid";
import Database from "better-sqlite3";
import { createApp } from "./index.js";
import { RepoManager } from "./repo-manager.js";
import { BlobStore } from "./blobs.js";
import { Firehose } from "./firehose.js";
import { loadConfig } from "./config.js";
import type { Config } from "./config.js";
import { BlockMap } from "@atproto/repo";
import { CID } from "@atproto/lex-data";

/** Create a CID string from raw bytes using SHA-256. */
async function makeCidStr(bytes: Uint8Array): Promise<string> {
	const cid = await createCid(CODEC_RAW, bytes);
	return cidToString(cid);
}

/** Create a minimal test Config. */
function testConfig(dataDir: string): Config {
	return {
		DID: "did:plc:test123",
		HANDLE: "test.example.com",
		PDS_HOSTNAME: "test.example.com",
		AUTH_TOKEN: "test-auth-token",
		SIGNING_KEY:
			"0000000000000000000000000000000000000000000000000000000000000001",
		SIGNING_KEY_PUBLIC: "zQ3shP2mWsZYWgvZM9GJ3EvMfRXQJwuTh6BdXLvJB9gFhT3Lr",
		JWT_SECRET: "test-jwt-secret",
		PASSWORD_HASH: "$2a$10$test",
		DATA_DIR: dataDir,
		PORT: 3000,
		IPFS_ENABLED: true,
		IPFS_NETWORKING: false,
		REPLICATE_DIDS: [],
		FIREHOSE_URL: "wss://localhost/xrpc/com.atproto.sync.subscribeRepos",
		FIREHOSE_ENABLED: false,
		RATE_LIMIT_ENABLED: false,
		RATE_LIMIT_READ_PER_MIN: 300,
		RATE_LIMIT_SYNC_PER_MIN: 30,
		RATE_LIMIT_SESSION_PER_MIN: 10,
		RATE_LIMIT_WRITE_PER_MIN: 200,
		RATE_LIMIT_CHALLENGE_PER_MIN: 20,
		RATE_LIMIT_MAX_CONNECTIONS: 100,
		RATE_LIMIT_FIREHOSE_PER_IP: 3,
		OAUTH_ENABLED: false, PUBLIC_URL: "http://localhost:3000",
	};
}

// ============================================
// IpfsService Unit Tests
// ============================================

describe("IpfsService", () => {
	let tmpDir: string;
	let db: InstanceType<typeof Database>;
	let service: IpfsService;

	beforeEach(() => {
		tmpDir = mkdtempSync(join(tmpdir(), "ipfs-test-"));
		db = new Database(join(tmpDir, "test.db"));
		service = new IpfsService({
			db,
			networking: false,
		});
	});

	afterEach(async () => {
		if (service.isRunning()) {
			await service.stop();
		}
		db.close();
		rmSync(tmpDir, { recursive: true, force: true });
	});

	describe("lifecycle", () => {
		it("start sets isRunning to true", async () => {
			expect(service.isRunning()).toBe(false);
			await service.start();
			expect(service.isRunning()).toBe(true);
		});

		it("stop sets isRunning to false", async () => {
			await service.start();
			await service.stop();
			expect(service.isRunning()).toBe(false);
		});

		it("getPeerId returns null without networking", async () => {
			await service.start();
			expect(service.getPeerId()).toBeNull();
		});

		it("getConnectionCount returns 0 without networking", async () => {
			await service.start();
			expect(service.getConnectionCount()).toBe(0);
		});
	});

	describe("putBlock + getBlock roundtrip", () => {
		it("stores and retrieves bytes by CID", async () => {
			await service.start();
			const bytes = new TextEncoder().encode("hello ipfs");
			const cidStr = await makeCidStr(bytes);

			await service.putBlock(cidStr, bytes);
			const retrieved = await service.getBlock(cidStr);

			expect(retrieved).not.toBeNull();
			expect(Buffer.from(retrieved!)).toEqual(Buffer.from(bytes));
		});
	});

	describe("putBlock + hasBlock", () => {
		it("returns true for stored block", async () => {
			await service.start();
			const bytes = new TextEncoder().encode("test block");
			const cidStr = await makeCidStr(bytes);

			await service.putBlock(cidStr, bytes);
			expect(await service.hasBlock(cidStr)).toBe(true);
		});

		it("returns false for unknown CID", async () => {
			await service.start();
			const bytes = new TextEncoder().encode("unknown block");
			const cidStr = await makeCidStr(bytes);
			expect(await service.hasBlock(cidStr)).toBe(false);
		});
	});

	describe("getBlock missing", () => {
		it("returns null for CID not in store", async () => {
			await service.start();
			const bytes = new TextEncoder().encode("missing block");
			const cidStr = await makeCidStr(bytes);
			expect(await service.getBlock(cidStr)).toBeNull();
		});
	});

	describe("putBlocks (BlockMap)", () => {
		it("stores multiple blocks from a BlockMap", async () => {
			await service.start();

			const entries: Array<{ bytes: Uint8Array; cidStr: string }> = [];
			for (let i = 0; i < 3; i++) {
				const bytes = new TextEncoder().encode(`block-${i}`);
				const cidStr = await makeCidStr(bytes);
				entries.push({ bytes, cidStr });
			}

			const blockMap = new BlockMap();
			for (const entry of entries) {
				blockMap.set(CID.parse(entry.cidStr), entry.bytes);
			}

			await service.putBlocks(blockMap);

			for (const entry of entries) {
				const retrieved = await service.getBlock(entry.cidStr);
				expect(retrieved).not.toBeNull();
				expect(Buffer.from(retrieved!)).toEqual(Buffer.from(entry.bytes));
			}
		});
	});

	describe("graceful no-ops before start", () => {
		it("putBlock does not throw", async () => {
			const bytes = new TextEncoder().encode("test");
			const cidStr = await makeCidStr(bytes);
			await expect(
				service.putBlock(cidStr, bytes),
			).resolves.toBeUndefined();
		});

		it("getBlock returns null", async () => {
			const bytes = new TextEncoder().encode("test");
			const cidStr = await makeCidStr(bytes);
			expect(await service.getBlock(cidStr)).toBeNull();
		});

		it("hasBlock returns false", async () => {
			const bytes = new TextEncoder().encode("test");
			const cidStr = await makeCidStr(bytes);
			expect(await service.hasBlock(cidStr)).toBe(false);
		});
	});

	describe("provideBlocks", () => {
		it("resolves without error when no networking", async () => {
			await service.start();
			const bytes = new TextEncoder().encode("provide-test");
			const cidStr = await makeCidStr(bytes);
			await expect(
				service.provideBlocks([cidStr]),
			).resolves.toBeUndefined();
		});
	});

	describe("gossipsub no-ops without networking", () => {
		it("onCommitNotification does not throw", async () => {
			await service.start();
			expect(() => service.onCommitNotification(() => {})).not.toThrow();
		});

		it("subscribeCommitTopics does not throw", async () => {
			await service.start();
			expect(() => service.subscribeCommitTopics(["did:plc:test"])).not.toThrow();
		});

		it("unsubscribeCommitTopics does not throw", async () => {
			await service.start();
			expect(() => service.unsubscribeCommitTopics(["did:plc:test"])).not.toThrow();
		});

		it("publishCommitNotification resolves without error", async () => {
			await service.start();
			await expect(
				service.publishCommitNotification("did:plc:test", "bafytest", "rev1"),
			).resolves.toBeUndefined();
		});
	});
});

// ============================================
// RASL Endpoint Integration Tests
// ============================================

describe("RASL endpoint", () => {
	let tmpDir: string;
	let db: InstanceType<typeof Database>;
	let ipfsService: IpfsService;
	let blobStore: BlobStore;
	let app: ReturnType<typeof createApp>;

	beforeEach(async () => {
		tmpDir = mkdtempSync(join(tmpdir(), "rasl-test-"));
		const config = testConfig(tmpDir);

		db = new Database(join(tmpDir, "test.db"));
		const repoManager = new RepoManager(db, config);
		repoManager.init();

		const firehose = new Firehose(repoManager);

		ipfsService = new IpfsService({
			db,
			networking: false,
		});
		await ipfsService.start();

		blobStore = new BlobStore(tmpDir, config.DID!);

		app = createApp(config, firehose, ipfsService, ipfsService, blobStore, undefined, undefined, repoManager);
	});

	afterEach(async () => {
		if (ipfsService.isRunning()) {
			await ipfsService.stop();
		}
		db.close();
		rmSync(tmpDir, { recursive: true, force: true });
	});

	it("fetches block from IPFS", async () => {
		const bytes = new TextEncoder().encode("ipfs block data");
		const cidStr = await makeCidStr(bytes);

		await ipfsService.putBlock(cidStr, bytes);

		const res = await app.request(
			`/.well-known/rasl/${cidStr}`,
			undefined,
			{},
		);
		expect(res.status).toBe(200);

		const body = new Uint8Array(await res.arrayBuffer());
		expect(Buffer.from(body)).toEqual(Buffer.from(bytes));
	});

	it("falls back to SQLite", async () => {
		const bytes = new TextEncoder().encode("sqlite block data");
		const cidStr = await makeCidStr(bytes);

		// Insert directly into blocks table (not in IPFS)
		db.prepare("INSERT INTO blocks (cid, bytes, rev) VALUES (?, ?, ?)").run(
			cidStr,
			Buffer.from(bytes),
			"test-rev",
		);

		const res = await app.request(
			`/.well-known/rasl/${cidStr}`,
			undefined,
			{},
		);
		expect(res.status).toBe(200);

		const body = new Uint8Array(await res.arrayBuffer());
		expect(Buffer.from(body)).toEqual(Buffer.from(bytes));
	});

	it("falls back to blob store", async () => {
		const bytes = new TextEncoder().encode("blob data");
		const blobRef = await blobStore.putBlob(bytes, "application/octet-stream");
		const cidStr = blobRef.ref.$link;

		const res = await app.request(
			`/.well-known/rasl/${cidStr}`,
			undefined,
			{},
		);
		expect(res.status).toBe(200);

		const body = new Uint8Array(await res.arrayBuffer());
		expect(Buffer.from(body)).toEqual(Buffer.from(bytes));
	});

	it("returns 404 for missing block", async () => {
		const bytes = new TextEncoder().encode("nonexistent");
		const cidStr = await makeCidStr(bytes);

		const res = await app.request(
			`/.well-known/rasl/${cidStr}`,
			undefined,
			{},
		);
		expect(res.status).toBe(404);

		const json = (await res.json()) as { error: string };
		expect(json.error).toBe("BlockNotFound");
	});

	it("returns correct response headers", async () => {
		const bytes = new TextEncoder().encode("header test");
		const cidStr = await makeCidStr(bytes);

		await ipfsService.putBlock(cidStr, bytes);

		const res = await app.request(
			`/.well-known/rasl/${cidStr}`,
			undefined,
			{},
		);
		expect(res.status).toBe(200);
		expect(res.headers.get("Content-Type")).toBe("application/octet-stream");
		expect(res.headers.get("Cache-Control")).toBe(
			"public, max-age=31536000, immutable",
		);
		expect(res.headers.get("ETag")).toBe(`"${cidStr}"`);
	});
});

// ============================================
// Config Tests
// ============================================

describe("config", () => {
	const envKeys = [
		"DID",
		"HANDLE",
		"PDS_HOSTNAME",
		"AUTH_TOKEN",
		"SIGNING_KEY",
		"SIGNING_KEY_PUBLIC",
		"JWT_SECRET",
		"PASSWORD_HASH",
		"IPFS_ENABLED",
		"IPFS_NETWORKING",
		"DATA_DIR",
		"PORT",
		"EMAIL",
	];
	const savedEnv: Record<string, string | undefined> = {};

	beforeEach(() => {
		for (const key of envKeys) {
			savedEnv[key] = process.env[key];
		}
		process.env.DID = "did:plc:test";
		process.env.HANDLE = "test.example.com";
		process.env.PDS_HOSTNAME = "test.example.com";
		process.env.AUTH_TOKEN = "token";
		process.env.SIGNING_KEY = "key";
		process.env.SIGNING_KEY_PUBLIC = "pubkey";
		process.env.JWT_SECRET = "secret";
		process.env.PASSWORD_HASH = "hash";
		delete process.env.IPFS_ENABLED;
		delete process.env.IPFS_NETWORKING;
	});

	afterEach(() => {
		for (const key of envKeys) {
			if (savedEnv[key] === undefined) {
				delete process.env[key];
			} else {
				process.env[key] = savedEnv[key];
			}
		}
	});

	it("IPFS_ENABLED defaults to true", () => {
		const config = loadConfig("/nonexistent/.env");
		expect(config.IPFS_ENABLED).toBe(true);
	});

	it('IPFS_ENABLED set to "false" returns false', () => {
		process.env.IPFS_ENABLED = "false";
		const config = loadConfig("/nonexistent/.env");
		expect(config.IPFS_ENABLED).toBe(false);
	});

	it("IPFS_NETWORKING defaults to true", () => {
		const config = loadConfig("/nonexistent/.env");
		expect(config.IPFS_NETWORKING).toBe(true);
	});

	it('IPFS_NETWORKING set to "false" returns false', () => {
		process.env.IPFS_NETWORKING = "false";
		const config = loadConfig("/nonexistent/.env");
		expect(config.IPFS_NETWORKING).toBe(false);
	});
});