diff --git a/.antd-version b/.antd-version index 87a1cf5..6345c21 100644 --- a/.antd-version +++ b/.antd-version @@ -1 +1 @@ -v0.12.0 +v0.13.0 diff --git a/Dockerfile b/Dockerfile index e185cb4..99dd3e2 100644 --- a/Dockerfile +++ b/Dockerfile @@ -7,7 +7,7 @@ # platform); keep ANTD_IMAGE in lockstep with .antd-version. release.yml # passes the pinned tag explicitly; this default keeps `docker compose # up --build` and bare `docker build` working out of the box. -ARG ANTD_IMAGE=ghcr.io/withautonomi/antd:v0.12.0 +ARG ANTD_IMAGE=ghcr.io/withautonomi/antd:v0.13.0 FROM ${ANTD_IMAGE} AS antd # Build frontend on the native arch — JS output is arch-independent. diff --git a/README.md b/README.md index 8093133..7463258 100644 --- a/README.md +++ b/README.md @@ -75,7 +75,14 @@ data_dir = "./data" # Required jwt_secret = "your-secret-key-at-least-32-chars" -wallet_encryption_key = "64-hex-char-key-for-aes-256-gcm" +wallet_encryption_key = "64-hex-char-key-for-aes-256-gcm" # local payment backend only; optional with payment_backend = "hosted" + +# Payments: "local" (default) signs uploads with a wallet you add in the admin UI; +# "hosted" pays through the Autonomi Pay gateway from prepaid credits — no wallet, +# no EVM RPC on this instance. Requires antd >= 0.13.0. +# payment_backend = "hosted" +# payment_gateway_url = "https://pay.example.com" +# payment_gateway_api_key = "pgk_..." # Bootstrap admin — seeds the first admin on a fresh DB (self-registration is # off by default; the server won't start with no admin and no seed). @@ -183,7 +190,10 @@ curl -X POST /api/v2/tokens \ | `INDELIBLE_JWT_SECRET` | **Required.** Secret for JWT signing; **minimum 32 characters** (the server refuses to start below this). Generate with `openssl rand -hex 32` | -- | | `INDELIBLE_JWT_SECRET_PREVIOUS` | Comma-separated **verify-only** former JWT secrets, kept during a rotation so live sessions survive. New tokens always sign with `INDELIBLE_JWT_SECRET`; these only verify already-issued tokens until they expire. See [key-rotation guide](docs/guides/key-rotation.md#rotating-the-jwt-secret) | -- | | `INDELIBLE_JWT_SECRET_FILE` | Path to a file holding the JWT secret (Docker/K8s secrets); takes precedence over `INDELIBLE_JWT_SECRET` | -- | -| `INDELIBLE_WALLET_ENCRYPTION_KEY` | **Required.** 64-char hex key for wallet encryption (AES-256-GCM) | -- | +| `INDELIBLE_WALLET_ENCRYPTION_KEY` | **Required for the local payment backend.** 64-char hex key for wallet + OIDC client-secret encryption (AES-256-GCM). Optional with `INDELIBLE_PAYMENT_BACKEND=hosted` (no wallet exists); set it there only if you want OIDC login, whose client secrets it also encrypts | -- | +| `INDELIBLE_PAYMENT_BACKEND` | Who pays for uploads: `local` (this instance's wallet signs) or `hosted` (the Autonomi Pay gateway pays from the tenant's prepaid credits; no wallet, no EVM RPC on this instance; antd >= 0.13.0). Unknown values refuse to start. Not to be confused with the upload API's `payment_mode` (`auto`/`merkle`/`single`), which is how a payment is structured on-chain | `local` | +| `INDELIBLE_PAYMENT_GATEWAY_URL` | Gateway base URL. **Required** when the backend is `hosted` | -- | +| `INDELIBLE_PAYMENT_GATEWAY_API_KEY` | Tenant API key issued by the gateway (Bearer). Required when the backend is `hosted` | -- | | `INDELIBLE_WALLET_ENCRYPTION_KEY_FILE` | Path to a file holding the wallet encryption key (Docker/K8s secrets); takes precedence over `INDELIBLE_WALLET_ENCRYPTION_KEY` | -- | | `INDELIBLE_WALLET_ENCRYPTION_KEY_PREVIOUS` | Comma-separated **decrypt-only** former wallet keys, so the running service can read rows not yet re-encrypted during a rotation. See [key-rotation guide](docs/guides/key-rotation.md#rotating-the-wallet-encryption-key) | -- | | `INDELIBLE_SECRETS_BACKEND` | Where key material is sourced from. `env` sources from env / config-file / `_FILE`. Other backends (Vault, cloud KMS) plug in behind the same provider seam | `env` | diff --git a/cmd/indelible/main.go b/cmd/indelible/main.go index d173ed0..5a022a7 100644 --- a/cmd/indelible/main.go +++ b/cmd/indelible/main.go @@ -85,6 +85,15 @@ func main() { slog.SetDefault(logger) slog.Info("starting indelible", "version", buildinfo.Version, "port", cfg.Port, "db_driver", cfg.DBDriver()) + if cfg.PaymentBackend.Hosted() { + // V2-929: uploads are paid by the gateway from prepaid credits — no + // wallet record, no EVM RPC, and the wallet encryption key is optional + // (it only gates OIDC client-secret storage here). + slog.Info("payment backend: hosted — uploads paid by the gateway, no wallet or EVM RPC on this instance", + "gateway", cfg.PaymentGatewayURL, "wallet_key_configured", cfg.WalletKeyConfigured()) + } else { + slog.Info("payment backend: local — uploads signed with the instance wallet", "network", cfg.Network) + } // Managed antd var antdMgr *managedantd.Manager diff --git a/docs/docs.go b/docs/docs.go index e2bb6ad..d6aadc6 100644 --- a/docs/docs.go +++ b/docs/docs.go @@ -144,6 +144,157 @@ const docTemplate = `{ } } }, + "/admin/billing": { + "get": { + "security": [ + { + "BearerAuth": [] + } + ], + "description": "Hosted-mode billing: gateway credits and credited top-up history", + "produces": [ + "application/json" + ], + "tags": [ + "Admin: Billing" + ], + "summary": "Billing summary", + "responses": { + "200": { + "description": "OK", + "schema": { + "type": "object", + "additionalProperties": true + } + }, + "400": { + "description": "Not in hosted payment mode", + "schema": { + "type": "object", + "additionalProperties": { + "type": "string" + } + } + } + } + } + }, + "/admin/billing/topup-checkout": { + "post": { + "security": [ + { + "BearerAuth": [] + } + ], + "description": "Create a Stripe Checkout session at the gateway; returns the hosted payment page URL and the exact credit", + "consumes": [ + "application/json" + ], + "produces": [ + "application/json" + ], + "tags": [ + "Admin: Billing" + ], + "summary": "Start a card top-up", + "parameters": [ + { + "description": "Amount in USD cents and return URLs", + "name": "body", + "in": "body", + "required": true, + "schema": { + "$ref": "#/definitions/internal_handlers.topupCheckoutRequest" + } + } + ], + "responses": { + "200": { + "description": "session_id, url, credit_atto", + "schema": { + "type": "object", + "additionalProperties": true + } + }, + "400": { + "description": "Bad Request", + "schema": { + "type": "object", + "additionalProperties": { + "type": "string" + } + } + }, + "502": { + "description": "Gateway unreachable", + "schema": { + "type": "object", + "additionalProperties": { + "type": "string" + } + } + } + } + } + }, + "/admin/billing/topup-sync": { + "post": { + "security": [ + { + "BearerAuth": [] + } + ], + "description": "Ask the gateway to retrieve the Checkout session from Stripe and credit it if paid (idempotent; webhook-loss fallback)", + "consumes": [ + "application/json" + ], + "produces": [ + "application/json" + ], + "tags": [ + "Admin: Billing" + ], + "summary": "Sync a top-up", + "parameters": [ + { + "description": "Checkout session id", + "name": "body", + "in": "body", + "required": true, + "schema": { + "$ref": "#/definitions/internal_handlers.topupSyncRequest" + } + } + ], + "responses": { + "200": { + "description": "credited, payment_status", + "schema": { + "type": "object", + "additionalProperties": true + } + }, + "400": { + "description": "Bad Request", + "schema": { + "type": "object", + "additionalProperties": { + "type": "string" + } + } + }, + "502": { + "description": "Gateway unreachable", + "schema": { + "type": "object", + "additionalProperties": { + "type": "string" + } + } + } + } + } + }, "/admin/departments": { "get": { "security": [ @@ -5663,7 +5814,7 @@ const docTemplate = `{ "BearerAuth": [] } ], - "description": "Get an exact cost quote by sending the file bytes. antd runs self-encryption and queries the live network for chunk pricing — no estimation, no scaling. Returns a structured estimated_cost object with cost, chunk_count, gas, and payment_mode.", + "description": "Get an exact cost quote by sending the file bytes. antd runs self-encryption and queries the live network for chunk pricing — no estimation, no scaling. Returns a structured estimated_cost object with cost, chunk_count, gas, and payment_mode (antd's on-chain payment strategy: auto | merkle | single). With the hosted payment backend the gateway debits gross — batch total plus a per-batch network fee (V2-1098) — so the response additionally carries gateway_fee_per_batch_atto, estimated_batch_count, and estimated_total_with_fee_atto (V2-1113).", "consumes": [ "multipart/form-data" ], @@ -7264,6 +7415,29 @@ const docTemplate = `{ } } }, + "internal_handlers.topupCheckoutRequest": { + "type": "object", + "properties": { + "amount_usd_cents": { + "type": "integer" + }, + "cancel_url": { + "type": "string" + }, + "success_url": { + "description": "Absolute URLs back into this instance's UI; validated by the gateway\n(http/https only). Stripe substitutes {CHECKOUT_SESSION_ID} in\nsuccess_url if the placeholder is present.", + "type": "string" + } + } + }, + "internal_handlers.topupSyncRequest": { + "type": "object", + "properties": { + "session_id": { + "type": "string" + } + } + }, "internal_handlers.updateGroupRequest": { "type": "object", "properties": { @@ -7411,12 +7585,23 @@ const docTemplate = `{ "filename": { "type": "string" }, + "gateway_fee_atto": { + "description": "GatewayFeeAtto itemizes the gateway's per-batch network fee out of the\ngross actual_cost (V2-1098); absent when no fee was charged.", + "type": "string" + }, + "gateway_payment_key": { + "type": "string" + }, "last_quoted_cost": { "type": "string" }, "original_filename": { "type": "string" }, + "payment_backend": { + "description": "Payment provenance (V2-1086): \"local\" (instance wallet) or \"hosted\"\n(gateway credits) + the gateway's batch key; absent when nothing was paid.", + "type": "string" + }, "processing_at": { "type": "string" }, diff --git a/docs/swagger.json b/docs/swagger.json index bb322c1..3f32fa8 100644 --- a/docs/swagger.json +++ b/docs/swagger.json @@ -138,6 +138,157 @@ } } }, + "/admin/billing": { + "get": { + "security": [ + { + "BearerAuth": [] + } + ], + "description": "Hosted-mode billing: gateway credits and credited top-up history", + "produces": [ + "application/json" + ], + "tags": [ + "Admin: Billing" + ], + "summary": "Billing summary", + "responses": { + "200": { + "description": "OK", + "schema": { + "type": "object", + "additionalProperties": true + } + }, + "400": { + "description": "Not in hosted payment mode", + "schema": { + "type": "object", + "additionalProperties": { + "type": "string" + } + } + } + } + } + }, + "/admin/billing/topup-checkout": { + "post": { + "security": [ + { + "BearerAuth": [] + } + ], + "description": "Create a Stripe Checkout session at the gateway; returns the hosted payment page URL and the exact credit", + "consumes": [ + "application/json" + ], + "produces": [ + "application/json" + ], + "tags": [ + "Admin: Billing" + ], + "summary": "Start a card top-up", + "parameters": [ + { + "description": "Amount in USD cents and return URLs", + "name": "body", + "in": "body", + "required": true, + "schema": { + "$ref": "#/definitions/internal_handlers.topupCheckoutRequest" + } + } + ], + "responses": { + "200": { + "description": "session_id, url, credit_atto", + "schema": { + "type": "object", + "additionalProperties": true + } + }, + "400": { + "description": "Bad Request", + "schema": { + "type": "object", + "additionalProperties": { + "type": "string" + } + } + }, + "502": { + "description": "Gateway unreachable", + "schema": { + "type": "object", + "additionalProperties": { + "type": "string" + } + } + } + } + } + }, + "/admin/billing/topup-sync": { + "post": { + "security": [ + { + "BearerAuth": [] + } + ], + "description": "Ask the gateway to retrieve the Checkout session from Stripe and credit it if paid (idempotent; webhook-loss fallback)", + "consumes": [ + "application/json" + ], + "produces": [ + "application/json" + ], + "tags": [ + "Admin: Billing" + ], + "summary": "Sync a top-up", + "parameters": [ + { + "description": "Checkout session id", + "name": "body", + "in": "body", + "required": true, + "schema": { + "$ref": "#/definitions/internal_handlers.topupSyncRequest" + } + } + ], + "responses": { + "200": { + "description": "credited, payment_status", + "schema": { + "type": "object", + "additionalProperties": true + } + }, + "400": { + "description": "Bad Request", + "schema": { + "type": "object", + "additionalProperties": { + "type": "string" + } + } + }, + "502": { + "description": "Gateway unreachable", + "schema": { + "type": "object", + "additionalProperties": { + "type": "string" + } + } + } + } + } + }, "/admin/departments": { "get": { "security": [ @@ -5657,7 +5808,7 @@ "BearerAuth": [] } ], - "description": "Get an exact cost quote by sending the file bytes. antd runs self-encryption and queries the live network for chunk pricing — no estimation, no scaling. Returns a structured estimated_cost object with cost, chunk_count, gas, and payment_mode.", + "description": "Get an exact cost quote by sending the file bytes. antd runs self-encryption and queries the live network for chunk pricing — no estimation, no scaling. Returns a structured estimated_cost object with cost, chunk_count, gas, and payment_mode (antd's on-chain payment strategy: auto | merkle | single). With the hosted payment backend the gateway debits gross — batch total plus a per-batch network fee (V2-1098) — so the response additionally carries gateway_fee_per_batch_atto, estimated_batch_count, and estimated_total_with_fee_atto (V2-1113).", "consumes": [ "multipart/form-data" ], @@ -7258,6 +7409,29 @@ } } }, + "internal_handlers.topupCheckoutRequest": { + "type": "object", + "properties": { + "amount_usd_cents": { + "type": "integer" + }, + "cancel_url": { + "type": "string" + }, + "success_url": { + "description": "Absolute URLs back into this instance's UI; validated by the gateway\n(http/https only). Stripe substitutes {CHECKOUT_SESSION_ID} in\nsuccess_url if the placeholder is present.", + "type": "string" + } + } + }, + "internal_handlers.topupSyncRequest": { + "type": "object", + "properties": { + "session_id": { + "type": "string" + } + } + }, "internal_handlers.updateGroupRequest": { "type": "object", "properties": { @@ -7405,12 +7579,23 @@ "filename": { "type": "string" }, + "gateway_fee_atto": { + "description": "GatewayFeeAtto itemizes the gateway's per-batch network fee out of the\ngross actual_cost (V2-1098); absent when no fee was charged.", + "type": "string" + }, + "gateway_payment_key": { + "type": "string" + }, "last_quoted_cost": { "type": "string" }, "original_filename": { "type": "string" }, + "payment_backend": { + "description": "Payment provenance (V2-1086): \"local\" (instance wallet) or \"hosted\"\n(gateway credits) + the gateway's batch key; absent when nothing was paid.", + "type": "string" + }, "processing_at": { "type": "string" }, diff --git a/docs/swagger.yaml b/docs/swagger.yaml index 90530c7..3840f2d 100644 --- a/docs/swagger.yaml +++ b/docs/swagger.yaml @@ -677,6 +677,24 @@ definitions: uuid: type: string type: object + internal_handlers.topupCheckoutRequest: + properties: + amount_usd_cents: + type: integer + cancel_url: + type: string + success_url: + description: |- + Absolute URLs back into this instance's UI; validated by the gateway + (http/https only). Stripe substitutes {CHECKOUT_SESSION_ID} in + success_url if the placeholder is present. + type: string + type: object + internal_handlers.topupSyncRequest: + properties: + session_id: + type: string + type: object internal_handlers.updateGroupRequest: properties: description: @@ -774,10 +792,22 @@ definitions: type: integer filename: type: string + gateway_fee_atto: + description: |- + GatewayFeeAtto itemizes the gateway's per-batch network fee out of the + gross actual_cost (V2-1098); absent when no fee was charged. + type: string + gateway_payment_key: + type: string last_quoted_cost: type: string original_filename: type: string + payment_backend: + description: |- + Payment provenance (V2-1086): "local" (instance wallet) or "hosted" + (gateway credits) + the gateway's batch key; absent when nothing was paid. + type: string processing_at: type: string queued_at: @@ -937,6 +967,104 @@ paths: summary: Get upload analytics tags: - 'Admin: Analytics' + /admin/billing: + get: + description: 'Hosted-mode billing: gateway credits and credited top-up history' + produces: + - application/json + responses: + "200": + description: OK + schema: + additionalProperties: true + type: object + "400": + description: Not in hosted payment mode + schema: + additionalProperties: + type: string + type: object + security: + - BearerAuth: [] + summary: Billing summary + tags: + - 'Admin: Billing' + /admin/billing/topup-checkout: + post: + consumes: + - application/json + description: Create a Stripe Checkout session at the gateway; returns the hosted + payment page URL and the exact credit + parameters: + - description: Amount in USD cents and return URLs + in: body + name: body + required: true + schema: + $ref: '#/definitions/internal_handlers.topupCheckoutRequest' + produces: + - application/json + responses: + "200": + description: session_id, url, credit_atto + schema: + additionalProperties: true + type: object + "400": + description: Bad Request + schema: + additionalProperties: + type: string + type: object + "502": + description: Gateway unreachable + schema: + additionalProperties: + type: string + type: object + security: + - BearerAuth: [] + summary: Start a card top-up + tags: + - 'Admin: Billing' + /admin/billing/topup-sync: + post: + consumes: + - application/json + description: Ask the gateway to retrieve the Checkout session from Stripe and + credit it if paid (idempotent; webhook-loss fallback) + parameters: + - description: Checkout session id + in: body + name: body + required: true + schema: + $ref: '#/definitions/internal_handlers.topupSyncRequest' + produces: + - application/json + responses: + "200": + description: credited, payment_status + schema: + additionalProperties: true + type: object + "400": + description: Bad Request + schema: + additionalProperties: + type: string + type: object + "502": + description: Gateway unreachable + schema: + additionalProperties: + type: string + type: object + security: + - BearerAuth: [] + summary: Sync a top-up + tags: + - 'Admin: Billing' /admin/departments: get: description: Return the distinct department labels in use across API tokens, @@ -4851,10 +4979,13 @@ paths: post: consumes: - multipart/form-data - description: Get an exact cost quote by sending the file bytes. antd runs self-encryption + description: 'Get an exact cost quote by sending the file bytes. antd runs self-encryption and queries the live network for chunk pricing — no estimation, no scaling. Returns a structured estimated_cost object with cost, chunk_count, gas, and - payment_mode. + payment_mode (antd''s on-chain payment strategy: auto | merkle | single). + With the hosted payment backend the gateway debits gross — batch total plus + a per-batch network fee (V2-1098) — so the response additionally carries gateway_fee_per_batch_atto, + estimated_batch_count, and estimated_total_with_fee_atto (V2-1113).' parameters: - description: File to quote in: formData diff --git a/go.mod b/go.mod index 419c8a0..d29bbc0 100644 --- a/go.mod +++ b/go.mod @@ -4,7 +4,7 @@ go 1.25.13 require ( github.com/BurntSushi/toml v1.6.0 - github.com/WithAutonomi/ant-sdk/antd-go v0.12.0 + github.com/WithAutonomi/ant-sdk/antd-go v0.13.0 github.com/coreos/go-oidc/v3 v3.18.0 github.com/elimity-com/scim v0.0.0-20240320110924-172bf2aee9c8 github.com/ethereum/go-ethereum v1.17.3 diff --git a/go.sum b/go.sum index 93f6530..cd7b369 100644 --- a/go.sum +++ b/go.sum @@ -12,8 +12,8 @@ github.com/StackExchange/wmi v1.2.1 h1:VIkavFPXSjcnS+O8yTq7NI32k0R5Aj+v39y29VYDO github.com/StackExchange/wmi v1.2.1/go.mod h1:rcmrprowKIVzvc+NUiLncP2uuArMWLCbu9SBzvHz7e8= github.com/VictoriaMetrics/fastcache v1.13.0 h1:AW4mheMR5Vd9FkAPUv+NH6Nhw+fmbTMGMsNAoA/+4G0= github.com/VictoriaMetrics/fastcache v1.13.0/go.mod h1:hHXhl4DA2fTL2HTZDJFXWgW0LNjo6B+4aj2Wmng3TjU= -github.com/WithAutonomi/ant-sdk/antd-go v0.12.0 h1:Yt8lCXU6C2yiTA5sHCo3ozGn5sAEdqFRRRv6H+xXSZI= -github.com/WithAutonomi/ant-sdk/antd-go v0.12.0/go.mod h1:5JtaWzf87rJ3YqPRfQDyHJXrLEyakeAJyLIj7cspHVs= +github.com/WithAutonomi/ant-sdk/antd-go v0.13.0 h1:2w1Ol5AA2IcoIr8QLHy+BigRi0Lf5PDZC/H5MKVwX0k= +github.com/WithAutonomi/ant-sdk/antd-go v0.13.0/go.mod h1:5JtaWzf87rJ3YqPRfQDyHJXrLEyakeAJyLIj7cspHVs= github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= github.com/bits-and-blooms/bitset v1.20.0 h1:2F+rfL86jE2d/bmw7OhqUg2Sj/1rURkBn3MdfoPyRVU= diff --git a/internal/config/config.go b/internal/config/config.go index 87cf2de..28af7f0 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -15,6 +15,42 @@ import ( // Config holds all application configuration. Values can be set via // config file (TOML) or environment variables (INDELIBLE_ prefix). // Environment variables take precedence over file values. +// PaymentBackend names the payment system that settles uploads for this +// instance. It is the single place the backend set is defined: adding one +// means a new constant here, a case in each method below, and a payer in +// internal/worker — callers ask the backend what it needs (NeedsWallet, +// WantsSignedQuotes) rather than comparing its name. +type PaymentBackend string + +const ( + // PaymentBackendLocal signs payments with this instance's default wallet. + PaymentBackendLocal PaymentBackend = "local" + // PaymentBackendHosted delegates payment to the Autonomi Pay gateway, + // which pays from the tenant's prepaid credits; no wallet on the host. + PaymentBackendHosted PaymentBackend = "hosted" +) + +// Valid reports whether b is a known backend. +func (b PaymentBackend) Valid() bool { + switch b { + case PaymentBackendLocal, PaymentBackendHosted: + return true + } + return false +} + +// Hosted reports whether the gateway settles payments for this instance. +func (b PaymentBackend) Hosted() bool { return b == PaymentBackendHosted } + +// NeedsWallet reports whether uploads require a wallet record and its +// decrypted key on this instance. Only local signing does. +func (b PaymentBackend) NeedsWallet() bool { return b == PaymentBackendLocal } + +// WantsSignedQuotes reports whether upload prepare must return the signed +// quotes so the settling party can verify the batch offline before paying +// (V2-926). Only a remote payer needs them; local signing trusts its own antd. +func (b PaymentBackend) WantsSignedQuotes() bool { return b == PaymentBackendHosted } + type Config struct { Port int `toml:"port"` DBURL string `toml:"db_url"` @@ -75,6 +111,18 @@ type Config struct { EvmRPCURL string `toml:"evm_rpc_url"` // EVM RPC endpoint EvmTokenAddress string `toml:"evm_token_address"` // Payment token contract address + // PaymentBackend selects which payment system settles uploads (V2-929): + // PaymentBackendLocal signs with the instance wallet; PaymentBackendHosted + // POSTs each upload's payment batch to the gateway at PaymentGatewayURL. + // Exactly one backend is active per instance. Not to be confused with + // antd's per-request payment_mode (auto | merkle | single), which is how + // a payment is structured on-chain, not who pays for it. + PaymentBackend PaymentBackend `toml:"payment_backend"` // "local" (default) or "hosted" + PaymentGatewayURL string `toml:"payment_gateway_url"` // required when payment_backend=hosted + // PaymentGatewayAPIKey authenticates this instance's tenant account at + // the gateway (Bearer). Required when payment_backend=hosted. + PaymentGatewayAPIKey string `toml:"payment_gateway_api_key"` + // SMTP configuration for transactional emails (password reset, email verification) SMTP SMTPConfig `toml:"smtp"` @@ -94,8 +142,9 @@ type Config struct { walletKeyring *crypto.Keyring jwtKeyring *crypto.Keyring - // walletKeyUnconfigured is set by Load only for a reader replica that booted - // without a real wallet key (V2-518). It gates WalletKeyConfigured(). Kept as + // walletKeyUnconfigured is set by Load for an instance that booted without a + // real wallet key: a reader replica (V2-518) or a hosted-backend writer + // (V2-929). It gates WalletKeyConfigured(). Kept as // a flag (not derived from the key value) so a directly-constructed Config — // e.g. tests that use the all-zeros placeholder as a working key — still // reports the key as configured. @@ -107,8 +156,9 @@ type Config struct { func (c *Config) Secrets() secrets.Provider { return c.secrets } // WalletKeyConfigured reports whether the instance has a usable wallet/OIDC -// encryption key. It is false only on a reader replica that Load booted without -// one (V2-518). Callers that ENCRYPT wallet or OIDC secrets must refuse when +// encryption key. It is false when Load booted without one, which is allowed +// for a reader replica (V2-518) and for a writer on the hosted payment backend +// (V2-929). Callers that ENCRYPT wallet or OIDC secrets must refuse when // this is false, rather than seal data under the placeholder key into the shared // database (which the writer, holding the real key, could not decrypt). func (c *Config) WalletKeyConfigured() bool { @@ -382,6 +432,32 @@ func Load(path string) (*Config, error) { if v := os.Getenv("INDELIBLE_EVM_TOKEN_ADDRESS"); v != "" { cfg.EvmTokenAddress = v } + if v := os.Getenv("INDELIBLE_PAYMENT_BACKEND"); v != "" { + cfg.PaymentBackend = PaymentBackend(v) + } + if v := os.Getenv("INDELIBLE_PAYMENT_GATEWAY_URL"); v != "" { + cfg.PaymentGatewayURL = v + } + if v := os.Getenv("INDELIBLE_PAYMENT_GATEWAY_API_KEY"); v != "" { + cfg.PaymentGatewayAPIKey = v + } + + // Payment backend: default local; anything outside the known set is a + // typo, not a new backend, so refuse to start rather than silently + // signing with the wallet. Hosted needs the gateway address up front — + // failing here beats failing on the first upload. + if cfg.PaymentBackend == "" { + cfg.PaymentBackend = PaymentBackendLocal + } + if !cfg.PaymentBackend.Valid() { + return nil, fmt.Errorf("payment_backend %q is not supported (INDELIBLE_PAYMENT_BACKEND / payment_backend in config): use %q or %q", cfg.PaymentBackend, PaymentBackendLocal, PaymentBackendHosted) + } + if cfg.PaymentBackend.Hosted() && cfg.PaymentGatewayURL == "" { + return nil, fmt.Errorf("payment_backend=hosted requires payment_gateway_url (INDELIBLE_PAYMENT_GATEWAY_URL)") + } + if cfg.PaymentBackend.Hosted() && cfg.PaymentGatewayAPIKey == "" { + return nil, fmt.Errorf("payment_backend=hosted requires payment_gateway_api_key (INDELIBLE_PAYMENT_GATEWAY_API_KEY): without it the gateway answers 401 on the first upload") + } // Default antd binary if cfg.AntdBin == "" { @@ -412,22 +488,25 @@ func Load(path string) (*Config, error) { } } - // Require wallet encryption key — except on reader replicas (V2-518). A - // reader (WorkersEnabled=false) never decrypts an EVM wallet or OIDC client - // secret: the worker tier is off, and OIDC login / wallet admin run on the - // writer. So it boots without the key. An empty wallet keyring is still built - // below (NewKeyring tolerates ""), so the unused wallet/OIDC routes error - // cleanly rather than panic if reached. JWT_SECRET is still required for - // everyone — readers verify sessions and API tokens against the DB. + // Require the wallet encryption key only where a wallet can exist: a writer + // on the local payment backend. Two roles boot without it: + // - a reader replica (V2-518, WorkersEnabled=false) never decrypts an EVM + // wallet or OIDC client secret — the worker tier is off, and OIDC login / + // wallet admin run on the writer; + // - a hosted-backend writer (V2-929) has no wallet at all — the payment + // gateway's treasury signs, so there is nothing to encrypt. It MAY still + // set the key to keep OIDC client-secret storage available; without it, + // wallet/OIDC create refuse (503) exactly as on a reader. + // In both cases an all-zeros placeholder keyring is still built (NewKeyring + // tolerates it) so the unused wallet/OIDC routes error cleanly rather than + // panic, and the key is flagged unconfigured so encrypt entry points refuse + // rather than seal data under the placeholder. JWT_SECRET is still required + // for everyone — sessions and API tokens are verified against the DB. const placeholderWalletKey = "0000000000000000000000000000000000000000000000000000000000000000" if cfg.WalletEncryptionKey == "" || cfg.WalletEncryptionKey == placeholderWalletKey { - if cfg.WorkersEnabled { - return nil, fmt.Errorf("wallet_encryption_key is required (set INDELIBLE_WALLET_ENCRYPTION_KEY or wallet_encryption_key in config); generate with: openssl rand -hex 32") + if cfg.WorkersEnabled && cfg.PaymentBackend.NeedsWallet() { + return nil, fmt.Errorf("wallet_encryption_key is required for the local payment backend (set INDELIBLE_WALLET_ENCRYPTION_KEY or wallet_encryption_key in config; generate with: openssl rand -hex 32) — not needed with payment_backend=hosted") } - // Reader role: no real wallet key. Pin the all-zeros placeholder (a valid - // 32-byte key) so the *unused* wallet keyring still constructs and nothing - // nil-derefs, and flag the key as unconfigured so encrypt entry points - // (wallet/OIDC create) refuse rather than seal data under the placeholder. cfg.WalletEncryptionKey = placeholderWalletKey cfg.walletKeyUnconfigured = true } diff --git a/internal/config/config_test.go b/internal/config/config_test.go index ae71765..2007c3e 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -302,3 +302,122 @@ func TestLoad_DownloadCacheMaxBytesEnvInvalid(t *testing.T) { } } } + +func TestLoad_PaymentBackendDefaultsLocal(t *testing.T) { + setRequiredSecrets(t) + + cfg, err := Load("") + if err != nil { + t.Fatalf("Load: %v", err) + } + if cfg.PaymentBackend != PaymentBackendLocal { + t.Errorf("PaymentBackend = %q, want %q by default", cfg.PaymentBackend, PaymentBackendLocal) + } + if !cfg.PaymentBackend.NeedsWallet() || cfg.PaymentBackend.Hosted() || cfg.PaymentBackend.WantsSignedQuotes() { + t.Error("local backend should need a wallet, not be hosted, and not want signed quotes") + } +} + +func TestLoad_PaymentBackendUnknownRejected(t *testing.T) { + // A typo must not fall through to wallet signing. + setRequiredSecrets(t) + t.Setenv("INDELIBLE_PAYMENT_BACKEND", "hsoted") + + if _, err := Load(""); err == nil { + t.Fatal("expected Load to reject an unknown payment_backend") + } +} + +func TestLoad_PaymentBackendHostedRequiresGatewayURL(t *testing.T) { + setRequiredSecrets(t) + t.Setenv("INDELIBLE_PAYMENT_BACKEND", "hosted") + // Intentionally no INDELIBLE_PAYMENT_GATEWAY_URL. + + if _, err := Load(""); err == nil { + t.Fatal("expected Load to fail: hosted backend without a gateway URL") + } +} + +func TestLoad_PaymentBackendHostedFromEnv(t *testing.T) { + setRequiredSecrets(t) + t.Setenv("INDELIBLE_PAYMENT_BACKEND", "hosted") + t.Setenv("INDELIBLE_PAYMENT_GATEWAY_URL", "http://gateway.test:8090") + t.Setenv("INDELIBLE_PAYMENT_GATEWAY_API_KEY", "pgk_test") + + cfg, err := Load("") + if err != nil { + t.Fatalf("Load: %v", err) + } + if !cfg.PaymentBackend.Hosted() { + t.Errorf("PaymentBackend = %q, want hosted", cfg.PaymentBackend) + } + if cfg.PaymentBackend.NeedsWallet() { + t.Error("hosted backend must not require a wallet on the instance") + } + if !cfg.PaymentBackend.WantsSignedQuotes() { + t.Error("hosted backend must ask prepare for signed quotes (V2-926)") + } +} + +func TestLoad_HostedBackendBootsWithoutWalletKey(t *testing.T) { + // V2-929: a hosted-backend writer has no wallet to encrypt, so the wallet + // key is optional. It boots with workers ON, flags the key unconfigured + // (wallet/OIDC create refuse), and still builds a placeholder keyring. + t.Setenv("INDELIBLE_JWT_SECRET", "test-secret-at-least-32-bytes-long-xx") + t.Setenv("INDELIBLE_PAYMENT_BACKEND", "hosted") + t.Setenv("INDELIBLE_PAYMENT_GATEWAY_URL", "http://gateway.test:8090") + t.Setenv("INDELIBLE_PAYMENT_GATEWAY_API_KEY", "pgk_test") + // Intentionally no INDELIBLE_WALLET_ENCRYPTION_KEY; workers default to enabled. + + cfg, err := Load("") + if err != nil { + t.Fatalf("hosted Load without wallet key should succeed, got: %v", err) + } + if !cfg.WorkersEnabled { + t.Error("WorkersEnabled = false, want true (this is a writer)") + } + if cfg.WalletKeyConfigured() { + t.Error("WalletKeyConfigured() = true, want false without a real key") + } + if cfg.WalletKeyring() == nil { + t.Error("WalletKeyring() = nil, want a placeholder keyring") + } +} + +func TestLoad_HostedBackendKeepsWalletKeyWhenSet(t *testing.T) { + // Setting the key on a hosted writer keeps OIDC client-secret storage usable. + setRequiredSecrets(t) + t.Setenv("INDELIBLE_PAYMENT_BACKEND", "hosted") + t.Setenv("INDELIBLE_PAYMENT_GATEWAY_URL", "http://gateway.test:8090") + t.Setenv("INDELIBLE_PAYMENT_GATEWAY_API_KEY", "pgk_test") + + cfg, err := Load("") + if err != nil { + t.Fatalf("Load: %v", err) + } + if !cfg.WalletKeyConfigured() { + t.Error("WalletKeyConfigured() = false, want true when the key is set") + } +} + +func TestLoad_LocalBackendStillRequiresWalletKey(t *testing.T) { + // The relaxation is hosted-only; local signing still needs the key. + t.Setenv("INDELIBLE_JWT_SECRET", "test-secret-at-least-32-bytes-long-xx") + t.Setenv("INDELIBLE_PAYMENT_BACKEND", "local") + + if _, err := Load(""); err == nil { + t.Fatal("expected Load to fail: local backend, workers on, no wallet key") + } +} + +func TestLoad_PaymentBackendHostedRequiresAPIKey(t *testing.T) { + // Fail at boot, not on the first upload's 401. + setRequiredSecrets(t) + t.Setenv("INDELIBLE_PAYMENT_BACKEND", "hosted") + t.Setenv("INDELIBLE_PAYMENT_GATEWAY_URL", "http://gateway.test:8090") + // Intentionally no INDELIBLE_PAYMENT_GATEWAY_API_KEY. + + if _, err := Load(""); err == nil { + t.Fatal("expected Load to fail: hosted backend without a gateway API key") + } +} diff --git a/internal/database/busy_timeout_test.go b/internal/database/busy_timeout_test.go new file mode 100644 index 0000000..ac715cc --- /dev/null +++ b/internal/database/busy_timeout_test.go @@ -0,0 +1,99 @@ +package database + +import ( + "context" + "database/sql" + "path/filepath" + "sync" + "testing" +) + +// TestBusyTimeoutOnEveryPooledConnection pins the fix for the SQLITE_BUSY +// concurrent-finalize failure (exposed when V2-1112 coalescing made three +// uploads complete in the same instant): busy_timeout is a connection-level +// pragma, so it must ride the DSN (applied by modernc.org/sqlite to every +// connection it opens), not a one-shot Exec that covers a single pooled +// connection. +func TestBusyTimeoutOnEveryPooledConnection(t *testing.T) { + db, err := Open("sqlite://" + filepath.Join(t.TempDir(), "busy.db")) + if err != nil { + t.Fatal(err) + } + defer db.Close() + + // Hold several distinct pool connections open at once and read the pragma + // on each — with the old Exec-once approach only the first would be 5000. + ctx := context.Background() + const n = 4 + conns := make([]*sql.Conn, 0, n) + defer func() { + for _, c := range conns { + c.Close() + } + }() + for i := 0; i < n; i++ { + c, err := db.Conn(ctx) + if err != nil { + t.Fatal(err) + } + conns = append(conns, c) + } + for i, c := range conns { + var timeout int + if err := c.QueryRowContext(ctx, "PRAGMA busy_timeout").Scan(&timeout); err != nil { + t.Fatal(err) + } + if timeout != 5000 { + t.Fatalf("connection %d: busy_timeout = %d, want 5000", i, timeout) + } + } +} + +// TestConcurrentWritersDoNotHitBusy is the behavioral half: many goroutines +// writing transactionally at the same instant must all succeed (the +// busy_timeout makes writers queue instead of failing "database is locked"). +func TestConcurrentWritersDoNotHitBusy(t *testing.T) { + db, err := Open("sqlite://" + filepath.Join(t.TempDir(), "writers.db")) + if err != nil { + t.Fatal(err) + } + defer db.Close() + if _, err := db.Exec(`CREATE TABLE t (id INTEGER PRIMARY KEY AUTOINCREMENT, v TEXT)`); err != nil { + t.Fatal(err) + } + + const writers = 8 + var wg sync.WaitGroup + errs := make(chan error, writers) + for i := 0; i < writers; i++ { + wg.Add(1) + go func() { + defer wg.Done() + tx, err := db.Begin() + if err != nil { + errs <- err + return + } + if _, err := tx.Exec(`INSERT INTO t (v) VALUES (?)`, "x"); err != nil { + tx.Rollback() + errs <- err + return + } + errs <- tx.Commit() + }() + } + wg.Wait() + close(errs) + for err := range errs { + if err != nil { + t.Fatalf("concurrent writer failed: %v", err) + } + } + var count int + if err := db.QueryRow(`SELECT count(*) FROM t`).Scan(&count); err != nil { + t.Fatal(err) + } + if count != writers { + t.Fatalf("count = %d, want %d", count, writers) + } +} diff --git a/internal/database/database.go b/internal/database/database.go index eb4ba39..47cbb1d 100644 --- a/internal/database/database.go +++ b/internal/database/database.go @@ -44,15 +44,15 @@ func Open(dbURL string) (*DB, error) { return nil, fmt.Errorf("opening database: %w", err) } - // SQLite pragmas for performance. (foreign_keys is set per-connection via the - // DSN in parseURL — see the note there; a one-shot Exec would only cover one - // pooled connection.) + // journal_mode=WAL is database-persistent, so a one-shot Exec is enough (and + // doubles as the SQLITE_CANTOPEN probe for sqliteOpenError). busy_timeout and + // synchronous are *connection-level* and live in the DSN via parseURL — an + // Exec here would cover exactly one pooled connection, leaving the rest with + // busy_timeout=0: concurrent finalizes then fail SQLITE_BUSY the moment two + // uploads complete in the same instant (first seen when V2-1112 coalescing + // made three payments settle in one tx). if driver == "sqlite" { - if _, err := db.Exec(` - PRAGMA journal_mode=WAL; - PRAGMA busy_timeout=5000; - PRAGMA synchronous=NORMAL; - `); err != nil { + if _, err := db.Exec(`PRAGMA journal_mode=WAL;`); err != nil { db.Close() return nil, sqliteOpenError(err, dsn) } @@ -130,15 +130,17 @@ func parseURL(dbURL string) (driver, dsn string, err error) { id := memDBCounter.Add(1) dsn = fmt.Sprintf("file:memdb%d?mode=memory&cache=shared", id) } - // foreign_keys is a *connection-level* pragma in SQLite — setting it once - // after Open only covers a single pooled connection, leaving ON DELETE - // CASCADE / FK enforcement unreliable on the rest. modernc.org/sqlite - // applies _pragma params on every connection it opens, so set it here. + // foreign_keys, busy_timeout and synchronous are *connection-level* + // pragmas in SQLite — setting them once after Open only covers a single + // pooled connection (FK enforcement unreliable, busy_timeout=0 → + // SQLITE_BUSY under concurrent writers on the rest of the pool). + // modernc.org/sqlite applies _pragma params on every connection it + // opens, so set them here. sep := "?" if strings.Contains(dsn, "?") { sep = "&" } - dsn += sep + "_pragma=foreign_keys(1)" + dsn += sep + "_pragma=foreign_keys(1)&_pragma=busy_timeout(5000)&_pragma=synchronous(NORMAL)" return "sqlite", dsn, nil case strings.HasPrefix(dbURL, "postgres://"), strings.HasPrefix(dbURL, "postgresql://"): return "postgres", dbURL, nil diff --git a/internal/database/migrate_rollback_test.go b/internal/database/migrate_rollback_test.go new file mode 100644 index 0000000..b4ed203 --- /dev/null +++ b/internal/database/migrate_rollback_test.go @@ -0,0 +1,46 @@ +package database_test + +import ( + "strings" + "testing" + + "github.com/WithAutonomi/indelible/internal/database" + "github.com/WithAutonomi/indelible/internal/dbtest" +) + +// TestMigrationRollback_DropsHostedTransactionRows proves migration 016's +// Down completes with hosted-payment rows present (wallet_id NULL) on both +// drivers — under SQLite's foreign_keys pragma the previous COALESCE(wallet_id, +// 0) rewrite failed, and on Postgres SET NOT NULL fails on NULLs. The rows are +// dropped, deliberately (#163 review, V2-1269). +func TestMigrationRollback_DropsHostedTransactionRows(t *testing.T) { + db := dbtest.OpenEmptyDB(t) + driver := db.Driver() + if err := database.Migrate(db, driver); err != nil { + t.Fatalf("migrate up: %v", err) + } + if _, err := db.Exec(`INSERT INTO transactions (wallet_id, upload_id, tx_type, amount, balance_after, tx_hash) VALUES (NULL, NULL, 'hosted_payment', '1', '0', '0xabc')`); err != nil { + t.Fatalf("insert hosted row at head schema: %v", err) + } + + // Roll back past 016 (head is 017: gateway_fee, then 016: wallet-less). + for i := 0; i < 2; i++ { + if err := database.MigrateDown(db, driver); err != nil { + t.Fatalf("migrate down step %d: %v", i+1, err) + } + } + var n int + if err := db.QueryRow(`SELECT COUNT(*) FROM transactions`).Scan(&n); err != nil { + t.Fatalf("count after rollback: %v", err) + } + if n != 0 { + t.Fatalf("hosted rows must be dropped on rollback (pre-016 schema cannot hold them), found %d", n) + } + // And the column is NOT NULL again: a NULL insert must now be refused. + if _, err := db.Exec(`INSERT INTO transactions (wallet_id, upload_id, tx_type, amount, balance_after) VALUES (NULL, NULL, 'x', '1', '0')`); err == nil || !strings.Contains(strings.ToLower(err.Error()), "null") { + t.Fatalf("post-rollback schema must reject NULL wallet_id, got err=%v", err) + } + if err := database.Migrate(db, driver); err != nil { + t.Fatalf("migrate up again after rollback: %v", err) + } +} diff --git a/internal/database/migrations/postgres/015_payment_provenance.sql b/internal/database/migrations/postgres/015_payment_provenance.sql new file mode 100644 index 0000000..a9de121 --- /dev/null +++ b/internal/database/migrations/postgres/015_payment_provenance.sql @@ -0,0 +1,16 @@ +-- +goose Up + +-- V2-1086: per-upload payment provenance. payment_backend records WHO settled the +-- upload's payment was settled — 'local' (this instance's wallet signed) or +-- 'hosted' (the payment gateway paid from the tenant's credits). Without it +-- the instance-wide payment_backend config is the only tell, and history goes +-- ambiguous the moment an instance switches modes. gateway_payment_key is +-- the gateway's content-derived batch idempotency key — the stable join to +-- the gateway's payments ledger (and from there to the on-chain tx). Both +-- NULL for unpaid uploads (dedup/already_stored) and pre-existing rows. +ALTER TABLE uploads ADD COLUMN payment_backend TEXT; +ALTER TABLE uploads ADD COLUMN gateway_payment_key TEXT; + +-- +goose Down +ALTER TABLE uploads DROP COLUMN gateway_payment_key; +ALTER TABLE uploads DROP COLUMN payment_backend; diff --git a/internal/database/migrations/postgres/016_walletless_transactions.sql b/internal/database/migrations/postgres/016_walletless_transactions.sql new file mode 100644 index 0000000..80770d7 --- /dev/null +++ b/internal/database/migrations/postgres/016_walletless_transactions.sql @@ -0,0 +1,16 @@ +-- +goose Up + +-- V2-929 wallet-less hosted mode: hosted payments are settled by the +-- payment gateway's treasury, not by any wallet record — their transaction +-- rows carry wallet_id NULL. +ALTER TABLE transactions ALTER COLUMN wallet_id DROP NOT NULL; + +-- +goose Down + +-- SET NOT NULL fails while any hosted-payment row (wallet_id NULL — the +-- gateway's treasury paid, no wallet exists) is present, and the pre-016 +-- schema cannot hold them. They are DROPPED here, deliberately, so the +-- rollback completes; the gateway's ledger remains the record of those +-- payments. Export them first if you need them (#163 review, V2-1269). +DELETE FROM transactions WHERE wallet_id IS NULL; +ALTER TABLE transactions ALTER COLUMN wallet_id SET NOT NULL; diff --git a/internal/database/migrations/postgres/017_gateway_fee.sql b/internal/database/migrations/postgres/017_gateway_fee.sql new file mode 100644 index 0000000..19cbe33 --- /dev/null +++ b/internal/database/migrations/postgres/017_gateway_fee.sql @@ -0,0 +1,12 @@ +-- +goose Up + +-- V2-1098: the payment gateway's per-batch network fee, stamped with the +-- payment so the upload's cost display can itemize it ("cost includes the +-- fee that covered the chain's gas"). actual_cost holds the GROSS debit +-- (batch total + this fee) for hosted uploads — the number the tenant's +-- credits actually dropped by. NULL for local-mode and pre-existing rows, +-- and for hosted uploads paid before the gateway charged a fee. +ALTER TABLE uploads ADD COLUMN gateway_fee_atto TEXT; + +-- +goose Down +ALTER TABLE uploads DROP COLUMN gateway_fee_atto; diff --git a/internal/database/migrations/sqlite/015_payment_provenance.sql b/internal/database/migrations/sqlite/015_payment_provenance.sql new file mode 100644 index 0000000..a9de121 --- /dev/null +++ b/internal/database/migrations/sqlite/015_payment_provenance.sql @@ -0,0 +1,16 @@ +-- +goose Up + +-- V2-1086: per-upload payment provenance. payment_backend records WHO settled the +-- upload's payment was settled — 'local' (this instance's wallet signed) or +-- 'hosted' (the payment gateway paid from the tenant's credits). Without it +-- the instance-wide payment_backend config is the only tell, and history goes +-- ambiguous the moment an instance switches modes. gateway_payment_key is +-- the gateway's content-derived batch idempotency key — the stable join to +-- the gateway's payments ledger (and from there to the on-chain tx). Both +-- NULL for unpaid uploads (dedup/already_stored) and pre-existing rows. +ALTER TABLE uploads ADD COLUMN payment_backend TEXT; +ALTER TABLE uploads ADD COLUMN gateway_payment_key TEXT; + +-- +goose Down +ALTER TABLE uploads DROP COLUMN gateway_payment_key; +ALTER TABLE uploads DROP COLUMN payment_backend; diff --git a/internal/database/migrations/sqlite/016_walletless_transactions.sql b/internal/database/migrations/sqlite/016_walletless_transactions.sql new file mode 100644 index 0000000..45b5a93 --- /dev/null +++ b/internal/database/migrations/sqlite/016_walletless_transactions.sql @@ -0,0 +1,47 @@ +-- +goose Up + +-- V2-929 wallet-less hosted mode: hosted payments are settled by the +-- payment gateway's treasury, not by any wallet record — their transaction +-- rows carry wallet_id NULL. SQLite cannot drop NOT NULL in place, so the +-- table is rebuilt (same pattern as 014). +ALTER TABLE transactions RENAME TO transactions_old; +CREATE TABLE transactions ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + wallet_id INTEGER REFERENCES wallets(id), + upload_id INTEGER REFERENCES uploads(id), + tx_type TEXT NOT NULL, + amount TEXT NOT NULL, + balance_after TEXT NOT NULL, + tx_hash TEXT, + created_at DATETIME NOT NULL DEFAULT (datetime('now')) +); +INSERT INTO transactions (id, wallet_id, upload_id, tx_type, amount, balance_after, tx_hash, created_at) + SELECT id, wallet_id, upload_id, tx_type, amount, balance_after, tx_hash, created_at FROM transactions_old; +DROP TABLE transactions_old; +CREATE INDEX idx_transactions_wallet_id ON transactions(wallet_id); + +-- +goose Down + +-- Rolling back to a NOT NULL wallet_id cannot represent hosted-payment rows +-- (wallet_id NULL: the gateway's treasury paid, no wallet exists). They are +-- DROPPED here, deliberately: the previous COALESCE(wallet_id, 0) rewrite +-- pointed them at a wallet 0 that does not exist, which fails under the +-- foreign_keys pragma this app opens SQLite with (#163 review, V2-1269). +-- The gateway's ledger remains the record of those payments (payment_key on +-- uploads, dropped by 015's own Down); export them first if you need them. +ALTER TABLE transactions RENAME TO transactions_old; +DELETE FROM transactions_old WHERE wallet_id IS NULL; +CREATE TABLE transactions ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + wallet_id INTEGER NOT NULL REFERENCES wallets(id), + upload_id INTEGER REFERENCES uploads(id), + tx_type TEXT NOT NULL, + amount TEXT NOT NULL, + balance_after TEXT NOT NULL, + tx_hash TEXT, + created_at DATETIME NOT NULL DEFAULT (datetime('now')) +); +INSERT INTO transactions (id, wallet_id, upload_id, tx_type, amount, balance_after, tx_hash, created_at) + SELECT id, wallet_id, upload_id, tx_type, amount, balance_after, tx_hash, created_at FROM transactions_old; +DROP TABLE transactions_old; +CREATE INDEX idx_transactions_wallet_id ON transactions(wallet_id); diff --git a/internal/database/migrations/sqlite/017_gateway_fee.sql b/internal/database/migrations/sqlite/017_gateway_fee.sql new file mode 100644 index 0000000..19cbe33 --- /dev/null +++ b/internal/database/migrations/sqlite/017_gateway_fee.sql @@ -0,0 +1,12 @@ +-- +goose Up + +-- V2-1098: the payment gateway's per-batch network fee, stamped with the +-- payment so the upload's cost display can itemize it ("cost includes the +-- fee that covered the chain's gas"). actual_cost holds the GROSS debit +-- (batch total + this fee) for hosted uploads — the number the tenant's +-- credits actually dropped by. NULL for local-mode and pre-existing rows, +-- and for hosted uploads paid before the gateway charged a fee. +ALTER TABLE uploads ADD COLUMN gateway_fee_atto TEXT; + +-- +goose Down +ALTER TABLE uploads DROP COLUMN gateway_fee_atto; diff --git a/internal/evm/hosted.go b/internal/evm/hosted.go new file mode 100644 index 0000000..6e1dbad --- /dev/null +++ b/internal/evm/hosted.go @@ -0,0 +1,533 @@ +package evm + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "math/big" + "net/http" + "strings" + "sync" + "time" + + antd "github.com/WithAutonomi/ant-sdk/antd-go" +) + +// HostedPayer settles payments through a payment gateway's POST /pay instead +// of signing locally (payment_backend=hosted, V2-929 PoC). It satisfies the same +// method set as Signer so the upload worker can hold either behind one +// interface. The privateKeyHex arguments are ignored — the gateway holds the +// treasury key. +type HostedPayer struct { + gatewayURL string + apiKey string + client *http.Client + // pollWait bounds how long a 202-with-payment_key answer is polled via + // GET /payments/{key} before falling back to the preserve path. + pollWait time.Duration + // payAttempts/payRetryWait bound the transport-level resend of POST + // /pay (safe: the batch is idempotent gateway-side). Sized to outlast + // a gateway restart. + payAttempts int + payRetryWait time.Duration + // costs holds each paid batch's fee itemization keyed by payment_key + // until the worker collects it via PaymentCost (V2-1098) — the payer + // interface returns only the tx map, and one HostedPayer serves + // concurrent uploads, so a "last payment" field would race. + costsMu sync.Mutex + costs map[string]PaymentCost + // fee* memoize the gateway's per-batch network fee for the pre-spend + // gross estimate/ceiling (V2-1113) — the worker asks per upload, the + // gateway at most once per feeTTL (the wallet-status rate cadence, + // V2-1100). feeCached nil means never fetched successfully. + feeMu sync.Mutex + feeCached *big.Int + feeFetched time.Time + feeTTL time.Duration +} + +// PaymentCost is a paid batch's fee itemization: what the gateway debited +// beyond the batch total (V2-1098). +type PaymentCost struct { + FeeAtto string // the network fee, atto + TotalDebited string // batch total + fee — the gross debit +} + +// NewHostedPayer builds a payer that delegates to the gateway at gatewayURL, +// authenticating as this instance's tenant account via apiKey (Bearer). +func NewHostedPayer(gatewayURL, apiKey string) *HostedPayer { + return &HostedPayer{ + gatewayURL: strings.TrimRight(gatewayURL, "/"), + apiKey: apiKey, + // No overall timeout: /pay legitimately blocks for the gateway's + // sync wait, mirroring the local signer's bound. + client: &http.Client{}, + pollWait: 5 * time.Minute, + payAttempts: 6, + payRetryWait: 5 * time.Second, + feeTTL: time.Minute, + } +} + +// SetPollWait overrides the async-completion polling bound. +func (h *HostedPayer) SetPollWait(d time.Duration) { h.pollWait = d } + +// rememberCost stashes a paid response's fee itemization for the worker. +func (h *HostedPayer) rememberCost(paymentKey string, resp *hostedPayResponse) { + if paymentKey == "" || resp.FeeAmount == "" || resp.TotalDebited == "" { + return + } + h.costsMu.Lock() + defer h.costsMu.Unlock() + if h.costs == nil { + h.costs = map[string]PaymentCost{} + } + h.costs[paymentKey] = PaymentCost{FeeAtto: resp.FeeAmount, TotalDebited: resp.TotalDebited} +} + +// PaymentCost pops the fee itemization for a payment key, if the gateway +// reported one — the worker records the GROSS spend on the upload and its +// transaction so the customer's books match the gateway ledger (V2-1098). +func (h *HostedPayer) PaymentCost(paymentKey string) (PaymentCost, bool) { + h.costsMu.Lock() + defer h.costsMu.Unlock() + c, ok := h.costs[paymentKey] + if ok { + delete(h.costs, paymentKey) + } + return c, ok +} + +// hostedPayRequest is the /pay body. The tenant is never named here: the +// gateway resolves it from the Bearer API key and treats any account_id in +// the body as advisory at most, so sending one only invites confusion. +type hostedPayRequest struct { + PaymentType string `json:"payment_type,omitempty"` + Payments []antd.PaymentInfo `json:"payments"` + TokenAddress string `json:"token_address"` + PaymentVaultAddress string `json:"payment_vault_address"` + SignedQuotes []antd.SignedQuoteEntry `json:"signed_quotes,omitempty"` +} + +type hostedPayResponse struct { + Status string `json:"status"` + Replayed bool `json:"replayed"` + PayTxHash string `json:"pay_tx_hash"` + TxHashes map[string]string `json:"tx_hashes"` + TotalAmount string `json:"total_amount"` + // TotalUSDCents is the GROSS debit (batch total + network fee) in fiat + // at the gateway's rate (rounded up), when the gateway has a rate + // configured — the crypto-free number user-facing messages prefer + // (V2-1100); gross so "top up $X" always suffices (V2-1098). + TotalUSDCents int64 `json:"total_usd_cents"` + // FeeAmount/TotalDebited itemize the gateway's per-batch network fee + // (V2-1098): fee in atto and total_amount + fee — what the account was + // actually debited. Empty when the gateway charges no fee. + FeeAmount string `json:"fee_amount"` + TotalDebited string `json:"total_debited"` + // FeeUSDCents is the fee alone in fiat (rounded up), rate permitting. + FeeUSDCents int64 `json:"fee_usd_cents"` + Error string `json:"error"` + // PaymentKey is the gateway's batch idempotency key — persisted on the + // upload as its provenance join to the gateway ledger (V2-1086). + PaymentKey string `json:"payment_key"` +} + +// PayForQuotes submits the batch to the gateway and returns the +// quote_hash → tx_hash map antd's finalize expects. The signed quotes are +// relayed unmodified so the gateway can verify the batch offline before +// paying (V2-926). A gateway "unconfirmed" answer (202) is surfaced as +// ErrConfirmationTimeout so the worker preserves the upload for +// reconciliation instead of re-paying; a "rejected" answer is a permanent +// refusal — the worker abandons rather than retrying. +func (h *HostedPayer) PayForQuotes( + ctx context.Context, + _ string, // private key unused — the gateway signs + payments []antd.PaymentInfo, + signedQuotes []antd.SignedQuoteEntry, + tokenAddress string, + dataPaymentsAddress string, +) (map[string]string, string, error) { + body, err := json.Marshal(hostedPayRequest{ + PaymentType: "wave_batch", + Payments: payments, + TokenAddress: tokenAddress, + PaymentVaultAddress: dataPaymentsAddress, + SignedQuotes: signedQuotes, + }) + if err != nil { + return nil, "", fmt.Errorf("encoding /pay request: %w", err) + } + + // Transport failures are retried: the batch is content-addressed + // idempotent gateway-side (one payments record per idempotency key, + // ever — V2-924), so resending can never double-pay. This covers the + // gateway dying with our request in flight — the killed connection + // EOFs, the gateway restarts, and the resend lands on the idempotent + // replay path. Without it a crash in the narrow pre-202 window fails + // the upload even though the payment itself survives (V2-931 case I). + var resp *http.Response + for attempt := 1; ; attempt++ { + req, rerr := http.NewRequestWithContext(ctx, http.MethodPost, h.gatewayURL+"/pay", bytes.NewReader(body)) + if rerr != nil { + return nil, "", rerr + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Authorization", "Bearer "+h.apiKey) + resp, err = h.client.Do(req) + if err == nil { + break + } + if attempt >= h.payAttempts { + return nil, "", fmt.Errorf("payment gateway unreachable: %w", err) + } + select { + case <-ctx.Done(): + return nil, "", ctx.Err() + case <-time.After(h.payRetryWait): + } + } + defer resp.Body.Close() + + var payResp hostedPayResponse + raw, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) + if err := json.Unmarshal(raw, &payResp); err != nil { + return nil, "", fmt.Errorf("payment gateway returned %d: %s", resp.StatusCode, strings.TrimSpace(string(raw))) + } + + switch { + case resp.StatusCode == http.StatusOK && (payResp.Status == "paid" || payResp.Status == "nothing_to_pay"): + h.rememberCost(payResp.PaymentKey, &payResp) + return payResp.TxHashes, payResp.PaymentKey, nil + case resp.StatusCode == http.StatusAccepted && payResp.Status == "unconfirmed": + // Async contract (V-925/929): the gateway queued or broadcast the + // batch and handed back its payment_key — poll to completion. Only + // when polling exhausts (or no key was given) fall back to the + // preserve path via the same typed error as a local confirmation + // timeout, so classifyFailure never re-pays. + if payResp.PaymentKey != "" { + return h.pollPayment(ctx, payResp.PaymentKey, payResp.PayTxHash) + } + return nil, "", fmt.Errorf("%w (gateway tx %s)", ErrConfirmationTimeout, payResp.PayTxHash) + case payResp.Status == "insufficient_credits": + // The one refusal an operator fixes themselves: say what it costs + // and what to do (V2-930) — in fiat when the gateway has a rate + // (crypto-free counter, V2-1100), ANT only as the fallback. Both + // parts named when the gateway charges a network fee (V2-1098). + if payResp.TotalUSDCents > 0 { + if payResp.FeeUSDCents > 0 { + return nil, "", fmt.Errorf( + "insufficient gateway credits: this upload needs about $%d.%02d of storage credit (including a $%d.%02d network fee) — top up and retry (retrying is safe, nothing was paid)", + payResp.TotalUSDCents/100, payResp.TotalUSDCents%100, + payResp.FeeUSDCents/100, payResp.FeeUSDCents%100) + } + return nil, "", fmt.Errorf( + "insufficient gateway credits: this upload needs about $%d.%02d of storage credit — top up and retry (retrying is safe, nothing was paid)", + payResp.TotalUSDCents/100, payResp.TotalUSDCents%100) + } + if payResp.FeeAmount != "" { + return nil, "", fmt.Errorf( + "insufficient gateway credits: this upload needs %s ANT + %s ANT network fee — top up the account's credits and retry (retrying is safe, nothing was paid)", + attoToANT(payResp.TotalAmount), attoToANT(payResp.FeeAmount)) + } + return nil, "", fmt.Errorf( + "insufficient gateway credits: this upload needs %s ANT — top up the account's credits and retry (retrying is safe, nothing was paid)", + attoToANT(payResp.TotalAmount)) + default: + msg := payResp.Error + if msg == "" { + msg = strings.TrimSpace(string(raw)) + } + return nil, "", fmt.Errorf("payment gateway /pay failed (%d, %s): %s", resp.StatusCode, payResp.Status, msg) + } +} + +// attoToANT renders an atto amount (decimal string) as a human ANT figure — +// integer string math, never floats. Unparseable input passes through. +func attoToANT(atto string) string { + s := strings.TrimSpace(atto) + if s == "" || strings.ContainsAny(s, ".-") { + return atto + } + for _, c := range s { + if c < '0' || c > '9' { + return atto + } + } + if len(s) <= 18 { + s = strings.Repeat("0", 19-len(s)) + s + } + whole, frac := s[:len(s)-18], strings.TrimRight(s[len(s)-18:], "0") + if frac == "" { + return whole + } + return whole + "." + frac +} + +// pollPayment follows an async payment to its terminal state via +// GET /payments/{key}. Transport errors keep polling (the payment is safe +// server-side; the gateway may be restarting); the bound falls back to the +// preserve path, never a re-pay. +func (h *HostedPayer) pollPayment(ctx context.Context, paymentKey, lastTx string) (map[string]string, string, error) { + deadline := time.Now().Add(h.pollWait) + for { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, h.gatewayURL+"/payments/"+paymentKey, nil) + if err != nil { + return nil, "", err + } + req.Header.Set("Authorization", "Bearer "+h.apiKey) + resp, err := h.client.Do(req) + if err == nil { + var payResp hostedPayResponse + decodeErr := json.NewDecoder(io.LimitReader(resp.Body, 1<<20)).Decode(&payResp) + _ = resp.Body.Close() + if decodeErr == nil && resp.StatusCode == http.StatusOK { + if payResp.PayTxHash != "" { + lastTx = payResp.PayTxHash + } + switch payResp.Status { + case "paid": + h.rememberCost(paymentKey, &payResp) + return payResp.TxHashes, paymentKey, nil + case "failed", "rejected": + return nil, "", fmt.Errorf("payment gateway reported %s: %s", payResp.Status, payResp.Error) + } + } + } + if time.Now().After(deadline) { + return nil, "", fmt.Errorf("%w (gateway payment %s, tx %s)", ErrConfirmationTimeout, paymentKey, lastTx) + } + select { + case <-ctx.Done(): + return nil, "", ctx.Err() + case <-time.After(3 * time.Second): + } + } +} + +// AccountDetails is the gateway's GET /account answer as the display +// surfaces relay it: balance, the exact USD-per-ANT rate (V2-1100), the +// per-batch network fee (V2-1113), and the directional cost-per-GB estimate +// with its methodology basis (V2-1114). Every field beyond the balance is +// optional — empty/nil when the gateway has none configured, has too little +// paid history yet, or simply predates the field. Absence is never an error. +type AccountDetails struct { + BalanceAtto string + RateUSDPerANT string + FeePerBatchAtto string + // EstCostPerGBAtto is what ≈1 GB of fresh data costs at current prices, + // estimated by the gateway from its own recent paid history; "" = no + // estimate ("not enough data yet" client-side, never zero). + EstCostPerGBAtto string + // EstCostPerGBBasis is the estimate's basis object (median paid per + // quote, sample size, window, chunk/batch constants), relayed opaquely so + // the client tooltip hardcodes no methodology numbers. + EstCostPerGBBasis json.RawMessage +} + +// AccountDetails fetches GET /account once and returns everything the +// billing surfaces relay (see the AccountDetails type). +func (h *HostedPayer) AccountDetails(ctx context.Context) (AccountDetails, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, h.gatewayURL+"/account", nil) + if err != nil { + return AccountDetails{}, err + } + req.Header.Set("Authorization", "Bearer "+h.apiKey) + resp, err := h.client.Do(req) + if err != nil { + return AccountDetails{}, fmt.Errorf("payment gateway unreachable: %w", err) + } + defer resp.Body.Close() + var out struct { + Balance string `json:"balance"` + Rate string `json:"rate_usd_per_ant"` + Fee string `json:"fee_per_batch_atto"` + EstCostGB string `json:"est_cost_per_gb_atto"` + EstBasis json.RawMessage `json:"est_cost_per_gb_basis"` + Error string `json:"error"` + } + if err := json.NewDecoder(io.LimitReader(resp.Body, 1<<20)).Decode(&out); err != nil { + return AccountDetails{}, fmt.Errorf("decoding /account response: %w", err) + } + if resp.StatusCode != http.StatusOK { + return AccountDetails{}, fmt.Errorf("payment gateway /account failed (%d): %s", resp.StatusCode, out.Error) + } + return AccountDetails{ + BalanceAtto: out.Balance, + RateUSDPerANT: out.Rate, + FeePerBatchAtto: out.Fee, + EstCostPerGBAtto: out.EstCostGB, + EstCostPerGBBasis: out.EstBasis, + }, nil +} + +// AccountInfo returns the tenant's remaining gateway credit balance in atto +// plus the gateway's exact USD-per-ANT rate and per-batch network fee in +// atto (each empty when none configured, or when an older gateway predates +// the field) — the trio the crypto-free display converts with (V2-1100) and +// fee-aware estimates add with (V2-1113). Thin wrapper over AccountDetails +// for the callers that need no more. +func (h *HostedPayer) AccountInfo(ctx context.Context) (balance, rateUSDPerANT, feePerBatchAtto string, err error) { + d, err := h.AccountDetails(ctx) + return d.BalanceAtto, d.RateUSDPerANT, d.FeePerBatchAtto, err +} + +// AccountBalance returns the tenant's remaining gateway credit balance in +// atto (GET /account) — the meaningful "balance after" for hosted payments, +// where neither the wallet record nor the treasury is the payer's account. +func (h *HostedPayer) AccountBalance(ctx context.Context) (string, error) { + bal, _, _, err := h.AccountInfo(ctx) + return bal, err +} + +// FeePerBatch returns the gateway's per-batch network fee in atto — the +// V2-1098 surcharge every settled batch adds to the debit — so pre-spend +// surfaces can quote and gate on the same GROSS basis the gateway actually +// charges (V2-1113). Cached for feeTTL; while the gateway is unreachable the +// last known value keeps serving (fee changes are rare, refusing uploads over +// a stale fee lookup would be worse), and a gateway that reports no fee — +// none configured, or an older gateway without the field — counts as zero. +// +// The bool is false only when NO fetch has ever succeeded: then the fee is +// not "zero", it is unknown, and a caller gating spend on the gross cost must +// not treat it as zero (review of #163). Once a fetch has succeeded the value +// is known, even if stale. +func (h *HostedPayer) FeePerBatch(ctx context.Context) (*big.Int, bool) { + h.feeMu.Lock() + defer h.feeMu.Unlock() + if h.feeFetched.IsZero() || time.Since(h.feeFetched) >= h.feeTTL { + feeCtx, cancel := context.WithTimeout(ctx, 2*time.Second) + _, _, fee, err := h.AccountInfo(feeCtx) + cancel() + if err == nil { + if v, ok := new(big.Int).SetString(strings.TrimSpace(fee), 10); ok && v.Sign() > 0 { + h.feeCached = v + } else { + h.feeCached = new(big.Int) // absent/zero/junk → no fee + } + } + // Set even on error: retry after the normal interval, not per call. + h.feeFetched = time.Now() + } + if h.feeCached == nil { + return new(big.Int), false + } + return new(big.Int).Set(h.feeCached), true +} + +// relay performs one authenticated gateway call for the in-app billing +// surface (V2-1097) and hands back the gateway's status code + raw JSON so +// the caller can pass both through unmodified. Only transport-level failure +// is an error. +func (h *HostedPayer) relay(ctx context.Context, method, path string, body any) (int, []byte, error) { + var rd io.Reader + if body != nil { + enc, err := json.Marshal(body) + if err != nil { + return 0, nil, err + } + rd = bytes.NewReader(enc) + } + req, err := http.NewRequestWithContext(ctx, method, h.gatewayURL+path, rd) + if err != nil { + return 0, nil, err + } + if body != nil { + req.Header.Set("Content-Type", "application/json") + } + req.Header.Set("Authorization", "Bearer "+h.apiKey) + resp, err := h.client.Do(req) + if err != nil { + return 0, nil, fmt.Errorf("payment gateway unreachable: %w", err) + } + defer resp.Body.Close() + raw, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) + if err != nil { + return 0, nil, err + } + return resp.StatusCode, raw, nil +} + +// TopupCheckout relays a Stripe Checkout creation to the gateway +// (POST /topup/checkout): the gateway enforces bounds and validates the +// return URLs; the tenant API key never leaves the server. +func (h *HostedPayer) TopupCheckout(ctx context.Context, amountUSDCents int64, successURL, cancelURL string) (int, []byte, error) { + return h.relay(ctx, http.MethodPost, "/topup/checkout", map[string]any{ + "amount_usd_cents": amountUSDCents, + "success_url": successURL, + "cancel_url": cancelURL, + }) +} + +// TopupSync relays the deterministic credit fallback (POST /topup/sync) — +// used on return from Checkout so credits show without waiting on webhook +// delivery. Idempotent gateway-side. +func (h *HostedPayer) TopupSync(ctx context.Context, sessionID string) (int, []byte, error) { + return h.relay(ctx, http.MethodPost, "/topup/sync", map[string]any{"session_id": sessionID}) +} + +// Topups relays the tenant's credited top-up history (GET /topups). +func (h *HostedPayer) Topups(ctx context.Context) (int, []byte, error) { + return h.relay(ctx, http.MethodGet, "/topups", nil) +} + +// Credits relays the tenant's full credit history (GET /credits) — card +// top-ups and invoice-path grants alike, so the Billing screen answers +// "where did this credit come from" for every funding path. +func (h *HostedPayer) Credits(ctx context.Context) (int, []byte, error) { + return h.relay(ctx, http.MethodGet, "/credits", nil) +} + +// PayForMerkleTree is not supported by the gateway PoC (merkle hosted support +// is V2-934). +func (h *HostedPayer) PayForMerkleTree( + _ context.Context, + _ string, + _ int, + _ []antd.PoolCommitmentEntry, + _ uint64, + _ string, + _ string, +) (string, string, error) { + return "", "", fmt.Errorf("hosted payment mode does not support merkle payments yet (V2-934)") +} + +// GetBalances reports the gateway treasury's balances — in hosted mode the +// treasury is the paying wallet, so those are the balances worth recording. +func (h *HostedPayer) GetBalances(ctx context.Context, _ string, tokenAddress string) (string, string, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, h.gatewayURL+"/treasury?token="+tokenAddress, nil) + if err != nil { + return "", "", err + } + req.Header.Set("Authorization", "Bearer "+h.apiKey) + resp, err := h.client.Do(req) + if err != nil { + return "", "", fmt.Errorf("payment gateway unreachable: %w", err) + } + defer resp.Body.Close() + + var out struct { + TokenBalance string `json:"token_balance"` + GasBalance string `json:"gas_balance"` + Error string `json:"error"` + } + if err := json.NewDecoder(resp.Body).Decode(&out); err != nil { + return "", "", fmt.Errorf("decoding /treasury response: %w", err) + } + if resp.StatusCode != http.StatusOK { + return "", "", fmt.Errorf("payment gateway /treasury failed (%d): %s", resp.StatusCode, out.Error) + } + return out.TokenBalance, out.GasBalance, nil +} + +// SetConfirmationTimeout is a no-op: the confirmation bound lives on the +// gateway's signer in hosted mode. +func (h *HostedPayer) SetConfirmationTimeout(time.Duration) {} + +// RPCUrl reports the gateway endpoint; the worker only uses it for its +// rebuild-on-change check and logging. +func (h *HostedPayer) RPCUrl() string { + return h.gatewayURL +} diff --git a/internal/evm/hosted_test.go b/internal/evm/hosted_test.go new file mode 100644 index 0000000..769a49a --- /dev/null +++ b/internal/evm/hosted_test.go @@ -0,0 +1,539 @@ +package evm + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + antd "github.com/WithAutonomi/ant-sdk/antd-go" +) + +// stubGateway answers /pay and /account with canned bodies. +func stubGateway(t *testing.T, payStatus int, payBody map[string]any) *httptest.Server { + t.Helper() + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("Authorization") != "Bearer pgk_test" { + w.WriteHeader(http.StatusUnauthorized) + return + } + switch r.URL.Path { + case "/pay": + w.WriteHeader(payStatus) + _ = json.NewEncoder(w).Encode(payBody) + case "/account": + _ = json.NewEncoder(w).Encode(map[string]any{"account": "acme", "balance": "71358258928571428571"}) + default: + w.WriteHeader(http.StatusNotFound) + } + })) +} + +func TestHostedPayForQuotesPaid(t *testing.T) { + srv := stubGateway(t, http.StatusOK, map[string]any{ + "status": "paid", "pay_tx_hash": "0xabc", + "tx_hashes": map[string]string{"0xq1": "0xabc"}, + "payment_key": "deadbeef", + }) + defer srv.Close() + h := NewHostedPayer(srv.URL, "pgk_test") + hashes, key, err := h.PayForQuotes(context.Background(), "", []antd.PaymentInfo{{QuoteHash: "0xq1"}}, nil, "0xt", "0xv") + if err != nil || hashes["0xq1"] != "0xabc" || key != "deadbeef" { + t.Fatalf("paid path: %v %v %q", hashes, err, key) + } +} + +func TestHostedInsufficientCreditsIsHumanReadable(t *testing.T) { + srv := stubGateway(t, http.StatusPaymentRequired, map[string]any{ + "status": "insufficient_credits", + "error": "insufficient credits: balance 0 < total 35156250000000000", + "total_amount": "35156250000000000", + }) + defer srv.Close() + h := NewHostedPayer(srv.URL, "pgk_test") + _, _, err := h.PayForQuotes(context.Background(), "", []antd.PaymentInfo{{QuoteHash: "0xq1"}}, nil, "0xt", "0xv") + if err == nil { + t.Fatal("expected error") + } + msg := err.Error() + if !strings.Contains(msg, "0.03515625 ANT") || !strings.Contains(msg, "top up") { + t.Fatalf("message not operator-friendly: %q", msg) + } + if strings.Contains(msg, "35156250000000000") || strings.Contains(msg, "402") { + t.Fatalf("raw atto/status leaked into the message: %q", msg) + } +} + +func TestHostedPaymentCostFromPaidResponse(t *testing.T) { + srv := stubGateway(t, http.StatusOK, map[string]any{ + "status": "paid", "pay_tx_hash": "0xabc", + "tx_hashes": map[string]string{"0xq1": "0xabc"}, + "payment_key": "deadbeef", + "total_amount": "35156250000000000", + "fee_amount": "5000000000000000", + "total_debited": "40156250000000000", + }) + defer srv.Close() + h := NewHostedPayer(srv.URL, "pgk_test") + if _, _, err := h.PayForQuotes(context.Background(), "", []antd.PaymentInfo{{QuoteHash: "0xq1"}}, nil, "0xt", "0xv"); err != nil { + t.Fatal(err) + } + c, ok := h.PaymentCost("deadbeef") + if !ok || c.FeeAtto != "5000000000000000" || c.TotalDebited != "40156250000000000" { + t.Fatalf("payment cost: %+v ok=%v", c, ok) + } + // Pop semantics: collected once, then gone. + if _, ok := h.PaymentCost("deadbeef"); ok { + t.Fatal("cost must be popped on read") + } +} + +func TestHostedAsyncPollCarriesFee(t *testing.T) { + srv := asyncGateway(t, "paid", map[string]any{ + "pay_tx_hash": "0xdef", "tx_hashes": map[string]string{"0xq1": "0xdef"}, + "total_amount": "35156250000000000", "fee_amount": "5000000000000000", + "total_debited": "40156250000000000"}) + defer srv.Close() + h := NewHostedPayer(srv.URL, "pgk_test") + h.SetPollWait(30 * time.Second) + if _, _, err := h.PayForQuotes(context.Background(), "", []antd.PaymentInfo{{QuoteHash: "0xq1"}}, nil, "0xt", "0xv"); err != nil { + t.Fatal(err) + } + c, ok := h.PaymentCost("k123") + if !ok || c.TotalDebited != "40156250000000000" { + t.Fatalf("poll path lost the fee itemization: %+v ok=%v", c, ok) + } +} + +func TestHostedInsufficientCreditsNamesBothParts(t *testing.T) { + // Fiat form: gross + itemized fee, still no atto/ANT leak. + srv := stubGateway(t, http.StatusPaymentRequired, map[string]any{ + "status": "insufficient_credits", + "error": "insufficient credits: balance 0 < total 35156250000000000 + 5000000000000000 network fee", + "total_amount": "35156250000000000", + "fee_amount": "5000000000000000", + "total_debited": "40156250000000000", + "total_usd_cents": 2, + "fee_usd_cents": 1, + }) + defer srv.Close() + h := NewHostedPayer(srv.URL, "pgk_test") + _, _, err := h.PayForQuotes(context.Background(), "", []antd.PaymentInfo{{QuoteHash: "0xq1"}}, nil, "0xt", "0xv") + if err == nil { + t.Fatal("expected error") + } + msg := err.Error() + if !strings.Contains(msg, "$0.02") || !strings.Contains(msg, "$0.01 network fee") { + t.Fatalf("fiat message does not name both parts: %q", msg) + } + if strings.Contains(msg, " ANT") || strings.Contains(msg, "atto") { + t.Fatalf("crypto leaked into the fiat message: %q", msg) + } + + // ANT fallback (no rate): both parts in ANT. + srv2 := stubGateway(t, http.StatusPaymentRequired, map[string]any{ + "status": "insufficient_credits", + "total_amount": "35156250000000000", + "fee_amount": "5000000000000000", + "total_debited": "40156250000000000", + }) + defer srv2.Close() + h2 := NewHostedPayer(srv2.URL, "pgk_test") + _, _, err = h2.PayForQuotes(context.Background(), "", []antd.PaymentInfo{{QuoteHash: "0xq1"}}, nil, "0xt", "0xv") + if err == nil { + t.Fatal("expected error") + } + if msg := err.Error(); !strings.Contains(msg, "0.03515625 ANT") || !strings.Contains(msg, "0.005 ANT network fee") { + t.Fatalf("ANT fallback does not name both parts: %q", msg) + } +} + +func TestHostedAccountBalance(t *testing.T) { + srv := stubGateway(t, http.StatusOK, nil) + defer srv.Close() + h := NewHostedPayer(srv.URL, "pgk_test") + bal, err := h.AccountBalance(context.Background()) + if err != nil || bal != "71358258928571428571" { + t.Fatalf("balance: %q %v", bal, err) + } +} + +func TestAttoToANT(t *testing.T) { + for in, want := range map[string]string{ + "35156250000000000": "0.03515625", + "71358258928571428571": "71.358258928571428571", + "1000000000000000000": "1", + "0": "0", + "": "", + "not-a-number": "not-a-number", + } { + if got := attoToANT(in); got != want { + t.Errorf("attoToANT(%q) = %q, want %q", in, got, want) + } + } +} + +// asyncGateway stubs the V2-925 contract: /pay answers 202 with a +// payment_key; /payments/{key} advances queued → paid across polls. +func asyncGateway(t *testing.T, terminal string, terminalBody map[string]any) *httptest.Server { + t.Helper() + polls := 0 + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case r.URL.Path == "/pay": + w.WriteHeader(http.StatusAccepted) + _ = json.NewEncoder(w).Encode(map[string]any{ + "status": "unconfirmed", "payment_key": "k123", + "error": "payment queued for broadcast — poll GET /payments/k123"}) + case r.URL.Path == "/payments/k123": + polls++ + if polls == 1 { + _ = json.NewEncoder(w).Encode(map[string]any{"status": "queued"}) + return + } + body := map[string]any{"status": terminal} + for k, v := range terminalBody { + body[k] = v + } + _ = json.NewEncoder(w).Encode(body) + default: + w.WriteHeader(http.StatusNotFound) + } + })) +} + +func TestHostedAsyncPollToPaid(t *testing.T) { + srv := asyncGateway(t, "paid", map[string]any{ + "pay_tx_hash": "0xdef", "tx_hashes": map[string]string{"0xq1": "0xdef"}}) + defer srv.Close() + h := NewHostedPayer(srv.URL, "pgk_test") + h.SetPollWait(30 * time.Second) + hashes, key, err := h.PayForQuotes(context.Background(), "", []antd.PaymentInfo{{QuoteHash: "0xq1"}}, nil, "0xt", "0xv") + if err != nil || hashes["0xq1"] != "0xdef" || key != "k123" { + t.Fatalf("async paid: %v %q %v", hashes, key, err) + } +} + +func TestHostedAsyncPollToFailed(t *testing.T) { + srv := asyncGateway(t, "failed", map[string]any{"error": "transaction reverted: 0xbad"}) + defer srv.Close() + h := NewHostedPayer(srv.URL, "pgk_test") + h.SetPollWait(30 * time.Second) + _, _, err := h.PayForQuotes(context.Background(), "", []antd.PaymentInfo{{QuoteHash: "0xq1"}}, nil, "0xt", "0xv") + if err == nil || !strings.Contains(err.Error(), "reverted") { + t.Fatalf("async failed path: %v", err) + } + if errors.Is(err, ErrConfirmationTimeout) { + t.Fatal("a definitive failure must not map to the preserve path") + } +} + +// TestHostedBillingRelays proves the V2-1097 relay methods pass the +// gateway's status and body through unmodified — success and refusal alike +// — with the tenant key attached server-side. +func TestHostedBillingRelays(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("Authorization") != "Bearer pgk_test" { + w.WriteHeader(http.StatusUnauthorized) + return + } + switch r.URL.Path { + case "/topup/checkout": + var req map[string]any + _ = json.NewDecoder(r.Body).Decode(&req) + if req["amount_usd_cents"].(float64) == 2500 && + req["success_url"] != "http://app.local/admin/billing?topup={CHECKOUT_SESSION_ID}" { + t.Errorf("success_url not relayed: %v", req["success_url"]) + } + if req["amount_usd_cents"].(float64) == 100 { // below gateway bounds + w.WriteHeader(http.StatusBadRequest) + _ = json.NewEncoder(w).Encode(map[string]any{"error": "amount_usd_cents 100 outside bounds 500–1000000"}) + return + } + _ = json.NewEncoder(w).Encode(map[string]any{ + "session_id": "cs_test_1", "url": "https://checkout.stripe.com/pay/cs_test_1", "credit_atto": "71428571428571428571"}) + case "/topup/sync": + _ = json.NewEncoder(w).Encode(map[string]any{"credited": false, "payment_status": "unpaid"}) + case "/topups": + _ = json.NewEncoder(w).Encode(map[string]any{"topups": []map[string]any{{"id": 1, "session_id": "cs_test_1"}}}) + case "/credits": + _ = json.NewEncoder(w).Encode(map[string]any{"credits": []map[string]any{ + {"id": 2, "source": "card", "amount_usd_cents": 2500}, + {"id": 1, "source": "grant", "note": "invoice #77"}}}) + default: + w.WriteHeader(http.StatusNotFound) + } + })) + defer srv.Close() + h := NewHostedPayer(srv.URL, "pgk_test") + + status, raw, err := h.TopupCheckout(context.Background(), 2500, + "http://app.local/admin/billing?topup={CHECKOUT_SESSION_ID}", "http://app.local/admin/billing?cancelled=1") + if err != nil || status != http.StatusOK || !strings.Contains(string(raw), "cs_test_1") { + t.Fatalf("checkout relay: status=%d err=%v body=%s", status, err, raw) + } + + status, raw, err = h.TopupCheckout(context.Background(), 100, "http://app.local/x", "http://app.local/y") + if err != nil || status != http.StatusBadRequest || !strings.Contains(string(raw), "outside bounds") { + t.Fatalf("bounds refusal must pass through: status=%d err=%v body=%s", status, err, raw) + } + + status, raw, err = h.TopupSync(context.Background(), "cs_test_1") + if err != nil || status != http.StatusOK || !strings.Contains(string(raw), `"credited":false`) { + t.Fatalf("sync relay: status=%d err=%v body=%s", status, err, raw) + } + + status, raw, err = h.Topups(context.Background()) + if err != nil || status != http.StatusOK || !strings.Contains(string(raw), `"topups"`) { + t.Fatalf("topups relay: status=%d err=%v body=%s", status, err, raw) + } + + status, raw, err = h.Credits(context.Background()) + if err != nil || status != http.StatusOK || !strings.Contains(string(raw), `"source":"grant"`) { + t.Fatalf("credits relay: status=%d err=%v body=%s", status, err, raw) + } + + if _, _, err := NewHostedPayer("http://127.0.0.1:1", "pgk_test").Topups(context.Background()); err == nil { + t.Fatal("transport failure must surface as an error") + } +} + +// TestHostedFiatSurface proves the V2-1100 crypto-free plumbing: the 402 +// message speaks fiat when the gateway supplies cents (ANT only as the +// fallback), and AccountInfo carries the rate. +func TestHostedFiatSurface(t *testing.T) { + cents := int64(0) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/pay": + w.WriteHeader(http.StatusPaymentRequired) + _ = json.NewEncoder(w).Encode(map[string]any{ + "status": "insufficient_credits", "total_amount": "35156250000000000", + "total_usd_cents": cents, "tx_hashes": map[string]string{}}) + case "/account": + _ = json.NewEncoder(w).Encode(map[string]any{"account": "acme", "balance": "5", "rate_usd_per_ant": "0.35"}) + } + })) + defer srv.Close() + h := NewHostedPayer(srv.URL, "pgk_test") + + cents = 2 + _, _, err := h.PayForQuotes(context.Background(), "", nil, nil, "0xt", "0xv") + if err == nil || !strings.Contains(err.Error(), "about $0.02 of storage credit") { + t.Fatalf("fiat 402 message wrong: %v", err) + } + if strings.Contains(err.Error(), "ANT") { + t.Fatalf("fiat message must not mention ANT: %v", err) + } + + cents = 0 // no rate configured gateway-side → ANT fallback + _, _, err = h.PayForQuotes(context.Background(), "", nil, nil, "0xt", "0xv") + if err == nil || !strings.Contains(err.Error(), "0.03515625 ANT") { + t.Fatalf("ANT fallback message wrong: %v", err) + } + + bal, rate, fee, err := h.AccountInfo(context.Background()) + if err != nil || bal != "5" || rate != "0.35" { + t.Fatalf("AccountInfo: %q %q %v", bal, rate, err) + } + // Older gateway without fee_per_batch_atto: tolerated as empty, no error. + if fee != "" { + t.Fatalf("absent fee must relay as empty, got %q", fee) + } +} + +// TestHostedFeePerBatchRelay proves the V2-1113 fee relay: AccountInfo +// carries fee_per_batch_atto and FeePerBatch turns it into an exact big.Int +// — present, absent (older gateway), and zero all tolerated without error. +func TestHostedFeePerBatchRelay(t *testing.T) { + for name, tc := range map[string]struct { + account map[string]any + want string + }{ + "fee present": {map[string]any{"account": "acme", "balance": "5", + "rate_usd_per_ant": "0.35", "fee_per_batch_atto": "5000000000000000"}, "5000000000000000"}, + "fee absent (older gateway)": {map[string]any{"account": "acme", "balance": "5", + "rate_usd_per_ant": "0.35"}, "0"}, + "fee zero": {map[string]any{"account": "acme", "balance": "5", + "fee_per_batch_atto": "0"}, "0"}, + } { + t.Run(name, func(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _ = json.NewEncoder(w).Encode(tc.account) + })) + defer srv.Close() + h := NewHostedPayer(srv.URL, "pgk_test") + if got, _ := h.FeePerBatch(context.Background()); got.String() != tc.want { + t.Errorf("FeePerBatch = %s, want %s", got, tc.want) + } + }) + } +} + +// TestHostedFeePerBatchCache proves the fee lookup's cache discipline: one +// gateway call per TTL window, a mid-window fee change invisible until the +// window rolls, and an unreachable gateway serving the last known value +// rather than erroring or zeroing (a stale fee beats a wrongly-net ceiling). +func TestHostedFeePerBatchCache(t *testing.T) { + calls := 0 + fee := "5000000000000000" + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls++ + _ = json.NewEncoder(w).Encode(map[string]any{"balance": "5", "fee_per_batch_atto": fee}) + })) + defer srv.Close() + h := NewHostedPayer(srv.URL, "pgk_test") + + if got, _ := h.FeePerBatch(context.Background()); got.String() != "5000000000000000" { + t.Fatalf("first fetch: %s", got) + } + fee = "9000000000000000" + if got, _ := h.FeePerBatch(context.Background()); got.String() != "5000000000000000" { + t.Fatalf("within TTL the cached fee must serve, got %s", got) + } + if calls != 1 { + t.Fatalf("gateway asked %d times within one TTL, want 1", calls) + } + + h.feeTTL = 0 // expire the window + if got, _ := h.FeePerBatch(context.Background()); got.String() != "9000000000000000" { + t.Fatalf("expired window must refetch, got %s", got) + } + + // Gateway gone: the last known value keeps serving. + srv.Close() + if got, _ := h.FeePerBatch(context.Background()); got.String() != "9000000000000000" { + t.Fatalf("unreachable gateway must serve last known fee, got %s", got) + } + + // Never fetched successfully at all → zero AND flagged unknown, so a + // spend gate can refuse to treat it as a real zero (review of #163). + dead := NewHostedPayer("http://127.0.0.1:1", "pgk_test") + if got, known := dead.FeePerBatch(context.Background()); got.Sign() != 0 || known { + t.Fatalf("never-fetched fee must be 0 and unknown, got %s known=%v", got, known) + } +} + +// TestHostedPayTransportRetry proves the /pay resend (V2-931 case I root +// cause): the gateway dying mid-request EOFs the connection; the batch is +// idempotent gateway-side, so the payer resends and the upload survives the +// crash window instead of failing on a payment that actually went through. +func TestHostedPayTransportRetry(t *testing.T) { + drops := 2 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if drops > 0 { + drops-- + hj, ok := w.(http.Hijacker) + if !ok { + t.Fatal("no hijacker") + } + conn, _, _ := hj.Hijack() + _ = conn.Close() // client sees EOF — a killed gateway + return + } + _ = json.NewEncoder(w).Encode(map[string]any{ + "status": "paid", "pay_tx_hash": "0xabc", + "tx_hashes": map[string]string{"0xq1": "0xabc"}, "payment_key": "k9"}) + })) + defer srv.Close() + h := NewHostedPayer(srv.URL, "pgk_test") + h.payRetryWait = 10 * time.Millisecond + hashes, key, err := h.PayForQuotes(context.Background(), "", []antd.PaymentInfo{{QuoteHash: "0xq1"}}, nil, "0xt", "0xv") + if err != nil || hashes["0xq1"] != "0xabc" || key != "k9" { + t.Fatalf("retry path: %v %v %q", hashes, err, key) + } + + // Exhaustion still surfaces as unreachable. + dead := NewHostedPayer("http://127.0.0.1:1", "pgk_test") + dead.payAttempts, dead.payRetryWait = 2, time.Millisecond + _, _, err = dead.PayForQuotes(context.Background(), "", []antd.PaymentInfo{{QuoteHash: "0xq1"}}, nil, "0xt", "0xv") + if err == nil || !strings.Contains(err.Error(), "unreachable") { + t.Fatalf("exhaustion must report unreachable: %v", err) + } +} + +// TestHostedCostPerGBRelay proves the V2-1114 capacity-estimate relay: +// AccountDetails carries est_cost_per_gb_atto plus the opaque basis object +// when the gateway serves them, and reads absence (older gateway, thin paid +// history) as empty fields — never an error. AccountInfo's trio is +// unaffected either way. +func TestHostedCostPerGBRelay(t *testing.T) { + basis := `{"median_paid_per_quote_atto":"105468750000000","sample_quotes":128,"window":"7d","chunks_per_gb":256,"batches_per_gb":1}` + + t.Run("present", func(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write([]byte(`{"account":"acme","balance":"5","rate_usd_per_ant":"0.35",` + + `"fee_per_batch_atto":"5000000000000000",` + + `"est_cost_per_gb_atto":"32000000000000000","est_cost_per_gb_basis":` + basis + `}`)) + })) + defer srv.Close() + d, err := NewHostedPayer(srv.URL, "pgk_test").AccountDetails(context.Background()) + if err != nil { + t.Fatal(err) + } + if d.BalanceAtto != "5" || d.RateUSDPerANT != "0.35" || d.FeePerBatchAtto != "5000000000000000" { + t.Fatalf("trio regressed: %+v", d) + } + if d.EstCostPerGBAtto != "32000000000000000" { + t.Fatalf("est cost = %q", d.EstCostPerGBAtto) + } + // The basis relays opaquely — semantically the same JSON. + var got, want map[string]any + if json.Unmarshal(d.EstCostPerGBBasis, &got) != nil || json.Unmarshal([]byte(basis), &want) != nil { + t.Fatalf("basis not JSON: %s", d.EstCostPerGBBasis) + } + if len(got) != len(want) || got["sample_quotes"] != want["sample_quotes"] || got["window"] != "7d" { + t.Fatalf("basis relayed wrong: %s", d.EstCostPerGBBasis) + } + }) + + t.Run("absent (older gateway / thin history)", func(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _ = json.NewEncoder(w).Encode(map[string]any{"account": "acme", "balance": "5", "rate_usd_per_ant": "0.35"}) + })) + defer srv.Close() + d, err := NewHostedPayer(srv.URL, "pgk_test").AccountDetails(context.Background()) + if err != nil { + t.Fatalf("absence must never error: %v", err) + } + if d.EstCostPerGBAtto != "" || d.EstCostPerGBBasis != nil { + t.Fatalf("absent fields must relay empty: %+v", d) + } + }) +} + +// TestHostedFeePerBatchUnknownUntilFirstFetch proves the fee-unknown signal +// (review of #163): a gateway that is unreachable before any successful +// fetch yields known=false — the fee is not zero, it is unknown, and the +// worker must not gate spend as if it were zero. After one success the fee +// is known and stays known (stale) through a later outage. +func TestHostedFeePerBatchUnknownUntilFirstFetch(t *testing.T) { + up := true + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if !up { + http.Error(w, "down", http.StatusBadGateway) + return + } + _ = json.NewEncoder(w).Encode(map[string]any{"balance": "5", "fee_per_batch_atto": "5000000000000000"}) + })) + defer srv.Close() + h := NewHostedPayer(srv.URL, "pgk_test") + h.feeTTL = 0 // refetch on every call so the outage/recovery sequence is observable + + up = false + if fee, known := h.FeePerBatch(context.Background()); known || fee.Sign() != 0 { + t.Fatalf("before any successful fetch: fee=%s known=%v, want 0/false", fee, known) + } + up = true + if fee, known := h.FeePerBatch(context.Background()); !known || fee.String() != "5000000000000000" { + t.Fatalf("after a successful fetch: fee=%s known=%v, want 5000000000000000/true", fee, known) + } + up = false + if fee, known := h.FeePerBatch(context.Background()); !known || fee.String() != "5000000000000000" { + t.Fatalf("outage after a success must serve the stale fee as known: fee=%s known=%v", fee, known) + } +} diff --git a/internal/handlers/admin_billing.go b/internal/handlers/admin_billing.go new file mode 100644 index 0000000..4ffa464 --- /dev/null +++ b/internal/handlers/admin_billing.go @@ -0,0 +1,208 @@ +package handlers + +import ( + "context" + "encoding/json" + "log/slog" + "net/http" + "strings" + "time" + + "github.com/WithAutonomi/indelible/internal/config" + "github.com/WithAutonomi/indelible/internal/database" + "github.com/WithAutonomi/indelible/internal/evm" +) + +// Admin billing surface (V2-1097): the in-app home for hosted-mode funds. +// Every endpoint is a thin server-side relay to the payment gateway using +// the instance's tenant API key — the key must never reach the browser. +// The SPA supplies its own absolute return URLs (it knows its origin); the +// gateway validates them and Stripe lands the customer back on /admin/billing. + +// billingPayer builds the gateway client, or writes the refusal and returns +// nil when the instance is not in hosted mode (these endpoints have no +// meaning for local signing). +func billingPayer(w http.ResponseWriter, cfg *config.Config) *evm.HostedPayer { + if !cfg.PaymentBackend.Hosted() || cfg.PaymentGatewayURL == "" { + jsonError(w, "billing is only available with the hosted payment backend", http.StatusBadRequest) + return nil + } + return evm.NewHostedPayer(cfg.PaymentGatewayURL, cfg.PaymentGatewayAPIKey) +} + +// relayOut passes a gateway SUCCESS answer through unmodified — status code +// and body both — so a successful checkout or sync reads identically whether +// the caller hit the gateway directly or via this relay. +// +// A non-2xx answer is NOT echoed: the browser gets the gateway's status code +// and only its short operator-facing `error` string (or a generic message), +// while the raw body is logged server-side. Gateway internals stay off the +// wire (#163 review, V2-1269). +func relayOut(w http.ResponseWriter, op string, status int, raw []byte) { + if status >= 200 && status < 300 { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _, _ = w.Write(raw) + return + } + logged := raw + if len(logged) > 1024 { + logged = logged[:1024] + } + slog.Warn("billing relay: gateway refused", "op", op, "status", status, "gateway_body", string(logged)) + jsonError(w, gatewayErrorMessage(status, raw), status) +} + +// gatewayErrorMessage extracts the gateway's short `error` string from a +// refusal body — one line, bounded — or falls back to a generic message. +func gatewayErrorMessage(status int, raw []byte) string { + var g struct { + Error string `json:"error"` + } + msg := "" + if json.Unmarshal(raw, &g) == nil { + msg = strings.TrimSpace(g.Error) + } + if i := strings.IndexAny(msg, "\r\n"); i >= 0 { + msg = msg[:i] + } + if len(msg) > 200 { + msg = msg[:200] + } + if msg == "" { + return "payment gateway refused the request (HTTP " + http.StatusText(status) + ")" + } + return msg +} + +// @Summary Billing summary +// @Description Hosted-mode billing: gateway credits and credited top-up history +// @Tags Admin: Billing +// @Produce json +// @Success 200 {object} map[string]interface{} +// @Failure 400 {object} map[string]string "Not in hosted payment mode" +// @Router /admin/billing [get] +// @Security BearerAuth +func AdminBillingSummary(db *database.DB, cfg *config.Config) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + payer := billingPayer(w, cfg) + if payer == nil { + return + } + out := map[string]any{ + "payment_backend": string(config.PaymentBackendHosted), + "payment_gateway_url": cfg.PaymentGatewayURL, + } + ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second) + defer cancel() + // Credits and history are best-effort separately: a gateway hiccup + // on one must not blank the other. + if d, err := payer.AccountDetails(ctx); err == nil { + out["gateway_credit_atto"] = d.BalanceAtto + // Exact USD-per-ANT rate for fiat display (V2-1100). + if d.RateUSDPerANT != "" { + out["rate_usd_per_ant"] = d.RateUSDPerANT + } + // Per-batch network fee (V2-1098), relayed for fee-aware + // estimates and billing transparency (V2-1113). + if d.FeePerBatchAtto != "" { + out["fee_per_batch_atto"] = d.FeePerBatchAtto + } + // Directional cost-per-GB estimate + methodology basis (V2-1114): + // the "≈ N GB remaining at current prices" line beside the + // balance. Rides the live balance call this summary already makes + // (fresher than the wallet-status 60s cache, zero extra requests). + // Absent — older gateway or thin paid history — hides the line. + if d.EstCostPerGBAtto != "" { + out["est_cost_per_gb_atto"] = d.EstCostPerGBAtto + if len(d.EstCostPerGBBasis) > 0 { + out["est_cost_per_gb_basis"] = d.EstCostPerGBBasis + } + } + } + if status, raw, err := payer.Credits(ctx); err == nil && status == http.StatusOK { + var c struct { + Credits json.RawMessage `json:"credits"` + } + if json.Unmarshal(raw, &c) == nil && c.Credits != nil { + out["credits"] = c.Credits + } + } + jsonResponse(w, http.StatusOK, out) + } +} + +type topupCheckoutRequest struct { + AmountUSDCents int64 `json:"amount_usd_cents"` + // Absolute URLs back into this instance's UI; validated by the gateway + // (http/https only). Stripe substitutes {CHECKOUT_SESSION_ID} in + // success_url if the placeholder is present. + SuccessURL string `json:"success_url"` + CancelURL string `json:"cancel_url"` +} + +// @Summary Start a card top-up +// @Description Create a Stripe Checkout session at the gateway; returns the hosted payment page URL and the exact credit +// @Tags Admin: Billing +// @Accept json +// @Produce json +// @Param body body topupCheckoutRequest true "Amount in USD cents and return URLs" +// @Success 200 {object} map[string]interface{} "session_id, url, credit_atto" +// @Failure 400 {object} map[string]string +// @Failure 502 {object} map[string]string "Gateway unreachable" +// @Router /admin/billing/topup-checkout [post] +// @Security BearerAuth +func AdminBillingTopupCheckout(db *database.DB, cfg *config.Config) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + payer := billingPayer(w, cfg) + if payer == nil { + return + } + var req topupCheckoutRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + jsonError(w, "invalid JSON: "+err.Error(), http.StatusBadRequest) + return + } + status, raw, err := payer.TopupCheckout(r.Context(), req.AmountUSDCents, req.SuccessURL, req.CancelURL) + if err != nil { + jsonError(w, err.Error(), http.StatusBadGateway) + return + } + relayOut(w, "topup-checkout", status, raw) + } +} + +type topupSyncRequest struct { + SessionID string `json:"session_id"` +} + +// @Summary Sync a top-up +// @Description Ask the gateway to retrieve the Checkout session from Stripe and credit it if paid (idempotent; webhook-loss fallback) +// @Tags Admin: Billing +// @Accept json +// @Produce json +// @Param body body topupSyncRequest true "Checkout session id" +// @Success 200 {object} map[string]interface{} "credited, payment_status" +// @Failure 400 {object} map[string]string +// @Failure 502 {object} map[string]string "Gateway unreachable" +// @Router /admin/billing/topup-sync [post] +// @Security BearerAuth +func AdminBillingTopupSync(db *database.DB, cfg *config.Config) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + payer := billingPayer(w, cfg) + if payer == nil { + return + } + var req topupSyncRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil || req.SessionID == "" { + jsonError(w, "session_id required", http.StatusBadRequest) + return + } + status, raw, err := payer.TopupSync(r.Context(), req.SessionID) + if err != nil { + jsonError(w, err.Error(), http.StatusBadGateway) + return + } + relayOut(w, "topup-sync", status, raw) + } +} diff --git a/internal/handlers/admin_wallets.go b/internal/handlers/admin_wallets.go index 734b0f1..f343582 100644 --- a/internal/handlers/admin_wallets.go +++ b/internal/handlers/admin_wallets.go @@ -1,12 +1,14 @@ package handlers import ( + "context" "encoding/json" "errors" "fmt" "net/http" "strconv" "strings" + "time" "github.com/ethereum/go-ethereum/crypto" "github.com/go-chi/chi/v5" @@ -72,7 +74,26 @@ func AdminListWallets(db *database.DB, cfg *config.Config) http.HandlerFunc { resp = append(resp, toWalletResponse(wl)) } - jsonResponse(w, http.StatusOK, map[string]any{"wallets": resp}) + // Payment backend rides along so the wallet screen can state whether + // these wallets actually pay for uploads (hosted mode: they don't — + // the gateway settles from prepaid credits, V2-1086/V2-930). + hosted := cfg.PaymentBackend.Hosted() + out := map[string]any{ + "wallets": resp, + "payment_backend": string(cfg.PaymentBackend), + "payment_gateway_url": cfg.PaymentGatewayURL, + } + if hosted && cfg.PaymentGatewayURL != "" { + // Remaining credits, best-effort: an unreachable gateway must + // not break the wallets screen — the field is simply absent. + balCtx, cancel := context.WithTimeout(r.Context(), 3*time.Second) + defer cancel() + payer := evm.NewHostedPayer(cfg.PaymentGatewayURL, cfg.PaymentGatewayAPIKey) + if bal, err := payer.AccountBalance(balCtx); err == nil { + out["gateway_credit_atto"] = bal + } + } + jsonResponse(w, http.StatusOK, out) } } @@ -94,7 +115,7 @@ func AdminListWallets(db *database.DB, cfg *config.Config) http.HandlerFunc { // placeholder key into the shared DB — these operations belong on the writer. func requireWalletKey(w http.ResponseWriter, cfg *config.Config) bool { if !cfg.WalletKeyConfigured() { - jsonError(w, "wallet/OIDC management is unavailable on this instance (no wallet encryption key configured); perform it on the writer instance", http.StatusServiceUnavailable) + jsonError(w, "wallet/OIDC management is unavailable on this instance (no wallet encryption key configured): on a reader, perform it on the writer; on a hosted-backend writer, set INDELIBLE_WALLET_ENCRYPTION_KEY to enable OIDC client-secret storage", http.StatusServiceUnavailable) return false } return true diff --git a/internal/handlers/billing_relay_test.go b/internal/handlers/billing_relay_test.go new file mode 100644 index 0000000..2048fb4 --- /dev/null +++ b/internal/handlers/billing_relay_test.go @@ -0,0 +1,95 @@ +package handlers + +import ( + "bytes" + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/WithAutonomi/indelible/internal/config" +) + +// TestBillingRelay_GatewayErrorBodyNotEchoed proves the relay forwards a +// gateway refusal's status and short error string only — never the raw body +// (#163 review, V2-1269). +func TestBillingRelay_GatewayErrorBodyNotEchoed(t *testing.T) { + gw := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusBadRequest) + _, _ = w.Write([]byte(`{"error":"amount below minimum","internal":"stack: topup.go:42 ledger=pg://secret"}`)) + })) + defer gw.Close() + cfg := &config.Config{PaymentBackend: config.PaymentBackendHosted, PaymentGatewayURL: gw.URL, PaymentGatewayAPIKey: "pgk_test"} + + req := httptest.NewRequest(http.MethodPost, "/admin/billing/topup-checkout", + bytes.NewBufferString(`{"amount_usd_cents":100,"success_url":"http://x/s","cancel_url":"http://x/c"}`)) + rec := httptest.NewRecorder() + AdminBillingTopupCheckout(nil, cfg).ServeHTTP(rec, req) + + if rec.Code != http.StatusBadRequest { + t.Fatalf("status = %d, want 400 (gateway status preserved)", rec.Code) + } + body := rec.Body.String() + if strings.Contains(body, "internal") || strings.Contains(body, "secret") { + t.Fatalf("raw gateway body leaked to the browser: %s", body) + } + var out map[string]string + if err := json.Unmarshal(rec.Body.Bytes(), &out); err != nil || out["error"] != "amount below minimum" { + t.Fatalf("want {error: amount below minimum}, got %s", body) + } +} + +// TestBillingRelay_SuccessPassesThrough keeps the success contract: a 2xx +// gateway answer is relayed byte-for-byte. +func TestBillingRelay_SuccessPassesThrough(t *testing.T) { + const ok = `{"session_id":"cs_1","url":"https://checkout.test/cs_1","credit_atto":"2966"}` + gw := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(ok)) + })) + defer gw.Close() + cfg := &config.Config{PaymentBackend: config.PaymentBackendHosted, PaymentGatewayURL: gw.URL, PaymentGatewayAPIKey: "pgk_test"} + + req := httptest.NewRequest(http.MethodPost, "/admin/billing/topup-checkout", + bytes.NewBufferString(`{"amount_usd_cents":100,"success_url":"http://x/s","cancel_url":"http://x/c"}`)) + rec := httptest.NewRecorder() + AdminBillingTopupCheckout(nil, cfg).ServeHTTP(rec, req) + if rec.Code != http.StatusOK || rec.Body.String() != ok { + t.Fatalf("success must pass through unmodified: %d %s", rec.Code, rec.Body.String()) + } +} + +// TestGatewayPricingCache_KeyedByGateway proves two configs pointing at two +// gateways never serve each other's numbers, and each gateway is asked once +// per TTL (#163 review, V2-1269). +func TestGatewayPricingCache_KeyedByGateway(t *testing.T) { + mk := func(rate string, calls *int) *httptest.Server { + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + *calls++ + _ = json.NewEncoder(w).Encode(map[string]any{"balance": "0", "rate_usd_per_ant": rate}) + })) + } + var callsA, callsB int + a := mk("0.11", &callsA) + defer a.Close() + b := mk("0.22", &callsB) + defer b.Close() + cfgA := &config.Config{PaymentGatewayURL: a.URL, PaymentGatewayAPIKey: "pgk_a"} + cfgB := &config.Config{PaymentGatewayURL: b.URL, PaymentGatewayAPIKey: "pgk_b"} + + if got := cachedGatewayPricing(context.Background(), cfgA).rate; got != "0.11" { + t.Fatalf("gateway A rate = %q, want 0.11", got) + } + if got := cachedGatewayPricing(context.Background(), cfgB).rate; got != "0.22" { + t.Fatalf("gateway B rate = %q, want 0.22 (not A's cached value)", got) + } + if got := cachedGatewayPricing(context.Background(), cfgA).rate; got != "0.11" { + t.Fatalf("gateway A second read = %q, want 0.11", got) + } + if callsA != 1 || callsB != 1 { + t.Fatalf("within one TTL each gateway must be asked once: A=%d B=%d", callsA, callsB) + } +} diff --git a/internal/handlers/router.go b/internal/handlers/router.go index 347d994..10f5cb6 100644 --- a/internal/handlers/router.go +++ b/internal/handlers/router.go @@ -210,6 +210,12 @@ func NewRouter(cfg *config.Config, db *database.DB, antdInfo AntdInfoProvider, d r.Delete("/admin/tokens/bulk", AdminBulkRevokeTokens(db)) // Wallet management + // Billing (V2-1097): hosted-mode funds surface — server-side + // relays to the payment gateway (tenant key stays server-side). + r.Get("/admin/billing", AdminBillingSummary(db, cfg)) + r.Post("/admin/billing/topup-checkout", AdminBillingTopupCheckout(db, cfg)) + r.Post("/admin/billing/topup-sync", AdminBillingTopupSync(db, cfg)) + r.Get("/admin/wallets", AdminListWallets(db, cfg)) r.Post("/admin/wallets", AdminCreateWallet(db, cfg)) r.Put("/admin/wallets/{id}/default", AdminSetDefaultWallet(db, cfg)) diff --git a/internal/handlers/uploads.go b/internal/handlers/uploads.go index 2c59f78..4160fd0 100644 --- a/internal/handlers/uploads.go +++ b/internal/handlers/uploads.go @@ -7,6 +7,7 @@ import ( "fmt" "io" "log/slog" + "math/big" "net/http" "os" "path/filepath" @@ -38,15 +39,22 @@ type uploadResponse struct { DatamapAddress *string `json:"datamap_address"` EstimatedCost *string `json:"estimated_cost"` ActualCost *string `json:"actual_cost"` - ErrorMessage *string `json:"error_message"` - BackoffUntil *string `json:"backoff_until,omitempty"` - BackoffAttempt int `json:"backoff_attempt,omitempty"` - LastQuotedCost *string `json:"last_quoted_cost,omitempty"` - QueuedAt string `json:"queued_at"` - ProcessingAt *string `json:"processing_at"` - CompletedAt *string `json:"completed_at"` - FailedAt *string `json:"failed_at"` - CreatedAt string `json:"created_at"` + // Payment provenance (V2-1086): "local" (instance wallet) or "hosted" + // (gateway credits) + the gateway's batch key; absent when nothing was paid. + PaymentBackend *string `json:"payment_backend,omitempty"` + GatewayPaymentKey *string `json:"gateway_payment_key,omitempty"` + // GatewayFeeAtto itemizes the gateway's per-batch network fee out of the + // gross actual_cost (V2-1098); absent when no fee was charged. + GatewayFeeAtto *string `json:"gateway_fee_atto,omitempty"` + ErrorMessage *string `json:"error_message"` + BackoffUntil *string `json:"backoff_until,omitempty"` + BackoffAttempt int `json:"backoff_attempt,omitempty"` + LastQuotedCost *string `json:"last_quoted_cost,omitempty"` + QueuedAt string `json:"queued_at"` + ProcessingAt *string `json:"processing_at"` + CompletedAt *string `json:"completed_at"` + FailedAt *string `json:"failed_at"` + CreatedAt string `json:"created_at"` } func toUploadResponse(u *services.Upload) uploadResponse { @@ -70,6 +78,15 @@ func toUploadResponse(u *services.Upload) uploadResponse { if u.ActualCost.Valid { r.ActualCost = &u.ActualCost.String } + if u.PaymentBackend.Valid { + r.PaymentBackend = &u.PaymentBackend.String + } + if u.GatewayPaymentKey.Valid { + r.GatewayPaymentKey = &u.GatewayPaymentKey.String + } + if u.GatewayFeeAtto.Valid { + r.GatewayFeeAtto = &u.GatewayFeeAtto.String + } if u.ErrorMessage.Valid { r.ErrorMessage = &u.ErrorMessage.String } @@ -129,11 +146,14 @@ func CreateUpload(db *database.DB, cfg *config.Config) http.HandlerFunc { walletSvc := services.NewWalletService(db, cfg.WalletKeyring()) return func(w http.ResponseWriter, r *http.Request) { - // Pre-flight: reject early if no wallet is configured - wallet, err := walletSvc.GetDefault() - if err != nil || wallet == nil { - jsonErrorWithCode(w, "No wallet configured", "wallet_not_configured", http.StatusServiceUnavailable) - return + // Pre-flight: reject early if no wallet is configured. Hosted mode + // (V2-929) needs no wallet — the payment gateway's treasury signs. + if cfg.PaymentBackend.NeedsWallet() { + wallet, err := walletSvc.GetDefault() + if err != nil || wallet == nil { + jsonErrorWithCode(w, "No wallet configured", "wallet_not_configured", http.StatusServiceUnavailable) + return + } } // Disk back-pressure: the disk-alert worker sets "uploads_paused" when the @@ -555,7 +575,7 @@ func GetUpload(db *database.DB) http.HandlerFunc { // runs self-encryption + a real quote round-trip with the live network's pricer. // // @Summary Quote upload cost -// @Description Get an exact cost quote by sending the file bytes. antd runs self-encryption and queries the live network for chunk pricing — no estimation, no scaling. Returns a structured estimated_cost object with cost, chunk_count, gas, and payment_mode. +// @Description Get an exact cost quote by sending the file bytes. antd runs self-encryption and queries the live network for chunk pricing — no estimation, no scaling. Returns a structured estimated_cost object with cost, chunk_count, gas, and payment_mode (antd's on-chain payment strategy: auto | merkle | single). With the hosted payment backend the gateway debits gross — batch total plus a per-batch network fee (V2-1098) — so the response additionally carries gateway_fee_per_batch_atto, estimated_batch_count, and estimated_total_with_fee_atto (V2-1113). // @Tags Uploads // @Accept multipart/form-data // @Produce json @@ -625,12 +645,32 @@ func QuoteUpload(db *database.DB, cfg *config.Config) http.HandlerFunc { return } - jsonResponse(w, http.StatusOK, map[string]any{ + out := map[string]any{ "estimated_cost": est, "file_size": written, "original_filename": filepath.Base(header.Filename), "visibility": visibility, - }) + } + + // Hosted mode (V2-1113): the gateway debits GROSS — batch total plus + // a per-batch network fee (V2-1098) — so a quote without the fee + // understates what the credits will actually drop by. The worker + // settles one upload as exactly one gateway batch, so the estimate + // adds one fee. Still an estimate: full dedup at prepare time sends + // no batch and pays no fee. Exact big.Int math, atto in, atto out. + if cfg.PaymentBackend.Hosted() && cfg.PaymentGatewayURL != "" { + if fee := cachedGatewayPricing(r.Context(), cfg).fee; fee != "" { + if feeInt, ok := new(big.Int).SetString(fee, 10); ok && feeInt.Sign() > 0 { + out["gateway_fee_per_batch_atto"] = feeInt.String() + out["estimated_batch_count"] = 1 + if cost, ok := new(big.Int).SetString(est.Cost, 10); ok { + out["estimated_total_with_fee_atto"] = new(big.Int).Add(cost, feeInt).String() + } + } + } + } + + jsonResponse(w, http.StatusOK, out) } } @@ -1286,7 +1326,6 @@ func DeleteUpload(db *database.DB, cache *downloadcache.Store) http.HandlerFunc } } - // effectiveAllowlist resolves the content-type allowlist for an upload using // the override chain: token > user > system setting > built-in default. // Returns a comma-separated string of patterns (same shape as the setting). diff --git a/internal/handlers/wallet_status.go b/internal/handlers/wallet_status.go index 777797f..0fd4d4c 100644 --- a/internal/handlers/wallet_status.go +++ b/internal/handlers/wallet_status.go @@ -1,13 +1,82 @@ package handlers import ( + "context" + "encoding/json" "net/http" + "sync" + "time" "github.com/WithAutonomi/indelible/internal/config" "github.com/WithAutonomi/indelible/internal/database" + "github.com/WithAutonomi/indelible/internal/evm" "github.com/WithAutonomi/indelible/internal/services" ) +// gatewayPricingCache memoizes the gateway's USD-per-ANT rate (crypto-free +// display, V2-1100), per-batch network fee (fee-aware estimates, V2-1113) +// and cost-per-GB estimate + basis (capacity display, V2-1114): every +// authenticated view reads wallet-status, so the gateway is asked at most +// once per minute. +// +// Entries are keyed by gateway identity (URL + API key), never process-wide: +// two Configs in one process, or one Config re-pointed at another gateway, +// must never serve each other's numbers (#163 review, V2-1269). +var gatewayPricingCache struct { + sync.Mutex + byGateway map[string]*gatewayPricingEntry +} + +type gatewayPricingEntry struct { + pricing gatewayPricing + fetched time.Time +} + +// gatewayPricingTTL bounds how often any one gateway is asked. +const gatewayPricingTTL = time.Minute + +// gatewayPricing is the cached display trio+basis; every field optional +// (older gateway, nothing configured, thin history — all read as absent). +type gatewayPricing struct { + rate string + fee string + estCostGB string + estBasis json.RawMessage +} + +func cachedGatewayPricing(ctx context.Context, cfg *config.Config) gatewayPricing { + key := cfg.PaymentGatewayURL + "\x00" + cfg.PaymentGatewayAPIKey + gatewayPricingCache.Lock() + defer gatewayPricingCache.Unlock() + if gatewayPricingCache.byGateway == nil { + gatewayPricingCache.byGateway = map[string]*gatewayPricingEntry{} + } + e := gatewayPricingCache.byGateway[key] + if e == nil { + e = &gatewayPricingEntry{} + gatewayPricingCache.byGateway[key] = e + } + if !e.fetched.IsZero() && time.Since(e.fetched) < gatewayPricingTTL { + return e.pricing + } + rateCtx, cancel := context.WithTimeout(ctx, 2*time.Second) + defer cancel() + payer := evm.NewHostedPayer(cfg.PaymentGatewayURL, cfg.PaymentGatewayAPIKey) + d, err := payer.AccountDetails(rateCtx) + if err != nil { + // Best-effort display data: keep serving the stale values and try + // again after the normal interval. + e.fetched = time.Now() + return e.pricing + } + e.pricing = gatewayPricing{ + rate: d.RateUSDPerANT, fee: d.FeePerBatchAtto, + estCostGB: d.EstCostPerGBAtto, estBasis: d.EstCostPerGBBasis, + } + e.fetched = time.Now() + return e.pricing +} + // WalletStatus godoc // @Summary Check wallet configuration status // @Description Returns whether a default wallet is configured for uploads @@ -23,8 +92,39 @@ func WalletStatus(db *database.DB, cfg *config.Config) http.HandlerFunc { wallet, err := walletSvc.GetDefault() hasWallet := err == nil && wallet != nil - jsonResponse(w, http.StatusOK, map[string]any{ - "has_default_wallet": hasWallet, - }) + // The UI reads this as "can this instance pay for uploads". Hosted + // mode pays via the gateway with no wallet at all (V2-929). + hosted := cfg.PaymentBackend.Hosted() + out := map[string]any{ + "has_default_wallet": hasWallet || hosted, + "payment_backend": string(cfg.PaymentBackend), + } + // Crypto-free display (V2-1100): the gateway's USD-per-ANT rate, so + // every view can render costs and balances in fiat. Best-effort and + // cached — absent when the gateway has no rate or is unreachable. + if hosted && cfg.PaymentGatewayURL != "" { + p := cachedGatewayPricing(r.Context(), cfg) + if p.rate != "" { + out["gateway_rate_usd_per_ant"] = p.rate + } + // Fee-aware estimates (V2-1113): the per-batch network fee the + // gateway adds to every settled batch (V2-1098). The web app folds + // it into pre-upload estimates so they match the gross debit. + // Absent when the gateway charges none or predates the field. + if p.fee != "" { + out["gateway_fee_per_batch_atto"] = p.fee + } + // Capacity display (V2-1114): the gateway's directional + // cost-per-GB estimate and its methodology basis. Absent when the + // gateway predates the field or has too little paid history — + // the UI hides the "≈ N GB remaining" line entirely. + if p.estCostGB != "" { + out["gateway_est_cost_per_gb_atto"] = p.estCostGB + if len(p.estBasis) > 0 { + out["gateway_est_cost_per_gb_basis"] = p.estBasis + } + } + } + jsonResponse(w, http.StatusOK, out) } } diff --git a/internal/services/collection.go b/internal/services/collection.go index e56bb23..5cfffa8 100644 --- a/internal/services/collection.go +++ b/internal/services/collection.go @@ -251,7 +251,7 @@ func (s *CollectionService) ListFiles(collectionID int64, limit, offset int) ([] rows, err := s.db.Query( `SELECT u.id, u.uuid, u.user_id, u.token_id, u.filename, u.original_filename, u.file_size, u.content_type, u.visibility, u.status, u.status_detail, u.datamap_address, u.estimated_cost, u.actual_cost, u.error_message, u.temp_path, - u.data_map, u.backoff_until, u.backoff_attempt, u.last_quoted_cost, + u.data_map, u.backoff_until, u.backoff_attempt, u.last_quoted_cost, u.payment_backend, u.gateway_payment_key, u.gateway_fee_atto, u.queued_at, u.processing_at, u.completed_at, u.failed_at, u.created_at FROM uploads u INNER JOIN collection_files cf ON u.id = cf.upload_id diff --git a/internal/services/tag.go b/internal/services/tag.go index 1c553f6..de0fec5 100644 --- a/internal/services/tag.go +++ b/internal/services/tag.go @@ -169,7 +169,7 @@ func (s *TagService) Search(tagFilters map[string]string, query string, userID i // Fetch results selectSQL := `SELECT DISTINCT u.id, u.uuid, u.user_id, u.token_id, u.filename, u.original_filename, u.file_size, u.content_type, u.visibility, u.status, u.status_detail, u.datamap_address, u.estimated_cost, u.actual_cost, u.error_message, u.temp_path, - u.data_map, u.backoff_until, u.backoff_attempt, u.last_quoted_cost, + u.data_map, u.backoff_until, u.backoff_attempt, u.last_quoted_cost, u.payment_backend, u.gateway_payment_key, u.gateway_fee_atto, u.queued_at, u.processing_at, u.completed_at, u.failed_at, u.created_at ` + baseQuery + where + ` ORDER BY u.created_at DESC LIMIT ? OFFSET ?` queryArgs := make([]any, len(args), len(args)+2) @@ -206,7 +206,7 @@ func (s *TagService) SearchBySelector(userID int64, selectorClauses []string, se query := `SELECT DISTINCT u.id, u.uuid, u.user_id, u.token_id, u.filename, u.original_filename, u.file_size, u.content_type, u.visibility, u.status, u.status_detail, u.datamap_address, u.estimated_cost, u.actual_cost, u.error_message, u.temp_path, - u.data_map, u.backoff_until, u.backoff_attempt, u.last_quoted_cost, + u.data_map, u.backoff_until, u.backoff_attempt, u.last_quoted_cost, u.payment_backend, u.gateway_payment_key, u.gateway_fee_atto, u.queued_at, u.processing_at, u.completed_at, u.failed_at, u.created_at FROM uploads u WHERE u.user_id = ?` args := []interface{}{userID} @@ -269,7 +269,7 @@ func (s *TagService) SearchWithSelector(selectorClauses []string, selectorArgs [ // Fetch selectSQL := `SELECT DISTINCT u.id, u.uuid, u.user_id, u.token_id, u.filename, u.original_filename, u.file_size, u.content_type, u.visibility, u.status, u.status_detail, u.datamap_address, u.estimated_cost, u.actual_cost, u.error_message, u.temp_path, - u.data_map, u.backoff_until, u.backoff_attempt, u.last_quoted_cost, + u.data_map, u.backoff_until, u.backoff_attempt, u.last_quoted_cost, u.payment_backend, u.gateway_payment_key, u.gateway_fee_atto, u.queued_at, u.processing_at, u.completed_at, u.failed_at, u.created_at ` + baseSQL + ` ORDER BY u.created_at DESC LIMIT ? OFFSET ?` queryArgs := make([]any, len(args), len(args)+2) diff --git a/internal/services/transaction.go b/internal/services/transaction.go index 5d4b811..7c180e8 100644 --- a/internal/services/transaction.go +++ b/internal/services/transaction.go @@ -30,7 +30,12 @@ func NewTransactionService(db *database.DB) *TransactionService { } // Record logs a new transaction with an optional on-chain tx hash. +// walletID 0 records NULL — hosted payments belong to no wallet (V2-929). func (s *TransactionService) Record(walletID int64, uploadID *int64, txType, amount, balanceAfter, txHash string) (*Transaction, error) { + var wID sql.NullInt64 + if walletID != 0 { + wID = sql.NullInt64{Int64: walletID, Valid: true} + } var uID sql.NullInt64 if uploadID != nil { uID = sql.NullInt64{Int64: *uploadID, Valid: true} @@ -43,7 +48,7 @@ func (s *TransactionService) Record(walletID int64, uploadID *int64, txType, amo var id int64 err := s.db.QueryRow( `INSERT INTO transactions (wallet_id, upload_id, tx_type, amount, balance_after, tx_hash) VALUES (?, ?, ?, ?, ?, ?) RETURNING id`, - walletID, uID, txType, amount, balanceAfter, hash, + wID, uID, txType, amount, balanceAfter, hash, ).Scan(&id) if err != nil { return nil, err @@ -64,7 +69,7 @@ func (s *TransactionService) HasByUpload(uploadID int64) (bool, error) { func (s *TransactionService) GetByID(id int64) (*Transaction, error) { t := &Transaction{} err := s.db.QueryRow( - `SELECT id, wallet_id, upload_id, tx_type, amount, balance_after, tx_hash, created_at FROM transactions WHERE id = ?`, id, + `SELECT id, COALESCE(wallet_id, 0), upload_id, tx_type, amount, balance_after, tx_hash, created_at FROM transactions WHERE id = ?`, id, ).Scan(&t.ID, &t.WalletID, &t.UploadID, &t.TxType, &t.Amount, &t.BalanceAfter, &t.TxHash, &t.CreatedAt) if err != nil { return nil, err @@ -82,7 +87,7 @@ func (s *TransactionService) ListByWallet(walletID int64, limit, offset int) ([] s.db.QueryRow(`SELECT COUNT(*) FROM transactions WHERE wallet_id = ?`, walletID).Scan(&total) rows, err := s.db.Query( - `SELECT id, wallet_id, upload_id, tx_type, amount, balance_after, tx_hash, created_at + `SELECT id, COALESCE(wallet_id, 0), upload_id, tx_type, amount, balance_after, tx_hash, created_at FROM transactions WHERE wallet_id = ? ORDER BY created_at DESC LIMIT ? OFFSET ?`, walletID, limit, offset, ) @@ -136,7 +141,7 @@ func (s *TransactionService) List(walletID *int64, txType string, since, until * } rows, err := s.db.Query( - `SELECT id, wallet_id, upload_id, tx_type, amount, balance_after, tx_hash, created_at + `SELECT id, COALESCE(wallet_id, 0), upload_id, tx_type, amount, balance_after, tx_hash, created_at FROM transactions`+where+` ORDER BY created_at DESC LIMIT ? OFFSET ?`, append(args, limit, offset)..., ) diff --git a/internal/services/upload.go b/internal/services/upload.go index c90cc63..e8bf655 100644 --- a/internal/services/upload.go +++ b/internal/services/upload.go @@ -37,6 +37,15 @@ type Upload struct { BackoffUntil sql.NullTime BackoffAttempt int LastQuotedCost sql.NullString + // Payment provenance (V2-1086): how the payment was settled — "local" + // (this instance's wallet signed) or "hosted" (gateway credits) — and, + // when hosted, the gateway's batch idempotency key (the join to the + // gateway's payments ledger). NULL when nothing was paid. + PaymentBackend sql.NullString + GatewayPaymentKey sql.NullString + // GatewayFeeAtto is the gateway's per-batch network fee (V2-1098), + // itemized out of the gross actual_cost. NULL when no fee was charged. + GatewayFeeAtto sql.NullString QueuedAt time.Time ProcessingAt sql.NullTime CompletedAt sql.NullTime @@ -46,7 +55,7 @@ type Upload struct { const uploadColumns = `id, uuid, user_id, token_id, filename, original_filename, file_size, content_type, visibility, status, status_detail, datamap_address, estimated_cost, actual_cost, error_message, temp_path, - data_map, backoff_until, backoff_attempt, last_quoted_cost, + data_map, backoff_until, backoff_attempt, last_quoted_cost, payment_backend, gateway_payment_key, gateway_fee_atto, queued_at, processing_at, completed_at, failed_at, created_at` func scanUpload(scanner interface{ Scan(...any) error }) (*Upload, error) { @@ -54,7 +63,7 @@ func scanUpload(scanner interface{ Scan(...any) error }) (*Upload, error) { err := scanner.Scan( &u.ID, &u.UUID, &u.UserID, &u.TokenID, &u.Filename, &u.OriginalFilename, &u.FileSize, &u.ContentType, &u.Visibility, &u.Status, &u.StatusDetail, &u.DatamapAddress, &u.EstimatedCost, &u.ActualCost, &u.ErrorMessage, &u.TempPath, - &u.DataMap, &u.BackoffUntil, &u.BackoffAttempt, &u.LastQuotedCost, + &u.DataMap, &u.BackoffUntil, &u.BackoffAttempt, &u.LastQuotedCost, &u.PaymentBackend, &u.GatewayPaymentKey, &u.GatewayFeeAtto, &u.QueuedAt, &u.ProcessingAt, &u.CompletedAt, &u.FailedAt, &u.CreatedAt, ) return u, err @@ -426,6 +435,18 @@ func (s *UploadService) ListPrivatePublishCandidates(limit int) ([]*Upload, erro return out, rows.Err() } +// SetPaymentProvenance stamps how an upload's payment was settled (V2-1086): +// mode is "local" (this instance's wallet) or "hosted" (gateway credits); +// gatewayKey is the gateway's batch idempotency key and feeAtto its per-batch +// network fee (V2-1098) — both empty for local. +func (s *UploadService) SetPaymentProvenance(id int64, mode, gatewayKey, feeAtto string) error { + _, err := s.db.Exec( + `UPDATE uploads SET payment_backend = ?, gateway_payment_key = NULLIF(?, ''), gateway_fee_atto = NULLIF(?, '') WHERE id = ?`, + mode, gatewayKey, feeAtto, id, + ) + return err +} + // MarkFailed transitions an upload to "failed" with an error message. func (s *UploadService) MarkFailed(id int64, errMsg string) error { _, err := s.db.Exec( diff --git a/internal/worker/system_monitor.go b/internal/worker/system_monitor.go index f241abd..512ec74 100644 --- a/internal/worker/system_monitor.go +++ b/internal/worker/system_monitor.go @@ -263,6 +263,9 @@ func isAntdHardDown(err error) bool { } func (m *SystemMonitor) checkEvmRpcHealth() { + if !m.cfg.PaymentBackend.NeedsWallet() { + return // hosted backend: this instance never talks to an EVM RPC (V2-929) + } if m.cfg.EvmRPCURL == "" { return // not yet configured (set during first upload) } diff --git a/internal/worker/upload.go b/internal/worker/upload.go index f5b42cc..74d996d 100644 --- a/internal/worker/upload.go +++ b/internal/worker/upload.go @@ -107,6 +107,48 @@ func estimatedUploadCost(prepared *antd.PrepareUploadResult) *big.Int { return new(big.Int) } +// grossUploadCost is the ceiling's comparison basis (V2-1113): the net quote +// (estimatedUploadCost) plus, when a per-batch fee applies, one fee for the +// single gateway batch a prepared wave upload settles as — the same GROSS +// basis the gateway actually debits and 402s on (V2-1098), so a configured +// max_gas_fee refuses BEFORE spending exactly when the gateway would charge +// past it. A full-dedup prepare (no payments) sends no batch and pays no fee. +// feePerBatch nil means no fee applies (local mode, or fee unknown → zero). +func grossUploadCost(prepared *antd.PrepareUploadResult, feePerBatch *big.Int) *big.Int { + cost := estimatedUploadCost(prepared) + if feePerBatch != nil && feePerBatch.Sign() > 0 && len(prepared.Payments) > 0 { + cost.Add(cost, feePerBatch) + } + return cost +} + +// payer is the payment seam: either the local EVM signer or the hosted +// gateway client (payment_backend=hosted, V2-929 PoC). Both settle a prepared +// batch and answer balance queries; the worker never sees the difference. +// signedQuotes carries the opaque signed artifacts from the prepare response +// (V2-926) — the hosted gateway verifies them before paying; local signing +// ignores them. +type payer interface { + // PayForQuotes additionally returns the settling party's payment + // reference — the gateway's batch idempotency key for hosted payments, + // "" for local signing — stamped on the upload as provenance (V2-1086). + PayForQuotes(ctx context.Context, privateKeyHex string, payments []antd.PaymentInfo, signedQuotes []antd.SignedQuoteEntry, tokenAddress, dataPaymentsAddress string) (map[string]string, string, error) + PayForMerkleTree(ctx context.Context, privateKeyHex string, depth int, poolCommitments []antd.PoolCommitmentEntry, merklePaymentTimestamp uint64, tokenAddress, merklePaymentsAddress string) (winnerPoolHash, totalAmount string, err error) + GetBalances(ctx context.Context, walletAddress, tokenAddress string) (string, string, error) + SetConfirmationTimeout(d time.Duration) + RPCUrl() string +} + +// localPayer adapts *evm.Signer to the payer seam: local signing has no use +// for the relayed signed quotes, so it drops them. Keeps evm.Signer's own +// signature untouched (migrate.EvmPayer and the audit-anchor worker use it). +type localPayer struct{ *evm.Signer } + +func (l localPayer) PayForQuotes(ctx context.Context, privateKeyHex string, payments []antd.PaymentInfo, _ []antd.SignedQuoteEntry, tokenAddress, dataPaymentsAddress string) (map[string]string, string, error) { + hashes, err := l.Signer.PayForQuotes(ctx, privateKeyHex, payments, tokenAddress, dataPaymentsAddress) + return hashes, "", err +} + // UploadWorker processes queued file uploads in the background. type UploadWorker struct { uploadSvc *services.UploadService @@ -116,7 +158,7 @@ type UploadWorker struct { webhookSvc *services.WebhookDeliveryService settingsSvc *services.CachedSettingsService antdClient *antd.Client - evmSigner *evm.Signer // lazily initialized on first upload + evmSigner payer // lazily initialized on first upload cfg *config.Config // dlCache is the shared download cache store, seeded write-through from // upload temp files after a successful store (V2-822). Nil disables @@ -359,15 +401,21 @@ func (w *UploadWorker) processUpload(ctx context.Context, upload *services.Uploa return fmt.Errorf("Quota exceeded: %w", err) } - // Get default wallet — required for external signer payment - wallet, err := w.walletSvc.GetDefault() - if err != nil { - return fmt.Errorf("No wallet configured for payment") - } - - walletKey, err := w.walletSvc.DecryptKey(wallet) - if err != nil { - return fmt.Errorf("Failed to decrypt wallet key") + // Local mode signs with the default wallet. Hosted mode needs NO wallet + // at all (V2-929): the gateway's treasury signs, so a wallet record is + // neither required nor consulted. + var wallet *services.Wallet + var err error + walletKey := "" + if w.cfg.PaymentBackend.NeedsWallet() { + wallet, err = w.walletSvc.GetDefault() + if err != nil { + return fmt.Errorf("No wallet configured for payment") + } + walletKey, err = w.walletSvc.DecryptKey(wallet) + if err != nil { + return fmt.Errorf("Failed to decrypt wallet key") + } } // Phase 1: Prepare upload — encrypts file, collects network quotes. @@ -376,23 +424,78 @@ func (w *UploadWorker) processUpload(ctx context.Context, upload *services.Uploa // one EVM tx, and finalize returns a network address for the DataMap. // Private visibility: DataMap stays in-memory and is stored locally. var prepared *antd.PrepareUploadResult - if upload.Visibility == "public" { + switch { + case w.cfg.PaymentBackend.WantsSignedQuotes(): + // Remote payer (V2-926): ask for the signed quotes so the gateway can + // verify the batch offline before paying. Requires antd >= 0.13.0. + opts := antd.PrepareOptions{IncludeSignedQuotes: true} + if upload.Visibility == "public" { + opts.Visibility = "public" + } + prepared, err = w.antdClient.PrepareUploadWithOptions(ctx, tempPath, opts) + case upload.Visibility == "public": prepared, err = w.antdClient.PrepareUploadPublic(ctx, tempPath) - } else { + default: prepared, err = w.antdClient.PrepareUpload(ctx, tempPath) } if err != nil { return fmt.Errorf("Failed to prepare upload: %w", err) } + // Hosted mode needs the gateway client BEFORE the cost ceiling: the + // gateway debits GROSS — batch total + per-batch network fee (V2-1098) — + // so the ceiling compares that same basis (V2-1113) and must ask the + // gateway what the fee is. The ensure-payer block further down is a no-op + // once this has run. + if w.cfg.PaymentBackend.Hosted() { + if w.cfg.PaymentGatewayURL == "" { + return fmt.Errorf("payment_backend=hosted requires payment_gateway_url") + } + if w.evmSigner == nil { + w.evmSigner = evm.NewHostedPayer(w.cfg.PaymentGatewayURL, w.cfg.PaymentGatewayAPIKey) + } + } + // Cost ceiling — applies to wave-batch AND merkle. Wave cost is known upfront // (prepared.TotalAmount); merkle cost is the most the contract could charge - // (one winning candidate per pool). Either exceeding max_gas_fee backs off to - // a cheaper window rather than paying uncapped. Compared as big.Int so large - // atto-token amounts don't overflow. + // (one winning candidate per pool). Hosted mode compares gross — quote plus + // the gateway's per-batch network fee (V2-1113) — matching the actual debit + // and the gateway's own 402 threshold. Either exceeding max_gas_fee backs off + // to a cheaper window rather than paying uncapped. Compared as big.Int so + // large atto-token amounts don't overflow. if maxFeeStr, err := w.settingsSvc.Get("max_gas_fee"); err == nil { if maxFee, ok := new(big.Int).SetString(strings.TrimSpace(maxFeeStr), 10); ok && maxFee.Sign() > 0 { - estCost := estimatedUploadCost(prepared) + var feePerBatch *big.Int + if w.cfg.PaymentBackend.Hosted() { + if fp, ok := w.evmSigner.(interface { + FeePerBatch(context.Context) (*big.Int, bool) + }); ok { + fee, known := fp.FeePerBatch(ctx) + if !known { + // The gateway debits GROSS, but the fee could not be learned + // (never fetched successfully — gateway down since boot). A + // ceiling computed without it would pass exactly when the + // gateway might charge past it, so never spend on an + // unknown fee: back off like a too-high quote and retry the + // fetch on the next pass (review of #163). + attempt := upload.BackoffAttempt + 1 + if attempt > maxGasBackoffAttempts { + return fmt.Errorf("Gateway fee unavailable — cannot verify the cost ceiling; try again later") + } + backoffUntil := calcGasBackoff(attempt) + netCost := grossUploadCost(prepared, nil) + if err := w.uploadSvc.SetGasBackoff(upload.ID, backoffUntil, attempt, netCost.String()); err != nil { + return fmt.Errorf("Internal error scheduling retry") + } + slog.Warn("gateway fee unknown, deferring cost-ceiling check", + "uuid", upload.UUID, "net_quoted", netCost.String(), "max", maxFeeStr, + "attempt", attempt, "retry_at", backoffUntil.Format(time.RFC3339)) + return errGasBackoff + } + feePerBatch = fee + } + } + estCost := grossUploadCost(prepared, feePerBatch) if estCost.Cmp(maxFee) > 0 { attempt := upload.BackoffAttempt + 1 if attempt > maxGasBackoffAttempts { @@ -438,18 +541,20 @@ func (w *UploadWorker) processUpload(ctx context.Context, upload *services.Uploa // Cache antd's response only when our config is empty — preserves the // original "first PrepareUpload populates cfg" behaviour for installs // that rely on antd as authority. - if w.cfg.EvmRPCURL == "" && prepared.RPCUrl != "" { + if w.cfg.PaymentBackend.NeedsWallet() && w.cfg.EvmRPCURL == "" && prepared.RPCUrl != "" { w.cfg.EvmRPCURL = prepared.RPCUrl w.cfg.EvmTokenAddress = prepared.PaymentTokenAddress } - // Ensure EVM signer is connected to the resolved URL. - if w.evmSigner == nil || w.evmSigner.RPCUrl() != rpcURL { + // Ensure the payer is connected. Hosted mode (V2-929 PoC) delegates + // signing to the payment gateway and was ensured above, before the cost + // ceiling; otherwise connect the local EVM signer to the resolved URL. + if !w.cfg.PaymentBackend.Hosted() && (w.evmSigner == nil || w.evmSigner.RPCUrl() != rpcURL) { signer, err := evm.NewSigner(rpcURL) if err != nil { return fmt.Errorf("Failed to connect to EVM RPC: %w", err) } - w.evmSigner = signer + w.evmSigner = localPayer{signer} } // Optional operator override for how long we wait for a payment tx to @@ -488,7 +593,7 @@ func (w *UploadWorker) processUpload(ctx context.Context, upload *services.Uploa // Record the confirmed spend BEFORE finalize, so a finalize failure still // leaves an accounting record rather than losing the payment (V2-426). - w.recordPayment(ctx, wallet, upload, tokenAddr, paidAmount, txHash) + w.recordPayment(ctx, wallet, upload, tokenAddr, paidAmount, txHash, "", "") // Phase 3: Finalize merkle upload. A failure here means money is already // spent; re-running would submit a second merkle payment (not provably @@ -503,9 +608,10 @@ func (w *UploadWorker) processUpload(ctx context.Context, upload *services.Uploa // every chunk is already on-network (content-addressed dedup) — there's // nothing to pay, so skip signing an empty batch and finalize directly. var txHashes map[string]string + var gatewayKey string paymentMade := false if len(prepared.Payments) > 0 { - txHashes, err = w.evmSigner.PayForQuotes(ctx, walletKey, prepared.Payments, tokenAddr, prepared.PaymentVaultAddress) + txHashes, gatewayKey, err = w.evmSigner.PayForQuotes(ctx, walletKey, prepared.Payments, prepared.SignedQuotes, tokenAddr, prepared.PaymentVaultAddress) if err != nil { return fmt.Errorf("EVM payment failed: %w", err) } @@ -521,13 +627,30 @@ func (w *UploadWorker) processUpload(ctx context.Context, upload *services.Uploa } // Record the confirmed spend BEFORE finalize (V2-426). Only when a payment - // actually happened — a dedup re-Prepare pays nothing. + // actually happened — a dedup re-Prepare pays nothing. Hosted mode: the + // gateway may charge a per-batch network fee (V2-1098) — the customer's + // books record the GROSS debit (what the credits actually dropped by), + // with the fee itemized on the upload. + feeAtto := "" if paymentMade { - w.recordPayment(ctx, wallet, upload, tokenAddr, paidAmount, txHash) + if pc, ok := w.evmSigner.(interface { + PaymentCost(string) (evm.PaymentCost, bool) + }); ok && gatewayKey != "" { + if c, ok := pc.PaymentCost(gatewayKey); ok { + paidAmount = c.TotalDebited + feeAtto = c.FeeAtto + } + } + w.recordPayment(ctx, wallet, upload, tokenAddr, paidAmount, txHash, gatewayKey, feeAtto) } // Phase 3: Finalize wave-batch upload. Retrying re-Prepares at zero cost - // (dedup), so a finalize failure is safe to retry. + // (dedup), so a finalize failure is safe to retry. antd requires + // tx_hashes as an empty object — never null — when prepare reported no + // payments (full dedup), and a nil Go map marshals to null. + if txHashes == nil { + txHashes = map[string]string{} + } result, err = w.antdClient.FinalizeUpload(ctx, prepared.UploadID, txHashes, false) if err != nil { return fmt.Errorf("Failed to finalize upload: %w", errors.Join(errFinalizeFailed, err)) @@ -594,7 +717,33 @@ func (w *UploadWorker) processUpload(ctx context.Context, upload *services.Uploa // still leaves a queryable accounting record rather than losing the spend. Called // exactly once per real payment (a dedup re-Prepare pays nothing, so retries do // not double-record). -func (w *UploadWorker) recordPayment(ctx context.Context, wallet *services.Wallet, upload *services.Upload, tokenAddr, paidAmount, txHash string) { +func (w *UploadWorker) recordPayment(ctx context.Context, wallet *services.Wallet, upload *services.Upload, tokenAddr, paidAmount, txHash, gatewayKey, feeAtto string) { + // Provenance stamp (V2-1086): recorded with the payment so "how was this + // upload paid" survives instance-level payment_backend changes. feeAtto is + // the gateway's per-batch network fee when one was charged (V2-1098). + backend := string(w.cfg.PaymentBackend) + if err := w.uploadSvc.SetPaymentProvenance(upload.ID, backend, gatewayKey, feeAtto); err != nil { + slog.Warn("failed to stamp payment provenance", "error", err) + } + + if w.cfg.PaymentBackend.Hosted() { + // No wallet paid — the gateway's treasury did, debiting the tenant's + // credits. wallet_id NULL (hosted rows belong to no wallet, V-929), + // distinct tx_type, balance_after = remaining gateway credits. + creditBal := "" + if ab, ok := w.evmSigner.(interface { + AccountBalance(context.Context) (string, error) + }); ok { + if bal, err := ab.AccountBalance(ctx); err == nil { + creditBal = bal + } else { + slog.Warn("failed to query gateway credit balance", "error", err) + } + } + _, _ = w.txnSvc.Record(0, &upload.ID, "hosted_payment", paidAmount, creditBal, txHash) + return + } + if tokenBal, gasBal, err := w.evmSigner.GetBalances(ctx, wallet.Address, tokenAddr); err == nil { _ = w.walletSvc.UpdateBalance(wallet.ID, tokenBal, gasBal) _, _ = w.txnSvc.Record(wallet.ID, &upload.ID, "upload", paidAmount, tokenBal, txHash) diff --git a/internal/worker/upload_worker_test.go b/internal/worker/upload_worker_test.go index d298d4e..cfceb22 100644 --- a/internal/worker/upload_worker_test.go +++ b/internal/worker/upload_worker_test.go @@ -3,6 +3,7 @@ package worker import ( "errors" "fmt" + "math/big" "os" "path/filepath" "testing" @@ -436,6 +437,57 @@ func TestEstimatedUploadCost_Merkle(t *testing.T) { } } +// --- grossUploadCost (V2-1113 fee-aware ceiling basis) --- + +func TestGrossUploadCost_HostedAddsOneFeePerBatch(t *testing.T) { + // One prepared wave upload settles as exactly one gateway batch, so the + // gross basis is quote + one fee — exact atto, no floats. + p := &antd.PrepareUploadResult{PaymentType: "wave_batch", + TotalAmount: "35156250000000000", + Payments: []antd.PaymentInfo{{QuoteHash: "0xq1"}, {QuoteHash: "0xq2"}}} + fee := big.NewInt(0) + fee.SetString("5000000000000000", 10) + if got := grossUploadCost(p, fee); got.String() != "40156250000000000" { + t.Errorf("gross = %s, want 40156250000000000", got) + } + // The fee is per batch, not per payment: two payments, still one fee. +} + +func TestGrossUploadCost_NilFeeIsNet(t *testing.T) { + // Local mode (or unknown fee) passes nil → the V2-431 net basis, unchanged. + p := &antd.PrepareUploadResult{PaymentType: "wave_batch", + TotalAmount: "12345", Payments: []antd.PaymentInfo{{QuoteHash: "0xq1"}}} + if got := grossUploadCost(p, nil); got.String() != "12345" { + t.Errorf("nil fee = %s, want net 12345", got) + } + if got := grossUploadCost(p, new(big.Int)); got.String() != "12345" { + t.Errorf("zero fee = %s, want net 12345", got) + } +} + +func TestGrossUploadCost_FullDedupPaysNoFee(t *testing.T) { + // No payments → no gateway batch is sent → no fee, even in hosted mode. + p := &antd.PrepareUploadResult{PaymentType: "wave_batch", TotalAmount: "0"} + if got := grossUploadCost(p, big.NewInt(5)); got.Sign() != 0 { + t.Errorf("dedup gross = %s, want 0", got) + } +} + +func TestGrossUploadCost_CeilingComparesGross(t *testing.T) { + // The V2-1113 point: a quote under the ceiling whose GROSS crosses it + // must compare over — matching the gateway's own 402 threshold — while + // the same numbers in local mode (no fee) stay under. + p := &antd.PrepareUploadResult{PaymentType: "wave_batch", + TotalAmount: "100", Payments: []antd.PaymentInfo{{QuoteHash: "0xq1"}}} + maxFee := big.NewInt(105) + if grossUploadCost(p, nil).Cmp(maxFee) > 0 { + t.Error("net 100 must pass a 105 ceiling") + } + if grossUploadCost(p, big.NewInt(10)).Cmp(maxFee) <= 0 { + t.Error("gross 110 must refuse a 105 ceiling before spending") + } +} + // --- Constants --- func TestConstants(t *testing.T) { diff --git a/web/src/layouts/AppLayout.vue b/web/src/layouts/AppLayout.vue index f7d3c70..78aaa41 100644 --- a/web/src/layouts/AppLayout.vue +++ b/web/src/layouts/AppLayout.vue @@ -1,9 +1,10 @@