diff --git a/buyer/package.json b/buyer/package.json index 57c24bc..f471ac0 100644 --- a/buyer/package.json +++ b/buyer/package.json @@ -3,6 +3,9 @@ "version": "0.0.0", "private": true, "type": "module", + "exports": { + "./sdk": "./src/sdk/index.ts" + }, "scripts": { "typecheck": "tsc --noEmit", "build:ui": "vite build --config vite.config.ts", diff --git a/deploy/mainnet/docker-compose.gateway.yml b/deploy/mainnet/docker-compose.gateway.yml new file mode 100644 index 0000000..5dca5f0 --- /dev/null +++ b/deploy/mainnet/docker-compose.gateway.yml @@ -0,0 +1,62 @@ +# deploy/mainnet/docker-compose.gateway.yml — OpenAI-compatible Gateway — MAINNET +# +# Multi-tenant, custodial OpenAI gateway. Each API key is its own buyer wallet; +# every request posts an on-chain escrow from that user's wallet. See docs/gateway.md. +# +# Secrets live in gateway/.env (NOT committed): +# GATEWAY_MASTER_KEY= # losing this locks ALL user funds +# Back up BOTH gateway/.env (the master key) AND the gateway-mainnet-data volume +# (the gateway.db holds every user's encrypted wallet key + usage). + +name: marketplace-mainnet-gateway + +services: + gateway: + build: + context: ../.. + dockerfile: gateway/Dockerfile + container_name: marketplace-mainnet-gateway + restart: unless-stopped + env_file: + - path: ../../gateway/.env + required: true + environment: + INDEXER_URL: http://marketplace-mainnet-indexer:8090 + OGMIOS_URL: https://ogmios.vector.mainnet.apexfusion.org + LIVE_CHAIN: "1" + NETWORK_ID: "1" + GATEWAY_PORT: "8080" + GATEWAY_DB_DIR: /repo/data/gateway + volumes: + - gateway-mainnet-data:/repo/data/gateway + healthcheck: + test: ["CMD-SHELL", "wget -qO- http://localhost:8080/healthz >/dev/null || exit 1"] + interval: 20s + timeout: 5s + retries: 3 + start_period: 15s + networks: + - marketplace-mainnet-net + - traefik-net + labels: + - "traefik.enable=true" + - "traefik.docker.network=traefik-net" + - "traefik.http.routers.mp-gateway-mainnet.entrypoints=web" + - "traefik.http.routers.mp-gateway-mainnet.rule=Host(`api.marketplace.vector.apexfusion.org`)" + - "traefik.http.services.mp-gateway-mainnet.loadbalancer.server.port=8080" + logging: + driver: json-file + options: + max-size: "10m" + max-file: "3" + +networks: + marketplace-mainnet-net: + external: true + name: marketplace-mainnet-net + traefik-net: + external: true + name: traefik-net + +volumes: + gateway-mainnet-data: diff --git a/docs/gateway.md b/docs/gateway.md new file mode 100644 index 0000000..de48f10 --- /dev/null +++ b/docs/gateway.md @@ -0,0 +1,149 @@ +# OpenAI-Compatible API Gateway for the Vector Marketplace + +## Context + +Today the only way to send a prompt to the marketplace is the single-operator **buyer** web app (cookie auth, one shared wallet). We want external developers to use **any OpenAI SDK** against the marketplace. Design (locked through interview + a 5-lens adversarial audit of an earlier draft): + +- **Public, multi-tenant, custodial gateway.** Each API key is its **own buyer wallet** (own Ed25519 key, held AES-GCM-encrypted by the gateway). The user funds their own deposit address with AP3X; each request posts an on-chain escrow **from that user's wallet**. Not the deployed operator buyer. +- **Two surfaces, two escrow models:** + - **One-shot completions** — `POST /openai/v1/chat/completions` (stateless, OpenAI-standard). Routes to an `llm.text.generate.v1` supplier. **One escrow per call.** `stream:true` → buffered pseudo-stream. + - **Chat sessions** — a Vector REST extension. Routes to an `llm.chat.v1` supplier. **One escrow per session**: fixed price locked at open, **real token streaming** per turn (no per-turn charge), settled at session close. +- **Routing matches capability + model, not cheapest.** Per-prompt cost = the matched supplier's advertised price. +- **Gateway auto-maintains each user's wallet** (collateral + consolidation) so requests past #1 keep working. +- Onboarding is **self-serve via a small web UI**; the response is a **spec-clean OpenAI body + `x_vector` receipt extension**; custody is protected with **settle-on-success + a state-aware reclaim sweeper + a withdrawal path**. + +The audit confirmed 39 holes in the first draft (10 distinct themes). This document incorporates every fix; the "Audit-driven corrections" section maps each theme to its resolution. + +## Locked decisions + +| Area | Decision | +|---|---| +| Scope | Public multi-tenant OpenAI-compatible gateway; new `gateway/` pnpm package reusing the buyer `Marketplace` SDK | +| Identity | Each API key = its own custodial buyer wallet | +| Custody | AES-256-GCM, `GATEWAY_MASTER_KEY` (env), per-row nonce; decrypt in-memory at request time; **rotation CLI**; **withdrawal endpoint** is the user exit | +| One-shot | `/openai/v1/chat/completions` → `llm.text.generate.v1`, 1 escrow/call, buffered (+ pseudo-stream) | +| Chat session | `/openai/v1/chat/sessions*` → `llm.chat.v1`, 1 escrow/**session**, real SSE per turn, settle at close | +| Routing | match `model` + required capability; pick an available supplier (status `free`\|`unknown`); **not cheapest**; fallback to next match on busy/offline | +| Pricing | cost = matched supplier `price_lovelace`; **required balance = price + buyer_bond + supplier_bond + 5 ADA collateral + fee reserve** | +| Wallet health | gateway **auto-consolidates** each wallet to `{5-ADA collateral, working}` after settle / on a tick | +| Settlement | resolve the **Submitted** ref via indexer, `acceptResult`, **await confirm** before returning success; sweeper is backstop | +| Sweeper | state-aware: Open/Claimed past `deliver_by` → reclaim; Submitted → retry Accept in-window; never blind-reclaim Submitted | +| Response | OpenAI object `{id, object, created, model, choices, usage}` + `x_vector:{receipt, receipt_signature, escrow_ref}`; full streaming chunk transform + final usage chunk | +| Hardening | `/signup` IP rate-limit, per-key request rate-limit, `SdkRegistry` LRU cap, structured logging + audit trail | + +## Architecture + +New standalone pnpm package `gateway/` (add to `pnpm-workspace.yaml`). Boots **one** shared `ChainProvider` (Ogmios, `LIVE_CHAIN=1`) + indexer URL, and an `SdkRegistry` that lazily builds **one `Marketplace` per API key** bound to that user's `WalletKey` (LRU-capped, evicted by idle). All on-chain work for a key is serialized by a per-key promise-chain mutex. + +### Verified reuse points (do not re-implement) +- `Marketplace` ctor `MarketplaceOpts{chain, indexerUrl, walletKey, networkParams, historyStore?}` — `buyer/src/sdk/Marketplace.ts:117`. Per-user field is only `walletKey`. +- `deriveWalletKey` + ed25519 hook — `buyer/src/index.ts:24-64`. +- SDK methods: `discoverSuppliers`, `submitPrompt` (`Marketplace.ts:242`, verifies receipt but **does not settle**), `startChat`/`endChat` (chat session; `endChat` returns `acceptedRef` and awaits Accept), `acceptResult`, `reclaim`. +- Escrow value/bonds: buyer funds `totalLocked = price + buyer_bond + supplier_bond` at PostEscrow — `packages/shared/src/tx/escrow/postEscrow.ts:132-200`. On Accept supplier gets `payment+supplier_bond`, buyer gets `buyer_bond` back. +- Submitted-ref resolution: the buyer `/v1/accept` queries the indexer and matches the Submitted row by `posted_at` — `buyer/src/server.ts` `/v1/accept`. Gateway must mirror this. +- Wallet health: `planConsolidate` / `consolidateWallet` (`{collateral, working}` shape, `DEFAULT_COLLATERAL_LOVELACE = 5 ADA`) — `packages/shared/src/tx/wallet/`. Collateral assertion: `assertCollateralCandidate` — `packages/shared/src/tx/internal/liveCbor.ts`. +- Supplier chat-session SSE `{type:"token"|"done"|"error"}` + headers — `supplier/src/server.ts` `makeChatSessionHandlers`; buyer relay in `buyer/src/server.ts` `/v1/chat/message` (note: **verbatim** passthrough — the gateway must transform to OpenAI chunks). +- SQLite pattern (`better-sqlite3`, WAL) — `buyer/src/db/archive.ts`. Auth helpers `timingSafeCompareStrings`, `createRateLimiter` — `buyer/src/auth.ts`. +- Indexer `SupplierView` (`model`, `capability_id`, `price_lovelace`, `buyer_bond_lovelace`, `supplier_bond_lovelace`, `status` free/working/offline/unknown, `advert_status`) — `indexer/src/routes/suppliers.ts`. + +## Package layout (`gateway/`) +``` +src/ + index.ts # boot: config → shared chain/indexer → GatewayStore → SdkRegistry → app → listen + sweeper + wallet-health tick + config.ts # GATEWAY_MASTER_KEY(64hex), INDEXER_URL, OGMIOS_URL, LIVE_CHAIN, NETWORK_ID, GATEWAY_PORT, GATEWAY_DB_DIR, rate-limit knobs + wallet.ts # deriveWalletKey + ed25519 hook + genPrivKey + crypto/seal.ts # aes-256-gcm seal()/open(), per-row nonce + db/{schema.ts,store.ts} # api_keys, usage, sessions + sdk/registry.ts # per-key Marketplace cache (LRU) + per-key Mutex + routing/selectSupplier.ts # indexer query → capability+model match → available → fallback + openai/{shapes.ts,errors.ts,chatCompletions.ts,sessions.ts,models.ts} + account/routes.ts # POST /signup, GET /account, POST /account/withdraw + middleware/apiKeyAuth.ts, middleware/rateLimit.ts + walletHealth.ts # post-settle + ticker consolidation (reuse planConsolidate) + sweeper.ts # state-aware reclaim/accept-retry + ui/index.html + cli/rotate-master-key.ts # re-encrypt all rows old→new master key +``` +Build wiring: add `"exports": { "./sdk": "./dist/sdk/index.js" }` to `buyer/package.json`, depend `@marketplace/buyer: workspace:*` + `@marketplace/shared: workspace:*`. + +## SQLite schema +```sql +CREATE TABLE api_keys ( id TEXT PRIMARY KEY, key_hash TEXT UNIQUE NOT NULL, key_prefix TEXT NOT NULL, label TEXT, + wallet_pkh TEXT NOT NULL, deposit_address TEXT NOT NULL, + enc_priv_nonce TEXT NOT NULL, enc_priv_ct TEXT NOT NULL, enc_priv_tag TEXT NOT NULL, + master_key_version INTEGER NOT NULL DEFAULT 1, created_at INTEGER NOT NULL, disabled INTEGER NOT NULL DEFAULT 0 ); +CREATE TABLE usage ( id TEXT PRIMARY KEY, key_id TEXT NOT NULL, created_at INTEGER NOT NULL, kind TEXT NOT NULL, + model TEXT, capability_id TEXT, supplier_pkh TEXT, escrow_ref TEXT, cost_lovelace TEXT, + prompt_tokens INTEGER DEFAULT 0, completion_tokens INTEGER DEFAULT 0, status TEXT NOT NULL, failure_reason TEXT ); +CREATE TABLE sessions ( id TEXT PRIMARY KEY, key_id TEXT NOT NULL, escrow_ref TEXT NOT NULL, session_nonce TEXT NOT NULL, + supplier_base_url TEXT NOT NULL, model TEXT NOT NULL, state TEXT NOT NULL, opened_at INTEGER NOT NULL, closed_at INTEGER ); +``` +Raw API key never persisted (only `sha256(key)`); `key_prefix` for display. `master_key_version` enables rotation. + +## Endpoints + +**One-shot (OpenAI standard):** +- `POST /openai/v1/chat/completions` — Bearer. Parse `{model, messages, max_tokens?, stream?}`; **reject** `tools`/`tool_choice`/`functions` with 400; **ignore** `temperature/n/stop/top_p/...` (documented). Lifecycle below. +- `GET /openai/v1/models` — distinct `model` over Active suppliers → OpenAI list shape. + +**Chat sessions (Vector extension — documented as non-standard):** +- `POST /openai/v1/chat/sessions {model}` — open: route to `llm.chat.v1`, `startChat` (lock fixed price), persist session row → `{session_id, model}`. +- `POST /openai/v1/chat/sessions/{id}/messages {messages|content, stream}` — real SSE turn; transform supplier `{type:token}` → `chat.completion.chunk`; final usage chunk + `[DONE]`. No new escrow. +- `POST /openai/v1/chat/sessions/{id}/close` — settle: `endChat` (Submit+Accept, awaits), record usage, then run wallet-health. + +**Account / UI:** +- `POST /signup {label?}` — IP rate-limited. Generate priv → derive → seal → store → `{api_key (once), deposit_address, key_prefix}`. +- `GET /account` — Bearer. `{deposit_address, balance:{available_lovelace, locked_in_escrow_lovelace, ap3x}, collateral_ok:bool, spend:{total_cost_lovelace, request_count}, recent_usage[]}`. +- `POST /account/withdraw {to_address, amount_lovelace?}` — Bearer. Build a wallet→address transfer (new `buildWithdrawTx` in packages/shared; repo has no send tx today) sending available funds out. **The custodial exit path.** +- `GET /` — self-serve UI (create key, copy deposit address, live balances, usage, withdraw). + +## Request lifecycle — one-shot `/chat/completions` +All on-chain steps run inside the per-key `mutex.run(...)`. +1. **Route** (`selectSupplier`): indexer rows where `model` matches, `capability_id==="llm.text.generate.v1"`, `advert_status==="Active"`, live `status ∈ {free, unknown}`. Pick an available one (not cheapest). No match → 404 `model_not_found` (`type: invalid_request_error`). +2. **Pre-flight**: required = `price + buyer_bond + supplier_bond + 5_000_000 (collateral) + ~2 AP3X fee`. Verify total balance **and** a pure-ADA UTxO ≥5 ADA exists (`assertCollateralCandidate`-style). Else 402 `insufficient_funds` + `x_vector:{required_lovelace, available_lovelace, collateral_ok, deposit_address}`. +3. **Execute**: `submitPrompt({advertRef, messages, payment_lovelace:price, max_output_tokens})` (posts escrow → supplier → verifies receipt). +4. **Settle**: resolve the **Submitted** escrow ref via the indexer (mirror buyer `/v1/accept`: match by buyer pkh + `posted_at`), `acceptResult(submittedRef)`, **await confirmation**. On 409/offline mid-route, retry next matching supplier (bounded). On any post-escrow failure, record + leave for the sweeper (don't block the response on reclaim). +5. **Respond**: `{id:"chatcmpl-…", object:"chat.completion", created:, model, choices:[{index:0,message:{role:"assistant",content},finish_reason:"stop"}], usage:{prompt_tokens,completion_tokens,total_tokens}, x_vector:{receipt,receipt_signature,escrow_ref}}`. Record `usage`. +6. **Wallet health**: after settle, consolidate the wallet back to `{collateral, working}` (inside the mutex or queued). +7. `stream:true`: same flow; emit one (or chunked) `chat.completion.chunk` delta + final `finish_reason:"stop"` chunk **with `usage`** + `data: [DONE]`. Emit `: keepalive` SSE comments during the escrow-confirm gap. + +## Chat-session lifecycle +- **open** → `startChat` (escrow at fixed price), store `{escrow_ref, session_nonce, supplier_base_url}`. +- **messages** → POST the turn to `${supplier_base_url}/v1/chat/message` with `X-Escrow-Ref`; transform each `{type:token,value}` → `chat.completion.chunk`; `{type:done}` → final chunk + usage + `[DONE]`. (System messages: fold into the first turn's content; documented limitation since chat.v1 has no system channel.) +- **close** → `endChat` (Submit + Accept, awaits) → settle + bond refund; record usage; wallet-health. +- Sessions are reclaimable by the sweeper if abandoned before close. + +## Sweeper & wallet-health +- **Sweeper** (interval ~60s): per key, classify escrows. Open/Claimed past `deliver_by` → `reclaim`. Submitted **with a verified receipt** still in accept-window → retry `acceptResult`. Submitted past accept-window → leave for supplier Release (log). Never blind-`reclaim` Submitted. +- **Wallet-health tick**: periodically + after each settle, `planConsolidate` → `consolidateWallet` so every wallet keeps a ≥5 ADA collateral UTxO. Skip if already healthy. + +## Errors (OpenAI shape `{error:{message, type, code, param:null}}`) +no supplier → 404 `invalid_request_error`/`model_not_found`; underfunded/no-collateral → 402 `invalid_request_error`/`insufficient_funds`; bad/missing key → 401 `authentication_error`/`invalid_api_key`; rate limit → 429 `rate_limit_error` + `Retry-After`; supplier timeout → 504 `server_error`; supplier busy race (after retries) → 503 `server_error`/`overloaded`; receipt verify → 502 `server_error`/`receipt_verification_failed`; tx build → 409/502 `server_error`/`escrow_failed`; unsupported `tools`/`functions` → 400 `invalid_request_error`/`unsupported_parameter`. + +## Audit-driven corrections (themes → resolution) +1. **Settlement/Submitted-ref + fire-and-forget Accept** → lifecycle step 4 (indexer resolution + awaited Accept); sweeper backstop. +2. **No withdrawal** → `POST /account/withdraw` + new `buildWithdrawTx`. +3. **Master-key SPOF / offline-stranding** → documented backup, `rotate-master-key` CLI (`master_key_version`), withdrawal as exit; trustless recovery (pre-signed reclaim / multisig) explicitly **out of scope**, documented as residual risk. +4. **Stateless↔stateful chat** → sessions are explicit (open/messages/close), one escrow/session; system-message folding documented. +5. **Collateral/fragmentation** → corrected required-balance formula + pre-flight collateral check + auto wallet-health consolidation. +6. **~0.5 req/min throughput** (awaitTx in mutex) → documented per-key SLA; "move awaitTx out of the lock" noted as fast-follow. +7. **Sweeper can't reclaim Submitted** → state-aware sweeper. +8. **OpenAI wire format** → full response object + chunk transform + final usage chunk + keepalive + error taxonomy + param policy. +9. **Hardening** → `/signup` IP limit, per-key request limit, `SdkRegistry` LRU cap, `status ∈ {free,unknown}` + freshness + fallback retry. +10. **Economics/observability** → v1 = **no on-chain margin** (operator subsidizes infra; a fee needs a separate transfer or off-chain credit — folded into the deferred "billing" phase); add structured per-request logging + audit (request_id, key_id, model, supplier_pkh, escrow_ref, status, latency) + basic metrics/alerts (indexer stale, Ogmios slow, sweeper idle). +- **Doc fix**: clarify the **off-chain tx builder** enforces `payment==advert.price`; the validator only enforces the state machine, not pricing. + +## Out of scope for v1 (documented) +Real billing/markup & fee hook; trustless/self-custody recovery; multi-funding-UTxO optimization beyond the consolidation loop; moving `awaitTx` out of the per-key lock; real chat.v1 system-role support (needs a supplier API change). + +## Verification (testnet, end-to-end) +1. Indexer + Ogmios + ≥1 `llm.text.generate.v1` and ≥1 `llm.chat.v1` supplier for the same `model`, Active. +2. Boot gateway: `LIVE_CHAIN=1`, `GATEWAY_MASTER_KEY=$(openssl rand -hex 32)`, `OGMIOS_URL`, `INDEXER_URL`, `NETWORK_ID`, `GATEWAY_DB_DIR`, `GATEWAY_PORT`. +3. `POST /signup` → key + deposit address. Fund ≥ price + both bonds + 5 ADA collateral + ~2 ADA fees; confirm via `GET /account` (`collateral_ok:true`). +4. One-shot non-stream via curl and via the OpenAI Python/JS SDK (`base_url=…/openai/v1`) → valid `chat.completion` + `x_vector`; confirm the escrow **settled (Accepted)** and a `usage` row exists. +5. **Run a 2nd and 3rd request** → confirm wallet-health kept `collateral_ok` and they succeed (the audit's request-#2 failure must not reproduce). +6. One-shot `stream:true` → chunks + final usage + `[DONE]`. +7. Chat session: open → stream 2–3 turns → close; confirm **one** escrow for the whole session, settled at close. +8. `POST /account/withdraw` → unspent AP3X returns to an external address. +9. Underfunded / no-collateral key → 402 with guidance. Unknown model → 404. `tools:[…]` → 400. +10. Kill gateway mid-request → restart → sweeper reclaims Open/Claimed and retries Accept on Submitted; balances recover. diff --git a/gateway/.env.example b/gateway/.env.example new file mode 100644 index 0000000..f8ae443 --- /dev/null +++ b/gateway/.env.example @@ -0,0 +1,34 @@ +# gateway/.env.example — copy to .env and fill in. + +# 32-byte (64 hex char) master key used to AES-256-GCM-encrypt every user's +# custodial wallet private key at rest. Generate with: openssl rand -hex 32 +# LOSING THIS PERMANENTLY LOCKS ALL USER FUNDS. Back it up out-of-band. +GATEWAY_MASTER_KEY= + +# Internal indexer base URL (e.g. http://marketplace-indexer:8090). +INDEXER_URL= + +# Ogmios HTTP endpoint. Required (the gateway posts/settles escrows on chain). +OGMIOS_URL= + +# Must be "1" — the gateway signs and submits real transactions. +LIVE_CHAIN=1 + +# "0" testnet (addr_test…), "1" mainnet (addr…). Default "0". +NETWORK_ID=0 + +# HTTP port. Default 3010. +GATEWAY_PORT=3010 + +# Directory holding gateway.db (custodial keys + usage + sessions). BACK IT UP. +# Default ./data/gateway +GATEWAY_DB_DIR=./data/gateway + +# Optional tuning (defaults shown): +# SIGNUP_RATE_MAX=5 # signups per IP per window +# SIGNUP_RATE_WINDOW_MS=3600000 +# KEY_RATE_MAX=60 # requests per API key per window +# KEY_RATE_WINDOW_MS=60000 +# SWEEPER_INTERVAL_MS=60000 +# WALLET_HEALTH_INTERVAL_MS=600000 +# SDK_REGISTRY_MAX=500 # max cached per-key Marketplace instances diff --git a/gateway/Dockerfile b/gateway/Dockerfile new file mode 100644 index 0000000..d062f7d --- /dev/null +++ b/gateway/Dockerfile @@ -0,0 +1,70 @@ +# syntax=docker/dockerfile:1.7 +# +# gateway/Dockerfile — multi-stage image for the OpenAI-compatible gateway. +# +# Mirrors buyer/Dockerfile: we do NOT run `tsc` for the server because +# @marketplace/shared and @marketplace/buyer export .ts directly via +# package.json#exports — the tsx loader handles them at runtime. The gateway +# reuses the buyer's Marketplace SDK + lucid (buildAcceptTx / buildWithdrawTx), +# so it needs the same better-sqlite3 native rebuild + libsodium patch and the +# contracts/marketplace/plutus.json blueprint. +# +# Build: +# docker build -f gateway/Dockerfile -t marketplace-gateway . + +ARG NODE_VERSION=20.19.0 + +# ─── builder ────────────────────────────────────────────────────────────── +FROM node:${NODE_VERSION}-alpine AS builder +WORKDIR /repo + +RUN apk add --no-cache python3 make g++ +RUN corepack enable && corepack prepare pnpm@10.33.2 --activate + +COPY pnpm-workspace.yaml pnpm-lock.yaml package.json tsconfig.base.json .npmrc ./ +COPY packages/shared/package.json ./packages/shared/package.json +COPY buyer/package.json ./buyer/package.json +COPY gateway/package.json ./gateway/package.json + +# Install the gateway and its workspace deps (buyer + shared) with their deps. +RUN pnpm install --frozen-lockfile --filter '@marketplace/gateway...' + +# better-sqlite3 native binding (same step as buyer/indexer images). +RUN BSQ_DIR=$(ls -d node_modules/.pnpm/better-sqlite3@*/node_modules/better-sqlite3 2>/dev/null | head -1) \ + && [ -n "$BSQ_DIR" ] || (echo "FATAL: better-sqlite3 not in pnpm store" && exit 1) \ + && (cd "$BSQ_DIR" && (npx --yes prebuild-install --runtime=node || npm install --build-from-source --no-save)) \ + && ls -la "$BSQ_DIR/build/Release/better_sqlite3.node" + +# libsodium wrapper patch (lucid/CML dep) — same fix as the buyer image. +RUN set -e \ + && SRC=$(find node_modules/.pnpm -name "libsodium-sumo.mjs" -path "*/libsodium-sumo/dist/*" 2>/dev/null | head -1) \ + && DST=$(find node_modules/.pnpm -type d -path "*libsodium-wrappers-sumo@*/node_modules/libsodium-wrappers-sumo/dist/modules-sumo-esm" 2>/dev/null | head -1) \ + && [ -n "$SRC" ] && [ -n "$DST" ] || (echo "FATAL: libsodium-sumo paths not resolved" && exit 1) \ + && cp "$SRC" "$DST/" + +COPY packages/shared ./packages/shared +COPY buyer ./buyer +COPY gateway ./gateway +# loadBlueprint reads contracts/marketplace/plutus.json at runtime (Accept/Withdraw). +COPY contracts/marketplace/plutus.json ./contracts/marketplace/plutus.json + +# ─── runtime ────────────────────────────────────────────────────────────── +FROM node:${NODE_VERSION}-alpine AS runtime +WORKDIR /repo + +RUN apk add --no-cache tini +COPY --from=builder /repo /repo + +# Seed the gateway DB dir owned by node (so a fresh named volume inherits it). +RUN mkdir -p /repo/data/gateway && chown -R node:node /repo +USER node + +ENV NODE_ENV=production +ENV GATEWAY_PORT=8080 +EXPOSE 8080 + +HEALTHCHECK --interval=20s --timeout=5s --start-period=15s --retries=3 \ + CMD wget -qO- http://127.0.0.1:${GATEWAY_PORT}/healthz >/dev/null || exit 1 + +ENTRYPOINT ["/sbin/tini", "--"] +CMD ["node_modules/.bin/tsx", "gateway/src/index.ts"] diff --git a/gateway/package.json b/gateway/package.json new file mode 100644 index 0000000..660b7f8 --- /dev/null +++ b/gateway/package.json @@ -0,0 +1,28 @@ +{ + "name": "@marketplace/gateway", + "version": "0.0.0", + "private": true, + "type": "module", + "scripts": { + "typecheck": "tsc --noEmit", + "start": "tsx src/index.ts", + "dev": "tsx src/index.ts", + "rotate-master-key": "tsx src/cli/rotate-master-key.ts" + }, + "dependencies": { + "@marketplace/buyer": "workspace:*", + "@marketplace/shared": "workspace:*", + "@noble/ed25519": "^2.3.0", + "@noble/hashes": "^1.8.0", + "bech32": "^2.0.0", + "better-sqlite3": "^11.9.1", + "express": "^4.21.2" + }, + "devDependencies": { + "@types/better-sqlite3": "^7.6.12", + "@types/express": "^5.0.3", + "supertest": "^7.0.0", + "tsx": "^4.19.0", + "typescript": "^5.7.2" + } +} diff --git a/gateway/src/account/routes.ts b/gateway/src/account/routes.ts new file mode 100644 index 0000000..5f9357a --- /dev/null +++ b/gateway/src/account/routes.ts @@ -0,0 +1,155 @@ +/** + * gateway/src/account/routes.ts — signup, account, withdraw. + * + * POST /signup → mint a custodial wallet + API key (key shown once) + * GET /account → balance, collateral readiness, spend, recent usage + * POST /account/withdraw → move unspent AP3X out to an external address (exit) + */ + +import { randomBytes, randomUUID } from "crypto"; +import type { Request, Response } from "express"; +import { buildWithdrawTx } from "@marketplace/shared/tx/server"; +import type { GatewayDeps } from "../deps.js"; +import { genPrivKeyHex, deriveWalletKey } from "../wallet.js"; +import { seal } from "../crypto/seal.js"; +import { hashApiKey, requireKey } from "../middleware/apiKeyAuth.js"; +import { asyncHandler } from "../middleware/http.js"; +import { totalLovelace, hasCollateral } from "../onchain/preflight.js"; +import { badRequest } from "../openai/errors.js"; + +const ACTIVE_ESCROW_STATES = new Set(["Open", "Claimed", "Submitted"]); + +function ap3x(lovelace: bigint): string { + return (Number(lovelace) / 1e6).toFixed(2); +} + +// ─── signup ────────────────────────────────────────────────────────────────── + +export function makeSignupHandler(deps: GatewayDeps) { + return asyncHandler(async (req: Request, res: Response) => { + const body = (req.body ?? {}) as Record; + const label = typeof body.label === "string" ? body.label.slice(0, 120) : null; + + const privHex = genPrivKeyHex(); + const walletKey = deriveWalletKey(privHex, deps.config.networkId); + const sealed = seal(privHex, deps.config.masterKeyHex); + + const rawKey = `vk_${deps.config.networkId === 1 ? "live" : "test"}_${randomBytes(24).toString("hex")}`; + const keyPrefix = rawKey.slice(0, 12); + + deps.store.insertKey({ + id: randomUUID(), + key_hash: hashApiKey(rawKey), + key_prefix: keyPrefix, + label, + wallet_pkh: walletKey.pubKeyHash, + deposit_address: walletKey.address, + enc_priv_nonce: sealed.nonce, + enc_priv_ct: sealed.ct, + enc_priv_tag: sealed.tag, + master_key_version: 1, + created_at: Date.now(), + }); + + res.status(201).json({ + api_key: rawKey, + key_prefix: keyPrefix, + deposit_address: walletKey.address, + note: "Save api_key now — it is shown only once. Fund deposit_address with AP3X to use the gateway.", + }); + }); +} + +// ─── account ───────────────────────────────────────────────────────────────── + +export function makeAccountHandler(deps: GatewayDeps) { + return asyncHandler(async (req: Request, res: Response) => { + const keyRow = requireKey(req); + const utxos = await deps.chain.queryUtxosByAddress(keyRow.deposit_address); + const available = totalLovelace(utxos); + + let lockedInEscrow = 0n; + try { + const r = await deps.fetchFn(`${deps.config.indexerUrl}/escrows?buyer=${keyRow.wallet_pkh}`); + if (r.ok) { + const rows = (await r.json()) as Array>; + if (Array.isArray(rows)) { + for (const row of rows) { + if (!ACTIVE_ESCROW_STATES.has(String(row.state))) continue; + lockedInEscrow += + BigInt(String(row.payment_lovelace ?? "0")) + + BigInt(String(row.buyer_bond_lovelace ?? "0")) + + BigInt(String(row.supplier_bond_lovelace ?? "0")); + } + } + } + } catch { + /* best-effort; report 0 locked on indexer hiccup */ + } + + res.status(200).json({ + key_prefix: keyRow.key_prefix, + deposit_address: keyRow.deposit_address, + balance: { + available_lovelace: available.toString(), + locked_in_escrow_lovelace: lockedInEscrow.toString(), + ap3x: ap3x(available), + }, + collateral_ok: hasCollateral(utxos), + spend: { + total_cost_lovelace: deps.store.sumCostLovelace(keyRow.id).toString(), + request_count: deps.store.countUsage(keyRow.id), + }, + recent_usage: deps.store.listUsage(keyRow.id, 20).map((u) => ({ + created_at: u.created_at, + kind: u.kind, + model: u.model, + status: u.status, + cost_lovelace: u.cost_lovelace, + escrow_ref: u.escrow_ref, + failure_reason: u.failure_reason, + })), + }); + }); +} + +// ─── withdraw ──────────────────────────────────────────────────────────────── + +export function makeWithdrawHandler(deps: GatewayDeps) { + return asyncHandler(async (req: Request, res: Response) => { + const keyRow = requireKey(req); + const body = (req.body ?? {}) as Record; + + const toAddress = body.to_address; + if (typeof toAddress !== "string" || !/^addr(_test)?1[0-9a-z]+$/.test(toAddress)) { + throw badRequest("invalid_address", "`to_address` must be a bech32 Cardano address"); + } + + let amountLovelace: bigint | undefined; + if (body.amount_lovelace !== undefined && body.amount_lovelace !== null) { + try { + amountLovelace = BigInt(String(body.amount_lovelace)); + } catch { + throw badRequest("invalid_amount", "`amount_lovelace` must be an integer (string or number)"); + } + if (amountLovelace <= 0n) throw badRequest("invalid_amount", "`amount_lovelace` must be positive"); + } + + const ctx = deps.registry.getContext(keyRow); + await ctx.mutex.run(async () => { + const built = await buildWithdrawTx({ + chain: deps.chain, + walletKey: ctx.walletKey, + toAddress, + amountLovelace, + }); + await deps.chain.awaitTx(built.expectedTxHash, 120_000); + res.status(200).json({ + status: "submitted", + tx_hash: built.expectedTxHash, + amount_lovelace: built.amountLovelace.toString(), + to_address: built.toAddress, + }); + }); + }); +} diff --git a/gateway/src/chain.ts b/gateway/src/chain.ts new file mode 100644 index 0000000..8221230 --- /dev/null +++ b/gateway/src/chain.ts @@ -0,0 +1,16 @@ +/** + * gateway/src/chain.ts — the single shared LiveOgmiosProvider. + * + * The gateway always submits real transactions, so it always uses a Live + * provider (config enforces LIVE_CHAIN=1). One provider is shared across all + * per-key Marketplace instances; ChainProvider reads are concurrency-safe. + */ + +import { LiveOgmiosProvider } from "@marketplace/shared/chain"; +import type { GatewayConfig } from "./config.js"; + +export function buildChainProvider(config: GatewayConfig): LiveOgmiosProvider { + // eslint-disable-next-line no-console + console.log("[gateway] chain: LiveOgmiosProvider"); + return new LiveOgmiosProvider({ ogmiosUrl: config.ogmiosUrl }); +} diff --git a/gateway/src/cli/rotate-master-key.ts b/gateway/src/cli/rotate-master-key.ts new file mode 100644 index 0000000..1579d9e --- /dev/null +++ b/gateway/src/cli/rotate-master-key.ts @@ -0,0 +1,52 @@ +/** + * gateway/src/cli/rotate-master-key.ts — re-encrypt all custodial keys. + * + * Decrypts every api_keys row with the OLD master key and re-encrypts with the + * NEW one, bumping master_key_version. Run with the gateway stopped. + * + * GATEWAY_DB_DIR=./data/gateway \ + * pnpm --filter @marketplace/gateway rotate-master-key \ + * --old <64hex-old> --new <64hex-new> + */ + +import { GatewayStore } from "../db/store.js"; +import { seal, open as unseal } from "../crypto/seal.js"; + +const HEX64 = /^[0-9a-fA-F]{64}$/; + +function arg(name: string): string | undefined { + const i = process.argv.indexOf(`--${name}`); + return i >= 0 ? process.argv[i + 1] : undefined; +} + +function main(): void { + const oldKey = arg("old") ?? process.env.GATEWAY_MASTER_KEY_OLD; + const newKey = arg("new") ?? process.env.GATEWAY_MASTER_KEY; + const dbDir = arg("db") ?? process.env.GATEWAY_DB_DIR ?? "./data/gateway"; + + if (!oldKey || !HEX64.test(oldKey)) throw new Error("rotate: --old must be 64 hex chars"); + if (!newKey || !HEX64.test(newKey)) throw new Error("rotate: --new must be 64 hex chars"); + if (oldKey === newKey) throw new Error("rotate: --old and --new are identical"); + + const store = new GatewayStore(dbDir); + const rows = store.listAllKeys(); + let rotated = 0; + for (const row of rows) { + const priv = unseal( + { nonce: row.enc_priv_nonce, ct: row.enc_priv_ct, tag: row.enc_priv_tag }, + oldKey, + ); + const sealed = seal(priv, newKey); + store.updateKeyEncryption(row.id, { + enc_priv_nonce: sealed.nonce, + enc_priv_ct: sealed.ct, + enc_priv_tag: sealed.tag, + master_key_version: row.master_key_version + 1, + }); + rotated += 1; + } + // eslint-disable-next-line no-console + console.log(`[gateway] rotated ${rotated} key(s) at ${dbDir}; restart the gateway with the NEW GATEWAY_MASTER_KEY.`); +} + +main(); diff --git a/gateway/src/config.ts b/gateway/src/config.ts new file mode 100644 index 0000000..db59951 --- /dev/null +++ b/gateway/src/config.ts @@ -0,0 +1,113 @@ +/** + * gateway/src/config.ts — env loading for the OpenAI-compatible gateway. + * + * Mirrors buyer/src/config.ts conventions (requireField, HEX64_RE, LIVE_CHAIN + * semantics) but for a multi-tenant, custodial service. See docs/gateway.md. + * + * Required: + * GATEWAY_MASTER_KEY — 64-char hex (32 bytes). AES-256-GCM key wrapping every + * user's custodial wallet private key at rest. Losing it + * permanently locks all user funds. + * INDEXER_URL — internal indexer base URL. + * OGMIOS_URL — Ogmios HTTP endpoint (required; the gateway submits txs). + * + * Optional: + * LIVE_CHAIN — must be "1" (the gateway always needs a live provider). + * NETWORK_ID — "0" testnet (default) | "1" mainnet. + * GATEWAY_PORT — default 3010. + * GATEWAY_DB_DIR — default "./data/gateway". + * SIGNUP_RATE_MAX / SIGNUP_RATE_WINDOW_MS — per-IP signup limiter. + * KEY_RATE_MAX / KEY_RATE_WINDOW_MS — per-key request limiter. + * SWEEPER_INTERVAL_MS / WALLET_HEALTH_INTERVAL_MS — background tick cadences. + * SDK_REGISTRY_MAX — max cached per-key Marketplace instances (LRU). + */ + +const HEX64_RE = /^[0-9a-fA-F]{64}$/; +const POS_INT_RE = /^[1-9]\d*$/; + +export interface GatewayConfig { + masterKeyHex: string; + indexerUrl: string; + ogmiosUrl: string; + networkId: 0 | 1; + liveChain: boolean; + port: number; + dbDir: string; + signupRate: { max: number; windowMs: number }; + keyRate: { max: number; windowMs: number }; + sweeperIntervalMs: number; + walletHealthIntervalMs: number; + sdkRegistryMax: number; +} + +function requireField(env: Record, name: string): string { + const v = env[name]; + if (v === undefined || v === null || v === "") { + throw new Error(`loadConfig: missing required env var ${name}`); + } + return v; +} + +function posInt( + env: Record, + name: string, + fallback: number, +): number { + const v = env[name]; + if (v === undefined || v === "") return fallback; + if (!POS_INT_RE.test(v)) { + throw new Error(`loadConfig: ${name} must be a positive integer`); + } + return Number(v); +} + +export function loadConfig(env: Record): GatewayConfig { + const masterKeyHex = requireField(env, "GATEWAY_MASTER_KEY"); + if (!HEX64_RE.test(masterKeyHex)) { + throw new Error( + "loadConfig: GATEWAY_MASTER_KEY must be 64 hex chars (32 bytes); generate with: openssl rand -hex 32", + ); + } + + const indexerUrl = requireField(env, "INDEXER_URL").replace(/\/+$/, ""); + + // The gateway always posts/settles escrows, so a live provider is mandatory. + const liveChain = env.LIVE_CHAIN === "1"; + const ogmiosUrl = env.OGMIOS_URL ?? ""; + if (!liveChain) { + throw new Error("loadConfig: LIVE_CHAIN=1 is required (the gateway submits transactions)"); + } + if (ogmiosUrl === "") { + throw new Error("loadConfig: OGMIOS_URL is required when LIVE_CHAIN=1"); + } + + let networkId: 0 | 1 = 0; + const networkIdStr = env.NETWORK_ID; + if (networkIdStr !== undefined && networkIdStr !== "") { + if (networkIdStr !== "0" && networkIdStr !== "1") { + throw new Error('loadConfig: NETWORK_ID must be "0" (testnet) or "1" (mainnet)'); + } + networkId = networkIdStr === "1" ? 1 : 0; + } + + return { + masterKeyHex, + indexerUrl, + ogmiosUrl, + networkId, + liveChain, + port: posInt(env, "GATEWAY_PORT", 3010), + dbDir: env.GATEWAY_DB_DIR ?? "./data/gateway", + signupRate: { + max: posInt(env, "SIGNUP_RATE_MAX", 5), + windowMs: posInt(env, "SIGNUP_RATE_WINDOW_MS", 60 * 60 * 1000), + }, + keyRate: { + max: posInt(env, "KEY_RATE_MAX", 60), + windowMs: posInt(env, "KEY_RATE_WINDOW_MS", 60 * 1000), + }, + sweeperIntervalMs: posInt(env, "SWEEPER_INTERVAL_MS", 60 * 1000), + walletHealthIntervalMs: posInt(env, "WALLET_HEALTH_INTERVAL_MS", 10 * 60 * 1000), + sdkRegistryMax: posInt(env, "SDK_REGISTRY_MAX", 500), + }; +} diff --git a/gateway/src/crypto/seal.ts b/gateway/src/crypto/seal.ts new file mode 100644 index 0000000..e249b4e --- /dev/null +++ b/gateway/src/crypto/seal.ts @@ -0,0 +1,46 @@ +/** + * gateway/src/crypto/seal.ts — AES-256-GCM sealing for custodial private keys. + * + * Each user's wallet private key (64-char hex) is encrypted at rest with the + * GATEWAY_MASTER_KEY (32 bytes). A fresh 12-byte nonce per row; the 16-byte GCM + * auth tag is stored alongside. Plaintext is produced only in-memory at request + * time (open()) and never persisted or logged. + */ + +import { createCipheriv, createDecipheriv, randomBytes } from "crypto"; + +export interface Sealed { + /** 12-byte GCM nonce, hex. */ + nonce: string; + /** ciphertext, hex. */ + ct: string; + /** 16-byte GCM auth tag, hex. */ + tag: string; +} + +function masterKeyBuffer(masterKeyHex: string): Buffer { + const buf = Buffer.from(masterKeyHex, "hex"); + if (buf.length !== 32) { + throw new Error("seal: master key must decode to exactly 32 bytes"); + } + return buf; +} + +/** Encrypt `plaintext` (e.g. a wallet privKey hex) under the master key. */ +export function seal(plaintext: string, masterKeyHex: string): Sealed { + const key = masterKeyBuffer(masterKeyHex); + const nonce = randomBytes(12); + const cipher = createCipheriv("aes-256-gcm", key, nonce); + const ct = Buffer.concat([cipher.update(plaintext, "utf8"), cipher.final()]); + const tag = cipher.getAuthTag(); + return { nonce: nonce.toString("hex"), ct: ct.toString("hex"), tag: tag.toString("hex") }; +} + +/** Decrypt a Sealed blob under the master key. Throws on tamper / wrong key. */ +export function open(sealed: Sealed, masterKeyHex: string): string { + const key = masterKeyBuffer(masterKeyHex); + const decipher = createDecipheriv("aes-256-gcm", key, Buffer.from(sealed.nonce, "hex")); + decipher.setAuthTag(Buffer.from(sealed.tag, "hex")); + const pt = Buffer.concat([decipher.update(Buffer.from(sealed.ct, "hex")), decipher.final()]); + return pt.toString("utf8"); +} diff --git a/gateway/src/db/schema.ts b/gateway/src/db/schema.ts new file mode 100644 index 0000000..1524b17 --- /dev/null +++ b/gateway/src/db/schema.ts @@ -0,0 +1,61 @@ +/** + * gateway/src/db/schema.ts — SQLite schema for the gateway. + * + * Three tables: + * api_keys — one custodial wallet per key; the wallet priv is AES-GCM sealed + * (enc_priv_*). The raw API key is NEVER stored — only sha256(key). + * usage — one row per billable unit (a one-shot completion or a closed chat + * session), for /account spend reporting and audit. + * sessions — open chat sessions (one escrow per session), for streaming turns + * and sweeper recovery of abandoned sessions. + */ + +export const CREATE_TABLES_SQL = ` +CREATE TABLE IF NOT EXISTS api_keys ( + id TEXT PRIMARY KEY, + key_hash TEXT NOT NULL UNIQUE, + key_prefix TEXT NOT NULL, + label TEXT, + wallet_pkh TEXT NOT NULL, + deposit_address TEXT NOT NULL, + enc_priv_nonce TEXT NOT NULL, + enc_priv_ct TEXT NOT NULL, + enc_priv_tag TEXT NOT NULL, + master_key_version INTEGER NOT NULL DEFAULT 1, + created_at INTEGER NOT NULL, + disabled INTEGER NOT NULL DEFAULT 0 +); +CREATE INDEX IF NOT EXISTS idx_keys_hash ON api_keys(key_hash); + +CREATE TABLE IF NOT EXISTS usage ( + id TEXT PRIMARY KEY, + key_id TEXT NOT NULL, + created_at INTEGER NOT NULL, + kind TEXT NOT NULL, + model TEXT, + capability_id TEXT, + supplier_pkh TEXT, + escrow_ref TEXT, + cost_lovelace TEXT, + prompt_tokens INTEGER NOT NULL DEFAULT 0, + completion_tokens INTEGER NOT NULL DEFAULT 0, + status TEXT NOT NULL, + failure_reason TEXT +); +CREATE INDEX IF NOT EXISTS idx_usage_key ON usage(key_id, created_at DESC); + +CREATE TABLE IF NOT EXISTS sessions ( + id TEXT PRIMARY KEY, + key_id TEXT NOT NULL, + escrow_ref TEXT NOT NULL, + session_nonce TEXT NOT NULL, + supplier_base_url TEXT NOT NULL, + supplier_pkh TEXT NOT NULL, + model TEXT NOT NULL, + price_lovelace TEXT NOT NULL, + state TEXT NOT NULL, + opened_at INTEGER NOT NULL, + closed_at INTEGER +); +CREATE INDEX IF NOT EXISTS idx_sessions_key ON sessions(key_id, opened_at DESC); +`; diff --git a/gateway/src/db/store.ts b/gateway/src/db/store.ts new file mode 100644 index 0000000..3010a8e --- /dev/null +++ b/gateway/src/db/store.ts @@ -0,0 +1,196 @@ +/** + * gateway/src/db/store.ts — synchronous SQLite store (mirrors buyer/src/db/archive.ts). + * + * better-sqlite3 + WAL + prepared statements. Low write volume (≤1 row per + * request) so synchronous is fine and simpler than async. + */ + +import { mkdirSync } from "fs"; +import { join, resolve } from "path"; +import Database from "better-sqlite3"; +import type { Database as BetterDatabase } from "better-sqlite3"; +import { CREATE_TABLES_SQL } from "./schema.js"; + +export interface ApiKeyRow { + id: string; + key_hash: string; + key_prefix: string; + label: string | null; + wallet_pkh: string; + deposit_address: string; + enc_priv_nonce: string; + enc_priv_ct: string; + enc_priv_tag: string; + master_key_version: number; + created_at: number; + disabled: number; +} + +export interface UsageRow { + id: string; + key_id: string; + created_at: number; + kind: string; + model: string | null; + capability_id: string | null; + supplier_pkh: string | null; + escrow_ref: string | null; + cost_lovelace: string | null; + prompt_tokens: number; + completion_tokens: number; + status: string; + failure_reason: string | null; +} + +export interface SessionRow { + id: string; + key_id: string; + escrow_ref: string; + session_nonce: string; + supplier_base_url: string; + supplier_pkh: string; + model: string; + price_lovelace: string; + state: string; + opened_at: number; + closed_at: number | null; +} + +export class GatewayStore { + private readonly db: BetterDatabase; + + private readonly sInsertKey; + private readonly sGetKeyByHash; + private readonly sGetKeyById; + private readonly sListKeys; + private readonly sListAllKeys; + private readonly sUpdateKeyEncryption; + private readonly sInsertUsage; + private readonly sSumCost; + private readonly sCountByKey; + private readonly sListUsage; + private readonly sInsertSession; + private readonly sGetSession; + private readonly sSetSessionState; + private readonly sListOpenSessions; + + constructor(dbDir: string) { + const dir = resolve(dbDir); + mkdirSync(dir, { recursive: true }); + this.db = new Database(join(dir, "gateway.db")); + this.db.pragma("journal_mode = WAL"); + this.db.exec(CREATE_TABLES_SQL); + + this.sInsertKey = this.db.prepare(` + INSERT INTO api_keys ( + id, key_hash, key_prefix, label, wallet_pkh, deposit_address, + enc_priv_nonce, enc_priv_ct, enc_priv_tag, master_key_version, created_at, disabled + ) VALUES ( + @id, @key_hash, @key_prefix, @label, @wallet_pkh, @deposit_address, + @enc_priv_nonce, @enc_priv_ct, @enc_priv_tag, @master_key_version, @created_at, 0 + ) + `); + this.sGetKeyByHash = this.db.prepare(`SELECT * FROM api_keys WHERE key_hash = @key_hash`); + this.sGetKeyById = this.db.prepare(`SELECT * FROM api_keys WHERE id = @id`); + this.sListKeys = this.db.prepare(`SELECT * FROM api_keys WHERE disabled = 0`); + this.sListAllKeys = this.db.prepare(`SELECT * FROM api_keys`); + this.sUpdateKeyEncryption = this.db.prepare(` + UPDATE api_keys SET enc_priv_nonce = @enc_priv_nonce, enc_priv_ct = @enc_priv_ct, + enc_priv_tag = @enc_priv_tag, master_key_version = @master_key_version WHERE id = @id + `); + + this.sInsertUsage = this.db.prepare(` + INSERT INTO usage ( + id, key_id, created_at, kind, model, capability_id, supplier_pkh, escrow_ref, + cost_lovelace, prompt_tokens, completion_tokens, status, failure_reason + ) VALUES ( + @id, @key_id, @created_at, @kind, @model, @capability_id, @supplier_pkh, @escrow_ref, + @cost_lovelace, @prompt_tokens, @completion_tokens, @status, @failure_reason + ) + `); + this.sSumCost = this.db.prepare(` + SELECT COALESCE(SUM(CAST(cost_lovelace AS INTEGER)), 0) AS total + FROM usage WHERE key_id = @key_id AND status = 'completed' + `); + this.sCountByKey = this.db.prepare(`SELECT COUNT(*) AS n FROM usage WHERE key_id = @key_id`); + this.sListUsage = this.db.prepare(` + SELECT * FROM usage WHERE key_id = @key_id ORDER BY created_at DESC LIMIT @limit + `); + + this.sInsertSession = this.db.prepare(` + INSERT INTO sessions ( + id, key_id, escrow_ref, session_nonce, supplier_base_url, supplier_pkh, + model, price_lovelace, state, opened_at, closed_at + ) VALUES ( + @id, @key_id, @escrow_ref, @session_nonce, @supplier_base_url, @supplier_pkh, + @model, @price_lovelace, @state, @opened_at, NULL + ) + `); + this.sGetSession = this.db.prepare(`SELECT * FROM sessions WHERE id = @id`); + this.sSetSessionState = this.db.prepare(` + UPDATE sessions SET state = @state, closed_at = @closed_at WHERE id = @id + `); + this.sListOpenSessions = this.db.prepare(`SELECT * FROM sessions WHERE state = 'open'`); + } + + insertKey(row: Omit): void { + this.sInsertKey.run({ ...row, label: row.label ?? null }); + } + + getKeyByHash(keyHash: string): ApiKeyRow | undefined { + return this.sGetKeyByHash.get({ key_hash: keyHash }) as ApiKeyRow | undefined; + } + + getKeyById(id: string): ApiKeyRow | undefined { + return this.sGetKeyById.get({ id }) as ApiKeyRow | undefined; + } + + listKeys(): ApiKeyRow[] { + return this.sListKeys.all() as ApiKeyRow[]; + } + + listAllKeys(): ApiKeyRow[] { + return this.sListAllKeys.all() as ApiKeyRow[]; + } + + updateKeyEncryption( + id: string, + enc: { enc_priv_nonce: string; enc_priv_ct: string; enc_priv_tag: string; master_key_version: number }, + ): void { + this.sUpdateKeyEncryption.run({ id, ...enc }); + } + + insertUsage(row: UsageRow): void { + this.sInsertUsage.run(row); + } + + sumCostLovelace(keyId: string): bigint { + const r = this.sSumCost.get({ key_id: keyId }) as { total: number } | undefined; + return BigInt(r?.total ?? 0); + } + + countUsage(keyId: string): number { + const r = this.sCountByKey.get({ key_id: keyId }) as { n: number } | undefined; + return r?.n ?? 0; + } + + listUsage(keyId: string, limit: number): UsageRow[] { + return this.sListUsage.all({ key_id: keyId, limit }) as UsageRow[]; + } + + insertSession(row: Omit): void { + this.sInsertSession.run(row); + } + + getSession(id: string): SessionRow | undefined { + return this.sGetSession.get({ id }) as SessionRow | undefined; + } + + setSessionState(id: string, state: string, closedAt: number | null): void { + this.sSetSessionState.run({ id, state, closed_at: closedAt }); + } + + listOpenSessions(): SessionRow[] { + return this.sListOpenSessions.all() as SessionRow[]; + } +} diff --git a/gateway/src/deps.ts b/gateway/src/deps.ts new file mode 100644 index 0000000..9975cc5 --- /dev/null +++ b/gateway/src/deps.ts @@ -0,0 +1,16 @@ +/** + * gateway/src/deps.ts — shared dependency bundle passed to route factories. + */ + +import type { LiveOgmiosProvider } from "@marketplace/shared/chain"; +import type { GatewayConfig } from "./config.js"; +import type { GatewayStore } from "./db/store.js"; +import type { SdkRegistry } from "./sdk/registry.js"; + +export interface GatewayDeps { + config: GatewayConfig; + store: GatewayStore; + chain: LiveOgmiosProvider; + registry: SdkRegistry; + fetchFn: typeof globalThis.fetch; +} diff --git a/gateway/src/index.ts b/gateway/src/index.ts new file mode 100644 index 0000000..50367f6 --- /dev/null +++ b/gateway/src/index.ts @@ -0,0 +1,70 @@ +/** + * gateway/src/index.ts — entry point. + * + * Wires config → shared chain + store → SdkRegistry → Express app, and starts + * the reclaim sweeper + wallet-health ticker. Boot is wrapped in runMain() so + * importing this module (tests) does not start the server. + */ + +import { loadConfig } from "./config.js"; +import { buildChainProvider } from "./chain.js"; +import { GatewayStore } from "./db/store.js"; +import { SdkRegistry } from "./sdk/registry.js"; +import { createApp } from "./server.js"; +import { startSweeper } from "./sweeper.js"; +import { startWalletHealthTicker } from "./walletHealth.js"; +import type { GatewayDeps } from "./deps.js"; + +export function buildDeps(env: Record): GatewayDeps { + const config = loadConfig(env); + const store = new GatewayStore(config.dbDir); + const chain = buildChainProvider(config); + const registry = new SdkRegistry({ + chain, + indexerUrl: config.indexerUrl, + networkId: config.networkId, + masterKeyHex: config.masterKeyHex, + max: config.sdkRegistryMax, + }); + return { config, store, chain, registry, fetchFn: globalThis.fetch }; +} + +export function runMain(env: Record): void { + const deps = buildDeps(env); + const app = createApp(deps); + + const stopSweeper = startSweeper(deps); + const stopHealth = startWalletHealthTicker(deps); + + const server = app.listen(deps.config.port, () => { + // eslint-disable-next-line no-console + console.log( + `[gateway] listening on :${deps.config.port} (network=${deps.config.networkId}, indexer=${deps.config.indexerUrl})`, + ); + }); + + const shutdown = (): void => { + // eslint-disable-next-line no-console + console.log("[gateway] shutting down..."); + stopSweeper(); + stopHealth(); + server.close(() => process.exit(0)); + }; + process.once("SIGINT", shutdown); + process.once("SIGTERM", shutdown); +} + +const invokedDirectly = + typeof process !== "undefined" && + process.argv?.[1] !== undefined && + /gateway\/(src|dist)\/index\.(t|j)s$/.test(process.argv[1]); + +if (invokedDirectly) { + try { + runMain(process.env); + } catch (err) { + // eslint-disable-next-line no-console + console.error("[gateway] boot failed:", err); + process.exit(1); + } +} diff --git a/gateway/src/middleware/apiKeyAuth.ts b/gateway/src/middleware/apiKeyAuth.ts new file mode 100644 index 0000000..f533013 --- /dev/null +++ b/gateway/src/middleware/apiKeyAuth.ts @@ -0,0 +1,45 @@ +/** + * gateway/src/middleware/apiKeyAuth.ts — Bearer API-key authentication. + * + * Looks up sha256(rawKey) in api_keys (unique index). The raw key is never + * stored. Attaches the matched row to req.gatewayKey for downstream handlers. + */ + +import { createHash } from "crypto"; +import type { Request, Response, NextFunction } from "express"; +import type { GatewayStore, ApiKeyRow } from "../db/store.js"; +import { unauthorized } from "../openai/errors.js"; +import { sendError } from "./http.js"; + +export interface GatewayRequest extends Request { + gatewayKey?: ApiKeyRow; +} + +export function hashApiKey(raw: string): string { + return createHash("sha256").update(raw, "utf8").digest("hex"); +} + +export function makeApiKeyAuth(store: GatewayStore) { + return (req: Request, res: Response, next: NextFunction): void => { + const header = req.header("authorization") ?? ""; + const m = /^Bearer\s+(.+)$/i.exec(header.trim()); + if (!m) { + sendError(res, unauthorized("missing or malformed Authorization: Bearer header")); + return; + } + const row = store.getKeyByHash(hashApiKey(m[1].trim())); + if (!row || row.disabled) { + sendError(res, unauthorized()); + return; + } + (req as GatewayRequest).gatewayKey = row; + next(); + }; +} + +/** Read the authenticated key off the request (throws if middleware missing). */ +export function requireKey(req: Request): ApiKeyRow { + const row = (req as GatewayRequest).gatewayKey; + if (!row) throw unauthorized(); + return row; +} diff --git a/gateway/src/middleware/http.ts b/gateway/src/middleware/http.ts new file mode 100644 index 0000000..8db95f8 --- /dev/null +++ b/gateway/src/middleware/http.ts @@ -0,0 +1,25 @@ +/** + * gateway/src/middleware/http.ts — Express response/error helpers. + */ + +import type { Request, Response } from "express"; +import { GatewayError, toGatewayError, toErrorBody } from "../openai/errors.js"; + +/** Send a GatewayError (or any error) as an OpenAI-shaped JSON error response. */ +export function sendError(res: Response, err: unknown): void { + const ge = err instanceof GatewayError ? err : toGatewayError(err); + if (res.headersSent) { + res.end(); + return; + } + res.status(ge.httpStatus).json(toErrorBody(ge)); +} + +/** Wrap an async handler so rejections are mapped to error responses. */ +export function asyncHandler( + fn: (req: Request, res: Response) => Promise, +): (req: Request, res: Response) => void { + return (req, res) => { + fn(req, res).catch((err) => sendError(res, err)); + }; +} diff --git a/gateway/src/middleware/rateLimit.ts b/gateway/src/middleware/rateLimit.ts new file mode 100644 index 0000000..6d47cf5 --- /dev/null +++ b/gateway/src/middleware/rateLimit.ts @@ -0,0 +1,66 @@ +/** + * gateway/src/middleware/rateLimit.ts — per-IP and per-key limiters. + * + * Small sliding-window limiter (same shape as buyer/src/auth.ts createRateLimiter, + * inlined to avoid coupling the operator-auth module into the gateway). Used to + * bound public /signup abuse (per IP) and per-key request volume (429 + Retry-After). + */ + +import type { Request, Response, NextFunction } from "express"; +import { rateLimited } from "../openai/errors.js"; +import { sendError } from "./http.js"; +import { requireKey } from "./apiKeyAuth.js"; + +interface Bucket { + count: number; + windowStart: number; +} + +interface Limiter { + attempt(key: string): { allowed: boolean; retryAfterMs: number }; +} + +function createLimiter(opts: { max: number; windowMs: number }): Limiter { + const buckets = new Map(); + return { + attempt(key: string) { + const t = Date.now(); + const existing = buckets.get(key); + if (!existing || t - existing.windowStart >= opts.windowMs) { + buckets.set(key, { count: 1, windowStart: t }); + return { allowed: true, retryAfterMs: 0 }; + } + if (existing.count >= opts.max) { + return { allowed: false, retryAfterMs: Math.max(0, opts.windowMs - (t - existing.windowStart)) }; + } + existing.count += 1; + return { allowed: true, retryAfterMs: 0 }; + }, + }; +} + +type Keyer = (req: Request) => string; + +function makeLimiter(opts: { max: number; windowMs: number }, keyOf: Keyer) { + const limiter = createLimiter(opts); + return (req: Request, res: Response, next: NextFunction): void => { + const decision = limiter.attempt(keyOf(req)); + if (!decision.allowed) { + const retrySec = Math.ceil(decision.retryAfterMs / 1000); + res.setHeader("Retry-After", retrySec.toString()); + sendError(res, rateLimited(`rate limit exceeded; retry in ${retrySec}s`)); + return; + } + next(); + }; +} + +/** Per-IP limiter (for the public /signup endpoint). */ +export function ipRateLimit(opts: { max: number; windowMs: number }) { + return makeLimiter(opts, (req) => req.ip ?? "unknown"); +} + +/** Per-API-key limiter (apply AFTER apiKeyAuth). */ +export function keyRateLimit(opts: { max: number; windowMs: number }) { + return makeLimiter(opts, (req) => requireKey(req).id); +} diff --git a/gateway/src/onchain/preflight.ts b/gateway/src/onchain/preflight.ts new file mode 100644 index 0000000..da9a475 --- /dev/null +++ b/gateway/src/onchain/preflight.ts @@ -0,0 +1,55 @@ +/** + * gateway/src/onchain/preflight.ts — balance + collateral pre-flight. + * + * The buyer wallet funds the FULL escrow at PostEscrow: price + buyer_bond + + * supplier_bond (postEscrow.ts:132). Accept/Reclaim are Plutus script-spends + * that also need a pure-ADA collateral UTxO ≥ 5 ADA (assertCollateralCandidate). + * So required = price + buyer_bond + supplier_bond + collateral + fee reserve, + * AND the wallet must currently hold a pure-ADA UTxO ≥ collateral. We check both + * up front and return a clean 402 instead of failing mid-flight. + */ + +import type { ChainProvider, Utxo } from "@marketplace/shared/chain"; + +export const COLLATERAL_MIN_LOVELACE = 5_000_000n; +export const FEE_RESERVE_LOVELACE = 2_000_000n; + +export function totalLovelace(utxos: Utxo[]): bigint { + return utxos.reduce((sum, u) => sum + u.lovelace, 0n); +} + +/** True iff some UTxO is pure-ADA (no native assets) and ≥ collateral floor. */ +export function hasCollateral(utxos: Utxo[]): boolean { + return utxos.some( + (u) => u.lovelace >= COLLATERAL_MIN_LOVELACE && Object.keys(u.assets).length === 0, + ); +} + +export interface PreflightResult { + ok: boolean; + availableLovelace: bigint; + requiredLovelace: bigint; + collateralOk: boolean; +} + +export async function preflight( + chain: ChainProvider, + address: string, + cost: { priceLovelace: bigint; buyerBondLovelace: bigint; supplierBondLovelace: bigint }, +): Promise { + const utxos = await chain.queryUtxosByAddress(address); + const available = totalLovelace(utxos); + const collateralOk = hasCollateral(utxos); + const required = + cost.priceLovelace + + cost.buyerBondLovelace + + cost.supplierBondLovelace + + COLLATERAL_MIN_LOVELACE + + FEE_RESERVE_LOVELACE; + return { + ok: available >= required && collateralOk, + availableLovelace: available, + requiredLovelace: required, + collateralOk, + }; +} diff --git a/gateway/src/onchain/settle.ts b/gateway/src/onchain/settle.ts new file mode 100644 index 0000000..630e6a9 --- /dev/null +++ b/gateway/src/onchain/settle.ts @@ -0,0 +1,123 @@ +/** + * gateway/src/onchain/settle.ts — settlement + recovery helpers. + * + * submitPrompt returns the ORIGINAL Open escrow ref, but the supplier's + * Claim→Submit moves the live escrow to a NEW ref. To Accept (settle) we must + * resolve the current Submitted ref via the indexer (mirrors buyer /v1/accept: + * match the lineage by posted_at). Because the per-key mutex serializes a + * wallet's escrows, at settle time the wallet has at most one in-flight escrow, + * so a lone Submitted row is unambiguously ours. + * + * acceptResult/reclaim on the SDK are fire-and-forget; here we buildAcceptTx / + * buildReclaimTx and AWAIT confirmation so a returned success means settled. + */ + +import type { ChainProvider, OutputReference } from "@marketplace/shared/chain"; +import type { WalletKey } from "@marketplace/shared/tx"; +import { buildAcceptTx, buildReclaimTx } from "@marketplace/shared/tx"; +import { parseRef } from "../routing/selectSupplier.js"; +import { GatewayError } from "../openai/errors.js"; + +export interface EscrowRow { + utxo_ref: string; + state: string; + posted_at: number; +} + +export async function fetchEscrows( + indexerUrl: string, + buyerPkh: string, + fetchFn: typeof globalThis.fetch, +): Promise { + const res = await fetchFn(`${indexerUrl}/escrows?buyer=${buyerPkh}`); + if (!res.ok) throw new Error(`indexer /escrows returned ${res.status}`); + const body = (await res.json()) as unknown; + if (!Array.isArray(body)) throw new Error("indexer /escrows did not return an array"); + return body as EscrowRow[]; +} + +function sleep(ms: number): Promise { + return new Promise((r) => setTimeout(r, ms)); +} + +/** Poll the indexer for the current Submitted UTxO of our in-flight escrow. */ +export async function resolveSubmittedRef(opts: { + indexerUrl: string; + buyerPkh: string; + originalRefStr: string; + fetchFn?: typeof globalThis.fetch; + timeoutMs?: number; + intervalMs?: number; +}): Promise { + const fetchFn = opts.fetchFn ?? globalThis.fetch; + const timeoutMs = opts.timeoutMs ?? 60_000; + const intervalMs = opts.intervalMs ?? 2_000; + const deadline = Date.now() + timeoutMs; + + let lastErr: unknown; + for (;;) { + try { + const rows = await fetchEscrows(opts.indexerUrl, opts.buyerPkh, fetchFn); + + // 1. Direct: the original ref already shows Submitted. + const direct = rows.find((r) => r.utxo_ref === opts.originalRefStr && r.state === "Submitted"); + if (direct) return refOrThrow(direct.utxo_ref); + + // 2. Lineage: find the original row to read posted_at, then the Submitted + // row sharing it (the validator preserves posted_at across states). + const lineage = rows.find((r) => r.utxo_ref === opts.originalRefStr); + if (lineage) { + const submitted = rows.find((r) => r.posted_at === lineage.posted_at && r.state === "Submitted"); + if (submitted) return refOrThrow(submitted.utxo_ref); + } + + // 3. Fallback: the mutex guarantees ≤1 in-flight escrow for this wallet, + // so a single Submitted row is ours. + const submittedRows = rows.filter((r) => r.state === "Submitted"); + if (submittedRows.length === 1) return refOrThrow(submittedRows[0].utxo_ref); + } catch (err) { + lastErr = err; + } + if (Date.now() >= deadline) { + throw new GatewayError( + 502, + "server_error", + "escrow_settle_failed", + `could not resolve Submitted escrow within ${timeoutMs}ms${lastErr ? `: ${String(lastErr)}` : ""}`, + ); + } + await sleep(intervalMs); + } +} + +function refOrThrow(refStr: string): OutputReference { + const ref = parseRef(refStr); + if (!ref) { + throw new GatewayError(502, "server_error", "escrow_settle_failed", `bad escrow ref ${refStr}`); + } + return ref; +} + +/** Build + submit + await the Accept tx (Submitted → Accepted, terminal). */ +export async function acceptAndConfirm( + chain: ChainProvider, + walletKey: WalletKey, + escrowRef: OutputReference, + awaitTimeoutMs = 120_000, +): Promise { + const built = await buildAcceptTx({ chain, buyerKey: walletKey, escrowRef }); + await chain.awaitTx(built.expectedTxHash, awaitTimeoutMs); + return built.expectedTxHash; +} + +/** Build + submit + await a Reclaim tx (Open/Claimed → Reclaimed). */ +export async function reclaimAndConfirm( + chain: ChainProvider, + walletKey: WalletKey, + escrowRef: OutputReference, + awaitTimeoutMs = 120_000, +): Promise { + const built = await buildReclaimTx({ chain, buyerKey: walletKey, escrowRef }); + await chain.awaitTx(built.expectedTxHash, awaitTimeoutMs); + return built.expectedTxHash; +} diff --git a/gateway/src/openai/chatCompletions.ts b/gateway/src/openai/chatCompletions.ts new file mode 100644 index 0000000..7a90a75 --- /dev/null +++ b/gateway/src/openai/chatCompletions.ts @@ -0,0 +1,223 @@ +/** + * gateway/src/openai/chatCompletions.ts — POST /openai/v1/chat/completions. + * + * One escrow per call against an llm.text.generate.v1 supplier. Lifecycle (all + * inside the per-key mutex): route → preflight → submitPrompt → resolve the + * Submitted ref + Accept (awaited) → record usage → wallet-health → respond. + * stream:true returns the completion as a buffered pseudo-stream (one delta + + * final usage chunk + [DONE]) with keepalive comments during the escrow wait. + */ + +import { randomUUID } from "crypto"; +import type { Request, Response } from "express"; +import { TxConstructionError } from "@marketplace/shared/tx"; +import type { GatewayDeps } from "../deps.js"; +import type { ApiKeyRow, GatewayStore } from "../db/store.js"; +import type { KeyContext } from "../sdk/registry.js"; +import { requireKey } from "../middleware/apiKeyAuth.js"; +import { asyncHandler } from "../middleware/http.js"; +import { selectCandidates } from "../routing/selectSupplier.js"; +import { preflight } from "../onchain/preflight.js"; +import { resolveSubmittedRef, acceptAndConfirm } from "../onchain/settle.js"; +import { ensureWalletHealthy } from "../walletHealth.js"; +import { parseChatRequest, type ParsedChatRequest } from "./validate.js"; +import { + genId, + buildChatCompletion, + buildChunk, + usageFromReceipt, + sseData, + SSE_DONE, + type Usage, +} from "./shapes.js"; +import { notFound, paymentRequired, toGatewayError, toErrorBody } from "./errors.js"; + +const CAPABILITY = "llm.text.generate.v1"; + +function recordUsage( + store: GatewayStore, + p: { + keyId: string; + model: string; + supplierPkh: string | null; + escrowRef: string | null; + costLovelace: bigint | null; + usage: Usage | null; + status: "completed" | "failed"; + failureReason?: string; + }, +): void { + store.insertUsage({ + id: randomUUID(), + key_id: p.keyId, + created_at: Date.now(), + kind: "completion", + model: p.model, + capability_id: CAPABILITY, + supplier_pkh: p.supplierPkh, + escrow_ref: p.escrowRef, + cost_lovelace: p.costLovelace !== null ? p.costLovelace.toString() : null, + prompt_tokens: p.usage?.prompt_tokens ?? 0, + completion_tokens: p.usage?.completion_tokens ?? 0, + status: p.status, + failure_reason: p.failureReason ?? null, + }); +} + +export function makeChatCompletionsHandler(deps: GatewayDeps) { + return asyncHandler(async (req: Request, res: Response) => { + const keyRow = requireKey(req); + const parsed = parseChatRequest(req.body); + const ctx = deps.registry.getContext(keyRow); + await ctx.mutex.run(() => runOneShot(deps, keyRow, ctx, parsed, res)); + }); +} + +async function runOneShot( + deps: GatewayDeps, + keyRow: ApiKeyRow, + ctx: KeyContext, + parsed: ParsedChatRequest, + res: Response, +): Promise { + const { config, store, chain, fetchFn } = deps; + + const candidates = await selectCandidates({ + indexerUrl: config.indexerUrl, + model: parsed.model, + capabilityId: CAPABILITY, + fetchFn, + }); + if (candidates.length === 0) { + throw notFound("model_not_found", `no available supplier for model "${parsed.model}"`); + } + + const primary = candidates[0]; + const pf = await preflight(chain, ctx.walletKey.address, primary); + if (!pf.ok) { + throw paymentRequired( + pf.collateralOk + ? "insufficient balance for this request" + : "wallet has no pure-ADA UTxO ≥ 5 ADA for collateral; deposit a little more ADA", + { + x_vector: { + required_lovelace: pf.requiredLovelace.toString(), + available_lovelace: pf.availableLovelace.toString(), + collateral_ok: pf.collateralOk, + deposit_address: ctx.walletKey.address, + }, + }, + ); + } + + const id = genId(); + let keepalive: ReturnType | undefined; + if (parsed.stream) { + res.status(200); + res.setHeader("Content-Type", "text/event-stream"); + res.setHeader("Cache-Control", "no-cache, no-transform"); + res.setHeader("Connection", "keep-alive"); + res.setHeader("X-Accel-Buffering", "no"); + res.write(": connected\n\n"); + keepalive = setInterval(() => { + try { + res.write(": keepalive\n\n"); + } catch { + /* client gone */ + } + }, 10_000); + } + + try { + // Submit. Fall back to the next supplier ONLY on a pre-post + // TxConstructionError (no escrow was posted). A SupplierError means the + // escrow is already posted — do not retry (the sweeper recovers it). + let result: Awaited> | undefined; + let used: (typeof candidates)[number] | undefined; + let lastErr: unknown; + for (const cand of candidates) { + try { + result = await ctx.sdk.submitPrompt({ + advertRef: cand.advertRef, + messages: parsed.messages, + payment_lovelace: cand.priceLovelace, + max_output_tokens: parsed.maxTokens, + }); + used = cand; + break; + } catch (err) { + lastErr = err; + if (err instanceof TxConstructionError) continue; + throw err; + } + } + if (!result || !used) { + throw toGatewayError(lastErr ?? new Error("no supplier could be engaged")); + } + + // Settle: resolve the live Submitted ref, Accept, await confirmation. + const originalRefStr = `${result.escrowRef.txHash}#${result.escrowRef.index}`; + const submittedRef = await resolveSubmittedRef({ + indexerUrl: config.indexerUrl, + buyerPkh: ctx.walletKey.pubKeyHash, + originalRefStr, + fetchFn, + }); + await acceptAndConfirm(chain, ctx.walletKey, submittedRef); + + const usage = usageFromReceipt(result.receipt); + const vector = { + receipt: result.receipt, + receipt_signature: result.receiptSignature, + escrow_ref: originalRefStr, + }; + recordUsage(store, { + keyId: keyRow.id, + model: parsed.model, + supplierPkh: used.supplierPkh, + escrowRef: originalRefStr, + costLovelace: used.priceLovelace, + usage, + status: "completed", + }); + + // Re-shape the wallet ({collateral, working}) so the next request keeps a + // valid collateral UTxO. Best-effort; serialized by the mutex. + await ensureWalletHealthy(chain, ctx.walletKey).catch((e) => + // eslint-disable-next-line no-console + console.error("[gateway] wallet-health after completion:", e instanceof Error ? e.message : e), + ); + + if (parsed.stream) { + if (keepalive) clearInterval(keepalive); + res.write(sseData(buildChunk({ id, model: parsed.model, delta: { role: "assistant", content: result.response }, finishReason: null }))); + res.write(sseData(buildChunk({ id, model: parsed.model, delta: {}, finishReason: "stop", usage, vector }))); + res.write(SSE_DONE); + res.end(); + } else { + res.status(200).json(buildChatCompletion({ id, model: parsed.model, content: result.response, usage, vector })); + } + } catch (err) { + if (keepalive) clearInterval(keepalive); + recordUsage(store, { + keyId: keyRow.id, + model: parsed.model, + supplierPkh: null, + escrowRef: null, + costLovelace: null, + usage: null, + status: "failed", + failureReason: err instanceof Error ? err.message : String(err), + }); + if (parsed.stream && res.headersSent) { + try { + res.write(sseData(toErrorBody(toGatewayError(err)))); + } catch { + /* client gone */ + } + res.end(); + return; + } + throw err; + } +} diff --git a/gateway/src/openai/errors.ts b/gateway/src/openai/errors.ts new file mode 100644 index 0000000..493d2da --- /dev/null +++ b/gateway/src/openai/errors.ts @@ -0,0 +1,76 @@ +/** + * gateway/src/openai/errors.ts — OpenAI-shaped error taxonomy. + * + * OpenAI SDKs parse error.type to drive retry/severity, so every gateway error + * is mapped to a concrete {message, type, code} triple and the documented HTTP + * status. Marketplace SDK errors (SupplierError/IndexerError/ReceiptVerification + * Error) and TxConstructionError are translated here. + */ + +import { ReceiptVerificationError, IndexerError, SupplierError } from "@marketplace/buyer/sdk"; +import { TxConstructionError } from "@marketplace/shared/tx"; + +export type OpenAiErrorType = + | "invalid_request_error" + | "authentication_error" + | "rate_limit_error" + | "server_error"; + +export class GatewayError extends Error { + constructor( + public readonly httpStatus: number, + public readonly type: OpenAiErrorType, + public readonly code: string, + message: string, + /** Optional vendor extension fields merged alongside `error` in the body. */ + public readonly extra?: Record, + ) { + super(message); + this.name = "GatewayError"; + } +} + +export const badRequest = (code: string, message: string): GatewayError => + new GatewayError(400, "invalid_request_error", code, message); +export const unauthorized = (message = "invalid API key"): GatewayError => + new GatewayError(401, "authentication_error", "invalid_api_key", message); +export const paymentRequired = (message: string, extra?: Record): GatewayError => + new GatewayError(402, "invalid_request_error", "insufficient_funds", message, extra); +export const notFound = (code: string, message: string): GatewayError => + new GatewayError(404, "invalid_request_error", code, message); +export const rateLimited = (message: string): GatewayError => + new GatewayError(429, "rate_limit_error", "rate_limit_exceeded", message); + +/** Render the body OpenAI clients expect: {error:{message,type,code,param}}. */ +export function toErrorBody(err: GatewayError): Record { + return { + error: { message: err.message, type: err.type, code: err.code, param: null }, + ...(err.extra ?? {}), + }; +} + +/** Map any thrown value to a GatewayError. */ +export function toGatewayError(err: unknown): GatewayError { + if (err instanceof GatewayError) return err; + + if (err instanceof SupplierError) { + if (err.reason === "timeout") { + return new GatewayError(504, "server_error", "supplier_timeout", "supplier timed out"); + } + if (err.reason === "supplier_busy") { + return new GatewayError(503, "server_error", "overloaded", "all matching suppliers are busy"); + } + return new GatewayError(502, "server_error", "upstream_error", `supplier error: ${err.reason}`); + } + if (err instanceof ReceiptVerificationError) { + return new GatewayError(502, "server_error", "receipt_verification_failed", `receipt invalid: ${err.reason}`); + } + if (err instanceof IndexerError) { + return new GatewayError(502, "server_error", "indexer_error", `indexer error: ${err.reason}`); + } + if (err instanceof TxConstructionError) { + return new GatewayError(502, "server_error", "escrow_failed", `escrow build failed: ${err.reason}`); + } + const message = err instanceof Error ? err.message : String(err); + return new GatewayError(500, "server_error", "internal_error", message); +} diff --git a/gateway/src/openai/models.ts b/gateway/src/openai/models.ts new file mode 100644 index 0000000..1e58c24 --- /dev/null +++ b/gateway/src/openai/models.ts @@ -0,0 +1,18 @@ +/** + * gateway/src/openai/models.ts — GET /openai/v1/models. + * + * Distinct models advertised by Active suppliers, in the OpenAI list shape. + */ + +import type { Request, Response } from "express"; +import type { GatewayDeps } from "../deps.js"; +import { asyncHandler } from "../middleware/http.js"; +import { listModels } from "../routing/selectSupplier.js"; +import { buildModelsList } from "./shapes.js"; + +export function makeModelsHandler(deps: GatewayDeps) { + return asyncHandler(async (_req: Request, res: Response) => { + const models = await listModels({ indexerUrl: deps.config.indexerUrl, fetchFn: deps.fetchFn }); + res.status(200).json(buildModelsList(models)); + }); +} diff --git a/gateway/src/openai/sessions.ts b/gateway/src/openai/sessions.ts new file mode 100644 index 0000000..132355f --- /dev/null +++ b/gateway/src/openai/sessions.ts @@ -0,0 +1,353 @@ +/** + * gateway/src/openai/sessions.ts — chat sessions (Vector extension). + * + * One escrow per SESSION (fixed price locked at open, settled at close). Real + * token streaming per turn against an llm.chat.v1 supplier. Endpoints: + * POST /openai/v1/chat/sessions → open (startChat) + * POST /openai/v1/chat/sessions/:id/messages → one streamed turn + * POST /openai/v1/chat/sessions/:id/close → settle (endChat: Submit+Accept) + * + * The chat.v1 supplier has no system-role channel and accumulates the transcript + * server-side, so we send one user `content` per turn and mirror the transcript + * in-memory (for endChat's response_hash verification). A process restart loses + * the mirror; an abandoned session is reclaimed by the sweeper. + */ + +import { randomUUID } from "crypto"; +import type { Request, Response } from "express"; +import { TxConstructionError, type ChatMessage } from "@marketplace/shared/tx"; +import { SupplierError } from "@marketplace/buyer/sdk"; +import type { GatewayDeps } from "../deps.js"; +import type { ApiKeyRow, GatewayStore, SessionRow } from "../db/store.js"; +import type { KeyContext } from "../sdk/registry.js"; +import { requireKey } from "../middleware/apiKeyAuth.js"; +import { asyncHandler } from "../middleware/http.js"; +import { selectCandidates, parseRef } from "../routing/selectSupplier.js"; +import { preflight } from "../onchain/preflight.js"; +import { ensureWalletHealthy } from "../walletHealth.js"; +import { badRequest, notFound, paymentRequired, toGatewayError, toErrorBody } from "./errors.js"; +import { + genId, + nowSec, + buildChatCompletion, + buildChunk, + sseData, + SSE_DONE, + type Usage, +} from "./shapes.js"; + +const CAPABILITY = "llm.chat.v1"; + +/** In-memory per-session transcript mirror (sessionId → messages). */ +const transcripts = new Map(); + +function refStr(ref: { txHash: string; index: number }): string { + return `${ref.txHash}#${ref.index}`; +} + +// ─── open ────────────────────────────────────────────────────────────────── + +export function makeOpenSessionHandler(deps: GatewayDeps) { + return asyncHandler(async (req: Request, res: Response) => { + const keyRow = requireKey(req); + const body = (req.body ?? {}) as Record; + const model = body.model; + if (typeof model !== "string" || model === "") { + throw badRequest("invalid_model", "`model` is required"); + } + const ctx = deps.registry.getContext(keyRow); + await ctx.mutex.run(() => openSession(deps, keyRow, ctx, model, res)); + }); +} + +async function openSession( + deps: GatewayDeps, + keyRow: ApiKeyRow, + ctx: KeyContext, + model: string, + res: Response, +): Promise { + const candidates = await selectCandidates({ + indexerUrl: deps.config.indexerUrl, + model, + capabilityId: CAPABILITY, + fetchFn: deps.fetchFn, + }); + if (candidates.length === 0) { + throw notFound("model_not_found", `no available chat (llm.chat.v1) supplier for model "${model}"`); + } + const primary = candidates[0]; + const pf = await preflight(deps.chain, ctx.walletKey.address, primary); + if (!pf.ok) { + throw paymentRequired( + pf.collateralOk ? "insufficient balance to open a session" : "wallet has no ≥5 ADA pure-ADA collateral UTxO", + { + x_vector: { + required_lovelace: pf.requiredLovelace.toString(), + available_lovelace: pf.availableLovelace.toString(), + collateral_ok: pf.collateralOk, + deposit_address: ctx.walletKey.address, + }, + }, + ); + } + + // startChat posts the escrow + supplier Claim. Fall back to the next supplier + // only on a pre-post error (TxConstructionError, or a busy supplier caught by + // startChat's /status pre-check before any funds are locked). + let started: Awaited> | undefined; + let used: (typeof candidates)[number] | undefined; + let lastErr: unknown; + for (const cand of candidates) { + try { + started = await ctx.sdk.startChat({ advertRef: cand.advertRef, payment_lovelace: cand.priceLovelace }); + used = cand; + break; + } catch (err) { + lastErr = err; + if (err instanceof TxConstructionError) continue; + if (err instanceof SupplierError && err.reason === "supplier_busy") continue; + throw err; + } + } + if (!started || !used) throw toGatewayError(lastErr ?? new Error("no chat supplier could be engaged")); + + const sessionId = randomUUID(); + deps.store.insertSession({ + id: sessionId, + key_id: keyRow.id, + escrow_ref: refStr(started.escrowRef), + session_nonce: started.sessionNonce, + supplier_base_url: started.supplierBaseUrl, + supplier_pkh: used.supplierPkh, + model, + price_lovelace: used.priceLovelace.toString(), + state: "open", + opened_at: Date.now(), + }); + transcripts.set(sessionId, []); + + res.status(200).json({ id: sessionId, object: "chat.session", model, created: nowSec() }); +} + +// ─── messages (one streamed turn) ──────────────────────────────────────────── + +export function makeSessionMessageHandler(deps: GatewayDeps) { + return asyncHandler(async (req: Request, res: Response) => { + const keyRow = requireKey(req); + const session = loadOwnedSession(deps.store, req, keyRow); + if (session.state !== "open") { + throw badRequest("session_closed", `session ${session.id} is ${session.state}`); + } + const body = (req.body ?? {}) as Record; + const stream = body.stream !== false; // sessions default to streaming + const content = resolveTurnContent(session.id, body); + if (content === "") throw badRequest("content_required", "no user content in this turn"); + + const ctx = deps.registry.getContext(keyRow); + await ctx.mutex.run(() => streamTurn(deps, session, content, stream, res)); + }); +} + +function resolveTurnContent(sessionId: string, body: Record): string { + if (typeof body.content === "string") { + return foldSystem(sessionId, body.content, body); + } + if (Array.isArray(body.messages)) { + const msgs = body.messages as Array<{ role?: unknown; content?: unknown }>; + const lastUser = [...msgs].reverse().find((m) => m.role === "user" && typeof m.content === "string"); + const userContent = typeof lastUser?.content === "string" ? lastUser.content : ""; + return foldSystem(sessionId, userContent, body); + } + return ""; +} + +/** On the first turn only, prepend any system message (chat.v1 has no system channel). */ +function foldSystem(sessionId: string, content: string, body: Record): string { + const mirror = transcripts.get(sessionId) ?? []; + if (mirror.length > 0) return content; + if (Array.isArray(body.messages)) { + const sys = (body.messages as Array<{ role?: unknown; content?: unknown }>) + .filter((m) => m.role === "system" && typeof m.content === "string") + .map((m) => m.content as string) + .join("\n\n"); + if (sys) return `System: ${sys}\n\n${content}`; + } + return content; +} + +async function streamTurn( + deps: GatewayDeps, + session: SessionRow, + content: string, + stream: boolean, + res: Response, +): Promise { + const id = genId(); + const upstream = await deps.fetchFn(`${session.supplier_base_url.replace(/\/+$/, "")}/v1/chat/message`, { + method: "POST", + headers: { "content-type": "application/json", accept: "text/event-stream", "X-Escrow-Ref": session.escrow_ref }, + body: JSON.stringify({ content }), + }); + if (!upstream.ok || !upstream.body) { + const detail = await upstream.text().catch(() => ""); + throw toGatewayError(new SupplierError("supplier_http_error", { status: upstream.status, message: detail.slice(0, 200) })); + } + + if (stream) { + res.status(200); + res.setHeader("Content-Type", "text/event-stream"); + res.setHeader("Cache-Control", "no-cache, no-transform"); + res.setHeader("Connection", "keep-alive"); + res.setHeader("X-Accel-Buffering", "no"); + res.write(sseData(buildChunk({ id, model: session.model, delta: { role: "assistant" }, finishReason: null }))); + } + + let assistant = ""; + let errored: string | undefined; + await readSupplierSse(upstream.body as ReadableStream, (frame) => { + if (frame.type === "token" && typeof frame.value === "string") { + assistant += frame.value; + if (stream) res.write(sseData(buildChunk({ id, model: session.model, delta: { content: frame.value }, finishReason: null }))); + } else if (frame.type === "error") { + errored = typeof frame.message === "string" ? frame.message : "supplier stream error"; + } + }); + + if (errored !== undefined) { + if (stream) { + res.write(sseData(toErrorBody(toGatewayError(new SupplierError("upstream_error", { message: errored }))))); + res.end(); + return; + } + throw toGatewayError(new SupplierError("upstream_error", { message: errored })); + } + + // Mirror the transcript for endChat's response_hash verification. + const mirror = transcripts.get(session.id) ?? []; + mirror.push({ role: "user", content }); + mirror.push({ role: "assistant", content: assistant }); + transcripts.set(session.id, mirror); + + if (stream) { + res.write(sseData(buildChunk({ id, model: session.model, delta: {}, finishReason: "stop" }))); + res.write(SSE_DONE); + res.end(); + } else { + const usage: Usage = { prompt_tokens: 0, completion_tokens: 0, total_tokens: 0 }; + res.status(200).json(buildChatCompletion({ id, model: session.model, content: assistant, usage })); + } +} + +interface SupplierFrame { + type?: string; + value?: string; + message?: string; +} + +/** Read a supplier SSE stream of {type:token|done|error} frames. */ +async function readSupplierSse( + body: ReadableStream, + onFrame: (frame: SupplierFrame) => void, +): Promise { + const reader = body.getReader(); + const decoder = new TextDecoder(); + let buf = ""; + for (;;) { + const { value, done } = await reader.read(); + if (done) break; + buf += decoder.decode(value, { stream: true }); + let sep: number; + while ((sep = buf.indexOf("\n\n")) !== -1) { + const raw = buf.slice(0, sep); + buf = buf.slice(sep + 2); + for (const line of raw.split("\n")) { + const trimmed = line.trim(); + if (!trimmed.startsWith("data:")) continue; + const data = trimmed.slice(5).trim(); + if (data === "" || data === "[DONE]") continue; + try { + onFrame(JSON.parse(data) as SupplierFrame); + } catch { + /* ignore non-JSON keepalive/comment frames */ + } + } + } + } +} + +// ─── close (settle) ────────────────────────────────────────────────────────── + +export function makeCloseSessionHandler(deps: GatewayDeps) { + return asyncHandler(async (req: Request, res: Response) => { + const keyRow = requireKey(req); + const session = loadOwnedSession(deps.store, req, keyRow); + if (session.state !== "open") { + throw badRequest("session_closed", `session ${session.id} is already ${session.state}`); + } + const ctx = deps.registry.getContext(keyRow); + await ctx.mutex.run(() => closeSession(deps, keyRow, ctx, session, res)); + }); +} + +async function closeSession( + deps: GatewayDeps, + keyRow: ApiKeyRow, + ctx: KeyContext, + session: SessionRow, + res: Response, +): Promise { + const escrowRef = parseRef(session.escrow_ref); + if (!escrowRef) throw badRequest("bad_session", "session has an invalid escrow ref"); + + const result = await ctx.sdk.endChat({ + escrowRef, + sessionNonce: session.session_nonce, + supplierBaseUrl: session.supplier_base_url, + transcript: transcripts.get(session.id), + }); + + deps.store.setSessionState(session.id, "closed", Date.now()); + transcripts.delete(session.id); + + const prompt = result.receipt.prompt_tokens ?? 0; + const completion = result.receipt.completion_tokens ?? 0; + deps.store.insertUsage({ + id: randomUUID(), + key_id: keyRow.id, + created_at: Date.now(), + kind: "chat_session", + model: session.model, + capability_id: CAPABILITY, + supplier_pkh: session.supplier_pkh, + escrow_ref: session.escrow_ref, + cost_lovelace: session.price_lovelace, + prompt_tokens: prompt, + completion_tokens: completion, + status: "completed", + failure_reason: null, + }); + + await ensureWalletHealthy(deps.chain, ctx.walletKey).catch((e) => + // eslint-disable-next-line no-console + console.error("[gateway] wallet-health after session close:", e instanceof Error ? e.message : e), + ); + + res.status(200).json({ + status: "closed", + escrow_ref: session.escrow_ref, + accepted_ref: refStr(result.acceptedRef), + model: session.model, + x_vector: { receipt: result.receipt, receipt_signature: result.receiptSignature, escrow_ref: session.escrow_ref }, + }); +} + +function loadOwnedSession(store: GatewayStore, req: Request, keyRow: ApiKeyRow): SessionRow { + const raw = (req.params as Record).id; + const id = typeof raw === "string" ? raw : undefined; + const session = id ? store.getSession(id) : undefined; + if (!session || session.key_id !== keyRow.id) { + throw notFound("session_not_found", `no chat session ${id ?? "(none)"}`); + } + return session; +} diff --git a/gateway/src/openai/shapes.ts b/gateway/src/openai/shapes.ts new file mode 100644 index 0000000..89d99b4 --- /dev/null +++ b/gateway/src/openai/shapes.ts @@ -0,0 +1,106 @@ +/** + * gateway/src/openai/shapes.ts — OpenAI response object builders. + * + * Produces spec-clean ChatCompletion / ChatCompletionChunk / model-list objects + * plus the `x_vector` receipt extension. SSE helpers frame chunks for streaming. + */ + +import { randomBytes } from "crypto"; +import type { Receipt } from "@marketplace/shared/receipt"; + +export function genId(): string { + return `chatcmpl-${randomBytes(16).toString("hex")}`; +} + +export function nowSec(): number { + return Math.floor(Date.now() / 1000); +} + +export interface VectorReceipt { + receipt: Receipt; + receipt_signature: string; + escrow_ref: string; +} + +export interface Usage { + prompt_tokens: number; + completion_tokens: number; + total_tokens: number; +} + +export function usageFromReceipt(receipt: Receipt): Usage { + const prompt = receipt.prompt_tokens ?? 0; + const completion = receipt.completion_tokens ?? 0; + return { prompt_tokens: prompt, completion_tokens: completion, total_tokens: prompt + completion }; +} + +/** Non-streaming chat.completion object. */ +export function buildChatCompletion(args: { + id: string; + model: string; + content: string; + usage: Usage; + vector?: VectorReceipt; +}): Record { + return { + id: args.id, + object: "chat.completion", + created: nowSec(), + model: args.model, + choices: [ + { + index: 0, + message: { role: "assistant", content: args.content }, + finish_reason: "stop", + }, + ], + usage: args.usage, + ...(args.vector ? { x_vector: args.vector } : {}), + }; +} + +/** A streaming chat.completion.chunk. `delta` is the incremental content (or {}). */ +export function buildChunk(args: { + id: string; + model: string; + delta: { role?: string; content?: string } | Record; + finishReason: "stop" | null; + usage?: Usage; + vector?: VectorReceipt; +}): Record { + return { + id: args.id, + object: "chat.completion.chunk", + created: nowSec(), + model: args.model, + choices: [{ index: 0, delta: args.delta, finish_reason: args.finishReason }], + ...(args.usage ? { usage: args.usage } : {}), + ...(args.vector ? { x_vector: args.vector } : {}), + }; +} + +/** Frame any object as an SSE `data:` event. */ +export function sseData(obj: unknown): string { + return `data: ${JSON.stringify(obj)}\n\n`; +} + +export const SSE_DONE = "data: [DONE]\n\n"; + +export function buildModelsList(models: string[]): Record { + const created = nowSec(); + return { + object: "list", + data: models.map((id) => ({ id, object: "model", created, owned_by: "vector-marketplace" })), + }; +} + +/** Render OpenAI messages[] into a single prompt string for the chat.v1 session + * supplier (which has no system-role channel — see docs/gateway.md). */ +export function renderMessages(messages: Array<{ role: string; content: string }>): string { + return messages + .map((m) => { + const role = m.role === "assistant" ? "Assistant" : m.role === "system" ? "System" : "User"; + return `${role}: ${m.content}`; + }) + .join("\n\n"); +} diff --git a/gateway/src/openai/validate.ts b/gateway/src/openai/validate.ts new file mode 100644 index 0000000..1616997 --- /dev/null +++ b/gateway/src/openai/validate.ts @@ -0,0 +1,64 @@ +/** + * gateway/src/openai/validate.ts — request body validation for chat endpoints. + */ + +import type { ChatMessage } from "@marketplace/shared/tx"; +import { badRequest } from "./errors.js"; + +export interface ParsedChatRequest { + model: string; + messages: ChatMessage[]; + maxTokens?: number; + stream: boolean; +} + +const ROLES = new Set(["system", "user", "assistant"]); + +/** Validate an OpenAI chat.completions-style body. Rejects tools/functions; + * silently ignores temperature/n/stop/top_p/etc (documented). */ +export function parseChatRequest(body: unknown): ParsedChatRequest { + if (typeof body !== "object" || body === null) { + throw badRequest("invalid_body", "request body must be a JSON object"); + } + const b = body as Record; + + if ("tools" in b || "tool_choice" in b || "functions" in b || "function_call" in b) { + throw badRequest("unsupported_parameter", "tools and function-calling are not supported by this gateway"); + } + + const model = b.model; + if (typeof model !== "string" || model === "") { + throw badRequest("invalid_model", "`model` is required and must be a non-empty string"); + } + + const rawMessages = b.messages; + if (!Array.isArray(rawMessages) || rawMessages.length === 0) { + throw badRequest("invalid_messages", "`messages` is required and must be a non-empty array"); + } + const messages: ChatMessage[] = []; + for (const m of rawMessages) { + if ( + typeof m !== "object" || + m === null || + typeof (m as { role?: unknown }).role !== "string" || + typeof (m as { content?: unknown }).content !== "string" || + !ROLES.has((m as { role: string }).role) + ) { + throw badRequest("invalid_messages", "each message must be {role: system|user|assistant, content: string}"); + } + const mm = m as { role: "system" | "user" | "assistant"; content: string }; + messages.push({ role: mm.role, content: mm.content }); + } + + let maxTokens: number | undefined; + const mt = b.max_tokens ?? b.max_completion_tokens; + if (mt !== undefined && mt !== null) { + if (typeof mt !== "number" || !Number.isInteger(mt) || mt <= 0) { + throw badRequest("invalid_max_tokens", "`max_tokens` must be a positive integer"); + } + maxTokens = mt; + } + + const stream = b.stream === true; + return { model, messages, maxTokens, stream }; +} diff --git a/gateway/src/routing/selectSupplier.ts b/gateway/src/routing/selectSupplier.ts new file mode 100644 index 0000000..a5f61eb --- /dev/null +++ b/gateway/src/routing/selectSupplier.ts @@ -0,0 +1,119 @@ +/** + * gateway/src/routing/selectSupplier.ts — capability + model routing. + * + * Queries the indexer for suppliers advertising the requested model under the + * required capability. Per the design: match by capability + model, pick an + * AVAILABLE supplier (status free|unknown), NOT the cheapest. We return all + * eligible candidates so the caller can fall back to the next one when the + * chosen supplier turns out to be busy/offline at escrow-post time. + * + * `status==="unknown"` is treated as eligible: the indexer poller (~20s) leaves + * freshly-advertised suppliers as "unknown" until first polled, and excluding + * them would create artificial availability gaps. + */ + +import type { OutputReference } from "@marketplace/shared/chain"; + +/** Subset of the indexer's SupplierView we depend on. */ +interface SupplierView { + utxo_ref: string; + supplier_pkh: string; + capability_id: string; + model: string; + max_output_tokens: number; + price_lovelace: string; + supplier_bond_lovelace: string; + buyer_bond_lovelace: string; + endpoint_url: string; + advert_status: string; + status: string; +} + +export interface SupplierCandidate { + advertRef: OutputReference; + utxoRef: string; + supplierPkh: string; + model: string; + capabilityId: string; + endpointUrl: string; + priceLovelace: bigint; + buyerBondLovelace: bigint; + supplierBondLovelace: bigint; + maxOutputTokens: number; + status: string; +} + +const ESCROW_REF_RE = /^([0-9a-fA-F]{64})#(\d+)$/; + +export function parseRef(ref: string): OutputReference | null { + const m = ESCROW_REF_RE.exec(ref); + if (!m) return null; + return { txHash: m[1], index: Number(m[2]) }; +} + +export interface SelectSupplierOpts { + indexerUrl: string; + model: string; + capabilityId: string; + fetchFn?: typeof globalThis.fetch; +} + +/** + * Return eligible candidates for (model, capability), ordered for fallback. + * Eligible = advert Active, model matches, live status ∈ {free, unknown}. + */ +export async function selectCandidates(opts: SelectSupplierOpts): Promise { + const fetchFn = opts.fetchFn ?? globalThis.fetch; + const url = `${opts.indexerUrl}/suppliers?capability_id=${encodeURIComponent(opts.capabilityId)}`; + const res = await fetchFn(url); + if (!res.ok) { + throw new Error(`indexer /suppliers returned ${res.status}`); + } + const body = (await res.json()) as unknown; + if (!Array.isArray(body)) { + throw new Error("indexer /suppliers did not return an array"); + } + + const candidates: SupplierCandidate[] = []; + for (const raw of body as SupplierView[]) { + if (raw.capability_id !== opts.capabilityId) continue; + if (raw.model !== opts.model) continue; + if (raw.advert_status !== "Active") continue; + if (raw.status !== "free" && raw.status !== "unknown") continue; + const advertRef = parseRef(raw.utxo_ref); + if (!advertRef) continue; + candidates.push({ + advertRef, + utxoRef: raw.utxo_ref, + supplierPkh: raw.supplier_pkh, + model: raw.model, + capabilityId: raw.capability_id, + endpointUrl: raw.endpoint_url, + priceLovelace: BigInt(raw.price_lovelace), + buyerBondLovelace: BigInt(raw.buyer_bond_lovelace), + supplierBondLovelace: BigInt(raw.supplier_bond_lovelace), + maxOutputTokens: raw.max_output_tokens, + status: raw.status, + }); + } + // Prefer known-free over unknown so we route to confirmed-available first. + candidates.sort((a, b) => (a.status === "free" ? 0 : 1) - (b.status === "free" ? 0 : 1)); + return candidates; +} + +/** Distinct models across Active suppliers (for GET /openai/v1/models). */ +export async function listModels(opts: { + indexerUrl: string; + fetchFn?: typeof globalThis.fetch; +}): Promise { + const fetchFn = opts.fetchFn ?? globalThis.fetch; + const res = await fetchFn(`${opts.indexerUrl}/suppliers`); + if (!res.ok) throw new Error(`indexer /suppliers returned ${res.status}`); + const body = (await res.json()) as unknown; + if (!Array.isArray(body)) throw new Error("indexer /suppliers did not return an array"); + const models = new Set(); + for (const raw of body as SupplierView[]) { + if (raw.advert_status === "Active" && typeof raw.model === "string") models.add(raw.model); + } + return [...models].sort(); +} diff --git a/gateway/src/sdk/registry.ts b/gateway/src/sdk/registry.ts new file mode 100644 index 0000000..e4cb172 --- /dev/null +++ b/gateway/src/sdk/registry.ts @@ -0,0 +1,89 @@ +/** + * gateway/src/sdk/registry.ts — per-API-key Marketplace SDK cache + serializer. + * + * Each API key is its own custodial wallet, so each gets its own Marketplace + * instance bound to that wallet's WalletKey. The shared ChainProvider and + * indexer URL are reused across all keys. A per-key Mutex serializes all + * on-chain work for one wallet (a single wallet cannot post two escrows from + * the same UTxO concurrently). The cache is LRU-bounded to cap memory. + */ + +import { Marketplace, MemoryTaskHistoryStore } from "@marketplace/buyer/sdk"; +import type { LiveOgmiosProvider } from "@marketplace/shared/chain"; +import type { WalletKey } from "@marketplace/shared/tx"; +import { open as unseal } from "../crypto/seal.js"; +import { deriveWalletKey } from "../wallet.js"; +import type { ApiKeyRow } from "../db/store.js"; + +/** Tiny promise-chain mutex: run(fn) executes fns strictly one at a time. */ +export class Mutex { + private tail: Promise = Promise.resolve(); + + run(fn: () => Promise): Promise { + const result = this.tail.then(fn, fn); + // Keep the chain alive even if fn rejects; swallow here so the next + // run() isn't poisoned by a prior rejection. + this.tail = result.then( + () => undefined, + () => undefined, + ); + return result; + } +} + +export interface KeyContext { + sdk: Marketplace; + mutex: Mutex; + walletKey: WalletKey; +} + +export interface SdkRegistryDeps { + chain: LiveOgmiosProvider; + indexerUrl: string; + networkId: 0 | 1; + masterKeyHex: string; + max: number; +} + +export class SdkRegistry { + private readonly cache = new Map(); + + constructor(private readonly deps: SdkRegistryDeps) {} + + /** Build (or fetch from cache) the per-key SDK context. Decrypts the wallet + * key in-memory only when first constructing the instance. */ + getContext(keyRow: ApiKeyRow): KeyContext { + const existing = this.cache.get(keyRow.id); + if (existing) { + // LRU bump: re-insert to move to the end of iteration order. + this.cache.delete(keyRow.id); + this.cache.set(keyRow.id, existing); + return existing; + } + + const privHex = unseal( + { nonce: keyRow.enc_priv_nonce, ct: keyRow.enc_priv_ct, tag: keyRow.enc_priv_tag }, + this.deps.masterKeyHex, + ); + const walletKey = deriveWalletKey(privHex, this.deps.networkId); + const sdk = new Marketplace({ + chain: this.deps.chain, + indexerUrl: this.deps.indexerUrl, + walletKey, + networkParams: { networkId: this.deps.networkId }, + historyStore: new MemoryTaskHistoryStore(), + }); + const ctx: KeyContext = { sdk, mutex: new Mutex(), walletKey }; + this.cache.set(keyRow.id, ctx); + this.evictIfNeeded(); + return ctx; + } + + private evictIfNeeded(): void { + while (this.cache.size > this.deps.max) { + const oldest = this.cache.keys().next().value as string | undefined; + if (oldest === undefined) break; + this.cache.delete(oldest); + } + } +} diff --git a/gateway/src/server.ts b/gateway/src/server.ts new file mode 100644 index 0000000..74650a6 --- /dev/null +++ b/gateway/src/server.ts @@ -0,0 +1,63 @@ +/** + * gateway/src/server.ts — Express app wiring. + * + * Public: GET /healthz, GET /, POST /signup (IP rate-limited). + * Bearer: GET /account, POST /account/withdraw, + * POST /openai/v1/chat/completions, GET /openai/v1/models, + * POST /openai/v1/chat/sessions[/:id/messages|/close]. + */ + +import express, { type Express, type Request, type Response } from "express"; +import type { GatewayDeps } from "./deps.js"; +import { makeApiKeyAuth } from "./middleware/apiKeyAuth.js"; +import { ipRateLimit, keyRateLimit } from "./middleware/rateLimit.js"; +import { sendError } from "./middleware/http.js"; +import { notFound } from "./openai/errors.js"; +import { makeChatCompletionsHandler } from "./openai/chatCompletions.js"; +import { makeModelsHandler } from "./openai/models.js"; +import { + makeOpenSessionHandler, + makeSessionMessageHandler, + makeCloseSessionHandler, +} from "./openai/sessions.js"; +import { makeSignupHandler, makeAccountHandler, makeWithdrawHandler } from "./account/routes.js"; +import { INDEX_HTML } from "./ui/page.js"; + +export function createApp(deps: GatewayDeps): Express { + const app = express(); + // Behind traefik/Cloudflare — trust the proxy chain so req.ip is the client. + app.set("trust proxy", true); + app.use(express.json({ limit: "1mb" })); + + app.get("/healthz", (_req: Request, res: Response) => { + res.json({ ok: true }); + }); + + app.get("/", (_req: Request, res: Response) => { + res.status(200).type("html").send(INDEX_HTML); + }); + + // Public signup, rate-limited per IP. + app.post("/signup", ipRateLimit(deps.config.signupRate), makeSignupHandler(deps)); + + // Bearer-gated routes: auth → per-key rate limit → handler. + const auth = makeApiKeyAuth(deps.store); + const krl = keyRateLimit(deps.config.keyRate); + const gated = [auth, krl] as const; + + app.get("/account", auth, makeAccountHandler(deps)); + app.post("/account/withdraw", auth, makeWithdrawHandler(deps)); + + app.post("/openai/v1/chat/completions", ...gated, makeChatCompletionsHandler(deps)); + app.get("/openai/v1/models", auth, makeModelsHandler(deps)); + app.post("/openai/v1/chat/sessions", ...gated, makeOpenSessionHandler(deps)); + app.post("/openai/v1/chat/sessions/:id/messages", ...gated, makeSessionMessageHandler(deps)); + app.post("/openai/v1/chat/sessions/:id/close", ...gated, makeCloseSessionHandler(deps)); + + // OpenAI-shaped 404 for anything else. + app.use((_req: Request, res: Response) => { + sendError(res, notFound("not_found", "unknown route")); + }); + + return app; +} diff --git a/gateway/src/sweeper.ts b/gateway/src/sweeper.ts new file mode 100644 index 0000000..e3d16e5 --- /dev/null +++ b/gateway/src/sweeper.ts @@ -0,0 +1,82 @@ +/** + * gateway/src/sweeper.ts — state-aware custody safety net. + * + * Per key, classifies on-chain escrows and recovers stranded funds: + * Open|Claimed past the deliver_by deadline → Reclaim (returns funds + bond). + * Submitted (verified, in accept-window) → retry Accept (settle + refund). + * Submitted past the accept-window → leave for supplier Release (log). + * + * reclaim only works on Open/Claimed; Accept only on Submitted — so we branch by + * state and never blind-reclaim a Submitted escrow. The on-chain validator is + * the source of truth for deadlines, so premature/late attempts simply fail and + * are caught here; the age thresholds just avoid racing in-flight requests. + */ + +import type { GatewayDeps } from "./deps.js"; +import { fetchEscrows, acceptAndConfirm, reclaimAndConfirm } from "./onchain/settle.js"; +import { parseRef } from "./routing/selectSupplier.js"; + +// Don't touch escrows newer than this — an in-flight request is settling them. +const MIN_AGE_MS = 90_000; +// Reclaim Open/Claimed escrows older than this (past deliver_by ~10m + buffer). +const RECLAIM_AFTER_MS = 11 * 60 * 1000; + +function log(msg: string): void { + // eslint-disable-next-line no-console + console.log(`[gateway:sweeper] ${msg}`); +} + +export async function runSweepOnce(deps: GatewayDeps): Promise { + const now = Date.now(); + for (const keyRow of deps.store.listKeys()) { + let rows; + try { + rows = await fetchEscrows(deps.config.indexerUrl, keyRow.wallet_pkh, deps.fetchFn); + } catch (err) { + log(`indexer error for ${keyRow.key_prefix}: ${err instanceof Error ? err.message : String(err)}`); + continue; + } + + const ctx = deps.registry.getContext(keyRow); + for (const row of rows) { + const ref = parseRef(row.utxo_ref); + if (!ref) continue; + const ageMs = now - (row.posted_at ?? 0); + if (ageMs < MIN_AGE_MS) continue; + + try { + if (row.state === "Submitted") { + await ctx.mutex.run(() => acceptAndConfirm(deps.chain, ctx.walletKey, ref)); + log(`accepted stranded Submitted escrow ${row.utxo_ref}`); + } else if ((row.state === "Open" || row.state === "Claimed") && ageMs > RECLAIM_AFTER_MS) { + await ctx.mutex.run(() => reclaimAndConfirm(deps.chain, ctx.walletKey, ref)); + log(`reclaimed stranded ${row.state} escrow ${row.utxo_ref}`); + } + } catch { + // Expected when the validator rejects (wrong window/state) or the UTxO + // was already spent by an in-flight request; skip quietly. + } + } + } + + // Mark closed/settled chat sessions whose escrow is no longer active. + // (Best-effort housekeeping; the escrow recovery above is the load-bearing part.) +} + +/** Start the periodic sweeper. Returns a stop() function. */ +export function startSweeper(deps: GatewayDeps): () => void { + let running = false; + const tick = async (): Promise => { + if (running) return; + running = true; + try { + await runSweepOnce(deps); + } catch (err) { + log(`sweep failed: ${err instanceof Error ? err.message : String(err)}`); + } finally { + running = false; + } + }; + const handle = setInterval(() => void tick(), deps.config.sweeperIntervalMs); + return () => clearInterval(handle); +} diff --git a/gateway/src/ui/page.ts b/gateway/src/ui/page.ts new file mode 100644 index 0000000..6e5fbee --- /dev/null +++ b/gateway/src/ui/page.ts @@ -0,0 +1,81 @@ +/** + * gateway/src/ui/page.ts — self-serve web UI, inlined as a string so it ships + * in dist/ (tsc does not copy .html). Plain HTML + fetch, no build step. + */ + +export const INDEX_HTML = ` + + + + +Vector Marketplace — OpenAI Gateway + + + +

Vector Marketplace — OpenAI-compatible Gateway

+

Point any OpenAI SDK at using your API key as the bearer token.

+ +

1. Create an API key

+
+ + +
+
+ +

2. Fund your wallet

+

Send AP3X to your deposit address (you need price + bonds + ~5 ADA collateral + fees).

+ +

3. Account

+
+ + +
+
+ +

4. Withdraw

+
+ + + +
+
+ + + +`; diff --git a/gateway/src/wallet.ts b/gateway/src/wallet.ts new file mode 100644 index 0000000..7ca5844 --- /dev/null +++ b/gateway/src/wallet.ts @@ -0,0 +1,60 @@ +/** + * gateway/src/wallet.ts — per-user wallet key derivation. + * + * Copied from buyer/src/index.ts:24-64 (deriveWalletKey + ed25519 sha512 hook) + * so the gateway does not have to import the buyer entrypoint (which boots an + * Express server). genPrivKeyHex mints a fresh custodial wallet at signup. + */ + +import { createHash, randomBytes } from "crypto"; +import * as ed from "@noble/ed25519"; +import { blake2b } from "@noble/hashes/blake2b"; +import { bech32 } from "bech32"; +import type { WalletKey } from "@marketplace/shared/tx"; + +// Wire ed25519 sha512 hook (idempotent — same as buyer/src/index.ts + receipt/sign.ts). +ed.etc.sha512Sync = (...messages: Uint8Array[]): Uint8Array => { + const h = createHash("sha512"); + for (const m of messages) h.update(m); + return new Uint8Array(h.digest()); +}; + +function hexToBytes(hex: string): Uint8Array { + const bytes = new Uint8Array(hex.length / 2); + for (let i = 0; i < hex.length; i += 2) { + bytes[i / 2] = parseInt(hex.substring(i, i + 2), 16); + } + return bytes; +} + +function bytesToHex(b: Uint8Array): string { + let out = ""; + for (let i = 0; i < b.length; i++) out += b[i].toString(16).padStart(2, "0"); + return out; +} + +/** Mint a fresh 32-byte Ed25519 private key as 64-char hex. */ +export function genPrivKeyHex(): string { + return bytesToHex(new Uint8Array(randomBytes(32))); +} + +export function deriveWalletKey(privHex: string, networkId: 0 | 1): WalletKey { + const priv = hexToBytes(privHex); + const pub = ed.getPublicKey(priv); + const pubHex = bytesToHex(pub); + const pkh = blake2b(pub, { dkLen: 28 }); + const pkhHex = bytesToHex(pkh); + const header = networkId === 0 ? 0x60 : 0x61; + const payload = new Uint8Array(29); + payload[0] = header; + payload.set(pkh, 1); + const words = bech32.toWords(payload); + const hrp = networkId === 0 ? "addr_test" : "addr"; + const addr = bech32.encode(hrp, words, 1023); + return { + pubKeyHash: pkhHex, + pubKeyHex: pubHex, + privateKeyHex: privHex, + address: addr, + }; +} diff --git a/gateway/src/walletHealth.ts b/gateway/src/walletHealth.ts new file mode 100644 index 0000000..d6e3bd3 --- /dev/null +++ b/gateway/src/walletHealth.ts @@ -0,0 +1,49 @@ +/** + * gateway/src/walletHealth.ts — keep each custodial wallet usable. + * + * After every settled request (and on a periodic tick) we re-shape the wallet to + * {≥5 ADA collateral UTxO, working UTxO} via the shared consolidate builder, so + * the NEXT request's Accept/Reclaim script-spend always finds a pure-ADA + * collateral candidate. No-op when the wallet is already healthy. + */ + +import type { LiveOgmiosProvider } from "@marketplace/shared/chain"; +import type { WalletKey } from "@marketplace/shared/tx"; +import { runConsolidateWallet, DEFAULT_COLLATERAL_LOVELACE } from "@marketplace/shared/tx/server"; +import type { GatewayDeps } from "./deps.js"; + +export async function ensureWalletHealthy( + chain: LiveOgmiosProvider, + walletKey: WalletKey, +): Promise { + await runConsolidateWallet({ + chain, + walletKey, + collateralLovelace: DEFAULT_COLLATERAL_LOVELACE, + log: () => { + /* quiet; callers log failures */ + }, + }); +} + +/** Periodically re-shape every wallet so a collateral UTxO is always available. + * Serialized per key via the registry mutex; no-op for already-healthy wallets. */ +export function startWalletHealthTicker(deps: GatewayDeps): () => void { + let running = false; + const tick = async (): Promise => { + if (running) return; + running = true; + try { + for (const keyRow of deps.store.listKeys()) { + const ctx = deps.registry.getContext(keyRow); + await ctx.mutex.run(() => ensureWalletHealthy(deps.chain, ctx.walletKey)).catch(() => { + /* skip wallets that can't consolidate (e.g. drained) */ + }); + } + } finally { + running = false; + } + }; + const handle = setInterval(() => void tick(), deps.config.walletHealthIntervalMs); + return () => clearInterval(handle); +} diff --git a/gateway/test/http.test.ts b/gateway/test/http.test.ts new file mode 100644 index 0000000..a7eec75 --- /dev/null +++ b/gateway/test/http.test.ts @@ -0,0 +1,105 @@ +import { describe, it, expect } from "vitest"; +import { tmpdir } from "os"; +import { join } from "path"; +import { randomUUID } from "crypto"; +import request from "supertest"; +import { createApp } from "../src/server.js"; +import { GatewayStore } from "../src/db/store.js"; +import { SdkRegistry } from "../src/sdk/registry.js"; +import type { GatewayConfig } from "../src/config.js"; +import type { GatewayDeps } from "../src/deps.js"; + +const SUPPLIERS = [ + { utxo_ref: "aa".repeat(32) + "#0", supplier_pkh: "s1", capability_id: "llm.text.generate.v1", model: "qwen", max_output_tokens: 256, price_lovelace: "1000000", supplier_bond_lovelace: "1000000", buyer_bond_lovelace: "1000000", endpoint_url: "http://sup", advert_status: "Active", status: "free" }, +]; + +function makeDeps(fetchFn?: typeof globalThis.fetch): GatewayDeps { + const dbDir = join(tmpdir(), `gw-test-${randomUUID()}`); + const store = new GatewayStore(dbDir); + const config: GatewayConfig = { + masterKeyHex: "ab".repeat(32), + indexerUrl: "http://ix", + ogmiosUrl: "http://og", + networkId: 0, + liveChain: true, + port: 0, + dbDir, + signupRate: { max: 1000, windowMs: 60_000 }, + keyRate: { max: 1000, windowMs: 60_000 }, + sweeperIntervalMs: 60_000, + walletHealthIntervalMs: 600_000, + sdkRegistryMax: 100, + }; + const chain = { queryUtxosByAddress: async () => [] } as unknown as GatewayDeps["chain"]; + const registry = new SdkRegistry({ + chain, + indexerUrl: config.indexerUrl, + networkId: config.networkId, + masterKeyHex: config.masterKeyHex, + max: config.sdkRegistryMax, + }); + const defaultFetch = (async () => new Response("[]", { status: 200 })) as unknown as typeof globalThis.fetch; + return { config, store, chain, registry, fetchFn: fetchFn ?? defaultFetch }; +} + +describe("gateway HTTP", () => { + it("GET /healthz", async () => { + const res = await request(createApp(makeDeps())).get("/healthz"); + expect(res.status).toBe(200); + expect(res.body).toEqual({ ok: true }); + }); + + it("GET / serves the UI", async () => { + const res = await request(createApp(makeDeps())).get("/"); + expect(res.status).toBe(200); + expect(res.text).toContain("OpenAI-compatible Gateway"); + }); + + it("rejects unauthenticated chat completions with OpenAI error shape", async () => { + const res = await request(createApp(makeDeps())) + .post("/openai/v1/chat/completions") + .send({ model: "qwen", messages: [{ role: "user", content: "hi" }] }); + expect(res.status).toBe(401); + expect(res.body.error.type).toBe("authentication_error"); + expect(res.body.error.code).toBe("invalid_api_key"); + }); + + it("signup → account roundtrip", async () => { + const app = createApp(makeDeps()); + const signup = await request(app).post("/signup").send({ label: "test" }); + expect(signup.status).toBe(201); + expect(signup.body.api_key).toMatch(/^vk_test_[0-9a-f]{48}$/); + expect(signup.body.deposit_address).toMatch(/^addr_test1/); + + const key = signup.body.api_key as string; + const acct = await request(app).get("/account").set("authorization", `Bearer ${key}`); + expect(acct.status).toBe(200); + expect(acct.body.balance.available_lovelace).toBe("0"); + expect(acct.body.collateral_ok).toBe(false); + expect(acct.body.spend.request_count).toBe(0); + }); + + it("GET /openai/v1/models lists distinct models", async () => { + const fetchFn = (async (url: string) => + new Response(JSON.stringify(String(url).includes("/suppliers") ? SUPPLIERS : []), { status: 200 })) as unknown as typeof globalThis.fetch; + const app = createApp(makeDeps(fetchFn)); + const signup = await request(app).post("/signup").send({}); + const key = signup.body.api_key as string; + const res = await request(app).get("/openai/v1/models").set("authorization", `Bearer ${key}`); + expect(res.status).toBe(200); + expect(res.body.object).toBe("list"); + expect(res.body.data.map((m: { id: string }) => m.id)).toContain("qwen"); + }); + + it("rejects tools/functions with 400 before any chain work", async () => { + const app = createApp(makeDeps()); + const signup = await request(app).post("/signup").send({}); + const key = signup.body.api_key as string; + const res = await request(app) + .post("/openai/v1/chat/completions") + .set("authorization", `Bearer ${key}`) + .send({ model: "qwen", messages: [{ role: "user", content: "hi" }], tools: [{ type: "function" }] }); + expect(res.status).toBe(400); + expect(res.body.error.code).toBe("unsupported_parameter"); + }); +}); diff --git a/gateway/test/unit.test.ts b/gateway/test/unit.test.ts new file mode 100644 index 0000000..5532d0f --- /dev/null +++ b/gateway/test/unit.test.ts @@ -0,0 +1,153 @@ +import { describe, it, expect } from "vitest"; +import { seal, open } from "../src/crypto/seal.js"; +import { Mutex } from "../src/sdk/registry.js"; +import { selectCandidates, listModels, parseRef } from "../src/routing/selectSupplier.js"; +import { parseChatRequest } from "../src/openai/validate.js"; +import { buildChatCompletion, buildChunk, usageFromReceipt, renderMessages } from "../src/openai/shapes.js"; +import { totalLovelace, hasCollateral, preflight } from "../src/onchain/preflight.js"; +import { GatewayError } from "../src/openai/errors.js"; + +const MASTER = "ab".repeat(32); // 64 hex + +const delay = (ms: number) => new Promise((r) => setTimeout(r, ms)); + +function jsonResponse(body: unknown): typeof globalThis.fetch { + return (async () => new Response(JSON.stringify(body), { status: 200 })) as unknown as typeof globalThis.fetch; +} + +describe("crypto/seal", () => { + it("round-trips and binds the master key", () => { + const priv = "cd".repeat(32); + const sealed = seal(priv, MASTER); + expect(open(sealed, MASTER)).toBe(priv); + // Wrong key fails the GCM auth tag. + expect(() => open(sealed, "ff".repeat(32))).toThrow(); + // Tampered ciphertext fails. + expect(() => open({ ...sealed, ct: sealed.ct.replace(/.$/, "0") }, MASTER)).toThrow(); + }); +}); + +describe("Mutex", () => { + it("serializes in submission order and survives rejection", async () => { + const m = new Mutex(); + const order: number[] = []; + const p1 = m.run(async () => { + await delay(20); + order.push(1); + }); + const p2 = m.run(async () => { + order.push(2); + }); + await Promise.all([p1, p2]); + expect(order).toEqual([1, 2]); + + await m.run(async () => { + throw new Error("boom"); + }).catch(() => undefined); + expect(await m.run(async () => 42)).toBe(42); + }); +}); + +describe("routing/selectSupplier", () => { + const rows = [ + { utxo_ref: "aa".repeat(32) + "#0", supplier_pkh: "s1", capability_id: "llm.text.generate.v1", model: "m", max_output_tokens: 1, price_lovelace: "2", supplier_bond_lovelace: "1", buyer_bond_lovelace: "1", endpoint_url: "http://a", advert_status: "Active", status: "free" }, + { utxo_ref: "bb".repeat(32) + "#1", supplier_pkh: "s2", capability_id: "llm.text.generate.v1", model: "m", max_output_tokens: 1, price_lovelace: "9", supplier_bond_lovelace: "1", buyer_bond_lovelace: "1", endpoint_url: "http://b", advert_status: "Active", status: "unknown" }, + { utxo_ref: "cc".repeat(32) + "#0", supplier_pkh: "s3", capability_id: "llm.text.generate.v1", model: "m", max_output_tokens: 1, price_lovelace: "1", supplier_bond_lovelace: "1", buyer_bond_lovelace: "1", endpoint_url: "http://c", advert_status: "Retired", status: "free" }, + { utxo_ref: "dd".repeat(32) + "#0", supplier_pkh: "s4", capability_id: "llm.chat.v1", model: "m", max_output_tokens: 1, price_lovelace: "1", supplier_bond_lovelace: "1", buyer_bond_lovelace: "1", endpoint_url: "http://d", advert_status: "Active", status: "free" }, + { utxo_ref: "ee".repeat(32) + "#0", supplier_pkh: "s5", capability_id: "llm.text.generate.v1", model: "other", max_output_tokens: 1, price_lovelace: "1", supplier_bond_lovelace: "1", buyer_bond_lovelace: "1", endpoint_url: "http://e", advert_status: "Active", status: "free" }, + ]; + + it("matches capability + model + Active + (free|unknown), free first", async () => { + const got = await selectCandidates({ + indexerUrl: "http://ix", + model: "m", + capabilityId: "llm.text.generate.v1", + fetchFn: jsonResponse(rows), + }); + expect(got.map((c) => c.supplierPkh)).toEqual(["s1", "s2"]); // s3 Retired, s4 chat, s5 other model + expect(got[0].status).toBe("free"); + expect(got[1].status).toBe("unknown"); + expect(got[0].priceLovelace).toBe(2n); + }); + + it("routes chat.v1 separately", async () => { + const got = await selectCandidates({ + indexerUrl: "http://ix", + model: "m", + capabilityId: "llm.chat.v1", + fetchFn: jsonResponse(rows), + }); + expect(got.map((c) => c.supplierPkh)).toEqual(["s4"]); + }); + + it("lists distinct Active models", async () => { + expect(await listModels({ indexerUrl: "http://ix", fetchFn: jsonResponse(rows) })).toEqual(["m", "other"]); + }); + + it("parseRef", () => { + expect(parseRef("aa".repeat(32) + "#3")).toEqual({ txHash: "aa".repeat(32), index: 3 }); + expect(parseRef("nope")).toBeNull(); + }); +}); + +describe("openai/validate", () => { + it("accepts a valid body and ignores temperature", () => { + const p = parseChatRequest({ model: "m", messages: [{ role: "user", content: "hi" }], temperature: 0.7, max_tokens: 50 }); + expect(p.model).toBe("m"); + expect(p.maxTokens).toBe(50); + expect(p.stream).toBe(false); + }); + it("rejects tools/functions", () => { + expect(() => parseChatRequest({ model: "m", messages: [{ role: "user", content: "x" }], tools: [] })).toThrow(GatewayError); + }); + it("rejects empty messages and bad model", () => { + expect(() => parseChatRequest({ model: "m", messages: [] })).toThrow(); + expect(() => parseChatRequest({ messages: [{ role: "user", content: "x" }] })).toThrow(); + }); +}); + +describe("openai/shapes", () => { + const receipt = { prompt_hash: "p", response_hash: "r", model: "m", prompt_tokens: 3, completion_tokens: 5, wallclock_ms: 1, supplier_pkh: "s", escrow_ref: "x#0" }; + it("usage + completion shape", () => { + const usage = usageFromReceipt(receipt); + expect(usage).toEqual({ prompt_tokens: 3, completion_tokens: 5, total_tokens: 8 }); + const cc = buildChatCompletion({ id: "id", model: "m", content: "hello", usage, vector: { receipt, receipt_signature: "sig", escrow_ref: "x#0" } }); + expect(cc.object).toBe("chat.completion"); + expect((cc.choices as any)[0].message).toEqual({ role: "assistant", content: "hello" }); + expect((cc as any).x_vector.escrow_ref).toBe("x#0"); + }); + it("chunk shape", () => { + const ch = buildChunk({ id: "id", model: "m", delta: { content: "tok" }, finishReason: null }); + expect(ch.object).toBe("chat.completion.chunk"); + expect((ch.choices as any)[0].delta.content).toBe("tok"); + }); + it("renderMessages folds roles", () => { + expect(renderMessages([{ role: "system", content: "be nice" }, { role: "user", content: "hi" }])).toBe("System: be nice\n\nUser: hi"); + }); +}); + +describe("onchain/preflight", () => { + const pure = (lov: bigint) => ({ ref: { txHash: "x", index: 0 }, address: "a", lovelace: lov, assets: {}, datumHex: null, scriptRef: null }); + const mixed = (lov: bigint) => ({ ...pure(lov), assets: { "policy.tok": 1n } }); + + it("totalLovelace + hasCollateral", () => { + expect(totalLovelace([pure(3_000_000n), pure(4_000_000n)])).toBe(7_000_000n); + expect(hasCollateral([pure(4_000_000n)])).toBe(false); // < 5 ADA + expect(hasCollateral([mixed(6_000_000n)])).toBe(false); // has native asset + expect(hasCollateral([pure(6_000_000n)])).toBe(true); + }); + + it("preflight requires balance AND collateral", async () => { + const cost = { priceLovelace: 2_000_000n, buyerBondLovelace: 1_000_000n, supplierBondLovelace: 1_000_000n }; + // required = 2 + 1 + 1 + 5 (collateral) + 2 (fee) = 11 ADA + const fakeChain = { queryUtxosByAddress: async () => [pure(20_000_000n)] } as any; + const ok = await preflight(fakeChain, "a", cost); + expect(ok.ok).toBe(true); + expect(ok.requiredLovelace).toBe(11_000_000n); + + const poor = { queryUtxosByAddress: async () => [pure(4_000_000n)] } as any; + const bad = await preflight(poor, "a", cost); + expect(bad.ok).toBe(false); + expect(bad.collateralOk).toBe(false); + }); +}); diff --git a/gateway/tsconfig.json b/gateway/tsconfig.json new file mode 100644 index 0000000..e76fd54 --- /dev/null +++ b/gateway/tsconfig.json @@ -0,0 +1,9 @@ +{ + "extends": "../tsconfig.base.json", + "compilerOptions": { + "rootDir": "./src", + "outDir": "./dist", + "lib": ["ES2022"] + }, + "include": ["src/**/*.ts"] +} diff --git a/packages/shared/src/tx/server.ts b/packages/shared/src/tx/server.ts index 9dff896..afe5ceb 100644 --- a/packages/shared/src/tx/server.ts +++ b/packages/shared/src/tx/server.ts @@ -38,3 +38,10 @@ export type { ConsolidateWalletFlowResult, } from "./wallet/runConsolidateWallet.js"; export { runConsolidateWallet } from "./wallet/runConsolidateWallet.js"; + +// ── Wallet withdraw (custodial exit; live-only) ───────────────────────── +export type { + WithdrawWalletParams, + WithdrawWalletBuildResult, +} from "./wallet/withdrawWallet.js"; +export { buildWithdrawTx } from "./wallet/withdrawWallet.js"; diff --git a/packages/shared/src/tx/wallet/withdrawWallet.ts b/packages/shared/src/tx/wallet/withdrawWallet.ts new file mode 100644 index 0000000..c2eef64 --- /dev/null +++ b/packages/shared/src/tx/wallet/withdrawWallet.ts @@ -0,0 +1,81 @@ +/** + * withdrawWallet.ts — custodial-exit tx builder (wallet → external address). + * + * Mirrors consolidateWallet.ts but pays a chosen amount to an arbitrary bech32 + * address (lucid returns the change to the wallet). This is the withdrawal path + * for the gateway's custodial wallets: a user can move their unspent AP3X back + * out. No script, no datum, no native assets. + * + * `amountLovelace` omitted ⇒ withdraw all available minus a fee reserve. + * Live-only (needs a real Ogmios endpoint). + */ + +import type { LiveOgmiosProvider } from "../../chain/LiveOgmiosProvider.js"; +import type { BuildResult, WalletKey } from "../types.js"; +import { TxConstructionError } from "../types.js"; +import { OgmiosLucidProvider } from "../../chain/OgmiosLucidProvider.js"; +import { createLucidContext } from "../internal/lucidContext.js"; +import { DEFAULT_FEE_RESERVE } from "./planConsolidate.js"; + +export interface WithdrawWalletParams { + chain: LiveOgmiosProvider; + walletKey: WalletKey; + /** Destination bech32 address. */ + toAddress: string; + /** Lovelace to send. Omit to withdraw all available minus the fee reserve. */ + amountLovelace?: bigint; +} + +export interface WithdrawWalletBuildResult extends BuildResult { + amountLovelace: bigint; + toAddress: string; +} + +export async function buildWithdrawTx( + params: WithdrawWalletParams, +): Promise { + const { chain, walletKey, toAddress } = params; + + const provider = new OgmiosLucidProvider({ ogmiosUrl: chain.url, fetch: chain.fetchImpl }); + const { lucid } = await createLucidContext( + provider, + walletKey, + { networkId: 1, systemStartUnix: 0, slotLengthMs: 1000 }, + { usePresetProtocolParameters: true }, + ); + + const realWalletUtxos = await lucid.wallet().getUtxos(); + const total = realWalletUtxos.reduce((sum, u) => sum + (u.assets.lovelace ?? 0n), 0n); + const spendable = total - DEFAULT_FEE_RESERVE; + + const amount = params.amountLovelace ?? spendable; + if (amount <= 0n) { + throw new TxConstructionError("withdraw_amount_invalid", "nothing to withdraw"); + } + if (amount > spendable) { + throw new TxConstructionError( + "withdraw_insufficient", + `requested ${amount} exceeds withdrawable ${spendable} (total ${total} minus fee reserve)`, + ); + } + + let signed; + try { + const txBuilder = lucid + .newTx() + .pay.ToAddress(toAddress, { lovelace: amount }) + .addSignerKey(walletKey.pubKeyHash) + .setMinFee(500_000n); + const completed = await txBuilder.complete({ presetWalletInputs: realWalletUtxos, localUPLCEval: false }); + signed = await completed.sign.withWallet().complete(); + } catch (err) { + const msg = err instanceof Error ? err.message : String(err); + throw new TxConstructionError("withdraw_build_failed", msg); + } + + const txCborHex = signed.toCBOR(); + const expectedTxHash = signed.toHash(); + await chain.submitTx(txCborHex); + + return { txCborHex, expectedTxHash, amountLovelace: amount, toAddress }; +} diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 1432f74..2d29a3f 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -160,6 +160,46 @@ importers: specifier: ^5.4.0 version: 5.4.21(@types/node@25.6.0) + gateway: + dependencies: + '@marketplace/buyer': + specifier: workspace:* + version: link:../buyer + '@marketplace/shared': + specifier: workspace:* + version: link:../packages/shared + '@noble/ed25519': + specifier: ^2.3.0 + version: 2.3.0 + '@noble/hashes': + specifier: ^1.8.0 + version: 1.8.0 + bech32: + specifier: ^2.0.0 + version: 2.0.0 + better-sqlite3: + specifier: ^11.9.1 + version: 11.10.0 + express: + specifier: ^4.21.2 + version: 4.22.1 + devDependencies: + '@types/better-sqlite3': + specifier: ^7.6.12 + version: 7.6.13 + '@types/express': + specifier: ^5.0.3 + version: 5.0.6 + supertest: + specifier: ^7.0.0 + version: 7.2.2 + tsx: + specifier: ^4.19.0 + version: 4.21.0 + typescript: + specifier: ^5.7.2 + version: 5.9.3 + indexer: dependencies: '@marketplace/shared': diff --git a/pnpm-workspace.yaml b/pnpm-workspace.yaml index a1ea6a7..f8e6e38 100644 --- a/pnpm-workspace.yaml +++ b/pnpm-workspace.yaml @@ -1,6 +1,7 @@ packages: - "packages/*" - "buyer" + - "gateway" - "supplier" - "indexer" - "indexer-ui" diff --git a/vitest.config.ts b/vitest.config.ts index 3e7532c..1ac43ff 100644 --- a/vitest.config.ts +++ b/vitest.config.ts @@ -41,6 +41,7 @@ export default defineConfig({ "packages/**/*.{test,spec}.ts", "buyer/**/*.{test,spec}.ts", "buyer/**/*.{test,spec}.tsx", + "gateway/**/*.{test,spec}.ts", "supplier/**/*.{test,spec}.ts", "indexer/**/*.{test,spec}.ts", ],