diff --git a/.gitignore b/.gitignore index 12ecad4..8f1995b 100644 --- a/.gitignore +++ b/.gitignore @@ -1,8 +1,11 @@ node_modules/ dist/ data/ +data-node2/ +plans/ .env *.db *.db-journal *.db-wal *.db-shm +.claude/settings.local.json diff --git a/README.md b/README.md index 155920e..3fc6ce6 100644 --- a/README.md +++ b/README.md @@ -6,6 +6,7 @@ Peer-to-peer replication infrastructure for AT Protocol. Backs up and serves atp - Provides data on P2P networks (IPFS/libp2p) for other nodes to replicate - Fetches and stores data from P2P networks for serviced accounts - Mutual replication agreements between peers via on-protocol consent records +- Push-based offer discovery: nodes notify each other of replication offers P2PDS is infrastructure — like a torrent client for atproto data. It does not have its own identity. Users authenticate with their own atproto accounts, and records (`org.p2pds.peer`, `org.p2pds.replication.offer`) are published to the user's own repo via their PDS. @@ -14,9 +15,10 @@ P2PDS is infrastructure — like a torrent client for atproto data. It does not - **Runtime**: Node.js, TypeScript (ES2022, strict) - **Base**: Generalized from [Cirrus](https://github.com/ascorbic/cirrus) - **HTTP**: Hono -- **Database**: better-sqlite3 (sync API) -- **IPFS**: Helia (libp2p + DHT + bitswap + gossipsub) +- **Database**: better-sqlite3 (sync API) — all state in a single `pds.db` file +- **IPFS**: Helia with minimal libp2p (TCP + noise + yamux), SQLite-backed blockstore - **Identity**: AT Protocol DIDs via PLC directory +- **Auth**: OAuth (primary) or legacy JWT (fallback) - **Desktop**: Tauri v2 (optional, `apps/desktop/`) - **Content addressing**: [DASL](https://dasl.ing/) CIDs (CIDv1, SHA-256, dag-cbor/raw, base32lower) @@ -29,8 +31,8 @@ User's atproto account (any PDS: Bluesky, Cirrus, self-hosted) ┌─────────┐ │ p2pds │ ← replication infrastructure (local, cloud, or co-located) │ │ - │ SQLite │ block/blob tracking, sync state, challenge history - │ Helia │ IPFS storage, DHT announcements, bitswap, gossipsub + │ SQLite │ blocks, blobs, sync state, peer routing, challenge history + │ Helia │ IPFS storage, direct peer connections, bitswap │ Hono │ XRPC endpoints, admin dashboard, RASL └─────────┘ │ @@ -38,23 +40,49 @@ User's atproto account (any PDS: Bluesky, Cirrus, self-hosted) Other p2pds nodes (mutual replication via offer records) ``` -Configured with a list of DIDs to replicate: +### Storage -1. Resolves DIDs via PLC directory to find source PDSes -2. Fetches repos as CAR files from each DID's PDS (incremental via `since`) -3. Stores blocks in IPFS (Helia) and announces via DHT -4. Serves blocks via content-addressed RASL endpoint -5. Real-time sync via firehose (`com.atproto.sync.subscribeRepos`) -6. Gossipsub notifications for low-latency cross-node sync -7. Verifies block availability on remote peers via challenge-response protocol +All persistent state lives in a single SQLite database (`pds.db`): + +- **IPFS blocks** (`ipfs_blocks`) — replaces filesystem blockstore, avoids thousands of tiny files +- **IPFS datastore** (`ipfs_datastore`) — replaces filesystem datastore for libp2p peer/routing data +- **Replication state** — sync progress, peer info, block/blob tracking, firehose cursor +- **Challenge history** — proof-of-storage results and peer reliability scores +- **Incoming offers** — offers from other nodes awaiting accept/reject +- **Node identity** — DID + handle, established on first OAuth login + +### Identity model + +P2PDS starts without an identity. On first OAuth login, the user's DID becomes the node identity and is persisted in SQLite. Subsequent restarts load the identity from the database. This "lazy identity" model means: + +- No DID or signing key required in config +- Identity established interactively via the dashboard +- `RepoManager` is optional throughout (firehose, replication, startup all handle its absence) + +### Libp2p configuration + +Helia runs with a minimal libp2p stack: TCP transport, Noise encryption, Yamux multiplexing, and Identify only. No DHT, gossipsub, relay, autoNAT, UPnP, or WebRTC — those services peg CPU connecting to random peers. P2PDS dials known peers directly using multiaddrs from `org.p2pds.peer` records. + +### Replication flow + +1. User adds a DID via the dashboard → publishes an `org.p2pds.replication.offer` record +2. Node resolves the target's `org.p2pds.peer` record to find their p2pds endpoint +3. Node POSTs a notification to the target's `notifyOffer` endpoint +4. Target verifies the offer exists in the offerer's repo (anti-spoofing) +5. Target's dashboard shows the incoming offer with Accept/Reject buttons +6. Accepting creates a reciprocal offer + push notification back +7. Both nodes detect mutual agreement → promote to active replication +8. Sync loop: fetch repo, store blocks/blobs, verify, announce ### Design choices - **DHT only** for discovery/routing — no IPNI or centralized indexers - **Slow data is fine** as a tradeoff for resilience and decentralization - **Transport-agnostic verification** — RASL works over any HTTP transport -- **DASL-compliant content addressing** — all CIDs are CIDv1 + SHA-256 with either dag-cbor (`0x71`) or raw (`0x55`) codec, encoded as base32lower (`b` prefix). This is enforced at the library level by `@atcute/cid` and matches atproto's CID conventions. See [DASL CID spec](https://dasl.ing/cid.html). -- **No node identity** — p2pds acts on behalf of users, not as its own entity. Records publish to the user's own atproto repo. +- **DASL-compliant content addressing** — all CIDs are CIDv1 + SHA-256 with either dag-cbor (`0x71`) or raw (`0x55`) codec, encoded as base32lower (`b` prefix). Enforced by `@atcute/cid`. +- **No node identity** — p2pds acts on behalf of users, not as its own entity +- **SQLite everywhere** — single-file database, no filesystem blockstore churn +- **Consent-gated replication** — "Add" publishes an offer, not an immediate sync. Replication only begins when both sides agree. ### Deployment flexibility @@ -68,13 +96,15 @@ P2PDS defines two record types for the on-protocol interop surface: | NSID | Repo key | Purpose | |------|----------|---------| -| `org.p2pds.peer` | `self` | Binds an atproto DID to a libp2p PeerID + multiaddrs | +| `org.p2pds.peer` | `self` | Binds an atproto DID to a libp2p PeerID + multiaddrs + p2pds endpoint URL | | `org.p2pds.replication.offer` | `any` | Declares willingness to replicate a specific DID's data | Schemas are in `lexicons/` and validated by `src/lexicons.ts`. **Offer negotiation**: Peers publish offer records declaring willingness to replicate specific DIDs. When two peers have mutual offers (A offers to replicate B, B offers to replicate A), a replication agreement is automatically formed. Parameters are merged: `max(minCopies)`, `min(intervalSec)`, `max(priority)`. Revoking an offer = deleting the record. +**Push notifications**: When a node publishes an offer, it resolves the target's `org.p2pds.peer` record to find their p2pds HTTP endpoint and POSTs a notification. The receiving node verifies the offer exists in the sender's repo before storing it (prevents spoofing). + ## Verification Content-addressed retrieval is unforgeable: if a peer returns the correct bytes for a CID, they have the data. The verification stack exploits this property at multiple layers: @@ -88,24 +118,15 @@ Content-addressed retrieval is unforgeable: if a peer returns the correct bytes **Challenge-response protocol**: Three message types (`StorageChallenge` → `StorageChallengeResponse` → `StorageChallengeResult`). Deterministic generation from epoch + DIDs + nonce. Transport-agnostic with libp2p primary and HTTP fallback (`FailoverChallengeTransport`). Challenge history and peer reliability tracked in SQLite. -**L0** and **L1** run on a configurable timer (default 30 min). L1 samples are tuneable via `VerificationConfig.raslSampleSize` (default 50 blocks). - -### L2 (libp2p+HTTP Gateway) — future - -Reuses RASL verification logic over libp2p transports for NAT traversal and encryption without public IP. Requires [libp2p+HTTP Gateway spec](https://specs.ipfs.tech/http-gateways/libp2p-gateway/) in Helia. - -- Kubo (Go): [ipfs/kubo#10049](https://github.com/ipfs/kubo/issues/10049) (shipped) -- Helia (JS): [ipfs/helia#348](https://github.com/ipfs/helia/issues/348) (not yet) - ## Replication Sync loop (per DID, periodic with policy-driven intervals): 1. Resolve DID → PDS endpoint (via PLC directory) -2. Fetch repo (`com.atproto.sync.getRepo`, incremental via `since`) -3. Parse CAR, store blocks in IPFS, fetch and store blobs -4. Track block/blob CIDs, populate record paths via MST walk -5. Announce to DHT +2. Try libp2p peer-first sync if peer info is known +3. Fall back to HTTP PDS fetch (`com.atproto.sync.getRepo`, incremental via `since`) +4. Parse CAR, store blocks in IPFS, fetch and store blobs +5. Track block/blob CIDs, populate record paths via MST walk 6. Verify local block availability 7. If source PDS fails, fall back to peer endpoints @@ -125,10 +146,11 @@ Policies drive sync intervals, priority ordering, and `shouldReplicate` filterin ## App -- **Dashboard**: Server-rendered HTML at `/` (auto-refresh) +- **Dashboard**: Server-rendered HTML at `/` with auto-refresh, account search, incoming offer notifications - **API**: Authenticated XRPC endpoints for overview, per-DID status, network status, policies, sync history -- **DID management**: Add/remove DIDs at runtime via `addDid`/`removeDid` endpoints -- **Rate limiting**: Per-IP and per-DID limits across HTTP, gossipsub, and libp2p +- **DID management**: Add/remove/offer DIDs at runtime via dashboard or API +- **Incoming offers**: Accept/reject replication offers from other nodes via dashboard +- **Rate limiting**: Per-IP limits across all endpoint groups (meta, sync, session, read, write, challenge, app, notifyOffer) ## Desktop App @@ -149,23 +171,42 @@ npm test npm run dev ``` +### Two-node manual testing + +Scripts for running two p2pds nodes locally for manual testing: + +```bash +npm run start:both # Build and start both nodes on random ports +npm run start:node1 # Start node 1 only +npm run start:node2 # Start node 2 only (uses data-node2/.env) +npm run stop # Stop both nodes +npm run clean # Wipe data for both nodes (keeps .env files) +npm run logs # Show recent logs for both nodes +npm run test:add-did # Offer a DID on a running node +``` + +Node 2 requires a `data-node2/.env` file with a separate `OAUTH_ENABLED=true` and `DATA_DIR=./data-node2` config. Both nodes pick random ports and write them to `/tmp/p2pds-node{1,2}.port`. + ### Project structure ``` src/ index.ts Hono app with all routes server.ts HTTP server entry point + start.ts Server startup orchestrator config.ts Config interface + loadConfig() - validation.ts Record validator (atproto + p2pds lexicons) - lexicons.ts p2pds lexicon loader + validator - ipfs.ts IpfsService (Helia wrapper) + ipfs.ts IpfsService (Helia wrapper, SQLite-backed) + sqlite-blockstore.ts SQLite blockstore for Helia (replaces FsBlockstore) + sqlite-datastore.ts SQLite datastore for libp2p (replaces FsDatastore) repo-manager.ts Local repo management storage.ts SQLite block storage blobs.ts Blob storage - middleware/auth.ts Auth middleware + middleware/auth.ts Auth middleware (OAuth + legacy JWT) + oauth/ OAuth client, routes, session/state stores, PdsClient replication/ Sync, verification, challenges, offers, gossipsub policy/ Policy engine types, engine, presets xrpc/ XRPC endpoint handlers +scripts/ Two-node testing scripts lexicons/ Lexicon JSON schemas apps/desktop/ Tauri desktop app ``` @@ -174,24 +215,31 @@ apps/desktop/ Tauri desktop app Environment variables (or `.env` file): +| Variable | Required | Default | Description | +|----------|----------|---------|-------------| +| `OAUTH_ENABLED` | No | `false` | Enable OAuth login (recommended) | +| `PUBLIC_URL` | No | `http://localhost:$PORT` | Public URL for push notifications between nodes | +| `DATA_DIR` | No | `./data` | Data directory | +| `PORT` | No | `3000` | HTTP port | +| `IPFS_ENABLED` | No | `true` | Enable IPFS | +| `IPFS_NETWORKING` | No | `true` | Enable libp2p networking | +| `REPLICATE_DIDS` | No | | Comma-separated DIDs to replicate | +| `FIREHOSE_URL` | No | `wss://bsky.network/...` | Firehose WebSocket URL | +| `FIREHOSE_ENABLED` | No | `true` | Enable firehose sync | +| `POLICY_FILE` | No | | Path to policy JSON file | +| `RATE_LIMIT_ENABLED` | No | `true` | Enable rate limiting | + +**Legacy auth** (when `OAUTH_ENABLED=false`): + | Variable | Required | Description | |----------|----------|-------------| -| `DID` | Yes | Your atproto DID (e.g., `did:plc:...`) | -| `HANDLE` | Yes | Your handle (e.g., `user.example.com`) | +| `DID` | Yes | Your atproto DID | +| `HANDLE` | Yes | Your handle | | `PDS_HOSTNAME` | Yes | PDS hostname | | `AUTH_TOKEN` | Yes | Static auth token | -| `SIGNING_KEY` | Yes | Hex-encoded secp256k1 private key | -| `SIGNING_KEY_PUBLIC` | No | Multibase-encoded public key | | `JWT_SECRET` | Yes | JWT signing secret | | `PASSWORD_HASH` | Yes | Bcrypt password hash | -| `DATA_DIR` | No | Data directory (default: `./data`) | -| `PORT` | No | HTTP port (default: `3000`) | -| `IPFS_ENABLED` | No | Enable IPFS (default: `true`) | -| `IPFS_NETWORKING` | No | Enable IPFS networking (default: `true`) | -| `REPLICATE_DIDS` | No | Comma-separated DIDs to replicate | -| `FIREHOSE_URL` | No | Firehose WebSocket URL | -| `FIREHOSE_ENABLED` | No | Enable firehose sync (default: `false`) | -| `POLICY_FILE` | No | Path to policy JSON file | +| `SIGNING_KEY` | No | Hex-encoded secp256k1 private key | ## Status @@ -202,8 +250,11 @@ Environment variables (or `.env` file): 5. Challenge-response proof-of-storage — done 6. Policy engine — done 7. P2P offer negotiation — done -8. Admin dashboard + DID management — done -9. Rate limiting — done -10. Architecture refactor (user-DID model) — done -11. Lexicon definitions — done -12. Desktop app skeleton — done +8. Consent-gated replication — done +9. Incoming offer discovery via push notification — done +10. Admin dashboard + DID management — done +11. Rate limiting — done +12. Architecture refactor (user-DID model, lazy identity) — done +13. SQLite-backed IPFS storage — done +14. Lexicon definitions — done +15. Desktop app skeleton — done diff --git a/package.json b/package.json index 01bb381..75bbf68 100644 --- a/package.json +++ b/package.json @@ -10,7 +10,15 @@ "start": "node dist/server.js", "typecheck": "tsc --noEmit", "test": "vitest run", - "smoke-test": "bash scripts/smoke-test.sh" + "smoke-test": "bash scripts/smoke-test.sh", + "start:node1": "bash scripts/start-node.sh 1", + "start:node2": "bash scripts/start-node.sh 2", + "start:both": "bash scripts/start-both.sh", + "stop": "bash scripts/stop-both.sh", + "clean": "bash scripts/clean.sh", + "logs": "bash scripts/logs.sh", + "test:add-did": "bash scripts/test-add-did.sh", + "serve": "tsc && node dist/server.js > /tmp/p2pds-node1.log 2>&1 & echo $! > /tmp/p2pds-node1.pid && sleep 2 && cat /tmp/p2pds-node1.log" }, "dependencies": { "@atcute/atproto": "^3.1.10", diff --git a/scripts/clean.sh b/scripts/clean.sh new file mode 100755 index 0000000..fb0c850 --- /dev/null +++ b/scripts/clean.sh @@ -0,0 +1,15 @@ +#!/bin/bash +# Wipe all data for both nodes (DB, blobs, IPFS blockstore) +# Does NOT touch .env files +set -e + +# Stop any running nodes first +bash scripts/stop-both.sh 2>/dev/null || true + +echo "Cleaning node 1 data (data/)..." +rm -rf data/pds.db data/pds.db-shm data/pds.db-wal data/blobs data/ipfs + +echo "Cleaning node 2 data (data-node2/)..." +rm -rf data-node2/pds.db data-node2/pds.db-shm data-node2/pds.db-wal data-node2/blobs data-node2/ipfs + +echo "Done — both nodes clean" diff --git a/scripts/logs.sh b/scripts/logs.sh new file mode 100755 index 0000000..fca5b2c --- /dev/null +++ b/scripts/logs.sh @@ -0,0 +1,16 @@ +#!/bin/bash +# Show recent logs for p2pds nodes +NODE=${1:-0} +if [ "$NODE" = "1" ] || [ "$NODE" = "0" ]; then + if [ -f /tmp/p2pds-node1.log ]; then + echo "=== NODE 1 ===" + tail -30 /tmp/p2pds-node1.log + echo "" + fi +fi +if [ "$NODE" = "2" ] || [ "$NODE" = "0" ]; then + if [ -f /tmp/p2pds-node2.log ]; then + echo "=== NODE 2 ===" + tail -30 /tmp/p2pds-node2.log + fi +fi diff --git a/scripts/start-both.sh b/scripts/start-both.sh new file mode 100755 index 0000000..51c63b8 --- /dev/null +++ b/scripts/start-both.sh @@ -0,0 +1,51 @@ +#!/bin/bash +# Start both p2pds nodes for two-node testing +set -e + +# Stop previous instances if pid files exist +bash scripts/stop-both.sh 2>/dev/null || true + +# Build first +npm run build + +# Start node 1 (random port) +PORT1=$(python3 -c 'import socket; s=socket.socket(); s.bind(("",0)); print(s.getsockname()[1]); s.close()') +export PORT=$PORT1 +node dist/server.js > /tmp/p2pds-node1.log 2>&1 & +echo $! > /tmp/p2pds-node1.pid +echo "$PORT1" > /tmp/p2pds-node1.port +echo "Node 1 started (pid $!, port $PORT1, log: /tmp/p2pds-node1.log)" + +# Start node 2 (random port, different env) +PORT2=$(python3 -c 'import socket; s=socket.socket(); s.bind(("",0)); print(s.getsockname()[1]); s.close()') +set -a && source data-node2/.env && set +a +export PORT=$PORT2 +node dist/server.js > /tmp/p2pds-node2.log 2>&1 & +echo $! > /tmp/p2pds-node2.pid +echo "$PORT2" > /tmp/p2pds-node2.port +echo "Node 2 started (pid $!, port $PORT2, log: /tmp/p2pds-node2.log)" + +# Wait for both to be ready +echo "" +echo "Waiting for nodes..." +for node in 1 2; do + for i in $(seq 1 30); do + if grep -q "P2PDS_READY" "/tmp/p2pds-node${node}.log" 2>/dev/null; then + echo "Node $node ready (port $(cat /tmp/p2pds-node${node}.port))" + break + fi + if grep -q "Error:" "/tmp/p2pds-node${node}.log" 2>/dev/null; then + echo "Node $node FAILED:" + cat "/tmp/p2pds-node${node}.log" + break + fi + sleep 1 + done +done + +echo "" +echo "=== NODE 1 (port $PORT1) ===" +tail -10 /tmp/p2pds-node1.log +echo "" +echo "=== NODE 2 (port $PORT2) ===" +tail -10 /tmp/p2pds-node2.log diff --git a/scripts/start-node.sh b/scripts/start-node.sh new file mode 100755 index 0000000..c6c3605 --- /dev/null +++ b/scripts/start-node.sh @@ -0,0 +1,53 @@ +#!/bin/bash +# Start a single p2pds node: scripts/start-node.sh [1|2] +set -e +NODE=${1:-1} +PIDFILE="/tmp/p2pds-node${NODE}.pid" +PORTFILE="/tmp/p2pds-node${NODE}.port" +LOGFILE="/tmp/p2pds-node${NODE}.log" + +# Stop if already running +if [ -f "$PIDFILE" ]; then + kill "$(cat "$PIDFILE")" 2>/dev/null || true + rm -f "$PIDFILE" "$PORTFILE" + sleep 1 +fi + +npm run build + +# Pick a random high port +PORT=$(python3 -c 'import socket; s=socket.socket(); s.bind(("",0)); print(s.getsockname()[1]); s.close()') +export PORT + +if [ "$NODE" = "2" ]; then + set -a && source data-node2/.env && set +a + export PORT # override .env PORT with our random one +fi + +> "$LOGFILE" +node dist/server.js > "$LOGFILE" 2>&1 & +echo $! > "$PIDFILE" +echo "$PORT" > "$PORTFILE" + +# Wait for P2PDS_READY or failure +for i in $(seq 1 30); do + if grep -q "P2PDS_READY" "$LOGFILE" 2>/dev/null; then + echo "" + cat "$LOGFILE" + echo "" + echo "PID: $(cat "$PIDFILE") PORT: $PORT" + exit 0 + fi + if grep -q "Error:" "$LOGFILE" 2>/dev/null; then + echo "Node $NODE failed to start:" + cat "$LOGFILE" + rm -f "$PIDFILE" "$PORTFILE" + exit 1 + fi + sleep 1 +done + +echo "Node $NODE startup timed out. Log:" +cat "$LOGFILE" +rm -f "$PIDFILE" "$PORTFILE" +exit 1 diff --git a/scripts/stop-both.sh b/scripts/stop-both.sh new file mode 100755 index 0000000..655241c --- /dev/null +++ b/scripts/stop-both.sh @@ -0,0 +1,20 @@ +#!/bin/bash +# Stop both p2pds nodes using pid files, with fallback to pgrep +for node in node1 node2; do + pidfile="/tmp/p2pds-${node}.pid" + portfile="/tmp/p2pds-${node}.port" + if [ -f "$pidfile" ]; then + pid=$(cat "$pidfile") + kill "$pid" 2>/dev/null && echo "Stopped $node (pid $pid)" + rm -f "$pidfile" "$portfile" + fi +done + +# Fallback: kill any remaining p2pds server processes (compiled and tsx) +for pattern in "node dist/server" "tsx.*src/server" "tsx watch"; do + for pid in $(pgrep -f "$pattern" 2>/dev/null); do + kill "$pid" 2>/dev/null && echo "Stopped orphaned process $pid ($pattern)" + done +done + +sleep 2 diff --git a/scripts/test-add-did.sh b/scripts/test-add-did.sh new file mode 100755 index 0000000..c29db2d --- /dev/null +++ b/scripts/test-add-did.sh @@ -0,0 +1,56 @@ +#!/bin/bash +# Test adding a DID to a running p2pds node +# Usage: scripts/test-add-did.sh [port] +# If port is omitted, reads from /tmp/p2pds-node1.port + +if [ $# -eq 2 ]; then + PORT=$1; DID=$2 +elif [ $# -eq 1 ]; then + if [ -f /tmp/p2pds-node1.port ]; then + PORT=$(cat /tmp/p2pds-node1.port) + else + echo "Usage: test-add-did.sh [port] " + echo "No port given and /tmp/p2pds-node1.port not found. Start a node first." + exit 1 + fi + DID=$1 +else + echo "Usage: test-add-did.sh [port] " + exit 1 +fi + +# Extract auth token from dashboard HTML +HTML=$(curl -s "http://localhost:${PORT}/") +TOKEN=$(echo "$HTML" | grep 'const TOKEN' | sed 's/.*const TOKEN = "//;s/".*//') +if [ -z "$TOKEN" ]; then + echo "ERROR: Could not extract auth token from http://localhost:${PORT}/" + exit 1 +fi +echo "Auth token: ${TOKEN:0:10}..." + +echo "" +echo "Offering DID: $DID" +RESPONSE=$(curl -s -X POST "http://localhost:${PORT}/xrpc/org.p2pds.app.offerDid" \ + -H "Authorization: Bearer $TOKEN" \ + -H "Content-Type: application/json" \ + -d "{\"did\":\"$DID\"}") +echo "Response: $RESPONSE" + +echo "" +echo "Checking overview..." +curl -s "http://localhost:${PORT}/xrpc/org.p2pds.app.getOverview" \ + -H "Authorization: Bearer $TOKEN" | python3 -c " +import json, sys +d = json.load(sys.stdin) +print('DID:', d.get('did', '(none)')) +r = d.get('replication', {}) +print('Tracked DIDs:', r.get('trackedDids', [])) +states = r.get('syncStates', []) +for s in states: + print(f\" {s['did']}: status={s.get('status','?')}\") +offered = d.get('offeredDids', []) +if offered: + print('Offered DIDs:') + for o in offered: + print(f\" {o['did']}: offered at {o.get('offeredAt','?')}\") +" diff --git a/src/ipfs.ts b/src/ipfs.ts index 4e28626..fabe098 100644 --- a/src/ipfs.ts +++ b/src/ipfs.ts @@ -1,9 +1,10 @@ import { CID } from "multiformats"; -import { FsBlockstore } from "blockstore-fs"; -import { FsDatastore } from "datastore-fs"; import type { Helia } from "@helia/interface"; import type { BlockMap } from "@atproto/repo"; +import type Database from "better-sqlite3"; import { encode as cborEncode, decode as cborDecode } from "./cbor-compat.js"; +import { SqliteBlockstore } from "./sqlite-blockstore.js"; +import { SqliteDatastore } from "./sqlite-datastore.js"; import type { RateLimiter } from "./rate-limiter.js"; import { DEFAULT_RATE_LIMIT_CONFIG } from "./rate-limiter.js"; @@ -86,8 +87,7 @@ export interface NetworkService { } export interface IpfsConfig { - blocksPath: string; - datastorePath: string; + db: Database.Database; networking: boolean; } @@ -103,7 +103,7 @@ export interface IpfsConfig { */ export class IpfsService implements BlockStore, NetworkService { private helia: Helia | null = null; - private blockstore: FsBlockstore | null = null; + private blockstore: SqliteBlockstore | null = null; private config: IpfsConfig; private running = false; private commitHandlers: CommitNotificationHandler[] = []; @@ -124,35 +124,40 @@ export class IpfsService implements BlockStore, NetworkService { } async start(): Promise { - this.blockstore = new FsBlockstore(this.config.blocksPath); + this.blockstore = new SqliteBlockstore(this.config.db); if (this.config.networking) { - const { createHelia, libp2pDefaults } = await import("helia"); - const { gossipsub } = await import("@libp2p/gossipsub"); - const datastore = new FsDatastore(this.config.datastorePath); - - const libp2pConfig = libp2pDefaults(); - libp2pConfig.services.pubsub = gossipsub({ - emitSelf: false, - allowPublishToZeroTopicPeers: true, - maxInboundDataLength: MAX_GOSSIPSUB_MESSAGE_SIZE, - }); - // Connection manager limits - libp2pConfig.connectionManager = { - ...libp2pConfig.connectionManager, - maxConnections: 100, - maxIncomingPendingConnections: 10, - inboundConnectionThreshold: 5, - }; - + const { createHelia } = await import("helia"); + const { tcp } = await import("@libp2p/tcp"); + const { noise } = await import("@chainsafe/libp2p-noise"); + const { yamux } = await import("@chainsafe/libp2p-yamux"); + const { identify } = await import("@libp2p/identify"); + const datastore = new SqliteDatastore(this.config.db); + + // Minimal libp2p config: only what we need for direct peer connections. + // No DHT, gossipsub, relay, autoNAT, UPnP, WebRTC — those peg CPU + // connecting to random peers. We dial p2pds peers directly. + // Cast SQLite stores to Helia's expected interfaces. + // Our implementations are duck-type compatible but use Promise + // returns instead of AwaitGenerator, which Helia handles fine at runtime. this.helia = await createHelia({ - libp2p: libp2pConfig, - blockstore: this.blockstore, - datastore, + libp2p: { + addresses: { listen: ["/ip4/0.0.0.0/tcp/0"] }, + transports: [tcp()], + connectionEncrypters: [noise()], + streamMuxers: [yamux()], + services: { + identify: identify(), + }, + connectionManager: { + maxConnections: 20, + maxIncomingPendingConnections: 5, + inboundConnectionThreshold: 3, + }, + }, + blockstore: this.blockstore as any, + datastore: datastore as any, }); - this.setupGossipsubHandler(); - } else { - await this.blockstore.open(); } this.running = true; @@ -162,8 +167,6 @@ export class IpfsService implements BlockStore, NetworkService { if (this.helia) { await this.helia.stop(); this.helia = null; - } else if (this.blockstore) { - await this.blockstore.close(); } this.blockstore = null; this.running = false; @@ -195,7 +198,7 @@ export class IpfsService implements BlockStore, NetworkService { if (!this.blockstore) return null; try { const cid = CID.parse(cidStr); - // get() returns AsyncGenerator; collect chunks + // Both Helia's blockstore and SqliteBlockstore return AsyncGenerator const gen = this.helia ? this.helia.blockstore.get(cid, { offline: true }) : this.blockstore.get(cid); @@ -222,10 +225,10 @@ export class IpfsService implements BlockStore, NetworkService { if (!this.blockstore) return false; try { const cid = CID.parse(cidStr); - const store = this.helia - ? this.helia.blockstore - : this.blockstore; - return await store.has(cid); + if (this.helia) { + return await this.helia.blockstore.has(cid); + } + return await this.blockstore.has(cid); } catch { return false; } diff --git a/src/middleware/auth.ts b/src/middleware/auth.ts index b7dd282..3f05d7e 100644 --- a/src/middleware/auth.ts +++ b/src/middleware/auth.ts @@ -47,6 +47,11 @@ export async function requireAuth( return next(); } + // Legacy JWT auth requires PDS_HOSTNAME and JWT_SECRET + if (!c.env.PDS_HOSTNAME || !c.env.JWT_SECRET) { + return c.json({ error: "AuthenticationRequired", message: "Not authenticated" }, 401); + } + const serviceDid = `did:web:${c.env.PDS_HOSTNAME}`; // Try session JWT verification (HS256, signed with JWT_SECRET) diff --git a/src/server.ts b/src/server.ts index b723d17..bc2e923 100644 --- a/src/server.ts +++ b/src/server.ts @@ -1,12 +1,14 @@ import { loadConfig } from "./config.js"; -import { startServer } from "./start.js"; +import { startServer, type ServerHandle } from "./start.js"; const config = loadConfig(); -const handle = await startServer(config); -process.on("SIGINT", () => { - handle.close().finally(() => process.exit(0)); -}); -process.on("SIGTERM", () => { - handle.close().finally(() => process.exit(0)); -}); +// Register signal handlers BEFORE await so SIGTERM during startup exits cleanly +let handle: ServerHandle | undefined; +const shutdown = () => { + (handle?.close() ?? Promise.resolve()).finally(() => process.exit(0)); +}; +process.on("SIGINT", shutdown); +process.on("SIGTERM", shutdown); + +handle = await startServer(config); diff --git a/src/sqlite-blockstore.ts b/src/sqlite-blockstore.ts new file mode 100644 index 0000000..26f373f --- /dev/null +++ b/src/sqlite-blockstore.ts @@ -0,0 +1,86 @@ +/** + * SQLite-backed Blockstore for Helia. + * + * Replaces FsBlockstore to avoid creating thousands of tiny files + * that hammer macOS fseventsd. All blocks stored in a single SQLite table. + */ + +import type Database from "better-sqlite3"; +import { CID } from "multiformats"; + +export class SqliteBlockstore { + private db: Database.Database; + + constructor(db: Database.Database) { + this.db = db; + this.db.exec(` + CREATE TABLE IF NOT EXISTS ipfs_blocks ( + cid TEXT PRIMARY KEY, + bytes BLOB NOT NULL + ) + `); + } + + async put(key: CID, val: Uint8Array): Promise { + this.db + .prepare("INSERT OR REPLACE INTO ipfs_blocks (cid, bytes) VALUES (?, ?)") + .run(key.toString(), Buffer.from(val)); + return key; + } + + async * get(key: CID): AsyncGenerator { + const row = this.db + .prepare("SELECT bytes FROM ipfs_blocks WHERE cid = ?") + .get(key.toString()) as { bytes: Buffer } | undefined; + if (!row) { + throw new Error(`Block not found: ${key.toString()}`); + } + yield new Uint8Array(row.bytes); + } + + async has(key: CID): Promise { + const row = this.db + .prepare("SELECT 1 FROM ipfs_blocks WHERE cid = ?") + .get(key.toString()); + return row !== undefined; + } + + async delete(key: CID): Promise { + this.db + .prepare("DELETE FROM ipfs_blocks WHERE cid = ?") + .run(key.toString()); + } + + async * putMany(source: AsyncIterable<{ cid: CID; block: Uint8Array }> | Iterable<{ cid: CID; block: Uint8Array }>): AsyncGenerator { + for await (const { cid, block } of source) { + await this.put(cid, block); + yield cid; + } + } + + async * getMany(source: AsyncIterable | Iterable): AsyncGenerator<{ cid: CID; block: Uint8Array }> { + for await (const cid of source) { + let block: Uint8Array = new Uint8Array(0); + for await (const chunk of this.get(cid)) { + block = chunk; + } + yield { cid, block }; + } + } + + async * deleteMany(source: AsyncIterable | Iterable): AsyncGenerator { + for await (const cid of source) { + await this.delete(cid); + yield cid; + } + } + + async * getAll(): AsyncGenerator<{ cid: CID; block: Uint8Array }> { + const rows = this.db + .prepare("SELECT cid, bytes FROM ipfs_blocks") + .all() as Array<{ cid: string; bytes: Buffer }>; + for (const row of rows) { + yield { cid: CID.parse(row.cid), block: new Uint8Array(row.bytes) }; + } + } +} diff --git a/src/sqlite-datastore.ts b/src/sqlite-datastore.ts new file mode 100644 index 0000000..5eac5d8 --- /dev/null +++ b/src/sqlite-datastore.ts @@ -0,0 +1,128 @@ +/** + * SQLite-backed Datastore for libp2p. + * + * Replaces FsDatastore to avoid filesystem churn. + * Stores libp2p peer/routing data in a single SQLite table. + */ + +import type Database from "better-sqlite3"; +import { Key } from "interface-datastore"; +import type { Pair, Query, KeyQuery, Batch } from "interface-datastore"; +type DatastorePair = Pair; + +export class SqliteDatastore { + private db: Database.Database; + + constructor(db: Database.Database) { + this.db = db; + this.db.exec(` + CREATE TABLE IF NOT EXISTS ipfs_datastore ( + key TEXT PRIMARY KEY, + value BLOB NOT NULL + ) + `); + } + + async put(key: Key, val: Uint8Array): Promise { + this.db + .prepare("INSERT OR REPLACE INTO ipfs_datastore (key, value) VALUES (?, ?)") + .run(key.toString(), Buffer.from(val)); + return key; + } + + async get(key: Key): Promise { + const row = this.db + .prepare("SELECT value FROM ipfs_datastore WHERE key = ?") + .get(key.toString()) as { value: Buffer } | undefined; + if (!row) { + throw new Error(`Key not found: ${key.toString()}`); + } + return new Uint8Array(row.value); + } + + async has(key: Key): Promise { + const row = this.db + .prepare("SELECT 1 FROM ipfs_datastore WHERE key = ?") + .get(key.toString()); + return row !== undefined; + } + + async delete(key: Key): Promise { + this.db + .prepare("DELETE FROM ipfs_datastore WHERE key = ?") + .run(key.toString()); + } + + async * putMany(source: AsyncIterable | Iterable): AsyncGenerator { + for await (const { key, value } of source) { + await this.put(key, value); + yield key; + } + } + + async * getMany(source: AsyncIterable | Iterable): AsyncGenerator { + for await (const key of source) { + const value = await this.get(key); + yield { key, value }; + } + } + + async * deleteMany(source: AsyncIterable | Iterable): AsyncGenerator { + for await (const key of source) { + await this.delete(key); + yield key; + } + } + + query(q: Query): AsyncIterable { + const self = this; + return (async function* () { + const prefix = q.prefix ?? "/"; + const rows = self.db + .prepare("SELECT key, value FROM ipfs_datastore WHERE key LIKE ?") + .all(prefix + "%") as Array<{ key: string; value: Buffer }>; + for (const row of rows) { + yield { key: new Key(row.key), value: new Uint8Array(row.value) }; + } + })(); + } + + queryKeys(q: KeyQuery): AsyncIterable { + const self = this; + return (async function* () { + const prefix = q.prefix ?? "/"; + const rows = self.db + .prepare("SELECT key FROM ipfs_datastore WHERE key LIKE ?") + .all(prefix + "%") as Array<{ key: string }>; + for (const row of rows) { + yield new Key(row.key); + } + })(); + } + + batch(): Batch { + const ops: Array<{ type: "put"; key: Key; value: Uint8Array } | { type: "delete"; key: Key }> = []; + return { + put: (key: Key, value: Uint8Array) => { + ops.push({ type: "put", key, value }); + }, + delete: (key: Key) => { + ops.push({ type: "delete", key }); + }, + commit: async () => { + const insertStmt = this.db.prepare("INSERT OR REPLACE INTO ipfs_datastore (key, value) VALUES (?, ?)"); + const deleteStmt = this.db.prepare("DELETE FROM ipfs_datastore WHERE key = ?"); + const tx = this.db.transaction(() => { + for (const op of ops) { + if (op.type === "put") { + insertStmt.run(op.key.toString(), Buffer.from(op.value)); + } else { + deleteStmt.run(op.key.toString()); + } + } + }); + tx(); + }, + }; + } +} diff --git a/src/xrpc/server.ts b/src/xrpc/server.ts index 83c43f3..c51864b 100644 --- a/src/xrpc/server.ts +++ b/src/xrpc/server.ts @@ -23,6 +23,10 @@ export async function createSession( c: Context, repoManager: RepoManager, ): Promise { + if (!c.env.PASSWORD_HASH || !c.env.JWT_SECRET || !c.env.PDS_HOSTNAME) { + return c.json({ error: "NotSupported", message: "Legacy auth not configured — use OAuth" }, 501); + } + const body = await c.req.json<{ identifier: string; password: string; @@ -89,6 +93,10 @@ export async function refreshSession( c: Context, repoManager: RepoManager, ): Promise { + if (!c.env.JWT_SECRET || !c.env.PDS_HOSTNAME) { + return c.json({ error: "NotSupported", message: "Legacy auth not configured — use OAuth" }, 501); + } + const authHeader = c.req.header("Authorization"); if (!authHeader?.startsWith("Bearer ")) { @@ -163,6 +171,10 @@ export async function getSession( c: Context, repoManager: RepoManager, ): Promise { + if (!c.env.JWT_SECRET || !c.env.PDS_HOSTNAME) { + return c.json({ error: "NotSupported", message: "Legacy auth not configured — use OAuth" }, 501); + } + const authHeader = c.req.header("Authorization"); if (!authHeader?.startsWith("Bearer ")) {