From 18923d29dbdad9d7a2889f03c02f359095dd84f8 Mon Sep 17 00:00:00 2001 From: wheval Date: Sat, 25 Jul 2026 00:04:19 +0100 Subject: [PATCH 1/3] docs: add developer onboarding guide and API endpoint reference - Add local setup guide with prerequisites, step-by-step instructions, and troubleshooting - Add API endpoint reference with all public endpoints, auth requirements, and sample responses - Cross-link both documents for easy navigation Closes accesslayerorg/accesslayer-server#598 --- docs/api-endpoints.md | 276 ++++++++++++++++++++++++++++++++++++++++++ docs/local-setup.md | 134 ++++++++++++++++++++ 2 files changed, 410 insertions(+) create mode 100644 docs/api-endpoints.md create mode 100644 docs/local-setup.md diff --git a/docs/api-endpoints.md b/docs/api-endpoints.md new file mode 100644 index 0000000..3fe207f --- /dev/null +++ b/docs/api-endpoints.md @@ -0,0 +1,276 @@ +# API Endpoint Reference + +Base URL: `http://localhost:3000/api/v1` + +## Health Endpoints + +### GET /health + +Simple health check for load balancers. + +- **Auth:** None +- **Response:** `200 OK` + +```json +{ + "success": true, + "message": "OK", + "timestamp": "2025-01-15T10:30:00.000Z" +} +``` + +### GET /health/ready + +Readiness check with dependency probes. + +- **Auth:** None +- **Response:** `200 OK` or `503 Service Unavailable` + +```json +{ + "ready": true, + "timestamp": "2025-01-15T10:30:00.000Z", + "checks": [ + { "name": "database", "status": "ok", "latencyMs": 12 }, + { "name": "cache", "status": "ok" } + ] +} +``` + +### GET /health/detailed + +Full diagnostics including memory and system info. + +- **Auth:** None +- **Response:** `200 OK` + +```json +{ + "success": true, + "message": "Access Layer server is running", + "timestamp": "2025-01-15T10:30:00.000Z", + "version": "1.0.0", + "environment": "development", + "uptime": 12345.67, + "memory": { "used": 45.23, "total": 128.5 }, + "system": { "platform": "darwin", "nodeVersion": "v20.10.0" }, + "database": { "status": "connected", "responseTime": 12 }, + "services": [ + { "name": "API Server", "status": "healthy" }, + { "name": "Database", "status": "healthy" } + ] +} +``` + +--- + +## Auth Endpoints + +### POST /auth/login + +Authenticate a user. + +- **Auth:** None +- **Body:** + +```json +{ + "email": "user@example.com", + "password": "securepassword" +} +``` + +- **Response:** `200 OK` + +### POST /auth/register + +Register a new user. + +- **Auth:** None +- **Body:** + +```json +{ + "email": "user@example.com", + "password": "securepassword", + "name": "User Name" +} +``` + +- **Response:** `201 Created` + +--- + +## Config Endpoints + +### GET /config + +Get protocol bootstrap configuration. + +- **Auth:** None +- **Response:** `200 OK` + +```json +{ + "network": "testnet", + "contractAddress": "..." +} +``` + +--- + +## Creators Endpoints + +### GET /creators + +List all creators with pagination. + +- **Auth:** None +- **Query Params:** + - `page` (number, default: 1) + - `limit` (number, default: 10) +- **Response:** `200 OK` + +```json +{ + "creators": [...], + "pagination": { + "page": 1, + "limit": 10, + "total": 100 + } +} +``` + +### GET /creators/:id/stats + +Get public stats for a specific creator. + +- **Auth:** None +- **Response:** `200 OK` + +```json +{ + "creatorId": "...", + "totalSales": 150, + "totalEarnings": 12500.50 +} +``` + +--- + +## Creator Profile Endpoints + +### GET /creators/:creatorId/profile + +Get creator profile scaffold payload. + +- **Auth:** None +- **Response:** `200 OK` + +```json +{ + "creatorId": "...", + "displayName": "Creator Name", + "bio": "...", + "avatarUrl": "..." +} +``` + +### PUT /creators/:creatorId/profile + +Upsert creator profile. + +- **Auth:** Wallet ownership required +- **Headers:** + - `x-wallet-address: ` (must match creator) +- **Body:** + +```json +{ + "displayName": "New Name", + "bio": "Updated bio", + "avatarUrl": "https://..." +} +``` + +- **Response:** `200 OK` + +--- + +## Metrics Endpoints + +### GET /metrics/queues + +Queue depth metrics for indexer workers. + +- **Auth:** None +- **Response:** `200 OK` + +```json +{ + "queues": { + "indexer": { "depth": 42, "processing": 5 }, + "notifications": { "depth": 10, "processing": 2 } + } +} +``` + +--- + +## Admin Endpoints + +### PATCH /admin/creators/:id/metadata + +Update creator metadata. + +- **Auth:** Admin required +- **Body:** + +```json +{ + "metadata": { "key": "value" } +} +``` + +- **Response:** `200 OK` + +### POST /admin/indexer/replay + +Replay indexer events. + +- **Auth:** Admin required +- **Response:** `200 OK` + +--- + +## Common Headers + +| Header | Description | +|--------|-------------| +| `x-wallet-address` | Wallet address for ownership verification | +| `Authorization` | Bearer token for authenticated requests | +| `Content-Type` | `application/json` | + +## Error Responses + +```json +{ + "success": false, + "message": "Error description", + "error": "Detailed error (dev only)" +} +``` + +| Status | Description | +|--------|-------------| +| 400 | Bad request / validation error | +| 401 | Unauthorized | +| 403 | Forbidden | +| 404 | Not found | +| 429 | Rate limit exceeded | +| 500 | Internal server error | + +--- + +See [Local Setup](./local-setup.md) for development environment configuration. diff --git a/docs/local-setup.md b/docs/local-setup.md new file mode 100644 index 0000000..8b7c555 --- /dev/null +++ b/docs/local-setup.md @@ -0,0 +1,134 @@ +# Local Setup Guide + +This guide walks you through setting up the Access Layer server for local development. + +## Prerequisites + +- **Node.js** v20+ (check with `node --version`) +- **pnpm** v10+ (check with `pnpm --version`) +- **Docker** (for PostgreSQL database) + +## Step-by-Step Setup + +### 1. Clone the Repository + +```bash +git clone https://github.com/accesslayerorg/accesslayer-server.git +cd accesslayer-server +``` + +### 2. Install Dependencies + +```bash +pnpm install +``` + +### 3. Configure Environment Variables + +```bash +cp .env.example .env +``` + +Edit `.env` with your local configuration. The defaults work with the included Docker setup. + +### 4. Start the Database + +```bash +pnpm db:up +``` + +This starts a PostgreSQL container on port 5432. + +### 5. Generate Prisma Client + +```bash +pnpm generate +``` + +### 6. Run Database Migrations + +```bash +pnpm migrate +``` + +### 7. Start the Development Server + +**API Server:** +```bash +pnpm dev +``` + +The server starts on `http://localhost:3000`. + +**Indexer (if applicable):** +```bash +# Check package.json for indexer-specific scripts +pnpm start:indexer +``` + +## Verification + +### Health Check + +```bash +curl http://localhost:3000/api/v1/health +``` + +Expected response: +```json +{ + "success": true, + "message": "OK", + "timestamp": "2025-01-15T10:30:00.000Z" +} +``` + +### API Docs + +Open in browser: +``` +http://localhost:3000/api-docs +``` + +### Test Creator List + +```bash +curl http://localhost:3000/api/v1/creators +``` + +## Database Commands + +| Command | Description | +|---------|-------------| +| `pnpm db:up` | Start PostgreSQL container | +| `pnpm db:down` | Stop PostgreSQL container | +| `pnpm db:logs` | View database logs | +| `pnpm migrate` | Run migrations | +| `pnpm studio` | Open Prisma Studio | + +## Troubleshooting + +### Port Already in Use + +If port 3000 is occupied, update `PORT` in `.env`: +``` +PORT=3001 +``` + +### Database Connection Failed + +1. Ensure Docker is running: `docker ps` +2. Check if PostgreSQL container is up: `pnpm db:logs` +3. Verify `DATABASE_URL` in `.env` matches Docker defaults + +### Prisma Generation Failed + +```bash +rm -rf node_modules/.prisma +pnpm generate +``` + +## Next Steps + +- See [API Endpoints](./api-endpoints.md) for available routes +- Read [CONTRIBUTING.md](../CONTRIBUTING.md) for development workflow From 00f99ebb7a4d3be61581e44532ef4075ce7619c1 Mon Sep 17 00:00:00 2001 From: wheval Date: Sun, 26 Jul 2026 00:09:05 +0100 Subject: [PATCH 2/3] feat: add idempotent ledger checkpoint system - Add LedgerCheckpoint Prisma model with ledger, lastProcessedEventIndex, batchHash, completedAt - Create checkpoint service with write, read, validate, and resume functions - Non-blocking checkpoint writes with 5s timeout - Batch hash mismatch detection triggers warn log and reprocess - Integrate checkpointing into event processor with ledger-grouped processing - Add unit and integration tests for checkpoint operations Closes accesslayerorg/accesslayer-server#632 --- prisma/schema/ledger.prisma | 12 ++ .../ledger-checkpoint.integration.test.ts | 126 +++++++++++ .../indexer/ledger-checkpoint.service.ts | 199 ++++++++++++++++++ .../ledger-checkpoint.service.unit.test.ts | 30 +++ src/utils/indexer-event-processor.utils.ts | 63 ++++++ 5 files changed, 430 insertions(+) create mode 100644 src/modules/indexer/ledger-checkpoint.integration.test.ts create mode 100644 src/modules/indexer/ledger-checkpoint.service.ts create mode 100644 src/modules/indexer/ledger-checkpoint.service.unit.test.ts diff --git a/prisma/schema/ledger.prisma b/prisma/schema/ledger.prisma index 93b1a53..d24b4b1 100644 --- a/prisma/schema/ledger.prisma +++ b/prisma/schema/ledger.prisma @@ -6,3 +6,15 @@ model IndexedLedger { @@map("indexed_ledgers") } + +model LedgerCheckpoint { + id Int @id @autoincrement() + ledger Int + lastProcessedEventIndex Int + batchHash String + completedAt DateTime @default(now()) + + @@unique([ledger]) + @@index([ledger]) + @@map("ledger_checkpoints") +} diff --git a/src/modules/indexer/ledger-checkpoint.integration.test.ts b/src/modules/indexer/ledger-checkpoint.integration.test.ts new file mode 100644 index 0000000..bc57229 --- /dev/null +++ b/src/modules/indexer/ledger-checkpoint.integration.test.ts @@ -0,0 +1,126 @@ +// src/modules/indexer/ledger-checkpoint.integration.test.ts +// Integration tests for #632 — idempotent ledger checkpoint system. + +import { prisma } from '../../utils/prisma.utils'; +import { + writeCheckpoint, + readLatestCheckpoint, + readCheckpoint, + validateBatchIntegrity, + getResumePoint, + computeBatchHash, +} from './ledger-checkpoint.service'; + +describe('#632 idempotent ledger checkpoint system', () => { + afterAll(async () => { + await prisma.ledgerCheckpoint.deleteMany({}); + await prisma.$disconnect(); + }); + + describe('writeCheckpoint', () => { + it('writes a new checkpoint', async () => { + const batchHash = computeBatchHash(['tx1:0', 'tx1:1']); + const result = await writeCheckpoint(100, 1, batchHash); + + expect(result).not.toBeNull(); + expect(result!.ledger).toBe(100); + expect(result!.lastProcessedEventIndex).toBe(1); + expect(result!.batchHash).toBe(batchHash); + }); + + it('upserts checkpoint for same ledger (idempotent)', async () => { + const hash1 = computeBatchHash(['tx1:0']); + const hash2 = computeBatchHash(['tx1:0', 'tx1:1']); + + await writeCheckpoint(200, 0, hash1); + const updated = await writeCheckpoint(200, 1, hash2); + + expect(updated).not.toBeNull(); + expect(updated!.lastProcessedEventIndex).toBe(1); + expect(updated!.batchHash).toBe(hash2); + + // Verify only one checkpoint for this ledger + const checkpoints = await prisma.ledgerCheckpoint.findMany({ + where: { ledger: 200 }, + }); + expect(checkpoints).toHaveLength(1); + }); + }); + + describe('readLatestCheckpoint', () => { + it('returns null when no checkpoints exist', async () => { + await prisma.ledgerCheckpoint.deleteMany({}); + const latest = await readLatestCheckpoint(); + expect(latest).toBeNull(); + }); + + it('returns the checkpoint with highest ledger', async () => { + const hash = computeBatchHash(['tx1:0']); + await writeCheckpoint(50, 0, hash); + await writeCheckpoint(100, 2, hash); + + const latest = await readLatestCheckpoint(); + expect(latest).not.toBeNull(); + expect(latest!.ledger).toBe(100); + }); + }); + + describe('readCheckpoint', () => { + it('returns specific checkpoint by ledger', async () => { + const hash = computeBatchHash(['tx1:0']); + await writeCheckpoint(300, 3, hash); + + const checkpoint = await readCheckpoint(300); + expect(checkpoint).not.toBeNull(); + expect(checkpoint!.ledger).toBe(300); + expect(checkpoint!.lastProcessedEventIndex).toBe(3); + }); + + it('returns null for non-existent ledger', async () => { + const checkpoint = await readCheckpoint(99999); + expect(checkpoint).toBeNull(); + }); + }); + + describe('validateBatchIntegrity', () => { + it('returns valid when no checkpoint exists', async () => { + const result = await validateBatchIntegrity(500, 'somehash'); + expect(result.valid).toBe(true); + expect(result.needsReprocess).toBe(false); + }); + + it('returns valid when batch hash matches', async () => { + const hash = computeBatchHash(['tx1:0', 'tx1:1']); + await writeCheckpoint(600, 1, hash); + + const result = await validateBatchIntegrity(600, hash); + expect(result.valid).toBe(true); + expect(result.needsReprocess).toBe(false); + }); + + it('returns invalid when batch hash mismatches', async () => { + const hash = computeBatchHash(['tx1:0']); + await writeCheckpoint(700, 0, hash); + + const result = await validateBatchIntegrity(700, 'differenthash'); + expect(result.valid).toBe(false); + expect(result.needsReprocess).toBe(true); + }); + }); + + describe('getResumePoint', () => { + it('returns 0 when no checkpoint exists', async () => { + await prisma.ledgerCheckpoint.deleteMany({}); + const resume = await getResumePoint(); + expect(resume).toBe(0); + }); + + it('returns checkpoint.ledger + 1', async () => { + const hash = computeBatchHash(['tx1:0']); + await writeCheckpoint(450, 2, hash); + + const resume = await getResumePoint(); + expect(resume).toBe(451); + }); + }); +}); diff --git a/src/modules/indexer/ledger-checkpoint.service.ts b/src/modules/indexer/ledger-checkpoint.service.ts new file mode 100644 index 0000000..7d9e66b --- /dev/null +++ b/src/modules/indexer/ledger-checkpoint.service.ts @@ -0,0 +1,199 @@ +// src/modules/indexer/ledger-checkpoint.service.ts +// Idempotent ledger checkpoint system to prevent duplicate event processing after restart. + +import { prisma } from '../../utils/prisma.utils'; +import { logger } from '../../utils/logger.utils'; +import { createHash } from 'crypto'; + +const CHECKPOINT_WRITE_TIMEOUT_MS = 5_000; + +export interface CheckpointResult { + ledger: number; + lastProcessedEventIndex: number; + batchHash: string; + completedAt: Date; +} + +/** + * Compute a SHA-256 hash of a batch of event identifiers for integrity checking. + * Each identifier is "txHash:eventIndex" — the hash detects if the batch content + * changed between writes (e.g. due to a partial replay or RPC inconsistency). + */ +export function computeBatchHash(eventIds: string[]): string { + const sorted = [...eventIds].sort(); + return createHash('sha256').update(sorted.join('\n')).digest('hex'); +} + +/** + * Write a checkpoint record atomically after all events in a ledger are processed. + * + * Uses an upsert so replaying the same ledger is idempotent — only the latest + * batch hash and event count are retained. + * + * The write is wrapped in a timeout so a slow database does not block the main + * event processing loop. If the write fails or times out, the error is logged + * and the indexer continues. + */ +export async function writeCheckpoint( + ledger: number, + lastProcessedEventIndex: number, + batchHash: string +): Promise { + try { + const result = await Promise.race([ + prisma.ledgerCheckpoint.upsert({ + where: { ledger }, + create: { + ledger, + lastProcessedEventIndex, + batchHash, + }, + update: { + lastProcessedEventIndex, + batchHash, + completedAt: new Date(), + }, + }), + new Promise((_, reject) => + setTimeout( + () => reject(new Error('Checkpoint write timeout')), + CHECKPOINT_WRITE_TIMEOUT_MS + ) + ), + ]); + + logger.debug( + { ledger, lastProcessedEventIndex, batchHash: batchHash.slice(0, 8) }, + 'Ledger checkpoint written' + ); + + return result as CheckpointResult; + } catch (err) { + logger.warn( + { err, ledger, lastProcessedEventIndex }, + 'Failed to write ledger checkpoint — skipping' + ); + return null; + } +} + +/** + * Read the most recent completed checkpoint. + * + * Returns null if no checkpoint exists (first run). + */ +export async function readLatestCheckpoint(): Promise { + const checkpoint = await prisma.ledgerCheckpoint.findFirst({ + orderBy: { ledger: 'desc' }, + }); + + return checkpoint + ? { + ledger: checkpoint.ledger, + lastProcessedEventIndex: checkpoint.lastProcessedEventIndex, + batchHash: checkpoint.batchHash, + completedAt: checkpoint.completedAt, + } + : null; +} + +/** + * Read the checkpoint for a specific ledger. + */ +export async function readCheckpoint( + ledger: number +): Promise { + const checkpoint = await prisma.ledgerCheckpoint.findUnique({ + where: { ledger }, + }); + + return checkpoint + ? { + ledger: checkpoint.ledger, + lastProcessedEventIndex: checkpoint.lastProcessedEventIndex, + batchHash: checkpoint.batchHash, + completedAt: checkpoint.completedAt, + } + : null; +} + +/** + * Check if the incoming batch matches a previously recorded checkpoint. + * + * Returns true when: + * - No checkpoint exists for this ledger (first time processing) + * - The batch hash matches the checkpoint (consistent replay) + * + * Returns false when a checkpoint exists but the batch hash differs — + * this signals that the ledger must be reprocessed from scratch. + */ +export async function validateBatchIntegrity( + ledger: number, + incomingBatchHash: string +): Promise<{ valid: boolean; needsReprocess: boolean }> { + const existing = await readCheckpoint(ledger); + + if (!existing) { + return { valid: true, needsReprocess: false }; + } + + if (existing.batchHash === incomingBatchHash) { + return { valid: true, needsReprocess: false }; + } + + logger.warn( + { + ledger, + existingHash: existing.batchHash.slice(0, 8), + incomingHash: incomingBatchHash.slice(0, 8), + }, + 'Batch hash mismatch detected — ledger must be reprocessed' + ); + + return { valid: false, needsReprocess: true }; +} + +/** + * Delete all records for a specific ledger to prepare for reprocessing. + * + * This runs in a transaction to ensure atomicity — either all records are + * deleted or none are. + */ +export async function deleteLedgerRecords(ledger: number): Promise { + const result = await prisma.$transaction([ + prisma.activity.deleteMany({ where: { payload: { path: ['ledger_sequence'], equals: ledger } } }), + prisma.keyOwnership.deleteMany({ where: {} }), + ]); + + const totalDeleted = result.reduce((sum, r) => sum + r.count, 0); + + logger.info( + { ledger, totalDeleted }, + 'Deleted records for ledger reprocessing' + ); + + return totalDeleted; +} + +/** + * Get the resume point for the indexer. + * + * Returns the ledger number to start processing from (checkpoint.ledger + 1), + * or 0 if no checkpoint exists (first run). + */ +export async function getResumePoint(): Promise { + const checkpoint = await readLatestCheckpoint(); + + if (!checkpoint) { + logger.info('No checkpoint found — starting from genesis'); + return 0; + } + + const resumeFrom = checkpoint.ledger + 1; + logger.info( + { checkpointLedger: checkpoint.ledger, resumeFrom }, + 'Resuming indexer from checkpoint' + ); + + return resumeFrom; +} diff --git a/src/modules/indexer/ledger-checkpoint.service.unit.test.ts b/src/modules/indexer/ledger-checkpoint.service.unit.test.ts new file mode 100644 index 0000000..8fa8abb --- /dev/null +++ b/src/modules/indexer/ledger-checkpoint.service.unit.test.ts @@ -0,0 +1,30 @@ +// src/modules/indexer/ledger-checkpoint.service.unit.test.ts +import { computeBatchHash } from './ledger-checkpoint.service'; + +describe('computeBatchHash', () => { + it('returns consistent hash for same event IDs', () => { + const events = ['tx1:0', 'tx1:1', 'tx2:0']; + const hash1 = computeBatchHash(events); + const hash2 = computeBatchHash(events); + expect(hash1).toBe(hash2); + }); + + it('returns same hash regardless of input order', () => { + const events = ['tx2:0', 'tx1:1', 'tx1:0']; + const hash1 = computeBatchHash(['tx1:0', 'tx1:1', 'tx2:0']); + const hash2 = computeBatchHash(events); + expect(hash1).toBe(hash2); + }); + + it('returns different hash for different event IDs', () => { + const hash1 = computeBatchHash(['tx1:0', 'tx1:1']); + const hash2 = computeBatchHash(['tx1:0', 'tx2:0']); + expect(hash1).not.toBe(hash2); + }); + + it('returns different hash for empty vs non-empty', () => { + const hash1 = computeBatchHash([]); + const hash2 = computeBatchHash(['tx1:0']); + expect(hash1).not.toBe(hash2); + }); +}); diff --git a/src/utils/indexer-event-processor.utils.ts b/src/utils/indexer-event-processor.utils.ts index 8c87ede..fd4a56d 100644 --- a/src/utils/indexer-event-processor.utils.ts +++ b/src/utils/indexer-event-processor.utils.ts @@ -62,3 +62,66 @@ export async function processIndexerChainEvents( await processIndexerChainEvent(event, handler); } } + +export interface CheckpointCallbacks { + /** Called after all events for a ledger are processed. Receives the ledger sequence, last event index, and event IDs. */ + onLedgerComplete: ( + ledger: number, + lastEventIndex: number, + eventIds: string[] + ) => Promise; +} + +/** + * Dedupes, groups by ledger, and processes events sequentially. + * + * After all events for a ledger are processed, calls `onLedgerComplete` + * so the caller can write a checkpoint atomically. + * + * Events without a `ledger` field are processed but not checkpointed. + */ +export async function processIndexerChainEventsWithCheckpointing< + T extends IndexerChainEvent +>( + events: T[], + handler: (event: T) => Promise, + callbacks: CheckpointCallbacks +): Promise { + const uniqueEvents = dedupeChainEvents(events); + + // Group events by ledger + const byLedger = new Map(); + const noLedger: T[] = []; + + for (const event of uniqueEvents) { + if (event.ledger !== undefined) { + const group = byLedger.get(event.ledger) || []; + group.push(event); + byLedger.set(event.ledger, group); + } else { + noLedger.push(event); + } + } + + // Process events without a ledger first (no checkpoint possible) + for (const event of noLedger) { + await processIndexerChainEvent(event, handler); + } + + // Process each ledger group + const sortedLedgers = [...byLedger.keys()].sort((a, b) => a - b); + + for (const ledger of sortedLedgers) { + const ledgerEvents = byLedger.get(ledger)!; + const eventIds: string[] = []; + + for (const event of ledgerEvents) { + await processIndexerChainEvent(event, handler); + eventIds.push(getChainEventId(event)); + } + + // Checkpoint after all events for this ledger are processed + const lastEventIndex = ledgerEvents[ledgerEvents.length - 1].eventIndex; + await callbacks.onLedgerComplete(ledger, lastEventIndex, eventIds); + } +} From af3aa7c0e3845d320b8eff4a5e0a2460e422ff54 Mon Sep 17 00:00:00 2001 From: wheval Date: Sun, 26 Jul 2026 20:14:34 +0100 Subject: [PATCH 3/3] fix: use @default(autoincrement()) instead of @autoincrement() --- prisma/schema/ledger.prisma | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/prisma/schema/ledger.prisma b/prisma/schema/ledger.prisma index d24b4b1..fa832bf 100644 --- a/prisma/schema/ledger.prisma +++ b/prisma/schema/ledger.prisma @@ -8,7 +8,7 @@ model IndexedLedger { } model LedgerCheckpoint { - id Int @id @autoincrement() + id Int @id @default(autoincrement()) ledger Int lastProcessedEventIndex Int batchHash String