Merge pull request 'Build on Gitea, push to the Gitea registry' (#1) from ci/gitea-actions into main

This repo now lives on git.ffaerber.duckdns.org and its image is consumed by the
homelab swarm, so the build moves with it: .gitea/workflows/docker.yml builds on
the homelab runner and pushes to this Gitea's registry as an immutable short-SHA
tag plus :latest.

Build-only, on purpose. The GitHub workflow it replaces also SSHed into the
Hostinger VPS and ran `docker stack deploy` on every push to main, which
deploys the machine being retired and makes a deploy a side effect of a push.
The homelab pattern is a PR bumping the tag in ffaerber/homelab, so shipping is
a reviewed commit.

Plain `docker build` rather than buildx: the runner shares the host daemon
through a mounted socket, and buildx's gha cache backend does not exist outside
GitHub.

.github/workflows is removed from this copy so Gitea does not also try to run
it -- it would log in to ghcr with a Gitea token and SSH to a host whose secrets
are not here. The GitHub repository keeps its own copy untouched.

.gitea/workflows/docker.yml

@@ -0,0 +1,49 @@
+name: Docker
+# Builds the hosted-gateway image for the homelab instance and pushes it to this
+# Gitea's own registry. Deliberately build-only: the GitHub workflow this
+# replaces also SSHed into the Hostinger VPS and redeployed it, which is the
+# machine being retired. Here the image is deployed the way everything else in
+# the homelab is -- a PR bumping the tag in ffaerber/homelab's docker-compose.yml
+# -- so a deploy is a reviewed commit rather than a side effect of a push.
+on:
+  push:
+    branches: [main]
+  pull_request:
+  workflow_dispatch:
+
+jobs:
+  build:
+    # The homelab runner (a container on pluto, sharing the host docker daemon).
+    runs-on: ci-runner
+    steps:
+      - uses: actions/checkout@v4
+
+      # Plain docker rather than buildx: the runner already has the host daemon
+      # on a mounted socket, and the gha cache backend buildx wants does not
+      # exist here.
+      - name: Build
+        run: |
+          set -eu
+          SHA=$(echo "${{ github.sha }}" | cut -c1-7)
+          echo "IMAGE=git.ffaerber.duckdns.org/token-for-token/hosted-gateway" >> "$GITHUB_ENV"
+          echo "SHA=$SHA" >> "$GITHUB_ENV"
+          docker build --build-arg COMMIT_SHA=${{ github.sha }} \
+            -t "git.ffaerber.duckdns.org/token-for-token/hosted-gateway:$SHA" .
+
+      # Pull requests build to prove the Dockerfile still works, and stop there.
+      - name: Push
+        if: github.event_name != 'pull_request'
+        run: |
+          set -eu
+          echo "${{ secrets.GITEA_TOKEN }}" | docker login git.ffaerber.duckdns.org -u "${{ github.actor }}" --password-stdin
+          docker tag "$IMAGE:$SHA" "$IMAGE:latest"
+          docker push "$IMAGE:$SHA"
+          docker push "$IMAGE:latest"
+          docker logout git.ffaerber.duckdns.org
+          {
+            echo "### Pushed"
+            echo
+            echo '```'
+            echo "$IMAGE:$SHA"
+            echo '```'
+          } >> "$GITHUB_STEP_SUMMARY"

.github/workflows/docker.yml

@@ -1,60 +0,0 @@
-name: Docker Image
-
-on:
-  push:
-    branches: [main, develop]
-  pull_request:
-    branches: [main]
-
-jobs:
-  build-and-push:
-    runs-on: ubuntu-latest
-    permissions:
-      contents: read
-      packages: write
-
-    steps:
-      - uses: actions/checkout@v4
-
-      - uses: docker/setup-buildx-action@v3
-
-      - uses: docker/login-action@v3
-        with:
-          registry: ghcr.io
-          username: ${{ github.actor }}
-          password: ${{ secrets.GITHUB_TOKEN }}
-
-      - uses: docker/metadata-action@v5
-        id: meta
-        with:
-          images: ghcr.io/token-for-token/hosted-gateway
-          tags: |
-            type=sha
-            type=ref,event=branch
-            type=raw,value=latest,enable={{is_default_branch}}
-          labels: |
-            org.opencontainers.image.source=https://github.com/token-for-token/hosted-gateway
-
-      - uses: docker/build-push-action@v6
-        with:
-          context: .
-          push: ${{ github.event_name != 'pull_request' }}
-          tags: ${{ steps.meta.outputs.tags }}
-          labels: ${{ steps.meta.outputs.labels }}
-          build-args: |
-            COMMIT_SHA=${{ github.sha }}
-          cache-from: type=gha
-          cache-to: type=gha,mode=max
-
-      - name: docker stack deploy
-        if: github.event_name != 'pull_request'
-        uses: appleboy/ssh-action@v1.0.3
-        with:
-          username: root
-          host: ${{ secrets.SSH_KNOWN_HOSTS }}
-          key: ${{ secrets.SSH_PRIV_KEY }}
-          script: |
-            echo ${{ secrets.GITHUB_TOKEN }} | docker login ghcr.io -u ${{ github.actor }} --password-stdin
-            docker pull ghcr.io/token-for-token/hosted-gateway:main
-            docker stack deploy -c ~/devops/hostinger/prod-stack.yaml prod --with-registry-auth --resolve-image always
-            docker image prune -f --filter "until=24h"

README.md

@@ -40,7 +40,6 @@ src/
   env.ts               # zod-validated process.env
   auth/                # JWT, Google OAuth, email signup
   users/, accounts/, apiKeys/, pricing/, gateway/, chain/
-  lib/                 # shared coercion helpers (Decimal → wei string, query params)
   queue/               # connection, registry, lock, processor base, service
   t4t/                 # VENDORED from t4t-container
 prisma/schema.prisma

package.json

@@ -13,7 +13,6 @@
     "db:migrate": "bunx prisma migrate deploy",
     "db:seed": "bun run prisma/seed/index.ts",
     "test:e2e": "bun test --preload ./test/setup.ts ./test/**/*.e2e-spec.ts",
-    "test:unit": "bun test ./test/unit",
     "typecheck": "bunx tsc --noEmit"
   },
   "//comment": "Prisma 7 reads its config from prisma.config.ts at the package root, not from package.json#prisma.",

