diff --git a/src/api.ts b/src/api.ts index cfab0e8..f360716 100644 --- a/src/api.ts +++ b/src/api.ts @@ -8,6 +8,7 @@ import { usersRoutes } from './users/routes'; import { accountsRoutes } from './accounts/routes'; import { apiKeysRoutes } from './apiKeys/routes'; import { gatewayRoutes } from './gateway/routes'; +import { pssInboxRoutes } from './gateway/pssInbox'; import { beeRoutes } from './bee/routes'; import { getGatewayContext } from './gateway/context'; @@ -27,6 +28,7 @@ const app = new Elysia() .use(accountsRoutes) .use(apiKeysRoutes) .use(beeRoutes) + .use(pssInboxRoutes) .use(gatewayRoutes); // Start the HTTP server immediately so auth / accounts / api-keys routes are diff --git a/src/env.ts b/src/env.ts index 4052ca5..d367ad2 100644 --- a/src/env.ts +++ b/src/env.ts @@ -56,6 +56,10 @@ const EnvSchema = z.object({ // Server HOSTED_GATEWAY_PORT: z.coerce.number().int().positive().default(8080), + // Public base URL of this gateway, advertised to providers in job_notify + // envelopes as `clientReplyUrl`. Providers POST ack/deliver here when PSS + // reverse delivery is unreliable (e.g. Bee 2.8 push-routing skips us). + PUBLIC_API_URL: z.string().url().optional(), LOG_LEVEL: z.enum(['fatal', 'error', 'warn', 'info', 'debug', 'trace']).default('info'), // Operator PSS keypair file. All `prod_api` and `prod_worker` replicas diff --git a/src/gateway/context.ts b/src/gateway/context.ts index 6c29af6..6dc62b0 100644 --- a/src/gateway/context.ts +++ b/src/gateway/context.ts @@ -17,6 +17,7 @@ import { EciesCipher, jsonDecrypt } from '../t4t/lib/crypto'; import { loadOrCreatePssKey } from '../t4t/lib/keys'; import { ModelDiscovery } from '../t4t/modes/gateway/models'; import type { + Envelope, JobAckBody, JobDeliverBody, OpenAIChatResponse, @@ -66,6 +67,11 @@ export interface GatewayContext { clearFailureTimers: (routing: Hex) => void; rejectAndCleanup: (routing: Hex, err: Error) => void; scheduleAckTimeout: (routing: Hex, slot: FailureSlot) => void; + /** Process an inbound signed envelope from any transport (PSS or HTTPS). + * The caller is responsible for verifyEnvelope; the PSS subscription + * layer already does this. The HTTPS inbox endpoint must do it before + * invoking. */ + onEnvelope: (envelope: Envelope) => Promise; } let bootPromise: Promise | null = null; @@ -176,62 +182,66 @@ async function bootGatewayContext(): Promise { }, (ACK_WINDOW_SECONDS + FAILURE_GRACE_SECONDS) * 1000); } + // 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 => { + 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 (slot) { + if (slot.cancelTimer) { + clearTimeout(slot.cancelTimer); + slot.cancelTimer = undefined; + } + 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() }); + } + } 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; + try { + const bytes = await downloadChunk({ bee, postageBatchId, logger }, body.responseHash); + const payload = await jsonDecrypt(cipher, bytes); + 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); + } finally { + pending.delete(body.jobId); + jobMeta.delete(body.jobId); + clearFailureTimers(body.jobId); + } + } + }; + // Subscribe to the operator's client topic — ALL tenants' jobs come back // here (because all jobs go on-chain as the operator wallet). pss.subscribe({ topic: clientTopic(chain.address), - onEnvelope: async (envelope) => { - 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 (slot) { - if (slot.cancelTimer) { - clearTimeout(slot.cancelTimer); - slot.cancelTimer = undefined; - } - 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() }); - } - } 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; - try { - const bytes = await downloadChunk({ bee, postageBatchId, logger }, body.responseHash); - const payload = await jsonDecrypt(cipher, bytes); - 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); - } finally { - pending.delete(body.jobId); - jobMeta.delete(body.jobId); - clearFailureTimers(body.jobId); - } - } - }, + onEnvelope, }); logger.info( @@ -259,6 +269,7 @@ async function bootGatewayContext(): Promise { clearFailureTimers, rejectAndCleanup, scheduleAckTimeout, + onEnvelope, }; } diff --git a/src/gateway/handleChat.ts b/src/gateway/handleChat.ts index 309a735..38380ac 100644 --- a/src/gateway/handleChat.ts +++ b/src/gateway/handleChat.ts @@ -138,6 +138,7 @@ export async function handleChat(args: HandleChatArgs): Promise ctx.chain.wallet.signMessage({ account: ctx.chain.wallet.account!, message: msg }), diff --git a/src/gateway/pssInbox.ts b/src/gateway/pssInbox.ts new file mode 100644 index 0000000..0a1656d --- /dev/null +++ b/src/gateway/pssInbox.ts @@ -0,0 +1,44 @@ +import Elysia, { t } from 'elysia'; +import { getGatewayContext } from './context'; +import { verifyEnvelope } from '../t4t/lib/envelope'; +import type { Envelope } from '../t4t/lib/types'; +import { logger } from '../logger'; + +/** + * HTTPS fallback for the reverse-direction PSS path. The provider POSTs the + * same signed envelopes (`job_ack`, `job_deliver`) it would have sent via PSS. + * We verify the signature here — same trust check as the PSS subscription — and + * 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 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', + async ({ body, set }) => { + const env = body as Envelope; + if (!(await verifyEnvelope(env))) { + set.status = 400; + logger.warn({ from: env.from, type: env.type }, 'pss/inbox: invalid envelope signature'); + return { error: 'invalid envelope signature' }; + } + const ctx = await getGatewayContext(); + await ctx.onEnvelope(env); + logger.info({ from: env.from, type: env.type }, 'pss/inbox: envelope dispatched'); + return { ok: true }; + }, + { + body: t.Object({ + v: t.Number(), + type: t.Union([t.Literal('job_notify'), t.Literal('job_ack'), t.Literal('job_deliver')]), + from: t.String(), + to: t.String(), + ts: t.Number(), + nonce: t.String(), + sig: t.String(), + body: t.Any(), + }), + }, +); diff --git a/src/t4t/lib/types.ts b/src/t4t/lib/types.ts index 9457135..c5ee32e 100644 --- a/src/t4t/lib/types.ts +++ b/src/t4t/lib/types.ts @@ -36,6 +36,11 @@ export interface JobNotifyBody { clientPssPubKey: Hex /** Gateway's Bee overlay for PSS routing. */ clientSwarmOverlay: Hex + /** Optional HTTPS URL the provider may POST ack/deliver envelopes to, + * for clients (like the hosted gateway) where reverse-direction PSS + * delivery isn't reliable. The provider trusts this because the + * envelope is signed by the gateway's on-chain wallet. */ + clientReplyUrl?: string } export interface JobAckBody {