diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..049bd49 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,8 @@ +node_modules +.git +.github +dist +.env +.env.* +*.test.ts +CLAUDE.md diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml new file mode 100644 index 0000000..bdc659c --- /dev/null +++ b/.github/workflows/build.yml @@ -0,0 +1,57 @@ +name: Build and Push to GHCR + +on: + push: + branches: [master] + paths: + - 'src/**' + - 'Dockerfile' + - 'package.json' + - 'bun.lock*' + workflow_dispatch: + +env: + REGISTRY: ghcr.io + IMAGE_NAME: ${{ github.repository }} + +jobs: + build-and-push: + runs-on: ubuntu-latest + permissions: + contents: read + packages: write + + steps: + - name: Checkout repository + uses: actions/checkout@v4 + + - name: Log in to Container Registry + uses: docker/login-action@v3 + with: + registry: ${{ env.REGISTRY }} + username: ${{ github.actor }} + password: ${{ secrets.GITHUB_TOKEN }} + + - name: Extract metadata + id: meta + uses: docker/metadata-action@v5 + with: + images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }} + tags: | + type=raw,value=latest + type=sha,prefix= + + - name: Build and push Docker image + uses: docker/build-push-action@v5 + with: + context: . + push: true + tags: ${{ steps.meta.outputs.tags }} + labels: ${{ steps.meta.outputs.labels }} + + - name: Trigger deployment + uses: peter-evans/repository-dispatch@v3 + with: + token: ${{ secrets.DEPLOY_TOKEN }} + event-type: deploy + client-payload: '{"image_tag": "${{ github.sha }}"}' diff --git a/.github/workflows/deploy.yml b/.github/workflows/deploy.yml index 74ac393..77b6e4e 100644 --- a/.github/workflows/deploy.yml +++ b/.github/workflows/deploy.yml @@ -1,21 +1,54 @@ -name: Deploy +name: Deploy to VPS + on: - push: - branches: [master] + repository_dispatch: + types: [deploy] + workflow_dispatch: + +env: + VPS_HOST: 45.127.35.244 + VPS_USER: root + jobs: deploy: runs-on: ubuntu-latest steps: - - uses: actions/checkout@v4 - - uses: oven-sh/setup-bun@v2 + - name: Setup SSH + uses: webfactory/ssh-agent@v0.9.0 with: - bun-version: latest - - run: bun install - - run: bun run build - - run: bun test + ssh-private-key: ${{ secrets.VPS_SSH_KEY }} - - name: Deploy to Cloudflare Workers - uses: cloudflare/wrangler-action@v3 - with: - apiToken: ${{ secrets.CLOUDFLARE_API_TOKEN }} - accountId: ${{ secrets.CLOUDFLARE_ACCOUNT_ID }} + - name: Add VPS to known_hosts + run: | + mkdir -p ~/.ssh + ssh-keyscan -H ${{ env.VPS_HOST }} >> ~/.ssh/known_hosts + + - name: Pull and restart container + run: | + ssh ${{ env.VPS_USER }}@${{ env.VPS_HOST }} << 'EOF' + set -e + + echo "=== Logging in to GHCR ===" + echo "${{ secrets.GITHUB_TOKEN }}" | docker login ghcr.io -u ${{ github.actor }} --password-stdin + + echo "=== Pulling latest image ===" + docker pull ghcr.io/${{ github.repository }}:latest + + echo "=== Stopping old container ===" + cd /opt/edge-proxy + docker compose down || true + + echo "=== Starting new container ===" + docker compose up -d + + echo "=== Waiting for health check ===" + sleep 5 + curl -sf http://localhost:3000/health || echo "Health check failed" + + echo "=== Deployment complete ===" + EOF + + - name: Verify deployment + run: | + sleep 10 + curl -sf https://proxy.asepharyana.my.id/health || echo "External health check failed" diff --git a/CLAUDE.md b/CLAUDE.md index 0e5bbd2..979ae11 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -7,8 +7,50 @@ Key architecture facts: - Middleware stack: rate limiter, body limiter, structured logger, SSRF protection - WebSocket relay: bidirectional relay via `x-relay-target` header with `ws://` or `wss://` - Error classification: DNS errors -> 502, timeouts -> 504, SSRF blocks -> 403, rate limits -> 429 +- IPv6 support: dual-stack listen + outbound source rotation via `Bun.spawn` + `curl --interface` - The old Next.js `src/app/route.ts` still exists as a legacy file but is no longer the active entry point +## IPv6 Configuration + +### Environment Variables + +| Variable | Default | Description | +|----------|---------|-------------| +| `HOST` | `::` | Bind address. Use `::` for dual-stack (IPv4+IPv6) | +| `IPV6_SOURCES` | _(empty)_ | Comma-separated IPv6 source addresses for outbound rotation | + +### Setup + +1. Add IPv6 addresses to your interface: +```bash +ip -6 addr add 2001:df4:c140:1f::d6/128 dev eth0 +ip -6 addr add 2001:df4:c140:1f:ffff:ffff:ffff:ffff/128 dev eth0 +``` + +2. Configure the proxy with IPv6 source rotation: +```bash +IPV6_SOURCES=2001:df4:c140:1f::d6,2001:df4:c140:1f:ffff:ffff:ffff:ffff bun run src/index.ts +``` + +### How It Works + +- **Listen**: Server binds to `::` (all IPv6 interfaces) with `ipv6Only: false` (dual-stack) +- **Outbound**: When `IPV6_SOURCES` is configured, each outbound request rotates through the source addresses using `curl --interface ` +- **Failover**: Failed source addresses are automatically disabled after 3 consecutive failures + +### Source: `src/lib/ipv6-pool.ts` + +```typescript +import { IPv6SourcePool } from "./lib/ipv6-pool"; + +const pool = new IPv6SourcePool(); +pool.loadFromEnv(); // reads IPV6_SOURCES + +const source = pool.getNext(); // round-robin +pool.markSuccess(source); // reset failure count +pool.markFailed(source); // increment failure count +``` + Default to using Bun instead of Node.js. - Use `bun ` instead of `node ` or `ts-node ` diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..bbc62df --- /dev/null +++ b/Dockerfile @@ -0,0 +1,22 @@ +FROM oven/bun:1.3-alpine +WORKDIR /app + +# Install curl for IPv6 source rotation +RUN apk add --no-cache curl + +# Copy dependency manifests +COPY package.json bun.lock* ./ + +# Install all dependencies (no production deps) +RUN bun install + +# Copy source code +COPY src/ src/ +COPY public/ public/ +COPY proxy.txt ./ + +EXPOSE 3000 + +USER bun + +CMD ["bun", "run", "src/index.ts"] diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..57d3023 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,13 @@ +services: + edge-proxy: + image: ghcr.io/mytheclipse/proxy-bun:latest + container_name: edge-proxy + restart: unless-stopped + network_mode: host + environment: + HOST: "::" + PORT: "3000" + IPV6_SOURCES: "2001:df4:c140:1f::395,2001:df4:c140:1f::8,2001:df4:c140:1f::d9,2001:df4:c140:1f::db,2001:df4:c140:1f::dd,2001:df4:c140:1f::de,2001:df4:c140:1f::df,2001:df4:c140:1f:ffff:ffff:ffff:ffff" + RELAY_TIMEOUT_MS: "30000" + RATE_LIMIT_MAX: "1000" + RATE_LIMIT_WINDOW_MS: "60000" diff --git a/proxy.txt b/proxy.txt index 6d50863..e69de29 100644 --- a/proxy.txt +++ b/proxy.txt @@ -1,10 +0,0 @@ -38.154.203.95:5863:eoimpusi:27ije2xpgvl9 -198.105.121.200:6462:eoimpusi:27ije2xpgvl9 -64.137.96.74:6641:eoimpusi:27ije2xpgvl9 -209.127.138.10:5784:eoimpusi:27ije2xpgvl9 -38.154.185.97:6370:eoimpusi:27ije2xpgvl9 -84.247.60.125:6095:eoimpusi:27ije2xpgvl9 -142.111.67.146:5611:eoimpusi:27ije2xpgvl9 -191.96.254.138:6185:eoimpusi:27ije2xpgvl9 -31.58.9.4:6077:eoimpusi:27ije2xpgvl9 -104.239.107.47:5699:eoimpusi:27ije2xpgvl9 \ No newline at end of file diff --git a/src/index.ts b/src/index.ts index c163ade..61757e2 100644 --- a/src/index.ts +++ b/src/index.ts @@ -31,6 +31,7 @@ import { checkBodySize } from "./middleware/body-limiter"; import { createRateLimiter } from "./middleware/rate-limiter"; import { logRelayEvent } from "./middleware/logger"; import { ProxyPool, SessionProxyPool } from "./lib/proxy-pool"; +import { IPv6SourcePool } from "./lib/ipv6-pool"; import { handleChatCompletion, listModels } from "./lib/ai-proxy"; import { handleAnthropicMessages } from "./lib/anthropic-proxy"; import { fetchWithRetry, closeAllActiveReaders, isDevMode } from "./lib/fetch-utils"; @@ -40,6 +41,7 @@ import type { Server, ServerWebSocket } from "bun"; // --- Configuration ------------------------------------------------------------ const PORT = Number.parseInt(process.env.PORT ?? "3000", 10); +const HOST = process.env.HOST ?? "::"; const RELAY_TIMEOUT_MS = Number.parseInt( process.env.RELAY_TIMEOUT_MS ?? "30000", 10, @@ -87,6 +89,14 @@ proxyPool.tryLoad( const sessionPool = new SessionProxyPool(proxyPool); sessionPool.setFailureThreshold(3); +// --- IPv6 source pool (optional) ---------------------------------------------- + +const ipv6Pool = new IPv6SourcePool(); +ipv6Pool.loadFromEnv(); +if (ipv6Pool.configured) { + console.log(`[relay] IPv6 source pool loaded: ${ipv6Pool.size} addresses`); +} + // --- SSRF DNS rebinding protection -------------------------------------------- if (process.env.SSRF_DNS_CHECK === "true") { @@ -388,12 +398,16 @@ async function handleRelay( const targetUrlString = targetUrl.toString(); + // -- Get IPv6 source for outbound binding ---------------------------------- + const ipv6Source = ipv6Pool.getNext() ?? undefined; + // -- Execute upstream fetch with shared retry ------------------------------- const result = await fetchWithRetry( targetUrlString, fetchOptions, proxyPool, "relay", + ipv6Source, ); if (result.errorClassification) { @@ -477,6 +491,8 @@ function handleWebSocketUpgrade( // --- Server ------------------------------------------------------------------ const server: Server = Bun.serve({ + hostname: HOST, + ipv6Only: false, port: PORT, development: isDevMode() ? { hmr: true, console: true } : undefined, @@ -509,8 +525,9 @@ const server: Server = Bun.serve({ try { const body = await req.json(); const sessionId = crypto.randomUUID(); - console.log(`[index] POST /v1/chat/completions session=${sessionId.slice(0, 8)} model=${(body as any).model} poolSize=${proxyPool.size} sessionPool.active=${sessionPool.activeSessions}`); - return handleChatCompletion(body, proxyPool, sessionPool, sessionId); + const ipv6Source = ipv6Pool.getNext() ?? undefined; + console.log(`[index] POST /v1/chat/completions session=${sessionId.slice(0, 8)} model=${(body as any).model} poolSize=${proxyPool.size} sessionPool.active=${sessionPool.activeSessions} ipv6=${ipv6Source ?? "none"}`); + return handleChatCompletion(body, proxyPool, sessionPool, sessionId, ipv6Source); } catch { return new Response( JSON.stringify({ error: { message: "Invalid JSON body", type: "invalid_request_error" } }), @@ -532,8 +549,9 @@ const server: Server = Bun.serve({ try { const body = await req.json(); const sessionId = crypto.randomUUID(); - console.log(`[index] POST /v1/messages session=${sessionId.slice(0, 8)} model=${(body as any).model} poolSize=${proxyPool.size} sessionPool.active=${sessionPool.activeSessions}`); - return handleAnthropicMessages(body, proxyPool, sessionPool, sessionId); + const ipv6Source = ipv6Pool.getNext() ?? undefined; + console.log(`[index] POST /v1/messages session=${sessionId.slice(0, 8)} model=${(body as any).model} poolSize=${proxyPool.size} sessionPool.active=${sessionPool.activeSessions} ipv6=${ipv6Source ?? "none"}`); + return handleAnthropicMessages(body, proxyPool, sessionPool, sessionId, ipv6Source); } catch { return new Response( JSON.stringify({ @@ -659,7 +677,7 @@ const server: Server = Bun.serve({ // --- Startup ----------------------------------------------------------------- console.log( - `[relay] Edge Proxy Relay v${RELAY_VERSION} listening on http://localhost:${server.port}`, + `[relay] Edge Proxy Relay v${RELAY_VERSION} listening on http://${HOST}:${server.port}`, ); if (isDevMode()) { console.log("[relay] Development mode: HMR enabled"); diff --git a/src/lib/ai-proxy.ts b/src/lib/ai-proxy.ts index 84b15e6..6559f59 100644 --- a/src/lib/ai-proxy.ts +++ b/src/lib/ai-proxy.ts @@ -376,6 +376,7 @@ export async function handleChatCompletion( proxyPool?: ProxyPool, sessionPool?: SessionProxyPool, sessionId?: string, + ipv6Source?: string, ): Promise { // -- Input validation ------------------------------------------------------- const validationError = validateChatRequest(body); @@ -420,8 +421,8 @@ export async function handleChatCompletion( // -- Execute with session-aware or standard retry -------------------------- let result: FetchWithRetryResult = sessionPool && sessionId - ? await fetchWithSessionRetry(url, init, sessionPool, sessionId, `openai:${req.model}`) - : await fetchWithRetry(url, init, proxyPool, `openai:${req.model}`); + ? await fetchWithSessionRetry(url, init, sessionPool, sessionId, `openai:${req.model}`, undefined, ipv6Source) + : await fetchWithRetry(url, init, proxyPool, `openai:${req.model}`, ipv6Source); // -- Mimo Free: auth failure → invalidate JWT and retry once --------------- if ( @@ -437,8 +438,8 @@ export async function handleChatCompletion( }; result = sessionPool && sessionId - ? await fetchWithSessionRetry(url, init, sessionPool, sessionId, `openai:${req.model}`) - : await fetchWithRetry(url, init, proxyPool, `openai:${req.model}`); + ? await fetchWithSessionRetry(url, init, sessionPool, sessionId, `openai:${req.model}`, undefined, ipv6Source) + : await fetchWithRetry(url, init, proxyPool, `openai:${req.model}`, ipv6Source); } // -- aichat.org: session expiry → invalidate session and retry once --------- @@ -456,8 +457,8 @@ export async function handleChatCompletion( }; result = sessionPool && sessionId - ? await fetchWithSessionRetry(url, init, sessionPool, sessionId, `openai:${req.model}`) - : await fetchWithRetry(url, init, proxyPool, `openai:${req.model}`); + ? await fetchWithSessionRetry(url, init, sessionPool, sessionId, `openai:${req.model}`, undefined, ipv6Source) + : await fetchWithRetry(url, init, proxyPool, `openai:${req.model}`, ipv6Source); } if (result.errorClassification) { diff --git a/src/lib/anthropic-proxy.ts b/src/lib/anthropic-proxy.ts index f2d169d..0e435e8 100644 --- a/src/lib/anthropic-proxy.ts +++ b/src/lib/anthropic-proxy.ts @@ -473,6 +473,7 @@ export async function handleAnthropicMessages( proxyPool?: ProxyPool, sessionPool?: SessionProxyPool, sessionId?: string, + ipv6Source?: string, ): Promise { // -- Input validation ------------------------------------------------------- const validationError = validateAnthropicRequest(body); @@ -506,8 +507,8 @@ export async function handleAnthropicMessages( // -- Execute with session-aware or standard retry -------------------------- const result: FetchWithRetryResult = sessionPool && sessionId - ? await fetchWithSessionRetry(url, init, sessionPool, sessionId, `anthropic:${req.model}`) - : await fetchWithRetry(url, init, proxyPool, `anthropic:${req.model}`); + ? await fetchWithSessionRetry(url, init, sessionPool, sessionId, `anthropic:${req.model}`, undefined, ipv6Source) + : await fetchWithRetry(url, init, proxyPool, `anthropic:${req.model}`, ipv6Source); if (result.errorClassification) { if (sessionPool && sessionId) { diff --git a/src/lib/fetch-utils.ts b/src/lib/fetch-utils.ts index 1680170..dd316bb 100644 --- a/src/lib/fetch-utils.ts +++ b/src/lib/fetch-utils.ts @@ -6,6 +6,12 @@ */ import type { ProxyPool, SessionProxyPool } from "./proxy-pool"; +import type { IPv6SourcePool } from "./ipv6-pool"; + +// ─── Constants ───────────────────────────────────────────────────────── + +/** Default relay timeout (30 seconds). Used for curl IPv6 source requests. */ +const DEFAULT_TIMEOUT_MS = 30_000; // ─── Active stream tracking (for graceful shutdown) ─────────────────────── @@ -127,11 +133,109 @@ export function sanitizeErrorMessage(raw: string): string { return "Upstream error"; } +// ─── Fetch via curl (for IPv6 source binding) ──────────────────────────── + +/** + * Execute an HTTP request via curl with a specific source IPv6 address. + * Uses `Bun.spawn` to run curl with `--interface` to bind to the given address. + * + * This is used when outbound IPv6 source rotation is needed, since + * Bun's built-in fetch() does not support specifying a local address. + * + * @param url - Target URL + * @param init - Request init (method, headers, body) + * @param ipv6Source - IPv6 source address to bind to + * @param timeoutMs - Request timeout in milliseconds + */ +export async function fetchViaCurl( + url: string, + init: RequestInit, + ipv6Source: string, + timeoutMs: number, +): Promise { + const method = (init.method ?? "GET").toUpperCase(); + const args = [ + "curl", + "-6", + "--interface", ipv6Source, + "-X", method, + "-s", // silent mode + "--compressed", // auto-decompress gzip/brotli + "-o", "-", // output body to stdout + "-w", "\n%{http_code}", // append status code on new line after body + "--max-time", String(Math.ceil(timeoutMs / 1000)), + "--connect-timeout", "10", + ]; + + // Add headers + if (init.headers) { + const headers = init.headers instanceof Headers + ? Object.fromEntries(init.headers.entries()) + : init.headers; + for (const [key, value] of Object.entries(headers)) { + if (key.toLowerCase() === "host") continue; // curl sets Host automatically + args.push("-H", `${key}: ${value}`); + } + } + + // Add body for non-GET/HEAD methods + if (init.body && method !== "GET" && method !== "HEAD") { + if (typeof init.body === "string") { + args.push("-d", init.body); + } else if (init.body instanceof ArrayBuffer) { + args.push("--data-binary", "@-"); + } + // Note: streaming bodies not supported via curl path + } + + args.push(url); + + logProxy("fetchViaCurl", `${args.slice(0, 6).join(" ")}...`, { ipv6Source, url }); + + const proc = Bun.spawn(args, { + stdout: "pipe", + stderr: "pipe", + stdin: "pipe", + }); + + // Collect output with timeout + const timeout = setTimeout(() => { + proc.kill(); + }, timeoutMs + 5000); + + try { + const [stdout, stderr] = await Promise.all([ + new Response(proc.stdout).text(), + new Response(proc.stderr).text(), + ]); + + clearTimeout(timeout); + + // Parse: curl -w "\n%{http_code}" appends status code on a new line after body + const lastNewline = stdout.lastIndexOf("\n"); + const rawCode = parseInt(stdout.slice(lastNewline + 1), 10); + const statusCode = rawCode > 0 ? rawCode : 502; + const body = lastNewline >= 0 ? stdout.slice(0, lastNewline) : stdout; + + logProxy("fetchViaCurl", `response status=${statusCode}`, { ipv6Source, url }); + + return new Response(body, { + status: statusCode, + statusText: statusCode === 200 ? "OK" : "Error", + headers: { "Content-Type": "text/plain" }, + }); + } catch (err) { + clearTimeout(timeout); + proc.kill(); + throw err; + } +} + // ─── Fetch with retry (direct → proxy fallback) ───────────────────────── export interface FetchWithRetryResult { - response?: Response; - errorClassification?: { code: string; status: number; message: string }; + response?: Response; + errorClassification?: { code: string; status: number; message: string }; } /** @@ -141,12 +245,17 @@ export interface FetchWithRetryResult { * Falls back to direct when no pool is available. * Logs every failure to `console.warn` so the operator can diagnose without * the error body leaking to the downstream client. + * + * @param ipv6Source - Optional IPv6 source address for outbound binding. + * When provided, uses curl instead of fetch() to bind + * to the specified source address. */ export async function fetchWithRetry( - url: string, - init: RequestInit & { proxy?: string }, - proxyPool?: ProxyPool, - context?: string, + url: string, + init: RequestInit & { proxy?: string }, + proxyPool?: ProxyPool, + context?: string, + ipv6Source?: string, ): Promise { let response: Response | undefined; let lastError: unknown; @@ -184,8 +293,22 @@ export async function fetchWithRetry( const proxyShort = init.proxy ? init.proxy.replace(/https?:\/\//, "").replace(/@.*/, "@***") : "direct"; logProxy("fetchWithRetry", `attempt=${attempt + 1}/${maxAttempts} proxy=${proxyShort}`, { context }); - try { - response = await fetch(url, init); + try { + // Use curl for IPv6 source binding (only for direct connections) + if (ipv6Source && !init.proxy) { + response = await fetchViaCurl(url, init, ipv6Source, DEFAULT_TIMEOUT_MS); + // If curl failed with connection error (502, empty body), fallback to regular fetch + if (response.status === 502) { + const cloned = response.clone(); + const bodyText = await cloned.text(); + if (!bodyText) { + logProxy("fetchWithRetry", `IPv6 connection failed, falling back to regular fetch`, { ipv6Source, url }); + response = await fetch(url, init); + } + } + } else { + response = await fetch(url, init); + } if (response.ok) { if (usedProxy && proxyPool && proxyPool.size > 0) { proxyPool.markSuccess(); @@ -281,6 +404,8 @@ function classifyFetchErrorSafe(error: unknown): { * SSE streams: the initial request is retried normally. Once the response body * starts streaming, mid-stream errors are **not** retried; the session is * released and the error is returned to the caller. + * + * @param ipv6Source - Optional IPv6 source address for outbound binding. */ export async function fetchWithSessionRetry( url: string, @@ -289,12 +414,18 @@ export async function fetchWithSessionRetry( sessionId: string, context?: string, maxRetries?: number, + ipv6Source?: string, ): Promise { // Fallback when no session pool is available if (!sessionPool) { logProxy("fetchWithSessionRetry", "no session pool — direct fetch", { context, sessionId: sessionId.slice(0, 8) }); try { - const response = await fetch(url, init); + let response: Response; + if (ipv6Source) { + response = await fetchViaCurl(url, init, ipv6Source, DEFAULT_TIMEOUT_MS); + } else { + response = await fetch(url, init); + } return { response }; } catch (err) { return { errorClassification: classifyFetchErrorSafe(err) }; @@ -333,7 +464,13 @@ export async function fetchWithSessionRetry( }); try { - const response = await fetch(url, init); + // Use curl for IPv6 source binding (only for direct connections) + let response: Response; + if (ipv6Source && !init.proxy) { + response = await fetchViaCurl(url, init, ipv6Source, DEFAULT_TIMEOUT_MS); + } else { + response = await fetch(url, init); + } if (response.ok) { sessionPool.markSuccess(sessionId); diff --git a/src/lib/ipv6-pool.test.ts b/src/lib/ipv6-pool.test.ts new file mode 100644 index 0000000..7f495c1 --- /dev/null +++ b/src/lib/ipv6-pool.test.ts @@ -0,0 +1,118 @@ +import { test, expect, describe } from "bun:test"; +import { IPv6SourcePool } from "./ipv6-pool"; + +describe("IPv6SourcePool", () => { + test("should load addresses from comma-separated string", () => { + const pool = new IPv6SourcePool(); + pool.loadFromString("2001:db8::1,2001:db8::2,2001:db8::3"); + + expect(pool.size).toBe(3); + expect(pool.configured).toBe(true); + }); + + test("should return null when empty", () => { + const pool = new IPv6SourcePool(); + expect(pool.size).toBe(0); + expect(pool.configured).toBe(false); + expect(pool.getNext()).toBeNull(); + }); + + test("should rotate through addresses round-robin", () => { + const pool = new IPv6SourcePool(); + pool.loadFromString("2001:db8::1,2001:db8::2,2001:db8::3"); + + expect(pool.getNext()).toBe("2001:db8::1"); + expect(pool.getNext()).toBe("2001:db8::2"); + expect(pool.getNext()).toBe("2001:db8::3"); + expect(pool.getNext()).toBe("2001:db8::1"); // wraps around + }); + + test("should skip failed sources", () => { + const pool = new IPv6SourcePool(); + pool.loadFromString("2001:db8::1,2001:db8::2,2001:db8::3"); + pool.setFailureThreshold(1); + + pool.markFailed("2001:db8::2"); + + const results = [pool.getNext(), pool.getNext(), pool.getNext(), pool.getNext()]; + // Should skip 2001:db8::2 (disabled after 1 failure with threshold=1) + expect(results).not.toContain("2001:db8::2"); + }); + + test("should reset after all sources fail", () => { + const pool = new IPv6SourcePool(); + pool.loadFromString("2001:db8::1,2001:db8::2"); + + // Mark all as failed + pool.markFailed("2001:db8::1"); + pool.markFailed("2001:db8::1"); + pool.markFailed("2001:db8::1"); + pool.markFailed("2001:db8::2"); + pool.markFailed("2001:db8::2"); + pool.markFailed("2001:db8::2"); + + // Should auto-reset and return first + const addr = pool.getNext(); + expect(addr).toBe("2001:db8::1"); + }); + + test("should reset failure count on success", () => { + const pool = new IPv6SourcePool(); + pool.loadFromString("2001:db8::1,2001:db8::2"); + + pool.markFailed("2001:db8::1"); + pool.markFailed("2001:db8::1"); + pool.markSuccess("2001:db8::1"); + + // Should not be disabled after success + const addr = pool.getNext(); + expect(addr).toBe("2001:db8::1"); + }); + + test("should get address by index", () => { + const pool = new IPv6SourcePool(); + pool.loadFromString("2001:db8::1,2001:db8::2"); + + expect(pool.getAtIndex(0)).toBe("2001:db8::1"); + expect(pool.getAtIndex(1)).toBe("2001:db8::2"); + expect(pool.getAtIndex(2)).toBeNull(); + }); + + test("should handle whitespace in input", () => { + const pool = new IPv6SourcePool(); + pool.loadFromString(" 2001:db8::1 , 2001:db8::2 "); + + expect(pool.size).toBe(2); + expect(pool.getNext()).toBe("2001:db8::1"); + }); + + test("should handle empty input", () => { + const pool = new IPv6SourcePool(); + pool.loadFromString(""); + expect(pool.size).toBe(0); + }); + + test("should reset all sources", () => { + const pool = new IPv6SourcePool(); + pool.loadFromString("2001:db8::1,2001:db8::2"); + + pool.markFailed("2001:db8::1"); + pool.markFailed("2001:db8::1"); + pool.markFailed("2001:db8::1"); + pool.reset(); + + expect(pool.getNext()).toBe("2001:db8::1"); + }); + + test("should change failure threshold", () => { + const pool = new IPv6SourcePool(); + pool.loadFromString("2001:db8::1"); + pool.setFailureThreshold(1); + + pool.markFailed("2001:db8::1"); + // With threshold 1, should be disabled after 1 failure + // getNext should auto-reset since all failed + const addr = pool.getNext(); + expect(addr).toBe("2001:db8::1"); + }); +}); diff --git a/src/lib/ipv6-pool.ts b/src/lib/ipv6-pool.ts new file mode 100644 index 0000000..42bde3c --- /dev/null +++ b/src/lib/ipv6-pool.ts @@ -0,0 +1,186 @@ +/** + * IPv6 source address pool for outbound rotation. + * + * Loads multiple IPv6 source addresses and rotates through them + * round-robin. Each outbound request can pick the next source IP + * via getNext(), ensuring requests are distributed across addresses. + * + * --- Environment Variables ---------------------------------------------------- + * IPV6_SOURCES — Comma-separated list of IPv6 source addresses + * e.g. "2001:df4:c140:1f::d6,2001:df4:c140:1f:ffff:ffff:ffff:ffff" + */ + +const POOL_PREFIX = "[ipv6-pool]"; + +function logPool(msg: string, extra?: Record): void { + const ts = new Date().toISOString().slice(11, 23); + const parts = [`${POOL_PREFIX} ${ts}`, msg]; + if (extra) { + for (const [k, v] of Object.entries(extra)) { + parts.push(`${k}=${v ?? "null"}`); + } + } + console.log(parts.join(" ")); +} + +// --- Types ------------------------------------------------------------------- + +export interface IPv6SourceEntry { + address: string; + /** Number of consecutive failures for this source */ + failures: number; + /** Whether this source is temporarily disabled */ + disabled: boolean; +} + +// --- IPv6SourcePool ---------------------------------------------------------- + +export class IPv6SourcePool { + private sources: IPv6SourceEntry[] = []; + private currentIndex = 0; + private failureThreshold = 3; + + /** + * Load IPv6 source addresses from a comma-separated string. + * + * @param csv - Comma-separated IPv6 addresses (e.g. "addr1,addr2,addr3") + */ + loadFromString(csv: string): void { + if (!csv) return; + const addrs = csv + .split(",") + .map((a) => a.trim()) + .filter(Boolean); + + this.sources = addrs.map((addr) => ({ + address: addr, + failures: 0, + disabled: false, + })); + this.currentIndex = 0; + + logPool(`loaded ${this.sources.length} IPv6 source addresses`); + for (const src of this.sources) { + logPool(` source: ${src.address}`); + } + } + + /** + * Load from environment variable IPV6_SOURCES. + */ + loadFromEnv(): void { + const env = process.env.IPV6_SOURCES; + if (env) { + this.loadFromString(env); + } + } + + // -- Access ------------------------------------------------------------------ + + /** Total number of source addresses. */ + get size(): number { + return this.sources.length; + } + + /** Whether any source addresses are configured. */ + get configured(): boolean { + return this.sources.length > 0; + } + + /** + * Get the next IPv6 source address (round-robin). + * Skips sources that have exceeded the failure threshold. + * + * @returns The next IPv6 address, or null if pool is empty or all failed. + */ + getNext(): string | null { + if (this.sources.length === 0) return null; + + const startIndex = this.currentIndex; + let checked = 0; + + do { + const entry = this.sources[this.currentIndex]; + if (entry && !entry.disabled) { + const addr = entry.address; + // Advance index for next call + this.currentIndex = (this.currentIndex + 1) % this.sources.length; + logPool(`getNext -> ${addr} (index=${this.currentIndex})`); + return addr; + } + this.currentIndex = (this.currentIndex + 1) % this.sources.length; + checked++; + } while (this.currentIndex !== startIndex && checked <= this.sources.length); + + // All sources failed — reset and return first + logPool("all IPv6 sources failed, resetting"); + for (const src of this.sources) { + src.failures = 0; + src.disabled = false; + } + this.currentIndex = 0; + return this.sources[0]?.address ?? null; + } + + /** + * Get a specific source address by index. + * Returns null if out of bounds. + */ + getAtIndex(index: number): string | null { + return this.sources[index]?.address ?? null; + } + + // -- Failure tracking -------------------------------------------------------- + + /** + * Mark a source address as failed. + * If failures exceed threshold, the source is disabled until reset. + * + * @param address - The IPv6 address that failed + */ + markFailed(address: string): void { + const entry = this.sources.find((s) => s.address === address); + if (!entry) return; + + entry.failures++; + logPool(`markFailed ${address} (${entry.failures}/${this.failureThreshold})`); + + if (entry.failures >= this.failureThreshold) { + entry.disabled = true; + console.warn( + `[ipv6-pool] Source ${address} failed ${entry.failures}/${this.failureThreshold} times — disabled`, + ); + } + } + + /** + * Mark a source address as successful (resets failure count). + * + * @param address - The IPv6 address that succeeded + */ + markSuccess(address: string): void { + const entry = this.sources.find((s) => s.address === address); + if (!entry) return; + + if (entry.failures > 0) { + logPool(`markSuccess ${address} (was ${entry.failures} failures)`); + } + entry.failures = 0; + entry.disabled = false; + } + + /** Set the failure threshold (default 3). */ + setFailureThreshold(n: number): void { + this.failureThreshold = n; + } + + /** Reset all failure states. */ + reset(): void { + for (const src of this.sources) { + src.failures = 0; + src.disabled = false; + } + this.currentIndex = 0; + logPool("reset all sources"); + } +}