prisma/migrations/20260805120000_gateway_job_reservation_released_at/migration.sql

@@ -1,27 +0,0 @@
--- AlterTable
-ALTER TABLE "GatewayJob" ADD COLUMN     "reservationReleasedAt" TIMESTAMP(3);
-
--- Data repair: before this migration, `releaseReservation` bailed out whenever
--- the GatewayJob row was already in a terminal status — and both failure paths
--- in `gateway/context.ts` marked the row `cancelled` / `timed_out` *before*
--- calling it. So every job that failed to ACK or to deliver left its
--- `Account.reservedXbzzWei` permanently inflated, shrinking the tenant's
--- available balance for good. Give that stuck xBZZ back, flooring at 0 so a
--- partially-repaired database can never go negative.
-UPDATE "Account" a
-SET "reservedXbzzWei" = GREATEST(a."reservedXbzzWei" - leaked.total, 0)
-FROM (
-  SELECT j."userId" AS "userId", SUM(j."estimatedMaxXbzzWei") AS total
-  FROM "GatewayJob" j
-  WHERE j."status" IN ('cancelled', 'timed_out')
-    AND j."reservationReleasedAt" IS NULL
-  GROUP BY j."userId"
-) AS leaked
-WHERE a."userId" = leaked."userId";
-
--- Backfill the marker for every job whose reservation is no longer outstanding
--- (either repaired above, or already settled by `settleXbzz` on JobClaimed) so
--- a late JobClaimed event can't decrement the same reservation twice.
-UPDATE "GatewayJob"
-SET "reservationReleasedAt" = COALESCE("claimedAt", "updatedAt")
-WHERE "status" IN ('claimed', 'cancelled', 'timed_out', 'failed');

prisma/schema.prisma

@@ -176,13 +176,6 @@ model GatewayJob {
   ackDeadline          DateTime?
   deliveryDeadline     DateTime?
 
-  // Set the first time this job's off-chain reservation is released — by
-  // `settleXbzz` on JobClaimed, or by `releaseReservation` on cancel / timeout
-  // / dispatch failure. It is the compare-and-set marker that makes the
-  // release exactly-once: `status` cannot serve that role because a job can be
-  // marked terminal and released by different code paths in either order.
-  reservationReleasedAt DateTime?
-
   // Token accounting (from JobClaimed actualPayment)
   promptTokens         Int?
   completionTokens     Int?

src/accounts/ledger.ts

@@ -13,12 +13,6 @@ import { logger } from '../logger';
  * account), not as ledger rows — the ledger means "money moved" not "money
  * locked". On `JobClaimed` the reservation is released AND a `settle` ledger
  * row debits the actual cost.
- *
- * A reservation must be released exactly once, and the race is real: the
- * failure timers, `handleChat`'s catch block and the claim watcher can all
- * reach the same job. `GatewayJob.reservationReleasedAt` is the compare-and-set
- * marker — whichever transaction flips it from NULL owns the decrement, and
- * everyone else no-ops.
  */
 
 export class InsufficientBalanceError extends Error {
@@ -84,32 +78,9 @@ export async function reserveXbzz(userId: string, amountWei: bigint): Promise<vo
   });
 }
 
