johnandersen777.bsky.social/knotless

Switch repository

Loading…

// Runs the reference PDS (@atproto/pds via @atproto/dev-env) and a PLC
// directory on free ports, creates test accounts, prints one JSON line
// describing them, and shuts down when stdin closes.
//
//   node harness/pds.mjs [--accounts alice,bob,carol] [--rate-limits] [--blob-limit <bytes>]
//
// KNOTLESS_DEV_ENV names a built dev-env package to use instead of the npm
// one, such as an atproto checkout's packages/dev-env on the spaces alpha.
// The alpha's PDS migrates an account's database every time it opens it,
// for the alpha alone as it says, which costs a read more than the read:
// here each database is migrated the first time it is opened only.
//
// A gateway is an OAuth client, which the PDS reaches only over https at a
// hostname. So the harness makes a certificate for GATEWAY_HOST, trusts
// it, and resolves that name to 127.0.0.1; tests put harness/tls-proxy.mjs
// in front of a gateway with it.

import { fileURLToPath, pathToFileURL } from 'node:url'
import { dirname, join, resolve } from 'node:path'
import { execFileSync } from 'node:child_process'
import { mkdtempSync, readdirSync, readFileSync, realpathSync } from 'node:fs'
import { createRequire } from 'node:module'
import { AsyncLocalStorage } from 'node:async_hooks'
import { tmpdir } from 'node:os'
import dns from 'node:dns'
import tls from 'node:tls'

// Relative to the repository root.
const root = resolve(dirname(fileURLToPath(import.meta.url)), '..')
// The PDS's better-sqlite3 is built for the Node .node-version names, as
// CI's is; on another major it fails to load, saying only that two ABI
// numbers differ.
const pinned = readFileSync(join(root, '.node-version'), 'utf8').trim()
if (process.versions.node.split('.')[0] !== pinned.split('.')[0]) {
  console.error(
    `harness/pds.mjs runs on Node ${pinned}, as .node-version says, not ${process.version}: put a Node ${pinned.split('.')[0]} first on PATH (vite-plus and nvm read .node-version)`,
  )
  process.exit(1)
}
const devEnv = process.env.KNOTLESS_DEV_ENV
const { TestNetworkNoAppView } = await import(
  devEnv ? pathToFileURL(resolve(root, devEnv, 'dist/index.js')).href : '@atproto/dev-env'
)
if (devEnv) await migrateOnce(resolve(root, devEnv))
if (!process.env.KNOTLESS_PDS_FSYNC) await skipFsync(devEnv ? resolve(root, devEnv) : undefined)

/// The directory of the PDS's compiled modules that the dev-env at `dir`
/// (the npm package when there is none) runs.
function pdsDist(dir) {
  const from = dir
    ? pathToFileURL(join(dir, 'package.json'))
    : import.meta.resolve('@atproto/dev-env')
  return dirname(realpathSync(createRequire(from).resolve('@atproto/pds')))
}

/// Has the PDS's SQLite databases (an account's, the account manager's, the
/// sequencer's) commit without waiting for the disk: the same modules it
/// loads, patched before it starts. Each commit otherwise waits on an fsync,
/// 20ms on the runner's disk, and a write makes several, which made a push
/// or a pull cost most of a second when the PDS does the rest in a few
/// milliseconds. A test harness has no data to keep through a crash.
/// KNOTLESS_PDS_FSYNC=1 keeps the PDS as it is.
async function skipFsync(dir) {
  const { Database } = await import(pathToFileURL(join(pdsDist(dir), 'db/db.js')).href)
  const sqlite = Database.sqlite
  Database.sqlite = (location, opts) =>
    sqlite.call(Database, location, { ...opts, pragmas: { ...opts?.pragmas, synchronous: 'OFF' } })
}

/// Has the PDS that the dev-env at `dir` runs migrate each account's
/// database on its first opening alone, rather than on every one: the
/// same modules it loads, patched before it starts.
async function migrateOnce(dir) {
  const index = createRequire(join(dir, 'package.json')).resolve('@atproto/pds')
  const dist = dirname(realpathSync(index))
  const load = (file) => import(pathToFileURL(join(dist, file)).href)
  const { ActorStore } = await load('actor-store/actor-store.js')
  const { Migrator } = await load('db/migrator.js')
  const opened = new Set()
  const again = new AsyncLocalStorage()
  const open = ActorStore.prototype.openDb
  ActorStore.prototype.openDb = async function (did) {
    const { dbLocation } = await this.getLocation(did)
    const db = await again.run(opened.has(dbLocation), () => open.call(this, did))
    opened.add(dbLocation)
    return db
  }
  const migrate = Migrator.prototype.migrateToLatestOrThrow
  Migrator.prototype.migrateToLatestOrThrow = function () {
    return again.getStore() ? Promise.resolve([]) : migrate.call(this)
  }
}

