chadtmiller.com/pds.js

Switch repository

Loading…

#!/usr/bin/env node

// @pdsjs/readonly CLI - Run a read-only PDS server from CAR files

import fs from 'node:fs';
import { createServer } from 'node:http';
import path from 'node:path';
import Database from 'better-sqlite3';
import { createReadOnlyServer } from './index.js';

/**
 * Parse command line arguments
 * @returns {{ carFiles: string[], blobsDir: string | null, port: number, hostname: string, dataDir: string }}
 */
function parseArgs() {
  const args = process.argv.slice(2);
  /** @type {string[]} */
  const carFiles = [];
  let blobsDir = null;
  let port = 3000;
  let hostname = 'localhost';
  let dataDir = './pds-data';

  for (let i = 0; i < args.length; i++) {
    const arg = args[i];

    if (arg === '--car' || arg === '-c') {
      const value = args[++i];
      if (!value) {
        console.error('Error: --car requires a path');
        process.exit(1);
      }
      // Support glob patterns (Node.js 22+)
      if (value.includes('*')) {
        try {
          // Use dynamic import for glob (Node.js 22+)
          const { globSync } = fs;
          if (typeof globSync === 'function') {
            carFiles.push(...globSync(value));
          } else {
            console.error(
              'Warning: Glob patterns not supported in this Node.js version, please specify files individually',
            );
            carFiles.push(value);
          }
        } catch {
          console.error(
            'Warning: Glob patterns not supported, please specify files individually',
          );
          carFiles.push(value);
        }
      } else {
        carFiles.push(value);
      }
    } else if (arg === '--blobs' || arg === '-b') {
      blobsDir = args[++i];
      if (!blobsDir) {
        console.error('Error: --blobs requires a path');
        process.exit(1);
      }
    } else if (arg === '--port' || arg === '-p') {
      port = parseInt(args[++i], 10);
      if (Number.isNaN(port)) {
        console.error('Error: --port requires a number');
        process.exit(1);
      }
    } else if (arg === '--hostname' || arg === '-H') {
      hostname = args[++i];
      if (!hostname) {
        console.error('Error: --hostname requires a value');
        process.exit(1);
      }
    } else if (arg === '--data-dir' || arg === '-d') {
      dataDir = args[++i];
      if (!dataDir) {
        console.error('Error: --data-dir requires a path');
        process.exit(1);
      }
    } else if (arg === '--help' || arg === '-h') {
      printHelp();
      process.exit(0);
    } else if (arg === '--version' || arg === '-v') {
      const pkg = JSON.parse(
        fs.readFileSync(new URL('../package.json', import.meta.url), 'utf-8'),
      );
      console.log(pkg.version);
      process.exit(0);
    } else if (arg.startsWith('-')) {
      console.error(`Unknown option: ${arg}`);
      printHelp();
      process.exit(1);
    } else {
      // Positional arguments treated as CAR files
      carFiles.push(arg);
    }
  }

  // Check for CAR_DIR environment variable
  const carDir = process.env.PDS_CAR_DIR;
  if (carDir && carFiles.length === 0) {
    const files = fs.readdirSync(carDir).filter((f) => f.endsWith('.car'));
    carFiles.push(...files.map((f) => path.join(carDir, f)));
  }

  // Check for BLOBS_DIR environment variable
  if (!blobsDir && process.env.PDS_BLOBS_DIR) {
    blobsDir = process.env.PDS_BLOBS_DIR;
  }

  // Check for PORT environment variable
  if (process.env.PDS_PORT) {
    port = parseInt(process.env.PDS_PORT, 10);
  }

  // Check for HOSTNAME environment variable
  if (process.env.PDS_HOSTNAME) {
    hostname = process.env.PDS_HOSTNAME;
  }

  return { carFiles, blobsDir, port, hostname, dataDir };
}

function printHelp() {
  console.log(`
pds-readonly - Run a read-only PDS server from CAR files

Usage:
  pds-readonly [options] [car-files...]

Options:
  --car, -c <path>       Path to CAR file (can be specified multiple times)
  --blobs, -b <path>     Path to blobs directory (structure: blobs/{did}/{shard}/{cid})
  --port, -p <number>    Port to listen on (default: 3000)
  --hostname, -H <host>  PDS hostname for DID resolution (default: localhost)
  --data-dir, -d <path>  Directory for SQLite databases (default: ./pds-data)
  --help, -h             Show this help message
  --version, -v          Show version

Environment Variables:
  PDS_PORT               Port to listen on
  PDS_CAR_DIR            Directory containing CAR files
  PDS_BLOBS_DIR          Blobs directory path
  PDS_HOSTNAME           PDS hostname

Examples:
  # Load single repository
  pds-readonly --car ./repo.car --port 3000

  # Load multiple repositories
  pds-readonly --car ./alice.car --car ./bob.car

  # With blobs support
  pds-readonly --car ./repo.car --blobs ./blobs

  # Using environment variables
  PDS_CAR_DIR=./repos PDS_PORT=8080 pds-readonly
`);
}