-/**
- * Claim the right to release a job's reservation. Returns true for the caller
- * that flipped `reservationReleasedAt` from NULL, false for everyone after it
- * (including a job row that no longer exists).
- *
- * Concurrent callers serialize on the row lock `updateMany` takes; under READ
- * COMMITTED the loser re-evaluates the `reservationReleasedAt: null` predicate
- * against the committed row and matches nothing.
- */
-async function claimReservationRelease(
-  tx: Prisma.TransactionClient,
-  gatewayJobId: string,
-): Promise<boolean> {
-  const { count } = await tx.gatewayJob.updateMany({
-    where: { id: gatewayJobId, reservationReleasedAt: null },
-    data: { reservationReleasedAt: new Date() },
-  });
-  return count > 0;
-}
-
 /**
  * Settle a completed job: release the reservation and debit the actual cost.
- * The balance debit is idempotent via the `settle:<gatewayJobId>` ledger key;
- * the reservation release is idempotent via `reservationReleasedAt`, which also
- * covers the case where a failure path already released it (job cancelled
- * locally but claimed on-chain anyway).
+ * Idempotent via the `settle:<gatewayJobId>` ledger key.
  */
 export async function settleXbzz(
   userId: string,
@@ -129,14 +100,11 @@ export async function settleXbzz(
           idempotencyKey: key,
         },
       });
-      const releasing = await claimReservationRelease(tx, gatewayJobId);
       await tx.account.update({
         where: { userId },
         data: {
           balanceXbzzWei: { decrement: new Prisma.Decimal(actualXbzzWei.toString()) },
-          ...(releasing
-            ? { reservedXbzzWei: { decrement: new Prisma.Decimal(estimateXbzzWei.toString()) } }
-            : {}),
+          reservedXbzzWei: { decrement: new Prisma.Decimal(estimateXbzzWei.toString()) },
         },
       });
     });
@@ -150,10 +118,10 @@ export async function settleXbzz(
 }
 
 /**
- * Release a reservation without debiting (job cancelled, timed out, or never
- * dispatched). No ledger row — no money moved, only the lock lifted. Safe to
- * call from every failure path and in any order relative to the GatewayJob
- * status transition: `reservationReleasedAt` decides who actually decrements.
+ * Release a reservation without debiting (job cancelled or timed out).
+ * Idempotent via `release:<gatewayJobId>` — but we don't write a ledger row,
+ * we only adjust `reservedXbzzWei` and rely on the GatewayJob status
+ * transition to mark the release as having happened.
  */
 export async function releaseReservation(
   userId: string,
@@ -161,7 +129,14 @@ export async function releaseReservation(
   estimateXbzzWei: bigint,
 ): Promise<void> {
   await prisma.$transaction(async (tx) => {
-    if (!(await claimReservationRelease(tx, gatewayJobId))) return;
+    const job = await tx.gatewayJob.findUniqueOrThrow({
+      where: { id: gatewayJobId },
+      select: { status: true },
+    });
+    // Only release once — `cancelled` / `timed_out` / `failed` are terminal.
+    if (job.status === 'cancelled' || job.status === 'timed_out' || job.status === 'failed') {
+      return;
+    }
     await tx.account.update({
       where: { userId },
       data: { reservedXbzzWei: { decrement: new Prisma.Decimal(estimateXbzzWei.toString()) } },

src/accounts/routes.ts

@@ -2,8 +2,6 @@ import Elysia from 'elysia';
 import { prisma } from '../db';
 import { jwtGuard } from '../auth/jwt';
 import { getUsdEquivalent } from '../pricing/usdDisplay';
-import { weiString, weiToBigInt } from '../lib/decimal';
-import { parseLimit } from '../lib/query';
 
 export const accountsRoutes = new Elysia({ prefix: '/accounts' })
   .use(jwtGuard)
@@ -20,8 +18,8 @@ export const accountsRoutes = new Elysia({ prefix: '/accounts' })
       },
     });
 
-    const balanceWei = weiToBigInt(account.balanceXbzzWei);
-    const reservedWei = weiToBigInt(account.reservedXbzzWei);
+    const balanceWei = BigInt(account.balanceXbzzWei.toString());
+    const reservedWei = BigInt(account.reservedXbzzWei.toString());
     const available = balanceWei - reservedWei;
 
     const balanceUsd = await getUsdEquivalent(balanceWei).catch(() => null);
@@ -30,8 +28,8 @@ export const accountsRoutes = new Elysia({ prefix: '/accounts' })
     return {
       id: account.id,
       status: account.status,
-      balanceXbzzWei: weiString(account.balanceXbzzWei),
-      reservedXbzzWei: weiString(account.reservedXbzzWei),
+      balanceXbzzWei: account.balanceXbzzWei.toString(),
+      reservedXbzzWei: account.reservedXbzzWei.toString(),
       availableXbzzWei: available.toString(),
       balanceUsd,        // cosmetic; null if oracle unavailable
       availableUsd,
@@ -40,7 +38,7 @@ export const accountsRoutes = new Elysia({ prefix: '/accounts' })
     };
   })
   .get('/me/ledger', async ({ user, query }) => {
-    const limit = parseLimit(query.limit);
+    const limit = Math.min(Number(query.limit ?? 50), 200);
     const account = await prisma.account.findUniqueOrThrow({
       where: { userId: user.userId },
       select: { id: true },
@@ -59,5 +57,5 @@ export const accountsRoutes = new Elysia({ prefix: '/accounts' })
         note: true,
       },
     });
-    return entries.map((e) => ({ ...e, deltaXbzzWei: weiString(e.deltaXbzzWei) }));
+    return entries.map((e) => ({ ...e, deltaXbzzWei: e.deltaXbzzWei.toString() }));
   });