const GATEWAY_HOST = 'gateway.knotless.internal'
const certs = mkdtempSync(join(tmpdir(), 'knotless-tls-'))
const cert = join(certs, 'cert.pem')
const key = join(certs, 'key.pem')
execFileSync(
  'openssl',
  [
    'req',
    '-x509',
    '-newkey',
    'ec',
    '-pkeyopt',
    'ec_paramgen_curve:prime256v1',
    '-nodes',
    '-keyout',
    key,
    '-out',
    cert,
    '-days',
    '7',
    '-subj',
    `/CN=${GATEWAY_HOST}`,
    '-addext',
    `subjectAltName=DNS:${GATEWAY_HOST}`,
  ],
  { stdio: 'ignore' },
)
tls.setDefaultCACertificates([...tls.getCACertificates('default'), readFileSync(cert, 'utf8')])
const lookup = dns.lookup
dns.lookup = (hostname, options, callback) => {
  if (typeof options === 'function') [callback, options] = [options, {}]
  if (hostname !== GATEWAY_HOST) return lookup(hostname, options, callback)
  return options?.all
    ? callback(null, [{ address: '127.0.0.1', family: 4 }])
    : callback(null, '127.0.0.1', 4)
}

const arg = (name) => {
  const i = process.argv.indexOf(name)
  return i === -1 ? undefined : process.argv[i + 1]
}
const names = (arg('--accounts') ?? 'alice,bob,carol').split(',')

const net = await TestNetworkNoAppView.create({
  pds: {
    hostname: 'localhost',
    rateLimitsEnabled: process.argv.includes('--rate-limits'),
    ...(arg('--blob-limit') && { blobUploadLimit: Number(arg('--blob-limit')) }),
  },
})

const accounts = []
for (const name of names) {
  const password = `${name}-password`
  const res = await fetch(`${net.pds.url}/xrpc/com.atproto.server.createAccount`, {
    method: 'POST',
    headers: { 'content-type': 'application/json' },
    body: JSON.stringify({
      email: `${name}@test.com`,
      handle: `${name}.test`,
      password,
    }),
  })
  if (!res.ok) throw new Error(`createAccount ${name}: ${await res.text()}`)
  const { did, handle } = await res.json()
  accounts.push({ handle, did, password })
}

const described = await fetch(`${net.pds.url}/xrpc/com.atproto.server.describeServer`).then((res) =>
  res.json(),
)
if (described.spacesAlpha === true) await knowSpaces(net.pds.ctx.accountManager.db)

/// Puts knotless's space lexicons in the alpha PDS's store of the lexicons
/// it resolved, as resolved now and fresh for good. Its authorization page
/// resolves the lexicon of each space type a login's scope names, and
/// refuses the login if it cannot, through DNS and this PLC, which reach
/// none of knotless's published ones.
async function knowSpaces(db) {
  const dir = join(root, 'lexicons')
  const docs = readdirSync(dir, { recursive: true })
    .filter((file) => file.endsWith('.json'))
    .map((file) => JSON.parse(readFileSync(join(dir, file), 'utf8')))
    .filter((doc) => doc.defs?.main?.type === 'space')
  const now = new Date().toISOString()
  for (const doc of docs) {
    const row = {
      createdAt: now,
      updatedAt: '9999-01-01T00:00:00.000Z',
      lastSucceededAt: now,
      uri: `at://lexicons.knotless.dev/com.atproto.lexicon.schema/${doc.id}`,
      lexicon: JSON.stringify(doc),
    }
    await db.db
      .insertInto('lexicon')
      .values({ nsid: doc.id, ...row })
      .onConflict((oc) => oc.column('nsid').doUpdateSet(row))
      .execute()
  }
  if (docs.length < 2) throw new Error(`found ${docs.length} space lexicons under ${dir}`)
}

console.log(
  JSON.stringify({
    pds: net.pds.url,
    plc: net.plc.url,
    spaces: described.spacesAlpha === true,
    // Where the PDS keeps each account's store, for a test to spoil a
    // record in it as a server's fault might.
    actors: net.pds.ctx.cfg.actorStore.directory,
    gateway: { host: GATEWAY_HOST, cert, key },
    accounts,
  }),
)

const shutdown = async () => {
  await net.close()
  process.exit(0)
}
process.stdin.on('end', shutdown)
process.stdin.on('close', shutdown)
process.stdin.resume()
process.on('SIGTERM', shutdown)
process.on('SIGINT', shutdown)