async function main() {
  const { carFiles, blobsDir, port, hostname, dataDir } = parseArgs();

  if (carFiles.length === 0) {
    console.error('Error: At least one CAR file is required');
    console.error('Use --help for usage information');
    process.exit(1);
  }

  // Validate CAR files exist
  for (const carFile of carFiles) {
    if (!fs.existsSync(carFile)) {
      console.error(`Error: CAR file not found: ${carFile}`);
      process.exit(1);
    }
  }

  // Validate blobs directory if specified
  if (blobsDir && !fs.existsSync(blobsDir)) {
    console.warn(`Warning: Blobs directory not found: ${blobsDir}`);
    console.warn('Blob requests will return 404');
  }

  console.log('Starting read-only PDS server...');
  console.log(`Data directory: ${dataDir}`);
  console.log(`Hostname: ${hostname}`);
  if (blobsDir) {
    console.log(`Blobs directory: ${blobsDir}`);
  }

  // Create the read-only server
  const { server, loadCar } = createReadOnlyServer({
    hostname,
    dataDir,
    blobsDir: blobsDir || undefined,
    createDb: (dbPath) => new Database(dbPath),
  });

  // Load all CAR files
  console.log(`\nLoading ${carFiles.length} CAR file(s)...`);
  for (const carFile of carFiles) {
    try {
      console.log(`Loading: ${carFile}`);
      await loadCar(carFile);
    } catch (err) {
      console.error(`Error loading ${carFile}:`, err.message);
      process.exit(1);
    }
  }

  // Create HTTP server
  const httpServer = createServer(async (req, res) => {
    try {
      // Convert Node.js request to Web Request
      const url = new URL(req.url || '/', `http://${hostname}:${port}`);
      const headers = new Headers();
      for (const [key, value] of Object.entries(req.headers)) {
        if (value) {
          headers.set(key, Array.isArray(value) ? value.join(', ') : value);
        }
      }

      const body =
        req.method !== 'GET' && req.method !== 'HEAD'
          ? await readBody(req)
          : undefined;

      const request = new Request(url.toString(), {
        method: req.method,
        headers,
        body,
      });

      // Handle request
      const response = await server.fetch(request);

      // Convert Web Response to Node.js response
      res.statusCode = response.status;
      for (const [key, value] of response.headers) {
        res.setHeader(key, value);
      }

      if (response.body) {
        const reader = response.body.getReader();
        const pump = async () => {
          const { done, value } = await reader.read();
          if (done) {
            res.end();
            return;
          }
          res.write(value);
          await pump();
        };
        await pump();
      } else {
        const text = await response.text();
        res.end(text);
      }
    } catch (err) {
      console.error('Request error:', err);
      res.statusCode = 500;
      res.end(
        JSON.stringify({ error: 'InternalServerError', message: err.message }),
      );
    }
  });

  httpServer.listen(port, () => {
    console.log(`\nRead-only PDS server running at http://localhost:${port}`);
    console.log('\nAvailable endpoints:');
    console.log(`  GET  /xrpc/com.atproto.sync.listRepos`);
    console.log(`  GET  /xrpc/com.atproto.sync.getRepo?did=<did>`);
    console.log(
      `  GET  /xrpc/com.atproto.repo.listRecords?repo=<did>&collection=<nsid>`,
    );
    console.log(
      `  GET  /xrpc/com.atproto.repo.getRecord?repo=<did>&collection=<nsid>&rkey=<rkey>`,
    );
    console.log('\nWrite operations return 401 (AuthenticationRequired)');
  });

  // Handle shutdown
  process.on('SIGINT', () => {
    console.log('\nShutting down...');
    httpServer.close();
    process.exit(0);
  });

  process.on('SIGTERM', () => {
    console.log('\nShutting down...');
    httpServer.close();
    process.exit(0);
  });
}

/**
 * Read request body as Uint8Array
 * @param {import('node:http').IncomingMessage} req
 * @returns {Promise<Uint8Array<ArrayBuffer>>}
 */
function readBody(req) {
  return new Promise((resolve, reject) => {
    /** @type {Buffer[]} */
    const chunks = [];
    req.on('data', (chunk) => chunks.push(chunk));
    req.on('end', () => {
      const buffer = Buffer.concat(chunks);
      const arrayBuffer = /** @type {ArrayBuffer} */ (buffer.buffer);
      resolve(
        new Uint8Array(arrayBuffer, buffer.byteOffset, buffer.byteLength),
      );
    });
    req.on('error', reject);
  });
}

main().catch((err) => {
  console.error('Fatal error:', err);
  process.exit(1);
});