src/chain/claimWatcher.ts

@@ -15,7 +15,6 @@ import { redisConnection } from '../queue/connection';
 import { logger } from '../logger';
 import { prisma } from '../db';
 import { settleXbzz } from '../accounts/ledger';
-import { weiToBigInt } from '../lib/decimal';
 import { applyMarkup } from '../pricing/estimate';
 import { getGatewayContext } from '../gateway/context';
 
@@ -89,7 +88,7 @@ async function handleClaimed(jobId: Hex | undefined, paid: bigint | undefined):
   }
   if (job.status === 'claimed') return; // idempotent
 
-  const estimateWithMarkup = weiToBigInt(job.estimatedMaxXbzzWei);
+  const estimateWithMarkup = BigInt(job.estimatedMaxXbzzWei.toString());
   const actualWithMarkup = applyMarkup(paid);
 
   await settleXbzz(job.userId, job.id, estimateWithMarkup, actualWithMarkup);

src/gateway/context.ts

@@ -133,41 +133,19 @@ async function bootGatewayContext(): Promise<GatewayContext> {
     clearFailureTimers(routing);
   }
 
-  /** Lift the tenant's reservation, then record the terminal status. The order
-   *  no longer matters for correctness (`reservationReleasedAt` is the
-   *  exactly-once marker, not the job status), but a failure to release must
-   *  never stop us from settling the request — otherwise the caller hangs. */
-  async function releaseAndMark(
-    routing: Hex,
-    meta: JobMetaSlot,
-    status: 'cancelled' | 'timed_out' | 'failed',
-    reason: string,
-  ): Promise<void> {
-    await releaseReservation(meta.userId, meta.gatewayJobId, meta.estimateMaxXbzzWei).catch((err) =>
-      logger.error({ err, routing, gatewayJobId: meta.gatewayJobId }, 'reservation release failed'),
-    );
-    await markGatewayJob(meta.gatewayJobId, status, reason);
-  }
-
   async function onAckTimeout(routing: Hex): Promise<void> {
     const slot = failureTimers.get(routing);
     const meta = jobMeta.get(routing);
     if (!slot || !meta) return;
-    slot.cancelTimer = undefined;
     try {
       const tx = await cancelJob(chain, slot.onChainJobId);
       logger.warn({ tx, routing, onChainJobId: slot.onChainJobId }, 'no ACK — cancelled on-chain');
+      await markGatewayJob(meta.gatewayJobId, 'cancelled', 'no PSS ack within ACK_WINDOW');
+      await releaseReservation(meta.userId, meta.gatewayJobId, meta.estimateMaxXbzzWei);
+      rejectAndCleanup(routing, new Error('provider failed to ACK within window'));
     } catch (err) {
-      // Most likely the provider acked mid-flight, so the job is still live and
-      // cancelling it is no longer allowed. We are done waiting on the ACK
-      // either way — hand the job to the delivery deadline so *something* still
-      // settles it.
-      logger.warn({ err, routing }, 'cancelJob failed — falling through to the delivery deadline');
-      armDeliveryTimeout(routing, slot);
-      return;
+      logger.warn({ err, routing }, 'cancelJob failed — provider likely acked mid-flight');
     }
-    await releaseAndMark(routing, meta, 'cancelled', 'no PSS ack within ACK_WINDOW');
-    rejectAndCleanup(routing, new Error('provider failed to ACK within window'));
   }
 
   async function onDeliveryTimeout(routing: Hex): Promise<void> {
@@ -177,7 +155,8 @@ async function bootGatewayContext(): Promise<GatewayContext> {
     try {
       const tx = await timeoutJob(chain, slot.onChainJobId);
       logger.warn({ tx, routing, onChainJobId: slot.onChainJobId }, 'delivery missed — timeoutJob');
-      await releaseAndMark(routing, meta, 'timed_out', 'no PSS delivery before deadline');
+      await markGatewayJob(meta.gatewayJobId, 'timed_out', 'no PSS delivery before deadline');
+      await releaseReservation(meta.userId, meta.gatewayJobId, meta.estimateMaxXbzzWei);
       rejectAndCleanup(routing, new Error('provider failed to deliver before deadline'));
       return;
     } catch (err) {
@@ -186,14 +165,11 @@ async function bootGatewayContext(): Promise<GatewayContext> {
     try {
       const tx = await cancelJob(chain, slot.onChainJobId);
       logger.warn({ tx, routing, onChainJobId: slot.onChainJobId }, 'cancelJob fallback applied');
-      await releaseAndMark(routing, meta, 'cancelled', 'no PSS delivery; cancelJob fallback');
+      await markGatewayJob(meta.gatewayJobId, 'cancelled', 'no PSS delivery; cancelJob fallback');
+      await releaseReservation(meta.userId, meta.gatewayJobId, meta.estimateMaxXbzzWei);
       rejectAndCleanup(routing, new Error('provider failed to deliver before deadline'));
     } catch (err) {
       logger.error({ err, routing }, 'both timeoutJob and cancelJob failed — job stuck in escrow');
-      // The escrow entry is stuck, but the tenant shouldn't be: release the
-      // reservation. If the provider does eventually claim, `settleXbzz` still
-      // debits the actual cost and skips the (already taken) release.
-      await releaseAndMark(routing, meta, 'failed', 'delivery deadline missed; settlement failed');
       rejectAndCleanup(routing, new Error('delivery deadline missed; settlement failed'));
     }
   }
@@ -206,83 +182,50 @@ async function bootGatewayContext(): Promise<GatewayContext> {
     }, (ACK_WINDOW_SECONDS + FAILURE_GRACE_SECONDS) * 1000);
   }
 
-  /** Arm the delivery-deadline timer once. Every path that stops waiting for
-   *  an ACK must call this, otherwise nothing is left to settle the job and the
-   *  chat request hangs until the client gives up — holding a semaphore slot
-   *  and the tenant's reservation for as long as the process lives. */
-  function armDeliveryTimeout(routing: Hex, slot: FailureSlot): void {
-    if (slot.timeoutTimer) return;
-    const msUntilTimeout = Math.max(
-      0,
-      (slot.deliveryDeadline + FAILURE_GRACE_SECONDS) * 1000 - Date.now(),
-    );
-    slot.timeoutTimer = setTimeout(() => {
-      onDeliveryTimeout(routing).catch((err) =>
-        logger.error({ err, routing }, 'onDeliveryTimeout threw'),
-      );
-    }, msUntilTimeout);
-  }
-
-  /**
-   * An envelope only counts if it was signed by the provider this gateway
-   * actually posted the job to. `verifyEnvelope` proves the signature matches
-   * `envelope.from` — it says nothing about who `from` is, and the routing id
-   * plus our PSS public key both travel in the (unencrypted) `job_notify` body.
-   * Without this check anyone who observes a routing id can sign their own
-   * `job_deliver` and hand the tenant an arbitrary completion.
-   */
-  function envelopeSenderMatches(
-    envelope: Envelope,
-    meta: JobMetaSlot | undefined,
-  ): meta is JobMetaSlot {
-    if (!meta) return false;
-    return String(envelope.from).toLowerCase() === meta.provider.toLowerCase();
-  }
-
   // Handler for any verified envelope addressed to this gateway, regardless
   // of which transport (PSS subscription or HTTPS /pss/inbox) brought it.
   const onEnvelope = async (envelope: Envelope): Promise<void> => {
     if (envelope.type === 'job_ack') {
       const body = envelope.body as JobAckBody;
+      logger.info({ jobId: body.jobId, eta: body.estimatedCompletion }, 'ack received');
       const slot = failureTimers.get(body.jobId);
       const meta = jobMeta.get(body.jobId);
-      if (!envelopeSenderMatches(envelope, meta)) {
-        logger.warn(
-          { jobId: body.jobId, from: envelope.from, type: envelope.type },
-          'dropping envelope — not from the provider this job was posted to',
-        );
-        return;
-      }
-      logger.info({ jobId: body.jobId, eta: body.estimatedCompletion }, 'ack received');
       if (slot) {
         if (slot.cancelTimer) {
           clearTimeout(slot.cancelTimer);
           slot.cancelTimer = undefined;
         }
-        armDeliveryTimeout(body.jobId, slot);
+        if (!slot.timeoutTimer) {
+          const msUntilTimeout = Math.max(
+            0,
+            (slot.deliveryDeadline + FAILURE_GRACE_SECONDS) * 1000 - Date.now(),
+          );
+          slot.timeoutTimer = setTimeout(() => {
+            onDeliveryTimeout(body.jobId).catch((err) =>
+              logger.error({ err, routing: body.jobId }, 'onDeliveryTimeout threw'),
+            );
+          }, msUntilTimeout);
+        }
+      }
+      if (meta) {
+        await markGatewayJob(meta.gatewayJobId, 'acked', null, { ackedAt: new Date() });
       }
-      await markGatewayJob(meta.gatewayJobId, 'acked', null, { ackedAt: new Date() });
     } else if (envelope.type === 'job_deliver') {
       const body = envelope.body as JobDeliverBody;
       const slot = pending.get(body.jobId);
       const meta = jobMeta.get(body.jobId);
       if (!slot) return;
-      if (!envelopeSenderMatches(envelope, meta)) {
-        logger.warn(
-          { jobId: body.jobId, from: envelope.from, type: envelope.type },
-          'dropping envelope — not from the provider this job was posted to',
-        );
-        return;
-      }
       try {
         const bytes = await downloadChunk({ bee, postageBatchId, logger }, body.responseHash);
         const payload = await jsonDecrypt<ResponsePayload>(cipher, bytes);
-        await markGatewayJob(meta.gatewayJobId, 'delivered', null, {
-          deliveredAt: new Date(),
-          responseHash: body.responseHash,
-          promptTokens: payload.openaiResponse.usage?.prompt_tokens ?? null,
-          completionTokens: payload.openaiResponse.usage?.completion_tokens ?? null,
-        });
+        if (meta) {
+          await markGatewayJob(meta.gatewayJobId, 'delivered', null, {
+            deliveredAt: new Date(),
+            responseHash: body.responseHash,
+            promptTokens: payload.openaiResponse.usage?.prompt_tokens ?? null,
+            completionTokens: payload.openaiResponse.usage?.completion_tokens ?? null,
+          });
+        }
         slot.resolve(payload.openaiResponse);
       } catch (err) {
         slot.reject(err);

src/gateway/pssInbox.ts

@@ -11,11 +11,9 @@ import { logger } from '../logger';
  * hand off to the shared envelope handler.
  *
  * Lives outside the JWT-guarded routes because providers don't have JWTs.
- * Security model: anyone can POST, but an envelope only produces side effects
- * if (a) its signature recovers to `envelope.from`, checked here, and (b)
- * `envelope.from` is the provider the gateway posted that job to, checked in
- * `onEnvelope`. Everything else is dropped. The signature check alone is not
- * enough — it authenticates the sender, it does not authorize them.
+ * Security model: anyone can POST, but only envelopes signed by a wallet the
+ * gateway is actively waiting on (`pending.has(jobId)`) ever produce side
+ * effects. Unsigned or unmatched envelopes are dropped.
  */
 export const pssInboxRoutes = new Elysia().post(
   '/pss/inbox',

src/gateway/routes.ts

@@ -7,8 +7,6 @@ import { Semaphore } from '../queue/semaphore';
 import { env } from '../env';
 import { logger } from '../logger';
 import { InsufficientBalanceError } from '../accounts/ledger';
-import { weiString } from '../lib/decimal';
-import { parseLimit } from '../lib/query';
 
 const semaphore = new Semaphore(env.MAX_CONCURRENT_JOBS);
 
@@ -75,10 +73,7 @@ export const gatewayRoutes = new Elysia()
             }),
           ),
           temperature: t.Optional(t.Number()),
-          // Integer-and-positive is enforced here rather than in the pricing
-          // math: a fractional `max_tokens` makes `BigInt()` throw (500), and a
-          // negative one budgets a negative reservation.
-          max_tokens: t.Optional(t.Integer({ minimum: 1 })),
+          max_tokens: t.Optional(t.Number()),
           stream: t.Optional(t.Boolean()),
         },
         { additionalProperties: true },
@@ -86,7 +81,7 @@ export const gatewayRoutes = new Elysia()
     },
   )
   .get('/v1/jobs', async ({ user, query }) => {
-    const limit = parseLimit(query.limit);
+    const limit = Math.min(Number(query.limit ?? 50), 200);
     const jobs = await prisma.gatewayJob.findMany({
       where: { userId: user.userId },
       orderBy: { createdAt: 'desc' },
@@ -109,7 +104,7 @@ export const gatewayRoutes = new Elysia()
     });
     return jobs.map((j) => ({
       ...j,
-      estimatedMaxXbzzWei: weiString(j.estimatedMaxXbzzWei),
-      actualXbzzWei: j.actualXbzzWei ? weiString(j.actualXbzzWei) : null,
+      estimatedMaxXbzzWei: j.estimatedMaxXbzzWei.toString(),
+      actualXbzzWei: j.actualXbzzWei?.toString() ?? null,
     }));
   });

src/lib/decimal.ts

@@ -1,22 +0,0 @@
-import type { Prisma } from '@prisma/client';
-
-/**
- * Rendering helpers for the `Decimal(78, 0)` wei columns.
- *
- * `Prisma.Decimal` is decimal.js under the hood, and decimal.js' `toString()`
- * switches to exponential notation once the exponent reaches 21 — so a balance
- * of 1000 xBZZ stringifies as `"1e+21"`, not `"1000000000000000000000"`. That
- * string is not a valid BigInt literal (`BigInt("1e+21")` throws) and it is not
- * something an API client can parse as an integer either. `toFixed()` always
- * emits plain digits, which is what wei values need everywhere.
- */
-
-/** Plain-digit decimal string for a wei column — never exponential notation. */
-export function weiString(value: Prisma.Decimal): string {
-  return value.toFixed();
-}
-
-/** Same, as a BigInt. */
-export function weiToBigInt(value: Prisma.Decimal): bigint {
-  return BigInt(value.toFixed());
-}

src/lib/query.ts

@@ -1,14 +0,0 @@
-/**
- * Query-string coercion helpers. Everything in `query` is a string (or absent)
- * and is attacker-controlled, so it needs clamping before it reaches Prisma —
- * `take: NaN` from `?limit=abc` fails the query with a 500 rather than the 400
- * the caller deserves, and a negative `take` silently paginates backwards.
- */
-export function parseLimit(raw: unknown, fallback = 50, max = 200): number {
-  if (raw === undefined || raw === null) return fallback;
-  // `?limit=` parses as 0, which would clamp to 1 rather than mean "unset".
-  if (typeof raw === 'string' && raw.trim() === '') return fallback;
-  const n = Number(raw);
-  if (!Number.isFinite(n)) return fallback;
-  return Math.min(Math.max(Math.trunc(n), 1), max);
-}

src/pricing/estimate.ts

@@ -43,15 +43,7 @@ export function estimateMaxXbzz(
   req: ChatRequest,
   offering: Pick<ModelOffering, 'inputPricePerMillionTokens' | 'outputPricePerMillionTokens' | 'maxContextTokens'>,
 ): { rawWei: bigint; withMarkupWei: bigint } {
-  // `max_tokens` reaches us straight off the wire. A fractional value makes
-  // `BigInt()` throw and a negative one budgets a negative reservation (which
-  // would *credit* the caller), so anything that isn't a positive integer falls
-  // back to the default.
-  const requested = req.max_tokens;
-  const maxTokens =
-    typeof requested === 'number' && Number.isInteger(requested) && requested > 0
-      ? BigInt(requested)
-      : DEFAULT_MAX_TOKENS;
+  const maxTokens = BigInt(req.max_tokens ?? DEFAULT_MAX_TOKENS);
   const budget = maxTokens + HEADROOM_TOKENS;
   const inPay = BigInt(offering.inputPricePerMillionTokens) * budget;
   const outPay = BigInt(offering.outputPricePerMillionTokens) * budget;

src/queue/service.ts

@@ -50,13 +50,9 @@ export function createCudService<T>(
     },
 
     async delete(args: { where: any }): Promise<T> {
-      // Snapshot before the delete (the row is gone afterwards), but enqueue
-      // only once it has actually happened — otherwise a delete that throws
-      // still tells every consumer the entity is gone.
       const entity = await prismaDelegate.findUnique({ where: args.where });
-      const deleted = await prismaDelegate.delete(args);
-      await queue.add('deleted', { [queueName]: serializeBigInts(entity ?? deleted) });
-      return deleted;
+      await queue.add('deleted', { [queueName]: serializeBigInts(entity) });
+      return prismaDelegate.delete(args);
     },
   };
 }

test/unit/wei-and-limits.spec.ts

@@ -1,94 +0,0 @@
-/**
- * Pure-function regressions for the wei/query coercion helpers. No Postgres,
- * Redis, Bee or RPC — run with `bun run test:unit`.
- */
-import { describe, expect, it } from 'bun:test';
-import { Prisma } from '@prisma/client';
-import { weiString, weiToBigInt } from '../../src/lib/decimal';
-import { parseLimit } from '../../src/lib/query';
-
-// `src/pricing/estimate` pulls in the zod-validated env module, which exits the
-// process when required vars are missing. Fill in the minimum before importing.
-process.env.DATABASE_URL ??= 'postgresql://user:pass@localhost:5432/test';
-process.env.JWT_SECRET_KEY ??= 'test-secret-at-least-16-chars';
-process.env.OPERATOR_PRIVATE_KEY ??= '0x' + '11'.repeat(32);
-process.env.GNOSIS_RPC_URL ??= 'http://localhost:8545';
-process.env.REGISTRY_ADDRESS ??= '0x' + '22'.repeat(20);
-process.env.ESCROW_ADDRESS ??= '0x' + '33'.repeat(20);
-process.env.XBZZ_ADDRESS ??= '0x' + '44'.repeat(20);
-process.env.BEE_API_URL ??= 'http://localhost:1633';
-process.env.MARKUP_BPS ??= '2000';
-const { estimateMaxXbzz } = await import('../../src/pricing/estimate');
-
-const OFFERING = {
-  inputPricePerMillionTokens: 10n ** 18n,
-  outputPricePerMillionTokens: 2n * 10n ** 18n,
-  maxContextTokens: 0n,
-};
-
-describe('wei rendering', () => {
-  it('keeps large balances in plain notation', () => {
-    // decimal.js' toString() flips to exponential at 1e21 — i.e. at 1000 xBZZ,
-    // which is an ordinary balance. "1e+21" is not a valid BigInt literal.
-    const thousandXbzz = new Prisma.Decimal('1000000000000000000000');
-    expect(thousandXbzz.toString()).toBe('1e+21'); // the trap being guarded
-    expect(weiString(thousandXbzz)).toBe('1000000000000000000000');
-    expect(weiToBigInt(thousandXbzz)).toBe(1000000000000000000000n);
-  });
-
-  it('round-trips a full uint256', () => {
-    const max = (2n ** 256n - 1n).toString();
-    expect(weiToBigInt(new Prisma.Decimal(max))).toBe(2n ** 256n - 1n);
-  });
-
-  it('handles negative deltas and zero', () => {
-    expect(weiString(new Prisma.Decimal('-5000000000000000000000'))).toBe('-5000000000000000000000');
-    expect(weiString(new Prisma.Decimal(0))).toBe('0');
-  });
-});
-
-describe('parseLimit', () => {
-  it('clamps to the allowed range', () => {
-    expect(parseLimit('10')).toBe(10);
-    expect(parseLimit('1000')).toBe(200);
-    expect(parseLimit('-5')).toBe(1);
-    expect(parseLimit('0')).toBe(1);
-  });
-
-  it('falls back instead of handing Prisma a NaN take', () => {
-    expect(parseLimit('abc')).toBe(50);
-    expect(parseLimit(undefined)).toBe(50);
-    expect(parseLimit('')).toBe(50); // Number('') is 0, but empty means "unset"
-  });
-
-  it('truncates fractional input', () => {
-    expect(parseLimit('12.9')).toBe(12);
-  });
-});
-
-describe('estimateMaxXbzz', () => {
-  it('never returns a negative reservation', () => {
-    for (const max_tokens of [-1000, -1, 0]) {
-      const { rawWei, withMarkupWei } = estimateMaxXbzz({ model: 'm', max_tokens }, OFFERING);
-      expect(rawWei > 0n).toBe(true);
-      expect(withMarkupWei >= rawWei).toBe(true);
-    }
-  });
-
-  it('does not throw on a fractional max_tokens', () => {
-    const { rawWei } = estimateMaxXbzz({ model: 'm', max_tokens: 100.5 }, OFFERING);
-    const { rawWei: dflt } = estimateMaxXbzz({ model: 'm' }, OFFERING);
-    expect(rawWei).toBe(dflt);
-  });
-
-  it('scales with max_tokens', () => {
-    const small = estimateMaxXbzz({ model: 'm', max_tokens: 1 }, OFFERING).rawWei;
-    const large = estimateMaxXbzz({ model: 'm', max_tokens: 100_000 }, OFFERING).rawWei;
-    expect(large > small).toBe(true);
-  });
-
-  it('applies the markup on top of the on-chain amount', () => {
-    const { rawWei, withMarkupWei } = estimateMaxXbzz({ model: 'm', max_tokens: 1024 }, OFFERING);
-    expect(withMarkupWei).toBe((rawWei * 12_000n) / 10_000n);
-  });
-});